tegwick
989d5d4f5d
Standard, schema and validator for the founder's admission of one bounded executor run: exact plan by digest, window by risk class, SI MODE @ SCOPE ceiling with least-privilege coverage, blast radius by consumer, named source-pinned checks, revert per mutating step, budget, single use, and the NK-WP-0040 receipt. check-approval verifies a carrier record (Approval Engine or founder SSH signature): aal2 human approver, not the executor, consumed at most once in the window. Examples express the bind split; step 10 is refused as human-intrinsic. 22 tests; make targets; tool index. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 994874@bnt-lap001 Assistant-Session: 6017c416-d619-47eb-bd1a-f670a02d7d87
323 lines
14 KiB
Python
323 lines
14 KiB
Python
#!/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 '<root>'}:{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 steps:
|
|
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", "<document>: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": "<record>: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())
|