Support reading CCA cert from a pfx file. Tested.

Author: rayluo

.github/workflows/python-package.yml

@@ -18,6 +18,8 @@ jobs:
       TRAVIS: true
       LAB_APP_CLIENT_ID: ${{ secrets.LAB_APP_CLIENT_ID }}
       LAB_APP_CLIENT_SECRET: ${{ secrets.LAB_APP_CLIENT_SECRET }}
+      LAB_APP_CLIENT_CERT_BASE64: ${{ secrets.LAB_APP_CLIENT_CERT_BASE64 }}
+      LAB_APP_CLIENT_CERT_PFX_PATH: lab_cert.pfx
       LAB_OBO_CLIENT_SECRET: ${{ secrets.LAB_OBO_CLIENT_SECRET }}
       LAB_OBO_CONFIDENTIAL_CLIENT_ID: ${{ secrets.LAB_OBO_CONFIDENTIAL_CLIENT_ID }}
       LAB_OBO_PUBLIC_CLIENT_ID: ${{ secrets.LAB_OBO_PUBLIC_CLIENT_ID }}
@@ -43,6 +45,9 @@ jobs:
         python -m pip install --upgrade pip
         python -m pip install flake8 pytest
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+    - name: Populate lab cert.pfx
+      # https://docs.github.com/en/actions/security-guides/using-secrets-in-github-actions#storing-base64-binary-blobs-as-secrets
+      run: echo $LAB_APP_CLIENT_CERT_BASE64 | base64 -d > $LAB_APP_CLIENT_CERT_PFX_PATH
     - name: Test with pytest
       run: pytest --benchmark-skip
     - name: Lint with flake8

msal/application.py

@@ -65,6 +65,29 @@ def _str2bytes(raw):
         return raw
 
 
+def _load_private_key_from_pfx_path(pfx_path, passphrase_bytes):
+    # Cert concepts https://security.stackexchange.com/a/226758/125264
+    from cryptography.hazmat.primitives import hashes
+    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)
+    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
+
+
+def _load_private_key_from_pem_str(private_key_pem_str, passphrase_bytes):
+    from cryptography.hazmat.primitives import serialization
+    from cryptography.hazmat.backends import default_backend
+    return serialization.load_pem_private_key(  # cryptography 0.6+
+        _str2bytes(private_key_pem_str),
+        passphrase_bytes,
+        backend=default_backend(),  # It was a required param until 2020
+        )
+
+
 def _pii_less_home_account_id(home_account_id):
     parts = home_account_id.split(".")  # It could contain one or two parts
     parts[0] = "********"
@@ -254,6 +277,16 @@ def __init__(
                     "client_assertion": "...a JWT with claims aud, exp, iss, jti, nbf, and sub..."
                 }
 
+            .. admonition:: Supporting reading client cerficates from PFX files
+
+                *Added in version 1.29.0*:
+                Feed in a dictionary containing the path to a PFX file::
+
+                    {
+                        "private_key_pfx_path": "/path/to/your.pfx",
+                        "passphrase": "Passphrase if the private_key is encrypted (Optional. Added in version 1.6.0)",
+                    }
+
         :type client_credential: Union[dict, str]
 
         :param dict client_claims:
@@ -651,29 +684,37 @@ def _build_client(self, client_credential, authority, skip_regional_client=False
             default_headers['x-app-ver'] = self.app_version
         default_body = {"client_info": 1}
         if isinstance(client_credential, dict):
-            assert (("private_key" in client_credential
-                    and "thumbprint" in client_credential) or
-                    "client_assertion" in client_credential)
             client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT
-            if 'client_assertion' in client_credential:
+            # Use client_credential.get("...") rather than "..." in client_credential
+            # so that we can ignore an empty string came from an empty ENV VAR.
+            if client_credential.get("client_assertion"):
                 client_assertion = client_credential['client_assertion']
             else:
                 headers = {}
-                if 'public_certificate' in client_credential:
+                if client_credential.get('public_certificate'):
                     headers["x5c"] = extract_certs(client_credential['public_certificate'])
-                if not client_credential.get("passphrase"):
-                    unencrypted_private_key = client_credential['private_key']
+                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)
+                elif (
+                        client_credential.get("private_key")  # PEM blob
+                        and client_credential.get("thumbprint")):
+                    sha1_thumbprint = client_credential["thumbprint"]
+                    if passphrase_bytes:
+                        private_key = _load_private_key_from_pem_str(
+                            client_credential['private_key'], passphrase_bytes)
+                    else:  # PEM without passphrase
+                        private_key = client_credential['private_key']
                 else:
-                    from cryptography.hazmat.primitives import serialization
-                    from cryptography.hazmat.backends import default_backend
-                    unencrypted_private_key = serialization.load_pem_private_key(
-                        _str2bytes(client_credential["private_key"]),
-                        _str2bytes(client_credential["passphrase"]),
-                        backend=default_backend(),  # It was a required param until 2020
-                        )
+                    raise ValueError(
+                        "client_credential needs to follow this format "
+                        "https://msal-python.readthedocs.io/en/latest/#msal.ClientApplication.params.client_credential")
                 assertion = JwtAssertionCreator(
-                    unencrypted_private_key, algorithm="RS256",
-                    sha1_thumbprint=client_credential.get("thumbprint"), headers=headers)
+                    private_key, algorithm="RS256",
+                    sha1_thumbprint=sha1_thumbprint, headers=headers)
                 client_assertion = assertion.create_regenerative_assertion(
                     audience=authority.token_endpoint, issuer=self.client_id,
                     additional_claims=self.client_claims or {})

