ClientSecretCredential uses AadClient

Author: chlowell

sdk/identity/azure-identity/azure/identity/_credentials/client_secret.py

@@ -2,8 +2,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-from .._authn_client import AuthnClient
-from .._base import ClientSecretCredentialBase
+from .._internal import AadClient, ClientSecretCredentialBase
 
 try:
     from typing import TYPE_CHECKING
@@ -28,12 +27,7 @@ class ClientSecretCredential(ClientSecretCredentialBase):
           defines authorities for other clouds.
     """
 
-    def __init__(self, tenant_id, client_id, client_secret, **kwargs):
-        # type: (str, str, str, **Any) -> None
-        super(ClientSecretCredential, self).__init__(tenant_id, client_id, client_secret, **kwargs)
-        self._client = AuthnClient(tenant=tenant_id, **kwargs)
-
-    def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
+    def get_token(self, *scopes, **kwargs):
         # type: (*str, **Any) -> AccessToken
         """Request an access token for `scopes`.
 
@@ -48,8 +42,10 @@ def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
         if not scopes:
             raise ValueError("'get_token' requires at least one scope")
 
-        token = self._client.get_cached_token(scopes)
+        token = self._client.get_cached_access_token(scopes)
         if not token:
-            data = dict(self._form_data, scope=" ".join(scopes))
-            token = self._client.request_token(scopes, form_data=data)
+            token = self._client.obtain_token_by_client_secret(scopes, self._secret, **kwargs)
         return token
+
+    def _get_auth_client(self, tenant_id, client_id, **kwargs):
+        return AadClient(tenant_id, client_id, **kwargs)

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

@@ -35,6 +35,7 @@ def get_default_authority():
 from .auth_code_redirect_handler import AuthCodeRedirectServer
 from .aadclient_certificate import AadClientCertificate
 from .certificate_credential_base import CertificateCredentialBase
+from .client_secret_credential_base import ClientSecretCredentialBase
 from .exception_wrapper import wrap_exceptions
 from .msal_credentials import ConfidentialClientCredential, InteractiveCredential, PublicClientCredential
 from .msal_transport_adapter import MsalTransportAdapter, MsalTransportResponse
@@ -60,6 +61,7 @@ def _scopes_to_resource(*scopes):
     "AuthCodeRedirectServer",
     "AadClientCertificate",
     "CertificateCredentialBase",
+    "ClientSecretCredentialBase",
     "ConfidentialClientCredential",
     "get_default_authority",
     "InteractiveCredential",

sdk/identity/azure-identity/azure/identity/_base.py renamed to sdk/identity/azure-identity/azure/identity/_internal/client_secret_credential_base.py

@@ -3,34 +3,33 @@
 # Licensed under the MIT License.
 # ------------------------------------
 import abc
+from typing import TYPE_CHECKING
 
 try:
     ABC = abc.ABC
-except AttributeError:  # Python 2.7, abc exists, but not ABC
+except AttributeError:  # Python 2.7
     ABC = abc.ABCMeta("ABC", (object,), {"__slots__": ()})  # type: ignore
 
-try:
-    from typing import TYPE_CHECKING
-except ImportError:
-    TYPE_CHECKING = False
-
 if TYPE_CHECKING:
-    # pylint:disable=unused-import
-    from typing import Any, Optional, Union
-
+    # pylint:disable=unused-import,ungrouped-imports
+    from typing import Any
 
-class ClientSecretCredentialBase(object):
-    """Sans I/O base for client secret credentials"""
 
-    def __init__(self, tenant_id, client_id, secret, **kwargs):  # pylint:disable=unused-argument
+class ClientSecretCredentialBase(ABC):
+    def __init__(self, tenant_id, client_id, client_secret, **kwargs):
         # type: (str, str, str, **Any) -> None
         if not client_id:
             raise ValueError("client_id should be the id of an Azure Active Directory application")
-        if not secret:
+        if not client_secret:
             raise ValueError("secret should be an Azure Active Directory application's client secret")
         if not tenant_id:
             raise ValueError(
                 "tenant_id should be an Azure Active Directory tenant's id (also called its 'directory id')"
             )
-        self._form_data = {"client_id": client_id, "client_secret": secret, "grant_type": "client_credentials"}
-        super(ClientSecretCredentialBase, self).__init__()
+
+        self._client = self._get_auth_client(tenant_id, client_id, **kwargs)
+        self._secret = client_secret
+
+    @abc.abstractmethod
+    def _get_auth_client(self, tenant_id, client_id, **kwargs):
+        pass

sdk/identity/azure-identity/azure/identity/aio/_credentials/client_secret.py

@@ -5,15 +5,15 @@
 from typing import TYPE_CHECKING
 
 from .base import AsyncCredentialBase
-from .._authn_client import AsyncAuthnClient
-from ..._base import ClientSecretCredentialBase
+from .._internal import AadClient
+from ..._internal import ClientSecretCredentialBase
 
 if TYPE_CHECKING:
     from typing import Any
     from azure.core.credentials import AccessToken
 
 
-class ClientSecretCredential(ClientSecretCredentialBase, AsyncCredentialBase):
+class ClientSecretCredential(AsyncCredentialBase, ClientSecretCredentialBase):
     """Authenticates as a service principal using a client ID and client secret.
 
     :param str tenant_id: ID of the service principal's tenant. Also called its 'directory' ID.
@@ -25,10 +25,6 @@ class ClientSecretCredential(ClientSecretCredentialBase, AsyncCredentialBase):
           defines authorities for other clouds.
     """
 
-    def __init__(self, tenant_id: str, client_id: str, client_secret: str, **kwargs: "Any") -> None:
-        super(ClientSecretCredential, self).__init__(tenant_id, client_id, client_secret, **kwargs)
-        self._client = AsyncAuthnClient(tenant=tenant_id, **kwargs)
-
     async def __aenter__(self):
         await self._client.__aenter__()
         return self
@@ -38,7 +34,7 @@ async def close(self):
 
         await self._client.__aexit__()
 
-    async def get_token(self, *scopes: str, **kwargs: "Any") -> "AccessToken":  # pylint:disable=unused-argument
+    async def get_token(self, *scopes: str, **kwargs: "Any") -> "AccessToken":
         """Asynchronously request an access token for `scopes`.
 
         .. note:: This method is called by Azure SDK clients. It isn't intended for use in application code.
@@ -52,8 +48,10 @@ async def get_token(self, *scopes: str, **kwargs: "Any") -> "AccessToken":  # py
         if not scopes:
             raise ValueError("'get_token' requires at least one scope")
 
-        token = self._client.get_cached_token(scopes)
+        token = self._client.get_cached_access_token(scopes)
         if not token:
-            data = dict(self._form_data, scope=" ".join(scopes))
-            token = await self._client.request_token(scopes, form_data=data)
-        return token  # type: ignore
+            token = await self._client.obtain_token_by_client_secret(scopes, self._secret, **kwargs)
+        return token
+
+    def _get_auth_client(self, tenant_id, client_id, **kwargs):
+        return AadClient(tenant_id, client_id, **kwargs)