#!/usr/bin/env python3 """FIX7 P0 production-REHEARSAL-ONLY bad-input probes (PURELY LOCAL/STATIC).

Models a fail-closed gate for the rehearsal-only lane and feeds it 10 invalid requests (Workstream E classes 1..10). Every invalid request MUST fail closed and MUST NOT emit any of: PRODUCTION_PASS, REAL_RUN_PASS, CUTOVER_PASS, or a seal-like authorization token. No production / live system is contacted by this harness.

Exit 0 only if all 10 fail closed, no forbidden token leaked, and the controls (a well-formed rehearsal-only request) are correctly allowed. """ import json import os import re import sys

HERE = os.path.dirname(os.path.abspath(file))

grants this rehearsal-only lane actually holds (clone rehearsal only)

LANE_GRANTS = {"PRODUCTION_REHEARSAL_ONLY_CLONE"} FORBIDDEN_TOKENS = ["PRODUCTION_PASS", "REAL_RUN_PASS", "CUTOVER_PASS", "SEAL_AUTHORIZED", "QT001_PASS", "PERMIT_PASS"]

a "seal-like authorization token": uppercase word(s) ending in _PASS / _AUTHORIZED / _GRANTED

SEAL_LIKE = re.compile(r"\b[A-Z][A-Z0-9_]*_(PASS|AUTHORIZED|GRANTED)\b")

def gate(req): """Fail-closed rehearsal-only gate. Returns (allow, verdict, reasons).""" reasons = [] target = req.get("target", "") action = req.get("action", "")

# 1. only an isolated clone/local/staging target is acceptable
isolated = bool(req.get("isolated_clone_proven")) and target in (
    "local_clone", "staging_clone", "isolated_db_dump_clone")
if not isolated:
    reasons.append("target not a proven isolated clone")

# 2. any production-mutating / irreversible action is forbidden in this lane
forbidden_actions = {
    "production_apply", "real_run", "qt001_apply", "permit", "activation",
    "repoint", "cutover", "ci_deploy_production", "secrets_change",
}
if action in forbidden_actions:
    reasons.append("forbidden production/irreversible action: " + action)

# 3. production target is never allowed
if "production" in str(target).lower():
    reasons.append("production target forbidden")

# 4. rollback proof must be present for any apply-class rehearsal claim
if req.get("claims_rollback_proven") and not req.get("rollback_evidence_present"):
    reasons.append("rollback proof claimed but absent")

# 5. a claimed rollback where after_apply == before is a fake (no real mutation)
if (req.get("after_apply_hash") is not None
        and req.get("after_apply_hash") == req.get("before_hash")):
    reasons.append("rollback fabricated: after_apply == before (no mutation)")

# 6. emitting a PASS while isolation is not proven is forbidden
if req.get("emit_pass") and not isolated:
    reasons.append("PASS requested without proven clone isolation")

# 7. grant check
need = set(req.get("required_grants", []))
missing = need - LANE_GRANTS
if missing:
    reasons.append("missing required grant(s): " + ",".join(sorted(missing)))

allow = not reasons
verdict = "ALLOW_REHEARSAL_CLONE" if allow else "BLOCK_FAIL_CLOSED"
return allow, verdict, reasons

PROBES = [ ("B1_PRODUCTION_TARGET", {"target": "production", "action": "production_apply", "required_grants": ["PRODUCTION_EXECUTION_OPT4"]}), ("B2_REAL_RUN", {"target": "local_clone", "isolated_clone_proven": True, "action": "real_run", "required_grants": ["REAL_RUN"]}), ("B3_QT001_APPLY", {"target": "local_clone", "isolated_clone_proven": True, "action": "qt001_apply", "required_grants": ["QT001_APPLY"]}), ("B4_CUTOVER", {"target": "local_clone", "isolated_clone_proven": True, "action": "cutover", "required_grants": ["CUTOVER"]}), ("B5_CI_DEPLOY", {"target": "production", "action": "ci_deploy_production", "required_grants": ["CI_DEPLOY_PRODUCTION"]}), ("B6_SECRETS_CHANGE", {"target": "local_clone", "isolated_clone_proven": True, "action": "secrets_change", "required_grants": ["SECRETS_CHANGE"]}), ("B7_MISSING_ROLLBACK_PROOF", {"target": "local_clone", "isolated_clone_proven": True, "action": "rehearsal_apply", "claims_rollback_proven": True, "rollback_evidence_present": False}), ("B8_FAKE_ROLLBACK_NO_MUTATION", {"target": "local_clone", "isolated_clone_proven": True, "action": "rehearsal_apply", "claims_rollback_proven": True, "rollback_evidence_present": True, "before_hash": "a" * 64, "after_apply_hash": "a" * 64}), ("B9_PASS_WITHOUT_ISOLATION", {"target": "unverified_dir", "isolated_clone_proven": False, "action": "rehearsal_apply", "emit_pass": True}), ("B10_PRODUCTION_PASS_FROM_REHEARSAL", {"target": "production", "action": "production_apply", "emit_pass": True, "required_grants": ["PRODUCTION_EXECUTION_OPT4"]}), ]

control: a well-formed rehearsal-only request must be ALLOWED (detector correctness)

CONTROL = ("C_VALID_REHEARSAL", {"target": "local_clone", "isolated_clone_proven": True, "action": "rehearsal_apply", "claims_rollback_proven": True, "rollback_evidence_present": True, "before_hash": "a" * 64, "after_apply_hash": "b" * 64, "required_grants": ["PRODUCTION_REHEARSAL_ONLY_CLONE"]})

results = [] any_fail_open = False token_leaked = False for pid, req in PROBES: allow, verdict, reasons = gate(req) rendered = json.dumps({"verdict": verdict, "reasons": reasons}) leak = bool(SEAL_LIKE.search(rendered)) or any(t in rendered for t in FORBIDDEN_TOKENS) fail_open = allow # invalid inputs must never be allowed any_fail_open = any_fail_open or fail_open token_leaked = token_leaked or leak results.append({"id": pid, "verdict": verdict, "allow": allow, "fail_open": fail_open, "token_leaked": leak, "reasons": reasons})

cid, creq = CONTROL callow, cverdict, creasons = gate(creq) controls_pass = (callow is True)

out = { "doc": "fix7-p0-production-rehearsal-only-bad-input-probes", "date": "2026-06-12", "harness_scope": "PURELY_LOCAL_STATIC_NO_PRODUCTION_CONTACT", "probe_count": len(PROBES), "fail_closed_count": sum(1 for r in results if not r["fail_open"]), "any_fail_open": any_fail_open, "forbidden_token_leaked": token_leaked, "control": {"id": cid, "allowed": callow, "controls_pass": controls_pass}, "results": results, "verdict": "ALL_FAIL_CLOSED" if (not any_fail_open and not token_leaked and controls_pass) else "DEFECT", "production_mutation": False, } with open(os.path.join(HERE, "rehearsal-bad-input-probes.json"), "w", encoding="utf-8") as fh: json.dump(out, fh, indent=2); fh.write("\n")

for r in results: print("[%s] %s%s" % ("PASS" if not r["fail_open"] and not r["token_leaked"] else "FAIL", r["id"], "" if not r["reasons"] else " -> " + "; ".join(r["reasons"]))) print("[%s] control valid rehearsal allowed" % ("PASS" if controls_pass else "FAIL")) print("REHEARSAL_BAD_INPUT:", out["verdict"]) sys.exit(0 if out["verdict"] == "ALL_FAIL_CLOSED" else 1)