API Security — Rate Limiting, Input Sanitisation and Audit Logging

A public web application is immediately targeted by automated scanners, credential stuffers, and injection attempts. Security hardening is not a checklist item you add before launch — it is an ongoing discipline built into every layer. For the blog application, the highest-priority hardening tasks are: rate limiting to prevent brute-force and scraping, HTML sanitisation to prevent XSS from user-submitted post content, strict input validation to prevent mass assignment, and structured audit logging to detect and investigate incidents.

Rate Limiting with slowapi

pip install slowapi redis
# app/middleware/rate_limit.py
from slowapi import Limiter
from slowapi.util import get_remote_address

# Use Redis as the rate limit storage backend (shared across all workers)
limiter = Limiter(
    key_func    = get_remote_address,
    default_limits = ["200 per minute"],
    storage_uri = settings.redis_url,
)

# app/main.py — attach to the FastAPI app
from slowapi import _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded

app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# Apply limits per endpoint
@router.post("/auth/login")
@limiter.limit("10 per minute")   # strict on auth endpoints
async def login(request: Request, data: LoginRequest):
    ...

@router.get("/posts")
@limiter.limit("60 per minute")   # more generous on reads
async def list_posts(request: Request):
    ...
Note: Rate limits must be stored in Redis (not in-memory) when running multiple Uvicorn workers. An in-memory limiter means each worker has its own counter — 4 workers allow 4× the intended limit. Redis provides a shared atomic counter that all workers increment consistently. The Redis backend is also persistent across worker restarts, so a user cannot bypass limits by triggering a worker restart.
Tip: Apply different rate limits to different endpoint types: very strict for authentication (10 per minute per IP prevents brute-force), moderate for writes (post creation: 30 per hour per user), and generous for reads (post list: 200 per minute per IP). For authenticated users, rate limit by user ID rather than IP to avoid blocking legitimate users behind shared corporate NAT IPs that share one IP address with hundreds of colleagues.
Warning: Rate limiting alone is not sufficient against distributed attacks. A botnet with 1,000 IPs can each make 200 requests per minute — 200,000 total. Combine rate limiting with other defences: CAPTCHA on registration (hCaptcha, Cloudflare Turnstile), monitoring for unusual patterns (sudden spike in login failures), and IP reputation blocking (Cloudflare WAF, fail2ban). Rate limiting protects against accidental hammering and unsophisticated attacks; a proper WAF handles sophisticated ones.

HTML Sanitisation

pip install bleach
import bleach

# Allowed HTML tags and attributes for post bodies
ALLOWED_TAGS = [
    "p", "br", "strong", "em", "code", "pre", "blockquote",
    "h2", "h3", "h4", "ul", "ol", "li", "a", "img",
]
ALLOWED_ATTRS = {
    "a":   ["href", "title", "rel"],
    "img": ["src", "alt", "width", "height"],
}

def sanitise_html(raw_html: str) -> str:
    """
    Sanitise user-submitted HTML to prevent XSS.
    Called before storing post body to the database.
    """
    return bleach.clean(
        raw_html,
        tags       = ALLOWED_TAGS,
        attributes = ALLOWED_ATTRS,
        strip      = True,      # remove disallowed tags (not just escape them)
        strip_comments = True,
    )

# In the PostService.create() method:
post.body = sanitise_html(data.body)

Audit Logging

import structlog
from datetime import datetime, timezone

audit_log = structlog.get_logger("audit")

def log_audit(
    action:      str,
    user_id:     int | None,
    resource:    str,
    resource_id: int | None = None,
    details:     dict | None = None,
    request:     Request | None = None,
) -> None:
    """Write a structured audit log entry for write operations."""
    audit_log.info(
        action,
        timestamp   = datetime.now(timezone.utc).isoformat(),
        user_id     = user_id,
        resource    = resource,
        resource_id = resource_id,
        ip          = request.client.host if request else None,
        details     = details or {},
    )

# Usage — in post creation endpoint:
log_audit("post.created", current_user.id, "post", post.id,
          {"title": post.title, "status": post.status}, request)

Strict Update Validation (Prevent Mass Assignment)

# Pydantic schemas with strict update fields
class PostUpdate(BaseModel):
    title:  str  | None = None
    body:   str  | None = None
    status: str  | None = None
    # NOTE: author_id, created_at, view_count are NOT in this schema
    # A client cannot change the author or manipulate counters via PATCH

    model_config = ConfigDict(extra="forbid")   # reject any extra fields

Common Mistakes

Mistake 1 — In-memory rate limiter with multiple workers

❌ Wrong — each worker has its own counter, 4 workers = 4× the limit.

✅ Correct — use Redis as the shared rate limit storage backend.

Mistake 2 — Storing raw HTML without sanitisation

❌ Wrong — user submits <script>document.cookie</script> in post body, XSS attack when other users view the post.

✅ Correct — sanitise with bleach before storing, run DOMPurify in React before rendering.

🧠 Test Yourself

A user submits a post body containing <script>alert('XSS')</script>. With bleach.clean(body, strip=True), what is stored in the database?