CASAN/packages/casan-harness/scripts/bash/control-plane-settings.py

#!/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 contextlib
import hashlib
import json
import os
import shutil
import subprocess
import sys
from datetime import datetime, timezone

try:
    import fcntl
    _HAVE_FCNTL = True
except ImportError:  # non-POSIX (e.g. Windows): best-effort, no OS lock
    _HAVE_FCNTL = False


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"},
    # Plan-17 loop governance overrides (meta-loop, T5). Loosening a loop budget /
    # widening a convergence window is security-sensitive: it grants the agent more
    # autonomy, so it needs a real approval + SoD and is clamped to org_ceiling.
    "loop.max_steps": {"securitySensitive": True, "description": "Loop Governor: max steps per run"},
    "loop.max_tokens": {"securitySensitive": True, "description": "Loop Governor: max tokens per run"},
    "loop.max_wall_clock_sec": {"securitySensitive": True, "description": "Loop Governor: max wall-clock seconds per run"},
    "loop.max_cost_usd": {"securitySensitive": True, "description": "Loop Governor: max cost USD per run"},
    "loop.max_corrections_per_step": {"securitySensitive": True, "description": "Loop Governor: max corrections per step"},
    "loop.oscillation_repeat": {"securitySensitive": True, "description": "Convergence: identical-action repeats before OSCILLATING"},
    "loop.no_progress_window": {"securitySensitive": True, "description": "Convergence: no-progress window before STALLED"},
}

GENESIS_HASH = "0" * 64


def project_root() -> str:
    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")):
            return p
        p = os.path.dirname(p)
    return os.path.abspath(os.path.join(d, "..", "..", ".."))


def _tenant_id():
    # SEC-23 (MT-01): when a tenant id is set, the settings store (and its embedded
    # audit hash-chain) is partitioned per tenant so tenant A cannot read/modify
    # tenant B's governance state. An invalid id fails closed.
    import re
    t = os.environ.get("CASAN_TENANT_ID", "").strip()
    if not t:
        return None
    if not re.fullmatch(r"[A-Za-z0-9_-]+", t):
        raise SystemExit("CP_DENY tenant_id_invalid")
    return t


def store_path() -> str:
    explicit = os.environ.get("CASAN_CP_STORE_FILE")
    if explicit:
        return explicit
    tenant = _tenant_id()
    if tenant:
        base = os.environ.get(
            "CASAN_TENANT_STATE_ROOT",
            os.path.join(project_root(), ".specify/state/tenants"),
        )
        return os.path.join(base, tenant, "control-plane", "settings.json")
    return 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:
    # SEC-19 (ARCH-05): write atomically (tmp + rename) so a crash or a concurrent
    # reader never sees a half-written store / broken hash chain.
    path = store_path()
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(store, fh, indent=2, ensure_ascii=False)
        fh.write("\n")
        fh.flush()
        os.fsync(fh.fileno())
    os.replace(tmp, path)


@contextlib.contextmanager
def store_lock():
    """SEC-19 (ARCH-05): serialize the load→modify→save critical section so two
    concurrent `set`/`rollback` runs cannot lose a write or fork the audit chain.
    POSIX flock; a best-effort no-op where fcntl is unavailable."""
    lock_path = store_path() + ".lock"
    os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
    fh = open(lock_path, "w")
    try:
        if _HAVE_FCNTL:
            fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
        yield
    finally:
        try:
            if _HAVE_FCNTL:
                fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
        finally:
            fh.close()


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)})


# --- SEC-06 (H-06): sign the audit-chain HEAD -------------------------------
# The hash chain alone is recomputable: a file-writer who edits the store can
# recompute every hash and `verify-audit` (chain-only) would still PASS, so
# governance-report would falsely report CERTIFIED. Signing the head with an
# OFF-REPO private key (pubkey provisioned out-of-band; KMS in prod = SEC-02/16)
# means a recompute-attacker without the key cannot forge a matching signature.

def _enforced() -> bool:
    return os.environ.get("CASAN_PROFILE") == "prod" or os.environ.get("CASAN_VERIFY_STRICT") == "1"


def _cp_priv() -> str:
    key_dir = os.environ.get("CASAN_CP_KEY_DIR") or os.path.join(os.path.expanduser("~"), ".casan", "audit-keys")
    return os.path.join(key_dir, "cp-private.pem")


def _cp_pub() -> str:
    # Prod provisions this out-of-band (or via KMS). Default is adjacent to the
    # store for self-contained dev; an attacker who can also rewrite the pubkey is
    # covered by ARCH-01/SEC-16 (signed harness+policy bundle).
    return os.environ.get("CASAN_CP_PUB") or (store_path() + ".pub")


