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

Replace pyOpenSSL with standard library ssl for mTLS transport and update key decryption to use cryptography library.

This change also enhances security for handling private keys by:
- Using Linux memfd_create for RAM-backed in-memory files to avoid
writing secrets to physical storage.
- Encrypting plaintext keys on-the-fly before writing to fallback
temporary files on disk.
- Securely wiping temporary files with null bytes before deletion.

Fixes #16920
Closes #16921
fixes #17412

Author: nbayati

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"