#!/usr/bin/env python3 """CASAN Control Plane — governed settings store (Plan-13 core, harness-owned). This is the REUSABLE governance asset: it lives in the core harness, not in any generated app (e.g. OKR). Every write is (1) deny-by-default (only whitelisted keys), (2) approval-gated for security-sensitive keys, (3) versioned, and (4) recorded in a hash-linked audit chain so tampering is detectable. The standalone Control Plane web app (control-plane/) calls this CLI for all writes so the governance logic exists exactly once — in the harness. Store file: $CASAN_CP_STORE_FILE (default .specify/level5/control-plane-settings.json) """ import argparse import hashlib import json import os import sys from datetime import datetime, timezone SETTINGS_POLICY = { "compression.enabled": {"securitySensitive": False, "description": "Toggle context/token compression"}, "compression.mode": {"securitySensitive": False, "description": "extractive | structural | semantic-dedup | abstractive"}, "cost.absolute_cap_usd": {"securitySensitive": False, "description": "Absolute per-call cost cap"}, "model.primary": {"securitySensitive": False, "description": "Primary model spec (e.g. ollama:ornith:9b)"}, "security.strict": {"securitySensitive": True, "description": "H4 fail-closed strict mode"}, "kill_switch.global": {"securitySensitive": True, "description": "Global kill-switch engage/disengage"}, } GENESIS_HASH = "0" * 64 def project_root() -> str: return os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")) def store_path() -> str: return os.environ.get( "CASAN_CP_STORE_FILE", os.path.join(project_root(), ".specify/level5/control-plane-settings.json"), ) def now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def empty_store(): return {"settings": {}, "history": {}, "audit": []} def load_store(): path = store_path() if not os.path.isfile(path): return empty_store() with open(path, encoding="utf-8") as fh: data = json.load(fh) return {"settings": data.get("settings", {}), "history": data.get("history", {}), "audit": data.get("audit", [])} def save_store(store) -> None: path = store_path() os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as fh: json.dump(store, fh, indent=2, ensure_ascii=False) fh.write("\n") def hash_entry(entry) -> str: return hashlib.sha256(json.dumps(entry, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest() def append_audit(store, base) -> None: prev = store["audit"][-1] if store["audit"] else None prev_hash = prev["hash"] if prev else GENESIS_HASH seq = len(store["audit"]) + 1 without_hash = {"seq": seq, **base, "prevHash": prev_hash} store["audit"].append({**without_hash, "hash": hash_entry(without_hash)}) def do_set(key, value, actor, reason, approval): policy = SETTINGS_POLICY.get(key) if policy is None: print(f"SETTING_NOT_ALLOWED {key}", file=sys.stderr) return 2 if policy["securitySensitive"] and not (approval or "").strip(): print(f"APPROVAL_REQUIRED {key}", file=sys.stderr) return 3 store = load_store() prev = store["settings"].get(key) if prev is not None: store["history"].setdefault(key, []).append(prev) nxt = { "value": value, "version": (prev["version"] if prev else 0) + 1, "updatedAt": now_iso(), "actor": actor, "reason": reason, } store["settings"][key] = nxt append_audit(store, { "key": key, "action": "set", "value": value, "prevValue": prev["value"] if prev else None, "actor": actor, "reason": reason, "at": nxt["updatedAt"], }) save_store(store) print(json.dumps(nxt, ensure_ascii=False)) return 0 def do_rollback(key, actor, reason): store = load_store() history = store["history"].get(key, []) if not history: print(f"NO_PRIOR_VERSION {key}", file=sys.stderr) return 4 previous = history.pop() current = store["settings"].get(key) restored = { "value": previous["value"], "version": ((current["version"] if current else previous["version"]) + 1), "updatedAt": now_iso(), "actor": actor, "reason": reason, } store["settings"][key] = restored append_audit(store, { "key": key, "action": "rollback", "value": previous["value"], "prevValue": current["value"] if current else None, "actor": actor, "reason": reason, "at": restored["updatedAt"], }) save_store(store) print(json.dumps(restored, ensure_ascii=False)) return 0 def verify_audit(): store = load_store() prev_hash = GENESIS_HASH for entry in store["audit"]: rest = {k: v for k, v in entry.items() if k != "hash"} if rest.get("prevHash") != prev_hash or hash_entry(rest) != entry["hash"]: return {"ok": False, "brokenAt": entry["seq"]} prev_hash = entry["hash"] return {"ok": True, "brokenAt": None} def parse_value(raw): try: return json.loads(raw) except (ValueError, TypeError): return raw def main() -> int: ap = argparse.ArgumentParser() sub = ap.add_subparsers(dest="cmd", required=True) sub.add_parser("list-policy") sub.add_parser("get-all") g = sub.add_parser("get"); g.add_argument("key") s = sub.add_parser("set") s.add_argument("key"); s.add_argument("value") s.add_argument("--actor", required=True); s.add_argument("--reason", required=True); s.add_argument("--approval", default="") r = sub.add_parser("rollback") r.add_argument("key"); r.add_argument("--actor", required=True); r.add_argument("--reason", required=True) sub.add_parser("get-audit") sub.add_parser("verify-audit") e = sub.add_parser("effective") e.add_argument("key") e.add_argument("--default", default="") args = ap.parse_args() if args.cmd == "list-policy": print(json.dumps(SETTINGS_POLICY, ensure_ascii=False)); return 0 if args.cmd == "get-all": print(json.dumps(load_store()["settings"], ensure_ascii=False)); return 0 if args.cmd == "get": print(json.dumps(load_store()["settings"].get(args.key), ensure_ascii=False)); return 0 if args.cmd == "set": return do_set(args.key, parse_value(args.value), args.actor, args.reason, args.approval) if args.cmd == "rollback": return do_rollback(args.key, args.actor, args.reason) if args.cmd == "get-audit": print(json.dumps(load_store()["audit"], ensure_ascii=False)); return 0 if args.cmd == "verify-audit": verdict = verify_audit() print(f"CP_AUDIT ok={verdict['ok']} brokenAt={verdict['brokenAt']}") return 0 if verdict["ok"] else 1 if args.cmd == "effective": # Single read path for "effective setting": store value if set, else default. current = load_store()["settings"].get(args.key) if current is None: print(args.default) else: value = current["value"] print(value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)) return 0 return 2 if __name__ == "__main__": raise SystemExit(main())