From 63a999768dd85f4b7493aaf84c1ca0bd165468a6 Mon Sep 17 00:00:00 2001
From: Xeek <6032840+x3ek@users.noreply.github.com>
Date: Sat, 23 May 2026 06:52:57 -0500
Subject: [PATCH 1/6] 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>
---
 src/squishmark/routers/admin.py    |  24 ++++-
 src/squishmark/routers/auth.py     |   2 +
 src/squishmark/services/csrf.py    |  64 ++++++++++++
 tests/test_csrf.py                 | 156 +++++++++++++++++++++++++++++
 themes/blue-tech/admin/admin.html  |   7 ++
 themes/default/admin/admin.html    |   7 ++
 themes/terminal/admin/admin.html   |   1 +
 themes/terminal/static/js/admin.js |   6 ++
 8 files changed, 263 insertions(+), 4 deletions(-)
 create mode 100644 src/squishmark/services/csrf.py
 create mode 100644 tests/test_csrf.py

diff --git a/src/squishmark/routers/admin.py b/src/squishmark/routers/admin.py
index 7e817b6..99f18a8 100644
--- a/src/squishmark/routers/admin.py
+++ b/src/squishmark/routers/admin.py
@@ -14,6 +14,7 @@
 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
@@ -214,6 +215,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 +224,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
@@ -273,7 +276,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 +303,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 +332,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 +387,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:
diff --git a/src/squishmark/routers/auth.py b/src/squishmark/routers/auth.py
index 4503c31..8c6aa7d 100644
--- a/src/squishmark/routers/auth.py
+++ b/src/squishmark/routers/auth.py
@@ -116,6 +116,8 @@ 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_token", None)
 
     # Redirect to admin
     return RedirectResponse(url="/admin", status_code=302)
diff --git a/src/squishmark/services/csrf.py b/src/squishmark/services/csrf.py
new file mode 100644
index 0000000..b489468
--- /dev/null
+++ b/src/squishmark/services/csrf.py
@@ -0,0 +1,64 @@
+"""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).
+"""
+
+import logging
+import secrets
+
+from fastapi import HTTPException, Request
+
+from squishmark.config import get_settings
+
+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."""
+    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) -> None:
+    """FastAPI dependency that rejects requests missing or with an invalid CSRF token.
+
+    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
+
+    expected = request.session.get(SESSION_KEY) if hasattr(request, "session") else None
+    if not expected:
+        raise HTTPException(status_code=403, detail="CSRF token missing from session")
+
+    submitted = await _extract_submitted_token(request)
+    if not submitted or not secrets.compare_digest(submitted, expected):
+        raise HTTPException(status_code=403, detail="CSRF token invalid")
diff --git a/tests/test_csrf.py b/tests/test_csrf.py
new file mode 100644
index 0000000..67f43bc
--- /dev/null
+++ b/tests/test_csrf.py
@@ -0,0 +1,156 @@
+"""Tests for CSRF token generation and verification."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from squishmark.services.csrf import (
+    FORM_FIELD,
+    HEADER_NAME,
+    SESSION_KEY,
+    get_or_create_csrf_token,
+    verify_csrf_token,
+)
+
+
+def _request(
+    *,
+    session: dict | None = None,
+    header_token: str | None = None,
+    form_body: dict | None = None,
+    content_type: str = "application/json",
+) -> MagicMock:
+    """Build a mock Request with a session dict and optional token sources."""
+    request = MagicMock()
+    request.session = session if session is not None else {}
+    headers = {"content-type": content_type}
+    if header_token is not None:
+        headers[HEADER_NAME] = header_token
+    request.headers = headers
+    request.form = AsyncMock(return_value=form_body or {})
+    return request
+
+
+def test_get_or_create_csrf_token_mints_when_absent():
+    request = _request()
+    token = get_or_create_csrf_token(request)
+
+    assert token
+    assert len(token) > 20
+    assert request.session[SESSION_KEY] == token
+
+
+def test_get_or_create_csrf_token_returns_existing():
+    request = _request(session={SESSION_KEY: "preexisting-token"})
+
+    token = get_or_create_csrf_token(request)
+
+    assert token == "preexisting-token"
+
+
+def test_get_or_create_csrf_token_is_idempotent():
+    """Calling twice on the same request returns the same token."""
+    request = _request()
+
+    first = get_or_create_csrf_token(request)
+    second = get_or_create_csrf_token(request)
+
+    assert first == second
+
+
+@pytest.mark.asyncio
+async def test_verify_csrf_token_accepts_matching_header():
+    request = _request(
+        session={SESSION_KEY: "good-token"},
+        header_token="good-token",
+    )
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
+        await verify_csrf_token(request)  # should not raise
+
+
+@pytest.mark.asyncio
+async def test_verify_csrf_token_rejects_missing_session_token():
+    request = _request(session={}, header_token="anything")
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
+        with pytest.raises(HTTPException) as exc:
+            await verify_csrf_token(request)
+
+    assert exc.value.status_code == 403
+    assert "missing" in str(exc.value.detail).lower()
+
+
+@pytest.mark.asyncio
+async def test_verify_csrf_token_rejects_missing_submitted_token():
+    request = _request(session={SESSION_KEY: "good-token"})  # no header, no form
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
+        with pytest.raises(HTTPException) as exc:
+            await verify_csrf_token(request)
+
+    assert exc.value.status_code == 403
+    assert "invalid" in str(exc.value.detail).lower()
+
+
+@pytest.mark.asyncio
+async def test_verify_csrf_token_rejects_wrong_header():
+    request = _request(
+        session={SESSION_KEY: "good-token"},
+        header_token="wrong-token",
+    )
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
+        with pytest.raises(HTTPException) as exc:
+            await verify_csrf_token(request)
+
+    assert exc.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_verify_csrf_token_accepts_form_field_fallback():
+    """For form submissions without the header, the csrf_token form field is honored."""
+    request = _request(
+        session={SESSION_KEY: "good-token"},
+        form_body={FORM_FIELD: "good-token"},
+        content_type="application/x-www-form-urlencoded",
+    )
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
+        await verify_csrf_token(request)  # should not raise
+
+
+@pytest.mark.asyncio
+async def test_verify_csrf_token_header_takes_precedence_over_form():
+    """A valid header passes even if the form field is wrong."""
+    request = _request(
+        session={SESSION_KEY: "good-token"},
+        header_token="good-token",
+        form_body={FORM_FIELD: "wrong"},
+        content_type="application/x-www-form-urlencoded",
+    )
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
+        await verify_csrf_token(request)  # header wins, no raise
+
+
+@pytest.mark.asyncio
+async def test_verify_csrf_token_bypassed_in_dev_skip_auth():
+    """When debug and dev_skip_auth are both set, CSRF check is skipped."""
+    request = _request()  # no session, no token — would normally fail
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=True, dev_skip_auth=True)
+        await verify_csrf_token(request)  # should not raise
+
+
+@pytest.mark.asyncio
+async def test_verify_csrf_token_not_bypassed_in_prod_mode():
+    """dev_skip_auth without debug doesn't bypass."""
+    request = _request()
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=True)
+        with pytest.raises(HTTPException) as exc:
+            await verify_csrf_token(request)
+
+    assert exc.value.status_code == 403
diff --git a/themes/blue-tech/admin/admin.html b/themes/blue-tech/admin/admin.html
index 2e43089..c854a1b 100644
--- a/themes/blue-tech/admin/admin.html
+++ b/themes/blue-tech/admin/admin.html
@@ -3,6 +3,7 @@
 <head>
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <meta name="csrf-token" content="{{ csrf_token }}">
     <title>Admin - {{ site.title }}</title>
     <link rel="preconnect" href="https://fonts.googleapis.com">
     <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
