Add managed identity E2E tests + self-hosted CI stages (MSAL Go #641 parity)

.Pipelines/template-pipeline-stages.yml

@@ -124,6 +124,7 @@ stages:
           --ignore=tests/test_e2e.py \
           --ignore=tests/test_e2e_manual.py \
           --ignore=tests/test_fmi_e2e.py \
+          --ignore=tests/test_mi_e2e.py \
           --deselect tests/test_cryptography.py::CryptographyTestCase::test_ceiling_should_be_latest_cryptography_version_plus_three \
           --deselect tests/test_cryptography.py::CryptographyTestCase::test_should_be_run_with_latest_version_of_cryptography \
           2>&1 | tee test-results/pytest-unit.log
@@ -276,3 +277,123 @@ stages:
     - bash: rm -f "$(Agent.TempDirectory)/lab-auth.pfx"
       displayName: 'Remove lab certificate from agent'
       condition: always()
+
+# ══════════════════════════════════════════════════════════════════════════════
+# Stage 4 - MI E2E (IMDS) - REAL managed identity token acquisition on an Azure VM.
+#           Runs on the self-hosted "MISEManagedIdentity" pool, a Windows Azure VM
+#           that has the lab system-assigned + user-assigned identities assigned.
+#           Mirrors the MSAL Go "MI E2E - IMDS" stage. Only the IMDS cases in
+#           tests/test_mi_e2e.py run here (gated on MSAL_TEST_MI_IMDS); the Arc case
+#           self-skips. No lab certificate is needed - the VM's own managed identity
+#           is used.
+#
+#           Pool assumptions (self-hosted): Python 3.x on PATH and outbound access to
+#           PyPI for "pip install". Adjust if the pool differs.
+# ══════════════════════════════════════════════════════════════════════════════
+- stage: MIE2EImds
+  displayName: 'MI E2E - IMDS'
+  dependsOn: UnitTests
+  # Fork guard: never run untrusted forked-PR code on the self-hosted pool.
+  condition: and(eq(dependencies.UnitTests.result, 'Succeeded'), ne(variables['System.PullRequest.IsFork'], 'True'))
+  jobs:
+  - job: Pytest
+    displayName: 'Managed Identity E2E - VM / IMDS'
+    pool:
+      type: windows
+      isCustom: true
+      name: MISEManagedIdentity
+    timeoutInMinutes: 30
+    variables:
+      ob_outputDirectory: '$(Build.ArtifactStagingDirectory)'
+    steps:
+    # msal + its dependencies (requests, cryptography, PyJWT) and pytest are PRE-PROVISIONED on the
+    # self-hosted pool (like the Go/.NET toolchains already are), so there is no runtime pip install.
+    # This also avoids the Arc pool's egress TLS inspection blocking files.pythonhosted.org.
+    - task: PowerShell@2
+      displayName: 'Run pytest (MI E2E - IMDS)'
+      inputs:
+        targetType: 'inline'
+        workingDirectory: '$(System.DefaultWorkingDirectory)'
+        script: |
+          $ErrorActionPreference = 'Stop'
+          $py = (Get-Command python.exe -ErrorAction SilentlyContinue).Source
+          if (-not $py) {
+            $py = Get-ChildItem 'C:\Program Files\Python3*\python.exe','C:\Program Files (x86)\Python3*\python.exe','C:\Python3*\python.exe' -ErrorAction SilentlyContinue |
+              Select-Object -First 1 -ExpandProperty FullName
+          }
+          if (-not $py) { throw 'Python not found on this agent. Install Python 3.x on the pool machine.' }
+          Write-Host "Using Python: $py"
+          & $py --version
+          New-Item -ItemType Directory -Force -Path test-results | Out-Null
+          & $py -m pytest -vv --junitxml=test-results/junit-mi-e2e-imds.xml tests/test_mi_e2e.py
+      env:
+        PYTHONUNBUFFERED: '1'
+        MSAL_TEST_MI_IMDS: '1'
+
+    - task: PublishTestResults@2
+      displayName: 'Publish MI E2E (IMDS) test results'
+      condition: succeededOrFailed()
+      inputs:
+        testResultsFormat: 'JUnit'
+        testResultsFiles: 'test-results/junit-mi-e2e-imds.xml'
+        failTaskOnFailedTests: true
+        testRunTitle: 'MI E2E - IMDS'
+
+# ══════════════════════════════════════════════════════════════════════════════
+# Stage 5 - MI E2E (Azure Arc) - REAL managed identity token acquisition on an
+#           Azure Arc-enabled machine. Runs on the self-hosted "MISEAZUREARC" pool.
+#           Azure Arc supports the system-assigned identity only. Mirrors the MSAL Go
+#           "MI E2E - Azure Arc" stage. tests/test_mi_e2e.py's Arc case runs here; the
+#           IMDS cases self-skip (MSAL_TEST_MI_IMDS unset).
+#
+#           Pool note: as observed for MSAL Go, the Arc machine's egress inspection can
+#           reset TLS for package downloads. If "pip install" cannot reach PyPI, pre-provision
+#           the dependencies on the pool (or use an offline wheel cache) and drop the install step.
+# ══════════════════════════════════════════════════════════════════════════════
+- stage: MIE2EAzureArc
+  displayName: 'MI E2E - Azure Arc'
+  dependsOn: UnitTests
+  # Fork guard: never run untrusted forked-PR code on the self-hosted pool.
+  condition: and(eq(dependencies.UnitTests.result, 'Succeeded'), ne(variables['System.PullRequest.IsFork'], 'True'))
+  jobs:
+  - job: Pytest
+    displayName: 'Managed Identity E2E - Azure Arc'
+    pool:
+      type: windows
+      isCustom: true
+      name: MISEAZUREARC
+    timeoutInMinutes: 30
+    variables:
+      ob_outputDirectory: '$(Build.ArtifactStagingDirectory)'
+    steps:
+    # msal + its dependencies (requests, cryptography, PyJWT) and pytest are PRE-PROVISIONED on the
+    # self-hosted pool (like the Go/.NET toolchains already are), so there is no runtime pip install.
+    # This is required here because the Arc machine's egress TLS inspection blocks files.pythonhosted.org.
+    - task: PowerShell@2
+      displayName: 'Run pytest (MI E2E - Azure Arc)'
+      inputs:
+        targetType: 'inline'
+        workingDirectory: '$(System.DefaultWorkingDirectory)'
+        script: |
+          $ErrorActionPreference = 'Stop'
+          $py = (Get-Command python.exe -ErrorAction SilentlyContinue).Source
+          if (-not $py) {
+            $py = Get-ChildItem 'C:\Program Files\Python3*\python.exe','C:\Program Files (x86)\Python3*\python.exe','C:\Python3*\python.exe' -ErrorAction SilentlyContinue |
+              Select-Object -First 1 -ExpandProperty FullName
+          }
+          if (-not $py) { throw 'Python not found on this agent. Install Python 3.x on the pool machine.' }
+          Write-Host "Using Python: $py"
+          & $py --version
+          New-Item -ItemType Directory -Force -Path test-results | Out-Null
+          & $py -m pytest -vv --junitxml=test-results/junit-mi-e2e-arc.xml tests/test_mi_e2e.py
+      env:
+        PYTHONUNBUFFERED: '1'
+
+    - task: PublishTestResults@2
+      displayName: 'Publish MI E2E (Azure Arc) test results'
+      condition: succeededOrFailed()
+      inputs:
+        testResultsFormat: 'JUnit'
+        testResultsFiles: 'test-results/junit-mi-e2e-arc.xml'
+        failTaskOnFailedTests: true
+        testRunTitle: 'MI E2E - Azure Arc'

tests/test_mi_e2e.py

@@ -0,0 +1,133 @@
+"""End-to-end Managed Identity tests (real token acquisition).
+
+These tests perform REAL token acquisition and therefore only run on the
+self-hosted Azure DevOps pools that are actual Azure VM / Azure Arc machines with
+the lab managed identities assigned:
+
+  * IMDS tests -> the "MISEManagedIdentity" pool (an Azure VM). Gated on the
+                  MSAL_TEST_MI_IMDS environment variable, which that pipeline
+                  stage sets. (DEFAULT_TO_VM is also the fallback source on hosted
+                  agents, so an explicit flag is used instead of source detection.)
+  * Azure Arc  -> the "MISEAZUREARC" pool (an Azure Arc-enabled machine). Gated on
+                  the Azure Arc source being detected on the machine.
+
+They mirror the MSAL Go E2E tests
+(apps/tests/e2e/managedidentity_e2e_test.go and managedidentity_arc_e2e_test.go)
+and use the SAME lab identities and ARM resource, so both SDKs exercise the same
+lab configuration on the same machines.
+
+Everywhere else (hosted agents, local dev) the tests self-skip.
+"""
+import hashlib
+import os
+import unittest
+
+import requests
+
+from msal import (
+    ManagedIdentityClient,
+    SystemAssignedManagedIdentity,
+    UserAssignedManagedIdentity,
+)
+from msal.managed_identity import get_managed_identity_source, AZURE_ARC
+
+
+# Azure Resource Manager resource. Matches the ARM scope used by the MSAL .NET and
+# Go managed identity E2E tests.
+_ARM_RESOURCE = "https://management.azure.com"
+
+# User-assigned managed identities assigned to the MISEManagedIdentity VM. These are
+# the SAME values used by the MSAL Go / .NET IMDS E2E tests, so all SDKs exercise the
+# same lab configuration on the same VM.
+_UAMI_CLIENT_ID = "6325cd32-9911-41f3-819c-416cdf9104e7"
+_UAMI_OBJECT_ID = "ecb2ad92-3e30-4505-b79f-ac640d069f24"
+_UAMI_RESOURCE_ID = (
+    "/subscriptions/c1686c51-b717-4fe0-9af3-24a20a41fb0c/resourcegroups/"
+    "MSIV2-Testing-MSALNET/providers/Microsoft.ManagedIdentity/userAssignedIdentities/msiv2uami"
+)
+
+
+def _safe_error(result):
+    """Return a log-safe summary of a failed result.
+
+    Only the non-sensitive error fields are surfaced, so an assertion failure can
+    never spill an access token (or the whole result dict) into the CI logs.
+    """
+    return {
+        key: result[key]
+        for key in ("error", "error_description", "correlation_id")
+        if key in result
+    }
+
+
+def _acquire_token_twice_assert_caching(test, managed_identity):
+    """Acquire an ARM token twice for the given managed identity and assert the first
+    call reaches the identity provider while the second is served from the token cache.
+
+    Shared by the IMDS and Azure Arc E2E tests, mirroring the Go helper of the same name.
+    """
+    http_client = requests.Session()
+    client = ManagedIdentityClient(managed_identity, http_client=http_client)
+    try:
+        first = client.acquire_token_for_client(resource=_ARM_RESOURCE)
+        test.assertNotIn(
+            "error", first, "first acquisition failed: {}".format(_safe_error(first)))
+        test.assertIn("access_token", first)
+        test.assertEqual(
+            "identity_provider", first.get("token_source"),
+            "first call should reach the identity provider")
+
+        second = client.acquire_token_for_client(resource=_ARM_RESOURCE)
+        test.assertNotIn(
+            "error", second, "second acquisition failed: {}".format(_safe_error(second)))
+        test.assertIn("access_token", second)
+        test.assertEqual(
+            "cache", second.get("token_source"),
+            "second call should be served from the token cache")
+        # Compare tokens by SHA-256 digest so a mismatch never prints the actual
+        # token material into CI logs.
+        test.assertEqual(
+            hashlib.sha256(first["access_token"].encode("utf-8")).hexdigest(),
+            hashlib.sha256(second["access_token"].encode("utf-8")).hexdigest(),
+            "cached token should match the original token")
+    finally:
+        http_client.close()
+
+
+@unittest.skipUnless(
+    os.getenv("MSAL_TEST_MI_IMDS"),
+    "Set MSAL_TEST_MI_IMDS to run on the MISEManagedIdentity Azure VM (IMDS) pool")
+class ManagedIdentityImdsE2ETestCase(unittest.TestCase):
+    """Acquires ARM tokens over IMDS v1 for the system-assigned identity and each
+    user-assigned identity binding (client id / resource id / object id). Each test
+    asserts the first call reaches the identity provider and the second is cached."""
+
+    def test_system_assigned(self):
+        _acquire_token_twice_assert_caching(self, SystemAssignedManagedIdentity())
+
+    def test_user_assigned_client_id(self):
+        _acquire_token_twice_assert_caching(
+            self, UserAssignedManagedIdentity(client_id=_UAMI_CLIENT_ID))
+
+    def test_user_assigned_resource_id(self):
+        _acquire_token_twice_assert_caching(
+            self, UserAssignedManagedIdentity(resource_id=_UAMI_RESOURCE_ID))
+
+    def test_user_assigned_object_id(self):
+        _acquire_token_twice_assert_caching(
+            self, UserAssignedManagedIdentity(object_id=_UAMI_OBJECT_ID))
+
+
+@unittest.skipUnless(
+    get_managed_identity_source() == AZURE_ARC,
+    "Runs only on an Azure Arc-enabled machine (the MISEAZUREARC pool)")
+class ManagedIdentityAzureArcE2ETestCase(unittest.TestCase):
+    """Azure Arc supports the system-assigned identity only, so unlike the IMDS tests
+    there are no user-assigned variants."""
+
+    def test_system_assigned(self):
+        _acquire_token_twice_assert_caching(self, SystemAssignedManagedIdentity())
+
+
+if __name__ == "__main__":
+    unittest.main()