hexdev
9187fbb970
added access_token_ttl setting fixed public and private key mismatch while decoding jwt decode_jwt now returns None on expired tokens instead of raising exc generate_pair converts sub to string and uses dynamic expiration
55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
import hashlib
|
|
import logging
|
|
import secrets
|
|
from typing import Any
|
|
|
|
import jwt
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
|
|
|
|
from config import cfg
|
|
from schemas.dto import KeyPair
|
|
from schemas.jwt import JWTPayload
|
|
from schemas.providers import ProvidersType
|
|
|
|
ctx = PasswordHasher()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def hash_password(plain_password: str) -> ...:
|
|
return ctx.hash(plain_password)
|
|
|
|
|
|
def verify_password(hashed_password: str, plain_password: str) -> bool:
|
|
try:
|
|
ctx.verify(hashed_password, plain_password)
|
|
return True
|
|
except (VerifyMismatchError, VerificationError, InvalidHashError):
|
|
return False
|
|
except Exception:
|
|
logger.exception("unexpected error while comparing password and hash")
|
|
return False
|
|
|
|
|
|
def generate_jwt(payload: dict[str, Any]) -> str:
|
|
return jwt.encode(payload, cfg.private_key, "RS256")
|
|
|
|
|
|
def decode_jwt(token: str) -> dict[str, Any] | None:
|
|
try:
|
|
return jwt.decode(token, cfg.public_key, "RS256")
|
|
except jwt.ExpiredSignatureError:
|
|
return
|
|
|
|
|
|
def generate_pair(user_id: int, iss: ProvidersType) -> KeyPair:
|
|
payload = JWTPayload(sub=str(user_id), iss=iss)
|
|
access_token = generate_jwt(payload.model_dump())
|
|
refresh_token = secrets.token_urlsafe(32)
|
|
|
|
return KeyPair(access_token=access_token, refresh_token=refresh_token)
|
|
|
|
|
|
def hash_refresh_token(token: str):
|
|
return hashlib.sha256(token.encode()).hexdigest()
|