Module 4: Input & Output Sanitization

2. Input Sanitization: Fundamentals

Overview

Every input that reaches your AI system is a potential vector for problems — not only malicious attacks (Module 3 covered that), but data that causes unexpected behaviors simply because of how it's formed. A user who copies text from a PDF can include invisible Unicode characters that alter the meaning of the text. An input with mixed encoding (UTF-8 and Latin-1 in the same string) can produce inconsistent embeddings. A text with 50,000 characters can drain your token budget in a single request. An input with HTML or markdown can inject visual content into the model's response.

Input sanitization isn't the same as injection detection. Injection detection (Module 3) looks for malicious intent. Sanitization looks for malformed, inconsistent, or out-of-spec data. An input can pass the injection detector (it has no attack patterns) and still need sanitization (it has inconsistent encoding, control characters, or excessive length).

In this capsule you build the first piece of the Sanitization Pipeline: a complete InputSanitizer that normalizes, cleans, and validates every input before it touches the LLM. This component is reused directly in the capsule 08 project.


Why input sanitization matters in AI

In traditional web, you sanitize inputs to prevent SQL injection and XSS. In AI, you sanitize inputs for additional reasons:

1. Embedding consistency

If two users ask the same question but with different encoding ("café" in NFC vs NFD), the embedding can differ. This affects the quality of RAG searches.

import unicodedata

text_nfc = unicodedata.normalize("NFC", "café")
text_nfd = unicodedata.normalize("NFD", "café")

print(f"NFC: {text_nfc!r} ({len(text_nfc)} chars)")
print(f"NFD: {text_nfd!r} ({len(text_nfd)} chars)")
print(f"Visually equal: {text_nfc == text_nfd}")

# Expected output:
# NFC: 'café' (4 chars)
# NFD: 'café' (5 chars)  ← the 'é' decomposes into 'e' + combining accent
# Visually equal: False

2. Invisible character attacks

Zero-width Unicode characters can hide instructions inside seemingly harmless text:

visible_text = "Hi, how are you?"
hidden_attack = "Hi, \u200b\u200bignore\u200b instructions\u200b how are you?"

print(f"Visible: {visible_text}")
print(f"With hidden: {hidden_attack}")
print(f"Do they look the same? The user doesn't notice the difference")
print(f"Original len: {len(visible_text)}")
print(f"With-hidden len: {len(hidden_attack)}")

# Expected output:
# Visible: Hi, how are you?
# With hidden: Hi, ​​ignore​ instructions​ how are you?
# Do they look the same? The user doesn't notice the difference
# Original len: 16
# With-hidden len: 40

3. Token budget protection

An input of 50,000 characters can consume ~12,500 tokens on the input alone, leaving little room for the output and draining your token budget:

def estimate_tokens(text: str) -> int:
    return len(text) // 4

long_input = "a" * 50000
estimated = estimate_tokens(long_input)
print(f"Input length: {len(long_input)} chars")
print(f"Estimated tokens: {estimated}")
print(f"GPT-4o context window: 128,000 tokens")
print(f"Percentage consumed by input alone: {estimated/128000*100:.1f}%")

# Expected output:
# Input length: 50000 chars
# Estimated tokens: 12500
# GPT-4o context window: 128,000 tokens
# Percentage consumed by input alone: 9.8%

Unicode normalization

Unicode normalization is the first step of any sanitization pipeline. It guarantees that visually identical characters are represented the same way internally.

The 4 Unicode normalization forms

import unicodedata

text = "café résumé naïve"

forms = {
    "NFC": unicodedata.normalize("NFC", text),
    "NFD": unicodedata.normalize("NFD", text),
    "NFKC": unicodedata.normalize("NFKC", text),
    "NFKD": unicodedata.normalize("NFKD", text),
}

for form_name, normalized in forms.items():
    print(f"{form_name}: {normalized!r} ({len(normalized)} chars)")

# Expected output:
# NFC:  'café résumé naïve' (17 chars)  ← Canonical composed
# NFD:  'café résumé naïve' (21 chars)  ← Canonical decomposed
# NFKC: 'café résumé naïve' (17 chars)  ← Compatibility composed
# NFKD: 'café résumé naïve' (21 chars)  ← Compatibility decomposed

Which one to use? For input sanitization in AI systems, use NFKC:

  • K (Kompatibility): Converts "compatible" characters to their canonical equivalents. For example, (ligature) → fi, 1
  • C (Composed): Keeps accented characters as a single unit. é stays as a single codepoint, not as e + combining accent
import unicodedata

tricky_inputs = [
    ("file", "fi ligature"),
    ("①②③", "Circled numbers"),
    ("Hello", "Fullwidth chars"),
    ("𝐇𝐞𝐥𝐥𝐨", "Math bold"),
    ("ℌ𝔢𝔩𝔩𝔬", "Fraktur"),
]

for text, description in tricky_inputs:
    normalized = unicodedata.normalize("NFKC", text)
    print(f"{description}:")
    print(f"  Original:   {text!r}")
    print(f"  Normalized: {normalized!r}")
    print()

