|
1
|
"""Base64 JSON helpers used by the protocol-shaped local transport.""" |
|
2
|
|
|
3
|
from __future__ import annotations |
|
4
|
|
|
5
|
import base64 |
|
6
|
import json |
|
7
|
from typing import Any |
|
8
|
|
|
9
|
from pydantic import BaseModel |
|
10
|
|
|
11
|
from .errors import ProtocolError |
|
12
|
|
|
13
|
|
|
14
|
def encode_model(model: BaseModel) -> str: |
|
15
|
"""Serialize a model to standard Base64-encoded JSON.""" |
|
16
|
|
|
17
|
payload = model.model_dump_json(by_alias=True, exclude_none=True) |
|
18
|
return base64.b64encode(payload.encode("utf-8")).decode("ascii") |
|
19
|
|
|
20
|
|
|
21
|
def decode_json(value: str, *, header_name: str) -> dict[str, Any]: |
|
22
|
"""Decode a Base64 JSON header without leaking its raw value in errors.""" |
|
23
|
|
|
24
|
try: |
|
25
|
decoded = base64.b64decode(value, validate=True) |
|
26
|
payload = json.loads(decoded) |
|
27
|
except (ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc: |
|
28
|
raise ProtocolError( |
|
29
|
"malformed_protocol_header", |
|
30
|
f"{header_name} is not valid Base64-encoded JSON.", |
|
31
|
) from exc |
|
32
|
|
|
33
|
if not isinstance(payload, dict): |
|
34
|
raise ProtocolError( |
|
35
|
"malformed_protocol_header", |
|
36
|
f"{header_name} must decode to a JSON object.", |
|
37
|
) |
|
38
|
return payload |