Invalidate existing sessions on password change/reset (B-34)
Neither self-service password change nor the admin reset invalidated already-issued JWTs — a 24h-lifetime token stayed valid regardless, so a stolen token (or an attacker who already had the old password) kept working past a password change meant to lock them out. The admin reset exists precisely for the "account compromised" case and didn't evict the attacker at all. Add User.token_version (migration 943dbd74d983), embedded in every JWT as a "tv" claim and checked against the DB on every request in get_current_user/get_optional_user; a mismatch reads as session_expired. Both change-password and the admin reset bump it. change-password hands back a freshly minted token so the caller's own session keeps working instead of being logged out by its own request; the admin reset does not, since that session isn't the one making the call. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
+12
-4
@@ -39,12 +39,20 @@ def verify_password(password: str, password_hash: str) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def create_access_token(user_id: int) -> str:
|
||||
def create_access_token(user_id: int, token_version: int = 0) -> str:
|
||||
expires_at = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)
|
||||
payload = {"sub": str(user_id), "exp": expires_at}
|
||||
# "tv" lets get_current_user (app/auth/dependencies.py) reject a token issued
|
||||
# before the account's password was last changed (B-34): change-password and
|
||||
# the admin reset both bump User.token_version, so every token that still
|
||||
# carries the old value stops working immediately instead of staying valid
|
||||
# for up to jwt_expire_minutes after a compromise is supposedly handled.
|
||||
payload = {"sub": str(user_id), "tv": token_version, "exp": expires_at}
|
||||
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
|
||||
|
||||
|
||||
def decode_access_token(token: str) -> int:
|
||||
def decode_access_token(token: str) -> tuple[int, int]:
|
||||
payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
|
||||
return int(payload["sub"])
|
||||
# .get(..., 0) covers tokens issued before "tv" existed (pre-B-34 deploy) —
|
||||
# they carry no claim at all, and 0 is what a freshly migrated user's
|
||||
# token_version starts at, so those sessions keep working across the deploy.
|
||||
return int(payload["sub"]), int(payload.get("tv", 0))
|
||||
|
||||
Reference in New Issue
Block a user