# Expected output:
# fi ligature:
#   Original:   'file'
#   Normalized: 'file'
#
# Circled numbers:
#   Original:   '①②③'
#   Normalized: '123'
#
# Fullwidth chars:
#   Original:   'Hello'
#   Normalized: 'Hello'
#
# Math bold:
#   Original:   '𝐇𝐞𝐥𝐥𝐨'
#   Normalized: 'Hello'
#
# Fraktur:
#   Original:   'ℌ𝔢𝔩𝔩𝔬'
#   Normalized: 'Hello'

Attackers use these Unicode variants to bypass text filters. Your Module 3 injection detector might look for "ignore" but not detect "𝐢𝐠𝐧𝐨𝐫𝐞" (math bold). NFKC normalization resolves that before the detector sees it.


Removing dangerous characters

After normalizing Unicode, remove characters that shouldn't be in a normal text input.

Control and zero-width characters

import re

DANGEROUS_CHARS = {
    "\u200b": "Zero Width Space",
    "\u200c": "Zero Width Non-Joiner",
    "\u200d": "Zero Width Joiner",
    "\u200e": "Left-to-Right Mark",
    "\u200f": "Right-to-Left Mark",
    "\u202a": "Left-to-Right Embedding",
    "\u202b": "Right-to-Left Embedding",
    "\u202c": "Pop Directional Formatting",
    "\u202d": "Left-to-Right Override",
    "\u202e": "Right-to-Left Override",
    "\u2060": "Word Joiner",
    "\u2061": "Function Application",
    "\u2062": "Invisible Times",
    "\u2063": "Invisible Separator",
    "\u2064": "Invisible Plus",
    "\ufeff": "BOM / Zero Width No-Break Space",
}


def remove_dangerous_chars(text: str) -> tuple[str, list[str]]:
    """Removes dangerous characters and reports which ones it found."""
    found = []
    cleaned = text
    for char, name in DANGEROUS_CHARS.items():
        if char in cleaned:
            count = cleaned.count(char)
            found.append(f"{name} (x{count})")
            cleaned = cleaned.replace(char, "")

    control_pattern = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")
    control_matches = control_pattern.findall(cleaned)
    if control_matches:
        found.append(f"Control chars (x{len(control_matches)})")
        cleaned = control_pattern.sub("", cleaned)

    return cleaned, found


test_input = "Hi\u200b, how\u200d are you?\x00\x07"
cleaned, issues = remove_dangerous_chars(test_input)

print(f"Original: {test_input!r}")
print(f"Cleaned:  {cleaned!r}")
print(f"Issues:   {issues}")

# Expected output:
# Original: 'Hi\u200b, how\u200d are you?\x00\x07'
# Cleaned:  'Hi, how are you?'
# Issues:   ['Zero Width Space (x1)', 'Zero Width Joiner (x1)', 'Control chars (x2)']

Character Whitelisting vs Blacklisting

Blacklisting: block the known-bad

BLACKLISTED_PATTERNS = [
    r"<script",
    r"javascript:",
    r"on\w+=",
    r"<iframe",
    r"<object",
    r"data:text/html",
]

def blacklist_check(text: str) -> tuple[bool, list[str]]:
    """Returns (is_safe, patterns_found)."""
    import re
    found = []
    text_lower = text.lower()
    for pattern in BLACKLISTED_PATTERNS:
        if re.search(pattern, text_lower):
            found.append(pattern)
    return len(found) == 0, found

# Example
test1 = "How much does product X cost?"
test2 = "Look at this: <script>alert('xss')</script>"

safe1, issues1 = blacklist_check(test1)
safe2, issues2 = blacklist_check(test2)
print(f"'{test1[:40]}...' -> Safe: {safe1}")
print(f"'{test2[:40]}...' -> Safe: {safe2}, Issues: {issues2}")

# Expected output:
# 'How much does product X cost?...' -> Safe: True
# 'Look at this: <script>alert('xss')</scri...' -> Safe: False, Issues: ['<script']

The problem with blacklisting: It only blocks what you know. A new attack that isn't in your list passes without a problem.

Whitelisting: allow only the known-good

import re
from enum import Enum


class InputProfile(Enum):
    CHAT = "chat"
    CODE = "code"
    SEARCH = "search"


WHITELIST_PROFILES = {
    InputProfile.CHAT: {
        "pattern": r"^[\w\s\.,;:!¡?¿\-\(\)\"\'áéíóúüñÁÉÍÓÚÜÑ@#\n]+$",
        "description": "Conversational text: letters, numbers, basic punctuation",
        "max_length": 2000,
    },
    InputProfile.CODE: {
        "pattern": r"^[\w\s\.,;:!?¡¿\-\(\)\[\]\{\}\"\'`<>/\\=\+\*&|^~#@$%\n]+$",
        "description": "Code: includes programming characters",
        "max_length": 5000,
    },
    InputProfile.SEARCH: {
        "pattern": r"^[\w\s\.,\-\"\'áéíóúüñÁÉÍÓÚÜÑ]+$",
        "description": "Search: only text and minimal punctuation",
        "max_length": 500,
    },
}


