From 9334caf533b6651f5c468ddaed747623240e252c Mon Sep 17 00:00:00 2001 From: daddy Date: Tue, 15 Sep 2026 00:32:17 +0000 Subject: [PATCH] =?UTF-8?q?Add=20pluggable=20server-side=20TextFilter=20se?= =?UTF-8?q?rvice=20(social.md=20=C2=A74=20US-4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deterministic default (ListBasedTextFilter) backed by a configurable JSON term list, with normalization (case-fold, leetspeak substitutions, stripped punctuation/spacing between letters) so trivial evasion doesn't slip through. Supports both reject (blocking) and redact (asterisk replacement) modes. TextFilter is a Protocol, leaving an explicit extension point for a future third-party moderation provider. Standalone module, not wired into any router/endpoint -- sibling chat tasks depend on this one. No DB/migration work. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Y2FESCaMXeJpC9gSctiEeD --- backend/app/services/text_filter.py | 304 ++++++++++++++++++++ backend/app/services/text_filter_terms.json | 20 ++ backend/tests/test_text_filter.py | 194 +++++++++++++ 3 files changed, 518 insertions(+) create mode 100644 backend/app/services/text_filter.py create mode 100644 backend/app/services/text_filter_terms.json create mode 100644 backend/tests/test_text_filter.py diff --git a/backend/app/services/text_filter.py b/backend/app/services/text_filter.py new file mode 100644 index 0000000..107c795 --- /dev/null +++ b/backend/app/services/text_filter.py @@ -0,0 +1,304 @@ +"""Pluggable server-side text filtering, `docs/planning/social.md` §4/§5 (US-4). + +social.md §4 calls for a "Pluggable `TextFilter` interface with a +deterministic default (profanity list + normalization) so it runs without +external services; third-party hook later". This module is that interface +plus the default implementation: + +- `TextFilter` is a `Protocol` -- the extension point. A future third-party + moderation provider (a paid API, an ML classifier, etc.) just needs to + implement `check(text) -> FilterResult`; nothing else in this module + needs to change, and callers depend on the protocol, not a concrete + class. +- `ListBasedTextFilter` is the deterministic default: a configurable + term list (JSON file, see `text_filter_terms.json` next to this module) + matched against a normalized form of the input text so that trivial + evasion (case changes, leetspeak substitutions, and punctuation/spacing + inserted between letters) doesn't slip through. + +Nothing here talks to a database, an HTTP client, or the FastAPI app -- it +is a pure, side-effect-free (aside from reading the term list file once, at +construction) service so it can be unit tested in isolation. It is NOT +wired into any router/endpoint by this task; that's for the chat feature +tasks that depend on it (social.md §6) to do. + +Normalization approach +----------------------- +Matching naively against raw text misses "sh1t", "@ss", or "s.h.i.t". To +catch these without a real tokenizer, every character of the input is +independently normalized to at most one lowercase ASCII alphanumeric +character (case-folded, then run through a small leetspeak substitution +table); characters that don't reduce to `[a-z0-9]` (punctuation, spacing, +symbols, non-ASCII) are dropped entirely rather than kept as separators. +This collapses "s.h.i.t", "s h i t", and "SHIT" to the same normalized +string ("shit") that a plain lowercase term list entry matches, while a +side table (`index_map`) remembers which original character produced each +surviving normalized character -- so a match found in the normalized +string can still be translated back into a span of the *original* text for +redaction. + +This is a deliberate precision/recall tradeoff common to list-based +filters: stripping separators and folding leetspeak increases recall +against evasion at the cost of occasionally matching a term inside an +unrelated longer word (e.g. a 3-letter term inside a longer benign word). +That tradeoff is exactly why this is pluggable -- a deployment that needs +tighter precision can swap in a smarter `TextFilter` implementation behind +the same protocol. +""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from enum import Enum +from pathlib import Path +from typing import Iterable, Protocol, runtime_checkable + +# --------------------------------------------------------------------------- +# Public result / protocol +# --------------------------------------------------------------------------- + + +class FilterAction(str, Enum): + """What the caller should do with a checked message.""" + + ALLOW = "allow" + REDACT = "redact" + REJECT = "reject" + + +@dataclass(frozen=True) +class FilterResult: + """Outcome of running a piece of text through a `TextFilter`. + + - `action`: ALLOW (no matches), REDACT (matches replaced in + `filtered_text`), or REJECT (matches found; the message should be + blocked -- callers should not persist/deliver it even though + `filtered_text` is populated for logging/audit purposes). + - `filtered_text`: the text to actually store/deliver. Equal to the + input for ALLOW and REJECT; has matched spans replaced for REDACT. + - `matched_terms`: the distinct matched terms (original, un-normalized + list entries), in first-occurrence order. Empty for ALLOW. + """ + + action: FilterAction + filtered_text: str + matched_terms: tuple[str, ...] = () + + +@runtime_checkable +class TextFilter(Protocol): + """Extension point for text filtering/moderation providers. + + Any implementation -- the deterministic default below, or a future + third-party service call -- just needs this one method. + """ + + def check(self, text: str) -> FilterResult: + """Check `text` and return the action to take plus filtered text.""" + ... + + +class FilterMode(str, Enum): + """Configures what `ListBasedTextFilter` does when it finds a match.""" + + REDACT = "redact" + REJECT = "reject" + + +# --------------------------------------------------------------------------- +# Normalization +# --------------------------------------------------------------------------- + +# Common leetspeak / lookalike substitutions, applied after case-folding. +# Keys are already-lowercased single characters. +_LEET_MAP: dict[str, str] = { + "0": "o", + "1": "i", + "!": "i", + "3": "e", + "4": "a", + "@": "a", + "5": "s", + "$": "s", + "7": "t", + "+": "t", + "8": "b", + "|": "l", +} + +_ALNUM_RE = re.compile(r"[a-z0-9]") + + +def _normalize(text: str) -> tuple[str, list[int]]: + """Fold `text` to a lowercase-alnum-only string plus an index map. + + Returns `(normalized, index_map)` where `index_map[i]` is the index + into the *original* `text` that produced `normalized[i]`. Characters + that don't reduce to a single `[a-z0-9]` character (punctuation, + whitespace, symbols, most non-ASCII) are dropped from both outputs. + """ + normalized_chars: list[str] = [] + index_map: list[int] = [] + for i, ch in enumerate(text): + folded = ch.casefold() + # case-fold can expand to multiple characters for some unicode + # (e.g. German "ß" -> "ss"); such characters are not part of the + # leetspeak table and are not plain ascii alnum, so just skip them + # like any other non-matching character rather than guessing. + if len(folded) != 1: + continue + folded = _LEET_MAP.get(folded, folded) + if _ALNUM_RE.fullmatch(folded): + normalized_chars.append(folded) + index_map.append(i) + return "".join(normalized_chars), index_map + + +def _normalize_term(term: str) -> str: + normalized, _ = _normalize(term) + return normalized + + +# --------------------------------------------------------------------------- +# Default term list loading +# --------------------------------------------------------------------------- + +DEFAULT_TERMS_PATH = Path(__file__).with_name("text_filter_terms.json") + + +def load_terms(path: Path | str | None = None) -> list[str]: + """Load a term list from a JSON file shaped `{"terms": [...]}`. + + Any other top-level keys (e.g. `_readme`) are ignored. Defaults to the + bundled `text_filter_terms.json` next to this module. + """ + resolved = Path(path) if path is not None else DEFAULT_TERMS_PATH + with resolved.open("r", encoding="utf-8") as f: + data = json.load(f) + terms = data.get("terms", []) + if not isinstance(terms, list): + raise ValueError(f"{resolved}: 'terms' must be a list") + return [str(t) for t in terms] + + +# --------------------------------------------------------------------------- +# Span merging / redaction helpers +# --------------------------------------------------------------------------- + + +def _merge_spans(spans: list[tuple[int, int, str]]) -> list[tuple[int, int]]: + """Merge overlapping/adjacent (start, end) spans, sorted by start.""" + if not spans: + return [] + ordered = sorted((s, e) for s, e, _ in spans) + merged: list[tuple[int, int]] = [ordered[0]] + for start, end in ordered[1:]: + last_start, last_end = merged[-1] + if start <= last_end: + merged[-1] = (last_start, max(last_end, end)) + else: + merged.append((start, end)) + return merged + + +def _apply_redactions(text: str, merged_spans: list[tuple[int, int]], redaction_char: str) -> str: + if not merged_spans: + return text + pieces: list[str] = [] + cursor = 0 + for start, end in merged_spans: + pieces.append(text[cursor:start]) + pieces.append(redaction_char * (end - start)) + cursor = end + pieces.append(text[cursor:]) + return "".join(pieces) + + +# --------------------------------------------------------------------------- +# Default implementation +# --------------------------------------------------------------------------- + + +class ListBasedTextFilter: + """Deterministic `TextFilter`: normalized substring match against a term list. + + `mode` controls what happens on a match: + - `FilterMode.REDACT` (default): matched spans in the *original* text + are replaced with `redaction_char` repeated to the matched span's + original length; `action` is REDACT if anything matched, else ALLOW. + - `FilterMode.REJECT`: `filtered_text` is returned unchanged but + `action` is REJECT if anything matched (callers should treat this as + "do not store/deliver"), else ALLOW. + + An empty or whitespace-only string always returns ALLOW with no + matches -- there is nothing to filter. + """ + + def __init__( + self, + terms: Iterable[str] | None = None, + *, + mode: FilterMode = FilterMode.REDACT, + terms_path: Path | str | None = None, + redaction_char: str = "*", + ) -> None: + if terms is None: + terms = load_terms(terms_path) + # De-dupe while preserving order; drop blanks. + seen: dict[str, None] = {} + for t in terms: + t = t.strip() + if t: + seen.setdefault(t, None) + self._terms: tuple[str, ...] = tuple(seen) + # Pre-normalize once at construction; skip terms that normalize to + # nothing (e.g. a term that's pure punctuation). + self._normalized_terms: list[tuple[str, str]] = [ + (normalized, original) + for original in self._terms + if (normalized := _normalize_term(original)) + ] + self.mode = mode + self.redaction_char = redaction_char + + def check(self, text: str) -> FilterResult: + if not text or not text.strip(): + return FilterResult(action=FilterAction.ALLOW, filtered_text=text, matched_terms=()) + + normalized, index_map = _normalize(text) + if not normalized: + return FilterResult(action=FilterAction.ALLOW, filtered_text=text, matched_terms=()) + + spans: list[tuple[int, int, str]] = [] + for normalized_term, original_term in self._normalized_terms: + start = 0 + term_len = len(normalized_term) + while True: + idx = normalized.find(normalized_term, start) + if idx == -1: + break + end = idx + term_len + orig_start = index_map[idx] + orig_end = index_map[end - 1] + 1 + spans.append((orig_start, orig_end, original_term)) + start = end # non-overlapping: resume after this match + + if not spans: + return FilterResult(action=FilterAction.ALLOW, filtered_text=text, matched_terms=()) + + spans.sort(key=lambda s: s[0]) + matched_terms = tuple(dict.fromkeys(term for _, _, term in spans)) + + if self.mode is FilterMode.REJECT: + return FilterResult(action=FilterAction.REJECT, filtered_text=text, matched_terms=matched_terms) + + merged = _merge_spans(spans) + filtered_text = _apply_redactions(text, merged, self.redaction_char) + return FilterResult(action=FilterAction.REDACT, filtered_text=filtered_text, matched_terms=matched_terms) + + +def default_text_filter(mode: FilterMode = FilterMode.REDACT) -> ListBasedTextFilter: + """Convenience factory for the bundled default term list.""" + return ListBasedTextFilter(mode=mode) diff --git a/backend/app/services/text_filter_terms.json b/backend/app/services/text_filter_terms.json new file mode 100644 index 0000000..8fa788c --- /dev/null +++ b/backend/app/services/text_filter_terms.json @@ -0,0 +1,20 @@ +{ + "_readme": "Default deterministic term list for app.services.text_filter.ListBasedTextFilter. This is a minimal, illustrative starter list (common mild-to-strong English profanity, no slurs) meant to be replaced/extended per-deployment -- swap this file, or pass an explicit `terms`/`terms_path` to the filter, rather than editing this list in place for production moderation policy.", + "terms": [ + "fuck", + "shit", + "ass", + "asshole", + "bitch", + "bastard", + "damn", + "crap", + "dick", + "piss", + "slut", + "whore", + "cunt", + "douche", + "bullshit" + ] +} diff --git a/backend/tests/test_text_filter.py b/backend/tests/test_text_filter.py new file mode 100644 index 0000000..ebd363c --- /dev/null +++ b/backend/tests/test_text_filter.py @@ -0,0 +1,194 @@ +"""Unit tests for app.services.text_filter (docs/planning/social.md §4/§5 US-4). + +Pure, side-effect-free service -- no DB/Redis fixtures needed. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from app.services.text_filter import ( + DEFAULT_TERMS_PATH, + FilterAction, + FilterMode, + FilterResult, + ListBasedTextFilter, + TextFilter, + default_text_filter, + load_terms, +) + + +@pytest.fixture +def redact_filter() -> ListBasedTextFilter: + return ListBasedTextFilter(terms=["badword", "shoot"], mode=FilterMode.REDACT) + + +@pytest.fixture +def reject_filter() -> ListBasedTextFilter: + return ListBasedTextFilter(terms=["badword", "shoot"], mode=FilterMode.REJECT) + + +class TestAllowPath: + def test_clean_text_is_allowed_unchanged(self, redact_filter: ListBasedTextFilter) -> None: + result = redact_filter.check("hello there, how are you today?") + assert result.action is FilterAction.ALLOW + assert result.filtered_text == "hello there, how are you today?" + assert result.matched_terms == () + + def test_message_with_no_matches(self, reject_filter: ListBasedTextFilter) -> None: + # Distinct from the clean-text case above: exercises the reject-mode + # filter's no-match path specifically (mode should not matter when + # nothing matches). + text = "let's meet at the plaza around noon" + result = reject_filter.check(text) + assert result.action is FilterAction.ALLOW + assert result.filtered_text == text + assert result.matched_terms == () + + +class TestRedactPath: + def test_exact_match_is_redacted(self, redact_filter: ListBasedTextFilter) -> None: + result = redact_filter.check("don't be a badword about it") + assert result.action is FilterAction.REDACT + assert result.matched_terms == ("badword",) + assert "badword" not in result.filtered_text + assert result.filtered_text == "don't be a ******* about it" + + def test_multiple_distinct_matches(self, redact_filter: ListBasedTextFilter) -> None: + result = redact_filter.check("badword, don't shoot!") + assert result.action is FilterAction.REDACT + assert set(result.matched_terms) == {"badword", "shoot"} + assert "badword" not in result.filtered_text + assert "shoot" not in result.filtered_text + # Length is preserved (asterisks replace matched spans 1:1). + assert len(result.filtered_text) == len("badword, don't shoot!") + + def test_redaction_char_is_configurable(self) -> None: + f = ListBasedTextFilter(terms=["badword"], mode=FilterMode.REDACT, redaction_char="#") + result = f.check("that's a badword thing to say") + assert result.filtered_text == "that's a ####### thing to say" + + +class TestRejectPath: + def test_match_is_rejected_and_text_left_intact(self, reject_filter: ListBasedTextFilter) -> None: + text = "stop, badword!" + result = reject_filter.check(text) + assert result.action is FilterAction.REJECT + assert result.matched_terms == ("badword",) + # Reject mode does not mutate the text -- callers are expected to + # not store/deliver it at all, based on `action`. + assert result.filtered_text == text + + +class TestNormalization: + def test_leetspeak_variant_is_caught(self, redact_filter: ListBasedTextFilter) -> None: + # "b4dw0rd" -> normalizes to "badword" via the leet map (4->a, 0->o). + result = redact_filter.check("you're being a b4dw0rd right now") + assert result.action is FilterAction.REDACT + assert result.matched_terms == ("badword",) + + def test_separators_between_letters_are_stripped(self, redact_filter: ListBasedTextFilter) -> None: + result = redact_filter.check("that is b.a.d.w.o.r.d behavior") + assert result.action is FilterAction.REDACT + assert result.matched_terms == ("badword",) + + def test_spacing_between_letters_is_stripped(self, redact_filter: ListBasedTextFilter) -> None: + result = redact_filter.check("b a d w o r d, seriously") + assert result.action is FilterAction.REDACT + assert result.matched_terms == ("badword",) + + def test_case_folding(self, redact_filter: ListBasedTextFilter) -> None: + result = redact_filter.check("BADWORD in all caps") + assert result.action is FilterAction.REDACT + assert result.matched_terms == ("badword",) + + def test_symbol_lookalikes(self) -> None: + # @ -> a, $ -> s: "@ss" should normalize to "ass". + f = ListBasedTextFilter(terms=["ass"], mode=FilterMode.REDACT) + result = f.check("what an @$$ move") + assert result.action is FilterAction.REDACT + assert result.matched_terms == ("ass",) + + +class TestEmptyAndWhitespace: + def test_empty_string(self, redact_filter: ListBasedTextFilter) -> None: + result = redact_filter.check("") + assert result.action is FilterAction.ALLOW + assert result.filtered_text == "" + assert result.matched_terms == () + + def test_whitespace_only_string(self, redact_filter: ListBasedTextFilter) -> None: + text = " \t\n " + result = redact_filter.check(text) + assert result.action is FilterAction.ALLOW + assert result.filtered_text == text + assert result.matched_terms == () + + def test_punctuation_only_string_has_nothing_to_normalize(self, redact_filter: ListBasedTextFilter) -> None: + # No alnum characters survive normalization at all -- must not + # crash on an empty normalized string / empty index_map. + result = redact_filter.check("...!!!???") + assert result.action is FilterAction.ALLOW + assert result.matched_terms == () + + +class TestConfigurability: + def test_custom_term_list_is_used_instead_of_default(self) -> None: + f = ListBasedTextFilter(terms=["zzzcustomzzz"], mode=FilterMode.REDACT) + result = f.check("this contains zzzcustomzzz right here") + assert result.action is FilterAction.REDACT + assert result.matched_terms == ("zzzcustomzzz",) + # Default list terms should NOT match since we overrode the list. + result2 = f.check("this is fucking unrelated") + assert result2.action is FilterAction.ALLOW + + def test_duplicate_and_blank_terms_are_deduped(self) -> None: + f = ListBasedTextFilter(terms=["badword", "badword", " ", ""], mode=FilterMode.REDACT) + assert f._terms == ("badword",) + + def test_default_terms_file_loads_and_is_nonempty(self) -> None: + terms = load_terms() + assert isinstance(terms, list) + assert len(terms) > 0 + assert all(isinstance(t, str) and t for t in terms) + + def test_default_terms_path_exists(self) -> None: + assert DEFAULT_TERMS_PATH.exists() + assert DEFAULT_TERMS_PATH.suffix == ".json" + + def test_load_terms_from_explicit_path(self, tmp_path: Path) -> None: + custom = tmp_path / "custom_terms.json" + custom.write_text('{"terms": ["foo", "bar"]}', encoding="utf-8") + assert load_terms(custom) == ["foo", "bar"] + + def test_default_text_filter_factory_matches_default_list(self) -> None: + f = default_text_filter(mode=FilterMode.REJECT) + result = f.check("that's fucking ridiculous") + assert result.action is FilterAction.REJECT + assert "fuck" in result.matched_terms + + +class TestProtocol: + def test_list_based_filter_satisfies_text_filter_protocol(self, redact_filter: ListBasedTextFilter) -> None: + assert isinstance(redact_filter, TextFilter) + + def test_check_returns_filter_result(self, redact_filter: ListBasedTextFilter) -> None: + result = redact_filter.check("hi") + assert isinstance(result, FilterResult) + + def test_a_third_party_style_implementation_satisfies_the_protocol(self) -> None: + # Demonstrates the extension point: any object with a matching + # `check` method satisfies TextFilter without inheriting from + # anything -- this is how a future third-party provider would plug + # in without touching this module. + class AlwaysAllow: + def check(self, text: str) -> FilterResult: + return FilterResult(action=FilterAction.ALLOW, filtered_text=text, matched_terms=()) + + provider: TextFilter = AlwaysAllow() + assert isinstance(provider, TextFilter) + result = provider.check("anything at all, even fuck") + assert result.action is FilterAction.ALLOW