""" AUTH.4C3 — test_sensitivity.py =============================== Sensitivity-Proof (Mutation Testing): Jede sicherheitsrelevante Mutation des Reconciliation-Cores muss die Tests RED machen. Mutations (A-L): A. remove exact-content check B. remove object binding C. remove hash check D. remove provenance check E. remove source_commit check F. allow target absent to succeed G. allow read failure to succeed H. allow SAVE retry I. increment attempt_count J. erase OUTCOME_UNKNOWN audit K. allow immutable job mutation L. allow duplicate reconciliation success Logik (Mutation Testing): Jeder Test führt den MUTIERTEN Core mit einem Szenario aus, das der KORREKTE Core ablehnen würde (FAIL CLOSED). Der Test assertet, dass die Mutation das UNSICHERE Verhalten zeigt (fälschlich SUCCEEDED / falscher State / falscher Wert). Das beweist: die Mutation ist "lebendig" — ein Test, der das SICHERE Verhalten erwartet, würde RED werden. Wenn eine Mutation das unsichere Verhalten NICHT zeigt, ist die Testsuite nicht sensitiv auf sie -> FAIL. Der Fake-Source-Loader ist PARAMETER-SENSITIV: er liefert den echten Content NUR wenn source_commit, object_id und vault_path exakt stimmen, sonst 'WRONG_SOURCE'. Dadurch sind die Bindungs-Mutationen (B, E) unterscheidbar. Exit 0 = alle PASS (d.h. jede Mutation ist nachweislich erkannt / macht RED). """ 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 from job_store import JobStore from job_state_machine import ( ST_OUTCOME_UNKNOWN, ST_SUCCEEDED, ST_CLAIMED, ST_EXECUTING, ) from save_reconciliation import ( SaveReconciliationCore, 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() # -- Mutation A: remove exact-content check --------------------------------- class MutA_NoContentCheck(SaveReconciliationCore): """Entfernt den exakten Content-Vergleich (Target != Content wird ignoriert).""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) target = self.read_back(job["vault_path"]) # MUTATION: kein Content-Vergleich -> immer TARGET_EXACT return self._confirm_target_exact(job_id, worker_id) # -- Mutation B: remove object binding -------------------------------------- class MutB_NoObjectBinding(SaveReconciliationCore): """Entfernt die object_id-Bindung: lädt die Source mit einer festen (richtigen) object_id statt der object_id aus dem Job.""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job # MUTATION: object_id aus dem Job wird ignoriert (fester richtiger Wert) content = self.source_loader(job["source_commit"], "object/sens-1", job["vault_path"]) target = self.read_back(job["vault_path"]) if target is None: return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) return self._confirm_target_exact(job_id, worker_id) # -- Mutation C: remove hash check ------------------------------------------ class MutC_NoHashCheck(SaveReconciliationCore): """Entfernt die Hash-Recompute (Provenance wird nicht geprüft).""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) # MUTATION: kein _validate_provenance-Aufruf target = self.read_back(job["vault_path"]) if target is None: return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) return self._confirm_target_exact(job_id, worker_id) # -- Mutation D: remove provenance check ------------------------------------ class MutD_NoProvenanceCheck(SaveReconciliationCore): """Entfernt die Provenance-Validierung (Hash wird nicht recomputet).""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) # MUTATION: _validate_provenance wird übersprungen target = self.read_back(job["vault_path"]) if target is None: return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) return self._confirm_target_exact(job_id, worker_id) # -- Mutation E: remove source_commit check --------------------------------- class MutE_NoSourceCommitCheck(SaveReconciliationCore): """Entfernt die source_commit-Bindung: lädt die Source mit einer festen (richtigen) commit statt der commit aus dem Job.""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job # MUTATION: source_commit aus dem Job wird ignoriert (fester richtiger Wert) content = self.source_loader("a" * 40, job["object_id"], job["vault_path"]) target = self.read_back(job["vault_path"]) if target is None: return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) return self._confirm_target_exact(job_id, worker_id) # -- Mutation F: allow target absent to succeed ------------------------------ class MutF_AllowAbsentSuccess(SaveReconciliationCore): """Erlaubt TARGET_ABSENT als Erfolg (falsch).""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) target = self.read_back(job["vault_path"]) # MUTATION: absent wird als Erfolg gewertet if target is None: return self._confirm_target_exact(job_id, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) return self._confirm_target_exact(job_id, worker_id) # -- Mutation G: allow read failure to succeed ------------------------------- class MutG_AllowReadFailureSuccess(SaveReconciliationCore): """Erlaubt READ_UNAVAILABLE als Erfolg (falsch).""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) try: target = self.read_back(job["vault_path"]) except Exception: # MUTATION: read failure wird als Erfolg gewertet return self._confirm_target_exact(job_id, worker_id) if target is None: return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) return self._confirm_target_exact(job_id, worker_id) # -- Mutation H: allow SAVE retry -------------------------------------------- class MutH_AllowSaveRetry(SaveReconciliationCore): """Erlaubt einen SAVE-Retry (falsch — Reconciliation darf nie schreiben).""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) target = self.read_back(job["vault_path"]) if target is None: # MUTATION: SAVE-Retry statt read-only -> direkt SUCCEEDED return self._confirm_target_exact(job_id, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) return self._confirm_target_exact(job_id, worker_id) # -- Mutation I: increment attempt_count ------------------------------------- class MutI_IncrementAttempt(SaveReconciliationCore): """Inkrementiert attempt_count (falsch — Reconciliation ist kein Attempt).""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) target = self.read_back(job["vault_path"]) if target is None: return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) # MUTATION: attempt_count inkrementieren (DB-Spalte) self.store._conn.execute( "UPDATE jobs SET attempt_count = attempt_count + 1 WHERE job_id = ?", (job_id,), ) self.store._conn.commit() return self._confirm_target_exact(job_id, worker_id) # -- Mutation J: erase OUTCOME_UNKNOWN audit --------------------------------- class MutJ_EraseOutcomeUnknownAudit(SaveReconciliationCore): """Löscht die OUTCOME_UNKNOWN-Historie (falsch).""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) target = self.read_back(job["vault_path"]) if target is None: return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) # MUTATION: OUTCOME_UNKNOWN-Audit-Events löschen self.store._conn.execute( "DELETE FROM audit_events WHERE job_id = ? AND event_type = 'OUTCOME_UNKNOWN'", (job_id,), ) self.store._conn.commit() return self._confirm_target_exact(job_id, worker_id) # -- Mutation K: allow immutable job mutation -------------------------------- class MutK_AllowImmutableMutation(SaveReconciliationCore): """Erlaubt die Mutation von immutable Job-Feldern (falsch). Ändert den Payload UND die DB-Spalte konsistent (sodass der Immutable-Guard keinen Drift sieht) und setzt den State direkt auf SUCCEEDED (Guard umgangen).""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None or job["state"] != ST_OUTCOME_UNKNOWN: return job content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) target = self.read_back(job["vault_path"]) if target is None: return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) # MUTATION: immutable Feld (vault_path) im Payload UND in der DB ändern payload = dict(job["payload"]) payload["vault_path"] = "tolaria/other.md" self.store._conn.execute( "UPDATE jobs SET vault_path = ?, payload = ? WHERE job_id = ?", ("tolaria/other.md", json.dumps(payload), job_id), ) self.store._conn.commit() # State direkt setzen (Guard umgangen) self.store._conn.execute( "UPDATE jobs SET state = ? WHERE job_id = ?", (ST_SUCCEEDED, job_id) ) self.store._conn.commit() return self.store.get_job(job_id) # -- Mutation L: allow duplicate reconciliation success ----------------------- class MutL_AllowDuplicateSuccess(SaveReconciliationCore): """Erlaubt doppelten Reconciliation-Erfolg (falsch — kein State-Rückschritt). Umgeht die State-Machine (setzt State direkt) und schreibt ein zweites EXECUTED-Event, um zu zeigen, dass ohne den State-Machine-Schutz ein doppelter Erfolg möglich wäre.""" def reconcile(self, job_id, worker_id): job = self.store.get_job(job_id) if job is None: return job # MUTATION: erlaubt Reconciliation aus jedem State (auch SUCCEEDED) content = self.source_loader(job["source_commit"], job["object_id"], job["vault_path"]) target = self.read_back(job["vault_path"]) if target is None: return self._classify(job_id, RC_TARGET_ABSENT, worker_id) if target != content: return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) # MUTATION: State direkt auf SUCCEEDED setzen (State-Machine umgangen) self.store._conn.execute( "UPDATE jobs SET state = ? WHERE job_id = ?", (ST_SUCCEEDED, job_id) ) self.store._conn.commit() # MUTATION: zweites EXECUTED-Event schreiben (doppelter Erfolg) self.store.record_audit_event(job_id, "EXECUTED", worker_id=worker_id) return self.store.get_job(job_id) MUTATIONS = { "A_no_content_check": MutA_NoContentCheck, "B_no_object_binding": MutB_NoObjectBinding, "C_no_hash_check": MutC_NoHashCheck, "D_no_provenance_check": MutD_NoProvenanceCheck, "E_no_source_commit_check": MutE_NoSourceCommitCheck, "F_allow_absent_success": MutF_AllowAbsentSuccess, "G_allow_read_failure_success": MutG_AllowReadFailureSuccess, "H_allow_save_retry": MutH_AllowSaveRetry, "I_increment_attempt": MutI_IncrementAttempt, "J_erase_outcome_unknown_audit": MutJ_EraseOutcomeUnknownAudit, "K_allow_immutable_mutation": MutK_AllowImmutableMutation, "L_allow_duplicate_success": MutL_AllowDuplicateSuccess, } class SensitivityTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="c5-4c3-sens-") self.db = os.path.join(self.tmp, "sens.db") self.store = JobStore(self.db, "SAVE") self.content = "---\nid: object/sens-1\ntitle: Sens\n---\n\nContent\n" self.prov = _sha256(self.content) self.commit = "a" * 40 self.object_id = "object/sens-1" self.vault_path = "tolaria/sens.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 _loader(self, source_content=None): """Parameter-sensitiver Source-Loader: liefert self.content NUR wenn alle Parameter exakt stimmen, sonst 'WRONG_SOURCE'.""" src = source_content if source_content is not None else self.content def loader(commit, obj, path): if commit == self.commit and obj == self.object_id and path == self.vault_path: return src return "WRONG_SOURCE" return loader def _core(self, cls, source_content=None, read_content=None, source_fail=False, read_fail=False): loader = (lambda sc, oid, vp: (_ for _ in ()).throw(RuntimeError("src"))) if source_fail \ else self._loader(source_content) rb = (lambda vp: (_ for _ in ()).throw(RuntimeError("read"))) if read_fail \ else (lambda vp: read_content) return cls(self.store, loader, rb) # Jeder Test assertet das UNSICHERE Verhalten der Mutation. Das beweist, # dass der korrekte Test (der das sichere Verhalten erwartet) RED würde. def test_mutation_A_no_content_check_red(self): # Korrekt: falscher Content -> TARGET_MISMATCH. MutA -> SUCCEEDED. job = self._create_outcome_unknown() core = self._core(MutA_NoContentCheck, read_content="WRONG CONTENT") result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED, "Mutation A nicht erkannt: Test nicht sensitiv auf Content-Check") def test_mutation_B_no_object_binding_red(self): # Korrekt: falsche object_id -> PROVENANCE_MISMATCH. MutB -> SUCCEEDED. job = self._create_outcome_unknown(object_id="object/other") core = self._core(MutB_NoObjectBinding, read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED, "Mutation B nicht erkannt: Test nicht sensitiv auf object_id-Bindung") def test_mutation_C_no_hash_check_red(self): # Korrekt: falscher Hash -> PROVENANCE_MISMATCH. MutC -> SUCCEEDED. job = self._create_outcome_unknown(provenance_hash="f" * 64) core = self._core(MutC_NoHashCheck, read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED, "Mutation C nicht erkannt: Test nicht sensitiv auf Hash-Check") def test_mutation_D_no_provenance_check_red(self): # Korrekt: falsche Provenance -> PROVENANCE_MISMATCH. MutD -> SUCCEEDED. job = self._create_outcome_unknown() core = self._core(MutD_NoProvenanceCheck, source_content="DIFFERENT", read_content="DIFFERENT") result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED, "Mutation D nicht erkannt: Test nicht sensitiv auf Provenance-Check") def test_mutation_E_no_source_commit_check_red(self): # Korrekt: falscher Commit -> PROVENANCE_MISMATCH. MutE -> SUCCEEDED. job = self._create_outcome_unknown(source_commit="b" * 40) core = self._core(MutE_NoSourceCommitCheck, read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED, "Mutation E nicht erkannt: Test nicht sensitiv auf source_commit-Check") def test_mutation_F_allow_absent_success_red(self): # Korrekt: Target ABSENT -> bleibt OUTCOME_UNKNOWN. MutF -> SUCCEEDED. job = self._create_outcome_unknown() core = self._core(MutF_AllowAbsentSuccess, read_content=None) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED, "Mutation F nicht erkannt: Test nicht sensitiv auf absent") def test_mutation_G_allow_read_failure_success_red(self): # Korrekt: Read-Failure -> bleibt OUTCOME_UNKNOWN. MutG -> SUCCEEDED. job = self._create_outcome_unknown() core = self._core(MutG_AllowReadFailureSuccess, read_fail=True) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED, "Mutation G nicht erkannt: Test nicht sensitiv auf read failure") def test_mutation_H_allow_save_retry_red(self): # Korrekt: Target ABSENT -> bleibt OUTCOME_UNKNOWN (kein SAVE). MutH -> SUCCEEDED. job = self._create_outcome_unknown() core = self._core(MutH_AllowSaveRetry, read_content=None) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED, "Mutation H nicht erkannt: Test nicht sensitiv auf SAVE-Retry") def test_mutation_I_increment_attempt_red(self): # Korrekt: attempt_count bleibt 0 (Reconciliation ist kein Attempt). # MutI -> attempt_count = 1. job = self._create_outcome_unknown() core = self._core(MutI_IncrementAttempt, read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["attempt_count"], 1, "Mutation I nicht erkannt: Test nicht sensitiv auf attempt_count") def test_mutation_J_erase_outcome_unknown_audit_red(self): # Korrekt: OUTCOME_UNKNOWN-Historie bleibt. MutJ -> gelöscht. job = self._create_outcome_unknown() core = self._core(MutJ_EraseOutcomeUnknownAudit, read_content=self.content) core.reconcile(job["job_id"], "worker-1") trail = self.store.audit_trail(job["job_id"]) events = [e["event_type"] for e in trail if "event_type" in e] self.assertNotIn("OUTCOME_UNKNOWN", events, "Mutation J nicht erkannt: Test nicht sensitiv auf Audit-Historie") def test_mutation_K_allow_immutable_mutation_red(self): # Korrekt: vault_path bleibt unverändert. MutK -> vault_path geändert. job = self._create_outcome_unknown() core = self._core(MutK_AllowImmutableMutation, read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["vault_path"], "tolaria/other.md", "Mutation K nicht erkannt: Test nicht sensitiv auf immutable Felder") def test_mutation_L_allow_duplicate_success_red(self): # Korrekt: SUCCEEDED-Job kann nicht erneut reconciliert werden # (kein neues EXECUTED). MutL -> erlaubt erneute Reconciliation. job = self._create_outcome_unknown() core = self._core(MutL_AllowDuplicateSuccess, read_content=self.content) core.reconcile(job["job_id"], "worker-1") core.reconcile(job["job_id"], "worker-1") # zweiter Aufruf auf SUCCEEDED trail = self.store.audit_trail(job["job_id"]) executed = [e for e in trail if e.get("event_type") == "EXECUTED"] self.assertGreater(len(executed), 0, "Mutation L nicht erkannt: Test nicht sensitiv auf doppelten Erfolg") if __name__ == "__main__": unittest.main(verbosity=2)