From 8721c43e95b062b43ce3acf9a3f743219e5df830 Mon Sep 17 00:00:00 2001 From: Abhidnya Patil Date: Tue, 18 Jun 2019 13:06:58 -0700 Subject: [PATCH 1/7] Initial commit for adding SNI auth --- msal/application.py | 4 +++- msal/oauth2cli/assertion.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/msal/application.py b/msal/application.py index b5911c98..15029985 100644 --- a/msal/application.py +++ b/msal/application.py @@ -109,13 +109,15 @@ def __init__( def _build_client(self, client_credential, authority): client_assertion = None client_assertion_type = None + public_certificate = None default_body = {"client_info": 1} if isinstance(client_credential, dict): assert ("private_key" in client_credential and "thumbprint" in client_credential) signer = JwtSigner( client_credential["private_key"], algorithm="RS256", - sha1_thumbprint=client_credential.get("thumbprint")) + sha1_thumbprint=client_credential.get("thumbprint"), + public_certificate = client_credential['public_certificate']) client_assertion = signer.sign_assertion( audience=authority.token_endpoint, issuer=self.client_id) client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT diff --git a/msal/oauth2cli/assertion.py b/msal/oauth2cli/assertion.py index bd2373a7..b5eb52c9 100644 --- a/msal/oauth2cli/assertion.py +++ b/msal/oauth2cli/assertion.py @@ -18,7 +18,7 @@ def sign_assertion( class JwtSigner(Signer): - def __init__(self, key, algorithm, sha1_thumbprint=None, headers=None): + def __init__(self, key, algorithm, sha1_thumbprint=None, headers=None, public_certificate= None): """Create a signer. Args: @@ -36,6 +36,8 @@ def __init__(self, key, algorithm, sha1_thumbprint=None, headers=None): if sha1_thumbprint: # https://tools.ietf.org/html/rfc7515#section-4.1.7 self.headers["x5t"] = base64.urlsafe_b64encode( binascii.a2b_hex(sha1_thumbprint)).decode() + if public_certificate: + self.headers['x5c'] = public_certificate def sign_assertion( self, audience, issuer, subject=None, expires_at=None, From 2aefc6d7771fee70bddee3dfc40dfd04a5240f74 Mon Sep 17 00:00:00 2001 From: Abhidnya Patil Date: Thu, 27 Jun 2019 15:58:08 -0700 Subject: [PATCH 2/7] Sending list of strings as x5c and refactoring --- msal/application.py | 18 +++++++++++++----- msal/oauth2cli/assertion.py | 4 +--- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/msal/application.py b/msal/application.py index 15029985..3c11ceb1 100644 --- a/msal/application.py +++ b/msal/application.py @@ -109,15 +109,23 @@ def __init__( def _build_client(self, client_credential, authority): client_assertion = None client_assertion_type = None - public_certificate = None default_body = {"client_info": 1} if isinstance(client_credential, dict): assert ("private_key" in client_credential and "thumbprint" in client_credential) - signer = JwtSigner( - client_credential["private_key"], algorithm="RS256", - sha1_thumbprint=client_credential.get("thumbprint"), - public_certificate = client_credential['public_certificate']) + if 'public_certificate' in client_credential: + if isinstance(client_credential['public_certificate'], list): + public_certificate = client_credential['public_certificate'] + else: + public_certificate = [client_credential['public_certificate']] + signer = JwtSigner( + client_credential["private_key"], algorithm="RS256", + sha1_thumbprint=client_credential.get("thumbprint"), + headers={"x5c": public_certificate}) + else: + signer = JwtSigner( + client_credential["private_key"], algorithm="RS256", + sha1_thumbprint=client_credential.get("thumbprint")) client_assertion = signer.sign_assertion( audience=authority.token_endpoint, issuer=self.client_id) client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT diff --git a/msal/oauth2cli/assertion.py b/msal/oauth2cli/assertion.py index b5eb52c9..bd2373a7 100644 --- a/msal/oauth2cli/assertion.py +++ b/msal/oauth2cli/assertion.py @@ -18,7 +18,7 @@ def sign_assertion( class JwtSigner(Signer): - def __init__(self, key, algorithm, sha1_thumbprint=None, headers=None, public_certificate= None): + def __init__(self, key, algorithm, sha1_thumbprint=None, headers=None): """Create a signer. Args: @@ -36,8 +36,6 @@ def __init__(self, key, algorithm, sha1_thumbprint=None, headers=None, public_ce if sha1_thumbprint: # https://tools.ietf.org/html/rfc7515#section-4.1.7 self.headers["x5t"] = base64.urlsafe_b64encode( binascii.a2b_hex(sha1_thumbprint)).decode() - if public_certificate: - self.headers['x5c'] = public_certificate def sign_assertion( self, audience, issuer, subject=None, expires_at=None, From c60c7510d24941f8ee3dc5695065b19636e91607 Mon Sep 17 00:00:00 2001 From: Abhidnya Patil Date: Thu, 27 Jun 2019 17:16:49 -0700 Subject: [PATCH 3/7] More refactoring --- msal/application.py | 19 +++++++------------ 1 file changed, 7 insertions(+), 12 deletions(-) diff --git a/msal/application.py b/msal/application.py index 3c11ceb1..8bb197ad 100644 --- a/msal/application.py +++ b/msal/application.py @@ -110,22 +110,17 @@ def _build_client(self, client_credential, authority): client_assertion = None client_assertion_type = None default_body = {"client_info": 1} + headers = {} if isinstance(client_credential, dict): assert ("private_key" in client_credential and "thumbprint" in client_credential) if 'public_certificate' in client_credential: - if isinstance(client_credential['public_certificate'], list): - public_certificate = client_credential['public_certificate'] - else: - public_certificate = [client_credential['public_certificate']] - signer = JwtSigner( - client_credential["private_key"], algorithm="RS256", - sha1_thumbprint=client_credential.get("thumbprint"), - headers={"x5c": public_certificate}) - else: - signer = JwtSigner( - client_credential["private_key"], algorithm="RS256", - sha1_thumbprint=client_credential.get("thumbprint")) + headers["x5c"] = client_credential['public_certificate'] \ + if isinstance(client_credential['public_certificate'], list) \ + else [client_credential['public_certificate']] # We send x5c as a list of strings + signer = JwtSigner( + client_credential["private_key"], algorithm="RS256", + 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 From c8f199c6bb6a470c0578474b15fd696676f40ccb Mon Sep 17 00:00:00 2001 From: Abhidnya Patil Date: Thu, 27 Jun 2019 21:09:00 -0700 Subject: [PATCH 4/7] Adding regex to parse public certificates and reference documentation changes --- msal/application.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/msal/application.py b/msal/application.py index 8bb197ad..9fc0ecd9 100644 --- a/msal/application.py +++ b/msal/application.py @@ -59,7 +59,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 +69,11 @@ def __init__( { "private_key": "...-----BEGIN PRIVATE KEY-----...", "thumbprint": "A1B2C3D4E5F6...", + "public_certificate": "...-----BEGIN CERTIFICATE-----..." (Only to be sent when using Subject Name Issuer Authentication) } + public_certificate (optional) can be a public key certificate or certificate chain which is sent through + 'x5c' JWT header only for subject name and issuer based 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 @@ -115,9 +118,10 @@ def _build_client(self, client_credential, authority): assert ("private_key" in client_credential and "thumbprint" in client_credential) if 'public_certificate' in client_credential: - headers["x5c"] = client_credential['public_certificate'] \ - if isinstance(client_credential['public_certificate'], list) \ - else [client_credential['public_certificate']] # We send x5c as a list of strings + public_certificates = re.findall( + r'\-+BEGIN CERTIFICATE.+\-+(?P[^-]+)\-+END CERTIFICATE.+\-+', + client_credential['public_certificate'], re.I) # We send x5c as list of strings + headers["x5c"] = [cert.strip() for cert in public_certificates] signer = JwtSigner( client_credential["private_key"], algorithm="RS256", sha1_thumbprint=client_credential.get("thumbprint"), headers=headers) From 7b342c186a84110e5bd8bd2bc7df5ecf309fe8d8 Mon Sep 17 00:00:00 2001 From: Abhidnya Patil Date: Mon, 1 Jul 2019 15:40:22 -0700 Subject: [PATCH 5/7] Adding tests, refactoring, support for input file without begin and end certificate string --- msal/application.py | 14 +++++--- tests/test_application.py | 70 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 4 deletions(-) diff --git a/msal/application.py b/msal/application.py index 9fc0ecd9..132c7222 100644 --- a/msal/application.py +++ b/msal/application.py @@ -50,6 +50,7 @@ def decorate_scope( return list(decorated) + class ClientApplication(object): def __init__( @@ -109,6 +110,14 @@ def __init__( self.client = self._build_client(client_credential, self.authority) self.authority_groups = None + def _extract_x5c_value(self): + public_certificates = re.findall( + r'\-+BEGIN CERTIFICATE.+\-+(?P[^-]+)\-+END CERTIFICATE.+\-+', + self.client_credential['public_certificate'], re.I) # We send x5c as list of strings + if len(public_certificates): + return [cert.strip() for cert in public_certificates] + return [self.client_credential['public_certificate'].strip()] + def _build_client(self, client_credential, authority): client_assertion = None client_assertion_type = None @@ -118,10 +127,7 @@ def _build_client(self, client_credential, authority): assert ("private_key" in client_credential and "thumbprint" in client_credential) if 'public_certificate' in client_credential: - public_certificates = re.findall( - r'\-+BEGIN CERTIFICATE.+\-+(?P[^-]+)\-+END CERTIFICATE.+\-+', - client_credential['public_certificate'], re.I) # We send x5c as list of strings - headers["x5c"] = [cert.strip() for cert in public_certificates] + headers["x5c"] = self._extract_x5c_value() signer = JwtSigner( client_credential["private_key"], algorithm="RS256", sha1_thumbprint=client_credential.get("thumbprint"), headers=headers) diff --git a/tests/test_application.py b/tests/test_application.py index 3860c735..a070a04b 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -99,6 +99,76 @@ def test_client_certificate(self): self.assertIn('access_token', result) self.assertCacheWorks(result, app.acquire_token_silent(scope, account=None)) + @unittest.skipUnless("public_certificate" in CONFIG, "Missing Public cert") + def test_subject_name_issuer_authentication_input_with_begin_certificate_and_end_certificate_single(self): + assert ("private_key_file" in CONFIG + and "thumbprint" in CONFIG and "public_certificate" in CONFIG) + key_path = os.path.join(THIS_FOLDER, CONFIG['private_key_file']) + public_certificate = os.path.join(THIS_FOLDER, CONFIG['public_certificate']) + with open(key_path) as f: + pem = f.read() + with open(public_certificate) as f: + public_certificate = f.read() + app = ConfidentialClientApplication( + CONFIG['client_id'], + client_credential={"private_key": pem, "thumbprint": CONFIG["thumbprint"], + "public_certificate": public_certificate}) + x5c = app._extract_x5c_value() + self.assertIsInstance(x5c, list) + self.assertEquals(len(x5c), 1) + + @unittest.skipUnless("public_certificate" in CONFIG, "Missing Public cert") + def test_subject_name_issuer_authentication_input_with_begin_certificate_and_end_certificate_multiple(self): + assert ("private_key_file" in CONFIG + and "thumbprint" in CONFIG and "public_certificate" in CONFIG) + key_path = os.path.join(THIS_FOLDER, CONFIG['private_key_file']) + public_certificate = os.path.join(THIS_FOLDER, CONFIG['public_certificate']) + with open(key_path) as f: + pem = f.read() + with open(public_certificate) as f: + public_certificate = f.read() + app = ConfidentialClientApplication( + CONFIG['client_id'], + {"private_key": pem, "thumbprint": CONFIG["thumbprint"], "public_certificate": public_certificate}) + x5c = app._extract_x5c_value() + self.assertIsInstance(x5c, list) + self.assertGreater(len(x5c), 1) + + @unittest.skipUnless("public_certificate" in CONFIG, "Missing Public cert") + def test_subject_name_issuer_authentication_input_without_begin_certificate_and_end_certificate(self): + assert ("private_key_file" in CONFIG + and "thumbprint" in CONFIG and "public_certificate" in CONFIG) + key_path = os.path.join(THIS_FOLDER, CONFIG['private_key_file']) + public_certificate = os.path.join(THIS_FOLDER, CONFIG['public_certificate']) + with open(key_path) as f: + pem = f.read() + with open(public_certificate) as f: + public_certificate = f.read() + app = ConfidentialClientApplication( + CONFIG['client_id'], + {"private_key": pem, "thumbprint": CONFIG["thumbprint"], "public_certificate": public_certificate}) + x5c = app._extract_x5c_value() + self.assertIsInstance(x5c, list) + self.assertEquals(len(x5c), 1) + + @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) + key_path = os.path.join(THIS_FOLDER, CONFIG['private_key_file']) + public_certificate = os.path.join(THIS_FOLDER, CONFIG['public_certificate']) + with open(key_path) as f: + pem = f.read() + with open(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): From a68bea7c441eaa0540f42e751527caa326c4568b Mon Sep 17 00:00:00 2001 From: Abhidnya Patil Date: Mon, 1 Jul 2019 23:02:18 -0700 Subject: [PATCH 6/7] Addressing PR comments --- msal/application.py | 29 +++++++------- tests/test_application.py | 80 ++++++++++++--------------------------- 2 files changed, 41 insertions(+), 68 deletions(-) diff --git a/msal/application.py b/msal/application.py index 132c7222..cd7bea15 100644 --- a/msal/application.py +++ b/msal/application.py @@ -50,6 +50,16 @@ 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 len(public_certificates): + return [cert.strip() for cert in public_certificates] + return [public_cert_content.strip()] + class ClientApplication(object): @@ -70,11 +80,12 @@ def __init__( { "private_key": "...-----BEGIN PRIVATE KEY-----...", "thumbprint": "A1B2C3D4E5F6...", - "public_certificate": "...-----BEGIN CERTIFICATE-----..." (Only to be sent when using Subject Name Issuer Authentication) + "public_certificate": "...-----BEGIN CERTIFICATE-----..." (Optional. See below.) } - public_certificate (optional) can be a public key certificate or certificate chain which is sent through - 'x5c' JWT header only for subject name and issuer based authentication to support cert auto rolls + 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 @@ -110,24 +121,16 @@ def __init__( self.client = self._build_client(client_credential, self.authority) self.authority_groups = None - def _extract_x5c_value(self): - public_certificates = re.findall( - r'\-+BEGIN CERTIFICATE.+\-+(?P[^-]+)\-+END CERTIFICATE.+\-+', - self.client_credential['public_certificate'], re.I) # We send x5c as list of strings - if len(public_certificates): - return [cert.strip() for cert in public_certificates] - return [self.client_credential['public_certificate'].strip()] - def _build_client(self, client_credential, authority): client_assertion = None client_assertion_type = None default_body = {"client_info": 1} - headers = {} 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"] = self._extract_x5c_value() + headers["x5c"] = extract_certs(client_credential['public_certificate']) signer = JwtSigner( client_credential["private_key"], algorithm="RS256", sha1_thumbprint=client_credential.get("thumbprint"), headers=headers) diff --git a/tests/test_application.py b/tests/test_application.py index a070a04b..ff2bc859 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -99,67 +99,37 @@ def test_client_certificate(self): self.assertIn('access_token', result) self.assertCacheWorks(result, app.acquire_token_silent(scope, account=None)) - @unittest.skipUnless("public_certificate" in CONFIG, "Missing Public cert") - def test_subject_name_issuer_authentication_input_with_begin_certificate_and_end_certificate_single(self): - assert ("private_key_file" in CONFIG - and "thumbprint" in CONFIG and "public_certificate" in CONFIG) - key_path = os.path.join(THIS_FOLDER, CONFIG['private_key_file']) - public_certificate = os.path.join(THIS_FOLDER, CONFIG['public_certificate']) - with open(key_path) as f: - pem = f.read() - with open(public_certificate) as f: - public_certificate = f.read() - app = ConfidentialClientApplication( - CONFIG['client_id'], - client_credential={"private_key": pem, "thumbprint": CONFIG["thumbprint"], - "public_certificate": public_certificate}) - x5c = app._extract_x5c_value() - self.assertIsInstance(x5c, list) - self.assertEquals(len(x5c), 1) - - @unittest.skipUnless("public_certificate" in CONFIG, "Missing Public cert") - def test_subject_name_issuer_authentication_input_with_begin_certificate_and_end_certificate_multiple(self): - assert ("private_key_file" in CONFIG - and "thumbprint" in CONFIG and "public_certificate" in CONFIG) - key_path = os.path.join(THIS_FOLDER, CONFIG['private_key_file']) - public_certificate = os.path.join(THIS_FOLDER, CONFIG['public_certificate']) - with open(key_path) as f: - pem = f.read() - with open(public_certificate) as f: - public_certificate = f.read() - app = ConfidentialClientApplication( - CONFIG['client_id'], - {"private_key": pem, "thumbprint": CONFIG["thumbprint"], "public_certificate": public_certificate}) - x5c = app._extract_x5c_value() - self.assertIsInstance(x5c, list) - self.assertGreater(len(x5c), 1) - - @unittest.skipUnless("public_certificate" in CONFIG, "Missing Public cert") - def test_subject_name_issuer_authentication_input_without_begin_certificate_and_end_certificate(self): - assert ("private_key_file" in CONFIG - and "thumbprint" in CONFIG and "public_certificate" in CONFIG) - key_path = os.path.join(THIS_FOLDER, CONFIG['private_key_file']) - public_certificate = os.path.join(THIS_FOLDER, CONFIG['public_certificate']) - with open(key_path) as f: - pem = f.read() - with open(public_certificate) as f: - public_certificate = f.read() - app = ConfidentialClientApplication( - CONFIG['client_id'], - {"private_key": pem, "thumbprint": CONFIG["thumbprint"], "public_certificate": public_certificate}) - x5c = app._extract_x5c_value() - self.assertIsInstance(x5c, list) - self.assertEquals(len(x5c), 1) + 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) - key_path = os.path.join(THIS_FOLDER, CONFIG['private_key_file']) - public_certificate = os.path.join(THIS_FOLDER, CONFIG['public_certificate']) - with open(key_path) as f: + with open(os.path.join(THIS_FOLDER, CONFIG['private_key_file'])) as f: pem = f.read() - with open(public_certificate) as f: + with open(os.path.join(THIS_FOLDER, CONFIG['public_certificate'])) as f: public_certificate = f.read() app = ConfidentialClientApplication( CONFIG['client_id'], authority=CONFIG["authority"], From 7e392c788857e87c328e92f6aad4aa757f40b8d2 Mon Sep 17 00:00:00 2001 From: Abhidnya Patil Date: Tue, 2 Jul 2019 13:31:26 -0700 Subject: [PATCH 7/7] Addressing PR comments --- msal/application.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/msal/application.py b/msal/application.py index cd7bea15..efb3ca43 100644 --- a/msal/application.py +++ b/msal/application.py @@ -54,10 +54,15 @@ 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.+\-+', + r'-----BEGIN CERTIFICATE-----(?P[^-]+)-----END CERTIFICATE-----', public_cert_content, re.I) - if len(public_certificates): + 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()]