Security Patterns

Security Patterns

Comprehensive security patterns for building hardened applications. Each category has individual rule files in rules/ loaded on-demand.

Quick Reference

Total: 20 rules across 8 categories

Quick Start

# Argon2id password hashing
from argon2 import PasswordHasher
ph = PasswordHasher()
password_hash = ph.hash(password)
ph.verify(password_hash, password)

# JWT access token (15-min expiry)
import jwt
from datetime import datetime, timedelta, timezone
payload = {
    'sub': user_id, 'type': 'access',
    'exp': datetime.now(timezone.utc) + timedelta(minutes=15),
}
token = jwt.encode(payload, SECRET_KEY, algorithm='HS256')

// Zod v4 schema validation
import { z } from 'zod';
const UserSchema = z.object({
  email: z.email(),
  name: z.string().min(2).max(100),
  role: z.enum(['user', 'admin']).default('user'),
});
const result = UserSchema.safeParse(req.body);

# PII masking with Langfuse
import re
from langfuse import Langfuse

def mask_pii(data, **kwargs):
    if isinstance(data, str):
        data = re.sub(r'\b[\w.-]+@[\w.-]+\.\w+\b', '[REDACTED_EMAIL]', data)
        data = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', '[REDACTED_SSN]', data)
    return data

langfuse = Langfuse(mask=mask_pii)

Authentication

Secure authentication with OAuth 2.1, Passkeys/WebAuthn, JWT tokens, and role-based access control.

Key Decisions: Argon2id > bcrypt | Access tokens 15 min | PKCE required | Passkeys > TOTP > SMS

Defense-in-Depth

Multi-layer security architecture with no single point of failure.

Key Decisions: Immutable dataclass context | Query-level tenant filtering | No IDs in LLM prompts

sandbox.network.deniedDomains (CC 2.1.113+)

Network-layer blocklist enforced before Bash/WebFetch egress — pair with the hook-layer DENY_PATTERNS for defense in depth. Settings example:

"sandbox": {
  "network": {
    "deniedDomains": ["*.evil.com", "pastebin.com", "transfer.sh"]
  }
}