@@ -124,6 +125,12 @@ <h2>Cache</h2>
     </main>
 
     <script>
+        // Attach CSRF token to every HTMX request.
+        document.body.addEventListener('htmx:configRequest', (event) => {
+            const token = document.querySelector('meta[name="csrf-token"]')?.content;
+            if (token) event.detail.headers['X-CSRF-Token'] = token;
+        });
+
         // Toggle note form
         function toggleNoteForm() {
             document.getElementById('note-form').classList.toggle('hidden');
diff --git a/themes/default/admin/admin.html b/themes/default/admin/admin.html
index 5dbbda7..c0ed063 100644
--- a/themes/default/admin/admin.html
+++ b/themes/default/admin/admin.html
@@ -3,6 +3,7 @@
 <head>
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <meta name="csrf-token" content="{{ csrf_token }}">
     <title>Admin - {{ site.title }}</title>
     <link rel="stylesheet" href="/static/{{ theme_name }}/style.css">
     <link rel="stylesheet" href="/static/{{ theme_name }}/admin.css">
@@ -121,6 +122,12 @@ <h2>Cache</h2>
     </main>
 
     <script>
+        // Attach CSRF token to every HTMX request.
+        document.body.addEventListener('htmx:configRequest', (event) => {
+            const token = document.querySelector('meta[name="csrf-token"]')?.content;
+            if (token) event.detail.headers['X-CSRF-Token'] = token;
+        });
+
         // Toggle note form
         function toggleNoteForm() {
             document.getElementById('note-form').classList.toggle('hidden');
diff --git a/themes/terminal/admin/admin.html b/themes/terminal/admin/admin.html
index 92b4137..b113579 100644
--- a/themes/terminal/admin/admin.html
+++ b/themes/terminal/admin/admin.html
@@ -3,6 +3,7 @@
 <head>
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <meta name="csrf-token" content="{{ csrf_token }}">
     <title>Admin - {{ site.title }}</title>
     <link rel="preconnect" href="https://fonts.googleapis.com">
     <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
diff --git a/themes/terminal/static/js/admin.js b/themes/terminal/static/js/admin.js
index 451c1f1..51fb187 100644
--- a/themes/terminal/static/js/admin.js
+++ b/themes/terminal/static/js/admin.js
@@ -3,6 +3,12 @@ function toggleNoteForm() {
     if (form) form.classList.toggle('hidden');
 }
 
+// Attach CSRF token to every HTMX request.
+document.body.addEventListener('htmx:configRequest', function (event) {
+    var meta = document.querySelector('meta[name="csrf-token"]');
+    if (meta && meta.content) event.detail.headers['X-CSRF-Token'] = meta.content;
+});
+
 document.addEventListener('DOMContentLoaded', function () {
     var el = document.getElementById('views-data');
     if (!el) return;

From 8e2d042384e482b5a71ca3b2aa0754833cff19ee Mon Sep 17 00:00:00 2001
From: Xeek <6032840+x3ek@users.noreply.github.com>
Date: Sat, 23 May 2026 07:01:09 -0500
Subject: [PATCH 2/6] feat(admin): add GET /admin/csrf for JSON callers, use
 CSRF_SESSION_KEY constant
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 src/squishmark/routers/admin.py |  8 ++++++
 src/squishmark/routers/auth.py  |  3 ++-
 tests/test_admin_notes.py       | 43 +++++++++++++++++++++++++++++++++
 3 files changed, 53 insertions(+), 1 deletion(-)

diff --git a/src/squishmark/routers/admin.py b/src/squishmark/routers/admin.py
index 99f18a8..cadb040 100644
--- a/src/squishmark/routers/admin.py
+++ b/src/squishmark/routers/admin.py
@@ -251,6 +251,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) -> dict[str, str]:
+    """Return the current CSRF token for use in subsequent mutation requests."""
+    del admin  # auth side-effect only
+    return {"csrf_token": get_or_create_csrf_token(request)}
+
+
 # Analytics endpoints
 @router.get("/analytics")
 async def get_analytics(
diff --git a/src/squishmark/routers/auth.py b/src/squishmark/routers/auth.py
index 8c6aa7d..ff05653 100644
--- a/src/squishmark/routers/auth.py
+++ b/src/squishmark/routers/auth.py
@@ -7,6 +7,7 @@
 from fastapi.responses import RedirectResponse
 
 from squishmark.config import get_settings
+from squishmark.services.csrf import SESSION_KEY as CSRF_SESSION_KEY
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
@@ -117,7 +118,7 @@ async def oauth_callback(
         "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_token", None)
+    request.session.pop(CSRF_SESSION_KEY, None)
 
     # Redirect to admin
     return RedirectResponse(url="/admin", status_code=302)
diff --git a/tests/test_admin_notes.py b/tests/test_admin_notes.py
index c7ad3d2..fca04d2 100644
--- a/tests/test_admin_notes.py
+++ b/tests/test_admin_notes.py
@@ -300,6 +300,49 @@ async def test_edit_form_not_found_raises_404():
     assert exc_info.value.status_code == 404
 
 
+@pytest.mark.asyncio
+async def test_get_csrf_returns_session_token():
+    """GET /admin/csrf returns the current CSRF token for JSON API callers."""
+    from squishmark.routers.admin import get_csrf
+
+    request = MagicMock()
+    request.session = {}
+
+    result = await get_csrf(request=request, admin="test-admin")
+
+    assert "csrf_token" in result
+    assert result["csrf_token"]
+    # Returned token matches what's now stored on the session.
+    assert request.session["csrf_token"] == result["csrf_token"]
+
+
+@pytest.mark.asyncio
+async def test_get_csrf_idempotent_within_session():
+    """Calling GET /admin/csrf twice on the same session returns the same token."""
+    from squishmark.routers.admin import get_csrf
+
+    request = MagicMock()
+    request.session = {}
+
+    first = await get_csrf(request=request, admin="test-admin")
+    second = await get_csrf(request=request, admin="test-admin")
+
+    assert first["csrf_token"] == second["csrf_token"]
+
+
+@pytest.mark.asyncio
+async def test_oauth_callback_rotates_csrf_token():
+    """Successful OAuth callback clears any prior CSRF token from the session."""
+    from squishmark.services.csrf import SESSION_KEY
+
+    # Simulate the rotation step in isolation — the surrounding OAuth flow is HTTP-heavy.
+    session = {SESSION_KEY: "stale-token", "user": {"login": "x"}}
+    session.pop(SESSION_KEY, None)
+
+    assert SESSION_KEY not in session
+    assert session["user"] == {"login": "x"}
+
+
 @pytest.mark.asyncio
 async def test_get_current_admin_htmx_attaches_redirect_header():
     """HTMX requests with no session get an HX-Redirect header on 401."""

From 67dbf3d66653a424626449b15a39c48978a353e1 Mon Sep 17 00:00:00 2001
From: Xeek <6032840+x3ek@users.noreply.github.com>
Date: Sat, 23 May 2026 07:09:30 -0500
Subject: [PATCH 3/6] refactor(auth): make CSRF dep resolve after auth so 401
 fires before 403
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 src/squishmark/dependencies.py  | 44 ++++++++++++++++++++++++++++++++-
 src/squishmark/routers/admin.py | 37 +++------------------------
 src/squishmark/services/csrf.py | 18 +++++++++++---
 tests/test_admin_notes.py       | 43 +++++++++++++++++++++++++++-----
 tests/test_csrf.py              | 35 ++++++++++++++++++++------
 5 files changed, 126 insertions(+), 51 deletions(-)

diff --git a/src/squishmark/dependencies.py b/src/squishmark/dependencies.py
index b81a316..b1a6629 100644
--- a/src/squishmark/dependencies.py
+++ b/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)]
diff --git a/src/squishmark/routers/admin.py b/src/squishmark/routers/admin.py
index cadb040..6d29639 100644
--- a/src/squishmark/routers/admin.py
+++ b/src/squishmark/routers/admin.py
@@ -10,6 +10,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from squishmark.config import get_settings
+from squishmark.dependencies import AdminUser, get_current_admin
 from squishmark.models.content import Config
 from squishmark.models.db import Note, get_db_session
 from squishmark.services.analytics import AnalyticsService
@@ -19,6 +20,9 @@
 from squishmark.services.notes import NotesService
 from squishmark.services.theme import get_theme_engine, reset_theme_engine
 
+# Re-exported here for tests/callers that still import these from this module.
+__all__ = ["router", "get_current_admin", "AdminUser"]
+
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/admin", tags=["admin"])
@@ -66,39 +70,6 @@ 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)
-
-    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)]
 DbSession = Annotated[AsyncSession, Depends(get_db_session)]
 
 