def _head_paths():
    sp = store_path()
    return sp + ".head", sp + ".head.sig"


def chain_head(store) -> str:
    return store["audit"][-1]["hash"] if store["audit"] else GENESIS_HASH


def sign_head(store) -> None:
    head_file, sig_file = _head_paths()
    head = chain_head(store)
    with open(head_file, "w", encoding="utf-8") as fh:
        fh.write(head)
    ossl = shutil.which("openssl")
    if not ossl:
        return  # keyless dev: no signature — enforced mode rejects at verify time
    priv, pub = _cp_priv(), _cp_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)
    # Always re-export the pubkey so it matches the signing key (key-sync invariant).
    subprocess.run([ossl, "rsa", "-in", priv, "-pubout", "-out", pub], capture_output=True)
    subprocess.run([ossl, "dgst", "-sha256", "-sign", priv, "-out", sig_file, head_file], capture_output=True)


def verify_signature(store):
    """Returns state in {'signed','unsigned','invalid'} + a detail string."""
    head_file, sig_file = _head_paths()
    pub = _cp_pub()
    ossl = shutil.which("openssl")
    if not (ossl and os.path.isfile(head_file) and os.path.isfile(sig_file) and os.path.isfile(pub)):
        return "unsigned", "missing signature/pubkey/openssl"
    try:
        stored_head = open(head_file, encoding="utf-8").read().strip()
    except OSError:
        return "invalid", "head-file unreadable"
    if stored_head != chain_head(store):
        return "invalid", "head-file != recomputed head (chain recomputed?)"
    res = subprocess.run([ossl, "dgst", "-sha256", "-verify", pub, "-signature", sig_file, head_file],
                         capture_output=True)
    return ("signed", "ok") if res.returncode == 0 else ("invalid", "signature verify failed")


def _approval_enforced() -> bool:
    return os.environ.get("CASAN_PROFILE") == "prod" or os.environ.get("CASAN_APPROVAL_STRICT") == "1"


def check_approval(key, actor, approval):
    """SEC-07 (M-08): a security-sensitive setting change needs a REAL approval.
    Dev (default) keeps the backward-compatible "any non-empty --approval" gate;
    enforced mode requires a REGISTERED reviewer to cryptographically sign this
    change (verified by approval-verify.sh), so a bare string can no longer approve.
    Env contract mirrors governance-check: CASAN_APPROVER + CASAN_APPROVAL_SIG (or
    CASAN_APPROVAL_JWT). The signed assertion is bound to the key being changed."""
    if not _approval_enforced():
        return bool((approval or "").strip()), "dev_nonstrict"
    approver = os.environ.get("CASAN_APPROVER", "")
    if not approver:
        return False, "no_registered_approver"
    verifier = os.path.join(os.path.dirname(__file__), "approval-verify.sh")
    if not os.path.isfile(verifier):
        return False, "approval_verifier_missing"
    import tempfile
    fd, inp = tempfile.mkstemp(suffix=".approval-input")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            fh.write(key)  # the approval is bound to the key under change
        sig = os.environ.get("CASAN_APPROVAL_SIG", "-")
        rc = subprocess.run(["bash", verifier, "policy_change", actor, inp, approver, sig],
                            capture_output=True).returncode
    finally:
        try:
            os.unlink(inp)
        except OSError:
            pass
    return rc == 0, f"approval_verify_rc={rc}"


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"]:
        ok, reason_ = check_approval(key, actor, approval)
        if not ok:
            print(f"APPROVAL_REQUIRED {key} ({reason_})", file=sys.stderr)
            return 3
    with store_lock():  # SEC-19: atomic read-modify-write
        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)
        sign_head(store)
    print(json.dumps(nxt, ensure_ascii=False))
    return 0


def do_rollback(key, actor, reason):
    with store_lock():  # SEC-19: atomic read-modify-write
        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)
        sign_head(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"], "anchor": "chain-broken"}
        prev_hash = entry["hash"]
    # SEC-06: the chain is recomputable, so require a valid HEAD signature. A
    # present-but-mismatched signature is always a failure; a MISSING signature
    # fails only in enforced mode (dev stays permissive for backward compat).
    sig_state, detail = verify_signature(store)
    if sig_state == "invalid":
        return {"ok": False, "brokenAt": None, "anchor": "signature-invalid", "detail": detail}
    if sig_state == "unsigned" and _enforced():
        return {"ok": False, "brokenAt": None, "anchor": "unsigned-strict-fail", "detail": detail}
    return {"ok": True, "brokenAt": None, "anchor": sig_state}


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']} anchor={verdict.get('anchor')}")
        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())