``` ## Server-Side XSS Prevention (Flask) ```python from flask import request, render_template_string from markupsafe import escape @app.route('/greet') def greet(): name = request.args.get('name', '') return f"

Hello, {escape(name)}!

" # Or use Jinja2 templates (auto-escape by default) @app.route('/greet-template') def greet_template(): return render_template_string( "

Hello, {{ name }}!

", name=request.args.get('name', '') ) ``` ## Security Headers ```python SECURITY_HEADERS = { "Strict-Transport-Security": "max-age=31536000; includeSubDomains", "X-Content-Type-Options": "nosniff", "X-Frame-Options": "DENY", "X-XSS-Protection": "1; mode=block", "Referrer-Policy": "strict-origin-when-cross-origin", "Content-Security-Policy": "default-src 'self'", } @app.after_request def add_security_headers(response): for header, value in SECURITY_HEADERS.items(): response.headers[header] = value return response ``` ## SRI for CDN Scripts ```html ``` ## Anti-Patterns ```javascript // NEVER use innerHTML with user input element.innerHTML = userInput; // NEVER use dangerouslySetInnerHTML without sanitization

// NEVER trust Content-Type header for file validation if (file.type === 'image/png') {...} // Can be spoofed // ALWAYS use textContent or DOMPurify element.textContent = userInput; const safe = DOMPurify.sanitize(userInput); ``` **Incorrect — Using innerHTML with user content allows XSS script injection:** ```javascript const userComment = ""; element.innerHTML = userComment; // Script executes, stealing cookies/tokens ``` **Correct — Using textContent automatically escapes HTML and prevents XSS:** ```javascript const userComment = ""; element.textContent = userComment; // Renders as plain text: "" ``` --- ## References (11) ### Audit Logging # 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) | Event Type | What to Log | Example | |------------|-------------|---------| | Authentication | Success/failure, method | "User login via OAuth" | | Authorization | Decision, resource, action | "Access granted to analysis_123" | | Data Access | Read/write, resource type | "Read 10 documents" | | Data Modification | Before/after (hashed), resource | "Updated analysis status" | | LLM Calls | Model, tokens, latency (NOT prompt) | "GPT-4, 1500 tokens, 2.3s" | | Errors | Type, context (sanitized) | "ValidationError on /api/analyze" | ### Never Log (Sensitive Data) | Data Type | Why Not | Alternative | |-----------|---------|-------------| | Passwords | Security | Log "password changed" event | | API Keys | Security | Log key ID, not value | | Full Prompts | May contain PII | Log prompt hash, token count | | LLM Responses | May contain generated PII | Log response hash, length | | User Content | Privacy | Log content hash, length | | PII | GDPR/Privacy | Log anonymized or redacted | ## Implementation ### Sanitized Logger ```python 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 ```python 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 ```python # 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 | Environment | Retention | Reason | |-------------|-----------|--------| | Development | 7 days | Debugging | | Staging | 30 days | Testing | | Production | 1 year | Compliance | | Security Events | 7 years | Legal requirements | ## Integration with Langfuse ```python 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 ### Cc Permission Model # Claude Code Permission-Rule Semantics (security-relevant) How Claude Code's `allow` / `ask` / `deny` permission rules actually behave as of the supported floor (every behavior below had shipped by 2.1.166; current floor 2.1.220). 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: ```jsonc // .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**: ```jsonc { "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**: ```jsonc { "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. ## 6. Permission rules can match tool input parameters (2.1.178) Rules now match against a tool's **input parameters**, not just its name, with `*` wildcard support. Syntax: `Tool(param:value)`. The headline use is gating subagent spawns by model — `Agent(model:opus)` in a deny rule blocks any subagent attempted with Opus; `Agent(model:*)` denies all model-pinned subagent spawns. ```jsonc { "permissions": { "deny": ["Agent(model:opus)", "Bash(rm:*)"] } } ``` - **Use it for:** a *hard* block on expensive or capability-risky subagent models, or parameter-level gates (e.g. specific `Bash` subcommands) that tool-name rules can't express. Combine with tool-name + file-path rules for defense-in-depth. - **It is static** — a permission rule cannot run logic, read state, output advisory context, or prompt the user. For *advisory* model control (warn-don't-block, cost estimates, consent prompts) ork keeps its `model-cost-advisor` and `fable-spend-consent` hooks; the two are orthogonal (see `shared/rules/cc-native-first.md`). ## Recommended baseline posture ```jsonc // .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 # 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 ```python # 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 ```python 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 ```python 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 ```python # 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 ```python # ❌ 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 ```python 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 # Langfuse Mask Callback Pre-trace PII masking using Langfuse's mask callback for automatic redaction before data reaches the server. ## Basic Setup ```python 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 ```python 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](https://langfuse.com/docs/sdk/python#mask-callback) - [Langfuse Privacy](https://langfuse.com/docs/data-security-privacy) ### Ork Delta # Ork Delta: security-patterns OrchestKit-specific knowledge rescued when this skill's vendor-restatement files were removed (wrap + delta campaign, 2026-07-31). Each entry is a floor, scar, or house decision that upstream documentation will never carry. For everything else, see the "Upstream coverage (do not restate)" table in SKILL.md. ## Hash passwords with argon2-cffi, never passlib Why: House correction shipped in this skill's OWASP example files (v2.0.0, February 2026): passlib last released in 2020 and breaks on Python 3.13+ after PEP 594 removed the stdlib crypt module. The demos that showed passlib were rewritten to argon2-cffi; this line is the surviving record of that fix. Upstream: https://argon2-cffi.readthedocs.io/ and the OWASP Password Storage Cheat Sheet (https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html) ## Write Zod v4 API shapes, never v3 Why: security-patterns 2.0.0 (February 2026) standardized every example on Zod v4 after v3 shapes kept creeping back in from training-data bias. The four traps that fail at runtime on v4: `z.string().email()/.url()/.uuid()` are removed (use top-level `z.email()`, `z.url()`, `z.uuid()`); `ZodError.errors` is removed (use `.issues`); the `ZodIssueCode` enum is removed (issue codes live on `z.core`, pass string codes like `"custom"`); `z.setErrorMap`/`ZodErrorMap` are removed (pass the unified `error` param per schema, or `z.config(\{ customError \})` globally). Upstream: https://zod.dev/v4/changelog (context7: /colinhacks/zod) ## Keep identifiers out of LLM prompts; attribute deterministically after the call Why: OrchestKit house doctrine, carried through security-patterns since v1: `user_id`, `tenant_id`, and any UUID flow AROUND the model, never through it, and attribution comes from RequestContext plus pre-LLM source refs, never from model output. The forbidden pattern audit and builder implementations live in this skill at `references/prompt-audit.md` and `scripts/prompt_builder.py`. Upstream: OWASP LLM01 Prompt Injection (https://genai.owasp.org/llm-top-10/) ## Never bypass this repo's scanning gates with --no-verify Why: In OrchestKit itself the scanning rules are enforced, not advisory: `bin/git-hooks/pre-push:324` runs `tests/security/run-security-tests.sh` and `.github/workflows/ci.yml:162` runs gitleaks, so skipping the hook only relocates a secret or security failure to CI (repo CLAUDE.md house rule). Upstream: https://github.com/gitleaks/gitleaks and https://semgrep.dev/docs/ ### Output Guardrails # 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python # 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 ```python # ❌ 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 ```python 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 # 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 ```python 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 ```python 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 ```python 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 ```python # 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 ```python # 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 ```python # ❌ 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 ```python 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 # 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 ```python 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) ```python 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 ```python @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 ```python # 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 ```python # 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 ```python # ❌ 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 ```python 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 ``` ### Presidio Integration # Microsoft Presidio Integration Enterprise-grade PII detection and anonymization with Microsoft Presidio. ## Installation ```bash pip install presidio-analyzer presidio-anonymizer python -m spacy download en_core_web_lg ``` ## Basic Usage ```python from presidio_analyzer import AnalyzerEngine from presidio_anonymizer import AnonymizerEngine # Initialize engines (singleton recommended) analyzer = AnalyzerEngine() anonymizer = AnonymizerEngine() def detect_pii(text: str, language: str = "en") -> list: """Detect PII entities in text.""" return analyzer.analyze( text=text, language=language, entities=["PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER", "CREDIT_CARD", "US_SSN"] ) def anonymize_text(text: str, language: str = "en") -> str: """Detect and anonymize PII in text.""" results = analyzer.analyze(text=text, language=language) return anonymizer.anonymize(text=text, analyzer_results=results).text ``` ## Custom Operators ```python from presidio_anonymizer.entities import OperatorConfig operators = { "PERSON": OperatorConfig("replace", {"new_value": "[PERSON]"}), "CREDIT_CARD": OperatorConfig("mask", {"masking_char": "*", "chars_to_mask": 12}), "EMAIL_ADDRESS": OperatorConfig("hash", {"hash_type": "sha256"}), "US_SSN": OperatorConfig("redact"), } anonymized = anonymizer.anonymize(text=text, analyzer_results=results, operators=operators) ``` ## Custom Recognizers ```python from presidio_analyzer import Pattern, PatternRecognizer internal_id_recognizer = PatternRecognizer( supported_entity="INTERNAL_ID", patterns=[Pattern(name="internal_id", regex=r"ID-[A-Z]{2}-\d{6}", score=0.9)] ) analyzer.registry.add_recognizer(internal_id_recognizer) ``` ## References - [Presidio Documentation](https://microsoft.github.io/presidio/) - [Supported Entities](https://microsoft.github.io/presidio/supported_entities/) ### Prompt Audit # Prompt Audit ## Purpose Before any prompt is sent to an LLM, audit it for forbidden content: ``` ┌────────────────────────────────────────────────────────────┐ │ PROMPT AUDIT │ ├────────────────────────────────────────────────────────────┤ │ │ │ Prompt Template + Variables ──► Audit ──► Send to LLM │ │ │ │ │ ▼ │ │ ┌──────────────┐ │ │ │ FORBIDDEN │ │ │ │ PATTERNS │ │ │ ├──────────────┤ │ │ │ • user_id │ │ │ │ • tenant_id │ │ │ │ • UUIDs │ │ │ │ • API keys │ │ │ │ • Tokens │ │ │ │ • Secrets │ │ │ └──────────────┘ │ │ │ │ │ ┌───────────────┼───────────────┐ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ CLEAN │ │ WARNING │ │ BLOCK │ │ │ │ │ │ │ │ │ │ │ │ Proceed │ │ Log + │ │ Reject │ │ │ │ │ │ Proceed │ │ │ │ │ └──────────┘ └──────────┘ └──────────┘ │ │ │ └────────────────────────────────────────────────────────────┘ ``` ## OrchestKit Forbidden Patterns ### Critical (Block Immediately) | Pattern | Regex | Why Block | |---------|-------|-----------| | UUID | `[0-9a-f]\{8\}-[0-9a-f]\{4\}-...` | Hallucination, cross-tenant | | API Key | `api[_-]?key` | Secret exposure | | Token | `token\s*[:=]` | Auth exposure | | Password | `password\s*[:=]` | Credential exposure | | Secret | `secret\s*[:=]` | Generic secret | ### Warning (Log and Review) | Pattern | Regex | Why Warn | |---------|-------|----------| | user_id | `user[_-]?id` | Likely context leak | | tenant_id | `tenant[_-]?id` | Likely isolation leak | | analysis_id | `analysis[_-]?id` | Likely tracking leak | | document_id | `document[_-]?id` | Likely reference leak | | session_id | `session[_-]?id` | Likely auth leak | ## Implementation ### 1. Pattern Definitions ```python import re from enum import Enum from dataclasses import dataclass class AuditSeverity(Enum): CLEAN = "clean" WARNING = "warning" CRITICAL = "critical" @dataclass class AuditViolation: pattern: str severity: AuditSeverity match: str position: int # OrchestKit-specific patterns CRITICAL_PATTERNS = [ # UUIDs (r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', "UUID"), # Secrets (r'api[_-]?key\s*[:=]\s*["\']?\S+', "API_KEY"), (r'password\s*[:=]\s*["\']?\S+', "PASSWORD"), (r'secret\s*[:=]\s*["\']?\S+', "SECRET"), (r'token\s*[:=]\s*["\']?\S+', "TOKEN"), (r'bearer\s+\S+', "BEARER_TOKEN"), ] WARNING_PATTERNS = [ # OrchestKit identifiers (r'\buser[_-]?id\b', "USER_ID_FIELD"), (r'\btenant[_-]?id\b', "TENANT_ID_FIELD"), (r'\banalysis[_-]?id\b', "ANALYSIS_ID_FIELD"), (r'\bdocument[_-]?id\b', "DOCUMENT_ID_FIELD"), (r'\bartifact[_-]?id\b', "ARTIFACT_ID_FIELD"), (r'\bchunk[_-]?id\b', "CHUNK_ID_FIELD"), (r'\bsession[_-]?id\b', "SESSION_ID_FIELD"), (r'\btrace[_-]?id\b', "TRACE_ID_FIELD"), (r'\bworkflow[_-]?run[_-]?id\b', "WORKFLOW_ID_FIELD"), ] ``` ### 2. Audit Function ```python def audit_prompt(prompt: str) -> list[AuditViolation]: """ Audit prompt for forbidden patterns. Returns list of violations. """ violations = [] # Check critical patterns for pattern, name in CRITICAL_PATTERNS: for match in re.finditer(pattern, prompt, re.IGNORECASE): violations.append(AuditViolation( pattern=name, severity=AuditSeverity.CRITICAL, match=match.group()[:50], # Truncate for logging position=match.start(), )) # Check warning patterns for pattern, name in WARNING_PATTERNS: for match in re.finditer(pattern, prompt, re.IGNORECASE): violations.append(AuditViolation( pattern=name, severity=AuditSeverity.WARNING, match=match.group(), position=match.start(), )) return violations def has_critical_violations(violations: list[AuditViolation]) -> bool: """Check if any violations are critical""" return any(v.severity == AuditSeverity.CRITICAL for v in violations) ``` ### 3. Audit Decorator ```python from functools import wraps import structlog logger = structlog.get_logger() def audit_before_llm(func): """ Decorator that audits prompts before LLM call. Blocks on critical violations, logs warnings. """ @wraps(func) async def wrapper(*args, **kwargs): # Extract prompt from args/kwargs prompt = kwargs.get("prompt") or args[0] # Audit violations = audit_prompt(prompt) # Log warnings for v in violations: if v.severity == AuditSeverity.WARNING: logger.warning( "prompt_audit_warning", pattern=v.pattern, position=v.position, ) # Block on critical if has_critical_violations(violations): critical = [v for v in violations if v.severity == AuditSeverity.CRITICAL] raise PromptSecurityError( f"Prompt contains forbidden content: {[v.pattern for v in critical]}" ) # Proceed return await func(*args, **kwargs) return wrapper # Usage @audit_before_llm async def call_llm(prompt: str) -> str: return await llm.generate(prompt) ``` ### 4. Safe Prompt Builder ```python class SafePromptBuilder: """ Builds prompts with built-in audit. Prevents accidental ID inclusion. """ def __init__(self): self._parts: list[str] = [] self._context_ids: dict[str, Any] = {} # Stored but never in prompt def add_instruction(self, text: str) -> "SafePromptBuilder": """Add instruction text (audited)""" violations = audit_prompt(text) if has_critical_violations(violations): raise PromptSecurityError("Instruction contains forbidden content") self._parts.append(text) return self def add_content(self, content: str) -> "SafePromptBuilder": """Add user content (sanitized)""" # Strip any IDs from content clean_content = self._sanitize(content) self._parts.append(clean_content) return self def add_context_texts(self, texts: list[str]) -> "SafePromptBuilder": """Add context texts (sanitized)""" for text in texts: clean = self._sanitize(text) self._parts.append(f"- {clean}") return self def store_context_id(self, key: str, value: Any) -> "SafePromptBuilder": """Store ID for post-LLM attribution (never in prompt)""" self._context_ids[key] = value return self def _sanitize(self, text: str) -> str: """Remove any IDs from text""" # 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}', '[ID]', text, flags=re.IGNORECASE ) return text def build(self) -> tuple[str, dict]: """ Build prompt and return with stored context. Returns: (prompt, context_ids) """ prompt = "\n\n".join(self._parts) # Final audit violations = audit_prompt(prompt) if violations: logger.warning( "prompt_audit_final", violation_count=len(violations), ) if has_critical_violations(violations): raise PromptSecurityError("Built prompt contains forbidden content") return prompt, self._context_ids # Usage builder = SafePromptBuilder() prompt, context = ( builder .add_instruction("Analyze the following content:") .add_content(user_query) .add_context_texts(retrieved_docs) .store_context_id("user_id", ctx.user_id) # Stored, not in prompt .store_context_id("sources", source_refs) # Stored, not in prompt .build() ) ``` ## OrchestKit Integration ### Workflow Integration ```python # backend/app/workflows/agents/prompts/content_analysis.py from llm_safety import SafePromptBuilder def build_analysis_prompt( query: str, context_texts: list[str], ctx: RequestContext, ) -> tuple[str, dict]: """ Build content analysis prompt safely. Context IDs stored separately for attribution. """ return ( SafePromptBuilder() .add_instruction(""" You are an expert content analyzer. Analyze the following content and provide insights about: 1. Key concepts 2. Difficulty level 3. Prerequisites 4. Summary """) .add_instruction(f"User query: {query}") .add_instruction("Relevant context:") .add_context_texts(context_texts) .store_context_id("user_id", ctx.user_id) .store_context_id("tenant_id", ctx.tenant_id) .store_context_id("trace_id", ctx.trace_id) .build() ) ``` ### CI/CD Integration ```bash #!/bin/bash # scripts/audit_prompts.sh echo "Auditing prompt templates..." # Check for IDs in prompt files grep -rn \ "user_id\|tenant_id\|analysis_id\|document_id\|[0-9a-f]\{8\}-[0-9a-f]\{4\}" \ backend/app/**/prompts/ \ --include="*.py" \ --include="*.txt" \ --include="*.jinja2" if [ $? -eq 0 ]; then echo "❌ Found potential ID leaks in prompts!" exit 1 fi echo "✅ Prompt audit passed" ``` ## Testing ```python class TestPromptAudit: def test_detects_uuid(self): prompt = "Analyze doc 123e4567-e89b-12d3-a456-426614174000" violations = audit_prompt(prompt) assert len(violations) == 1 assert violations[0].severity == AuditSeverity.CRITICAL assert violations[0].pattern == "UUID" def test_detects_api_key(self): prompt = "Use api_key: sk-1234567890abcdef" violations = audit_prompt(prompt) assert any(v.pattern == "API_KEY" for v in violations) def test_warns_on_user_id_field(self): prompt = "For user_id please provide analysis" violations = audit_prompt(prompt) assert len(violations) == 1 assert violations[0].severity == AuditSeverity.WARNING def test_safe_builder_blocks_id(self): with pytest.raises(PromptSecurityError): ( SafePromptBuilder() .add_instruction("Analyze for user 123e4567-e89b-12d3-a456-426614174000") .build() ) def test_safe_builder_sanitizes_content(self): prompt, _ = ( SafePromptBuilder() .add_content("Doc ID: 123e4567-e89b-12d3-a456-426614174000") .build() ) assert "123e4567" not in prompt assert "[ID]" in prompt def test_context_ids_not_in_prompt(self): from uuid import uuid4 user_id = uuid4() prompt, context = ( SafePromptBuilder() .add_instruction("Analyze this") .store_context_id("user_id", user_id) .build() ) assert str(user_id) not in prompt assert context["user_id"] == user_id ``` ### Request Context Pattern # Request Context Pattern ## Purpose The RequestContext is an immutable object created at the gateway that carries identity and tracing information through the entire request lifecycle. It flows AROUND the LLM (never in prompts) and is used for: 1. **Authorization** - Who is making the request 2. **Data Filtering** - Scope queries to tenant/user 3. **Attribution** - Tag results with proper ownership 4. **Observability** - Correlate logs and traces ## Implementation ```python from dataclasses import dataclass from datetime import datetime, timezone from uuid import UUID from typing import FrozenSet @dataclass(frozen=True) # Immutable! class RequestContext: """ System context that NEVER appears in LLM prompts. Created at gateway, flows through all layers. """ # === Identity (WHO) === user_id: UUID tenant_id: UUID # For B2B multi-tenant session_id: str permissions: FrozenSet[str] # === Tracing (OBSERVABILITY) === request_id: str # Unique per request trace_id: str # Distributed tracing span_id: str # Current span # === Resource (WHAT) === resource_id: UUID | None = None # analysis_id, document_id, etc. resource_type: str | None = None # === Metadata (WHEN, WHERE) === timestamp: datetime = None client_ip: str = "" user_agent: str = "" def __post_init__(self): if self.timestamp is None: object.__setattr__(self, 'timestamp', datetime.now(timezone.utc)) ``` ## Creation at Gateway ```python from fastapi import Request, Depends import jwt # PyJWT async def get_request_context(request: Request) -> RequestContext: """FastAPI dependency that creates RequestContext from JWT""" # 1. Extract and verify JWT auth_header = request.headers.get("Authorization", "") if not auth_header.startswith("Bearer "): raise HTTPException(401, "Missing authorization") token = auth_header[7:] try: claims = jwt.decode(token, SECRET_KEY, algorithms=["HS256"]) except jwt.InvalidTokenError: raise HTTPException(401, "Invalid token") # 2. Build immutable context return RequestContext( user_id=UUID(claims["sub"]), tenant_id=UUID(claims["tenant_id"]), session_id=claims["session_id"], permissions=frozenset(claims.get("permissions", [])), request_id=request.headers.get("X-Request-ID", str(uuid4())), trace_id=generate_trace_id(), span_id=generate_span_id(), client_ip=request.client.host, user_agent=request.headers.get("User-Agent", ""), ) ``` ## Usage in Endpoints ```python @router.post("/api/v1/analyze") async def create_analysis( request: AnalyzeRequest, ctx: RequestContext = Depends(get_request_context), ): # Context is available throughout the request # Pass it to services, repositories, etc. # Authorization uses context await authorize(ctx, "analysis:create", None) # Data access uses context for filtering documents = await repo.find_by_user(ctx) # LLM call does NOT receive context # (see llm-safety-patterns skill) # Attribution uses context result = await save_result(llm_output, ctx) return result ``` ## OrchestKit Parameters In OrchestKit, these identifiers should be in RequestContext: | Parameter | Type | Source | Purpose | |-----------|------|--------|---------| | `user_id` | UUID | JWT | Data ownership | | `tenant_id` | UUID | JWT | Multi-tenant isolation | | `session_id` | str | JWT | Session tracking | | `analysis_id` | UUID | Generated | Current analysis job | | `trace_id` | str | Generated | Langfuse tracing | | `request_id` | str | Header/Generated | Request correlation | ## Why Immutable? The context is frozen (`frozen=True`) to prevent: 1. **Accidental modification** - Can't change user_id mid-request 2. **Security bypass** - Can't escalate permissions 3. **Thread safety** - Safe to pass between async tasks 4. **Hashability** - Can be used as dict key for caching ## Anti-Patterns ```python # BAD: Mutable context class RequestContext: user_id: UUID # Can be changed! # BAD: Context in prompt prompt = f"User {ctx.user_id} wants to analyze..." # BAD: Context not passed to services result = await service.process(content) # Missing ctx! # BAD: Context created inside service def process(self): ctx = RequestContext(...) # Should come from gateway! ``` --- ## Checklists (5) ### Auth Checklist # 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 | Endpoint | Limit | |----------|-------| | Login | 5 per minute | | Password reset | 3 per hour | | MFA verify | 5 per minute | | Registration | 10 per hour | | API general | 100 per minute | ## 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 ```python 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 # 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 ```bash # 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 # 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 ```python 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 ### Safety Checklist # 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 ### Validation Checklist # 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 ```typescript import { z } from 'zod'; import { zodResolver } from '@hookform/resolvers/zod'; import DOMPurify from 'dompurify'; ``` ### Python ```python from pydantic import BaseModel, EmailStr, Field from markupsafe import escape ``` ## Common Patterns ### Allowlist (✅ Do) ```typescript const allowed = ['name', 'email', 'createdAt']; if (!allowed.includes(sortColumn)) throw new Error('Invalid'); ``` ### Blocklist (❌ Don't) ```typescript 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 | Vulnerability | Prevention | |--------------|------------| | SQL Injection | Parameterized queries | | XSS | HTML escaping, CSP | | Path Traversal | Validate/sanitize paths | | SSRF | URL allowlist | | ReDoS | Avoid complex regex | | Buffer Overflow | Length limits |