from datetime import datetime, timedelta, timezone from jose import JWTError, jwt from passlib.context import CryptContext from app.core.config import Settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") ALGORITHM = "HS256" TOKEN_TYPE_USER = "user" TOKEN_TYPE_ADMIN = "admin" def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def create_user_access_token(user_id: int, settings: Settings) -> str: expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes) payload = { "sub": str(user_id), "exp": expire, "type": TOKEN_TYPE_USER, } return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM) def decode_user_access_token(token: str, settings: Settings) -> int | None: try: payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM]) except JWTError: return None if payload.get("type") != TOKEN_TYPE_USER: return None subject = payload.get("sub") if not subject: return None try: return int(subject) except (TypeError, ValueError): return None def create_admin_access_token(admin_id: int, settings: Settings) -> str: expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes) payload = { "sub": str(admin_id), "exp": expire, "type": TOKEN_TYPE_ADMIN, } return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM) def decode_admin_access_token(token: str, settings: Settings) -> int | None: try: payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM]) except JWTError: return None if payload.get("type") != TOKEN_TYPE_ADMIN: return None subject = payload.get("sub") if not subject: return None try: return int(subject) except (TypeError, ValueError): return None