#!/usr/bin/env python3 """Execution-specific adversarial bad-input probes for the FIX7 P0 NO-PRODUCTION implementation-execution lane.

A single fail-closed guard evaluate_execution_request(req) decides whether an execution request is permitted. It returns a list of reject codes; an empty list means "allowed" (only a fully no-production, fully-evidenced request). A request that touches any forbidden surface, lacks owner delegation, or carries a defective rollback proof MUST return a non-empty reject list and emit NO PASS/digest/cert/ seal-like token.

Rollback defects (12/13) are delegated to the byte-exact hardened validator (hardened_dryrun_validator.check_rollback_proof, sha256 e6547e69..56c47).

15 required probe classes (Workstream E) + a positive control.

Exit 0 => control passes (clean no-production request allowed) AND all 15 probes fail closed (any_fail_open == False) with no leaked official token. Exit 1 => any probe failed open, or the control was rejected. Writes execution-bad-input-probes.json (deterministic). """ import copy import json import os import sys

import hardened_dryrun_validator as V

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

OWNER_DELEGATION = "AUTHORIZE_IMPLEMENTATION_EXECUTION_NO_PRODUCTION" FORBIDDEN_ACTIONS = ["REAL_RUN", "QT001", "apply", "permit", "activation", "repoint", "cutover"] LEAK_TOKENS = ("PASS", "DIGEST", "CERT", "SEAL", "AUTHORIZED", "PROVEN_OK")

def evaluate_execution_request(req): """Fail-closed guard. Returns list of reject codes (empty == allowed).""" r = [] # owner delegation must be exactly the no-production execution grant if req.get("owner_delegation") != OWNER_DELEGATION: r.append("REJECT_NO_OWNER_DELEGATION") # target environment must be no-production env = str(req.get("target_env", "")) if env != "no-production-kb": r.append("REJECT_PRODUCTION_TARGET:" + (env or "<empty>")) # forbidden actions for a in req.get("actions", []): if a in FORBIDDEN_ACTIONS: r.append("REJECT_FORBIDDEN_ACTION:" + a) # production surfaces if str(req.get("pg_url", "")).strip(): r.append("REJECT_PRODUCTION_PG") if str(req.get("directus_target", "")).strip(): r.append("REJECT_PRODUCTION_DIRECTUS") if req.get("system_issues_mutation"): r.append("REJECT_PRODUCTION_SYSTEM_ISSUES") if req.get("ci_deploy"): r.append("REJECT_PRODUCTION_CI_DEPLOY") if req.get("secrets_change"): r.append("REJECT_SECRETS_CHANGE") if req.get("evidence_deletion"): r.append("REJECT_EVIDENCE_DELETION") # any forbidden surface flagged as "attempted" forbidden_attempted = bool( env not in ("", "no-production-kb") or any(a in FORBIDDEN_ACTIONS for a in req.get("actions", [])) or str(req.get("pg_url", "")).strip() or str(req.get("directus_target", "")).strip() or req.get("system_issues_mutation") or req.get("ci_deploy") or req.get("secrets_change") or req.get("evidence_deletion") ) # a request may NEVER claim execution PASS while a forbidden surface is attempted if req.get("claims_pass") and forbidden_attempted: r.append("REJECT_SUCCESS_CLAIM_WITH_FORBIDDEN_SURFACE") # rollback discipline: a claimed applied mutation needs a valid rollback proof rb = req.get("rollback_evidence") if req.get("mutation_applied"): if rb is None: r.append("REJECT_MISSING_ROLLBACK_PROOF") else: r.extend(V.check_rollback_proof(rb)) return r

--- a valid no-production request (positive control / probe base) ---

def good_request(): with open(os.path.join(HERE, "rollback-evidence.json"), "r", encoding="utf-8") as fh: rb = json.load(fh) return { "owner_delegation": OWNER_DELEGATION, "target_env": "no-production-kb", "actions": ["publish_kb_doc", "publish_governance_addendum"], "targets": ["operative-blueprint-doc", "governance-addendum"], "pg_url": "", "directus_target": "", "system_issues_mutation": False, "ci_deploy": False, "secrets_change": False, "evidence_deletion": False, "mutation_applied": True, "rollback_evidence": rb, "claims_pass": True, }

def expect_reject(name, req, expected_prefixes): codes = evaluate_execution_request(req) closed = bool(codes) and any( any(c.startswith(p) for p in expected_prefixes) for c in codes) leaked = [c for c in codes if any(t in c.upper() for t in LEAK_TOKENS)] ok = closed and not leaked print("[%s] %s -> %s%s" % ( "FAIL-CLOSED" if ok else "FAIL-OPEN", name, "; ".join(codes) if codes else "(allowed!)", "" if not leaked else " LEAK:" + ";".join(leaked))) return {"probe": name, "fail_closed": ok, "codes": codes, "expected_prefix": list(expected_prefixes), "leak": leaked}

def main(): results = []

# positive control: a clean no-production request is allowed
ctrl = evaluate_execution_request(good_request())
control_ok = not ctrl
print("[CONTROL] clean no-production request allowed: %s%s"
      % ("YES" if control_ok else "NO", "" if control_ok else " -> " + "; ".join(ctrl)))

g = good_request

# 1 production target requested
d = g(); d["target_env"] = "production"
results.append(expect_reject("E1 production target requested", d, ["REJECT_PRODUCTION_TARGET"]))
# 2 REAL_RUN requested
d = g(); d["actions"] = d["actions"] + ["REAL_RUN"]
results.append(expect_reject("E2 REAL_RUN requested", d, ["REJECT_FORBIDDEN_ACTION:REAL_RUN"]))
# 3 QT001 / apply requested
d = g(); d["actions"] = d["actions"] + ["QT001"]
results.append(expect_reject("E3 QT001/apply requested", d, ["REJECT_FORBIDDEN_ACTION:QT001"]))
# 4 permit/activation/repoint/cutover requested
d = g(); d["actions"] = d["actions"] + ["cutover"]
results.append(expect_reject("E4 cutover requested", d, ["REJECT_FORBIDDEN_ACTION:cutover"]))
# 5 production PG URL
d = g(); d["pg_url"] = "postgres://prod-host:5432/incomex"
results.append(expect_reject("E5 production PG URL", d, ["REJECT_PRODUCTION_PG"]))
# 6 production Directus target
d = g(); d["directus_target"] = "https://prod-directus/items/system_issues"
results.append(expect_reject("E6 production Directus target", d, ["REJECT_PRODUCTION_DIRECTUS"]))
# 7 production system_issues mutation
d = g(); d["system_issues_mutation"] = True
results.append(expect_reject("E7 production system_issues mutation", d, ["REJECT_PRODUCTION_SYSTEM_ISSUES"]))
# 8 production CI/deploy
d = g(); d["ci_deploy"] = True
results.append(expect_reject("E8 production CI/deploy", d, ["REJECT_PRODUCTION_CI_DEPLOY"]))
# 9 secrets/credential change
d = g(); d["secrets_change"] = True
results.append(expect_reject("E9 secrets/credential change", d, ["REJECT_SECRETS_CHANGE"]))
# 10 evidence deletion
d = g(); d["evidence_deletion"] = True
results.append(expect_reject("E10 evidence deletion", d, ["REJECT_EVIDENCE_DELETION"]))
# 11 owner delegation missing
d = g(); d["owner_delegation"] = ""
results.append(expect_reject("E11 owner delegation missing", d, ["REJECT_NO_OWNER_DELEGATION"]))
# 12 rollback claim with no mutation
d = g(); d["rollback_evidence"] = copy.deepcopy(d["rollback_evidence"])
d["rollback_evidence"]["staging_mutation_occurred"] = False
d["rollback_evidence"]["entries"] = []
results.append(expect_reject("E12 rollback claim with no mutation", d,
                             ["ROLLBACK_PROVEN_BUT_NO_MUTATION", "ROLLBACK_PROVEN_NO_ENTRIES"]))
# 13 after_apply_hash == before_hash
d = g(); d["rollback_evidence"] = copy.deepcopy(d["rollback_evidence"])
ent = d["rollback_evidence"]["entries"][0]
ent["before_hash"] = ent["after_apply_hash"]
ent["after_rollback_hash"] = ent["after_apply_hash"]
results.append(expect_reject("E13 after_apply_hash == before_hash", d,
                             ["ROLLBACK_APPLY_DID_NOT_MUTATE"]))
# 14 missing rollback proof
d = g(); d["rollback_evidence"] = None
results.append(expect_reject("E14 missing rollback proof", d, ["REJECT_MISSING_ROLLBACK_PROOF"]))
# 15 execution PASS while forbidden surface attempted
d = g(); d["ci_deploy"] = True; d["claims_pass"] = True
results.append(expect_reject("E15 PASS while forbidden surface attempted", d,
                             ["REJECT_SUCCESS_CLAIM_WITH_FORBIDDEN_SURFACE"]))

total = len(results)
passed = sum(1 for x in results if x["fail_closed"])
any_fail_open = (passed != total) or (not control_ok)

out = {
    "doc": "fix7-p0-no-production-implementation-execution-bad-input-probes",
    "date": "2026-06-12",
    "guard": "evaluate_execution_request (fail-closed; rollback defects delegated to hardened_dryrun_validator.check_rollback_proof)",
    "control_pass": control_ok,
    "total": total,
    "fail_closed": passed,
    "any_fail_open": any_fail_open,
    "results": results,
    "note": "All invalid execution requests fail closed and emit no PASS/digest/cert/seal token. Authorizes nothing; no production mutation.",
}
with open(os.path.join(HERE, "execution-bad-input-probes.json"), "w", encoding="utf-8") as fh:
    json.dump(out, fh, indent=2, sort_keys=True)
    fh.write("\n")

print("EXECUTION_BAD_INPUT_PROBES: %d/%d fail-closed; control_pass=%s; any_fail_open=%s"
      % (passed, total, control_ok, any_fail_open))
sys.exit(1 if any_fail_open else 0)

if name == "main": main()