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

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

@@ -17,42 +17,17 @@
 """
 
 import asyncio
-import contextlib
 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 +46,24 @@ 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:
+                context.load_cert_chain(
+                    certfile=cert_path,
+                    keyfile=key_path,
+                    password=passphrase_val or "",
+                )
             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) as exc:
+        raise exceptions.TransportError(
+            "Failed to load client certificate and key for mTLS."
+        ) from exc
 
 
 async def _run_in_executor(func, *args):

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

@@ -120,7 +120,7 @@ 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=""
         )
         super(MdsMtlsAdapter, self).__init__(*args, **kwargs)
 
@@ -146,6 +146,7 @@ def send(self, request, **kwargs):
             ssl.SSLError,
             requests.exceptions.SSLError,
             requests.exceptions.HTTPError,
+            requests.exceptions.ConnectionError,
         ) as e:
             _LOGGER.warning(
                 "mTLS connection to Compute Engine Metadata server failed. "

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

@@ -23,8 +23,6 @@
 import os
 import sys
 
-import cffi  # type: ignore
-
 from google.auth import exceptions
 
 _LOGGER = logging.getLogger(__name__)
@@ -45,13 +43,12 @@
 )
 
 
-# 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):
+    if sys.implementation.name != "cpython" or hasattr(sys, "getobjects"):
+        raise exceptions.MutualTLSChannelError(
+            "Custom TLS signing is only supported on standard release CPython runtimes."
+        )
     return ctypes.c_void_p.from_address(
         id(context) + ctypes.sizeof(ctypes.c_void_p) * 2
     )
@@ -274,7 +271,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"