import hashlib import logging import secrets from typing import Any import jwt from argon2 import PasswordHasher from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError from zxcvbn import zxcvbn 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, "EdDSA") def decode_jwt(token: str) -> dict[str, Any] | None: try: return jwt.decode(token, cfg.public_key, "EdDSA") 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, expires_at=payload.exp) def hash_refresh_token(token: str): return hashlib.sha256(token.encode()).hexdigest() def estimate_password_strength(password: str) -> bool: if len(password) < cfg.min_password_length: return False r = zxcvbn(password) return r.get("score", 0) > cfg.password_security_threshold