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

"""raw-HTML census — which branch of `is_active_tag` fires, and what narrowing it costs.

`docs/fp-sweep.py` answers *how often* the upload door costs a human. This answers
*why*, for the one detector that drives most of it, and *what a proposed narrowing
would actually buy* — end to end, as a change in `screen_output` disposition, not as
a count of regex hits.

WHY THIS EXISTS. `docs/LIMITATIONS.md` once attributed `active:raw-html` in 52 of
vendor-harvest's 98 non-WARN documents to the relative-URL-attribute over-reach.
Re-measured, that over-reach frees **one** document there. The claim was reasoning,
not measurement, and it stood in a published file for three releases. This script is
the measurement, so the next person changing that detector argues with numbers.

TWO METHOD TRAPS IT EXISTS TO AVOID:

- **Over-reach classes co-occur.** In one corpus, narrowing the URL-attribute branch
  alone frees 3 documents and taking `<base>` off the name branch alone frees 13 —
  but both together free 25. A document blocked by two classes is freed by neither
  alone. Measure candidates you intend to ship *together*, never one at a time.
- **Two of the three published populations are LIVING corpora**, re-harvested by
  their owning repo. A before/after split across two sessions mixes the narrowing's
  effect with corpus drift. Every candidate here runs against the same corpus state
  in one process, and `base` is re-measured rather than quoted from the doc.

The candidates are applied by replacing `active_content.is_active_tag` in-process.
That mirrors a real edit for the *scanner* path only: `neutralize` imports the
symbol by name, so a real edit would also change the mutator, which this script does
not simulate. See the raw-HTML bullets in `docs/LIMITATIONS.md`.

USAGE — corpus roots are arguments, never hardcoded; the corpora live in private
consumer repos and their paths must not reach a public mirror:

    python docs/rawhtml-census.py LABEL=/path/to/corpus [LABEL=/path ...]
                                  [--ext=.md,.txt] [--include=/subtree/]
"""
from __future__ import annotations

import re
import sys
from collections import Counter
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))

from llm_ingestion_guard import (  # noqa: E402
    PRESET_USER_UPLOAD, Disposition, screen_output,
)
from llm_ingestion_guard import active_content as ac  # noqa: E402

BENIGN = Disposition.WARN

# Attribute parser — only needed to read a URL attribute's VALUE, which
# `_URL_ATTR_RE` (a presence test) deliberately does not capture.
_ATTR_KV_RE = re.compile(
    r"""\b(?P<k>[A-Za-z_:][\w:.\-]*)\s*=\s*(?P<v>"[^"]*"|'[^']*'|[^\s>]+)"""
)
_URL_ATTR_NAMES = frozenset({
    "src", "href", "xlink:href", "srcset", "data", "poster", "formaction",
    "action", "background", "cite", "codebase", "longdesc",
})


def has_external_url_attr(attrs: str) -> bool:
    """True if any URL-bearing attribute points at an attacker-reachable target."""
    for m in _ATTR_KV_RE.finditer(attrs):
        if m.group("k").lower() not in _URL_ATTR_NAMES:
            continue
        value = m.group("v")
        if value[:1] in "\"'":
            value = value[1:-1]
        if ac._has_external_target(value.strip()):
            return True
    return False


def _variant(*, drop: frozenset[str] = frozenset(), external_only: bool = False):
    """Build an `is_active_tag` replacement: names minus ``drop``, URL branch gated."""
    keep = frozenset(ac._ACTIVE_TAGS - drop)

    def is_active_tag(name: str, attrs: str) -> bool:
        if name.lower() in keep or ac._EVENT_ATTR_RE.search(attrs):
            return True
        if not ac._URL_ATTR_RE.search(attrs):
            return False
        return has_external_url_attr(attrs) if external_only else True

    return is_active_tag


CANDIDATES = [
    ("base", _variant()),
    # The URL-attribute branch requires an EXTERNAL target — the rule the markdown
    # paths already apply. A relative `href` reaches no attacker-controlled host.
    ("A: url-attr external-only", _variant(external_only=True)),
    # `<base>` off the NAME set. HTML's `<base>` has its whole affordance in `href`,
    # which the URL-attribute branch still catches; APIM policy XML's `<base />` is
    # attribute-less and has no affordance in any renderer.
    ("base-url: <base> needs a URL", _variant(drop=frozenset({"base"}))),
    ("A + base-url (both)",
     _variant(drop=frozenset({"base"}), external_only=True)),
    # The CEILING: no narrowing can free more than switching the detector off.
    ("NONE (ceiling)", lambda name, attrs: False),
]


