diff --git a/backend/api/auth.py b/backend/api/auth.py index 7649cab..364eebb 100644 --- a/backend/api/auth.py +++ b/backend/api/auth.py @@ -699,6 +699,7 @@ async def password_reset_request( @router.post("/password-reset/confirm") async def password_reset_confirm( + request: Request, body: PasswordResetConfirmRequest, session: AsyncSession = Depends(get_db), ): @@ -740,6 +741,12 @@ async def password_reset_confirm( # Update password and revoke all sessions (forces re-auth through TOTP if enabled) user.password_hash = auth_service.hash_password(body.new_password) await auth_service.revoke_all_refresh_tokens(session, user.id) + # Revoke any pre-reset access tokens still within their TTL window (T-7.2-01) + await request.app.state.redis.set( + f"user_nbf:{user.id}", + int(time.time()), + ex=settings.access_token_expire_minutes * 60, + ) await session.commit() # Do NOT issue tokens (AUTH-05 — user must pass TOTP gate on next login) diff --git a/backend/tests/test_auth_api.py b/backend/tests/test_auth_api.py index 47f786e..9ec2e9b 100644 --- a/backend/tests/test_auth_api.py +++ b/backend/tests/test_auth_api.py @@ -695,3 +695,26 @@ async def test_disable_totp_writes_user_nbf_to_redis(authed_client, db_session: nbf_bytes = await app.state.redis.get(f"user_nbf:{user.id}") assert nbf_bytes is not None assert int(nbf_bytes.decode() if isinstance(nbf_bytes, (bytes, bytearray)) else nbf_bytes) > 0 + + +@pytest.mark.asyncio +async def test_password_reset_confirm_writes_user_nbf_to_redis(authed_client, db_session: AsyncSession): + """POST /api/auth/password-reset/confirm must write user_nbf:{user_id} to Redis (CR-02 gap closure).""" + from main import app + from services.auth import create_password_reset_token + from sqlalchemy import select as sa_select + + await _register(authed_client, handle="nbfw_prc1", email="nbfw_prc1@example.com") + result = await db_session.execute(sa_select(User).where(User.email == "nbfw_prc1@example.com")) + user = result.scalar_one() + + reset_token = create_password_reset_token(str(user.id)) + resp = await authed_client.post( + "/api/auth/password-reset/confirm", + json={"token": reset_token, "new_password": "NewSecure!Pass99#"}, + ) + assert resp.status_code == 200 + + nbf_bytes = await app.state.redis.get(f"user_nbf:{user.id}") + assert nbf_bytes is not None + assert int(nbf_bytes.decode() if isinstance(nbf_bytes, (bytes, bytearray)) else nbf_bytes) > 0