Wildcards supported (*.example.com, evil.com/*/malicious/*). Plugins ship a baseline list in src/settings/ork.settings.json; project settings can extend it. Use for: prompt-injection exfil sinks, known-bad registries, paste services that bypass audit.

Input Validation

Validate and sanitize all untrusted input using Zod v4 and Pydantic.

Key Decisions: Allowlist over blocklist | Server-side always | Validate magic bytes not extensions

OWASP Top 10

Protection against the most critical web application security risks.

Key Decisions: Parameterized queries only | Hardcode JWT algorithm | SameSite=Strict cookies

LLM Safety

Security patterns for LLM integrations including context separation and output validation.

Key Decisions: IDs flow around LLM, never through | Attribution is deterministic | Audit every prompt

Context Separation (CRITICAL)

Sensitive IDs and data flow AROUND the LLM, never through it. The LLM sees only content — mapping back to entities happens deterministically after.

# CORRECT: IDs bypass the LLM
context = {"user_id": user_id, "tenant_id": tenant_id}  # kept server-side
llm_input = f"Summarize this document:\n{doc_text}"       # no IDs in prompt
llm_output = call_llm(llm_input)
result = {"summary": llm_output, **context}               # IDs reattached after

Output Validation Pipeline

Every LLM response MUST pass a 4-stage guardrail pipeline before reaching the user:

def validate_llm_output(raw_output: str, schema, sources: list[str]) -> str:
    # 1. Schema — does it match expected structure?
    parsed = schema.parse(raw_output)
    # 2. Grounding — are claims supported by source documents?
    assert_grounded(parsed, sources)
    # 3. Safety — toxicity, PII leakage, prompt leakage
    assert_safe(parsed, max_toxicity=0.5)
    # 4. Size — prevent token-bomb responses
    assert len(parsed.text) < MAX_OUTPUT_CHARS
    return parsed.text

PII Masking

PII detection and masking for LLM observability pipelines and logging.

Key Decisions: Presidio for enterprise | Replace with type tokens | Use mask callback at init

Scanning

Automated security scanning for dependencies, code, and secrets.

Key Decisions: Pre-commit hooks for shift-left | Block on critical/high | Gitleaks + detect-secrets baseline

Advanced Guardrails

Production LLM safety with NeMo Guardrails, Guardrails AI validators, and DeepTeam red-teaming.

Key Decisions: NeMo for flows, Guardrails AI for validators | Toxicity 0.5 threshold | Red-team pre-release + quarterly

Managed Hook Hierarchy (CC 2.1.49)

Plugin settings follow a 3-tier precedence:

Security hooks shipped by OrchestKit are managed defaults — users can disable them but are warned. Enterprise admins can lock settings via managed profiles.

CC 2.1.166 — managed-settings enforcement fix: before 2.1.166 a single invalid entry in managed settings silently disabled enforcement of all remaining valid policies — one typo could void your entire security lockdown. Require 2.1.166+ when relying on managed profiles, and validate the file before deploying it. The same release fixed allowedMcpServers/deniedMcpServers predicates not matching when they use ${VAR} references.

CC 2.1.160 — write prompts: Claude Code now prompts before writing shell startup files (.zshenv, .zlogin, .bash_login, ~/.config/git/) and — under acceptEdits — build-tool configs that grant code execution (.npmrc, .yarnrc*, bunfig.toml, .bazelrc, .pre-commit-config.yaml, .devcontainer/). Treat these as defense-in-depth defaults: approve deliberately rather than blanket-allowing.

Permission-rule semantics (≥ 2.1.166): allow/ask/deny rules gained security-relevant behavior — Read deny now hides files from Glob/Grep, deny tool-names accept globs ("*" = default-deny), explicit WebFetch(domain:…) overrides the preapproved-host auto-allow, relayed SendMessage from other sessions carries no authority, and org-managed rules apply for the whole session. See references/cc-permission-model.md for the full model + a recommended baseline settings.json.

Anti-Patterns (FORBIDDEN)

# Authentication
user.password = request.form['password']       # Plaintext password storage
response_type=token                             # Implicit OAuth grant (deprecated)
return "Email not found"                        # Information disclosure

# Input Validation
"SELECT * FROM users WHERE name = '" + name + "'"  # SQL injection
if (file.type === 'image/png') {...}               # Trusting Content-Type header

# LLM Safety
prompt = f"Analyze for user {user_id}"             # ID in prompt
artifact.user_id = llm_output["user_id"]           # Trusting LLM-generated IDs

# PII
logger.info(f"User email: {user.email}")           # Raw PII in logs
langfuse.trace(input=raw_prompt)                   # Unmasked observability data

Detailed Documentation

Load on demand with Read("${CLAUDE_SKILL_DIR}/references/<file>"):

Related Skills

• api-design-framework - API security patterns
• ork:rag-retrieval - RAG pipeline patterns requiring tenant-scoped retrieval
• llm-evaluation - Output quality assessment including hallucination detection

Capability Details

authentication

Keywords: password, hashing, JWT, token, OAuth, PKCE, passkey, WebAuthn, RBAC, session Solves:

• Implement secure authentication with modern standards
• JWT token management with proper expiry
• OAuth 2.1 with PKCE flow
• Passkeys/WebAuthn registration and login
• Role-based access control

defense-in-depth

Keywords: defense in depth, security layers, multi-layer, request context, tenant isolation Solves:

• How to secure AI applications end-to-end
• Implement 8-layer security architecture
• Create immutable request context
• Ensure tenant isolation at query level

cc-subprocess-hardening (CC 2.1.98)

Keywords: subprocess, sandbox, PID namespace, env scrub, script caps Solves:

• Limit runaway hook scripts: CLAUDE_CODE_SCRIPT_CAPS=100
• Strip credentials from subprocesses: CLAUDE_CODE_SUBPROCESS_ENV_SCRUB=1
• PID namespace isolation on Linux for subprocess sandboxing
• Prevent Bash permission bypasses via backslash escapes and compound commands

CC 2.1.128 — SDK host "Always allow" persistence: when a user picks "Always allow" from a Bash permission prompt in an SDK host, the grant now persists via .claude/settings.local.json instead of evaporating at session end. Audit your SDK consumers' .gitignore to confirm .claude/settings.local.json is excluded — committing it leaks per-developer Bash auth grants. Project-committed .claude/settings.json is unchanged; only the user-machine-local file receives the new entries.

CC 2.1.169 — managed MCP enforcement + OTEL cert-path trust: two policy-bypass classes closed. Enterprise allowedMcpServers/deniedMcpServers policies were NOT enforced on reconnect, IDE-typed configs, --mcp-config servers in the first post-install session, or before remote settings loaded — treat any pre-2.1.169 managed-MCP audit as incomplete on those paths. And untrusted project settings could set OTEL client-certificate paths without trust confirmation (a cloned repo could point telemetry at an attacker cert); now gated behind trust. Both fixes are active at ork's floor (2.1.170).

CC 2.1.163 — home-path deny rules now cover `$HOME` Bash refs: before this fix a Read(~/.ssh/**)-style deny rule blocked the Read tool but NOT a Bash command that reached the same file via $HOME/.ssh/... — a silent secrets-read bypass. If you gate home-directory secrets (e.g. ~/.aws/credentials, ~/.ssh/*, ~/.gnupg/*) through permission deny rules, pin your CC floor to >= 2.1.163; older builds (< 2.1.163) leave the Bash path open — ork's floor is now 2.1.170, which already includes this fix.

input-validation

Keywords: schema, validate, Zod, Pydantic, sanitize, HTML, XSS, file upload Solves:

• Validate input against schemas (Zod v4, Pydantic)
• Prevent injection attacks with allowlists
• Sanitize HTML and prevent XSS
• Validate file uploads by magic bytes

owasp-top-10

Keywords: OWASP, sql injection, broken access control, CSRF, XSS, SSRF Solves:

• Fix OWASP Top 10 vulnerabilities
• Prevent SQL and command injection
• Implement CSRF protection
• Fix broken authentication

llm-safety

Keywords: prompt injection, context separation, guardrails, hallucination, LLM output Solves:

• Prevent prompt injection attacks
• Implement context separation (IDs around LLM)
• Validate LLM output with guardrail pipeline
• Deterministic post-LLM attribution

pii-masking

Keywords: PII, masking, Presidio, Langfuse, redact, GDPR, privacy Solves:

• Detect and mask PII in LLM pipelines
• Integrate masking with Langfuse observability
• Implement pre-logging redaction
• GDPR-compliant data handling

Authentication Security Checklist

Password Security

• [ ] Use Argon2id (preferred) or bcrypt for hashing
• [ ] Minimum 12 character password requirement
• [ ] Check against common password lists
• [ ] No password hints or security questions
• [ ] Rate limit password attempts (5 per minute)
• [ ] Account lockout after 10 failed attempts

Token Security

• [ ] Access tokens: 15 min - 1 hour lifetime
• [ ] Refresh tokens: 7-30 days with rotation
• [ ] Store access tokens in memory only (not localStorage)
• [ ] Store refresh tokens in HTTPOnly cookies
• [ ] Implement refresh token rotation
• [ ] Revoke all tokens on password change

Session Security

• [ ] SESSION_COOKIE_SECURE=True (HTTPS only)
• [ ] SESSION_COOKIE_HTTPONLY=True (no JS access)
• [ ] SESSION_COOKIE_SAMESITE='Strict'
• [ ] Session timeout (1 hour inactivity)
• [ ] Regenerate session ID on login

OAuth 2.1 Compliance

• [ ] Use PKCE for ALL clients
• [ ] No implicit grant
• [ ] No password grant
• [ ] State parameter for CSRF protection
• [ ] Validate redirect_uri exactly
• [ ] Use HTTPS for all endpoints

Passkeys/WebAuthn (If Implemented)

• [ ] Require user verification (biometric)
• [ ] Require resident keys for passwordless
• [ ] Validate RP ID matches origin
• [ ] Track sign count for replay protection
• [ ] Allow multiple passkeys per user

Multi-Factor Authentication

• [ ] Offer MFA (TOTP, Passkeys)
• [ ] TOTP: 6 digits, 30-second window
• [ ] Backup codes (10 one-time use)
• [ ] Remember device option (30 days max)
• [ ] Require MFA for sensitive operations

Rate Limiting

Error Messages

• [ ] Generic "Invalid credentials" (don't reveal which is wrong)
• [ ] Don't reveal if email exists in forgot password
• [ ] Log detailed errors server-side only
• [ ] No stack traces in production

Secure Headers

response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['X-Frame-Options'] = 'DENY'
response.headers['Content-Security-Policy'] = "default-src 'self'"

Audit Logging

• [ ] Log all authentication attempts
• [ ] Log password changes
• [ ] Log MFA setup/disable
• [ ] Log token revocations
• [ ] Log suspicious activity (multiple failed attempts)

Review Checklist

Before deployment:

• [ ] No hardcoded secrets in code
• [ ] Secrets in environment variables
• [ ] HTTPS enforced everywhere
• [ ] Rate limiting configured
• [ ] Audit logging enabled
• [ ] Password hashing uses Argon2id or bcrypt
• [ ] Token lifetimes appropriate
• [ ] MFA available

Common Vulnerabilities to Avoid

• [ ] No password in URL parameters
• [ ] No session ID in URL
• [ ] No sensitive data in JWT payload
• [ ] No implicit OAuth grant
• [ ] No predictable session IDs
• [ ] No client-side token storage in localStorage

Pre-Deployment Security Checklist

Before deploying any AI feature, verify all 8 layers:

Layer 0: Edge Protection

• [ ] WAF rules active for OWASP Top 10
• [ ] Rate limiting configured per user/IP
• [ ] DDoS protection enabled
• [ ] HTTPS enforced (no HTTP)

Layer 1: Gateway / Authentication

• [ ] JWT validation active
• [ ] Token expiry enforced
• [ ] RequestContext created from JWT (not user input)
• [ ] Permissions extracted from token

Layer 2: Input Validation

• [ ] Pydantic/Zod models for all request bodies
• [ ] Size limits on all inputs
• [ ] PII detection on user-provided content
• [ ] Injection pattern detection (SQL, XSS, prompt)

Layer 3: Authorization

• [ ] Every endpoint has authorization check
• [ ] RBAC/ABAC policies defined
• [ ] Cross-tenant access blocked
• [ ] Resource-level access verified

Layer 4: Data Access

• [ ] All queries use parameterized values (no f-strings)
• [ ] All queries include tenant_id filter
• [ ] Repository pattern enforces tenant scope
• [ ] Vector search includes tenant filter

Layer 5: LLM Orchestration

• [ ] No user_id in prompts
• [ ] No tenant_id in prompts
• [ ] No analysis_id in prompts
• [ ] No document_id in prompts
• [ ] No UUIDs in prompts
• [ ] Prompt audit check passes

Layer 6: Output Validation

• [ ] LLM output parsed with schema
• [ ] Content guardrails active (toxicity, PII)
• [ ] Hallucination detection for critical fields
• [ ] Output size limits enforced

Layer 7: Attribution & Storage

• [ ] Attribution uses RequestContext (not LLM output)
• [ ] Source references from pre-LLM lookup
• [ ] Audit event logged
• [ ] Data encrypted at rest

Layer 8: Observability

• [ ] Structured logging active
• [ ] Sensitive data redacted from logs
• [ ] Langfuse tracing enabled
• [ ] Metrics exported (latency, errors, tokens)
• [ ] Alerts configured for anomalies

---

Quick Verification Commands

# Check for IDs in prompt templates
grep -rn "user_id\|tenant_id\|analysis_id\|document_id" backend/app/**/prompts/

# Check for raw SQL (should use parameterized)
grep -rn "f\"SELECT\|f'SELECT" backend/app/

# Check for missing tenant filter
grep -rn "SELECT.*FROM" backend/app/ | grep -v "tenant_id"

# Run security linter
poetry run bandit -r backend/app/ -f json

# Check for hardcoded secrets
grep -rn "api_key\s*=\s*['\"]" backend/

---

Sign-off required before merge:

• [ ] Developer self-review
• [ ] Security checklist verified
• [ ] Code reviewer approved
• [ ] CI/CD security scans pass

Pre-LLM Call Checklist

Before ANY LLM Call in OrchestKit

Use this checklist before sending any prompt to an LLM:

Phase 1: Context Available

• [ ] RequestContext obtained from JWT (not user input)
• [ ] user_id available in context
• [ ] tenant_id available in context
• [ ] trace_id set for observability

Phase 2: Data Isolation

• [ ] Query includes WHERE tenant_id = :tenant_id
• [ ] Query includes WHERE user_id = :user_id (if user-scoped)
• [ ] Vector search filtered by tenant
• [ ] Full-text search filtered by tenant

Phase 3: Source References Captured

• [ ] document_ids saved for attribution
• [ ] chunk_ids saved for attribution
• [ ] Retrieval timestamp recorded
• [ ] Similarity scores captured (for debugging)

Phase 4: Content Extraction

• [ ] Only content text extracted (no metadata with IDs)
• [ ] Content stripped of any embedded UUIDs
• [ ] Content stripped of any ID field names

Phase 5: Prompt Building

• [ ] Prompt contains ONLY content text
• [ ] No user_id in prompt
• [ ] No tenant_id in prompt
• [ ] No analysis_id in prompt
• [ ] No document_id in prompt
• [ ] No UUIDs in prompt
• [ ] No API keys or secrets in prompt

Phase 6: Prompt Audit

• [ ] audit_prompt() called on final prompt
• [ ] No critical violations detected
• [ ] Warnings logged for review

Phase 7: LLM Call

• [ ] Timeout configured
• [ ] Error handling in place
• [ ] Response parsing ready
• [ ] Langfuse trace started

---

Quick Verification Script

from llm_safety import audit_prompt, has_critical_violations

def verify_llm_ready(
    prompt: str,
    ctx: RequestContext,
    source_refs: SourceReference,
) -> bool:
    """Quick verification before LLM call"""

    # Check context
    assert ctx.user_id is not None, "Missing user_id"
    assert ctx.tenant_id is not None, "Missing tenant_id"

    # Check source refs captured
    assert len(source_refs.document_ids) >= 0, "Source refs not captured"

    # Audit prompt
    violations = audit_prompt(prompt)
    if has_critical_violations(violations):
        raise PromptSecurityError(violations)

    return True

---

Post-LLM Attribution Checklist

After LLM returns:

• [ ] Output parsed with schema validation
• [ ] Output checked for hallucinated IDs
• [ ] Output checked for grounding
• [ ] Content safety validated
• [ ] Attribution attached from RequestContext
• [ ] Source links created from captured refs
• [ ] Audit event logged
• [ ] Langfuse trace completed

---

Sign-off: Run verify_llm_ready() before every LLM call

LLM Safety Checklist

Input Safety

• [ ] Validate input length
• [ ] Detect prompt injection attempts
• [ ] Sanitize user content
• [ ] Rate limit requests

Output Safety

• [ ] Content filtering
• [ ] PII detection and redaction
• [ ] Harmful content detection
• [ ] Bias monitoring

System Prompts

• [ ] Clear boundaries
• [ ] Role definition
• [ ] Refusal instructions
• [ ] No secrets in prompts

Guardrails

• [ ] Input guardrails
• [ ] Output guardrails
• [ ] Topic restrictions
• [ ] Sensitive content handling

Monitoring

• [ ] Log flagged content
• [ ] Alert on violations
• [ ] Human review queue
• [ ] Incident response plan

Input Validation Checklist

Core Principles

• [ ] Never trust user input - validate everything
• [ ] Validate server-side - client-side is UX only
• [ ] Use allowlists - not blocklists
• [ ] Validate type, length, format, range
• [ ] Sanitize output - escape when rendering

Schema Definition

• [ ] Define schema for all API endpoints
• [ ] Use strict types (no any)
• [ ] Set reasonable min/max lengths
• [ ] Use enums for fixed value sets
• [ ] Add custom error messages
• [ ] Handle optional vs required properly

String Validation

• [ ] Trim whitespace where appropriate
• [ ] Set maximum length (prevent DoS)
• [ ] Use regex for format validation
• [ ] Escape HTML for display
• [ ] Validate email with proper regex
• [ ] Validate URLs against allowlist domains

Number Validation

• [ ] Use integer for IDs
• [ ] Set min/max bounds
• [ ] Handle NaN and Infinity
• [ ] Use coercion for query params

File Validation

• [ ] Check file extension
• [ ] Validate MIME type
• [ ] Verify magic bytes (actual content)
• [ ] Set maximum file size
• [ ] Scan for malware (production)
• [ ] Generate new filename (no user input)

Database Query Safety

• [ ] Use parameterized queries
• [ ] Allowlist sort columns
• [ ] Validate pagination limits
• [ ] Escape identifiers if dynamic

Error Messages

• [ ] Generic errors for users
• [ ] Detailed errors in logs only
• [ ] Don't reveal system internals
• [ ] Don't reveal valid usernames/emails

Validation Libraries

TypeScript/JavaScript

import { z } from 'zod';
import { zodResolver } from '@hookform/resolvers/zod';
import DOMPurify from 'dompurify';

Python

from pydantic import BaseModel, EmailStr, Field
from markupsafe import escape

Common Patterns

Allowlist (✅ Do)

const allowed = ['name', 'email', 'createdAt'];
if (!allowed.includes(sortColumn)) throw new Error('Invalid');

Blocklist (❌ Don't)

const blocked = ['password', 'secret'];
if (blocked.includes(field)) throw new Error('Invalid');
// Problem: Forgets to block new sensitive fields

Type Coercion

• [ ] Use z.coerce.* for query parameters
• [ ] Handle empty strings appropriately
• [ ] Consider timezone for dates
• [ ] Parse numbers from strings safely

Async Validation

• [ ] Use for uniqueness checks (email, username)
• [ ] Rate limit async validations
• [ ] Cache validation results where appropriate
• [ ] Handle race conditions

Security Headers

Content-Security-Policy: default-src 'self'
X-Content-Type-Options: nosniff
X-Frame-Options: DENY
X-XSS-Protection: 1; mode=block

Review Checklist

Before PR:

• [ ] All endpoints have input validation
• [ ] Server-side validation implemented
• [ ] Allowlists used instead of blocklists
• [ ] Error messages don't leak info
• [ ] File uploads validate content, not just extension
• [ ] SQL queries use parameterized statements
• [ ] HTML output is escaped
• [ ] Maximum lengths set on all strings

Common Vulnerabilities to Prevent

Authentication Implementation Examples

Password Hashing (Argon2id)

from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

ph = PasswordHasher(
    time_cost=3,        # Number of iterations
    memory_cost=65536,  # 64 MB
    parallelism=4,      # Number of threads
)

def hash_password(password: str) -> str:
    """Hash password with Argon2id."""
    return ph.hash(password)

def verify_password(password_hash: str, password: str) -> bool:
    """Verify password against hash."""
    try:
        ph.verify(password_hash, password)
        return True
    except VerifyMismatchError:
        return False

# Check if rehash needed (parameters changed)
def needs_rehash(password_hash: str) -> bool:
    return ph.check_needs_rehash(password_hash)

JWT Access Token

import jwt
from datetime import datetime, timedelta, timezone

SECRET_KEY = os.environ["JWT_SECRET_KEY"]
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15

def create_access_token(user_id: str, roles: list[str] = None) -> str:
    """Create short-lived access token."""
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "type": "access",
        "roles": roles or [],
        "iat": now,
        "exp": now + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verify_access_token(token: str) -> dict | None:
    """Verify and decode access token."""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "access":
            return None
        return payload
    except jwt.ExpiredSignatureError:
        return None
    except jwt.InvalidTokenError:
        return None

Session Management

from flask import Flask, session
from datetime import datetime, timedelta, timezone

app = Flask(__name__)

# Secure session configuration
app.config.update(
    SECRET_KEY=os.environ["SESSION_SECRET"],
    SESSION_COOKIE_SECURE=True,       # HTTPS only
    SESSION_COOKIE_HTTPONLY=True,     # No JavaScript access
    SESSION_COOKIE_SAMESITE='Strict', # CSRF protection
    PERMANENT_SESSION_LIFETIME=timedelta(hours=1),
)

@app.route('/login', methods=['POST'])
def login():
    user = authenticate(request.form['email'], request.form['password'])
    if user:
        session.permanent = True
        session['user_id'] = user.id
        session['created_at'] = datetime.now(timezone.utc).isoformat()
        return redirect('/dashboard')
    return render_template('login.html', error='Invalid credentials')

@app.route('/logout', methods=['POST'])
def logout():
    session.clear()
    return redirect('/login')

Rate Limiting

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    app,
    key_func=get_remote_address,
    default_limits=["200 per day", "50 per hour"],
    storage_uri="redis://localhost:6379",
)

@app.route('/api/auth/login', methods=['POST'])
@limiter.limit("5 per minute")  # Strict rate limit for login
def login():
    # Login logic
    pass

@app.route('/api/auth/password-reset', methods=['POST'])
@limiter.limit("3 per hour")  # Very strict for password reset
def password_reset():
    # Always return success (don't reveal if email exists)
    return {"message": "If email exists, reset link sent"}

Role-Based Access Control

from functools import wraps
from flask import abort, g

def require_role(*roles):
    """Decorator to require specific role(s)."""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            if not g.current_user:
                abort(401)
            if not any(role in g.current_user.roles for role in roles):
                abort(403)
            return f(*args, **kwargs)
        return wrapper
    return decorator

def require_permission(permission: str):
    """Decorator to require specific permission."""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            if not g.current_user:
                abort(401)
            if not g.current_user.has_permission(permission):
                abort(403)
            return f(*args, **kwargs)
        return wrapper
    return decorator

# Usage
@app.route('/admin/users')
@require_role('admin')
def admin_users():
    return get_all_users()

@app.route('/api/patients/<id>')
@require_permission('patients:read')
def get_patient(id):
    return get_patient_by_id(id)

Multi-Factor Authentication (TOTP)

import pyotp
import qrcode
from io import BytesIO
import base64

def generate_totp_secret() -> str:
    """Generate new TOTP secret for user."""
    return pyotp.random_base32()

def get_totp_provisioning_uri(secret: str, email: str, issuer: str = "MyApp") -> str:
    """Get provisioning URI for authenticator app."""
    totp = pyotp.TOTP(secret)
    return totp.provisioning_uri(name=email, issuer_name=issuer)

def get_totp_qr_code(provisioning_uri: str) -> str:
    """Generate QR code as base64 image."""
    qr = qrcode.QRCode(version=1, box_size=10, border=5)
    qr.add_data(provisioning_uri)
    qr.make(fit=True)
    
    img = qr.make_image(fill_color="black", back_color="white")
    buffer = BytesIO()
    img.save(buffer, format="PNG")
    
    return base64.b64encode(buffer.getvalue()).decode()

def verify_totp(secret: str, code: str) -> bool:
    """Verify TOTP code."""
    totp = pyotp.TOTP(secret)
    return totp.verify(code, valid_window=1)  # Allow 1 period window

# Usage
@app.route('/api/auth/mfa/setup', methods=['POST'])
@login_required
def setup_mfa():
    secret = generate_totp_secret()
    uri = get_totp_provisioning_uri(secret, g.current_user.email)
    qr = get_totp_qr_code(uri)
    
    # Store secret temporarily until verified
    session['pending_mfa_secret'] = secret
    
    return {"qr_code": qr, "secret": secret}

@app.route('/api/auth/mfa/verify', methods=['POST'])
@login_required
def verify_mfa_setup():
    code = request.json['code']
    secret = session.get('pending_mfa_secret')
    
    if verify_totp(secret, code):
        g.current_user.mfa_secret = secret
        g.current_user.mfa_enabled = True
        db.session.commit()
        return {"success": True}
    
    return {"error": "Invalid code"}, 400

Complete Login Flow with MFA

@app.route('/api/auth/login', methods=['POST'])
@limiter.limit("5 per minute")
def login():
    email = request.json.get('email')
    password = request.json.get('password')
    
    user = User.query.filter_by(email=email).first()
    
    # Don't reveal if user exists
    if not user or not verify_password(user.password_hash, password):
        return {"error": "Invalid credentials"}, 401
    
    # Check if MFA required
    if user.mfa_enabled:
        # Create temporary token for MFA step
        mfa_token = create_mfa_pending_token(user.id)
        return {"mfa_required": True, "mfa_token": mfa_token}
    
    # No MFA - issue tokens
    return issue_tokens(user)

@app.route('/api/auth/mfa', methods=['POST'])
@limiter.limit("5 per minute")
def verify_mfa():
    mfa_token = request.json.get('mfa_token')
    code = request.json.get('code')
    
    # Verify MFA pending token
    user_id = verify_mfa_pending_token(mfa_token)
    if not user_id:
        return {"error": "Invalid or expired MFA token"}, 401
    
    user = User.query.get(user_id)
    
    # Verify TOTP code
    if not verify_totp(user.mfa_secret, code):
        return {"error": "Invalid MFA code"}, 401
    
    return issue_tokens(user)

def issue_tokens(user):
    """Issue access and refresh tokens."""
    access_token = create_access_token(user.id, user.roles)
    refresh_token = create_refresh_token(user.id)
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "Bearer",
        "expires_in": ACCESS_TOKEN_EXPIRE_MINUTES * 60,
    }

