58 lines
1.5 KiB
Python
58 lines
1.5 KiB
Python
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
|
|
|