test(mi-e2e): redact tokens in logs, close session, fork-guard self-hosted stages

Address PR review feedback:
- Assert access_token present on the cached (second) acquisition before indexing.
- Never log token material: summarize error results to safe fields only, and
  compare the cached token to the original via SHA-256 digest instead of raw values.
- Close the requests.Session in a finally block.
- Add a stage-level fork guard so forked-PR code never runs on the self-hosted
  MISEManagedIdentity / MISEAZUREARC pools.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 5ad57850-7d1f-4b9a-bf9f-723d69a9687c

Author: gladjohn

.Pipelines/template-pipeline-stages.yml

@@ -293,7 +293,8 @@ stages:
 - stage: MIE2EImds
   displayName: 'MI E2E - IMDS'
   dependsOn: UnitTests
-  condition: eq(dependencies.UnitTests.result, 'Succeeded')
+  # 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'
@@ -352,7 +353,8 @@ stages:
 - stage: MIE2EAzureArc
   displayName: 'MI E2E - Azure Arc'
   dependsOn: UnitTests
-  condition: eq(dependencies.UnitTests.result, 'Succeeded')
+  # 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'

tests/test_mi_e2e.py

@@ -18,6 +18,7 @@
 
 Everywhere else (hosted agents, local dev) the tests self-skip.
 """
+import hashlib
 import os
 import unittest
 
@@ -46,29 +47,51 @@
 )
 
 
+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.
     """
-    client = ManagedIdentityClient(managed_identity, http_client=requests.Session())
-
-    first = client.acquire_token_for_client(resource=_ARM_RESOURCE)
-    test.assertNotIn("error", first, "first acquisition failed: {}".format(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(second))
-    test.assertEqual(
-        "cache", second.get("token_source"),
-        "second call should be served from the token cache")
-    test.assertEqual(
-        first["access_token"], second["access_token"],
-        "cached token should match the original token")
+    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(