Address review comments on mTLS PoP cert/key handling

Four small, low-risk fixes from the code review:

- application.py `_private_key_to_unencrypted_pem`: pass an explicit
  `backend=default_backend()` to `load_pem_private_key` (required on the
  supported `cryptography` floor, 2.5) and wrap the call so a bad or
  encrypted-without-passphrase key surfaces a clear, actionable ValueError
  instead of a cryptic low-level error. Mirrors the sibling
  `_load_private_key_from_pem_str`.

- token_cache.py: coerce `key_id` to a str via a new `_key_id_to_str`
  helper before building the AccessToken cache key, so a non-str key_id
  never raises TypeError at the `"-" + key_id` concatenation. The normal
  ASCII-str path is byte-identical, so existing mtls_pop cache entries are
  unaffected. Defensive only: current callers already pass an ASCII str.

- test_application.py: hoist `import os` to the top of the module and drop
  the mid-file `import os as _os` alias.

Adds regression tests: bytes/other key_id coercion (test_token_cache.py)
and the clearer private-key load errors (test_mtls_transport.py).

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: Robbie-Microsoft

msal/application.py

@@ -120,10 +120,18 @@ def _private_key_to_unencrypted_pem(private_key, passphrase_bytes=None):
     The result is suitable for ``ssl.SSLContext.load_cert_chain`` via a temp file.
     """
     from cryptography.hazmat.primitives import serialization
+    from cryptography.hazmat.backends import default_backend
     if isinstance(private_key, (str, bytes)):
-        key_obj = serialization.load_pem_private_key(
-            _str2bytes(private_key) if isinstance(private_key, str) else private_key,
-            passphrase_bytes)
+        try:
+            key_obj = serialization.load_pem_private_key(
+                _str2bytes(private_key) if isinstance(private_key, str) else private_key,
+                passphrase_bytes,
+                backend=default_backend(),  # Required param until cryptography 3.1
+                )
+        except (TypeError, ValueError) as exc:
+            raise ValueError(
+                "Could not load the private key for mTLS Proof-of-Possession. "
+                "If the key is encrypted, provide its 'passphrase'.") from exc
     else:  # Already a cryptography private-key object
         key_obj = private_key
     return key_obj.private_bytes(

msal/token_cache.py

@@ -14,6 +14,17 @@
 logger = logging.getLogger(__name__)
 _GRANT_TYPE_BROKER = "broker"
 
+
+def _key_id_to_str(key_id):
+    """Coerce a key_id (base64url x5t#S256) into a str for the AT cache key.
+
+    It is normally an ASCII str, but tolerate bytes (decoded as ASCII) or any
+    other type so that building the cache key never raises ``TypeError``.
+    """
+    if isinstance(key_id, bytes):
+        return key_id.decode("ascii")
+    return key_id if isinstance(key_id, str) else str(key_id)
+
 # Fields in the request data dict that should NOT be included in the extended
 # cache key hash. Everything else in data IS included, because those are extra
 # body parameters going on the wire and must differentiate cached tokens.
@@ -170,8 +181,9 @@ def __init__(self):
                         ).lower()
                         # key_id is a base64url x5t#S256 and is case-sensitive,
                         # so it is appended AFTER lower-casing the rest, to keep
-                        # ATs bound to different keys/certs isolated.
-                        + ("-" + key_id if key_id else ""),
+                        # ATs bound to different keys/certs isolated. Coerce it to
+                        # a str first so a non-str key_id never breaks caching.
+                        + ("-" + _key_id_to_str(key_id) if key_id else ""),
             self.CredentialType.ID_TOKEN:
                 lambda home_account_id=None, environment=None, client_id=None,
                         realm=None, **ignored_payload_from_a_real_token:

tests/test_application.py

@@ -3,6 +3,7 @@
 import base64
 import json
 import logging
+import os
 import sys
 import time
 import warnings
@@ -1589,8 +1590,7 @@ def mock_post(url, headers=None, data=None, *args, **kwargs):
             "not receive a context dict")
 
 
-import os as _os
-_MTLS_PFX = _os.path.join(_os.path.dirname(__file__), "certificate-with-password.pfx")
+_MTLS_PFX = os.path.join(os.path.dirname(__file__), "certificate-with-password.pfx")
 _MTLS_CERT_CRED = {
     "private_key_pfx_path": _MTLS_PFX,
     "passphrase": "password",

tests/test_mtls_transport.py

@@ -3,7 +3,8 @@
 import unittest
 
 from msal import mtls
-from msal.application import _load_mtls_cert_material
+from msal.application import (
+    _load_mtls_cert_material, _private_key_to_unencrypted_pem)
 
 
 class TestMtlsEndpointTransform(unittest.TestCase):
@@ -142,5 +143,38 @@ def test_http_client_builds_session_lazily(self):
         client.close()
 
 
+class TestUnencryptedPemLoading(unittest.TestCase):
+    """_private_key_to_unencrypted_pem pins the cryptography backend and turns
+    low-level load failures into a clear, actionable ValueError."""
+
+    @staticmethod
+    def _encrypted_pem(passphrase=b"secret"):
+        from cryptography.hazmat.primitives import serialization
+        from cryptography.hazmat.primitives.asymmetric import rsa
+        from cryptography.hazmat.backends import default_backend
+        key = rsa.generate_private_key(
+            public_exponent=65537, key_size=2048, backend=default_backend())
+        return key.private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            serialization.BestAvailableEncryption(passphrase))
+
+    def test_encrypted_key_without_passphrase_raises_clear_error(self):
+        with self.assertRaises(ValueError) as cm:
+            _private_key_to_unencrypted_pem(self._encrypted_pem(), None)
+        self.assertIn("passphrase", str(cm.exception))
+
+    def test_garbage_key_raises_clear_error(self):
+        with self.assertRaises(ValueError) as cm:
+            _private_key_to_unencrypted_pem(b"not a real pem", None)
+        self.assertIn("private key for mTLS", str(cm.exception))
+
+    def test_encrypted_key_with_passphrase_round_trips_to_unencrypted_pem(self):
+        pem = _private_key_to_unencrypted_pem(
+            self._encrypted_pem(b"secret"), b"secret")
+        self.assertIn(b"PRIVATE KEY", pem)
+        self.assertNotIn(b"ENCRYPTED", pem)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_token_cache.py

@@ -4,7 +4,8 @@
 import time
 import warnings
 
-from msal.token_cache import TokenCache, SerializableTokenCache, _compute_ext_cache_key
+from msal.token_cache import (
+    TokenCache, SerializableTokenCache, _compute_ext_cache_key, _key_id_to_str)
 from tests import unittest
 
 
@@ -260,6 +261,19 @@ def test_access_tokens_with_different_key_id(self):
                     home_account_id="uid.utid",
                 ))
 
+    def test_key_id_is_coerced_to_str_for_cache_key(self):
+        # A bytes (or otherwise non-str) key_id must not raise a TypeError while
+        # building the AT cache key (the "-" + key_id concatenation once assumed
+        # str). This is defensive: current callers already pass an ASCII str.
+        self.assertEqual("THUMB", _key_id_to_str(b"THUMB"))
+        self.assertEqual("THUMB", _key_id_to_str("THUMB"))
+        self.assertEqual("123", _key_id_to_str(123))  # Any other type -> str()
+        # Adding + searching an AT with a bytes key_id no longer raises; the
+        # helper below stores and then finds it by that same key_id.
+        self._test_data_should_be_saved_and_searchable_in_access_token(
+            {"key_id": b"THUMB"})
+        self.assertEqual(1, len(self.cache._cache["AccessToken"]))
+
     def test_bearer_and_mtls_pop_tokens_coexist_and_isolate(self):
         # The crux of mTLS PoP cache isolation (plan C6): a Bearer token and an
         # mtls_pop token for the SAME app/scope/tenant must coexist, and an