2. Input Sanitization

Description

Input sanitization is the first layer of the guardrails pipeline. It normalizes and cleans the text before sending it to the LLM: removes control characters, normalizes spaces, applies length limits with real token counting, and validates encoding. It's the cheapest layer to implement (<1ms of latency, $0) and prevents an entire class of problems: volume attacks, malicious characters, and inconsistencies in the input.

What sanitization solves

Without sanitization, these inputs reach the LLM unmodified:

# Input 1: Control characters that confuse the tokenizer
"Analyze this text\x00\x01\x08 and give me a result"
# The NULL character and other control chars can cause unexpected behavior

# Input 2: Volume attack (economic DoS)
"a" * 500_000  # ~125K tokens → ~$0.02 per request × 1000 requests = $20 in minutes
# Without a length limit → a malicious user empties your account

# Input 3: Invalid encoding that crashes the parser
b"texto\xff\xfe".decode("latin-1")  # Invalid bytes for UTF-8
# Can cause a UnicodeDecodeError during processing

# Input 4: Multiple spaces that consume extra tokens
"This  text   has    many     spaces"
# 5 word tokens + 8 space tokens = unnecessary cost

Complete sanitizer with all layers

# src/guardrails/input_sanitizer.py
import re
import unicodedata
from typing import Optional

class InputSanitizationResult:
    """Result of sanitization with metadata."""
    def __init__(
        self,
        text: str,
        was_truncated: bool = False,
        original_length: int = 0,
        had_control_chars: bool = False,
        had_invalid_encoding: bool = False
    ):
        self.text = text
        self.was_truncated = was_truncated
        self.original_length = original_length
        self.had_control_chars = had_control_chars
        self.had_invalid_encoding = had_invalid_encoding
    
    def __bool__(self):
        return bool(self.text)
    
    @property
    def was_modified(self) -> bool:
        return self.was_truncated or self.had_control_chars or self.had_invalid_encoding

def sanitize_input(
    text: str,
    max_chars: int = 40_000,
    normalize_spaces: bool = True,
    strip_control_chars: bool = True,
    fix_encoding: bool = True
) -> InputSanitizationResult:
    """
    Sanitizes the user input before sending it to the LLM.
    
    Pipeline:
    1. Validate type and emptiness
    2. Fix encoding (if enabled)
    3. Normalize unicode (NFKC)
    4. Remove control characters
    5. Normalize spaces
    6. Truncate by maximum length
    
    Args:
        text: The user input
        max_chars: Maximum characters (default: 40K ≈ 10K tokens)
        normalize_spaces: Collapse multiple spaces into one
        strip_control_chars: Remove ASCII control characters
        fix_encoding: Repair invalid UTF-8 bytes
    
    Returns:
        InputSanitizationResult with the clean text and metadata
    """
    if not text or not isinstance(text, str):
        return InputSanitizationResult(text="")
    
    original_length = len(text)
    had_control_chars = False
    had_invalid_encoding = False
    
    # Step 1: Fix encoding (remove invalid UTF-8 bytes)
    if fix_encoding:
        encoded = text.encode("utf-8", errors="replace")
        fixed = encoded.decode("utf-8", errors="replace")
        if fixed != text:
            had_invalid_encoding = True
        text = fixed
    
    # Step 2: Normalize Unicode to NFKC
    # NFKC converts: ① → 1, fi → fi, ½ → 1/2
    # Useful for normalizing text from varied sources
    text = unicodedata.normalize("NFKC", text)
    
    # Step 3: Remove ASCII control characters
    if strip_control_chars:
        # Remove: NUL (0x00), BEL (0x07), BS (0x08), FF (0x0C), etc.
        # KEEP: TAB (0x09), LF (0x0A), CR (0x0D) — they are valid line breaks
        original = text
        text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)
        if text != original:
            had_control_chars = True
    
    # Step 4: Normalize spaces
    if normalize_spaces:
        # Collapse multiple spaces into one (but not newlines)
        text = re.sub(r' {2,}', ' ', text)
        # Trim spaces at the start and end
        text = text.strip()
    
    # Step 5: Truncate by maximum length
    was_truncated = False
    if len(text) > max_chars:
        text = text[:max_chars]
        was_truncated = True
    
    return InputSanitizationResult(
        text=text,
        was_truncated=was_truncated,
        original_length=original_length,
        had_control_chars=had_control_chars,
        had_invalid_encoding=had_invalid_encoding
    )

