Update to Graph SDK

infra/main.bicep

@@ -195,4 +195,5 @@ module keyVaultSecrets './core/security/keyvault-secret.bicep' = [for secret in
   }
 }]
 
-output SERVICE_APP_URI string = containerApp.outputs.uri
+output SERVICE_APP_URI string = containerApp.outputs.uri
+output AZURE_KEY_VAULT_NAME string = keyVault.outputs.name

pyproject.toml

@@ -0,0 +1,6 @@
+[tool.ruff]
+line-length = 120
+target-version = "py311"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP"]

requirements.txt

@@ -1,2 +1,2 @@
-urllib3
-azure-identity
+azure-identity
+msgraph-sdk

scripts/auth_init.py

@@ -1,89 +1,80 @@
+import asyncio
 import os
 import subprocess
 
-import urllib3
 from azure.identity import AzureDeveloperCliCredential
-
-
-def get_auth_headers(credential):
-    return {
-        "Authorization": "Bearer "
-        + credential.get_token("https://graph.microsoft.com/.default").token
-    }
-
-
-def check_for_application(credential, app_id):
-    resp = urllib3.request(
-        "GET",
-        f"https://graph.microsoft.com/v1.0/applications/{app_id}",
-        headers=get_auth_headers(credential),
-    )
-    if resp.status != 200:
-        print("Application not found")
+from kiota_abstractions.api_error import APIError
+from msgraph import GraphServiceClient
+from msgraph.generated.applications.item.add_password.add_password_post_request_body import (
+    AddPasswordPostRequestBody,
+)
+from msgraph.generated.models.application import Application
+from msgraph.generated.models.implicit_grant_settings import ImplicitGrantSettings
+from msgraph.generated.models.password_credential import PasswordCredential
+from msgraph.generated.models.web_application import WebApplication
+
+
+async def check_for_application(client: GraphServiceClient, app_id: str) -> bool:
+    try:
+        await client.applications.by_application_id(app_id).get()
+    except APIError:
         return False
     return True
 
 
