security(07.2-02): add user_nbf Redis NBF check to get_current_user

- Added import logging + _logger = logging.getLogger(__name__) to deps/auth.py
- Inserted user_nbf Redis check block after decode_access_token in get_current_user
- Check reads user_nbf:{payload['sub']} from Redis; rejects if payload['iat'] < stored timestamp
- Uses strict < comparison (token issued exactly at event second is allowed — D-02)
- Handles both bytes and str returns from Redis for FakeRedis/aioredis compatibility
- except HTTPException: raise guard precedes broad except Exception (T-7.2-02 Pitfall 1)
- Broad except logs warning and fails open (D-04 — Redis outage must not block requests)
- Promoted all 3 Wave 0 NBF xfail stubs to passing assertions in test_auth_deps.py
This commit is contained in:
curo1305
2026-06-05 19:19:06 +02:00
parent c00c1fbaa7
commit 30ad9fd015
2 changed files with 34 additions and 9 deletions
+28
View File
@@ -20,6 +20,7 @@ Usage in route handlers:
): ):
... ...
""" """
import logging
import uuid import uuid
from fastapi import Depends, HTTPException, Request, status from fastapi import Depends, HTTPException, Request, status
@@ -30,6 +31,8 @@ from db.models import User
from deps.db import get_db from deps.db import get_db
from services import auth as auth_service from services import auth as auth_service
_logger = logging.getLogger(__name__)
# HTTPBearer parses the Authorization: Bearer <token> header. # HTTPBearer parses the Authorization: Bearer <token> header.
# auto_error=True (default) raises 403 if no Authorization header is present. # auto_error=True (default) raises 403 if no Authorization header is present.
security = HTTPBearer() security = HTTPBearer()
@@ -56,6 +59,31 @@ async def get_current_user(
headers={"WWW-Authenticate": "Bearer"}, headers={"WWW-Authenticate": "Bearer"},
) from exc ) from exc
# ── user_nbf check (D-02, D-03, D-04) ──────────────────────────────────────
# Reject tokens issued before a security event (password change, TOTP enroll,
# deactivation). The user_nbf:{user_id} key is written by Wave 2 (Plan 03).
# Fail-open on Redis errors (D-04): a Redis outage must never deny service.
# CRITICAL: the `except HTTPException: raise` guard MUST precede the broad
# `except Exception` — otherwise the intentional 401 is swallowed (T-7.2-02).
try:
redis_client = request.app.state.redis
nbf_bytes = await redis_client.get(f"user_nbf:{payload['sub']}")
if nbf_bytes is not None:
# Tolerate both bytes and str returns (FakeRedis in tests may return
# either depending on what was stored; real aioredis returns bytes).
nbf_str = nbf_bytes.decode() if isinstance(nbf_bytes, (bytes, bytearray)) else nbf_bytes
if payload["iat"] < int(nbf_str):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Session invalidated",
headers={"WWW-Authenticate": "Bearer"},
)
except HTTPException:
raise # re-raise the 401 we just constructed (T-7.2-02: Pitfall 1 guard)
except Exception as exc:
_logger.warning("Redis user_nbf check failed (fail-open): %s", exc)
# ── end user_nbf check ──────────────────────────────────────────────────────
try: try:
user_uuid = uuid.UUID(payload["sub"]) user_uuid = uuid.UUID(payload["sub"])
except (KeyError, ValueError) as exc: except (KeyError, ValueError) as exc:
+6 -9
View File
@@ -186,7 +186,6 @@ def test_deps_auth_has_http_403():
# by replacing `assert False, "stub"` with the actual assertion logic. # by replacing `assert False, "stub"` with the actual assertion logic.
@pytest.mark.xfail(strict=False, reason="Phase 7.2 Wave 1 — NBF check not yet implemented in get_current_user")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_current_user_rejects_token_when_iat_before_user_nbf(auth_client, db_session): async def test_get_current_user_rejects_token_when_iat_before_user_nbf(auth_client, db_session):
"""Token with iat < user_nbf in Redis should return 401 'Session invalidated'.""" """Token with iat < user_nbf in Redis should return 401 'Session invalidated'."""
@@ -204,11 +203,11 @@ async def test_get_current_user_rejects_token_when_iat_before_user_nbf(auth_clie
resp = await auth_client.get( resp = await auth_client.get(
"/test/me", headers={"Authorization": f"Bearer {token}"} "/test/me", headers={"Authorization": f"Bearer {token}"}
) )
# Wave 1 will implement the NBF check; this assertion is the target behaviour assert resp.status_code == 401
assert False, "stub — Wave 1 will fill in: assert resp.status_code == 401 and 'Session invalidated' in resp.json()['detail']" assert resp.json()["detail"] == "Session invalidated"
assert resp.headers.get("WWW-Authenticate") == "Bearer"
@pytest.mark.xfail(strict=False, reason="Phase 7.2 Wave 1 — NBF check not yet implemented in get_current_user")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_current_user_allows_token_when_iat_after_user_nbf(auth_client, db_session): async def test_get_current_user_allows_token_when_iat_after_user_nbf(auth_client, db_session):
"""Token with iat > user_nbf in Redis should return 200 (token is valid).""" """Token with iat > user_nbf in Redis should return 200 (token is valid)."""
@@ -226,11 +225,9 @@ async def test_get_current_user_allows_token_when_iat_after_user_nbf(auth_client
resp = await auth_client.get( resp = await auth_client.get(
"/test/me", headers={"Authorization": f"Bearer {token}"} "/test/me", headers={"Authorization": f"Bearer {token}"}
) )
# Wave 1 will implement the NBF check; this assertion is the target behaviour assert resp.status_code == 200
assert False, "stub — Wave 1 will fill in: assert resp.status_code == 200"
@pytest.mark.xfail(strict=False, reason="Phase 7.2 Wave 1 — NBF check not yet implemented in get_current_user")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_current_user_failopen_on_redis_error(auth_client, db_session): async def test_get_current_user_failopen_on_redis_error(auth_client, db_session):
"""Redis error during NBF check must fail-open (return 200, not crash).""" """Redis error during NBF check must fail-open (return 200, not crash)."""
@@ -249,5 +246,5 @@ async def test_get_current_user_failopen_on_redis_error(auth_client, db_session)
resp = await auth_client.get( resp = await auth_client.get(
"/test/me", headers={"Authorization": f"Bearer {token}"} "/test/me", headers={"Authorization": f"Bearer {token}"}
) )
# Wave 1 will implement fail-open (D-04): assert resp.status_code == 200 # D-04: Redis outage must fail-open; request proceeds normally
assert False, "stub — Wave 1 will fill in: assert resp.status_code == 200" assert resp.status_code == 200