Support SNI via PFX

Opt in via client_credential={
    "public_certificate": True,
    "private_key_pfx_path": "/path/to/cert.pfx",
}

Author: rayluo

msal/application.py

@@ -61,17 +61,23 @@ def _str2bytes(raw):
         return raw
 
 
-def _load_private_key_from_pfx_path(pfx_path, passphrase_bytes):
+def _parse_pfx(pfx_path, passphrase_bytes):
     # Cert concepts https://security.stackexchange.com/a/226758/125264
-    from cryptography.hazmat.primitives import hashes
+    from cryptography.hazmat.primitives import hashes, serialization
     from cryptography.hazmat.primitives.serialization import pkcs12
     with open(pfx_path, 'rb') as f:
         private_key, cert, _ = pkcs12.load_key_and_certificates(  # cryptography 2.5+
             # https://cryptography.io/en/latest/hazmat/primitives/asymmetric/serialization/#cryptography.hazmat.primitives.serialization.pkcs12.load_key_and_certificates
             f.read(), passphrase_bytes)
+    if not (private_key and cert):
+        raise ValueError("Your PFX file shall contain both private key and cert")
+    cert_pem = cert.public_bytes(encoding=serialization.Encoding.PEM).decode()  # cryptography 1.0+
+    x5c = [
+        '\n'.join(cert_pem.splitlines()[1:-1])  # Strip the "--- header ---" and "--- footer ---"
+    ]
     sha1_thumbprint = cert.fingerprint(hashes.SHA1()).hex()  # cryptography 0.7+
         # https://cryptography.io/en/latest/x509/reference/#x-509-certificate-object
-    return private_key, sha1_thumbprint
+    return private_key, sha1_thumbprint, x5c
 
 
 def _load_private_key_from_pem_str(private_key_pem_str, passphrase_bytes):
