zone-engine/tools/check_layer_conformance.py

#!/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/.

Per GH-DEC-2026-017 (amendments A9, A11, A12): INTENT.md governs and
layer.yaml is a derived artifact that must be marked derived and name
INTENT.md as its source; layer values are compared against §3's closed
four-token vocabulary after an ASCII case-fold, and neither form is
re-spelled; neither form carries a standard version.

Per GH-DEC-2026-020 (A12 r2), "carries a standard version" is a property of
the content, not of a key name. The declaration is every key and value of the
INTENT.md frontmatter and of layer.yaml; a version-bearing `standard:` or
`companion:` path, a `companion_version`, or any `*_version` key other than
`schema_version` is a pin and fails. Comments and `schema_version` are not
reached. Stance, claims, and evidence-classification maps are not
declarations and are never read here. The version belongs to the run: every
run prints VALIDATED_AGAINST and the scope it ranged over.

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 re
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"

# The standard text this checker was built and validated against. A
# declaration carries no version (A12 r2); the run states it instead
# (GH-DEC-2026-020 §4). Bump when re-validated against a newer accepted text.
VALIDATED_AGAINST = "net-kingdom/canon/standards/security-layer-model_v0.7.md"

# What a run ranges over. Nothing else in the tree is read; in particular no
# stance, claims, or evidence-classification map (GH-DEC-2026-020 §3).
SCOPE = "INTENT.md frontmatter, layer.yaml (A12 across all keys and values), tools/**/*.py"

# A12 r2: keys that are version pins. schema_version is the file format's own
# version and is not reached.
VERSION_KEY = re.compile(r"(^|_)version$", re.IGNORECASE)
NOT_REACHED_KEYS = {"schema_version"}
# A version in a path (`..._v0.7.md`, `...-v1.md`) anywhere in a value.
VERSIONED_PATH = re.compile(r"[_-]v\d+(\.\d+)*(\.[A-Za-z]+)?\b", re.IGNORECASE)
# A bare version token (`v0.2`, `0.7`) in the value of a standard/companion pin.
PIN_KEYS = {"standard", "companion", "framework"}
VERSION_TOKEN = re.compile(r"\bv?\d+\.\d+(\.\d+)*\b", re.IGNORECASE)

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")

# §3 as amended by A9: closed, four tokens, case-insensitive. Stored folded.
LAYER_VOCABULARY = {"taxonomy", "tooling", "engine", "staff"}


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", "derived", "derived_from", "tooling_contacts"):
        if key not in data:
            raise ConformanceError(f"{path.name} missing required key '{key}' (§11)")
    if data["derived"] is not True:
        raise ConformanceError(
            f"{path.name} must be marked 'derived: true' (§11, GH-DEC-2026-017 §1)"
        )
    if data["derived_from"] != "INTENT.md":
        raise ConformanceError(
            f"{path.name} derives from {data['derived_from']!r}; §11 names INTENT.md"
        )
    _reject_versions(path.name, data)
    return dict(data)


def load_intent(path: Path = INTENT) -> dict[str, Any]:
    if not path.exists():
        raise ConformanceError(f"no INTENT.md at {path}")
    front = _frontmatter(path)
    if "layer" not in front:
        raise ConformanceError("INTENT.md frontmatter has no 'layer:' key (§11)")
    _reject_versions("INTENT.md frontmatter", front)
    return front


def declaration_versions(data: Any, where: str = "") -> list[str]:
    """Every standard/companion version pin in a declaration (A12 r2).

    Walks all keys and values. Comments never reach here (YAML drops them);
    schema_version is skipped by name.
    """
    found: list[str] = []
    if isinstance(data, Mapping):
        for key, value in data.items():
            name = str(key)
            here = f"{where}.{name}" if where else name
            if name.lower() in NOT_REACHED_KEYS:
                continue
            if VERSION_KEY.search(name):
                found.append(f"key '{here}' is a version pin")
                continue
            if name.lower() in PIN_KEYS and isinstance(value, (str, int, float)):
                if VERSION_TOKEN.search(str(value)) or VERSIONED_PATH.search(str(value)):
                    found.append(f"'{here}: {value}' carries a version")
                    continue
            found.extend(declaration_versions(value, here))
    elif isinstance(data, list):
        for index, item in enumerate(data):
            found.extend(declaration_versions(item, f"{where}[{index}]"))
    elif isinstance(data, str) and VERSIONED_PATH.search(data):
        found.append(f"'{where}: {data}' is a version-bearing path")
    return found


def _reject_versions(source: str, data: Any) -> None:
    hits = declaration_versions(data)
    if hits:
        raise ConformanceError(
            f"{source} carries a standard version; a layer declaration MUST NOT, in any "
            f"key or value (GH-DEC-2026-017 §5, GH-DEC-2026-020, A12 r2): " + "; ".join(hits)
        )


def _norm(value: Any) -> str:
    """ASCII case-fold (§3 as amended by A9). Only A-Z are folded."""
    text = str(value or "").strip()
    return text.translate(str.maketrans("ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz"))


def check_declaration_agrees(
    decl: Mapping[str, Any], intent: Mapping[str, Any]
) -> list[str]:
    errors: list[str] = []
    for source, value in (("INTENT.md", intent.get("layer")), ("layer.yaml", decl.get("layer"))):
        if _norm(value) not in LAYER_VOCABULARY:
            errors.append(
                f"{source} layer '{value}' is not in §3's closed vocabulary "
                f"{sorted(LAYER_VOCABULARY)} (case-insensitive)"
            )
    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()
    print(f"checker validated against: {VALIDATED_AGAINST}")
    print(f"scope: {SCOPE}")
    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"derived from {decl['derived_from']}"
        )
        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(
            "OK: Engine/PIP declaration agrees with INTENT.md and carries no standard "
            f"version; no Tooling client or decision surface in {TOOLS.relative_to(ROOT)}; "
            f"validated against {VALIDATED_AGAINST}"
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())