Spaces:
Running
Running
| """Password hashing and JWT token helpers.""" | |
| from datetime import datetime, timedelta, timezone | |
| from typing import Any | |
| import bcrypt | |
| import jwt | |
| from app.config import get_settings | |
| settings = get_settings() | |
| def hash_password(password: str) -> str: | |
| return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") | |
| def verify_password(password: str, password_hash: str) -> bool: | |
| try: | |
| return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8")) | |
| except ValueError: | |
| return False | |
| def _create_token(subject: str, token_type: str, expires_delta: timedelta) -> str: | |
| now = datetime.now(timezone.utc) | |
| payload: dict[str, Any] = { | |
| "sub": subject, | |
| "type": token_type, | |
| "iat": now, | |
| "exp": now + expires_delta, | |
| } | |
| return jwt.encode(payload, settings.secret_key, algorithm=settings.jwt_algorithm) | |
| def create_access_token(user_id: int) -> str: | |
| return _create_token( | |
| str(user_id), "access", timedelta(minutes=settings.access_token_expire_minutes) | |
| ) | |
| def create_refresh_token(user_id: int) -> str: | |
| return _create_token( | |
| str(user_id), "refresh", timedelta(days=settings.refresh_token_expire_days) | |
| ) | |
| def decode_token(token: str, expected_type: str) -> int | None: | |
| """Return the user id encoded in the token, or None if invalid.""" | |
| try: | |
| payload = jwt.decode( | |
| token, settings.secret_key, algorithms=[settings.jwt_algorithm] | |
| ) | |
| except jwt.PyJWTError: | |
| return None | |
| if payload.get("type") != expected_type: | |
| return None | |
| try: | |
| return int(payload["sub"]) | |
| except (KeyError, ValueError): | |
| return None | |