From 57a245a24a1e6b6a29455af651240e123acb4a45 Mon Sep 17 00:00:00 2001 From: rwagwani Date: Mon, 26 Oct 2026 11:02:01 +0000 Subject: [PATCH 1/4] fix(guardrail): fold disguised letters before to_ascii erases them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit is_safe ran to_ascii before any check, and to_ascii deletes every non-ASCII run. A fullwidth, accented or lookalike spelling of a blocklist entry (fullwidth "snow", "snów", "snw" with a Cyrillic o) therefore reached the matcher with letters missing and passed, while a reader still saw the entry. The passes now run first on the prompt as written, where censor_prompt folds code points (NFKC, combining marks removed) before matching. The to_ascii pass is kept after it, because some separators ("boston·dynamics") are only turned into a space by to_ascii. On non-ASCII prompts the passes also run on two more readings: invisible characters deleted rather than read as a space, and letters from other scripts that are drawn like Latin ones folded to Latin, using a vendored table generated from Unicode confusables.txt. Every pass that ran before still runs, so a prompt blocked before is still blocked. The list setup that __init__ did inline is moved into load_lists, shared with a new from_lists constructor, so the tests run the real entry point, is_safe, over the real bookkeeping instead of a copy of it. This is the Unicode part of the earlier combined change. The whitelist phrases and the list-maintenance tools move to their own pull requests. Co-Authored-By: Claude Opus 5.5 --- .../guardrail/blocklist/blocklist.py | 125 +++++++- .../guardrail/blocklist/blocklist_test.py | 271 ++++++++++++++++-- .../guardrail/blocklist/lookalikes.py | 225 +++++++++++++++ .../guardrail/blocklist/lookalikes_test.py | 71 +++++ 4 files changed, 669 insertions(+), 23 deletions(-) create mode 100644 cosmos_framework/auxiliary/guardrail/blocklist/lookalikes.py create mode 100644 cosmos_framework/auxiliary/guardrail/blocklist/lookalikes_test.py diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py index aa4f4e171..6b54e7612 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py @@ -10,6 +10,7 @@ import nltk from better_profanity import profanity +from cosmos_framework.auxiliary.guardrail.blocklist.lookalikes import fold_lookalikes from cosmos_framework.auxiliary.guardrail.blocklist.utils import read_keyword_list_from_dir, to_ascii from cosmos_framework.auxiliary.guardrail.common.core import ( GUARDRAIL1_CHECKPOINT, @@ -37,6 +38,14 @@ INVISIBLE_CHARS = re.compile(r"[­​-‏‪-‮⁠-⁤]") +class _NoLemmatizer: + """Leaves every token as it is. The default for Blocklist.from_lists.""" + + @staticmethod + def lemmatize(token: str) -> str: + return token + + class Blocklist(ContentSafetyGuardrail): def __init__( self, @@ -57,10 +66,49 @@ def __init__( self.guardrail_partial_match_min_chars = guardrail_partial_match_min_chars self.guardrail_partial_match_letter_count = guardrail_partial_match_letter_count - # Load blocklist and whitelist keywords - self.blocklist_words = read_keyword_list_from_dir(os.path.join(self.checkpoint_dir, "custom")) - self.whitelist_words = read_keyword_list_from_dir(os.path.join(self.checkpoint_dir, "whitelist")) - self.exact_match_words = read_keyword_list_from_dir(os.path.join(self.checkpoint_dir, "exact_match")) + self.load_lists( + blocklist_words=read_keyword_list_from_dir(os.path.join(self.checkpoint_dir, "custom")), + whitelist_words=read_keyword_list_from_dir(os.path.join(self.checkpoint_dir, "whitelist")), + exact_match_words=read_keyword_list_from_dir(os.path.join(self.checkpoint_dir, "exact_match")), + ) + + @classmethod + def from_lists( + cls, + blocklist_words: list[str], + whitelist_words: list[str] | None = None, + exact_match_words: list[str] | None = None, + lemmatizer=None, + guardrail_partial_match_min_chars: int = 6, + guardrail_partial_match_letter_count: float = 0.4, + ) -> "Blocklist": + """Build a Blocklist from keyword lists in memory, without the checkpoint. + + For the tests and the list-maintenance tools. It goes through load_lists + like __init__ does, so callers exercise the production bookkeeping rather + than a copy of it. The matcher is a fresh instance, not the module-level + singleton production shares. ``lemmatizer`` defaults to leaving tokens + unchanged; pass nltk's WordNetLemmatizer to match production exactly. + """ + from better_profanity.better_profanity import Profanity + + blocklist = cls.__new__(cls) + blocklist.profanity = Profanity() + blocklist.lemmatizer = lemmatizer if lemmatizer is not None else _NoLemmatizer() + blocklist.guardrail_partial_match_min_chars = guardrail_partial_match_min_chars + blocklist.guardrail_partial_match_letter_count = guardrail_partial_match_letter_count + blocklist.load_lists(blocklist_words, whitelist_words or [], exact_match_words or []) + return blocklist + + def load_lists(self, blocklist_words: list[str], whitelist_words: list[str], exact_match_words: list[str]) -> None: + """Load the keyword lists into the matcher. + + The single place that turns lists into matcher state, shared by __init__ + and from_lists. + """ + self.blocklist_words = blocklist_words + self.whitelist_words = whitelist_words + self.exact_match_words = exact_match_words self.profanity.load_censor_words(custom_words=self.blocklist_words, whitelist_words=self.whitelist_words) log.debug(f"Loaded {len(self.blocklist_words)} words/phrases from blocklist") @@ -86,9 +134,22 @@ def normalize_for_matching(prompt: str) -> str: The collapse also closes a plainer hole: a multi-word entry was defeated by typing two spaces between its words. + + The code-point folding is a separate method because is_safe also needs it + on its own: see alternate_spellings. """ + return Blocklist.fold_invisible_characters(Blocklist.normalize_code_points(prompt)) + + @staticmethod + def normalize_code_points(prompt: str) -> str: + """NFKC, combining marks removed, runs of whitespace collapsed.""" prompt = unicodedata.normalize("NFKC", prompt) prompt = "".join(c for c in unicodedata.normalize("NFKD", prompt) if not unicodedata.combining(c)) + return " ".join(prompt.split()) + + @staticmethod + def fold_invisible_characters(prompt: str) -> str: + """Rewrite zero-width and bidirectional characters to a space.""" return " ".join(INVISIBLE_CHARS.sub(" ", prompt).split()) def censor_prompt(self, input_prompt: str) -> tuple[bool, str]: @@ -226,9 +287,61 @@ def is_safe(self, input_prompt: str = "") -> tuple[bool, str]: # Check if the input is empty if not input_prompt: return False, "Input is empty" - input_prompt = to_ascii(input_prompt) - # Check full sentence for censored words + safe, message = self.check_passes(input_prompt) + if not safe or input_prompt.isascii(): + return safe, message + + # The passes run once more on each other way the prompt can be read. + # Like the raw pass, these only add checks: a prompt blocked as written + # is still blocked. + for spelling in self.alternate_spellings(input_prompt): + safe, message = self.check_passes(spelling) + if not safe: + return safe, message + return safe, message + + @classmethod + def alternate_spellings(cls, prompt: str) -> list[str]: + """Other readings of a non-ASCII prompt that the passes must also see. + + * Invisible characters deleted rather than turned into a space. The + space reading catches "snowwhite" as the entry "snow white", + but splits "snow white" into "sn ow white", which matches + nothing. Deleting them reads it as the entry. + * Letters from other scripts drawn like Latin ones ("snw", with + a Cyrillic o) folded to Latin. They survive NFKC, and to_ascii deletes + them, so every pass reads the word with a letter missing. NFKC runs + first, so a word mixing fullwidth and lookalike letters is recognized + as one disguised word. + + Each reading is returned only if it differs from the prompt. + """ + normalized = cls.normalize_code_points(prompt) + joined = " ".join(INVISIBLE_CHARS.sub("", normalized).split()) + spellings: list[str] = [] + for spelling in (joined, fold_lookalikes(normalized), fold_lookalikes(joined)): + if spelling != normalized and spelling not in spellings: + spellings.append(spelling) + return spellings + + def check_passes(self, input_prompt: str) -> tuple[bool, str]: + """Run every blocklist pass on one spelling of the prompt.""" + # An extra first pass sees the prompt as written. to_ascii rewrites every + # non-ASCII run to a space, which erases exactly the spellings + # censor_prompt was taught to fold: a fullwidth or accented letter is + # gone rather than normalized, so the word it spells never reaches the + # matcher and the prompt reads as clean. censor_prompt does its own + # code-point folding, so it needs the original text to fold. + censored, message = self.censor_prompt(input_prompt) + if censored: + return False, message + + # The ASCII pass is kept rather than replaced. Some separators survive the + # folding -- "boston·dynamics" stays one token on the raw pass -- and + # only to_ascii turns them into the space the entry needs. Adding a pass + # means nothing that was blocked before can pass now. + input_prompt = to_ascii(input_prompt) censored, message = self.censor_prompt(input_prompt) if censored: return False, message diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py index 87fd618b0..33348d0cb 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py @@ -47,20 +47,44 @@ def test_partial_match_with_threshold(): assert match is False +class _IdentityLemmatizer: + """Stand-in for WordNetLemmatizer, so a test needs no downloaded corpus. + + The lemmatizer belongs to the second pass, which re-checks the prompt with + words reduced to their stems. Nothing here tests lemmatization, and taking a + dependency on nltk data would make these tests fail in any environment + without the checkpoint's nltk_data directory. + """ + + @staticmethod + def lemmatize(token: str) -> str: + return token + + def _blocklist_with_words(words: list[str], whitelist: list[str] | None = None) -> Blocklist: """Build a Blocklist around a small word list, without downloading a checkpoint. - censor_prompt only needs the profanity matcher and the whitelist, so the - heavyweight __init__ (checkpoint download, nltk data) is bypassed here. + The heavyweight __init__ (checkpoint download, nltk data) is bypassed, but + from_lists loads the lists through the same code as __init__, so a test + exercises the real bookkeeping and can call the real entry point, is_safe. """ - from better_profanity.better_profanity import Profanity + return Blocklist.from_lists( + words, + whitelist, + lemmatizer=_IdentityLemmatizer(), + guardrail_partial_match_min_chars=100, + ) + - whitelist = whitelist or [] - bl = Blocklist.__new__(Blocklist) - bl.profanity = Profanity() - bl.profanity.load_censor_words(custom_words=words, whitelist_words=whitelist) - bl.whitelist_words = whitelist - return bl +def _blocked(bl: Blocklist, prompt: str) -> tuple[bool, str]: + """Run the real entry point, is_safe, and report whether it blocked. + + is_safe rather than censor_prompt: is_safe also runs to_ascii, a lemmatized + pass and the exact-match check, and a hole in any of them does not show up + when censor_prompt is called on its own. + """ + safe, message = bl.is_safe(prompt) + return not safe, message @pytest.mark.L1 @@ -276,9 +300,9 @@ def test_invisible_characters_do_not_split_a_blocked_phrase(name, prompt): """ bl = _blocklist_with_words(["snow white"]) - blocked, _ = bl.censor_prompt(prompt) + safe, _ = bl.is_safe(prompt) - assert blocked is True, f"{name} was not folded away" + assert safe is False, f"{name} was not folded away" @pytest.mark.L1 @@ -295,9 +319,95 @@ def test_unusual_spaces_do_not_split_a_blocked_phrase(name, prompt): """Any Unicode space between the words of an entry must still match.""" bl = _blocklist_with_words(["snow white"]) - blocked, _ = bl.censor_prompt(prompt) + safe, _ = bl.is_safe(prompt) + + assert safe is False, f"{name} was not folded away" + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("name", "prompt"), + [ + ("zero width space", "a sn\u200bow white poster"), + ("zero width joiner", "a snow wh\u200dite poster"), + ("soft hyphen", "a sn\u00adow white poster"), + ("several", "a s\u200bn\u200bo\u200bw white poster"), + ("with a lookalike", "a sn\u200b\u043ew white poster"), + ], +) +def test_invisible_characters_inside_one_word_of_a_phrase(name, prompt): + """An invisible character inside one word of a multi-word entry must not hide it. + + Turning it into a space reads "snow white" as "sn ow white", which + matches nothing, while a reader sees the entry. Deleting it is the other + reading, and both are checked. + """ + bl = _blocklist_with_words(["snow white"]) + + safe, _ = bl.is_safe(prompt) - assert blocked is True, f"{name} was not folded away" + assert safe is False, f"{name} let the entry through" + + +@pytest.mark.L1 +def test_invisible_characters_between_words_still_read_as_a_space(): + """Deleting invisible characters is an extra reading; the space reading stays.""" + bl = _blocklist_with_words(["snow white"]) + + safe, _ = bl.is_safe("a snow\u200bwhite poster") + + assert safe is False + + +@pytest.mark.L1 +def test_deleting_invisible_characters_does_not_block_ordinary_text(): + """Text that uses invisible characters legitimately must still come back safe.""" + bl = _blocklist_with_words(["snow white", "deskin"]) + + for prompt in ( + "a family emoji \U0001f468\u200d\U0001f469\u200d\U0001f467 on a desk", + "a long\u00adword with a soft hyphen", + "two\u200bwords joined by a zero width space", + ): + safe, message = bl.is_safe(prompt) + assert safe is True, f"{prompt!r} was wrongly reported as blocked: {message}" + + +class _PluralLemmatizer: + """Strips a trailing "s", the way WordNet reduces "dynamics" to "dynamic". + + With the identity stand-in, the lemmatized pass re-checks the ASCII prompt + unchanged and catches anything the first pass missed. An entry whose words + lemmatize to something else, such as "boston dynamics", gets no such rescue, + which is the case these tests need. + """ + + @staticmethod + def lemmatize(token: str) -> str: + return token[:-1] if token.endswith("s") else token + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("name", "prompt"), + [ + ("middle dot", "a boston·dynamics robot"), + ("bullet", "a boston•dynamics robot"), + ("arrow", "a boston→dynamics robot"), + ], +) +def test_separators_the_folding_keeps_still_split_a_blocked_phrase(name, prompt): + """A separator NFKC leaves alone reaches the raw pass as part of one token. + + Only to_ascii turns it into the space the entry needs, so the ASCII pass has + to run as well as the raw one, not instead of it. + """ + bl = _blocklist_with_words(["boston dynamics"]) + bl.lemmatizer = _PluralLemmatizer() + + safe, _ = bl.is_safe(prompt) + + assert safe is False, f"{name} separator let the entry through" @pytest.mark.L1 @@ -305,9 +415,9 @@ def test_fullwidth_letters_are_folded_to_ascii(): """Fullwidth forms are drawn as the ASCII letters they normalize to.""" bl = _blocklist_with_words(["snow white"]) - blocked, _ = bl.censor_prompt("a ｓｎｏｗ white poster") + safe, _ = bl.is_safe("a ｓｎｏｗ white poster") - assert blocked is True + assert safe is False @pytest.mark.L1 @@ -316,8 +426,8 @@ def test_combining_marks_do_not_make_a_new_word(): bl = _blocklist_with_words(["snow white"]) for prompt in ("a snów white poster", unicodedata.normalize("NFD", "a snów white poster")): - blocked, _ = bl.censor_prompt(prompt) - assert blocked is True, f"{prompt!r} was not folded away" + safe, _ = bl.is_safe(prompt) + assert safe is False, f"{prompt!r} was not folded away" @pytest.mark.L1 @@ -336,6 +446,133 @@ def test_normalization_does_not_block_ordinary_text(): assert blocked is False, f"{prompt!r} was wrongly reported as blocked: {message}" +@pytest.mark.L1 +def test_first_pass_on_raw_non_ascii_does_not_block_ordinary_text(monkeypatch): + """The first pass now sees the prompt before to_ascii, so pin what that means. + + Running censor_prompt ahead of to_ascii is what lets the matcher fold a + fullwidth or accented spelling. The cost is that non-ASCII text no longer + arrives flattened to spaces, so ordinary prose in another script -- or with + an accent, an em dash, a CJK sentence -- reaches the matcher as written and + must still come back safe. + + word_tokenize is stubbed to a split so the test does not depend on the + nltk data that ships in the checkpoint directory; the second pass is not + what is under test here. + """ + monkeypatch.setattr( + "cosmos_framework.auxiliary.guardrail.blocklist.blocklist.nltk.word_tokenize", + str.split, + ) + bl = _blocklist_with_words(["snow white"], whitelist=["flat"]) + + for prompt in ( + "café scene with Élodie", + "an em dash — and an arrow → in the text", + "こんにちは from the model", + "a robot on a flat desk", + "Grüße aus München", + ): + safe, message = bl.is_safe(prompt) + assert safe is True, f"{prompt!r} was wrongly reported as blocked: {message}" + + +# Written as escapes: the point of these letters is that they cannot be told +# apart from Latin ones on screen. +_CYRILLIC_A = "\u0430" +_CYRILLIC_O = "\u043e" +_GREEK_O = "\u03bf" + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("name", "entry", "prompt"), + [ + ("cyrillic o", "snow white", f"a sn{_CYRILLIC_O}w white poster"), + ("greek o", "snow white", f"a sn{_GREEK_O}w white poster"), + ("capital cyrillic o", "snow white", "a SN\u041eW WHITE poster"), + ("single-word entry", "batman", f"a b{_CYRILLIC_A}tman toy"), + ("every letter swapped", "pace", "a \u0440\u0430\u0441\u0435 car"), + ("mixed with fullwidth", "snow white", f"a \uff53\uff4e{_CYRILLIC_O}\uff57 white poster"), + ("mixed with an invisible split", "batman", f"a b\u200b{_CYRILLIC_A}tman toy"), + ], +) +def test_lookalike_letters_do_not_spell_a_new_word(name, entry, prompt): + """A letter from another script drawn like a Latin one must not hide the entry. + + NFKC leaves these letters alone and to_ascii deletes them, so every other + pass reads the word with a letter missing. + """ + bl = _blocklist_with_words([entry]) + + safe, _ = bl.is_safe(prompt) + + assert safe is False, f"{name} let the entry through" + + +@pytest.mark.L1 +def test_lookalike_folding_leaves_real_words_alone(): + """A word with letters that have no Latin twin is real text, not a disguise. + + The Russian word for "pine" starts with three lookalikes (c, o, c) and then + has letters with no Latin twin. Folding the lookalikes anyway, and letting + to_ascii drop the rest, leaves Latin fragments that fuse into the entry. + """ + bl = _blocklist_with_words(["coca"]) + + for prompt in ("\u0441\u043e\u0441\u043d\u0430 \u0443 \u043e\u043a\u043d\u0430", "\u0441\u043e\u0441\u043d\u0430"): + safe, message = bl.is_safe(prompt) + assert safe is True, f"{prompt!r} was wrongly reported as blocked: {message}" + + +@pytest.mark.L1 +def test_ascii_prompt_skips_the_lookalike_pass(monkeypatch): + """English prompts are untouched: the lookalike pass never runs on ASCII.""" + + def fail(prompt): + raise AssertionError(f"lookalike pass ran on {prompt!r}") + + monkeypatch.setattr("cosmos_framework.auxiliary.guardrail.blocklist.blocklist.fold_lookalikes", fail) + bl = _blocklist_with_words(["snow white"]) + + assert bl.is_safe("a robot sitting at a table")[0] is True + assert bl.is_safe("a snow white poster")[0] is False + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("name", "prompt"), + [ + ("arrow", "a snow→white poster"), + ("emoji", "a snow🙂white poster"), + ("ellipsis", "a snow…white poster"), + ("middle dot", "a snow·white poster"), + ], +) +def test_non_ascii_separator_is_caught_by_the_flattened_pass(monkeypatch, name, prompt): + """The two passes cover different evasions, and both are needed. + + A visible non-ASCII character between the words of an entry is not an + invisible mark, so normalize_for_matching leaves it in place and the first + pass -- which now sees raw text -- does not match. to_ascii rewrites it to a + space, so the second pass does. The reverse holds for a fullwidth or + accented spelling, which only the first pass can fold. + + Neither pass is redundant: dropping either one opens the family the other + covers. This test fails if the flattened pass is ever removed or moved + ahead of the first. + """ + monkeypatch.setattr( + "cosmos_framework.auxiliary.guardrail.blocklist.blocklist.nltk.word_tokenize", + str.split, + ) + bl = _blocklist_with_words(["snow white"]) + + safe, _ = bl.is_safe(prompt) + + assert safe is False, f"{name} separator was not caught by either pass" + + @pytest.mark.L1 def test_normalization_preserves_the_earlier_sentinel_and_markdown_behaviour(): """Folding runs after the sentinel strip and must not resurrect the "*" bug.""" diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/lookalikes.py b/cosmos_framework/auxiliary/guardrail/blocklist/lookalikes.py new file mode 100644 index 000000000..1501fb5fa --- /dev/null +++ b/cosmos_framework/auxiliary/guardrail/blocklist/lookalikes.py @@ -0,0 +1,225 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +"""Fold letters from other scripts that are drawn like Latin letters. + +A Cyrillic o (U+043E) and a Latin o render identically, but they are +different code points, and compatibility normalization (NFKC) leaves the +Cyrillic one alone. "snow white" typed with that one Cyrillic letter reads +as the entry to a human while reaching the matcher as a word it does not know; +to_ascii then deletes the letter outright, leaving "snw". + +The table below comes from the Unicode confusables data (UTS #39, +confusables.txt, version 18.0.0): every single code point whose skeleton is one +Latin letter, minus the ones NFKC already folds. It is vendored rather than +fetched so the guardrail has no new dependency; regenerate it with + + python -m cosmos_framework.auxiliary.guardrail.blocklist.lookalikes confusables.txt + +Folding every listed character everywhere would also rewrite real Russian, +Greek or Ukrainian text into Latin fragments that can fuse into an entry, so a +word is folded only when it looks like a disguised Latin word: it contains a +Latin letter, or every letter in it is a lookalike. The Russian word for +"pine" contains U+043D, which has no Latin twin, so it is left alone. +""" + +import sys +import unicodedata +from collections.abc import Iterable + + +def _expand(by_letter: dict[str, str]) -> dict[str, str]: + return {char: letter for letter, chars in by_letter.items() for char in chars} + + +# Generated by format_table from confusables.txt 18.0.0; do not edit by hand. +LOOKALIKES: dict[str, str] = _expand( + { + "a": ( + "\u0251\u0391\u03b1\u0410\u0430\u13aa\u15c5\u237a\ua4ee\uab64\U000102a0\U00016f40\U0001ccd6" + "\U0001d6a8\U0001d6c2\U0001d6e2\U0001d6fc\U0001d71c\U0001d736\U0001d756\U0001d770\U0001d790" + "\U0001d7aa\U0001df5a\U0001df6a" + ), + "b": ( + "\u0184\u0392\u0412\u042c\u07d5\u13cf\u13f4\u1472\u15af\u15f7\u2c82\ua4d0\ua557\ua7b4\U00010282" + "\U000102a1\U00010301\U0001031c\U00016eb6\U0001ccd7\U0001d207\U0001d6a9\U0001d6e3\U0001d71d" + "\U0001d757\U0001d791" + ), + "c": ( + "\u03f2\u03f9\u0421\u0441\u1004\u105a\u13df\u1c83\u1d04\u2ca4\u2ca5\ua4da\uabaf\U000102a2" + "\U00010302\U00010415\U0001043d\U0001051b\U000118e9\U000118f2\U0001ccd8\U0001d134\U0001f74c" + ), + "d": "\u0501\u13a0\u13e7\u146f\u15de\u15ea\ua4d2\ua4d3\U0001018b\U0001ccd9", + "e": ( + "\u0395\u0415\u0435\u04bd\u13ac\u212e\u22ff\u2d39\ua4f0\ua5cb\uab32\U00010286\U000118a6\U000118ae" + "\U0001ccda\U0001d6ac\U0001d6e6\U0001d720\U0001d75a\U0001d794\U0001df81" + ), + "f": ( + "\u017f\u0192\u0284\u03dc\u0584\u07d3\u15b4\u1e9d\ua4dd\ua798\ua799\uab35\U0001017e\U00010287" + "\U000102a5\U00010525\U000118a2\U000118c2\U0001ccdb\U0001d213\U0001d7ca" + ), + "g": "\u018d\u0261\u050c\u0581\u13c0\u13f3\u1d83\ua4d6\U0001ccdc", + "h": ( + "\u0397\u041d\u04ba\u04bb\u0570\u10b9\u13bb\u13c2\u157c\u2c8e\ua4e7\U000102cf\U0001ccdd\U0001d6ae" + "\U0001d6e8\U0001d722\U0001d75c\U0001d796" + ), + "i": ( + "\xa1\u0131\u0269\u026a\u03b9\u0456\u0582\u13a5\u2373\u2c93\ua647\uab75\U000118c3\U0001d6a4" + "\U0001d6ca\U0001d704\U0001d73e\U0001d778\U0001d7b2" + ), + "j": "\u0237\u037f\u03f3\u0408\u0458\u0575\u13ab\u148d\ua4d9\ua7b2\U0001ccdf\U0001d6a5", + "k": ( + "\u039a\u041a\u13e6\u16d5\u2c94\ua4d7\U00010518\U0001cce0\U0001d6b1\U0001d6eb\U0001d725\U0001d75f\U0001d799" + ), + "l": ( + "\u0196\u01c0\u0399\u0406\u04c0\u04cf\u05c0\u05d5\u05df\u0627\u0661\u06f1\u07ca\u0964\u104a\u13de" + "\u14aa\u16c1\u16d0\u2110\u2111\u2160\u21bf\u2223\u23fd\u2502\u2503\u2c92\u2cd0\u2d4a\u2d4f\ua4e1" + "\ua4f2\ua56f\ua781\ua7ae\ua7fe\ua830\ua8ce\uaa5d\ufe31\ufe8d\ufe8e\uff11\uff29\uff5c\uffe8" + "\U0001028a\U00010309\U00010320\U0001041b\U0001050e\U00010526\U00010926\U00010c3e\U00010ca5" + "\U00010cfa\U00011047\U000110c0\U00011141\U000111c5\U000113d4\U0001144b\U000115c5\U00011641" + "\U000118a3\U000118b2\U00011c41\U00011dda\U00011de1\U00016d63\U00016eaa\U00016f16\U00016f28" + "\U0001ccde\U0001cce1\U0001ccf1\U0001d100\U0001d22a\U0001d377\U0001d408\U0001d43c\U0001d470" + "\U0001d4d8\U0001d540\U0001d574\U0001d5a8\U0001d5dc\U0001d610\U0001d644\U0001d678\U0001d6b0" + "\U0001d6ea\U0001d724\U0001d75e\U0001d798\U0001d7cf\U0001d7d9\U0001d7e3\U0001d7ed\U0001d7f7" + "\U0001e141\U0001e8c7\U0001ed01\U0001ee00\U0001ee80\U0001fbf1" + ), + "m": ( + "\u039c\u03fa\u041c\u13b7\u15f0\u16d6\u2c98\ua4df\U000102b0\U00010311\U00010c21\U0001cce2" + "\U0001d6b3\U0001d6ed\U0001d727\U0001d761\U0001d79b" + ), + "n": ( + "\u039d\u0578\u057c\u2c9a\ua4e0\U00010513\U00011abe\U0001cce3\U0001d6b4\U0001d6ee\U0001d728" + "\U0001d762\U0001d79c" + ), + "o": ( + "\u039f\u03bf\u03c3\u03ed\u041e\u043e\u0555\u0585\u05e1\u0647\u0665\u06be\u06c1\u06d5\u06f5\u07c0" + "\u07cb\u0840\u0966\u09e6\u0a66\u0ae6\u0b20\u0b66\u0be6\u0c02\u0c66\u0c82\u0ce6\u0d02\u0d20\u0d66" + "\u0d82\u0e50\u0ed0\u101d\u1040\u10ff\u110b\u11bc\u12d0\u17e0\u1a45\u1a80\u1a90\u1bea\u1c82\u1cbf" + "\u1d0f\u1d11\u2c9e\u2c9f\u2d54\u3007\u3147\ua4f3\uab3d\ufba6\ufba7\ufba8\ufba9\ufbaa\ufbab\ufbac" + "\ufbad\ufee9\ufeea\ufeeb\ufeec\uff10\uffb7\U00010292\U000102ab\U0001030f\U00010404\U0001042c" + "\U000104c2\U000104ea\U00010516\U0001092c\U00010c17\U00010d07\U00011124\U00011302\U000114d0" + "\U000118b5\U000118c8\U000118d7\U000118e0\U00011de0\U00016ae9\U0001cce4\U0001ccf0\U0001d6b6" + "\U0001d6d0\U0001d6d4\U0001d6f0\U0001d70a\U0001d70e\U0001d72a\U0001d744\U0001d748\U0001d764" + "\U0001d77e\U0001d782\U0001d79e\U0001d7b8\U0001d7bc\U0001d7ce\U0001d7d8\U0001d7e2\U0001d7ec" + "\U0001d7f6\U0001e140\U0001e2f0\U0001ee24\U0001ee84\U0001fbf0" + ), + "p": ( + "\xfe\u01bf\u03a1\u03c1\u03f1\u03f8\u0420\u0440\u13e2\u146d\u2374\u2ca2\u2ca3\u2cce\u2ccf\ua4d1" + "\U00010295\U0001cce5\U0001d6b8\U0001d6d2\U0001d6e0\U0001d6f2\U0001d70c\U0001d71a\U0001d72c" + "\U0001d746\U0001d754\U0001d766\U0001d780\U0001d78e\U0001d7a0\U0001d7ba\U0001d7c8" + ), + "q": "\u051a\u051b\u0563\u0566\u2d55\U0001cce6", + "r": ( + "\u01a6\u024c\u0433\u13a1\u13d2\u1587\u1d26\u2c85\ua4e3\uab47\uab48\uab81\U000104b4\U00016a19" + "\U00016f35\U0001cce7\U0001d216" + ), + "s": ( + "\u01bd\u0405\u0455\u054f\u0d1f\u10bd\u10fd\u13d5\u13da\u1cbd\ua4e2\ua576\ua731\uabaa\U00010296" + "\U00010420\U00010448\U000118c1\U00016ad6\U00016f3a\U0001cce8" + ), + "t": ( + "\u03a4\u0422\u07e0\u13a2\u22a4\u27d9\u2ca6\u3112\u4e05\ua4d4\ua50b\U00010297\U000102b1\U00010315" + "\U000118bc\U00016f0a\U0001cce9\U0001d373\U0001d6bb\U0001d6f5\U0001d72f\U0001d769\U0001d7a3" + "\U0001f768" + ), + "u": ( + "\u028b\u03c5\u054d\u057d\u1200\u144c\u1d1c\u222a\u22c3\ua4f4\ua79f\uab4e\uab52\U000104ce" + "\U000104f6\U000118b8\U000118d8\U00016f42\U0001ccea\U0001d6d6\U0001d710\U0001d74a\U0001d784" + "\U0001d7be" + ), + "v": ( + "\u03bd\u0474\u0475\u05d8\u0667\u06f7\u13d9\u142f\u1d20\u2228\u22c1\u2d38\ua4e6\ua6df\uaba9" + "\U0001051d\U00010c1f\U00011706\U000118a0\U000118c0\U00016f08\U0001cceb\U0001cefc\U0001d20d" + "\U0001d6ce\U0001d708\U0001d742\U0001d77c\U0001d7b6\U0001e145" + ), + "w": ( + "\u026f\u0448\u0461\u051c\u051d\u0561\u13b3\u13d4\u1d21\u29e2\u2cbd\ua4ea\ua7fa\uab83\uaba4" + "\U0001170a\U0001170e\U0001170f\U000118e6\U000118ef\U0001ccec\U0001d20c\U0001df7d" + ), + "x": ( + "\xd7\u03a7\u0425\u0445\u1541\u157d\u166d\u166e\u16b7\u1763\u1bec\u1cf5\u2573\u292b\u292c\u2a2f" + "\u2cac\u2d5d\ua4eb\ua7b3\U00010290\U000102b4\U00010317\U00010322\U00010527\U00010c13\U00010c82" + "\U00010cc2\U00010cfc\U000118ec\U0001cced\U0001d6be\U0001d6f8\U0001d732\U0001d76c\U0001d7a6" + ), + "y": ( + "\u0263\u028f\u03a5\u03b3\u03d2\u0423\u0443\u04ae\u04af\u07cc\u10e7\u13a9\u13bd\u1d8c\u1eff\u213d" + "\u2ca8\u2ca9\u311a\u4e2b\ua4ec\uab5a\U000102b2\U00010c20\U000118a4\U000118c4\U000118dc\U00016f43" + "\U0001ccee\U0001d6bc\U0001d6c4\U0001d6f6\U0001d6fe\U0001d730\U0001d738\U0001d76a\U0001d772" + "\U0001d7a4\U0001d7ac" + ), + "z": ( + "\u0396\u10cd\u13c3\u1d22\u2c6b\u2c6c\u2c8c\u2c8d\u2d2d\ua4dc\ua6c9\uab93\U000102f5\U00010507" + "\U000118a9\U000118e5\U00011abc\U0001ccef\U0001d6ad\U0001d6e7\U0001d721\U0001d75b\U0001d795" + ), + } +) + + +def table_from_confusables(lines: Iterable[str]) -> dict[str, str]: + """Build the lookalike table from the lines of confusables.txt.""" + table: dict[str, str] = {} + for line in lines: + line = line.split("#", 1)[0].strip() + if not line: + continue + source, target, _ = (field.strip() for field in line.split(";")) + source = "".join(chr(int(code, 16)) for code in source.split()) + target = "".join(chr(int(code, 16)) for code in target.split()).lower() + if len(source) != 1 or source.isascii(): + continue + if len(target) != 1 or not (target.isascii() and target.isalpha()): + continue + if unicodedata.normalize("NFKC", source).lower() == target: + continue + table[source] = target + return table + + +def _is_disguised(word: str) -> bool: + """Whether a word looks like a Latin word spelled with lookalikes.""" + if not any(char in LOOKALIKES for char in word): + return False + if any(char.isascii() and char.isalpha() for char in word): + return True + return all(char in LOOKALIKES for char in word if char.isalpha()) + + +def fold_lookalikes(prompt: str) -> str: + """Rewrite lookalike letters to Latin, in the words that look disguised. + + Returns the prompt unchanged when it holds no lookalike, which includes + every ASCII prompt. + """ + if prompt.isascii(): + return prompt + return " ".join( + "".join(LOOKALIKES.get(char, char) for char in word) if _is_disguised(word) else word for word in prompt.split() + ) + + +def format_table(table: dict[str, str], width: int = 96) -> str: + """Render a table as the LOOKALIKES literal above, escaped and grouped by letter.""" + by_letter: dict[str, list[str]] = {} + for char, letter in sorted(table.items()): + by_letter.setdefault(letter, []).append(char) + lines = ["LOOKALIKES: dict[str, str] = _expand(", " {"] + for letter, chars in sorted(by_letter.items()): + chunks = [""] + for char in chars: + escaped = char.encode("unicode_escape").decode("ascii") + if len(chunks[-1]) + len(escaped) > width: + chunks.append("") + chunks[-1] += escaped + if len(chunks) == 1: + lines.append(f' "{letter}": "{chunks[0]}",') + else: + lines.append(f' "{letter}": (') + lines.extend(f' "{chunk}"' for chunk in chunks) + lines.append(" ),") + lines += [" }", ")"] + return "\n".join(lines) + + +if __name__ == "__main__": + with open(sys.argv[1], encoding="utf-8-sig") as source_file: + print(format_table(table_from_confusables(source_file))) diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/lookalikes_test.py b/cosmos_framework/auxiliary/guardrail/blocklist/lookalikes_test.py new file mode 100644 index 000000000..39c4deeb1 --- /dev/null +++ b/cosmos_framework/auxiliary/guardrail/blocklist/lookalikes_test.py @@ -0,0 +1,71 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +import unicodedata + +import pytest + +from cosmos_framework.auxiliary.guardrail.blocklist.lookalikes import ( + LOOKALIKES, + fold_lookalikes, + format_table, + table_from_confusables, +) + +_CONFUSABLES_SAMPLE = """\ +# confusables.txt excerpt; fields are source ; target ; type +043E ;\t006F ;\tMA\t# ( о → o ) CYRILLIC SMALL LETTER O → LATIN SMALL LETTER O +0410 ;\t0041 ;\tMA\t# ( А → A ) CYRILLIC CAPITAL LETTER A → LATIN CAPITAL LETTER A +FF4F ;\t006F ;\tMA\t# ( ｏ → o ) FULLWIDTH LATIN SMALL LETTER O → LATIN SMALL LETTER O +0031 ;\t006C ;\tMA\t# ( 1 → l ) DIGIT ONE → LATIN SMALL LETTER L +0448 ;\t0077 ;\tMA\t# ( ш → w ) CYRILLIC SMALL LETTER SHA → LATIN SMALL LETTER W +0444 ;\t0278 ;\tMA\t# ( ф → ɸ ) CYRILLIC SMALL LETTER EF → LATIN SMALL LETTER PHI +0433 0335 ;\t0072 0335 ;\tMA\t# two code points on each side +""" + + +@pytest.mark.L1 +def test_table_from_confusables_keeps_single_letters_nfkc_misses(): + """Only one code point onto one Latin letter, and only where NFKC does not already fold it.""" + table = table_from_confusables(_CONFUSABLES_SAMPLE.splitlines()) + + assert table == {"о": "o", "А": "a", "ш": "w"} + + +@pytest.mark.L1 +def test_format_table_round_trips(): + """The vendored literal is what format_table prints, so regenerating it is exact.""" + namespace: dict = {} + exec(format_table(LOOKALIKES), {"_expand": lambda d: {c: k for k, v in d.items() for c in v}}, namespace) + + assert namespace["LOOKALIKES"] == LOOKALIKES + + +@pytest.mark.L1 +def test_vendored_table_shape(): + """Every key is one non-ASCII code point NFKC leaves alone; every value is a Latin letter.""" + assert len(LOOKALIKES) > 700 + for char, letter in LOOKALIKES.items(): + assert len(char) == 1 and not char.isascii(), repr(char) + assert letter in "abcdefghijklmnopqrstuvwxyz", repr(letter) + assert unicodedata.normalize("NFKC", char).lower() != letter, repr(char) + assert LOOKALIKES["о"] == "o" + assert LOOKALIKES["ο"] == "o" + assert LOOKALIKES["а"] == "a" + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("prompt", "expected"), + [ + ("plain ascii text", "plain ascii text"), + ("a snоw poster", "a snow poster"), + ("расе", "pace"), + ("сосна", "сосна"), + ("café こんにちは", "café こんにちは"), + ("snоw сосна", "snow сосна"), + ], +) +def test_fold_lookalikes_folds_only_disguised_words(prompt, expected): + """A word is folded when it mixes in Latin letters or is made only of lookalikes.""" + assert fold_lookalikes(prompt) == expected From 3b004d75f57e8d83b82755cadff9aa8840d62c97 Mon Sep 17 00:00:00 2001 From: rwagwani Date: Mon, 26 Oct 2026 11:02:52 +0000 Subject: [PATCH 2/4] fix(guardrail): run the exact-match check on the folded reading too Review on #306: censor_prompt folds fullwidth and accented letters itself, but the lemmatized pass and the exact-match check only see the prompt after to_ascii, which deletes those letters. alternate_spellings left out the plain folded reading, so an entry that is only on the exact-match list still passed when spelled in fullwidth or with an accent. alternate_spellings now includes the code-point-folded reading, so the whole chain of passes, exact-match included, runs on it. Tests cover fullwidth, precomposed and decomposed accents, lookalike and invisible spellings of an exact-match-only entry, and ordinary text that must stay allowed. Without the change, the fullwidth and both accent cases fail. Co-Authored-By: Claude Opus 5.5 --- .../guardrail/blocklist/blocklist.py | 9 ++- .../guardrail/blocklist/blocklist_test.py | 62 ++++++++++++++++++- 2 files changed, 68 insertions(+), 3 deletions(-) diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py index 6b54e7612..8ebf6dadc 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py @@ -305,6 +305,11 @@ def is_safe(self, input_prompt: str = "") -> tuple[bool, str]: def alternate_spellings(cls, prompt: str) -> list[str]: """Other readings of a non-ASCII prompt that the passes must also see. + * The code points folded (NFKC, combining marks removed). censor_prompt + folds them itself, but the lemmatized pass and the exact-match check + only see the prompt after to_ascii, which deletes a fullwidth or + accented letter rather than folding it. Without this reading an entry + that is only on the exact-match list is missed when spelled that way. * Invisible characters deleted rather than turned into a space. The space reading catches "snowwhite" as the entry "snow white", but splits "snow white" into "sn ow white", which matches @@ -320,8 +325,8 @@ def alternate_spellings(cls, prompt: str) -> list[str]: normalized = cls.normalize_code_points(prompt) joined = " ".join(INVISIBLE_CHARS.sub("", normalized).split()) spellings: list[str] = [] - for spelling in (joined, fold_lookalikes(normalized), fold_lookalikes(joined)): - if spelling != normalized and spelling not in spellings: + for spelling in (normalized, joined, fold_lookalikes(normalized), fold_lookalikes(joined)): + if spelling != prompt and spelling not in spellings: spellings.append(spelling) return spellings diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py index 33348d0cb..791ba7497 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py @@ -61,7 +61,9 @@ def lemmatize(token: str) -> str: return token -def _blocklist_with_words(words: list[str], whitelist: list[str] | None = None) -> Blocklist: +def _blocklist_with_words( + words: list[str], whitelist: list[str] | None = None, exact_match: list[str] | None = None +) -> Blocklist: """Build a Blocklist around a small word list, without downloading a checkpoint. The heavyweight __init__ (checkpoint download, nltk data) is bypassed, but @@ -71,6 +73,7 @@ def _blocklist_with_words(words: list[str], whitelist: list[str] | None = None) return Blocklist.from_lists( words, whitelist, + exact_match, lemmatizer=_IdentityLemmatizer(), guardrail_partial_match_min_chars=100, ) @@ -584,3 +587,60 @@ def test_normalization_preserves_the_earlier_sentinel_and_markdown_behaviour(): blocked, message = bl.censor_prompt("a bad\x00word in the text") assert blocked is True assert "badword" not in message + + +def _fullwidth(text: str) -> str: + """Lowercase ASCII letters rewritten as their fullwidth presentation forms.""" + return "".join(chr(ord(c) - 0x61 + 0xFF41) if "a" <= c <= "z" else c for c in text) + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("name", "prompt"), + [ + ("verbatim", "a chromebook on the desk"), + ("fullwidth", f"a {_fullwidth('chromebook')} on the desk"), + ("accent", "a chromébook on the desk"), + ("decomposed accent", unicodedata.normalize("NFD", "a chromébook on the desk")), + ("lookalike", f"a chr{_CYRILLIC_O}mebook on the desk"), + ("invisible split", "a chrome\u200bbook on the desk"), + ], +) +def test_exact_match_entry_is_folded_like_the_other_lists(monkeypatch, name, prompt): + """An entry only on the exact-match list is caught however it is spelled. + + The exact-match check runs on the prompt after to_ascii, which deletes a + fullwidth or accented letter instead of folding it, so the folded reading + has to go through the whole chain of passes, not only censor_prompt. + The custom list holds an unrelated entry, so only the exact-match check can + block these prompts. + """ + monkeypatch.setattr( + "cosmos_framework.auxiliary.guardrail.blocklist.blocklist.nltk.word_tokenize", + str.split, + ) + bl = _blocklist_with_words(["deskin"], exact_match=["chromebook"]) + + safe, message = bl.is_safe(prompt) + + assert safe is False, f"{name} spelling of an exact-match entry was let through" + assert "Exact Match Word" in message + + +@pytest.mark.L1 +def test_exact_match_folding_does_not_block_ordinary_text(monkeypatch): + """The folded reading must not invent exact matches in clean text.""" + monkeypatch.setattr( + "cosmos_framework.auxiliary.guardrail.blocklist.blocklist.nltk.word_tokenize", + str.split, + ) + bl = _blocklist_with_words(["deskin"], exact_match=["chromebook"]) + + for prompt in ( + "a chrome bowl on the desk", + "café scene with Élodie", + f"a {_fullwidth('laptop')} on the desk", + "\u0441\u043e\u0441\u043d\u0430 \u0443 \u043e\u043a\u043d\u0430", + ): + safe, message = bl.is_safe(prompt) + assert safe is True, f"{prompt!r} was wrongly reported as blocked: {message}" From 63d5b3185292a64b4a8234335b116936369eef45 Mon Sep 17 00:00:00 2001 From: rwagwani Date: Mon, 26 Oct 2026 11:13:46 +0000 Subject: [PATCH 3/4] fix(guardrail): let the whitelist hold phrases, so prose stops matching entries The matcher fuses neighbouring words, which is what catches "shot gun" as the entry "shotgun", but it also turns ordinary prose into an entry: "a robot sitting at a desk in the background" is blocked as "deskin". The matcher's whitelist only holds single tokens, and whitelisting either half cannot help, because fusing does not consult the whitelist. A whitelist entry with a space in it is now a phrase. A prompt is cleared only if every match lies entirely inside a whitelisted phrase. Matches that straddle a phrase boundary, and matches elsewhere in the prompt, stay blocked. Phrases are located before invisible characters are folded, so "deskin", which renders as the entry, cannot reach the exemption. With no phrase in the prompt the result and message are exactly what they were. The matches are found by trying every span on its own against the matcher, not reconstructed from the censored output. Review on #306 showed why: the censored output collapses each match to one token, so two adjacent matches ("nice camera" + "shot gun") were split from the token count alone as "nice camera shot" + "gun", both inside whitelisted phrases, and the prompt was allowed. The matcher is also greedy, so a match sharing a token with an earlier one ("in gun" after "desk in") never appeared in the output at all. Also logs a warning at load time when a whitelisted word is censored anyway as the leetspeak variant of another entry ("flat" from "fiat"), since that whitelist entry then does nothing. Moved here from #306, so the Unicode fix there can land on its own. Co-Authored-By: Claude Opus 5.5 --- .../guardrail/blocklist/blocklist.py | 161 +++++++++++- .../guardrail/blocklist/blocklist_test.py | 237 ++++++++++++++++++ 2 files changed, 387 insertions(+), 11 deletions(-) diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py index 8ebf6dadc..d45fdf652 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py @@ -4,11 +4,13 @@ import argparse import os import re +import string import unicodedata from difflib import SequenceMatcher import nltk from better_profanity import profanity +from better_profanity.constants import ALLOWED_CHARACTERS from cosmos_framework.auxiliary.guardrail.blocklist.lookalikes import fold_lookalikes from cosmos_framework.auxiliary.guardrail.blocklist.utils import read_keyword_list_from_dir, to_ascii @@ -37,6 +39,13 @@ # while reaching the matcher as a token it does not recognise. INVISIBLE_CHARS = re.compile(r"[­​-‏‪-‮⁠-⁤]") +# Punctuation trimmed from a token before it is compared with a whitelisted +# phrase, so "a desk in, the corner" still matches the phrase "desk in". The +# characters better_profanity treats as part of a word (" $ ' * @) are kept: +# they are the ones its leetspeak mapping substitutes, so stripping them would +# let "de$k in" read as the whitelisted phrase. +_STRIP_CHARS = "".join(sorted(set(string.punctuation) - ALLOWED_CHARACTERS)) + class _NoLemmatizer: """Leaves every token as it is. The default for Blocklist.from_lists.""" @@ -110,11 +119,30 @@ def load_lists(self, blocklist_words: list[str], whitelist_words: list[str], exa self.whitelist_words = whitelist_words self.exact_match_words = exact_match_words - self.profanity.load_censor_words(custom_words=self.blocklist_words, whitelist_words=self.whitelist_words) + # The matcher's whitelist is a set of single tokens, so a whitelisted + # phrase never reaches it. Phrases are handled in censor_prompt instead, + # by exempting the matches that fall inside one. + self.whitelist_phrases = sorted((w for w in self.whitelist_words if " " in w.strip()), key=len, reverse=True) + whitelist_tokens = [w for w in self.whitelist_words if " " not in w.strip()] + + self.profanity.load_censor_words(custom_words=self.blocklist_words, whitelist_words=whitelist_tokens) log.debug(f"Loaded {len(self.blocklist_words)} words/phrases from blocklist") - log.debug(f"Whitelisted {len(self.whitelist_words)} words/phrases from whitelist") + log.debug(f"Whitelisted {len(whitelist_tokens)} words and {len(self.whitelist_phrases)} phrases") log.debug(f"Loaded {len(self.exact_match_words)} exact match words/phrases from blocklist") + for token in self.unprotected_whitelist_tokens(self.profanity, whitelist_tokens): + # better_profanity drops a whitelisted word before it expands the + # leetspeak variants of the other entries, so a whitelisted word can + # still be censored as some *other* entry's variant -- "flat" is what + # "fiat" becomes under the i/l mapping. Whitelisting it then silently + # does nothing, which is worth saying out loud at load time. + log.warning(f"Whitelisted word is still censored as a variant of another blocklist entry: {token!r}") + + @staticmethod + def unprotected_whitelist_tokens(matcher, whitelist_tokens: list[str]) -> list[str]: + """Whitelisted words the loaded matcher censors anyway.""" + return [token for token in whitelist_tokens if matcher.censor(token, censor_char=CENSOR_SENTINEL) != token] + @staticmethod def normalize_for_matching(prompt: str) -> str: """Fold away the ways a word can be written to look normal but not match. @@ -135,8 +163,9 @@ def normalize_for_matching(prompt: str) -> str: The collapse also closes a plainer hole: a multi-word entry was defeated by typing two spaces between its words. - The code-point folding is a separate method because is_safe also needs it - on its own: see alternate_spellings. + The two foldings are separate methods because a whitelisted phrase has to + be matched between them (see censor_prompt), and is_safe also needs the + code-point folding on its own (see alternate_spellings). """ return Blocklist.fold_invisible_characters(Blocklist.normalize_code_points(prompt)) @@ -152,6 +181,81 @@ def fold_invisible_characters(prompt: str) -> str: """Rewrite zero-width and bidirectional characters to a space.""" return " ".join(INVISIBLE_CHARS.sub(" ", prompt).split()) + @staticmethod + def align_folded_to_source(source: list[str], folded: list[str]) -> list[int]: + """For each folded token, the index of the source token it came from. + + Folding invisible characters to a space can split one source token into + several, so the two streams are not index-for-index. Walk them together, + consuming folded tokens until their concatenation reproduces the source + token, which is exactly what the fold did to it. + """ + mapping: list[int] = [] + source_index = 0 + buffer = "" + for token in folded: + mapping.append(min(source_index, len(source) - 1)) + buffer += token + while ( + source_index < len(source) + and buffer + and len(buffer) >= len(INVISIBLE_CHARS.sub("", source[source_index])) + ): + buffer = "" + source_index += 1 + return mapping + + def fused_matches(self, tokens: list[str]) -> list[tuple[int, int]]: + """(start, end) token indices of every span that is a match on its own, end exclusive. + + Censoring the whole prompt is not enough to place the matches. The + matcher replaces a matched span -- which may cover several tokens -- with + a single censored token, so two adjacent matches ("nice camera" + "shot + gun") cannot be told apart by counting tokens. It also matches greedily + left to right, so a match that shares a token with an earlier one ("in + gun" after "desk in") is never reported at all. + + So every span is tried on its own: it is a match when the matcher reduces + it to a single censored token. A span must start and end on a token that + holds a word, so trailing punctuation does not widen a match past a + phrase. The matcher joins at most MAX_NUMBER_COMBINATIONS + 1 words, and + every word-bearing token holds at least one, which bounds the width. + """ + has_word = [any(c in ALLOWED_CHARACTERS for c in token) for token in tokens] + max_words = self.profanity.MAX_NUMBER_COMBINATIONS + 1 + matches: list[tuple[int, int]] = [] + for start in range(len(tokens)): + if not has_word[start]: + continue + ends, words = [], 0 + for end in range(start + 1, len(tokens) + 1): + words += has_word[end - 1] + if words > max_words: + break + if has_word[end - 1]: + ends.append(end) + # A match starting here lies inside the widest window, and the + # matcher reading that window from its first token would censor + # something; when it does not, no narrower span needs trying. + if CENSOR_SENTINEL not in self.profanity.censor(" ".join(tokens[start : ends[-1]]), CENSOR_SENTINEL): + continue + for end in ends: + censored = self.profanity.censor(" ".join(tokens[start:end]), censor_char=CENSOR_SENTINEL).split() + if len(censored) == 1 and CENSOR_SENTINEL in censored[0]: + matches.append((start, end)) + return matches + + def whitelisted_phrase_spans(self, tokens: list[str]) -> list[tuple[int, int]]: + """(start, end) token indices of every whitelisted phrase occurrence.""" + lowered = [token.strip(_STRIP_CHARS).lower() for token in tokens] + spans: list[tuple[int, int]] = [] + for phrase in self.whitelist_phrases: + parts = phrase.strip().lower().split() + for start in range(len(lowered) - len(parts) + 1): + if lowered[start : start + len(parts)] == parts: + spans.append((start, start + len(parts))) + return spans + def censor_prompt(self, input_prompt: str) -> tuple[bool, str]: """Censor the prompt using the blocklist with better-profanity fuzzy matching. @@ -169,15 +273,50 @@ def censor_prompt(self, input_prompt: str) -> tuple[bool, str]: # than substituting keeps the stricter reading: "n\x00ike" fuses back to # a blocked word instead of being split into two harmless tokens. input_prompt = input_prompt.replace(CENSOR_SENTINEL, "") - input_prompt = self.normalize_for_matching(input_prompt) - # The whitelist is handed to load_censor_words(), so the matcher already - # leaves whitelisted words alone. Restoring them a second time here is - # what introduced the token misalignment described in the commit message. - censored_prompt = self.profanity.censor(input_prompt, censor_char=CENSOR_SENTINEL) - if CENSOR_SENTINEL in censored_prompt: + # Whitelisted single words are handed to load_censor_words(), so the + # matcher already leaves them alone; restoring them a second time after + # censoring is what introduced the token misalignment removed earlier. + # Whitelisted phrases never reach the matcher, so they are located here -- + # between the two foldings, for the reason given below. + input_prompt = self.normalize_code_points(input_prompt) + + # Phrase occurrences are located before invisible characters are folded, + # and that order matters: "deskin" renders as the entry with no + # space in it, so folding first would turn an evasion into the + # whitelisted phrase and exempt it. + source_tokens = input_prompt.split() + exempt_spans = self.whitelisted_phrase_spans(source_tokens) + + folded_prompt = self.fold_invisible_characters(input_prompt) + censored_prompt = self.profanity.censor(folded_prompt, censor_char=CENSOR_SENTINEL) + if CENSOR_SENTINEL not in censored_prompt: + return False, "" + if not exempt_spans: display_prompt = censored_prompt.replace(CENSOR_SENTINEL, CENSOR) return True, f"Prompt blocked by censorship: Censored Prompt: {display_prompt}" - return False, "" + + # The exemption is applied to the matches, not to the prompt. Masking the + # phrase out before matching would also delete any match that overlaps + # it; dropping only the matches that lie entirely inside a whitelisted + # phrase leaves the ones that straddle its boundary blocked. + folded_tokens = folded_prompt.split() + to_source = self.align_folded_to_source(source_tokens, folded_tokens) + + def exempt(start: int, end: int) -> bool: + first, last = to_source[start], to_source[end - 1] + return any(low <= first and last < high for low, high in exempt_spans) + + # Every match must lie inside a phrase, including matches the censored + # output does not show; see fused_matches. + remaining = [(start, end) for start, end in self.fused_matches(folded_tokens) if not exempt(start, end)] + if not remaining: + return False, "" + + display_tokens = list(folded_tokens) + for start, end in remaining: + for index in range(start, end): + display_tokens[index] = CENSOR + return True, f"Prompt blocked by censorship: Censored Prompt: {' '.join(display_tokens)}" @staticmethod def check_partial_match( diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py index 791ba7497..89614c5b3 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py @@ -644,3 +644,240 @@ def test_exact_match_folding_does_not_block_ordinary_text(monkeypatch): ): safe, message = bl.is_safe(prompt) assert safe is True, f"{prompt!r} was wrongly reported as blocked: {message}" + + +@pytest.mark.L1 +def test_whitelisted_phrase_clears_the_prose_that_fuses_into_an_entry(): + """The reported false positive: "desk" + "in" fuses into the entry "deskin". + + Whitelisting the phrase exempts that one spelling. Whitelisting either half + on its own cannot: the fusing does not consult the whitelist. + """ + bl = _blocklist_with_words(["deskin"], whitelist=["desk in"]) + + blocked, message = _blocked(bl, "a robot sitting at a desk in the background") + + assert blocked is False, message + + +@pytest.mark.L1 +def test_whitelisting_a_phrase_leaves_every_other_spelling_blocked(): + """The exemption is one spelling wide, not a hole in the entry.""" + bl = _blocklist_with_words(["deskin"], whitelist=["desk in"]) + + for prompt in ( + "deskin the animal", + "d eskin the animal", + "de skin the animal", + "des kin the animal", + "de sk in the animal", + "d3skin the animal", + ): + blocked, _ = _blocked(bl, prompt) + assert blocked is True, f"{prompt!r} was not blocked" + + +@pytest.mark.L1 +def test_whitelisted_phrase_is_restored_in_the_blocked_message(): + """The placeholder is internal; the user-facing quote shows their own words.""" + bl = _blocklist_with_words(["deskin", "badword"], whitelist=["desk in"]) + + blocked, message = _blocked(bl, "a badword next to a desk in the room") + + assert blocked is True + assert "desk in" in message + assert "zzwhitelisted" not in message + + +@pytest.mark.L1 +def test_whitelisted_word_behaviour_is_unchanged_by_phrase_support(): + """Single-word whitelisting still goes to the matcher, as before.""" + bl = _blocklist_with_words(["snow white"], whitelist=["flat", "desk in"]) + + blocked, _ = _blocked(bl, "the floor is flat") + assert blocked is False + + blocked, _ = _blocked(bl, "a Snow White poster on a flat wall") + assert blocked is True + + +@pytest.mark.L1 +def test_invisible_character_cannot_reach_the_phrase_exemption(): + """An exemption must not be reachable by a spelling the reader cannot see. + + "deskin" renders as the entry with nothing between its halves. The + invisible character is folded to a space for matching, which would turn it + into the whitelisted phrase, so the phrase is matched before that folding. + """ + bl = _blocklist_with_words(["deskin"], whitelist=["desk in"]) + + blocked, _ = _blocked(bl, "desk​in the animal") + assert blocked is True + + blocked, _ = _blocked(bl, "desk­in the animal") + assert blocked is True + + +@pytest.mark.L1 +def test_phrase_exemption_survives_spacing_and_case(): + """The exempted spelling is the phrase as a reader writes it, however spaced.""" + bl = _blocklist_with_words(["deskin"], whitelist=["desk in"]) + + for prompt in ("a desk in the room", "a desk in the room", "a DESK IN the room", "Desk In The Room"): + blocked, _ = _blocked(bl, prompt) + assert blocked is False, f"{prompt!r} was blocked" + + +@pytest.mark.L1 +def test_whitelisted_word_censored_as_another_entrys_variant_is_reported(): + """Whitelisting a word does not guarantee the matcher spares it. + + better_profanity removes whitelisted words before it expands the leetspeak + variants of the remaining entries, so a whitelisted word that happens to be + a variant of another entry stays censored -- "flat" is what "fiat" becomes + under the i/l mapping. The whitelist entry then silently does nothing, so + the loader names it rather than leaving it to be found in production. + """ + from better_profanity.better_profanity import Profanity + + matcher = Profanity() + matcher.load_censor_words(custom_words=["fiat"], whitelist_words=["flat"]) + + assert Blocklist.unprotected_whitelist_tokens(matcher, ["flat"]) == ["flat"] + + unaffected = Profanity() + unaffected.load_censor_words(custom_words=["snow white"], whitelist_words=["flat"]) + + assert Blocklist.unprotected_whitelist_tokens(unaffected, ["flat"]) == [] + + +@pytest.mark.L1 +def test_match_straddling_a_whitelisted_phrase_stays_blocked(): + """The exemption covers the phrase, not everything that touches it. + + Masking the phrase out before matching also deleted any match that overlaps + it: with "in gun" on the blocklist, "a desk in gun on the table" stopped + being blocked as soon as "desk in" was whitelisted. Only a match lying + entirely inside the phrase may be dropped. + """ + bl = _blocklist_with_words(["in gun"], whitelist=["desk in"]) + + blocked, _ = _blocked(bl, "a desk in gun on the table") + + assert blocked is True + + +@pytest.mark.L1 +def test_fusion_across_the_phrase_boundary_stays_blocked(): + """Whitelisting "camera shot" must not stop "shot" + "gun" fusing into "shotgun". + + The fused match takes one token from inside the phrase and one from outside + it, so it straddles the boundary and is kept. + """ + bl = _blocklist_with_words(["shotgun"], whitelist=["camera shot"]) + + blocked, _ = _blocked(bl, "a camera shot gun in frame") + assert blocked is True + + blocked, message = _blocked(bl, "a camera shot of the room") + assert blocked is False, message + + +@pytest.mark.L1 +def test_second_occurrence_of_a_phrase_does_not_hide_a_real_match(): + """One exempt occurrence must not exempt a different match elsewhere.""" + bl = _blocklist_with_words(["deskin", "in gun"], whitelist=["desk in"]) + + blocked, _ = _blocked(bl, "a desk in room and a desk in gun") + + assert blocked is True + + +@pytest.mark.L1 +def test_align_folded_to_source_maps_a_split_token_back(): + """Folding an invisible character splits one source token into two.""" + source = ["a", "desk​in", "room"] + folded = ["a", "desk", "in", "room"] + + assert Blocklist.align_folded_to_source(source, folded) == [0, 1, 1, 2] + + +@pytest.mark.L1 +def test_adjacent_matches_are_each_located_before_the_exemption(): + """Two matches side by side must each be checked against the phrases. + + Reported in review on #306. "nice camera" and "shot gun" are two fused + matches next to each other. The matcher reduces them to two censored + tokens, and a split guessed from the token count alone put the boundary + after "shot": "nice camera shot" and "gun", each of which lies inside a + whitelisted phrase, so a prompt containing a real "shot gun" was allowed. + """ + bl = _blocklist_with_words(["nicecamera", "shotgun"], whitelist=["nice camera shot", "gun show"]) + + blocked, _ = _blocked(bl, "nice camera shot gun show") + + assert blocked is True + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("name", "prompt"), + [ + ("adjacent, phrase exempts the first", "a desk in shot gun"), + ("adjacent, phrase exempts the second", "a shot gun desk in room"), + ("three in a row", "desk in shot gun bat man"), + ("with punctuation", "a desk in, shot gun."), + ("invisible split beside a phrase", "a desk in shot\u200bgun"), + ("fullwidth beside a phrase", "a desk in \uff53\uff48\uff4f\uff54 gun"), + ], +) +def test_real_match_beside_an_exempt_match_stays_blocked(name, prompt): + """An exempt match must not take a real match next to it along with it.""" + bl = _blocklist_with_words(["deskin", "shotgun", "batman"], whitelist=["desk in"]) + + blocked, _ = _blocked(bl, prompt) + + assert blocked is True, f"{name}: the real match was exempted" + + +@pytest.mark.L1 +@pytest.mark.parametrize( + "prompt", + [ + "a desk in a room", + "a desk in, a desk in the corner", + "a desk in desk in the corner", + "two phrases: desk in and camera shot", + ], +) +def test_only_exempt_matches_leave_the_prompt_allowed(prompt): + """Every match lying inside a phrase, however many, still clears the prompt.""" + bl = _blocklist_with_words(["deskin", "shotgun"], whitelist=["desk in", "camera shot"]) + + blocked, message = _blocked(bl, prompt) + + assert blocked is False, message + + +@pytest.mark.L1 +def test_fused_matches_locate_adjacent_matches(): + """The spans come from the matcher, not from the token count.""" + bl = _blocklist_with_words(["nicecamera", "shotgun"]) + + assert bl.fused_matches("nice camera shot gun show".split()) == [(0, 2), (2, 4)] + + +@pytest.mark.L1 +def test_fused_matches_include_a_match_the_greedy_censor_hides(): + """ "in gun" shares "in" with "desk in"; censoring the prompt reports only the first.""" + bl = _blocklist_with_words(["deskin", "in gun"]) + + assert bl.fused_matches("a desk in gun".split()) == [(1, 3), (2, 4)] + + +@pytest.mark.L1 +def test_fused_matches_locate_a_multi_token_match(): + """A multi-word match collapses to one censored token; recover its width.""" + bl = _blocklist_with_words(["snow white"]) + + assert bl.fused_matches("a snow white poster on a wall".split()) == [(1, 3)] From 967e7452fad42edde3f141ec14abe1d4767b2ba5 Mon Sep 17 00:00:00 2001 From: rwagwani Date: Mon, 26 Oct 2026 12:43:54 +0000 Subject: [PATCH 4/4] fix(guardrail): locate matches ending in a one-letter word, and bound the cost A second review of the span fix found that better_profanity ignores a one-letter word at the very end of its input. Censored on its own, the span "batma n" (or "plan b", or a one-letter entry) matched nothing, so the match was never located, and a whitelisted phrase anywhere in the prompt cleared it: "snow white batma n more" passed with "snow white" whitelisted. Each span is now censored with a word after it, so the matcher reads it as it would inside a prompt. As a second guard, a prompt that was censored but in which no span matches on its own stays blocked. Locating every match is also bounded. A 32-token chunk that the matcher clears in one call is skipped, results are cached across the readings is_safe checks, and past 4000 matcher calls the prompt stays blocked, as it is without phrase support. The reviewer's worst case, 800 tokens of repeated phrase plus a non-ASCII character, goes from 25.5 s to 2.1 s. align_folded_to_source also skips a source token made only of invisible characters, which folds to nothing and shifted every later token by one. Co-Authored-By: Claude Opus 5.5 --- .../guardrail/blocklist/blocklist.py | 108 ++++++++++++++---- .../guardrail/blocklist/blocklist_test.py | 78 +++++++++++++ 2 files changed, 164 insertions(+), 22 deletions(-) diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py index d45fdf652..71bb453bb 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py @@ -46,6 +46,23 @@ # let "de$k in" read as the whitelisted phrase. _STRIP_CHARS = "".join(sorted(set(string.punctuation) - ALLOWED_CHARACTERS)) +# Read after a span when it is censored on its own. better_profanity ignores a +# one-letter word at the very end of its input, so "plan b" alone matches +# nothing while "plan b more" matches; following every span with a word makes +# it read the span as it would inside a prompt. Long and unlike any entry, so +# it never joins a match itself. +_SPAN_END = "zqxjspanendzqxj" + +# Bounds the per-instance cache of span censoring results. +_SPAN_CACHE_LIMIT = 100_000 + +# Tokens fused_matches clears with one matcher call before trying each start. +_SPAN_CHUNK = 32 + +# Matcher calls fused_matches may make for one prompt. A prompt needing more is +# left blocked, as it is without phrase support, rather than slowed down. +_SPAN_BUDGET = 4_000 + class _NoLemmatizer: """Leaves every token as it is. The default for Blocklist.from_lists.""" @@ -122,6 +139,8 @@ def load_lists(self, blocklist_words: list[str], whitelist_words: list[str], exa # The matcher's whitelist is a set of single tokens, so a whitelisted # phrase never reaches it. Phrases are handled in censor_prompt instead, # by exempting the matches that fall inside one. + self._span_cache: dict[str, tuple[str, bool]] = {} + self._span_calls = 0 self.whitelist_phrases = sorted((w for w in self.whitelist_words if " " in w.strip()), key=len, reverse=True) whitelist_tokens = [w for w in self.whitelist_words if " " not in w.strip()] @@ -194,6 +213,9 @@ def align_folded_to_source(source: list[str], folded: list[str]) -> list[int]: source_index = 0 buffer = "" for token in folded: + # A source token made only of invisible characters folds to nothing. + while source_index < len(source) - 1 and not INVISIBLE_CHARS.sub("", source[source_index]): + source_index += 1 mapping.append(min(source_index, len(source) - 1)) buffer += token while ( @@ -205,7 +227,7 @@ def align_folded_to_source(source: list[str], folded: list[str]) -> list[int]: source_index += 1 return mapping - def fused_matches(self, tokens: list[str]) -> list[tuple[int, int]]: + def fused_matches(self, tokens: list[str]) -> list[tuple[int, int]] | None: """(start, end) token indices of every span that is a match on its own, end exclusive. Censoring the whole prompt is not enough to place the matches. The @@ -216,35 +238,70 @@ def fused_matches(self, tokens: list[str]) -> list[tuple[int, int]]: gun" after "desk in") is never reported at all. So every span is tried on its own: it is a match when the matcher reduces - it to a single censored token. A span must start and end on a token that - holds a word, so trailing punctuation does not widen a match past a + it to a single censored token. Each span is read with a word after it, + for the reason given at _SPAN_END. A span must start and end on a token + that holds a word, so trailing punctuation does not widen a match past a phrase. The matcher joins at most MAX_NUMBER_COMBINATIONS + 1 words, and every word-bearing token holds at least one, which bounds the width. + + Returns None when the prompt needs more than _SPAN_BUDGET matcher calls; + the caller then leaves it blocked. """ has_word = [any(c in ALLOWED_CHARACTERS for c in token) for token in tokens] max_words = self.profanity.MAX_NUMBER_COMBINATIONS + 1 + calls_before = self._span_calls + + def window_end(start: int, word_limit: int) -> int: + """End of the widest window from start holding at most word_limit words.""" + end, words = start, 0 + while end < len(tokens) and words + has_word[end] <= word_limit: + words += has_word[end] + end += 1 + return end + matches: list[tuple[int, int]] = [] - for start in range(len(tokens)): - if not has_word[start]: + for chunk in range(0, len(tokens), _SPAN_CHUNK): + # A match starting in this chunk lies inside the chunk plus the + # widest match after it. If the matcher censors nothing there, no + # match lies inside it: one it skipped would have to sit under an + # earlier match, which would have been censored. + words_in_chunk = sum(has_word[chunk : chunk + _SPAN_CHUNK]) + if not self._censors_anything(" ".join(tokens[chunk : window_end(chunk, words_in_chunk + max_words)])): continue - ends, words = [], 0 - for end in range(start + 1, len(tokens) + 1): - words += has_word[end - 1] - if words > max_words: - break - if has_word[end - 1]: - ends.append(end) - # A match starting here lies inside the widest window, and the - # matcher reading that window from its first token would censor - # something; when it does not, no narrower span needs trying. - if CENSOR_SENTINEL not in self.profanity.censor(" ".join(tokens[start : ends[-1]]), CENSOR_SENTINEL): - continue - for end in ends: - censored = self.profanity.censor(" ".join(tokens[start:end]), censor_char=CENSOR_SENTINEL).split() - if len(censored) == 1 and CENSOR_SENTINEL in censored[0]: - matches.append((start, end)) + for start in range(chunk, min(chunk + _SPAN_CHUNK, len(tokens))): + if not has_word[start]: + continue + if self._span_calls - calls_before > _SPAN_BUDGET: + return None + ends = [end for end in range(start + 1, window_end(start, max_words) + 1) if has_word[end - 1]] + # The same argument for the widest window from this start. + if not self._censors_anything(" ".join(tokens[start : ends[-1]])): + continue + matches.extend((start, end) for end in ends if self._is_one_match(" ".join(tokens[start:end]))) return matches + def _censors_anything(self, text: str) -> bool: + return CENSOR_SENTINEL in self._censor_span(text)[0] + + def _is_one_match(self, text: str) -> bool: + censored, end_survived = self._censor_span(text) + return end_survived and len(censored.split()) == 1 and CENSOR_SENTINEL in censored + + def _censor_span(self, text: str) -> tuple[str, bool]: + """The span censored with _SPAN_END after it, and whether _SPAN_END survived. + + Cached, because neighbouring spans and the alternate readings is_safe + checks repeat the same text many times over. + """ + if text not in self._span_cache: + if len(self._span_cache) >= _SPAN_CACHE_LIMIT: + self._span_cache.clear() + self._span_calls += 1 + censored = self.profanity.censor(f"{text} {_SPAN_END}", censor_char=CENSOR_SENTINEL) + body, _, tail = censored.rpartition(" ") + self._span_cache[text] = (body, True) if tail == _SPAN_END else (censored, False) + return self._span_cache[text] + def whitelisted_phrase_spans(self, tokens: list[str]) -> list[tuple[int, int]]: """(start, end) token indices of every whitelisted phrase occurrence.""" lowered = [token.strip(_STRIP_CHARS).lower() for token in tokens] @@ -308,7 +365,14 @@ def exempt(start: int, end: int) -> bool: # Every match must lie inside a phrase, including matches the censored # output does not show; see fused_matches. - remaining = [(start, end) for start, end in self.fused_matches(folded_tokens) if not exempt(start, end)] + matches = self.fused_matches(folded_tokens) + if not matches: + # Over the budget, or the prompt as a whole was censored yet no span + # on its own is a match. Stay blocked rather than let the exemption + # clear a match it could not locate. + display_prompt = censored_prompt.replace(CENSOR_SENTINEL, CENSOR) + return True, f"Prompt blocked by censorship: Censored Prompt: {display_prompt}" + remaining = [(start, end) for start, end in matches if not exempt(start, end)] if not remaining: return False, "" diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py index 89614c5b3..499e7f13c 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py @@ -881,3 +881,81 @@ def test_fused_matches_locate_a_multi_token_match(): bl = _blocklist_with_words(["snow white"]) assert bl.fused_matches("a snow white poster on a wall".split()) == [(1, 3)] + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("name", "prompt"), + [ + ("entry ending in a one-letter word", "snow white plan b more"), + ("last letter split off", "snow white batma n more"), + ("last letter split off by a hyphen", "snow white batma-n more"), + ("leetspeak with the last letter split off", "snow white b@tma n more"), + ("one-letter entry", "snow white x y"), + ], +) +def test_a_phrase_elsewhere_does_not_clear_a_match_ending_in_one_letter(name, prompt): + """Found in a second review of the span fix. + + better_profanity ignores a one-letter word at the very end of its input, so + a span such as "batma n" censored on its own matched nothing, the match was + never located, and a whitelisted phrase anywhere in the prompt cleared it. + """ + bl = _blocklist_with_words(["batman", "plan b", "x"], whitelist=["snow white"]) + + blocked, _ = _blocked(bl, prompt) + + assert blocked is True, f"{name} was cleared by an unrelated phrase" + + +@pytest.mark.L1 +def test_a_match_the_spans_cannot_locate_stays_blocked(monkeypatch): + """If no span matches on its own although the prompt was censored, stay blocked.""" + bl = _blocklist_with_words(["batman"], whitelist=["snow white"]) + monkeypatch.setattr(bl, "fused_matches", lambda tokens: []) + + blocked, _ = bl.censor_prompt("snow white batman") + + assert blocked is True + + +@pytest.mark.L1 +def test_a_prompt_over_the_span_budget_stays_blocked(monkeypatch): + """Locating every match has a cost cap; past it the prompt is blocked, as without phrases.""" + monkeypatch.setattr("cosmos_framework.auxiliary.guardrail.blocklist.blocklist._SPAN_BUDGET", 3) + bl = _blocklist_with_words(["deskin"], whitelist=["desk in"]) + + blocked, _ = bl.censor_prompt(" ".join(["a desk in the room"] * 20)) + + assert blocked is True + + +@pytest.mark.L1 +def test_matches_are_found_on_both_sides_of_a_chunk_boundary(): + """The chunk pre-check must not skip a match that starts near a chunk's end.""" + bl = _blocklist_with_words(["deskin", "shotgun"], whitelist=["desk in"]) + filler = ["word"] * 30 + + for offset in range(28, 36): + tokens = ["word"] * offset + ["shot", "gun"] + filler + ["desk", "in"] + assert (offset, offset + 2) in bl.fused_matches(tokens), offset + blocked, _ = _blocked(bl, " ".join(tokens)) + assert blocked is True, offset + + +@pytest.mark.L1 +def test_align_folded_to_source_skips_a_token_of_only_invisible_characters(): + """Found in review: such a token folds to nothing and must not shift the rest.""" + source = ["\u200b", "snow", "white", "batman"] + folded = ["snow", "white", "batman"] + + assert Blocklist.align_folded_to_source(source, folded) == [1, 2, 3] + + +@pytest.mark.L1 +def test_invisible_only_token_does_not_pull_a_match_into_a_phrase(): + bl = _blocklist_with_words(["batman"], whitelist=["snow white"]) + + blocked, _ = bl.censor_prompt("\u200b snow white batman") + + assert blocked is True