From 2c765ae5a1e4a355420a75a27cf4175cdee0b52c Mon Sep 17 00:00:00 2001 From: Bogdan Gavril Date: Mon, 9 Nov 2026 12:47:51 +0000 Subject: [PATCH] app: validate public certificate encodings Parse public certificate data before constructing x5c and serialize only certificate DER. Preserve PEM chain order and whitespace-tolerant tagless base64 DER input while rejecting malformed or trailing data. Document the input validation and cover supported forms and setup errors. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 14b467bc-6ea2-4b1f-bd35-38f826d5a90c --- msal/application.py | 35 ++++++++---- tests/test_application.py | 90 ++++++++++++++++++++++++------- tests/test_optional_thumbprint.py | 34 +++++++++--- 3 files changed, 125 insertions(+), 34 deletions(-) diff --git a/msal/application.py b/msal/application.py index 1aa14c28..f4409747 100644 --- a/msal/application.py +++ b/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[^-]+)-----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.*?)-----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 `_ which is an approach to allow easier certificate rotation. diff --git a/tests/test_application.py b/tests/test_application.py index 7e5a64dd..69dd6d1b 100644 --- a/tests/test_application.py +++ b/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): diff --git a/tests/test_optional_thumbprint.py b/tests/test_optional_thumbprint.py index 56ec4013..12a8ce71 100644 --- a/tests/test_optional_thumbprint.py +++ b/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"