From 373425fc0db4f281adf57dcbdf6d889c4304a2b7 Mon Sep 17 00:00:00 2001 From: Red Queen Date: Thu, 27 Aug 2026 10:09:18 +0000 Subject: [PATCH] AUTH.3E: Audit-Trail trennt REQUESTED/AUTHORIZED/EXECUTED (Fresh Checker Punkt 27) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fresh Checker identifizierte Defekt: audit_trail() lieferte nur einen Job-Snapshot, kein persistentes Audit-Event-Log. Design §18 verlangt Trennung der Phasen. - job_store: audit_events-Tabelle + _record_audit/record_audit_event; REQUESTED bei create_job - delete_executor_core: AUTHORIZED nach AUTH.3D-Validierung, EXECUTED nach Mutation - save_executor_core: AUTHORIZED nach Provenance-Validierung, EXECUTED nach Mutation - test_job_channel: +4 AuditTrailTests (REQUESTED/AUTHORIZED/EXECUTED) REPAIR_CYCLES=1. Volle Regression 444 PASS (76+50+19+299). --- .../c5-sync-service/delete_executor_core.py | 7 ++ tolaria/c5-sync-service/job_store.py | 88 ++++++++++++--- tolaria/c5-sync-service/save_executor_core.py | 7 ++ tolaria/c5-sync-service/test_job_channel.py | 101 ++++++++++++++++++ 4 files changed, 188 insertions(+), 15 deletions(-) diff --git a/tolaria/c5-sync-service/delete_executor_core.py b/tolaria/c5-sync-service/delete_executor_core.py index b74e46a..610b569 100644 --- a/tolaria/c5-sync-service/delete_executor_core.py +++ b/tolaria/c5-sync-service/delete_executor_core.py @@ -120,11 +120,18 @@ class DeleteExecutorCore: self.store.mark_failed(job_id, gate["code"], worker_id) return self.store.get_job(job_id) + # Audit: Executor hat Job + AUTH.3D-Approval validiert (AUTHORIZED). + # Ein REQUESTED-Eintrag ist KEINE Autorisierung — nur AUTHORIZED + # (nach AUTH.3D-Validierung) autorisiert die Mutation. + self.store.record_audit_event(job_id, "AUTHORIZED", worker_id=worker_id) + # Tolaria-DELETE ausführen (fail-closed) result = self.tolaria_delete(job["vault_path"]) status = result.get("status") if status == "ok": self.store.mark_succeeded(job_id, worker_id) + # Audit: Mutation ausgeführt (EXECUTED). + self.store.record_audit_event(job_id, "EXECUTED", worker_id=worker_id) elif status == "error" and result.get("uncertain"): self.store.mark_outcome_unknown(job_id, worker_id) else: diff --git a/tolaria/c5-sync-service/job_store.py b/tolaria/c5-sync-service/job_store.py index 07791dd..599ed6a 100644 --- a/tolaria/c5-sync-service/job_store.py +++ b/tolaria/c5-sync-service/job_store.py @@ -110,6 +110,22 @@ class JobStore: self._conn.execute(""" CREATE INDEX IF NOT EXISTS idx_jobs_type ON jobs(job_type) """) + # Audit-Ereignis-Log (REQUESTED / AUTHORIZED / EXECUTED). + # Trennt die Phasen: RQ hat Job erzeugt (REQUESTED), Executor hat + # Job + Approval validiert (AUTHORIZED), Mutation ausgeführt (EXECUTED). + self._conn.execute(""" + CREATE TABLE IF NOT EXISTS audit_events ( + event_id TEXT PRIMARY KEY, + job_id TEXT NOT NULL, + event_type TEXT NOT NULL, + worker_id TEXT, + result_code TEXT, + created_at INTEGER NOT NULL + ) + """) + self._conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_audit_job ON audit_events(job_id) + """) self._conn.commit() # -- Job-Type-Scope ----------------------------------------------------- @@ -154,6 +170,9 @@ class JobStore: except sqlite3.IntegrityError as e: # UNIQUE-Verletzung (idempotency_key bereits vergeben) -> Anomalie. raise JobStoreError(f"create_job failed (idempotency_key collision): {e}") from e + # Audit: RQ hat den Job erzeugt (REQUESTED). Kein REQUESTED-Eintrag ist + # eine Autorisierung — nur AUTHORIZED (nach Validierung) autorisiert. + self._record_audit(job_id, "REQUESTED") return self.get_job(job_id) def get_job(self, job_id: str) -> Optional[Dict[str, Any]]: @@ -181,6 +200,28 @@ class JobStore: # -- State-Transition (mit Immutable-Fields-Schutz) --------------------- + def _record_audit(self, job_id: str, event_type: str, *, + worker_id: Optional[str] = None, + result_code: Optional[str] = None) -> None: + """Schreibt ein Audit-Ereignis (REQUESTED/AUTHORIZED/EXECUTED).""" + now = int(time.time() * 1000) + self._conn.execute( + """ + INSERT INTO audit_events + (event_id, job_id, event_type, worker_id, result_code, created_at) + VALUES (?, ?, ?, ?, ?, ?) + """, + (str(uuid.uuid4()), job_id, event_type, worker_id, result_code, now), + ) + self._conn.commit() + + def record_audit_event(self, job_id: str, event_type: str, *, + worker_id: Optional[str] = None, + result_code: Optional[str] = None) -> None: + """Öffentlicher Audit-Eintrag (AUTHORIZED/EXECUTED) für Executor-Cores.""" + self._record_audit(job_id, event_type, worker_id=worker_id, + result_code=result_code) + def _transition(self, job_id: str, to_state: str, *, result_code: Optional[str] = None, worker_id: Optional[str] = None) -> Dict[str, Any]: """ @@ -275,22 +316,39 @@ class JobStore: # -- Audit -------------------------------------------------------------- def audit_trail(self, job_id: str) -> List[Dict[str, Any]]: - """Gibt den Audit-Trail eines Jobs zurück (aus der jobs-Tabelle).""" + """ + Gibt den Audit-Trail eines Jobs zurück: die chronologische Liste der + Audit-Ereignisse (REQUESTED / AUTHORIZED / EXECUTED) plus den aktuellen + Job-Snapshot. Ein REQUESTED-Eintrag ist KEINE Autorisierung — nur + AUTHORIZED (nach AUTH.3D-Validierung) autorisiert eine Mutation. + """ + events = self._conn.execute( + "SELECT * FROM audit_events WHERE job_id = ? ORDER BY created_at", + (job_id,), + ).fetchall() + trail = [{ + "event_id": e["event_id"], + "job_id": e["job_id"], + "event_type": e["event_type"], + "worker_id": e["worker_id"], + "result_code": e["result_code"], + "created_at": e["created_at"], + } for e in events] job = self.get_job(job_id) - if job is None: - return [] - return [{ - "job_id": job["job_id"], - "mission_id": job["mission_id"], - "job_type": job["job_type"], - "object_id": job["object_id"], - "state": job["state"], - "worker_id": job["worker_id"], - "attempt_count": job["attempt_count"], - "result_code": job["result_code"], - "created_at": job["created_at"], - "updated_at": job["updated_at"], - }] + if job is not None: + trail.append({ + "job_id": job["job_id"], + "mission_id": job["mission_id"], + "job_type": job["job_type"], + "object_id": job["object_id"], + "state": job["state"], + "worker_id": job["worker_id"], + "attempt_count": job["attempt_count"], + "result_code": job["result_code"], + "created_at": job["created_at"], + "updated_at": job["updated_at"], + }) + return trail # -- Helpers ------------------------------------------------------------ diff --git a/tolaria/c5-sync-service/save_executor_core.py b/tolaria/c5-sync-service/save_executor_core.py index d2d3e1f..1447c44 100644 --- a/tolaria/c5-sync-service/save_executor_core.py +++ b/tolaria/c5-sync-service/save_executor_core.py @@ -109,11 +109,18 @@ class SaveExecutorCore: self.store.mark_failed(job_id, RC_PROVENANCE_MISMATCH, worker_id) return self.store.get_job(job_id) + # Audit: Executor hat Job + Provenance validiert (AUTHORIZED). + # Ein REQUESTED-Eintrag ist KEINE Autorisierung — nur AUTHORIZED + # autorisiert die Mutation. + self.store.record_audit_event(job_id, "AUTHORIZED", worker_id=worker_id) + # Tolaria-SAVE ausführen (fail-closed) result = self.tolaria_save(job["vault_path"], content) status = result.get("status") if status == "ok": self.store.mark_succeeded(job_id, worker_id) + # Audit: Mutation ausgeführt (EXECUTED). + self.store.record_audit_event(job_id, "EXECUTED", worker_id=worker_id) elif status == "error" and result.get("uncertain"): self.store.mark_outcome_unknown(job_id, worker_id) else: diff --git a/tolaria/c5-sync-service/test_job_channel.py b/tolaria/c5-sync-service/test_job_channel.py index 7dc3f5d..a0579f3 100644 --- a/tolaria/c5-sync-service/test_job_channel.py +++ b/tolaria/c5-sync-service/test_job_channel.py @@ -624,5 +624,106 @@ class StateMachineTests(unittest.TestCase): transition(ST_OUTCOME_UNKNOWN, ST_READY) +class AuditTrailTests(unittest.TestCase): + """Audit-Trail trennt REQUESTED / AUTHORIZED / EXECUTED (Design §18).""" + + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + self.save_db = os.path.join(self.tmpdir, "c5a_save.db") + self.delete_db = os.path.join(self.tmpdir, "c5a_delete.db") + self.http_calls = [] + self.approvals = {} + self.content = "hello world" + self.provenance = _sha256(self.content) + + def _source_loader(self, commit, object_id): + return self.content + + def _tolaria_save(self, vault_path, content): + self.http_calls.append(("SAVE", vault_path)) + return {"status": "ok"} + + def _approval_loader(self, approval_id): + return self.approvals.get(approval_id) + + def _approval_verify(self, approval): + return {"valid": approval.get("signature_valid", False), "reason": "ok"} + + def _tolaria_delete(self, vault_path): + self.http_calls.append(("DELETE", vault_path)) + return {"status": "ok"} + + def test_audit_save_requested_authorized_executed(self): + # SAVE: REQUESTED (RQ erzeugt) -> AUTHORIZED (Executor validiert) + # -> EXECUTED (Mutation ausgeführt). + store = JobStore(self.save_db, "SAVE") + job = _make_save_job(provenance_hash=self.provenance) + store.create_job(job) + store.mark_ready(job["job_id"]) + exec_core = SaveExecutorCore(store, self._source_loader, self._tolaria_save) + exec_core.process_job(job["job_id"], "save-worker") + trail = store.audit_trail(job["job_id"]) + types = [e["event_type"] for e in trail if "event_type" in e] + self.assertIn("REQUESTED", types) + self.assertIn("AUTHORIZED", types) + self.assertIn("EXECUTED", types) + # REQUESTED kommt vor AUTHORIZED vor EXECUTED + self.assertLess(types.index("REQUESTED"), types.index("AUTHORIZED")) + self.assertLess(types.index("AUTHORIZED"), types.index("EXECUTED")) + store.close() + + def test_audit_delete_requested_authorized_executed(self): + # DELETE: REQUESTED -> AUTHORIZED (nach AUTH.3D-Validierung) -> EXECUTED. + store = JobStore(self.delete_db, "DELETE") + job = _make_delete_job() + self.approvals[job["approval_id"]] = _valid_approval(job) + store.create_job(job) + store.mark_ready(job["job_id"]) + exec_core = DeleteExecutorCore( + store, self._approval_loader, self._approval_verify, self._tolaria_delete + ) + exec_core.process_job(job["job_id"], "delete-worker") + trail = store.audit_trail(job["job_id"]) + types = [e["event_type"] for e in trail if "event_type" in e] + self.assertIn("REQUESTED", types) + self.assertIn("AUTHORIZED", types) + self.assertIn("EXECUTED", types) + self.assertLess(types.index("REQUESTED"), types.index("AUTHORIZED")) + self.assertLess(types.index("AUTHORIZED"), types.index("EXECUTED")) + store.close() + + def test_audit_denied_delete_has_no_executed(self): + # DELETE ohne Approval -> denied. Es gibt REQUESTED, aber KEIN + # AUTHORIZED und KEIN EXECUTED (keine Mutation autorisiert). + store = JobStore(self.delete_db, "DELETE") + job = _make_delete_job() + store.create_job(job) + store.mark_ready(job["job_id"]) + exec_core = DeleteExecutorCore( + store, self._approval_loader, self._approval_verify, self._tolaria_delete + ) + result = exec_core.process_job(job["job_id"], "delete-worker") + self.assertEqual(result["result_code"], "APPROVAL_MISSING") + trail = store.audit_trail(job["job_id"]) + types = [e["event_type"] for e in trail if "event_type" in e] + self.assertIn("REQUESTED", types) + self.assertNotIn("AUTHORIZED", types) + self.assertNotIn("EXECUTED", types) + self.assertEqual(self.http_calls, []) + store.close() + + def test_audit_requested_is_not_authorization(self): + # Ein REQUESTED-Eintrag allein autorisiert nichts: ein DELETE-Job, + # der nur REQUESTED ist (nie AUTHORIZED), führt zu keiner Mutation. + store = JobStore(self.delete_db, "DELETE") + job = _make_delete_job() + store.create_job(job) + trail = store.audit_trail(job["job_id"]) + types = [e["event_type"] for e in trail if "event_type" in e] + self.assertEqual(types, ["REQUESTED"]) + self.assertEqual(self.http_calls, []) + store.close() + + if __name__ == "__main__": unittest.main()