This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
|
||||
class Microsoft365OAuthError(RuntimeError):
|
||||
"""Safe OAuth boundary error that never includes tokens or client secrets."""
|
||||
|
||||
|
||||
def acquire_exchange_access_token(*, client_factory=None) -> str:
|
||||
tenant_id = settings.M365_TENANT_ID.strip()
|
||||
client_id = settings.M365_CLIENT_ID.strip()
|
||||
client_secret = settings.M365_CLIENT_SECRET.strip()
|
||||
if not all((tenant_id, client_id, client_secret)):
|
||||
raise Microsoft365OAuthError("Microsoft 365 OAuth is niet volledig geconfigureerd.")
|
||||
|
||||
if client_factory is None:
|
||||
import msal
|
||||
|
||||
client_factory = msal.ConfidentialClientApplication
|
||||
factory = client_factory
|
||||
client = factory(
|
||||
client_id,
|
||||
authority=f"https://login.microsoftonline.com/{tenant_id}",
|
||||
client_credential=client_secret,
|
||||
)
|
||||
result = client.acquire_token_for_client(scopes=[settings.M365_EXCHANGE_SCOPE])
|
||||
token = result.get("access_token") if isinstance(result, dict) else None
|
||||
if not isinstance(token, str) or not token:
|
||||
category = "unknown"
|
||||
if isinstance(result, dict):
|
||||
category = str(result.get("error") or "unknown")[:80]
|
||||
raise Microsoft365OAuthError(
|
||||
f"Microsoft 365 OAuth-token kon niet worden verkregen ({category})."
|
||||
)
|
||||
return token
|
||||
|
||||
|
||||
def xoauth2_bytes(username: str, access_token: str) -> bytes:
|
||||
if not username or "\x00" in username or "\r" in username or "\n" in username:
|
||||
raise Microsoft365OAuthError("Ongeldig Microsoft 365-mailboxadres.")
|
||||
return f"user={username}\x01auth=Bearer {access_token}\x01\x01".encode()
|
||||
|
||||
|
||||
def xoauth2_b64(username: str, access_token: str) -> str:
|
||||
return base64.b64encode(xoauth2_bytes(username, access_token)).decode("ascii")
|
||||
Reference in New Issue
Block a user