#!/usr/bin/env python3 """CASAN Plan-16 SEC-16 (ARCH-01) — signed harness + policy bundle. Every gate is bypassable by editing the gate itself: change `security-check.sh`, `prompt-filter.yaml`, `thresholds.yaml`, `model-digest.pin` or `reviewers.registry` and the control is simply gone — no input needed. This binds the harness code + policy files to a manifest whose head is signed with an OFF-REPO key. Before a run, the harness verifies its self-hash against the signed manifest and REFUSES to run on any mismatch (enforced mode). An attacker who edits a gate cannot re-sign the manifest without the off-repo key. Usage: bundle-integrity.py generate # hash harness+policy, write + sign manifest bundle-integrity.py verify [--strict] # re-check; exit 1 on drift / bad signature Env: CASAN_BUNDLE_ROOT (default .specify), CASAN_BUNDLE_MANIFEST, CASAN_BUNDLE_KEY_DIR, CASAN_BUNDLE_PUB, CASAN_PROFILE=prod / CASAN_VERIFY_STRICT=1 (enforce signature + drift). """ import argparse import fnmatch import glob import hashlib import json import os def _casan_app_root(): # Plan-01: walk UP for the `.specify` state marker (harness code lives in # packages/casan-harness/, so a fixed __file__ depth would mis-root). _d = os.path.abspath(os.path.dirname(__file__)) _p = _d while _p != os.path.dirname(_p): if os.path.isdir(os.path.join(_p, ".specify")) or os.path.isdir(os.path.join(_p, "packages/casan-harness")): return _p _p = os.path.dirname(_p) return os.path.abspath(os.path.join(_d, "..", "..", "..")) import shutil import subprocess import sys # Harness code + policy/data files whose modification would silently disable a # control. Globs are relative to the bundle root (default .specify). INCLUDE_GLOBS = [ "scripts/bash/*.sh", "scripts/bash/*.py", "**/prompt-filter.yaml", "**/thresholds.yaml", "**/compression-policy.yaml", "**/*.pin", "**/reviewers.registry", "**/pii-rules.yaml", "**/pii-rules.json", "**/redteam-vectors.yaml", "**/attack-catalog.yaml", ] # Never bind volatile artifacts (they change every run) or the manifest itself. EXCLUDE_SUBSTR = ["/logs/", "/output/", "bundle-integrity", "test-integrity"] def bundle_root() -> str: return os.environ.get("CASAN_BUNDLE_ROOT") or os.path.join( _casan_app_root(), ".specify") def manifest_path() -> str: return os.environ.get("CASAN_BUNDLE_MANIFEST") or os.path.join( bundle_root(), "level5", "central-governance", "harness-bundle-manifest.json") def _enforced() -> bool: return os.environ.get("CASAN_PROFILE") == "prod" or os.environ.get("CASAN_VERIFY_STRICT") == "1" def bundle_files(): root = bundle_root() found = set() for pat in INCLUDE_GLOBS: for path in glob.glob(os.path.join(root, pat), recursive=True): if os.path.isfile(path) and not any(s in path.replace(os.sep, "/") for s in EXCLUDE_SUBSTR): found.add(path) return sorted(found) def hash_file(path) -> str: return hashlib.sha256(open(path, "rb").read()).hexdigest() def build_manifest(): root = bundle_root() files = {os.path.relpath(f, root): hash_file(f) for f in bundle_files()} return {"files": files, "file_count": len(files)} def manifest_head(manifest) -> str: core = json.dumps(manifest["files"], sort_keys=True, separators=(",", ":")) return hashlib.sha256(core.encode()).hexdigest() def _priv(): d = os.environ.get("CASAN_BUNDLE_KEY_DIR") or os.path.join(os.path.expanduser("~"), ".casan", "audit-keys") return os.path.join(d, "harness-bundle-private.pem") def _pub(): return os.environ.get("CASAN_BUNDLE_PUB") or (manifest_path() + ".pub") def _sig(): return manifest_path() + ".sig" def _head_file(): return manifest_path() + ".head" def sign(manifest): open(_head_file(), "w", encoding="utf-8").write(manifest_head(manifest)) ossl = shutil.which("openssl") if not ossl: return priv, pub = _priv(), _pub() os.makedirs(os.path.dirname(priv), exist_ok=True) if not os.path.isfile(priv): subprocess.run([ossl, "genpkey", "-algorithm", "RSA", "-pkeyopt", "rsa_keygen_bits:2048", "-out", priv], capture_output=True) try: os.chmod(priv, 0o600) except OSError: pass os.makedirs(os.path.dirname(pub) or ".", exist_ok=True) subprocess.run([ossl, "rsa", "-in", priv, "-pubout", "-out", pub], capture_output=True) subprocess.run([ossl, "dgst", "-sha256", "-sign", priv, "-out", _sig(), _head_file()], capture_output=True) def check_signature(manifest): ossl = shutil.which("openssl") if not (ossl and os.path.isfile(_head_file()) and os.path.isfile(_sig()) and os.path.isfile(_pub())): return "unsigned" if open(_head_file(), encoding="utf-8").read().strip() != manifest_head(manifest): return "invalid" res = subprocess.run([ossl, "dgst", "-sha256", "-verify", _pub(), "-signature", _sig(), _head_file()], capture_output=True) return "signed" if res.returncode == 0 else "invalid" def do_generate(): manifest = build_manifest() os.makedirs(os.path.dirname(manifest_path()), exist_ok=True) with open(manifest_path(), "w", encoding="utf-8") as fh: json.dump(manifest, fh, indent=2) fh.write("\n") sign(manifest) print(f"BUNDLE_INTEGRITY_GENERATED files={manifest['file_count']}") return 0 def do_verify(strict): mp = manifest_path() if not os.path.isfile(mp): # No manifest provisioned. Enforced mode treats this as fail-closed; dev is lax. if strict or _enforced(): print("BUNDLE_INTEGRITY_FAIL no_manifest_in_enforced_mode", file=sys.stderr) return 1 print("BUNDLE_INTEGRITY_SKIP no manifest (dev)") return 0 manifest = json.load(open(mp, encoding="utf-8")) root = bundle_root() stored = manifest.get("files", {}) drift = [] for rel, want in stored.items(): path = os.path.join(root, rel) if not os.path.isfile(path): drift.append(f"{rel}:removed") elif hash_file(path) != want: drift.append(f"{rel}:modified") # A NEW harness script / policy file that is not in the manifest is also drift. current = {os.path.relpath(f, root) for f in bundle_files()} for rel in current - set(stored): drift.append(f"{rel}:unmanifested") sig_state = check_signature(manifest) if sig_state == "invalid": print("BUNDLE_INTEGRITY_FAIL manifest_signature_invalid", file=sys.stderr) return 1 if sig_state == "unsigned" and (strict or _enforced()): print("BUNDLE_INTEGRITY_FAIL manifest_unsigned_in_enforced_mode", file=sys.stderr) return 1 if drift: print("BUNDLE_INTEGRITY_FAIL drift " + " ".join(sorted(drift)[:20]), file=sys.stderr) return 1 print(f"BUNDLE_INTEGRITY_OK files={len(stored)} anchor={sig_state}") return 0 def main() -> int: ap = argparse.ArgumentParser() sub = ap.add_subparsers(dest="cmd", required=True) sub.add_parser("generate") v = sub.add_parser("verify") v.add_argument("--strict", action="store_true") args = ap.parse_args() if args.cmd == "generate": return do_generate() if args.cmd == "verify": return do_verify(args.strict) return 2 if __name__ == "__main__": raise SystemExit(main())