feat(guard): active-content detector wired into the output gate (review MAJOR #1)
Close the EchoLeak wiring hole (CVE-2025-32711 class): markdown images/ links, reference definitions, autolinks, raw active HTML and data: URIs now surface as report-only findings (active:*, OWASP LLM05) in scan_output step 6, so screen_output and okf.import_bundle dispose of them instead of admitting them with findings=[]. - new active_content.py: canonical home of the shared pattern table + scan_active_content; neutralize refactored to import it (mutating API and behavior unchanged, all neutralize tests pass as-is) - images/links flagged only for absolute/protocol-relative URLs: relative in-bundle links are legitimate wiki/OKF mechanism (principle 5) - evidence carries defanged URLs only (hxxps://evil[.]example) - EchoLeak vectors planted in both showcases; detach proofs cover them - README export list + checklist step 6, CLAUDE.md context line updated Suite: 321 -> 341 passed. Core invariant intact (dependencies=[]).
This commit is contained in:
parent
31166d0af0
commit
4d53765c63
9 changed files with 454 additions and 101 deletions
|
|
@ -15,6 +15,11 @@ human-auditable form: URLs get a non-resolvable scheme and bracketed dots
|
|||
so a renderer shows it as literal text instead of executing it. The visible
|
||||
information survives review; only the machine-actionable affordance dies.
|
||||
|
||||
The pattern table this mutator rewrites is shared with the report-only detector
|
||||
:func:`~llm_ingestion_guard.active_content.scan_active_content` and lives in
|
||||
``active_content`` — detection feeds the standard gate; defanging stays the
|
||||
separate, opt-in mutation below.
|
||||
|
||||
Two properties are load-bearing and mirror the sanitizer:
|
||||
|
||||
1. **Opt-in and separate.** Calling this function *is* the opt-in to mutate.
|
||||
|
|
@ -36,73 +41,20 @@ from __future__ import annotations
|
|||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .active_content import (
|
||||
AUTOLINK_RE,
|
||||
DATA_URI_RE,
|
||||
HTML_TAG_RE,
|
||||
MD_IMAGE_RE,
|
||||
MD_LINK_RE,
|
||||
MD_REFDEF_RE,
|
||||
URL_IN_TEXT_RE,
|
||||
defang_url,
|
||||
is_active_tag,
|
||||
redact,
|
||||
)
|
||||
from .report import Finding, Report, Severity, Source
|
||||
|
||||
# --- URL defang -------------------------------------------------------------
|
||||
# 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).
|
||||
_URL_IN_TEXT_RE = re.compile(r"[A-Za-z][A-Za-z0-9+.\-]*://[^\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)
|
||||
|
||||
|
||||
# --- Active-content constructs ----------------------------------------------
|
||||
# Markdown image / inline link: `[text](url "title")`. `url` stops at the first
|
||||
# `)` or whitespace (balanced-paren URLs matched conservatively — see scope note).
|
||||
_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:...>`.
|
||||
_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.
|
||||
_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",
|
||||
})
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class NeutralizeResult:
|
||||
|
|
@ -112,12 +64,6 @@ class NeutralizeResult:
|
|||
report: Report
|
||||
|
||||
|
||||
def _redact(s: str, show_start: int = 16, show_end: int = 6) -> str:
|
||||
if len(s) <= show_start + show_end + 3:
|
||||
return s
|
||||
return f"{s[:show_start]}...{s[-show_end:]}"
|
||||
|
||||
|
||||
def neutralize(text: str, source: Source = Source.OUTPUT) -> NeutralizeResult:
|
||||
"""Defang active-content constructs in ``text`` and report each class.
|
||||
|
||||
|
|
@ -131,7 +77,7 @@ def neutralize(text: str, source: Source = Source.OUTPUT) -> NeutralizeResult:
|
|||
def _flag(label: str, severity: Severity, count: int, evidence: str) -> None:
|
||||
report.add(Finding(
|
||||
label=label, severity=severity, source=source, detector="neutralize",
|
||||
count=count, evidence=_redact(evidence), owasp="LLM05",
|
||||
count=count, evidence=redact(evidence), owasp="LLM05",
|
||||
))
|
||||
|
||||
# 1. Markdown images — the zero-click auto-fetch primitive (EchoLeak). Run
|
||||
|
|
@ -139,11 +85,11 @@ def neutralize(text: str, source: Source = Source.OUTPUT) -> NeutralizeResult:
|
|||
img_ev: list[str] = []
|
||||
|
||||
def _img(m: re.Match[str]) -> str:
|
||||
defanged = _defang_url(m.group("url"))
|
||||
defanged = defang_url(m.group("url"))
|
||||
img_ev.append(defanged)
|
||||
return f'![{m.group("alt")}]({defanged}{m.group("title")})'
|
||||
|
||||
out, n_img = _MD_IMAGE_RE.subn(_img, out)
|
||||
out, n_img = MD_IMAGE_RE.subn(_img, out)
|
||||
if n_img:
|
||||
_flag("neutralize:markdown-image", Severity.HIGH, n_img, img_ev[0])
|
||||
|
||||
|
|
@ -151,11 +97,11 @@ def neutralize(text: str, source: Source = Source.OUTPUT) -> NeutralizeResult:
|
|||
link_ev: list[str] = []
|
||||
|
||||
def _link(m: re.Match[str]) -> str:
|
||||
defanged = _defang_url(m.group("url"))
|
||||
defanged = defang_url(m.group("url"))
|
||||
link_ev.append(defanged)
|
||||
return f'[{m.group("text")}]({defanged}{m.group("title")})'
|
||||
|
||||
out, n_link = _MD_LINK_RE.subn(_link, out)
|
||||
out, n_link = MD_LINK_RE.subn(_link, out)
|
||||
if n_link:
|
||||
_flag("neutralize:markdown-link", Severity.MEDIUM, n_link, link_ev[0])
|
||||
|
||||
|
|
@ -163,11 +109,11 @@ def neutralize(text: str, source: Source = Source.OUTPUT) -> NeutralizeResult:
|
|||
ref_ev: list[str] = []
|
||||
|
||||
def _ref(m: re.Match[str]) -> str:
|
||||
defanged = _defang_url(m.group("url"))
|
||||
defanged = defang_url(m.group("url"))
|
||||
ref_ev.append(defanged)
|
||||
return m.group("pre") + defanged
|
||||
|
||||
out, n_ref = _MD_REFDEF_RE.subn(_ref, out)
|
||||
out, n_ref = MD_REFDEF_RE.subn(_ref, out)
|
||||
if n_ref:
|
||||
_flag("neutralize:reference-link", Severity.MEDIUM, n_ref, ref_ev[0])
|
||||
|
||||
|
|
@ -175,11 +121,11 @@ def neutralize(text: str, source: Source = Source.OUTPUT) -> NeutralizeResult:
|
|||
auto_ev: list[str] = []
|
||||
|
||||
def _auto(m: re.Match[str]) -> str:
|
||||
defanged = _defang_url(m.group("url"))
|
||||
defanged = defang_url(m.group("url"))
|
||||
auto_ev.append(defanged)
|
||||
return f"<{defanged}>"
|
||||
|
||||
out, n_auto = _AUTOLINK_RE.subn(_auto, out)
|
||||
out, n_auto = AUTOLINK_RE.subn(_auto, out)
|
||||
if n_auto:
|
||||
_flag("neutralize:autolink", Severity.MEDIUM, n_auto, auto_ev[0])
|
||||
|
||||
|
|
@ -188,21 +134,15 @@ def neutralize(text: str, source: Source = Source.OUTPUT) -> NeutralizeResult:
|
|||
|
||||
def _html(m: re.Match[str]) -> str:
|
||||
tag = m.group(0)
|
||||
attrs = m.group("attrs") or ""
|
||||
active = (
|
||||
m.group("name").lower() in _ACTIVE_TAGS
|
||||
or _EVENT_ATTR_RE.search(attrs)
|
||||
or _URL_ATTR_RE.search(attrs)
|
||||
)
|
||||
if not active:
|
||||
if not is_active_tag(m.group("name"), m.group("attrs") or ""):
|
||||
return tag
|
||||
html_state["count"] += 1
|
||||
if not html_state["ev"]:
|
||||
html_state["ev"] = tag
|
||||
inert = _URL_IN_TEXT_RE.sub(lambda u: _defang_url(u.group(0)), tag)
|
||||
inert = URL_IN_TEXT_RE.sub(lambda u: defang_url(u.group(0)), tag)
|
||||
return inert.replace("<", "<").replace(">", ">")
|
||||
|
||||
out = _HTML_TAG_RE.sub(_html, out)
|
||||
out = HTML_TAG_RE.sub(_html, out)
|
||||
if html_state["count"]:
|
||||
_flag("neutralize:raw-html", Severity.HIGH, html_state["count"], html_state["ev"])
|
||||
|
||||
|
|
@ -211,11 +151,11 @@ def neutralize(text: str, source: Source = Source.OUTPUT) -> NeutralizeResult:
|
|||
data_ev: list[str] = []
|
||||
|
||||
def _data(m: re.Match[str]) -> str:
|
||||
defanged = _defang_url(m.group(0))
|
||||
defanged = defang_url(m.group(0))
|
||||
data_ev.append(defanged)
|
||||
return defanged
|
||||
|
||||
out, n_data = _DATA_URI_RE.subn(_data, out)
|
||||
out, n_data = DATA_URI_RE.subn(_data, out)
|
||||
if n_data:
|
||||
_flag("neutralize:data-uri", Severity.HIGH, n_data, data_ev[0])
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue