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>
59 lines
2.4 KiB
Python
59 lines
2.4 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
import logging
|
|
|
|
import jwt
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import InvalidHashError, VerificationError
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Shared by registration (app/auth/routes.py) and the self-service password change
|
|
# (app/api/routes/users.py) so the two can't enforce different minimums.
|
|
MIN_PASSWORD_LENGTH = 8
|
|
|
|
_hasher = PasswordHasher()
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return _hasher.hash(password)
|
|
|
|
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
|
"""Any failure to verify reads as "wrong password", never as a server error.
|
|
|
|
Catching only VerifyMismatchError left the other two cases as unhandled 500s
|
|
(B-13): VerificationError covers argon2's other verification failures, and
|
|
InvalidHashError fires when the stored hash can't be parsed at all — which is a
|
|
data problem worth logging, but from the caller's side it still just means this
|
|
password does not open this account.
|
|
"""
|
|
try:
|
|
return _hasher.verify(password_hash, password)
|
|
except InvalidHashError:
|
|
logger.error("stored password hash is unparseable — password verification cannot succeed")
|
|
return False
|
|
except VerificationError:
|
|
return False
|
|
|
|
|
|
def create_access_token(user_id: int, token_version: int = 0) -> str:
|
|
expires_at = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)
|
|
# "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) -> tuple[int, int]:
|
|
payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
|
|
# .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))
|