add default scopes for known clouds

Author: chlowell

sdk/identity/azure-identity/azure/identity/_internal/__init__.py

@@ -61,6 +61,7 @@ def _scopes_to_resource(*scopes):
     "InteractiveCredential",
     "MsalTransportAdapter",
     "MsalTransportResponse",
+    "normalize_authority",
     "PublicClientCredential",
     "wrap_exceptions",
 ]

sdk/identity/azure-identity/azure/identity/_internal/msal_credentials.py

@@ -20,7 +20,8 @@
 
 from .exception_wrapper import wrap_exceptions
 from .msal_transport_adapter import MsalTransportAdapter
-from .._exceptions import AuthenticationRequiredError
+from .._constants import KnownAuthorities
+from .._exceptions import AuthenticationRequiredError, CredentialUnavailableError
 from .._internal import get_default_authority, normalize_authority
 from .._auth_record import AuthenticationRecord
 
@@ -41,6 +42,13 @@
 
 _LOGGER = logging.getLogger(__name__)
 
+_DEFAULT_AUTHENTICATE_SCOPES = {
+    "https://" + KnownAuthorities.AZURE_CHINA: ("https://management.core.chinacloudapi.cn//.default",),
+    "https://" + KnownAuthorities.AZURE_GERMANY: ("https://management.core.cloudapi.de//.default",),
+    "https://" + KnownAuthorities.AZURE_GOVERNMENT: ("https://management.core.usgovcloudapi.net//.default",),
+    "https://" + KnownAuthorities.AZURE_PUBLIC_CLOUD: ("https://management.core.windows.net//.default",),
+}
+
 
 def _decode_client_info(raw):
     """Taken from msal.oauth2cli.oidc"""
@@ -91,11 +99,9 @@ class MsalCredential(ABC):
 
     def __init__(self, client_id, client_credential=None, **kwargs):
         # type: (str, Optional[Union[str, Mapping[str, str]]], **Any) -> None
-        tenant_id = kwargs.pop("tenant_id", None) or "organizations"
         authority = kwargs.pop("authority", None)
-        authority = normalize_authority(authority) if authority else get_default_authority()
-
-        self._base_url = "/".join((authority, tenant_id.strip("/")))
+        self._authority = normalize_authority(authority) if authority else get_default_authority()
+        self._tenant_id = kwargs.pop("tenant_id", None) or "organizations"
 
         self._client_credential = client_credential
         self._client_id = client_id
@@ -130,7 +136,7 @@ def _create_app(self, cls):
             app = cls(
                 client_id=self._client_id,
                 client_credential=self._client_credential,
-                authority=self._base_url,
+                authority="{}/{}".format(self._authority, self._tenant_id),
                 token_cache=self._cache,
             )
 
@@ -242,15 +248,24 @@ def authenticate(self, **kwargs):
         # type: (**Any) -> AuthenticationRecord
         """Interactively authenticate a user.
 
-        :keyword Sequence[str] scopes: optional scopes to request during authentication, such as those provided by
+        :keyword Sequence[str] scopes: scopes to request during authentication, such as those provided by
           :func:`AuthenticationRequiredError.scopes`. If provided, successful authentication will cache an access token
           for these scopes.
         :rtype: ~azure.identity.AuthenticationRecord
         :raises ~azure.core.exceptions.ClientAuthenticationError: authentication failed. The error's ``message``
           attribute gives a reason.
         """
 
-        scopes = kwargs.pop("scopes", None) or ("https://management.azure.com/.default",)
+        scopes = kwargs.pop("scopes", None)
+        if not scopes:
+            if self._authority not in _DEFAULT_AUTHENTICATE_SCOPES:
+                # the credential is configured to use a cloud whose ARM scope we can't determine
+                raise CredentialUnavailableError(
+                    message="Authenticating in this environment requires a value for the 'scopes' keyword argument."
+                )
+
+            scopes = _DEFAULT_AUTHENTICATE_SCOPES[self._authority]
+
         _ = self.get_token(*scopes, _allow_prompt=True, **kwargs)
         return self.authentication_record  # type: ignore
 

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

@@ -142,6 +142,34 @@ def validate_scopes(*scopes, **_):
     assert request_token.call_count == 1, "validation method wasn't called"
 
 
+@pytest.mark.parametrize(
+    "authority,expected_scope",
+    (
+        (KnownAuthorities.AZURE_CHINA, "https://management.core.chinacloudapi.cn//.default"),
+        (KnownAuthorities.AZURE_GERMANY, "https://management.core.cloudapi.de//.default"),
+        (KnownAuthorities.AZURE_GOVERNMENT, "https://management.core.usgovcloudapi.net//.default"),
+        (KnownAuthorities.AZURE_PUBLIC_CLOUD, "https://management.core.windows.net//.default"),
+    ),
+)
+def test_authenticate_default_scopes(authority, expected_scope):
+    """when given no scopes, authenticate should default to the ARM scope appropriate for the configured authority"""
+
+    def validate_scopes(*scopes):
+        assert scopes == (expected_scope,)
+        return {"access_token": "**", "expires_in": 42}
+
+    request_token = Mock(wraps=validate_scopes)
+    MockCredential(authority=authority, request_token=request_token).authenticate()
+    assert request_token.call_count == 1
+
+
+def test_authenticate_unknown_cloud():
+    """authenticate should raise when given no scopes in an unknown cloud"""
+
+    with pytest.raises(CredentialUnavailableError):
+        MockCredential(authority="localhost").authenticate()
+
+
 @pytest.mark.parametrize("option", (True, False))
 def test_authenticate_ignores_disable_automatic_authentication(option):
     """authenticate should prompt for authentication regardless of the credential's configuration"""