From 782211c9a749e0995de497ea09004697e66f22cb Mon Sep 17 00:00:00 2001
From: Negar Bayati <negarb@google.com>
Date: Wed, 17 Jun 2026 07:01:56 +0000
Subject: [PATCH 01/10] feat(auth): add mTLS helper methods and endpoint
 configuration for handwritten SDK mTLS support

- Introduced `GOOGLE_API_USE_MTLS_ENDPOINT` environment variable to control whether an mTLS endpoint should be used (`always`, `never`, or `auto`).
- Added several new helper functions in `google.auth.transport.mtls` to facilitate SSL context creation and client certificate loading:
  - `load_client_cert_into_context`: Loads a client certificate and key into a provided SSL context.
  - `make_client_cert_ssl_context`: Creates a default SSL context loaded with a specific client certificate and key.
  - `load_default_client_cert`: Discovers and loads the default client certificate into a provided SSL context if mTLS is enabled.
  - `get_default_ssl_context`: Returns a default SSL context pre-loaded with the default client certificate, or `None` if unavailable.
  - `should_use_mtls_endpoint`: Determines if an mTLS endpoint should be used based on the new environment variable and certificate availability.
- Fixed outdated docstrings for `default_client_cert_source` and `default_client_encrypted_cert_source` to correctly state they raise `MutualTLSChannelError` instead of `DefaultClientCertSourceError`.
- Updated `default_client_cert_source` to also catch `ClientCertError` when loading credentials.
- Added comprehensive unit tests for the new mTLS helper methods.
---
 .../google/auth/environment_vars.py           |   3 +
 .../google-auth/google/auth/transport/mtls.py | 178 +++++++++++-
 .../google-auth/tests/transport/test_mtls.py  | 263 ++++++++++++++++++
 3 files changed, 441 insertions(+), 3 deletions(-)

diff --git a/packages/google-auth/google/auth/environment_vars.py b/packages/google-auth/google/auth/environment_vars.py
index b7ff66c8b54a..7d82d288a24c 100644
--- a/packages/google-auth/google/auth/environment_vars.py
+++ b/packages/google-auth/google/auth/environment_vars.py
@@ -133,3 +133,6 @@
     "GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES"
 )
 """Environment variable to prevent agent token sharing for GCP services."""
+
+GOOGLE_API_USE_MTLS_ENDPOINT = "GOOGLE_API_USE_MTLS_ENDPOINT"
+"""Environment variable controlling whether to use mTLS endpoint or not."""
diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py
index 666a6ca1fd91..c3c1cf186c2b 100644
--- a/packages/google-auth/google/auth/transport/mtls.py
+++ b/packages/google-auth/google/auth/transport/mtls.py
@@ -14,12 +14,19 @@
 
 """Utilites for mutual TLS."""
 
+import logging
 from os import getenv
+import ssl
+from typing import Optional
 
+from google.auth import environment_vars
 from google.auth import exceptions
 from google.auth.transport import _mtls_helper
 
 
+_LOGGER = logging.getLogger(__name__)
+
+
 def has_default_client_cert_source(include_context_aware=True):
     """Check if default client SSL credentials exists on the device.
 
@@ -60,7 +67,7 @@ def default_client_cert_source():
             client certificate bytes and private key bytes, both in PEM format.
 
     Raises:
-        google.auth.exceptions.DefaultClientCertSourceError: If the default
+        google.auth.exceptions.MutualTLSChannelError: If the default
             client SSL credentials don't exist or are malformed.
     """
     if not has_default_client_cert_source(include_context_aware=True):
@@ -71,7 +78,12 @@ def default_client_cert_source():
     def callback():
         try:
             _, cert_bytes, key_bytes = _mtls_helper.get_client_cert_and_key()
-        except (OSError, RuntimeError, ValueError) as caught_exc:
+        except (
+            exceptions.ClientCertError,
+            OSError,
+            RuntimeError,
+            ValueError,
+        ) as caught_exc:
             new_exc = exceptions.MutualTLSChannelError(caught_exc)
             raise new_exc from caught_exc
 
@@ -96,7 +108,7 @@ def default_client_encrypted_cert_source(cert_path, key_path):
             returns the cert_path, key_path and passphrase bytes.
 
     Raises:
-        google.auth.exceptions.DefaultClientCertSourceError: If any problem
+        google.auth.exceptions.MutualTLSChannelError: If any problem
             occurs when loading or saving the client certificate and key.
     """
     if not has_default_client_cert_source(include_context_aware=True):
@@ -140,3 +152,163 @@ def should_use_client_cert():
        bool: indicating whether the client certificate should be used for mTLS.
     """
     return _mtls_helper.check_use_client_cert()
+
+
+def load_client_cert_into_context(
+    ctx: ssl.SSLContext,
+    cert_bytes: bytes,
+    key_bytes: bytes,
+    passphrase: Optional[bytes] = None,
+) -> None:
+    """Load a client certificate and key into an SSL context.
+
+    Args:
+        ctx (ssl.SSLContext): The SSL context to load the certificate and key into.
+        cert_bytes (bytes): The client certificate bytes in PEM format.
+        key_bytes (bytes): The client private key bytes in PEM format.
+        passphrase (Optional[bytes]): The passphrase for the client private key.
+
+    Raises:
+        google.auth.exceptions.MutualTLSChannelError: If the SSL context is invalid,
+            or if loading the certificate and key fails.
+    """
+    if ctx is None or not hasattr(ctx, "load_cert_chain"):
+        raise exceptions.MutualTLSChannelError(
+            "Failed to load client certificate and key for mTLS. The provided context "
+            "object is invalid or does not support loading certificate chains."
+        )
+
+    try:
+        with _mtls_helper.secure_cert_key_paths(
+            cert_bytes, key_bytes, passphrase=passphrase
+        ) as (
+            cert_path,
+            key_path,
+            passphrase_val,
+        ):
+            ctx.load_cert_chain(
+                certfile=cert_path, keyfile=key_path, password=passphrase_val
+            )
+    except (
+        ssl.SSLError,
+        OSError,
+        ValueError,
+        RuntimeError,
+    ) as caught_exc:
+        new_exc = exceptions.MutualTLSChannelError(
+            "Failed to load client certificate and key for mTLS."
+        )
+        raise new_exc from caught_exc
+
+
+def make_client_cert_ssl_context(
+    cert_bytes: bytes,
+    key_bytes: bytes,
+    passphrase: Optional[bytes] = None,
+) -> ssl.SSLContext:
+    """Create a default SSL context loaded with the client certificate and key.
+
+    Args:
+        cert_bytes (bytes): The client certificate bytes in PEM format.
+        key_bytes (bytes): The client private key bytes in PEM format.
+        passphrase (Optional[bytes]): The passphrase for the client private key.
+
+    Returns:
+        ssl.SSLContext: The SSL context loaded with the client certificate and key.
+
+    Raises:
+        google.auth.exceptions.MutualTLSChannelError: If loading the certificate and key fails.
+    """
+    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
+    load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase=passphrase)
+    return ctx
+
+
+def load_default_client_cert(ctx: ssl.SSLContext) -> bool:
+    """Load the default client certificate and key into an SSL context if configured.
+
+    If client certificates are enabled and a default client certificate source is
+    found, the certificate and key are loaded into the SSL context.
+
+    Args:
+        ctx (ssl.SSLContext): The SSL context to load the default client certificate
+            and key into.
+
+    Returns:
+        bool: True if client certificates are enabled and the default client
+            certificate was successfully loaded. False if client certificates
+            are disabled or if no default certificate source is configured.
+
+    Raises:
+        google.auth.exceptions.ClientCertError: If the default client certificate
+            source exists but cannot be loaded or parsed.
+        google.auth.exceptions.MutualTLSChannelError: If the default client certificate
+            or key is malformed.
+    """
+    if not should_use_client_cert() or not has_default_client_cert_source():
+        return False
+    (
+        has_cert,
+        cert_bytes,
+        key_bytes,
+        passphrase,
+    ) = _mtls_helper.get_client_ssl_credentials()
+    if not has_cert:
+        return False
+    load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase)
+    return True
+
+
+def get_default_ssl_context() -> Optional[ssl.SSLContext]:
+    """Get a default SSL context loaded with the default client certificate.
+
+    Returns:
+        ssl.SSLContext: An SSL context loaded with the default client
+            certificate, or None if client certificates are not configured
+            or available.
+
+    Raises:
+        google.auth.exceptions.ClientCertError: If the default client certificate
+            source exists but cannot be loaded or parsed.
+        google.auth.exceptions.MutualTLSChannelError: If the default client certificate
+            or key is malformed.
+    """
+    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
+    return ctx if load_default_client_cert(ctx) else None
+
+
+def should_use_mtls_endpoint(
+    client_cert_available: Optional[bool] = None,
+) -> bool:
+    """Determine whether to use an mTLS endpoint.
+
+    This relies on the GOOGLE_API_USE_MTLS_ENDPOINT environment variable. If set to
+    "always", returns True. If set to "never", returns False. If set to "auto"
+    or unset, returns whether a client certificate is available.
+
+    Args:
+        client_cert_available (bool): indicating if a client certificate
+            is available. If None, this is determined by checking if client
+            certificates are enabled and a default source is present.
+
+    Returns:
+        bool: indicating if an mTLS endpoint should be used.
+    """
+    if client_cert_available is None:
+        client_cert_available = should_use_client_cert()
+
+    use_mtls_endpoint = getenv(environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT, "auto")
+    use_mtls_endpoint = use_mtls_endpoint.lower()
+    if use_mtls_endpoint == "always":
+        return True
+    if use_mtls_endpoint == "never":
+        return False
+    if use_mtls_endpoint == "auto":
+        return client_cert_available
+
+    _LOGGER.warning(
+        "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value %r. Accepted "
+        "values: never, auto, always. Defaulting to auto.",
+        use_mtls_endpoint,
+    )
+    return client_cert_available
diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py
index 405cb496cad2..69cbe792575d 100644
--- a/packages/google-auth/tests/transport/test_mtls.py
+++ b/packages/google-auth/tests/transport/test_mtls.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import contextlib
+import ssl
 from unittest import mock
 
 import pytest  # type: ignore