def documents(root: Path, exts: tuple[str, ...], include: str) -> list[Path]:
    files = []
    for p in sorted(root.rglob("*")):
        if not p.is_file() or p.suffix not in exts:
            continue
        rel = p.relative_to(root)
        if any(part.startswith(".") for part in rel.parts):
            continue
        if include and include not in f"/{rel}":
            continue
        files.append(p)
    return files


def masked_text(text: str) -> str:
    """Reproduce `scan_active_content`'s masking up to the raw-HTML pass.

    Order matters: an image is not also a link, and a construct already consumed
    by an earlier pass must not be re-counted as a tag.
    """
    masked = text[: ac.MAX_SCAN_CHARS]
    for pattern in (ac.MD_IMAGE_RE, ac.MD_LINK_RE, ac.MD_REFDEF_RE, ac.AUTOLINK_RE):
        masked = pattern.sub(lambda m: " " * len(m.group(0)), masked)
    return masked


def branch_census(texts: list[str]) -> tuple[Counter, Counter, Counter]:
    """Count active-tag occurrences by the branch that fires first."""
    names: Counter[str] = Counter()
    relative: Counter[str] = Counter()
    external: Counter[str] = Counter()
    for text in texts:
        for m in ac.HTML_TAG_RE.finditer(masked_text(text)):
            name, attrs = m.group("name"), m.group("attrs") or ""
            if name.lower() in ac._ACTIVE_TAGS:
                names[name.lower()] += 1
            elif ac._EVENT_ATTR_RE.search(attrs):
                names["(on*= handler)"] += 1
            elif ac._URL_ATTR_RE.search(attrs):
                (external if has_external_url_attr(attrs) else relative)[name] += 1
    return names, relative, external


def main() -> None:
    exts: tuple[str, ...] = (".md", ".txt")
    include = ""
    specs = []
    for arg in sys.argv[1:]:
        if arg.startswith("--ext="):
            exts = tuple(e if e.startswith(".") else f".{e}"
                         for e in arg.split("=", 1)[1].split(","))
        elif arg.startswith("--include="):
            include = arg.split("=", 1)[1]
        else:
            specs.append(arg)

    if not specs:
        print(__doc__)
        raise SystemExit(2)

    original = ac.is_active_tag
    try:
        for spec in specs:
            if "=" not in spec:
                raise SystemExit(f"expected LABEL=PATH, got {spec!r}")
            label, _, path = spec.partition("=")
            root = Path(path).expanduser()
            if not root.is_dir():
                raise SystemExit(f"{label}: {root} is not a directory")

            texts = [t for t in (p.read_text(encoding="utf-8", errors="replace")
                                 for p in documents(root, exts, include)) if t.strip()]
            n = len(texts)
            print(f"\n## {label} — {n} documents", flush=True)

            baseline = None
            for name, fn in CANDIDATES:
                ac.is_active_tag = fn
                non_warn = sum(
                    1 for t in texts
                    if screen_output(t, PRESET_USER_UPLOAD).disposition is not BENIGN
                )
                if baseline is None:
                    baseline = non_warn
                    print(f"   {name:>28}: {non_warn:4d} ({non_warn / n:5.1%})", flush=True)
                else:
                    print(f"   {name:>28}: {non_warn:4d} ({non_warn / n:5.1%})"
                          f"   frees {baseline - non_warn}", flush=True)
            ac.is_active_tag = original

            names, relative, external = branch_census(texts)
            print(f"   name branch      : {dict(names.most_common(8))}")
            print(f"   url-attr relative: {dict(relative.most_common(8))}  <- A frees these")
            print(f"   url-attr external: {dict(external.most_common(8))}  <- A keeps these")
    finally:
        ac.is_active_tag = original

    print("\n---")
    print("Candidates are measured TOGETHER as well as alone: over-reach classes\n"
          "co-occur, and a document blocked by two of them is freed by neither\n"
          "alone. Populations are never summed — each has its own denominator.")


if __name__ == "__main__":
    main()