feat: add experimental enterprise cert support (#1052)

* feat: add experimental enterprise cert support

* fix test issue

* resolve comments

.gitignore

@@ -45,4 +45,6 @@ pytype_output/
 .python-version
 .DS_Store
 cert_path
-key_path
+key_path
+env/
+.vscode/

google/auth/transport/_custom_tls_signer.py

@@ -0,0 +1,235 @@
+# Copyright 2022 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http:https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Code for configuring client side TLS to offload the signing operation to
+signing libraries.
+"""
+
+import ctypes
+import json
+import logging
+import os
+import sys
+
+import cffi # type: ignore
+import six
+
+from google.auth import exceptions
+
+_LOGGER = logging.getLogger(__name__)
+
+# C++ offload lib requires google-auth lib to provide the following callback:
+# using SignFunc = int (*)(unsigned char *sig, size_t *sig_len,
+# const unsigned char *tbs, size_t tbs_len)
+# The bytes to be signed and the length are provided via `tbs` and `tbs_len`,
+# the callback computes the signature, and write the signature and its length
+# into `sig` and `sig_len`.
+# If the signing is successful, the callback returns 1, otherwise it returns 0.
+SIGN_CALLBACK_CTYPE = ctypes.CFUNCTYPE(
+ ctypes.c_int, # return type
+ ctypes.POINTER(ctypes.c_ubyte), # sig
+ ctypes.POINTER(ctypes.c_size_t), # sig_len
+ ctypes.POINTER(ctypes.c_ubyte), # tbs
+ ctypes.c_size_t, # tbs_len
+)
+
+
+# Cast SSL_CTX* to void*
+def _cast_ssl_ctx_to_void_p(ssl_ctx):
+ return ctypes.cast(int(cffi.FFI().cast("intptr_t", ssl_ctx)), ctypes.c_void_p)
+
+
+# Load offload library and set up the function types.
+def load_offload_lib(offload_lib_path):
+ _LOGGER.debug("loading offload library from %s", offload_lib_path)
+
+ # winmode parameter is only available for python 3.8+.
+ lib = (
+ ctypes.CDLL(offload_lib_path, winmode=0)
+ if sys.version_info >= (3, 8) and os.name == "nt"
+ else ctypes.CDLL(offload_lib_path)
+ )
+
+ # Set up types for:
+ # int ConfigureSslContext(SignFunc sign_func, const char *cert, SSL_CTX *ctx)
+ lib.ConfigureSslContext.argtypes = [
+ SIGN_CALLBACK_CTYPE,
+ ctypes.c_char_p,
+ ctypes.c_void_p,
+ ]
+ lib.ConfigureSslContext.restype = ctypes.c_int
+
+ return lib
+
+
+# Load signer library and set up the function types.
+# See: https://github.com/googleapis/enterprise-certificate-proxy/blob/main/cshared/main.go
+def load_signer_lib(signer_lib_path):
+ _LOGGER.debug("loading signer library from %s", signer_lib_path)
+
+ # winmode parameter is only available for python 3.8+.
+ lib = (
+ ctypes.CDLL(signer_lib_path, winmode=0)
+ if sys.version_info >= (3, 8) and os.name == "nt"
+ else ctypes.CDLL(signer_lib_path)
+ )
+
+ # Set up types for:
+ # func GetCertPemForPython(configFilePath *C.char, certHolder *byte, certHolderLen int)
+ lib.GetCertPemForPython.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int]
+ # Returns: certLen
+ lib.GetCertPemForPython.restype = ctypes.c_int
+
+ # Set up types for:
+ # func SignForPython(configFilePath *C.char, digest *byte, digestLen int,
+ # sigHolder *byte, sigHolderLen int)
+ lib.SignForPython.argtypes = [
+ ctypes.c_char_p,
+ ctypes.c_char_p,
+ ctypes.c_int,
+ ctypes.c_char_p,
+ ctypes.c_int,
+ ]
+ # Returns: the signature length
+ lib.SignForPython.restype = ctypes.c_int
+
+ return lib
+
+
+# Computes SHA256 hash.
+def _compute_sha256_digest(to_be_signed, to_be_signed_len):
+ from cryptography.hazmat.primitives import hashes
+
+ data = ctypes.string_at(to_be_signed, to_be_signed_len)
+ hash = hashes.Hash(hashes.SHA256())
+ hash.update(data)
+ return hash.finalize()
+
+
+# Create the signing callback. The actual signing work is done by the
+# `SignForPython` method from the signer lib.
+def get_sign_callback(signer_lib, config_file_path):
+ def sign_callback(sig, sig_len, tbs, tbs_len):
+ _LOGGER.debug("calling sign callback...")
+
+ digest = _compute_sha256_digest(tbs, tbs_len)
+ digestArray = ctypes.c_char * len(digest)
+
+ # reserve 2000 bytes for the signature, shoud be more then enough.
+ # RSA signature is 256 bytes, EC signature is 70~72.
+ sig_holder_len = 2000
+ sig_holder = ctypes.create_string_buffer(sig_holder_len)
+
+ signature_len = signer_lib.SignForPython(
+ config_file_path.encode(), # configFilePath
+ digestArray.from_buffer(bytearray(digest)), # digest
+ len(digest), # digestLen
+ sig_holder, # sigHolder
+ sig_holder_len, # sigHolderLen
+ )
+
+ if signature_len == 0:
+ # signing failed, return 0
+ return 0
+
+ sig_len[0] = signature_len
+ bs = bytearray(sig_holder)
+ for i in range(signature_len):
+ sig[i] = bs[i]
+
+ return 1
+
+ return SIGN_CALLBACK_CTYPE(sign_callback)
+
+
+# Obtain the certificate bytes by calling the `GetCertPemForPython` method from
+# the signer lib. The method is called twice, the first time is to compute the
+# cert length, then we create a buffer to hold the cert, and call it again to
+# fill the buffer.
+def get_cert(signer_lib, config_file_path):
+ # First call to calculate the cert length
+ cert_len = signer_lib.GetCertPemForPython(
+ config_file_path.encode(), # configFilePath
+ None, # certHolder
+ 0, # certHolderLen
+ )
+ if cert_len == 0:
+ raise exceptions.MutualTLSChannelError("failed to get certificate")
+
+ # Then we create an array to hold the cert, and call again to fill the cert
+ cert_holder = ctypes.create_string_buffer(cert_len)
+ signer_lib.GetCertPemForPython(
+ config_file_path.encode(), # configFilePath
+ cert_holder, # certHolder
+ cert_len, # certHolderLen
+ )
+ return bytes(cert_holder)
+
+
+class CustomTlsSigner(object):
+ def __init__(self, enterprise_cert_file_path):
+ """
+ This class loads the offload and signer library, and calls APIs from
+ these libraries to obtain the cert and a signing callback, and attach
+ them to SSL context. The cert and the signing callback will be used
+ for client authentication in TLS handshake.
+
+ Args:
+ enterprise_cert_file_path (str): the path to a enterprise cert JSON
+ file. The file should contain the following field:
+
+ {
+ "libs": {
+ "signer_library": "...",
+ "offload_library": "..."
+ }
+ }
+ """
+ self._enterprise_cert_file_path = enterprise_cert_file_path
+ self._cert = None
+ self._sign_callback = None
+
+ def load_libraries(self):
+ try:
+ with open(self._enterprise_cert_file_path, "r") as f:
+ enterprise_cert_json = json.load(f)
+ libs = enterprise_cert_json["libs"]
+ signer_library = libs["signer_library"]
+ offload_library = libs["offload_library"]
+ except (KeyError, ValueError) as caught_exc:
+ new_exc = exceptions.MutualTLSChannelError(
+ "enterprise cert file is invalid", caught_exc
+ )
+ six.raise_from(new_exc, caught_exc)
+ self._offload_lib = load_offload_lib(offload_library)
+ self._signer_lib = load_signer_lib(signer_library)
+
+ def set_up_custom_key(self):
+ # We need to keep a reference of the cert and sign callback so it won't
+ # be garbage collected, otherwise it will crash when used by signer lib.
+ self._cert = get_cert(self._signer_lib, self._enterprise_cert_file_path)
+ self._sign_callback = get_sign_callback(
+ self._signer_lib, self._enterprise_cert_file_path
+ )
+
+ def attach_to_ssl_context(self, ctx):
+ # In the TLS handshake, the signing operation will be done by the
+ # sign_callback.
+ if not self._offload_lib.ConfigureSslContext(
+ self._sign_callback,
+ ctypes.c_char_p(self._cert),
+ _cast_ssl_ctx_to_void_p(ctx._ctx._context),
+ ):
+ raise exceptions.MutualTLSChannelError("failed to configure SSL context")

google/auth/transport/requests.py

@@ -245,6 +245,65 @@ def proxy_manager_for(self, *args, **kwargs):
  return super(_MutualTlsAdapter, self).proxy_manager_for(*args, **kwargs)
 
 
+class _MutualTlsOffloadAdapter(requests.adapters.HTTPAdapter):
+ """
+ A TransportAdapter that enables mutual TLS and offloads the client side
+ signing operation to the signing library.
+
+ Args:
+ enterprise_cert_file_path (str): the path to a enterprise cert JSON
+ file. The file should contain the following field:
+
+ {
+ "libs": {
+ "signer_library": "...",
+ "offload_library": "..."
+ }
+ }
+
+ Raises:
+ ImportError: if certifi or pyOpenSSL is not installed
+ google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
+ creation failed for any reason.
+ """
+
+ def __init__(self, enterprise_cert_file_path):
+ import certifi
+ import urllib3.contrib.pyopenssl
+
+ from google.auth.transport import _custom_tls_signer
+
+ # Call inject_into_urllib3 to activate certificate checking. See the
+ # following links for more info:
+ # (1) doc: https://github.com/urllib3/urllib3/blob/cb9ebf8aac5d75f64c8551820d760b72b619beff/src/urllib3/contrib/pyopenssl.py#L31-L32
+ # (2) mTLS example: https://github.com/urllib3/urllib3/issues/474#issuecomment-253168415
+ urllib3.contrib.pyopenssl.inject_into_urllib3()
+
+ self.signer = _custom_tls_signer.CustomTlsSigner(enterprise_cert_file_path)
+ self.signer.load_libraries()
+ self.signer.set_up_custom_key()
+
+ poolmanager = create_urllib3_context()
+ poolmanager.load_verify_locations(cafile=certifi.where())
+ self.signer.attach_to_ssl_context(poolmanager)
+ self._ctx_poolmanager = poolmanager
+
+ proxymanager = create_urllib3_context()
+ proxymanager.load_verify_locations(cafile=certifi.where())
+ self.signer.attach_to_ssl_context(proxymanager)
+ self._ctx_proxymanager = proxymanager
+
+ super(_MutualTlsOffloadAdapter, self).__init__()
+
+ def init_poolmanager(self, *args, **kwargs):
+ kwargs["ssl_context"] = self._ctx_poolmanager
+ super(_MutualTlsOffloadAdapter, self).init_poolmanager(*args, **kwargs)
+
+ def proxy_manager_for(self, *args, **kwargs):
+ kwargs["ssl_context"] = self._ctx_proxymanager
+ return super(_MutualTlsOffloadAdapter, self).proxy_manager_for(*args, **kwargs)
+
+
 class AuthorizedSession(requests.Session):
  """A Requests Session class with credentials.
 

noxfile.py

@@ -118,6 +118,7 @@ def unit_prev_versions(session):
  "--cov=google.oauth2",
  "--cov=tests",
  "tests",
+ "--ignore=tests/transport/test__custom_tls_signer.py", # enterprise cert is for python 3.6+
  )
 
 

setup.py

@@ -38,6 +38,9 @@
  ],
  "pyopenssl": "pyopenssl>=20.0.0",
  "reauth": "pyu2f>=0.1.5",
+ # Enterprise cert only works for OpenSSL 1.1.1. Newer versions of these
+ # dependencies are built with OpenSSL 3.0 so we need to fix the version.
+ "enterprise_cert": ["cryptography==36.0.2", "pyopenssl==22.0.0"],
 }
 
 with io.open("README.rst", "r") as fh:

tests/data/enterprise_cert_invalid.json

@@ -0,0 +1,3 @@
+{
+ "libs": {}
+}

tests/data/enterprise_cert_valid.json

@@ -0,0 +1,6 @@
+{
+ "libs": {
+ "signer_library": "/path/to/signer/lib",
+ "offload_library": "/path/to/offload/lib"
+ }
+}