@@ -135,6 +137,12 @@ def test_default_client_cert_source(
     with pytest.raises(exceptions.MutualTLSChannelError):
         callback()
 
+    # Test bad callback which throws ClientCertError.
+    get_client_cert_and_key.side_effect = exceptions.ClientCertError()
+    callback = mtls.default_client_cert_source()
+    with pytest.raises(exceptions.MutualTLSChannelError):
+        callback()
+
 
 @mock.patch(
     "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True
@@ -173,3 +181,258 @@ def test_should_use_client_cert(check_use_client_cert):
 
     check_use_client_cert.return_value = False
     assert not mtls.should_use_client_cert()
+
+
+@contextlib.contextmanager
+def _fake_secure_paths(cert_bytes, key_bytes, passphrase=None):
+    yield "cert_path", "key_path", passphrase
+
+
+@mock.patch(
+    "google.auth.transport._mtls_helper.secure_cert_key_paths",
+    side_effect=_fake_secure_paths,
+)
+def test_load_client_cert_into_context_success(mock_secure_paths):
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    result = mtls.load_client_cert_into_context(
+        mock_ctx, b"cert", b"key", passphrase=b"passphrase"
+    )
+    assert result is None
+    mock_ctx.load_cert_chain.assert_called_once_with(
+        certfile="cert_path", keyfile="key_path", password=b"passphrase"
+    )
+
+
+@mock.patch(
+    "google.auth.transport._mtls_helper.secure_cert_key_paths",
+    side_effect=_fake_secure_paths,
+)
+def test_load_client_cert_into_context_error(mock_secure_paths):
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    mock_ctx.load_cert_chain.side_effect = ssl.SSLError("boom")
+    with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
+        mtls.load_client_cert_into_context(mock_ctx, b"cert", b"key")
+    assert "Failed to load client certificate and key" in str(exc_info.value)
+    assert isinstance(exc_info.value.__cause__, ssl.SSLError)
+
+
+def test_load_client_cert_into_context_invalid_ctx():
+    with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
+        mtls.load_client_cert_into_context(None, b"cert", b"key")
+    assert (
+        "The provided context object is invalid or does not support loading certificate chains"
+        in str(exc_info.value)
+    )
+    assert exc_info.value.__cause__ is None
+
+
+@mock.patch(
+    "google.auth.transport._mtls_helper.secure_cert_key_paths",
+    side_effect=_fake_secure_paths,
+)
+def test_load_client_cert_into_context_load_chain_type_error(mock_secure_paths):
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    mock_ctx.load_cert_chain.side_effect = TypeError("invalid password type")
+    with pytest.raises(TypeError) as exc_info:
+        mtls.load_client_cert_into_context(mock_ctx, b"cert", b"key")
+    assert "invalid password type" in str(exc_info.value)
+
+
+@mock.patch("google.auth.transport.mtls.load_client_cert_into_context", autospec=True)
+@mock.patch("ssl.create_default_context", autospec=True)
+def test_make_client_cert_ssl_context(mock_create_context, mock_load_cert):
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    mock_create_context.return_value = mock_ctx
+
+    result = mtls.make_client_cert_ssl_context(b"cert", b"key", b"passphrase")
+
+    assert result == mock_ctx
+    mock_create_context.assert_called_once_with(ssl.Purpose.SERVER_AUTH)
+    mock_load_cert.assert_called_once_with(
+        mock_ctx, b"cert", b"key", passphrase=b"passphrase"
+    )
+
+
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
+def test_load_default_client_cert_disabled(mock_should_use):
+    mock_should_use.return_value = False
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    assert mtls.load_default_client_cert(mock_ctx) is False
+    mock_ctx.load_cert_chain.assert_not_called()
+
+
+@mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
+def test_load_default_client_cert_no_source(mock_should_use, mock_has_source):
+    mock_should_use.return_value = True
+    mock_has_source.return_value = False
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    assert mtls.load_default_client_cert(mock_ctx) is False
+    mock_ctx.load_cert_chain.assert_not_called()
+
+
+@mock.patch(
+    "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True
+)
+@mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
+def test_load_default_client_cert_no_cert(
+    mock_should_use, mock_has_source, mock_get_credentials
+):
+    mock_should_use.return_value = True
+    mock_has_source.return_value = True
+    mock_get_credentials.return_value = (False, None, None, None)
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    assert mtls.load_default_client_cert(mock_ctx) is False
+    mock_ctx.load_cert_chain.assert_not_called()
+
+
+@mock.patch("google.auth.transport.mtls.load_client_cert_into_context", autospec=True)
+@mock.patch(
+    "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True
+)
+@mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
+def test_load_default_client_cert_success(
+    mock_should_use, mock_has_source, mock_get_credentials, mock_load_cert
+):
+    mock_should_use.return_value = True
+    mock_has_source.return_value = True
+    mock_get_credentials.return_value = (True, b"cert", b"key", b"passphrase")
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+
+    assert mtls.load_default_client_cert(mock_ctx) is True
+    mock_load_cert.assert_called_once_with(mock_ctx, b"cert", b"key", b"passphrase")
+
+
+@mock.patch(
+    "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True
+)
+@mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
+def test_load_default_client_cert_propagates_client_cert_error(
+    mock_should_use, mock_has_source, mock_get_credentials
+):
+    mock_should_use.return_value = True
+    mock_has_source.return_value = True
+    mock_get_credentials.side_effect = exceptions.ClientCertError("credentials failure")
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+
+    with pytest.raises(exceptions.ClientCertError) as exc_info:
+        mtls.load_default_client_cert(mock_ctx)
+    assert "credentials failure" in str(exc_info.value)
+
+
+@mock.patch("google.auth.transport.mtls.load_default_client_cert", autospec=True)
+@mock.patch("ssl.create_default_context", autospec=True)
+def test_get_default_ssl_context_configured(mock_create_context, mock_load_default):
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    mock_create_context.return_value = mock_ctx
+    mock_load_default.return_value = True
+
+    result = mtls.get_default_ssl_context()
+
+    assert result == mock_ctx
+    mock_create_context.assert_called_once_with(ssl.Purpose.SERVER_AUTH)
+    mock_load_default.assert_called_once_with(mock_ctx)
+
+
+@mock.patch("google.auth.transport.mtls.load_default_client_cert", autospec=True)
+@mock.patch("ssl.create_default_context", autospec=True)
+def test_get_default_ssl_context_unconfigured(mock_create_context, mock_load_default):
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    mock_create_context.return_value = mock_ctx
+    mock_load_default.return_value = False
+
+    result = mtls.get_default_ssl_context()
+
+    assert result is None
+    mock_create_context.assert_called_once_with(ssl.Purpose.SERVER_AUTH)
+    mock_load_default.assert_called_once_with(mock_ctx)
+
+
+@pytest.mark.parametrize(
+    "env_val,client_cert_available,expected",
+    [
+        ("always", True, True),
+        ("always", False, True),
+        ("never", True, False),
+        ("never", False, False),
+        ("auto", True, True),
+        ("auto", False, False),
+        (None, True, True),  # Defaults to auto
+        (None, False, False),  # Defaults to auto
+        ("ALWAYS", True, True),
+        ("ALWAYS", False, True),
+        ("NEVER", True, False),
+        ("NEVER", False, False),
+        ("AUTO", True, True),
+        ("AUTO", False, False),
+        ("invalid_val", True, True),
+        ("invalid_val", False, False),
+    ],
+)
+@mock.patch(
+    "google.auth.environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT",
+    "GOOGLE_API_USE_MTLS_ENDPOINT",
+)
+@mock.patch("google.auth.transport.mtls.getenv", autospec=True)
+def test_should_use_mtls_endpoint(
+    mock_getenv, env_val, client_cert_available, expected
+):
+    mock_getenv.side_effect = (
+        lambda var, default=None: env_val
+        if (var == "GOOGLE_API_USE_MTLS_ENDPOINT" and env_val is not None)
+        else default
+    )
+    result = mtls.should_use_mtls_endpoint(client_cert_available)
+    assert result == expected
+
+
+@mock.patch(
+    "google.auth.environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT",
+    "GOOGLE_API_USE_MTLS_ENDPOINT",
+)
+@mock.patch("google.auth.transport.mtls.getenv", autospec=True)
+def test_should_use_mtls_endpoint_invalid_value(mock_getenv, caplog):
+    mock_getenv.side_effect = (
+        lambda var, default=None: "invalid_value"
+        if var == "GOOGLE_API_USE_MTLS_ENDPOINT"
+        else default
+    )
+    with caplog.at_level("WARNING"):
+        assert mtls.should_use_mtls_endpoint(True) is True
+        assert "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value" in caplog.text
+        assert "Defaulting to auto" in caplog.text
+
+    caplog.clear()
+
+    with caplog.at_level("WARNING"):
+        assert mtls.should_use_mtls_endpoint(False) is False
+        assert "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value" in caplog.text
+        assert "Defaulting to auto" in caplog.text
+
+
+@mock.patch(
+    "google.auth.environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT",
+    "GOOGLE_API_USE_MTLS_ENDPOINT",
+)
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
+@mock.patch("google.auth.transport.mtls.getenv", autospec=True)
+def test_should_use_mtls_endpoint_default_client_cert(
+    mock_getenv, mock_should_use_client_cert
+):
+    mock_getenv.side_effect = (
+        lambda var, default=None: "auto"
+        if var == "GOOGLE_API_USE_MTLS_ENDPOINT"
+        else default
+    )
+    mock_should_use_client_cert.return_value = True
+    assert mtls.should_use_mtls_endpoint() is True
+    mock_should_use_client_cert.assert_called_once()
+
+    mock_should_use_client_cert.reset_mock()
+
+    mock_should_use_client_cert.return_value = False
+    assert mtls.should_use_mtls_endpoint() is False
+    mock_should_use_client_cert.assert_called_once()

From 28bc123ae3e7750783209ed523131140c6ec0c83 Mon Sep 17 00:00:00 2001
From: nbayati <99771966+nbayati@users.noreply.github.com>
Date: Fri, 3 Jul 2026 13:49:17 -0700
Subject: [PATCH 02/10] refactor: make mtls cert loaders private and enforce
 strict ssl.SSLContext typing

---
 .../google-auth/google/auth/transport/mtls.py   | 10 +++++-----
 .../google-auth/tests/transport/test_mtls.py    | 17 +++++++++--------
 2 files changed, 14 insertions(+), 13 deletions(-)

diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py
index c3c1cf186c2b..4f77be6e3e98 100644
--- a/packages/google-auth/google/auth/transport/mtls.py
+++ b/packages/google-auth/google/auth/transport/mtls.py
@@ -154,7 +154,7 @@ def should_use_client_cert():
     return _mtls_helper.check_use_client_cert()
 
 
-def load_client_cert_into_context(
+def _load_client_cert_into_context(
     ctx: ssl.SSLContext,
     cert_bytes: bytes,
     key_bytes: bytes,
@@ -172,7 +172,7 @@ def load_client_cert_into_context(
         google.auth.exceptions.MutualTLSChannelError: If the SSL context is invalid,
             or if loading the certificate and key fails.
     """
-    if ctx is None or not hasattr(ctx, "load_cert_chain"):
+    if not isinstance(ctx, ssl.SSLContext):
         raise exceptions.MutualTLSChannelError(
             "Failed to load client certificate and key for mTLS. The provided context "
             "object is invalid or does not support loading certificate chains."
@@ -201,7 +201,7 @@ def load_client_cert_into_context(
         raise new_exc from caught_exc
 
 
-def make_client_cert_ssl_context(
+def _make_client_cert_ssl_context(
     cert_bytes: bytes,
     key_bytes: bytes,
     passphrase: Optional[bytes] = None,
@@ -220,7 +220,7 @@ def make_client_cert_ssl_context(
         google.auth.exceptions.MutualTLSChannelError: If loading the certificate and key fails.
     """
     ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
-    load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase=passphrase)
+    _load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase=passphrase)
     return ctx
 
 
@@ -255,7 +255,7 @@ def load_default_client_cert(ctx: ssl.SSLContext) -> bool:
     ) = _mtls_helper.get_client_ssl_credentials()
     if not has_cert:
         return False
-    load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase)
+    _load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase)
     return True
 
 
diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py
index 69cbe792575d..53be7d785959 100644
--- a/packages/google-auth/tests/transport/test_mtls.py
+++ b/packages/google-auth/tests/transport/test_mtls.py
@@ -194,7 +194,7 @@ def _fake_secure_paths(cert_bytes, key_bytes, passphrase=None):
 )
 def test_load_client_cert_into_context_success(mock_secure_paths):
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
-    result = mtls.load_client_cert_into_context(
+    result = mtls._load_client_cert_into_context(
         mock_ctx, b"cert", b"key", passphrase=b"passphrase"
     )
     assert result is None
@@ -211,14 +211,15 @@ def test_load_client_cert_into_context_error(mock_secure_paths):
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
     mock_ctx.load_cert_chain.side_effect = ssl.SSLError("boom")
     with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
-        mtls.load_client_cert_into_context(mock_ctx, b"cert", b"key")
+        mtls._load_client_cert_into_context(mock_ctx, b"cert", b"key")
     assert "Failed to load client certificate and key" in str(exc_info.value)
     assert isinstance(exc_info.value.__cause__, ssl.SSLError)
 
 
-def test_load_client_cert_into_context_invalid_ctx():
+@pytest.mark.parametrize("invalid_ctx", [None, object()])
+def test_load_client_cert_into_context_invalid_ctx(invalid_ctx):
     with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
-        mtls.load_client_cert_into_context(None, b"cert", b"key")
+        mtls._load_client_cert_into_context(invalid_ctx, b"cert", b"key")
     assert (
         "The provided context object is invalid or does not support loading certificate chains"
         in str(exc_info.value)
@@ -234,17 +235,17 @@ def test_load_client_cert_into_context_load_chain_type_error(mock_secure_paths):
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
     mock_ctx.load_cert_chain.side_effect = TypeError("invalid password type")
     with pytest.raises(TypeError) as exc_info:
-        mtls.load_client_cert_into_context(mock_ctx, b"cert", b"key")
+        mtls._load_client_cert_into_context(mock_ctx, b"cert", b"key")
     assert "invalid password type" in str(exc_info.value)
 
 
-@mock.patch("google.auth.transport.mtls.load_client_cert_into_context", autospec=True)
+@mock.patch("google.auth.transport.mtls._load_client_cert_into_context", autospec=True)
 @mock.patch("ssl.create_default_context", autospec=True)
 def test_make_client_cert_ssl_context(mock_create_context, mock_load_cert):
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
     mock_create_context.return_value = mock_ctx
 
-    result = mtls.make_client_cert_ssl_context(b"cert", b"key", b"passphrase")
+    result = mtls._make_client_cert_ssl_context(b"cert", b"key", b"passphrase")
 
     assert result == mock_ctx
     mock_create_context.assert_called_once_with(ssl.Purpose.SERVER_AUTH)
@@ -287,7 +288,7 @@ def test_load_default_client_cert_no_cert(
     mock_ctx.load_cert_chain.assert_not_called()
 
 
-@mock.patch("google.auth.transport.mtls.load_client_cert_into_context", autospec=True)
+@mock.patch("google.auth.transport.mtls._load_client_cert_into_context", autospec=True)
 @mock.patch(
     "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True
 )

From 172dd399747f1bc43fabd2875e35bb06fcc5c527 Mon Sep 17 00:00:00 2001
From: nbayati <99771966+nbayati@users.noreply.github.com>
Date: Fri, 3 Jul 2026 13:52:40 -0700
Subject: [PATCH 03/10] fix: improve mtls exception handling and edge cases

---
 .../google-auth/google/auth/transport/mtls.py | 33 +++++++++++++------
 .../google-auth/tests/transport/test_mtls.py  | 10 +++---
 2 files changed, 29 insertions(+), 14 deletions(-)

diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py
index 4f77be6e3e98..bff0b8c50729 100644
--- a/packages/google-auth/google/auth/transport/mtls.py
+++ b/packages/google-auth/google/auth/transport/mtls.py
@@ -186,18 +186,22 @@ def _load_client_cert_into_context(
             key_path,
             passphrase_val,
         ):
+            password_val = (
+                passphrase_val.decode("utf-8")
+                if isinstance(passphrase_val, bytes)
+                else passphrase_val
+            )
             ctx.load_cert_chain(
-                certfile=cert_path, keyfile=key_path, password=passphrase_val
+                certfile=cert_path, keyfile=key_path, password=password_val
             )
     except (
         ssl.SSLError,
         OSError,
         ValueError,
         RuntimeError,
+        TypeError,
     ) as caught_exc:
-        new_exc = exceptions.MutualTLSChannelError(
-            "Failed to load client certificate and key for mTLS."
-        )
+        new_exc = exceptions.MutualTLSChannelError(caught_exc)
         raise new_exc from caught_exc
 
 
@@ -247,12 +251,21 @@ def load_default_client_cert(ctx: ssl.SSLContext) -> bool:
     """
     if not should_use_client_cert() or not has_default_client_cert_source():
         return False
-    (
-        has_cert,
-        cert_bytes,
-        key_bytes,
-        passphrase,
-    ) = _mtls_helper.get_client_ssl_credentials()
+    try:
+        (
+            has_cert,
+            cert_bytes,
+            key_bytes,
+            passphrase,
+        ) = _mtls_helper.get_client_ssl_credentials()
+    except (
+        exceptions.ClientCertError,
+        OSError,
+        RuntimeError,
+        ValueError,
+    ) as caught_exc:
+        new_exc = exceptions.MutualTLSChannelError(caught_exc)
+        raise new_exc from caught_exc
     if not has_cert:
         return False
     _load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase)
diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py
index 53be7d785959..25fa094ca154 100644
--- a/packages/google-auth/tests/transport/test_mtls.py
+++ b/packages/google-auth/tests/transport/test_mtls.py
@@ -199,7 +199,7 @@ def test_load_client_cert_into_context_success(mock_secure_paths):
     )
     assert result is None
     mock_ctx.load_cert_chain.assert_called_once_with(
-        certfile="cert_path", keyfile="key_path", password=b"passphrase"
+        certfile="cert_path", keyfile="key_path", password="passphrase"
     )
 
 
@@ -212,7 +212,7 @@ def test_load_client_cert_into_context_error(mock_secure_paths):
     mock_ctx.load_cert_chain.side_effect = ssl.SSLError("boom")
     with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
         mtls._load_client_cert_into_context(mock_ctx, b"cert", b"key")
-    assert "Failed to load client certificate and key" in str(exc_info.value)
+    assert "boom" in str(exc_info.value)
     assert isinstance(exc_info.value.__cause__, ssl.SSLError)
 
 
@@ -234,9 +234,10 @@ def test_load_client_cert_into_context_invalid_ctx(invalid_ctx):
 def test_load_client_cert_into_context_load_chain_type_error(mock_secure_paths):
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
     mock_ctx.load_cert_chain.side_effect = TypeError("invalid password type")
-    with pytest.raises(TypeError) as exc_info:
+    with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
         mtls._load_client_cert_into_context(mock_ctx, b"cert", b"key")
     assert "invalid password type" in str(exc_info.value)
+    assert isinstance(exc_info.value.__cause__, TypeError)
 
 
 @mock.patch("google.auth.transport.mtls._load_client_cert_into_context", autospec=True)
@@ -319,9 +320,10 @@ def test_load_default_client_cert_propagates_client_cert_error(
     mock_get_credentials.side_effect = exceptions.ClientCertError("credentials failure")
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
 
-    with pytest.raises(exceptions.ClientCertError) as exc_info:
+    with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
         mtls.load_default_client_cert(mock_ctx)
     assert "credentials failure" in str(exc_info.value)
+    assert isinstance(exc_info.value.__cause__, exceptions.ClientCertError)
 
 
 @mock.patch("google.auth.transport.mtls.load_default_client_cert", autospec=True)

From 27fae9f5d92f972273f3a3d0f82351cc8b40ce2b Mon Sep 17 00:00:00 2001
From: nbayati <99771966+nbayati@users.noreply.github.com>
Date: Fri, 3 Jul 2026 13:54:26 -0700
Subject: [PATCH 04/10] fix: logic polish and docstring correction for mtls

---
 .../google-auth/google/auth/transport/mtls.py | 17 +++++++++--------
 .../google-auth/tests/transport/test_mtls.py  | 19 +++++--------------
 2 files changed, 14 insertions(+), 22 deletions(-)

diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py
index bff0b8c50729..5efdec5092a8 100644
--- a/packages/google-auth/google/auth/transport/mtls.py
+++ b/packages/google-auth/google/auth/transport/mtls.py
@@ -286,6 +286,9 @@ def get_default_ssl_context() -> Optional[ssl.SSLContext]:
         google.auth.exceptions.MutualTLSChannelError: If the default client certificate
             or key is malformed.
     """
+    if not should_use_client_cert() or not has_default_client_cert_source():
+        return None
+
     ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
     return ctx if load_default_client_cert(ctx) else None
 
@@ -300,9 +303,9 @@ def should_use_mtls_endpoint(
     or unset, returns whether a client certificate is available.
 
     Args:
-        client_cert_available (bool): indicating if a client certificate
+        client_cert_available (Optional[bool]): indicating if a client certificate
             is available. If None, this is determined by checking if client
-            certificates are enabled and a default source is present.
+            certificates are enabled using :func:`should_use_client_cert`.
 
     Returns:
         bool: indicating if an mTLS endpoint should be used.
@@ -311,7 +314,7 @@ def should_use_mtls_endpoint(
         client_cert_available = should_use_client_cert()
 
     use_mtls_endpoint = getenv(environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT, "auto")
-    use_mtls_endpoint = use_mtls_endpoint.lower()
+    use_mtls_endpoint = use_mtls_endpoint.strip().lower()
     if use_mtls_endpoint == "always":
         return True
     if use_mtls_endpoint == "never":
@@ -319,9 +322,7 @@ def should_use_mtls_endpoint(
     if use_mtls_endpoint == "auto":
         return client_cert_available
 
-    _LOGGER.warning(
-        "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value %r. Accepted "
-        "values: never, auto, always. Defaulting to auto.",
-        use_mtls_endpoint,
+    raise exceptions.MutualTLSChannelError(
+        f"Unsupported {environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT} value "
+        f"'{use_mtls_endpoint}'. Accepted values: never, auto, always."
     )
-    return client_cert_available
diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py
index 25fa094ca154..f964825652fa 100644
--- a/packages/google-auth/tests/transport/test_mtls.py
+++ b/packages/google-auth/tests/transport/test_mtls.py
@@ -371,8 +371,6 @@ def test_get_default_ssl_context_unconfigured(mock_create_context, mock_load_def
         ("NEVER", False, False),
         ("AUTO", True, True),
         ("AUTO", False, False),
-        ("invalid_val", True, True),
-        ("invalid_val", False, False),
     ],
 )
 @mock.patch(
@@ -397,23 +395,16 @@ def test_should_use_mtls_endpoint(
     "GOOGLE_API_USE_MTLS_ENDPOINT",
 )
 @mock.patch("google.auth.transport.mtls.getenv", autospec=True)
-def test_should_use_mtls_endpoint_invalid_value(mock_getenv, caplog):
+def test_should_use_mtls_endpoint_invalid_value(mock_getenv):
     mock_getenv.side_effect = (
         lambda var, default=None: "invalid_value"
         if var == "GOOGLE_API_USE_MTLS_ENDPOINT"
         else default
     )
-    with caplog.at_level("WARNING"):
-        assert mtls.should_use_mtls_endpoint(True) is True
-        assert "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value" in caplog.text
-        assert "Defaulting to auto" in caplog.text
-
-    caplog.clear()
-
-    with caplog.at_level("WARNING"):
-        assert mtls.should_use_mtls_endpoint(False) is False
-        assert "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value" in caplog.text
-        assert "Defaulting to auto" in caplog.text
+    with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
+        mtls.should_use_mtls_endpoint(True)
+    assert "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value" in str(exc_info.value)
+    assert "'invalid_value'" in str(exc_info.value)
 
 
 @mock.patch(

From 93d6b82159d008e1595c97298c8d6d30f7565167 Mon Sep 17 00:00:00 2001
From: nbayati <99771966+nbayati@users.noreply.github.com>
Date: Fri, 3 Jul 2026 14:03:00 -0700
Subject: [PATCH 05/10] test: fill mtls test suite gaps and improve mock rigor

---
 .../tests/transport/test__mtls_helper.py      | 100 ++++++++++++++++++
 .../google-auth/tests/transport/test_mtls.py  |  81 ++++++++------
 2 files changed, 149 insertions(+), 32 deletions(-)

diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py
index 281bd4111662..0a21e5384c02 100644
--- a/packages/google-auth/tests/transport/test__mtls_helper.py
+++ b/packages/google-auth/tests/transport/test__mtls_helper.py
@@ -1204,6 +1204,31 @@ def test_falls_back_to_tempfile_when_memfd_fails(
             assert key_path == "/tmp/key"
             assert passphrase == b"new_pass"
 
+    @mock.patch.object(sys, "platform", "linux")
+    @mock.patch.object(os, "memfd_create", create=True)
+    @mock.patch.object(_mtls_helper, "_memfd_cert_key_paths", autospec=True)
+    @mock.patch.object(_mtls_helper, "_tempfile_cert_key_paths", autospec=True)
+    def test_tier3_fallback_failure_propagation(
+        self, mock_tempfile_cm, mock_memfd_cm, mock_memfd_create
+    ):
+        mock_memfd_ctx = mock.MagicMock()
+        mock_memfd_ctx.__enter__.side_effect = _mtls_helper._MemfdCreationError(
+            "memfd failed"
+        )
+        mock_memfd_cm.return_value = mock_memfd_ctx
+
+        mock_tempfile_ctx = mock.MagicMock()
+        mock_tempfile_ctx.__enter__.side_effect = OSError("tempfile failed")
+        mock_tempfile_cm.return_value = mock_tempfile_ctx
+
+        with pytest.raises(OSError) as exc_info:
+            with _mtls_helper.secure_cert_key_paths(
+                pytest.public_cert_bytes, pytest.private_key_bytes, b"passphrase"
+            ):
+                pass
+
+        assert "tempfile failed" in str(exc_info.value)
+
     @mock.patch.object(sys, "platform", "darwin")
     @mock.patch.object(_mtls_helper, "_tempfile_cert_key_paths", autospec=True)
     def test_uses_tempfile_directly_on_unsupported_os(self, mock_tempfile_cm):
@@ -1378,6 +1403,46 @@ def _redirect_mkstemp(dir=None):
             # Verify cert file is still cleaned up even if key cleanup raised KeyboardInterrupt
             assert not os.path.exists(cert_path)
 
+    @mock.patch.object(os, "access", return_value=True)
+    @mock.patch.object(os.path, "isdir", return_value=True)
+    @mock.patch.object(_mtls_helper, "_encrypt_key_if_plaintext", autospec=True)
+    def test_cleanup_on_key_write_error(
+        self, mock_encrypt, mock_isdir, mock_access, tmpdir
+    ):
+        import builtins
+
+        original_mkstemp = tempfile.mkstemp
+
+        cert_path_ref = []
+
+        def _redirect_mkstemp(dir=None):
+            fd, path = original_mkstemp(dir=str(tmpdir))
+            cert_path_ref.append(path)
+            return fd, path
+
+        with mock.patch.object(tempfile, "mkstemp", side_effect=_redirect_mkstemp):
+            mock_encrypt.return_value = (b"encrypted_key", b"new_pass")
+
+            import os
+            original_fdopen = os.fdopen
+            call_count = [0]
+
+            def _failing_fdopen(fd, *args, **kwargs):
+                call_count[0] += 1
+                if call_count[0] == 2:
+                    raise OSError("key write failed")
+                return original_fdopen(fd, *args, **kwargs)
+
+            with pytest.raises(OSError):
+                with mock.patch("os.fdopen", side_effect=_failing_fdopen):
+                    with _mtls_helper._tempfile_cert_key_paths(
+                        b"cert", b"key", b"pass"
+                    ):
+                        pass
+
+            assert len(cert_path_ref) > 0
+            assert not os.path.exists(cert_path_ref[0])
+
     @mock.patch.object(os, "access", return_value=True)
     @mock.patch.object(os.path, "isdir", return_value=True)
     @mock.patch.object(_mtls_helper, "_encrypt_key_if_plaintext", autospec=True)
@@ -1472,6 +1537,21 @@ def test_encrypts_plaintext_key(self):
         )
         assert decrypted
 
+        original_key = serialization.load_pem_private_key(
+            pytest.private_key_bytes, password=None
+        )
+        decrypted_bytes = decrypted.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+        original_bytes = original_key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+        assert decrypted_bytes == original_bytes
+
     @mock.patch("secrets.token_hex", return_value="0123456789abcdef0123456789abcdef")
     def test_default_passphrase_generation(self, mock_secrets):
         encrypted_bytes, passphrase = _mtls_helper._encrypt_key_if_plaintext(
@@ -1495,6 +1575,24 @@ def test_encrypts_plaintext_key_string_passphrase(self):
         assert encrypted_bytes != pytest.private_key_bytes
         assert b"ENCRYPTED PRIVATE KEY" in encrypted_bytes
 
+        decrypted = serialization.load_pem_private_key(
+            encrypted_bytes, password=b"my_passphrase_str"
+        )
+        original_key = serialization.load_pem_private_key(
+            pytest.private_key_bytes, password=None
+        )
+        decrypted_bytes = decrypted.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+        original_bytes = original_key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+        assert decrypted_bytes == original_bytes
+
     def test_returns_unsupported_algorithm_asis(self):
         import cryptography.exceptions
 
@@ -1537,6 +1635,7 @@ def test_success(
 
         mock_open.assert_called_once_with("/path/to/secret", "r+b")
         mock_fh.write.assert_called_once_with(b"\0" * 10)
+        mock_fh.flush.assert_called_once()
         mock_fsync.assert_called_once()
         mock_remove.assert_called_once_with("/path/to/secret")
 
@@ -1583,5 +1682,6 @@ def test_remove_oserror_ignored(
         _mtls_helper._secure_wipe_and_remove("/path/to/secret")
 
         mock_open.assert_called_once_with("/path/to/secret", "r+b")
+        mock_fh.flush.assert_called_once()
         mock_fsync.assert_called_once()
         mock_remove.assert_called_once_with("/path/to/secret")
diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py
index f964825652fa..a52ac914f6dc 100644
--- a/packages/google-auth/tests/transport/test_mtls.py
+++ b/packages/google-auth/tests/transport/test_mtls.py
@@ -198,6 +198,7 @@ def test_load_client_cert_into_context_success(mock_secure_paths):
         mock_ctx, b"cert", b"key", passphrase=b"passphrase"
     )
     assert result is None
+    mock_secure_paths.assert_called_once_with(b"cert", b"key", passphrase=b"passphrase")
     mock_ctx.load_cert_chain.assert_called_once_with(
         certfile="cert_path", keyfile="key_path", password="passphrase"
     )
@@ -207,13 +208,37 @@ def test_load_client_cert_into_context_success(mock_secure_paths):
     "google.auth.transport._mtls_helper.secure_cert_key_paths",
     side_effect=_fake_secure_paths,
 )
-def test_load_client_cert_into_context_error(mock_secure_paths):
+def test_load_client_cert_into_context_success_no_passphrase(mock_secure_paths):
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
-    mock_ctx.load_cert_chain.side_effect = ssl.SSLError("boom")
+    result = mtls._load_client_cert_into_context(mock_ctx, b"cert", b"key")
+    assert result is None
+    mock_secure_paths.assert_called_once_with(b"cert", b"key", passphrase=None)
+    mock_ctx.load_cert_chain.assert_called_once_with(
+        certfile="cert_path", keyfile="key_path", password=None
+    )
+
+
+@pytest.mark.parametrize(
+    "exception",
+    [
+        ssl.SSLError("mock error message"),
+        OSError("mock error message"),
+        ValueError("mock error message"),
+        RuntimeError("mock error message"),
+        TypeError("mock error message"),
+    ],
+)
+@mock.patch(
+    "google.auth.transport._mtls_helper.secure_cert_key_paths",
+    side_effect=_fake_secure_paths,
+)
+def test_load_client_cert_into_context_error(mock_secure_paths, exception):
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    mock_ctx.load_cert_chain.side_effect = exception
     with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
         mtls._load_client_cert_into_context(mock_ctx, b"cert", b"key")
-    assert "boom" in str(exc_info.value)
-    assert isinstance(exc_info.value.__cause__, ssl.SSLError)
+    assert "mock error message" in str(exc_info.value)
+    assert isinstance(exc_info.value.__cause__, type(exception))
 
 
 @pytest.mark.parametrize("invalid_ctx", [None, object()])
@@ -227,19 +252,6 @@ def test_load_client_cert_into_context_invalid_ctx(invalid_ctx):
     assert exc_info.value.__cause__ is None
 
 
-@mock.patch(
-    "google.auth.transport._mtls_helper.secure_cert_key_paths",
-    side_effect=_fake_secure_paths,
-)
-def test_load_client_cert_into_context_load_chain_type_error(mock_secure_paths):
-    mock_ctx = mock.Mock(spec=ssl.SSLContext)
-    mock_ctx.load_cert_chain.side_effect = TypeError("invalid password type")
-    with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
-        mtls._load_client_cert_into_context(mock_ctx, b"cert", b"key")
-    assert "invalid password type" in str(exc_info.value)
-    assert isinstance(exc_info.value.__cause__, TypeError)
-
-
 @mock.patch("google.auth.transport.mtls._load_client_cert_into_context", autospec=True)
 @mock.patch("ssl.create_default_context", autospec=True)
 def test_make_client_cert_ssl_context(mock_create_context, mock_load_cert):
@@ -289,14 +301,17 @@ def test_load_default_client_cert_no_cert(
     mock_ctx.load_cert_chain.assert_not_called()
 
 
-@mock.patch("google.auth.transport.mtls._load_client_cert_into_context", autospec=True)
+@mock.patch(
+    "google.auth.transport._mtls_helper.secure_cert_key_paths",
+    side_effect=_fake_secure_paths,
+)
 @mock.patch(
     "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True
 )
 @mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
 @mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
 def test_load_default_client_cert_success(
-    mock_should_use, mock_has_source, mock_get_credentials, mock_load_cert
+    mock_should_use, mock_has_source, mock_get_credentials, mock_secure_paths
 ):
     mock_should_use.return_value = True
     mock_has_source.return_value = True
@@ -304,7 +319,9 @@ def test_load_default_client_cert_success(
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
 
     assert mtls.load_default_client_cert(mock_ctx) is True
-    mock_load_cert.assert_called_once_with(mock_ctx, b"cert", b"key", b"passphrase")
+    mock_ctx.load_cert_chain.assert_called_once_with(
+        certfile="cert_path", keyfile="key_path", password="passphrase"
+    )
 
 
 @mock.patch(
@@ -354,6 +371,18 @@ def test_get_default_ssl_context_unconfigured(mock_create_context, mock_load_def
     mock_load_default.assert_called_once_with(mock_ctx)
 
 
+@mock.patch("google.auth.transport.mtls.load_default_client_cert", autospec=True)
+@mock.patch("ssl.create_default_context", autospec=True)
+def test_get_default_ssl_context_exception(mock_create_context, mock_load_default):
+    mock_ctx = mock.Mock(spec=ssl.SSLContext)
+    mock_create_context.return_value = mock_ctx
+    mock_load_default.side_effect = exceptions.ClientCertError("mock error message")
+
+    with pytest.raises(exceptions.ClientCertError) as exc_info:
+        mtls.get_default_ssl_context()
+    assert "mock error message" in str(exc_info.value)
+
+
 @pytest.mark.parametrize(
     "env_val,client_cert_available,expected",
     [
@@ -373,10 +402,6 @@ def test_get_default_ssl_context_unconfigured(mock_create_context, mock_load_def
         ("AUTO", False, False),
     ],
 )
-@mock.patch(
-    "google.auth.environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT",
-    "GOOGLE_API_USE_MTLS_ENDPOINT",
-)
 @mock.patch("google.auth.transport.mtls.getenv", autospec=True)
 def test_should_use_mtls_endpoint(
     mock_getenv, env_val, client_cert_available, expected
@@ -390,10 +415,6 @@ def test_should_use_mtls_endpoint(
     assert result == expected
 
 
-@mock.patch(
-    "google.auth.environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT",
-    "GOOGLE_API_USE_MTLS_ENDPOINT",
-)
 @mock.patch("google.auth.transport.mtls.getenv", autospec=True)
 def test_should_use_mtls_endpoint_invalid_value(mock_getenv):
     mock_getenv.side_effect = (
@@ -407,10 +428,6 @@ def test_should_use_mtls_endpoint_invalid_value(mock_getenv):
     assert "'invalid_value'" in str(exc_info.value)
 
 
-@mock.patch(
-    "google.auth.environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT",
-    "GOOGLE_API_USE_MTLS_ENDPOINT",
-)
 @mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
 @mock.patch("google.auth.transport.mtls.getenv", autospec=True)
 def test_should_use_mtls_endpoint_default_client_cert(

From 805bb049548fa99ae9e7cadd1f81fb8c3dbdb33c Mon Sep 17 00:00:00 2001
From: nbayati <99771966+nbayati@users.noreply.github.com>
Date: Fri, 3 Jul 2026 15:41:27 -0700
Subject: [PATCH 06/10] test: add missing unit tests

---
 .../tests/transport/test__mtls_helper.py      |  3 --
 .../google-auth/tests/transport/test_mtls.py  | 46 ++++++++++++-------
 2 files changed, 30 insertions(+), 19 deletions(-)

diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py
index 0a21e5384c02..e7b80e92d0ab 100644
--- a/packages/google-auth/tests/transport/test__mtls_helper.py
+++ b/packages/google-auth/tests/transport/test__mtls_helper.py
@@ -1409,8 +1409,6 @@ def _redirect_mkstemp(dir=None):
     def test_cleanup_on_key_write_error(
         self, mock_encrypt, mock_isdir, mock_access, tmpdir
     ):
-        import builtins
-
         original_mkstemp = tempfile.mkstemp
 
         cert_path_ref = []
@@ -1423,7 +1421,6 @@ def _redirect_mkstemp(dir=None):
         with mock.patch.object(tempfile, "mkstemp", side_effect=_redirect_mkstemp):
             mock_encrypt.return_value = (b"encrypted_key", b"new_pass")
 
-            import os
             original_fdopen = os.fdopen
             call_count = [0]
 
diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py
index a52ac914f6dc..94d08906a91b 100644
--- a/packages/google-auth/tests/transport/test_mtls.py
+++ b/packages/google-auth/tests/transport/test_mtls.py
@@ -77,10 +77,8 @@ def side_effect(path):
 @mock.patch("google.auth.transport._mtls_helper._check_config_path", autospec=True)
 def test_has_default_client_cert_source_env_var_success(check_config_path, mock_getenv):
     # 1. Mock getenv to return our test path
-    mock_getenv.side_effect = (
-        lambda var: "path/to/cert.json"
-        if var == "GOOGLE_API_CERTIFICATE_CONFIG"
-        else None
+    mock_getenv.side_effect = lambda var: (
+        "path/to/cert.json" if var == "GOOGLE_API_CERTIFICATE_CONFIG" else None
     )
 
     # 2. Mock _check_config_path side effect
@@ -105,8 +103,8 @@ def test_has_default_client_cert_source_env_var_invalid_config_path(
     check_config_path, mock_getenv
 ):
     # Set the env var but make the check fail
-    mock_getenv.side_effect = (
-        lambda var: "invalid/path" if var == "GOOGLE_API_CERTIFICATE_CONFIG" else None
+    mock_getenv.side_effect = lambda var: (
+        "invalid/path" if var == "GOOGLE_API_CERTIFICATE_CONFIG" else None
     )
     check_config_path.return_value = None
 
@@ -406,8 +404,8 @@ def test_get_default_ssl_context_exception(mock_create_context, mock_load_defaul
 def test_should_use_mtls_endpoint(
     mock_getenv, env_val, client_cert_available, expected
 ):
-    mock_getenv.side_effect = (
-        lambda var, default=None: env_val
+    mock_getenv.side_effect = lambda var, default=None: (
+        env_val
         if (var == "GOOGLE_API_USE_MTLS_ENDPOINT" and env_val is not None)
         else default
     )
@@ -417,10 +415,8 @@ def test_should_use_mtls_endpoint(
 
 @mock.patch("google.auth.transport.mtls.getenv", autospec=True)
 def test_should_use_mtls_endpoint_invalid_value(mock_getenv):
-    mock_getenv.side_effect = (
-        lambda var, default=None: "invalid_value"
-        if var == "GOOGLE_API_USE_MTLS_ENDPOINT"
-        else default
+    mock_getenv.side_effect = lambda var, default=None: (
+        "invalid_value" if var == "GOOGLE_API_USE_MTLS_ENDPOINT" else default
     )
     with pytest.raises(exceptions.MutualTLSChannelError) as exc_info:
         mtls.should_use_mtls_endpoint(True)
@@ -433,10 +429,8 @@ def test_should_use_mtls_endpoint_invalid_value(mock_getenv):
 def test_should_use_mtls_endpoint_default_client_cert(
     mock_getenv, mock_should_use_client_cert
 ):
-    mock_getenv.side_effect = (
-        lambda var, default=None: "auto"
-        if var == "GOOGLE_API_USE_MTLS_ENDPOINT"
-        else default
+    mock_getenv.side_effect = lambda var, default=None: (
+        "auto" if var == "GOOGLE_API_USE_MTLS_ENDPOINT" else default
     )
     mock_should_use_client_cert.return_value = True
     assert mtls.should_use_mtls_endpoint() is True
@@ -447,3 +441,23 @@ def test_should_use_mtls_endpoint_default_client_cert(
     mock_should_use_client_cert.return_value = False
     assert mtls.should_use_mtls_endpoint() is False
     mock_should_use_client_cert.assert_called_once()
+
+
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
+@mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
+def test_get_default_ssl_context_not_should_use(mock_has_default, mock_should_use):
+    mock_should_use.return_value = False
+    mock_has_default.return_value = True
+
+    result = mtls.get_default_ssl_context()
+    assert result is None
+
+
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
+@mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
+def test_get_default_ssl_context_no_default_source(mock_has_default, mock_should_use):
+    mock_should_use.return_value = True
+    mock_has_default.return_value = False
+
+    result = mtls.get_default_ssl_context()
+    assert result is None

From f167b3db23a51ee8655943e782f657df0d96a795 Mon Sep 17 00:00:00 2001
From: nbayati <99771966+nbayati@users.noreply.github.com>
Date: Sat, 4 Jul 2026 10:24:08 -0700
Subject: [PATCH 07/10] chore: remove unused sync
 `_make_client_cert_ssl_context`

Removes the private `_make_client_cert_ssl_context` method and its corresponding unit test from `google.auth.transport.mtls`.

Initially, this method was added to mirror the async API (`aio.transport.mtls.make_client_cert_ssl_context`). However, upon further review, we decided that external client libraries do not need it (they should use `get_default_ssl_context` or `load_default_client_cert` instead), and there is no need for it internally within the synchronous transports (e.g., `requests`, `urllib3`) which handle their mTLS context initialization independently. Removing this dead code reduces the internal API surface area and eliminates unnecessary maintenance overhead.
---
 .../google-auth/google/auth/transport/mtls.py | 23 -------------------
 .../google-auth/tests/transport/test_mtls.py  | 15 ------------
 2 files changed, 38 deletions(-)

diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py
index 5efdec5092a8..0588f86db796 100644
--- a/packages/google-auth/google/auth/transport/mtls.py
+++ b/packages/google-auth/google/auth/transport/mtls.py
@@ -205,29 +205,6 @@ def _load_client_cert_into_context(
         raise new_exc from caught_exc
 
 
-def _make_client_cert_ssl_context(
-    cert_bytes: bytes,
-    key_bytes: bytes,
-    passphrase: Optional[bytes] = None,
-) -> ssl.SSLContext:
-    """Create a default SSL context loaded with the client certificate and key.
-
-    Args:
-        cert_bytes (bytes): The client certificate bytes in PEM format.
-        key_bytes (bytes): The client private key bytes in PEM format.
-        passphrase (Optional[bytes]): The passphrase for the client private key.
-
-    Returns:
-        ssl.SSLContext: The SSL context loaded with the client certificate and key.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If loading the certificate and key fails.
-    """
-    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
-    _load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase=passphrase)
-    return ctx
-
-
 def load_default_client_cert(ctx: ssl.SSLContext) -> bool:
     """Load the default client certificate and key into an SSL context if configured.
 
diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py
index 94d08906a91b..131268832dfa 100644
--- a/packages/google-auth/tests/transport/test_mtls.py
+++ b/packages/google-auth/tests/transport/test_mtls.py
@@ -250,21 +250,6 @@ def test_load_client_cert_into_context_invalid_ctx(invalid_ctx):
     assert exc_info.value.__cause__ is None
 
 
-@mock.patch("google.auth.transport.mtls._load_client_cert_into_context", autospec=True)
-@mock.patch("ssl.create_default_context", autospec=True)
-def test_make_client_cert_ssl_context(mock_create_context, mock_load_cert):
-    mock_ctx = mock.Mock(spec=ssl.SSLContext)
-    mock_create_context.return_value = mock_ctx
-
-    result = mtls._make_client_cert_ssl_context(b"cert", b"key", b"passphrase")
-
-    assert result == mock_ctx
-    mock_create_context.assert_called_once_with(ssl.Purpose.SERVER_AUTH)
-    mock_load_cert.assert_called_once_with(
-        mock_ctx, b"cert", b"key", passphrase=b"passphrase"
-    )
-
-
 @mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
 def test_load_default_client_cert_disabled(mock_should_use):
     mock_should_use.return_value = False

From ae950565db2847875549b4870d4a10e3c19f75c1 Mon Sep 17 00:00:00 2001
From: nbayati <99771966+nbayati@users.noreply.github.com>
Date: Sat, 4 Jul 2026 10:31:25 -0700
Subject: [PATCH 08/10] fix: default to auto for empty mtls endpoint
 environment variable

---
 packages/google-auth/google/auth/transport/mtls.py | 2 +-
 packages/google-auth/tests/transport/test_mtls.py  | 2 ++
 2 files changed, 3 insertions(+), 1 deletion(-)

diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py
index 0588f86db796..520f40a838fb 100644
--- a/packages/google-auth/google/auth/transport/mtls.py
+++ b/packages/google-auth/google/auth/transport/mtls.py
@@ -291,7 +291,7 @@ def should_use_mtls_endpoint(
         client_cert_available = should_use_client_cert()
 
     use_mtls_endpoint = getenv(environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT, "auto")
-    use_mtls_endpoint = use_mtls_endpoint.strip().lower()
+    use_mtls_endpoint = (use_mtls_endpoint or "auto").strip().lower()
     if use_mtls_endpoint == "always":
         return True
     if use_mtls_endpoint == "never":
diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py
index 131268832dfa..eef75b0f1549 100644
--- a/packages/google-auth/tests/transport/test_mtls.py
+++ b/packages/google-auth/tests/transport/test_mtls.py
@@ -383,6 +383,8 @@ def test_get_default_ssl_context_exception(mock_create_context, mock_load_defaul
         ("NEVER", False, False),
         ("AUTO", True, True),
         ("AUTO", False, False),
+        ("", True, True),
+        ("", False, False),
     ],
 )
 @mock.patch("google.auth.transport.mtls.getenv", autospec=True)

From 57b252565d811821c352ca318eb39af7951b953a Mon Sep 17 00:00:00 2001
From: nbayati <99771966+nbayati@users.noreply.github.com>
Date: Sat, 4 Jul 2026 10:48:38 -0700
Subject: [PATCH 09/10] fix: resolve mypy type narrowing in mtls.py

---
 packages/google-auth/google/auth/transport/mtls.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/packages/google-auth/google/auth/transport/mtls.py b/packages/google-auth/google/auth/transport/mtls.py
index 520f40a838fb..f4c48ff36f5e 100644
--- a/packages/google-auth/google/auth/transport/mtls.py
+++ b/packages/google-auth/google/auth/transport/mtls.py
@@ -191,6 +191,10 @@ def _load_client_cert_into_context(
                 if isinstance(passphrase_val, bytes)
                 else passphrase_val
             )
+            if cert_path is None or key_path is None:
+                raise exceptions.MutualTLSChannelError(
+                    "Failed to generate temporary file paths for the client certificate and key."
+                )
             ctx.load_cert_chain(
                 certfile=cert_path, keyfile=key_path, password=password_val
             )

From f8289d23a89a3929e43f4f423e89a36d0aea3c56 Mon Sep 17 00:00:00 2001
From: nbayati <99771966+nbayati@users.noreply.github.com>
Date: Sun, 5 Jul 2026 10:29:40 -0700
Subject: [PATCH 10/10] test: mock mtls endpoint checks in
 get_default_ssl_context tests

---
 .../google-auth/tests/transport/test_mtls.py  | 24 ++++++++++++++++---
 1 file changed, 21 insertions(+), 3 deletions(-)

diff --git a/packages/google-auth/tests/transport/test_mtls.py b/packages/google-auth/tests/transport/test_mtls.py
index eef75b0f1549..ad08df712516 100644
--- a/packages/google-auth/tests/transport/test_mtls.py
+++ b/packages/google-auth/tests/transport/test_mtls.py
@@ -326,9 +326,15 @@ def test_load_default_client_cert_propagates_client_cert_error(
     assert isinstance(exc_info.value.__cause__, exceptions.ClientCertError)
 
 
+@mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
 @mock.patch("google.auth.transport.mtls.load_default_client_cert", autospec=True)
 @mock.patch("ssl.create_default_context", autospec=True)
-def test_get_default_ssl_context_configured(mock_create_context, mock_load_default):
+def test_get_default_ssl_context_configured(
+    mock_create_context, mock_load_default, mock_should_use, mock_has_source
+):
+    mock_should_use.return_value = True
+    mock_has_source.return_value = True
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
     mock_create_context.return_value = mock_ctx
     mock_load_default.return_value = True
@@ -340,9 +346,15 @@ def test_get_default_ssl_context_configured(mock_create_context, mock_load_defau
     mock_load_default.assert_called_once_with(mock_ctx)
 
 
+@mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
 @mock.patch("google.auth.transport.mtls.load_default_client_cert", autospec=True)
 @mock.patch("ssl.create_default_context", autospec=True)
-def test_get_default_ssl_context_unconfigured(mock_create_context, mock_load_default):
+def test_get_default_ssl_context_unconfigured(
+    mock_create_context, mock_load_default, mock_should_use, mock_has_source
+):
+    mock_should_use.return_value = True
+    mock_has_source.return_value = True
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
     mock_create_context.return_value = mock_ctx
     mock_load_default.return_value = False
@@ -354,9 +366,15 @@ def test_get_default_ssl_context_unconfigured(mock_create_context, mock_load_def
     mock_load_default.assert_called_once_with(mock_ctx)
 
 
+@mock.patch("google.auth.transport.mtls.has_default_client_cert_source", autospec=True)
+@mock.patch("google.auth.transport.mtls.should_use_client_cert", autospec=True)
 @mock.patch("google.auth.transport.mtls.load_default_client_cert", autospec=True)
 @mock.patch("ssl.create_default_context", autospec=True)
-def test_get_default_ssl_context_exception(mock_create_context, mock_load_default):
+def test_get_default_ssl_context_exception(
+    mock_create_context, mock_load_default, mock_should_use, mock_has_source
+):
+    mock_should_use.return_value = True
+    mock_has_source.return_value = True
     mock_ctx = mock.Mock(spec=ssl.SSLContext)
     mock_create_context.return_value = mock_ctx
     mock_load_default.side_effect = exceptions.ClientCertError("mock error message")