Passing encrypted key as credential

msal/application.py

@@ -90,6 +90,14 @@ def _merge_claims_challenge_and_capabilities(capabilities, claims_challenge):
     return json.dumps(claims_dict)
 
 
+def _str2bytes(raw):
+    # A conversion based on duck-typing rather than six.text_type
+    try:
+        return raw.encode(encoding="utf-8")
+    except:
+        return raw
+
+
 class ClientApplication(object):
 
     ACQUIRE_TOKEN_SILENT_ID = "84"
@@ -124,6 +132,7 @@ def __init__(
                     "private_key": "...-----BEGIN PRIVATE KEY-----...",
                     "thumbprint": "A1B2C3D4E5F6...",
                     "public_certificate": "...-----BEGIN CERTIFICATE-----..." (Optional. See below.)
+                    "passphrase": "Passphrase if the private_key is encrypted (Optional)"
                 }
 
             *Added in version 0.5.0*:
@@ -252,8 +261,18 @@ def _build_client(self, client_credential, authority):
             headers = {}
             if 'public_certificate' in client_credential:
                 headers["x5c"] = extract_certs(client_credential['public_certificate'])
+            if not client_credential.get("passphrase"):
+                unencrypted_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
+                    )
             assertion = JwtAssertionCreator(
-                client_credential["private_key"], algorithm="RS256",
+                unencrypted_private_key, algorithm="RS256",
                 sha1_thumbprint=client_credential.get("thumbprint"), headers=headers)
             client_assertion = assertion.create_regenerative_assertion(
                 audience=authority.token_endpoint, issuer=self.client_id,

setup.py

@@ -74,6 +74,16 @@
     install_requires=[
         'requests>=2.0.0,<3',
         'PyJWT[crypto]>=1.0.0,<2',
+
+        'cryptography>=0.6,<4',
+            # load_pem_private_key() is available since 0.6
+            # https://github.com/pyca/cryptography/blob/master/CHANGELOG.rst#06---2014-09-29
+            #
+            # Not sure what should be used as an upper bound here
+            # https://github.com/pyca/cryptography/issues/5532
+            # We will go with "<4" for now, which is also what our another dependency,
+            # pyjwt, currently use.
+
     ]
 )
 

tests/test_application.py

@@ -1,6 +1,7 @@
 # Note: Since Aug 2019 we move all e2e tests into test_e2e.py,
 # so this test_application file contains only unit tests without dependency.
 from msal.application import *
+from msal.application import _str2bytes
 import msal
 from msal.application import _merge_claims_challenge_and_capabilities
 from tests import unittest
@@ -39,6 +40,14 @@ def test_extract_multiple_tag_enclosed_certs(self):
         self.assertEqual(["my_cert1", "my_cert2"], extract_certs(pem))
 
 
+class TestBytesConversion(unittest.TestCase):
+    def test_string_to_bytes(self):
+        self.assertEqual(type(_str2bytes("some string")), type(b"bytes"))
+
+    def test_bytes_to_bytes(self):
+        self.assertEqual(type(_str2bytes(b"some bytes")), type(b"bytes"))
+
+
 class TestClientApplicationAcquireTokenSilentErrorBehaviors(unittest.TestCase):
 
     def setUp(self):