"""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()