diff --git a/src/squishmark/services/csrf.py b/src/squishmark/services/csrf.py
index b489468..0dd9e88 100644
--- a/src/squishmark/services/csrf.py
+++ b/src/squishmark/services/csrf.py
@@ -3,15 +3,18 @@
 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).
+``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 HTTPException, Request
+from fastapi import Depends, HTTPException, Request
 
 from squishmark.config import get_settings
+from squishmark.dependencies import get_current_admin
 
 logger = logging.getLogger(__name__)
 
@@ -44,12 +47,21 @@ async def _extract_submitted_token(request: Request) -> str | None:
     return None
 
 
-async def verify_csrf_token(request: Request) -> 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``.
     """
+    del admin  # auth side-effect only — value not used here
     settings = get_settings()
     if settings.debug and settings.dev_skip_auth:
         logger.warning("CSRF bypassed - dev_skip_auth is enabled")
diff --git a/tests/test_admin_notes.py b/tests/test_admin_notes.py
index fca04d2..e5547c9 100644
--- a/tests/test_admin_notes.py
+++ b/tests/test_admin_notes.py
@@ -332,15 +332,46 @@ async def test_get_csrf_idempotent_within_session():
 
 @pytest.mark.asyncio
 async def test_oauth_callback_rotates_csrf_token():
-    """Successful OAuth callback clears any prior CSRF token from the session."""
+    """A successful OAuth callback clears any prior CSRF token from the session.
+
+    Exercises ``oauth_callback`` end-to-end with mocked GitHub HTTP calls so a
+    regression that drops the rotation line will fail this test.
+    """
+    from squishmark.routers.auth import oauth_callback
     from squishmark.services.csrf import SESSION_KEY
 
-    # Simulate the rotation step in isolation — the surrounding OAuth flow is HTTP-heavy.
-    session = {SESSION_KEY: "stale-token", "user": {"login": "x"}}
-    session.pop(SESSION_KEY, None)
+    request = MagicMock()
+    request.session = {SESSION_KEY: "stale-pre-auth-token"}
+    request.url_for = MagicMock(return_value="http://localhost:8000/auth/callback")
+
+    fake_settings = MagicMock(
+        debug=False,
+        dev_skip_auth=False,
+        secret_key="0123456789abcdef-rest",  # state validates against secret_key[:16]
+        github_client_id="cid",
+        github_client_secret="csecret",
+    )
+
+    token_response = MagicMock(status_code=200)
+    token_response.json = MagicMock(return_value={"access_token": "gho_test"})
+    user_response = MagicMock(status_code=200)
+    user_response.json = MagicMock(return_value={"login": "alice", "name": "Alice", "avatar_url": "u"})
+
+    mock_client = AsyncMock()
+    mock_client.post = AsyncMock(return_value=token_response)
+    mock_client.get = AsyncMock(return_value=user_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock(return_value=None)
+
+    with (
+        patch("squishmark.routers.auth.get_settings", return_value=fake_settings),
+        patch("squishmark.routers.auth.httpx.AsyncClient", return_value=mock_client),
+    ):
+        result = await oauth_callback(request=request, code="abc", state="0123456789abcdef")
 
-    assert SESSION_KEY not in session
-    assert session["user"] == {"login": "x"}
+    assert result.status_code == 302
+    assert SESSION_KEY not in request.session
+    assert request.session["user"]["login"] == "alice"
 
 
 @pytest.mark.asyncio
diff --git a/tests/test_csrf.py b/tests/test_csrf.py
index 67f43bc..2853e30 100644
--- a/tests/test_csrf.py
+++ b/tests/test_csrf.py
@@ -1,10 +1,13 @@
 """Tests for CSRF token generation and verification."""
 
+import inspect
+from typing import get_args
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi import HTTPException
 
+from squishmark.dependencies import get_current_admin
 from squishmark.services.csrf import (
     FORM_FIELD,
     HEADER_NAME,
@@ -14,6 +17,22 @@
 )
 
 
+def test_verify_csrf_token_depends_on_get_current_admin():
+    """``verify_csrf_token`` must declare get_current_admin as a dependency.
+
+    This forces FastAPI to resolve auth BEFORE the CSRF check, so unauthenticated
+    HTMX requests get the 401 + HX-Redirect from auth instead of a misleading
+    403 for a CSRF token they couldn't have obtained.
+    """
+    sig = inspect.signature(verify_csrf_token)
+    admin_param = sig.parameters.get("admin")
+    assert admin_param is not None, "verify_csrf_token must accept 'admin' parameter"
+    # Annotation is Annotated[str, Depends(get_current_admin)]
+    depends_objs = [arg for arg in get_args(admin_param.annotation) if hasattr(arg, "dependency")]
+    assert depends_objs, "admin parameter must use Depends(...)"
+    assert depends_objs[0].dependency is get_current_admin
+
+
 def _request(
     *,
     session: dict | None = None,
@@ -67,7 +86,7 @@ async def test_verify_csrf_token_accepts_matching_header():
     )
     with patch("squishmark.services.csrf.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
-        await verify_csrf_token(request)  # should not raise
+        await verify_csrf_token(request, admin="test-admin")  # should not raise
 
 
 @pytest.mark.asyncio
@@ -76,7 +95,7 @@ async def test_verify_csrf_token_rejects_missing_session_token():
     with patch("squishmark.services.csrf.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
         with pytest.raises(HTTPException) as exc:
-            await verify_csrf_token(request)
+            await verify_csrf_token(request, admin="test-admin")
 
     assert exc.value.status_code == 403
     assert "missing" in str(exc.value.detail).lower()
@@ -88,7 +107,7 @@ async def test_verify_csrf_token_rejects_missing_submitted_token():
     with patch("squishmark.services.csrf.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
         with pytest.raises(HTTPException) as exc:
-            await verify_csrf_token(request)
+            await verify_csrf_token(request, admin="test-admin")
 
     assert exc.value.status_code == 403
     assert "invalid" in str(exc.value.detail).lower()
@@ -103,7 +122,7 @@ async def test_verify_csrf_token_rejects_wrong_header():
     with patch("squishmark.services.csrf.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
         with pytest.raises(HTTPException) as exc:
-            await verify_csrf_token(request)
+            await verify_csrf_token(request, admin="test-admin")
 
     assert exc.value.status_code == 403
 
@@ -118,7 +137,7 @@ async def test_verify_csrf_token_accepts_form_field_fallback():
     )
     with patch("squishmark.services.csrf.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
-        await verify_csrf_token(request)  # should not raise
+        await verify_csrf_token(request, admin="test-admin")  # should not raise
 
 
 @pytest.mark.asyncio
@@ -132,7 +151,7 @@ async def test_verify_csrf_token_header_takes_precedence_over_form():
     )
     with patch("squishmark.services.csrf.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
-        await verify_csrf_token(request)  # header wins, no raise
+        await verify_csrf_token(request, admin="test-admin")  # header wins, no raise
 
 
 @pytest.mark.asyncio
@@ -141,7 +160,7 @@ async def test_verify_csrf_token_bypassed_in_dev_skip_auth():
     request = _request()  # no session, no token — would normally fail
     with patch("squishmark.services.csrf.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=True, dev_skip_auth=True)
-        await verify_csrf_token(request)  # should not raise
+        await verify_csrf_token(request, admin="test-admin")  # should not raise
 
 
 @pytest.mark.asyncio
@@ -151,6 +170,6 @@ async def test_verify_csrf_token_not_bypassed_in_prod_mode():
     with patch("squishmark.services.csrf.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=True)
         with pytest.raises(HTTPException) as exc:
-            await verify_csrf_token(request)
+            await verify_csrf_token(request, admin="test-admin")
 
     assert exc.value.status_code == 403

From 9e4b8e58efcc20c6542787fb0c3f9a6b615e44e6 Mon Sep 17 00:00:00 2001
From: Xeek <6032840+x3ek@users.noreply.github.com>
Date: Sat, 23 May 2026 07:40:55 -0500
Subject: [PATCH 4/6] refactor(deps): consolidate is_htmx + fix stale test
 patch targets
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 src/squishmark/dependencies.py  | 4 ++--
 src/squishmark/routers/admin.py | 7 +------
 tests/test_admin_notes.py       | 4 ++--
 3 files changed, 5 insertions(+), 10 deletions(-)

diff --git a/src/squishmark/dependencies.py b/src/squishmark/dependencies.py
index b1a6629..97f5c79 100644
--- a/src/squishmark/dependencies.py
+++ b/src/squishmark/dependencies.py
@@ -24,7 +24,7 @@ def is_admin(request: Request) -> bool:
     return user.get("login") in settings.admin_users_list
 
 
-def _is_htmx(request: Request) -> bool:
+def is_htmx(request: Request) -> bool:
     """Return True when the request was made by HTMX."""
     return request.headers.get("HX-Request") == "true"
 
@@ -46,7 +46,7 @@ async def get_current_admin(request: Request) -> str:
         logger.warning("Auth bypassed - returning dev-admin user")
         return "dev-admin"
 
-    htmx_headers = {"HX-Redirect": "/auth/login"} if _is_htmx(request) else None
+    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
diff --git a/src/squishmark/routers/admin.py b/src/squishmark/routers/admin.py
index 6d29639..8b96d77 100644
--- a/src/squishmark/routers/admin.py
+++ b/src/squishmark/routers/admin.py
@@ -10,7 +10,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from squishmark.config import get_settings
-from squishmark.dependencies import AdminUser, get_current_admin
+from squishmark.dependencies import AdminUser, get_current_admin, is_htmx
 from squishmark.models.content import Config
 from squishmark.models.db import Note, get_db_session
 from squishmark.services.analytics import AnalyticsService
@@ -28,11 +28,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."""
diff --git a/tests/test_admin_notes.py b/tests/test_admin_notes.py
index e5547c9..ecbae94 100644
--- a/tests/test_admin_notes.py
+++ b/tests/test_admin_notes.py
@@ -383,7 +383,7 @@ async def test_get_current_admin_htmx_attaches_redirect_header():
     request.session = {}
     request.headers = {"HX-Request": "true"}
 
