feat(admin): add CSRF protection to mutation endpoints (#79)

* feat(admin): add CSRF protection to mutation endpoints

Closes #79.

Admin POST/PUT/DELETE routes now require a server-validated CSRF token in addition to the session cookie. Token is per-session, minted lazily when the dashboard renders, rotated on OAuth login, and validated via a FastAPI dependency on each mutation route. Clients send it in the X-CSRF-Token header (HTMX listener reads the token from a <meta> tag) with a csrf_token form-field fallback. Dev-skip-auth mode bypasses CSRF to match auth bypass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* feat(admin): add GET /admin/csrf for JSON callers, use CSRF_SESSION_KEY constant

Address Copilot review on PR #83:

1. auth.py now uses SESSION_KEY constant from services.csrf instead of hardcoding 'csrf_token' — avoids key drift if the storage key ever changes.

2. New GET /admin/csrf endpoint returns the current token in JSON, matching the issue's acceptance criterion for non-HTML clients. The module docstring already promised JSON API support; this makes good on it.

Tests added for the new endpoint (idempotent within session) and the auth rotation behavior.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(auth): make CSRF dep resolve after auth so 401 fires before 403

Address Copilot review pass 2 on PR #83:

Route-level dependencies=[Depends(verify_csrf_token)] resolves before parameter deps (including AdminUser), so unauthenticated HTMX requests would get 403 (CSRF token missing) instead of the intended 401 + HX-Redirect. Fix by:

1. Move get_current_admin from routers/admin.py to dependencies.py so services/csrf.py can import it without a circular dep. Re-exported from admin.py so existing test patches keep working.

2. verify_csrf_token now depends on get_current_admin via Annotated[str, Depends(...)]. FastAPI resolves auth first; if it raises 401, CSRF never runs.

Also rewrote test_oauth_callback_rotates_csrf_token to actually exercise oauth_callback with mocked httpx calls (per Copilot feedback — the previous version only tested dict.pop). Added a structural test that locks in the new CSRF→auth dep ordering.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(deps): consolidate is_htmx + fix stale test patch targets

Address Copilot review pass 3 on PR #83:

1. Centralize is_htmx in dependencies.py and import it from routers/admin.py. Previously the prior commit duplicated the helper (as _is_htmx in dependencies.py alongside the existing is_htmx in admin.py).

2. Update test patches that target squishmark.routers.admin.get_settings to squishmark.dependencies.get_settings. After moving get_current_admin to dependencies.py the old patches were silently no-ops — tests passed only because the real settings happened to produce the same outcome for an empty session.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(csrf): self-review pass — integration tests, generic error, polish

Independent code review pass on PR #83. Six changes:

1. tests/test_csrf_integration.py (new): five TestClient-based tests that actually exercise the route-level CSRF dep and the auth-before-CSRF ordering — previously all four 'dependencies=[Depends(verify_csrf_token)]' lines were uncovered (existing tests call handler functions directly and bypass FastAPI's DI). Includes the crucial 'unauth HTMX → 401+HX-Redirect, not 403' assertion.

2. services/csrf.py: collapse the two distinct error messages ('CSRF token missing from session' vs 'CSRF token invalid') into one generic 'CSRF validation failed'. The prior split was an information disclosure — it told an attacker whether they had a session cookie to bind to.

3. services/csrf.py: document the assumption that Starlette caches request.form() so the dep + route handler can both read it without losing the body.

4. routers/admin.py: drop the __all__ re-export of get_current_admin / AdminUser. Test imports updated to point at squishmark.dependencies directly — there were no production callers.

5. routers/admin.py: GET /admin/csrf now returns a typed CSRFTokenResponse Pydantic model instead of a raw dict, matching the convention of NoteResponse and CacheRefreshResponse.

6. tests/test_admin_notes.py: oauth_callback rotation test now also asserts a fresh token can be minted after rotation, not just that the stale one is gone.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* feat(csrf): log rejection reason + rotation event for operability

Two structured logs added so production issues are diagnosable:

1. verify_csrf_token now emits WARNING on every rejection with reason (no-session-token / no-submitted-token / token-mismatch), method, path, admin, and htmx flag. The user-facing detail stays generic ('CSRF validation failed') to avoid info disclosure — the log is the only place the specific failure mode lives. Caplog test locks the format in.

2. oauth_callback emits INFO on CSRF token rotation, tying the rotation to the user that just logged in. Useful for audit trails and login-flow debugging.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: x3ek

src/squishmark/dependencies.py

@@ -1,11 +1,14 @@
 """FastAPI dependency injection utilities."""
 
+import logging
 from typing import Annotated
 
-from fastapi import Depends, Request
+from fastapi import Depends, HTTPException, Request
 
 from squishmark.config import Settings, get_settings
 
+logger = logging.getLogger(__name__)
+
 # Type alias for settings dependency
 SettingsDep = Annotated[Settings, Depends(get_settings)]
 
@@ -19,3 +22,42 @@ def is_admin(request: Request) -> bool:
     if user is None:
         return False
     return user.get("login") in settings.admin_users_list
+
+
+def is_htmx(request: Request) -> bool:
+    """Return True when the request was made by HTMX."""
+    return request.headers.get("HX-Request") == "true"
+
+
+async def get_current_admin(request: Request) -> str:
+    """
+    Get the current admin user from session.
+
+    Raises HTTPException 401 if not authenticated.
+    Raises HTTPException 403 if not an admin.
+
+    For HTMX requests, attaches an ``HX-Redirect`` header so the browser
+    is redirected to the login page without any client JavaScript.
+    """
+    settings = get_settings()
+
+    # Dev mode auth bypass (requires both flags)
+    if settings.debug and settings.dev_skip_auth:
+        logger.warning("Auth bypassed - returning dev-admin user")
+        return "dev-admin"
+
+    htmx_headers = {"HX-Redirect": "/auth/login"} if is_htmx(request) else None
+
+    # Check for user in session (set by OAuth callback)
+    user = request.session.get("user") if hasattr(request, "session") else None
+
+    if user is None:
+        raise HTTPException(status_code=401, detail="Not authenticated", headers=htmx_headers)
+
+    if user["login"] not in settings.admin_users_list:
+        raise HTTPException(status_code=403, detail="Not authorized", headers=htmx_headers)
+
+    return user["login"]
+
+
+AdminUser = Annotated[str, Depends(get_current_admin)]

src/squishmark/routers/admin.py

@@ -10,10 +10,12 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from squishmark.config import get_settings
+from squishmark.dependencies import AdminUser, is_htmx
 from squishmark.models.content import Config
 from squishmark.models.db import Note, get_db_session
 from squishmark.services.analytics import AnalyticsService
 from squishmark.services.cache import get_cache
+from squishmark.services.csrf import get_or_create_csrf_token, verify_csrf_token
 from squishmark.services.github import get_github_service
 from squishmark.services.notes import NotesService
 from squishmark.services.theme import get_theme_engine, reset_theme_engine
@@ -23,11 +25,6 @@
 router = APIRouter(prefix="/admin", tags=["admin"])
 
 
-def is_htmx(request: Request) -> bool:
-    """Return True when the request was made by HTMX."""
-    return request.headers.get("HX-Request") == "true"
-
-
 # Pydantic models for request/response
 class NoteCreate(BaseModel):
     """Request body for creating a note."""
@@ -65,39 +62,12 @@ class CacheRefreshResponse(BaseModel):
     duration_ms: float
 
 
-# Dependency for getting the current admin user
-async def get_current_admin(request: Request) -> str:
-    """
-    Get the current admin user from session.
-
-    Raises HTTPException 401 if not authenticated.
-    Raises HTTPException 403 if not an admin.
-
-    For HTMX requests, attaches an ``HX-Redirect`` header so the browser
-    is redirected to the login page without any client JavaScript.
-    """
-    settings = get_settings()
-
-    # Dev mode auth bypass (requires both flags)
-    if settings.debug and settings.dev_skip_auth:
-        logger.warning("Auth bypassed - returning dev-admin user")
-        return "dev-admin"
-
-    htmx_headers = {"HX-Redirect": "/auth/login"} if is_htmx(request) else None
-
-    # Check for user in session (set by OAuth callback)
-    user = request.session.get("user") if hasattr(request, "session") else None
-
-    if user is None:
-        raise HTTPException(status_code=401, detail="Not authenticated", headers=htmx_headers)
+class CSRFTokenResponse(BaseModel):
+    """Response body for the CSRF token endpoint."""
 
-    if user["login"] not in settings.admin_users_list:
-        raise HTTPException(status_code=403, detail="Not authorized", headers=htmx_headers)
+    csrf_token: str
 
-    return user["login"]
 
-
-AdminUser = Annotated[str, Depends(get_current_admin)]
 DbSession = Annotated[AsyncSession, Depends(get_db_session)]
 
 
@@ -214,6 +184,7 @@ async def admin_dashboard(
     cache = get_cache()
 
     # Render admin template
+    csrf_token = get_or_create_csrf_token(request)
     theme_engine = await get_theme_engine(github_service)
     try:
         html = await theme_engine.render_admin(
@@ -222,6 +193,7 @@ async def admin_dashboard(
             analytics=analytics,
             notes=[_to_note_response(n) for n in notes],
             cache_size=cache.size,
+            csrf_token=csrf_token,
         )
     except Exception:
         # Fallback if admin template doesn't exist
@@ -248,6 +220,14 @@ async def admin_dashboard(
     return HTMLResponse(content=html)
 
 
+# CSRF token endpoint (for JSON API callers that can't scrape the meta tag)
+@router.get("/csrf")
+async def get_csrf(request: Request, admin: AdminUser) -> CSRFTokenResponse:
+    """Return the current CSRF token for use in subsequent mutation requests."""
+    del admin  # auth side-effect only
+    return CSRFTokenResponse(csrf_token=get_or_create_csrf_token(request))
+
+
 # Analytics endpoints
 @router.get("/analytics")
 async def get_analytics(
@@ -273,7 +253,12 @@ async def list_notes(
     return [_to_note_response(n) for n in notes]
 
 
-@router.post("/notes", status_code=201, response_model=None)
+@router.post(
+    "/notes",
+    status_code=201,
+    response_model=None,
+    dependencies=[Depends(verify_csrf_token)],
+)
 async def create_note(
     request: Request,
     admin: AdminUser,
@@ -295,7 +280,11 @@ async def create_note(
     return response
 
 
-@router.put("/notes/{note_id}", response_model=None)
+@router.put(
+    "/notes/{note_id}",
+    response_model=None,
+    dependencies=[Depends(verify_csrf_token)],
+)
 async def update_note(
     request: Request,
     admin: AdminUser,
@@ -320,7 +309,11 @@ async def update_note(
     return response
 
 
-@router.delete("/notes/{note_id}", response_model=None)
+@router.delete(
+    "/notes/{note_id}",
+    response_model=None,
+    dependencies=[Depends(verify_csrf_token)],
+)
 async def delete_note(
     request: Request,
     admin: AdminUser,
@@ -371,7 +364,7 @@ async def view_note(
 
 
 # Cache management
-@router.post("/cache/refresh")
+@router.post("/cache/refresh", dependencies=[Depends(verify_csrf_token)])
 async def refresh_cache(
     admin: AdminUser,
 ) -> CacheRefreshResponse:

src/squishmark/routers/auth.py

@@ -1,12 +1,16 @@
 """Authentication routes for GitHub OAuth."""
 
+import logging
 from urllib.parse import urlencode
 
 import httpx
 from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import RedirectResponse
 
 from squishmark.config import get_settings
+from squishmark.services.csrf import SESSION_KEY as CSRF_SESSION_KEY
+
+logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
@@ -116,6 +120,9 @@ async def oauth_callback(
         "name": user_data.get("name"),
         "avatar_url": user_data.get("avatar_url"),
     }
+    # Rotate CSRF token on login so a stale pre-auth token can't be replayed.
+    request.session.pop(CSRF_SESSION_KEY, None)
+    logger.info("CSRF token rotated on OAuth login (user=%s)", user_data["login"])
 
     # Redirect to admin
     return RedirectResponse(url="/admin", status_code=302)

src/squishmark/services/csrf.py

@@ -0,0 +1,97 @@
+"""CSRF token generation and verification for admin mutation endpoints.
+
+Tokens are stored in the signed session cookie under ``SESSION_KEY`` and
+validated on POST/PUT/DELETE requests via the ``verify_csrf_token`` dependency.
+Clients send the token in the ``X-CSRF-Token`` header (HTMX, JSON API) or a
+``csrf_token`` form field (plain form fallback). JSON callers that cannot
+read the meta tag can fetch the current token from ``GET /admin/csrf``.
+"""
+
+import logging
+import secrets
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, Request
+
+from squishmark.config import get_settings
+from squishmark.dependencies import get_current_admin, is_htmx
+
+logger = logging.getLogger(__name__)
+
+SESSION_KEY = "csrf_token"
+HEADER_NAME = "X-CSRF-Token"
+FORM_FIELD = "csrf_token"
+
+
+def get_or_create_csrf_token(request: Request) -> str:
+    """Return the session's CSRF token, minting a new one if absent."""
+    token = request.session.get(SESSION_KEY)
+    if not token:
+        token = secrets.token_urlsafe(32)
+        request.session[SESSION_KEY] = token
+    return token
+
+
+async def _extract_submitted_token(request: Request) -> str | None:
+    """Read the submitted CSRF token from header or form body.
+
+    For form requests we call ``request.form()``; Starlette caches the parsed
+    form on the request instance, so the route handler's subsequent
+    ``request.form()`` call is served from cache and does not re-read the body.
+    If Starlette ever stops caching, both paths still parse independently — but
+    we'd duplicate work, not lose data.
+    """
+    header_token = request.headers.get(HEADER_NAME)
+    if header_token:
+        return header_token
+
+    content_type = request.headers.get("content-type", "")
+    if content_type.startswith(("application/x-www-form-urlencoded", "multipart/form-data")):
+        form = await request.form()
+        value = form.get(FORM_FIELD)
+        if isinstance(value, str):
+            return value
+    return None
+
+
+async def verify_csrf_token(
+    request: Request,
+    admin: Annotated[str, Depends(get_current_admin)],
+) -> None:
+    """FastAPI dependency that rejects requests missing or with an invalid CSRF token.
+
+    Depends on :func:`get_current_admin` so authentication resolves first. This
+    matters for unauthenticated HTMX requests: they get the 401 + ``HX-Redirect``
+    response from auth instead of a misleading 403 for a CSRF token they had no
+    way to obtain.
+
+    Skipped when ``debug`` and ``dev_skip_auth`` are both set, matching the
+    auth-bypass behavior in ``get_current_admin``.
+    """
+    settings = get_settings()
+    if settings.debug and settings.dev_skip_auth:
+        logger.warning("CSRF bypassed - dev_skip_auth is enabled")
+        return
+
+    # Single user-facing error regardless of which check fails — distinguishing
+    # "no session token" from "wrong submitted token" would tell an attacker
+    # whether they got a session to bind to. Server-side log records the
+    # specific reason so operators can diagnose without that side-channel.
+    expected = request.session.get(SESSION_KEY) if hasattr(request, "session") else None
+    submitted = await _extract_submitted_token(request)
+    if not expected or not submitted or not secrets.compare_digest(submitted, expected):
+        if not expected:
+            reason = "no-session-token"
+        elif not submitted:
+            reason = "no-submitted-token"
+        else:
+            reason = "token-mismatch"
+        logger.warning(
+            "CSRF rejected: reason=%s method=%s path=%s admin=%s htmx=%s",
+            reason,
+            request.method,
+            request.url.path,
+            admin,
+            is_htmx(request),
+        )
+        raise HTTPException(status_code=403, detail="CSRF validation failed")