OWASP Top 10 (2025) - Vulnerable vs Secure Code

Real examples showing vulnerable code and their secure alternatives. Categories follow the OWASP Top 10:2025 ordering.

A01: Broken Access Control

❌ Vulnerable: Direct Object Reference

@app.get("/api/documents/{doc_id}")
def get_document(doc_id: int):
    # Anyone can access any document by guessing IDs
    return db.query(Document).get(doc_id)

✅ Secure: Authorization Check

@app.get("/api/documents/{doc_id}")
def get_document(doc_id: int, current_user: User = Depends(get_current_user)):
    doc = db.query(Document).get(doc_id)
    if doc.owner_id != current_user.id and not current_user.is_admin:
        raise HTTPException(403, "Access denied")
    return doc

A02: Security Misconfiguration

❌ Vulnerable: Debug in Production

app = Flask(__name__)
app.run(debug=True)  # Exposes debugger, allows code execution

✅ Secure: Environment-based Config

app = Flask(__name__)
app.run(debug=os.getenv("FLASK_ENV") == "development")

❌ Vulnerable: CORS Allow All

CORS(app, origins="*", allow_credentials=True)

✅ Secure: Explicit Origins

CORS(app, origins=["https://app.example.com"], allow_credentials=True)

A04: Cryptographic Failures

❌ Vulnerable: Weak Hashing

import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()

✅ Secure: Modern Password Hashing

from passlib.hash import argon2
password_hash = argon2.hash(password)
# Verify: argon2.verify(password, password_hash)

A05: Injection

In OWASP Top 10:2025, Cross-Site Scripting (XSS) is folded into the Injection category.

❌ Vulnerable: SQL Injection

query = f"SELECT * FROM users WHERE name = '{name}'"
cursor.execute(query)  # name = "'; DROP TABLE users; --"

✅ Secure: Parameterized Query

cursor.execute("SELECT * FROM users WHERE name = %s", (name,))
# Or with ORM:
db.query(User).filter(User.name == name).first()

❌ Vulnerable: Command Injection

import os
os.system(f"convert {filename} output.png")  # filename = "; rm -rf /"

✅ Secure: Use subprocess with list args

import subprocess
subprocess.run(["convert", filename, "output.png"], check=True)

❌ Vulnerable: XSS — Unescaped Output

element.innerHTML = userInput;  // userInput = "<script>stealCookies()</script>"

✅ Secure: Text Content or Sanitization

element.textContent = userInput;  // Automatically escaped
// Or with sanitization:
element.innerHTML = DOMPurify.sanitize(userInput);

React (Safe by Default)

// ✅ Safe - React escapes by default
<div>{userInput}</div>

// ❌ Dangerous - explicitly bypasses escaping
<div dangerouslySetInnerHTML={{__html: userInput}} />

A08: Software or Data Integrity Failures

Insecure deserialization is classified here in OWASP Top 10:2025.

❌ Vulnerable: Pickle from Untrusted Source

import pickle
data = pickle.loads(user_input)  # Can execute arbitrary code

✅ Secure: Use JSON

import json
data = json.loads(user_input)  # Only parses data, no code execution

Quick Reference

Input Validation Patterns

API Request Validation (TypeScript)

import { z } from 'zod';

// Request body schema
const CreateUserSchema = z.object({
  email: z.email(),
  password: z.string().min(8).max(100),
  name: z.string().min(2).max(100).transform(s => s.trim()),
  role: z.enum(['user', 'admin']).default('user'),
  metadata: z.record(z.string()).optional(),
});

type CreateUserRequest = z.infer<typeof CreateUserSchema>;

// Express middleware
function validateBody<T extends z.ZodSchema>(schema: T) {
  return (req: Request, res: Response, next: NextFunction) => {
    const result = schema.safeParse(req.body);
    if (!result.success) {
      return res.status(400).json({
        error: 'Validation failed',
        details: result.error.flatten().fieldErrors,
      });
    }
    req.body = result.data;
    next();
  };
}

// Usage
app.post('/api/users', validateBody(CreateUserSchema), async (req, res) => {
  const user = req.body as CreateUserRequest;
  // user is fully typed and validated
});

Query Parameter Validation

const PaginationSchema = z.object({
  page: z.coerce.number().int().positive().default(1),
  limit: z.coerce.number().int().min(1).max(100).default(20),
  sort: z.enum(['name', 'email', 'createdAt']).default('createdAt'),
  order: z.enum(['asc', 'desc']).default('desc'),
});

function validateQuery<T extends z.ZodSchema>(schema: T) {
  return (req: Request, res: Response, next: NextFunction) => {
    const result = schema.safeParse(req.query);
    if (!result.success) {
      return res.status(400).json({
        error: 'Invalid query parameters',
        details: result.error.flatten().fieldErrors,
      });
    }
    req.query = result.data;
    next();
  };
}

app.get('/api/users', validateQuery(PaginationSchema), (req, res) => {
  const { page, limit, sort, order } = req.query;
  // All values are properly typed and defaulted
});

Discriminated Union for Polymorphic Data

const NotificationSchema = z.discriminatedUnion('type', [
  z.object({
    type: z.literal('email'),
    email: z.email(),
    subject: z.string().min(1),
    body: z.string().min(1),
  }),
  z.object({
    type: z.literal('sms'),
    phone: z.string().regex(/^\+[1-9]\d{1,14}$/),
    message: z.string().max(160),
  }),
  z.object({
    type: z.literal('push'),
    deviceToken: z.string().min(1),
    title: z.string().max(50),
    body: z.string().max(200),
  }),
]);

type Notification = z.infer<typeof NotificationSchema>;

// Type-safe handling
function sendNotification(notification: Notification) {
  switch (notification.type) {
    case 'email':
      return sendEmail(notification.email, notification.subject, notification.body);
    case 'sms':
      return sendSMS(notification.phone, notification.message);
    case 'push':
      return sendPush(notification.deviceToken, notification.title, notification.body);
  }
}

Allowlist Validation

// Only allow specific values
const SortColumnSchema = z.enum(['name', 'email', 'createdAt', 'updatedAt']);

// For dynamic allowlists
function createAllowlistSchema<T extends string>(allowed: readonly T[]) {
  return z.enum(allowed as [T, ...T[]]);
}

const allowedColumns = ['name', 'email', 'createdAt'] as const;
const DynamicSortSchema = createAllowlistSchema(allowedColumns);

File Upload Validation

const FileUploadSchema = z.object({
  file: z.object({
    name: z.string(),
    type: z.enum(['image/jpeg', 'image/png', 'image/webp', 'application/pdf']),
    size: z.number().max(5 * 1024 * 1024, 'File must be under 5MB'),
  }),
});

// Validate file content (magic bytes)
const imageMagicBytes: Record<string, number[]> = {
  'image/jpeg': [0xFF, 0xD8, 0xFF],
  'image/png': [0x89, 0x50, 0x4E, 0x47],
  'image/webp': [0x52, 0x49, 0x46, 0x46],
  'application/pdf': [0x25, 0x50, 0x44, 0x46],
};

function validateFileContent(buffer: Buffer, mimeType: string): boolean {
  const expected = imageMagicBytes[mimeType];
  if (!expected) return false;
  return expected.every((byte, i) => buffer[i] === byte);
}

URL Validation with Domain Allowlist

const ALLOWED_DOMAINS = ['api.example.com', 'cdn.example.com'] as const;

const UrlSchema = z.url()
  .refine(
    (url) => {
      const { hostname, protocol } = new URL(url);
      return protocol === 'https:' && ALLOWED_DOMAINS.includes(hostname as any);
    },
    { message: 'URL must be HTTPS and from allowed domains' }
  );

// Usage
UrlSchema.parse('https://api.example.com/data'); // OK
UrlSchema.parse('https://evil.com/data');        // Error
UrlSchema.parse('http://api.example.com/data');  // Error (not HTTPS)

Python (Pydantic) Validation

from pydantic import BaseModel, EmailStr, Field, field_validator
from typing import Literal, Union

# Basic model
class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(min_length=2, max_length=100)
    age: int = Field(ge=0, le=150)
    
    @field_validator('name')
    @classmethod
    def strip_and_title(cls, v: str) -> str:
        return v.strip().title()

# Discriminated union
class EmailNotification(BaseModel):
    type: Literal['email']
    email: EmailStr
    subject: str
    body: str

class SMSNotification(BaseModel):
    type: Literal['sms']
    phone: str
    message: str = Field(max_length=160)

Notification = Union[EmailNotification, SMSNotification]

# Allowlist validation
ALLOWED_COLUMNS = frozenset(['name', 'email', 'created_at'])

def validate_sort_column(column: str) -> str:
    if column not in ALLOWED_COLUMNS:
        raise ValueError(f"Invalid sort column: {column}")
    return column

HTML Sanitization

from markupsafe import escape

@app.route('/comment', methods=['POST'])
def create_comment():
    # Escape HTML to prevent XSS
    content = escape(request.form['content'])
    db.execute("INSERT INTO comments (content) VALUES (?)", [content])

import DOMPurify from 'dompurify';

// Sanitize HTML input
const sanitizedHtml = DOMPurify.sanitize(userInput, {
  ALLOWED_TAGS: ['b', 'i', 'em', 'strong', 'a'],
  ALLOWED_ATTR: ['href'],
});

Form Validation with React Hook Form

import { useForm } from 'react-hook-form';
import { zodResolver } from '@hookform/resolvers/zod';

const SignupSchema = z.object({
  email: z.email('Invalid email'),
  password: z.string()
    .min(8, 'Password must be at least 8 characters')
    .regex(/[A-Z]/, 'Must contain uppercase')
    .regex(/[0-9]/, 'Must contain number'),
  confirmPassword: z.string(),
}).refine(data => data.password === data.confirmPassword, {
  message: "Passwords don't match",
  path: ['confirmPassword'],
});

type SignupForm = z.infer<typeof SignupSchema>;

function SignupForm() {
  const { register, handleSubmit, formState: { errors } } = useForm<SignupForm>({
    resolver: zodResolver(SignupSchema),
  });

  return (
    <form onSubmit={handleSubmit(onSubmit)}>
      <input {...register('email')} placeholder="Email" />
      {errors.email && <span className="error">{errors.email.message}</span>}
      
      <input {...register('password')} type="password" placeholder="Password" />
      {errors.password && <span className="error">{errors.password.message}</span>}
      
      <input {...register('confirmPassword')} type="password" placeholder="Confirm" />
      {errors.confirmPassword && <span className="error">{errors.confirmPassword.message}</span>}
      
      <button type="submit">Sign Up</button>
    </form>
  );
}

Audit Logging

Purpose

Audit logs answer: Who did What, When, Where, and Why?

They're required for:

• Security incident investigation
• Compliance (SOC2, GDPR, HIPAA)
• Debugging production issues
• Usage analytics

What to Log

Always Log (Audit Events)

Never Log (Sensitive Data)

Implementation

Sanitized Logger

import structlog
import re
import hashlib
from typing import Any

