Security Guide¶

@pkg_router.register(
    PkgID.GET_AUTHORS,
    json_schema=GetAuthorsModel,
    roles=["viewer"]  # Read-only access
)
async def get_authors_handler(request: RequestModel) -> ResponseModel:
    ...

@pkg_router.register(
    PkgID.DELETE_AUTHOR,
    roles=["admin"]  # Admin only
)
async def delete_author_handler(request: RequestModel) -> ResponseModel:
    ...

from app.dependencies.permissions import require_roles

@router.get("/authors", dependencies=[Depends(require_roles("viewer"))])
async def get_authors():
    ...

@router.delete("/authors/{id}", dependencies=[Depends(require_roles("admin"))])
async def delete_author(id: int):
    ...

@router.get("/health")  # No require_roles = public endpoint
async def health_check():
    ...

# app/auth.py - Already implemented
class AuthBackend(AuthenticationBackend):
    async def authenticate(self, conn):
        # 1. Extract token from header/query
        # 2. Decode and verify signature
        # 3. Check expiration
        # 4. Verify audience
        # 5. Validate issuer
        # 6. Check not-before (nbf) claim
        # 7. Verify token wasn't revoked

# Implement token blacklist in Redis
from app.storage.redis import RRedis

async def revoke_token(token_jti: str, exp: int):
    """Revoke a token by adding to blacklist."""
    redis = RRedis()
    ttl = exp - int(time.time())
    await redis.setex(f"revoked_token:{token_jti}", ttl, "1")

async def is_token_revoked(token_jti: str) -> bool:
    """Check if token is revoked."""
    redis = RRedis()
    return await redis.exists(f"revoked_token:{token_jti}")

# ✅ SAFE: Parameterized query
stmt = select(Author).where(Author.name == user_input)
result = await session.execute(stmt)

# ❌ NEVER DO THIS: String concatenation
query = f"SELECT * FROM authors WHERE name = '{user_input}'"

from pydantic import BaseModel, validator

class AuthorCreate(BaseModel):
    name: str
    bio: str | None = None

    @validator('name')
    def validate_name(cls, v):
        if not v or len(v) > 100:
            raise ValueError('Name must be 1-100 characters')
        if not v.replace(' ', '').isalnum():
            raise ValueError('Name must be alphanumeric')
        return v

# app/storage/redis.py
redis_client = Redis(
    host=settings.REDIS_IP,
    port=settings.REDIS_PORT,
    password=settings.REDIS_PASSWORD,
    ssl=True,  # Enable SSL in production
    ssl_cert_reqs='required',
    ssl_ca_certs='/path/to/ca.crt',
    socket_connect_timeout=5,
    socket_timeout=5,
    decode_responses=True,
)

import boto3
from botocore.exceptions import ClientError

def get_secret(secret_name):
    """Retrieve secret from AWS Secrets Manager."""
    client = boto3.client('secretsmanager', region_name='us-east-1')

    try:
        response = client.get_secret_value(SecretId=secret_name)
        return response['SecretString']
    except ClientError as e:
        raise Exception(f"Error retrieving secret: {e}")

# Usage
db_password = get_secret('prod/db/password')

import hvac

client = hvac.Client(url='https://vault.example.com:8200')
client.auth.approle.login(
    role_id='app-role-id',
    secret_id='app-secret-id',
)

# Read secret
secret = client.secrets.kv.v2.read_secret_version(
    path='production/database',
)
db_password = secret['data']['data']['password']

from pydantic import BaseModel, Field, validator
from typing import Literal

class CreateAuthorInput(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    bio: str | None = Field(None, max_length=1000)
    status: Literal['active', 'inactive'] = 'active'

    @validator('name')
    def sanitize_name(cls, v):
        # Remove potentially dangerous characters
        import re
        v = re.sub(r'[<>\"\'&]', '', v)
        return v.strip()

from html import escape

# Escape user-generated content before rendering
safe_bio = escape(author.bio)

# app/middlewares/security_headers.py
from starlette.middleware.base import BaseHTTPMiddleware

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)

        # XSS Protection
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"

        # HTTPS Enforcement
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"

        # CSP
        response.headers["Content-Security-Policy"] = (
            "default-src 'self'; "
            "script-src 'self' 'unsafe-inline' https://cdn.example.com; "
            "style-src 'self' 'unsafe-inline'; "
            "img-src 'self' data: https:; "
            "font-src 'self' data:; "
            "connect-src 'self' wss://api.example.com; "
            "frame-ancestors 'none'"
        )

        # Permissions Policy
        response.headers["Permissions-Policy"] = (
            "geolocation=(), "
            "microphone=(), "
            "camera=()"
        )

        return response

# app/__init__.py
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://app.example.com",
        "https://admin.example.com"
    ],  # Specific origins only
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
    max_age=600,  # Cache preflight requests for 10 minutes
)

# Already implemented in app/api/ws/websocket.py
class PackageAuthWebSocketEndpoint(WebSocketEndpoint):
    async def on_connect(self, websocket: WebSocket):
        # 1. Extract token from query params
        # 2. Validate token with Keycloak
        # 3. Check rate limits
        # 4. Accept or reject connection

# Already implemented in app/routing.py
class PackageRouter:
    async def handle_request(self, request: RequestModel, user: UserModel):
        # 1. Validate JSON schema
        # 2. Check RBAC permissions
        # 3. Rate limit messages
        # 4. Sanitize input

@router.get("/users/{user_id}/data")
async def get_user_data(user_id: str):
    """Export all user data (GDPR compliance)."""
    # Return all data associated with user

@router.delete("/users/{user_id}/gdpr-delete")
async def gdpr_delete_user(user_id: str):
    """Permanently delete all user data."""
    # Delete from all tables
    # Anonymize audit logs