fix/security headers now adds cache-control and pragma

Author: chintakjoshi

app/middleware/security_headers.py

@@ -10,6 +10,19 @@
 class SecurityHeadersMiddleware(BaseHTTPMiddleware):
     """Ensure every response carries mandatory security headers."""
 
+    _NO_STORE_EXACT_PATHS = frozenset(
+        {
+            "/auth/csrf",
+            "/auth/login",
+            "/auth/token",
+            "/auth/logout",
+            "/auth/reauth",
+            "/auth/otp/verify/login",
+            "/auth/otp/verify/action",
+            "/auth/oauth/google/callback",
+            "/auth/saml/callback",
+        }
+    )
     _DEFAULT_CONTENT_SECURITY_POLICY = (
         "default-src 'none'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'"
     )
@@ -39,6 +52,9 @@ def _headers_for_path(cls, path: str) -> dict[str, str]:
             if path.startswith("/docs")
             else cls._DEFAULT_CONTENT_SECURITY_POLICY
         )
+        if path in cls._NO_STORE_EXACT_PATHS:
+            headers["Cache-Control"] = "no-store"
+            headers["Pragma"] = "no-cache"
         return headers
 
     async def dispatch(self, request: Request, call_next) -> Response:

tests/integration/test_auth_router_real.py

@@ -9,6 +9,12 @@
 from app.core.jwt import get_jwt_service
 
 
+def _assert_no_store_headers(headers: dict[str, str]) -> None:
+    """Assert auth-state responses are marked as non-cacheable."""
+    assert headers.get("cache-control") == "no-store"
+    assert headers.get("pragma") == "no-cache"
+
+
 @pytest.mark.asyncio
 async def test_auth_login_refresh_logout_happy_path(
     app_factory,
@@ -27,6 +33,7 @@ async def test_auth_login_refresh_logout_happy_path(
             json={"email": "alice@example.com", "password": "Password123!"},
         )
         assert login_response.status_code == 200
+        _assert_no_store_headers(dict(login_response.headers))
         login_payload = login_response.json()
         assert login_payload["access_token"]
         assert login_payload["refresh_token"]
@@ -46,6 +53,7 @@ async def test_auth_login_refresh_logout_happy_path(
             json={"refresh_token": login_payload["refresh_token"]},
         )
         assert refresh_response.status_code == 200
+        _assert_no_store_headers(dict(refresh_response.headers))
         refresh_payload = refresh_response.json()
         assert refresh_payload["refresh_token"] != login_payload["refresh_token"]
         refresh_access_claims = jwt_service.verify_token(
@@ -64,12 +72,14 @@ async def test_auth_login_refresh_logout_happy_path(
             headers={"authorization": f"Bearer {refresh_payload['access_token']}"},
         )
         assert logout_response.status_code == 204
+        _assert_no_store_headers(dict(logout_response.headers))
 
         refresh_after_logout = await client.post(
             "/auth/token",
             json={"refresh_token": refresh_payload["refresh_token"]},
         )
         assert refresh_after_logout.status_code == 401
+        _assert_no_store_headers(dict(refresh_after_logout.headers))
         assert refresh_after_logout.json()["code"] == "session_expired"
 
 
@@ -88,6 +98,7 @@ async def test_auth_cookie_login_refresh_logout_happy_path(
     ) as client:
         csrf_response = await client.get("/auth/csrf")
         assert csrf_response.status_code == 200
+        _assert_no_store_headers(dict(csrf_response.headers))
         csrf_token = csrf_response.json()["csrf_token"]
 
         login_response = await client.post(
@@ -99,6 +110,7 @@ async def test_auth_cookie_login_refresh_logout_happy_path(
             },
         )
         assert login_response.status_code == 200
+        _assert_no_store_headers(dict(login_response.headers))
         assert login_response.json() == {
             "authenticated": True,
             "session_transport": "cookie",
@@ -122,6 +134,7 @@ async def test_auth_cookie_login_refresh_logout_happy_path(
             },
         )
         assert refresh_response.status_code == 200
+        _assert_no_store_headers(dict(refresh_response.headers))
         assert refresh_response.json() == {
             "authenticated": True,
             "session_transport": "cookie",
@@ -146,6 +159,7 @@ async def test_auth_cookie_login_refresh_logout_happy_path(
             },
         )
         assert logout_response.status_code == 204
+        _assert_no_store_headers(dict(logout_response.headers))
 
         replacement_csrf = await client.get("/auth/csrf")
         assert replacement_csrf.status_code == 200

tests/integration/test_middleware_stack.py

@@ -21,6 +21,10 @@
     "x-content-type-options": "nosniff",
     "strict-transport-security": "max-age=63072000; includeSubDomains; preload",
 }
+_NO_STORE_HEADERS = {
+    "cache-control": "no-store",
+    "pragma": "no-cache",
+}
 
 
 class _InMemoryRateLimitRedis:
@@ -100,6 +104,10 @@ async def server_error() -> None:
     async def login() -> dict[str, bool]:
         return {"ok": True}
 
+    @app.post("/auth/token")
+    async def token() -> dict[str, bool]:
+        return {"ok": True}
+
     return app
 
 
@@ -109,6 +117,12 @@ def _assert_security_headers(headers: dict[str, str]) -> None:
         assert headers.get(header_name) == expected_value
 
 
+def _assert_no_store_headers(headers: dict[str, str]) -> None:
+    """Assert token-bearing responses are marked as non-cacheable."""
+    for header_name, expected_value in _NO_STORE_HEADERS.items():
+        assert headers.get(header_name) == expected_value
+
+
 @pytest.mark.asyncio
 async def test_headers_present_on_success_and_error_responses() -> None:
     """Correlation ID and security headers are present on 2xx/4xx/5xx."""
@@ -128,10 +142,34 @@ async def test_headers_present_on_success_and_error_responses() -> None:
     assert client_error.status_code == 401
     assert client_error.headers.get("x-correlation-id")
     _assert_security_headers(dict(client_error.headers))
+    for header_name in _NO_STORE_HEADERS:
+        assert header_name not in client_error.headers
 
     assert server_error.status_code == 500
     assert server_error.headers.get("x-correlation-id")
     _assert_security_headers(dict(server_error.headers))
+    for header_name in _NO_STORE_HEADERS:
+        assert header_name not in server_error.headers
+
+
+@pytest.mark.asyncio
+async def test_token_bearing_auth_routes_are_marked_no_store() -> None:
+    """Login and token routes include Cache-Control no-store and Pragma no-cache."""
+    app = _build_test_app()
+
+    async with AsyncClient(
+        transport=ASGITransport(app=app), base_url="http://testserver"
+    ) as client:
+        login_response = await client.post("/auth/login")
+        token_response = await client.post("/auth/token")
+
+    assert login_response.status_code == 200
+    _assert_security_headers(dict(login_response.headers))
+    _assert_no_store_headers(dict(login_response.headers))
+
+    assert token_response.status_code == 200
+    _assert_security_headers(dict(token_response.headers))
+    _assert_no_store_headers(dict(token_response.headers))
 
 
 @pytest.mark.asyncio
@@ -148,6 +186,8 @@ async def test_rate_limit_rejects_auth_login_with_required_error_code() -> None:
     assert first.status_code == 200
     assert second.status_code == 429
     assert second.json() == {"detail": "Rate limit exceeded.", "code": "rate_limited"}
+    _assert_no_store_headers(dict(first.headers))
+    _assert_no_store_headers(dict(second.headers))
 
 
 @pytest.mark.asyncio