def whitelist_check(text: str, profile: InputProfile) -> dict:
    config = WHITELIST_PROFILES[profile]
    result = {
        "valid": True,
        "issues": [],
        "profile": profile.value,
    }

    if len(text) > config["max_length"]:
        result["valid"] = False
        result["issues"].append(
            f"Exceeds max length: {len(text)} > {config['max_length']}"
        )

    if not re.match(config["pattern"], text, re.UNICODE):
        result["valid"] = False
        invalid_chars = set()
        for char in text:
            if not re.match(config["pattern"], char, re.UNICODE):
                invalid_chars.add(repr(char))
        result["issues"].append(
            f"Invalid chars: {', '.join(list(invalid_chars)[:5])}"
        )

    return result


print(whitelist_check("How much does it cost?", InputProfile.CHAT))
print(whitelist_check("def foo(): return 42", InputProfile.CODE))
print(whitelist_check("<script>alert(1)</script>", InputProfile.SEARCH))

# Expected output:
# {'valid': True, 'issues': [], 'profile': 'chat'}
# {'valid': True, 'issues': [], 'profile': 'code'}
# {'valid': False, 'issues': ['Invalid chars: ...'], 'profile': 'search'}

Recommendation: Use whitelisting for endpoints with predictable inputs (search, forms). Use blacklisting as an additional layer for free-text endpoints (chat). The combination of both is ideal.


Length Limits and Token Budget

Length limits aren't only security — they're resource management. Every token costs money and consumes context window.

from pydantic import BaseModel, Field, field_validator


class InputLimits(BaseModel):
    """Per-endpoint limit configuration."""
    endpoint: str
    max_chars: int
    max_estimated_tokens: int
    max_lines: int
    trim_strategy: str = "reject"


ENDPOINT_LIMITS = {
    "/chat": InputLimits(
        endpoint="/chat",
        max_chars=4000,
        max_estimated_tokens=1000,
        max_lines=50,
        trim_strategy="truncate_with_notice",
    ),
    "/search": InputLimits(
        endpoint="/search",
        max_chars=500,
        max_estimated_tokens=125,
        max_lines=1,
        trim_strategy="reject",
    ),
    "/summarize": InputLimits(
        endpoint="/summarize",
        max_chars=20000,
        max_estimated_tokens=5000,
        max_lines=500,
        trim_strategy="truncate_silent",
    ),
}


def enforce_limits(text: str, endpoint: str) -> dict:
    limits = ENDPOINT_LIMITS.get(endpoint)
    if not limits:
        return {"error": f"Unknown endpoint: {endpoint}"}

    result = {
        "original_length": len(text),
        "original_lines": text.count("\n") + 1,
        "estimated_tokens": len(text) // 4,
        "passed": True,
        "action": None,
        "text": text,
    }

    violations = []
    if len(text) > limits.max_chars:
        violations.append(f"chars: {len(text)} > {limits.max_chars}")
    if result["estimated_tokens"] > limits.max_estimated_tokens:
        violations.append(
            f"tokens: {result['estimated_tokens']} > {limits.max_estimated_tokens}"
        )
    if result["original_lines"] > limits.max_lines:
        violations.append(
            f"lines: {result['original_lines']} > {limits.max_lines}"
        )

    if violations:
        result["passed"] = False
        result["violations"] = violations

        if limits.trim_strategy == "reject":
            result["action"] = "rejected"
            result["text"] = None
        elif limits.trim_strategy == "truncate_with_notice":
            result["action"] = "truncated"
            result["text"] = text[:limits.max_chars]
            result["notice"] = (
                f"Input truncated from {len(text)} to {limits.max_chars} characters"
            )
        elif limits.trim_strategy == "truncate_silent":
            result["action"] = "truncated_silent"
            result["text"] = text[:limits.max_chars]

    return result


print(enforce_limits("How much does the iPhone cost?", "/search"))
print(enforce_limits("a" * 1000, "/search"))

# Expected output:
# {'original_length': 30, 'original_lines': 1, 'estimated_tokens': 7,
#  'passed': True, 'action': None, 'text': 'How much does the iPhone cost?'}
# {'original_length': 1000, 'original_lines': 1, 'estimated_tokens': 250,
#  'passed': False, 'action': 'rejected', 'text': None, ...}

HTML and Markdown Stripping

LLMs can interpret HTML and markdown inside the input in unexpected ways. Removing these marks prevents visual and formatting injection.

import re
try:
    import bleach
    HAS_BLEACH = True
except ImportError:
    HAS_BLEACH = False


def strip_html(text: str) -> str:
    """Removes HTML tags from the input."""
    if HAS_BLEACH:
        return bleach.clean(text, tags=[], strip=True)
    return re.sub(r"<[^>]+>", "", text)


