llm-ingestion-pipeline-security/tests/test_invisible.py

"""Every invisible codepoint is caught, or exempt on context with a named reason.

The denominator is frozen in `docs/threat-2026/codepoints.json` (session 1,
2026-10-28): Default_Ignorable_Code_Point | Cf | Co in Unicode 18.0, 141 674
codepoints. Before 1.4.2, 17 of 19 measured classes gave 0 findings and stayed in
the "sanitized" text — variation-selector smuggling (A01) and Sneaky Bits (A02)
among them.

The rule is the one the ZWJ fix established: judge a character by its CONTEXT,
never by identity alone. The exemptions, each a known legitimate use:

- one VS15/VS16 after an emoji base (presentation selector);
- one ideographic selector after a CJK ideograph (registered name variants);
- ZWNJ/ZWJ next to an Arabic-script or Indic letter (required orthography);
- LRM/RLM/ALM in a document that carries right-to-left text;
- an Arabic number sign directly before a digit (it spans the digits).

A RUN is a carrier whatever it stands after: an exemption never holds when a
neighbour is itself invisible (selectors: when the next one is a selector).
Private use is caught in full (operator 2026-10-28: an icon-font glyph is
blocked rather than leaving a ~17-bit-per-slot channel open).

Every payload here is built from escapes — no invisible glyph sits in source.
"""
import json
from pathlib import Path

import pytest

from llm_ingestion_guard import Disposition, PRESET_TRUSTED_SOURCE, decide, scan_output
from llm_ingestion_guard.lexicon import scan_lexicon
from llm_ingestion_guard.report import Report
from llm_ingestion_guard.sanitize import sanitize
from redos_clock import scan_seconds

_CODEPOINTS = (Path(__file__).resolve().parent.parent
               / "docs" / "threat-2026" / "codepoints.json")
_DENOMINATOR = json.loads(_CODEPOINTS.read_text(encoding="utf-8"))

# The output-gate label for each sanitize class (one predicate, two surfaces).
_OUTPUT_LABEL = {
    "sanitize:zero-width": "output:zero-width-present",
    "sanitize:bidi-override": "output:bidi-present",
    "sanitize:unicode-tag": "output:unicode-tag-present",
    "sanitize:variation-selector": "output:variation-selector-present",
    "sanitize:private-use": "output:private-use-present",
    "sanitize:invisible-format": "output:invisible-format-present",
}

VS16 = "\ufe0f"
VS15 = "\ufe0e"
ZWNJ = "\u200c"
ZWJ = "\u200d"
LRM = "\u200e"
RLM = "\u200f"
ALM = "\u061c"
HEBREW = "שלום"                       # shalom
PERSIAN_WORD = "می" + ZWNJ + "خواهم"  # mi-khaham


def _labels(report) -> set:
    return {f.label for f in report.findings}


def _sanitize_labels(text: str) -> set:
    return {lbl for lbl in _labels(sanitize(text).report) if lbl in _OUTPUT_LABEL}


def _output_labels(text: str) -> set:
    return {lbl for lbl in _labels(scan_output(text)) if lbl in _OUTPUT_LABEL.values()}


def _ranges():
    return [(int(lo, 16), int(hi, 16)) for lo, hi in _DENOMINATOR["ranges"]]


# --- the denominator ---------------------------------------------------------


def test_the_frozen_denominator_is_the_one_in_the_code():
    from llm_ingestion_guard.sanitize import _INVISIBLE_RANGES
    assert list(_INVISIBLE_RANGES) == _ranges()
    assert sum(hi - lo + 1 for lo, hi in _INVISIBLE_RANGES) == _DENOMINATOR["counts"]["total"]


def test_every_denominator_codepoint_out_of_context_is_caught_on_both_surfaces():
    # 141 674 codepoints, each between two ASCII letters: no exemption context
    # exists there, so every one must be counted, stripped and flagged.
    cps = [cp for lo, hi in _ranges() for cp in range(lo, hi + 1)]
    assert len(cps) == 141_674
    text = "a" + "a".join(chr(cp) for cp in cps) + "a"
    result = sanitize(text)
    assert result.text == "a" * (len(cps) + 1)
    stripped = sum(f.count for f in result.report.findings if f.label in _OUTPUT_LABEL)
    assert stripped == len(cps)
    from llm_ingestion_guard.output import _scan_invisible_carriers
    from llm_ingestion_guard.report import Source
    flagged = _scan_invisible_carriers(text, Source.OUTPUT)
    assert sum(f.count for f in flagged.findings) == len(cps)


