[Identity] Add multi-tenant live testing (#34257)

In the CI, there are credential envvars that correspond to a
multi-tenant enabled application. Here live testing is provided for that
along with corresponding recordings.

Signed-off-by: Paul Van Eck <[email protected]>

Author: pvaneck

sdk/identity/azure-identity/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "python",
   "TagPrefix": "python/identity/azure-identity",
-  "Tag": "python/identity/azure-identity_ef8f51ecfd"
+  "Tag": "python/identity/azure-identity_cb8dd6f319"
 }

sdk/identity/azure-identity/tests/conftest.py

@@ -7,10 +7,19 @@
 
 from unittest import mock
 import pytest
-from devtools_testutils import test_proxy, add_general_regex_sanitizer, is_live, add_body_key_sanitizer
+from devtools_testutils import (
+    test_proxy,
+    is_live,
+    add_general_regex_sanitizer,
+    add_body_key_sanitizer,
+    add_header_regex_sanitizer,
+    add_remove_header_sanitizer,
+    set_custom_default_matcher,
+)
 from azure.identity._constants import DEVELOPER_SIGN_ON_CLIENT_ID, EnvironmentVariables
 
 RECORD_IMDS = "--record-imds"
+TEST_ID = "00000000-0000-0000-0000-000000000000"
 
 
 def pytest_addoption(parser):
@@ -165,7 +174,11 @@ def event_loop():
 
 
 @pytest.fixture(scope="session", autouse=True)
-def add_sanitizers(test_proxy):
+def add_sanitizers(test_proxy, environment_variables):
+    set_custom_default_matcher(
+        excluded_headers="x-client-current-telemetry,x-client-last-telemetry,x-client-os,"
+        "x-client-sku,x-client-ver,x-client-cpu,x-client-brkrver,x-ms-lib-capability"  # cspell:ignore brkrver
+    )
     if EnvironmentVariables.MSI_ENDPOINT in os.environ:
         url = os.environ.get(EnvironmentVariables.MSI_ENDPOINT)
         PLAYBACK_URL = "https://msi-endpoint/token"
@@ -190,6 +203,17 @@ def add_sanitizers(test_proxy):
         add_general_regex_sanitizer(regex=os.environ["OBO_USERNAME"], value="username")
     add_body_key_sanitizer(json_path="$..access_token", value="access_token")
 
+    # Multi-tenant environment variables sanitization
+    sanitization_mapping = {
+        "AZURE_IDENTITY_MULTI_TENANT_TENANT_ID": TEST_ID,
+        "AZURE_IDENTITY_MULTI_TENANT_CLIENT_ID": TEST_ID,
+        "AZURE_IDENTITY_MULTI_TENANT_CLIENT_SECRET": TEST_ID,
+    }
+    environment_variables.sanitize_batch(sanitization_mapping)
+    add_header_regex_sanitizer(key="Set-Cookie", value="[set-cookie;]")
+    add_remove_header_sanitizer(headers="Cookie")
+    add_header_regex_sanitizer(key="client-request-id", value="sanitized")
+
 
 @pytest.fixture(scope="session", autouse=True)
 def patch_async_sleep():

sdk/identity/azure-identity/tests/test_environment_credential.py

@@ -34,19 +34,19 @@ def test_incomplete_configuration():
 
 
 @pytest.mark.parametrize(
-    "credential_name,environment_variables",
+    "credential_name,envvars",
     (
         ("ClientSecretCredential", EnvironmentVariables.CLIENT_SECRET_VARS),
         ("CertificateCredential", EnvironmentVariables.CERT_VARS),
         ("UsernamePasswordCredential", EnvironmentVariables.USERNAME_PASSWORD_VARS),
     ),
 )
-def test_passes_authority_argument(credential_name, environment_variables):
+def test_passes_authority_argument(credential_name, envvars):
     """the credential pass the 'authority' keyword argument to its inner credential"""
 
     authority = "authority"
 
-    with mock.patch.dict("os.environ", {variable: "foo" for variable in environment_variables}, clear=True):
+    with mock.patch.dict("os.environ", {variable: "foo" for variable in envvars}, clear=True):
         with mock.patch(EnvironmentCredential.__module__ + "." + credential_name) as mock_credential:
             EnvironmentCredential(authority=authority)
 

sdk/identity/azure-identity/tests/test_environment_credential_async.py

