1
0
Fork 0

feat: scaffold package + report and sanitize modules (TDD)

Build order steps 1-2 of docs/PLAN.md:
- pyproject.toml (llm-ingestion-guard, stdlib-only core, extras [ml]/[judge]/[dev]), LICENSE (MIT)
- report: Finding/Report/Severity/Source shared type (pure data)
- sanitize: carrier stripping (zero-width, BIDI, Unicode-tag, HTML comment,
  data: URI) with the byte-identical / removes-only invariant
- docs/PLAN.md: v1 implementation plan (positioning A, gap-expanded scope,
  llm-security reuse map)

15 tests passing.

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:
Kjell Tore Guttormsen 2026-07-04 09:24:20 +02:00
commit a9c4ccd8c7
8 changed files with 564 additions and 0 deletions

View file

@ -0,0 +1,11 @@
"""llm-ingestion-guard — a write-time defensive layer for LLM ingestion pipelines.
Query-time guardrails guard the answer; this library guards the *artifact*. It
packages the ingestion-side security contract sanitize -> fence -> tool-less
quarantined transform -> per-stage capability isolation -> scan output before
persist -> fail-secure as composable, stdlib-first, framework-agnostic code.
The public API is wired up as modules land; see docs/PLAN.md for the build order.
"""
__version__ = "0.1.0"

View file

@ -0,0 +1,99 @@
"""report — the structured findings type shared across all detectors.
Pure data. Detection everywhere in this library is ``text -> findings`` (design
principle 3): a detector never mutates its input, performs no I/O, and returns
``Finding`` objects collected into a ``Report``. Disposition (WARN /
QUARANTINE_REVIEW / FAIL_SECURE) is decided by the caller from these findings
it is not baked into the Finding itself (design principle 4).
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Iterable, Optional
class Severity(str, Enum):
"""Finding severity tiers. Ordered by :func:`severity_rank`."""
CRITICAL = "critical"
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
INFO = "info"
_RANK = {
Severity.CRITICAL: 4,
Severity.HIGH: 3,
Severity.MEDIUM: 2,
Severity.LOW: 1,
Severity.INFO: 0,
}
def severity_rank(severity: Severity) -> int:
"""Return an integer rank for a severity — higher is more severe."""
return _RANK[severity]
class Source(str, Enum):
"""Which side of the pipeline a finding came from.
``INPUT`` the untrusted content fed into the model. ``OUTPUT`` the
model's emitted text, scanned before it is persisted (the RAG-poisoning
gate).
"""
INPUT = "input"
OUTPUT = "output"
@dataclass(frozen=True)
class Finding:
"""A single detection. Immutable and hashable.
``label`` is the finding class (e.g. ``"override:ignore-previous"``),
``detector`` is the producing module (``sanitize`` / ``lexicon`` /
``entropy`` / ``output`` / ...). ``evidence`` is a redacted, human-readable
fragment never raw payload content in alerts.
"""
label: str
severity: Severity
source: Source
detector: str
count: int = 1
offset: Optional[int] = None
evidence: Optional[str] = None
owasp: Optional[str] = None
@dataclass
class Report:
"""A mutable collection of findings with convenience aggregates."""
findings: list[Finding] = field(default_factory=list)
def add(self, finding: Finding) -> None:
self.findings.append(finding)
def extend(self, findings: Iterable[Finding]) -> None:
self.findings.extend(findings)
@property
def found(self) -> bool:
return bool(self.findings)
def max_severity(self) -> Optional[Severity]:
"""The most severe finding's severity, or ``None`` if empty."""
if not self.findings:
return None
return max((f.severity for f in self.findings), key=severity_rank)
def counts(self) -> dict[Severity, int]:
"""Count of findings per severity tier (all tiers present, zero-filled)."""
counts = {severity: 0 for severity in Severity}
for finding in self.findings:
counts[finding.severity] += 1
return counts

View file