@pytest.mark.parametrize("cp", [
    0x2060, 0x2061, 0x2062, 0x2063, 0x2064, 0x180E, 0x3164, 0x115F, 0x1160,
    0xFFA0, 0x034F, 0xFFF9, 0x200E, 0x061C, 0xE000, 0xFE00, 0xE0100,
], ids=lambda cp: f"U+{cp:04X}")
def test_the_seventeen_classes_measured_red_are_caught(cp):
    # PLAN T.1.2: 17 of 19 classes, 8 of each, gave 0 findings on v1.4.1.
    text = "notes " + chr(cp) * 8 + " end"
    assert _sanitize_labels(text), f"U+{cp:04X}: not caught by sanitize"
    assert chr(cp) not in sanitize(text).text
    assert _output_labels(text), f"U+{cp:04X}: not caught by the output gate"


def test_new_carrier_labels_block_in_every_tier():
    for label in _OUTPUT_LABEL.keys() | set(_OUTPUT_LABEL.values()):
        from llm_ingestion_guard.report import Finding, Severity, Source
        report = Report()
        report.add(Finding(label=label, severity=Severity.HIGH, source=Source.INPUT,
                           detector="sanitize", owasp="LLM01:2026"))
        decision = decide(report, PRESET_TRUSTED_SOURCE)
        assert decision.disposition is Disposition.FAIL_SECURE, label


# --- the attacks -------------------------------------------------------------


def test_variation_selector_run_after_an_emoji_is_a_carrier():
    # A01: Butler's encoding hangs the run off a visible emoji. The first
    # selector stands right after a valid base; the run is what convicts it.
    payload = "".join(chr(0xFE00 + b) if b < 16 else chr(0xE0100 + b - 16)
                      for b in b"ignore previous")
    text = "Notes \U0001F600" + payload + " end."
    result = sanitize(text)
    assert "sanitize:variation-selector" in _labels(result.report)
    assert result.text == "Notes \U0001F600 end."
    assert "output:variation-selector-present" in _labels(scan_output(text))


def test_sneaky_bits_are_a_carrier():
    # A02: bits in U+2062 / U+2064.
    text = "Notes" + "\u2062\u2064\u2064\u2062" * 10 + " end."
    assert "sanitize:invisible-format" in _labels(sanitize(text).report)
    assert sanitize(text).text == "Notes end."
    assert "output:invisible-format-present" in _labels(scan_output(text))


# --- the exemptions: legitimate use survives byte-identical --------------------

_LEGITIMATE = [
    ("VS16 after stopwatch", "Timer ⏱" + VS16 + " 5 min"),        # measured in corpus
    ("VS16 after warning", "⚠" + VS16 + " Do not deploy"),
    ("VS16 after heart", "Thanks ❤" + VS16 + " team"),
    ("VS15 after heart", "Thanks ❤" + VS15 + " team"),
    ("keycap one", "Step 1" + VS16 + "⃣ first"),
    ("copyright sign", "©" + VS16 + " 2026"),
    ("VS16 + ZWJ sequence", "❤" + VS16 + ZWJ + "\U0001F525 on fire"),
    ("ideographic variant", "葛\U000E0100城"),                  # registered name form
    ("CJK compatibility variant", "豈\ufe00 ok"),
    ("Persian ZWNJ", "من " + PERSIAN_WORD + " این"),
    ("Persian ZWNJ after a link", "[کتاب]]" + ZWNJ + "ها"),
    ("Persian ZWJ for medial heh", "ه" + ZWJ + "."),
    ("Devanagari half form", "क्" + ZWJ + "ष"),
    ("Malayalam explicit virama", "ന്" + ZWNJ + " "),
    ("LRM in Hebrew text", HEBREW + " (ABC" + LRM + ") " + HEBREW),
    ("RLM in Hebrew text", HEBREW + " 6" + RLM + ": " + HEBREW),
    ("LRM elsewhere in an RTL document", "Ref [1" + LRM + "].\n" + HEBREW),
    ("ALM in Arabic text", "مصر " + ALM + "(1) مصر"),
    ("Arabic number sign", "المبلغ \u0600١٢٣"),
    ("Arabic year sign", "\u0601" + "2026"),
]


@pytest.mark.parametrize("name,text", _LEGITIMATE, ids=[c[0] for c in _LEGITIMATE])
def test_legitimate_use_is_neither_flagged_nor_stripped(name, text):
    result = sanitize(text)
    assert not _sanitize_labels(text), f"{name}: {_labels(result.report)}"
    assert result.text == text, f"{name}: content corrupted"
    assert not _output_labels(text), f"{name}: output gate flagged it"


