diff --git a/backend/api/auth.py b/backend/api/auth.py index 0c44f95..7649cab 100644 --- a/backend/api/auth.py +++ b/backend/api/auth.py @@ -20,6 +20,7 @@ Security invariants: from __future__ import annotations import hashlib +import time import uuid from typing import Literal, Optional @@ -503,6 +504,12 @@ async def change_password( ip_address=_ip, metadata_={"sessions_revoked": revoked}, ) + # Revoke any pre-change access tokens still within their TTL window (T-7.2-01) + await request.app.state.redis.set( + f"user_nbf:{current_user.id}", + int(time.time()), + ex=settings.access_token_expire_minutes * 60, + ) await session.commit() return {"message": "Password updated", "sessions_revoked": revoked} @@ -596,6 +603,12 @@ async def enable_totp( ip_address=_ip, metadata_={"sessions_revoked": revoked}, ) + # Revoke any pre-enroll access tokens still within their TTL window (T-7.2-01) + await redis_client.set( + f"user_nbf:{current_user.id}", + int(time.time()), + ex=settings.access_token_expire_minutes * 60, + ) await session.commit() return {"backup_codes": plain_codes, "sessions_revoked": revoked} @@ -636,6 +649,12 @@ async def disable_totp( ip_address=_ip, metadata_={"sessions_revoked": revoked}, ) + # Revoke any pre-revoke access tokens still within their TTL window (T-7.2-01) + await request.app.state.redis.set( + f"user_nbf:{current_user.id}", + int(time.time()), + ex=settings.access_token_expire_minutes * 60, + ) await session.commit() return {"message": "TOTP disabled", "sessions_revoked": revoked} diff --git a/backend/tests/test_auth_api.py b/backend/tests/test_auth_api.py index daf61fc..47f786e 100644 --- a/backend/tests/test_auth_api.py +++ b/backend/tests/test_auth_api.py @@ -615,10 +615,10 @@ async def test_disable_totp_revokes_other_sessions(authed_client, db_session: As # assert int(nbf_bytes.decode() if isinstance(nbf_bytes, (bytes, bytearray)) else nbf_bytes) > 0 -@pytest.mark.xfail(strict=False, reason="Phase 7.2 Wave 2 — user_nbf write not yet added to change_password handler") @pytest.mark.asyncio async def test_change_password_writes_user_nbf_to_redis(authed_client, db_session: AsyncSession): """POST /api/auth/change-password must write user_nbf:{user_id} to Redis.""" + from main import app from sqlalchemy import select as sa_select await _register(authed_client, handle="nbfw_cp1", email="nbfw_cp1@example.com") @@ -636,14 +636,15 @@ async def test_change_password_writes_user_nbf_to_redis(authed_client, db_sessio ) assert resp.status_code == 200 - # Wave 2 will add the user_nbf write; this is the target assertion: - pytest.xfail("Phase 7.2 Wave 2 stub — user_nbf write not yet implemented in change_password") + 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.xfail(strict=False, reason="Phase 7.2 Wave 2 — user_nbf write not yet added to enable_totp handler") @pytest.mark.asyncio async def test_enable_totp_writes_user_nbf_to_redis(authed_client, db_session: AsyncSession): """POST /api/auth/totp/enable must write user_nbf:{user_id} to Redis.""" + from main import app from sqlalchemy import select as sa_select await _register(authed_client, handle="nbfw_et1", email="nbfw_et1@example.com") @@ -664,14 +665,15 @@ async def test_enable_totp_writes_user_nbf_to_redis(authed_client, db_session: A ) assert resp.status_code == 200 - # Wave 2 will add the user_nbf write; this is the target assertion: - pytest.xfail("Phase 7.2 Wave 2 stub — user_nbf write not yet implemented in enable_totp") + 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.xfail(strict=False, reason="Phase 7.2 Wave 2 — user_nbf write not yet added to disable_totp handler") @pytest.mark.asyncio async def test_disable_totp_writes_user_nbf_to_redis(authed_client, db_session: AsyncSession): """DELETE /api/auth/totp must write user_nbf:{user_id} to Redis.""" + from main import app from sqlalchemy import select as sa_select await _register(authed_client, handle="nbfw_dt1", email="nbfw_dt1@example.com") @@ -690,5 +692,6 @@ async def test_disable_totp_writes_user_nbf_to_redis(authed_client, db_session: ) assert resp.status_code == 200 - # Wave 2 will add the user_nbf write; this is the target assertion: - pytest.xfail("Phase 7.2 Wave 2 stub — user_nbf write not yet implemented in disable_totp") + 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