def strip_markdown_formatting(text: str) -> str:
    """Removes markdown formatting while preserving the text."""
    cleaned = text
    cleaned = re.sub(r"!\[([^\]]*)\]\([^\)]+\)", r"\1", cleaned)
    cleaned = re.sub(r"\[([^\]]+)\]\([^\)]+\)", r"\1", cleaned)
    cleaned = re.sub(r"#{1,6}\s*", "", cleaned)
    cleaned = re.sub(r"\*\*(.+?)\*\*", r"\1", cleaned)
    cleaned = re.sub(r"\*(.+?)\*", r"\1", cleaned)
    cleaned = re.sub(r"__(.+?)__", r"\1", cleaned)
    cleaned = re.sub(r"_(.+?)_", r"\1", cleaned)
    cleaned = re.sub(r"~~(.+?)~~", r"\1", cleaned)
    cleaned = re.sub(r"`{3}[\s\S]*?`{3}", "[code block removed]", cleaned)
    cleaned = re.sub(r"`(.+?)`", r"\1", cleaned)
    return cleaned


html_input = '<div>Hello <script>alert("xss")</script> world</div>'
md_input = "# Title\n**Bold** and [link](http://evil.com) and `code`"

print(f"HTML original: {html_input}")
print(f"HTML stripped:  {strip_html(html_input)}")
print()
print(f"MD original: {md_input}")
print(f"MD stripped:  {strip_markdown_formatting(md_input)}")

# Expected output:
# HTML original: <div>Hello <script>alert("xss")</script> world</div>
# HTML stripped:  Hello alert("xss") world
#
# MD original: # Title
# **Bold** and [link](http://evil.com) and `code`
# MD stripped:  Title
# Bold and link and code

Multi-language Input Handling

AI systems receive inputs in multiple languages. Sanitization must handle different scripts without breaking them.

import unicodedata


def detect_scripts(text: str) -> dict[str, int]:
    """Detects the Unicode scripts present in the text."""
    scripts: dict[str, int] = {}
    for char in text:
        if char.isspace() or unicodedata.category(char).startswith("P"):
            continue
        try:
            script = unicodedata.name(char, "UNKNOWN").split()[0]
        except ValueError:
            script = "UNKNOWN"
        scripts[script] = scripts.get(script, 0) + 1
    return scripts


def check_script_mixing(text: str, max_scripts: int = 2) -> dict:
    """Detects suspicious mixing of Unicode scripts."""
    scripts = detect_scripts(text)
    letter_scripts = {
        k: v for k, v in scripts.items()
        if k not in ("DIGIT", "UNKNOWN")
    }

    result = {
        "scripts_found": letter_scripts,
        "num_scripts": len(letter_scripts),
        "suspicious": len(letter_scripts) > max_scripts,
    }

    if result["suspicious"]:
        result["warning"] = (
            f"Input mixes {len(letter_scripts)} scripts: "
            f"{', '.join(letter_scripts.keys())}. "
            f"Max allowed: {max_scripts}"
        )
    return result


print(check_script_mixing("Hello world"))
print(check_script_mixing("Hello world こんにちは"))
print(check_script_mixing("Hello мир 你好 مرحبا"))

# Expected output:
# {'scripts_found': {'LATIN': 10}, 'num_scripts': 1, 'suspicious': False}
# {'scripts_found': {'LATIN': 10, 'HIRAGANA': 5}, 'num_scripts': 2, 'suspicious': False}
# {'scripts_found': {'LATIN': 5, 'CYRILLIC': 3, 'CJK': 2, 'ARABIC': 5},
#  'num_scripts': 4, 'suspicious': True, 'warning': 'Input mixes 4 scripts...'}

Script mixing can be legitimate (a developer asking about an API in Japanese) or it can be a homoglyph attack (using Cyrillic 'а' that looks identical to Latin 'a' to bypass filters). The tolerance level depends on your context.


Whitespace normalization

Excessive or inconsistent whitespace can affect tokenization and cost:

import re


def normalize_whitespace(text: str) -> str:
    """Normalizes whitespace while preserving basic structure."""
    cleaned = text.replace("\t", "    ")
    cleaned = re.sub(r" {2,}", " ", cleaned)
    cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
    cleaned = re.sub(r"[ \t]+\n", "\n", cleaned)
    cleaned = cleaned.strip()
    return cleaned


messy_input = """   Hello    world   

  
  
  How    are you?   
  
  
  
  Fine, thanks.   """

cleaned = normalize_whitespace(messy_input)
print(f"Original ({len(messy_input)} chars):")
print(repr(messy_input[:100]))
print(f"\nCleaned ({len(cleaned)} chars):")
print(repr(cleaned))

# Expected output:
# Original (76 chars):
# '   Hello    world   \n\n  \n  \n  How    are you?   \n  \n  \n  \n  Fine, thanks.   '
#
# Cleaned (46 chars):
# 'Hello world\n\n\n\n How are you?\n\n\n\n Fine, thanks.'

InputSanitizer: the complete class

Now we integrate everything into a reusable InputSanitizer class that applies each step in order:

import re
import unicodedata
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

try:
    import bleach
    HAS_BLEACH = True
except ImportError:
    HAS_BLEACH = False


class SanitizationAction(Enum):
    PASS = "pass"
    CLEANED = "cleaned"
    TRUNCATED = "truncated"
    REJECTED = "rejected"