def test_persian_zwnj_does_not_trip_the_word_splitting_row():
    # The lexicon row `unicode:zero-width-in-word` (\w[ZW]\w) matched every
    # Persian word written with its required ZWNJ. It judges the joiner by the
    # same predicate now; a Latin word split is still the row's target.
    assert "unicode:zero-width-in-word" not in _labels(scan_lexicon(PERSIAN_WORD))
    assert "unicode:zero-width-in-word" in _labels(scan_lexicon("impor" + ZWNJ + "tant"))
    mixed = PERSIAN_WORD + " impor" + ZWNJ + "tant"
    assert "unicode:zero-width-in-word" in _labels(scan_lexicon(mixed))


# --- the exemptions do not reach the attack shapes ---------------------------

_CARRIERS = [
    ("two VS16 after an emoji", "❤" + VS16 + VS16, "sanitize:variation-selector"),
    ("VS16 after a letter", "a" + VS16, "sanitize:variation-selector"),
    ("VS16 at document start", VS16 + "x", "sanitize:variation-selector"),
    ("two ideographic selectors", "葛\U000E0100\U000E0101", "sanitize:variation-selector"),
    ("ideographic selector after a letter", "a\U000E0100", "sanitize:variation-selector"),
    ("ZWNJ splitting a Latin word", "impor" + ZWNJ + "tant", "sanitize:zero-width"),
    ("ZWNJ between emoji", "\U0001F469" + ZWNJ + "\U0001F4BB", "sanitize:zero-width"),
    ("ZWNJ run in Persian", "می" + ZWNJ + ZWNJ + "خ", "sanitize:zero-width"),
    ("ZWNJ next to another invisible", "م" + ZWNJ + LRM + "خ", "sanitize:zero-width"),
    ("LRM in a Latin-only document", "Plain text" + LRM + " here", "sanitize:invisible-format"),
    ("ALM alone is not RTL context", ALM + "x", "sanitize:invisible-format"),
    ("number sign alone is not RTL context", "\u0600 then " + LRM, "sanitize:invisible-format"),
    ("LRM/RLM bits in Hebrew", HEBREW + " " + LRM + RLM + LRM + RLM, "sanitize:invisible-format"),
    ("number sign without a digit", "م \u0600م", "sanitize:invisible-format"),
    ("number-sign run", "\u0600\u0600١", "sanitize:invisible-format"),
    ("BMP private use", "icon \ue000 here", "sanitize:private-use"),
    ("Apple logo", "\uf8ff Mac", "sanitize:private-use"),
    ("plane-15 private use", "x\U000F0001y", "sanitize:private-use"),
    ("Mongolian vowel separator", "ᠠ\u180eᠡ", "sanitize:invisible-format"),
    ("interlinear annotation", "\ufff9hidden\ufffb", "sanitize:invisible-format"),
]


@pytest.mark.parametrize("name,text,label", _CARRIERS, ids=[c[0] for c in _CARRIERS])
def test_carrier_shapes_are_caught(name, text, label):
    assert label in _sanitize_labels(text), f"{name}: {_labels(sanitize(text).report)}"
    assert _OUTPUT_LABEL[label] in _output_labels(text), name


# --- one predicate, two surfaces ---------------------------------------------


@pytest.mark.parametrize(
    "text",
    [c[1] for c in _LEGITIMATE] + [c[1] for c in _CARRIERS],
    ids=[c[0] for c in _LEGITIMATE] + [c[0] for c in _CARRIERS],
)
def test_sanitize_and_the_output_gate_agree_per_class(text):
    # Anything sanitize strips, the output gate flags — class for class, count
    # for count. A split is how a carrier reaches a persisted artifact.
    stripped = {(f.label, f.count) for f in sanitize(text).report.findings
                if f.label in _OUTPUT_LABEL}
    from llm_ingestion_guard.output import _scan_invisible_carriers
    from llm_ingestion_guard.report import Source
    flagged = {(f.label, f.count) for f in _scan_invisible_carriers(text, Source.OUTPUT).findings}
    assert {(_OUTPUT_LABEL[lbl], n) for lbl, n in stripped} == flagged


# --- self-safety --------------------------------------------------------------


def test_document_level_rtl_context_stays_linear():
    # The RTL context is one lookup per document, not one per mark: a rescan
    # per mark would be quadratic in a document made of marks.
    text = HEBREW + (" x" + LRM) * 300_000
    assert scan_seconds(sanitize, text) < 2.0
    from llm_ingestion_guard.output import _scan_invisible_carriers
    from llm_ingestion_guard.report import Source
    assert scan_seconds(lambda t: _scan_invisible_carriers(t, Source.OUTPUT), text) < 2.0