feat(01-03): apply security middleware stack and database health check

- Add TrustedHostMiddleware for Host header validation
- Add CORSMiddleware with configurable origins
- Add rate limiting with RateLimitExceeded handler
- Add custom middleware for security headers (HSTS, X-Frame-Options, etc.)
- Add /health/db endpoint that checks database connectivity
- Mark health endpoints as rate limit exempt
- Fix linting issues in migration file (Rule 3 - Blocking)
This commit is contained in:
Mikkel Georgsen 2026-01-25 20:20:00 +00:00
parent 09f89617e7
commit 0d1a008d2f
3 changed files with 119 additions and 28 deletions

View file

@ -5,44 +5,64 @@ Revises:
Create Date: 2026-01-25 20:11:11.446731 Create Date: 2026-01-25 20:11:11.446731
""" """
from typing import Sequence, Union
from alembic import op from collections.abc import Sequence
import sqlalchemy as sa import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic. # revision identifiers, used by Alembic.
revision: str = 'de1460a760b0' revision: str = "de1460a760b0"
down_revision: Union[str, Sequence[str], None] = None down_revision: str | Sequence[str] | None = None
branch_labels: Union[str, Sequence[str], None] = None branch_labels: str | Sequence[str] | None = None
depends_on: Union[str, Sequence[str], None] = None depends_on: str | Sequence[str] | None = None
def upgrade() -> None: def upgrade() -> None:
"""Upgrade schema.""" """Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ### # ### commands auto generated by Alembic - please adjust! ###
op.create_table('builds', op.create_table(
sa.Column('id', sa.UUID(), nullable=False), "builds",
sa.Column('config_hash', sa.String(length=64), nullable=False), sa.Column("id", sa.UUID(), nullable=False),
sa.Column('status', sa.Enum('PENDING', 'BUILDING', 'COMPLETED', 'FAILED', 'CACHED', name='buildstatus'), nullable=False), sa.Column("config_hash", sa.String(length=64), nullable=False),
sa.Column('iso_path', sa.String(length=512), nullable=True), sa.Column(
sa.Column('error_message', sa.Text(), nullable=True), "status",
sa.Column('build_log', sa.Text(), nullable=True), sa.Enum(
sa.Column('started_at', sa.DateTime(timezone=True), nullable=True), "PENDING", "BUILDING", "COMPLETED", "FAILED", "CACHED",
sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True), name="buildstatus",
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), ),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), nullable=False,
sa.PrimaryKeyConstraint('id') ),
sa.Column("iso_path", sa.String(length=512), nullable=True),
sa.Column("error_message", sa.Text(), nullable=True),
sa.Column("build_log", sa.Text(), nullable=True),
sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.text("now()"),
nullable=False,
),
sa.Column(
"updated_at",
sa.DateTime(timezone=True),
server_default=sa.text("now()"),
nullable=False,
),
sa.PrimaryKeyConstraint("id"),
) )
op.create_index(op.f('ix_builds_config_hash'), 'builds', ['config_hash'], unique=True) op.create_index(
op.create_index('ix_builds_status', 'builds', ['status'], unique=False) op.f("ix_builds_config_hash"), "builds", ["config_hash"], unique=True
)
op.create_index("ix_builds_status", "builds", ["status"], unique=False)
# ### end Alembic commands ### # ### end Alembic commands ###
def downgrade() -> None: def downgrade() -> None:
"""Downgrade schema.""" """Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ### # ### commands auto generated by Alembic - please adjust! ###
op.drop_index('ix_builds_status', table_name='builds') op.drop_index("ix_builds_status", table_name="builds")
op.drop_index(op.f('ix_builds_config_hash'), table_name='builds') op.drop_index(op.f("ix_builds_config_hash"), table_name="builds")
op.drop_table('builds') op.drop_table("builds")
# ### end Alembic commands ### # ### end Alembic commands ###

View file

@ -1,17 +1,44 @@
"""Health check endpoints.""" """Health check endpoints."""
from fastapi import APIRouter from typing import Any
from fastapi import APIRouter, Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.api.deps import get_db
from backend.app.core.security import limiter
router = APIRouter() router = APIRouter()
@router.get("") @router.get("")
@limiter.exempt
async def health_check() -> dict[str, str]: async def health_check() -> dict[str, str]:
"""Basic health check endpoint.""" """Basic health check endpoint."""
return {"status": "healthy"} return {"status": "healthy"}
@router.get("/ready") @router.get("/ready")
@limiter.exempt
async def readiness_check() -> dict[str, str]: async def readiness_check() -> dict[str, str]:
"""Readiness check endpoint (DB check added in plan 01-02).""" """Readiness check endpoint."""
return {"status": "ready"} return {"status": "ready"}
@router.get("/db")
@limiter.exempt
async def database_health_check(
db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
"""Health check that verifies database connectivity.
Returns:
Status indicating healthy/unhealthy and database connection state.
"""
try:
result = await db.execute(text("SELECT 1"))
result.scalar()
return {"status": "healthy", "database": "connected"}
except Exception as e:
return {"status": "unhealthy", "database": "error", "detail": str(e)}

View file

@ -1,9 +1,14 @@
"""FastAPI application entry point.""" """FastAPI application entry point."""
from fastapi import FastAPI from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from slowapi import _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from backend.app.api.v1.router import api_router from backend.app.api.v1.router import api_router
from backend.app.core.config import settings from backend.app.core.config import settings
from backend.app.core.security import limiter
app = FastAPI( app = FastAPI(
title="Debate API", title="Debate API",
@ -14,6 +19,45 @@ app = FastAPI(
debug=settings.debug, debug=settings.debug,
) )
# Middleware order matters - first added = outermost layer
# See: 01-RESEARCH.md Pattern 3: FastAPI Security Middleware Stack
# 1. Trusted Host (reject requests with invalid Host header)
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=settings.allowed_hosts_list,
)
# 2. CORS (handle cross-origin requests)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins_list,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
allow_headers=["*"],
max_age=600, # Cache preflight requests for 10 minutes
)
# 3. Rate limiting
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
# 4. Custom middleware for security headers
@app.middleware("http")
async def security_headers_middleware(request: Request, call_next: object) -> Response:
"""Add security headers to all responses."""
response: Response = await call_next(request) # type: ignore[misc]
response.headers["Strict-Transport-Security"] = (
"max-age=31536000; includeSubDomains"
)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
return response
# Include API routers # Include API routers
app.include_router(api_router, prefix="/api/v1") app.include_router(api_router, prefix="/api/v1")