@@ -231,47 +237,71 @@ def __init__(
 
         :param client_credential:
             For :class:`PublicClientApplication`, you use `None` here.
+
             For :class:`ConfidentialClientApplication`,
-            it can be a string containing client secret,
-            or an X509 certificate container in this form::
+            it supports many different input formats for different scenarios.
 
-                {
-                    "private_key": "...-----BEGIN PRIVATE KEY-----... in PEM format",
-                    "thumbprint": "A1B2C3D4E5F6...",
-                    "public_certificate": "...-----BEGIN CERTIFICATE-----... (Optional. See below.)",
-                    "passphrase": "Passphrase if the private_key is encrypted (Optional. Added in version 1.6.0)",
-                }
+            .. admonition:: Support using a client secret.
 
-            MSAL Python requires a "private_key" in PEM format.
-            If your cert is in a PKCS12 (.pfx) format, you can also
-            `convert it to PEM and get the thumbprint <https://github.com/Azure/azure-sdk-for-python/blob/07d10639d7e47f4852eaeb74aef5d569db499d6e/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py#L101-L123>`_.
+                Just feed in a string, such as ``"your client secret"``.
 
-            The thumbprint is available in your app's registration in Azure Portal.
-            Alternatively, you can `calculate the thumbprint <https://github.com/Azure/azure-sdk-for-python/blob/07d10639d7e47f4852eaeb74aef5d569db499d6e/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py#L94-L97>`_.
+            .. admonition:: Support using a certificate in X.509 (.pem) format
 
-            *Added in version 0.5.0*:
-            public_certificate (optional) is public key certificate
-            which will be sent through 'x5c' JWT header only for
-            subject name and issuer authentication to support cert auto rolls.
-
-            Per `specs <https://tools.ietf.org/html/rfc7515#section-4.1.6>`_,
-            "the certificate containing
-            the public key corresponding to the key used to digitally sign the
-            JWS MUST be the first certificate.  This MAY be followed by
-            additional certificates, with each subsequent certificate being the
-            one used to certify the previous one."
-            However, your certificate's issuer may use a different order.
-            So, if your attempt ends up with an error AADSTS700027 -
-            "The provided signature value did not match the expected signature value",
-            you may try use only the leaf cert (in PEM/str format) instead.
-
-            *Added in version 1.13.0*:
-            It can also be a completely pre-signed assertion that you've assembled yourself.
-            Simply pass a container containing only the key "client_assertion", like this::
+                Feed in a dict in this form::
 
-                {
-                    "client_assertion": "...a JWT with claims aud, exp, iss, jti, nbf, and sub..."
-                }
+                    {
+                        "private_key": "...-----BEGIN PRIVATE KEY-----... in PEM format",
+                        "thumbprint": "A1B2C3D4E5F6...",
+                        "passphrase": "Passphrase if the private_key is encrypted (Optional. Added in version 1.6.0)",
+                    }
+
+                MSAL Python requires a "private_key" in PEM format.
+                If your cert is in PKCS12 (.pfx) format,
+                you can convert it to X.509 (.pem) format,
+                by ``openssl pkcs12 -in file.pfx -out file.pem -nodes``.
+
+                The thumbprint is available in your app's registration in Azure Portal.
+                Alternatively, you can `calculate the thumbprint <https://github.com/Azure/azure-sdk-for-python/blob/07d10639d7e47f4852eaeb74aef5d569db499d6e/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py#L94-L97>`_.
+
+            .. admonition:: Support Subject Name/Issuer Auth with a cert in .pem
+
+                `Subject Name/Issuer Auth
+                <https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/60>`_
+                is an approach to allow easier certificate rotation.
+
+                *Added in version 0.5.0*::
+
+                    {
+                        "private_key": "...-----BEGIN PRIVATE KEY-----... in PEM format",
+                        "thumbprint": "A1B2C3D4E5F6...",
+                        "public_certificate": "...-----BEGIN CERTIFICATE-----...",
+                        "passphrase": "Passphrase if the private_key is encrypted (Optional. Added in version 1.6.0)",
+                    }
+
+                ``public_certificate`` (optional) is public key certificate
+                which will be sent through 'x5c' JWT header only for
+                subject name and issuer authentication to support cert auto rolls.
+
+                Per `specs <https://tools.ietf.org/html/rfc7515#section-4.1.6>`_,
+                "the certificate containing
+                the public key corresponding to the key used to digitally sign the
+                JWS MUST be the first certificate.  This MAY be followed by
+                additional certificates, with each subsequent certificate being the
+                one used to certify the previous one."
+                However, your certificate's issuer may use a different order.
+                So, if your attempt ends up with an error AADSTS700027 -
+                "The provided signature value did not match the expected signature value",
+                you may try use only the leaf cert (in PEM/str format) instead.
+
+            .. admonition:: Supporting raw assertion obtained from elsewhere
+
+                *Added in version 1.13.0*:
+                It can also be a completely pre-signed assertion that you've assembled yourself.
+                Simply pass a container containing only the key "client_assertion", like this::
+
+                    {
+                        "client_assertion": "...a JWT with claims aud, exp, iss, jti, nbf, and sub..."
+                    }
 
             .. admonition:: Supporting reading client cerficates from PFX files
 
@@ -280,14 +310,26 @@ def __init__(
 
                     {
                         "private_key_pfx_path": "/path/to/your.pfx",
-                        "passphrase": "Passphrase if the private_key is encrypted (Optional. Added in version 1.6.0)",
+                        "passphrase": "Passphrase if the private_key is encrypted (Optional)",
                     }
 
                 The following command will generate a .pfx file from your .key and .pem file::
 
                     openssl pkcs12 -export -out certificate.pfx -inkey privateKey.key -in certificate.pem
 
-        :type client_credential: Union[dict, str]
+            .. admonition:: Support Subject Name/Issuer Auth with a cert in .pfx
+
+                *Added in version 1.30.0*:
+                If your .pfx file contains both the private key and public cert,
+                you can opt in for Subject Name/Issuer Auth like this::
+
+                    {
+                        "private_key_pfx_path": "/path/to/your.pfx",
+                        "public_certificate": True,
+                        "passphrase": "Passphrase if the private_key is encrypted (Optional)",
+                    }
+
+        :type client_credential: Union[dict, str, None]
 
         :param dict client_claims:
             *Added in version 0.5.0*:
@@ -699,14 +741,15 @@ def _build_client(self, client_credential, authority, skip_regional_client=False
                 client_assertion = client_credential['client_assertion']
             else:
                 headers = {}
-                if client_credential.get('public_certificate'):
-                    headers["x5c"] = extract_certs(client_credential['public_certificate'])
                 passphrase_bytes = _str2bytes(
                     client_credential["passphrase"]
                     ) if client_credential.get("passphrase") else None
                 if client_credential.get("private_key_pfx_path"):
-                    private_key, sha1_thumbprint = _load_private_key_from_pfx_path(
-                        client_credential["private_key_pfx_path"], passphrase_bytes)
+                    private_key, sha1_thumbprint, x5c = _parse_pfx(
+                        client_credential["private_key_pfx_path"],
+                        passphrase_bytes)
+                    if client_credential.get("public_certificate") is True and x5c:
+                        headers["x5c"] = x5c
                 elif (
                         client_credential.get("private_key")  # PEM blob
                         and client_credential.get("thumbprint")):
@@ -720,6 +763,10 @@ def _build_client(self, client_credential, authority, skip_regional_client=False
                     raise ValueError(
                         "client_credential needs to follow this format "
                         "https://msal-python.readthedocs.io/en/latest/#msal.ClientApplication.params.client_credential")
+                if ("x5c" not in headers  # So we did not run the pfx code path
+                    and isinstance(client_credential.get('public_certificate'), str)
+                ):  # Then we treat the public_certificate value as PEM content
+                    headers["x5c"] = extract_certs(client_credential['public_certificate'])
                 assertion = JwtAssertionCreator(
                     private_key, algorithm="RS256",
                     sha1_thumbprint=sha1_thumbprint, headers=headers)

tests/test_cryptography.py

@@ -8,7 +8,7 @@
 import requests
 
 from msal.application import (
-    _str2bytes, _load_private_key_from_pem_str, _load_private_key_from_pfx_path)
+    _str2bytes, _load_private_key_from_pem_str, _parse_pfx)
 
 
 latest_cryptography_version = ET.fromstring(
@@ -48,7 +48,7 @@ def test_latest_cryptography_should_support_our_usage_without_warnings(self):
                 _load_private_key_from_pem_str(f.read(), passphrase_bytes)
             pfx = sibling("certificate-with-password.pfx")  # Created by:
                 # openssl pkcs12 -export -inkey test/certificate-with-password.pem -in tests/certificate-with-password.pem -out tests/certificate-with-password.pfx
-            _load_private_key_from_pfx_path(pfx, passphrase_bytes)
+            _parse_pfx(pfx, passphrase_bytes)
             self.assertEqual(0, len(encountered_warnings),
                 "Did cryptography deprecate the functions that we used?")
 

tests/test_e2e.py

@@ -80,7 +80,7 @@ def _get_hint(html_mode=None, username=None, lab_name=None, username_uri=None):
             else "the upn from {}".format(_render(
                 username_uri, description="here" if html_mode else None)),
         lab=_render(
-            "https://aka.ms/GetLabUserSecret?Secret=" + (lab_name or "msidlabXYZ"),
+            "https://aka.ms/GetLabSecret?Secret=" + (lab_name or "msidlabXYZ"),
             description="this password api" if html_mode else None,
             ),
         )
@@ -463,7 +463,10 @@ def get_lab_app(
         # id came from https://docs.msidlab.com/accounts/confidentialclient.html
         client_id = os.getenv(env_client_id)
         # Cert came from https://ms.portal.azure.com/#@microsoft.onmicrosoft.com/asset/Microsoft_Azure_KeyVault/Certificate/https://msidlabs.vault.azure.net/certificates/LabVaultAccessCert
-        client_credential = {"private_key_pfx_path": os.getenv(env_client_cert_path)}
+        client_credential = {
+            "private_key_pfx_path": os.getenv(env_client_cert_path),
+            "public_certificate": True,  # Opt in for SNI
+            }
     elif os.getenv(env_client_id) and os.getenv(env_name2):
         # Data came from here
         # https://docs.msidlab.com/accounts/confidentialclient.html
@@ -529,7 +532,7 @@ def get_lab_user_secret(cls, lab_name="msidlab4"):
         lab_name = lab_name.lower()
         if lab_name not in cls._secrets:
             logger.info("Querying lab user password for %s", lab_name)
-            url = "https://msidlab.com/api/LabUserSecret?secret=%s" % lab_name
+            url = "https://msidlab.com/api/LabSecret?secret=%s" % lab_name
             resp = cls.session.get(url)
             cls._secrets[lab_name] = resp.json()["value"]
         return cls._secrets[lab_name]
@@ -860,7 +863,7 @@ def test_adfs2019_onprem_acquire_token_by_auth_code(self):
 
         # https://msidlab.com/api/user?usertype=onprem&federationprovider=ADFSv2019
         username = "..."  # The upn from the link above
-        password="***"  # From https://aka.ms/GetLabUserSecret?Secret=msidlabXYZ
+        password="***"  # From https://aka.ms/GetLabSecret?Secret=msidlabXYZ
         """
         config = self.get_lab_user(usertype="onprem", federationProvider="ADFSv2019")
         config["authority"] = "https://fs.%s.com/adfs" % config["lab_name"]
@@ -953,7 +956,7 @@ def test_b2c_acquire_token_by_auth_code(self):
 
             username="[email protected]"
                 # This won't work https://msidlab.com/api/user?usertype=b2c
-            password="***"  # From https://aka.ms/GetLabUserSecret?Secret=msidlabb2c
+            password="***"  # From https://aka.ms/GetLabSecret?Secret=msidlabb2c
         """
         config = self.get_lab_app_object(azureenvironment="azureb2ccloud")
         self._test_acquire_token_by_auth_code(