1
0
Fork 0

docs(redos): sweep every regex surface, not just the lexicon

0.3.3 swept the 83 lexicon patterns arm-by-arm and left the other tables on
0.3.2's hand-written rows -- the class of sweep that misses arms. Generalise the
generator over all eleven regex-bearing modules and make the collector mechanical
on both axes: walk each module namespace for compiled patterns (a pattern added
later is swept without anyone listing it) and derive each one's call mode by
grepping the module source, since `.sub()`/`.finditer()` visit every start
position where `.match()` cannot. Patterns reached only through a helper
parameter get the worst mode, marked `*`, so the fallback can over-measure but
never miss.

Sweeps 137 patterns across 11 tables where 0.3.3 covered 83 in one.
This commit is contained in:
Kjell Tore Guttormsen 2026-08-01 19:57:59 +02:00
commit abbfe5f0fd

View file

@ -19,13 +19,26 @@ exponent is 0.96-1.05, i.e. LINEAR (0.2s at the 1M cap). Machine load inflates
the small measurements. Always re-measure a flagged arm across 4+ doublings and the small measurements. Always re-measure a flagged arm across 4+ doublings and
read the exponent before concluding anything; x4 on doubling is quadratic, x2 is read the exponent before concluding anything; x4 on doubling is quadratic, x2 is
linear. The two real findings above sat at 297s and 55s, not near the floor. linear. The two real findings above sat at 297s and 55s, not near the floor.
TABLES (`python docs/redos-sweep.py [table ...]`, default: all). The lexicon is
one of six regex surfaces; 0.3.2 swept the other five with HAND-WRITTEN rows,
which is the class of sweep this script exists because it misses arms. The
collector is therefore MECHANICAL on both axes -- it walks each module's
namespace for compiled patterns (so a pattern added later is swept without
anyone remembering to list it) and derives each one's CALL MODE by grepping the
module source for `NAME.<method>(`. Mode matters: `.match()`/`.fullmatch()` are
anchored at position 0 and cannot pay the per-start rescan cost that makes a run
quadratic, so timing them with `search` would manufacture unreachable flags,
while `.sub()`/`.finditer()` scan every position and must be timed as such.
""" """
from __future__ import annotations from __future__ import annotations
import importlib
import json import json
import re import re
import sys import sys
import time import time
from dataclasses import dataclass
from pathlib import Path from pathlib import Path
SRC = Path(__file__).resolve().parent.parent / "src" / "llm_ingestion_guard" SRC = Path(__file__).resolve().parent.parent / "src" / "llm_ingestion_guard"
@ -120,35 +133,191 @@ def build(unit: str, n: int) -> str:
return (unit * (n // len(unit) + 1))[:n] return (unit * (n // len(unit) + 1))[:n]
def t(rx: re.Pattern[str], text: str) -> float: def t(rx: re.Pattern[str], text: str, mode: str = "search") -> float:
"""Time one scan of ``text`` in the mode the production code actually uses."""
mode = mode.rstrip("*")
start = time.monotonic() start = time.monotonic()
rx.search(text) if mode == "finditer":
for _ in rx.finditer(text):
pass
elif mode == "sub":
rx.sub("", text)
elif mode == "match":
rx.match(text)
elif mode == "fullmatch":
rx.fullmatch(text)
else:
rx.search(text)
return time.monotonic() - start return time.monotonic() - start
def main() -> None: # --- targets ----------------------------------------------------------------
@dataclass(frozen=True)
class Target:
table: str
id: str
regex: re.Pattern[str]
src: str
mode: str
_MODE_ORDER = ("sub", "finditer", "search", "match", "fullmatch")
_CALL_RE = r"\b%s\b(?:\[[^\]]*\]|\.\w+)*\s*\.\s*(search|match|fullmatch|finditer|subn?)\("
def _mode_for(names: tuple[str, ...], module_src: str) -> str:
"""Derive the call mode from the module source: worst scanning mode wins.
``names`` are the identifiers a pattern is reachable through (its own name
plus, for a pattern living in a table, the table's name). A pattern in a
table is usually never named directly -- it is reached through the loop
variable of ``for p in TABLE:`` -- so those bindings are resolved too, else
the whole egress table reads as ``search`` when it is really ``finditer``.
A pattern with NO direct reference at all is reached indirectly some other
way -- ``active_content`` passes each construct regex into a local ``_scan``
helper that calls ``pattern.sub`` on the parameter -- so it falls back to the
worst scanning mode rather than the mildest. ``sub``/``finditer`` visit every
start position where ``search`` stops at the first match, so the fallback can
only over-measure, never miss. Fallback rows print with a ``*``.
"""
idents = set(names)
for name in names:
for m in re.finditer(r"for\s+([\w,\s]+?)\s+in\s+%s\b" % re.escape(name),
module_src):
idents.update(v.strip() for v in m.group(1).split(",") if v.strip())
found = set()
for ident in idents:
for m in re.finditer(_CALL_RE % re.escape(ident), module_src):
found.add("sub" if m.group(1).startswith("sub") else m.group(1))
for mode in _MODE_ORDER:
if mode in found:
return mode
return "sub*"
def _walk(value, path: str, depth: int = 0):
"""Yield (path, compiled_pattern) for patterns nested in tables/dataclasses."""
if isinstance(value, re.Pattern):
yield path, value
elif depth < 3 and isinstance(value, (list, tuple, frozenset, set)):
for i, item in enumerate(value):
yield from _walk(item, f"{path}[{i}]", depth + 1)
elif (depth < 3 and not isinstance(value, type)
and hasattr(value, "__dataclass_fields__")):
ident = getattr(value, "id", None)
for f in value.__dataclass_fields__:
sub = getattr(value, f)
if isinstance(sub, re.Pattern):
yield (f"{ident}" if ident else f"{path}.{f}"), sub
def collect_module(mod_name: str, table: str, only=None, skip=()) -> list[Target]:
"""Every compiled pattern reachable from a module's namespace."""
mod = importlib.import_module(f"llm_ingestion_guard.{mod_name}")
module_src = (SRC / f"{mod_name}.py").read_text()
out: list[Target] = []
for name, value in vars(mod).items():
if name.startswith("__") or name in skip:
continue
if only is not None and name not in only:
continue
for path, rx in _walk(value, name):
out.append(Target(table, path, rx, rx.pattern,
_mode_for((name, path), module_src)))
return out
def collect_lexicon() -> list[Target]:
raw = json.loads((SRC / "injection_lexicon.json").read_text())["patterns"] raw = json.loads((SRC / "injection_lexicon.json").read_text())["patterns"]
src_by_id = {e["id"]: e["regex"] for e in raw} src_by_id = {e["id"]: e["regex"] for e in raw}
# scan_lexicon drives every entry through `.search()` (lexicon.py:289,352).
return [Target("lexicon", p.id, p.regex, src_by_id[p.id], "search")
for p in load_lexicon()]
def collect_egress() -> list[Target]:
return collect_module("output", "egress", only={"_SECRET_PATTERNS"})
def collect_output() -> list[Target]:
return collect_module("output", "output", skip={"_SECRET_PATTERNS"})
TABLES = {
"lexicon": collect_lexicon,
# The normalizers run `.sub()` over EVERY input before any pattern matches;
# 0.3.3 swept the 83 patterns and not these.
"normalize": lambda: collect_module("lexicon", "normalize"),
"active_content": lambda: collect_module("active_content", "active_content"),
"entropy": lambda: collect_module("entropy", "entropy") + [
# Inline literals in is_base64_like / is_hex_blob (entropy.py:111,118),
# invisible to a namespace walk.
Target("entropy", "is_base64_like", re.compile(r"[A-Za-z0-9+/]{20,}={0,3}"),
r"[A-Za-z0-9+/]{20,}={0,3}", "fullmatch"),
Target("entropy", "is_hex_blob", re.compile(r"(?:0x)?[0-9a-fA-F]{32,}"),
r"(?:0x)?[0-9a-fA-F]{32,}", "fullmatch"),
],
"egress": collect_egress,
"output": collect_output,
# Not "detector tables", but the same regex surface on the same paths.
# Leaving them out would reproduce the 0.3.2 mistake at module granularity.
"sanitize": lambda: collect_module("sanitize", "sanitize"),
"neutralize": lambda: collect_module("neutralize", "neutralize"),
"okf": lambda: collect_module("okf", "okf"),
"contract": lambda: collect_module("contract", "contract"),
"fence": lambda: collect_module("fence", "fence"),
}
EXTRA_UNITS = ["[", "<a:", "<a ", "![", "a://:"]
def sweep(targets: list[Target]) -> list[tuple]:
flagged = [] flagged = []
print(f"== arm-by-arm sweep, N={N1}->{N2}, flag ratio>={RATIO_FLAG} ==") for tgt in targets:
for p in load_lexicon(): units = sample(tgt.src) + EXTRA_UNITS
src = src_by_id[p.id]
units = sample(src) + ["[", "<a:", "<a ", "![", "a://:"]
hits = [] hits = []
for u in units: for u in units:
t1 = t(p.regex, build(u, N1)) t1 = t(tgt.regex, build(u, N1), tgt.mode)
if t1 > HARD_CAP: if t1 > HARD_CAP:
hits.append((u, t1, float("inf"), float("inf"))) hits.append((u, t1, float("inf"), float("inf")))
continue continue
t2 = t(p.regex, build(u, N2)) t2 = t(tgt.regex, build(u, N2), tgt.mode)
ratio = t2 / t1 if t1 > 0 else 0.0 ratio = t2 / t1 if t1 > 0 else 0.0
if t2 >= NOISE_FLOOR and ratio >= RATIO_FLAG: if t2 >= NOISE_FLOOR and ratio >= RATIO_FLAG:
hits.append((u, t1, t2, ratio)) hits.append((u, t1, t2, ratio))
for u, t1, t2, r in sorted(hits, key=lambda h: -h[2])[:3]: for u, t1, t2, r in sorted(hits, key=lambda h: -h[2])[:3]:
flagged.append((p.id, u, t1, t2, r)) flagged.append((tgt, u, t1, t2, r))
print(f" {p.id:40} unit={u!r:14} {t1:.4f}->{t2:.4f}s x{r:.1f}") print(f" {tgt.table}/{tgt.id:34} [{tgt.mode:9}] "
ids = {f[0] for f in flagged} f"unit={u!r:14} {t1:.4f}->{t2:.4f}s x{r:.1f}")
print(f"\ncandidates: {len(ids)} patterns / 83, {len(flagged)} arms") return flagged
def main() -> None:
argv = sys.argv[1:]
listing = "--list" in argv
wanted = [a for a in argv if a != "--list"] or list(TABLES)
unknown = [w for w in wanted if w not in TABLES]
if unknown:
sys.exit(f"unknown table(s): {unknown}; known: {list(TABLES)}")
print(f"== arm-by-arm sweep, N={N1}->{N2}, flag ratio>={RATIO_FLAG} ==")
total, flagged = 0, []
for name in wanted:
targets = TABLES[name]()
total += len(targets)
print(f"-- {name}: {len(targets)} patterns")
if listing:
for tgt in targets:
print(f" {tgt.id:34} [{tgt.mode:9}] {tgt.src[:70]}")
continue
flagged += sweep(targets)
if listing:
print(f"\n{total} patterns across {len(wanted)} table(s)")
return
ids = {f[0].id for f in flagged}
print(f"\ncandidates: {len(ids)} patterns / {total}, {len(flagged)} arms")
if __name__ == "__main__": if __name__ == "__main__":