"""Password hashing and signed access tokens using only the Python standard library.""" import base64 import hashlib import hmac import json import secrets import time from app.core.config import Settings _ITERATIONS = 600_000 def token_digest(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() def hash_password(password: str) -> str: salt = secrets.token_bytes(16) digest = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, _ITERATIONS) return "pbkdf2_sha256${}${}${}".format( _ITERATIONS, base64.urlsafe_b64encode(salt).decode(), base64.urlsafe_b64encode(digest).decode(), ) def verify_password(password: str, encoded: str) -> bool: try: algorithm, iterations, salt, digest = encoded.split("$", 3) if algorithm != "pbkdf2_sha256": return False actual = hashlib.pbkdf2_hmac( "sha256", password.encode(), base64.urlsafe_b64decode(salt), int(iterations) ) return hmac.compare_digest(actual, base64.urlsafe_b64decode(digest)) except (ValueError, TypeError): return False def _secret(settings: Settings) -> bytes: if settings.auth_token_secret: if settings.app_env == "production" and len(settings.auth_token_secret) < 32: raise ValueError("AUTH_TOKEN_SECRET must be at least 32 characters in production") return settings.auth_token_secret.encode() if settings.app_env == "test": return b"test-only-auth-secret-not-for-production" raise ValueError("AUTH_TOKEN_SECRET must be configured outside test") def issue_token( user_id: int, settings: Settings, token_version: int = 0, session_id: str | None = None, expires_at: int | None = None, ) -> tuple[str, int]: expires_at = expires_at or int(time.time()) + settings.auth_token_ttl_seconds payload = base64.urlsafe_b64encode( json.dumps( {"sub": user_id, "exp": expires_at, "ver": token_version, "sid": session_id}, separators=(",", ":"), ).encode() ).rstrip(b"=") signature = hmac.new(_secret(settings), payload, hashlib.sha256).digest() return ( f"{payload.decode()}.{base64.urlsafe_b64encode(signature).rstrip(b'=').decode()}", expires_at, ) def verify_token_claims(token: str, settings: Settings) -> tuple[int, int, str | None] | None: try: payload_text, signature_text = token.split(".", 1) payload = payload_text.encode() expected = hmac.new(_secret(settings), payload, hashlib.sha256).digest() signature = base64.urlsafe_b64decode(signature_text + "=" * (-len(signature_text) % 4)) if not hmac.compare_digest(expected, signature): return None data = json.loads(base64.urlsafe_b64decode(payload + b"=" * (-len(payload) % 4))) user_id = data["sub"] version = data.get("ver", 0) session_id = data.get("sid") return ( (user_id, version, session_id) if isinstance(user_id, int) and isinstance(version, int) and (session_id is None or isinstance(session_id, str)) and data["exp"] > time.time() else None ) except (KeyError, TypeError, ValueError, json.JSONDecodeError): return None def verify_token(token: str, settings: Settings) -> int | None: claims = verify_token_claims(token, settings) return claims[0] if claims else None def issue_refresh_token() -> str: return secrets.token_urlsafe(48) def derive_refresh_token(token: str, idempotency_key: str, settings: Settings) -> str: digest = hmac.new( _secret(settings), f"refresh-v1\0{token}\0{idempotency_key}".encode(), hashlib.sha256 ).digest() return base64.urlsafe_b64encode(digest).rstrip(b"=").decode()