Lesson 6: Capstone — Harden a Production AI App

Build a security-hardened FastAPI wrapper around the Anthropic API that passes a full red-team test battery: prompt injection defense, PII redaction, multi-tenant isolation, and a GDPR erasure endpoint — all with structured audit logs.

User request
 └─ normalize_input() — strip zero-width, normalize Unicode
 └─ is_injection_attempt() — LLM-based detector (block if YES)
 └─ redact_pii() — Presidio scan, tokenize PII before sending to LLM
 └─ tenant_chat() — scoped system prompt + tenant vector store
 └─ detokenize_response() — restore PII tokens in output
 └─ log_ai_decision() — append-only compliance audit log
 └─ Response to user

Phase 1 — Input Defense Layer

Implement normalize_input(text): strip zero-width characters (U+200B–U+200D), normalize to Unicode NFKC, detect and remove base64-encoded instructions.
Implement is_injection_attempt(text) using a haiku classifier. Return 403 with a generic error if the classifier returns YES.
Add XML wrapping: <user_input>{message}</user_input> around all user content before it reaches the system prompt boundary.
Write 10 injection test cases (direct, indirect, encoding variants) and assert all are blocked. Add these to your test suite as regression tests.

Phase 2 — PII Redaction Pipeline

Install Presidio and spaCy. Run detect_pii() on all user inputs before they reach the LLM. Log detected entity types (NOT values) for monitoring.
Implement PiiVault: tokenize PII before the LLM call, detokenize the response before returning to the user. Verify round-trip integrity with 5 test cases.
Scan LLM responses for PII with Presidio. If a response contains unredacted PII that the user didn't provide, log a pii_leakage_alert event and redact before returning.

Phase 3 — Multi-Tenant Isolation

Create two ChromaDB collections: tenant_acme and tenant_beta. Ingest unique documents into each (e.g., ACME's pricing, BETA's pricing).
Implement tenant_chat(ctx, messages, query) using a tenant-scoped system prompt and retrieve_for_tenant().
Write a cross-tenant penetration test: while authenticated as tenant_beta, ask about tenant_acme's pricing. Assert the response contains no tenant_acme data.
Verify the semantic cache key includes tenant_id hash. Confirm the same query returns different responses for different tenants (no cache bleeding).

Phase 4 — Compliance Infrastructure

Implement log_ai_decision() writing structured JSON to an append-only file. Fields: timestamp, user_id_hash, tenant_id, model, query_hash, context_doc_count, latency_ms.
Implement POST /users/{user_id}/erase — GDPR erasure endpoint. Deletes: vector store documents, conversation history, cache entries. Returns a signed erasure report.
Test the erasure endpoint: ingest 3 documents for user X, call the erasure endpoint, then verify retrieval returns 0 results for that user.

# Full secure pipeline — FastAPI handler
from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.post("/secure-chat")
async def secure_chat(request: SecureChatRequest) -> SecureChatResponse:
    # Phase 1: Input defense
    clean_input = normalize_input(request.message)
    if is_injection_attempt(clean_input):
        raise HTTPException(status_code=403, detail="Request blocked by security filter.")

    # Phase 2: PII tokenization
    vault = PiiVault()
    safe_input = vault.tokenize(clean_input)

    # Phase 3: Tenant-scoped LLM call
    ctx = TenantContext(
        tenant_id=request.tenant_id,
        user_id=request.user_id,
        tier=get_user_tier(request.user_id),
    )
    raw_response = tenant_chat(ctx, request.history, safe_input)

    # Phase 2: Detokenize + scan output
    response = vault.detokenize(raw_response)
    if detect_pii(response):
        log_compliance_event({"event": "pii_in_output", "tenant_id": ctx.tenant_id})
        response = redact_pii(response)

    # Phase 4: Audit log
    log_ai_decision(
        user_id=request.user_id,
        tenant_id=request.tenant_id,
        query=request.message,
        retrieved_context=[],   # Fill from tenant_chat
        model="claude-haiku-4-5-20251001",
        response=response,
        latency_ms=0,           # Fill with timing
    )
    return SecureChatResponse(answer=response)

Graduation criteria: All 10 injection test cases blocked, cross-tenant penetration test shows 0 data leakage, erasure endpoint verified to delete all user data, and every request generates a valid audit log entry with no raw PII.

User request └─ normalize_input() — strip zero-width, normalize Unicode └─ is_injection_attempt() — LLM-based detector (block if YES) └─ redact_pii() — Presidio scan, tokenize PII before sending to LLM └─ tenant_chat() — scoped system prompt + tenant vector store └─ detokenize_response() — restore PII tokens in output └─ log_ai_decision() — append-only compliance audit log └─ Response to user

# Full secure pipeline — FastAPI handler from fastapi import FastAPI, HTTPException app = FastAPI() @app.post("/secure-chat") async def secure_chat(request: SecureChatRequest) -> SecureChatResponse: # Phase 1: Input defense clean_input = normalize_input(request.message) if is_injection_attempt(clean_input): raise HTTPException(status_code=403, detail="Request blocked by security filter.") # Phase 2: PII tokenization vault = PiiVault() safe_input = vault.tokenize(clean_input) # Phase 3: Tenant-scoped LLM call ctx = TenantContext( tenant_id=request.tenant_id, user_id=request.user_id, tier=get_user_tier(request.user_id), ) raw_response = tenant_chat(ctx, request.history, safe_input) # Phase 2: Detokenize + scan output response = vault.detokenize(raw_response) if detect_pii(response): log_compliance_event({"event": "pii_in_output", "tenant_id": ctx.tenant_id}) response = redact_pii(response) # Phase 4: Audit log log_ai_decision( user_id=request.user_id, tenant_id=request.tenant_id, query=request.message, retrieved_context=[], # Fill from tenant_chat model="claude-haiku-4-5-20251001", response=response, latency_ms=0, # Fill with timing ) return SecureChatResponse(answer=response)