kings-guard/scripts/check_layer_conformance.py
tegwick 9daea96c43 Finish KG-WP-0003: stream completeness and live qonto observation
Classify evidence as load-bearing or attributive, draft the emission-cadence
declaration for Taxonomy, treat silence as a stream finding, keep completeness
separate from record richness, forbid immune memory as a state plane, and make
containment proposals reconstructable to their origin. Observe real
qonto-assistant audit events; deny-class completeness stays unknown until the
source publishes a heartbeat.

Assistant: grok
Assistant-Session: 01a05ef1-9e5a-70f2-b0ff-0b05d6b38ae9
2026-09-02 00:11:57 +02:00

215 lines
8.3 KiB
Python

#!/usr/bin/env python3
"""Check kings-guard against the NetKingdom security layer model (§5, §11).
Read-only. kings-guard's whole position under the standard rests on one claim:
it holds no direct client for any Tooling-layer system, and the
capabilities that would need one sit at zero instead (§11 blocked-clean)
That claim has been asserted in prose since KG-DEC-2026-001. §11 (v0.6) requires
a machine-readable declaration because prose cannot distinguish a declaration
from a transcribed review. This script is what makes the claim checkable: it
fails if a Tooling client appears in src/ without a matching layer.yaml entry.
The failure it exists to catch is a *convenience* — someone reaching for an
OpenBao or cluster client during an incident because the engine surface still
does not exist (§9.2). That is precisely the "small convenience" §6 warns about,
and it would arrive as a one-line import.
Review dates are reported, never enforced: a date-triggered failure breaks the
build on a calendar day with no code change.
Exit 0 clean, 1 undeclared contact found, 2 declaration malformed.
"""
from __future__ import annotations
import argparse
import ast
import re
import sys
from datetime import date
from pathlib import Path
import yaml
ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src" / "kings_guard"
DECL = ROOT / "layer.yaml"
# Import roots that would constitute a direct Tooling-layer client under §4.
# Matched against the top-level module of every import in src/.
TOOLING_IMPORTS = {
"hvac": "OpenBao / Vault client",
"bao": "OpenBao client",
"kubernetes": "cluster client",
"kubernetes_asyncio": "cluster client",
"psycopg": "direct database connection",
"psycopg2": "direct database connection",
"asyncpg": "direct database connection",
"sqlalchemy": "direct database connection",
"pymysql": "direct database connection",
"redis": "direct datastore connection",
"ldap3": "direct LDAP client (key-cape tooling)",
"python_ldap": "direct LDAP client (key-cape tooling)",
"docker": "container runtime client",
}
# §3.4 rule 1 — no standing credential held in the repository or its
# configuration. Filenames that would be a standing secret, and assignments
# of well-known secret env vars to string literals in src/.
CREDENTIAL_FILENAMES = {
".env",
".env.local",
".env.production",
"credentials.json",
"secrets.yaml",
"secrets.yml",
"id_rsa",
"id_ed25519",
"id_ecdsa",
}
CREDENTIAL_LITERAL = re.compile(
r"""(?x)
\b(?:VAULT_TOKEN|OPENBAO_TOKEN|BAO_TOKEN|AWS_SECRET_ACCESS_KEY|
PRIVATE_KEY|BEGIN\ (?:RSA\ )?PRIVATE\ KEY)
"""
)
SKIP_CREDENTIAL_SCAN_DIRS = {".git", ".venv", "__pycache__", ".pytest_cache", ".ruff_cache"}
def load_declaration() -> dict:
if not DECL.exists():
print(f"FAIL: no declaration at {DECL.relative_to(ROOT)} (§11)", file=sys.stderr)
raise SystemExit(2)
try:
data = yaml.safe_load(DECL.read_text())
except yaml.YAMLError as exc:
print(f"FAIL: {DECL.name} is not parseable: {exc}", file=sys.stderr)
raise SystemExit(2) from exc
for key in ("layer", "repository", "tooling_contacts", "standard_version"):
if key not in data:
print(f"FAIL: {DECL.name} missing required key '{key}' (§11)", file=sys.stderr)
raise SystemExit(2)
if data["layer"] != "staff":
print(f"FAIL: declared layer is '{data['layer']}', expected 'staff'", file=sys.stderr)
raise SystemExit(2)
return data
def imported_modules(path: Path) -> set[str]:
"""Top-level module name of every import in one file."""
try:
tree = ast.parse(path.read_text())
except SyntaxError:
return set()
found: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
found.update(alias.name.split(".")[0] for alias in node.names)
elif isinstance(node, ast.ImportFrom):
if node.level == 0 and node.module:
found.add(node.module.split(".")[0])
return found
def scan() -> list[tuple[Path, str, str]]:
hits: list[tuple[Path, str, str]] = []
for path in sorted(SRC.rglob("*.py")):
for module in sorted(imported_modules(path)):
if module in TOOLING_IMPORTS:
hits.append((path, module, TOOLING_IMPORTS[module]))
return hits
def scan_standing_credentials() -> list[tuple[Path, str]]:
"""§3.4 rule 1: no standing credential in the repository or its config."""
hits: list[tuple[Path, str]] = []
for path in ROOT.rglob("*"):
if not path.is_file():
continue
if any(part in SKIP_CREDENTIAL_SCAN_DIRS for part in path.parts):
continue
if path.name in CREDENTIAL_FILENAMES:
hits.append((path, f"credential-shaped file {path.name}"))
continue
if path.suffix in {".pem", ".key"} and "test" not in path.parts:
hits.append((path, f"key material file {path.name}"))
if SRC.is_dir():
for path in sorted(SRC.rglob("*.py")):
text = path.read_text(encoding="utf-8")
if CREDENTIAL_LITERAL.search(text):
hits.append((path, "standing-credential literal or private-key block"))
return hits
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--report", action="store_true", help="print the declaration summary")
args = parser.parse_args()
decl = load_declaration()
declared = {c.get("id") for c in decl.get("tooling_contacts") or []}
hits = scan()
credential_hits = scan_standing_credentials()
undeclared = [h for h in hits if h[1] not in declared]
rules = decl.get("agent_principal_rules") or {}
checks = decl.get("agent_principal_rule_checks") or {}
if args.report:
print(f"kings-guard — layer {decl['layer']}, standard v{decl['standard_version']}")
print(f" tooling contacts declared: {len(declared)}")
print(f" unowned capabilities (§11 blocked-clean): "
f"{len(decl.get('unowned_capabilities') or [])}")
today = date.today()
for cap in decl.get("unowned_capabilities") or []:
review = cap.get("review")
stale = ""
if review and date.fromisoformat(str(review)) < today:
stale = " [REVIEW OVERDUE]"
print(f" - {cap['id']}: {cap.get('owner_status', '?')}{stale}")
print(" agent-principal rule checks (§3.4):")
for name, meta in checks.items():
form = meta.get("form", "unspecified") if isinstance(meta, dict) else "unspecified"
print(f" - {name}: {form} (claimed={rules.get(name)})")
if undeclared:
print("", file=sys.stderr)
print("FAIL: undeclared Tooling-layer client (§11 undeclared violation)", file=sys.stderr)
for path, module, what in undeclared:
rel = path.relative_to(ROOT)
print(f" {rel}: imports '{module}'{what}", file=sys.stderr)
print("", file=sys.stderr)
print(" A Staff repository may not hold a direct Tooling client (§5).", file=sys.stderr)
print(" Route it through the owning engine, or if none exists, raise an", file=sys.stderr)
print(" engine gap — do not declare this to make the check pass.", file=sys.stderr)
return 1
if credential_hits:
print("", file=sys.stderr)
print("FAIL: standing credential material (§3.4 rule 1)", file=sys.stderr)
for path, what in credential_hits:
try:
rel = path.relative_to(ROOT)
except ValueError:
rel = path
print(f" {rel}: {what}", file=sys.stderr)
print("", file=sys.stderr)
print(" Authority is per task, time-bounded, and attributable to the", file=sys.stderr)
print(" principal acted for. Do not hold a standing secret here.", file=sys.stderr)
return 1
if rules.get("no_standing_credential") is not True:
print("FAIL: layer.yaml does not claim no_standing_credential (§3.4 rule 1)", file=sys.stderr)
return 2
if not args.report:
print(
f"OK: no direct Tooling client in {SRC.relative_to(ROOT)}; "
"no standing credential (§5, §11, §3.4 rule 1)"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())