1
0
Fork 0
llm-ingestion-pipeline-secu.../src/llm_ingestion_guard/lexicon.py
Kjell Tore Guttormsen b8028ba870 fix(lexicon): two quadratic patterns, reachable through the output gate too
The input-path duty `8deca93` scoped. All 83 lexicon patterns measured arm by
arm; two are quadratic, same shape 0.3.2 fixed -- a run in front of a required
literal that may cross the pattern's own opening anchor. Exponent 1.98 over five
points, so quadratic, not exponential.

  markdown:link-anchor-injection  `[`          1.91s @8k   ~8.3h at the cap
  markdown:link-anchor-injection  `[system](`  0.006s @8k  ~89s at the cap
  markdown:link-ref-comment       `[//]: # (`  0.22s @8k   ~1.0h at the cap

Not input-path-only: `scan_lexicon` runs on the output path, so `scan_output("["
* 100_000)` took 334.7s. 0.3.2's "last quadratic site on the output path" was
false when written -- its sweep drove `[` only through `scan_active_content`.

Fix is anchor exclusion, not bounding (bounding attacker-controlled content is a
one-line bypass). The excluded char is `(`, not the obvious `[`: excluding `[`
drops `[//]: # (see [x] then ignore this)`, which no other pattern catches. The
anchors contain `(` too, so it telescopes at zero measured recall cost.

N is per row deliberately. The URL arm ran 0.9s UNFIXED at N=100_000 -- under the
2.0s bound, so that row could not have failed. Measured at N=300_000 instead,
where crafted (8.10s) and legitimate (0.926s) separate 8.8x.
2026-07-31 21:50:51 +02:00

399 lines
15 KiB
Python