class SanitizedLogger:
    """Logger that automatically redacts sensitive data"""

    REDACT_PATTERNS = {
        r"password": "[PASSWORD_REDACTED]",
        r"api[_-]?key": "[API_KEY_REDACTED]",
        r"secret": "[SECRET_REDACTED]",
        r"token": "[TOKEN_REDACTED]",
        r"authorization": "[AUTH_REDACTED]",
        r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}": "[EMAIL_REDACTED]",
    }

    HASH_FIELDS = {"prompt", "response", "content"}

    def __init__(self):
        self._logger = structlog.get_logger()

    def _sanitize(self, data: dict[str, Any]) -> dict[str, Any]:
        """Sanitize sensitive fields"""
        result = {}
        for key, value in data.items():
            # Hash content fields instead of logging
            if key.lower() in self.HASH_FIELDS:
                result[f"{key}_hash"] = hashlib.sha256(
                    str(value).encode()
                ).hexdigest()[:16]
                result[f"{key}_length"] = len(str(value))
                continue

            # Redact sensitive patterns
            str_value = str(value)
            for pattern, replacement in self.REDACT_PATTERNS.items():
                if re.search(pattern, key, re.IGNORECASE):
                    result[key] = replacement
                    break
                str_value = re.sub(pattern, replacement, str_value, flags=re.IGNORECASE)
            else:
                result[key] = str_value

        return result

    def audit(self, event: str, **kwargs):
        """Log an audit event with automatic sanitization"""
        sanitized = self._sanitize(kwargs)
        self._logger.info(
            event,
            audit=True,
            **sanitized,
        )

    def info(self, msg: str, **kwargs):
        self._logger.info(msg, **self._sanitize(kwargs))

    def error(self, msg: str, **kwargs):
        self._logger.error(msg, **self._sanitize(kwargs))

Audit Event Structure

from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from uuid import UUID

class AuditAction(Enum):
    CREATE = "create"
    READ = "read"
    UPDATE = "update"
    DELETE = "delete"
    LOGIN = "login"
    LOGOUT = "logout"
    LLM_CALL = "llm_call"
    SEARCH = "search"

@dataclass
class AuditEvent:
    """Structured audit event"""
    # WHO
    user_id: UUID
    tenant_id: UUID
    session_id: str

    # WHAT
    action: AuditAction
    resource_type: str
    resource_id: UUID | None

    # WHEN
    timestamp: datetime

    # WHERE
    request_id: str
    trace_id: str
    ip_address: str
    user_agent: str

    # OUTCOME
    success: bool
    error_code: str | None = None

    # CONTEXT (sanitized)
    metadata: dict | None = None

Usage in OrchestKit

# Authentication
logger.audit(
    "user.login",
    user_id=user.id,
    tenant_id=user.tenant_id,
    method="oauth",
    success=True,
)

# Data Access
logger.audit(
    "documents.search",
    user_id=ctx.user_id,
    tenant_id=ctx.tenant_id,
    query_hash=hash(query),  # Not the actual query
    result_count=len(results),
    success=True,
)

# LLM Call
logger.audit(
    "llm.generate",
    user_id=ctx.user_id,
    tenant_id=ctx.tenant_id,
    model="gpt-4",
    input_tokens=1500,
    output_tokens=500,
    latency_ms=2300,
    prompt_hash=hash(prompt),  # Not the actual prompt!
    success=True,
)

# Authorization Failure
logger.audit(
    "authorization.denied",
    user_id=ctx.user_id,
    tenant_id=ctx.tenant_id,
    action="analysis:delete",
    resource_id=analysis_id,
    reason="missing permission",
    success=False,
)

Log Retention

Integration with Langfuse

from langfuse import Langfuse

langfuse = Langfuse()

# Create trace for observability
trace = langfuse.trace(
    name="analysis",
    user_id=str(ctx.user_id),  # Langfuse supports user tracking
    session_id=ctx.session_id,
    metadata={
        "tenant_id": str(ctx.tenant_id),
        "request_id": ctx.request_id,
    },
)

# Log LLM call
generation = trace.generation(
    name="content_analysis",
    model="gpt-4",
    input=prompt,  # Langfuse handles securely
    output=response,
)

Compliance Considerations

GDPR

• Log data access but not the data itself
• Provide audit trail for subject access requests
• Log data deletion events

SOC2

• Log all authentication events
• Log all authorization decisions
• Log all data modifications
• Retain logs for audit period

HIPAA

• Log all access to PHI
• Log user ID, timestamp, action
• Never log PHI content in logs

Claude Code Permission-Rule Semantics (security-relevant)

How Claude Code's allow / ask / deny permission rules actually behave as of the supported floor (≥ 2.1.166 for every behavior below; current floor 2.1.168). These are the facts OrchestKit security guidance depends on — get them wrong and a "locked down" config has holes. Each behavior shipped in a specific release; all are guaranteed present at the floor.

1. Read deny rules hide files from Glob/Grep (2.1.162)

A Read deny rule is now a real secrecy boundary, not just a read block:

// .claude/settings.json
{ "permissions": { "deny": ["Read(./.env*)", "Read(./secrets/**)"] } }

Matching files no longer appear in Glob or Grep results either — before 2.1.162 the agent could still discover (and infer from) denied paths via search even though it couldn't read them. Treat Read(deny) as "this path does not exist for the agent."

• Use it for: secrets, key material, customer data dumps, .env families.
• Pitfall: a deny rule with a typo silently protects nothing — there is no

"unknown path" warning (unlike deny tool names, see §2). Verify with a probe Glob after deploying.

2. Glob support in deny-rule tool-name position (2.1.166)

The tool-name slot of a deny rule accepts globs, enabling a default-deny baseline:

{ "permissions": {
    "deny":  ["*"],                       // deny ALL tools…
    "allow": ["Read(./src/**)", "Grep", "Glob"]   // …then re-allow the minimum
} }

• "*" in a deny rule denies every tool.
• Allow rules reject non-MCP globs (you cannot allow: ["*"]) — allow stays explicit by design.
• Unknown tool names in deny rules emit a startup warning (catches typos — the

safety net §1 lacks). Watch the startup log when authoring deny lists.

3. Explicit WebFetch rules override the preapproved-host auto-allow (2.1.162)

CC auto-allows a built-in set of preapproved WebFetch domains. Before 2.1.162 your explicit rules were ignored for those hosts; now explicit `WebFetch(domain:…)` deny/ask/allow takes precedence:

{ "permissions": { "deny": ["WebFetch(domain:raw.githubusercontent.com)"] } }
// now actually blocks it, even though it's normally preapproved

• Use it for: blocking exfiltration sinks / paste hosts even when they're on the

default allow-list; forcing ask on a sensitive internal domain.

4. Cross-session SendMessage relays carry no user authority (2.1.166)

Multi-agent hardening: a message relayed via SendMessage from another Claude session no longer inherits the originating user's authority.

• Receivers refuse relayed permission requests.
• auto mode blocks them outright.

Implication for OrchestKit's multi-agent flows (agent-orchestration, mcp-patterns): a peer or compromised session cannot escalate by asking your session to approve a tool call on its behalf. Design fan-out so privileged actions run in the session that legitimately holds the authority — do not route approvals through relays.

5. Org-managed permission rules apply for the whole session (2.1.163)

