Validate public certificate encodings before x5c construction

msal/application.py

@@ -1,3 +1,4 @@
+import base64
 import functools
 import json
 import time
@@ -52,17 +53,28 @@ def _init_broker(enable_pii_log):  # Make it a function to allow mocking
 def extract_certs(public_cert_content):
     # Parses raw public certificate file contents and returns a list of strings
     # Usage: headers = {"x5c": extract_certs(open("my_cert.pem").read())}
+    from cryptography import x509
+    from cryptography.hazmat.backends import default_backend
+    from cryptography.hazmat.primitives import serialization
+
     public_certificates = re.findall(
-        r'-----BEGIN CERTIFICATE-----(?P<cert_value>[^-]+)-----END CERTIFICATE-----',
-        public_cert_content, re.I)
-    if public_certificates:
-        return [cert.strip() for cert in public_certificates]
-    # The public cert tags are not found in the input,
-    # let's make best effort to exclude a private key pem file.
-    if "PRIVATE KEY" in public_cert_content:
+        r'-----BEGIN CERTIFICATE-----(?P<cert_value>.*?)-----END CERTIFICATE-----',
+        public_cert_content, re.I | re.S)
+    certificates = []
+    try:
+        for value in public_certificates or [public_cert_content]:
+            der = base64.b64decode("".join(value.split()), validate=True)
+            cert = x509.load_der_x509_certificate(der, default_backend())
+            certificate_der = cert.public_bytes(serialization.Encoding.DER)
+            # Some parsers accept trailing data after the certificate.
+            if certificate_der != der:
+                raise ValueError()
+            certificates.append(base64.b64encode(certificate_der).decode("ascii"))
+    except ValueError:
         raise ValueError(
-            "We expect your public key but detect a private key instead")
-    return [public_cert_content.strip()]
+            "Invalid public certificate. Provide PEM certificates or a "
+            "base64-encoded DER certificate.") from None
+    return certificates
 
 
 def _merge_claims_challenge_and_capabilities(capabilities, claims_challenge):
