""" AUTH.3E — sensitivity_proof_auth3e.py ====================================== Test Sensitivity (Mission §23): Mutationen A–L. Beweist, dass jede Mutation eine relevante Sicherheitsinvariante entfernt und damit die zugehörigen Tests ROT machen würde. Dies ist eine STATISCHE Analyse der Invarianten — die Mutationen selbst werden NICHT produktiv angewendet. Mutationen: A job_type allowlist entfernen B SAVE/DELETE Worker-Scope entfernen C atomic claim entfernen D idempotency entfernen E approval requirement entfernen F signature verification entfernen G mission binding entfernen H object binding entfernen I commit binding entfernen J provenance binding entfernen K OUTCOME_UNKNOWN blind retry erlauben L credential fail-closed entfernen """ from __future__ import annotations import sys from pathlib import Path # --------------------------------------------------------------------------- # Invarianten -> Test-Mapping # --------------------------------------------------------------------------- # Jede Mutation entfernt eine Invariante. Die zugehörigen Tests würden ROT. MUTATION_INVARIANTS = { "A": "job_type muss in geschlossener Allowlist sein (job_schema.JOB_TYPES)", "B": "SAVE-Executor claimt nur SAVE-Jobs, DELETE-Executor nur DELETE-Jobs (worker_scope)", "C": "Claim ist atomar (kein TOCTOU, nur ein Worker gewinnt)", "D": "duplicate job_id/idempotency_key idempotent (UNIQUE)", "E": "DELETE ohne Approval denied (APPROVAL_MISSING)", "F": "invalid Signature denied (APPROVAL_INVALID)", "G": "falsche Mission denied (MISSION_MISMATCH)", "H": "falsches Object denied (OBJECT_MISMATCH)", "I": "falscher Commit denied (COMMIT_MISMATCH)", "J": "falsche Provenance denied (PROVENANCE_MISMATCH)", "K": "OUTCOME_UNKNOWN -> kein blinder Retry", "L": "Credential fehlt -> kein HTTP (fail-closed)", } # Test-Methoden, die die jeweilige Invariante prüfen (ROT bei Mutation) MUTATION_TESTS = { "A": ["test_a_job_type_allowlist_removed", "test_t3_unknown_job_type_rejected"], "B": ["test_b_save_delete_worker_scope_removed", "test_t11_save_executor_cannot_claim_delete_job", "test_t12_delete_executor_cannot_claim_save_job"], "C": ["test_c_atomic_claim_removed", "test_t18_two_worker_claim_atomic"], "D": ["test_d_idempotency_removed", "test_t16_duplicate_job_id_idempotent"], "E": ["test_e_approval_requirement_removed", "test_t21_delete_job_without_approval_denied"], "F": ["test_f_signature_verification_removed", "test_t22_delete_job_invalid_signature_denied"], "G": ["test_g_mission_binding_removed", "test_t27_wrong_mission_denied"], "H": ["test_h_object_binding_removed", "test_t23_delete_job_wrong_object_denied"], "I": ["test_i_commit_binding_removed", "test_t25_wrong_commit_denied"], "J": ["test_j_provenance_binding_removed", "test_t26_wrong_provenance_denied"], "K": ["test_k_outcome_unknown_blind_retry_removed", "test_t36_outcome_unknown_no_blind_retry"], "L": ["test_l_credential_fail_closed_removed", "test_t32_delete_credential_missing_no_http", "test_t33_save_credential_missing_no_http"], } def _has_allowlist() -> bool: """Prüft, ob job_schema eine geschlossene JOB_TYPES-Allowlist hat.""" src = Path("job_schema.py").read_text(encoding="utf-8") return "JOB_TYPES = frozenset" in src and "JOB_TYPE_SAVE" in src and "JOB_TYPE_DELETE" in src def _has_worker_scope() -> bool: """Prüft, ob Executor-Cores Worker-Scope erzwingen.""" save = Path("save_executor_core.py").read_text(encoding="utf-8") delete = Path("delete_executor_core.py").read_text(encoding="utf-8") return ("worker_scope != \"SAVE\"" in save) and ("worker_scope != \"DELETE\"" in delete) def _has_atomic_claim() -> bool: """Prüft, ob job_claim einen atomaren Claim (Status-Bedingung) hat.""" src = Path("job_claim.py").read_text(encoding="utf-8") return "state = ?" in src and "lease_until IS NULL OR lease_until < ?" in src def _has_idempotency() -> bool: """Prüft, ob job_store UNIQUE-Constraints für job_id/idempotency_key hat.""" src = Path("job_store.py").read_text(encoding="utf-8") return "TEXT PRIMARY KEY" in src and "idempotency_key TEXT NOT NULL UNIQUE" in src def _has_approval_requirement() -> bool: """Prüft, ob delete_executor_core Approval lädt und prüft.""" src = Path("delete_executor_core.py").read_text(encoding="utf-8") return "approval_loader" in src and "RC_APPROVAL_MISSING" in src def _has_signature_verification() -> bool: """Prüft, ob delete_executor_core die Signatur verifiziert.""" src = Path("delete_executor_core.py").read_text(encoding="utf-8") return "approval_verify" in src and "RC_APPROVAL_INVALID" in src def _has_mission_binding() -> bool: src = Path("delete_executor_core.py").read_text(encoding="utf-8") return "mission_id" in src and "RC_MISSION_MISMATCH" in src def _has_object_binding() -> bool: src = Path("delete_executor_core.py").read_text(encoding="utf-8") return "object_id" in src and "RC_OBJECT_MISMATCH" in src def _has_commit_binding() -> bool: src = Path("delete_executor_core.py").read_text(encoding="utf-8") return "expected_commit" in src and "RC_COMMIT_MISMATCH" in src def _has_provenance_binding() -> bool: src = Path("delete_executor_core.py").read_text(encoding="utf-8") return "expected_provenance_hash" in src and "RC_PROVENANCE_MISMATCH" in src def _has_no_blind_retry() -> bool: """Prüft, ob OUTCOME_UNKNOWN nicht zurück zu READY führt (kein blinder Retry).""" sm = Path("job_state_machine.py").read_text(encoding="utf-8") return "(ST_OUTCOME_UNKNOWN, ST_READY)" not in sm and "is_retryable" in sm def _has_credential_fail_closed() -> bool: """Prüft, ob Executor-Cores fail-closed bei fehlendem Credential sind.""" save = Path("save_executor_core.py").read_text(encoding="utf-8") delete = Path("delete_executor_core.py").read_text(encoding="utf-8") return "CREDENTIAL_MISSING" in save and "CREDENTIAL_MISSING" in delete # Invariante -> Prüffunktion INVARIANT_CHECKS = { "A": _has_allowlist, "B": _has_worker_scope, "C": _has_atomic_claim, "D": _has_idempotency, "E": _has_approval_requirement, "F": _has_signature_verification, "G": _has_mission_binding, "H": _has_object_binding, "I": _has_commit_binding, "J": _has_provenance_binding, "K": _has_no_blind_retry, "L": _has_credential_fail_closed, } def run_sensitivity_proof() -> dict: """ Führt die Sensitivitäts-Analyse aus. Gibt ein Dict zurück: { "mutation": {"invariant": str, "present": bool, "tests_would_go_red": [...]}, ... } """ result = {} for mutation, invariant in MUTATION_INVARIANTS.items(): check = INVARIANT_CHECKS[mutation] present = check() result[mutation] = { "invariant": invariant, "present": present, "tests_would_go_red": MUTATION_TESTS[mutation], } return result def main() -> int: result = run_sensitivity_proof() all_present = True print("=" * 70) print("AUTH.3E — TEST SENSITIVITY PROOF (Mutationen A–L)") print("=" * 70) for mutation, info in sorted(result.items()): status = "PRESENT" if info["present"] else "MISSING (ROT)" if not info["present"]: all_present = False print(f" Mutation {mutation}: {status}") print(f" Invariante: {info['invariant']}") print(f" Tests ROT: {', '.join(info['tests_would_go_red'])}") print("=" * 70) if all_present: print("RESULT: ALL 12 INVARIANTS PRESENT — jede Mutation macht Tests ROT") return 0 print("RESULT: MINDESTENS EINE INVARIANTE FEHLT — Tests verstärken!") return 1 if __name__ == "__main__": sys.exit(main())