Spaces:
Paused
Paused
| from __future__ import annotations | |
| from datetime import datetime, timedelta, timezone | |
| from typing import Any | |
| from jose import JWTError, jwt | |
| from passlib.context import CryptContext | |
| from app.configs.settings import Settings | |
| pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto") | |
| class TokenError(ValueError): | |
| """Raised when a JWT cannot be decoded or validated.""" | |
| def hash_password(password: str) -> str: | |
| return pwd_context.hash(password) | |
| def verify_password(password: str, hashed_password: str) -> bool: | |
| return pwd_context.verify(password, hashed_password) | |
| def create_token( | |
| settings: Settings, | |
| subject: str, | |
| token_type: str, | |
| expires_delta: timedelta, | |
| extra_claims: dict[str, Any] | None = None, | |
| ) -> str: | |
| now = datetime.now(timezone.utc) | |
| payload: dict[str, Any] = { | |
| "sub": subject, | |
| "type": token_type, | |
| "iat": int(now.timestamp()), | |
| "exp": int((now + expires_delta).timestamp()), | |
| } | |
| if extra_claims: | |
| payload.update(extra_claims) | |
| return jwt.encode(payload, settings.security.jwt_secret, algorithm=settings.security.jwt_algorithm) | |
| def decode_token(settings: Settings, token: str, expected_type: str | None = None) -> dict[str, Any]: | |
| try: | |
| payload = jwt.decode( | |
| token, | |
| settings.security.jwt_secret, | |
| algorithms=[settings.security.jwt_algorithm], | |
| ) | |
| except JWTError as exc: | |
| raise TokenError("Invalid token.") from exc | |
| if expected_type and payload.get("type") != expected_type: | |
| raise TokenError("Unexpected token type.") | |
| return payload | |