-def create_application(credential):
-    resp = urllib3.request(
-        "POST",
-        "https://graph.microsoft.com/v1.0/applications",
-        headers=get_auth_headers(credential),
-        json={
-            "displayName": "WebApp",
-            "signInAudience": "AzureADandPersonalMicrosoftAccount",
-            "web": {
-                "redirectUris": ["http://localhost:5000/.auth/login/aad/callback"],
-                "implicitGrantSettings": {"enableIdTokenIssuance": True},
-            },
-        },
-        timeout=urllib3.Timeout(connect=10, read=10),
+async def create_application(client: GraphServiceClient) -> Application:
+    request_body = Application(
+        display_name="WebApp",
+        sign_in_audience="AzureADandPersonalMicrosoftAccount",
+        web=WebApplication(
+            redirect_uris=["http://localhost:5000/.auth/login/aad/callback"],
+            implicit_grant_settings=ImplicitGrantSettings(enable_id_token_issuance=True),
+        ),
     )
+    return await client.applications.post(request_body)
 
-    app_id = resp.json()["id"]
-    client_id = resp.json()["appId"]
-
-    return app_id, client_id
 
-
-def add_client_secret(credential, app_id):
-    resp = urllib3.request(
-        "POST",
-        f"https://graph.microsoft.com/v1.0/applications/{app_id}/addPassword",
-        headers=get_auth_headers(credential),
-        json={"passwordCredential": {"displayName": "WebAppSecret"}},
-        timeout=urllib3.Timeout(connect=10, read=10),
+async def add_client_secret(client: GraphServiceClient, app_id: str) -> str:
+    request_body = AddPasswordPostRequestBody(
+        password_credential=PasswordCredential(display_name="WebAppSecret"),
     )
-    client_secret = resp.json()["secretText"]
-    return client_secret
+    result = await client.applications.by_application_id(app_id).add_password.post(request_body)
+    return result.secret_text
 
 
 def update_azd_env(name, val):
     subprocess.run(f"azd env set {name} {val}", shell=True)
 
 
-if __name__ == "__main__":
+async def main():
     if os.getenv("AZURE_USE_AUTHENTICATION", "false") != "true":
         print("AZURE_USE_AUTHENTICATION is false, not setting up authentication")
         exit(0)
 
     print("AZURE_USE_AUTHENTICATION is true, setting up authentication...")
     credential = AzureDeveloperCliCredential(tenant_id=os.getenv("AZURE_AUTH_TENANT_ID"))
 
+    scopes = ["https://graph.microsoft.com/.default"]
+    client = GraphServiceClient(credentials=credential, scopes=scopes)
+
     app_id = os.getenv("AZURE_AUTH_APP_ID", "no-id")
     if app_id != "no-id":
         print(f"Checking if application {app_id} exists")
-        if check_for_application(credential, app_id):
+        if await check_for_application(client, app_id):
             print("Application already exists, not creating new one")
             exit(0)
 
     print("Creating application registration")
-    app_id, client_id = create_application(credential)
+    app = await create_application(client)
 
-    print(f"Adding client secret to {app_id}")
-    client_secret = add_client_secret(credential, app_id)
+    print(f"Adding client secret to {app.id}")
+    client_secret = await add_client_secret(client, app.id)
 
     print("Updating azd env with AZURE_AUTH_APP_ID, AZURE_AUTH_CLIENT_ID, AZURE_AUTH_CLIENT_SECRET")
-    update_azd_env("AZURE_AUTH_APP_ID", app_id)
-    update_azd_env("AZURE_AUTH_CLIENT_ID", client_id)
-    update_azd_env("AZURE_AUTH_CLIENT_SECRET", client_secret)
+    update_azd_env("AZURE_AUTH_APP_ID", app.id)
+    update_azd_env("AZURE_AUTH_CLIENT_ID", app.app_id)
+    update_azd_env("AZURE_AUTH_CLIENT_SECRET", client_secret)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

scripts/auth_update.py

@@ -1,37 +1,39 @@
+import asyncio
 import os
 
-import urllib3
 from azure.identity import AzureDeveloperCliCredential
+from msgraph import GraphServiceClient
+from msgraph.generated.models.application import Application
+from msgraph.generated.models.web_application import WebApplication
 
 
-def update_redirect_uris(credential, app_id, uri):
-    urllib3.request(
-        "PATCH",
-        f"https://graph.microsoft.com/v1.0/applications/{app_id}",
-        headers={
-            "Authorization": "Bearer "
-            + credential.get_token("https://graph.microsoft.com/.default").token,
-        },
-        json={
-            "web": {
-                "redirectUris": [
-                    "http://localhost:3000/api/auth/callback/azure-ad",
-                    f"{uri}/api/auth/callback/azure-ad",
-                ]
-            }
-        },
+async def update_redirect_uris(client: GraphServiceClient, app_id: str, uri: str):
+    request_body = Application(
+        web=WebApplication(
+            redirect_uris=[
+                "http://localhost:3000/api/auth/callback/azure-ad",
+                f"{uri}/api/auth/callback/azure-ad",
+            ]
+        ),
     )
+    await client.applications.by_application_id(app_id).patch(request_body)
 
 
-if __name__ == "__main__":
+async def main():
     if os.getenv("AZURE_USE_AUTHENTICATION", "false") != "true":
         print("AZURE_USE_AUTHENTICATION is false, not updating authentication")
         exit(0)
 
     print("AZURE_USE_AUTHENTICATION is true, updating authentication...")
     credential = AzureDeveloperCliCredential(tenant_id=os.getenv("AZURE_AUTH_TENANT_ID"))
+    scopes = ["https://graph.microsoft.com/.default"]
+    client = GraphServiceClient(credentials=credential, scopes=scopes)
 
     app_id = os.getenv("AZURE_AUTH_APP_ID")
     uri = os.getenv("SERVICE_APP_URI")
     print(f"Updating application registration {app_id} with redirect URI for {uri}")
-    update_redirect_uris(credential, app_id, uri)
+    await update_redirect_uris(client, app_id, uri)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())