CertificateCredential uses AadClient

Author: chlowell

sdk/identity/azure-identity/azure/identity/_base.py

@@ -3,13 +3,6 @@
 # Licensed under the MIT License.
 # ------------------------------------
 import abc
-import binascii
-
-from cryptography import x509
-from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.hazmat.backends import default_backend
-from msal.oauth2cli import JwtSigner
-import six
 
 try:
     ABC = abc.ABC
@@ -41,48 +34,3 @@ def __init__(self, tenant_id, client_id, secret, **kwargs):  # pylint:disable=un
             )
         self._form_data = {"client_id": client_id, "client_secret": secret, "grant_type": "client_credentials"}
         super(ClientSecretCredentialBase, self).__init__()
-
-
-class CertificateCredentialBase(ABC):
-    """Sans I/O base for certificate credentials"""
-
-    def __init__(self, tenant_id, client_id, certificate_path, **kwargs):  # pylint:disable=unused-argument
-        # type: (str, str, str, **Any) -> None
-        if not certificate_path:
-            raise ValueError(
-                "'certificate_path' must be the path to a PEM file containing an x509 certificate and its private key"
-            )
-
-        super(CertificateCredentialBase, self).__init__()
-
-        password = kwargs.pop("password", None)
-        if isinstance(password, six.text_type):
-            password = password.encode(encoding="utf-8")
-
-        with open(certificate_path, "rb") as f:
-            pem_bytes = f.read()
-
-        private_key = serialization.load_pem_private_key(pem_bytes, password=password, backend=default_backend())
-        cert = x509.load_pem_x509_certificate(pem_bytes, default_backend())
-        fingerprint = cert.fingerprint(hashes.SHA1())   #nosec
-
-        self._client = self._get_auth_client(tenant_id, **kwargs)
-        self._client_id = client_id
-        self._signer = JwtSigner(private_key, "RS256", sha1_thumbprint=binascii.hexlify(fingerprint))
-
-    def _get_request_data(self, *scopes):
-        assertion = self._signer.sign_assertion(audience=self._client.auth_url, issuer=self._client_id)
-        if isinstance(assertion, six.binary_type):
-            assertion = assertion.decode("utf-8")
-
-        return {
-            "client_assertion": assertion,
-            "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
-            "client_id": self._client_id,
-            "grant_type": "client_credentials",
-            "scope": " ".join(scopes),
-        }
-
-    @abc.abstractmethod
-    def _get_auth_client(self, tenant_id, **kwargs):
-        pass

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

@@ -4,8 +4,7 @@
 # ------------------------------------
 from typing import TYPE_CHECKING
 
-from .._authn_client import AuthnClient
-from .._base import CertificateCredentialBase
+from .._internal import AadClient, CertificateCredentialBase
 
 if TYPE_CHECKING:
     from azure.core.credentials import AccessToken
@@ -42,11 +41,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 = self._get_request_data(*scopes)
-            token = self._client.request_token(scopes, form_data=data)
+            token = self._client.obtain_token_by_client_certificate(scopes, self._certificate, **kwargs)
         return token
 
-    def _get_auth_client(self, tenant_id, **kwargs):
-        return AuthnClient(tenant=tenant_id, **kwargs)
+    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

@@ -34,6 +34,7 @@ def get_default_authority():
 from .aad_client_base import AadClientBase
 from .auth_code_redirect_handler import AuthCodeRedirectServer
 from .aadclient_certificate import AadClientCertificate
+from .certificate_credential_base import CertificateCredentialBase
 from .exception_wrapper import wrap_exceptions
 from .msal_credentials import ConfidentialClientCredential, InteractiveCredential, PublicClientCredential
 from .msal_transport_adapter import MsalTransportAdapter, MsalTransportResponse
@@ -58,6 +59,7 @@ def _scopes_to_resource(*scopes):
     "AadClientBase",
     "AuthCodeRedirectServer",
     "AadClientCertificate",
+    "CertificateCredentialBase",
     "ConfidentialClientCredential",
     "get_default_authority",
     "InteractiveCredential",

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

@@ -127,9 +127,8 @@ def _get_client_certificate_request(self, scopes, certificate):
         }
 
         request = HttpRequest(
-            "POST", self._token_endpoint, headers={"Content-Type": "application/x-www-form-urlencoded"}
+            "POST", self._token_endpoint, headers={"Content-Type": "application/x-www-form-urlencoded"}, data=data
         )
-        request.set_formdata_body(data)
         return request
 
     def _get_jwt_assertion(self, certificate):

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

@@ -0,0 +1,47 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+import abc
+
+import six
+from azure.identity._internal import AadClientCertificate
+
+try:
+    ABC = abc.ABC
+except AttributeError:  # Python 2.7, abc exists, but not ABC
+    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
+
+
+class CertificateCredentialBase(ABC):
+    def __init__(self, tenant_id, client_id, certificate_path, **kwargs):
+        # type: (str, str, str, **Any) -> None
+        if not certificate_path:
+            raise ValueError(
+                "'certificate_path' must be the path to a PEM file containing an x509 certificate and its private key"
+            )
+
+        super(CertificateCredentialBase, self).__init__()
+
+        password = kwargs.pop("password", None)
+        if isinstance(password, six.text_type):
+            password = password.encode(encoding="utf-8")
+
+        with open(certificate_path, "rb") as f:
+            pem_bytes = f.read()
+
+        self._certificate = AadClientCertificate(pem_bytes, password=password)
+        self._client = self._get_auth_client(tenant_id, client_id, **kwargs)
+
+    @abc.abstractmethod
+    def _get_auth_client(self, tenant_id, client_id, **kwargs):
+        pass

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

@@ -5,8 +5,8 @@
 from typing import TYPE_CHECKING
 
 from .base import AsyncCredentialBase
-from .._authn_client import AsyncAuthnClient
-from ..._base import CertificateCredentialBase
+from .._internal import AadClient
+from ..._internal import CertificateCredentialBase
 
 if TYPE_CHECKING:
     from typing import Any
@@ -51,11 +51,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 = self._get_request_data(*scopes)
-            token = await self._client.request_token(scopes, form_data=data)
-        return token  # type: ignore
+            token = await self._client.obtain_token_by_client_certificate(scopes, self._certificate, **kwargs)
+        return token
 
-    def _get_auth_client(self, tenant_id, **kwargs):
-        return AsyncAuthnClient(tenant=tenant_id, **kwargs)
+    def _get_auth_client(self, tenant_id, client_id, **kwargs):
+        return AadClient(tenant_id, client_id, **kwargs)