refactor(auth): replace pyOpenSSL with standard ssl and cryptography

packages/google-auth/google/auth/aio/transport/mtls.py

@@ -17,42 +17,18 @@
 """
 
 import asyncio
-import contextlib
+import inspect
 import logging
-import os
 import ssl
-import tempfile
 from typing import Optional
 
 from google.auth import exceptions
-import google.auth.transport._mtls_helper
+from google.auth.transport._mtls_helper import secure_cert_key_paths
 import google.auth.transport.mtls
 
 _LOGGER = logging.getLogger(__name__)
 
 
-@contextlib.contextmanager
-def _create_temp_file(content: bytes):
- """Creates a temporary file with the given content.
-
- Args:
- content (bytes): The content to write to the file.
-
- Yields:
- str: The path to the temporary file.
- """
- # Create a temporary file that is readable only by the owner.
- fd, file_path = tempfile.mkstemp()
- try:
- with os.fdopen(fd, "wb") as f:
- f.write(content)
- yield file_path
- finally:
- # Securely delete the file after use.
- if os.path.exists(file_path):
- os.remove(file_path)
-
-
 def make_client_cert_ssl_context(
  cert_bytes: bytes, key_bytes: bytes, passphrase: Optional[bytes] = None
 ) -> ssl.SSLContext:
@@ -71,19 +47,29 @@ def make_client_cert_ssl_context(
  Raises:
  google.auth.exceptions.TransportError: If there is an error loading the certificate.
  """
- with _create_temp_file(cert_bytes) as cert_path, _create_temp_file(
- key_bytes
- ) as key_path:
- try:
+ try:
+ with secure_cert_key_paths(cert_bytes, key_bytes, passphrase=passphrase) as (
+ cert_path,
+ key_path,
+ passphrase_val,
+ ):
  context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
- context.load_cert_chain(
- certfile=cert_path, keyfile=key_path, password=passphrase
- )
+ if cert_path:
+ password = (
+ passphrase_val.decode("utf-8")
+ if isinstance(passphrase_val, bytes)
+ else passphrase_val
+ )
+ context.load_cert_chain(
+ certfile=cert_path,
+ keyfile=key_path,
+ password=password,
+ )
  return context
-except (ssl.SSLError, OSError, IOError, ValueError, RuntimeError) as exc:
-raise exceptions.TransportError(
-"Failed to load client certificate and key for mTLS."
-) from exc
+ except (ssl.SSLError, OSError, IOError, ValueError, RuntimeError, TypeError) as exc:
+ raise exceptions.TransportError(
+ "Failed to load client certificate and key for mTLS."
+ ) from exc
 
 
 async def _run_in_executor(func, *args):
