llm-ingestion-pipeline-security/docs/catalog-census.py

"""catalog-number census — every place an OWASP LLM Top 10 number is used, and what it means.

OWASP published the 2026 edition of the Top 10 for LLM Applications with a new
order: only LLM01 and LLM02 kept their meaning (`docs/PLAN.md` § T.5). A bare
`LLM05` written against 2025 now reads as Data and Model Poisoning, and the
`owasp` field carries that misreading to every consumer of a finding. Row 2 of the
2026 done-criterion (`docs/PLAN.md` § T.2) counts those places; this script is the
count, so the number is reproducible instead of a grep someone remembers.

WHAT IT CLASSIFIES. Every `LLMnn` token in a tracked file, with slash lists
expanded (`LLM01/02/04` is three numbers), lands in exactly one class:

- ``old`` — used with its 2025 meaning and no year. Maps mechanically (§ T.5).
- ``wrong`` — used with a meaning no edition gave the number, or one too doubtful
  to map mechanically. Needs a decision; each rule below carries its reason.
- ``same`` — LLM01 / LLM02 used with the meaning both editions share. Listed,
  not counted: the number is right, only the year is missing (session 3).
- ``current`` — carries `:2026`, the newest edition. Listed, not counted.
- ``history`` — CHANGELOG, the v1 plan and published reviews. Snapshots of what
  was true when written; they are never corrected (STATE conventions), so they
  sit outside the denominator by definition (§ T.2 row 2).
- ``migration-plan`` — the 2026 section of `docs/PLAN.md`. It names both editions
  by construction (its mapping table IS the old numbers); counting it would make
  row 2 unreachable by the very text that plans the move.
- ``self`` — this script and the session-1 artefacts in `docs/threat-2026/` (the
  catalog register and the frozen lists), which name both editions by construction.

The DENOMINATOR is ``old + wrong``.

WHY THE `wrong` RULES MATCH ON CONTEXT, NOT LINE NUMBERS. A rule names a file, a
substring of the token's line or of up to three lines before it (stopping at a
line that carries another token, so a rule cannot reach a neighbour), the number, and how
many sites it must match. A rule that matches a different count than it states
raises: an edit that moves or removes a judged site must re-judge it, not let
the rule go stale and the site silently fall back to ``old``.

THE FROZEN DENOMINATOR is measured at a fixed revision, read from git objects,
not the working tree, so it is reproducible after every later edit:

    python3 docs/catalog-census.py --rev 6702d7a --tsv docs/threat-2026/census-6702d7a.tsv

The sha256 printed last is over the listing (every class, every site); the
listing and its hash are frozen in `docs/threat-2026/` and `docs/PLAN.md`.
Without ``--rev`` it measures the tracked files of the working tree.
"""
from __future__ import annotations

import argparse
import hashlib
import re
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent

TITLES_2025 = {
    "01": "Prompt Injection",
    "02": "Sensitive Information Disclosure",
    "03": "Supply Chain",
    "04": "Data and Model Poisoning",
    "05": "Improper Output Handling",
    "06": "Excessive Agency",
    "07": "System Prompt Leakage",
    "08": "Vector and Embedding Weaknesses",
    "09": "Misinformation",
    "10": "Unbounded Consumption",
}

# 2025 number -> 2026 number, same risk. Verified against the publisher in
# session 1; the source and check date live in docs/threat-2026/catalogs.json.
MAP_2025_TO_2026 = {
    "01": "01", "02": "02", "03": "04", "04": "05", "05": "10",
    "06": "03", "07": "08", "08": "09", "09": "07", "10": "06",
}

HISTORY_FILES = {
    "CHANGELOG.md": "released entries are snapshots",
    "docs/PLAN-v1.md": "the v1 plan",
    "docs/review-2026-07.md": "published review",
    "docs/GATE-G-v1.md": "published gate record",
    "docs/2026-09-07-limitations-44-maaling.md": "published measurement",
}

SELF_PREFIXES = ("docs/catalog-census.py", "docs/threat-2026/", "tests/test_catalog_numbers.py")

# The newest edition. `tests/test_catalog_numbers.py` asserts it equals the
# register's `latest`, so the two cannot drift apart.
CURRENT_YEAR = "2026"

# `docs/PLAN.md` is split by its top-level headings.
PLAN = "docs/PLAN.md"
PLAN_SECTIONS = (
    ("# Trusselbildet 2026", "migration-plan"),
    ("# Implementation Plan", "history"),
)

_FAIL_SECURE = ("Fail-secure disposition is not Unbounded Consumption in any edition; "
                "it is the guard's own failure semantics")
_PARSER_SURFACE = ("A frontmatter shape the parser refuses to interpret is parser-surface "
                   "reduction, not resource consumption; no edition defines it")
_NOT_MODEL_OUTPUT = ("Validates received bundle metadata, not model output; "
                     "Improper Output Handling is about LLM output passed downstream")
_FACTUAL_POISONING = ("Factual poisoning of the ingested store (PoisonedRAG) is Data and "
                      "Model Poisoning or Vector and Embedding Weaknesses, not Misinformation")