Counting tokens with tiktoken (precise limits)

Character limits are an approximation. For precise limits, use tiktoken:

# pip install tiktoken

import tiktoken
from functools import lru_cache

@lru_cache(maxsize=4)
def get_encoder(model: str = "gpt-4o-mini") -> tiktoken.Encoding:
    """Loads the encoder once and caches it."""
    try:
        return tiktoken.encoding_for_model(model)
    except KeyError:
        # Fallback to the standard encoding if the model isn't in tiktoken
        return tiktoken.get_encoding("cl100k_base")

def count_tokens(text: str, model: str = "gpt-4o-mini") -> int:
    """Counts the exact tokens for a specific model."""
    encoder = get_encoder(model)
    return len(encoder.encode(text))

def truncate_to_tokens(
    text: str,
    max_tokens: int = 3_000,
    model: str = "gpt-4o-mini"
) -> tuple[str, int]:
    """
    Truncates text so it doesn't exceed max_tokens.
    
    Returns:
        (truncated_text, tokens_used)
    """
    encoder = get_encoder(model)
    tokens = encoder.encode(text)
    
    if len(tokens) <= max_tokens:
        return text, len(tokens)
    
    truncated_tokens = tokens[:max_tokens]
    truncated_text = encoder.decode(truncated_tokens)
    return truncated_text, max_tokens

# Sanitization function with a token limit:
def sanitize_with_token_limit(
    text: str,
    max_tokens: int = 3_000,
    model: str = "gpt-4o-mini"
) -> InputSanitizationResult:
    """Sanitizes and then truncates to a specific number of tokens."""
    # First sanitize (cleans the text)
    result = sanitize_input(text, max_chars=max_tokens * 6)  # 6 chars/token as a buffer
    
    if not result.text:
        return result
    
    # Then truncate exactly by tokens
    truncated, tokens_used = truncate_to_tokens(result.text, max_tokens, model)
    
    if truncated != result.text:
        result.text = truncated
        result.was_truncated = True
    
    return result

Length rules by model

# Maximum limits by model:
MODEL_CONTEXT_LIMITS = {
    "gpt-4o-mini":  128_000,  # total context tokens
    "gpt-4o":       128_000,
    "gpt-4":         8_192,
    "gpt-3.5-turbo": 16_385,
    "claude-3-haiku": 200_000,
}

# Practical rule to calculate max input tokens:
def calculate_max_input_tokens(
    model: str,
    system_prompt: str,
    reserve_output_tokens: int = 500
) -> int:
    """
    Calculates the maximum allowed input tokens.
    
    max_input = context_limit - system_prompt_tokens - output_reserve
    """
    max_context = MODEL_CONTEXT_LIMITS.get(model, 8_192)
    system_tokens = count_tokens(system_prompt, model)
    
    return max_context - system_tokens - reserve_output_tokens

# Real example:
SYSTEM_PROMPT = "You are a sentiment analyzer. Respond with JSON."
MAX_INPUT = calculate_max_input_tokens("gpt-4o-mini", SYSTEM_PROMPT, 500)
# → 128_000 - 10 - 500 = 127_490 tokens available for input
# → In practice: use 2_000-4_000 for cost reasons, not for the limit

Limits by cost, not just by capacity

The maximum context is not the practical limit — cost is:

# Context limit: 128K tokens
# Cost limit for an /analyze endpoint:
#   - Input: 3_000 tokens × $0.15/1M = $0.00045
#   - With 10K requests/day: $4.50/day ← OK

# With 50K-token input:
#   - Input: 50_000 tokens × $0.15/1M = $0.0075 per request
#   - With 10K requests/day: $75/day ← Might be too expensive

# Recommendation by endpoint type:
ENDPOINT_TOKEN_LIMITS = {
    "/analyze": 3_000,           # Short analysis texts
    "/summarize": 8_000,         # Medium documents
    "/summarize-long": 32_000,   # Long documents
    "/chat": 4_000,              # Conversation (per turn)
}

