nexo312/trading-system-docs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

AUTH.4C3: OUTCOME_UNKNOWN read-only reconciliation contract — RECONCILED state, immutable-field guard, provenance recompute, fail-closed classification, 61 isolated tests (reconciliation 30, incident fixture 5, adversarial 14, sensitivity 12)

Browse source
This commit is contained in:
Red Queen 2026-08-27 15:57:06 +00:00
parent 2e631fd5fa
commit dbc31ad1bb
7 changed files with 1807 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
tolaria/c5-sync-service/job_state_machine.py
View file

@ -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. Deterministische Job-State-Machine für den Executor Command Channel.
States: States:
CREATED, READY, CLAIMED, EXECUTING, SUCCEEDED, FAILED, REJECTED, OUTCOME_UNKNOWN CREATED, READY, CLAIMED, EXECUTING, SUCCEEDED, FAILED, REJECTED, OUTCOME_UNKNOWN,
RECONCILED
Eigenschaften: Eigenschaften:
* Geschlossene Transition-Allowlist * Geschlossene Transition-Allowlist
@ -12,6 +13,10 @@ Eigenschaften:
* DELETE: KEIN RETRYABLE — ein gestarteter DELETE wird bei Lease-Expiry in * DELETE: KEIN RETRYABLE — ein gestarteter DELETE wird bei Lease-Expiry in
OUTCOME_UNKNOWN überführt, NICHT zurück zu READY (kein blinder Retry) OUTCOME_UNKNOWN überführt, NICHT zurück zu READY (kein blinder Retry)
* SAVE: RETRYABLE nur bei transienten Fehlern (TolariaUnavailable) * 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). Isoliert implementiert (KEIN produktiver Container).
""" """
@ -31,10 +36,11 @@ ST_SUCCEEDED = "SUCCEEDED"
ST_FAILED = "FAILED" ST_FAILED = "FAILED"
ST_REJECTED = "REJECTED" ST_REJECTED = "REJECTED"
ST_OUTCOME_UNKNOWN = "OUTCOME_UNKNOWN" ST_OUTCOME_UNKNOWN = "OUTCOME_UNKNOWN"
ST_RECONCILED = "RECONCILED" # AUTH.4C3: read-only Reconciliation abgeschlossen
ALL_STATES = frozenset({ ALL_STATES = frozenset({
ST_CREATED, ST_READY, ST_CLAIMED, ST_EXECUTING, 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 (ST_EXECUTING, ST_REJECTED), # Gate-Fehler während Execution
# OUTCOME_UNKNOWN (terminal für DELETE; SAVE kann via Human/Admin reaktiviert werden) # OUTCOME_UNKNOWN (terminal für DELETE; SAVE kann via Human/Admin reaktiviert werden)
(ST_OUTCOME_UNKNOWN, ST_REJECTED), # expliziter Human-/Admin-Eingriff (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) # Terminal-States (keine weiteren Transitionen ohne expliziten Eingriff)

66
tolaria/c5-sync-service/job_store.py
View file

@ -35,6 +35,7 @@ from job_state_machine import (
ST_FAILED, ST_FAILED,
ST_OUTCOME_UNKNOWN, ST_OUTCOME_UNKNOWN,
ST_READY, ST_READY,
ST_RECONCILED,
ST_REJECTED, ST_REJECTED,
ST_SUCCEEDED, ST_SUCCEEDED,
InvalidTransitionError, InvalidTransitionError,
@ -233,6 +234,12 @@ class JobStore:
raise JobNotFoundError(f"job not found: {job_id}") raise JobNotFoundError(f"job not found: {job_id}")
from_state = job["state"] from_state = job["state"]
transition(from_state, to_state) # wirft bei ungültiger Transition 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) now = int(time.time() * 1000)
self._conn.execute( self._conn.execute(
""" """
@ -260,6 +267,65 @@ class JobStore:
def mark_outcome_unknown(self, job_id: str, worker_id: str) -> Dict[str, Any]: 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) 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 ------------------------------------------------------ # -- Claim / Lease ------------------------------------------------------
def claim_job(self, job_id: str, worker_id: str, lease_seconds: int = 60) -> Dict[str, Any]: def claim_job(self, job_id: str, worker_id: str, lease_seconds: int = 60) -> Dict[str, Any]:

247
tolaria/c5-sync-service/save_reconciliation.py Normal file
View file

@ -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"]

288
tolaria/c5-sync-service/test_adversarial.py Normal file
View file

@ -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)

203
tolaria/c5-sync-service/test_productive_incident_fixture.py Normal file
View file

@ -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)

508
tolaria/c5-sync-service/test_reconciliation.py Normal file
View file

@ -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)

482
tolaria/c5-sync-service/test_sensitivity.py Normal file
View file

@ -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)