69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
from dataclasses import dataclass
|
|
from time import monotonic
|
|
from urllib.parse import urljoin
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
|
|
|
|
class OAuth2ConfigurationError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class OAuth2Token:
|
|
access_token: str
|
|
expires_at: float
|
|
token_type: str = "Bearer"
|
|
|
|
def is_valid(self) -> bool:
|
|
return bool(self.access_token) and monotonic() < self.expires_at
|
|
|
|
|
|
class OAuth2ClientCredentials:
|
|
def __init__(self):
|
|
self._token = None
|
|
|
|
@property
|
|
def token_url(self):
|
|
provider_private_url = settings.OAUTH2_PROVIDER_PRIVATE_URL.rstrip("/")
|
|
if not provider_private_url:
|
|
raise OAuth2ConfigurationError("OAUTH2_PROVIDER_PRIVATE_URL is required.")
|
|
return urljoin(f"{provider_private_url}/", "apis/applications/")
|
|
|
|
def get_access_token(self):
|
|
if self._token and self._token.is_valid():
|
|
return self._token.access_token
|
|
|
|
if not settings.OAUTH2_CLIENT_ID or not settings.OAUTH2_CLIENT_SECRET:
|
|
raise OAuth2ConfigurationError(
|
|
"OAUTH2_CLIENT_ID and OAUTH2_CLIENT_SECRET are required."
|
|
)
|
|
|
|
response = requests.post(
|
|
self.token_url,
|
|
data={
|
|
"grant_type": "client_credentials",
|
|
"scope": settings.OAUTH2_SCOPES,
|
|
},
|
|
auth=(settings.OAUTH2_CLIENT_ID, settings.OAUTH2_CLIENT_SECRET),
|
|
timeout=settings.SERVICE_REQUEST_TIMEOUT,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
|
|
access_token = payload.get("access_token")
|
|
if not access_token:
|
|
raise OAuth2ConfigurationError("OAuth2 response did not include access_token.")
|
|
|
|
expires_in = int(payload.get("expires_in", 3600))
|
|
token_type = payload.get("token_type", "Bearer")
|
|
self._token = OAuth2Token(
|
|
access_token=access_token,
|
|
expires_at=monotonic() + max(expires_in - 60, 1),
|
|
token_type=token_type,
|
|
)
|
|
return self._token.access_token
|
|
|
|
def authorization_header(self):
|
|
return {"Authorization": f"Bearer {self.get_access_token()}"}
|