Complete sanitizer tests

# tests/unit/guardrails/test_input_sanitizer.py
import pytest
from src.guardrails.input_sanitizer import sanitize_input, count_tokens

class TestSanitizeInput:
    """Tests for the input sanitizer."""
    
    # ─── Happy path ───────────────────────────────────────────────
    
    def test_normal_text_unchanged(self):
        result = sanitize_input("Hi, how are you?")
        assert result.text == "Hi, how are you?"
        assert not result.was_modified
    
    def test_trims_leading_trailing_spaces(self):
        result = sanitize_input("  text with spaces  ")
        assert result.text == "text with spaces"
    
    # ─── Control characters ───────────────────────────────────────
    
    def test_removes_null_bytes(self):
        result = sanitize_input("hello\x00world")
        assert "\x00" not in result.text
        assert result.had_control_chars
    
    def test_removes_control_chars(self):
        malicious = "text\x01\x02\x03normal"
        result = sanitize_input(malicious)
        assert "\x01" not in result.text
        assert "\x02" not in result.text
        assert "textnormal" == result.text
    
    def test_preserves_newlines(self):
        """Line breaks are valid and must be preserved."""
        text = "line 1\nline 2\r\nline 3"
        result = sanitize_input(text)
        assert "\n" in result.text  # LF is preserved
    
    def test_preserves_tabs(self):
        text = "column1\tcolumn2"
        result = sanitize_input(text)
        assert "\t" in result.text
    
    # ─── Space normalization ──────────────────────────────────────
    
    def test_collapses_multiple_spaces(self):
        result = sanitize_input("text  with   many    spaces")
        assert result.text == "text with many spaces"
    
    # ─── Length limits ────────────────────────────────────────────
    
    def test_truncates_at_max_chars(self):
        long_text = "a" * 100_000
        result = sanitize_input(long_text, max_chars=1_000)
        assert len(result.text) == 1_000
        assert result.was_truncated
    
    def test_short_text_not_truncated(self):
        result = sanitize_input("short text", max_chars=1_000)
        assert not result.was_truncated
    
    # ─── Empty / None ─────────────────────────────────────────────
    
    def test_empty_string_returns_empty(self):
        result = sanitize_input("")
        assert result.text == ""
        assert not bool(result)
    
    def test_whitespace_only_returns_empty(self):
        result = sanitize_input("   \t\n   ")
        assert result.text == ""
    
    def test_none_returns_empty(self):
        result = sanitize_input(None)
        assert result.text == ""
    
    # ─── Unicode ──────────────────────────────────────────────────
    
    def test_preserves_spanish_chars(self):
        text = "Comunicación en español con ñ, tildes: á é í ó ú"
        result = sanitize_input(text)
        assert "ñ" in result.text
        assert "á" in result.text
    
    def test_preserves_emojis(self):
        text = "I love 😊 this product 🎉"
        result = sanitize_input(text)
        assert "😊" in result.text
    
    def test_normalizes_unicode_nfkc(self):
        """Equivalent Unicode characters are normalized to canonical form."""
        # ① (circled digit one) → 1
        text = "Important ① point"
        result = sanitize_input(text)
        assert "1" in result.text  # Normalized to "1"

class TestCountTokens:
    def test_approximate_token_count(self):
        """A short word ≈ 1 token."""
        assert 1 <= count_tokens("hello") <= 3
    
    def test_long_text_has_more_tokens(self):
        short = count_tokens("hi")
        long = count_tokens("hello world this is a long sentence with many words")
        assert long > short

Integration with FastAPI

# src/app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from src.guardrails.input_sanitizer import sanitize_with_token_limit

app = FastAPI()

class AnalyzeRequest(BaseModel):
    text: str = Field(min_length=1, max_length=200_000)  # Basic Pydantic pre-validation

@app.post("/analyze")
async def analyze_endpoint(request: AnalyzeRequest):
    # Sanitize before passing to the LLM
    sanitized = sanitize_with_token_limit(request.text, max_tokens=3_000)
    
    if not sanitized:
        raise HTTPException(status_code=400, detail="Empty or invalid input")
    
    if sanitized.was_truncated:
        # Optionally: notify the user that it was truncated
        # Or simply proceed without warning (depends on the UX)
        pass
    
    result = analyze_sentiment(sanitized.text)
    return result