@@ -357,6 +369,11 @@ def __init__(
 
                 ``public_certificate`` (optional) is public key certificate
                 which will be sent through 'x5c' JWT header.
+                With a manual thumbprint, PEM certificate chains or tagless
+                base64-encoded DER certificates are accepted. These encodings
+                are validated before constructing 'x5c'; malformed data raises
+                ``ValueError``. PEM chains retain their order, and other
+                labeled PEM blocks are not included in 'x5c'.
                 This is useful when you use `Subject Name/Issuer Authentication
                 <https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/60>`_
                 which is an approach to allow easier certificate rotation.

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.
 import base64
+import datetime
 import json
 import logging
 import os
@@ -9,6 +10,9 @@
 import warnings
 from unittest.mock import patch, Mock
 from requests.exceptions import ConnectionError as RequestsConnectionError
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec
 import msal
 from msal.application import (
     extract_certs,
@@ -34,29 +38,79 @@
 
 class TestHelperExtractCerts(unittest.TestCase):  # It is used by SNI scenario
 
+    @classmethod
+    def setUpClass(cls):
+        key = ec.generate_private_key(ec.SECP256R1())
+        name = x509.Name([x509.NameAttribute(x509.NameOID.COMMON_NAME, "test")])
+        cls.certs = [
+            x509.CertificateBuilder().subject_name(name).issuer_name(name)
+            .public_key(key.public_key()).serial_number(serial)
+            .not_valid_before(datetime.datetime(2020, 1, 1))
+            .not_valid_after(datetime.datetime(2030, 1, 1))
+            .sign(key, hashes.SHA256())
+            for serial in (1, 2)]
+        cls.pem = [
+            cert.public_bytes(serialization.Encoding.PEM).decode("ascii")
+            for cert in cls.certs]
+        cls.der = [
+            cert.public_bytes(serialization.Encoding.DER) for cert in cls.certs]
+        cls.encoded = [base64.b64encode(der).decode("ascii") for der in cls.der]
+
     def test_extract_a_tag_less_public_cert(self):
-        pem = "my_cert"
-        self.assertEqual(["my_cert"], extract_certs(pem))
+        value = self.encoded[0]
+        self.assertEqual([value], extract_certs(
+            " \n" + value[:40] + "\n\t" + value[40:] + " \n"))
 
     def test_extract_a_tag_enclosed_cert(self):
-        pem = """
-        -----BEGIN CERTIFICATE-----
-        my_cert
-        -----END CERTIFICATE-----
-        """
-        self.assertEqual(["my_cert"], extract_certs(pem))
+        self.assertEqual([self.encoded[0]], extract_certs(self.pem[0]))
 
     def test_extract_multiple_tag_enclosed_certs(self):
-        pem = """
-        -----BEGIN CERTIFICATE-----
-        my_cert1
-        -----END CERTIFICATE-----
-
-        -----BEGIN CERTIFICATE-----
-        my_cert2
-        -----END CERTIFICATE-----
-        """
-        self.assertEqual(["my_cert1", "my_cert2"], extract_certs(pem))
+        self.assertEqual(self.encoded, extract_certs("".join(self.pem)))
+        self.assertEqual(
+            self.encoded[::-1], extract_certs("".join(self.pem[::-1])))
+
+    def test_extract_only_certificates_from_combined_pem(self):
+        with open(os.path.join(
+                os.path.dirname(__file__), "certificate-with-password.pem")) as f:
+            combined = f.read()
+        cert = x509.load_pem_x509_certificate(combined.encode("ascii"))
+        expected = base64.b64encode(
+            cert.public_bytes(serialization.Encoding.DER)).decode("ascii")
+        self.assertEqual([expected], extract_certs(combined))
+
+    def test_reject_invalid_certificate_encodings(self):
+        public_key_der = self.certs[0].public_key().public_bytes(
+            serialization.Encoding.DER,
+            serialization.PublicFormat.SubjectPublicKeyInfo)
+        invalid_values = [
+            "", " \n\t", "not base64!", "é", "AA--AA",
+            base64.b64encode(b"not a certificate").decode("ascii"),
+            base64.b64encode(public_key_der).decode("ascii"),
+            base64.b64encode(self.der[0] + b"trailing bytes").decode("ascii"),
+            base64.b64encode(self.der[0] + self.der[1]).decode("ascii"),
+        ]
+        for value in invalid_values:
+            for content in (
+                    value,
+                    "-----BEGIN CERTIFICATE-----\n{}\n"
+                    "-----END CERTIFICATE-----".format(value)):
+                with self.subTest(content=content):
+                    with self.assertRaisesRegex(ValueError, "certificate"):
+                        extract_certs(content)
+        with self.assertRaises(ValueError):
+            extract_certs(self.pem[0] + "-----BEGIN CERTIFICATE-----\n"
+                "invalid\n-----END CERTIFICATE-----")
+        with self.assertRaises(ValueError):
+            extract_certs("-----BEGIN PUBLIC KEY-----\n{}\n"
+                "-----END PUBLIC KEY-----".format(self.encoded[0]))
+
+    def test_reject_trailing_der_when_parser_accepts_it(self):
+        value = base64.b64encode(
+            self.der[0] + b"trailing bytes").decode("ascii")
+        with patch("cryptography.x509.load_der_x509_certificate",
+                return_value=self.certs[0]):
+            with self.assertRaisesRegex(ValueError, "certificate"):
+                extract_certs(value)
 
 
 class TestBytesConversion(unittest.TestCase):

tests/test_optional_thumbprint.py

@@ -1,5 +1,9 @@
+import base64
 import unittest
+from pathlib import Path
 from unittest.mock import Mock, patch
+from cryptography import x509
+from cryptography.hazmat.primitives import serialization
 from msal.application import ConfidentialClientApplication
 
 
@@ -9,17 +13,15 @@
 class TestClientCredentialWithOptionalThumbprint(unittest.TestCase):
     """Test that thumbprint is optional when public_certificate is provided"""
 
-    # Sample test certificate and private key (PEM format)
-    # These are minimal valid PEM structures for testing
+    # Assertion signing is mocked; certificate parsing uses a real fixture.
     test_private_key = """-----BEGIN PRIVATE KEY-----
 MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC7VJTUt9Us8cKj
 MzEfYyjiWA4R4/M2bS1+fWIcPm15j7uo6xKvRr4PNx5bKMDFqMdW6/xfqFWX0nZK
 -----END PRIVATE KEY-----"""
 
-    test_certificate = """-----BEGIN CERTIFICATE-----
-MIIC5jCCAc6gAwIBAgIJALdYQVsVsNZHMA0GCSqGSIb3DQEBCwUAMBYxFDASBgNV
-BAMMC0V4YW1wbGUgQ0EwHhcNMjQwMTAxMDAwMDAwWhcNMjUwMTAxMDAwMDAwWjAW
------END CERTIFICATE-----"""
+    test_certificate = x509.load_pem_x509_certificate(
+        Path(__file__).with_name("certificate-with-password.pem").read_bytes()
+        ).public_bytes(serialization.Encoding.PEM).decode("ascii")
 
     def _setup_mocks(self, mock_authority_class, authority="https://login.microsoftonline.com/common"):
         """Helper to setup Authority mock"""
@@ -157,13 +159,18 @@ def test_pem_with_both_uses_manual_thumbprint_as_sha1(
         )
 
         # Verify manual thumbprint takes precedence (backward compatibility)
-        self._verify_assertion_params(
+        call_args = self._verify_assertion_params(
             mock_jwt_creator_class,
             expected_algorithm='RS256',
             expected_thumbprint_type='sha1',
             expected_thumbprint_value=manual_thumbprint,
             has_x5c=True  # x5c should still be present
         )
+        cert = x509.load_pem_x509_certificate(self.test_certificate.encode("ascii"))
+        self.assertEqual(
+            [base64.b64encode(cert.public_bytes(
+                serialization.Encoding.DER)).decode("ascii")],
+            call_args[1]["headers"]["x5c"])
 
     @patch('cryptography.x509.load_pem_x509_certificate')
     @patch('msal.application._extract_cert_and_thumbprints')
@@ -191,6 +198,19 @@ def test_pem_with_adfs_uses_sha1(
             expected_thumbprint_type='sha1'
         )
 
+    def test_manual_thumbprint_rejects_malformed_public_certificate(
+            self, mock_jwt_creator_class, mock_authority_class):
+        self._setup_mocks(mock_authority_class)
+        with self.assertRaisesRegex(ValueError, "certificate"):
+            ConfidentialClientApplication(
+                "my_client_id",
+                client_credential={
+                    "private_key": self.test_private_key,
+                    "thumbprint": "A1B2C3D4E5F6",
+                    "public_certificate": "bm90IGEgY2VydGlmaWNhdGU=",
+                })
+        mock_jwt_creator_class.assert_not_called()
+
     def test_pem_with_neither_raises_error(self, mock_jwt_creator_class, mock_authority_class):
         """Test that providing neither thumbprint nor certificate raises ValueError"""
         authority = "https://login.microsoftonline.com/common"