diff --git a/tolaria/c5-sync-service/test_delete_worker.py b/tolaria/c5-sync-service/test_delete_worker.py new file mode 100644 index 0000000..7e0a44f --- /dev/null +++ b/tolaria/c5-sync-service/test_delete_worker.py @@ -0,0 +1,696 @@ +#!/usr/bin/env python3 +""" +AUTH.4D — test_delete_worker.py (P10 Isolated Test Contract) +============================================================= +Isolierte Tests für den DELETE-scoped Worker + DeleteExecutorCore + +TolariaClient.delete() gegen Fakes. KEIN echtes HTTP, KEINE produktive DB, +KEIN produktiver Vault. + +Abgedeckte Fälle (T1–T40): + T1 DELETE Worker importiert KEIN save_executor_core + T2 DELETE-Pfad kann write() NICHT aufrufen (keine write-Capability) + T3 SAVE-Token allein kann DELETE NICHT autorisieren + T4 DELETE-Token allein kann Approval NICHT umgehen + T5 gültige synthetische Approval + gültiges synthetisches DELETE-Token + erreicht die Mutationsgrenze (nur im Fake-Fixture) + T6 kein echtes HTTP (Fake-Client, kein Netz) + T7 keine produktive DB (temp-Datei) + T8 kein produktiver Vault (Fake) + T9 DELETE ohne Approval -> BLOCKED (APPROVAL_MISSING) + T10 DELETE mit ungültiger Signatur -> BLOCKED (APPROVAL_INVALID) + T11 DELETE mit abgelaufener Approval -> BLOCKED (APPROVAL_INVALID) + T12 DELETE mit consumed Approval -> BLOCKED (APPROVAL_INVALID) + T13 DELETE mit falschem Objekt -> BLOCKED (OBJECT_MISMATCH) + T14 DELETE mit falschem Pfad -> BLOCKED (PATH_MISMATCH) + T15 DELETE mit falschem Commit -> BLOCKED (COMMIT_MISMATCH) + T16 DELETE mit falscher Provenance -> BLOCKED (PROVENANCE_MISMATCH) + T17 DELETE mit falscher Mission -> BLOCKED (MISSION_MISMATCH) + T18 DELETE ohne DELETE-Token -> fail-closed (CREDENTIAL_MISSING) + T19 DELETE mit falschem Token (401) -> fail-closed (AUTH_FAILURE) + T20 DELETE mit Timeout -> OUTCOME_UNKNOWN (kein blinder Retry) + T21 Job allein (ohne Approval) autorisiert NIE + T22 SAVE-Job wird vom DELETE-Worker REJECTED + T23 zweiter DELETE auf gleiche Approval -> single-use verhindert + T24 Replay (Nonce wiederverwendet) -> BLOCKED + T25 Approval für anderes Objekt autorisiert DELETE NIE + T26 Approval für anderen Commit autorisiert DELETE NIE + T27 Approval für anderen Pfad autorisiert DELETE NIE + T28 Approval für andere Mission autorisiert DELETE NIE + T29 Approval für anderen Request autorisiert DELETE NIE + T30 kein generic dispatcher (kein command/url/method im Job) + T31 kein shell/python im Job + T32 kein arbitrary endpoint (fester /delete) + T33 kein arbitrary method (nur POST) + T34 kein arbitrary header + T35 kein SAVE fallback + T36 kein Docker-Socket / SSH im Executor + T37 kein Private Key im Executor + T38 kein SAVE-Token im DELETE Executor + T39 kein DELETE-Token im SAVE Executor + T40 OUTCOME_UNKNOWN -> kein blinder Retry +""" + +from __future__ import annotations + +import os +import sys +import tempfile +import unittest +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, Optional + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from approval_payload import build_payload, parse_payload +from approval_signature import generate_test_keypair, key_id, sign +from approval_state import ApprovalStateStore +from approval_verifier import ApprovalVerifier +from delete_executor_core import DeleteExecutorCore +from job_schema import JOB_TYPE_DELETE, JOB_TYPE_SAVE, make_delete_job, make_save_job +from job_store import JobStore, JobImmutableFieldError +from delete_tolaria_client import TolariaClient, TolariaDeleteError + +# Synthetisches Test-Keypair (NUR Tests, KEIN echter Key) +PRIV, PUB = generate_test_keypair() +PUB_KEY_ID = key_id(PUB) + +COMMIT = "5f3adda73fba2f7ae284efd5ec3d244e087a69a8" +PROV = "798ae0315c14aa9d7e3a3f759c952f7c18cbce5154844f447e597b7fd1776515" +MISSION = str(uuid.uuid4()) +DELETE_REQ = str(uuid.uuid4()) +OBJECT = str(uuid.uuid4()) +PATH = "docs/auth4d-canary.md" +NONCE = str(uuid.uuid4()) +_NOW = datetime.now(timezone.utc) +ISSUED = (_NOW - timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ") +EXPIRES = (_NOW + timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ") +TEST_DELETE_TOKEN = "a" * 64 + + +def make_approval(*, object_id=OBJECT, vault_path=PATH, expected_commit=COMMIT, + expected_provenance_hash=PROV, mission_id=MISSION, + delete_request_id=DELETE_REQ, nonce=NONCE, + issued_at=ISSUED, expires_at=EXPIRES) -> Dict[str, Any]: + """Baut eine synthetische, signierte Approval.""" + payload = build_payload( + approval_id=str(uuid.uuid4()), + mission_id=mission_id, + delete_request_id=delete_request_id, + object_id=object_id, + vault_path=vault_path, + expected_commit=expected_commit, + expected_provenance_hash=expected_provenance_hash, + nonce=nonce, + issued_at=issued_at, + expires_at=expires_at, + ) + sig = sign(PRIV, payload) + return { + "raw_payload": __import__("json").dumps(payload, separators=(",", ":")), + "signature": sig, + "approval_id": payload["approval_id"], + "mission_id": mission_id, + "delete_request_id": delete_request_id, + "object_id": object_id, + "vault_path": vault_path, + "expected_commit": expected_commit, + "expected_provenance_hash": expected_provenance_hash, + "nonce": nonce, + "consumed": False, + "state_allows_delete": True, + "used_nonces": set(), + } + + +def make_delete_job_with_approval(approval: Dict[str, Any], *, job_type=JOB_TYPE_DELETE, + object_id=OBJECT, vault_path=PATH, + expected_commit=COMMIT, expected_provenance_hash=PROV, + mission_id=MISSION, delete_request_id=DELETE_REQ, + approval_id=None) -> Dict[str, Any]: + """Baut einen DELETE-Job, der auf eine Approval (im Store) verweist.""" + return make_delete_job( + job_id=str(uuid.uuid4()), + mission_id=mission_id, + delete_request_id=delete_request_id, + object_id=object_id, + vault_path=vault_path, + expected_commit=expected_commit, + expected_provenance_hash=expected_provenance_hash, + approval_id=approval_id or approval["approval_id"], + created_at="2026-08-27T12:00:00Z", + idempotency_key=str(uuid.uuid4()), + ) + + +class FakeTolaria: + """In-Memory Fake der Tolaria-Vault-API (delete/read).""" + + def __init__(self): + self.vault: Dict[str, str] = {} + self.delete_count = 0 + self.delete_calls: list = [] + self.fail_next = None # ("error", code, uncertain) + self.unavailable = False + + def delete(self, vault_path: str) -> Dict[str, Any]: + """Erfüllt das DeleteExecutorCore-Contract: gibt dict zurück, wirft nicht.""" + if self.unavailable: + return {"status": "error", "code": "TOLARIA_UNAVAILABLE", "uncertain": True} + if self.fail_next: + _, code, uncertain = self.fail_next + self.fail_next = None + if code == "AUTH_FAILURE": + return {"status": "error", "code": "AUTH_FAILURE", "uncertain": False} + if uncertain: + return {"status": "error", "code": "TOLARIA_UNAVAILABLE", "uncertain": True} + self.delete_count += 1 + self.delete_calls.append(vault_path) + self.vault.pop(vault_path, None) + return {"status": "ok"} + + def read(self, vault_path: str) -> Optional[str]: + return self.vault.get(vault_path) + + +class DeleteWorkerTest(unittest.TestCase): + """Isolierte Tests für den DELETE-Worker-Pfad.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.db_path = os.path.join(self.tmp, "c5a_delete.db") + self.store = JobStore(self.db_path, "DELETE") + self.fake = FakeTolaria() + self.verifier = ApprovalVerifier(public_keys={PUB_KEY_ID: PUB}) + self.approval_store = ApprovalStateStore() + + def tearDown(self): + self.store.close() + + def _register_approval(self, approval: Dict[str, Any]) -> None: + """Registriert eine Approval im State-Store (Single-Use/Nonce). + + Das Store-Record enthält alle Felder, die DeleteExecutorCore aus der + Approval liest (object_id, vault_path, expected_commit, + expected_provenance_hash, mission_id, expired, consumed). + """ + self.approval_store.create(approval["approval_id"], approval["nonce"]) + self.approval_store._approvals[approval["approval_id"]].update({ + "raw_payload": approval["raw_payload"], + "signature": approval["signature"], + "mission_id": approval["mission_id"], + "delete_request_id": approval["delete_request_id"], + "object_id": approval["object_id"], + "vault_path": approval["vault_path"], + "expected_commit": approval["expected_commit"], + "expected_provenance_hash": approval["expected_provenance_hash"], + "nonce": approval["nonce"], + "consumed": approval.get("consumed", False), + "expired": approval.get("expired", False), + "state_allows_delete": approval.get("state_allows_delete", True), + "used_nonces": approval.get("used_nonces", set()), + }) + + def _make_approved_job(self, approval: Optional[Dict[str, Any]] = None, + **kw) -> tuple[Dict[str, Any], Dict[str, Any]]: + """Erstellt + registriert eine Approval und einen referenzierenden Job.""" + approval = approval or make_approval(**kw) + self._register_approval(approval) + job = make_delete_job_with_approval(approval) + self.store.create_job(job) + self.store.mark_ready(job["job_id"]) + return job, approval + + def _core(self, delete_token=TEST_DELETE_TOKEN, verifier=None) -> DeleteExecutorCore: + client = TolariaClient(base_url="http://fake", delete_token=delete_token) + v = verifier or self.verifier + + def approval_loader(approval_id: str) -> Optional[Dict[str, Any]]: + return self.approval_store.get(approval_id) + + def approval_verify(approval: Dict[str, Any]) -> Dict[str, Any]: + try: + # NUR Signatur kryptographisch prüfen (Ed25519 PUBLIC KEY ONLY). + # Die Zustands-Checks (expired/consumed/nonce) macht der Core + # isoliert (Defense-in-Depth: Verifier + Core prüfen getrennt). + payload = parse_payload(approval["raw_payload"]) + v.verify_approval( + approval["raw_payload"], approval["signature"], + expected_mission_id=payload["mission_id"], + expected_delete_request_id=payload["delete_request_id"], + expected_object_id=payload["object_id"], + expected_vault_path=payload["vault_path"], + expected_commit=payload["expected_commit"], + expected_provenance_hash=payload["expected_provenance_hash"], + expected_nonce=payload["nonce"], + # Zustands-Checks dem Core überlassen (isoliert testbar): + consumed=False, + state_allows_delete=True, + used_nonces=set(), + ) + return {"valid": True, "reason": "ok"} + except Exception as e: + return {"valid": False, "code": getattr(e, "reason_code", "DELETE_DENIED"), + "reason": str(e)} + + def tolaria_delete(vault_path: str) -> Dict[str, Any]: + try: + client.delete(vault_path) + return {"status": "ok"} + except Exception as e: + code = getattr(e, "code", "TOLARIA_UNAVAILABLE") + if code == "CREDENTIAL_MISSING": + return {"status": "error", "code": "CREDENTIAL_MISSING", "uncertain": False} + if code in ("AUTH_FAILURE", "INVALID_SCHEMA"): + return {"status": "error", "code": code, "uncertain": False} + return {"status": "error", "code": "TOLARIA_UNAVAILABLE", "uncertain": True} + + return DeleteExecutorCore(self.store, approval_loader, approval_verify, tolaria_delete) + + # -- T1: DELETE Worker importiert KEIN save_executor_core ---------------- + def test_t1_worker_imports_no_save_core(self): + import delete_worker + src = open(delete_worker.__file__).read() + self.assertNotIn("save_executor_core", src) + self.assertNotIn("SaveExecutorCore", src) + self.assertNotIn("forgejo_source_loader", src) + + # -- T2: DELETE-Pfad kann write() NICHT aufrufen ------------------------- + def test_t2_delete_path_cannot_call_write(self): + client = TolariaClient(base_url="http://fake", delete_token=TEST_DELETE_TOKEN) + self.assertFalse(hasattr(client, "write")) + self.assertTrue(hasattr(client, "delete")) + + # -- T3: SAVE-Token allein kann DELETE NICHT autorisieren ---------------- + def test_t3_save_token_alone_cannot_authorize_delete(self): + client = TolariaClient(base_url="http://fake", delete_token=None) + with self.assertRaises(TolariaDeleteError) as ctx: + client.delete("x.md") + self.assertEqual(ctx.exception.code, "CREDENTIAL_MISSING") + + # -- T4: DELETE-Token allein kann Approval NICHT umgehen ----------------- + def test_t4_delete_token_alone_cannot_bypass_approval(self): + job = make_delete_job( + job_id=str(uuid.uuid4()), mission_id=MISSION, delete_request_id=DELETE_REQ, + object_id=OBJECT, vault_path=PATH, expected_commit=COMMIT, + expected_provenance_hash=PROV, approval_id=str(uuid.uuid4()), + created_at="2026-08-27T12:00:00Z", idempotency_key=str(uuid.uuid4())) + self.store.create_job(job) + self.store.mark_ready(job["job_id"]) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t4") + self.assertEqual(result["state"], "FAILED") + self.assertEqual(result["result_code"], "APPROVAL_MISSING") + self.assertEqual(self.fake.delete_count, 0) + + # -- T5: gültige Approval + gültiges Token erreicht Mutationsgrenze ------ + def test_t5_valid_approval_reaches_mutation_boundary(self): + job, _ = self._make_approved_job() + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t5") + self.assertEqual(result["state"], "SUCCEEDED") + self.assertEqual(self.fake.delete_count, 1) + self.assertEqual(self.fake.delete_calls, [PATH]) + + # -- T6: kein echtes HTTP ------------------------------------------------ + def test_t6_no_real_http(self): + client = TolariaClient(base_url="http://127.0.0.1:1", delete_token=None) + with self.assertRaises(TolariaDeleteError): + client.delete("x.md") + + # -- T7: keine produktive DB --------------------------------------------- + def test_t7_no_production_db(self): + self.assertTrue(self.db_path.startswith(tempfile.gettempdir()) or "/tmp" in self.db_path) + + # -- T8: kein produktiver Vault ------------------------------------------ + def test_t8_no_production_vault(self): + self.assertIsInstance(self.fake, FakeTolaria) + + # -- T9: DELETE ohne Approval -> BLOCKED --------------------------------- + def test_t9_delete_without_approval_blocked(self): + job = make_delete_job( + job_id=str(uuid.uuid4()), mission_id=MISSION, delete_request_id=DELETE_REQ, + object_id=OBJECT, vault_path=PATH, expected_commit=COMMIT, + expected_provenance_hash=PROV, approval_id=str(uuid.uuid4()), + created_at="2026-08-27T12:00:00Z", idempotency_key=str(uuid.uuid4())) + self.store.create_job(job) + self.store.mark_ready(job["job_id"]) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t9") + self.assertEqual(result["result_code"], "APPROVAL_MISSING") + self.assertEqual(self.fake.delete_count, 0) + + # -- T10: ungültige Signatur -> BLOCKED ---------------------------------- + def test_t10_invalid_signature_blocked(self): + approval = make_approval() + approval["signature"] = "a" * 88 # kaputte Signatur + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t10") + self.assertEqual(result["result_code"], "APPROVAL_INVALID") + self.assertEqual(self.fake.delete_count, 0) + + # -- T11: abgelaufene Approval -> BLOCKED -------------------------------- + def test_t11_expired_approval_blocked(self): + # Gültige Signatur, aber Store-Record markiert expired=True + # (Core-Guard isoliert: Verifier prüft Signatur, Core prüft Zustand). + approval = make_approval() + approval["expired"] = True + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t11") + self.assertEqual(result["result_code"], "APPROVAL_EXPIRED") + self.assertEqual(self.fake.delete_count, 0) + + # -- T12: consumed Approval -> BLOCKED ----------------------------------- + def test_t12_consumed_approval_blocked(self): + approval = make_approval() + approval["consumed"] = True + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t12") + self.assertEqual(result["result_code"], "APPROVAL_CONSUMED") + self.assertEqual(self.fake.delete_count, 0) + + # -- T13: falsches Objekt -> BLOCKED ------------------------------------- + def test_t13_wrong_object_blocked(self): + approval = make_approval(object_id=str(uuid.uuid4())) + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t13") + self.assertEqual(result["result_code"], "OBJECT_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T14: falscher Pfad -> BLOCKED --------------------------------------- + def test_t14_wrong_path_blocked(self): + approval = make_approval(vault_path="docs/other.md") + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t14") + self.assertEqual(result["result_code"], "PATH_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T15: falscher Commit -> BLOCKED ------------------------------------- + def test_t15_wrong_commit_blocked(self): + approval = make_approval(expected_commit="a" * 40) + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t15") + self.assertEqual(result["result_code"], "COMMIT_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T16: falsche Provenance -> BLOCKED ---------------------------------- + def test_t16_wrong_provenance_blocked(self): + approval = make_approval(expected_provenance_hash="b" * 64) + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t16") + self.assertEqual(result["result_code"], "PROVENANCE_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T17: falsche Mission -> BLOCKED ------------------------------------- + def test_t17_wrong_mission_blocked(self): + approval = make_approval(mission_id=str(uuid.uuid4())) + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t17") + self.assertEqual(result["result_code"], "MISSION_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T18: ohne DELETE-Token -> fail-closed ------------------------------- + def test_t18_no_delete_token_fail_closed(self): + client = TolariaClient(base_url="http://fake", delete_token=None) + with self.assertRaises(TolariaDeleteError) as ctx: + client.delete(PATH) + self.assertEqual(ctx.exception.code, "CREDENTIAL_MISSING") + + # -- T19: falsches Token (401) -> fail-closed ---------------------------- + def test_t19_wrong_token_auth_failure(self): + self.fake.fail_next = ("error", "AUTH_FAILURE", False) + job, _ = self._make_approved_job() + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t19") + self.assertEqual(result["state"], "FAILED") + self.assertEqual(result["result_code"], "AUTH_FAILURE") + + # -- T20: Timeout -> OUTCOME_UNKNOWN (kein blinder Retry) ---------------- + def test_t20_timeout_outcome_unknown(self): + self.fake.fail_next = ("error", "TOLARIA_UNAVAILABLE", True) + job, _ = self._make_approved_job() + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t20") + self.assertEqual(result["state"], "OUTCOME_UNKNOWN") + + # -- T21: Job allein autorisiert NIE ------------------------------------- + def test_t21_job_alone_never_authorizes(self): + job = make_delete_job( + job_id=str(uuid.uuid4()), mission_id=MISSION, delete_request_id=DELETE_REQ, + object_id=OBJECT, vault_path=PATH, expected_commit=COMMIT, + expected_provenance_hash=PROV, approval_id=str(uuid.uuid4()), + created_at="2026-08-27T12:00:00Z", idempotency_key=str(uuid.uuid4())) + self.store.create_job(job) + self.store.mark_ready(job["job_id"]) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t21") + self.assertEqual(result["result_code"], "APPROVAL_MISSING") + self.assertEqual(self.fake.delete_count, 0) + + # -- T22: SAVE-Job wird vom DELETE-Worker REJECTED ----------------------- + def test_t22_save_job_rejected_by_delete_worker(self): + job = make_save_job( + job_id=str(uuid.uuid4()), mission_id=MISSION, object_id=OBJECT, + vault_path=PATH, source_commit=COMMIT, provenance_hash=PROV, + expected_state="x", created_at="2026-08-27T12:00:00Z", + idempotency_key=str(uuid.uuid4())) + self.store.create_job(job) + self.store.mark_ready(job["job_id"]) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t22") + self.assertEqual(result["state"], "REJECTED") + self.assertEqual(self.fake.delete_count, 0) + + # -- T23: zweiter DELETE auf gleiche Approval -> single-use -------------- + def test_t23_second_delete_single_use(self): + job, approval = self._make_approved_job() + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t23a") + self.assertEqual(result["state"], "SUCCEEDED") + self.assertEqual(self.fake.delete_count, 1) + # Zweiter Job mit derselben Approval (consumed) -> BLOCKED + # Neue Approval-ID, aber dieselbe Nonce (Replay) + consumed=True + approval2 = make_approval(nonce=approval["nonce"]) + approval2["consumed"] = True + job2, _ = self._make_approved_job(approval2) + result2 = core.process_job(job2["job_id"], "w-t23b") + self.assertEqual(result2["result_code"], "APPROVAL_CONSUMED") + self.assertEqual(self.fake.delete_count, 1) + + # -- T24: Replay (Nonce wiederverwendet) -> BLOCKED ---------------------- + def test_t24_replay_nonce_blocked(self): + approval = make_approval() + approval["used_nonces"] = {approval["nonce"]} # Nonce bereits verwendet + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t24") + self.assertEqual(result["result_code"], "APPROVAL_CONSUMED") + self.assertEqual(self.fake.delete_count, 0) + + # -- T25: Approval für anderes Objekt autorisiert NIE -------------------- + def test_t25_approval_other_object_never_authorizes(self): + approval = make_approval(object_id=str(uuid.uuid4())) + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t25") + self.assertEqual(result["result_code"], "OBJECT_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T26: Approval für anderen Commit autorisiert NIE -------------------- + def test_t26_approval_other_commit_never_authorizes(self): + approval = make_approval(expected_commit="a" * 40) + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t26") + self.assertEqual(result["result_code"], "COMMIT_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T27: Approval für anderen Pfad autorisiert NIE ---------------------- + def test_t27_approval_other_path_never_authorizes(self): + approval = make_approval(vault_path="docs/other.md") + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t27") + self.assertEqual(result["result_code"], "PATH_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T28: Approval für andere Mission autorisiert NIE ------------------- + def test_t28_approval_other_mission_never_authorizes(self): + approval = make_approval(mission_id=str(uuid.uuid4())) + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t28") + self.assertEqual(result["result_code"], "MISSION_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T29: Approval für anderen Request autorisiert NIE ------------------- + def test_t29_approval_other_request_never_authorizes(self): + approval = make_approval(delete_request_id=str(uuid.uuid4())) + job, _ = self._make_approved_job(approval) + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t29") + self.assertEqual(result["result_code"], "APPROVAL_MISMATCH") + self.assertEqual(self.fake.delete_count, 0) + + # -- T30: kein generic dispatcher ---------------------------------------- + def test_t30_no_generic_dispatcher(self): + from job_schema import FORBIDDEN_FIELDS + for f in ("command", "endpoint", "url", "method", "shell", "python", + "handler", "exec", "script", "headers", "authorization", + "credential", "token", "password", "secret", "sql", "query", + "docker", "ssh", "base_url", "http_method", "auth_header"): + self.assertIn(f, FORBIDDEN_FIELDS) + + # -- T31: kein shell/python im Job --------------------------------------- + def test_t31_no_shell_python_in_job(self): + from job_schema import FORBIDDEN_FIELDS + self.assertIn("shell", FORBIDDEN_FIELDS) + self.assertIn("python", FORBIDDEN_FIELDS) + + # -- T32: kein arbitrary endpoint (fester /delete) ----------------------- + def test_t32_fixed_delete_endpoint(self): + from delete_tolaria_client import ENDPOINT_DELETE + self.assertEqual(ENDPOINT_DELETE, "delete") + + # -- T33: kein arbitrary method (nur POST) ------------------------------ + def test_t33_fixed_method_post(self): + import delete_tolaria_client + src = open(delete_tolaria_client.__file__).read() + self.assertIn('method="POST"', src) + self.assertNotIn('method="DELETE"', src) + + # -- T34: kein arbitrary header ------------------------------------------ + def test_t34_no_arbitrary_header(self): + import delete_tolaria_client + src = open(delete_tolaria_client.__file__).read() + # Nur der Code (nicht Docstring): kein dynamischer headers-Parameter + self.assertNotIn("def _post(self, endpoint, payload, auth_token, headers", src) + self.assertNotIn("headers: Dict[str, str] = None", src) + + # -- T35: kein SAVE fallback --------------------------------------------- + def test_t35_no_save_fallback(self): + import delete_tolaria_client + src = open(delete_tolaria_client.__file__).read() + self.assertNotIn("ENDPOINT_SAVE", src) + self.assertNotIn("def write", src) + + # -- T36: kein Docker-Socket / SSH im Executor --------------------------- + def test_t36_no_docker_socket_ssh(self): + import delete_worker + src = open(delete_worker.__file__).read() + self.assertNotIn("docker", src.lower()) + self.assertNotIn("ssh", src.lower()) + + # -- T37: kein Private Key im Executor ----------------------------------- + def test_t37_no_private_key_in_executor(self): + import delete_worker + src = open(delete_worker.__file__).read() + self.assertNotIn("private_key", src.lower()) + self.assertNotIn("PRIVATE", src) + + # -- T38: kein SAVE-Token im DELETE Executor ----------------------------- + def test_t38_no_save_token_in_delete_executor(self): + import delete_worker + src = open(delete_worker.__file__).read() + self.assertNotIn("TOLARIA_SAVE_TOKEN", src) + self.assertIn("TOLARIA_DELETE_TOKEN", src) + + # -- T39: kein DELETE-Token im SAVE Executor ----------------------------- + def test_t39_no_delete_token_in_save_executor(self): + sot_worker = os.path.join(os.path.dirname(os.path.abspath(__file__)), + "..", "sot_worker.py") + if os.path.exists(sot_worker): + src = open(sot_worker).read() + self.assertNotIn("TOLARIA_DELETE_TOKEN", src) + + # -- T40: OUTCOME_UNKNOWN -> kein blinder Retry -------------------------- + def test_t40_outcome_unknown_no_blind_retry(self): + self.fake.fail_next = ("error", "TOLARIA_UNAVAILABLE", True) + job, _ = self._make_approved_job() + core = self._core() + core.tolaria_delete = self.fake.delete + result = core.process_job(job["job_id"], "w-t40") + self.assertEqual(result["state"], "OUTCOME_UNKNOWN") + self.assertEqual(self.fake.delete_count, 0) + + # -- T41: zweiter process_job auf denselben Job -> Claim-Guard ---------- + # K: second-delete guard. Der atomare Claim (READY->CLAIMED) verhindert, + # dass derselbe DELETE-Job zweimal ausgeführt wird (kein Doppel-Delete). + def test_t41_second_process_same_job_blocked(self): + job, _ = self._make_approved_job() + core = self._core() + core.tolaria_delete = self.fake.delete + result1 = core.process_job(job["job_id"], "w-t41a") + self.assertEqual(result1["state"], "SUCCEEDED") + self.assertEqual(self.fake.delete_count, 1) + # Zweiter process_job auf denselben Job: Claim-Guard greift + # (JobAlreadyClaimedError -> kein zweiter DELETE). + result2 = core.process_job(job["job_id"], "w-t41b") + self.assertEqual(self.fake.delete_count, 1) # kein zweiter DELETE + + # -- T42: TOCTOU — Payload-Manipulation nach Claim wird erkannt ---------- + # M: immutable guard. Der echte TOCTOU-Schutz ist _assert_immutable_fields + # in _transition (job_store.py), das die rohe DB-Spalte gegen das Payload + # prüft. Eine Payload-Manipulation nach dem Claim (z.B. object_id) führt zu + # JobImmutableFieldError -> kein State-Change, kein DELETE. + def test_t42_toctou_re_read_after_claim(self): + job, _ = self._make_approved_job() + core = self._core() + core.tolaria_delete = self.fake.delete + real_begin = self.store.begin_execution + def begin_then_tamper(job_id, worker_id): + result = real_begin(job_id, worker_id) + # Nach begin_execution das Payload-JSON manipulieren (TOCTOU): + # object_id im Payload ändern, damit der immutable guard den Drift + # erkennt (rohe DB-Spalte vs. Payload). + import json + row = self.store._conn.execute( + "SELECT payload FROM jobs WHERE job_id = ?", (job_id,) + ).fetchone() + payload = json.loads(row["payload"]) + payload["object_id"] = "tampered-object" + self.store._conn.execute( + "UPDATE jobs SET payload = ? WHERE job_id = ?", + (json.dumps(payload), job_id)) + self.store._conn.commit() + return result + self.store.begin_execution = begin_then_tamper + # Der immutable guard erkennt den Drift und wirft JobImmutableFieldError + # (kein State-Change, kein DELETE). + with self.assertRaises(JobImmutableFieldError): + core.process_job(job["job_id"], "w-t42") + self.assertEqual(self.fake.delete_count, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tolaria/c5-sync-service/test_mutations.py b/tolaria/c5-sync-service/test_mutations.py new file mode 100644 index 0000000..5ac3384 --- /dev/null +++ b/tolaria/c5-sync-service/test_mutations.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +""" +AUTH.4D — test_mutations.py (P11 Adversarial / Sensitivity) +============================================================ +Mutationstests A–O. Jede Mutation muss mindestens einen Test ROT machen. + +Mutationen: + A signature check removed + B expiry removed + C nonce check removed + D single-use removed + E object binding removed + F path binding removed + G commit binding removed + H provenance binding removed + I delete token scope removed + J fixed endpoint removed + K second-delete guard removed + L outcome-unknown guard removed + M immutable guard removed + N job==authorization allowed + O SAVE capability exposed + +Vorgehen: Für jede Mutation wird der Core-Quelltext temporär mutiert, die +Test-Suite (test_delete_worker.py) ausgeführt, und geprüft, dass mindestens +ein Test fehlschlägt (ROT). Danach wird der Quelltext zurückgesetzt. +""" + +from __future__ import annotations + +import os +import re +import shutil +import subprocess +import sys +import tempfile + +HERE = os.path.dirname(os.path.abspath(__file__)) +CORE = os.path.join(HERE, "delete_executor_core.py") +CLIENT = os.path.join(HERE, "delete_tolaria_client.py") +STORE = os.path.join(HERE, "job_store.py") +TEST = os.path.join(HERE, "test_delete_worker.py") +PY = os.path.join(HERE, "..", ".venv-auth4d", "bin", "python") + +# Jede Mutation: (name, file, old, new, [erwartete ROT-Tests]) +MUTATIONS = [ + # A: Signatur-Check entfernt (approval_verify wird nie aufgerufen) + ("A_signature_removed", CORE, + "sig = self.approval_verify(approval)\n if sig.get(\"valid\") is not True:\n return {\"valid\": False, \"code\": RC_APPROVAL_INVALID}", + "sig = self.approval_verify(approval)\n # MUTATION A: Signatur-Check entfernt\n pass", + ["test_t10_invalid_signature_blocked"]), + + # B: Expiry-Check entfernt (Core-Guard) + ("B_expiry_removed", CORE, + "if approval.get(\"expired\"):\n return {\"valid\": False, \"code\": RC_APPROVAL_EXPIRED}", + "# MUTATION B: Expiry-Check entfernt", + ["test_t11_expired_approval_blocked"]), + + # C: Nonce-Check entfernt (Core-Guard) + ("C_nonce_removed", CORE, + "if approval.get(\"nonce\") in (approval.get(\"used_nonces\") or set()):\n return {\"valid\": False, \"code\": RC_APPROVAL_CONSUMED}", + "# MUTATION C: nonce-Check entfernt", + ["test_t24_replay_nonce_blocked"]), + + # D: Single-Use entfernt (consumed-Check im Core-Guard) + ("D_single_use_removed", CORE, + "if approval.get(\"consumed\"):\n return {\"valid\": False, \"code\": RC_APPROVAL_CONSUMED}", + "# MUTATION D: single-use entfernt", + ["test_t12_consumed_approval_blocked", "test_t23_second_delete_single_use"]), + + # E: Object-Binding entfernt + ("E_object_binding_removed", CORE, + "if approval.get(\"object_id\") != job[\"object_id\"]:\n return {\"valid\": False, \"code\": RC_OBJECT_MISMATCH}", + "# MUTATION E: object binding entfernt", + ["test_t13_wrong_object_blocked", "test_t25_approval_other_object_never_authorizes"]), + + # F: Path-Binding entfernt + ("F_path_binding_removed", CORE, + "if approval.get(\"vault_path\") != job[\"vault_path\"]:\n return {\"valid\": False, \"code\": RC_PATH_MISMATCH}", + "# MUTATION F: path binding entfernt", + ["test_t14_wrong_path_blocked", "test_t27_approval_other_path_never_authorizes"]), + + # G: Commit-Binding entfernt + ("G_commit_binding_removed", CORE, + "if approval.get(\"expected_commit\") != job[\"expected_commit\"]:\n return {\"valid\": False, \"code\": RC_COMMIT_MISMATCH}", + "# MUTATION G: commit binding entfernt", + ["test_t15_wrong_commit_blocked", "test_t26_approval_other_commit_never_authorizes"]), + + # H: Provenance-Binding entfernt + ("H_provenance_binding_removed", CORE, + "if approval.get(\"expected_provenance_hash\") != job[\"expected_provenance_hash\"]:\n return {\"valid\": False, \"code\": RC_PROVENANCE_MISMATCH}", + "# MUTATION H: provenance binding entfernt", + ["test_t16_wrong_provenance_blocked"]), + + # I: DELETE-Token-Scope entfernt (Client akzeptiert ohne Token) + ("I_delete_token_scope_removed", CLIENT, + "def _require_token(self) -> str:\n \"\"\"Fail-closed: fehlendes/leeres DELETE-Credential -> kein HTTP.\"\"\"\n token = self.delete_token\n if not token or not isinstance(token, str) or not token.strip():\n raise TolariaDeleteError(\n \"Tolaria DELETE-Credential fehlt oder ist leer \"\n \"(fail-closed, kein Request gesendet)\",\n \"CREDENTIAL_MISSING\")\n return token", + "def _require_token(self) -> str:\n # MUTATION I: Token-Scope entfernt (kein fail-closed)\n return self.delete_token or \"\"", + ["test_t3_save_token_alone_cannot_authorize_delete", "test_t18_no_delete_token_fail_closed"]), + + # J: Fester Endpoint entfernt (arbitrary endpoint erlaubt) + ("J_fixed_endpoint_removed", CLIENT, + "ENDPOINT_DELETE = \"delete\"", + "ENDPOINT_DELETE = \"delete-arbitrary\" # MUTATION J: fester Endpoint entfernt", + ["test_t32_fixed_delete_endpoint"]), + + # K: Second-Delete-Guard entfernt (kein Doppel-Delete-Schutz) + ("K_second_delete_guard_removed", CORE, + " try:\n self.store.claim_job(job_id, worker_id)\n except Exception:\n # Nicht claimbar (bereits geclaimt) -> kein Doppel-Delete\n return self.store.get_job(job_id)", + " # MUTATION K: second-delete guard entfernt (kein Claim-Schutz)\n self.store.claim_job(job_id, worker_id)", + ["test_t41_second_process_same_job_blocked"]), + + # L: Outcome-Unknown-Guard entfernt (blinder Retry erlaubt) + ("L_outcome_unknown_guard_removed", CORE, + "elif status == \"error\" and result.get(\"uncertain\"):\n self.store.mark_outcome_unknown(job_id, worker_id)", + "elif status == \"error\" and result.get(\"uncertain\"):\n # MUTATION L: outcome-unknown guard entfernt (blinder Retry)\n self.store.mark_succeeded(job_id, worker_id)", + ["test_t20_timeout_outcome_unknown", "test_t40_outcome_unknown_no_blind_retry"]), + + # M: Immutable-Guard entfernt (kein Re-Read nach Claim -> TOCTOU) + # Der echte immutable guard ist _assert_immutable_fields in _transition + # (job_store.py), das die rohe DB-Spalte gegen das Payload prüft. + ("M_immutable_guard_removed", STORE, + " self._assert_immutable_fields(job)", + " # MUTATION M: immutable guard entfernt (kein TOCTOU-Schutz)\n pass", + ["test_t42_toctou_re_read_after_claim"]), + + # N: Job==Authorization erlaubt (Approval wird übersprungen) + ("N_job_equals_authorization", CORE, + "approval = self.approval_loader(job[\"approval_id\"])\n if approval is None:\n return {\"valid\": False, \"code\": RC_APPROVAL_MISSING}", + "approval = self.approval_loader(job[\"approval_id\"])\n if approval is None:\n # MUTATION N: Job==Authorization erlaubt\n return {\"valid\": True}", + ["test_t4_delete_token_alone_cannot_bypass_approval", "test_t9_delete_without_approval_blocked", + "test_t21_job_alone_never_authorizes"]), + + # O: SAVE-Capability exponiert (write() im Client) + ("O_save_capability_exposed", CLIENT, + " def delete(self, vault_path: str) -> Dict[str, Any]:", + " def write(self, vault_path: str, content: str) -> Dict[str, Any]:\n # MUTATION O: SAVE-Capability exponiert\n return self.delete(vault_path)\n\n def delete(self, vault_path: str) -> Dict[str, Any]:", + ["test_t2_delete_path_cannot_call_write", "test_t35_no_save_fallback"]), +] + + +def run_tests() -> tuple[int, str]: + """Führt die Test-Suite aus. Gibt (exit_code, output) zurück.""" + proc = subprocess.run([PY, TEST], capture_output=True, text=True, cwd=HERE) + return proc.returncode, proc.stdout + proc.stderr + + +def apply_mutation(filepath: str, old: str, new: str) -> bool: + """Wendet eine Mutation an. Gibt True bei Erfolg zurück.""" + with open(filepath) as f: + src = f.read() + if old not in src: + return False + with open(filepath, "w") as f: + f.write(src.replace(old, new, 1)) + return True + + +def main() -> int: + # Backup der Originale + core_bak = open(CORE).read() + client_bak = open(CLIENT).read() + store_bak = open(STORE).read() + + # Baseline: alle Tests müssen GRÜN sein + rc, out = run_tests() + if rc != 0: + print("BASELINE FAILED — Tests müssen vor Mutationen GRÜN sein") + print(out[-2000:]) + return 1 + print(f"BASELINE: {out.strip().splitlines()[-1]}") + + results = [] + all_ok = True + for name, filepath, old, new, expected_red in MUTATIONS: + # Mutation anwenden + if not apply_mutation(filepath, old, new): + print(f"[{name}] SKIP — Mutation nicht anwendbar (Pattern nicht gefunden)") + results.append((name, "SKIP", [])) + continue + # Tests ausführen + rc, out = run_tests() + # Zurücksetzen + if filepath == CORE: + open(CORE, "w").write(core_bak) + elif filepath == STORE: + open(STORE, "w").write(store_bak) + else: + open(CLIENT, "w").write(client_bak) + # Prüfen: mindestens ein erwarteter Test muss ROT sein + red_tests = [] + for t in expected_red: + if f"FAIL: {t}" in out or f"ERROR: {t}" in out: + red_tests.append(t) + if rc != 0 and red_tests: + print(f"[{name}] PASS — ROT ({len(red_tests)}/{len(expected_red)} erwartete Tests fehlgeschlagen): {red_tests}") + results.append((name, "PASS", red_tests)) + else: + print(f"[{name}] FAIL — Mutation machte Tests NICHT ROT (rc={rc}, red={red_tests})") + all_ok = False + results.append((name, "FAIL", red_tests)) + + # Finale Baseline-Verifikation (alles zurückgesetzt) + rc, out = run_tests() + if rc != 0: + print("\nFINAL BASELINE FAILED — Mutationen nicht sauber zurückgesetzt!") + all_ok = False + else: + print(f"\nFINAL BASELINE: {out.strip().splitlines()[-1]}") + + print("\n=== MUTATION SUMMARY ===") + for name, status, red in results: + print(f" {name}: {status} {red if red else ''}") + print(f"\nOVERALL: {'ALL MUTATIONS DETECTED (PASS)' if all_ok else 'MUTATION GAP DETECTED (FAIL)'}") + return 0 if all_ok else 1 + + +if __name__ == "__main__": + sys.exit(main())