feat(lexicon): JSON injection lexicon + variant-set scan (TDD) [skip-docs]
Build order step 4 — the load-bearing port from the llm-security seed
(injection-patterns.mjs + string-utils.mjs), stdlib-only.
- injection_lexicon.json: 83 patterns (CRITICAL/HIGH/HYBRID/MEDIUM) as the
single source of truth (regex + id + severity + owasp + desc), compiled once
by a thin loader. Decoupled from the engine for a future TS port.
- scan_lexicon(text, source, max_scan_chars) -> Report: matches every pattern
against a deduped variant set (raw / normalized / homoglyph-folded / rot13),
plus unicode-tag presence signal and the cognitive-load trap.
- normalize_for_scan chain ported: unicode-tags -> bidi -> HTML-entities ->
unicode/hex/URL escapes -> whole-string base64 (reuses entropy.try_decode_base64)
-> collapse letter-spacing; plus fold_homoglyphs / rot13.
- Self-safety (OWASP LLM10): input-size cap (scan prefix + flag oversize) and
ReDoS-safe port — the two nested-.*? sub-agent patterns bounded to
(?:\S+\s+){0,N}?; verified true positives still fire.
- Non-Latin data (homoglyph map, BIDI block) built from explicit code points;
JSON non-ASCII kept as \uXXXX escapes.
24 tests; 55 green total.
[skip-docs]: README positioning + honest-limitations is a deliberate build-order
step-11 deliverable (steps 1-3 likewise left README frozen). README status line
("pre-implementation") is stale and flagged for the step-11 refresh.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K8GmKRCdsPjWYAKWsNgeQS
This commit is contained in:
parent
e5e91df369
commit
f397cd94e1
3 changed files with 1266 additions and 0 deletions
378
src/llm_ingestion_guard/lexicon.py
Normal file
378
src/llm_ingestion_guard/lexicon.py
Normal file
|
|
@ -0,0 +1,378 @@
|
|||
"""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 — the two sub-agent patterns
|
||||
whose seed form nested ``.*?`` are ported with *bounded* token-gap quantifiers
|
||||
(``(?:\\S+\\s+){0,N}?``), since Python's ``re`` has no timeout.
|
||||
|
||||
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 .entropy import try_decode_base64
|
||||
from .report import Finding, Report, Severity, Source
|
||||
|
||||
# --- self-safety: input-size cap (OWASP LLM10) ------------------------------
|
||||
# Large enough for a real ingested document; beyond it we scan the prefix and
|
||||
# flag, so runtime stays bounded even on a decompression-bomb-sized input.
|
||||
MAX_SCAN_CHARS = 1_000_000
|
||||
|
||||
_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 = {
|
||||
"<": "<", ">": ">", "&": "&", """: '"', "'": "'",
|
||||
" ": " ", "&tab;": "\t", "&newline;": "\n",
|
||||
"(": "(", ")": ")", "[": "[", "]": "]",
|
||||
"{": "{", "}": "}", "/": "/", "\": "\\",
|
||||
":": ":", ";": ";", ",": ",", ".": ".",
|
||||
"!": "!", "?": "?", "#": "#", "%": "%",
|
||||
"=": "=", "+": "+", "−": "-", "*": "*",
|
||||
"|": "|", "˜": "~", "`": "`", "^": "^",
|
||||
"_": "_", "&at;": "@", "$": "$",
|
||||
}
|
||||
|
||||
|
||||
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) < 2500:
|
||||
return None
|
||||
tail = text[2000:]
|
||||
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 = 40 # 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
|
||||
Loading…
Add table
Add a link
Reference in a new issue