fix(07.2): add user_nbf write to password_reset_confirm — CR-02 gap closure
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user