294 lines
11 KiB
Python
294 lines
11 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Check zone-engine against the NetKingdom security layer model (§5, §6, §11).
|
||
|
|
|
||
|
|
Read-only. This repository's position is:
|
||
|
|
|
||
|
|
Engine / PIP for zone identity and membership, offline reference
|
||
|
|
conformance, no Tooling-layer client, no HTTP authorization decision
|
||
|
|
surface, no PEP.
|
||
|
|
|
||
|
|
§11 requires a machine-readable declaration that agrees with INTENT.md
|
||
|
|
frontmatter. This script is what makes that claim checkable: it fails if
|
||
|
|
layer.yaml is missing, disagrees with INTENT.md, or if a Tooling client or
|
||
|
|
decision surface appears under tools/.
|
||
|
|
|
||
|
|
The failure it exists to catch is a *convenience* — a live lookup, an
|
||
|
|
OpenBao client, or an /authorize helper "just for this consumer". That is
|
||
|
|
this repository's original §7 falsifier, now statute §6.
|
||
|
|
|
||
|
|
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 or decision surface, 2 declaration
|
||
|
|
malformed or in disagreement with INTENT.md.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import ast
|
||
|
|
import sys
|
||
|
|
from datetime import date
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Iterable, Mapping
|
||
|
|
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
TOOLS = ROOT / "tools"
|
||
|
|
DECL = ROOT / "layer.yaml"
|
||
|
|
INTENT = ROOT / "INTENT.md"
|
||
|
|
|
||
|
|
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",
|
||
|
|
}
|
||
|
|
|
||
|
|
HTTP_SURFACE_IMPORTS = {
|
||
|
|
"flask": "HTTP framework",
|
||
|
|
"fastapi": "HTTP framework",
|
||
|
|
"starlette": "HTTP framework",
|
||
|
|
"aiohttp": "HTTP framework",
|
||
|
|
"tornado": "HTTP framework",
|
||
|
|
"bottle": "HTTP framework",
|
||
|
|
"quart": "HTTP framework",
|
||
|
|
"sanic": "HTTP framework",
|
||
|
|
"django": "HTTP framework",
|
||
|
|
"gunicorn": "HTTP server",
|
||
|
|
"uvicorn": "HTTP server",
|
||
|
|
"hypercorn": "HTTP server",
|
||
|
|
"waitress": "HTTP server",
|
||
|
|
}
|
||
|
|
|
||
|
|
DECISION_PATHS = ("/authorize", "/v1/check", "/v1/authorize")
|
||
|
|
|
||
|
|
|
||
|
|
class ConformanceError(ValueError):
|
||
|
|
"""Declaration is missing, unparseable, or disagrees with INTENT.md."""
|
||
|
|
|
||
|
|
|
||
|
|
def _frontmatter(path: Path) -> dict[str, Any]:
|
||
|
|
text = path.read_text()
|
||
|
|
if not text.startswith("---\n") or "\n---\n" not in text[4:]:
|
||
|
|
raise ConformanceError(f"{path.name} requires YAML frontmatter")
|
||
|
|
raw = text.split("\n---\n", 1)[0][4:]
|
||
|
|
value = yaml.safe_load(raw) or {}
|
||
|
|
if not isinstance(value, Mapping):
|
||
|
|
raise ConformanceError(f"{path.name} frontmatter must be a mapping")
|
||
|
|
return dict(value)
|
||
|
|
|
||
|
|
|
||
|
|
def load_declaration(path: Path = DECL) -> dict[str, Any]:
|
||
|
|
if not path.exists():
|
||
|
|
raise ConformanceError(f"no declaration at {path.name} (§11)")
|
||
|
|
try:
|
||
|
|
data = yaml.safe_load(path.read_text())
|
||
|
|
except yaml.YAMLError as exc:
|
||
|
|
raise ConformanceError(f"{path.name} is not parseable: {exc}") from exc
|
||
|
|
if not isinstance(data, Mapping):
|
||
|
|
raise ConformanceError(f"{path.name} must be a mapping")
|
||
|
|
for key in ("layer", "role", "repository", "tooling_contacts", "standard_version"):
|
||
|
|
if key not in data:
|
||
|
|
raise ConformanceError(f"{path.name} missing required key '{key}' (§11)")
|
||
|
|
return dict(data)
|
||
|
|
|
||
|
|
|
||
|
|
def load_intent(path: Path = INTENT) -> dict[str, Any]:
|
||
|
|
if not path.exists():
|
||
|
|
raise ConformanceError(f"no INTENT.md at {path}")
|
||
|
|
return _frontmatter(path)
|
||
|
|
|
||
|
|
|
||
|
|
def _norm(value: Any) -> str:
|
||
|
|
return str(value or "").strip().lower()
|
||
|
|
|
||
|
|
|
||
|
|
def check_declaration_agrees(
|
||
|
|
decl: Mapping[str, Any], intent: Mapping[str, Any]
|
||
|
|
) -> list[str]:
|
||
|
|
errors: list[str] = []
|
||
|
|
if _norm(decl.get("layer")) != "engine":
|
||
|
|
errors.append(f"declared layer is '{decl.get('layer')}', expected 'engine'")
|
||
|
|
if _norm(decl.get("role")) != "pip":
|
||
|
|
errors.append(f"declared role is '{decl.get('role')}', expected 'pip'")
|
||
|
|
if _norm(intent.get("layer")) != _norm(decl.get("layer")):
|
||
|
|
errors.append(
|
||
|
|
f"INTENT.md layer '{intent.get('layer')}' disagrees with "
|
||
|
|
f"layer.yaml '{decl.get('layer')}'"
|
||
|
|
)
|
||
|
|
if _norm(intent.get("role")) != _norm(decl.get("role")):
|
||
|
|
errors.append(
|
||
|
|
f"INTENT.md role '{intent.get('role')}' disagrees with "
|
||
|
|
f"layer.yaml '{decl.get('role')}'"
|
||
|
|
)
|
||
|
|
if decl.get("repository") != "zone-engine":
|
||
|
|
errors.append(f"declared repository is '{decl.get('repository')}'")
|
||
|
|
if decl.get("pep_stance"):
|
||
|
|
errors.append("pep_stance is set; this repository is not PEP-shaped")
|
||
|
|
contacts = decl.get("tooling_contacts")
|
||
|
|
if contacts not in ([], None):
|
||
|
|
errors.append(
|
||
|
|
"tooling_contacts is not empty; a new contact needs a §10 cut, "
|
||
|
|
"not a quiet row"
|
||
|
|
)
|
||
|
|
return errors
|
||
|
|
|
||
|
|
|
||
|
|
def imported_modules(path: Path) -> set[str]:
|
||
|
|
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 string_literals(path: Path) -> list[str]:
|
||
|
|
try:
|
||
|
|
tree = ast.parse(path.read_text())
|
||
|
|
except SyntaxError:
|
||
|
|
return []
|
||
|
|
found: list[str] = []
|
||
|
|
for node in ast.walk(tree):
|
||
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
||
|
|
found.append(node.value)
|
||
|
|
return found
|
||
|
|
|
||
|
|
|
||
|
|
def scan_tooling(directory: Path) -> list[tuple[Path, str, str]]:
|
||
|
|
hits: list[tuple[Path, str, str]] = []
|
||
|
|
for path in sorted(directory.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_decision_surfaces(directory: Path) -> list[tuple[Path, str, str]]:
|
||
|
|
hits: list[tuple[Path, str, str]] = []
|
||
|
|
for path in sorted(directory.rglob("*.py")):
|
||
|
|
if path.resolve() == Path(__file__).resolve():
|
||
|
|
continue
|
||
|
|
for module in sorted(imported_modules(path)):
|
||
|
|
if module in HTTP_SURFACE_IMPORTS:
|
||
|
|
hits.append((path, module, HTTP_SURFACE_IMPORTS[module]))
|
||
|
|
if module == "http":
|
||
|
|
# http.server is the stdlib decision-adjacent live surface.
|
||
|
|
text = path.read_text()
|
||
|
|
if "http.server" in text:
|
||
|
|
hits.append((path, "http.server", "stdlib HTTP server"))
|
||
|
|
for literal in string_literals(path):
|
||
|
|
for needle in DECISION_PATHS:
|
||
|
|
if needle in literal:
|
||
|
|
hits.append((path, literal, f"authorization path {needle}"))
|
||
|
|
return hits
|
||
|
|
|
||
|
|
|
||
|
|
def overdue_reviews(decl: Mapping[str, Any], today: date | None = None) -> list[str]:
|
||
|
|
today = today or date.today()
|
||
|
|
notes: list[str] = []
|
||
|
|
for cap in decl.get("unowned_capabilities") or []:
|
||
|
|
review = cap.get("review")
|
||
|
|
if not review:
|
||
|
|
continue
|
||
|
|
try:
|
||
|
|
due = date.fromisoformat(str(review))
|
||
|
|
except ValueError:
|
||
|
|
notes.append(f"{cap.get('id')}: unparseable review date {review!r}")
|
||
|
|
continue
|
||
|
|
if due < today:
|
||
|
|
notes.append(f"{cap.get('id')}: review overdue ({review})")
|
||
|
|
return notes
|
||
|
|
|
||
|
|
|
||
|
|
def evaluate(
|
||
|
|
*,
|
||
|
|
root: Path | None = None,
|
||
|
|
tools: Path | None = None,
|
||
|
|
decl_path: Path | None = None,
|
||
|
|
intent_path: Path | None = None,
|
||
|
|
) -> tuple[int, list[str], list[str]]:
|
||
|
|
"""Return (exit_code, errors, reports). Exit 2 for declaration, 1 for tree."""
|
||
|
|
root = root or ROOT
|
||
|
|
tools = tools or (root / "tools")
|
||
|
|
decl_path = decl_path or (root / "layer.yaml")
|
||
|
|
intent_path = intent_path or (root / "INTENT.md")
|
||
|
|
try:
|
||
|
|
decl = load_declaration(decl_path)
|
||
|
|
intent = load_intent(intent_path)
|
||
|
|
except ConformanceError as exc:
|
||
|
|
return 2, [str(exc)], []
|
||
|
|
errors = check_declaration_agrees(decl, intent)
|
||
|
|
if errors:
|
||
|
|
return 2, errors, []
|
||
|
|
reports = overdue_reviews(decl)
|
||
|
|
tree_errors: list[str] = []
|
||
|
|
for path, module, what in scan_tooling(tools):
|
||
|
|
rel = path.relative_to(root) if path.is_relative_to(root) else path
|
||
|
|
tree_errors.append(f"{rel}: imports '{module}' — {what}")
|
||
|
|
for path, token, what in scan_decision_surfaces(tools):
|
||
|
|
rel = path.relative_to(root) if path.is_relative_to(root) else path
|
||
|
|
tree_errors.append(f"{rel}: {what} ({token})")
|
||
|
|
if tree_errors:
|
||
|
|
return 1, tree_errors, reports
|
||
|
|
return 0, [], reports
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv: Iterable[str] | None = None) -> int:
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
parser.add_argument("--report", action="store_true", help="print the declaration summary")
|
||
|
|
args = parser.parse_args(list(argv) if argv is not None else None)
|
||
|
|
|
||
|
|
code, errors, reports = evaluate()
|
||
|
|
if args.report:
|
||
|
|
try:
|
||
|
|
decl = load_declaration()
|
||
|
|
except ConformanceError as exc:
|
||
|
|
print(f"FAIL: {exc}", file=sys.stderr)
|
||
|
|
return 2
|
||
|
|
print(
|
||
|
|
f"zone-engine — layer {decl['layer']}, role {decl['role']}, "
|
||
|
|
f"standard v{decl['standard_version']}"
|
||
|
|
)
|
||
|
|
print(f" tooling contacts declared: {len(decl.get('tooling_contacts') or [])}")
|
||
|
|
print(f" pep_stance: {decl.get('pep_stance')!r}")
|
||
|
|
for note in reports:
|
||
|
|
print(f" review: {note}")
|
||
|
|
|
||
|
|
if code == 2:
|
||
|
|
print("FAIL: layer declaration malformed or disagrees with INTENT.md (§11)", file=sys.stderr)
|
||
|
|
for item in errors:
|
||
|
|
print(f" {item}", file=sys.stderr)
|
||
|
|
return 2
|
||
|
|
if code == 1:
|
||
|
|
print("FAIL: undeclared Tooling client or HTTP decision surface (§6, §11)", file=sys.stderr)
|
||
|
|
for item in errors:
|
||
|
|
print(f" {item}", file=sys.stderr)
|
||
|
|
print("", file=sys.stderr)
|
||
|
|
print(" A live API, Tooling client, or /authorize helper is a layer change.", file=sys.stderr)
|
||
|
|
print(" Do not declare it to make the check pass.", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
if not args.report:
|
||
|
|
print(f"OK: Engine/PIP declaration agrees with INTENT.md; no Tooling client or decision surface in {TOOLS.relative_to(ROOT)}")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|