Add support for certificate credential authentication using JWT assertions (#247)

Author: jannispl

README.md

@@ -69,13 +69,13 @@ In all cases, when registering, make sure your client is set up to use an OAuth
 It is also highly recommended to use a scope that will grant "offline" access (i.e., a way to [refresh the OAuth 2.0 authentication token](https://oauth.net/2/refresh-tokens/) without user intervention).
 The [sample configuration file](https://github.com/simonrob/email-oauth2-proxy/blob/main/emailproxy.config) provides example scope values for several common providers.
 
-- Office 365: register a new [Microsoft identity application](https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app)
+- Office 365: register a new [Microsoft identity application](https://learn.microsoft.com/entra/identity-platform/quickstart-register-app)
 - Gmail / Google Workspace: register a [Google API desktop app client](https://developers.google.com/identity/protocols/oauth2/native-app)
 - AOL and Yahoo Mail (and subproviders such as AT&T) are not currently allowing new client registrations with the OAuth email scope – the only option here is to reuse the credentials from an existing client that does have this permission.
 
 The proxy supports [Google Cloud service accounts](https://cloud.google.com/iam/docs/service-account-overview) for access to Google Workspace Gmail.
-It also supports the [client credentials grant (CCG)](https://learn.microsoft.com/en-us/entra/identity-platform/v2-oauth2-client-creds-grant-flow) and [resource owner password credentials grant (ROPCG)](https://learn.microsoft.com/en-us/entra/identity-platform/v2-oauth-ropc) OAuth 2.0 flows.
-Please note that currently only Office 365 is known to support the CCG and ROPCG methods.
+It also supports the [client credentials grant (CCG)](https://learn.microsoft.com/entra/identity-platform/v2-oauth2-client-creds-grant-flow) and [resource owner password credentials grant (ROPCG)](https://learn.microsoft.com/entra/identity-platform/v2-oauth-ropc) OAuth 2.0 flows, and [certificate credentials (JWT)](https://learn.microsoft.com/entra/identity-platform/certificate-credentials).
+Please note that currently only Office 365 is known to support the CCG, ROPCG and certificate credentials methods.
 See the [sample configuration file](https://github.com/simonrob/email-oauth2-proxy/blob/main/emailproxy.config) for further details.
 
 

emailproxy.config

@@ -123,6 +123,10 @@ documentation = Accounts are specified using your email address as the section h
 	- It is possible to create Office 365 clients that do not require a secret to be sent. If this is the case for your
 	setup, delete the `client_secret` line from your account's configuration entry (do not leave the default value).
 
+	- To use O365 certificate credentials instead of a client secret, delete the `client_secret` line and instead
+	provide a `jwt_certificate_path` (e.g., /path/to/certificate.pem) and `jwt_key_path` (e.g., /path/to/key.pem).
+	Further documentation and examples can be found at https://github.com/simonrob/email-oauth2-proxy/pull/247.
+
 	- The proxy supports the client credentials grant (CCG) and resource owner password credentials grant (ROPCG) OAuth
 	2.0 flows (both currently only known to be available for Office 365). To use either of these flows, add an account
 	entry as normal, but do not add a `permission_url` value (it does not apply, and its absence signals to the proxy to

emailproxy.py

@@ -6,7 +6,7 @@
 __author__ = 'Simon Robinson'
 __copyright__ = 'Copyright (c) 2024 Simon Robinson'
 __license__ = 'Apache 2.0'
-__version__ = '2024-05-23'  # ISO 8601 (YYYY-MM-DD)
+__version__ = '2024-05-25'  # ISO 8601 (YYYY-MM-DD)
 __package_version__ = '.'.join([str(int(i)) for i in __version__.split('-')])  # for pyproject.toml usage only
 
 import abc
@@ -176,6 +176,7 @@ class NSObject:
 AUTHENTICATION_TIMEOUT = 600
 
 TOKEN_EXPIRY_MARGIN = 600  # seconds before its expiry to refresh the OAuth 2.0 token
+JWT_LIFETIME = 300  # seconds to add to the current time and use for the `exp` value in JWT certificate credentials
 
 LOG_FILE_MAX_SIZE = 32 * 1024 * 1024  # when using a log file, its maximum size in bytes before rollover (0 = no limit)
 LOG_FILE_MAX_BACKUPS = 10  # the number of log files to keep when LOG_FILE_MAX_SIZE is exceeded (0 = disable rollover)
@@ -713,6 +714,8 @@ def get_oauth2_credentials(username, password, reload_remote_accounts=True):
         client_secret = AppConfig.get_option_with_catch_all_fallback(config, username, 'client_secret')
         client_secret_encrypted = AppConfig.get_option_with_catch_all_fallback(config, username,
                                                                                'client_secret_encrypted')
+        jwt_certificate_path = AppConfig.get_option_with_catch_all_fallback(config, username, 'jwt_certificate_path')
+        jwt_key_path = AppConfig.get_option_with_catch_all_fallback(config, username, 'jwt_key_path')
 
         # note that we don't require permission_url here because it is not needed for the client credentials grant flow,
         # and likewise for client_secret here because it can be optional for Office 365 configurations
@@ -772,13 +775,55 @@ def get_oauth2_credentials(username, password, reload_remote_accounts=True):
                             APP_NAME, username)
                 else:
                     Log.info('Warning: found both `client_secret_encrypted` and `client_secret` for account', username,
-                             ' - the un-encrypted value will be used. Removing the un-encrypted value is recommended')
+                             '- the un-encrypted value will be used. Removing the un-encrypted value is recommended')
+
+            # O365 certificate credentials - see: learn.microsoft.com/entra/identity-platform/certificate-credentials
+            jwt_client_assertion = None
+            if jwt_certificate_path and jwt_key_path:
+                if client_secret or client_secret_encrypted:
+                    client_secret_type = '`client_secret%s`' % ('_encrypted' if client_secret_encrypted else '')
+                    Log.info('Warning: found both certificate credentials and', client_secret_type, 'for account',
+                             username, '- the', client_secret_type, 'value will be used. To use certificate',
+                             'credentials, remove the client secret value')
+
+                else:
+                    try:
+                        # noinspection PyUnresolvedReferences
+                        import jwt
+                    except ImportError:
+                        return False, ('Unable to load jwt, which is a requirement when using certificate credentials '
+                                       '(`jwt_` options). Please run `python -m pip install -r requirements-core.txt`')
+                    import uuid
+                    from cryptography import x509
+                    from cryptography.hazmat.primitives import serialization
+
+                    try:
+                        jwt_now = datetime.datetime.now(datetime.timezone.utc)
+                        jwt_certificate_fingerprint = x509.load_pem_x509_certificate(
+                            pathlib.Path(jwt_certificate_path).read_bytes()).fingerprint(hashes.SHA256())
+                        jwt_client_assertion = jwt.encode(
+                            {
+                                'aud': token_url,
+                                'exp': jwt_now + datetime.timedelta(seconds=JWT_LIFETIME),
+                                'iss': client_id,
+                                'jti': str(uuid.uuid4()),
+                                'nbf': jwt_now,
+                                'sub': client_id
+                            },
+                            serialization.load_pem_private_key(pathlib.Path(jwt_key_path).read_bytes(), password=None),
+                            algorithm='RS256',
+                            headers={
+                                'x5t#S256': base64.urlsafe_b64encode(jwt_certificate_fingerprint).decode('utf-8')
+                            })
+                    except FileNotFoundError:
+                        return (False, 'Unable to create credentials assertion for account %s - please check that the '
+                                       '`jwt_certificate_path` and `jwt_key_path` values are correct' % username)
 
             if access_token or refresh_token:  # if possible, refresh the existing token(s)
                 if not access_token or access_token_expiry - current_time < TOKEN_EXPIRY_MARGIN:
                     if refresh_token:
                         response = OAuth2Helper.refresh_oauth2_access_token(token_url, client_id, client_secret,
-                                                                            username,
+                                                                            jwt_client_assertion, username,
                                                                             cryptographer.decrypt(refresh_token))
 
                         access_token = response['access_token']
@@ -822,8 +867,9 @@ def get_oauth2_credentials(username, password, reload_remote_accounts=True):
                                    '`permission_url`' % (APP_NAME, username))
 
                 response = OAuth2Helper.get_oauth2_authorisation_tokens(token_url, redirect_uri, client_id,
-                                                                        client_secret, auth_result, oauth2_scope,
-                                                                        oauth2_flow, username, password)
+                                                                        client_secret, jwt_client_assertion,
+                                                                        auth_result, oauth2_scope, oauth2_flow,
+                                                                        username, password)
 
                 if AppConfig.get_global('encrypt_client_secret_on_first_use', fallback=False):
                     if client_secret:
@@ -1051,8 +1097,8 @@ def get_oauth2_authorisation_code(permission_url, redirect_uri, redirect_listen_
                 time.sleep(1)
 
     @staticmethod
-    def get_oauth2_authorisation_tokens(token_url, redirect_uri, client_id, client_secret, authorisation_code,
-                                        oauth2_scope, oauth2_flow, username, password):
+    def get_oauth2_authorisation_tokens(token_url, redirect_uri, client_id, client_secret, jwt_client_assertion,
+                                        authorisation_code, oauth2_scope, oauth2_flow, username, password):
         """Requests OAuth 2.0 access and refresh tokens from token_url using the given client_id, client_secret,
         authorisation_code and redirect_uri, returning a dict with 'access_token', 'expires_in', and 'refresh_token'
         on success, or throwing an exception on failure (e.g., HTTP 400)"""
@@ -1064,6 +1110,12 @@ def get_oauth2_authorisation_tokens(token_url, redirect_uri, client_id, client_s
                   'redirect_uri': redirect_uri, 'grant_type': oauth2_flow}
         if not client_secret:
             del params['client_secret']  # client secret can be optional for O365, but we don't want a None entry
+
+            # certificate credentials are only used when no client secret is provided
+            if jwt_client_assertion:
+                params['client_assertion_type'] = 'urn:ietf:params:oauth:client-assertion-type:jwt-bearer'
+                params['client_assertion'] = jwt_client_assertion
+
         if oauth2_flow != 'authorization_code':
             del params['code']  # CCG/ROPCG flows have no code, but we need the scope and (for ROPCG) username+password
             params['scope'] = oauth2_scope
@@ -1115,13 +1167,19 @@ def get_service_account_authorisation_token(key_type, key_path_or_contents, oaut
         return {'access_token': credentials.token, 'expires_in': int(credentials.expiry.timestamp() - time.time())}
 
     @staticmethod
-    def refresh_oauth2_access_token(token_url, client_id, client_secret, username, refresh_token):
+    def refresh_oauth2_access_token(token_url, client_id, client_secret, jwt_client_assertion, username, refresh_token):
         """Obtains a new access token from token_url using the given client_id, client_secret and refresh token,
         returning a dict with 'access_token', 'expires_in', and 'refresh_token' on success; exception on failure"""
         params = {'client_id': client_id, 'client_secret': client_secret, 'refresh_token': refresh_token,
                   'grant_type': 'refresh_token'}
         if not client_secret:
             del params['client_secret']  # client secret can be optional for O365, but we don't want a None entry
+
+            # certificate credentials are only used when no client secret is provided
+            if jwt_client_assertion:
+                params['client_assertion_type'] = 'urn:ietf:params:oauth:client-assertion-type:jwt-bearer'
+                params['client_assertion'] = jwt_client_assertion
+
         try:
             response = urllib.request.urlopen(
                 urllib.request.Request(token_url, data=urllib.parse.urlencode(params).encode('utf-8'),

pyproject.toml

@@ -1,5 +1,5 @@
 [build-system]
-requires = ["setuptools>=61.0", "pyasyncore; python_version >= '3.12'", "cryptography"] # core requirements are needed for version detection, which requires importing the script
+requires = ["setuptools>=61.0", "pyasyncore; python_version >= '3.12'", "cryptography"] # core requirements are needed for version detection when building for PyPI, which requires importing (but not running) the script on `ubuntu-latest`
 build-backend = "setuptools.build_meta"
 
 [project]

requirements-core.txt

@@ -13,3 +13,6 @@ pyoslog>=0.3.0; sys_platform == 'darwin'
 
 # required only if using the --external-auth option in --no-gui mode
 prompt_toolkit
+
+# required only if using JWT certificate credentials (O365)
+pyjwt>=2.4