# (path, context substring, number, expected sites, reason)
WRONG_RULES = (
    ("README.md", "LLM10 · fail-secure", "10", 1, _FAIL_SECURE),
    ("src/llm_ingestion_guard/coverage.py", "_fail_closed_probe, owasp=", "10", 1,
     _FAIL_SECURE),
    ("src/llm_ingestion_guard/coverage.py", "nested: x", "10", 1, _PARSER_SURFACE),
    ("src/llm_ingestion_guard/coverage.py", "desc: |", "10", 1, _PARSER_SURFACE),
    ("src/llm_ingestion_guard/coverage.py", "flow sequence, quoted element", "10", 1,
     _PARSER_SURFACE),
    ("src/llm_ingestion_guard/coverage.py", "mapping key off the allowlist", "10", 1,
     _PARSER_SURFACE),
    ("src/llm_ingestion_guard/coverage.py", "validate_resource_url(", "05", 3,
     _NOT_MODEL_OUTPUT),
    ("src/llm_ingestion_guard/coverage.py", "T5a cross-link dangerous scheme", "05", 1,
     _NOT_MODEL_OUTPUT),
    ("src/llm_ingestion_guard/coverage.py", "_semantic_poisoning_gap", "09", 1,
     _FACTUAL_POISONING),
    ("tests/test_coverage_matrix.py", '"LLM06", "LLM09"', "09", 1,
     "Asserts the matrix carries the semantic-poisoning row's LLM09; follows that row"),
    ("src/llm_ingestion_guard/coverage.py", "def _raise_case(", "01", 1,
     "Default tags the OKF path-traversal, absolute-path, reserved-name, non-.md and "
     "bundle-escape rows as Prompt Injection; none of them is"),
)

_TOKEN = re.compile(r"\bLLM(\d\d)((?:/(?:LLM)?\d\d)*)(?::(\d{4}))?(?!\d)")
_VALID = set(TITLES_2025)


def _git(*args: str) -> str:
    return subprocess.run(("git", "-C", str(ROOT)) + args, check=True,
                          capture_output=True, text=True).stdout


def _files(rev: str | None) -> list[tuple[str, str]]:
    names = (_git("ls-tree", "-r", "--name-only", rev) if rev else _git("ls-files")).split("\n")
    out = []
    for name in sorted(n for n in names if n):
        if name.startswith("dist/") or ".local." in name:
            continue
        try:
            text = (_git("show", f"{rev}:{name}") if rev
                    else (ROOT / name).read_text(encoding="utf-8"))
        except (UnicodeDecodeError, subprocess.CalledProcessError, FileNotFoundError):
            continue
        out.append((name, text))
    return out


def _plan_class(lines: list[str]) -> list[str | None]:
    """Per-line section class for docs/PLAN.md (None = live text)."""
    current: str | None = None
    out = []
    for line in lines:
        if line.startswith("# "):
            current = next((cls for head, cls in PLAN_SECTIONS if line.startswith(head)), None)
        out.append(current)
    return out


def census(rev: str | None) -> list[tuple[str, ...]]:
    rows = []
    rule_hits = [0] * len(WRONG_RULES)
    for path, text in _files(rev):
        lines = text.split("\n")
        sections = _plan_class(lines) if path == PLAN else [None] * len(lines)
        for i, line in enumerate(lines):
            for m in _TOKEN.finditer(line):
                numbers = [m.group(1)] + re.findall(r"\d\d", m.group(2))
                year = m.group(3) or ""
                for nn in numbers:
                    if nn not in _VALID:
                        continue
                    reason = ""
                    if path.startswith(SELF_PREFIXES):
                        cls = "self"
                    elif path in HISTORY_FILES:
                        cls, reason = "history", HISTORY_FILES[path]
                    elif sections[i]:
                        cls = sections[i]
                    elif year == CURRENT_YEAR:
                        cls = "current"
                    else:
                        cls = "same" if nn in ("01", "02") else "old"
                        start = i
                        while start > max(0, i - 3) and not _TOKEN.search(lines[start - 1]):
                            start -= 1
                        context = "\n".join(lines[start:i + 1])
                        for k, (rpath, sub, rnn, _, rreason) in enumerate(WRONG_RULES):
                            if rpath == path and rnn == nn and sub in context:
                                cls, reason = "wrong", rreason
                                rule_hits[k] += 1
                                break
                    rows.append((cls, path, str(i + 1), f"LLM{nn}", year,
                                 TITLES_2025[nn], f"LLM{MAP_2025_TO_2026[nn]}:2026", reason))
    stale = [(r[0], r[1], r[3], hits) for r, hits in zip(WRONG_RULES, rule_hits) if hits != r[3]]
    # The rules judge the sites of the FROZEN denominator, so they are held to
    # their counts there. In a later tree a judged site is meant to be fixed --
    # it then carries the year and never reaches a rule -- and a site a rule no
    # longer reaches still counts as `old` or `same`, which the permanent test
    # refuses as well. A stale rule there changes a label, never the total.
    if stale and rev is not None:
        raise SystemExit(f"stale wrong-rules (path, substring, expected, matched): {stale}")
    return rows


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--rev", help="measure this git revision instead of the working tree")
    ap.add_argument("--tsv", help="also write the full listing to this path")
    args = ap.parse_args(argv)
    rows = census(args.rev)
    header = "class\tpath\tline\tnumber\tyear\tmeaning_2025\tsame_risk_2026\treason"
    body = "\n".join([header] + ["\t".join(r) for r in rows]) + "\n"
    if args.tsv:
        Path(args.tsv).write_text(body, encoding="utf-8")
    counts: dict[str, int] = {}
    for r in rows:
        counts[r[0]] = counts.get(r[0], 0) + 1
    files = sorted({r[1] for r in rows if r[0] in ("old", "wrong")})
    print(f"revision: {args.rev or 'working tree'}")
    for cls in ("old", "wrong", "same", "current", "history", "migration-plan", "self"):
        print(f"  {cls:<15} {counts.get(cls, 0)}")
    print(f"denominator (old + wrong): {counts.get('old', 0) + counts.get('wrong', 0)}"
          f" in {len(files)} files")
    print(f"sha256: {hashlib.sha256(body.encode('utf-8')).hexdigest()}")
    return 0


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