-    with patch("squishmark.routers.admin.get_settings") as mock_settings:
+    with patch("squishmark.dependencies.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False, admin_users_list=[])
         with pytest.raises(HTTPException) as exc_info:
             await get_current_admin(request)
@@ -514,7 +514,7 @@ async def test_get_current_admin_non_htmx_omits_redirect_header():
     request.session = {}
     request.headers = {}
 
-    with patch("squishmark.routers.admin.get_settings") as mock_settings:
+    with patch("squishmark.dependencies.get_settings") as mock_settings:
         mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False, admin_users_list=[])
         with pytest.raises(HTTPException) as exc_info:
             await get_current_admin(request)

From 3ddad10cb50d40f4677a9b619914734b2761455d Mon Sep 17 00:00:00 2001
From: Xeek <6032840+x3ek@users.noreply.github.com>
Date: Sat, 23 May 2026 07:58:20 -0500
Subject: [PATCH 5/6] =?UTF-8?q?refactor(csrf):=20self-review=20pass=20?=
 =?UTF-8?q?=E2=80=94=20integration=20tests,=20generic=20error,=20polish?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 src/squishmark/routers/admin.py |  15 ++--
 src/squishmark/services/csrf.py |  19 +++--
 tests/test_admin_notes.py       |  18 +++--
 tests/test_csrf.py              |  17 ++++-
 tests/test_csrf_integration.py  | 123 ++++++++++++++++++++++++++++++++
 5 files changed, 172 insertions(+), 20 deletions(-)
 create mode 100644 tests/test_csrf_integration.py