Exercises

Exercise 1: Implement a minimal sanitize_input

Implement a version of sanitize_input that only does: trim, collapse spaces, and limit to 5000 characters:

def sanitize_input(text: str, max_chars: int = 5000) -> str:
    if not text or not isinstance(text, str):
        return ""
    # Trim + collapse spaces
    cleaned = " ".join(text.strip().split())
    # Length limit
    return cleaned[:max_chars]

# Basic tests:
assert sanitize_input("  hello  world  ") == "hello world"
assert sanitize_input("") == ""
assert len(sanitize_input("a" * 10000)) == 5000

Exercise 2: Control character test

Write 5 tests for control character filtering. Include: NULL byte, BEL, mixed text, preserve newlines, preserve tabs:

def test_null_byte_removed():
    assert "\x00" not in sanitize_input("text\x00")

def test_bel_removed():
    assert "\x07" not in sanitize_input("text\x07normal")

def test_mixed_control_and_normal():
    result = sanitize_input("hello\x01world")
    assert result == "helloworld"  # Control char removed, no space

def test_newline_preserved():
    assert "\n" in sanitize_input("line1\nline2")

def test_tab_preserved():
    assert "\t" in sanitize_input("col1\tcol2")

Exercise 3: Calculate the token limit for your app

Your app has:

• Model: gpt-4o-mini
• System prompt: "You are a sentiment analyzer. Respond ONLY with JSON."
• Maximum output: 200 tokens

How many tokens can you use for the user input? In approximate characters?

# Count the system prompt's tokens:
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o-mini")
system_tokens = len(enc.encode("You are a sentiment analyzer. Respond ONLY with JSON."))
# ≈ 11 tokens

# gpt-4o-mini context limit: 128,000 tokens
# max_input = 128,000 - 11 (system) - 200 (output) = 127,789 tokens available

# In practice (due to cost):
# 3,000 tokens per request × $0.15/1M = $0.00045 per request
# With 100K requests/day: $45/day — acceptable

# In approximate characters (4 chars/token for English):
# 3,000 tokens × 4 = 12,000 characters → max_chars = 12_000

Exercise 4: Sanitization for different input types

For each input type, decide what level of sanitization you need:

1. Free-form chat text from the user
2. Python code submitted by the user
3. Parsed PDF document (extracted text)
4. URL submitted by the user

Exercise 5: Sanitization logging

Implement logging for when a modified input is sanitized:

import logging
import hashlib

logger = logging.getLogger("guardrails.sanitizer")

def sanitize_and_log(text: str, user_id: str = None) -> InputSanitizationResult:
    result = sanitize_input(text)
    
    if result.was_modified:
        # Log without including the original content (privacy)
        input_hash = hashlib.sha256(text.encode()).hexdigest()[:8]
        logger.info(
            "input_sanitized",
            extra={
                "input_hash": input_hash,
                "original_length": result.original_length,
                "cleaned_length": len(result.text),
                "was_truncated": result.was_truncated,
                "had_control_chars": result.had_control_chars,
                "user_id": user_id
            }
        )
    
    return result

Summary

• Sanitization = first layer of the pipeline: normalize, clean, limit length
• 5-step pipeline: fix encoding → normalize unicode → remove control chars → collapse spaces → truncate
• Token limit is more precise than character limit — use tiktoken for production apps
• The practical limit is cost, not technical capacity — define it according to your per-request budget
• Mandatory tests: NULL bytes, empty text, very long text, valid special characters (ñ, emojis)
• Be conservative: better not to remove valid characters than to remove too many

Additional resources

1. tiktoken — OpenAI's official tokenizer for counting exact tokens
2. OpenAI Tokenizer (web) — Visualize tokens interactively
3. Unicode NFKC normalization — Why normalize unicode
4. OWASP Input Validation Cheat Sheet — Complete validation guide
5. Python unicodedata — Standard module for unicode
6. Python re module — For the control char patterns