Enterprise lockdown reliability fix: org-managed permission rules now apply for the entire session even when the managed-settings fetch completes during startup on a fresh config directory (previously a first-run race left the session unmanaged). Also in 2.1.163: a Read(~/Desktop/**)-style home-dir deny now also blocks Bash commands that reach the path via $HOME.

• Require 2.1.163+ when relying on org-managed profiles for a security boundary.
• Pairs with the 2.1.166 managed-settings enforcement fix (see SKILL.md "Managed Hook

Hierarchy") — one invalid entry no longer voids the rest of the policy.

Recommended baseline posture

// .claude/settings.json — default-deny, explicit re-allow, secrecy on secrets
{ "permissions": {
    "deny":  ["*", "Read(./.env*)", "Read(./secrets/**)",
              "WebFetch(domain:pastebin.com)"],
    "allow": ["Read(./src/**)", "Read(./docs/**)", "Grep", "Glob",
              "Bash(npm run test:*)"],
    "ask":   ["Bash(git push:*)"]
} }

Verify after deploy: a Grep for a denied secret returns nothing (§1), the startup log shows no "unknown tool" warnings (§2), and a denied preapproved WebFetch domain is actually refused (§3).

Context Separation Pattern

OWASP LLM Top 10 (2025): This pattern mitigates LLM07: System Prompt Leakage — keeping identifiers and internal context out of prompts prevents the model from echoing or leaking system-level data.

The Problem

When identifiers appear in LLM prompts, several security issues arise:

┌─────────────────────────────────────────────────────────┐
│  WHAT HAPPENS WHEN IDs GO INTO PROMPTS                  │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  "Analyze document doc_abc123 for user usr_xyz789"      │
│                     │                    │              │
│                     ▼                    ▼              │
│              ┌──────────────────────────────┐           │
│              │           LLM                │           │
│              │                              │           │
│              │  May hallucinate:            │           │
│              │  - doc_abc124 (off by one)   │           │
│              │  - doc_xyz789 (mixed up)     │           │
│              │  - usr_other (cross-tenant)  │           │
│              └──────────────────────────────┘           │
│                                                         │
│  RISKS:                                                 │
│  • Hallucinated IDs don't exist → crashes              │
│  • Mixed IDs → wrong data attribution                  │
│  • Cross-tenant IDs → security breach                  │
│  • IDs in logs/traces → data leakage                   │
│                                                         │
└─────────────────────────────────────────────────────────┘

The Solution: Context Separation

┌─────────────────────────────────────────────────────────┐
│  CORRECT: CONTEXT FLOWS AROUND LLM                      │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  RequestContext ─────────────────────────────────────►  │
│  (user_id, tenant_id, etc.)                    │        │
│         │                                      │        │
│         │   ┌──────────────────────┐          │        │
│         │   │                      │          │        │
│         ▼   │       LLM            │          ▼        │
│  ┌──────────┤                      ├─────────────┐     │
│  │ Content  │  Sees ONLY:          │  Content +  │     │
│  │ (text)   │  - Document text     │  Context    │     │
│  │          │  - Query text        │  (merged)   │     │
│  └──────────┤  - Instructions      ├─────────────┘     │
│             │                      │                    │
│             │  NO IDs!             │                    │
│             └──────────────────────┘                    │
│                                                         │
└─────────────────────────────────────────────────────────┘

Implementation

1. Define What's Forbidden

# OrchestKit parameters that NEVER go in prompts
FORBIDDEN_IN_PROMPTS = {
    # User identity
    "user_id",      # UUID - hallucination risk
    "tenant_id",    # UUID - cross-tenant risk
    "session_id",   # String - auth context

    # Resource references
    "analysis_id",  # UUID - job tracking
    "document_id",  # UUID - source tracking
    "artifact_id",  # UUID - output tracking
    "chunk_id",     # UUID - RAG reference

    # System context
    "trace_id",     # String - observability
    "request_id",   # String - request tracking
    "workflow_run_id",  # UUID - workflow tracking

    # Secrets
    "api_key",      # String - never!
    "token",        # String - never!
}

2. Separate Context from Content

from dataclasses import dataclass
from uuid import UUID

@dataclass
class ContentPayload:
    """What the LLM sees - content only"""
    query: str
    context_texts: list[str]
    instructions: str

@dataclass
class ContextPayload:
    """What flows around the LLM - never in prompt"""
    user_id: UUID
    tenant_id: UUID
    analysis_id: UUID
    source_refs: list[UUID]
    trace_id: str

async def analyze_content(
    content: ContentPayload,
    context: ContextPayload,
) -> AnalysisResult:
    """
    Content goes TO the LLM.
    Context goes AROUND the LLM.
    """
    # Build prompt from content only
    prompt = build_prompt(
        query=content.query,
        context_texts=content.context_texts,
        instructions=content.instructions,
        # NO context payload fields here!
    )

    # LLM sees content only
    llm_output = await llm.generate(prompt)

    # Reattach context to output
    return AnalysisResult(
        content=llm_output,
        user_id=context.user_id,      # From context
        tenant_id=context.tenant_id,   # From context
        analysis_id=context.analysis_id,  # From context
        sources=context.source_refs,   # From context
    )

3. Audit Prompts Before Sending

import re

UUID_PATTERN = r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'

def audit_prompt(prompt: str) -> list[str]:
    """
    Check for forbidden patterns before sending to LLM.
    Raises if any IDs detected.
    """
    violations = []

    # Check for UUIDs
    if re.search(UUID_PATTERN, prompt, re.IGNORECASE):
        violations.append("UUID detected in prompt")

    # Check for ID field names
    for forbidden in FORBIDDEN_IN_PROMPTS:
        pattern = rf'\b{forbidden}\b'
        if re.search(pattern, prompt, re.IGNORECASE):
            violations.append(f"Forbidden field '{forbidden}' in prompt")

    return violations

# Usage in prompt building
def build_safe_prompt(content: ContentPayload) -> str:
    prompt = f"""
    Analyze the following content:

    {content.query}

    Context:
    {chr(10).join(content.context_texts)}
    """

    # Audit before returning
    violations = audit_prompt(prompt)
    if violations:
        raise PromptSecurityError(
            f"Prompt contains forbidden content: {violations}"
        )

    return prompt

OrchestKit Integration Points

Content Analysis Workflow

# backend/app/workflows/agents/content_analyzer.py

async def analyze(state: AnalysisState) -> AnalysisState:
    # Context is in state, but NOT passed to prompt
    ctx = state.request_context

    # Build content-only payload
    content = ContentPayload(
        query=state.analysis_request.query,
        context_texts=[doc.content for doc in state.retrieved_docs],
        instructions=get_analysis_instructions(),
    )

    # Context payload for attribution
    context = ContextPayload(
        user_id=ctx.user_id,
        tenant_id=ctx.tenant_id,
        analysis_id=state.analysis_id,
        source_refs=[doc.id for doc in state.retrieved_docs],
        trace_id=ctx.trace_id,
    )

    result = await analyze_content(content, context)
    return state.with_result(result)

Common Mistakes

# ❌ BAD: ID in prompt
prompt = f"Analyze document {doc_id} for user {user_id}"

# ❌ BAD: ID in f-string
prompt = f"Context from analysis {analysis_id}:\n{context}"

# ❌ BAD: ID in instruction
prompt = f"You are analyzing for tenant {tenant_id}. Be helpful."

# ✅ GOOD: Content only
prompt = f"Analyze the following document:\n{document_content}"

# ✅ GOOD: No IDs visible
prompt = f"""
Analyze this content and provide insights:

{content}

Relevant context:
{context_texts}
"""

Testing Context Separation

import pytest

class TestContextSeparation:

    def test_prompt_contains_no_uuids(self):
        content = ContentPayload(
            query="What are the key concepts?",
            context_texts=["Machine learning basics..."],
            instructions="Provide clear analysis",
        )

        prompt = build_safe_prompt(content)

        assert not re.search(UUID_PATTERN, prompt)

    def test_prompt_contains_no_forbidden_fields(self):
        content = ContentPayload(...)
        prompt = build_safe_prompt(content)

        for forbidden in FORBIDDEN_IN_PROMPTS:
            assert forbidden not in prompt.lower()

    def test_audit_catches_leaked_uuid(self):
        bad_prompt = "Analyze doc 123e4567-e89b-12d3-a456-426614174000"

        violations = audit_prompt(bad_prompt)

        assert len(violations) > 0
        assert "UUID" in violations[0]

Langfuse Mask Callback

Pre-trace PII masking using Langfuse's mask callback for automatic redaction before data reaches the server.

Basic Setup

from langfuse import Langfuse
import re

PII_PATTERNS = {
    "email": re.compile(r'\b[\w.-]+@[\w.-]+\.\w{2,}\b'),
    "phone": re.compile(r'\b(?:\+1[-.]?)?\(?\d{3}\)?[-.]?\d{3}[-.]?\d{4}\b'),
    "ssn": re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
}

def mask_pii(data: dict) -> dict:
    """Mask PII in Langfuse trace data before sending."""
    def redact_string(value: str) -> str:
        for entity_type, pattern in PII_PATTERNS.items():
            value = pattern.sub(f'[REDACTED_{entity_type.upper()}]', value)
        return value

    def redact_recursive(obj):
        if isinstance(obj, str):
            return redact_string(obj)
        elif isinstance(obj, dict):
            return {k: redact_recursive(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [redact_recursive(item) for item in obj]
        return obj

    return redact_recursive(data)

langfuse = Langfuse(mask=mask_pii)

With Presidio

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def presidio_mask(data: dict) -> dict:
    """Enterprise-grade PII masking with Presidio."""
    def anonymize_string(value: str) -> str:
        if len(value) < 5:
            return value
        results = analyzer.analyze(text=value, language="en")
        if results:
            return anonymizer.anonymize(text=value, analyzer_results=results).text
        return value

    def process_recursive(obj):
        if isinstance(obj, str):
            return anonymize_string(obj)
        elif isinstance(obj, dict):
            return {k: process_recursive(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [process_recursive(item) for item in obj]
        return obj

    return process_recursive(data)

langfuse = Langfuse(mask=presidio_mask)

References

• Langfuse Mask Callback
• Langfuse Privacy

LLM Guard Sanitization

Input/output sanitization for LLM pipelines using LLM Guard's Anonymize and Deanonymize scanners.

Installation

pip install llm-guard
python -m spacy download en_core_web_trf  # High-accuracy model

Basic Input Sanitization

from llm_guard.input_scanners import Anonymize
from llm_guard.input_scanners.anonymize_helpers import BERT_LARGE_NER_CONF
from llm_guard.vault import Vault

# Vault stores original values for deanonymization
vault = Vault()

# Initialize scanner with configuration
scanner = Anonymize(
    vault=vault,
    preamble="",  # Text prepended to sanitized output
    allowed_names=["John Doe"],  # Names to NOT anonymize
    hidden_names=["Acme Corp"],  # Always anonymize these
    recognizer_conf=BERT_LARGE_NER_CONF,
    language="en"
)

def sanitize_input(prompt: str) -> tuple[str, bool, float]:
    """
    Sanitize user input before sending to LLM.

    Returns:
        (sanitized_prompt, is_valid, risk_score)
    """
    sanitized_prompt, is_valid, risk_score = scanner.scan(prompt)
    return sanitized_prompt, is_valid, risk_score

# Usage
prompt = "My name is Jane Smith and my email is jane@company.com"
sanitized, valid, risk = sanitize_input(prompt)
# Result: "My name is [REDACTED_PERSON_1] and my email is [REDACTED_EMAIL_1]"

Output Deanonymization

from llm_guard.output_scanners import Deanonymize

# Use the same vault from input sanitization
deanonymize_scanner = Deanonymize(vault=vault)

def deanonymize_output(sanitized_prompt: str, model_output: str) -> str:
    """
    Restore original values in model output.

    Args:
        sanitized_prompt: The prompt that was sent to the LLM
        model_output: The LLM's response

    Returns:
        Output with original values restored
    """
    restored_output, is_valid, risk_score = deanonymize_scanner.scan(
        sanitized_prompt,
        model_output
    )
    return restored_output

# Example flow
original_prompt = "Schedule a meeting with Jane Smith at jane@company.com"
sanitized_prompt, _, _ = scanner.scan(original_prompt)
# sanitized_prompt = "Schedule a meeting with [PERSON_1] at [EMAIL_1]"

llm_response = await llm.generate(sanitized_prompt)
# llm_response = "Meeting scheduled with [PERSON_1]. Confirmation sent to [EMAIL_1]."

final_response = deanonymize_output(sanitized_prompt, llm_response)
# final_response = "Meeting scheduled with Jane Smith. Confirmation sent to jane@company.com."

Output Sensitive Data Detection

from llm_guard.output_scanners import Sensitive

# Detect PII in LLM outputs (without prior anonymization)
sensitive_scanner = Sensitive(
    entity_types=["PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER", "CREDIT_CARD"],
    redact=True,  # Replace detected PII with [REDACTED]
    threshold=0.5  # Confidence threshold (0-1)
)

def check_output_for_pii(prompt: str, output: str) -> tuple[str, bool, float]:
    """
    Check LLM output for leaked PII.

    Returns:
        (sanitized_output, is_valid, risk_score)
    """
    sanitized_output, is_valid, risk_score = sensitive_scanner.scan(prompt, output)
    return sanitized_output, is_valid, risk_score

Full Pipeline Integration

from llm_guard.input_scanners import Anonymize
from llm_guard.output_scanners import Deanonymize, Sensitive
from llm_guard.vault import Vault
from langfuse import observe, get_client

class SecureLLMPipeline:
    def __init__(self):
        self.vault = Vault()
        self.anonymize = Anonymize(vault=self.vault, language="en")
        self.deanonymize = Deanonymize(vault=self.vault)
        self.sensitive_check = Sensitive(redact=True)

    @observe(name="secure_llm_call")
    async def process(self, user_input: str) -> str:
        """Secure LLM pipeline with full PII protection."""

        # Step 1: Anonymize input
        sanitized_input, input_valid, input_risk = self.anonymize.scan(user_input)

        get_client().update_current_observation(
            metadata={
                "input_risk_score": input_risk,
                "pii_detected_in_input": not input_valid
            }
        )

        # Step 2: Call LLM with sanitized input
        llm_response = await self.llm.generate(sanitized_input)

        # Step 3: Check output for leaked PII
        checked_output, output_valid, output_risk = self.sensitive_check.scan(
            sanitized_input,
            llm_response
        )

        # Step 4: Deanonymize for user (restore original names)
        final_output = self.deanonymize.scan(sanitized_input, checked_output)[0]

        get_client().update_current_observation(
            metadata={
                "output_risk_score": output_risk,
                "pii_leaked_in_output": not output_valid
            }
        )

        return final_output

Configuration Options

Anonymize Scanner

from llm_guard.input_scanners import Anonymize
from llm_guard.input_scanners.anonymize_helpers import (
    BERT_LARGE_NER_CONF,
    BERT_BASE_NER_CONF,
    DISTILBERT_NER_CONF
)

scanner = Anonymize(
    vault=vault,
    preamble="",                          # Prepend to output
    allowed_names=["Claude", "GPT"],      # Don't anonymize these
    hidden_names=["Internal Corp"],       # Always anonymize these
    entity_types=[                        # Entities to detect
        "PERSON",
        "EMAIL_ADDRESS",
        "PHONE_NUMBER",
        "CREDIT_CARD",
        "US_SSN",
        "IP_ADDRESS",
        "LOCATION"
    ],
    use_faker=True,                       # Replace with fake data
    recognizer_conf=BERT_LARGE_NER_CONF,  # NER model config
    threshold=0.5,                        # Confidence threshold
    language="en"                         # Language
)

Recognizer Configurations

Handling Overlapping Entities

LLM Guard handles overlapping entities automatically:

# Input: "Contact John Smith at john.smith@example.com"
# PERSON: "John Smith" (indices 8-18)
# EMAIL: "john.smith@example.com" (indices 22-45)
# - john.smith overlaps with PERSON

# LLM Guard prioritizes:
# 1. Higher confidence score wins
# 2. Longer span wins if scores equal

Testing

import pytest
from llm_guard.input_scanners import Anonymize
from llm_guard.vault import Vault

def test_anonymization():
    vault = Vault()
    scanner = Anonymize(vault=vault)

    test_input = "Contact John at john@example.com or 555-123-4567"
    sanitized, is_valid, risk = scanner.scan(test_input)

    # Verify PII is removed
    assert "John" not in sanitized
    assert "john@example.com" not in sanitized
    assert "555-123-4567" not in sanitized

    # Verify placeholders are present
    assert "[PERSON" in sanitized or "REDACTED" in sanitized

def test_deanonymization():
    vault = Vault()
    anonymize = Anonymize(vault=vault)
    deanonymize = Deanonymize(vault=vault)

    original = "Send email to Alice"
    sanitized, _, _ = anonymize.scan(original)

    # Simulate LLM response
    response = f"Email sent to {sanitized.split()[-1]}"

    restored, _, _ = deanonymize.scan(sanitized, response)
    assert "Alice" in restored

References

• LLM Guard Documentation
• Anonymize Scanner
• Sensitive Scanner
• LLM Guard GitHub

Logging Redaction Patterns

Pre-logging PII redaction with structlog and loguru.

Structlog Processor

import re
import structlog
from typing import Any

# Pre-compile patterns
PII_PATTERNS = {
    "email": re.compile(r'\b[\w.-]+@[\w.-]+\.\w{2,}\b'),
    "phone": re.compile(r'\b(?:\+1[-.]?)?\(?\d{3}\)?[-.]?\d{3}[-.]?\d{4}\b'),
    "ssn": re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
    "credit_card": re.compile(r'\b(?:\d[ -]*?){13,19}\b'),
    "ip": re.compile(r'\b(?:\d{1,3}\.){3}\d{1,3}\b'),
}

def redact_pii(logger, method_name: str, event_dict: dict) -> dict:
    """
    Structlog processor to redact PII from all log fields.
    """
    def redact_value(value: Any) -> Any:
        if isinstance(value, str):
            result = value
            for entity_type, pattern in PII_PATTERNS.items():
                result = pattern.sub(f'[REDACTED_{entity_type.upper()}]', result)
            return result
        elif isinstance(value, dict):
            return {k: redact_value(v) for k, v in value.items()}
        elif isinstance(value, list):
            return [redact_value(item) for item in value]
        return value

    return {k: redact_value(v) for k, v in event_dict.items()}


# Configure structlog with PII redaction
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        redact_pii,  # Add PII redaction processor
        structlog.processors.JSONRenderer()
    ]
)

logger = structlog.get_logger()

# Usage - PII is automatically redacted
logger.info(
    "user_registered",
    email="john@example.com",  # -> [REDACTED_EMAIL]
    phone="555-123-4567"       # -> [REDACTED_PHONE]
)

Structlog with Presidio

import structlog
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

# Singleton Presidio engines
_analyzer = None
_anonymizer = None

def get_presidio_engines():
    global _analyzer, _anonymizer
    if _analyzer is None:
        _analyzer = AnalyzerEngine()
        _anonymizer = AnonymizerEngine()
    return _analyzer, _anonymizer

def presidio_redact_processor(logger, method_name: str, event_dict: dict) -> dict:
    """Use Presidio for enterprise-grade PII redaction in logs."""
    analyzer, anonymizer = get_presidio_engines()

    def redact_value(value):
        if isinstance(value, str) and len(value) > 5:
            try:
                results = analyzer.analyze(text=value, language="en")
                if results:
                    anonymized = anonymizer.anonymize(
                        text=value,
                        analyzer_results=results
                    )
                    return anonymized.text
            except Exception:
                pass  # Fallback to original on error
        elif isinstance(value, dict):
            return {k: redact_value(v) for k, v in value.items()}
        elif isinstance(value, list):
            return [redact_value(item) for item in value]
        return value

    return {k: redact_value(v) for k, v in event_dict.items()}

structlog.configure(
    processors=[
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        presidio_redact_processor,
        structlog.processors.JSONRenderer()
    ]
)

Loguru Filter

import re
from loguru import logger

PII_PATTERNS = {
    "email": re.compile(r'\b[\w.-]+@[\w.-]+\.\w{2,}\b'),
    "phone": re.compile(r'\b(?:\+1[-.]?)?\(?\d{3}\)?[-.]?\d{3}[-.]?\d{4}\b'),
    "ssn": re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
    "credit_card": re.compile(r'\b(?:\d[ -]*?){13,19}\b'),
}

def pii_filter(record):
    """Loguru filter to redact PII from log messages."""
    message = record["message"]
    for entity_type, pattern in PII_PATTERNS.items():
        message = pattern.sub(f'[REDACTED_{entity_type.upper()}]', message)
    record["message"] = message
    return True

# Configure loguru with PII filter
logger.remove()  # Remove default handler
logger.add(
    "logs/app.log",
    filter=pii_filter,
    format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}",
    serialize=True  # JSON format
)

# Usage
logger.info("User john@example.com logged in from 192.168.1.1")
# Output: "User [REDACTED_EMAIL] logged in from [REDACTED_IP]"

Loguru with Custom Patcher

from loguru import logger

def pii_patcher(record):
    """Patch record to redact PII in extra fields."""
    if "extra" in record:
        for key, value in record["extra"].items():
            if isinstance(value, str):
                for entity_type, pattern in PII_PATTERNS.items():
                    value = pattern.sub(f'[REDACTED_{entity_type.upper()}]', value)
                record["extra"][key] = value
    return record

logger = logger.patch(pii_patcher)

# Usage with bound variables
logger.bind(user_email="jane@example.com").info("Processing user request")
# The email in extra will be redacted

Field-Specific Redaction

import structlog
from typing import Any

# Fields that should always be redacted
SENSITIVE_FIELDS = {
    "email", "phone", "ssn", "credit_card", "password",
    "api_key", "token", "secret", "authorization"
}

# Fields that should be partially masked
PARTIAL_MASK_FIELDS = {
    "user_id": lambda v: f"{str(v)[:4]}...{str(v)[-4:]}" if len(str(v)) > 8 else "***"
}

def smart_redact_processor(logger, method_name: str, event_dict: dict) -> dict:
    """Smart redaction based on field names."""
    result = {}

    for key, value in event_dict.items():
        key_lower = key.lower()

        # Full redaction for sensitive fields
        if key_lower in SENSITIVE_FIELDS:
            result[key] = "[REDACTED]"

        # Partial masking
        elif key_lower in PARTIAL_MASK_FIELDS:
            result[key] = PARTIAL_MASK_FIELDS[key_lower](value)

        # Pattern-based redaction for other string fields
        elif isinstance(value, str):
            result[key] = redact_pii_patterns(value)

        else:
            result[key] = value

    return result

def redact_pii_patterns(value: str) -> str:
    """Apply PII patterns to a string."""
    for entity_type, pattern in PII_PATTERNS.items():
        value = pattern.sub(f'[REDACTED_{entity_type.upper()}]', value)
    return value

Context Manager for Sensitive Operations

import structlog
from contextlib import contextmanager

@contextmanager
def sensitive_logging_context():
    """
    Context manager that increases redaction sensitivity.
    """
    # Bind a flag to indicate we're in a sensitive context
    token = structlog.contextvars.bind_contextvars(
        _sensitive_context=True
    )
    try:
        yield
    finally:
        structlog.contextvars.unbind_contextvars("_sensitive_context")

def enhanced_redact_processor(logger, method_name: str, event_dict: dict) -> dict:
    """Enhanced redaction in sensitive contexts."""
    is_sensitive = event_dict.pop("_sensitive_context", False)

    if is_sensitive:
        # In sensitive context, redact everything that looks like data
        for key, value in event_dict.items():
            if isinstance(value, str) and len(value) > 3:
                event_dict[key] = "[REDACTED_SENSITIVE]"
    else:
        # Normal PII redaction
        event_dict = redact_pii(logger, method_name, event_dict)

    return event_dict

# Usage
logger = structlog.get_logger()

with sensitive_logging_context():
    logger.info("processing_payment", card="4111111111111111")
    # Everything is redacted in this context

Testing Log Redaction

import pytest
import structlog
from io import StringIO

def test_pii_redaction_in_logs():
    """Verify PII is redacted from logs."""
    output = StringIO()

    structlog.configure(
        processors=[
            redact_pii,
            structlog.processors.JSONRenderer()
        ],
        logger_factory=structlog.WriteLoggerFactory(file=output)
    )

    logger = structlog.get_logger()
    logger.info("test", email="test@example.com", ssn="123-45-6789")

    log_output = output.getvalue()

    assert "test@example.com" not in log_output
    assert "123-45-6789" not in log_output
    assert "[REDACTED_EMAIL]" in log_output
    assert "[REDACTED_SSN]" in log_output

References

• Structlog Documentation
• Loguru Documentation
• Structlog Processors Guide

OAuth 2.1 & Passkeys Reference

OAuth 2.1 Overview

OAuth 2.1 consolidates OAuth 2.0 best practices and security requirements:

Key Changes from OAuth 2.0

• PKCE required for ALL clients (not just public)
• Implicit grant removed (security vulnerability)
• Password grant removed (credential anti-pattern)
• Bearer tokens must use TLS
• Refresh token rotation mandatory

PKCE Flow (Required)

import hashlib
import base64
import secrets

def generate_pkce_pair():
    """Generate code_verifier and code_challenge for PKCE."""
    # Generate random code_verifier (43-128 chars)
    code_verifier = secrets.token_urlsafe(64)
    
    # Create code_challenge using S256
    digest = hashlib.sha256(code_verifier.encode()).digest()
    code_challenge = base64.urlsafe_b64encode(digest).rstrip(b'=').decode()
    
    return code_verifier, code_challenge

# Usage
verifier, challenge = generate_pkce_pair()

# Step 1: Authorization request
auth_url = f"""https://auth.example.com/authorize?
    response_type=code
    &client_id={client_id}
    &redirect_uri={redirect_uri}
    &code_challenge={challenge}
    &code_challenge_method=S256
    &state={state}
    &scope=openid profile"""

# Step 2: Exchange code for tokens
token_response = requests.post(
    "https://auth.example.com/token",
    data={
        "grant_type": "authorization_code",
        "code": auth_code,
        "redirect_uri": redirect_uri,
        "client_id": client_id,
        "code_verifier": verifier,  # PKCE verification
    }
)

Token Lifetimes (2026 Recommendations)

DPoP (Demonstrating Proof of Possession)

Binds tokens to client cryptographic keys:

import jwt
import time
import uuid

def create_dpop_proof(http_method: str, http_uri: str, private_key) -> str:
    """Create DPoP proof for request."""
    claims = {
        "jti": str(uuid.uuid4()),
        "htm": http_method,
        "htu": http_uri,
        "iat": int(time.time()),
    }
    
    headers = {
        "typ": "dpop+jwt",
        "alg": "ES256",
        "jwk": private_key.public_key().export_key(),
    }
    
    return jwt.encode(claims, private_key, algorithm="ES256", headers=headers)

# Usage
dpop_proof = create_dpop_proof("POST", "https://api.example.com/token", private_key)

response = requests.post(
    "https://api.example.com/token",
    headers={"DPoP": dpop_proof},
    data={"grant_type": "refresh_token", "refresh_token": rt},
)

Passkeys / WebAuthn

Overview

Passkeys replace passwords with cryptographic credentials:

• Phishing-resistant: Bound to origin
• Passwordless: No secrets to remember
• Multi-device: Synced via platform
• Biometric: Face ID, Touch ID, fingerprint

Registration Flow

from webauthn import (
    generate_registration_options,
    verify_registration_response,
)
from webauthn.helpers.structs import (
    AuthenticatorSelectionCriteria,
    ResidentKeyRequirement,
    UserVerificationRequirement,
)

# Step 1: Generate registration options
options = generate_registration_options(
    rp_id="example.com",
    rp_name="Example App",
    user_id=user.id.encode(),
    user_name=user.email,
    user_display_name=user.name,
    authenticator_selection=AuthenticatorSelectionCriteria(
        resident_key=ResidentKeyRequirement.REQUIRED,
        user_verification=UserVerificationRequirement.REQUIRED,
    ),
)

# Send options to client
return jsonify(options)

# Step 2: Verify registration response
verification = verify_registration_response(
    credential=client_response,
    expected_challenge=stored_challenge,
    expected_rp_id="example.com",
    expected_origin="https://example.com",
)

# Store credential
db.save_credential(
    user_id=user.id,
    credential_id=verification.credential_id,
    public_key=verification.credential_public_key,
    sign_count=verification.sign_count,
)

Authentication Flow

from webauthn import (
    generate_authentication_options,
    verify_authentication_response,
)

# Step 1: Generate authentication options
options = generate_authentication_options(
    rp_id="example.com",
    allow_credentials=[
        {"id": cred.credential_id, "type": "public-key"}
        for cred in user.credentials
    ],
)

# Step 2: Verify authentication response
verification = verify_authentication_response(
    credential=client_response,
    expected_challenge=stored_challenge,
    expected_rp_id="example.com",
    expected_origin="https://example.com",
    credential_public_key=stored_credential.public_key,
    credential_current_sign_count=stored_credential.sign_count,
)

# Update sign count (replay protection)
stored_credential.sign_count = verification.new_sign_count
db.save(stored_credential)

# Issue session/tokens
return create_session(user)

Frontend Implementation

// Registration
async function registerPasskey(options: PublicKeyCredentialCreationOptions) {
  const credential = await navigator.credentials.create({
    publicKey: options,
  });
  
  // Send credential to server
  await fetch('/api/auth/passkey/register', {
    method: 'POST',
    body: JSON.stringify(credential),
  });
}

// Authentication
async function authenticateWithPasskey(options: PublicKeyCredentialRequestOptions) {
  const credential = await navigator.credentials.get({
    publicKey: options,
  });
  
  // Send credential to server
  const response = await fetch('/api/auth/passkey/authenticate', {
    method: 'POST',
    body: JSON.stringify(credential),
  });
  
  return response.json();
}

// Conditional UI (autofill)
if (window.PublicKeyCredential?.isConditionalMediationAvailable) {
  const available = await PublicKeyCredential.isConditionalMediationAvailable();
  if (available) {
    // Show passkey autofill in username field
    const credential = await navigator.credentials.get({
      publicKey: options,
      mediation: 'conditional',
    });
  }
}

Refresh Token Rotation

import secrets
import hashlib
from datetime import datetime, timedelta, timezone

def rotate_refresh_token(old_token: str, db) -> tuple[str, str]:
    """Rotate refresh token on use (security best practice)."""
    old_hash = hashlib.sha256(old_token.encode()).hexdigest()
    
    # Find and validate old token
    token_record = db.query("""
        SELECT user_id, version FROM refresh_tokens
        WHERE token_hash = ? AND expires_at > NOW() AND revoked = FALSE
    """, [old_hash]).fetchone()
    
    if not token_record:
        raise InvalidTokenError("Refresh token invalid or expired")
    
    user_id, version = token_record
    
    # Revoke old token
    db.execute(
        "UPDATE refresh_tokens SET revoked = TRUE WHERE token_hash = ?",
        [old_hash]
    )
    
    # Create new tokens
    new_access_token = create_access_token(user_id)
    
    new_refresh_token = secrets.token_urlsafe(32)
    new_hash = hashlib.sha256(new_refresh_token.encode()).hexdigest()
    
    db.execute("""
        INSERT INTO refresh_tokens (user_id, token_hash, expires_at, version)
        VALUES (?, ?, ?, ?)
    """, [user_id, new_hash, datetime.now(timezone.utc) + timedelta(days=7), version + 1])
    
    return new_access_token, new_refresh_token

External Links

• OAuth 2.1 Draft Spec
• WebAuthn Spec
• FIDO Alliance
• py_webauthn Library

Output Guardrails

Purpose

After LLM returns, validate the output before using it:

┌────────────────────────────────────────────────────────────┐
│                  OUTPUT VALIDATION                         │
├────────────────────────────────────────────────────────────┤
│                                                            │
│  LLM Response ──► Guardrails ──► Validated Output          │
│                       │                                    │
│                       ▼                                    │
│              ┌────────────────┐                            │
│              │   VALIDATORS   │                            │
│              ├────────────────┤                            │
│              │ □ Schema       │  Does it match expected?   │
│              │ □ No IDs       │  No hallucinated UUIDs?    │
│              │ □ Grounded     │  Supported by context?     │
│              │ □ Safe         │  No toxic content?         │
│              │ □ Size         │  Within limits?            │
│              └────────────────┘                            │
│                       │                                    │
│           ┌──────────┴──────────┐                         │
│           ▼                     ▼                         │
│    ┌──────────┐          ┌──────────┐                     │
│    │   PASS   │          │   FAIL   │                     │
│    │          │          │          │                     │
│    │ Continue │          │ Retry or │                     │
│    │          │          │ Error    │                     │
│    └──────────┘          └──────────┘                     │
│                                                            │
└────────────────────────────────────────────────────────────┘

Implementation

1. Validation Result Type

from dataclasses import dataclass
from enum import Enum

class ValidationStatus(Enum):
    PASSED = "passed"
    FAILED = "failed"
    WARNING = "warning"

@dataclass
class ValidationResult:
    status: ValidationStatus
    reason: str | None = None
    details: dict | None = None

    @property
    def is_valid(self) -> bool:
        return self.status in (ValidationStatus.PASSED, ValidationStatus.WARNING)

2. Schema Validation

from pydantic import BaseModel, ValidationError
from typing import TypeVar

T = TypeVar("T", bound=BaseModel)

def validate_schema(
    llm_output: dict,
    schema: type[T],
) -> tuple[T | None, ValidationResult]:
    """
    Validate LLM output matches expected schema.
    """
    try:
        parsed = schema.model_validate(llm_output)
        return parsed, ValidationResult(
            status=ValidationStatus.PASSED,
        )
    except ValidationError as e:
        return None, ValidationResult(
            status=ValidationStatus.FAILED,
            reason=f"Schema validation failed: {e.error_count()} errors",
            details={"errors": e.errors()},
        )

# Usage
class AnalysisOutput(BaseModel):
    summary: str
    key_concepts: list[str]
    difficulty: str

parsed, result = validate_schema(llm_response, AnalysisOutput)
if not result.is_valid:
    raise ValidationError(result.reason)

3. No Hallucinated IDs

import re

UUID_PATTERN = r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'

def validate_no_ids(output: str) -> ValidationResult:
    """
    Ensure LLM didn't hallucinate any identifiers.
    """
    # Check for UUIDs
    uuids = re.findall(UUID_PATTERN, output, re.IGNORECASE)
    if uuids:
        return ValidationResult(
            status=ValidationStatus.FAILED,
            reason=f"Found {len(uuids)} hallucinated UUIDs",
            details={"uuids": uuids},
        )

    # Check for ID-like patterns
    id_patterns = [
        r'user_id[:\s]+\S+',
        r'doc_id[:\s]+\S+',
        r'id[:\s]+[a-f0-9]{8,}',
    ]

    for pattern in id_patterns:
        matches = re.findall(pattern, output, re.IGNORECASE)
        if matches:
            return ValidationResult(
                status=ValidationStatus.WARNING,
                reason=f"Found ID-like pattern: {matches[0]}",
                details={"matches": matches},
            )

    return ValidationResult(status=ValidationStatus.PASSED)

4. Grounding Validation

def validate_grounding(
    output: str,
    context_texts: list[str],
    threshold: float = 0.3,
) -> ValidationResult:
    """
    Check if LLM output is grounded in provided context.
    Uses simple keyword overlap for speed.
    """
    # Extract key terms from output
    output_terms = set(extract_key_terms(output))

    # Extract key terms from context
    context_terms = set()
    for text in context_texts:
        context_terms.update(extract_key_terms(text))

    # Calculate overlap
    if not output_terms:
        return ValidationResult(
            status=ValidationStatus.WARNING,
            reason="No key terms in output",
        )

    overlap = len(output_terms & context_terms) / len(output_terms)

    if overlap < threshold:
        return ValidationResult(
            status=ValidationStatus.WARNING,
            reason=f"Low grounding score: {overlap:.2%}",
            details={
                "overlap": overlap,
                "threshold": threshold,
                "ungrounded_terms": list(output_terms - context_terms)[:10],
            },
        )

    return ValidationResult(
        status=ValidationStatus.PASSED,
        details={"grounding_score": overlap},
    )

def extract_key_terms(text: str) -> list[str]:
    """Extract meaningful terms from text"""
    import re
    # Simple: words 4+ chars, lowercased
    words = re.findall(r'\b[a-zA-Z]{4,}\b', text.lower())
    # Filter common words
    stopwords = {'this', 'that', 'with', 'from', 'have', 'been', 'will', 'would'}
    return [w for w in words if w not in stopwords]

5. Content Safety

async def validate_content_safety(
    output: str,
) -> ValidationResult:
    """
    Check for toxic/harmful content.
    Uses simple pattern matching + optional LLM check.
    """
    # Quick pattern check
    toxic_patterns = [
        r'\b(hate|violence|harm|kill)\b',
        r'\b(password|secret|api.?key)\b',
    ]

    for pattern in toxic_patterns:
        if re.search(pattern, output, re.IGNORECASE):
            return ValidationResult(
                status=ValidationStatus.FAILED,
                reason=f"Potentially unsafe content detected",
            )

    # PII detection
    pii_patterns = {
        "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        "phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
        "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
    }

    detected_pii = []
    for pii_type, pattern in pii_patterns.items():
        if re.search(pattern, output):
            detected_pii.append(pii_type)

    if detected_pii:
        return ValidationResult(
            status=ValidationStatus.WARNING,
            reason=f"Potential PII detected: {detected_pii}",
            details={"pii_types": detected_pii},
        )

    return ValidationResult(status=ValidationStatus.PASSED)

6. Size Limits

def validate_size(
    output: str,
    max_chars: int = 50000,
    max_tokens: int = 10000,
) -> ValidationResult:
    """
    Ensure output is within size limits.
    """
    if len(output) > max_chars:
        return ValidationResult(
            status=ValidationStatus.FAILED,
            reason=f"Output exceeds {max_chars} chars: {len(output)}",
        )

    # Rough token estimate
    estimated_tokens = len(output) // 4
    if estimated_tokens > max_tokens:
        return ValidationResult(
            status=ValidationStatus.WARNING,
            reason=f"Output may exceed token limit: ~{estimated_tokens}",
        )

    return ValidationResult(status=ValidationStatus.PASSED)

7. Combined Validator

from dataclasses import dataclass

@dataclass
class GuardrailsConfig:
    validate_schema: bool = True
    validate_no_ids: bool = True
    validate_grounding: bool = True
    validate_safety: bool = True
    validate_size: bool = True
    grounding_threshold: float = 0.3
    max_output_chars: int = 50000

async def run_guardrails(
    llm_output: dict,
    context_texts: list[str],
    schema: type[BaseModel],
    config: GuardrailsConfig = GuardrailsConfig(),
) -> tuple[BaseModel | None, list[ValidationResult]]:
    """
    Run all guardrails on LLM output.
    Returns parsed output and all validation results.
    """
    results = []
    parsed = None

    # 1. Schema validation
    if config.validate_schema:
        parsed, result = validate_schema(llm_output, schema)
        results.append(result)
        if not result.is_valid:
            return None, results  # Stop early

    output_str = str(llm_output)

    # 2. No hallucinated IDs
    if config.validate_no_ids:
        result = validate_no_ids(output_str)
        results.append(result)

    # 3. Grounding check
    if config.validate_grounding:
        result = validate_grounding(
            output_str,
            context_texts,
            config.grounding_threshold,
        )
        results.append(result)

    # 4. Content safety
    if config.validate_safety:
        result = await validate_content_safety(output_str)
        results.append(result)

    # 5. Size limits
    if config.validate_size:
        result = validate_size(output_str, config.max_output_chars)
        results.append(result)

    # Check for failures
    failures = [r for r in results if r.status == ValidationStatus.FAILED]
    if failures:
        return None, results

    return parsed, results

OrchestKit Integration

# backend/app/workflows/agents/content_analyzer.py

async def analyze_with_guardrails(state: AnalysisState) -> AnalysisState:
    """Run LLM with output guardrails"""

    # Call LLM
    llm_response = await llm.generate(state.prompt)

    # Run guardrails
    parsed, validations = await run_guardrails(
        llm_output=llm_response,
        context_texts=state.context_texts,
        schema=AnalysisOutput,
    )

    # Log validations
    for v in validations:
        if v.status != ValidationStatus.PASSED:
            logger.warning(
                "guardrail_issue",
                status=v.status.value,
                reason=v.reason,
                trace_id=state.request_context.trace_id,
            )

    if parsed is None:
        raise GuardrailError(
            "LLM output failed validation",
            validations=[v for v in validations if not v.is_valid],
        )

    return state.with_output(parsed)

Common Mistakes

# ❌ BAD: No validation
artifact.content = llm_response["content"]  # Could be anything!

# ❌ BAD: Only schema validation
parsed = AnalysisOutput.parse_obj(response)  # Ignores content issues

# ❌ BAD: Trusting LLM completely
if llm_response.get("is_safe", True):  # LLM said it's safe!
    use_response(llm_response)

# ✅ GOOD: Full guardrail pipeline
parsed, results = await run_guardrails(
    llm_output=response,
    context_texts=context,
    schema=AnalysisOutput,
)

Testing Guardrails

class TestGuardrails:

    def test_detects_hallucinated_uuid(self):
        output = "Analysis for doc 123e4567-e89b-12d3-a456-426614174000"
        result = validate_no_ids(output)
        assert result.status == ValidationStatus.FAILED

    def test_detects_low_grounding(self):
        output = "This is about quantum physics and black holes"
        context = ["Python programming tutorial"]
        result = validate_grounding(output, context)
        assert result.status == ValidationStatus.WARNING

    async def test_detects_pii(self):
        output = "Contact john@example.com for details"
        result = await validate_content_safety(output)
        assert result.status == ValidationStatus.WARNING
        assert "email" in result.details["pii_types"]

    async def test_full_pipeline_passes(self):
        valid_output = {
            "summary": "Introduction to machine learning",
            "key_concepts": ["ML", "training", "models"],
            "difficulty": "intermediate",
        }
        context = ["Machine learning is a subset of AI..."]

        parsed, results = await run_guardrails(
            llm_output=valid_output,
            context_texts=context,
            schema=AnalysisOutput,
        )

        assert parsed is not None
        assert all(r.is_valid for r in results)

Post-LLM Attribution

The Principle

Attribution is DETERMINISTIC, not LLM-generated.

>

The LLM produces content. We attach context from our records.

┌────────────────────────────────────────────────────────────┐
│                   POST-LLM PHASE                           │
├────────────────────────────────────────────────────────────┤
│                                                            │
│                ┌─────────────────────┐                     │
│                │       LLM           │                     │
│                │                     │                     │
│                │  Output: content    │                     │
│                │  (text, analysis)   │                     │
│                └──────────┬──────────┘                     │
│                           │                                │
│                           ▼                                │
│              ┌────────────────────────┐                    │
│              │   ATTRIBUTION LAYER    │                    │
│              │                        │                    │
│  From Pre-LLM:                        From Context:        │
│  ├─ source_refs ─────────────────────► source_ids         │
│  └─ chunk_ids                          │                   │
│                                        │                   │
│  From RequestContext:                  │                   │
│  ├─ user_id ─────────────────────────► user_id            │
│  ├─ tenant_id ───────────────────────► tenant_id          │
│  ├─ trace_id ────────────────────────► trace_id           │
│  └─ analysis_id ─────────────────────► analysis_id        │
│                                        │                   │
│  Generated:                            │                   │
│  ├─ new UUID ────────────────────────► artifact_id        │
│  └─ timestamp ───────────────────────► created_at         │
│              │                        │                    │
│              └────────────┬───────────┘                    │
│                           │                                │
│                           ▼                                │
│              ┌────────────────────────┐                    │
│              │    COMPLETE RESULT     │                    │
│              │                        │                    │
│              │  content + attribution │                    │
│              │  (ready for storage)   │                    │
│              └────────────────────────┘                    │
│                                                            │
└────────────────────────────────────────────────────────────┘

Implementation

1. Attribution Data Structure

from dataclasses import dataclass
from datetime import datetime
from uuid import UUID, uuid4

@dataclass
class AttributedResult:
    """LLM output with deterministic attribution"""

    # Generated identifier
    id: UUID

    # From RequestContext (system-provided)
    user_id: UUID
    tenant_id: UUID
    analysis_id: UUID
    trace_id: str

    # From Pre-LLM refs (deterministic)
    source_document_ids: list[UUID]
    source_chunk_ids: list[UUID]

    # From LLM (content only)
    content: str
    key_concepts: list[str]
    difficulty_level: str
    summary: str

    # Metadata
    created_at: datetime
    model_used: str
    processing_time_ms: float

2. Attribution Function

async def attribute_llm_output(
    llm_output: dict,
    ctx: RequestContext,
    source_refs: SourceReference,
    model_name: str,
    processing_time_ms: float,
) -> AttributedResult:
    """
    Attach context to LLM output.
    All attribution comes from our records, not the LLM.
    """

    # Validate LLM output has no IDs
    if contains_identifiers(llm_output):
        raise SecurityError("LLM output contains identifiers")

    return AttributedResult(
        # New ID for this artifact
        id=uuid4(),

        # From RequestContext (verified from JWT)
        user_id=ctx.user_id,
        tenant_id=ctx.tenant_id,
        analysis_id=ctx.resource_id,
        trace_id=ctx.trace_id,

        # From Pre-LLM capture (deterministic)
        source_document_ids=source_refs.document_ids,
        source_chunk_ids=source_refs.chunk_ids,

        # From LLM (content only)
        content=llm_output["analysis"],
        key_concepts=llm_output.get("key_concepts", []),
        difficulty_level=llm_output.get("difficulty", "intermediate"),
        summary=llm_output.get("summary", ""),

        # Metadata
        created_at=datetime.now(timezone.utc),
        model_used=model_name,
        processing_time_ms=processing_time_ms,
    )

def contains_identifiers(output: dict) -> bool:
    """Check if LLM output contains any identifiers"""
    import re

    output_str = str(output)

    # Check for UUIDs
    if re.search(UUID_PATTERN, output_str):
        return True

    # Check for ID field names in content
    for field in ["user_id", "tenant_id", "document_id"]:
        if field in output_str.lower():
            return True

    return False

3. Storage with Attribution

async def save_attributed_result(
    result: AttributedResult,
    db: AsyncSession,
) -> None:
    """
    Save result with all attribution intact.
    Attribution comes from our context, not LLM.
    """

    # Create artifact record
    artifact = Artifact(
        id=result.id,
        user_id=result.user_id,
        tenant_id=result.tenant_id,
        analysis_id=result.analysis_id,
        content=result.content,
        key_concepts=result.key_concepts,
        difficulty_level=result.difficulty_level,
        summary=result.summary,
        created_at=result.created_at,
        model_used=result.model_used,
    )
    db.add(artifact)

    # Create source links
    for doc_id in result.source_document_ids:
        link = ArtifactSourceLink(
            artifact_id=result.id,
            document_id=doc_id,
            tenant_id=result.tenant_id,  # Denormalized for RLS
        )
        db.add(link)

    await db.commit()

    # Audit log
    logger.audit(
        "artifact.created",
        artifact_id=result.id,
        user_id=result.user_id,
        tenant_id=result.tenant_id,
        source_count=len(result.source_document_ids),
    )

OrchestKit Integration

Content Analysis Workflow

# backend/app/workflows/agents/content_analyzer.py

async def create_analysis_artifact(state: AnalysisState) -> AnalysisState:
    """Create artifact with proper attribution"""

    # LLM output (content only)
    llm_output = state.llm_response

    # Attribute using our context
    attributed = await attribute_llm_output(
        llm_output=llm_output,
        ctx=state.request_context,          # From JWT
        source_refs=state.source_refs,       # From pre-LLM
        model_name=state.model_used,
        processing_time_ms=state.llm_time_ms,
    )

    # Save with attribution
    await save_attributed_result(attributed, state.db)

    return state.with_artifact(attributed)

Artifact Retrieval

# backend/app/api/artifacts.py

@router.get("/{artifact_id}")
async def get_artifact(
    artifact_id: UUID,
    ctx: RequestContext = Depends(get_request_context),
    db: AsyncSession = Depends(get_db),
):
    """Get artifact with source attribution"""

    # Query with tenant filter
    artifact = await db.execute(
        """
        SELECT a.*, array_agg(asl.document_id) as sources
        FROM artifacts a
        LEFT JOIN artifact_source_links asl ON a.id = asl.artifact_id
        WHERE a.id = :id
          AND a.tenant_id = :tenant_id  -- ALWAYS filter
        GROUP BY a.id
        """,
        {
            "id": artifact_id,
            "tenant_id": ctx.tenant_id,
        }
    )

    if not artifact:
        raise HTTPException(404)

    return ArtifactResponse(
        id=artifact.id,
        content=artifact.content,
        sources=artifact.sources,  # Deterministic from our records
        created_at=artifact.created_at,
    )

Common Mistakes

# ❌ BAD: Asking LLM for attribution
prompt = "Analyze this and tell me which document it came from"
response = llm.generate(prompt)
doc_id = response["source_document"]  # HALLUCINATED!

# ❌ BAD: Trusting LLM-provided IDs
llm_output = {"analysis": "...", "user_id": "abc123"}
artifact.user_id = llm_output["user_id"]  # WRONG!

# ❌ BAD: Generating IDs in prompt
prompt = f"Generate a unique ID for this analysis: {analysis_id}"

# ✅ GOOD: Attribution from our records
artifact.user_id = ctx.user_id  # From JWT
artifact.sources = source_refs.document_ids  # From pre-LLM

# ✅ GOOD: Generate IDs ourselves
artifact.id = uuid4()  # We generate

# ✅ GOOD: LLM provides content only
artifact.content = llm_output["analysis"]  # Just the text

Testing Attribution

class TestAttribution:

    async def test_attribution_from_context_not_llm(self, ctx):
        """Attribution must come from our context"""

        # LLM returns content only
        llm_output = {
            "analysis": "This is the analysis",
            "key_concepts": ["ML", "AI"],
        }

        source_refs = SourceReference(
            document_ids=[uuid4(), uuid4()],
            chunk_ids=[uuid4()],
        )

        result = await attribute_llm_output(
            llm_output=llm_output,
            ctx=ctx,
            source_refs=source_refs,
        )

        # Attribution from context, not LLM
        assert result.user_id == ctx.user_id
        assert result.tenant_id == ctx.tenant_id
        assert result.source_document_ids == source_refs.document_ids

    async def test_rejects_llm_with_ids(self, ctx):
        """Reject LLM output that contains IDs"""

        bad_output = {
            "analysis": "Result for user 123e4567-e89b-12d3-a456-426614174000",
        }

        with pytest.raises(SecurityError):
            await attribute_llm_output(bad_output, ctx, source_refs)

    async def test_source_links_created(self, ctx, db):
        """Source links are created with artifact"""

        result = await attribute_llm_output(...)
        await save_attributed_result(result, db)

        links = await db.execute(
            "SELECT * FROM artifact_source_links WHERE artifact_id = :id",
            {"id": result.id}
        )

        assert len(links) == len(result.source_document_ids)

Pre-LLM Filtering

Purpose

Before ANY data reaches the LLM, it must be: 1. Scoped to the current tenant/user 2. Filtered for relevance 3. Stripped of identifiers 4. Captured for later attribution

┌────────────────────────────────────────────────────────────┐
│                    PRE-LLM PHASE                           │
├────────────────────────────────────────────────────────────┤
│                                                            │
│  User Query ──► Tenant Filter ──► Content Extract ──► LLM  │
│       │              │                   │                 │
│       │              │                   │                 │
│       ▼              ▼                   ▼                 │
│  ┌─────────┐   ┌───────────┐     ┌─────────────┐          │
│  │ Query   │   │ Documents │     │ Text Only   │          │
│  │ Text    │   │ for THIS  │     │ (no IDs)    │          │
│  │         │   │ tenant    │     │             │          │
│  └─────────┘   └───────────┘     └─────────────┘          │
│                      │                                     │
│                      ▼                                     │
│              ┌─────────────┐                              │
│              │ Save Refs   │                              │
│              │ for Later   │  ◄── For post-LLM attribution│
│              │ Attribution │                              │
│              └─────────────┘                              │
│                                                            │
└────────────────────────────────────────────────────────────┘

Implementation

1. Tenant-Scoped Retrieval

from uuid import UUID
from dataclasses import dataclass

@dataclass
class SourceReference:
    """Tracks what was retrieved for attribution"""
    document_ids: list[UUID]
    chunk_ids: list[UUID]
    similarity_scores: list[float]
    retrieval_timestamp: datetime

async def retrieve_with_isolation(
    query: str,
    ctx: RequestContext,
    limit: int = 10,
) -> tuple[list[str], SourceReference]:
    """
    Retrieve documents scoped to tenant/user.
    Returns: (content_texts, source_references)
    """
    # Embed query
    query_embedding = await embed(query)

    # Search with MANDATORY tenant filter
    results = await db.execute(
        """
        SELECT id, chunk_id, content,
               1 - (embedding <-> :query) as similarity
        FROM document_chunks
        WHERE tenant_id = :tenant_id    -- REQUIRED
          AND user_id = :user_id        -- REQUIRED
          AND embedding <-> :query < 0.5
        ORDER BY embedding <-> :query
        LIMIT :limit
        """,
        {
            "tenant_id": ctx.tenant_id,  # From JWT
            "user_id": ctx.user_id,       # From JWT
            "query": query_embedding,
            "limit": limit,
        }
    )

    # Separate content from references
    content_texts = [r.content for r in results]
    source_refs = SourceReference(
        document_ids=[r.id for r in results],
        chunk_ids=[r.chunk_id for r in results],
        similarity_scores=[r.similarity for r in results],
        retrieval_timestamp=datetime.now(timezone.utc),
    )

    return content_texts, source_refs

2. Content Extraction (Strip IDs)

def extract_content_only(documents: list[Document]) -> list[str]:
    """
    Extract text content, stripping any embedded IDs.
    """
    contents = []
    for doc in documents:
        # Get content
        text = doc.content

        # Remove any embedded IDs (defensive)
        text = strip_identifiers(text)

        contents.append(text)

    return contents

def strip_identifiers(text: str) -> str:
    """Remove any identifiers that might have leaked into content"""
    import re

    # Remove UUIDs
    text = re.sub(
        r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
        '[REDACTED]',
        text,
        flags=re.IGNORECASE
    )

    # Remove common ID patterns
    patterns = [
        r'user_id:\s*\S+',
        r'tenant_id:\s*\S+',
        r'doc_id:\s*\S+',
    ]
    for pattern in patterns:
        text = re.sub(pattern, '[REDACTED]', text, flags=re.IGNORECASE)

    return text

3. Full Pre-LLM Pipeline

@dataclass
class PreLLMResult:
    """Complete pre-LLM preparation result"""
    query: str
    context_texts: list[str]
    source_refs: SourceReference
    preparation_time_ms: float

async def prepare_for_llm(
    query: str,
    ctx: RequestContext,
) -> PreLLMResult:
    """
    Complete pre-LLM preparation:
    1. Retrieve with tenant isolation
    2. Extract content only
    3. Save references for attribution
    """
    start = time.monotonic()

    # Step 1: Tenant-scoped retrieval
    raw_results, source_refs = await retrieve_with_isolation(
        query=query,
        ctx=ctx,
    )

    # Step 2: Extract and clean content
    context_texts = [strip_identifiers(text) for text in raw_results]

    # Step 3: Audit for any remaining IDs
    for text in context_texts:
        violations = audit_prompt(text)
        if violations:
            logger.warning(
                "ID found in content, redacting",
                violations=violations,
            )

    elapsed = (time.monotonic() - start) * 1000

    return PreLLMResult(
        query=query,
        context_texts=context_texts,
        source_refs=source_refs,
        preparation_time_ms=elapsed,
    )

OrchestKit Integration

In Content Analysis Workflow

# backend/app/workflows/agents/retriever.py

async def retrieve_context(state: AnalysisState) -> AnalysisState:
    """RAG retrieval with tenant isolation"""

    ctx = state.request_context

    # Pre-LLM preparation
    pre_llm = await prepare_for_llm(
        query=state.analysis_request.query,
        ctx=ctx,
    )

    # Store for later phases
    return state.copy(
        context_texts=pre_llm.context_texts,
        source_refs=pre_llm.source_refs,
        # NO IDs in state that goes to LLM
    )

In Library Search

# backend/app/services/search.py

async def search_libraries(
    query: str,
    ctx: RequestContext,
) -> SearchResult:
    """Search golden dataset with isolation"""

    # Always filter by tenant
    results = await db.execute(
        """
        SELECT id, title, url, summary, content
        FROM golden_dataset
        WHERE tenant_id = :tenant_id
          AND search_vector @@ plainto_tsquery(:query)
        ORDER BY ts_rank(search_vector, plainto_tsquery(:query)) DESC
        LIMIT 20
        """,
        {
            "tenant_id": ctx.tenant_id,
            "query": query,
        }
    )

    # Return content and refs separately
    return SearchResult(
        items=[r.content for r in results],  # Content for LLM
        refs=[r.id for r in results],         # IDs for attribution
    )

Common Mistakes

# ❌ BAD: Query without tenant filter
results = await db.execute("SELECT * FROM documents")

# ❌ BAD: Tenant filter as optional
async def search(tenant_id: UUID | None = None):
    query = "SELECT * FROM documents"
    if tenant_id:  # Can be bypassed!
        query += f" WHERE tenant_id = '{tenant_id}'"

# ❌ BAD: Trusting client-provided tenant
async def search(request: Request):
    tenant_id = request.query_params["tenant_id"]  # Attacker controls!

# ❌ BAD: Including IDs in content
results = [{"id": doc.id, "content": doc.content} for doc in docs]

# ✅ GOOD: Mandatory tenant filter from context
results = await db.execute(
    "SELECT content FROM documents WHERE tenant_id = :tid",
    {"tid": ctx.tenant_id}  # From verified JWT
)

# ✅ GOOD: Content separate from refs
content = [doc.content for doc in docs]  # For LLM
refs = [doc.id for doc in docs]           # For attribution

Testing Pre-LLM Filtering

class TestPreLLMFiltering:

    async def test_retrieval_respects_tenant(
        self,
        tenant_a_ctx,
        tenant_b_ctx,
    ):
        # Create doc for tenant B
        await create_document(
            tenant_id=tenant_b_ctx.tenant_id,
            content="Secret data",
        )

        # Search as tenant A
        result = await prepare_for_llm(
            query="secret",
            ctx=tenant_a_ctx,
        )

        # Must not find tenant B's data
        assert len(result.context_texts) == 0

    async def test_content_has_no_uuids(self, ctx):
        result = await prepare_for_llm(
            query="test query",
            ctx=ctx,
        )

        for text in result.context_texts:
            assert not re.search(UUID_PATTERN, text)

    async def test_source_refs_captured(self, ctx):
        result = await prepare_for_llm(
            query="test query",
            ctx=ctx,
        )

        # Refs saved for attribution
        assert len(result.source_refs.document_ids) > 0
        assert result.source_refs.retrieval_timestamp is not None

[Rule Name]

[Brief description — 1-2 sentences.]

Incorrect:

// Bad pattern

Correct:

// Good pattern

Key rules:

• [Rule 1]
• [Rule 2]
• [Rule 3]

Reference: [link]