@ -0,0 +1,98 @@
"""sanitize — carrier stripping for untrusted content.
Removes the invisible/steganographic carrier classes that smuggle instructions
past a human reader but reach the model: zero-width characters, BIDI overrides,
Unicode-tag steganography, HTML comments, and ``data:`` URIs.
Contract (BRIEF §5, §9): this function only ever *removes* never rewrites.
Clean input returns byte-identical with an empty report, and the output is
always a subsequence of the input. Per-class counts are reported so the caller
can gate (WARN / block) on them. Ported from the ``llm-security`` unicode-scanner
and string-utils primitives.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from .report import Finding, Report, Severity, Source
# Invisible / steganographic character classes (codepoints).
_ZERO_WIDTH = frozenset({0x200B, 0x200C, 0x200D, 0xFEFF, 0x00AD})
_BIDI = frozenset({0x202A, 0x202B, 0x202C, 0x202D, 0x202E, 0x2066, 0x2067, 0x2068, 0x2069})
_TAG_LO, _TAG_HI = 0xE0000, 0xE007F # Unicode Tags block (U+E0000U+E007F)
# Span carriers. Lazy `.*?` + explicit terminator — no catastrophic backtracking.
_HTML_COMMENT_RE = re.compile(r"<!--.*?-->", re.DOTALL)
# `data:` not preceded by a letter (so "metadata:" / "userdata:" do not match),
# consuming up to the next whitespace / quote / angle bracket / closing paren.
_DATA_URI_RE = re.compile(r"(?<![A-Za-z])data:[^\s'\"<>)]+", re.IGNORECASE)
@dataclass(frozen=True)
class SanitizeResult:
"""The cleaned text plus a report of what was stripped."""
text: str
report: Report
def _redact(s: str, show_start: int = 12, show_end: int = 4) -> str:
if len(s) <= show_start + show_end + 3:
return s
return f"{s[:show_start]}...{s[-show_end:]}"
def _decode_tags(codepoints: list[int]) -> str:
"""Decode Unicode-tag codepoints to their hidden ASCII (cp - 0xE0000)."""
out = []
for cp in codepoints:
ascii_cp = cp - 0xE0000
out.append(chr(ascii_cp) if 0x20 <= ascii_cp <= 0x7E else "?")
return "".join(out)
def sanitize(text: str, source: Source = Source.INPUT) -> SanitizeResult:
"""Strip carrier classes from ``text`` and report per-class counts."""
report = Report()
# Character-class carriers: single pass, keep everything else verbatim.
zero_width = 0
bidi = 0
tag_cps: list[int] = []
kept: list[str] = []
for ch in text:
cp = ord(ch)
if cp in _ZERO_WIDTH:
zero_width += 1
elif cp in _BIDI:
bidi += 1
elif _TAG_LO <= cp <= _TAG_HI:
tag_cps.append(cp)
else:
kept.append(ch)
# Preserve object identity (and byte-identity) when nothing was stripped.
cleaned = "".join(kept) if (zero_width or bidi or tag_cps) else text
# Span carriers.
cleaned, n_comments = _HTML_COMMENT_RE.subn("", cleaned)
cleaned, n_data = _DATA_URI_RE.subn("", cleaned)
if zero_width:
report.add(Finding(label="sanitize:zero-width", severity=Severity.HIGH,
source=source, detector="sanitize", count=zero_width, owasp="LLM01"))
if bidi:
report.add(Finding(label="sanitize:bidi-override", severity=Severity.HIGH,
source=source, detector="sanitize", count=bidi, owasp="LLM01"))
if tag_cps:
report.add(Finding(label="sanitize:unicode-tag", severity=Severity.CRITICAL,
source=source, detector="sanitize", count=len(tag_cps),
evidence=_redact(_decode_tags(tag_cps)), owasp="LLM01"))
if n_comments:
report.add(Finding(label="sanitize:html-comment", severity=Severity.MEDIUM,
source=source, detector="sanitize", count=n_comments, owasp="LLM01"))
if n_data:
report.add(Finding(label="sanitize:data-uri", severity=Severity.MEDIUM,
source=source, detector="sanitize", count=n_data, owasp="LLM01"))
return SanitizeResult(text=cleaned, report=report)