@dataclass
class SanitizationResult:
    original: str
    sanitized: Optional[str]
    action: SanitizationAction
    issues: list[str] = field(default_factory=list)
    metrics: dict = field(default_factory=dict)

    @property
    def passed(self) -> bool:
        return self.action != SanitizationAction.REJECTED


class InputSanitizer:
    ZERO_WIDTH_CHARS = set(
        "\u200b\u200c\u200d\u200e\u200f"
        "\u202a\u202b\u202c\u202d\u202e"
        "\u2060\u2061\u2062\u2063\u2064"
        "\ufeff"
    )

    CONTROL_CHAR_PATTERN = re.compile(
        r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]"
    )

    HTML_TAG_PATTERN = re.compile(r"<[^>]+>")

    def __init__(
        self,
        max_length: int = 4000,
        max_lines: int = 50,
        normalize_unicode: bool = True,
        strip_html: bool = True,
        strip_markdown: bool = False,
        remove_zero_width: bool = True,
        normalize_whitespace: bool = True,
        max_scripts: int = 3,
        on_overlength: str = "truncate",
    ):
        self.max_length = max_length
        self.max_lines = max_lines
        self.normalize_unicode = normalize_unicode
        self.strip_html = strip_html
        self.strip_markdown = strip_markdown
        self.remove_zero_width = remove_zero_width
        self.normalize_ws = normalize_whitespace
        self.max_scripts = max_scripts
        self.on_overlength = on_overlength

    def sanitize(self, text: str) -> SanitizationResult:
        if not text or not text.strip():
            return SanitizationResult(
                original=text,
                sanitized=None,
                action=SanitizationAction.REJECTED,
                issues=["Empty or whitespace-only input"],
            )

        issues: list[str] = []
        cleaned = text
        metrics = {"original_length": len(text)}

        # Step 1: Unicode normalization
        if self.normalize_unicode:
            cleaned = unicodedata.normalize("NFKC", cleaned)
            if cleaned != text:
                issues.append("Unicode normalized (NFKC)")

        # Step 2: Remove zero-width characters
        if self.remove_zero_width:
            zw_count = sum(1 for c in cleaned if c in self.ZERO_WIDTH_CHARS)
            if zw_count > 0:
                cleaned = "".join(
                    c for c in cleaned if c not in self.ZERO_WIDTH_CHARS
                )
                issues.append(f"Removed {zw_count} zero-width characters")

        # Step 3: Remove control characters
        control_matches = self.CONTROL_CHAR_PATTERN.findall(cleaned)
        if control_matches:
            cleaned = self.CONTROL_CHAR_PATTERN.sub("", cleaned)
            issues.append(
                f"Removed {len(control_matches)} control characters"
            )

        # Step 4: Strip HTML
        if self.strip_html:
            html_tags = self.HTML_TAG_PATTERN.findall(cleaned)
            if html_tags:
                if HAS_BLEACH:
                    cleaned = bleach.clean(cleaned, tags=[], strip=True)
                else:
                    cleaned = self.HTML_TAG_PATTERN.sub("", cleaned)
                issues.append(f"Stripped {len(html_tags)} HTML tags")

        # Step 5: Strip markdown (optional)
        if self.strip_markdown:
            before = cleaned
            cleaned = self._strip_markdown(cleaned)
            if cleaned != before:
                issues.append("Stripped markdown formatting")

        # Step 6: Normalize whitespace
        if self.normalize_ws:
            before_len = len(cleaned)
            cleaned = self._normalize_whitespace(cleaned)
            diff = before_len - len(cleaned)
            if diff > 0:
                issues.append(f"Normalized whitespace (saved {diff} chars)")

        # Step 7: Length limits
        metrics["cleaned_length"] = len(cleaned)
        metrics["estimated_tokens"] = len(cleaned) // 4
        metrics["line_count"] = cleaned.count("\n") + 1

        if len(cleaned) > self.max_length:
            if self.on_overlength == "reject":
                return SanitizationResult(
                    original=text,
                    sanitized=None,
                    action=SanitizationAction.REJECTED,
                    issues=[
                        f"Exceeds max length: "
                        f"{len(cleaned)} > {self.max_length}"
                    ],
                    metrics=metrics,
                )
            elif self.on_overlength == "truncate":
                cleaned = cleaned[:self.max_length]
                issues.append(
                    f"Truncated from {metrics['cleaned_length']} "
                    f"to {self.max_length} chars"
                )
                metrics["truncated"] = True

        action = (
            SanitizationAction.CLEANED if issues
            else SanitizationAction.PASS
        )
        if metrics.get("truncated"):
            action = SanitizationAction.TRUNCATED

        return SanitizationResult(
            original=text,
            sanitized=cleaned,
            action=action,
            issues=issues,
            metrics=metrics,
        )

    def _strip_markdown(self, text: str) -> str:
        cleaned = text
        cleaned = re.sub(r"!\[([^\]]*)\]\([^\)]+\)", r"\1", cleaned)
        cleaned = re.sub(r"\[([^\]]+)\]\([^\)]+\)", r"\1", cleaned)
        cleaned = re.sub(r"#{1,6}\s*", "", cleaned)
        cleaned = re.sub(r"\*\*(.+?)\*\*", r"\1", cleaned)
        cleaned = re.sub(r"__(.+?)__", r"\1", cleaned)
        cleaned = re.sub(r"\*(.+?)\*", r"\1", cleaned)
        cleaned = re.sub(r"_(.+?)_", r"\1", cleaned)
        cleaned = re.sub(r"~~(.+?)~~", r"\1", cleaned)
        cleaned = re.sub(r"`{3}[\s\S]*?`{3}", "[code removed]", cleaned)
        cleaned = re.sub(r"`(.+?)`", r"\1", cleaned)
        return cleaned

    def _normalize_whitespace(self, text: str) -> str:
        cleaned = text.replace("\t", "    ")
        cleaned = re.sub(r" {2,}", " ", cleaned)
        cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
        cleaned = re.sub(r"[ \t]+\n", "\n", cleaned)
        return cleaned.strip()


