tegwick
ee472acc87
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
117 lines
5.2 KiB
Python
117 lines
5.2 KiB
Python
"""Closed JSON decoding of the existing Memo model, not a second memo schema."""
|
|
|
|
import base64
|
|
from dataclasses import MISSING, fields, is_dataclass
|
|
from enum import Enum
|
|
import hashlib
|
|
import re
|
|
from types import UnionType
|
|
from typing import get_args, get_origin, get_type_hints
|
|
|
|
from .intake import check_one_question, scan_secret_markers
|
|
from .memo import Memo
|
|
|
|
|
|
def _decode(value, model, depth=0):
|
|
if depth > 16:
|
|
raise ValueError("command nesting too deep")
|
|
origin, args = get_origin(model), get_args(model)
|
|
if origin is UnionType:
|
|
if value is None and type(None) in args:
|
|
return None
|
|
return _decode(value, next(arg for arg in args if arg is not type(None)), depth + 1)
|
|
if is_dataclass(model):
|
|
if not isinstance(value, dict):
|
|
raise ValueError("object required")
|
|
definitions = {field.name: field for field in fields(model)}
|
|
if value.keys() - definitions.keys():
|
|
raise ValueError("unknown field")
|
|
required = {name for name, field in definitions.items()
|
|
if field.default is MISSING and field.default_factory is MISSING}
|
|
if not required <= value.keys():
|
|
raise ValueError("missing field")
|
|
hints = get_type_hints(model)
|
|
return model(**{key: _decode(item, hints[key], depth + 1) for key, item in value.items()})
|
|
if origin is tuple:
|
|
if not isinstance(value, list) or len(value) > 1000:
|
|
raise ValueError("bounded array required")
|
|
return tuple(_decode(item, args[0], depth + 1) for item in value)
|
|
if isinstance(model, type) and issubclass(model, Enum):
|
|
return model(value)
|
|
if model is dict:
|
|
if not isinstance(value, dict):
|
|
raise ValueError("object required")
|
|
_json(value, depth + 1)
|
|
return value
|
|
if model in (str, bool, int):
|
|
if type(value) is not model or (model is str and len(value) > 100000):
|
|
raise ValueError("invalid field type or size")
|
|
return value
|
|
raise ValueError("unsupported model field")
|
|
|
|
|
|
def _json(value, depth):
|
|
if depth > 16:
|
|
raise ValueError("command nesting too deep")
|
|
if isinstance(value, dict):
|
|
for key, item in value.items():
|
|
if not isinstance(key, str):
|
|
raise ValueError("invalid object key")
|
|
_json(item, depth + 1)
|
|
elif isinstance(value, list):
|
|
for item in value:
|
|
_json(item, depth + 1)
|
|
elif value is not None and type(value) not in (str, int, bool):
|
|
raise ValueError("unsupported JSON value")
|
|
|
|
|
|
def decode_memo(data):
|
|
scan_secret_markers(data)
|
|
memo = _decode(data, Memo)
|
|
if not re.fullmatch(r"[A-Za-z0-9._:-]{1,256}", memo.id) or memo.version < 1:
|
|
raise ValueError("invalid memo identity")
|
|
if memo.sealed or memo.approval_id is not None or memo.approval_binding_digest is not None:
|
|
raise ValueError("unbound draft required")
|
|
check_one_question(memo.question)
|
|
for value in (memo.question, memo.requested_act, memo.brief, memo.binding.principal.id,
|
|
memo.binding.principal.kind, memo.binding.principal.display_name,
|
|
memo.binding.target.id, memo.binding.target.kind, memo.binding.target.label):
|
|
if not value.strip():
|
|
raise ValueError("nonempty text required")
|
|
for items, attribute in ((memo.packet, "item_id"), (memo.highlights, "id")):
|
|
ids = [getattr(item, attribute) for item in items]
|
|
if len(set(ids)) != len(ids) or any(not ident.strip() for ident in ids):
|
|
raise ValueError("duplicate or empty identifier")
|
|
for item in memo.packet:
|
|
if not re.fullmatch(r"sha256:[a-f0-9]{64}", item.hash):
|
|
raise ValueError("invalid packet digest")
|
|
return memo
|
|
|
|
|
|
def decode_documents(documents, memo):
|
|
"""Validate command-local content; never claim custody from a supplied hash."""
|
|
if not isinstance(documents, list) or len(documents) > 100:
|
|
raise ValueError("bounded document list required")
|
|
permitted = {item.hash for item in memo.packet}
|
|
decoded, seen = [], set()
|
|
for document in documents:
|
|
if not isinstance(document, dict) or set(document) != {"digest", "media_type", "content_base64"}:
|
|
raise ValueError("invalid document fields")
|
|
digest, media = document["digest"], document["media_type"]
|
|
if (not isinstance(digest, str) or digest not in permitted or digest in seen
|
|
or not isinstance(media, str) or not media.strip() or len(media) > 256
|
|
or not isinstance(document["content_base64"], str)):
|
|
raise ValueError("invalid document")
|
|
content = base64.b64decode(document["content_base64"], validate=True)
|
|
if len(content) > 192 * 1024 or "sha256:" + hashlib.sha256(content).hexdigest() != digest:
|
|
raise ValueError("packet content mismatch")
|
|
# UTF-8 packet text receives the same bounded secret-marker check as intake.
|
|
try:
|
|
text = content.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
text = None
|
|
if text is not None:
|
|
scan_secret_markers(text)
|
|
seen.add(digest)
|
|
decoded.append((digest, media, content))
|
|
return decoded
|