From 57a245a24a1e6b6a29455af651240e123acb4a45 Mon Sep 17 00:00:00 2001 From: rwagwani Date: Mon, 26 Oct 2026 11:02:01 +0000 Subject: [PATCH 1/5] 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/5] 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/5] 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/5] 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 From 18a87ac8e8825034d241c19057a005b6c34714cb Mon Sep 17 00:00:00 2001 From: rwagwani Date: Mon, 26 Oct 2026 11:17:45 +0000 Subject: [PATCH 5/5] feat(guardrail): add list-maintenance tools that measure blocklist changes hygiene.py flags blocklist entries that two ordinary dictionary words fuse into, which is how prose collides with an entry ("desk in" -> "deskin"). It reads lists with the production reader and checks what it advises. coverage.py measures what a list change stops catching. It writes five spellings of every entry (verbatim, split by a space, split by a zero-width space, fullwidth, lookalike) and runs them through is_safe, the deployed entry point, with the exact-match list and nltk data beside the lists when present. Review on #306 pointed out that keying probes by entry index and spelling, and ignoring keys present on only one side, let a removed or moved entry drop out of the comparison with a clean exit. Probes are now identified by a digest of the entry plus the spelling, so position does not matter. The probe set can be frozen with --write-probes and replayed with --probes, so an entry removed from the list keeps its probes and shows up as lost. Every result file records the digest of the probe set it ran, and --compare exits 2 with "incomparable" when the two runs did not evaluate the same probes, instead of reporting nothing lost. A file that is not a run, or holds no probes, is also incomparable rather than a crash or a clean result. Probe ids keep the entry's case, so case variants keep their own probes, and each result records digests of the lists it measured and the lemmatizer used, and --compare prints any of those that changed. Moved here from #306, so the Unicode fix there can land on its own. Co-Authored-By: Claude Opus 5.5 --- .../auxiliary/guardrail/blocklist/coverage.py | 281 ++++++++++++++++++ .../guardrail/blocklist/coverage_test.py | 260 ++++++++++++++++ .../auxiliary/guardrail/blocklist/hygiene.py | 270 +++++++++++++++++ .../guardrail/blocklist/hygiene_test.py | 131 ++++++++ 4 files changed, 942 insertions(+) create mode 100644 cosmos_framework/auxiliary/guardrail/blocklist/coverage.py create mode 100644 cosmos_framework/auxiliary/guardrail/blocklist/coverage_test.py create mode 100644 cosmos_framework/auxiliary/guardrail/blocklist/hygiene.py create mode 100644 cosmos_framework/auxiliary/guardrail/blocklist/hygiene_test.py diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/coverage.py b/cosmos_framework/auxiliary/guardrail/blocklist/coverage.py new file mode 100644 index 000000000..527d5c527 --- /dev/null +++ b/cosmos_framework/auxiliary/guardrail/blocklist/coverage.py @@ -0,0 +1,281 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +"""Measure what a blocklist change stops catching. + +A corpus of ordinary model output answers one question: does a change block +things it did not block before. Any change that *unblocks* -- a whitelist entry, +a data move, a new exemption mechanism -- carries the opposite risk, and a +corpus cannot show it: the phrasings that would newly slip through are exactly +the ones ordinary output does not contain. + +This builds that missing side. For every entry it writes five prompts, one per +way the entry can be spelled while still reading as itself: + + verbatim the entry as written + spaced split in the middle, the fusing evasion ("de skin") + invisible split with a zero-width space + fullwidth spelled in fullwidth letters + lookalike every letter that has one swapped for a Cyrillic or Greek twin + +Freeze the probes once, run both versions over that same frozen set, and +compare: any prompt that was blocked and now is not is a spelling the change +gave up. A probe is identified by a digest of its entry plus the spelling, not +by the entry's position, so reordering the list does not change what is +compared, and an entry removed from the list keeps its probes in the frozen set +-- its spellings then show up as lost. Two result files that were not run over +the same probes are reported as incomparable, never as a clean comparison. +Entry text is never printed, so output can be pasted into a review. A frozen +probe file does hold the probe prompts, and so the entries, in plain text: keep +it with the lists, not in a ticket. + + coverage.py --write-probes probes.json + coverage.py --probes probes.json --whitelist-dir --json before.json + coverage.py --probes probes.json --whitelist-dir --json after.json + coverage.py --compare before.json after.json + +The exact-match list and nltk data beside are picked up on their +own; the whitelist is not, so pass --whitelist-dir to measure the production +configuration. Only blocklist entries are probed, not exact-match entries. +""" + +import argparse +import hashlib +import json +import sys +import unicodedata +from collections.abc import Callable, Iterable +from pathlib import Path + +# Lowercase ASCII letter -> its fullwidth presentation form. +FULLWIDTH = {chr(code): chr(code - 0x61 + 0xFF41) for code in range(0x61, 0x7B)} + +ZERO_WIDTH_SPACE = "​" + + +def _lookalike_spelling() -> dict[str, str]: + """Latin letter -> a Cyrillic twin where one exists, else a Greek one.""" + from cosmos_framework.auxiliary.guardrail.blocklist.lookalikes import LOOKALIKES + + twins: dict[str, str] = {} + for script in ("CYRILLIC SMALL LETTER", "GREEK SMALL LETTER"): + for char, letter in sorted(LOOKALIKES.items()): + if letter not in twins and unicodedata.name(char, "").startswith(script): + twins[letter] = char + return twins + + +LOOKALIKE = _lookalike_spelling() + +PROMPT_TEMPLATE = "a photo of {body} in the scene" + + +def spellings(entry: str) -> dict[str, str]: + """The five prompts probing one entry. + + Splitting at the midpoint is arbitrary but stable: the point is to produce a + spelling the matcher has to fuse back together, not to find the split a + person would choose. + """ + entry = entry.strip() + middle = max(1, len(entry) // 2) + return { + "verbatim": PROMPT_TEMPLATE.format(body=entry), + "spaced": PROMPT_TEMPLATE.format(body=f"{entry[:middle]} {entry[middle:]}"), + "invisible": PROMPT_TEMPLATE.format(body=f"{entry[:middle]}{ZERO_WIDTH_SPACE}{entry[middle:]}"), + "fullwidth": PROMPT_TEMPLATE.format(body="".join(FULLWIDTH.get(c, c) for c in entry)), + "lookalike": PROMPT_TEMPLATE.format(body="".join(LOOKALIKE.get(c, c) for c in entry.lower())), + } + + +def probe_id(entry: str, spelling: str) -> str: + """A stable, opaque name for one probe: a digest of the entry, and the spelling. + + The digest stands in for the entry text, which must not be printed, and does + not depend on where the entry sits in the list. + """ + digest = hashlib.sha256(entry.strip().encode("utf-8")).hexdigest()[:16] + return f"{digest}:{spelling}" + + +def build_probes(entries: Iterable[str]) -> dict[str, str]: + """Every probe prompt for the given entries, keyed by probe_id.""" + probes: dict[str, str] = {} + for entry in entries: + for spelling, prompt in spellings(entry).items(): + probes[probe_id(entry, spelling)] = prompt + return probes + + +def corpus_id(probes: dict[str, str]) -> str: + """A digest of the whole probe set, recorded with each run's results.""" + return hashlib.sha256(json.dumps(probes, sort_keys=True).encode("utf-8")).hexdigest()[:16] + + +def measure(probes: dict[str, str], is_blocked: Callable[[str], bool]) -> dict[str, bool]: + """Whether each probe prompt is blocked, keyed by probe id.""" + return {key: bool(is_blocked(prompt)) for key, prompt in probes.items()} + + +def summarize(results: dict[str, bool]) -> dict[str, dict[str, int]]: + """Blocked / passed counts per spelling, plus a total.""" + summary: dict[str, dict[str, int]] = {} + for key, blocked in results.items(): + bucket = summary.setdefault(key.split(":", 1)[1], {"blocked": 0, "passed": 0}) + bucket["blocked" if blocked else "passed"] += 1 + summary["total"] = { + "blocked": sum(1 for v in results.values() if v), + "passed": sum(1 for v in results.values() if not v), + } + return summary + + +def compare(before: dict, after: dict) -> dict: + """What the change gave up, and what it started catching. + + Takes two result files as written by a run. `lost` is the one that matters + for an unblocking change: prompts that were blocked and are not any more. + The diff is only meaningful over the same probes, so `comparable` is False + when the two runs used different probe sets or report different probe ids, + and `lost` must then not be read as a complete answer. + """ + before_results, after_results = before["results"], after["results"] + shared = set(before_results) & set(after_results) + only_in_before = sorted(set(before_results) - set(after_results)) + only_in_after = sorted(set(after_results) - set(before_results)) + return { + "comparable": before.get("corpus") == after.get("corpus") and not only_in_before and not only_in_after, + "corpus": [before.get("corpus"), after.get("corpus")], + "lost": sorted(key for key in shared if before_results[key] and not after_results[key]), + "gained": sorted(key for key in shared if not before_results[key] and after_results[key]), + "only_in_before": only_in_before, + "only_in_after": only_in_after, + } + + +def _build_is_blocked(blocklist_dir: str, whitelist_dir: str | None, add_phrase: str | None): + """A predicate running the deployed entry point, is_safe, over the given lists. + + is_safe and not censor_prompt: is_safe is what callers run, and it does work + of its own before the matcher -- to_ascii, a lemmatized pass, the exact-match + list -- so a sweep through censor_prompt can report a spelling as caught that + production lets through. When the lists sit in a checkpoint's blocklist + directory, the sibling exact_match list and nltk_data are used as well, so + the sweep sees what production sees. + """ + from cosmos_framework.auxiliary.guardrail.blocklist.blocklist import Blocklist + from cosmos_framework.auxiliary.guardrail.blocklist.utils import read_keyword_list_from_dir + + entries = read_keyword_list_from_dir(blocklist_dir) + whitelist = read_keyword_list_from_dir(whitelist_dir) if whitelist_dir else [] + if add_phrase: + whitelist = whitelist + [add_phrase] + + root = Path(blocklist_dir).parent + exact_match_dir = root / "exact_match" + exact_match = read_keyword_list_from_dir(str(exact_match_dir)) if exact_match_dir.is_dir() else [] + + lemmatizer = None + nltk_dir = root / "nltk_data" + if nltk_dir.is_dir(): + import nltk + + nltk.data.path.append(str(nltk_dir)) + lemmatizer = nltk.WordNetLemmatizer() + else: + print(f"no nltk_data next to {blocklist_dir}; the lemmatized pass sees words unchanged", file=sys.stderr) + + probe = Blocklist.from_lists(entries, whitelist, exact_match, lemmatizer=lemmatizer) + setup = { + "blocklist": _digest(entries), + "whitelist": _digest(whitelist), + "exact_match": _digest(exact_match), + "lemmatizer": "wordnet" if lemmatizer is not None else "none", + } + return entries, (lambda text: not probe.is_safe(text)[0]), setup + + +def _digest(words: list[str]) -> str: + """A digest of a keyword list, recorded so a result says what it measured.""" + return hashlib.sha256("\n".join(words).encode("utf-8")).hexdigest()[:16] + + +def parse_args(argv=None): + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("blocklist_dir", nargs="?", help="directory of blocklist keyword files") + parser.add_argument("--whitelist-dir", help="directory of whitelist keyword files") + parser.add_argument("--add-phrase", help="whitelist this phrase in addition, to model a proposed change") + parser.add_argument("--write-probes", help="write the probes for blocklist_dir here and stop, to freeze them") + parser.add_argument("--probes", help="run this frozen probe file instead of probes built from blocklist_dir") + parser.add_argument("--json", help="write the per-prompt results here") + parser.add_argument("--compare", nargs=2, metavar=("BEFORE", "AFTER"), help="diff two result files") + return parser.parse_args(argv) + + +def main(argv=None) -> int: + """Exit 0 when nothing was lost, 1 when something was, 2 when the runs cannot be compared.""" + args = parse_args(argv) + + if args.compare: + before = json.loads(Path(args.compare[0]).read_text()) + after = json.loads(Path(args.compare[1]).read_text()) + for path, run in zip(args.compare, (before, after)): + if not isinstance(run, dict) or not run.get("results") or "corpus" not in run: + print(f"incomparable: {path} is not a result file written by a run, or holds no probes") + return 2 + diff = compare(before, after) + if not diff["comparable"]: + print("incomparable: the two runs did not evaluate the same probes") + print(f" probe sets: {diff['corpus'][0]} vs {diff['corpus'][1]}") + print(f" probes only in before: {len(diff['only_in_before'])}") + print(f" probes only in after : {len(diff['only_in_after'])}") + print(" freeze the probes with --write-probes and run both versions with --probes") + return 2 + for name in sorted(set(before.get("setup", {})) | set(after.get("setup", {}))): + was, now = before.get("setup", {}).get(name), after.get("setup", {}).get(name) + if was != now: + print(f"setup changed: {name} {was} -> {now}") + print(f"lost (blocked before, not after): {len(diff['lost'])}") + for key in diff["lost"]: + print(f" {key}") + print(f"gained (not blocked before, now) : {len(diff['gained'])}") + return 1 if diff["lost"] else 0 + + if not args.blocklist_dir: + print("a blocklist directory is required unless --compare is used", file=sys.stderr) + return 2 + + entries, is_blocked, setup = _build_is_blocked(args.blocklist_dir, args.whitelist_dir, args.add_phrase) + if args.write_probes: + probes = build_probes(entries) + Path(args.write_probes).write_text(json.dumps({"corpus": corpus_id(probes), "probes": probes}, indent=0)) + print(f"probes: {len(probes)} from {len(entries)} entries, set {corpus_id(probes)}") + print(f"written: {args.write_probes}") + return 0 + + if args.probes: + frozen = json.loads(Path(args.probes).read_text()) + probes = frozen["probes"] + if corpus_id(probes) != frozen["corpus"]: + print(f"{args.probes} was edited after it was written; refusing to run it", file=sys.stderr) + return 2 + else: + probes = build_probes(entries) + + results = measure(probes, is_blocked) + summary = summarize(results) + print(f"entries: {len(entries)} prompts: {len(results)} probe set: {corpus_id(probes)}") + for spelling in ("verbatim", "spaced", "invisible", "fullwidth", "lookalike", "total"): + if spelling in summary: + counts = summary[spelling] + print(f" {spelling:10s} blocked={counts['blocked']:5d} passed={counts['passed']:5d}") + if args.json: + Path(args.json).write_text( + json.dumps({"corpus": corpus_id(probes), "setup": setup, "summary": summary, "results": results}, indent=0) + ) + print(f"written: {args.json}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/coverage_test.py b/cosmos_framework/auxiliary/guardrail/blocklist/coverage_test.py new file mode 100644 index 000000000..e3a85c205 --- /dev/null +++ b/cosmos_framework/auxiliary/guardrail/blocklist/coverage_test.py @@ -0,0 +1,260 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +import json + +import pytest + +from cosmos_framework.auxiliary.guardrail.blocklist.coverage import ( + ZERO_WIDTH_SPACE, + _build_is_blocked, + build_probes, + compare, + corpus_id, + main, + measure, + probe_id, + spellings, + summarize, +) +from cosmos_framework.auxiliary.guardrail.blocklist.lookalikes import fold_lookalikes + + +@pytest.mark.L1 +def test_spellings_cover_the_five_evasions(): + """Each probe must still read as the entry to a person.""" + probes = spellings("deskin") + + assert probes["verbatim"] == "a photo of deskin in the scene" + assert probes["spaced"] == "a photo of des kin in the scene" + assert probes["invisible"] == f"a photo of des{ZERO_WIDTH_SPACE}kin in the scene" + assert probes["fullwidth"] == "a photo of ｄｅｓｋｉｎ in the scene" + assert probes["lookalike"] != probes["verbatim"] + assert probes["lookalike"].isascii() is False + assert fold_lookalikes(probes["lookalike"]) == probes["verbatim"] + + +@pytest.mark.L1 +def test_spellings_handle_a_short_entry(): + """A two-character entry must still produce a split, not an empty half.""" + probes = spellings("ab") + + assert probes["spaced"] == "a photo of a b in the scene" + + +@pytest.mark.L1 +def test_probe_ids_do_not_depend_on_position_or_show_the_entry(): + """Reordering the list must not change what is compared, and ids are safe to paste.""" + forward = build_probes(["deskin", "chromebook"]) + backward = build_probes(["chromebook", "deskin"]) + + assert forward == backward + assert corpus_id(forward) == corpus_id(backward) + assert probe_id("deskin", "spaced") in forward + assert probe_id(" deskin ", "spaced") == probe_id("deskin", "spaced") + assert probe_id("Deskin", "spaced") != probe_id("deskin", "spaced") + assert not any("deskin" in key or "chromebook" in key for key in forward) + + +@pytest.mark.L1 +def test_measure_keys_results_by_probe_id(): + probes = build_probes(["deskin", "chromebook"]) + + results = measure(probes, lambda text: "deskin" in text) + + assert results[probe_id("deskin", "verbatim")] is True + assert results[probe_id("chromebook", "verbatim")] is False + assert set(results) == set(probes) + + +@pytest.mark.L1 +def test_summarize_counts_per_spelling(): + results = {"a:verbatim": True, "a:spaced": False, "b:verbatim": True, "b:spaced": True} + + summary = summarize(results) + + assert summary["verbatim"] == {"blocked": 2, "passed": 0} + assert summary["spaced"] == {"blocked": 1, "passed": 1} + assert summary["total"] == {"blocked": 3, "passed": 1} + + +@pytest.mark.L1 +def test_compare_names_what_a_change_gave_up(): + """`lost` is the answer an unblocking change has to provide.""" + before = {"corpus": "c", "results": {"a:verbatim": True, "a:spaced": True, "b:verbatim": False}} + after = {"corpus": "c", "results": {"a:verbatim": True, "a:spaced": False, "b:verbatim": True}} + + diff = compare(before, after) + + assert diff["comparable"] is True + assert diff["lost"] == ["a:spaced"] + assert diff["gained"] == ["b:verbatim"] + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("name", "after"), + [ + ("different probe set", {"corpus": "other", "results": {"a:spaced": True}}), + ("probe missing after", {"corpus": "c", "results": {}}), + ("probe added after", {"corpus": "c", "results": {"a:spaced": True, "b:spaced": True}}), + ], +) +def test_compare_refuses_runs_over_different_probes(tmp_path, capsys, name, after): + """Results over different probes are incomparable, never a clean comparison.""" + before = {"corpus": "c", "results": {"a:spaced": True}} + (tmp_path / "before.json").write_text(json.dumps(before)) + (tmp_path / "after.json").write_text(json.dumps(after)) + + assert compare(before, after)["comparable"] is False, name + assert main(["--compare", str(tmp_path / "before.json"), str(tmp_path / "after.json")]) == 2 + assert "incomparable" in capsys.readouterr().out + + +@pytest.mark.L1 +def test_compare_mode_exits_non_zero_when_something_was_lost(tmp_path, capsys): + """CI can gate on it: a lost spelling is a failing exit code.""" + (tmp_path / "before.json").write_text(json.dumps({"corpus": "c", "results": {"a:spaced": True}})) + (tmp_path / "after.json").write_text(json.dumps({"corpus": "c", "results": {"a:spaced": False}})) + + assert main(["--compare", str(tmp_path / "before.json"), str(tmp_path / "after.json")]) == 1 + assert "a:spaced" in capsys.readouterr().out + + +@pytest.mark.L1 +def test_compare_mode_exits_zero_when_nothing_was_lost(tmp_path): + (tmp_path / "before.json").write_text(json.dumps({"corpus": "c", "results": {"a:spaced": True}})) + (tmp_path / "after.json").write_text(json.dumps({"corpus": "c", "results": {"a:spaced": True}})) + + assert main(["--compare", str(tmp_path / "before.json"), str(tmp_path / "after.json")]) == 0 + + +def _write_list(directory, words): + directory.mkdir(parents=True, exist_ok=True) + (directory / "list").write_text("".join(f"{word}\n" for word in words), encoding="utf-8") + return str(directory) + + +@pytest.mark.L1 +def test_removing_an_entry_shows_as_lost_over_frozen_probes(tmp_path): + """Removing an entry must not silently remove its probes from the comparison.""" + before_dir = _write_list(tmp_path / "before" / "custom", ["deskin", "batman"]) + after_dir = _write_list(tmp_path / "after" / "custom", ["deskin"]) + probes, before, after = (str(tmp_path / name) for name in ("probes.json", "before.json", "after.json")) + + assert main([before_dir, "--write-probes", probes]) == 0 + assert main([before_dir, "--probes", probes, "--json", before]) == 0 + assert main([after_dir, "--probes", probes, "--json", after]) == 0 + + diff = compare(json.loads(open(before).read()), json.loads(open(after).read())) + assert diff["comparable"] is True + assert probe_id("batman", "verbatim") in diff["lost"] + assert not any(key.startswith(probe_id("deskin", "")[:16]) for key in diff["lost"]) + assert main(["--compare", before, after]) == 1 + + +@pytest.mark.L1 +def test_removing_an_entry_without_frozen_probes_is_incomparable(tmp_path): + """Each run building its own probes cannot hide the removed entry either.""" + before_dir = _write_list(tmp_path / "before" / "custom", ["deskin", "batman"]) + after_dir = _write_list(tmp_path / "after" / "custom", ["deskin"]) + before, after = str(tmp_path / "before.json"), str(tmp_path / "after.json") + + assert main([before_dir, "--json", before]) == 0 + assert main([after_dir, "--json", after]) == 0 + + assert main(["--compare", before, after]) == 2 + + +@pytest.mark.L1 +def test_reordering_the_list_compares_cleanly(tmp_path): + """Moving an entry changes nothing, because probes are not keyed by position.""" + before_dir = _write_list(tmp_path / "before" / "custom", ["deskin", "batman"]) + after_dir = _write_list(tmp_path / "after" / "custom", ["batman", "deskin"]) + before, after = str(tmp_path / "before.json"), str(tmp_path / "after.json") + + assert main([before_dir, "--json", before]) == 0 + assert main([after_dir, "--json", after]) == 0 + + assert main(["--compare", before, after]) == 0 + + +@pytest.mark.L1 +def test_an_edited_probe_file_is_refused(tmp_path): + custom = _write_list(tmp_path / "custom", ["deskin"]) + probes = tmp_path / "probes.json" + assert main([custom, "--write-probes", str(probes)]) == 0 + + frozen = json.loads(probes.read_text()) + frozen["probes"].popitem() + probes.write_text(json.dumps(frozen)) + + assert main([custom, "--probes", str(probes)]) == 2 + + +@pytest.mark.L1 +def test_sweep_over_a_synthetic_list_runs_the_deployed_path(tmp_path): + """End to end on a small list, so the machinery is exercised without a checkpoint.""" + custom = _write_list(tmp_path / "custom", ["deskin"]) + + assert main([custom, "--json", str(tmp_path / "out.json")]) == 0 + + written = json.loads((tmp_path / "out.json").read_text()) + assert written["results"][probe_id("deskin", "verbatim")] is True + assert written["corpus"] == corpus_id(build_probes(["deskin"])) + assert written["summary"]["total"]["blocked"] >= 1 + + +@pytest.mark.L1 +def test_sweep_uses_the_exact_match_list_beside_the_blocklist(tmp_path): + """The sweep runs is_safe, so an entry only on the exact-match list still counts.""" + (tmp_path / "custom").mkdir() + (tmp_path / "custom" / "list").write_text("deskin\n", encoding="utf-8") + (tmp_path / "exact_match").mkdir() + (tmp_path / "exact_match" / "list").write_text("chromebook\n", encoding="utf-8") + + _, is_blocked, _ = _build_is_blocked(str(tmp_path / "custom"), None, None) + + assert is_blocked("a photo of chromebook in the scene") + assert not is_blocked("a photo of a laptop in the scene") + + +@pytest.mark.L1 +@pytest.mark.parametrize( + ("name", "content"), + [ + ("no results key", {"summary": {}}), + ("empty results", {"corpus": "c", "results": {}}), + ("no corpus", {"results": {"a:spaced": True}}), + ("not an object", ["a:spaced"]), + ], +) +def test_compare_refuses_a_file_that_is_not_a_run(tmp_path, capsys, name, content): + """Found in review: a malformed file must not crash into the "lost" exit code, or compare clean.""" + good = {"corpus": "c", "results": {"a:spaced": True}} + (tmp_path / "good.json").write_text(json.dumps(good)) + (tmp_path / "bad.json").write_text(json.dumps(content)) + + assert main(["--compare", str(tmp_path / "good.json"), str(tmp_path / "bad.json")]) == 2, name + assert main(["--compare", str(tmp_path / "bad.json"), str(tmp_path / "good.json")]) == 2, name + assert "incomparable" in capsys.readouterr().out + + +@pytest.mark.L1 +def test_case_variants_of_an_entry_keep_their_own_probes(): + """Found in review: folding case in the id collapsed entries whose probes differ.""" + probes = build_probes(["Batman", "batman"]) + + assert len(probes) == 2 * len(spellings("batman")) + + +@pytest.mark.L1 +def test_a_run_records_the_lists_it_measured(tmp_path): + custom = _write_list(tmp_path / "custom", ["deskin"]) + whitelist = _write_list(tmp_path / "whitelist", ["flat"]) + + assert main([custom, "--whitelist-dir", whitelist, "--json", str(tmp_path / "out.json")]) == 0 + + setup = json.loads((tmp_path / "out.json").read_text())["setup"] + assert set(setup) == {"blocklist", "whitelist", "exact_match", "lemmatizer"} + assert "deskin" not in json.dumps(setup) diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/hygiene.py b/cosmos_framework/auxiliary/guardrail/blocklist/hygiene.py new file mode 100644 index 000000000..c14acddde --- /dev/null +++ b/cosmos_framework/auxiliary/guardrail/blocklist/hygiene.py @@ -0,0 +1,270 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +"""Hygiene check for blocklist entries that ordinary prose can trip. + +`better_profanity` catches evasions such as "n ike" by fusing adjacent tokens +and testing the fused string against the blocklist. The side effect is that a +single-word entry which is itself two ordinary English words glued together +also matches ordinary prose: the entry "deskin" blocks "a robot sitting at a +desk in the background", because "desk" + "in" fuses back into the entry. + +The check is purely syntactic -- it asks whether an entry has a two-way split +whose halves are both dictionary words -- so it flags entries whose split is +grammatical but not idiomatic ("deskinning" -> "desk" + "inning") alongside the +ones prose really produces. It is therefore meant to be read by a list owner, +not to auto-delete entries. Multi-word entries are never flagged: they cannot +be fused out of prose. + +The remedy for a flagged entry is to whitelist the phrase it collides with: +"desk in" on the whitelist exempts that one spelling while every other spelling +of the entry stays blocked. Moving the entry to the exact-match list also works +but gives up more -- it stops the fused and leetspeak spellings from matching at +all. Either way the decision belongs to the list owners. + +Two modes: + + gate check candidate entries before they are added to the list; exits + non-zero if any candidate is flagged, so it can run in CI. + + audit report which entries already on a list have the shape. + +By default the audit quotes only entries from PUBLIC_EXAMPLES, a few trademark +names that are safe to quote; every other entry is printed only with +--show-entries, so the default output can be pasted into a ticket without +reproducing content-safety terms. +""" + +import argparse +import sys +from collections.abc import Callable, Iterable +from pathlib import Path + +# A single letter is a word in WordNet ("a", "I"), which would make every entry +# splittable. Two characters is the shortest half worth reporting. +DEFAULT_MIN_PART_LEN = 2 + +# Sentence used to ask the live matcher whether a split really is blocked. The +# surrounding words are inert; only the fused pair matters. +PROBE_TEMPLATE = "a {left} {right} nearby" + +# Entries safe to name outside NVIDIA when illustrating the problem: trademarks +# and product names, no content-safety terms. +PUBLIC_EXAMPLES = frozenset({"deskin", "chromebook", "gameboy", "batman", "shotgun", "playstation"}) + + +def wordnet_vocabulary(nltk_data_dir: str | None = None) -> Callable[[str], bool]: + """Return an `is_word` predicate backed by WordNet. + + The blocklist checkpoint ships its own `nltk_data`; pass that directory to + avoid depending on a system-wide corpus. + """ + import nltk + + if nltk_data_dir: + nltk.data.path.append(str(nltk_data_dir)) + from nltk.corpus import wordnet + + wordnet.synsets("test") # fail here, not part-way through a scan + cache: dict[str, bool] = {} + + def is_word(word: str) -> bool: + if word not in cache: + cache[word] = bool(wordnet.synsets(word)) + return cache[word] + + return is_word + + +def colliding_splits( + entry: str, + is_word: Callable[[str], bool], + min_part_len: int = DEFAULT_MIN_PART_LEN, +) -> list[tuple[str, str]]: + """Every two-way split of `entry` whose halves are both dictionary words. + + Ordered longest-left-half first, which puts the split prose actually + produces ("desk in") ahead of the incidental ones ("de skin"). + """ + entry = entry.strip().lower() + if " " in entry: + return [] + splits = [ + (entry[:i], entry[i:]) + for i in range(min_part_len, len(entry) - min_part_len + 1) + if is_word(entry[:i]) and is_word(entry[i:]) + ] + return sorted(splits, key=lambda split: -len(split[0])) + + +def read_keyword_files(directory: str | Path) -> list[str]: + """Read every keyword file in a directory, exactly as the guardrail loads it. + + Delegates to the production reader, so an entry the guardrail enforces is + never invisible to the audit. That reader keeps every stripped line, + including any that start with "#"; only blank lines are dropped here, since + an empty entry cannot be split. + """ + from cosmos_framework.auxiliary.guardrail.blocklist.utils import read_keyword_list_from_dir + + return [entry for entry in read_keyword_list_from_dir(str(directory)) if entry] + + +def build_reachability_probe( + blocklist_words: Iterable[str], whitelist_words: Iterable[str] = () +) -> Callable[[str, str], bool]: + """Return a predicate telling whether the deployed path really blocks a split. + + A split is only a problem if the deployed path fires on the spaced form; + this drops splits that are theoretical for the list being checked. + + The probe runs censor_prompt, not a bare matcher. A whitelist entry that is + a phrase never reaches better_profanity -- exempting it is something + censor_prompt does on top -- so a bare matcher would keep reporting a split + as reachable after the phrase had been whitelisted, which is the remedy this + tool prints. The advice has to be checkable by the check that gives it. + """ + from cosmos_framework.auxiliary.guardrail.blocklist.blocklist import Blocklist + + probe_blocklist = Blocklist.from_lists(list(blocklist_words), list(whitelist_words)) + + def is_reachable(left: str, right: str) -> bool: + return probe_blocklist.censor_prompt(PROBE_TEMPLATE.format(left=left, right=right))[0] + + return is_reachable + + +def check_entry( + entry: str, + is_word: Callable[[str], bool], + min_part_len: int = DEFAULT_MIN_PART_LEN, + is_reachable: Callable[[str, str], bool] | None = None, +) -> list[tuple[str, str]]: + """Colliding splits for one entry, optionally confirmed against a matcher.""" + splits = colliding_splits(entry, is_word, min_part_len) + if is_reachable is not None: + splits = [split for split in splits if is_reachable(*split)] + return splits + + +def audit_entries( + entries: Iterable[str], + is_word: Callable[[str], bool], + min_part_len: int = DEFAULT_MIN_PART_LEN, + is_reachable: Callable[[str, str], bool] | None = None, +) -> list[tuple[str, list[tuple[str, str]]]]: + """Flagged entries, deduplicated and lowercased, in list order.""" + findings = [] + for entry in sorted({e.strip().lower() for e in entries}): + splits = check_entry(entry, is_word, min_part_len, is_reachable) + if splits: + findings.append((entry, splits)) + return findings + + +def _format_splits(splits: list[tuple[str, str]]) -> str: + return ", ".join(f"{left}|{right}" for left, right in splits) + + +def _cmd_gate(args, is_word, is_reachable) -> int: + flagged = 0 + for candidate in args.entry: + splits = check_entry(candidate, is_word, args.min_part_len, is_reachable) + if not splits: + print(f"ok {candidate}") + continue + flagged += 1 + left, right = splits[0] + print(f"FLAGGED {candidate} splits into dictionary words: {_format_splits(splits)}") + print(f" prose such as {PROBE_TEMPLATE.format(left=left, right=right)!r} would be blocked") + print(f" whitelist the phrase '{left} {right}' to exempt that one spelling,") + print(" or waive with a note if the split is not English anyone writes") + if flagged and not args.warn_only: + print(f"\n{flagged} of {len(args.entry)} candidate(s) flagged", file=sys.stderr) + return 1 + return 0 + + +def _cmd_audit(args, is_word, is_reachable) -> int: + entries = read_keyword_files(args.blocklist_dir) + unique = sorted({entry.strip().lower() for entry in entries}) + single = [entry for entry in unique if " " not in entry] + findings = audit_entries(single, is_word, args.min_part_len, is_reachable) + + print(f"blocklist directory : {args.blocklist_dir}") + print(f"entries (lines) : {len(entries)}") + print(f"entries (unique, lowered) : {len(unique)}") + print(f" multi-word : {len(unique) - len(single)}") + print(f" single-word : {len(single)}") + share = f" ({len(findings) / len(single):.0%} of single-word)" if single else "" + print(f"flagged single-word entries: {len(findings)}{share}") + if is_reachable is not None: + print(" confirmed against the matcher: the spaced form really is blocked") + + public = [(entry, splits) for entry, splits in findings if entry in PUBLIC_EXAMPLES] + if public: + print("\nexamples that are safe to quote outside NVIDIA:") + for entry, splits in public: + left, right = splits[0] + print(f" {entry:<14} -> '{left} {right}'") + print(f" ... and {len(findings) - len(public)} more; pass --show-entries to list them") + + if args.show_entries: + print("\nall flagged entries:") + for entry, splits in findings: + print(f" {entry:<24} {_format_splits(splits)}") + return 0 + + +def parse_args(argv=None): + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("--nltk-data", help="directory holding the WordNet corpus, e.g. the checkpoint's nltk_data") + parser.add_argument( + "--min-part-len", + type=int, + default=DEFAULT_MIN_PART_LEN, + help="shortest half that counts as a word (default: %(default)s)", + ) + parser.add_argument( + "--confirm-reachable", + metavar="BLOCKLIST_DIR", + help="keep only splits the matcher built from this directory really blocks", + ) + parser.add_argument("--whitelist-dir", help="whitelist directory, used when confirming reachability") + + sub = parser.add_subparsers(dest="mode", required=True) + gate = sub.add_parser("gate", help="check candidate entries before adding them") + gate.add_argument("entry", nargs="+") + gate.add_argument("--warn-only", action="store_true", help="report flagged candidates but exit 0") + + audit = sub.add_parser("audit", help="report collisions in an existing list") + audit.add_argument("blocklist_dir") + audit.add_argument("--show-entries", action="store_true", help="print entry text, which may include unsafe terms") + return parser.parse_args(argv) + + +def main(argv=None) -> int: + args = parse_args(argv) + is_word = wordnet_vocabulary(args.nltk_data) + + source = args.confirm_reachable or (args.blocklist_dir if args.mode == "audit" else None) + is_reachable = None + if source: + whitelist = read_keyword_files(args.whitelist_dir) if args.whitelist_dir else [] + entries = read_keyword_files(source) + if args.mode == "gate": + # The candidate is not on the list yet -- that is the point of the + # gate. Probing a matcher built only from the existing list can + # never fire on the candidate's own split, so every split would be + # discarded as unreachable and the gate would pass anything. + entries = entries + list(args.entry) + is_reachable = build_reachability_probe(entries, whitelist) + + return _cmd_gate(args, is_word, is_reachable) if args.mode == "gate" else _cmd_audit(args, is_word, is_reachable) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/hygiene_test.py b/cosmos_framework/auxiliary/guardrail/blocklist/hygiene_test.py new file mode 100644 index 000000000..a78ec6e4b --- /dev/null +++ b/cosmos_framework/auxiliary/guardrail/blocklist/hygiene_test.py @@ -0,0 +1,131 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +import pytest + +from cosmos_framework.auxiliary.guardrail.blocklist.hygiene import ( + audit_entries, + build_reachability_probe, + check_entry, + colliding_splits, + read_keyword_files, +) + +# Stand-in for WordNet so the tests need no corpus download. Only the words the +# cases below split into are members. +VOCABULARY = {"de", "desk", "in", "skin", "skinned", "inning", "chrome", "book", "game", "boy", "a"} + + +def is_word(word: str) -> bool: + return word in VOCABULARY + + +@pytest.mark.L1 +def test_entry_that_splits_into_two_words_is_flagged(): + """The reported false positive: "desk" + "in" fuses back into the entry.""" + assert ("desk", "in") in colliding_splits("deskin", is_word) + + +@pytest.mark.L1 +def test_longest_left_half_is_reported_first(): + """ "desk in" is the split prose produces; "de skin" is incidental.""" + splits = colliding_splits("deskin", is_word) + assert splits[0] == ("desk", "in") + assert ("de", "skin") in splits + + +@pytest.mark.L1 +def test_entry_without_a_dictionary_split_is_not_flagged(): + assert colliding_splits("huracan", is_word) == [] + + +@pytest.mark.L1 +def test_multi_word_entries_are_never_flagged(): + """A phrase cannot be fused out of prose, so it has no collision shape.""" + assert colliding_splits("game boy", is_word) == [] + + +@pytest.mark.L1 +def test_single_letter_halves_do_not_count(): + """ "a" is a dictionary word; counting it would flag every entry.""" + assert colliding_splits("askin", is_word) == [] + + +@pytest.mark.L1 +def test_entry_is_lowercased_and_stripped(): + assert colliding_splits(" DeskIn ", is_word) == colliding_splits("deskin", is_word) + + +@pytest.mark.L1 +def test_reachability_probe_keeps_only_splits_the_matcher_blocks(): + """A split is a problem only if the deployed matcher fires on the spaced form.""" + is_reachable = build_reachability_probe(["deskin"]) + + assert is_reachable("desk", "in") is True + assert is_reachable("chrome", "book") is False + + +@pytest.mark.L1 +def test_check_entry_drops_splits_that_the_matcher_does_not_block(): + """ "chromebook" is not on this list, so its split is theoretical here.""" + is_reachable = build_reachability_probe(["deskin"]) + + assert check_entry("deskin", is_word, is_reachable=is_reachable) == [("desk", "in"), ("de", "skin")] + assert check_entry("chromebook", is_word, is_reachable=is_reachable) == [] + + +@pytest.mark.L1 +def test_audit_deduplicates_case_variants(): + findings = audit_entries(["deskin", "DeskIn", "huracan"], is_word) + + assert [entry for entry, _ in findings] == ["deskin"] + + +@pytest.mark.L1 +def test_read_keyword_files_matches_what_the_guardrail_loads(tmp_path): + """A "#" line is live in production, so the audit has to see it too.""" + (tmp_path / "list").write_text("deskin\n\n#hashtagged\nchromebook\n", encoding="utf-8") + + assert read_keyword_files(tmp_path) == ["deskin", "#hashtagged", "chromebook"] + + +@pytest.mark.L1 +def test_probe_honours_a_whitelisted_phrase(): + """The remedy this tool prints must be visible to the tool's own probe. + + A whitelist entry that is a phrase never reaches better_profanity; exempting + it is something censor_prompt does on top. A probe built on a bare matcher + therefore kept reporting "desk in" as reachable after the phrase had been + whitelisted -- telling the caller to do something that changed nothing. + """ + without = build_reachability_probe(["deskin"]) + with_phrase = build_reachability_probe(["deskin"], ["desk in"]) + + assert without("desk", "in") is True + assert with_phrase("desk", "in") is False + # Only that one spelling is exempt; the entry's other splits still fire. + assert with_phrase("de", "skin") is True + + +@pytest.mark.L1 +def test_gate_probe_includes_the_candidate_being_gated(tmp_path): + """gate checks an entry that is not on the list yet, which is the point. + + Building the probe from the existing list alone means the matcher can never + fire on the candidate's own split, so every split is discarded as + unreachable and the gate passes anything handed to it. + """ + from cosmos_framework.auxiliary.guardrail.blocklist import hygiene + + blocklist_dir = tmp_path / "custom" + blocklist_dir.mkdir() + (blocklist_dir / "list").write_text("snow white\n", encoding="utf-8") + + argv = ["--confirm-reachable", str(blocklist_dir), "gate", "deskin"] + monkey_vocabulary = hygiene.wordnet_vocabulary + + try: + hygiene.wordnet_vocabulary = lambda _dir=None: is_word + assert hygiene.main(argv) == 1 # flagged, so non-zero exit + finally: + hygiene.wordnet_vocabulary = monkey_vocabulary