copilot-swe-agent[bot]
a4aaebfe66
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
411 lines
17 KiB
Python
411 lines
17 KiB
Python
"""Tests for the CSRF protection middleware (app/middleware/csrf.py)."""
|
||
|
||
import secrets
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
from fastapi import Request
|
||
from starlette.responses import JSONResponse, RedirectResponse
|
||
|
||
from app.middleware.csrf import CSRF_EXEMPT_PATHS, CSRF_PROTECTED_METHODS, CSRFMiddleware
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Unit tests – CSRFMiddleware._get_submitted_token
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestGetSubmittedToken:
|
||
"""Unit tests for the CSRF token extraction helper."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_header_token(self):
|
||
"""Token is read from the X-CSRF-Token request header."""
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.headers = {"X-CSRF-Token": "abc123"}
|
||
mock_request.form = AsyncMock(return_value={})
|
||
|
||
token = await CSRFMiddleware._get_submitted_token(mock_request)
|
||
assert token == "abc123"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_form_token_for_urlencoded(self):
|
||
"""Token is read from the form body for URL-encoded POST data."""
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.headers = {"content-type": "application/x-www-form-urlencoded"}
|
||
mock_request.body = AsyncMock(return_value=b"csrf_token=form_token_xyz")
|
||
mock_request.form = AsyncMock(return_value={"csrf_token": "form_token_xyz"})
|
||
|
||
token = await CSRFMiddleware._get_submitted_token(mock_request)
|
||
assert token == "form_token_xyz"
|
||
# body() must have been awaited so the body is cached for downstream re-reads.
|
||
mock_request.body.assert_awaited_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_header_takes_priority_over_form(self):
|
||
"""Header token takes priority over form body token."""
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.headers = {
|
||
"X-CSRF-Token": "header_token",
|
||
"content-type": "application/x-www-form-urlencoded",
|
||
}
|
||
mock_request.form = AsyncMock(return_value={"csrf_token": "form_token"})
|
||
|
||
token = await CSRFMiddleware._get_submitted_token(mock_request)
|
||
assert token == "header_token"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_none_when_no_token(self):
|
||
"""Returns None when no token is present in header or body."""
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.headers = {"content-type": "application/json"}
|
||
mock_request.form = AsyncMock(return_value={})
|
||
|
||
token = await CSRFMiddleware._get_submitted_token(mock_request)
|
||
assert token is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_none_for_multipart_without_header(self):
|
||
"""Multipart bodies without a header should return None (not parsed)."""
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.headers = {"content-type": "multipart/form-data; boundary=----boundary"}
|
||
mock_request.form = AsyncMock(return_value={"csrf_token": "should_not_be_read"})
|
||
|
||
token = await CSRFMiddleware._get_submitted_token(mock_request)
|
||
assert token is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_body_is_cached_before_form_parse(self):
|
||
"""body() is called before form() so downstream handlers can re-read the body.
|
||
|
||
This covers the Starlette BaseHTTPMiddleware body-replay bug: if form()
|
||
is called without first calling body(), _stream_consumed is set to True
|
||
but _body remains unset. _CachedRequest.wrapped_receive then forwards
|
||
an empty body to downstream apps (e.g. the /auth endpoint) causing
|
||
form_keys=[] and login failures. Calling body() first caches _body so
|
||
wrapped_receive correctly replays the full body.
|
||
"""
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.headers = {"content-type": "application/x-www-form-urlencoded"}
|
||
call_order: list[str] = []
|
||
|
||
async def _body():
|
||
call_order.append("body")
|
||
return b"csrf_token=tok&username=alice&password=test_password"
|
||
|
||
async def _form():
|
||
call_order.append("form")
|
||
return {"csrf_token": "tok", "username": "alice", "password": "test_password"}
|
||
|
||
mock_request.body = _body
|
||
mock_request.form = _form
|
||
|
||
token = await CSRFMiddleware._get_submitted_token(mock_request)
|
||
|
||
assert token == "tok"
|
||
# body() must be called BEFORE form() to ensure body caching.
|
||
assert call_order == ["body", "form"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_body_is_not_called_when_header_present(self):
|
||
"""When the CSRF token is in the X-CSRF-Token header, body() is not called.
|
||
|
||
For header-based token submission (AJAX / fetch requests) we skip body
|
||
parsing entirely, so the body stream remains unconsumed and downstream
|
||
handlers can read it normally.
|
||
"""
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.headers = {
|
||
"X-CSRF-Token": "header_tok",
|
||
"content-type": "application/x-www-form-urlencoded",
|
||
}
|
||
mock_request.body = AsyncMock(return_value=b"username=alice")
|
||
mock_request.form = AsyncMock(return_value={"csrf_token": "header_tok"})
|
||
|
||
token = await CSRFMiddleware._get_submitted_token(mock_request)
|
||
|
||
assert token == "header_tok"
|
||
# body() must NOT be called – the header path returns early.
|
||
mock_request.body.assert_not_awaited()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handles_form_parse_exception_gracefully(self):
|
||
"""A broken form body does not crash the middleware."""
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.headers = {"content-type": "application/x-www-form-urlencoded"}
|
||
mock_request.body = AsyncMock(return_value=b"")
|
||
mock_request.form = AsyncMock(side_effect=Exception("parse error"))
|
||
|
||
token = await CSRFMiddleware._get_submitted_token(mock_request)
|
||
assert token is None
|
||
|
||
|
||
# Unit tests – CSRFMiddleware.dispatch
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestCSRFMiddlewareDispatch:
|
||
"""Unit tests for the CSRFMiddleware.dispatch method."""
|
||
|
||
def _make_middleware(self, auth_enabled: bool = True):
|
||
mock_app = AsyncMock()
|
||
mock_config = MagicMock()
|
||
mock_config.auth_enabled = auth_enabled
|
||
return CSRFMiddleware(mock_app, mock_config)
|
||
|
||
def _make_request(self, method="GET", path="/", session=None, headers=None, state=None):
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.method = method
|
||
mock_request.url = MagicMock()
|
||
mock_request.url.path = path
|
||
mock_request.session = session if session is not None else {}
|
||
mock_request.headers = headers or {}
|
||
mock_request.state = MagicMock()
|
||
mock_request.state.csrf_token = None
|
||
return mock_request
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_noop_when_auth_disabled(self):
|
||
"""Middleware is a no-op when AUTH_ENABLED is False."""
|
||
middleware = self._make_middleware(auth_enabled=False)
|
||
request = self._make_request(method="POST", path="/api/test")
|
||
|
||
next_response = MagicMock()
|
||
call_next = AsyncMock(return_value=next_response)
|
||
|
||
result = await middleware.dispatch(request, call_next)
|
||
|
||
call_next.assert_called_once_with(request)
|
||
assert result is next_response
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generates_token_when_not_in_session(self):
|
||
"""A new CSRF token is generated and stored in the session when absent."""
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(method="GET", session={})
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
await middleware.dispatch(request, call_next)
|
||
|
||
assert "csrf_token" in request.session
|
||
token = request.session["csrf_token"]
|
||
assert len(token) == 64 # secrets.token_hex(32) -> 64 hex chars
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reuses_existing_token_from_session(self):
|
||
"""An existing session token is reused instead of regenerating."""
|
||
existing_token = secrets.token_hex(32)
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(method="GET", session={"csrf_token": existing_token})
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
await middleware.dispatch(request, call_next)
|
||
|
||
assert request.session["csrf_token"] == existing_token
|
||
assert request.state.csrf_token == existing_token
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_attaches_token_to_request_state(self):
|
||
"""Token is always attached to request.state.csrf_token."""
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(method="GET", session={})
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
await middleware.dispatch(request, call_next)
|
||
|
||
assert request.state.csrf_token is not None
|
||
assert len(request.state.csrf_token) == 64
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_safe_methods_pass_without_token(self):
|
||
"""GET/HEAD/OPTIONS requests pass through without CSRF validation."""
|
||
middleware = self._make_middleware()
|
||
|
||
for method in ("GET", "HEAD", "OPTIONS"):
|
||
request = self._make_request(method=method, session={})
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
result = await middleware.dispatch(request, call_next)
|
||
call_next.assert_called_once_with(request)
|
||
call_next.reset_mock()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_post_with_valid_header_token_passes(self):
|
||
"""POST with a matching X-CSRF-Token header passes validation."""
|
||
token = secrets.token_hex(32)
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(
|
||
method="POST",
|
||
path="/api/process/",
|
||
session={"csrf_token": token},
|
||
headers={"X-CSRF-Token": token},
|
||
)
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
with patch.object(CSRFMiddleware, "_get_submitted_token", new=AsyncMock(return_value=token)):
|
||
result = await middleware.dispatch(request, call_next)
|
||
|
||
call_next.assert_called_once_with(request)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_post_with_invalid_token_returns_403_for_api(self):
|
||
"""POST with a wrong token on an API route returns HTTP 403."""
|
||
token = secrets.token_hex(32)
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(
|
||
method="POST",
|
||
path="/api/process/",
|
||
session={"csrf_token": token},
|
||
)
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
with patch.object(CSRFMiddleware, "_get_submitted_token", new=AsyncMock(return_value="wrong_token")):
|
||
result = await middleware.dispatch(request, call_next)
|
||
|
||
assert isinstance(result, JSONResponse)
|
||
assert result.status_code == 403
|
||
call_next.assert_not_called()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_post_with_missing_token_returns_403_for_api(self):
|
||
"""POST with no CSRF token on an API route returns HTTP 403."""
|
||
token = secrets.token_hex(32)
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(
|
||
method="POST",
|
||
path="/api/settings/bulk-update",
|
||
session={"csrf_token": token},
|
||
)
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
with patch.object(CSRFMiddleware, "_get_submitted_token", new=AsyncMock(return_value=None)):
|
||
result = await middleware.dispatch(request, call_next)
|
||
|
||
assert isinstance(result, JSONResponse)
|
||
assert result.status_code == 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_post_with_invalid_token_redirects_for_frontend(self):
|
||
"""POST with a wrong token on a frontend route redirects to /login."""
|
||
token = secrets.token_hex(32)
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(
|
||
method="POST",
|
||
path="/auth",
|
||
session={"csrf_token": token},
|
||
)
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
with patch.object(CSRFMiddleware, "_get_submitted_token", new=AsyncMock(return_value="bad_token")):
|
||
result = await middleware.dispatch(request, call_next)
|
||
|
||
assert isinstance(result, RedirectResponse)
|
||
assert result.status_code == 302
|
||
assert "/login?error=Invalid+request" in result.headers["location"]
|
||
call_next.assert_not_called()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_with_valid_token_passes(self):
|
||
"""DELETE with a matching token passes through."""
|
||
token = secrets.token_hex(32)
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(
|
||
method="DELETE",
|
||
path="/api/files/1",
|
||
session={"csrf_token": token},
|
||
)
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
with patch.object(CSRFMiddleware, "_get_submitted_token", new=AsyncMock(return_value=token)):
|
||
result = await middleware.dispatch(request, call_next)
|
||
|
||
call_next.assert_called_once_with(request)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_callback_is_exempt(self):
|
||
"""OAuth callback path is exempt from CSRF validation even on POST."""
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(
|
||
method="POST",
|
||
path="/oauth-callback",
|
||
session={"csrf_token": secrets.token_hex(32)},
|
||
)
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
with patch.object(CSRFMiddleware, "_get_submitted_token", new=AsyncMock(return_value=None)):
|
||
result = await middleware.dispatch(request, call_next)
|
||
|
||
call_next.assert_called_once_with(request)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_qr_auth_claim_is_exempt(self):
|
||
"""QR auth claim path is exempt from CSRF validation.
|
||
|
||
The mobile app calls this endpoint without a browser session and
|
||
therefore without a CSRF token. The cryptographically-random,
|
||
single-use challenge token provides equivalent protection.
|
||
"""
|
||
middleware = self._make_middleware()
|
||
request = self._make_request(
|
||
method="POST",
|
||
path="/api/qr-auth/claim",
|
||
session={},
|
||
)
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
|
||
with patch.object(CSRFMiddleware, "_get_submitted_token", new=AsyncMock(return_value=None)):
|
||
result = await middleware.dispatch(request, call_next)
|
||
|
||
call_next.assert_called_once_with(request)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Integration tests – via TestClient
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.security
|
||
class TestCSRFIntegration:
|
||
"""Integration tests for CSRF protection using the FastAPI TestClient.
|
||
|
||
The shared ``client`` fixture runs with ``AUTH_ENABLED=False`` (see
|
||
``conftest.py``), so the CSRF middleware is disabled by design. The tests
|
||
below verify constants and confirm that the middleware is a no-op in that
|
||
configuration.
|
||
"""
|
||
|
||
def test_csrf_constants(self):
|
||
"""Verify the constant sets have the expected members."""
|
||
assert "POST" in CSRF_PROTECTED_METHODS
|
||
assert "PUT" in CSRF_PROTECTED_METHODS
|
||
assert "DELETE" in CSRF_PROTECTED_METHODS
|
||
assert "PATCH" in CSRF_PROTECTED_METHODS
|
||
assert "GET" not in CSRF_PROTECTED_METHODS
|
||
assert "/oauth-callback" in CSRF_EXEMPT_PATHS
|
||
assert "/api/qr-auth/claim" in CSRF_EXEMPT_PATHS
|
||
|
||
def test_csrf_middleware_noop_when_auth_disabled(self):
|
||
"""When AUTH_ENABLED=False the middleware dispatch is a no-op (no validation)."""
|
||
# Build a middleware instance with auth disabled.
|
||
mock_app = AsyncMock()
|
||
mock_config = MagicMock()
|
||
mock_config.auth_enabled = False
|
||
middleware = CSRFMiddleware(mock_app, mock_config)
|
||
|
||
import asyncio
|
||
|
||
mock_request = MagicMock(spec=Request)
|
||
mock_request.method = "POST"
|
||
mock_request.url = MagicMock()
|
||
mock_request.url.path = "/api/process/"
|
||
mock_request.session = {}
|
||
|
||
call_next = AsyncMock(return_value=MagicMock())
|
||
asyncio.run(middleware.dispatch(mock_request, call_next))
|
||
|
||
# call_next must have been called (request was not blocked).
|
||
call_next.assert_called_once_with(mock_request)
|
||
# Session should remain untouched (no token generated).
|
||
assert "csrf_token" not in mock_request.session
|