""" AUTH.4C3 — test_adversarial.py =============================== Adversarial-Tests für den OUTCOME_UNKNOWN-RECONCILIATION-CONTRACT. Ziel: Manipulationen und Angriffe auf den Reconciliation-Pfad müssen FAIL CLOSED sein. Kein Angriff darf zu einem falschen SUCCEEDED oder zu einem SAVE führen. Exit 0 = alle PASS. """ from __future__ import annotations import hashlib import json import os import shutil import sys import tempfile import unittest import uuid sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from job_schema import make_save_job, JobRejectedError from job_store import JobStore, JobImmutableFieldError from job_state_machine import ( ST_OUTCOME_UNKNOWN, ST_SUCCEEDED, ST_CLAIMED, ST_EXECUTING, ST_RECONCILED, ST_FAILED, ST_REJECTED, ) from save_reconciliation import ( SaveReconciliationCore, ReconciliationError, RC_TARGET_ABSENT, RC_TARGET_MISMATCH, RC_READ_UNAVAILABLE, RC_PROVENANCE_MISMATCH, ) def _sha256(s: str) -> str: return hashlib.sha256(s.encode("utf-8")).hexdigest() class AdversarialTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="c5-4c3-adv-") self.db = os.path.join(self.tmp, "adv.db") self.store = JobStore(self.db, "SAVE") self.content = "---\nid: object/adv-1\ntitle: Adv\n---\n\nContent\n" self.prov = _sha256(self.content) self.commit = "a" * 40 self.object_id = "object/adv-1" self.vault_path = "tolaria/adv.md" def tearDown(self): self.store.close() shutil.rmtree(self.tmp, ignore_errors=True) def _make_job(self, **overrides): job = make_save_job( job_id=str(uuid.uuid4()), mission_id=str(uuid.uuid4()), object_id=self.object_id, vault_path=self.vault_path, source_commit=self.commit, provenance_hash=self.prov, expected_state="present", created_at="2026-08-27T00:00:00Z", idempotency_key=str(uuid.uuid4()), ) job.update(overrides) return job def _create_outcome_unknown(self, **overrides): job = self._make_job(**overrides) self.store.create_job(job) self.store.mark_ready(job["job_id"]) self.store._transition(job["job_id"], ST_CLAIMED) self.store._transition(job["job_id"], ST_EXECUTING) self.store._transition(job["job_id"], ST_OUTCOME_UNKNOWN) return self.store.get_job(job["job_id"]) def _core(self, source_content=None, read_content=None, source_fail=False, read_fail=False): src = source_content if source_content is not None else self.content return SaveReconciliationCore( self.store, (lambda sc, oid, vp: (_ for _ in ()).throw(RuntimeError("src"))) if source_fail else (lambda sc, oid, vp: src), (lambda vp: (_ for _ in ()).throw(RuntimeError("read"))) if read_fail else (lambda vp: read_content), ) # -- Manipulated job path ------------------------------------------------ def test_manipulated_job_path(self): # Job-Pfad im Payload geändert -> Drift-Guard muss greifen job = self._create_outcome_unknown() job["payload"]["vault_path"] = "tolaria/other.md" with self.assertRaises(JobImmutableFieldError): self.store._assert_immutable_fields(job) # -- Manipulated object_id ---------------------------------------------- def test_manipulated_object_id(self): job = self._create_outcome_unknown() job["payload"]["object_id"] = "object/other" with self.assertRaises(JobImmutableFieldError): self.store._assert_immutable_fields(job) # -- Manipulated provenance --------------------------------------------- # provenance_hash ist NUR im Payload (keine DB-Spalte). Der Guard prüft # nur DB-Spalten (object_id, vault_path). Der Reconciliation-Core schützt # provenance_hash durch _validate_provenance (Hash-Recompute): ein # manipulierter Hash -> PROVENANCE_MISMATCH -> bleibt OUTCOME_UNKNOWN. def test_manipulated_provenance(self): job = self._create_outcome_unknown() # DB-Payload manipulieren (nicht nur in-memory Dict) payload = dict(job["payload"]) payload["provenance_hash"] = "f" * 64 self.store._conn.execute( "UPDATE jobs SET payload = ? WHERE job_id = ?", (json.dumps(payload), job["job_id"]), ) self.store._conn.commit() # Reconciliation erkennt die Manipulation (Hash-Recompute mismatch) core = self._core(read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_PROVENANCE_MISMATCH) # -- Manipulated source_commit ------------------------------------------ # source_commit ist NUR im Payload (keine DB-Spalte). Der Reconciliation- # Core schützt source_commit durch den parameter-sensitiven Source-Loader: # ein manipulierter Commit -> WRONG_SOURCE -> Hash-Recompute mismatch -> # PROVENANCE_MISMATCH -> bleibt OUTCOME_UNKNOWN. def test_manipulated_source_commit(self): job = self._create_outcome_unknown() # DB-Payload manipulieren (nicht nur in-memory Dict) payload = dict(job["payload"]) payload["source_commit"] = "b" * 40 self.store._conn.execute( "UPDATE jobs SET payload = ? WHERE job_id = ?", (json.dumps(payload), job["job_id"]), ) self.store._conn.commit() # Parameter-sensitiver Loader: liefert Content NUR bei exaktem Commit def loader(commit, oid, vp): if commit == self.commit and oid == self.object_id and vp == self.vault_path: return self.content return "WRONG_SOURCE" core = SaveReconciliationCore(self.store, loader, lambda vp: self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_PROVENANCE_MISMATCH) # -- Stale source ------------------------------------------------------- def test_stale_source(self): # Source-Loader liefert alten Content (Source hat sich geändert) job = self._create_outcome_unknown() stale = "---\nid: object/adv-1\ntitle: Adv\n---\n\nSTALE\n" core = self._core(source_content=stale, read_content=stale) result = core.reconcile(job["job_id"], "worker-1") # Stale Source -> Hash recompute != Job-Hash -> PROVENANCE_MISMATCH self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_PROVENANCE_MISMATCH) # -- Target substitution ------------------------------------------------ def test_target_substitution(self): # Read-Back liefert anderen Content (Target wurde substituiert) job = self._create_outcome_unknown() core = self._core(read_content="SUBSTITUTED CONTENT") result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_TARGET_MISMATCH) # -- Unicode path ambiguity --------------------------------------------- def test_unicode_path_ambiguity(self): # Unicode-Homoglyphen im Pfad -> Schema-Reject with self.assertRaises(JobRejectedError): make_save_job( job_id=str(uuid.uuid4()), mission_id=str(uuid.uuid4()), object_id=self.object_id, vault_path="tolaria/file.md", # ligature source_commit=self.commit, provenance_hash=self.prov, expected_state="present", created_at="2026-08-27T00:00:00Z", idempotency_key=str(uuid.uuid4()), ) # -- Traversal ---------------------------------------------------------- def test_traversal(self): with self.assertRaises(JobRejectedError): make_save_job( job_id=str(uuid.uuid4()), mission_id=str(uuid.uuid4()), object_id=self.object_id, vault_path="../../etc/passwd", source_commit=self.commit, provenance_hash=self.prov, expected_state="present", created_at="2026-08-27T00:00:00Z", idempotency_key=str(uuid.uuid4()), ) # -- Duplicate reconciliation ------------------------------------------ def test_duplicate_reconciliation(self): job = self._create_outcome_unknown() core = self._core(read_content=self.content) r1 = core.reconcile(job["job_id"], "worker-1") r2 = core.reconcile(job["job_id"], "worker-1") self.assertEqual(r1["state"], ST_SUCCEEDED) self.assertEqual(r2["state"], ST_SUCCEEDED) # Kein State-Rückschritt, kein neues EXECUTED trail = self.store.audit_trail(job["job_id"]) executed = [e for e in trail if e.get("event_type") == "EXECUTED"] self.assertEqual(len(executed), 0) # -- Concurrent reconciliation ------------------------------------------ def test_concurrent_reconciliation(self): job = self._create_outcome_unknown() core1 = self._core(read_content=self.content) core2 = self._core(read_content=self.content) r1 = core1.reconcile(job["job_id"], "worker-1") r2 = core2.reconcile(job["job_id"], "worker-2") self.assertEqual(r1["state"], ST_SUCCEEDED) self.assertEqual(r2["state"], ST_SUCCEEDED) # -- DB row modification ------------------------------------------------ def test_db_row_modification(self): # Direkte DB-Manipulation: provenance_hash im Payload geändert. # Die eigentliche Verteidigung gegen DB-Manipulation ist die # PROVENANCE-RECOMPUTE: Reconciliation berechnet den Hash aus der # autoritativen Source neu und vergleicht mit dem gespeicherten # provenance_hash. Ein manipulierter Hash -> PROVENANCE_MISMATCH. job = self._create_outcome_unknown() # Payload-JSON in der DB manipulieren (provenance_hash auf falschen Wert) payload = dict(job["payload"]) payload["provenance_hash"] = "f" * 64 self.store._conn.execute( "UPDATE jobs SET payload = ? WHERE job_id = ?", (__import__("json").dumps(payload), job["job_id"]), ) self.store._conn.commit() # Reconciliation recomputet den Hash aus der Source (self.content) # und vergleicht mit dem manipulierten provenance_hash -> MISMATCH core = self._core(read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") # Manipulierter Hash -> PROVENANCE_MISMATCH -> FAIL CLOSED self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_PROVENANCE_MISMATCH) # -- Fake exact flag ---------------------------------------------------- def test_fake_exact_flag(self): # Angreifer versucht, ein "exact"-Flag zu setzen (gibt es nicht) job = self._create_outcome_unknown() # Reconciliation-Core hat kein "exact"-Flag, das gesetzt werden kann core = self._core(read_content="WRONG") result = core.reconcile(job["job_id"], "worker-1") # Falscher Content -> TARGET_MISMATCH, kein SUCCEEDED self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_TARGET_MISMATCH) # -- Forged audit event ------------------------------------------------- def test_forged_audit_event(self): # Angreifer versucht, ein EXECUTED-Event zu forgen (ohne Mutation) job = self._create_outcome_unknown() # Reconciliation erzeugt KEIN EXECUTED-Event core = self._core(read_content=self.content) core.reconcile(job["job_id"], "worker-1") trail = self.store.audit_trail(job["job_id"]) executed = [e for e in trail if e.get("event_type") == "EXECUTED"] self.assertEqual(len(executed), 0) # -- Force SUCCEEDED without target proof -------------------------------- def test_force_succeeded_without_target_proof(self): # Angreifer versucht, SUCCEEDED zu erzwingen, obwohl Target ABSENT job = self._create_outcome_unknown() core = self._core(read_content=None) # absent result = core.reconcile(job["job_id"], "worker-1") # Target ABSENT -> NICHT SUCCEEDED self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_TARGET_ABSENT) if __name__ == "__main__": unittest.main(verbosity=2)