103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
"""
|
|
Authentication module for Microsoft Graph API.
|
|
"""
|
|
|
|
import os
|
|
import msal
|
|
import logging
|
|
from rich import print
|
|
from rich.panel import Panel
|
|
|
|
# Comprehensive logging suppression for authentication-related libraries
|
|
logging.getLogger("msal").setLevel(logging.ERROR)
|
|
logging.getLogger("urllib3").setLevel(logging.ERROR)
|
|
logging.getLogger("requests").setLevel(logging.ERROR)
|
|
logging.getLogger("requests_oauthlib").setLevel(logging.ERROR)
|
|
logging.getLogger("aiohttp").setLevel(logging.ERROR)
|
|
logging.getLogger("aiohttp.access").setLevel(logging.ERROR)
|
|
logging.getLogger("asyncio").setLevel(logging.ERROR)
|
|
logging.getLogger("azure").setLevel(logging.ERROR)
|
|
logging.getLogger("azure.core").setLevel(logging.ERROR)
|
|
|
|
|
|
def ensure_directory_exists(path):
|
|
if not os.path.exists(path):
|
|
os.makedirs(path)
|
|
|
|
|
|
def get_access_token(scopes):
|
|
"""
|
|
Authenticate with Microsoft Graph API and obtain an access token.
|
|
|
|
Args:
|
|
scopes (list): List of scopes to request.
|
|
|
|
Returns:
|
|
tuple: (access_token, headers) where access_token is the token string
|
|
and headers is a dict with Authorization header.
|
|
|
|
Raises:
|
|
ValueError: If environment variables are missing.
|
|
Exception: If authentication fails.
|
|
"""
|
|
# Read Azure app credentials from environment variables
|
|
client_id = os.getenv("AZURE_CLIENT_ID")
|
|
tenant_id = os.getenv("AZURE_TENANT_ID")
|
|
|
|
if not client_id or not tenant_id:
|
|
raise ValueError(
|
|
"Please set the AZURE_CLIENT_ID and AZURE_TENANT_ID environment variables."
|
|
)
|
|
|
|
# Token cache
|
|
cache = msal.SerializableTokenCache()
|
|
cache_file = "token_cache.bin"
|
|
|
|
if os.path.exists(cache_file):
|
|
cache.deserialize(open(cache_file, "r").read())
|
|
|
|
# Authentication
|
|
authority = f"https://login.microsoftonline.com/{tenant_id}"
|
|
app = msal.PublicClientApplication(
|
|
client_id, authority=authority, token_cache=cache
|
|
)
|
|
accounts = app.get_accounts()
|
|
|
|
if accounts:
|
|
token_response = app.acquire_token_silent(scopes, account=accounts[0])
|
|
else:
|
|
flow = app.initiate_device_flow(scopes=scopes)
|
|
if "user_code" not in flow:
|
|
raise Exception("Failed to create device flow")
|
|
|
|
print(
|
|
Panel(
|
|
flow["message"],
|
|
border_style="magenta",
|
|
padding=2,
|
|
title="MSAL Login Flow Link",
|
|
)
|
|
)
|
|
|
|
token_response = app.acquire_token_by_device_flow(flow)
|
|
|
|
if token_response is None:
|
|
raise Exception("Token response is None - authentication failed")
|
|
|
|
if "access_token" not in token_response:
|
|
error_description = token_response.get("error_description", "Unknown error")
|
|
error_code = token_response.get("error", "unknown_error")
|
|
raise Exception(f"Failed to acquire token - {error_code}: {error_description}")
|
|
|
|
# Save token cache
|
|
with open(cache_file, "w") as f:
|
|
f.write(cache.serialize())
|
|
|
|
access_token = token_response["access_token"]
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Prefer": 'outlook.body-content-type="text",IdType="ImmutableId"',
|
|
}
|
|
|
|
return access_token, headers
|