@@ -187,9 +173,9 @@ async def get_client_cert_and_key(client_cert_callback=None):
  """
  if client_cert_callback:
  result = client_cert_callback()
-try:
+if inspect.isawaitable(result):
  cert, key = await result
-except TypeError:
+else:
  cert, key = result
  return True, cert, key
 

packages/google-auth/google/auth/compute_engine/_mtls.py

@@ -120,8 +120,9 @@ def __init__(
  self.ssl_context = ssl.create_default_context()
  self.ssl_context.load_verify_locations(cafile=mds_mtls_config.ca_cert_path)
  self.ssl_context.load_cert_chain(
- certfile=mds_mtls_config.client_combined_cert_path
+ certfile=mds_mtls_config.client_combined_cert_path, password=""
  )
+ self._fallback_adapter = HTTPAdapter()
  super(MdsMtlsAdapter, self).__init__(*args, **kwargs)
 
  def init_poolmanager(self, *args, **kwargs):
@@ -146,6 +147,8 @@ def send(self, request, **kwargs):
  ssl.SSLError,
  requests.exceptions.SSLError,
  requests.exceptions.HTTPError,
+ requests.exceptions.ConnectionError,
+ requests.exceptions.Timeout,
  ) as e:
  _LOGGER.warning(
  "mTLS connection to Compute Engine Metadata server failed. "
@@ -157,6 +160,9 @@ def send(self, request, **kwargs):
  http_fallback_url = urlunparse(parsed_original_url._replace(scheme="http"))
  request.url = http_fallback_url
 
- # Use a standard HTTPAdapter for the fallback
- http_adapter = HTTPAdapter()
- return http_adapter.send(request, **kwargs)
+ # Use the cached standard HTTPAdapter for the fallback
+ return self._fallback_adapter.send(request, **kwargs)
+
+ def close(self):
+ self._fallback_adapter.close()
+ super(MdsMtlsAdapter, self).close()

packages/google-auth/google/auth/identity_pool.py

@@ -152,28 +152,34 @@ def __init__(self, trust_chain_path, leaf_cert_callback):
 
  @_helpers.copy_docstring(SubjectTokenSupplier)
  def get_subject_token(self, context, request):
- # Import OpennSSL inline because it is an extra import only required by customers
- # using mTLS.
- from OpenSSL import crypto
+ from cryptography import x509
 
- leaf_cert = crypto.load_certificate(
- crypto.FILETYPE_PEM, self._leaf_cert_callback()
- )
+ try:
+ leaf_cert_data = self._leaf_cert_callback()
+ except Exception as e:
+ raise exceptions.RefreshError("Failed to retrieve leaf certificate.") from e
+
+ try:
+ if isinstance(leaf_cert_data, str):
+ leaf_cert_data = leaf_cert_data.encode("utf-8")
+ leaf_cert = x509.load_pem_x509_certificate(leaf_cert_data)
+ except Exception as e:
+ raise exceptions.RefreshError("Failed to parse leaf certificate.") from e
  trust_chain = self._read_trust_chain()
  cert_chain = []
 
- cert_chain.append(_X509Supplier._encode_cert(leaf_cert))
+ cert_chain.append(_encode_cert(leaf_cert))
 
  if trust_chain is None or len(trust_chain) == 0:
  return json.dumps(cert_chain)
 
  # Append the first cert if it is not the leaf cert.
- first_cert = _X509Supplier._encode_cert(trust_chain[0])
+ first_cert = _encode_cert(trust_chain[0])
  if first_cert != cert_chain[0]:
  cert_chain.append(first_cert)
 
  for i in range(1, len(trust_chain)):
- encoded = _X509Supplier._encode_cert(trust_chain[i])
+ encoded = _encode_cert(trust_chain[i])
  # Check if the current cert is the leaf cert and raise an exception if it is.
  if encoded == cert_chain[0]:
  raise exceptions.RefreshError(
@@ -184,9 +190,7 @@ def get_subject_token(self, context, request):
  return json.dumps(cert_chain)
 
  def _read_trust_chain(self):
- # Import OpennSSL inline because it is an extra import only required by customers
- # using mTLS.
- from OpenSSL import crypto
+ from cryptography import x509
 
  certificate_trust_chain = []
  # If no trust chain path was provided, return an empty list.
@@ -204,9 +208,7 @@ def _read_trust_chain(self):
  cert_data = b"-----BEGIN CERTIFICATE-----" + cert_block
  try:
  # Load each certificate and add it to the trust chain.
- cert = crypto.load_certificate(
- crypto.FILETYPE_PEM, cert_data
- )
+ cert = x509.load_pem_x509_certificate(cert_data)
  certificate_trust_chain.append(cert)
  except Exception as e:
  raise exceptions.RefreshError(
@@ -215,19 +217,22 @@ def _read_trust_chain(self):
  )
  ) from e
  return certificate_trust_chain
- except FileNotFoundError:
+ except FileNotFoundError as e:
  raise exceptions.RefreshError(
  "Trust chain file '{}' was not found.".format(self._trust_chain_path)
- )
+ ) from e
+ except OSError as e:
+ raise exceptions.RefreshError(
+ "Error accessing trust chain file '{}'.".format(self._trust_chain_path)
+ ) from e
+
 
- def _encode_cert(cert):
- # Import OpennSSL inline because it is an extra import only required by customers
- # using mTLS.
- from OpenSSL import crypto
+def _encode_cert(cert):
+ from cryptography.hazmat.primitives import serialization
 
-return base64.b64encode(
- crypto.dump_certificate(crypto.FILETYPE_ASN1, cert)
- ).decode("utf-8")
+ return base64.b64encode(cert.public_bytes(serialization.Encoding.DER)).decode(
+"utf-8"
+ )
 
 
 def _parse_token_data(token_content, format_type="text", subject_token_field_name=None):

packages/google-auth/google/auth/transport/_custom_tls_signer.py

@@ -21,9 +21,9 @@
 import json
 import logging
 import os
+import ssl
 import sys
-
-import cffi # type: ignore
+import sysconfig
 
 from google.auth import exceptions
 
@@ -45,16 +45,23 @@
 )
 
 
-# Cast SSL_CTX* to void*
-def _cast_ssl_ctx_to_void_p_pyopenssl(ssl_ctx):
- return ctypes.cast(int(cffi.FFI().cast("intptr_t", ssl_ctx)), ctypes.c_void_p)
-
-
 # Cast SSL_CTX* to void*
 def _cast_ssl_ctx_to_void_p_stdlib(context):
- return ctypes.c_void_p.from_address(
- id(context) + ctypes.sizeof(ctypes.c_void_p) * 2
- )
+ if not issubclass(type(context), ssl.SSLContext):
+ raise TypeError("context must be an instance of ssl.SSLContext, not a mock")
+
+ if (
+ sys.implementation.name != "cpython"
+ or hasattr(sys, "getobjects")
+ or sysconfig.get_config_var("Py_DEBUG")
+ or sysconfig.get_config_var("Py_GIL_DISABLED") == 1
+ ):
+ raise exceptions.MutualTLSChannelError(
+ "Custom TLS signing is only supported on standard release CPython runtimes."
+ )
+
+ offset = sys.getsizeof(object())
+ return ctypes.c_void_p.from_address(id(context) + offset)
 
 
 # Load offload library and set up the function types.
@@ -274,7 +281,7 @@ def attach_to_ssl_context(self, ctx):
  if not self._offload_lib.ConfigureSslContext(
  self._sign_callback,
  ctypes.c_char_p(self._cert),
-_cast_ssl_ctx_to_void_p_pyopenssl(ctx._ctx._context),
+_cast_ssl_ctx_to_void_p_stdlib(ctx),
  ):
  raise exceptions.MutualTLSChannelError(
  "failed to configure ECP Offload SSL context"