feat(plan-01): Phase 1 — relocate harness code to packages/casan-harness (symlink facade)
Physically move the pure-code subtrees out of .specify into the package, leaving compat symlinks at the old .specify/<dir> paths so every existing reference (internal CASAN_HARNESS_ROOT + external CI/docker/mjs) keeps resolving. Runtime state stays put. Moved (git mv): scripts/ tests/ security/ templates/ config/ governance/ memory/ .specify/<dir> -> packages/casan-harness/<dir> (+ .specify/<dir> symlink) Stays in .specify (state/governance/domain, handled later): logs/ agentops/ level5/ init-options.json traceability-map.json Python `.resolve()` self-location followed the compat symlink into packages and lost the app root; generate-casan-demo-context.py, generate-agentops-dashboard.py and dashboard-server.py now walk UP for the `.specify` state marker instead of a fixed parent depth (fixes "missing trace files" in run-casan4). Full gate: PASS=64 FAIL=0 SKIP=3 (CASAN_CI_STEP_TIMEOUT_SEC=1200 — track-a ~450s runs close to the 600s default and can tip over under load; this is timing variance, not a regression — it passed cleanly with headroom). Runtime log/audit artifacts kept unstaged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
thanhnv
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
2c765c9a45
commit
664bd1f00c
@@ -0,0 +1,199 @@
|
||||
#!/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
|
||||
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(
|
||||
os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")), ".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())
|
||||
Reference in New Issue
Block a user