Capstone: Secure Multi-Tenant Application

# tenant_middleware.py, JWT + RBAC enforcement
import jwt
from functools import wraps
from dataclasses import dataclass

@dataclass
class TokenClaims:
    user_id: str
    tenant_id: str
    roles: list[str]

def decode_token(token: str) -> TokenClaims | None:
    """Verify and decode access token."""
    try:
        secret = settings.jwt_secret.get_secret_value()
        payload = jwt.decode(
            token, secret,
            algorithms=[settings.jwt_algorithm],
            options={"require": ["exp", "sub", "tenant_id"]},
        )
        if payload.get("type") != "access":
            return None
        return TokenClaims(
            user_id=payload["sub"],
            tenant_id=payload["tenant_id"],
            roles=payload.get("roles", []),
        )
    except jwt.InvalidTokenError:
        return None

def require_auth(required_role: str | None = None):
    """Decorator: validate JWT + optionally check role."""
    def decorator(func):
        @wraps(func)
        def wrapper(bearer_token: str, *args, **kwargs):
            if not bearer_token.startswith("Bearer "):
                raise PermissionError("Missing or malformed Authorization header")
            token = bearer_token[7:]
            claims = decode_token(token)
            if not claims:
                raise PermissionError("Invalid or expired token")
            if required_role and required_role not in claims.roles:
                raise PermissionError(f"Role '{required_role}' required")
            # Inject claims into function
            return func(claims, *args, **kwargs)
        return wrapper
    return decorator

@require_auth()
def get_my_profile(claims: TokenClaims) -> dict:
    return {"user_id": claims.user_id, "tenant": claims.tenant_id, "roles": claims.roles}

@require_auth(required_role="admin")
def delete_user(claims: TokenClaims, target_user_id: str) -> str:
    return f"Admin {claims.user_id} deleted user {target_user_id}"

# Test
auth_header = f"Bearer {tokens['access_token']}"
profile = get_my_profile(auth_header)
print(f"Profile: {profile}")

try:
    delete_user(auth_header, "user-99")
except PermissionError as e:
    print(f"Access denied: {e}")   # alice is 'member', not 'admin'

# secure_api.py, FastAPI with security headers + input validation
from fastapi import FastAPI, Request, Response, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field, field_validator
import re

app = FastAPI(title="Secure Multi-Tenant API", docs_url=None)   # Disable docs in production
security = HTTPBearer()

# ── Security headers middleware ───────────────────────────────
@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response: Response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    return response

# ── Input validation with Pydantic ───────────────────────────
class CreateRecordRequest(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    content: str = Field(..., min_length=1, max_length=10000)
    tags: list[str] = Field(default=[], max_length=10)

    @field_validator("title", "content")
    @classmethod
    def no_null_bytes(cls, v: str) -> str:
        if "\x00" in v:
            raise ValueError("Null bytes not allowed")
        return v.strip()

    @field_validator("tags")
    @classmethod
    def validate_tags(cls, tags: list[str]) -> list[str]:
        pattern = re.compile(r"^[a-zA-Z0-9-_]+$")
        for tag in tags:
            if not pattern.match(tag):
                raise ValueError(f"Invalid tag: {tag!r}. Only alphanumeric, hyphen, underscore allowed.")
        return tags

# ── Protected endpoint ────────────────────────────────────────
@app.post("/api/records")
async def create_record(
    body: CreateRecordRequest,
    credentials: HTTPAuthorizationCredentials = Depends(security),
):
    claims = decode_token(credentials.credentials)
    if not claims:
        raise HTTPException(status_code=401, detail="Invalid token")

    # Encrypt record with tenant's key
    record_data = f"{body.title}:{body.content}".encode()
    encrypted = encrypt_record(claims.tenant_id, record_data)

    return {
        "record_id": secrets.token_hex(16),
        "tenant_id": claims.tenant_id,
        "encrypted_size": len(encrypted),
        "status": "created",
    }

print("FastAPI app with:")
print("  ✅ Security headers middleware")
print("  ✅ JWT authentication (Depends)")
print("  ✅ Pydantic input validation (injection prevention)")
print("  ✅ Per-tenant AES-GCM encryption")
print("  ✅ Tenant isolation via claims")