from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone from uuid import uuid4 from jose import JWTError, jwt from passlib.context import CryptContext from app.core.config import get_settings from app.core.exceptions import UnauthorizedError pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") settings = get_settings() ACCESS_TOKEN_TYPE = "access" # nosec B105 @dataclass(slots=True) class TokenPayload: user_id: int def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: return pwd_context.hash(password) def create_access_token(subject: str) -> str: now = datetime.now(timezone.utc) expire = now + timedelta(minutes=settings.access_token_expire_minutes) to_encode = { "sub": subject, "iat": int(now.timestamp()), "nbf": int(now.timestamp()), "exp": int(expire.timestamp()), "jti": str(uuid4()), "type": ACCESS_TOKEN_TYPE, "iss": settings.jwt_issuer, "aud": settings.jwt_audience, } return jwt.encode(to_encode, settings.jwt_secret_key, algorithm=settings.jwt_algorithm) def decode_access_token(token: str) -> TokenPayload: try: payload = jwt.decode( token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm], issuer=settings.jwt_issuer, audience=settings.jwt_audience, ) subject = payload.get("sub") token_type = payload.get("type") if subject is None or token_type != ACCESS_TOKEN_TYPE: raise UnauthorizedError("Could not validate credentials") return TokenPayload(user_id=int(subject)) except (JWTError, ValueError) as exc: raise UnauthorizedError("Could not validate credentials") from exc