trading-system-docs/tolaria/c5-sync-service/test_approval_auth3d.py
Red Queen 9c8d239ae5 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.
2026-08-27 09:36:21 +00:00

475 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
AUTH.3D — test_approval_auth3d.py
Test Contract T1T30 + 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):
"""T1T30 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 AJ 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()