""" 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()