daddy/modbison
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add pluggable server-side TextFilter service (social.md §4 US-4)

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y2FESCaMXeJpC9gSctiEeD
This commit is contained in:
daddy
2026-09-15 00:32:17 +00:00
parent f40940a491
commit 9334caf533
3 changed files with 518 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/text_filter.py
+304
View File
@@ -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)
backend/app/services/text_filter_terms.json
+20
View File
@@ -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"
]
}