69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import Depends, Header, HTTPException, status
|
|
import jwt
|
|
|
|
from api.config import settings
|
|
from api.tokens import decode_access_token
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AuthUser:
|
|
user_id: int
|
|
first_name: str | None = None
|
|
last_name: str | None = None
|
|
username: str | None = None
|
|
|
|
|
|
def _extract_bearer(authorization: str | None) -> str | None:
|
|
if not authorization:
|
|
return None
|
|
if authorization.lower().startswith("bearer "):
|
|
return authorization[7:].strip()
|
|
return None
|
|
|
|
|
|
async def require_api_token(
|
|
authorization: str | None = Header(default=None),
|
|
x_api_token: str | None = Header(default=None, alias="X-API-Token"),
|
|
) -> None:
|
|
if not settings.api_token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail="Admin API_TOKEN is not configured",
|
|
)
|
|
token = (x_api_token or "").strip() or _extract_bearer(authorization)
|
|
if not token or token != settings.api_token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or missing API token",
|
|
)
|
|
|
|
|
|
async def require_user(
|
|
authorization: str | None = Header(default=None),
|
|
) -> AuthUser:
|
|
token = _extract_bearer(authorization)
|
|
if not token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Authorization Bearer token required",
|
|
)
|
|
try:
|
|
payload = decode_access_token(token)
|
|
user_id = int(payload["uid"])
|
|
except (jwt.PyJWTError, KeyError, TypeError, ValueError) as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired session",
|
|
) from exc
|
|
|
|
return AuthUser(
|
|
user_id=user_id,
|
|
first_name=payload.get("fn"),
|
|
last_name=payload.get("ln"),
|
|
username=payload.get("un"),
|
|
)
|