nexo312/trading-system-docs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
trading-system-docs/tolaria/c5-sync-service/test_delete_worker.py
Rain Ocampo dcc6d8a098 AUTH.4D: P10/P11 test suite + mutation/sensitivity tests 
- test_delete_worker.py: 42 tests (T1-T42) against productive delete_* SoT code
  (T29 delete_request_id mismatch, T41 second-delete guard, T42 immutable guard)
- test_mutations.py: 15 mutations A-O, all detected (P11 sensitivity)

AUTH.4D P14. No deployment, no token injection, no key provisioning.
2026-08-28 02:30:19 +00:00

696 lines
33 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/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()
Reference in a new issue View git blame Copy permalink