diff --git a/msal/application.py b/msal/application.py index b5911c98..efb3ca43 100644 --- a/msal/application.py +++ b/msal/application.py @@ -50,6 +50,22 @@ def decorate_scope( return list(decorated) +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())} + 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: + raise ValueError( + "We expect your public key but detect a private key instead") + return [public_cert_content.strip()] + + class ClientApplication(object): def __init__( @@ -59,7 +75,7 @@ def __init__( verify=True, proxies=None, timeout=None): """Create an instance of application. - :param client_id: Your app has a clinet_id after you register it on AAD. + :param client_id: Your app has a client_id after you register it on AAD. :param client_credential: For :class:`PublicClientApplication`, you simply use `None` here. For :class:`ConfidentialClientApplication`, @@ -69,8 +85,12 @@ def __init__( { "private_key": "...-----BEGIN PRIVATE KEY-----...", "thumbprint": "A1B2C3D4E5F6...", + "public_certificate": "...-----BEGIN CERTIFICATE-----..." (Optional. See below.) } + public_certificate (optional) is public key certificate which is + sent through 'x5c' JWT header only for + subject name and issuer authentication to support cert auto rolls :param str authority: A URL that identifies a token authority. It should be of the format https://login.microsoftonline.com/your_tenant @@ -113,9 +133,12 @@ def _build_client(self, client_credential, authority): if isinstance(client_credential, dict): assert ("private_key" in client_credential and "thumbprint" in client_credential) + headers = {} + if 'public_certificate' in client_credential: + headers["x5c"] = extract_certs(client_credential['public_certificate']) signer = JwtSigner( client_credential["private_key"], algorithm="RS256", - sha1_thumbprint=client_credential.get("thumbprint")) + sha1_thumbprint=client_credential.get("thumbprint"), headers=headers) client_assertion = signer.sign_assertion( audience=authority.token_endpoint, issuer=self.client_id) client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT diff --git a/tests/test_application.py b/tests/test_application.py index 3860c735..ff2bc859 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -99,6 +99,46 @@ def test_client_certificate(self): self.assertIn('access_token', result) self.assertCacheWorks(result, app.acquire_token_silent(scope, account=None)) + def test_extract_a_tag_less_public_cert(self): + pem = "my_cert" + self.assertEqual(["my_cert"], extract_certs(pem)) + + def test_extract_a_tag_enclosed_cert(self): + pem = """ + -----BEGIN CERTIFICATE----- + my_cert + -----END CERTIFICATE----- + """ + self.assertEqual(["my_cert"], extract_certs(pem)) + + 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)) + + @unittest.skipUnless("public_certificate" in CONFIG, "Missing Public cert") + def test_subject_name_issuer_authentication(self): + assert ("private_key_file" in CONFIG + and "thumbprint" in CONFIG and "public_certificate" in CONFIG) + with open(os.path.join(THIS_FOLDER, CONFIG['private_key_file'])) as f: + pem = f.read() + with open(os.path.join(THIS_FOLDER, CONFIG['public_certificate'])) as f: + public_certificate = f.read() + app = ConfidentialClientApplication( + CONFIG['client_id'], authority=CONFIG["authority"], + client_credential={"private_key": pem, "thumbprint": CONFIG["thumbprint"], + "public_certificate": public_certificate}) + scope = CONFIG.get("scope", []) + result = app.acquire_token_for_client(scope) + self.assertIn('access_token', result) + self.assertCacheWorks(result, app.acquire_token_silent(scope, account=None)) @unittest.skipUnless("client_id" in CONFIG, "client_id missing") class TestPublicClientApplication(Oauth2TestCase):