tegwick
17f92c9f11
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 3775@bnt-lap001 Assistant-Session: cfacb6a5-ad5b-4f79-ac67-053e3b0d34a3
207 lines
7.9 KiB
Python
207 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
# /// script
|
|
# dependencies = ["jsonschema>=4.10,<5"]
|
|
# ///
|
|
"""Validate platform-root custody records (NK-ADR-0017, NK-WP-0045-T03).
|
|
|
|
Checks the JSON Schema (`canon/schemas/custody-record_v0.1.schema.json`) and
|
|
the semantics the schema cannot express: unique entries, at most one active
|
|
grant per exact (issuer, subject), revision ordering, supersession links,
|
|
append-only successors, and secret-looking values. It also provides the
|
|
fail-closed read rule a consumer applies to an observation (15-minute bound).
|
|
|
|
Offline and read-only: it never contacts an issuer, OpenBao or a directory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
import pathlib
|
|
import re
|
|
import sys
|
|
from typing import Any
|
|
|
|
import jsonschema
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[2]
|
|
SCHEMA = ROOT / "canon/schemas/custody-record_v0.1.schema.json"
|
|
MAX_CACHE_SECONDS = 900
|
|
|
|
# Value shapes that look like secret material (OpenBao/Vault tokens, PEM keys,
|
|
# JWTs). Field names are rejected by the schema; this catches smuggled values.
|
|
SECRET_VALUE = re.compile(
|
|
r"(\bhv[sbr]\.[A-Za-z0-9_-]{8,}|\bs\.[A-Za-z0-9]{20,}"
|
|
r"|-----BEGIN [A-Z ]*PRIVATE KEY-----"
|
|
r"|\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.)"
|
|
)
|
|
|
|
|
|
def load_schema() -> dict[str, Any]:
|
|
return json.loads(SCHEMA.read_text(encoding="utf-8"))
|
|
|
|
|
|
def parse_time(value: str) -> dt.datetime:
|
|
return dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
|
|
|
|
def key(entry: dict[str, Any]) -> tuple[str, str]:
|
|
return (entry["subject"]["issuer"], entry["subject"]["subject"])
|
|
|
|
|
|
def is_active(entry: dict[str, Any]) -> bool:
|
|
return "revoked" not in entry and "superseded_by" not in entry
|
|
|
|
|
|
def schema_errors(document: Any, schema: dict[str, Any]) -> list[str]:
|
|
validator = jsonschema.Draft202012Validator(
|
|
schema, format_checker=jsonschema.FormatChecker()
|
|
)
|
|
errors = []
|
|
for error in sorted(validator.iter_errors(document), key=lambda e: list(e.path)):
|
|
where = "/".join(str(part) for part in error.path) or "<root>"
|
|
errors.append(f"schema: {where}: {error.message}")
|
|
return errors
|
|
|
|
|
|
def _walk_strings(node: Any, path: str = ""):
|
|
if isinstance(node, dict):
|
|
for name, value in node.items():
|
|
yield from _walk_strings(value, f"{path}/{name}")
|
|
elif isinstance(node, list):
|
|
for index, value in enumerate(node):
|
|
yield from _walk_strings(value, f"{path}/{index}")
|
|
elif isinstance(node, str):
|
|
yield path or "<root>", node
|
|
|
|
|
|
def semantic_errors(record: dict[str, Any]) -> list[str]:
|
|
errors: list[str] = []
|
|
for where, value in _walk_strings(record):
|
|
if SECRET_VALUE.search(value):
|
|
errors.append(f"{where}: value looks like secret material")
|
|
|
|
revision = record["revision"]
|
|
entries = record["entries"]
|
|
by_id: dict[str, dict[str, Any]] = {}
|
|
for entry in entries:
|
|
if entry["entry_id"] in by_id:
|
|
errors.append(f"{entry['entry_id']}: duplicate entry_id")
|
|
by_id[entry["entry_id"]] = entry
|
|
|
|
active: dict[tuple[str, str], str] = {}
|
|
for entry in entries:
|
|
eid = entry["entry_id"]
|
|
granted = entry["granted"]
|
|
if granted["revision"] > revision:
|
|
errors.append(f"{eid}: granted.revision is after record revision")
|
|
if is_active(entry):
|
|
if key(entry) in active:
|
|
errors.append(
|
|
f"{eid}: second active grant for {key(entry)} "
|
|
f"(already {active[key(entry)]})"
|
|
)
|
|
active[key(entry)] = eid
|
|
revoked = entry.get("revoked")
|
|
if revoked:
|
|
if not granted["revision"] < revoked["revision"] <= revision:
|
|
errors.append(f"{eid}: revoked.revision must follow the grant and not exceed the record revision")
|
|
if parse_time(revoked["at"]) < parse_time(granted["at"]):
|
|
errors.append(f"{eid}: revoked before granted")
|
|
link = entry.get("superseded_by")
|
|
if link:
|
|
successor = by_id.get(link["entry_id"])
|
|
if successor is None or link["entry_id"] == eid:
|
|
errors.append(f"{eid}: superseded_by names no other entry")
|
|
else:
|
|
if successor["granted"]["revision"] != link["revision"]:
|
|
errors.append(f"{eid}: superseded_by.revision must equal the successor's granted.revision")
|
|
if parse_time(successor["granted"]["at"]) < parse_time(granted["at"]):
|
|
errors.append(f"{eid}: successor granted before the superseded entry")
|
|
if not granted["revision"] < link["revision"] <= revision:
|
|
errors.append(f"{eid}: superseded_by.revision must follow the grant and not exceed the record revision")
|
|
return errors
|
|
|
|
|
|
def validate(record: Any, schema: dict[str, Any] | None = None) -> list[str]:
|
|
errors = schema_errors(record, schema or load_schema())
|
|
if errors:
|
|
return errors
|
|
return semantic_errors(record)
|
|
|
|
|
|
def successor_errors(previous: dict[str, Any], current: dict[str, Any]) -> list[str]:
|
|
"""A later read must be a strictly newer, append-only revision."""
|
|
errors: list[str] = []
|
|
if previous["record_id"] != current["record_id"]:
|
|
return ["record_id changed between revisions"]
|
|
if current["revision"] < previous["revision"]:
|
|
return ["revision went backwards: stale read, keep the newer observation"]
|
|
if current["revision"] == previous["revision"]:
|
|
if current != previous:
|
|
errors.append("same revision with different content")
|
|
return errors
|
|
now = {entry["entry_id"]: entry for entry in current["entries"]}
|
|
for entry in previous["entries"]:
|
|
later = now.get(entry["entry_id"])
|
|
if later is None:
|
|
errors.append(f"{entry['entry_id']}: entry removed; custody actions append, never delete")
|
|
continue
|
|
for field in ("subject", "grant", "evidence", "granted"):
|
|
if later[field] != entry[field]:
|
|
errors.append(f"{entry['entry_id']}: {field} rewritten")
|
|
for field in ("revoked", "superseded_by"):
|
|
if field in entry and later.get(field) != entry[field]:
|
|
errors.append(f"{entry['entry_id']}: {field} rewritten")
|
|
return errors
|
|
|
|
|
|
def observe_root_grant(
|
|
record: dict[str, Any] | None,
|
|
issuer: str,
|
|
subject: str,
|
|
tenant: str,
|
|
read_at: dt.datetime,
|
|
now: dt.datetime,
|
|
) -> dict[str, Any] | None:
|
|
"""Return the ADR-0017 section 3 observation, or None (fail closed).
|
|
|
|
None when: the tenant is not `platform`; the record is missing, invalid or
|
|
older than the cache bound; or no active entry matches the exact
|
|
(issuer, subject).
|
|
"""
|
|
if tenant != "platform" or record is None or validate(record):
|
|
return None
|
|
bound = min(record["freshness"]["max_cache_seconds"], MAX_CACHE_SECONDS)
|
|
if not dt.timedelta(0) <= now - read_at <= dt.timedelta(seconds=bound):
|
|
return None
|
|
for entry in record["entries"]:
|
|
if key(entry) == (issuer, subject) and is_active(entry):
|
|
return {
|
|
"subject": dict(entry["subject"]),
|
|
"grant": dict(entry["grant"]),
|
|
"source": {"record_id": record["record_id"], "revision": record["revision"]},
|
|
"observed_at": read_at.isoformat().replace("+00:00", "Z"),
|
|
}
|
|
return None
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
parser.add_argument("paths", nargs="+", type=pathlib.Path)
|
|
args = parser.parse_args(argv)
|
|
schema = load_schema()
|
|
failed = False
|
|
for path in args.paths:
|
|
errors = validate(json.loads(path.read_text(encoding="utf-8")), schema)
|
|
for error in errors:
|
|
print(f"{path}: {error}", file=sys.stderr)
|
|
failed = failed or bool(errors)
|
|
if not errors:
|
|
print(f"{path}: ok")
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|