Skip to content

Serialization

serialization

Serialization layer for encrypting session state and event data.

Converts Python dictionaries and JSON strings into self-describing encrypted envelopes using any EncryptionBackend-conformant backend. Each envelope carries a version byte and backend identifier prefix to support future backend migrations. The BACKEND_REGISTRY maps recognized backend IDs to human-readable names and is used by _parse_envelope() for validation. Registered backends: Fernet (0x01) and AES-GCM (0x02).

The layer is stateless — four async module-level functions, no classes. The encryption backend is passed per call.

Examples:

Encrypt and decrypt a session state dictionary:

from adk_secure_sessions.serialization import (
    encrypt_session,
    decrypt_session,
    BACKEND_FERNET,
)

envelope = await encrypt_session(state, backend, BACKEND_FERNET)
restored = await decrypt_session(envelope, backend)
See Also

adk_secure_sessions.protocols: Encryption backend protocol that backends must conform to.

ENVELOPE_VERSION_1 module-attribute 

ENVELOPE_VERSION_1: int = 1

Current envelope format version byte.

BACKEND_FERNET module-attribute 

BACKEND_FERNET: int = 1

Backend identifier for Fernet encryption.

BACKEND_AES_GCM module-attribute 

BACKEND_AES_GCM: int = 2

Backend identifier for AES-256-GCM encryption.

BACKEND_REGISTRY module-attribute 

BACKEND_REGISTRY: dict[int, str] = {
    BACKEND_FERNET: "Fernet",
    BACKEND_AES_GCM: "AES-GCM",
}

Mapping of supported backend IDs to human-readable names.

Used by _parse_envelope() to validate incoming envelopes and generate error messages for unrecognized backends.

encrypt_session async 

encrypt_session(
    data: dict[str, Any],
    backend: EncryptionBackend,
    backend_id: int,
) -> bytes

Serialize a session state dict to an encrypted envelope.

PARAMETER DESCRIPTION
data

JSON-serializable Python dictionary.

TYPE: dict[str, Any]

backend

Any EncryptionBackend-conformant object.

TYPE: EncryptionBackend

backend_id

Integer identifying the backend.

TYPE: int

RETURNS DESCRIPTION
bytes

Encrypted envelope bytes: [version][backend_id][ciphertext].
RAISES DESCRIPTION
SerializationError

If data cannot be serialized to JSON.

Examples:

envelope = await encrypt_session(
    {"ssn": "123-45-6789"}, backend, BACKEND_FERNET
)
Source code in src/adk_secure_sessions/serialization.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
async def encrypt_session(
    data: dict[str, Any],
    backend: EncryptionBackend,
    backend_id: int,
) -> bytes:
    """Serialize a session state dict to an encrypted envelope.

    Args:
        data: JSON-serializable Python dictionary.
        backend: Any ``EncryptionBackend``-conformant object.
        backend_id: Integer identifying the backend.

    Returns:
        Encrypted envelope bytes: ``[version][backend_id][ciphertext]``.

    Raises:
        SerializationError: If *data* cannot be serialized to JSON.

    Examples:
        ```python
        envelope = await encrypt_session(
            {"ssn": "123-45-6789"}, backend, BACKEND_FERNET
        )
        ```
    """
    try:
        plaintext = json.dumps(data).encode()
    except (TypeError, ValueError) as exc:
        msg = "Failed to serialize session data to JSON"
        raise SerializationError(msg) from exc
    ciphertext = await backend.encrypt(plaintext)
    return _build_envelope(ENVELOPE_VERSION_1, backend_id, ciphertext)

decrypt_session async 

decrypt_session(
    envelope: bytes, backend: EncryptionBackend
) -> dict[str, Any]

Decrypt an encrypted envelope back to a session state dict.

PARAMETER DESCRIPTION
envelope

Encrypted envelope bytes (>= 3 bytes).

TYPE: bytes

backend

Any EncryptionBackend-conformant object.

TYPE: EncryptionBackend

RETURNS DESCRIPTION
dict[str, Any]

Original Python dictionary.
RAISES DESCRIPTION
DecryptionError

If envelope is invalid, tampered, or backend fails.
SerializationError