"""lexicon — the load-bearing injection-pattern gate.
A ``text -> findings`` detector (design principle 3): pure, no I/O beyond
loading its own data file once, no mutation. It scans text against a JSON
pattern table (CRITICAL / HIGH / MEDIUM, with cross-domain HYBRID patterns at
HIGH) ported from the ``llm-security`` seed, and — because obfuscation is the
whole game — it matches every pattern against a *set of variants* of the input,
not just the raw text:
* **raw** — the text as given.
* **normalized** — :func:`normalize_for_scan` peels obfuscation layers
(Unicode-tag steganography, BIDI overrides, HTML entities, unicode/hex/URL
escapes, whole-string base64), then collapses letter-spacing.
* **homoglyph-folded** — Cyrillic/Greek look-alikes folded to Latin, so
``ign``+Cyrillic-``o``+``re`` matches the same pattern as ``ignore``.
* **rot13** — for inputs long enough to carry a sentence, a rot13 layer
catches phrases hidden in comment blocks.
Findings are deduplicated by pattern id (the same class matched in two variants
reports once). Disposition (WARN / QUARANTINE_REVIEW / FAIL_SECURE) is the
caller's — this module only reports.
**Self-safety (OWASP LLM10).** A scanner that hangs on crafted input *is* the
DoS. Two guards land here (the shared guard ``entropy`` deferred to this
module): an input-size cap (:data:`MAX_SCAN_CHARS`; oversize input is scanned up
to the cap and flagged) and ReDoS-safe patterns, since Python's ``re`` has no
timeout. The pattern table needs *two* remedies, not one:
* **Bounded token gaps** — the two sub-agent patterns whose seed form nested
``.*?`` are ported with ``(?:\\S+\\s+){0,N}?``.
* **Anchor exclusion** — a run in front of a *required* literal is quadratic
whenever it may cross the pattern's own opening anchor, with no nesting
involved. Measured across all 83 patterns arm by arm, two markdown patterns
had this defect; both now exclude the anchor character from the run. The
exclusion is ``(`` rather than ``[`` in both cases: it telescopes just as
well (the anchors contain ``(`` too) and costs no measured recall, whereas
excluding ``[`` drops a link-ref comment carrying a nested bracket that no
other pattern catches. Bounding the runs instead is the wrong fix here for
the reason ``active_content`` documents — the content is attacker-controlled,
so padding past a bound would be a one-line bypass.
The cap does not mitigate this on its own: it bounds the *input*, and quadratic
work on a bounded input is still hours. See
``tests/test_lexicon.py::test_crafted_redos_payload_stays_bounded_in_the_lexicon``.
The pattern table ships as JSON (``injection_lexicon.json``) — the single source
of truth, decoupled from this engine for a future TS port. Non-Latin data in
*this* module (the homoglyph map, the BIDI code block) is built from explicit
code points, never literal look-alike/invisible characters in source.
"""
from __future__ import annotations
import json
import re
import unicodedata
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import unquote
from .calibration import (
COGNITIVE_LOAD_MIN_LEN,
COGNITIVE_LOAD_TAIL_START,
MAX_SCAN_CHARS,
ROT13_MIN_LEN as _ROT13_MIN_LEN,
)
from .entropy import try_decode_base64
from .report import Finding, Report, Severity, Source
# Self-safety input-size cap (OWASP LLM10), rot13 variant floor, and the
# cognitive-load-trap lengths all live in `calibration` (the Node port shares
# them). MAX_SCAN_CHARS is re-exported here for `output` and existing callers.
_LEXICON_FILE = "injection_lexicon.json"
_FLAG_MAP = {"i": re.IGNORECASE, "m": re.MULTILINE, "s": re.DOTALL}
@dataclass(frozen=True)
class LexiconPattern:
"""One compiled lexicon entry. ``id`` is the machine label used for dedup
and carried onto :class:`~llm_ingestion_guard.report.Finding.label`."""
id: str
regex: re.Pattern[str]
severity: Severity
owasp: str
desc: str
_LEXICON_CACHE: list[LexiconPattern] | None = None
def load_lexicon() -> list[LexiconPattern]:
"""Load and compile the JSON pattern table once; cached thereafter."""
global _LEXICON_CACHE
if _LEXICON_CACHE is None:
path = Path(__file__).with_name(_LEXICON_FILE)
data = json.loads(path.read_text(encoding="utf-8"))
patterns: list[LexiconPattern] = []
for entry in data["patterns"]:
flags = 0
for ch in entry.get("flags", ""):
flags |= _FLAG_MAP[ch]
patterns.append(
LexiconPattern(
id=entry["id"],
regex=re.compile(entry["regex"], flags),
severity=Severity(entry["severity"]),
owasp=entry.get("owasp", "LLM01"),
desc=entry.get("desc", ""),
)
)
_LEXICON_CACHE = patterns
return _LEXICON_CACHE
# --- normalization primitives (ported from string-utils.mjs) ----------------
# Confusable map — characters that LOOK Latin but are Cyrillic/Greek. Kept small
# and surgical (injection vocabulary only); Latin-Extended letters used in real
# languages (aa/ae/oe, accented vowels, etc.) are deliberately excluded. Built
# from explicit code points so no look-alike character hides in this source.
_HOMOGLYPH_MAP = {
# Cyrillic -> Latin (lower)
chr(0x0430): "a", chr(0x0435): "e", chr(0x043E): "o", chr(0x0441): "c",
chr(0x0440): "p", chr(0x0445): "x", chr(0x0443): "y", chr(0x0456): "i",
chr(0x0458): "j", chr(0x0455): "s", chr(0x04CF): "l",
# Cyrillic -> Latin (upper)
chr(0x0410): "A", chr(0x0415): "E", chr(0x041E): "O", chr(0x0421): "C",
chr(0x0420): "P", chr(0x0425): "X", chr(0x0423): "Y",
# Greek -> Latin (unambiguous look-alikes)
chr(0x03B1): "a", chr(0x03BF): "o", chr(0x03C1): "p", chr(0x03B9): "i",
chr(0x03BD): "v", chr(0x03C4): "t",
chr(0x0391): "A", chr(0x039F): "O", chr(0x03A1): "P", chr(0x03A4): "T",
}
# BIDI override / isolate code points (reorder text visually): U+202A-202E and
# U+2066-2069. Filtered out char-by-char — no invisible char in this source.
_BIDI_CPS = frozenset(range(0x202A, 0x202F)) | frozenset(range(0x2066, 0x206A))
_HTML_HEX_RE = re.compile(r"&#x([0-9a-fA-F]{1,6});")
_HTML_DEC_RE = re.compile(r"&#(\d{1,7});")
_HTML_NAMED_RE = re.compile(r"&[a-zA-Z]{2,8};")
_U_BRACE_RE = re.compile(r"\\u\{([0-9a-fA-F]{1,6})\}")
_U4_RE = re.compile(r"\\u([0-9a-fA-F]{4})")
_HEX_ESC_RE = re.compile(r"\\x([0-9a-fA-F]{2})")
# 4+ single letters separated by single spaces -> collapse ("i g n o r e").
_SPACING_RE = re.compile(r"\b([A-Za-z]) (?:[A-Za-z] ){2,}[A-Za-z]\b")
_HTML_NAMED = {
"&lt;": "<", "&gt;": ">", "&amp;": "&", "&quot;": '"', "&apos;": "'",
"&nbsp;": " ", "&tab;": "\t", "&newline;": "\n",
"&lpar;": "(", "&rpar;": ")", "&lsqb;": "[", "&rsqb;": "]",
"&lcub;": "{", "&rcub;": "}", "&sol;": "/", "&bsol;": "\\",
"&colon;": ":", "&semi;": ";", "&comma;": ",", "&period;": ".",
"&excl;": "!", "&quest;": "?", "&num;": "#", "&percnt;": "%",
"&equals;": "=", "&plus;": "+", "&minus;": "-", "&ast;": "*",
"&vert;": "|", "&tilde;": "~", "&grave;": "`", "&Hat;": "^",
"&lowbar;": "_", "&at;": "@", "&dollar;": "$",
}
def _chr_or(cp: int, original: str) -> str:
return chr(cp) if cp <= 0x10FFFF else original
def rot13(s: str) -> str:
"""Caesar shift by 13 over ASCII letters; its own inverse."""
out = []
for ch in s:
o = ord(ch)
if 65 <= o <= 90:
out.append(chr((o - 65 + 13) % 26 + 65))
elif 97 <= o <= 122:
out.append(chr((o - 97 + 13) % 26 + 97))
else:
out.append(ch)
return "".join(out)
def fold_homoglyphs(s: str) -> str:
"""Fold confusable Cyrillic/Greek characters to their Latin look-alikes.
NFKC first (collapses Mathematical-Alphanumeric, width variants, ligatures),
then the surgical :data:`_HOMOGLYPH_MAP`. Pure-ASCII input short-circuits.
"""
if not s or all(ord(ch) < 128 for ch in s):
return s
normalized = unicodedata.normalize("NFKC", s)
return "".join(_HOMOGLYPH_MAP.get(ch, ch) for ch in normalized)
def collapse_letter_spacing(s: str) -> str:
"""Collapse letter-spaced evasion: ``i g n o r e`` -> ``ignore`` (>=4)."""
return _SPACING_RE.sub(lambda m: m.group(0).replace(" ", ""), s)
def contains_unicode_tags(s: str) -> bool:
"""True if ``s`` holds invisible Unicode-Tag (U+E0000 block) or PUA chars."""
for ch in s:
cp = ord(ch)
if 0xE0001 <= cp <= 0xE007F:
return True
if 0xF0000 <= cp <= 0xFFFFD:
return True
if 0x100000 <= cp <= 0x10FFFD:
return True
return False
def _decode_unicode_tags(s: str) -> str:
"""Reveal Unicode-Tag steganography: U+E0001-E007F -> ASCII (cp-0xE0000)."""
if not any(0xE0001 <= ord(ch) <= 0xE007F for ch in s):
return s
out = []
for ch in s:
cp = ord(ch)
out.append(chr(cp - 0xE0000) if 0xE0001 <= cp <= 0xE007F else ch)
return "".join(out)
def _strip_bidi_overrides(s: str) -> str:
if not any(ord(ch) in _BIDI_CPS for ch in s):
return s
return "".join(ch for ch in s if ord(ch) not in _BIDI_CPS)
def _decode_html_entities(s: str) -> str:
if "&" not in s:
return s
s = _HTML_HEX_RE.sub(lambda m: _chr_or(int(m.group(1), 16), m.group(0)), s)
s = _HTML_DEC_RE.sub(lambda m: _chr_or(int(m.group(1), 10), m.group(0)), s)
s = _HTML_NAMED_RE.sub(lambda m: _HTML_NAMED.get(m.group(0), m.group(0)), s)
return s
def _decode_unicode_escapes(s: str) -> str:
s = _U_BRACE_RE.sub(lambda m: _chr_or(int(m.group(1), 16), m.group(0)), s)
s = _U4_RE.sub(lambda m: chr(int(m.group(1), 16)), s)
return s
def _decode_hex_escapes(s: str) -> str:
return _HEX_ESC_RE.sub(lambda m: chr(int(m.group(1), 16)), s)
def _decode_url_encoding(s: str) -> str:
if "%" not in s:
return s
return unquote(s)
def normalize_for_scan(s: str) -> str:
"""Peel known obfuscation layers so patterns match the decoded text.
Order mirrors the seed ``normalizeForScan``: Unicode-tags and BIDI first
(once), then up to 3 iterations of HTML-entity / unicode-escape /
hex-escape / URL / whole-string-base64 decoding (to catch layered
encodings), then a letter-spacing collapse. Whole-string base64 reuses
:func:`entropy.try_decode_base64` — a blob embedded *inside* larger text is
``entropy``'s decode-and-rescan job, not this one.
"""
result = _decode_unicode_tags(s)
result = _strip_bidi_overrides(result)
for _ in range(3):
prev = result
result = _decode_html_entities(result)
result = _decode_unicode_escapes(result)
result = _decode_hex_escapes(result)
result = _decode_url_encoding(result)
decoded = try_decode_base64(result)
if decoded is not None:
result = decoded
if result == prev:
break
return collapse_letter_spacing(result)
# --- cognitive-load trap (injection buried deep in verbose output) ----------
def check_cognitive_load_trap(text: str) -> str | None:
"""Return the id of a CRITICAL pattern found *only past the first 2000
chars* of long text (>=2500), else ``None``. Placement is the signal: an
override buried at the tail of verbose output is a human-in-the-loop trap.
"""
if len(text) < COGNITIVE_LOAD_MIN_LEN:
return None
tail = text[COGNITIVE_LOAD_TAIL_START:]
for pattern in load_lexicon():
if pattern.severity is Severity.CRITICAL and pattern.regex.search(tail):
return pattern.id
return None
# --- variant set + scan ------------------------------------------------------
# _ROT13_MIN_LEN (imported from calibration): shorter strings hit
# rot13-look-alike false positives.
def _build_variants(text: str) -> list[tuple[str, str]]:
"""The deduplicated (name, string) variants to match every pattern against."""
normalized = normalize_for_scan(text)
folded = fold_homoglyphs(text)
folded_normalized = fold_homoglyphs(normalized)
variants: list[tuple[str, str]] = [("raw", text)]
seen = {text}
def add(name: str, value: str) -> None:
if value not in seen:
seen.add(value)
variants.append((name, value))
add("normalized", normalized)
add("folded", folded)
add("folded-normalized", folded_normalized)
if len(text) > _ROT13_MIN_LEN:
add("rot13", rot13(text))
if len(normalized) > _ROT13_MIN_LEN:
add("rot13-normalized", rot13(normalized))
return variants
def scan_lexicon(
text: str,
source: Source = Source.INPUT,
max_scan_chars: int = MAX_SCAN_CHARS,
) -> Report:
"""Scan ``text`` for injection patterns across its obfuscation variants."""
report = Report()
truncated = len(text) > max_scan_chars
scan_text = text[:max_scan_chars] if truncated else text
if truncated:
report.add(
Finding(
label="lexicon:oversize-input",
severity=Severity.MEDIUM,
source=source,
detector="lexicon",
count=len(text),
owasp="LLM10",
evidence=f"input {len(text)} chars exceeds cap {max_scan_chars}; scanned prefix only",
)
)
patterns = load_lexicon()
seen: set[str] = set()
for variant_name, variant in _build_variants(scan_text):
for pattern in patterns:
if pattern.id in seen:
continue
match = pattern.regex.search(variant)
if match is None:
continue
seen.add(pattern.id)
# Offset is only meaningful in the raw text; decoding/folding/rot13
# shift positions, so leave it None and name the variant instead.
offset = match.start() if variant_name == "raw" else None
report.add(
Finding(
label=pattern.id,
severity=pattern.severity,
source=source,
detector="lexicon",
offset=offset,
owasp=pattern.owasp,
evidence=f"{pattern.desc} [{variant_name}]",
)
)
# Unicode-tag / PUA presence is a distinct HIGH signal regardless of decoded
# content. A hidden CRITICAL injection is already caught via the normalized
# variant (which decodes the tags), so no separate escalation is needed here.
if contains_unicode_tags(scan_text):
report.add(
Finding(
label="lexicon:unicode-tags-present",
severity=Severity.HIGH,
source=source,
detector="lexicon",
owasp="LLM01",
evidence="invisible Unicode-Tag/PUA characters present",
)
)
trap = check_cognitive_load_trap(scan_text)
if trap is not None:
report.add(
Finding(
label="hitl-trap:cognitive-load",
severity=Severity.MEDIUM,
source=source,
detector="lexicon",
owasp="LLM01",
evidence=f"critical injection buried after 2000 chars ({trap})",
)
)
return report