dcc6d8a098
- 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.
219 lines
10 KiB
Python
219 lines
10 KiB
Python
#!/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())
|