If decrypted bytes are not valid JSON.

Examples:

state = await decrypt_session(envelope, backend)
Source code in src/adk_secure_sessions/serialization.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def decrypt_session(
    envelope: bytes,
    backend: EncryptionBackend,
) -> dict[str, Any]:
    """Decrypt an encrypted envelope back to a session state dict.

    Args:
        envelope: Encrypted envelope bytes (>= 3 bytes).
        backend: Any ``EncryptionBackend``-conformant object.

    Returns:
        Original Python dictionary.

    Raises:
        DecryptionError: If envelope is invalid, tampered, or backend fails.
        SerializationError: If decrypted bytes are not valid JSON.

    Examples:
        ```python
        state = await decrypt_session(envelope, backend)
        ```
    """
    _version, _backend_id, ciphertext = _parse_envelope(envelope)
    plaintext = await backend.decrypt(ciphertext)
    try:
        data = json.loads(plaintext)
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        msg = "Failed to deserialize decrypted data from JSON"
        raise SerializationError(msg) from exc
    return data

encrypt_json async 

encrypt_json(
    json_str: str,
    backend: EncryptionBackend,
    backend_id: int,
) -> bytes

Encrypt a pre-serialized JSON string into an encrypted envelope.

PARAMETER DESCRIPTION
json_str

Valid JSON string (e.g., from model_dump_json()).

TYPE: str

backend

Any EncryptionBackend-conformant object.

TYPE: EncryptionBackend

backend_id

Integer identifying the backend.

TYPE: int

RETURNS DESCRIPTION
bytes

Encrypted envelope bytes: [version][backend_id][ciphertext].

Examples:

envelope = await encrypt_json(event.model_dump_json(), backend, BACKEND_FERNET)
Source code in src/adk_secure_sessions/serialization.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
async def encrypt_json(
    json_str: str,
    backend: EncryptionBackend,
    backend_id: int,
) -> bytes:
    """Encrypt a pre-serialized JSON string into an encrypted envelope.

    Args:
        json_str: Valid JSON string (e.g., from ``model_dump_json()``).
        backend: Any ``EncryptionBackend``-conformant object.
        backend_id: Integer identifying the backend.

    Returns:
        Encrypted envelope bytes: ``[version][backend_id][ciphertext]``.

    Examples:
        ```python
        envelope = await encrypt_json(event.model_dump_json(), backend, BACKEND_FERNET)
        ```
    """
    plaintext = json_str.encode("utf-8")
    ciphertext = await backend.encrypt(plaintext)
    return _build_envelope(ENVELOPE_VERSION_1, backend_id, ciphertext)

decrypt_json async 

decrypt_json(
    envelope: bytes, backend: EncryptionBackend
) -> str

Decrypt an encrypted envelope back to a JSON string.

PARAMETER DESCRIPTION
envelope

Encrypted envelope bytes (>= 3 bytes).

TYPE: bytes

backend

Any EncryptionBackend-conformant object.

TYPE: EncryptionBackend

RETURNS DESCRIPTION
str

Original JSON string.
RAISES DESCRIPTION
DecryptionError

If envelope is invalid, tampered, backend fails, or decrypted bytes are not valid UTF-8.

Examples:

json_str = await decrypt_json(envelope, backend)
Source code in src/adk_secure_sessions/serialization.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
async def decrypt_json(
    envelope: bytes,
    backend: EncryptionBackend,
) -> str:
    """Decrypt an encrypted envelope back to a JSON string.

    Args:
        envelope: Encrypted envelope bytes (>= 3 bytes).
        backend: Any ``EncryptionBackend``-conformant object.

    Returns:
        Original JSON string.

    Raises:
        DecryptionError: If envelope is invalid, tampered, backend fails,
            or decrypted bytes are not valid UTF-8.

    Examples:
        ```python
        json_str = await decrypt_json(envelope, backend)
        ```
    """
    _version, _backend_id, ciphertext = _parse_envelope(envelope)
    plaintext = await backend.decrypt(ciphertext)
    try:
        return plaintext.decode("utf-8")
    except UnicodeDecodeError as exc:
        msg = "Failed to decode decrypted data as UTF-8"
        raise DecryptionError(msg) from exc