diff --git a/src/squishmark/routers/admin.py b/src/squishmark/routers/admin.py
index 8b96d77..f802d5a 100644
--- a/src/squishmark/routers/admin.py
+++ b/src/squishmark/routers/admin.py
@@ -10,7 +10,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from squishmark.config import get_settings
-from squishmark.dependencies import AdminUser, get_current_admin, is_htmx
+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
@@ -20,9 +20,6 @@
 from squishmark.services.notes import NotesService
 from squishmark.services.theme import get_theme_engine, reset_theme_engine
 
-# Re-exported here for tests/callers that still import these from this module.
-__all__ = ["router", "get_current_admin", "AdminUser"]
-
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/admin", tags=["admin"])
@@ -65,6 +62,12 @@ class CacheRefreshResponse(BaseModel):
     duration_ms: float
 
 
+class CSRFTokenResponse(BaseModel):
+    """Response body for the CSRF token endpoint."""
+
+    csrf_token: str
+
+
 DbSession = Annotated[AsyncSession, Depends(get_db_session)]
 
 
@@ -219,10 +222,10 @@ async def admin_dashboard(
 
 # 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) -> dict[str, str]:
+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 {"csrf_token": get_or_create_csrf_token(request)}
+    return CSRFTokenResponse(csrf_token=get_or_create_csrf_token(request))
 
 
 # Analytics endpoints
diff --git a/src/squishmark/services/csrf.py b/src/squishmark/services/csrf.py
index 0dd9e88..3837a70 100644
--- a/src/squishmark/services/csrf.py
+++ b/src/squishmark/services/csrf.py
@@ -33,7 +33,14 @@ def get_or_create_csrf_token(request: Request) -> str:
 
 
 async def _extract_submitted_token(request: Request) -> str | None:
-    """Read the submitted CSRF token from header or form body."""
+    """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
@@ -67,10 +74,10 @@ async def verify_csrf_token(
         logger.warning("CSRF bypassed - dev_skip_auth is enabled")
         return
 
+    # Single error message 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.
     expected = request.session.get(SESSION_KEY) if hasattr(request, "session") else None
-    if not expected:
-        raise HTTPException(status_code=403, detail="CSRF token missing from session")
-
     submitted = await _extract_submitted_token(request)
-    if not submitted or not secrets.compare_digest(submitted, expected):
-        raise HTTPException(status_code=403, detail="CSRF token invalid")
+    if not expected or not submitted or not secrets.compare_digest(submitted, expected):
+        raise HTTPException(status_code=403, detail="CSRF validation failed")
diff --git a/tests/test_admin_notes.py b/tests/test_admin_notes.py
index ecbae94..cb8dbde 100644
--- a/tests/test_admin_notes.py
+++ b/tests/test_admin_notes.py
@@ -310,10 +310,9 @@ async def test_get_csrf_returns_session_token():
 
     result = await get_csrf(request=request, admin="test-admin")
 
-    assert "csrf_token" in result
-    assert result["csrf_token"]
+    assert result.csrf_token
     # Returned token matches what's now stored on the session.
-    assert request.session["csrf_token"] == result["csrf_token"]
+    assert request.session["csrf_token"] == result.csrf_token
 
 
 @pytest.mark.asyncio
@@ -327,7 +326,7 @@ async def test_get_csrf_idempotent_within_session():
     first = await get_csrf(request=request, admin="test-admin")
     second = await get_csrf(request=request, admin="test-admin")
 
-    assert first["csrf_token"] == second["csrf_token"]
+    assert first.csrf_token == second.csrf_token
 
 
 @pytest.mark.asyncio
@@ -373,11 +372,18 @@ async def test_oauth_callback_rotates_csrf_token():
     assert SESSION_KEY not in request.session
     assert request.session["user"]["login"] == "alice"
 
+    # And a fresh token mints on the next get_or_create call, distinct from the stale one.
+    from squishmark.services.csrf import get_or_create_csrf_token
+
+    new_token = get_or_create_csrf_token(request)
+    assert new_token
+    assert new_token != "stale-pre-auth-token"
+
 
 @pytest.mark.asyncio
 async def test_get_current_admin_htmx_attaches_redirect_header():
     """HTMX requests with no session get an HX-Redirect header on 401."""
-    from squishmark.routers.admin import get_current_admin
+    from squishmark.dependencies import get_current_admin
 
     request = MagicMock()
     request.session = {}
@@ -508,7 +514,7 @@ async def test_render_partial_restores_theme_on_render_exception():
 @pytest.mark.asyncio
 async def test_get_current_admin_non_htmx_omits_redirect_header():
     """Non-HTMX requests get a plain 401 with no HX-Redirect header."""
-    from squishmark.routers.admin import get_current_admin
+    from squishmark.dependencies import get_current_admin
 
     request = MagicMock()
     request.session = {}
diff --git a/tests/test_csrf.py b/tests/test_csrf.py
index 2853e30..72e86dd 100644
--- a/tests/test_csrf.py
+++ b/tests/test_csrf.py
@@ -98,7 +98,6 @@ async def test_verify_csrf_token_rejects_missing_session_token():
             await verify_csrf_token(request, admin="test-admin")
 
     assert exc.value.status_code == 403
-    assert "missing" in str(exc.value.detail).lower()
 
 
 @pytest.mark.asyncio
@@ -110,7 +109,21 @@ async def test_verify_csrf_token_rejects_missing_submitted_token():
             await verify_csrf_token(request, admin="test-admin")
 
     assert exc.value.status_code == 403
-    assert "invalid" in str(exc.value.detail).lower()
+
+
+@pytest.mark.asyncio
+async def test_verify_csrf_token_error_message_is_generic():
+    """The error detail must not leak which check (session vs submitted) failed."""
+    request_no_session = _request(session={}, header_token="anything")
+    request_no_submission = _request(session={SESSION_KEY: "good-token"})
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
+        with pytest.raises(HTTPException) as exc_a:
+            await verify_csrf_token(request_no_session, admin="x")
+        with pytest.raises(HTTPException) as exc_b:
+            await verify_csrf_token(request_no_submission, admin="x")
+
+    assert exc_a.value.detail == exc_b.value.detail
 
 
 @pytest.mark.asyncio
diff --git a/tests/test_csrf_integration.py b/tests/test_csrf_integration.py
new file mode 100644
index 0000000..26b6da7
--- /dev/null
+++ b/tests/test_csrf_integration.py
@@ -0,0 +1,123 @@
+"""Integration tests for CSRF enforcement on admin mutation routes.
+
+These exercise the full FastAPI dependency chain via ``TestClient`` — unlike
+``test_csrf.py`` (which calls ``verify_csrf_token`` directly) or
+``test_admin_notes.py`` (which calls route handlers directly), the tests here
+confirm that ``dependencies=[Depends(verify_csrf_token)]`` on the route
+decorators actually fires, and in the correct order relative to auth.
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import FastAPI, Request
+from fastapi.testclient import TestClient
+from starlette.middleware.sessions import SessionMiddleware
+
+from squishmark.models.db import get_db_session
+from squishmark.routers.admin import router as admin_router
+from squishmark.services.csrf import SESSION_KEY
+
+
+def _app() -> FastAPI:
+    """Minimal FastAPI app wired with admin router + SessionMiddleware."""
+    app = FastAPI()
+    app.add_middleware(SessionMiddleware, secret_key="test-secret-key", session_cookie="s")
+    app.include_router(admin_router)
+    # DB session dep — never reached by these tests, but FastAPI resolves all deps
+    # eagerly so it has to return something.
+    app.dependency_overrides[get_db_session] = lambda: AsyncMock()
+    return app
+
+
+@pytest.fixture
+def settings_prod():
+    """Patch settings to behave like production: debug off, dev_skip_auth off,
+    'alice' is the only admin."""
+    fake = MagicMock(
+        debug=False,
+        dev_skip_auth=False,
+        admin_users_list=["alice"],
+    )
+    with (
+        patch("squishmark.dependencies.get_settings", return_value=fake),
+        patch("squishmark.services.csrf.get_settings", return_value=fake),
+        patch("squishmark.routers.admin.get_settings", return_value=fake),
+    ):
+        yield fake
+
+
+def _client_with_session(session_data: dict) -> TestClient:
+    """Build a TestClient whose first request seeds ``request.session`` with
+    ``session_data``. TestClient's cookie jar persists across requests, so once
+    the session cookie is set, subsequent requests on the same client carry it.
+    """
+    app = _app()
+
+    @app.get("/_test/seed-session")
+    async def seed(request: Request) -> dict:
+        for k, v in session_data.items():
+            request.session[k] = v
+        return {"ok": True}
+
+    client = TestClient(app)
+    resp = client.get("/_test/seed-session")
+    assert resp.status_code == 200, resp.text
+    return client
+
+
+def test_unauthenticated_htmx_mutation_gets_401_redirect_not_403_csrf(settings_prod):
+    """The crux of the dep-ordering fix: an unauthenticated HTMX POST must hit
+    the auth check (401 + HX-Redirect), not CSRF (403)."""
+    del settings_prod
+    client = TestClient(_app())
+    resp = client.post(
+        "/admin/notes",
+        json={"path": "/x", "text": "y"},
+        headers={"HX-Request": "true"},
+    )
+    assert resp.status_code == 401
+    assert resp.headers.get("HX-Redirect") == "/auth/login"
+
+
+def test_authenticated_mutation_without_csrf_token_gets_403(settings_prod):
+    """An admin who somehow omits the CSRF token is rejected."""
+    del settings_prod
+    client = _client_with_session({"user": {"login": "alice"}})
+    resp = client.post(
+        "/admin/notes",
+        json={"path": "/x", "text": "y"},
+    )
+    assert resp.status_code == 403
+
+
+def test_authenticated_mutation_with_stale_csrf_token_gets_403(settings_prod):
+    """An admin sending a token that doesn't match the session value is rejected."""
+    del settings_prod
+    client = _client_with_session({"user": {"login": "alice"}, SESSION_KEY: "real-token"})
+    resp = client.post(
+        "/admin/notes",
+        json={"path": "/x", "text": "y"},
+        headers={"X-CSRF-Token": "stale-or-attacker-token"},
+    )
+    assert resp.status_code == 403
+
+
+def test_get_admin_csrf_endpoint_returns_token_for_admin(settings_prod):
+    """JSON callers can fetch the current token via GET /admin/csrf."""
+    del settings_prod
+    client = _client_with_session({"user": {"login": "alice"}})
+    resp = client.get("/admin/csrf")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert "csrf_token" in body
+    assert isinstance(body["csrf_token"], str)
+    assert len(body["csrf_token"]) > 20
+
+
+def test_get_admin_csrf_endpoint_is_auth_gated(settings_prod):
+    """Unauthenticated callers must not be able to obtain a token."""
+    del settings_prod
+    client = TestClient(_app())
+    resp = client.get("/admin/csrf")
+    assert resp.status_code == 401

From cc403fcfa61554d3d88d13412354f552baf1e05b Mon Sep 17 00:00:00 2001
From: Xeek <6032840+x3ek@users.noreply.github.com>
Date: Sat, 23 May 2026 08:01:10 -0500
Subject: [PATCH 6/6] feat(csrf): log rejection reason + rotation event for
 operability
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 src/squishmark/routers/auth.py  |  4 ++++
 src/squishmark/services/csrf.py | 22 ++++++++++++++++++----
 tests/test_csrf.py              | 33 +++++++++++++++++++++++++++++++++
 3 files changed, 55 insertions(+), 4 deletions(-)

diff --git a/src/squishmark/routers/auth.py b/src/squishmark/routers/auth.py
index ff05653..ce769f9 100644
--- a/src/squishmark/routers/auth.py
+++ b/src/squishmark/routers/auth.py
@@ -1,5 +1,6 @@
 """Authentication routes for GitHub OAuth."""
 
+import logging
 from urllib.parse import urlencode
 
 import httpx
@@ -9,6 +10,8 @@
 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"])
 
 GITHUB_AUTHORIZE_URL = "https://github.com/login/oauth/authorize"
@@ -119,6 +122,7 @@ async def oauth_callback(
     }
     # 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)
diff --git a/src/squishmark/services/csrf.py b/src/squishmark/services/csrf.py
index 3837a70..9a25232 100644
--- a/src/squishmark/services/csrf.py
+++ b/src/squishmark/services/csrf.py
@@ -14,7 +14,7 @@
 from fastapi import Depends, HTTPException, Request
 
 from squishmark.config import get_settings
-from squishmark.dependencies import get_current_admin
+from squishmark.dependencies import get_current_admin, is_htmx
 
 logger = logging.getLogger(__name__)
 
@@ -68,16 +68,30 @@ async def verify_csrf_token(
     Skipped when ``debug`` and ``dev_skip_auth`` are both set, matching the
     auth-bypass behavior in ``get_current_admin``.
     """
-    del admin  # auth side-effect only — value not used here
     settings = get_settings()
     if settings.debug and settings.dev_skip_auth:
         logger.warning("CSRF bypassed - dev_skip_auth is enabled")
         return
 
-    # Single error message regardless of which check fails — distinguishing
+    # 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.
+    # 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")
diff --git a/tests/test_csrf.py b/tests/test_csrf.py
index 72e86dd..62f06f5 100644
--- a/tests/test_csrf.py
+++ b/tests/test_csrf.py
@@ -39,6 +39,8 @@ def _request(
     header_token: str | None = None,
     form_body: dict | None = None,
     content_type: str = "application/json",
+    method: str = "POST",
+    path: str = "/admin/notes",
 ) -> MagicMock:
     """Build a mock Request with a session dict and optional token sources."""
     request = MagicMock()
@@ -48,6 +50,8 @@ def _request(
         headers[HEADER_NAME] = header_token
     request.headers = headers
     request.form = AsyncMock(return_value=form_body or {})
+    request.method = method
+    request.url.path = path
     return request
 
 
@@ -126,6 +130,35 @@ async def test_verify_csrf_token_error_message_is_generic():
     assert exc_a.value.detail == exc_b.value.detail
 
 
+@pytest.mark.asyncio
+async def test_verify_csrf_token_logs_rejection_reason(caplog):
+    """Server-side log records the specific failure mode so operators can debug.
+
+    The user response is generic ('CSRF validation failed') to avoid info
+    disclosure, so the log is the only place this detail lives.
+    """
+    import logging
+
+    caplog.set_level(logging.WARNING, logger="squishmark.services.csrf")
+    request = _request(
+        session={SESSION_KEY: "good"},
+        header_token="wrong",
+        method="POST",
+        path="/admin/notes",
+    )
+    with patch("squishmark.services.csrf.get_settings") as mock_settings:
+        mock_settings.return_value = MagicMock(debug=False, dev_skip_auth=False)
+        with pytest.raises(HTTPException):
+            await verify_csrf_token(request, admin="alice")
+
+    msg = caplog.text
+    assert "CSRF rejected" in msg
+    assert "reason=token-mismatch" in msg
+    assert "method=POST" in msg
+    assert "path=/admin/notes" in msg
+    assert "admin=alice" in msg
+
+
 @pytest.mark.asyncio
 async def test_verify_csrf_token_rejects_wrong_header():
     request = _request(