@@ -70,18 +70,18 @@ async def test_incomplete_configuration():
 
 
 @pytest.mark.parametrize(
-    "credential_name,environment_variables",
+    "credential_name,envvars",
     (
         ("ClientSecretCredential", EnvironmentVariables.CLIENT_SECRET_VARS),
         ("CertificateCredential", EnvironmentVariables.CERT_VARS),
     ),
 )
-def test_passes_authority_argument(credential_name, environment_variables):
+def test_passes_authority_argument(credential_name, envvars):
     """the credential pass the 'authority' keyword argument to its inner credential"""
 
     authority = "authority"
 
-    with mock.patch.dict(ENVIRON, {variable: "foo" for variable in environment_variables}, clear=True):
+    with mock.patch.dict(ENVIRON, {variable: "foo" for variable in envvars}, clear=True):
         with mock.patch(EnvironmentCredential.__module__ + "." + credential_name) as mock_credential:
             EnvironmentCredential(authority=authority)
 

sdk/identity/azure-identity/tests/test_multi_tenant_auth.py

@@ -0,0 +1,35 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See LICENSE.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+import os
+
+import pytest
+from devtools_testutils import AzureRecordedTestCase, is_live
+from azure.core import PipelineClient
+from azure.core.rest import HttpRequest, HttpResponse
+from azure.identity import ClientSecretCredential
+
+
+class TestMultiTenantAuth(AzureRecordedTestCase):
+    def _send_request(self, credential: ClientSecretCredential) -> HttpResponse:
+        client = PipelineClient(base_url="https://graph.microsoft.com")
+        token = credential.get_token("https://graph.microsoft.com/.default")
+        headers = {"Authorization": "Bearer " + token.token, "ConsistencyLevel": "eventual"}
+        request = HttpRequest("GET", "https://graph.microsoft.com/v1.0/applications/$count", headers=headers)
+        response = client.send_request(request)
+        return response
+
+    @pytest.mark.skipif(
+        is_live() and not os.environ.get("AZURE_IDENTITY_MULTI_TENANT_CLIENT_ID"),
+        reason="Multi-tenant envvars not configured.",
+    )
+    def test_multi_tenant_client_secret_graph_call(self, recorded_test, environment_variables):
+        client_id = environment_variables.get("AZURE_IDENTITY_MULTI_TENANT_CLIENT_ID")
+        tenant_id = environment_variables.get("AZURE_IDENTITY_MULTI_TENANT_TENANT_ID")
+        client_secret = environment_variables.get("AZURE_IDENTITY_MULTI_TENANT_CLIENT_SECRET")
+        credential = ClientSecretCredential(tenant_id, client_id, client_secret)
+        response = self._send_request(credential)
+        assert response.status_code == 200
+        assert int(response.text()) > 0

sdk/identity/azure-identity/tests/test_multi_tenant_auth_async.py

@@ -0,0 +1,37 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See LICENSE.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+import os
+
+import pytest
+from devtools_testutils import AzureRecordedTestCase, is_live
+from azure.core import AsyncPipelineClient
+from azure.core.rest import HttpRequest, HttpResponse
+from azure.identity.aio import ClientSecretCredential
+
+
+class TestMultiTenantAuthAsync(AzureRecordedTestCase):
+    async def _send_request(self, credential: ClientSecretCredential) -> HttpResponse:
+        client = AsyncPipelineClient(base_url="https://graph.microsoft.com")
+        token = await credential.get_token("https://graph.microsoft.com/.default")
+        headers = {"Authorization": "Bearer " + token.token, "ConsistencyLevel": "eventual"}
+        request = HttpRequest("GET", "https://graph.microsoft.com/v1.0/applications/$count", headers=headers)
+        response = await client.send_request(request, stream=False)
+        return response
+
+    @pytest.mark.asyncio
+    @pytest.mark.skipif(
+        is_live() and not os.environ.get("AZURE_IDENTITY_MULTI_TENANT_CLIENT_ID"),
+        reason="Multi-tenant envvars not configured.",
+    )
+    async def test_multi_tenant_client_secret_graph_call(self, recorded_test, environment_variables):
+        client_id = environment_variables.get("AZURE_IDENTITY_MULTI_TENANT_CLIENT_ID")
+        tenant_id = environment_variables.get("AZURE_IDENTITY_MULTI_TENANT_TENANT_ID")
+        client_secret = environment_variables.get("AZURE_IDENTITY_MULTI_TENANT_CLIENT_SECRET")
+        credential = ClientSecretCredential(tenant_id, client_id, client_secret)
+        async with credential:
+            response = await self._send_request(credential)
+            assert response.status_code == 200
+            assert int(response.text()) > 0