feat(07.2-03): add import time + user_nbf writes to api/auth.py (change_password, enable_totp, disable_totp)

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}

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:
+    nbf_bytes = await app.state.redis.get(f"user_nbf:{user.id}")
-    pytest.xfail("Phase 7.2 Wave 2 stub — user_nbf write not yet implemented in change_password")
+    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:
+    nbf_bytes = await app.state.redis.get(f"user_nbf:{user.id}")
-    pytest.xfail("Phase 7.2 Wave 2 stub — user_nbf write not yet implemented in enable_totp")
+    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:
+    nbf_bytes = await app.state.redis.get(f"user_nbf:{user.id}")
-    pytest.xfail("Phase 7.2 Wave 2 stub — user_nbf write not yet implemented in disable_totp")
+    assert nbf_bytes is not None
+    assert int(nbf_bytes.decode() if isinstance(nbf_bytes, (bytes, bytearray)) else nbf_bytes) > 0