71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
import hmac
|
|
|
|
from fastapi import HTTPException, Request, status
|
|
|
|
from crypto_spot_bot.config import Settings
|
|
|
|
|
|
class ApiAuthorizer:
|
|
"""Authenticate API calls either directly or through an authenticated proxy."""
|
|
|
|
def __init__(self, settings: Settings):
|
|
self.settings = settings
|
|
|
|
async def require(self, request: Request) -> None:
|
|
if self._proxy_authenticated(request) or self._token_authenticated(
|
|
request, self.settings.api_auth_token
|
|
):
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="API authentication required",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
async def require_training(self, request: Request) -> None:
|
|
expected = self.settings.training_worker_token or self.settings.api_auth_token
|
|
if self._proxy_authenticated(request) or self._token_authenticated(request, expected):
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="training worker authentication required",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
def configured(self) -> bool:
|
|
return bool(
|
|
self.settings.api_auth_token
|
|
or self.settings.training_worker_token
|
|
or self.settings.trusted_proxy_user_header
|
|
)
|
|
|
|
def _proxy_authenticated(self, request: Request) -> bool:
|
|
header = self.settings.trusted_proxy_user_header
|
|
if not header:
|
|
return False
|
|
return bool(request.headers.get(header, "").strip())
|
|
|
|
def _token_authenticated(self, request: Request, expected: str) -> bool:
|
|
if not expected:
|
|
return False
|
|
candidates = [request.headers.get("X-TradeBot-Token", "").strip()]
|
|
authorization = request.headers.get("Authorization", "").strip()
|
|
if authorization.lower().startswith("bearer "):
|
|
candidates.append(authorization[7:].strip())
|
|
elif authorization.lower().startswith("basic "):
|
|
decoded = _decode_basic(authorization[6:].strip())
|
|
if decoded:
|
|
candidates.append(decoded)
|
|
return any(candidate and hmac.compare_digest(candidate, expected) for candidate in candidates)
|
|
|
|
|
|
def _decode_basic(value: str) -> str:
|
|
try:
|
|
return base64.b64decode(value, validate=True).decode("utf-8")
|
|
except (binascii.Error, UnicodeDecodeError, ValueError):
|
|
return ""
|