"""Tests for carrier stripping (build order step 2). Core invariants (BRIEF §9): clean input returns byte-identical with an all-zero report; the sanitizer only ever *removes* — its output is always a subsequence of the input. """ import time from llm_ingestion_guard.sanitize import sanitize from llm_ingestion_guard.report import Severity, Source def _is_subsequence(sub: str, full: str) -> bool: it = iter(full) return all(ch in it for ch in sub) def test_clean_input_is_byte_identical(): text = "Hello, world. This is clean prose, with punctuation and a URL https://x.io/y." result = sanitize(text) assert result.text == text assert result.report.found is False def test_output_is_always_a_subsequence_of_input(): text = "a​bc data:text/plain;base64,QQ== d‮e" result = sanitize(text) assert _is_subsequence(result.text, text) assert len(result.text) <= len(text) def test_zero_width_removed_and_counted(): result = sanitize("ig​no​re") assert result.text == "ignore" zw = [f for f in result.report.findings if "zero-width" in f.label] assert len(zw) == 1 assert zw[0].count == 2 assert zw[0].source is Source.INPUT def test_bidi_override_removed(): result = sanitize("abc‮def") assert "‮" not in result.text assert any("bidi" in f.label for f in result.report.findings) def test_unicode_tag_removed_decoded_and_critical(): # Tag chars U+E0068 U+E0069 encode the hidden ASCII "hi". text = "visible" + chr(0xE0068) + chr(0xE0069) result = sanitize(text) assert result.text == "visible" tag = [f for f in result.report.findings if "unicode-tag" in f.label][0] assert tag.severity is Severity.CRITICAL assert tag.evidence is not None and "hi" in tag.evidence def test_html_comment_removed(): result = sanitize("beforeafter") assert result.text == "beforeafter" assert any("html-comment" in f.label for f in result.report.findings) def test_data_uri_removed(): result = sanitize("click data:text/html;base64,PHNjcmlwdD4= now") assert "data:text/html" not in result.text assert any("data-uri" in f.label for f in result.report.findings) def test_data_uri_does_not_match_inside_a_word(): # "metadata:" must not be mistaken for a data: URI. text = "the metadata: field is clean" result = sanitize(text) assert result.text == text assert result.report.found is False def test_output_source_is_respected(): result = sanitize("x​y", source=Source.OUTPUT) assert all(f.source is Source.OUTPUT for f in result.report.findings) # --- self-safety (OWASP LLM10): ReDoS on the comment stripper ---------------- # `sanitize` is step 1 of `prepare_input` -- the first thing every ingested # document hits -- and unlike `scan_lexicon`/`scan_output` it applies NO input # cap, so a quadratic run here has no ceiling at all. `` is a lazy run # in front of a REQUIRED literal: crafted input that repeats the opener and never # supplies `-->` makes every start position rescan the tail. Measured 20.1s at # 100_000 chars, exponent 1.96-2.14 over four doublings. Found by # docs/redos-sweep.py once it was generalised past the lexicon table. _REDOS_N = 100_000 def test_crafted_comment_payload_stays_bounded(): payload = ("" payload = (unit * (_REDOS_N // len(unit) + 1))[:_REDOS_N] start = time.monotonic() sanitize(payload) assert time.monotonic() - start < 0.5 def test_comment_stripping_survives_the_redos_fix(): # Recall parity for every comment shape the lazy regex used to handle: # nested markup, newlines (the pattern was DOTALL), and an unterminated # comment, which must be left alone rather than swallowed to end-of-input. assert sanitize("a z").text == "a z" assert sanitize("a z").text == "a z" assert sanitize("a b c").text == "a b c" assert sanitize("a b").text == "a --> b"