286 lines
10 KiB
Python
286 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""CASAN Approval Inbox + Delegation resolver (Plan-13 HITL surface).
|
|
|
|
Harness-owned primitive for the Control Panel:
|
|
* delegation-policy.yaml resolves whether an action needs human approval;
|
|
* pending proposals are stored in a versioned JSON file;
|
|
* approve/reject decisions are append-only oversight events with a hash chain;
|
|
* Separation of Duties is enforced in the primitive (proposer != approver).
|
|
|
|
The web app may present this data, but the governance state lives in the harness.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
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")) 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, "..", "..", ".."))
|
|
|
|
|
|
ROOT = project_root()
|
|
HARNESS_ROOT = os.path.join(ROOT, "packages", "casan-harness")
|
|
TENANT_STORE = os.path.join(HARNESS_ROOT, "scripts", "bash", "tenant-store.sh")
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def inbox_path() -> str:
|
|
if os.environ.get("CASAN_APPROVAL_INBOX_FILE"):
|
|
return guarded_override(os.environ["CASAN_APPROVAL_INBOX_FILE"])
|
|
if os.environ.get("CASAN_TENANT_ID"):
|
|
return tenant_path("approvals/approval-inbox.json")
|
|
return os.path.join(ROOT, ".specify", "level5", "approval-inbox.json")
|
|
|
|
|
|
def tenant_path(logical: str) -> str:
|
|
r = subprocess.run(["bash", TENANT_STORE, "resolve", logical], cwd=ROOT, capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
|
|
return r.stdout.strip()
|
|
|
|
|
|
def guarded_override(path: str) -> str:
|
|
if path and os.environ.get("CASAN_TENANT_ID"):
|
|
r = subprocess.run(["bash", TENANT_STORE, "guard", path], cwd=ROOT, capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
|
|
return path
|
|
|
|
|
|
def policy_path() -> str:
|
|
return os.environ.get("CASAN_DELEGATION_POLICY") or os.path.join(
|
|
HARNESS_ROOT, "config", "delegation-policy.yaml"
|
|
)
|
|
|
|
|
|
def empty_store():
|
|
return {"proposals": [], "oversight": []}
|
|
|
|
|
|
def load_store():
|
|
path = inbox_path()
|
|
if not os.path.isfile(path):
|
|
return empty_store()
|
|
with open(path, encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
return {
|
|
"proposals": data.get("proposals", []),
|
|
"oversight": data.get("oversight", []),
|
|
}
|
|
|
|
|
|
def save_store(store):
|
|
path = inbox_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")
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def hash_entry(entry) -> str:
|
|
return hashlib.sha256(json.dumps(entry, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def append_oversight(store, base):
|
|
prev = store["oversight"][-1] if store["oversight"] else None
|
|
prev_hash = prev["hash"] if prev else GENESIS_HASH
|
|
seq = len(store["oversight"]) + 1
|
|
core = {"seq": seq, **base, "prevHash": prev_hash}
|
|
store["oversight"].append({**core, "hash": hash_entry(core)})
|
|
|
|
|
|
def load_policy():
|
|
path = policy_path()
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
# File is JSON-in-YAML syntax: valid YAML, dependency-free JSON parse.
|
|
return json.load(fh)
|
|
except Exception as exc:
|
|
raise SystemExit(f"DELEGATION_POLICY_DENY unreadable policy: {exc}")
|
|
|
|
|
|
def resolve(policy, project, action, risk, sensitive):
|
|
projects = policy.get("projects", {})
|
|
project_cfg = projects.get(project, {})
|
|
level = project_cfg.get("actions", {}).get(action) or project_cfg.get("default_level") or policy.get("default_level", "L1")
|
|
level_cfg = policy.get("levels", {}).get(level)
|
|
if not level_cfg:
|
|
return {"project": project, "action": action, "level": level, "requires_approval": True, "reason": "unknown_level_fail_closed"}
|
|
high = sensitive or risk in {"high", "critical", "security-sensitive"}
|
|
requires = bool(level_cfg.get("high_requires_approval" if high else "standard_requires_approval", True))
|
|
return {
|
|
"project": project,
|
|
"action": action,
|
|
"level": level,
|
|
"risk": risk,
|
|
"sensitive": bool(sensitive),
|
|
"requires_approval": requires,
|
|
"reason": "high_risk_gate" if high and requires else ("level_requires_approval" if requires else "delegated_by_policy"),
|
|
}
|
|
|
|
|
|
def cmd_resolve(args):
|
|
verdict = resolve(load_policy(), args.project, args.action, args.risk, args.sensitive)
|
|
print(json.dumps(verdict, ensure_ascii=False))
|
|
return 0 if not verdict["requires_approval"] else 2
|
|
|
|
|
|
def cmd_submit(args):
|
|
policy_verdict = resolve(load_policy(), args.project, args.action, args.risk, args.sensitive)
|
|
store = load_store()
|
|
raw = f"{args.action}|{args.target}|{args.proposer}|{args.reason}|{now_iso()}"
|
|
pid = "AP-" + hashlib.sha256(raw.encode("utf-8")).hexdigest()[:12]
|
|
try:
|
|
payload = json.loads(args.payload) if args.payload else {}
|
|
except ValueError:
|
|
print("APPROVAL_SUBMIT_DENY payload_not_json", file=sys.stderr)
|
|
return 64
|
|
status = "pending" if policy_verdict["requires_approval"] else "auto_allowed"
|
|
proposal = {
|
|
"id": pid,
|
|
"status": status,
|
|
"project": args.project,
|
|
"action": args.action,
|
|
"target": args.target,
|
|
"risk": args.risk,
|
|
"sensitive": bool(args.sensitive),
|
|
"proposer": args.proposer,
|
|
"reason": args.reason,
|
|
"payload": payload,
|
|
"delegation": policy_verdict,
|
|
"created_at": now_iso(),
|
|
"decided_at": None,
|
|
"approver": None,
|
|
"decision_reason": None,
|
|
}
|
|
store["proposals"].append(proposal)
|
|
append_oversight(store, {
|
|
"event": "proposal_submitted",
|
|
"proposal_id": pid,
|
|
"actor": args.proposer,
|
|
"action": args.action,
|
|
"target": args.target,
|
|
"status": status,
|
|
"reason": args.reason,
|
|
"at": proposal["created_at"],
|
|
})
|
|
save_store(store)
|
|
print(json.dumps(proposal, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
def cmd_list(args):
|
|
store = load_store()
|
|
proposals = store["proposals"]
|
|
if args.status != "all":
|
|
proposals = [p for p in proposals if p.get("status") == args.status]
|
|
print(json.dumps({"count": len(proposals), "proposals": proposals, "oversight": store["oversight"][-50:]}, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
def cmd_decide(args):
|
|
if args.decision not in {"approve", "reject"}:
|
|
print("APPROVAL_DECIDE_DENY decision_must_be_approve_or_reject", file=sys.stderr)
|
|
return 64
|
|
store = load_store()
|
|
proposal = next((p for p in store["proposals"] if p.get("id") == args.id), None)
|
|
if proposal is None:
|
|
print(f"APPROVAL_DECIDE_DENY unknown_id {args.id}", file=sys.stderr)
|
|
return 1
|
|
if proposal.get("status") != "pending":
|
|
print(f"APPROVAL_DECIDE_DENY not_pending status={proposal.get('status')}", file=sys.stderr)
|
|
return 3
|
|
if proposal.get("proposer") == args.approver:
|
|
print(f"APPROVAL_DECIDE_DENY sod_self_approval actor={args.approver}", file=sys.stderr)
|
|
return 3
|
|
proposal["status"] = "approved" if args.decision == "approve" else "rejected"
|
|
proposal["approver"] = args.approver
|
|
proposal["decision_reason"] = args.reason
|
|
proposal["decided_at"] = now_iso()
|
|
append_oversight(store, {
|
|
"event": f"proposal_{proposal['status']}",
|
|
"proposal_id": proposal["id"],
|
|
"actor": args.approver,
|
|
"action": proposal.get("action"),
|
|
"target": proposal.get("target"),
|
|
"status": proposal["status"],
|
|
"reason": args.reason,
|
|
"at": proposal["decided_at"],
|
|
})
|
|
save_store(store)
|
|
print(json.dumps(proposal, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
def cmd_verify(args):
|
|
store = load_store()
|
|
prev = GENESIS_HASH
|
|
for entry in store["oversight"]:
|
|
rest = {k: v for k, v in entry.items() if k != "hash"}
|
|
if rest.get("prevHash") != prev or hash_entry(rest) != entry.get("hash"):
|
|
print(f"APPROVAL_INBOX_AUDIT ok=false brokenAt={entry.get('seq')}")
|
|
return 1
|
|
prev = entry["hash"]
|
|
print(f"APPROVAL_INBOX_AUDIT ok=true records={len(store['oversight'])} head={prev}")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
r = sub.add_parser("resolve")
|
|
r.add_argument("--project", default="default")
|
|
r.add_argument("--action", required=True)
|
|
r.add_argument("--risk", default="standard")
|
|
r.add_argument("--sensitive", action="store_true")
|
|
s = sub.add_parser("submit")
|
|
s.add_argument("--project", default="default")
|
|
s.add_argument("--action", required=True)
|
|
s.add_argument("--target", required=True)
|
|
s.add_argument("--risk", default="standard")
|
|
s.add_argument("--sensitive", action="store_true")
|
|
s.add_argument("--proposer", required=True)
|
|
s.add_argument("--reason", required=True)
|
|
s.add_argument("--payload", default="{}")
|
|
l = sub.add_parser("list")
|
|
l.add_argument("--status", default="pending", choices=["pending", "approved", "rejected", "auto_allowed", "all"])
|
|
d = sub.add_parser("decide")
|
|
d.add_argument("--id", required=True)
|
|
d.add_argument("--decision", required=True)
|
|
d.add_argument("--approver", required=True)
|
|
d.add_argument("--reason", required=True)
|
|
sub.add_parser("verify-audit")
|
|
args = ap.parse_args()
|
|
if args.cmd == "resolve":
|
|
return cmd_resolve(args)
|
|
if args.cmd == "submit":
|
|
return cmd_submit(args)
|
|
if args.cmd == "list":
|
|
return cmd_list(args)
|
|
if args.cmd == "decide":
|
|
return cmd_decide(args)
|
|
if args.cmd == "verify-audit":
|
|
return cmd_verify(args)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|