ops-warden/scripts/check_layer_conformance.py

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

Read-only. Makes §11's second mechanical check real:

    every direct Tooling client in a Staff repository maps to a declared
    §5.1, §5.2, or §5.3 entry

The failure this catches is a *new* direct OpenBao contact appearing in
src/warden/ without an entry in layer.yaml — an undeclared violation (§11),
which is a finding rather than a tracked gap. It deliberately does NOT check
the review dates: a date-triggered failure breaks the build on a calendar day
with no code change (the reasoning recorded in WARDEN-WP-0033-T05), so
staleness is reported and left to `--report`, never to CI.

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

import argparse
import re
import sys
from datetime import date
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src" / "warden"
DECL = ROOT / "layer.yaml"

VALID_SHAPES = {"5.1", "5.2", "5.3"}

# A direct Tooling contact is an *invocation*, not a mention. Matching the word
# "bao" caught help text, a docstring, and the dev-tier doubles library that
# simulates bao rather than calling it — three false positives on first run.
# So match the two shapes that actually execute:
#   1. an HTTP request built against the OpenBao address
#   2. an argv list whose first element is the bao binary
TOOLING_PATTERNS = (
    # httpx call whose URL is built from the configured OpenBao/Vault address
    re.compile(r"""\bhttpx\.\w+\(|url\s*=\s*f?["'].*\{self\._cfg\.addr\}"""),
    # argv construction: [bao_bin, ...] / ["bao", ...] / [bao_binary, ...]
    re.compile(r"""\[\s*(?:["']bao["']|bao_bin\b|bao_binary\b)\s*,"""),
)

# httpx alone is not a Tooling contact — policy.py calls an Engine and worker.py
# calls the State Hub. A module matching only the httpx pattern counts as a
# contact only if it also references the OpenBao address configuration.
ADDR_HINT = re.compile(r"""_cfg\.addr|VAULT_ADDR|BAO_ADDR""")

# Modules that talk to an Engine or to something outside the §4 catalog. Listed
# in layer.yaml under non_tooling_clients and excluded from the scan with it.
def _excluded(decl: dict) -> set[str]:
    return {e["module"].split("/")[-1] for e in decl.get("non_tooling_clients", [])}


def load_declaration() -> dict:
    if not DECL.exists():
        print(f"MISSING: {DECL} — ops-warden must declare in its own voice (§11)")
        raise SystemExit(2)
    decl = yaml.safe_load(DECL.read_text())
    for key in ("layer", "repository", "standard_version", "tooling_contacts"):
        if key not in decl:
            print(f"MALFORMED: layer.yaml has no {key!r}")
            raise SystemExit(2)
    for c in decl["tooling_contacts"]:
        if c.get("shape") not in VALID_SHAPES:
            print(f"MALFORMED: {c.get('id')} has shape {c.get('shape')!r}, not one of {sorted(VALID_SHAPES)}")
            raise SystemExit(2)
        # §5.3 carries four fields, machine-readably. That is the whole point of
        # the shape; a gap missing them is prose wearing a schema.
        if c["shape"] == "5.3":
            for field in ("capability", "intended_owner", "blocked_on", "review"):
                if not c.get(field):
                    print(f"MALFORMED: §5.3 entry {c['id']!r} is missing {field!r}")
                    raise SystemExit(2)
        # §5.2's test is the supplied-authority property.
        if c["shape"] == "5.2" and c.get("supplied_authority") != "none":
            print(f"MALFORMED: §5.2 conduit {c['id']!r} must declare supplied_authority: none")
            raise SystemExit(2)
    return decl


def scan_modules() -> dict[str, list[int]]:
    """Return {module_name: [line numbers]} for direct Tooling contacts."""
    found: dict[str, list[int]] = {}
    for path in sorted(SRC.rglob("*.py")):
        if path.name.startswith("test_"):
            continue
        text = path.read_text()
        hits: list[int] = []
        for n, line in enumerate(text.splitlines(), 1):
            stripped = line.strip()
            if stripped.startswith("#") or stripped.startswith('"'):
                continue
            if any(p.search(line) for p in TOOLING_PATTERNS):
                hits.append(n)
        if hits:
            # An httpx-only match needs the OpenBao address to be a Tooling
            # contact; otherwise it is an Engine or non-catalogued call.
            argv_shape = any(TOOLING_PATTERNS[1].search(ln) for ln in text.splitlines())
            if argv_shape or ADDR_HINT.search(text):
                found[path.name] = hits
    return found


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--report", action="store_true", help="also print the declaration and gap review dates")
    args = ap.parse_args()

    decl = load_declaration()
    declared = {c["module"].split("/")[-1] for c in decl["tooling_contacts"]}
    excluded = _excluded(decl)
    found = scan_modules()

    undeclared = {m: lines for m, lines in found.items() if m not in declared and m not in excluded}
    # A voluntary declaration has no fixed argv shape to detect (an
    # operator-configured command). Over-declaring is safe; not reporting it as
    # stale keeps the signal meaningful.
    voluntary = {
        c["module"].split("/")[-1]
        for c in decl["tooling_contacts"]
        if c.get("detection") == "voluntary"
    }
    stale_decls = declared - set(found) - voluntary

    if args.report:
        print(f"{decl['repository']} — layer: {decl['layer']} (model v{decl['standard_version']})")
        print(f"declared by {decl['declared_by']}\n")
        for c in decl["tooling_contacts"]:
            line = f"  §{c['shape']}  {c['id']:<28} {c['module']}"
            if c["shape"] == "5.3":
                overdue = str(c["review"]) < date.today().isoformat()
                line += f"  -> {c['intended_owner']}  review {c['review']}"
                if overdue:
                    line += "  [REVIEW OVERDUE]"
            print(line)
        gaps = [c for c in decl["tooling_contacts"] if c["shape"] == "5.3"]
        print(f"\n{len(gaps)} declared gap(s) — tracked non-conformance, not conformance (§11).")

    ok = True
    if undeclared:
        ok = False
        print("\nUNDECLARED TOOLING CONTACT — a finding under §11, not a tracked gap:")
        for m, lines in sorted(undeclared.items()):
            print(f"  src/warden/{m}: line(s) {', '.join(map(str, lines[:6]))}")
        print("\nAdd a §5.1/§5.2/§5.3 entry to layer.yaml, or route it through an engine.")

    if stale_decls:
        print("\nNote: declared but no contact found (module removed or refactored?):")
        for m in sorted(stale_decls):
            print(f"  {m}")

    if ok and not args.report:
        print(f"PASS — {len(found)} module(s) with Tooling contact, all declared.")
    elif ok:
        print("\nPASS — every direct Tooling contact maps to a declared shape.")
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())