diff --git a/tolaria/c5-sync-service/job_state_machine.py b/tolaria/c5-sync-service/job_state_machine.py index 9235a49..f3b3d50 100644 --- a/tolaria/c5-sync-service/job_state_machine.py +++ b/tolaria/c5-sync-service/job_state_machine.py @@ -1,10 +1,11 @@ """ -AUTH.3E — job_state_machine.py -=============================== +AUTH.4C3 — job_state_machine.py (ERWEITERT) +============================================ Deterministische Job-State-Machine für den Executor Command Channel. States: - CREATED, READY, CLAIMED, EXECUTING, SUCCEEDED, FAILED, REJECTED, OUTCOME_UNKNOWN + CREATED, READY, CLAIMED, EXECUTING, SUCCEEDED, FAILED, REJECTED, OUTCOME_UNKNOWN, + RECONCILED Eigenschaften: * Geschlossene Transition-Allowlist @@ -12,6 +13,10 @@ Eigenschaften: * DELETE: KEIN RETRYABLE — ein gestarteter DELETE wird bei Lease-Expiry in OUTCOME_UNKNOWN überführt, NICHT zurück zu READY (kein blinder Retry) * SAVE: RETRYABLE nur bei transienten Fehlern (TolariaUnavailable) + * AUTH.4C3: OUTCOME_UNKNOWN -> RECONCILED -> SUCCEEDED (read-only Reconciliation). + RECONCILED ist ein persistenter Zwischenzustand, der die read-only + Reconciliation dokumentiert. SUCCEEDED bleibt der terminale Erfolgs-State. + KEIN stilles OUTCOME_UNKNOWN -> SUCCEEDED ohne Reconciliation-Evidence. Isoliert implementiert (KEIN produktiver Container). """ @@ -31,10 +36,11 @@ ST_SUCCEEDED = "SUCCEEDED" ST_FAILED = "FAILED" ST_REJECTED = "REJECTED" ST_OUTCOME_UNKNOWN = "OUTCOME_UNKNOWN" +ST_RECONCILED = "RECONCILED" # AUTH.4C3: read-only Reconciliation abgeschlossen ALL_STATES = frozenset({ ST_CREATED, ST_READY, ST_CLAIMED, ST_EXECUTING, - ST_SUCCEEDED, ST_FAILED, ST_REJECTED, ST_OUTCOME_UNKNOWN, + ST_SUCCEEDED, ST_FAILED, ST_REJECTED, ST_OUTCOME_UNKNOWN, ST_RECONCILED, }) # --------------------------------------------------------------------------- @@ -58,6 +64,9 @@ ALLOWED_TRANSITIONS: FrozenSet[Tuple[str, str]] = frozenset({ (ST_EXECUTING, ST_REJECTED), # Gate-Fehler während Execution # OUTCOME_UNKNOWN (terminal für DELETE; SAVE kann via Human/Admin reaktiviert werden) (ST_OUTCOME_UNKNOWN, ST_REJECTED), # expliziter Human-/Admin-Eingriff + # AUTH.4C3: read-only Reconciliation (nur SAVE) + (ST_OUTCOME_UNKNOWN, ST_RECONCILED), # TARGET_EXACT bestätigt (read-only) + (ST_RECONCILED, ST_SUCCEEDED), # terminal bestätigt, KEINE neue Mutation }) # Terminal-States (keine weiteren Transitionen ohne expliziten Eingriff) diff --git a/tolaria/c5-sync-service/job_store.py b/tolaria/c5-sync-service/job_store.py index 599ed6a..0f1466e 100644 --- a/tolaria/c5-sync-service/job_store.py +++ b/tolaria/c5-sync-service/job_store.py @@ -35,6 +35,7 @@ from job_state_machine import ( ST_FAILED, ST_OUTCOME_UNKNOWN, ST_READY, + ST_RECONCILED, ST_REJECTED, ST_SUCCEEDED, InvalidTransitionError, @@ -233,6 +234,12 @@ class JobStore: raise JobNotFoundError(f"job not found: {job_id}") from_state = job["state"] transition(from_state, to_state) # wirft bei ungültiger Transition + # AUTH.4C3: Immutable-Fields-Guard. Sicherheitsrelevante Felder dürfen + # nach OUTCOME_UNKNOWN nicht mehr verändert werden. Der Guard verifiziert + # die immutable Felder im persistierten Job gegen den Payload (der beim + # Claim/Execution nicht verändert wird). Falls ein Feld fehlt oder + # inkonsistent ist -> FAIL CLOSED (kein State-Change). + self._assert_immutable_fields(job) now = int(time.time() * 1000) self._conn.execute( """ @@ -260,6 +267,65 @@ class JobStore: def mark_outcome_unknown(self, job_id: str, worker_id: str) -> Dict[str, Any]: return self._transition(job_id, ST_OUTCOME_UNKNOWN, worker_id=worker_id) + def mark_reconciled(self, job_id: str, worker_id: str) -> Dict[str, Any]: + """AUTH.4C3: OUTCOME_UNKNOWN -> RECONCILED (read-only Reconciliation).""" + return self._transition(job_id, ST_RECONCILED, worker_id=worker_id) + + def set_result_code(self, job_id: str, result_code: str) -> Dict[str, Any]: + """ + AUTH.4C3: Setzt NUR den result_code eines Jobs (Diagnose-Information), + OHNE State-Transition. Der State bleibt unverändert (z.B. OUTCOME_UNKNOWN). + Kein Mutation-Retry, kein State-Change. Verbessert Operator-Visibility: + der Operator sieht, warum die Reconciliation nicht erfolgreich war. + """ + job = self.get_job(job_id) + if job is None: + raise JobNotFoundError(f"job not found: {job_id}") + now = int(time.time() * 1000) + self._conn.execute( + "UPDATE jobs SET result_code = ?, updated_at = ? WHERE job_id = ?", + (result_code, now, job_id), + ) + self._conn.commit() + return self.get_job(job_id) + + # -- Immutable-Fields-Guard (AUTH.4C3) ---------------------------------- + + def _assert_immutable_fields(self, job: Dict[str, Any]) -> None: + """ + Verifiziert, dass die sicherheitsrelevanten immutable Felder im + persistierten Job konsistent sind. Prüft die rohen DB-Spalten + (object_id, vault_path) gegen das Payload-JSON. Falls ein immutable Feld + im Payload von seinem DB-Spalten-Wert abweicht -> FAIL CLOSED + (JobImmutableFieldError, kein State-Change). Verhindert, dass ein + manipulierter Job (z.B. geänderte object_id/vault_path) eine Transition + durchläuft. + + Hinweis: _row_to_dict merged das Payload in das Top-Level-Dict, daher + ist ein Payload-vs-Top-Level-Vergleich wirkungslos. Stattdessen wird + die rohe DB-Spalte direkt gelesen und gegen das Payload geprüft. + """ + payload = job.get("payload") + if not isinstance(payload, dict): + return + job_id = job.get("job_id") + if not job_id: + return + row = self._conn.execute( + "SELECT object_id, vault_path FROM jobs WHERE job_id = ?", (job_id,) + ).fetchone() + if row is None: + return + db_object_id, db_vault_path = row + # object_id: DB-Spalte muss mit Payload übereinstimmen + if "object_id" in payload and payload["object_id"] != db_object_id: + raise JobImmutableFieldError( + f"job {job_id} immutable field drift: object_id") + # vault_path: DB-Spalte muss mit Payload übereinstimmen + if "vault_path" in payload and payload["vault_path"] != db_vault_path: + raise JobImmutableFieldError( + f"job {job_id} immutable field drift: vault_path") + # -- Claim / Lease ------------------------------------------------------ def claim_job(self, job_id: str, worker_id: str, lease_seconds: int = 60) -> Dict[str, Any]: diff --git a/tolaria/c5-sync-service/save_reconciliation.py b/tolaria/c5-sync-service/save_reconciliation.py new file mode 100644 index 0000000..b425b90 --- /dev/null +++ b/tolaria/c5-sync-service/save_reconciliation.py @@ -0,0 +1,247 @@ +""" +AUTH.4C3 — save_reconciliation.py +================================= +Read-only OUTCOME_UNKNOWN-RECONCILIATION-CONTRACT für den SAVE-Executor. + +Schließt den Contract Gap: OUTCOME_UNKNOWN + Read-Back beweist exakt, dass das +gewünschte Ziel bereits erreicht wurde -> terminal bestätigter Erfolg, OHNE +zweiten SAVE, OHNE Retry, OHNE manuellen SUCCEEDED-State, OHNE SQL, OHNE +Überschreiben historischer Unsicherheit. + +CORE SAFETY INVARIANT: + OUTCOME_UNKNOWN bedeutet: der Executor darf NICHT wissen, ob die Mutation + stattgefunden hat. Daher darf OUTCOME_UNKNOWN NIEMALS direkt automatisch + erneut mutieren. Einzige sichere nächste Aktion: READ-ONLY RECONCILIATION. + Erst wenn die autoritative Zielseite exakt beweist TARGET_ALREADY_REACHED, + darf der Job ohne neue Mutation terminal bestätigt werden. + +KRITISCH: + Dieses Modul besitzt KEIN write-Callable. Es gibt KEINE tolaria_save / + write()-Dependency Injection. Reconciliation ist technisch unfähig zu + schreiben. Dependencies: source_loader, read_back, job_store, audit. + +Isoliert implementiert (KEIN produktiver Container). Nutzt job_store + +job_state_machine + job_schema. +""" + +from __future__ import annotations + +import hashlib +from typing import Any, Callable, Dict, Optional + +from job_schema import JOB_TYPE_SAVE +from job_store import JobStore +from job_state_machine import ( + ST_OUTCOME_UNKNOWN, + ST_RECONCILED, + ST_SUCCEEDED, +) + +# --------------------------------------------------------------------------- +# Reconciliation-Ergebnis-Klassifikation +# --------------------------------------------------------------------------- +# Nur TARGET_EXACT darf terminalen Erfolg bestätigen. +RC_TARGET_EXACT = "TARGET_EXACT" +RC_TARGET_ABSENT = "TARGET_ABSENT" +RC_TARGET_MISMATCH = "TARGET_MISMATCH" +RC_READ_UNAVAILABLE = "READ_UNAVAILABLE" +RC_INVALID_JOB = "INVALID_JOB" +RC_PROVENANCE_MISMATCH = "PROVENANCE_MISMATCH" +RC_ALREADY_RECONCILED = "ALREADY_RECONCILED" + +# Audit-Events (AUTH.4C3) +EV_RECONCILIATION_STARTED = "RECONCILIATION_STARTED" +EV_RECONCILIATION_TARGET_EXACT = "RECONCILIATION_TARGET_EXACT" +EV_CONFIRMED_EXECUTED = "CONFIRMED_EXECUTED" + +# Immutable Job-Felder (müssen nach OUTCOME_UNKNOWN unveränderlich sein) +IMMUTABLE_RECONCILIATION_FIELDS = frozenset({ + "object_id", "vault_path", "source_commit", "provenance_hash", + "idempotency_key", "mission_id", +}) + + +class ReconciliationError(Exception): + """Basis-Fehler für Reconciliation.""" + + +class ReconciliationWriteError(ReconciliationError): + """Wird geworfen, wenn Reconciliation versucht zu schreiben (darf nie passieren).""" + + +class SaveReconciliationCore: + """ + Read-only Reconciliation-Core für OUTCOME_UNKNOWN-SAVE-Jobs. + + source_loader: Callable[[str, str, str], str] — lädt Content aus autoritativer + Source (source_commit, object_id, vault_path). Identisch zu SaveExecutorCore. + read_back: Callable[[str], Optional[str]] — liest den aktuellen Tolaria-Target + (vault_path) -> content oder None. Credential-frei (Least Privilege). + store: JobStore — für State-Transition + Audit. + + KEIN tolaria_save / write-Callable. Dieses Modul kann technisch NICHT schreiben. + """ + + def __init__( + self, + store: JobStore, + source_loader: Callable[[str, str, str], str], + read_back: Callable[[str], Optional[str]], + ): + if store.worker_scope != "SAVE": + raise ReconciliationError( + "SaveReconciliationCore requires worker_scope='SAVE'") + self.store = store + self.source_loader = source_loader + self.read_back = read_back + + # -- Hauptverarbeitung -------------------------------------------------- + + def reconcile(self, job_id: str, worker_id: str) -> Dict[str, Any]: + """ + Führt die read-only Reconciliation eines OUTCOME_UNKNOWN-SAVE-Jobs aus. + + Ablauf: + 1. Job laden (muss existieren) + 2. State prüfen (muss OUTCOME_UNKNOWN sein) + 3. Job-Type prüfen (nur SAVE) + 4. Immutable Felder validieren + 5. Content aus autoritativer Source rekonstruieren (exact commit) + 6. Provenance/Hash recompute + validieren + 7. Tolaria-Target lesen (credential-frei) + 8. Exakt vergleichen -> klassifizieren + 9. Nur TARGET_EXACT -> RECONCILED -> SUCCEEDED (KEIN SAVE) + + Returns: finaler Job-State. + """ + job = self.store.get_job(job_id) + if job is None: + raise ReconciliationError(f"job not found: {job_id}") + + # State muss OUTCOME_UNKNOWN sein (nur dieser Zustand ist reconciliierbar) + if job["state"] != ST_OUTCOME_UNKNOWN: + # Idempotenz: bereits RECONCILED -> finalisiere zu SUCCEEDED + if job["state"] == ST_RECONCILED: + return self._finalize_reconciled(job_id, worker_id) + # SUCCEEDED bleibt terminal (kein State-Rückschritt) + if job["state"] == ST_SUCCEEDED: + return job + # Andere States (READY/CLAIMED/EXECUTING/FAILED/REJECTED/CREATED) + # dürfen den Recovery-Pfad NICHT nutzen. + raise ReconciliationError( + f"job {job_id} state={job['state']} not reconciliable (must be OUTCOME_UNKNOWN)") + + # Job-Type: nur SAVE + if job["job_type"] != JOB_TYPE_SAVE: + raise ReconciliationError( + f"job {job_id} job_type={job['job_type']} not reconciliable (must be SAVE)") + + # Immutable Felder validieren (müssen vorhanden + konsistent sein) + self._validate_immutable_fields(job) + + # Audit: Reconciliation beginnt (read-only) + self.store.record_audit_event( + job_id, EV_RECONCILIATION_STARTED, worker_id=worker_id) + + # 1. Content aus autoritativer Source rekonstruieren (exact commit) + try: + content = self.source_loader( + job["source_commit"], job["object_id"], job["vault_path"]) + except Exception: + # Source nicht verfügbar -> READ_UNAVAILABLE (kein Erfolg) + return self._classify(job_id, RC_READ_UNAVAILABLE, worker_id) + + # 2. Provenance/Hash recompute + validieren + if not self._validate_provenance(job, content): + return self._classify(job_id, RC_PROVENANCE_MISMATCH, worker_id) + + # 3. Tolaria-Target lesen (credential-frei) + try: + target_content = self.read_back(job["vault_path"]) + except Exception: + # Read-Back nicht verfügbar -> READ_UNAVAILABLE (kein Erfolg) + return self._classify(job_id, RC_READ_UNAVAILABLE, worker_id) + + # 4. Exakt vergleichen -> klassifizieren + if target_content is None: + # Target ABSENT -> NICHT automatisch SAVE wiederholen + return self._classify(job_id, RC_TARGET_ABSENT, worker_id) + + if target_content != content: + # Target MISMATCH -> FAIL CLOSED (kein Überschreiben, kein Repair) + return self._classify(job_id, RC_TARGET_MISMATCH, worker_id) + + # 5. TARGET_EXACT: alle Felder exakt bestätigt + return self._confirm_target_exact(job_id, worker_id) + + # -- TARGET_EXACT-Bestätigung ------------------------------------------- + + def _confirm_target_exact(self, job_id: str, worker_id: str) -> Dict[str, Any]: + """ + TARGET_EXACT bestätigt: KEIN SAVE. Job wird regulär finalisiert. + OUTCOME_UNKNOWN -> RECONCILED -> SUCCEEDED. + """ + # Audit: TARGET_EXACT bestätigt (read-only) + self.store.record_audit_event( + job_id, EV_RECONCILIATION_TARGET_EXACT, worker_id=worker_id) + + # OUTCOME_UNKNOWN -> RECONCILED (persistenter Zwischenzustand) + self.store.mark_reconciled(job_id, worker_id) + + # RECONCILED -> SUCCEEDED (terminal, KEINE neue Mutation) + self.store.mark_succeeded(job_id, worker_id) + + # Audit: Mutation als BEREITS ausgeführt bestätigt (NICHT neu ausgeführt) + self.store.record_audit_event( + job_id, EV_CONFIRMED_EXECUTED, worker_id=worker_id) + + return self.store.get_job(job_id) + + # -- Idempotenz: bereits RECONCILED ------------------------------------- + + def _finalize_reconciled(self, job_id: str, worker_id: str) -> Dict[str, Any]: + """ + Job ist bereits RECONCILED (Crash nach RECONCILED vor SUCCEEDED, oder + wiederholte Reconciliation). Finalisiere zu SUCCEEDED. Kein State-Rückschritt, + kein neues EXECUTED, kein SAVE. + """ + # Audit: ALREADY_RECONCILED (idempotent) + self.store.record_audit_event( + job_id, EV_RECONCILIATION_TARGET_EXACT, worker_id=worker_id) + self.store.mark_succeeded(job_id, worker_id) + self.store.record_audit_event( + job_id, EV_CONFIRMED_EXECUTED, worker_id=worker_id) + return self.store.get_job(job_id) + + # -- Klassifikation (kein Erfolg) --------------------------------------- + + def _classify(self, job_id: str, result: str, worker_id: str) -> Dict[str, Any]: + """ + Nicht-TARGET_EXACT-Ergebnis: Job bleibt OUTCOME_UNKNOWN (kein State-Change). + Kein automatischer Retry. Human Review. Der result_code wird im Job + gesetzt (Diagnose-Information), der State bleibt unverändert. + """ + # Audit: Reconciliation-Ergebnis (kein Erfolg, kein State-Change) + self.store.record_audit_event( + job_id, f"RECONCILIATION_{result}", worker_id=worker_id, + result_code=result) + # result_code im Job setzen (kein State-Change) + self.store.set_result_code(job_id, result) + return self.store.get_job(job_id) + + # -- Validierung -------------------------------------------------------- + + def _validate_immutable_fields(self, job: Dict[str, Any]) -> None: + """Prüft, dass alle immutable Reconciliation-Felder vorhanden sind.""" + missing = IMMUTABLE_RECONCILIATION_FIELDS - set(job.keys()) + if missing: + raise ReconciliationError( + f"job {job['job_id']} missing immutable fields: {sorted(missing)}") + + def _validate_provenance(self, job: Dict[str, Any], content: str) -> bool: + """ + RECOMPUTE: berechnet den Provenance-Hash aus dem rekonstruierten Content + und vergleicht mit dem Job-Feld. RQ darf den Hash nicht blind bestimmen. + """ + computed = hashlib.sha256(content.encode("utf-8")).hexdigest() + return computed == job["provenance_hash"] diff --git a/tolaria/c5-sync-service/test_adversarial.py b/tolaria/c5-sync-service/test_adversarial.py new file mode 100644 index 0000000..7fa0675 --- /dev/null +++ b/tolaria/c5-sync-service/test_adversarial.py @@ -0,0 +1,288 @@ +""" +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) diff --git a/tolaria/c5-sync-service/test_productive_incident_fixture.py b/tolaria/c5-sync-service/test_productive_incident_fixture.py new file mode 100644 index 0000000..86bc260 --- /dev/null +++ b/tolaria/c5-sync-service/test_productive_incident_fixture.py @@ -0,0 +1,203 @@ +""" +AUTH.4C3 — test_productive_incident_fixture.py +============================================== +ISOLIERTE Fixture, die exakt den realen AUTH.4C2-Incident nachbildet. + +Real-Incident (AUTH.4C2, produktiv): + JOB_ID: f18608df-0469-4760-8d84-d6bf1917e772 + STATE: OUTCOME_UNKNOWN + attempt_count: 1 + Canary: PRESENT + Path: tolaria/auth4c2-canary.md + Object ID: object/cfeafffb-415e-4743-af44-e8c92efb9ac5 + Hash: 798ae0315c14aa9d7e3a3f759c952f7c18cbce5154844f447e597b7fd1776515 + Tolaria Markdown Count: 85 + +Erwartung (AUTH.4C3): + Reconciliation bestätigt Erfolg (TARGET_EXACT) + SAVE_CALL_COUNT = 0 + attempt_count bleibt 1 + Historie bleibt erhalten (OUTCOME_UNKNOWN sichtbar im Audit) + KEINE produktive DB. KEIN produktiver Job. KEIN SAVE. + +Exit 0 = alle PASS. +""" + +from __future__ import annotations + +import hashlib +import os +import shutil +import sys +import tempfile +import unittest + +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, + EV_RECONCILIATION_STARTED, EV_RECONCILIATION_TARGET_EXACT, + EV_CONFIRMED_EXECUTED, +) + +# Exakte Werte aus dem realen Incident (ISOLIERTE Fixture, KEINE produktive DB) +INCIDENT_JOB_ID = "f18608df-0469-4760-8d84-d6bf1917e772" +INCIDENT_OBJECT_ID = "object/cfeafffb-415e-4743-af44-e8c92efb9ac5" +INCIDENT_PATH = "tolaria/auth4c2-canary.md" +INCIDENT_HASH = "798ae0315c14aa9d7e3a3f759c952f7c18cbce5154844f447e597b7fd1776515" +INCIDENT_ATTEMPTS = 1 + + +class ProductiveIncidentFixtureTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + # Canary-Content: der EXAKTE reale Inhalt aus dem produktiven Vault + # (read-only verifiziert, Hash stimmt exakt mit dem Incident überein) + cls.canary_content = ( + "---\n" + "id: object/cfeafffb-415e-4743-af44-e8c92efb9ac5\n" + "type: Note\n" + "title: AUTH.4C2 Canary\n" + "tags: [auth4c2, canary, c5-save]\n" + "created: 2026-08-27\n" + "mission: AUTH.4C2\n" + "---\n" + "\n" + "# AUTH.4C2 First Productive SAVE Canary\n" + "\n" + "Dieser Eintrag ist der kontrollierte produktive SAVE-Canary der Mission AUTH.4C2.\n" + "\n" + "- **Mission:** AUTH.4C2 — First Productive SAVE Canary\n" + "- **Zweck:** Beweis exakt eines produktiven SAVEs über den isolierten c5-save-executor-Pfad\n" + "- **Pfad:** tolaria/auth4c2-canary.md\n" + "- **Quelle:** Forgejo Source of Truth (commit-bound)\n" + "- **Fachlich:** neutral — kein Trading-Inhalt, keine Secrets\n" + "\n" + "Dieser Canary bleibt kontrolliert vorhanden für spätere AUTH.4E / Cleanup-Mission.\n" + ) + # Verifiziere: Hash des Canary-Contents == Incident-Hash + computed = hashlib.sha256(cls.canary_content.encode("utf-8")).hexdigest() + assert computed == INCIDENT_HASH, ( + f"Fixture-Hash mismatch: computed={computed} expected={INCIDENT_HASH}" + ) + + def setUp(self): + # Jeder Test bekommt eine FRISCHE isolierte DB (create_job ist idempotent + # per job_id — dieselbe INCIDENT_JOB_ID darf nicht über Tests kollidieren) + self.tmp = tempfile.mkdtemp(prefix="c5-4c3-incident-") + self.db = os.path.join(self.tmp, "incident.db") + self.store = JobStore(self.db, "SAVE") + + def tearDown(self): + self.store.close() + shutil.rmtree(self.tmp, ignore_errors=True) + + def _create_incident_job(self): + """Erzeugt den Job exakt wie im realen Incident (OUTCOME_UNKNOWN, attempts=1).""" + job = make_save_job( + job_id=INCIDENT_JOB_ID, + mission_id="00000000-0000-0000-0000-000000000001", + object_id=INCIDENT_OBJECT_ID, + vault_path=INCIDENT_PATH, + source_commit="a" * 40, + provenance_hash=INCIDENT_HASH, + expected_state="present", + created_at="2026-08-27T00:00:00Z", + idempotency_key="00000000-0000-0000-0000-000000000002", + ) + self.store.create_job(job) + self.store.mark_ready(job["job_id"]) + # Simuliere Execution -> OUTCOME_UNKNOWN (attempt_count=1 via Claim) + 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) + # attempt_count auf 1 setzen (wie im Incident) + self.store._conn.execute( + "UPDATE jobs SET attempt_count = ? WHERE job_id = ?", + (INCIDENT_ATTEMPTS, job["job_id"]), + ) + self.store._conn.commit() + return self.store.get_job(job["job_id"]) + + def test_incident_reconciliation_confirms_success(self): + """TARGET_EXACT: Reconciliation bestätigt Erfolg, Job -> SUCCEEDED.""" + job = self._create_incident_job() + self.assertEqual(job["state"], ST_OUTCOME_UNKNOWN) + self.assertEqual(job["attempt_count"], INCIDENT_ATTEMPTS) + + # Read-Back liefert exakt den Canary-Content (Target PRESENT + exact) + core = SaveReconciliationCore( + self.store, + lambda sc, oid, vp: self.canary_content, # source_loader + lambda vp: self.canary_content, # read_back + ) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_SUCCEEDED) + + def test_incident_zero_save_calls(self): + """SAVE_CALL_COUNT = 0: Reconciliation ruft niemals write() auf.""" + job = self._create_incident_job() + save_calls = {"count": 0} + + def fake_save(*args, **kwargs): + save_calls["count"] += 1 + raise AssertionError("Reconciliation darf NIE write() aufrufen!") + + core = SaveReconciliationCore( + self.store, + lambda sc, oid, vp: self.canary_content, + lambda vp: self.canary_content, + ) + # Reconciliation-Core hat KEIN write-Callable (technisch unmöglich) + self.assertFalse(hasattr(core, "tolaria_save")) + self.assertFalse(hasattr(core, "write")) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_SUCCEEDED) + self.assertEqual(save_calls["count"], 0) + + def test_incident_attempt_count_unchanged(self): + """attempt_count bleibt 1 (Reconciliation ist KEIN neuer Mutation Attempt).""" + job = self._create_incident_job() + core = SaveReconciliationCore( + self.store, + lambda sc, oid, vp: self.canary_content, + lambda vp: self.canary_content, + ) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_SUCCEEDED) + self.assertEqual(result["attempt_count"], INCIDENT_ATTEMPTS) + + def test_incident_history_preserved(self): + """Historie bleibt erhalten: OUTCOME_UNKNOWN sichtbar im Audit-Trail.""" + job = self._create_incident_job() + core = SaveReconciliationCore( + self.store, + lambda sc, oid, vp: self.canary_content, + lambda vp: self.canary_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] + # Reconciliation-Evidence vorhanden + self.assertIn(EV_RECONCILIATION_STARTED, events) + self.assertIn(EV_RECONCILIATION_TARGET_EXACT, events) + self.assertIn(EV_CONFIRMED_EXECUTED, events) + # KEIN EXECUTED (Mutation wurde NICHT neu ausgeführt) + self.assertNotIn("EXECUTED", events) + # Der Job-Snapshot im Trail zeigt den finalen State SUCCEEDED, + # aber die OUTCOME_UNKNOWN-Historie ist in den Events sichtbar + # (der Job durchlief OUTCOME_UNKNOWN -> RECONCILED -> SUCCEEDED). + + def test_incident_no_production_db(self): + """Fixture nutzt eine ISOLIERTE temporäre DB, keine produktive DB.""" + self.assertNotEqual(self.db, "/opt/data/forgejo/trading-system-docs/tolaria/c5-sync-service/c5a_save.db") + self.assertTrue(self.db.startswith(tempfile.gettempdir()) or "/tmp" in self.db or "c5-4c3-incident" in self.db) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tolaria/c5-sync-service/test_reconciliation.py b/tolaria/c5-sync-service/test_reconciliation.py new file mode 100644 index 0000000..8719aa9 --- /dev/null +++ b/tolaria/c5-sync-service/test_reconciliation.py @@ -0,0 +1,508 @@ +""" +AUTH.4C3 — test_reconciliation.py +================================= +Isolierte Tests für den OUTCOME_UNKNOWN-RECONCILIATION-CONTRACT. + +Testumgebung: + * JobStore auf temporärer SQLite-DB + * Fake-Source-Loader (liefert Content aus Test-Repo) + * Fake-Read-Back (liefert Tolaria-Target) + * KEIN write-Callable im Reconciliation-Core (technisch unmöglich) + +Führt die AUTH.4C3-Tests T1-T30 aus. Exit 0 = alle PASS. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import sys +import tempfile +import unittest +import uuid +from typing import Any, Dict, Optional + +# Module aus dem Build-Verzeichnis importieren +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_RECONCILED, ST_SUCCEEDED, ST_FAILED, ST_REJECTED, + ST_READY, ST_CLAIMED, ST_EXECUTING, ST_CREATED, + InvalidTransitionError, transition, is_valid_transition, +) +from save_reconciliation import ( + SaveReconciliationCore, ReconciliationError, + RC_TARGET_EXACT, RC_TARGET_ABSENT, RC_TARGET_MISMATCH, + RC_READ_UNAVAILABLE, RC_INVALID_JOB, RC_PROVENANCE_MISMATCH, + RC_ALREADY_RECONCILED, + EV_RECONCILIATION_STARTED, EV_RECONCILIATION_TARGET_EXACT, + EV_CONFIRMED_EXECUTED, +) + + +def _sha256(s: str) -> str: + return hashlib.sha256(s.encode("utf-8")).hexdigest() + + +class FakeSourceLoader: + """Fake-Source-Loader. Liefert Content aus einem Dict (commit -> path -> content).""" + + def __init__(self, content: str, fail: bool = False): + self.content = content + self.fail = fail + self.calls = 0 + + def __call__(self, source_commit: str, object_id: str, vault_path: str) -> str: + self.calls += 1 + if self.fail: + raise RuntimeError("source unavailable") + return self.content + + +class FakeReadBack: + """Fake-Read-Back. Liefert Tolaria-Target oder None (absent) / wirft (unavailable).""" + + def __init__(self, content: Optional[str] = None, fail: bool = False): + self.content = content + self.fail = fail + self.calls = 0 + + def __call__(self, vault_path: str) -> Optional[str]: + self.calls += 1 + if self.fail: + raise RuntimeError("read unavailable") + return self.content + + +class AUTH4C3ReconciliationTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.tmp = tempfile.mkdtemp(prefix="c5-4c3-test-") + cls.db = os.path.join(cls.tmp, "test_save.db") + cls.store = JobStore(cls.db, "SAVE") + cls.content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nHello World\n" + cls.prov = _sha256(cls.content) + cls.commit = "a" * 40 + cls.object_id = "object/test-obj-1" + cls.vault_path = "tolaria/test.md" + + @classmethod + def tearDownClass(cls): + cls.store.close() + shutil.rmtree(cls.tmp, ignore_errors=True) + + def _make_job(self, **overrides) -> Dict[str, Any]: + 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) -> Dict[str, Any]: + """Erzeugt einen Job im OUTCOME_UNKNOWN-Zustand (wie der produktive Incident).""" + job = self._make_job(**overrides) + self.store.create_job(job) + self.store.mark_ready(job["job_id"]) + # Simuliere: Job wurde geclaimt + execution -> OUTCOME_UNKNOWN + # (direkt via _transition, da wir keinen vollen Execution-Pfad brauchen) + 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: Optional[str] = None, + read_content: Optional[str] = None, + source_fail: bool = False, read_fail: bool = False) -> SaveReconciliationCore: + src = source_content if source_content is not None else self.content + loader = FakeSourceLoader(src, fail=source_fail) + rb = FakeReadBack(read_content, fail=read_fail) + core = SaveReconciliationCore(self.store, loader, rb) + return core + + # -- T1: OUTCOME_UNKNOWN + exact target -> confirmed success ------------- + + def test_t1_exact_target_confirmed_success(self): + job = self._create_outcome_unknown() + core = self._core(read_content=self.content) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_SUCCEEDED) + # Audit: RECONCILIATION_STARTED + TARGET_EXACT + CONFIRMED_EXECUTED + trail = self.store.audit_trail(job["job_id"]) + events = [e["event_type"] for e in trail if "event_type" in e] + self.assertIn(EV_RECONCILIATION_STARTED, events) + self.assertIn(EV_RECONCILIATION_TARGET_EXACT, events) + self.assertIn(EV_CONFIRMED_EXECUTED, events) + + # -- T2: exact target -> zero SAVE calls --------------------------------- + + def test_t2_zero_save_calls(self): + job = self._create_outcome_unknown() + core = self._core(read_content=self.content) + # Reconciliation-Core hat KEIN write-Callable. Wir prüfen, dass der + # Core keine write-Methode besitzt und dass der Job SUCCEEDED wird + # ohne dass irgendein SAVE aufgerufen werden könnte. + self.assertFalse(hasattr(core, "tolaria_save")) + self.assertFalse(hasattr(core, "write")) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_SUCCEEDED) + + # -- T3: attempt_count unchanged ---------------------------------------- + + def test_t3_attempt_count_unchanged(self): + job = self._create_outcome_unknown() + self.assertEqual(job["attempt_count"], 0) # direkt via _transition, kein Claim + core = self._core(read_content=self.content) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_SUCCEEDED) + self.assertEqual(result["attempt_count"], 0) + + # -- T4: exact target second reconcile -> idempotent --------------------- + + def test_t4_second_reconcile_idempotent(self): + job = self._create_outcome_unknown() + core = self._core(read_content=self.content) + result1 = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result1["state"], ST_SUCCEEDED) + # Zweite Reconciliation: Job ist SUCCEEDED -> bleibt terminal + result2 = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result2["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"] + confirmed = [e for e in trail if e.get("event_type") == EV_CONFIRMED_EXECUTED] + self.assertEqual(len(executed), 0) # kein neues EXECUTED + self.assertGreaterEqual(len(confirmed), 1) + + # -- T5: target absent -> remains unknown -------------------------------- + + def test_t5_target_absent_remains_unknown(self): + job = self._create_outcome_unknown() + core = self._core(read_content=None) # absent + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) + self.assertEqual(result["result_code"], RC_TARGET_ABSENT) + + # -- T6: target mismatch content -> fail-closed -------------------------- + + def test_t6_target_mismatch_content_fail_closed(self): + job = self._create_outcome_unknown() + core = self._core(read_content="DIFFERENT CONTENT") + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) + self.assertEqual(result["result_code"], RC_TARGET_MISMATCH) + + # -- T7: target mismatch object -> fail-closed --------------------------- + + def test_t7_target_mismatch_object_fail_closed(self): + # object_id im Job geändert -> source_loader liefert anderen Content + job = self._create_outcome_unknown(object_id="object/other-obj") + # Source-Loader liefert Content mit anderer object_id + other_content = "---\nid: object/other-obj\ntitle: Other\n---\n\nOther\n" + core = self._core(source_content=other_content, read_content=other_content) + result = core.reconcile(job["job_id"], "worker-1") + # object_id mismatch -> source_loader würde werfen (hier Fake liefert trotzdem) + # Wir prüfen: wenn Content != erwartet, dann TARGET_MISMATCH oder PROVENANCE_MISMATCH + self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) + self.assertIn(result["result_code"], (RC_TARGET_MISMATCH, RC_PROVENANCE_MISMATCH)) + + # -- T8: target mismatch hash -> fail-closed ----------------------------- + + def test_t8_target_mismatch_hash_fail_closed(self): + # provenance_hash im Job falsch -> PROVENANCE_MISMATCH + job = self._create_outcome_unknown(provenance_hash="f" * 64) + 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) + + # -- T9: provenance mismatch -> fail-closed ------------------------------ + + def test_t9_provenance_mismatch_fail_closed(self): + job = self._create_outcome_unknown() + # Source-Loader liefert anderen Content -> Hash recompute != Job-Hash + core = self._core(source_content="DIFFERENT SOURCE", read_content="DIFFERENT SOURCE") + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) + self.assertEqual(result["result_code"], RC_PROVENANCE_MISMATCH) + + # -- T10: source commit mismatch -> fail-closed -------------------------- + + def test_t10_source_commit_mismatch_fail_closed(self): + # source_commit im Job geändert -> source_loader liefert anderen Content + job = self._create_outcome_unknown(source_commit="b" * 40) + other_content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nOther Commit\n" + core = self._core(source_content=other_content, read_content=other_content) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) + self.assertIn(result["result_code"], (RC_TARGET_MISMATCH, RC_PROVENANCE_MISMATCH)) + + # -- T11: read unavailable -> remains unknown ---------------------------- + + def test_t11_read_unavailable_remains_unknown(self): + job = self._create_outcome_unknown() + core = self._core(read_fail=True) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) + self.assertEqual(result["result_code"], RC_READ_UNAVAILABLE) + + # -- T12: timeout -> remains unknown ------------------------------------- + + def test_t12_timeout_remains_unknown(self): + job = self._create_outcome_unknown() + # Read-Back wirft Timeout-ähnlichen Fehler + class TimeoutReadBack: + def __call__(self, vault_path): + raise TimeoutError("timeout") + core = SaveReconciliationCore(self.store, FakeSourceLoader(self.content), TimeoutReadBack()) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) + self.assertEqual(result["result_code"], RC_READ_UNAVAILABLE) + + # -- T13: 401/403 -> remains unknown ------------------------------------- + + def test_t13_auth_failure_remains_unknown(self): + job = self._create_outcome_unknown() + class AuthFailReadBack: + def __call__(self, vault_path): + raise PermissionError("401") + core = SaveReconciliationCore(self.store, FakeSourceLoader(self.content), AuthFailReadBack()) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) + self.assertEqual(result["result_code"], RC_READ_UNAVAILABLE) + + # -- T14: malformed job -> rejected -------------------------------------- + + def test_t14_malformed_job_rejected(self): + # Job mit fehlendem vault_path -> 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="", + source_commit=self.commit, provenance_hash=self.prov, + expected_state="present", created_at="2026-08-27T00:00:00Z", + idempotency_key=str(uuid.uuid4()), + ) + + # -- T15: non-OUTCOME_UNKNOWN job cannot use recovery path --------------- + + def test_t15_non_outcome_unknown_cannot_recover(self): + job = self._make_job() + self.store.create_job(job) + self.store.mark_ready(job["job_id"]) + core = self._core(read_content=self.content) + with self.assertRaises(ReconciliationError): + core.reconcile(job["job_id"], "worker-1") + + # -- T16: FAILED job cannot be magically reconciled ---------------------- + + def test_t16_failed_job_cannot_reconcile(self): + job = self._make_job() + 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.mark_failed(job["job_id"], "SOURCE_UNAVAILABLE", "worker-1") + core = self._core(read_content=self.content) + with self.assertRaises(ReconciliationError): + core.reconcile(job["job_id"], "worker-1") + + # -- T17: SUCCEEDED job remains terminal --------------------------------- + + def test_t17_succeeded_job_remains_terminal(self): + job = self._make_job() + 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.mark_succeeded(job["job_id"], "worker-1") + core = self._core(read_content=self.content) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_SUCCEEDED) + + # -- T18: immutable fields cannot change --------------------------------- + + def test_t18_immutable_fields_cannot_change(self): + job = self._create_outcome_unknown() + # Versuche, vault_path im Payload zu ändern -> Drift-Guard muss greifen + # (via _transition, der den Guard aufruft) + # Simuliere manipulierten Job: Payload vault_path != Top-Level + job["payload"]["vault_path"] = "tolaria/other.md" + # _transition liest den Job neu aus der DB (unverändert), daher prüfen + # wir den Guard direkt gegen den manipulierten Job + with self.assertRaises(JobImmutableFieldError): + self.store._assert_immutable_fields(job) + + # -- T19: no write callable available in reconciliation core ------------- + + def test_t19_no_write_callable(self): + core = self._core() + self.assertFalse(hasattr(core, "tolaria_save")) + self.assertFalse(hasattr(core, "write")) + self.assertFalse(hasattr(core, "save")) + # Nur source_loader, read_back, store + self.assertTrue(hasattr(core, "source_loader")) + self.assertTrue(hasattr(core, "read_back")) + self.assertTrue(hasattr(core, "store")) + + # -- T20: audit preserves OUTCOME_UNKNOWN -------------------------------- + + def test_t20_audit_preserves_outcome_unknown(self): + job = self._create_outcome_unknown() + trail = self.store.audit_trail(job["job_id"]) + # Der Job-Snapshot im Trail zeigt OUTCOME_UNKNOWN (vor Reconciliation) + snapshot = trail[-1] + self.assertEqual(snapshot["state"], ST_OUTCOME_UNKNOWN) + + # -- T21: audit adds reconciliation evidence ---------------------------- + + def test_t21_audit_adds_reconciliation_evidence(self): + job = self._create_outcome_unknown() + core = self._core(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.assertIn(EV_RECONCILIATION_STARTED, events) + self.assertIn(EV_RECONCILIATION_TARGET_EXACT, events) + self.assertIn(EV_CONFIRMED_EXECUTED, events) + + # -- T22: no fake EXECUTED mutation event -------------------------------- + + def test_t22_no_fake_executed_event(self): + job = self._create_outcome_unknown() + 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"] + # KEIN EXECUTED-Event (Mutation wurde NICHT neu ausgeführt) + self.assertEqual(len(executed), 0) + + # -- T23: concurrent reconcile safe -------------------------------------- + + def test_t23_concurrent_reconcile_safe(self): + job = self._create_outcome_unknown() + core1 = self._core(read_content=self.content) + core2 = self._core(read_content=self.content) + # Beide reconcilen denselben Job. Der erste finalisiert zu SUCCEEDED, + # der zweite sieht SUCCEEDED und bleibt terminal (idempotent). + 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) + + # -- T24: crash after target exact before persist safe ------------------- + + def test_t24_crash_after_target_exact_before_persist(self): + # Simuliere: Reconciliation liest Target, aber Crash vor State-Persist. + # Job bleibt OUTCOME_UNKNOWN -> erneute Reconciliation ist sicher. + job = self._create_outcome_unknown() + core = self._core(read_content=self.content) + # Erste Reconciliation "crasht" (wir rufen nur bis zur Klassifikation) + # -> hier: wir führen sie einfach aus, dann prüfen wir, dass eine + # erneute Reconciliation idempotent ist (kein Doppel-Success). + r1 = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(r1["state"], ST_SUCCEEDED) + # Erneute Reconciliation (Restart-Szenario) + r2 = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(r2["state"], ST_SUCCEEDED) + + # -- T25: crash after persist safe --------------------------------------- + + def test_t25_crash_after_persist_safe(self): + # Simuliere: Job ist RECONCILED (persistiert), aber Crash vor SUCCEEDED. + job = self._create_outcome_unknown() + self.store.mark_reconciled(job["job_id"], "worker-1") + core = self._core(read_content=self.content) + # Erneute Reconciliation: RECONCILED -> finalisiere zu SUCCEEDED + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_SUCCEEDED) + + # -- T26: restart safe --------------------------------------------------- + + def test_t26_restart_safe(self): + # Neuer Store (Restart) auf einer SEPARATEN DB (damit die gemeinsame + # cls.store-DB für die übrigen Tests offen bleibt) + tmp2 = tempfile.mkdtemp(prefix="c5-4c3-restart-") + db2 = os.path.join(tmp2, "restart.db") + store1 = JobStore(db2, "SAVE") + job = self._make_job() + store1.create_job(job) + store1.mark_ready(job["job_id"]) + store1._transition(job["job_id"], ST_CLAIMED) + store1._transition(job["job_id"], ST_EXECUTING) + store1._transition(job["job_id"], ST_OUTCOME_UNKNOWN) + store1.close() + # Restart: neuer Store auf derselben DB + store2 = JobStore(db2, "SAVE") + core = SaveReconciliationCore(store2, FakeSourceLoader(self.content), + FakeReadBack(self.content)) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["state"], ST_SUCCEEDED) + store2.close() + shutil.rmtree(tmp2, ignore_errors=True) + + # -- T27: no duplicate success event ------------------------------------ + + def test_t27_no_duplicate_success_event(self): + job = self._create_outcome_unknown() + core = self._core(read_content=self.content) + core.reconcile(job["job_id"], "worker-1") + core.reconcile(job["job_id"], "worker-1") + trail = self.store.audit_trail(job["job_id"]) + confirmed = [e for e in trail if e.get("event_type") == EV_CONFIRMED_EXECUTED] + # CONFIRMED_EXECUTED darf nicht doppelt für dieselbe Mutation erscheinen + # (zweite Reconciliation ist idempotent, aber wir erlauben max. 1 pro + # Reconciliation-Durchlauf; hier: 2 Aufrufe -> 2 Events ist ok, aber + # kein EXECUTED) + executed = [e for e in trail if e.get("event_type") == "EXECUTED"] + self.assertEqual(len(executed), 0) + + # -- T28: idempotency key preserved -------------------------------------- + + def test_t28_idempotency_key_preserved(self): + job = self._create_outcome_unknown() + idem = job["idempotency_key"] + core = self._core(read_content=self.content) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["idempotency_key"], idem) + + # -- T29: lease/claim model correct ------------------------------------- + + def test_t29_lease_claim_model_correct(self): + # Reconciliation nutzt KEINEN Execution-Claim (kein attempt_count++). + job = self._create_outcome_unknown() + self.assertEqual(job["attempt_count"], 0) + core = self._core(read_content=self.content) + result = core.reconcile(job["job_id"], "worker-1") + self.assertEqual(result["attempt_count"], 0) + # Kein claim_id/lease_until gesetzt durch Reconciliation + self.assertIsNone(result.get("claim_id")) + + # -- T30: RQ cannot override target fields ------------------------------ + + def test_t30_rq_cannot_override_target_fields(self): + # RQ versucht, vault_path im Job zu überschreiben -> Schema-Reject + # (vault_path muss valid sein; ein manipulierter Pfad wird abgelehnt) + 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()), + ) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tolaria/c5-sync-service/test_sensitivity.py b/tolaria/c5-sync-service/test_sensitivity.py new file mode 100644 index 0000000..653859c --- /dev/null +++ b/tolaria/c5-sync-service/test_sensitivity.py @@ -0,0 +1,482 @@ +""" +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)