Phillip Tarrant
9a8506970c
Head Hen CMS end-to-end: dashboard lists all posts (drafts + published), Markdown editor with live preview + drag-drop image upload, Pillow media pipeline re-encoding every upload to JPEG, post CRUD + publish toggle + hard delete, About page edit, and double-submit CSRF cookie enforced on every admin mutating endpoint (Phase 3's TODO markers resolved). Slug auto-generated on create and server-locked once a post has been published. Unpublish preserves `published_at` so re-publish keeps original date ordering. Every admin write invalidates the read-side Post/Page TTL caches and records an `auth_events` audit row. CSRF middleware is narrow by design — issues/refreshes the `cb_csrf` cookie only on `GET /admin*`, and mutating endpoints opt in via `require_csrf_form` or `require_csrf_header` Depends. Public routes, healthz, and pre-auth login stay untouched. 64 new tests cover slugs, CSRF, media, admin posts/pages services, and end-to-end CMS routes. Tests never mock the DB — real temp SQLite files per the CLAUDE.md mandate. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
350 lines
14 KiB
Python
350 lines
14 KiB
Python
"""FastAPI application factory and module-level app instance.
|
|
|
|
The factory pattern (``create_app``) keeps test setup straightforward and
|
|
lets us swap in alternate configurations without module-level side
|
|
effects. ``app = create_app()`` at import time is what Uvicorn references
|
|
via ``app.main:app``.
|
|
|
|
Phase 2 additions:
|
|
- Build a shared SQLAlchemy :class:`~sqlalchemy.Engine` from
|
|
``settings.database_url`` and attach the per-connection
|
|
PRAGMA listener (WAL + foreign keys).
|
|
- Apply SQL migrations from :mod:`app.models.migrations`.
|
|
- Run the idempotent seed (welcome post, About page, system user).
|
|
- Instantiate :class:`PostService` and :class:`PageService` and
|
|
expose them on ``app.state`` for route-level DI.
|
|
|
|
Phase 3 additions:
|
|
- Build an ``itsdangerous.URLSafeTimedSerializer`` on
|
|
``settings.secret_key`` and attach to ``app.state``.
|
|
- Instantiate :class:`AuditService`, :class:`EmailService`,
|
|
:class:`SessionService`, :class:`AuthService` and attach them to
|
|
``app.state``.
|
|
- Create a SlowAPI :class:`Limiter` and register the
|
|
``RateLimitExceeded`` exception handler (renders
|
|
``admin/rate_limited.html`` + HTTP 429 + audit row).
|
|
- Include the admin router.
|
|
|
|
Phase 4 additions:
|
|
- Build a separate :class:`itsdangerous.URLSafeTimedSerializer`
|
|
salted with ``"csrf"`` and wrap it in a :class:`CSRFService`.
|
|
- Instantiate :class:`MarkdownService`,
|
|
:class:`AdminPostsService`, :class:`AdminPagesService`, and
|
|
:class:`MediaService` and attach them to ``app.state``.
|
|
- Mount the admin CMS router.
|
|
- Install a lightweight middleware that issues / refreshes the
|
|
CSRF cookie on admin GET requests and exposes the token via
|
|
``request.state.csrf_token`` for template rendering.
|
|
- Mount ``settings.media_root`` at ``/media`` as a StaticFiles
|
|
route so uploaded images are publicly reachable under the
|
|
Markdown URLs the admin inserts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import structlog
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from itsdangerous import URLSafeTimedSerializer
|
|
from slowapi.errors import RateLimitExceeded
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
from app import __version__
|
|
from app.config import get_settings
|
|
from app.db import build_engine, run_migrations
|
|
from app.logging_config import configure_logging
|
|
from app.models.seed import run_seed
|
|
from app.routes.admin import router as admin_router
|
|
from app.routes.admin_cms import router as admin_cms_router
|
|
from app.routes.health import router as health_router
|
|
from app.routes.public import router as public_router
|
|
from app.services.admin_pages import AdminPagesService
|
|
from app.services.admin_posts import AdminPostsService
|
|
from app.services.audit import AuditService
|
|
from app.services.auth import AuthService
|
|
from app.services.csrf import CSRF_COOKIE_NAME, CSRFService
|
|
from app.services.email import EmailService
|
|
from app.services.markdown import MarkdownService
|
|
from app.services.media import MediaService
|
|
from app.services.pages import PageService
|
|
from app.services.posts import PostService
|
|
from app.services.rate_limit import create_limiter
|
|
from app.services.sessions import SessionService
|
|
|
|
|
|
# Resolve the package root once so template / static paths stay correct
|
|
# regardless of the current working directory at startup (running under
|
|
# uvicorn from the repo root vs. pytest from anywhere vs. inside Docker).
|
|
_PACKAGE_ROOT: Path = Path(__file__).resolve().parent
|
|
_TEMPLATES_DIR: Path = _PACKAGE_ROOT / "templates"
|
|
_STATIC_DIR: Path = _PACKAGE_ROOT / "static"
|
|
|
|
|
|
class CSRFCookieMiddleware(BaseHTTPMiddleware):
|
|
"""Issue / refresh the CSRF cookie on admin GET responses.
|
|
|
|
The middleware is intentionally narrow: it only fires for requests
|
|
whose path starts with ``/admin`` AND method is GET. That's the
|
|
set of responses that render HTML with a form waiting to be
|
|
submitted; mutating endpoints never set the cookie themselves.
|
|
|
|
For every such request we:
|
|
|
|
1. Read the existing ``cb_csrf`` cookie (if any).
|
|
2. Call :meth:`CSRFService.issue` which reuses the underlying
|
|
nonce when the cookie is still valid — this means a tab that
|
|
GETs the dashboard, then POSTs 30 minutes later, still
|
|
matches.
|
|
3. Stash the token on ``request.state.csrf_token`` so route
|
|
handlers pass it into Jinja.
|
|
4. After the downstream response is produced, set / refresh the
|
|
cookie header.
|
|
"""
|
|
|
|
def __init__(self, app, csrf_service: CSRFService) -> None:
|
|
"""Store the service by reference; BaseHTTPMiddleware takes the ASGI app."""
|
|
super().__init__(app)
|
|
self._csrf_service: CSRFService = csrf_service
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
"""Run the admin-GET issue hook around the downstream handler."""
|
|
should_issue = (
|
|
request.url.path.startswith("/admin")
|
|
and request.method == "GET"
|
|
)
|
|
if should_issue:
|
|
existing = request.cookies.get(CSRF_COOKIE_NAME)
|
|
token, cookie_value = self._csrf_service.issue(existing)
|
|
request.state.csrf_token = token
|
|
else:
|
|
cookie_value = None
|
|
|
|
response: Response = await call_next(request)
|
|
|
|
if should_issue and cookie_value is not None:
|
|
response.set_cookie(
|
|
value=cookie_value,
|
|
**self._csrf_service.cookie_params(),
|
|
)
|
|
return response
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
"""Build and return the FastAPI application.
|
|
|
|
Responsibilities (in strict order):
|
|
|
|
1. Load validated configuration via :func:`get_settings`.
|
|
2. Initialize structured logging *before* any logger is used.
|
|
3. Build the SQLAlchemy engine and install the PRAGMA listener.
|
|
4. Apply SQL migrations (idempotent — no-op after first boot).
|
|
5. Run the seed (idempotent — marked via ``schema_migrations``).
|
|
6. Instantiate services and attach them to ``app.state`` so route
|
|
dependencies can resolve them via ``request.app.state``.
|
|
7. Mount static files, attach the shared :class:`Jinja2Templates`,
|
|
and register routers (including admin).
|
|
8. Wire the SlowAPI limiter and its exception handler.
|
|
9. Emit a single ``app_started`` structured log event.
|
|
"""
|
|
# Parse + validate configuration first so a bad environment fails fast
|
|
# with a clear pydantic error before we touch logging / FastAPI.
|
|
settings = get_settings()
|
|
|
|
# Configure structlog *before* acquiring any logger in the app so the
|
|
# very first log line already flows through our processor chain.
|
|
configure_logging(settings.app_env)
|
|
|
|
# --- Database plumbing --------------------------------------------------
|
|
# Engine is a process-wide resource. Built here so that migrations
|
|
# and seed both run on the same pool/config as the running app.
|
|
engine = build_engine(settings.database_url)
|
|
run_migrations(engine)
|
|
run_seed(engine)
|
|
|
|
# Ensure the media storage directory exists — Starlette's
|
|
# StaticFiles mount refuses to start without a real directory.
|
|
media_root = Path(settings.media_root)
|
|
media_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
application = FastAPI(
|
|
title="Chicken Babies R Us",
|
|
version=__version__,
|
|
# Docs stay on for Phase 0; a later phase can gate these behind
|
|
# admin auth or disable them entirely in production.
|
|
docs_url="/docs",
|
|
redoc_url="/redoc",
|
|
)
|
|
|
|
# Serve CSS, images, and other static assets from app/static. The
|
|
# `check_dir=False` would let Starlette skip the existence check; we
|
|
# leave it at its default so a missing directory surfaces loudly in
|
|
# dev. In prod the directory is baked into the container image.
|
|
application.mount(
|
|
"/static",
|
|
StaticFiles(directory=_STATIC_DIR),
|
|
name="static",
|
|
)
|
|
|
|
# Public mount for admin-uploaded images. Kept separate from /static
|
|
# so admin uploads never collide with the brand assets shipped with
|
|
# the container image.
|
|
application.mount(
|
|
"/media",
|
|
StaticFiles(directory=str(media_root)),
|
|
name="media",
|
|
)
|
|
|
|
# Single shared Jinja2 environment. Storing it on ``app.state`` keeps
|
|
# route modules free of an import dependency on this module (which
|
|
# would be circular once admin/auth routers are added in later
|
|
# phases). Route handlers pull it via a ``Depends(get_templates)``
|
|
# function defined next to the routes.
|
|
templates = Jinja2Templates(directory=_TEMPLATES_DIR)
|
|
application.state.templates = templates
|
|
|
|
# Store the engine + services on ``app.state`` so the
|
|
# dependency-injection helpers in :mod:`app.services.*` can find
|
|
# them without importing this module (circular-import-safe).
|
|
application.state.engine = engine
|
|
application.state.post_service = PostService(engine)
|
|
application.state.page_service = PageService(engine)
|
|
|
|
# --- Phase 3 wiring -----------------------------------------------------
|
|
# itsdangerous signer: signs (and later verifies) session-cookie
|
|
# values using SECRET_KEY and the salt "session". The same instance
|
|
# is shared by every request — cheap to construct, no state beyond
|
|
# the key.
|
|
signer = URLSafeTimedSerializer(settings.secret_key, salt="session")
|
|
application.state.signer = signer
|
|
|
|
# Audit first — EmailService and AuthService both depend on it
|
|
# (EmailService indirectly via the request-path contract: failures
|
|
# are logged, never raised).
|
|
audit_service = AuditService(engine)
|
|
email_service = EmailService(settings, templates)
|
|
session_service = SessionService(engine, signer, settings)
|
|
auth_service = AuthService(
|
|
engine, email_service, session_service, audit_service, settings
|
|
)
|
|
|
|
application.state.audit_service = audit_service
|
|
application.state.email_service = email_service
|
|
application.state.session_service = session_service
|
|
application.state.auth_service = auth_service
|
|
|
|
# --- Phase 4 wiring -----------------------------------------------------
|
|
# CSRF signer: separate salt so a session cookie never validates
|
|
# as a CSRF token (domain separation via salt).
|
|
csrf_signer = URLSafeTimedSerializer(settings.secret_key, salt="csrf")
|
|
csrf_service = CSRFService(
|
|
csrf_signer,
|
|
production=(settings.app_env == "production"),
|
|
)
|
|
application.state.csrf_service = csrf_service
|
|
|
|
markdown_service = MarkdownService()
|
|
application.state.markdown_service = markdown_service
|
|
|
|
admin_posts_service = AdminPostsService(
|
|
engine=engine,
|
|
markdown=markdown_service,
|
|
post_service=application.state.post_service,
|
|
page_service=application.state.page_service,
|
|
audit=audit_service,
|
|
)
|
|
admin_pages_service = AdminPagesService(
|
|
engine=engine,
|
|
markdown=markdown_service,
|
|
page_service=application.state.page_service,
|
|
post_service=application.state.post_service,
|
|
audit=audit_service,
|
|
)
|
|
media_service = MediaService(
|
|
engine=engine,
|
|
media_root=str(media_root),
|
|
audit=audit_service,
|
|
)
|
|
|
|
application.state.admin_posts_service = admin_posts_service
|
|
application.state.admin_pages_service = admin_pages_service
|
|
application.state.media_service = media_service
|
|
|
|
# CSRF cookie middleware — narrow to admin GETs; everything else
|
|
# passes through untouched so public routes are unaffected.
|
|
application.add_middleware(CSRFCookieMiddleware, csrf_service=csrf_service)
|
|
|
|
# SlowAPI limiter + exception handler. The limiter is a module-level
|
|
# singleton in app.services.rate_limit (because @limiter.limit has
|
|
# to be applied at endpoint-definition time, before include_router).
|
|
# We still attach it to app.state so SlowAPI's request-path
|
|
# middleware can reach it via request.app.state.limiter.
|
|
limiter = create_limiter()
|
|
application.state.limiter = limiter
|
|
|
|
application.add_exception_handler(
|
|
RateLimitExceeded, _make_rate_limit_handler(templates, audit_service)
|
|
)
|
|
|
|
# Register routers. Kept explicit (no dynamic discovery) so the set of
|
|
# mounted endpoints is trivially auditable.
|
|
application.include_router(health_router)
|
|
application.include_router(public_router)
|
|
application.include_router(admin_router)
|
|
application.include_router(admin_cms_router)
|
|
|
|
# Single structured startup event. Do NOT include secret material.
|
|
logger = structlog.get_logger(__name__)
|
|
logger.info(
|
|
"app_started",
|
|
app_env=settings.app_env,
|
|
version=__version__,
|
|
commit_sha=settings.git_commit_sha,
|
|
)
|
|
|
|
return application
|
|
|
|
|
|
def _make_rate_limit_handler(
|
|
templates: Jinja2Templates,
|
|
audit_service: AuditService,
|
|
):
|
|
"""Build the FastAPI exception handler for ``RateLimitExceeded``.
|
|
|
|
Renders ``admin/rate_limited.html`` at HTTP 429 and writes a
|
|
``rate_limited`` audit row scoped to the IP path. Email-scope
|
|
rate-limits are handled inside AuthService and don't come through
|
|
this handler.
|
|
"""
|
|
|
|
async def _handler(request: Request, exc: RateLimitExceeded):
|
|
# Best-effort endpoint path for the audit detail; the limiter
|
|
# doesn't surface a structured endpoint name so we use the URL
|
|
# path which is already stable / non-sensitive.
|
|
endpoint = request.url.path
|
|
ip = request.client.host if request.client else ""
|
|
ua = request.headers.get("user-agent", "")
|
|
audit_service.record(
|
|
"rate_limited",
|
|
ip=ip,
|
|
user_agent=ua,
|
|
detail={"scope": "ip", "endpoint": endpoint},
|
|
)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/rate_limited.html",
|
|
{},
|
|
status_code=429,
|
|
)
|
|
|
|
return _handler
|
|
|
|
|
|
# Module-level ASGI handle. Uvicorn / gunicorn import this as
|
|
# ``app.main:app``. Building it at import time is intentional: it fails
|
|
# loudly at container start if configuration is invalid.
|
|
app: FastAPI = create_app()
|