""" AUTH.3D — approval_payload.py Kanonischer Human DELETE Approval Payload. Deterministische Canonicalization: feste Feldreihenfolge, keine Duplikate, keine Whitespace-Varianz, UTF-8, keine Unicode-Normalisierung (exakte Bytes), Pfad-Normalisierung. Keine Signatur über frei formatierte Texte. NUR isolierter Code + Tests. KEIN produktives Deployment. """ from __future__ import annotations import hashlib import json import re from typing import Any, Dict, List, Optional APPROVAL_VERSION = 1 OPERATION_DELETE = "DELETE" # Feste Feldreihenfolge für die Canonicalization (deterministisch). _CANONICAL_FIELDS: List[str] = [ "approval_version", "approval_id", "mission_id", "delete_request_id", "operation", "object_id", "vault_path", "expected_commit", "expected_provenance_hash", "nonce", "issued_at", "expires_at", ] _REQUIRED_FIELDS: set = set(_CANONICAL_FIELDS) # Pfad-Normalisierung: keine doppelten Slashes, kein "./", kein "..", kein trailing slash. _DOUBLE_SLASH = re.compile(r"//+") _DOT_SEGMENT = re.compile(r"(^|/)\.(/|$)") _DOTDOT_SEGMENT = re.compile(r"(^|/)\.\.(/|$)") class ApprovalPayloadError(ValueError): """Fehler bei Payload-Erzeugung/Canonicalization.""" def _normalize_path(path: str) -> str: """Normalisiert einen Vault-Pfad deterministisch (keine Varianz).""" if not isinstance(path, str) or not path: raise ApprovalPayloadError("vault_path muss ein nicht-leerer String sein") p = path.strip() if not p.startswith("/"): p = "/" + p p = _DOUBLE_SLASH.sub("/", p) p = _DOT_SEGMENT.sub("/", p) p = _DOTDOT_SEGMENT.sub("/", p) # trailing slash entfernen (ausser Root) if len(p) > 1 and p.endswith("/"): p = p.rstrip("/") return p def _validate_iso8601_utc(value: str, field: str) -> str: """Validiert ISO-8601-UTC (z.B. 2026-08-27T09:00:00Z).""" if not isinstance(value, str) or not value: raise ApprovalPayloadError(f"{field} muss ein ISO-8601-UTC-String sein") # Einfache, strikte Form: YYYY-MM-DDTHH:MM:SSZ if not re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$", value): raise ApprovalPayloadError(f"{field} muss ISO-8601-UTC (…Z) sein: {value!r}") return value def _validate_hex(value: str, field: str, length: int) -> str: if not isinstance(value, str) or not re.fullmatch(r"[0-9a-fA-F]{" + str(length) + r"}", value): raise ApprovalPayloadError(f"{field} muss ein {length}-Zeichen-Hex-String sein") return value.lower() def _validate_uuid(value: str, field: str) -> str: if not isinstance(value, str) or not re.fullmatch( r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", value ): raise ApprovalPayloadError(f"{field} muss eine UUID sein") return value.lower() def build_payload( *, approval_id: str, mission_id: str, delete_request_id: str, object_id: str, vault_path: str, expected_commit: str, expected_provenance_hash: str, nonce: str, issued_at: str, expires_at: str, approval_version: int = APPROVAL_VERSION, operation: str = OPERATION_DELETE, ) -> Dict[str, Any]: """ Baut und validiert einen kanonischen Approval Payload. Wirft ApprovalPayloadError bei ungültigen Feldern. Gibt ein dict mit exakt den 12 kanonischen Feldern zurück (keine optionalen Felder). """ if approval_version != APPROVAL_VERSION: raise ApprovalPayloadError(f"Unbekannte approval_version: {approval_version}") if operation != OPERATION_DELETE: raise ApprovalPayloadError(f"operation muss {OPERATION_DELETE} sein, nicht {operation!r}") payload: Dict[str, Any] = { "approval_version": approval_version, "approval_id": _validate_uuid(approval_id, "approval_id"), "mission_id": _validate_uuid(mission_id, "mission_id"), "delete_request_id": _validate_uuid(delete_request_id, "delete_request_id"), "operation": operation, "object_id": _validate_uuid(object_id, "object_id"), "vault_path": _normalize_path(vault_path), "expected_commit": _validate_hex(expected_commit, "expected_commit", 40), "expected_provenance_hash": _validate_hex(expected_provenance_hash, "expected_provenance_hash", 64), "nonce": _validate_uuid(nonce, "nonce"), "issued_at": _validate_iso8601_utc(issued_at, "issued_at"), "expires_at": _validate_iso8601_utc(expires_at, "expires_at"), } return payload def canonicalize(payload: Dict[str, Any]) -> bytes: """ Deterministische Canonicalization des Payloads zu Bytes. - Feste Feldreihenfolge (_CANONICAL_FIELDS) - Keine Duplikate, keine optionalen Felder - UTF-8, keine Whitespace-Varianz (compact separators) - Keine Unicode-Normalisierung (exakte Bytes) """ if not isinstance(payload, dict): raise ApprovalPayloadError("Payload muss ein dict sein") # Nur die kanonischen Felder, in fester Reihenfolge. ordered: Dict[str, Any] = {} for field in _CANONICAL_FIELDS: if field not in payload: raise ApprovalPayloadError(f"Payload fehlt Pflichtfeld: {field}") ordered[field] = payload[field] # Keine zusätzlichen Felder erlauben (verhindert canonicalization ambiguity). extra = set(payload.keys()) - set(_CANONICAL_FIELDS) if extra: raise ApprovalPayloadError(f"Payload hat unerlaubte Felder: {sorted(extra)}") # compact separators -> keine Whitespace-Varianz; sort_keys=False (Reihenfolge fix). text = json.dumps(ordered, ensure_ascii=True, separators=(",", ":"), sort_keys=False) return text.encode("utf-8") def payload_hash(payload: Dict[str, Any]) -> str: """SHA-256 über die kanonisierten Payload-Bytes (hex).""" return hashlib.sha256(canonicalize(payload)).hexdigest() def parse_payload(raw: str) -> Dict[str, Any]: """ Parst einen JSON-String in einen Payload und validiert ihn. - Duplicate JSON keys -> Fehler (verhindert canonicalization ambiguity) - Malformed JSON -> Fehler """ if not isinstance(raw, str) or not raw.strip(): raise ApprovalPayloadError("Payload muss ein nicht-leerer JSON-String sein") try: obj = json.loads(raw, object_pairs_hook=_reject_duplicate_keys) except json.JSONDecodeError as e: raise ApprovalPayloadError(f"Malformed JSON: {e}") from e if not isinstance(obj, dict): raise ApprovalPayloadError("Payload muss ein JSON-Objekt sein") # Validierung + Canonicalization (wirft bei ungültigen Feldern). canonicalize(obj) return obj def _reject_duplicate_keys(pairs: List[tuple]) -> Dict[str, Any]: """Reject duplicate JSON keys (canonicalization ambiguity).""" d: Dict[str, Any] = {} for k, v in pairs: if k in d: raise ApprovalPayloadError(f"Duplicate JSON key: {k}") d[k] = v return d