setup.cfg

@@ -46,13 +46,13 @@ install_requires =
     # therefore is insusceptible to CVE-2022-29217 so no need to bump to PyJWT 2.4+
     PyJWT[crypto]>=1.0.0,<3
 
-    # load_pem_private_key() is available since 0.6
-    # https://github.com/pyca/cryptography/blob/master/CHANGELOG.rst#06---2014-09-29
+    # load_key_and_certificates() is available since 2.5
+    # https://cryptography.io/en/latest/hazmat/primitives/asymmetric/serialization/#cryptography.hazmat.primitives.serialization.pkcs12.load_key_and_certificates
     #
     # And we will use the cryptography (X+3).0.0 as the upper bound,
     # based on their latest deprecation policy
     # https://cryptography.io/en/latest/api-stability/#deprecation
-    cryptography>=0.6,<45
+    cryptography>=2.5,<45
 
 
 [options.extras_require]

tests/certificate-with-password.pfx

@@ -7,7 +7,8 @@
 
 import requests
 
-from msal.application import _str2bytes
+from msal.application import (
+    _str2bytes, _load_private_key_from_pem_str, _load_private_key_from_pfx_path)
 
 
 latest_cryptography_version = ET.fromstring(
@@ -26,6 +27,10 @@ def get_current_ceiling():
     raise RuntimeError("Unable to find cryptography info from setup.cfg")
 
 
+def sibling(filename):
+    return os.path.join(os.path.dirname(__file__), filename)
+
+
 class CryptographyTestCase(TestCase):
 
     def test_should_be_run_with_latest_version_of_cryptography(self):
@@ -37,18 +42,13 @@ def test_should_be_run_with_latest_version_of_cryptography(self):
             cryptography.__version__, latest_cryptography_version))
 
     def test_latest_cryptography_should_support_our_usage_without_warnings(self):
-        with open(os.path.join(
-                os.path.dirname(__file__), "certificate-with-password.pem")) as f:
-            cert = f.read()
+        passphrase_bytes = _str2bytes("password")
         with warnings.catch_warnings(record=True) as encountered_warnings:
-            # The usage was copied from application.py
-            from cryptography.hazmat.primitives import serialization
-            from cryptography.hazmat.backends import default_backend
-            unencrypted_private_key = serialization.load_pem_private_key(
-                 _str2bytes(cert),
-                _str2bytes("password"),
-                backend=default_backend(),  # It was a required param until 2020
-                )
+            with open(sibling("certificate-with-password.pem")) as f:
+                _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)
             self.assertEqual(0, len(encountered_warnings),
                 "Did cryptography deprecate the functions that we used?")
 

tests/test_cryptography.py

@@ -446,6 +446,7 @@ def test_device_flow(self):
 def get_lab_app(
         env_client_id="LAB_APP_CLIENT_ID",
         env_name2="LAB_APP_CLIENT_SECRET",  # A var name that hopefully avoids false alarm
+        env_client_cert_path="LAB_APP_CLIENT_CERT_PFX_PATH",
         authority="https://login.microsoftonline.com/"
             "72f988bf-86f1-41af-91ab-2d7cd011db47",  # Microsoft tenant ID
         timeout=None,
@@ -458,21 +459,23 @@ def get_lab_app(
         "Reading ENV variables %s and %s for lab app defined at "
         "https://docs.msidlab.com/accounts/confidentialclient.html",
         env_client_id, env_name2)
-    if os.getenv(env_client_id) and os.getenv(env_name2):
-        # A shortcut mainly for running tests on developer's local development machine
-        # or it could be setup on Travis CI
-        #   https://docs.travis-ci.com/user/environment-variables/#defining-variables-in-repository-settings
+    if os.getenv(env_client_id) and os.getenv(env_client_cert_path):
+        # 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)}
+    elif os.getenv(env_client_id) and os.getenv(env_name2):
         # Data came from here
         # https://docs.msidlab.com/accounts/confidentialclient.html
         client_id = os.getenv(env_client_id)
-        client_secret = os.getenv(env_name2)
+        client_credential = os.getenv(env_name2)
     else:
         logger.info("ENV variables are not defined. Fall back to MSI.")
         # See also https://microsoft.sharepoint-df.com/teams/MSIDLABSExtended/SitePages/Programmatically-accessing-LAB-API's.aspx
         raise unittest.SkipTest("MSI-based mechanism has not been implemented yet")
     return msal.ConfidentialClientApplication(
             client_id,
-            client_credential=client_secret,
+            client_credential=client_credential,
             authority=authority,
             http_client=MinimalHttpClient(timeout=timeout),
             **kwargs)