#!/usr/bin/env python3 """Validate NetKingdom Bounded Execution Admissions (standard v0.1, NK-WP-0056-T01). Offline and deterministic. Three commands: validate [--as-of T] FILE... schema plus rules BEA-1..BEA-14 for each admission check-approval ADMISSION RECORD rules BEA-C1..BEA-C7: does a carrier record (Approval Engine approval or founder SSH signature) admit exactly this admission, and was it consumed at most once, by the executor, in the window digest FILE print plan_digest and admission_digest (for authoring) Findings name rules and locations, never field values. Nothing here contacts a service, reads a secret or authorizes anything: a valid admission is a well-formed request, not an admission. """ from __future__ import annotations import argparse import datetime as dt import hashlib import json import pathlib import re import sys from typing import Any import jsonschema import yaml ROOT = pathlib.Path(__file__).resolve().parents[2] SCHEMA = ROOT / "canon/schemas/bounded-execution-admission_v0.1.schema.json" SECRET_VALUE = re.compile( r"(\bhv[sbr]\.[A-Za-z0-9_-]{8,}|\bs\.[A-Za-z0-9]{20,}" r"|-----BEGIN [A-Z ]*PRIVATE KEY-----|AGE-SECRET-KEY-1[A-Z0-9]{10,}" r"|\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\." r"|://[^/\s:@]+:[^/\s@]{3,}@)" ) MAX_WINDOW = {"low": dt.timedelta(hours=72), "medium": dt.timedelta(hours=24), "high": dt.timedelta(hours=4)} MAX_ATTEMPTS = {"low": 3, "medium": 3, "high": 1} CARRIER_RECORD = "netkingdom.io/bounded-execution-approval/v0.1" # -------------------------------------------------------------------------------------------------- # Loading and digests # -------------------------------------------------------------------------------------------------- def load(path: pathlib.Path) -> Any: text = path.read_text(encoding="utf-8") data = json.loads(text) if path.suffix == ".json" else yaml.safe_load(text) return json.loads(json.dumps(data, default=str)) # YAML dates become strings, as JSON would carry them def canonical(value: Any) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def plan_digest(document: dict) -> str: return "sha256:" + hashlib.sha256(canonical(document["spec"]["plan"])).hexdigest() def admission_digest(document: dict) -> str: """What a carrier binds: the whole admission, including plan_digest.""" return "sha256:" + hashlib.sha256(canonical(document)).hexdigest() def parse_time(value: str) -> dt.datetime: return dt.datetime.fromisoformat(value.replace("Z", "+00:00")) def strings(value: Any): if isinstance(value, str): yield value elif isinstance(value, dict): for item in value.values(): yield from strings(item) elif isinstance(value, list): for item in value: yield from strings(item) # -------------------------------------------------------------------------------------------------- # Rules # -------------------------------------------------------------------------------------------------- def covers(grant: dict, need: dict) -> bool: return grant["mode"] == need["mode"] and (need["scope"] == grant["scope"] or need["scope"].startswith(grant["scope"] + "/")) def evaluate(document: Any, schema: dict, as_of: dt.datetime | None = None) -> dict: findings: list[dict] = [] def fail(rule: str, where: str) -> None: findings.append({"rule": rule, "where": where}) validator = jsonschema.Draft202012Validator(schema, format_checker=jsonschema.FormatChecker()) schema_errors = sorted(f"{'/'.join(str(p) for p in e.absolute_path) or ''}:{e.validator}" for e in validator.iter_errors(document)) if schema_errors: return {"valid": False, "findings": [{"rule": "BEA-1", "where": e} for e in schema_errors]} spec = document["spec"] risk = spec["risk_class"] steps = spec["plan"]["steps"] reverts = spec["plan"].get("revert_steps", []) every_step = steps + reverts ceiling = spec["scope_ceiling"] checks = {c["id"]: c for c in spec["checks"]} pinned = {f["path"]: f["sha256"] for f in spec["plan"]["source"]["files"]} # BEA-2 the admission binds the exact plan. if spec["plan_digest"] != plan_digest(document): fail("BEA-2", "spec/plan_digest") # BEA-3 a bounded window, tighter for higher risk. start, end = parse_time(spec["window"]["not_before"]), parse_time(spec["window"]["not_after"]) window = end - start if window <= dt.timedelta(0): fail("BEA-3", "spec/window") elif window > MAX_WINDOW[risk]: fail("BEA-3", "spec/window:longer-than-risk-class-allows") # BEA-4 the executor is not a person and not an approver; an agent only when harness-launched (SI-14). executor = spec["executor"] if executor["principal_type"] == "HUMAN": fail("BEA-4", "spec/executor/principal_type") if executor["identity"] in spec["approval"]["approvers"]: fail("BEA-4", "spec/executor/identity:is-an-approver") if executor["principal_type"] == "AGENT" and executor.get("harness_launched") is not True: fail("BEA-4", "spec/executor/harness_launched") # BEA-5 the ceiling never holds sovereign or rule-changing authority; whole-service ADMINISTER is # root-class, high risk and one consumer at a time. for i, grant in enumerate(ceiling): where = f"spec/scope_ceiling/{i}" if grant["mode"] in ("SOVEREIGN", "GOVERN"): fail("BEA-5", where + ":mode") if grant["mode"] == "CONSTRUCT" and "PRODUCTION" in grant["environments"]: fail("BEA-5", where + ":construct-in-production") if grant["mode"] == "ADMINISTER": if spec["blast_radius"]["max_concurrent"] != 1: fail("BEA-5", where + ":administer-needs-max-concurrent-1") if "/" not in grant["scope"]: users = [s for s in every_step if covers(grant, s["requires"])] if risk != "high" or any(s["class"] != "root-class" for s in users): fail("BEA-5", where + ":whole-service-administer-is-root-class-high-risk") # BEA-6 least privilege: every step fits the ceiling, and every ceiling grant is used. for i, step in enumerate(every_step): if not any(covers(g, step["requires"]) for g in ceiling): fail("BEA-6", f"step/{step['id']}/requires:outside-ceiling") for i, grant in enumerate(ceiling): if not any(covers(grant, s["requires"]) for s in every_step): fail("BEA-6", f"spec/scope_ceiling/{i}:unused") # BEA-7 classes: human-intrinsic never; root-class only at high risk. for step in every_step: if step["class"] == "human-intrinsic": fail("BEA-7", f"step/{step['id']}/class:human-intrinsic") if step["class"] == "root-class" and risk != "high": fail("BEA-7", f"step/{step['id']}/class:root-class-needs-high-risk") # BEA-8 blast radius. radius = spec["blast_radius"] for step in every_step: if step["consumer"] not in radius["consumers"]: fail("BEA-8", f"step/{step['id']}/consumer:outside-blast-radius") if radius["max_concurrent"] > len(radius["consumers"]): fail("BEA-8", "spec/blast_radius/max_concurrent") if risk in ("medium", "high") and radius["max_concurrent"] != 1: fail("BEA-8", "spec/blast_radius/max_concurrent:one-at-a-time-for-medium-and-high") # BEA-9 named machine checks, pinned to the admitted source. if len(checks) != len(spec["checks"]): fail("BEA-9", "spec/checks:duplicate-id") used = set() for step in every_step: # revert steps verify the restored state with their own checks if step["mutates"] and not step.get("checks_after"): fail("BEA-9", f"step/{step['id']}/checks_after:missing") for ref in step.get("checks_after", []): used.add(ref) if ref not in checks: fail("BEA-9", f"step/{step['id']}/checks_after:undeclared") for cid, check in checks.items(): if cid not in used: fail("BEA-9", f"check/{cid}:unused") if pinned.get(check["ref"]) != check["sha256"]: fail("BEA-9", f"check/{cid}/ref:not-pinned-in-plan-source") # BEA-10 every mutating step can be undone, or is declared irreversible at high risk. ids = [s["id"] for s in every_step] if len(set(ids)) != len(ids): fail("BEA-10", "plan:duplicate-step-id") revert_ids = {s["id"] for s in reverts} for step in steps: if not step["mutates"]: continue if step.get("irreversible"): if risk != "high": fail("BEA-10", f"step/{step['id']}/irreversible:needs-high-risk") elif step.get("revert") not in revert_ids: fail("BEA-10", f"step/{step['id']}/revert:missing") for step in reverts: if not step["mutates"] or step.get("revert") or step.get("irreversible"): fail("BEA-10", f"revert_step/{step['id']}:must-mutate-and-not-nest") # BEA-11 stop-and-revert follows the plan: a revertable plan reverts on failure. policy = spec["stop_and_revert"] if revert_ids and policy["on_check_failure"] != "revert": fail("BEA-11", "spec/stop_and_revert/on_check_failure") if not revert_ids and policy["on_check_failure"] == "revert": fail("BEA-11", "spec/stop_and_revert/on_check_failure:nothing-to-revert") # BEA-12 the budget fits the window and the checks fit the budget. budget = spec["budget"] if budget["wall_time_seconds"] > max(window.total_seconds(), 0): fail("BEA-12", "spec/budget/wall_time_seconds:exceeds-window") if budget["attempts"] > MAX_ATTEMPTS[risk]: fail("BEA-12", "spec/budget/attempts") check_time = sum(c["timeout_seconds"] + c.get("settle_seconds", 0) for c in spec["checks"]) if check_time > budget["wall_time_seconds"]: fail("BEA-12", "spec/budget/wall_time_seconds:shorter-than-checks") # BEA-13 the executed source is pinned and the plan names it. if not spec["plan"]["source"]["files"]: fail("BEA-13", "spec/plan/source/files") # BEA-14 no secret-shaped values anywhere. if any(SECRET_VALUE.search(s) for s in strings(document)): fail("BEA-14", ":secret-shaped-value") report = {"valid": not findings, "findings": findings, "plan_digest": plan_digest(document), "admission_digest": admission_digest(document)} if as_of is not None: report["window_state"] = ("not-yet" if as_of < start else "expired" if as_of >= end else "open") return report def check_approval(document: dict, record: Any) -> dict: """BEA-C1..C7: a carrier record admits exactly this admission, consumed at most once in the window.""" findings: list[dict] = [] def fail(rule: str, where: str) -> None: findings.append({"rule": rule, "where": where}) spec = document["spec"] required = ("apiVersion", "carrier", "approval_ref", "admission_id", "admission_digest", "approver", "approved_at", "consumptions") if not isinstance(record, dict) or any(k not in record for k in required) \ or record.get("apiVersion") != CARRIER_RECORD: return {"admitted": False, "findings": [{"rule": "BEA-C1", "where": ":shape"}]} if record["carrier"] not in spec["approval"]["carriers"]: fail("BEA-C1", "record/carrier") if record["admission_id"] != spec["id"] or record["admission_digest"] != admission_digest(document): fail("BEA-C2", "record/admission_digest") approver = record["approver"] if isinstance(record["approver"], dict) else {} if approver.get("name") not in spec["approval"]["approvers"] or approver.get("principal_type") != "HUMAN": fail("BEA-C3", "record/approver") if approver.get("assurance") != "aal2": fail("BEA-C4", "record/approver/assurance") if approver.get("name") == spec["executor"]["identity"]: fail("BEA-C3", "record/approver:is-the-executor") start, end = parse_time(spec["window"]["not_before"]), parse_time(spec["window"]["not_after"]) approved = parse_time(record["approved_at"]) if approved >= end: fail("BEA-C5", "record/approved_at:after-window") consumptions = record["consumptions"] if isinstance(record["consumptions"], list) else [] if len(consumptions) > 1: fail("BEA-C6", "record/consumptions:more-than-once") for i, use in enumerate(consumptions): at = parse_time(use["at"]) if not (start <= at < end) or at < approved: fail("BEA-C7", f"record/consumptions/{i}/at") if use.get("by") != spec["executor"]["identity"]: fail("BEA-C7", f"record/consumptions/{i}/by") return {"admitted": not findings, "consumed": len(consumptions) == 1, "findings": findings} # -------------------------------------------------------------------------------------------------- # CLI # -------------------------------------------------------------------------------------------------- def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) sub = parser.add_subparsers(dest="command", required=True) v = sub.add_parser("validate") v.add_argument("paths", nargs="+", type=pathlib.Path) v.add_argument("--as-of", help="evaluate the window at this time (ISO 8601)") c = sub.add_parser("check-approval") c.add_argument("admission", type=pathlib.Path) c.add_argument("record", type=pathlib.Path) d = sub.add_parser("digest") d.add_argument("path", type=pathlib.Path) args = parser.parse_args(argv) schema = json.loads(SCHEMA.read_text(encoding="utf-8")) if args.command == "digest": document = load(args.path) print(json.dumps({"plan_digest": plan_digest(document), "admission_digest": admission_digest(document)})) return 0 if args.command == "check-approval": document = load(args.admission) if not evaluate(document, schema)["valid"]: print(json.dumps({"admitted": False, "findings": [{"rule": "BEA-C0", "where": "admission:invalid"}]})) return 1 report = check_approval(document, load(args.record)) print(json.dumps(report, sort_keys=True)) return 0 if report["admitted"] else 1 as_of = parse_time(args.as_of) if args.as_of else None worst = 0 for path in args.paths: report = evaluate(load(path), schema, as_of) report["file"] = str(path) print(json.dumps(report, sort_keys=True)) worst = max(worst, 0 if report["valid"] else 1) return worst if __name__ == "__main__": sys.exit(main())