diff --git a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py index aa4f4e171..71bb453bb 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist.py @@ -4,12 +4,15 @@ 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 from cosmos_framework.auxiliary.guardrail.common.core import ( GUARDRAIL1_CHECKPOINT, @@ -36,6 +39,38 @@ # 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)) + +# 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.""" + + @staticmethod + def lemmatize(token: str) -> str: + return token + class Blocklist(ContentSafetyGuardrail): def __init__( @@ -57,16 +92,76 @@ 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")), + ) - self.profanity.load_censor_words(custom_words=self.blocklist_words, whitelist_words=self.whitelist_words) + @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 + + # 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()] + + 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. @@ -86,11 +181,138 @@ 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 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)) + + @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()) + @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: + # 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 ( + 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]] | 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 + 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. 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 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 + 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] + 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. @@ -108,15 +330,57 @@ 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. + 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, "" + + 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( @@ -226,9 +490,66 @@ 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. + + * 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 + 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 (normalized, joined, fold_lookalikes(normalized), fold_lookalikes(joined)): + if spelling != prompt 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..499e7f13c 100644 --- a/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py +++ b/cosmos_framework/auxiliary/guardrail/blocklist/blocklist_test.py @@ -47,20 +47,47 @@ def test_partial_match_with_threshold(): assert match is False -def _blocklist_with_words(words: list[str], whitelist: list[str] | None = None) -> Blocklist: +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, exact_match: 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, + exact_match, + 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 +303,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 +322,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 blocked is True, f"{name} was not folded away" + 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 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 +418,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 +429,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 +449,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.""" @@ -347,3 +587,375 @@ 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}" + + +@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)] + + +@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 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 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