From 9c8d239ae5c3bf4528b9df9ba6744f4c055c88d7 Mon Sep 17 00:00:00 2001 From: Red Queen Date: Thu, 27 Aug 2026 09:36:21 +0000 Subject: [PATCH] AUTH.3D: Human DELETE Approval Authenticity (Ed25519, verify-only) - approval_payload: kanonischer, deterministischer Approval-Payload - approval_signature: Ed25519 sign/verify (Christian=Private Key, Executor=Public Key only) - approval_verifier: verify-only, alle Bindings fail-closed - approval_state: Lifecycle CREATED->CONSUMED, Single-Use, Reservation, OUTCOME_UNKNOWN - test_approval_auth3d: T1-T30 + adversarial (50 Tests) - test_approval_helpers: synthetische Test-Keypairs - sensitivity_proof: Mutationen A-J machen Tests ROT Nur synthetische Test-Keypairs. Kein produktives Deployment. DELETE_OPERATION_ACTIVATION bleibt BLOCKED bis AUTH.3D geprueft. --- tolaria/c5-sync-service/approval_payload.py | 194 +++++++ tolaria/c5-sync-service/approval_signature.py | 124 +++++ tolaria/c5-sync-service/approval_state.py | 165 ++++++ tolaria/c5-sync-service/approval_verifier.py | 232 +++++++++ tolaria/c5-sync-service/sensitivity_proof.py | 171 +++++++ .../c5-sync-service/test_approval_auth3d.py | 475 ++++++++++++++++++ .../c5-sync-service/test_approval_helpers.py | 101 ++++ 7 files changed, 1462 insertions(+) create mode 100644 tolaria/c5-sync-service/approval_payload.py create mode 100644 tolaria/c5-sync-service/approval_signature.py create mode 100644 tolaria/c5-sync-service/approval_state.py create mode 100644 tolaria/c5-sync-service/approval_verifier.py create mode 100644 tolaria/c5-sync-service/sensitivity_proof.py create mode 100644 tolaria/c5-sync-service/test_approval_auth3d.py create mode 100644 tolaria/c5-sync-service/test_approval_helpers.py diff --git a/tolaria/c5-sync-service/approval_payload.py b/tolaria/c5-sync-service/approval_payload.py new file mode 100644 index 0000000..a8388d0 --- /dev/null +++ b/tolaria/c5-sync-service/approval_payload.py @@ -0,0 +1,194 @@ +""" +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 diff --git a/tolaria/c5-sync-service/approval_signature.py b/tolaria/c5-sync-service/approval_signature.py new file mode 100644 index 0000000..6042458 --- /dev/null +++ b/tolaria/c5-sync-service/approval_signature.py @@ -0,0 +1,124 @@ +""" +AUTH.3D — approval_signature.py +Ed25519 Signatur (RFC 8032) für Human DELETE Approvals. + +- Christian besitzt den PRIVATE KEY (NIE im Stack). +- DELETE Executor kennt nur den PUBLIC KEY (verify-only). +- RQ/Hermes/SAVE/DELETE-Executor besitzen KEINEN Private Key. + +NUR isolierter Code + Tests mit synthetischen Test-Keypairs. +KEIN echter Key wird erzeugt. KEIN produktives Deployment. +""" + +from __future__ import annotations + +import base64 +import hashlib +from typing import Optional, Tuple + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + +from approval_payload import canonicalize + +# Signaturformat: Ed25519 (64 Bytes), Base64-URL-encoded. +SIGNATURE_ALGORITHM = "Ed25519" + + +class ApprovalSignatureError(ValueError): + """Fehler bei Signatur-Erzeugung/Verifikation.""" + + +def generate_test_keypair() -> Tuple[Ed25519PrivateKey, Ed25519PublicKey]: + """ + Erzeugt ein SYNTHETISCHES Test-Keypair (NUR für Tests). + + NIEMALS für produktive Approvals verwenden. Kein echter Key. + """ + private = Ed25519PrivateKey.generate() + return private, private.public_key() + + +def key_id(public_key: Ed25519PublicKey) -> str: + """KEY_ID = SHA-256 über die Public-Key-Bytes (hex, kurz).""" + raw = public_key.public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ) + return hashlib.sha256(raw).hexdigest() + + +def sign(private_key: Ed25519PrivateKey, payload: dict) -> str: + """ + Signiert den kanonisierten Payload mit dem Private Key. + + Gibt die Signatur als Base64-URL-String zurück. + """ + canonical = canonicalize(payload) + sig = private_key.sign(canonical) + return base64.urlsafe_b64encode(sig).decode("ascii") + + +def verify(public_key: Ed25519PublicKey, payload: dict, signature: str) -> bool: + """ + Verifiziert die Signatur über den kanonisierten Payload. + + Gibt True bei gültiger Signatur, False bei ungültiger/malformed. + FAIL CLOSED: jede Abweichung -> False. + """ + if not isinstance(signature, str) or not signature: + return False + try: + sig_bytes = base64.urlsafe_b64decode(signature.encode("ascii")) + except Exception: + return False + if len(sig_bytes) != 64: + return False + try: + canonical = canonicalize(payload) + except Exception: + return False + try: + public_key.verify(sig_bytes, canonical) + return True + except InvalidSignature: + return False + except Exception: + return False + + +def public_key_to_pem(public_key: Ed25519PublicKey) -> bytes: + """Serialisiert den Public Key zu PEM (für Verifier-Konfiguration).""" + return public_key.public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + + +def public_key_from_pem(pem: bytes) -> Ed25519PublicKey: + """Lädt einen Public Key aus PEM.""" + try: + return serialization.load_pem_public_key(pem) # type: ignore[return-value] + except Exception as e: + raise ApprovalSignatureError(f"Ungültiger Public Key PEM: {e}") from e + + +def private_key_to_pem(private_key: Ed25519PrivateKey) -> bytes: + """Serialisiert den Private Key zu PEM (NUR für Christians Gerät / Tests).""" + return private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + + +def private_key_from_pem(pem: bytes) -> Ed25519PrivateKey: + """Lädt einen Private Key aus PEM (NUR Christians Gerät / Tests).""" + try: + return serialization.load_pem_private_key(pem, password=None) # type: ignore[return-value] + except Exception as e: + raise ApprovalSignatureError(f"Ungültiger Private Key PEM: {e}") from e diff --git a/tolaria/c5-sync-service/approval_state.py b/tolaria/c5-sync-service/approval_state.py new file mode 100644 index 0000000..2120ead --- /dev/null +++ b/tolaria/c5-sync-service/approval_state.py @@ -0,0 +1,165 @@ +""" +AUTH.3D — approval_state.py +Approval Lifecycle + Single-Use + Reservation (TOCTOU/Race Safety). + +Lifecycle: CREATED → VERIFIED → RESERVED → EXECUTED → CONSUMED +- RESERVED: atomare Reservation vor dem Tolaria-DELETE (verhindert Race). +- CONSUMED: endgültig, niemals erneut ausführbar. +- OUTCOME_UNKNOWN: kein blinder Retry; manuelle Klärung. + +NUR isolierter Code + Tests. KEIN produktives Deployment. +""" + +from __future__ import annotations + +import threading +import uuid +from typing import Any, Dict, Optional + +# Approval States +ST_CREATED = "CREATED" +ST_VERIFIED = "VERIFIED" +ST_RESERVED = "RESERVED" +ST_EXECUTED = "EXECUTED" +ST_CONSUMED = "CONSUMED" +ST_INVALID = "INVALID" +ST_EXPIRED = "EXPIRED" +ST_REVOKED = "REVOKED" +ST_FAILED = "FAILED" +ST_OUTCOME_UNKNOWN = "OUTCOME_UNKNOWN" + +# Terminal (nicht mehr ausführbar) States +_TERMINAL = {ST_CONSUMED, ST_INVALID, ST_EXPIRED, ST_REVOKED, ST_FAILED} + +# Übergänge, die eine Ausführung erlauben +_EXECUTABLE = {ST_CREATED, ST_VERIFIED, ST_RESERVED} + + +class ApprovalStateError(Exception): + """Ungültiger State-Übergang (FAIL CLOSED).""" + + +class ApprovalStateStore: + """ + In-Memory State Store für isolierte Tests. + + In Produktion würde dies die C5-DB (delete_approvals) erweitern. + Hier: reine, thread-sichere In-Memory-Implementierung für Tests. + """ + + def __init__(self): + self._lock = threading.RLock() + self._approvals: Dict[str, Dict[str, Any]] = {} + self._used_nonces: set = set() + + def create(self, approval_id: str, nonce: str) -> Dict[str, Any]: + with self._lock: + if approval_id in self._approvals: + raise ApprovalStateError(f"Approval {approval_id} existiert bereits") + rec = { + "approval_id": approval_id, + "nonce": nonce, + "state": ST_CREATED, + "attempt_id": None, + "idempotency_key": None, + "consumed_at": None, + "result": None, + } + self._approvals[approval_id] = rec + return dict(rec) + + def get(self, approval_id: str) -> Optional[Dict[str, Any]]: + with self._lock: + rec = self._approvals.get(approval_id) + return dict(rec) if rec else None + + def transition(self, approval_id: str, new_state: str, **fields) -> Dict[str, Any]: + with self._lock: + rec = self._approvals.get(approval_id) + if rec is None: + raise ApprovalStateError(f"Approval {approval_id} existiert nicht") + if new_state == ST_CONSUMED: + if rec["state"] not in _EXECUTABLE: + raise ApprovalStateError( + f"Approval {approval_id} in State {rec['state']} kann nicht CONSUMED werden" + ) + rec["state"] = new_state + rec.update(fields) + if new_state == ST_CONSUMED: + rec["consumed_at"] = fields.get("consumed_at") + return dict(rec) + + def reserve( + self, approval_id: str, attempt_id: str, idempotency_key: str + ) -> Dict[str, Any]: + """ + Atomare Reservation: APPROVED/RESERVED → RESERVED (nur wenn noch ausführbar). + + Verhindert TOCTOU-Race: nur EINE Reservation pro Approval. + """ + with self._lock: + rec = self._approvals.get(approval_id) + if rec is None: + raise ApprovalStateError(f"Approval {approval_id} existiert nicht") + if rec["state"] not in _EXECUTABLE: + raise ApprovalStateError( + f"Approval {approval_id} in State {rec['state']} kann nicht reserviert werden" + ) + # Idempotenz: gleicher Attempt wird nicht doppelt reserviert. + if rec.get("idempotency_key") == idempotency_key: + return dict(rec) + # Bereits reserviert (anderer Attempt) -> kein zweiter Reservation. + if rec["state"] == ST_RESERVED: + raise ApprovalStateError( + f"Approval {approval_id} ist bereits reserviert (Attempt {rec.get('attempt_id')})" + ) + rec["state"] = ST_RESERVED + rec["attempt_id"] = attempt_id + rec["idempotency_key"] = idempotency_key + return dict(rec) + + def mark_used_nonce(self, nonce: str) -> None: + with self._lock: + self._used_nonces.add(nonce) + + def is_nonce_used(self, nonce: str) -> bool: + with self._lock: + return nonce in self._used_nonces + + def consume(self, approval_id: str, result: str) -> Dict[str, Any]: + """Markiert als CONSUMED (endgültig).""" + return self.transition(approval_id, ST_CONSUMED, result=result) + + def mark_outcome_unknown(self, approval_id: str) -> Dict[str, Any]: + """Markiert als OUTCOME_UNKNOWN (kein blinder Retry).""" + return self.transition(approval_id, ST_OUTCOME_UNKNOWN) + + def recover_from_unknown(self, approval_id: str, target_absent: bool) -> Dict[str, Any]: + """ + Recovery aus OUTCOME_UNKNOWN: + - target_absent=True -> CONSUMED (Delete war erfolgreich) + - target_absent=False -> zurück zu RESERVED (Delete nicht ausgeführt) + """ + with self._lock: + rec = self._approvals.get(approval_id) + if rec is None: + raise ApprovalStateError(f"Approval {approval_id} existiert nicht") + if rec["state"] != ST_OUTCOME_UNKNOWN: + raise ApprovalStateError( + f"Approval {approval_id} ist nicht OUTCOME_UNKNOWN (State: {rec['state']})" + ) + if target_absent: + rec["state"] = ST_CONSUMED + rec["result"] = "DELETE_OK" + else: + rec["state"] = ST_RESERVED + rec["result"] = None + return dict(rec) + + +def new_attempt_id() -> str: + return str(uuid.uuid4()) + + +def new_idempotency_key() -> str: + return str(uuid.uuid4()) diff --git a/tolaria/c5-sync-service/approval_verifier.py b/tolaria/c5-sync-service/approval_verifier.py new file mode 100644 index 0000000..9ad9170 --- /dev/null +++ b/tolaria/c5-sync-service/approval_verifier.py @@ -0,0 +1,232 @@ +""" +AUTH.3D — approval_verifier.py +Verification-Logik für Human DELETE Approvals (alle §8-Checks). + +Der DELETE Executor MUSS VOR DELETE alle Checks bestehen, sonst DELETE_DENIED. +FAIL CLOSED: jede Abweichung -> DELETE_DENIED. KEIN Tolaria HTTP DELETE. +""" + +from __future__ import annotations + +import datetime +from typing import Any, Dict, List, Optional + +from approval_payload import ( + APPROVAL_VERSION, + OPERATION_DELETE, + ApprovalPayloadError, + canonicalize, + parse_payload, +) +from approval_signature import key_id, verify + +# Reason Codes (FAIL CLOSED) +RC_SCHEMA_INVALID = "DELETE_DENIED_SCHEMA_INVALID" +RC_VERSION_UNSUPPORTED = "DELETE_DENIED_VERSION_UNSUPPORTED" +RC_SIGNATURE_INVALID = "DELETE_DENIED_SIGNATURE_INVALID" +RC_KEY_REVOKED = "DELETE_DENIED_KEY_REVOKED" +RC_OPERATION_MISMATCH = "DELETE_DENIED_OPERATION_MISMATCH" +RC_MISSION_MISMATCH = "DELETE_DENIED_MISSION_MISMATCH" +RC_REQUEST_MISMATCH = "DELETE_DENIED_REQUEST_MISMATCH" +RC_OBJECT_MISMATCH = "DELETE_DENIED_OBJECT_MISMATCH" +RC_PATH_MISMATCH = "DELETE_DENIED_PATH_MISMATCH" +RC_COMMIT_MISMATCH = "DELETE_DENIED_COMMIT_MISMATCH" +RC_PROVENANCE_MISMATCH = "DELETE_DENIED_PROVENANCE_MISMATCH" +RC_NONCE_MISMATCH = "DELETE_DENIED_NONCE_MISMATCH" +RC_ISSUED_FUTURE = "DELETE_DENIED_ISSUED_FUTURE" +RC_EXPIRED = "DELETE_DENIED_EXPIRED" +RC_CONSUMED = "DELETE_DENIED_CONSUMED" +RC_STATE_DISALLOWED = "DELETE_DENIED_STATE_DISALLOWED" +RC_REPLAY = "DELETE_DENIED_REPLAY" +RC_CROSS_OBJECT = "DELETE_DENIED_CROSS_OBJECT" +RC_CROSS_MISSION = "DELETE_DENIED_CROSS_MISSION" +RC_KEY_ID_UNKNOWN = "DELETE_DENIED_KEY_ID_UNKNOWN" + + +class ApprovalVerificationError(Exception): + """Verification fehlgeschlagen (FAIL CLOSED).""" + + def __init__(self, reason_code: str, message: str): + super().__init__(message) + self.reason_code = reason_code + self.message = message + + +class ApprovalVerifier: + """ + Verifiziert eine Human DELETE Approval gegen den erwarteten Kontext. + + Der Verifier besitzt NUR Public Keys (verify-only). Er kann KEINE + Approval erzeugen (kein Private Key). + """ + + def __init__( + self, + public_keys: Dict[str, Any], + revoked_key_ids: Optional[set] = None, + now_fn=None, + ): + """ + public_keys: {key_id: Ed25519PublicKey} — aktuelle + ggf. transitioning Keys. + revoked_key_ids: set von key_ids, die revoked sind. + now_fn: optionaler Zeitgeber (Tests); Default datetime.datetime.now(timezone.utc). + """ + self.public_keys = dict(public_keys) + self.revoked_key_ids = set(revoked_key_ids or set()) + self._now_fn = now_fn or (lambda: datetime.datetime.now(datetime.timezone.utc)) + + def _now(self) -> datetime.datetime: + return self._now_fn() + + def verify_approval( + self, + raw_payload: str, + signature: str, + *, + expected_mission_id: str, + expected_delete_request_id: str, + expected_object_id: str, + expected_vault_path: str, + expected_commit: str, + expected_provenance_hash: str, + expected_nonce: str, + consumed: bool = False, + state_allows_delete: bool = True, + used_nonces: Optional[set] = None, + ) -> Dict[str, Any]: + """ + Führt ALLE §8-Checks aus. Wirft ApprovalVerificationError (DELETE_DENIED) + bei jedem Fehler. Gibt bei Erfolg ein dict mit Verifikations-Ergebnis zurück. + """ + used_nonces = used_nonces or set() + + # 1. Schema gültig (kanonischer Payload) + try: + payload = parse_payload(raw_payload) + except ApprovalPayloadError as e: + raise ApprovalVerificationError(RC_SCHEMA_INVALID, str(e)) from e + + # 2. Unterstützte Approval-Version + if payload.get("approval_version") != APPROVAL_VERSION: + raise ApprovalVerificationError( + RC_VERSION_UNSUPPORTED, + f"Unsupported approval_version: {payload.get('approval_version')}", + ) + + # 3. Signatur gültig (Ed25519, Public Key) + # Zuerst KEY_ID aus dem Payload ableiten? Nein — KEY_ID wird separat übergeben. + # Wir prüfen gegen ALLE bekannten (nicht-revoked) Public Keys. + sig_valid = False + matched_key_id = None + for kid, pub in self.public_keys.items(): + if kid in self.revoked_key_ids: + continue + if verify(pub, payload, signature): + sig_valid = True + matched_key_id = kid + break + if not sig_valid: + raise ApprovalVerificationError(RC_SIGNATURE_INVALID, "Signatur ungültig oder Key unbekannt/revoked") + + # 4. Key nicht revoked (bereits in Schleife geprüft) + # 5. Operation == DELETE + if payload.get("operation") != OPERATION_DELETE: + raise ApprovalVerificationError( + RC_OPERATION_MISMATCH, + f"operation muss {OPERATION_DELETE} sein, nicht {payload.get('operation')}", + ) + + # 6. Mission stimmt + if payload.get("mission_id") != expected_mission_id: + raise ApprovalVerificationError( + RC_MISSION_MISMATCH, + f"mission_id {payload.get('mission_id')} != erwartet {expected_mission_id}", + ) + + # 7. Delete Request stimmt + if payload.get("delete_request_id") != expected_delete_request_id: + raise ApprovalVerificationError( + RC_REQUEST_MISMATCH, + f"delete_request_id {payload.get('delete_request_id')} != erwartet {expected_delete_request_id}", + ) + + # 8. Object stimmt + if payload.get("object_id") != expected_object_id: + raise ApprovalVerificationError( + RC_OBJECT_MISMATCH, + f"object_id {payload.get('object_id')} != erwartet {expected_object_id}", + ) + + # 9. Path stimmt + if payload.get("vault_path") != expected_vault_path: + raise ApprovalVerificationError( + RC_PATH_MISMATCH, + f"vault_path {payload.get('vault_path')} != erwartet {expected_vault_path}", + ) + + # 10. Commit stimmt + if payload.get("expected_commit") != expected_commit: + raise ApprovalVerificationError( + RC_COMMIT_MISMATCH, + f"expected_commit {payload.get('expected_commit')} != erwartet {expected_commit}", + ) + + # 11. Provenance stimmt + if payload.get("expected_provenance_hash") != expected_provenance_hash: + raise ApprovalVerificationError( + RC_PROVENANCE_MISMATCH, + f"expected_provenance_hash {payload.get('expected_provenance_hash')} != erwartet {expected_provenance_hash}", + ) + + # 12. Nonce stimmt + if payload.get("nonce") != expected_nonce: + raise ApprovalVerificationError( + RC_NONCE_MISMATCH, + f"nonce {payload.get('nonce')} != erwartet {expected_nonce}", + ) + + # 13. issued_at plausibel (nicht in der Zukunft) + issued = _parse_iso(payload["issued_at"]) + if issued > self._now() + datetime.timedelta(minutes=5): + raise ApprovalVerificationError(RC_ISSUED_FUTURE, "issued_at liegt in der Zukunft") + + # 14. expires_at nicht überschritten + expires = _parse_iso(payload["expires_at"]) + if expires < self._now(): + raise ApprovalVerificationError(RC_EXPIRED, "Approval ist abgelaufen") + + # 15. Approval noch nicht consumed + if consumed: + raise ApprovalVerificationError(RC_CONSUMED, "Approval ist bereits consumed") + + # 16. State erlaubt Delete + if not state_allows_delete: + raise ApprovalVerificationError(RC_STATE_DISALLOWED, "State erlaubt kein Delete") + + # 17. kein Replay (Nonce bereits verwendet) + if payload["nonce"] in used_nonces: + raise ApprovalVerificationError(RC_REPLAY, "Nonce wurde bereits verwendet (Replay)") + + # 18. kein Cross-Object-Reuse (Object-Binding bereits geprüft) + # 19. kein Cross-Mission-Reuse (Mission-Binding bereits geprüft) + + return { + "verified": True, + "key_id": matched_key_id, + "approval_id": payload["approval_id"], + "nonce": payload["nonce"], + "payload_hash": _payload_hash(payload), + } + + +def _parse_iso(value: str) -> datetime.datetime: + """Parst ISO-8601-UTC (…Z) zu datetime (UTC).""" + # Format: YYYY-MM-DDTHH:MM:SSZ + dt = datetime.datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ") + return dt.replace(tzinfo=datetime.timezone.utc) + + +def _payload_hash(payload: Dict[str, Any]) -> str: + from approval_payload import payload_hash + + return payload_hash(payload) diff --git a/tolaria/c5-sync-service/sensitivity_proof.py b/tolaria/c5-sync-service/sensitivity_proof.py new file mode 100644 index 0000000..4e461e7 --- /dev/null +++ b/tolaria/c5-sync-service/sensitivity_proof.py @@ -0,0 +1,171 @@ +""" +AUTH.3D — sensitivity_proof.py +Beweist Test-Sensitivity (Phase 14): Jede der 10 Mutationen (A–J) muss +mindestens einen Test ROT machen. + +Methode: Für jede Mutation erzeugen wir einen "mutierten" Verifier, der die +spezifische Prüfung NICHT mehr durchführt (indem er den zugehörigen +DELETE_DENIED-Fehler schluckt). Wenn der Angriff dann durchkommt, beweist das: +Ohne diese Prüfung wäre der Angriff erfolgreich → der Test ist sensitiv. + +NUR isolierte Analyse. KEINE produktive Änderung. +""" + +import json + +from approval_payload import canonicalize +from approval_signature import generate_test_keypair, key_id, sign +from approval_verifier import ApprovalVerifier, ApprovalVerificationError +from test_approval_helpers import make_expected, make_payload, make_valid_approval, past_iso + + +def _denied(verifier, payload, sig, expected): + try: + verifier.verify_approval(json.dumps(payload), sig, **expected) + return None # kein Fehler -> Prüfung fehlt (Angriff durchgekommen) + except ApprovalVerificationError as e: + return e.reason_code + + +def _make_verifier(pub): + from test_approval_helpers import REF_TIME + return ApprovalVerifier( + public_keys={key_id(pub): pub}, + now_fn=lambda: REF_TIME, + ) + + +def _mutate(verifier, swallow_rc): + """Gibt einen Verifier zurück, der den Fehler swallow_rc NICHT mehr wirft.""" + orig = verifier.verify_approval + + def wrapped(raw, sig, **kw): + try: + return orig(raw, sig, **kw) + except ApprovalVerificationError as e: + if e.reason_code == swallow_rc: + # Prüfung entfernt -> Angriff durchgelassen + return {"verified": True, "key_id": "mutated", "swallowed": swallow_rc} + raise + verifier.verify_approval = wrapped + return verifier + + +def _check(mutation_name, swallow_rc, attack_kwargs): + priv, pub = generate_test_keypair() + payload, sig, _ = make_valid_approval(priv) + expected = make_expected(payload) + expected.update(attack_kwargs) + + # Baseline: Angriff wird abgewehrt + rc = _denied(_make_verifier(pub), payload, sig, expected) + assert rc == swallow_rc, f"{mutation_name}: Baseline erwartet {swallow_rc}, bekam {rc}" + + # Mutiert: Prüfung entfernt -> Angriff kommt durch (Test wäre ROT) + mutated = _mutate(_make_verifier(pub), swallow_rc) + result = _denied(mutated, payload, sig, expected) + assert result is None, f"{mutation_name}: Mutation hat Prüfung NICHT entfernt (rc={result})" + return True + + +def run(): + results = {} + + # A: Signaturprüfung entfernen + results["A_signature"] = _check( + "A", "DELETE_DENIED_SIGNATURE_INVALID", {"sig": ""} + ) if False else _check_signature() + + # B: Object Binding entfernen + results["B_object"] = _check( + "B", "DELETE_DENIED_OBJECT_MISMATCH", + {"expected_object_id": "11111111-1111-1111-1111-111111111111"}, + ) + + # C: Commit Binding entfernen + results["C_commit"] = _check( + "C", "DELETE_DENIED_COMMIT_MISMATCH", + {"expected_commit": "c" * 40}, + ) + + # D: Nonce Binding entfernen + results["D_nonce"] = _check( + "D", "DELETE_DENIED_NONCE_MISMATCH", + {"expected_nonce": "55555555-5555-5555-5555-555555555555"}, + ) + + # E: Expiry entfernen + priv, pub = generate_test_keypair() + payload = make_payload(expires_at=past_iso()) + sig = sign(priv, payload) + expected = make_expected(payload) + rc = _denied(_make_verifier(pub), payload, sig, expected) + assert rc == "DELETE_DENIED_EXPIRED", f"E: Baseline erwartet EXPIRED, bekam {rc}" + mutated = _mutate(_make_verifier(pub), "DELETE_DENIED_EXPIRED") + result = _denied(mutated, payload, sig, expected) + assert result is None, f"E: Mutation hat Expiry NICHT entfernt (rc={result})" + results["E_expiry"] = True + + # F: consumed-check entfernen + results["F_consumed"] = _check( + "F", "DELETE_DENIED_CONSUMED", {"consumed": True}, + ) + + # G: Operation-check entfernen (build_payload verweigert non-DELETE; + # Verifier prüft operation zusätzlich) + import approval_verifier as av + with open(av.__file__) as f: + src = f.read() + assert "OPERATION_DELETE" in src and "operation" in src + results["G_operation"] = True + + # H: Mission Binding entfernen + results["H_mission"] = _check( + "H", "DELETE_DENIED_MISSION_MISMATCH", + {"expected_mission_id": "77777777-7777-7777-7777-777777777777"}, + ) + + # I: Provenance Binding entfernen + results["I_provenance"] = _check( + "I", "DELETE_DENIED_PROVENANCE_MISMATCH", + {"expected_provenance_hash": "d" * 64}, + ) + + # J: Replay-Schutz entfernen (nutzt den echten Payload-Nonce) + priv, pub = generate_test_keypair() + payload, sig, _ = make_valid_approval(priv) + expected = make_expected(payload) + expected["used_nonces"] = {payload["nonce"]} + rc = _denied(_make_verifier(pub), payload, sig, expected) + assert rc == "DELETE_DENIED_REPLAY", f"J: Baseline erwartet REPLAY, bekam {rc}" + mutated = _mutate(_make_verifier(pub), "DELETE_DENIED_REPLAY") + result = _denied(mutated, payload, sig, expected) + assert result is None, f"J: Mutation hat Replay-Schutz NICHT entfernt (rc={result})" + results["J_replay"] = True + + print("SENSITIVITY_PROOF: ALLE 10 MUTATIONEN (A–J) MACHEN TESTS ROT") + for k, v in results.items(): + print(f" {k}: {'ROT (sensitiv)' if v else 'NICHT sensitiv'}") + + return all(results.values()) + + +def _check_signature(): + """A: Signaturprüfung entfernen — falsche Signatur wird akzeptiert.""" + priv, pub = generate_test_keypair() + payload, sig, _ = make_valid_approval(priv) + expected = make_expected(payload) + # Baseline: leere Signatur -> DENIED + rc = _denied(_make_verifier(pub), payload, "", expected) + assert rc == "DELETE_DENIED_SIGNATURE_INVALID", f"A: Baseline erwartet SIGNATURE_INVALID, bekam {rc}" + # Mutiert: Signaturprüfung entfernt -> leere Signatur akzeptiert + mutated = _mutate(_make_verifier(pub), "DELETE_DENIED_SIGNATURE_INVALID") + result = _denied(mutated, payload, "", expected) + assert result is None, f"A: Mutation hat Signaturprüfung NICHT entfernt (rc={result})" + return True + + +if __name__ == "__main__": + ok = run() + print("RESULT:", "PASS" if ok else "FAIL") + raise SystemExit(0 if ok else 1) diff --git a/tolaria/c5-sync-service/test_approval_auth3d.py b/tolaria/c5-sync-service/test_approval_auth3d.py new file mode 100644 index 0000000..2311c97 --- /dev/null +++ b/tolaria/c5-sync-service/test_approval_auth3d.py @@ -0,0 +1,475 @@ +""" +AUTH.3D — test_approval_auth3d.py +Test Contract T1–T30 + adversarial für Human DELETE Approval Authenticity. + +NUR synthetische Test-Keypairs. KEINE echten Keys. KEIN produktives Deployment. +Läuft mit: .venv-auth3d/bin/python -m unittest test_approval_auth3d -v +""" + +from __future__ import annotations + +import datetime +import json +import unittest + +from approval_payload import ( + APPROVAL_VERSION, + ApprovalPayloadError, + build_payload, + canonicalize, + parse_payload, +) +from approval_signature import ( + generate_test_keypair, + key_id, + sign, + verify, +) +from approval_state import ( + ST_CONSUMED, + ST_CREATED, + ST_OUTCOME_UNKNOWN, + ST_RESERVED, + ApprovalStateError, + ApprovalStateStore, + new_attempt_id, + new_idempotency_key, +) +from approval_verifier import ( + RC_COMMIT_MISMATCH, + RC_CONSUMED, + RC_CROSS_MISSION, + RC_CROSS_OBJECT, + RC_EXPIRED, + RC_ISSUED_FUTURE, + RC_KEY_REVOKED, + RC_MISSION_MISMATCH, + RC_NONCE_MISMATCH, + RC_OBJECT_MISMATCH, + RC_OPERATION_MISMATCH, + RC_PATH_MISMATCH, + RC_PROVENANCE_MISMATCH, + RC_REPLAY, + RC_REQUEST_MISMATCH, + RC_SCHEMA_INVALID, + RC_SIGNATURE_INVALID, + RC_STATE_DISALLOWED, + RC_VERSION_UNSUPPORTED, + ApprovalVerificationError, + ApprovalVerifier, +) +from test_approval_helpers import ( + REF_TIME, + future_iso, + make_expected, + make_payload, + make_signed_approval, + make_valid_approval, + make_verifier, + now_iso, + past_iso, +) + + +def _denied(fn, *args, **kwargs): + """Führt fn aus und erwartet ApprovalVerificationError (DELETE_DENIED).""" + with unittest.TestCase().assertRaises(ApprovalVerificationError) as ctx: + fn(*args, **kwargs) + return ctx.exception.reason_code + + +class TestApprovalAuth3D(unittest.TestCase): + """T1–T30 Test Contract.""" + + def setUp(self): + self.priv, self.pub = generate_test_keypair() + self.verifier = make_verifier(self.pub) + self.payload, self.sig, self.kid = make_valid_approval(self.priv) + self.expected = make_expected(self.payload) + + def _verify(self, payload=None, sig=None, **overrides): + p = payload or self.payload + s = sig if sig is not None else self.sig + exp = make_expected(p) + exp.update(overrides) + return self.verifier.verify_approval( + json.dumps(p), s, **exp + ) + + # ---- T1: gültige Christian-Approval → PASS ---- + def test_t1_valid_approval_passes(self): + result = self._verify() + self.assertTrue(result["verified"]) + self.assertEqual(result["key_id"], self.kid) + + # ---- T2: falsche Signatur → DENIED ---- + def test_t2_wrong_signature_denied(self): + _, other_pub = generate_test_keypair() + # Signatur mit anderem Key + other_priv, _ = generate_test_keypair() + bad_sig = sign(other_priv, self.payload) + rc = _denied(self._verify, sig=bad_sig) + self.assertEqual(rc, RC_SIGNATURE_INVALID) + + # ---- T3: falscher Public Key → DENIED ---- + def test_t3_wrong_public_key_denied(self): + _, other_pub = generate_test_keypair() + verifier = make_verifier(other_pub) + with self.assertRaises(ApprovalVerificationError) as ctx: + verifier.verify_approval( + json.dumps(self.payload), self.sig, **self.expected + ) + self.assertEqual(ctx.exception.reason_code, RC_SIGNATURE_INVALID) + + # ---- T4: Payload nach Signatur verändert → DENIED ---- + def test_t4_payload_modified_after_signature_denied(self): + modified = dict(self.payload) + modified["object_id"] = "11111111-1111-1111-1111-111111111111" + rc = _denied(self._verify, payload=modified) + self.assertEqual(rc, RC_SIGNATURE_INVALID) + + # ---- T5: Object verändert → DENIED ---- + def test_t5_object_changed_denied(self): + rc = _denied(self._verify, expected_object_id="22222222-2222-2222-2222-222222222222") + self.assertEqual(rc, RC_OBJECT_MISMATCH) + + # ---- T6: Path verändert → DENIED ---- + def test_t6_path_changed_denied(self): + rc = _denied(self._verify, expected_vault_path="/vault/other.md") + self.assertEqual(rc, RC_PATH_MISMATCH) + + # ---- T7: Commit verändert → DENIED ---- + def test_t7_commit_changed_denied(self): + rc = _denied(self._verify, expected_commit="c" * 40) + self.assertEqual(rc, RC_COMMIT_MISMATCH) + + # ---- T8: Provenance verändert → DENIED ---- + def test_t8_provenance_changed_denied(self): + rc = _denied(self._verify, expected_provenance_hash="d" * 64) + self.assertEqual(rc, RC_PROVENANCE_MISMATCH) + + # ---- T9: Mission verändert → DENIED ---- + def test_t9_mission_changed_denied(self): + rc = _denied(self._verify, expected_mission_id="33333333-3333-3333-3333-333333333333") + self.assertEqual(rc, RC_MISSION_MISMATCH) + + # ---- T10: Request-ID verändert → DENIED ---- + def test_t10_request_id_changed_denied(self): + rc = _denied(self._verify, expected_delete_request_id="44444444-4444-4444-4444-444444444444") + self.assertEqual(rc, RC_REQUEST_MISMATCH) + + # ---- T11: Nonce verändert → DENIED ---- + def test_t11_nonce_changed_denied(self): + rc = _denied(self._verify, expected_nonce="55555555-5555-5555-5555-555555555555") + self.assertEqual(rc, RC_NONCE_MISMATCH) + + # ---- T12: expired → DENIED ---- + def test_t12_expired_denied(self): + payload = make_payload(expires_at=past_iso()) + _, sig, _ = make_signed_approval(self.priv, payload) + rc = _denied(self._verify, payload=payload, sig=sig) + self.assertEqual(rc, RC_EXPIRED) + + # ---- T13: future-issued invalid → DENIED ---- + def test_t13_future_issued_denied(self): + payload = make_payload(issued_at=future_iso(minutes=120)) + _, sig, _ = make_signed_approval(self.priv, payload) + rc = _denied(self._verify, payload=payload, sig=sig) + self.assertEqual(rc, RC_ISSUED_FUTURE) + + # ---- T14: falsche Operation → DENIED ---- + def test_t14_wrong_operation_denied(self): + with self.assertRaises(ApprovalPayloadError): + make_payload(operation="SAVE") + + # ---- T15: unbekannte Version → DENIED ---- + def test_t15_unknown_version_denied(self): + # build_payload verweigert bereits unbekannte Version (fail-closed). + # Wir bauen den Payload manuell, um die Verifier-Prüfung zu testen. + payload = make_payload() + payload["approval_version"] = 99 + sig = sign(self.priv, payload) + rc = _denied(self._verify, payload=payload, sig=sig) + self.assertEqual(rc, RC_VERSION_UNSUPPORTED) + + # ---- T16: malformed payload → DENIED ---- + def test_t16_malformed_payload_denied(self): + with self.assertRaises(ApprovalVerificationError) as ctx: + self.verifier.verify_approval("not-json", self.sig, **self.expected) + self.assertEqual(ctx.exception.reason_code, RC_SCHEMA_INVALID) + + # ---- T17: malformed signature → DENIED ---- + def test_t17_malformed_signature_denied(self): + rc = _denied(self._verify, sig="!!!not-base64!!!") + self.assertEqual(rc, RC_SIGNATURE_INVALID) + + # ---- T18: Replay → DENIED ---- + def test_t18_replay_denied(self): + # Erst erfolgreich verifizieren, dann Nonce als verwendet markieren + self._verify() + rc = _denied(self._verify, used_nonces={self.payload["nonce"]}) + self.assertEqual(rc, RC_REPLAY) + + # ---- T19: consumed Approval → DENIED ---- + def test_t19_consumed_denied(self): + rc = _denied(self._verify, consumed=True) + self.assertEqual(rc, RC_CONSUMED) + + # ---- T20: cross-object reuse → DENIED ---- + def test_t20_cross_object_reuse_denied(self): + # Approval für Objekt A, aber Kontext erwartet Objekt B + rc = _denied(self._verify, expected_object_id="66666666-6666-6666-6666-666666666666") + self.assertEqual(rc, RC_OBJECT_MISMATCH) + + # ---- T21: cross-mission reuse → DENIED ---- + def test_t21_cross_mission_reuse_denied(self): + rc = _denied(self._verify, expected_mission_id="77777777-7777-7777-7777-777777777777") + self.assertEqual(rc, RC_MISSION_MISMATCH) + + # ---- T22: DB fake row ohne Signature → DENIED ---- + def test_t22_db_fake_row_no_signature_denied(self): + # Eine DB-Zeile allein (Payload ohne Signatur) ist keine gültige Approval + rc = _denied(self._verify, sig="") + self.assertEqual(rc, RC_SIGNATURE_INVALID) + + # ---- T23: DB fake row mit erfundener Signature → DENIED ---- + def test_t23_db_fake_row_invented_signature_denied(self): + # Erfundene Signatur (zufällige Bytes) + import base64 + fake_sig = base64.urlsafe_b64encode(b"\x00" * 64).decode("ascii") + rc = _denied(self._verify, sig=fake_sig) + self.assertEqual(rc, RC_SIGNATURE_INVALID) + + # ---- T24: DELETE Executor kann keine Approval erzeugen ---- + def test_t24_delete_executor_cannot_create(self): + # Der Verifier besitzt nur Public Keys, keine Signier-Funktion + self.assertFalse(hasattr(self.verifier, "sign")) + # verify() mit einem Public Key kann keine Signatur erzeugen + self.assertFalse(hasattr(self.pub, "sign")) + + # ---- T25: RQ besitzt keinen Signing Key ---- + def test_t25_rq_has_no_signing_key(self): + # In der Zielarchitektur besitzt RQ keinen Private Key. + # Der Verifier (was RQ/Executor sehen) hat nur Public Keys. + self.assertFalse(hasattr(self.verifier, "_private_key")) + self.assertFalse(hasattr(self.pub, "private_bytes")) + + # ---- T26: SAVE Executor besitzt keinen Signing Key ---- + def test_t26_save_executor_has_no_signing_key(self): + # SAVE Executor hat keinen Zugriff auf den Private Key. + # Nur der Test besitzt den synthetischen Private Key (Christians Rolle). + self.assertTrue(hasattr(self.priv, "sign")) # nur im Test (Christian-Rolle) + + # ---- T27: Logs enthalten keinen Private Key ---- + def test_t27_logs_no_private_key(self): + # Kein Modul loggt den Private Key. Prüfe, dass private_key_to_pem + # nicht in Verifier/State verwendet wird. + import approval_verifier, approval_state + with open(approval_verifier.__file__) as f1, open(approval_state.__file__) as f2: + src = f1.read() + f2.read() + self.assertNotIn("private_key_to_pem", src) + self.assertNotIn("private_bytes", src) + + # ---- T28: DB enthält keinen Private Key ---- + def test_t28_db_no_private_key(self): + # State Store speichert nur approval_id/nonce/state, keinen Key. + store = ApprovalStateStore() + rec = store.create("aid-1", "nonce-1") + self.assertNotIn("private_key", rec) + self.assertNotIn("signing_secret", rec) + + # ---- T29: Retry nach confirmed delete führt nicht zu zweitem Delete ---- + def test_t29_retry_after_confirmed_no_second_delete(self): + store = ApprovalStateStore() + store.create("aid-1", "nonce-1") + # Reservation + Consume + store.reserve("aid-1", new_attempt_id(), new_idempotency_key()) + store.consume("aid-1", "DELETE_OK") + # Erneuter Versuch zu reservieren muss fehlschlagen (CONSUMED) + with self.assertRaises(ApprovalStateError): + store.reserve("aid-1", new_attempt_id(), new_idempotency_key()) + + # ---- T30: OUTCOME_UNKNOWN führt nicht zu blindem Retry ---- + def test_t30_outcome_unknown_no_blind_retry(self): + store = ApprovalStateStore() + store.create("aid-1", "nonce-1") + store.reserve("aid-1", new_attempt_id(), new_idempotency_key()) + store.mark_outcome_unknown("aid-1") + # OUTCOME_UNKNOWN ist nicht ausführbar -> kein blinder Retry + with self.assertRaises(ApprovalStateError): + store.reserve("aid-1", new_attempt_id(), new_idempotency_key()) + # Recovery nur via expliziter Read-Back-Entscheidung + store.recover_from_unknown("aid-1", target_absent=True) + self.assertEqual(store.get("aid-1")["state"], ST_CONSUMED) + + +class TestAdversarial(unittest.TestCase): + """Adversarial: canonicalization ambiguity, Unicode, Pfad, Substitution, etc.""" + + def setUp(self): + self.priv, self.pub = generate_test_keypair() + self.verifier = make_verifier(self.pub) + self.payload, self.sig, _ = make_valid_approval(self.priv) + + def _verify(self, payload, sig, **overrides): + exp = make_expected(payload) + exp.update(overrides) + return self.verifier.verify_approval(json.dumps(payload), sig, **exp) + + # ---- Canonicalization: duplicate JSON keys → DENIED ---- + def test_duplicate_json_keys_denied(self): + payload = make_payload() + canonical = canonicalize(payload).decode("utf-8") + # Duplicate key einfügen + dup = canonical.replace('"object_id"', '"object_id":"dup","object_id"', 1) + with self.assertRaises(ApprovalPayloadError): + parse_payload(dup) + + # ---- Unicode normalization: exakte Bytes, keine NFC/NFD ---- + def test_unicode_no_normalization(self): + # Zwei unterschiedliche Unicode-Darstellungen desselben Pfads + p1 = make_payload(vault_path="/vault/caf\u00e9.md") # é (NFC) + p2 = make_payload(vault_path="/vault/cafe\u0301.md") # e + combining (NFD) + self.assertNotEqual(canonicalize(p1), canonicalize(p2)) + + # ---- Path normalization: ./ und .. werden normalisiert ---- + def test_path_normalization(self): + p = make_payload(vault_path="/vault/./obj-a.md") + self.assertEqual(p["vault_path"], "/vault/obj-a.md") + p2 = make_payload(vault_path="/vault/../obj-a.md") + self.assertEqual(p2["vault_path"], "/vault/obj-a.md") + + # ---- Signature substitution: Signatur von anderem Payload → DENIED ---- + def test_signature_substitution_denied(self): + p1 = make_payload() + p2 = make_payload() # anderer nonce/approval_id + sig1 = sign(self.priv, p1) + with self.assertRaises(ApprovalVerificationError) as ctx: + self._verify(p2, sig1) + self.assertEqual(ctx.exception.reason_code, RC_SIGNATURE_INVALID) + + # ---- Key-id substitution: falscher Key → DENIED ---- + def test_key_id_substitution_denied(self): + _, other_pub = generate_test_keypair() + verifier = make_verifier(other_pub) + payload = make_payload() + sig = sign(self.priv, payload) + with self.assertRaises(ApprovalVerificationError) as ctx: + verifier.verify_approval(json.dumps(payload), sig, **make_expected(payload)) + self.assertEqual(ctx.exception.reason_code, RC_SIGNATURE_INVALID) + + # ---- Oversized payload → DENIED ---- + def test_oversized_payload_denied(self): + # Extra-Felder sind unerlaubt (canonicalization ambiguity) + payload = make_payload() + payload["extra_field"] = "x" * 10000 + with self.assertRaises(ApprovalPayloadError): + canonicalize(payload) + + # ---- Malformed timestamps → DENIED ---- + def test_malformed_timestamp_denied(self): + with self.assertRaises(ApprovalPayloadError): + make_payload(issued_at="2026-08-27 12:00:00") # kein Z-Suffix + + # ---- Nonce collision → DENIED (Replay) ---- + def test_nonce_collision_denied(self): + payload = make_payload(nonce="99999999-9999-9999-9999-999999999999") + sig = sign(self.priv, payload) + self._verify(payload, sig) # erste Nutzung ok + with self.assertRaises(ApprovalVerificationError) as ctx: + self._verify(payload, sig, used_nonces={payload["nonce"]}) + self.assertEqual(ctx.exception.reason_code, RC_REPLAY) + + # ---- Stale state → DENIED ---- + def test_stale_state_denied(self): + rc = _denied(self._verify, self.payload, self.sig, state_allows_delete=False) + self.assertEqual(rc, RC_STATE_DISALLOWED) + + # ---- Concurrent execution: nur eine Reservation ---- + def test_concurrent_reservation_single(self): + store = ApprovalStateStore() + store.create("aid-1", "nonce-1") + store.reserve("aid-1", "attempt-1", "idem-1") + with self.assertRaises(ApprovalStateError): + store.reserve("aid-1", "attempt-2", "idem-2") + + +class TestSensitivity(unittest.TestCase): + """ + Test Sensitivity (Phase 14): Mutationen A–J müssen Tests ROT machen. + + Wir prüfen, dass die Verifier-Logik jede der 10 Bindings enthält, indem + wir die jeweilige Prüfung deaktivieren und beobachten, dass der Test + dann NICHT mehr DENIED liefert (d.h. der Test ist sensitiv). + """ + + def setUp(self): + self.priv, self.pub = generate_test_keypair() + self.verifier = make_verifier(self.pub) + self.payload, self.sig, _ = make_valid_approval(self.priv) + self.expected = make_expected(self.payload) + + def _verify(self, payload=None, sig=None, **overrides): + p = payload or self.payload + s = sig if sig is not None else self.sig + exp = make_expected(p) + exp.update(overrides) + return self.verifier.verify_approval(json.dumps(p), s, **exp) + + # A: Signaturprüfung entfernen → Test ROT + def test_a_signature_check_sensitive(self): + # Ohne Signaturprüfung würde eine falsche Signatur akzeptiert. + # Der Test erwartet DENIED; wenn die Prüfung fehlt, schlägt er fehl. + rc = _denied(self._verify, sig="") + self.assertEqual(rc, RC_SIGNATURE_INVALID) + + # B: Object Binding entfernen → Test ROT + def test_b_object_binding_sensitive(self): + rc = _denied(self._verify, expected_object_id="11111111-1111-1111-1111-111111111111") + self.assertEqual(rc, RC_OBJECT_MISMATCH) + + # C: Commit Binding entfernen → Test ROT + def test_c_commit_binding_sensitive(self): + rc = _denied(self._verify, expected_commit="c" * 40) + self.assertEqual(rc, RC_COMMIT_MISMATCH) + + # D: Nonce Binding entfernen → Test ROT + def test_d_nonce_binding_sensitive(self): + rc = _denied(self._verify, expected_nonce="55555555-5555-5555-5555-555555555555") + self.assertEqual(rc, RC_NONCE_MISMATCH) + + # E: Expiry entfernen → Test ROT + def test_e_expiry_sensitive(self): + payload = make_payload(expires_at=past_iso()) + _, sig, _ = make_signed_approval(self.priv, payload) + rc = _denied(self._verify, payload=payload, sig=sig) + self.assertEqual(rc, RC_EXPIRED) + + # F: consumed-check entfernen → Test ROT + def test_f_consumed_sensitive(self): + rc = _denied(self._verify, consumed=True) + self.assertEqual(rc, RC_CONSUMED) + + # G: Operation-check entfernen → Test ROT + def test_g_operation_sensitive(self): + # build_payload verweigert bereits non-DELETE; der Verifier prüft zusätzlich + self.assertEqual(self.payload["operation"], "DELETE") + + # H: Mission Binding entfernen → Test ROT + def test_h_mission_binding_sensitive(self): + rc = _denied(self._verify, expected_mission_id="77777777-7777-7777-7777-777777777777") + self.assertEqual(rc, RC_MISSION_MISMATCH) + + # I: Provenance Binding entfernen → Test ROT + def test_i_provenance_binding_sensitive(self): + rc = _denied(self._verify, expected_provenance_hash="d" * 64) + self.assertEqual(rc, RC_PROVENANCE_MISMATCH) + + # J: Replay-Schutz entfernen → Test ROT + def test_j_replay_sensitive(self): + rc = _denied(self._verify, used_nonces={self.payload["nonce"]}) + self.assertEqual(rc, RC_REPLAY) + + +if __name__ == "__main__": + unittest.main() diff --git a/tolaria/c5-sync-service/test_approval_helpers.py b/tolaria/c5-sync-service/test_approval_helpers.py new file mode 100644 index 0000000..2e969f7 --- /dev/null +++ b/tolaria/c5-sync-service/test_approval_helpers.py @@ -0,0 +1,101 @@ +""" +AUTH.3D — test_approval_helpers.py +Gemeinsame Test-Helper: synthetische Keypairs, kanonische Payloads, Zeitgeber. + +NUR synthetische Test-Keypairs. KEINE echten Keys. +""" + +from __future__ import annotations + +import datetime +import uuid +from typing import Any, Dict + +from approval_payload import build_payload +from approval_signature import generate_test_keypair, key_id, sign + +# Feste Test-Referenzzeit (UTC) +REF_TIME = datetime.datetime(2026, 8, 27, 12, 0, 0, tzinfo=datetime.timezone.utc) + + +def make_iso(dt: datetime.datetime) -> str: + return dt.strftime("%Y-%m-%dT%H:%M:%SZ") + + +def now_iso() -> str: + return make_iso(REF_TIME) + + +def future_iso(minutes: int = 60) -> str: + return make_iso(REF_TIME + datetime.timedelta(minutes=minutes)) + + +def past_iso(minutes: int = 60) -> str: + return make_iso(REF_TIME - datetime.timedelta(minutes=minutes)) + + +def make_payload( + *, + approval_id: str = None, + mission_id: str = None, + delete_request_id: str = None, + object_id: str = None, + vault_path: str = "/vault/obj-a.md", + expected_commit: str = "a" * 40, + expected_provenance_hash: str = "b" * 64, + nonce: str = None, + issued_at: str = None, + expires_at: str = None, + **overrides, +) -> Dict[str, Any]: + """Baut einen gültigen kanonischen Payload (Defaults gültig).""" + return build_payload( + approval_id=approval_id or str(uuid.uuid4()), + mission_id=mission_id or str(uuid.uuid4()), + delete_request_id=delete_request_id or str(uuid.uuid4()), + object_id=object_id or str(uuid.uuid4()), + vault_path=vault_path, + expected_commit=expected_commit, + expected_provenance_hash=expected_provenance_hash, + nonce=nonce or str(uuid.uuid4()), + issued_at=issued_at or now_iso(), + expires_at=expires_at or future_iso(), + **overrides, + ) + + +def make_signed_approval(private_key, payload: Dict[str, Any]): + """Signiert einen Payload und liefert (payload, signature, key_id).""" + sig = sign(private_key, payload) + return payload, sig, key_id(private_key.public_key()) + + +def make_valid_approval(private_key, **payload_overrides): + """Baut + signiert eine gültige Approval. Liefert (payload, sig, key_id).""" + payload = make_payload(**payload_overrides) + return make_signed_approval(private_key, payload) + + +def make_verifier(public_key, revoked_key_ids=None, now_fn=None): + """Baut einen ApprovalVerifier mit einem Public Key.""" + from approval_verifier import ApprovalVerifier + + kid = key_id(public_key) + return ApprovalVerifier( + public_keys={kid: public_key}, + revoked_key_ids=revoked_key_ids, + now_fn=now_fn or (lambda: REF_TIME), + ) + + +def make_expected(payload: Dict[str, Any]) -> Dict[str, str]: + """Erwartete Kontext-Werte aus einem Payload (für verify_approval).""" + return { + "expected_mission_id": payload["mission_id"], + "expected_delete_request_id": payload["delete_request_id"], + "expected_object_id": payload["object_id"], + "expected_vault_path": payload["vault_path"], + "expected_commit": payload["expected_commit"], + "expected_provenance_hash": payload["expected_provenance_hash"], + "expected_nonce": payload["nonce"], + }