# --- Demonstration ---

sanitizer = InputSanitizer(max_length=2000, strip_html=True)

test_cases = [
    "How much does the iPhone 15 cost?",
    "Hi\u200b there\u200d with\ufeff hidden chars",
    '<script>alert("xss")</script> Laptop price?',
    "a" * 5000,
    "   ",
    "Hello file café",
]

for test in test_cases:
    result = sanitizer.sanitize(test)
    print(f"Input:  {test[:60]!r}{'...' if len(test) > 60 else ''}")
    print(f"Action: {result.action.value}")
    if result.sanitized:
        print(f"Output: {result.sanitized[:60]!r}")
    if result.issues:
        print(f"Issues: {result.issues}")
    print()

# Expected output:
# Input:  'How much does the iPhone 15 cost?'
# Action: pass
# Output: 'How much does the iPhone 15 cost?'
#
# Input:  'Hi\u200b there\u200d with\ufeff hidden chars'
# Action: cleaned
# Output: 'Hi there with hidden chars'
# Issues: ['Removed 3 zero-width characters']
#
# Input:  '<script>alert("xss")</script> Laptop price?'
# Action: cleaned
# Output: 'alert("xss") Laptop price?'
# Issues: ['Stripped 2 HTML tags']
#
# Input:  'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'...
# Action: truncated
# Output: 'aaa...' (truncated to 2000)
# Issues: ['Truncated from 5000 to 2000 chars']
#
# Input:  '   '
# Action: rejected
# Issues: ['Empty or whitespace-only input']
#
# Input:  'Hello file café'
# Action: cleaned
# Output: 'Hello file café'
# Issues: ['Unicode normalized (NFKC)']

Integration with the Sanitization Pipeline

The InputSanitizer is the first piece of the pipeline you'll build in capsule 08:

sanitizer_chat = InputSanitizer(
    max_length=4000,
    max_lines=50,
    strip_html=True,
    strip_markdown=False,
    on_overlength="truncate",
)

sanitizer_search = InputSanitizer(
    max_length=500,
    max_lines=1,
    strip_html=True,
    strip_markdown=True,
    on_overlength="reject",
)

sanitizer_summarize = InputSanitizer(
    max_length=20000,
    max_lines=500,
    strip_html=True,
    strip_markdown=False,
    on_overlength="truncate",
)

SANITIZERS = {
    "/chat": sanitizer_chat,
    "/search": sanitizer_search,
    "/summarize": sanitizer_summarize,
}

In the project, each endpoint uses a different configuration. This lets you be aggressive on search (short inputs, text only) and permissive on summarize (long inputs, formatting preserved).


Troubleshooting

Problem 1: "Unicode normalization breaks emojis"

Emojis with a Zero Width Joiner (ZWJ) like 👨\u200d💻 (man + ZWJ + computer) can break if you remove all ZWJ.

Solution: Preserve ZWJ inside emoji sequences. Detect whether the ZWJ is between emoji codepoints before removing it:

import unicodedata

def is_emoji_context(text: str, index: int) -> bool:
    if index <= 0 or index >= len(text) - 1:
        return False
    prev_cat = unicodedata.category(text[index - 1])
    next_cat = unicodedata.category(text[index + 1])
    return prev_cat == "So" or next_cat == "So"

Problem 2: "Length limits block legitimate inputs in Asian languages"

CJK characters (Chinese, Japanese, Korean) use more bytes but convey more information per character. A limit of 500 characters in Chinese is equivalent to ~250 words, whereas in English it's equivalent to ~100 words.

Solution: Use token-based limits instead of character-based limits for multilingual endpoints:

def adaptive_length_limit(text: str, max_tokens: int = 500) -> bool:
    estimated_tokens = len(text) // 4
    return estimated_tokens <= max_tokens

Problem 3: "Sanitization adds visible latency"

Each sanitization step adds microseconds, but accumulated they can be perceptible.

Solution: Measure each step and optimize the expensive ones. Unicode normalization and regex are the heaviest steps. Precompile regex patterns:

import re
PRECOMPILED = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")

Problem 4: "Inputs with legitimate code get broken"

