user-engine/scripts/check_layer_conformance.py

#!/usr/bin/env python3
"""Check user-engine against the NetKingdom security layer model (§5, §11).

Read-only. user-engine is Engine/PIP and holds no catalogued Tooling client.
PostgreSQL is the modeled-concept store, declared under own_store.

The failure this exists to catch is a convenience: an OpenBao, Vault, LDAP,
or cluster client arriving as one import. That is an undeclared violation.

Exit 0 clean, 1 undeclared contact found, 2 declaration malformed.
"""
from __future__ import annotations

import argparse
import ast
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src" / "user_engine"
DECL = ROOT / "layer.yaml"
sys.path.insert(0, str(ROOT / "src"))

from user_engine.layer_yaml import load_mapping  # noqa: E402

TOOLING_IMPORTS = {
    "hvac": "OpenBao / Vault client",
    "bao": "OpenBao client",
    "kubernetes": "cluster client",
    "kubernetes_asyncio": "cluster client",
    "ldap3": "direct LDAP client (key-cape tooling)",
    "python_ldap": "direct LDAP client (key-cape tooling)",
    "docker": "container runtime client",
    "redis": "direct datastore connection",
}

OWN_STORE_IMPORTS = {
    "psycopg": "PostgreSQL modeled-concept store",
    "psycopg2": "PostgreSQL modeled-concept store",
    "asyncpg": "PostgreSQL modeled-concept store",
    "sqlalchemy": "database client",
    "pymysql": "database client",
}


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 = load_mapping(DECL)
    except (ValueError, OSError) as exc:
        print(f"FAIL: {DECL.name} is not parseable: {exc}", file=sys.stderr)
        raise SystemExit(2) from exc
    for key in ("layer", "role", "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"] != "engine":
        print(
            f"FAIL: declared layer is {data['layer']!r}, expected 'engine'",
            file=sys.stderr,
        )
        raise SystemExit(2)
    if str(data["role"]).lower() != "pip":
        print(f"FAIL: declared role is {data['role']!r}, expected 'pip'", file=sys.stderr)
        raise SystemExit(2)
    if data["tooling_contacts"] not in ([], None):
        print(
            "FAIL: tooling_contacts must be empty; catalogued Tooling clients "
            "are undeclared violations for this Engine",
            file=sys.stderr,
        )
        raise SystemExit(2)
    return data


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 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_own_store() -> 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 OWN_STORE_IMPORTS:
                hits.append((path, module, OWN_STORE_IMPORTS[module]))
    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()
    hits = scan()
    own_store_hits = scan_own_store()
    own_store = decl.get("own_store") or []

    if args.report:
        print(
            f"user-engine — layer {decl['layer']}/{decl['role']}, "
            f"standard v{decl['standard_version']}"
        )
        print("  tooling contacts declared: 0")
        print(f"  own-store declarations: {len(own_store)}")
        print(f"  own-store imports: {len(own_store_hits)}")
        print(f"  pep stance: {decl.get('pep_stance')}")

    if hits:
        print("", file=sys.stderr)
        print(
            "FAIL: undeclared Tooling-layer client (§11 undeclared violation)",
            file=sys.stderr,
        )
        for path, module, what in hits:
            print(f"  {path.relative_to(ROOT)}: imports {module!r} — {what}", file=sys.stderr)
        return 1

    if own_store_hits and not own_store:
        print(
            "FAIL: modeled-concept store import with no own_store declaration",
            file=sys.stderr,
        )
        for path, module, what in own_store_hits:
            print(f"  {path.relative_to(ROOT)}: imports {module!r} — {what}", file=sys.stderr)
        return 1

    if not args.report:
        print(
            f"OK: no catalogued Tooling client in {SRC.relative_to(ROOT)} "
            "(Engine/PIP, §11)"
        )
    return 0


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