import base64 from datetime import timedelta import hashlib import hmac import json import time from typing import Any from app.config import settings from app.timezone import now def _b64encode(raw: bytes) -> str: return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") def _b64decode(raw: str) -> bytes: padding = "=" * (-len(raw) % 4) return base64.urlsafe_b64decode(raw + padding) def create_token(payload: dict[str, Any]) -> str: expires_at = now() + timedelta(minutes=settings.token_expire_minutes) body = { **payload, "exp": int(expires_at.timestamp()), "iat": int(time.time()), } encoded_body = _b64encode(json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode()) signature = hmac.new( settings.jwt_secret.encode(), encoded_body.encode(), hashlib.sha256, ).digest() return f"{encoded_body}.{_b64encode(signature)}" def verify_token(token: str) -> dict[str, Any]: try: encoded_body, encoded_signature = token.split(".", 1) except ValueError as exc: raise ValueError("invalid token") from exc expected = hmac.new( settings.jwt_secret.encode(), encoded_body.encode(), hashlib.sha256, ).digest() actual = _b64decode(encoded_signature) if not hmac.compare_digest(expected, actual): raise ValueError("invalid token") payload = json.loads(_b64decode(encoded_body)) if payload.get("exp", 0) < int(time.time()): raise ValueError("token expired") return payload