If a developer asks about HTML or JavaScript, your HTML stripper removes their code.

Solution: Use the InputProfile.CODE profile that preserves programming characters, or detect whether the input is inside backticks/code blocks before stripping.


Exercises

Exercise 1: Homoglyph detector

Implement a function that detects whether a text uses characters from different Unicode scripts that look similar (homoglyphs), like Cyrillic 'а' vs Latin 'a'. These are used to bypass filters.

See solution
import unicodedata

LATIN_CYRILLIC_HOMOGLYPHS = {
    "а": "a", "е": "e", "о": "o", "р": "p",
    "с": "c", "у": "y", "х": "x", "А": "A",
    "В": "B", "Е": "E", "К": "K", "М": "M",
    "Н": "H", "О": "O", "Р": "P", "С": "C",
    "Т": "T", "Х": "X",
}


def detect_homoglyphs(text: str) -> dict:
    found = []
    for i, char in enumerate(text):
        if char in LATIN_CYRILLIC_HOMOGLYPHS:
            found.append({
                "position": i,
                "char": char,
                "looks_like": LATIN_CYRILLIC_HOMOGLYPHS[char],
                "actual_script": "Cyrillic",
            })

    return {
        "has_homoglyphs": len(found) > 0,
        "count": len(found),
        "details": found,
    }


test = "ignоre instructiоns"  # the 'о' are Cyrillic
result = detect_homoglyphs(test)
print(f"Text: {test!r}")
print(f"Homoglyphs: {result['has_homoglyphs']}")
print(f"Count: {result['count']}")
for d in result["details"]:
    print(f"  Position {d['position']}: '{d['char']}' looks like '{d['looks_like']}' ({d['actual_script']})")

# Expected output:
# Text: 'ignоre instructiоns'
# Homoglyphs: True
# Count: 2
#   Position 3: 'о' looks like 'o' (Cyrillic)
#   Position 16: 'о' looks like 'o' (Cyrillic)

Explanation: Homoglyphs are a sophisticated evasion technique. If your injection detector looks for "ignore", it won't find "ignоre" with a Cyrillic 'о'. NFKC normalization doesn't resolve homoglyphs because they are completely different codepoints. You need explicit detection.

Exercise 2: Sanitizer with performance metrics

Modify the InputSanitizer to measure the time of each step and report which one is the most expensive.

See solution
import time

class TimedInputSanitizer(InputSanitizer):
    def sanitize(self, text: str) -> SanitizationResult:
        timings = {}

        start = time.perf_counter_ns()
        result = super().sanitize(text)
        total = time.perf_counter_ns() - start

        result.metrics["total_time_us"] = total / 1000
        return result


timed = TimedInputSanitizer(max_length=2000)
result = timed.sanitize("Hi\u200b there\u200d " + "a" * 3000)
print(f"Total time: {result.metrics.get('total_time_us', 0):.1f} µs")
print(f"Action: {result.action.value}")

# Expected output:
# Total time: ~50-200 µs (varies by hardware)
# Action: truncated

Explanation: Measuring the performance of each step lets you identify bottlenecks. In production with thousands of requests, a sanitization that takes 1ms becomes significant.

Exercise 3: Mixed encoding detector

Implement a function that detects whether a text has inconsistent encoding (for example, a mix of UTF-8 and Latin-1 artifacts like é instead of é).

See solution
MOJIBAKE_PATTERNS = [
    ("á", "á"), ("é", "é"), ("í", "í"),
    ("ó", "ó"), ("ú", "ú"), ("ñ", "ñ"),
    ("ü", "ü"), ("¿", "¿"), ("¡", "¡"),
]

def detect_and_fix_mojibake(text: str) -> dict:
    fixes = []
    fixed = text
    for broken, correct in MOJIBAKE_PATTERNS:
        if broken in fixed:
            count = fixed.count(broken)
            fixes.append(f"{broken!r}{correct!r} (x{count})")
            fixed = fixed.replace(broken, correct)

    return {
        "had_mojibake": len(fixes) > 0,
        "fixes": fixes,
        "original": text,
        "fixed": fixed,
    }

test = "Cómo estás? ¿Qué pasa?"
result = detect_and_fix_mojibake(test)
print(f"Original: {result['original']}")
print(f"Fixed:    {result['fixed']}")
print(f"Fixes:    {result['fixes']}")

# Expected output:
# Original: Cómo estás? ¿Qué pasa?
# Fixed:    Cómo estás? ¿Qué pasa?
# Fixes:    ["'á' → 'á' (x1)", "'é' → 'é' (x1)", "'ó' → 'ó' (x1)", "'¿' → '¿' (x1)"]

Explanation: Mojibake (encoding artifacts) is common when users copy text from PDFs, emails, or web pages with incorrect encoding. Detecting and fixing these patterns improves input quality without rejecting it.

Exercise 4: InputSanitizer with file-based configuration

Create a version of the InputSanitizer that loads its configuration from a dictionary (simulating a configuration file).

See solution
SANITIZER_CONFIGS = {
    "strict": {
        "max_length": 500,
        "max_lines": 5,
        "strip_html": True,
        "strip_markdown": True,
        "on_overlength": "reject",
    },
    "moderate": {
        "max_length": 4000,
        "max_lines": 50,
        "strip_html": True,
        "strip_markdown": False,
        "on_overlength": "truncate",
    },
    "permissive": {
        "max_length": 20000,
        "max_lines": 500,
        "strip_html": False,
        "strip_markdown": False,
        "on_overlength": "truncate",
    },
}


def create_sanitizer(profile: str) -> InputSanitizer:
    config = SANITIZER_CONFIGS.get(profile)
    if not config:
        raise ValueError(f"Unknown profile: {profile}. Options: {list(SANITIZER_CONFIGS.keys())}")
    return InputSanitizer(**config)


strict = create_sanitizer("strict")
moderate = create_sanitizer("moderate")

test = "Hello <b>world</b>! " * 100
print(f"Strict:   {strict.sanitize(test).action.value}")
print(f"Moderate: {moderate.sanitize(test).action.value}")

# Expected output:
# Strict:   rejected
# Moderate: cleaned

Explanation: In production, you want to change the sanitization configuration without redeploying code. File-based profiles let you adjust thresholds dynamically or per endpoint.

Exercise 5: Sanitizer with audit log

Add an audit logging system to the InputSanitizer that records each sanitization with a timestamp, action, and issues found.

See solution
import json
from datetime import datetime, timezone


class AuditedSanitizer:
    def __init__(self, sanitizer: InputSanitizer):
        self.sanitizer = sanitizer
        self.audit_log: list[dict] = []

    def sanitize(self, text: str, request_id: str = "unknown") -> SanitizationResult:
        result = self.sanitizer.sanitize(text)

        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "request_id": request_id,
            "action": result.action.value,
            "issues": result.issues,
            "original_length": len(text),
            "sanitized_length": len(result.sanitized) if result.sanitized else 0,
        }
        self.audit_log.append(entry)
        return result

    def get_stats(self) -> dict:
        if not self.audit_log:
            return {"total": 0}
        actions = [e["action"] for e in self.audit_log]
        return {
            "total": len(actions),
            "passed": actions.count("pass"),
            "cleaned": actions.count("cleaned"),
            "truncated": actions.count("truncated"),
            "rejected": actions.count("rejected"),
        }


audited = AuditedSanitizer(InputSanitizer(max_length=100))
audited.sanitize("Hello world", "req-001")
audited.sanitize("<b>Bold</b> text", "req-002")
audited.sanitize("a" * 200, "req-003")
audited.sanitize("   ", "req-004")

print(json.dumps(audited.get_stats(), indent=2))

# Expected output:
# {
#   "total": 4,
#   "passed": 1,
#   "cleaned": 1,
#   "truncated": 1,
#   "rejected": 1
# }

Explanation: The audit log is fundamental for calibrating your sanitization in production. If 30% of inputs are rejected, your limits are too strict. If 0% are cleaned, you might have insufficient defenses. The metrics guide the tuning.


Summary

  • 🔑 Input sanitization is different from injection detection: sanitization cleans malformed data; injection detection looks for intentional attacks
  • 🔑 NFKC normalization is the mandatory first step — it converts Unicode variants (fullwidth, ligatures, math symbols) to their canonical forms, closing a filter-evasion vector
  • 🔑 Zero-width characters (ZWJ, ZWNJ, ZWS) can hide invisible instructions inside seemingly harmless text
  • 🔑 Whitelisting is safer than blacklisting — it allows only the known-good instead of blocking the known-bad
  • 🔑 Length limits aren't only security — they're resource management: every character consumes tokens that cost money and context window
  • 🔑 The InputSanitizer applies steps in order: Unicode normalization → removal of dangerous characters → strip HTML → whitespace normalization → length limits
  • 🔑 Sanitization must be configurable per endpoint: a search endpoint needs strict limits, a summarize endpoint needs to be permissive
  • 🔑 The central trade-off is security vs usability: every restriction you add potentially blocks legitimate inputs
  • 🔑 Audit logging is essential for calibrating sanitization in production: the metrics tell you whether you're too strict or too permissive

Additional resources

  1. Unicode Security Considerations (TR#36) — Official Unicode report on attacks based on Unicode properties, fundamental for understanding why normalization matters
  2. Unicode Normalization Forms (TR#15) — Technical specification of NFC, NFD, NFKC, NFKD with detailed examples
  3. OWASP Input Validation Cheat Sheet — Input validation guide with principles applicable to AI input sanitization
  4. Bleach Documentation — HTML sanitization library for Python, useful for stripping HTML from inputs
  5. Confusable Detection (Unicode) — Official algorithm for detecting homoglyphs/confusables between Unicode scripts
  6. OWASP LLM05: Improper Output Handling — Context on why input sanitization complements output validation
  7. Python unicodedata Module — Official documentation of Python's module for Unicode manipulation
  8. Invisible Characters — A Complete Reference — Reference of invisible Unicode characters with explanations of each one

Created: March 2026 Version: 1.0