googleapis · ohmayr · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026
@@ -37,15 +37,19 @@
 version is at least 1.4.0.
 """
 
+from typing import Mapping
+
 from google.auth.crypt import base
 from google.auth.crypt import es
 from google.auth.crypt import es256
+from google.auth.crypt import pqc
 from google.auth.crypt import rsa
 
 EsSigner = es.EsSigner
 EsVerifier = es.EsVerifier
 ES256Signer = es256.ES256Signer
 ES256Verifier = es256.ES256Verifier
+PqcSigner = pqc.PqcSigner
 
 
 # Aliases to maintain the v1.0.0 interface, as the crypt module was split
@@ -56,6 +60,36 @@
 RSAVerifier = rsa.RSAVerifier
 
 
+def from_service_account_info(info: Mapping[str, str]) -> base.Signer:
+    """Create a Signer instance from a service account info dictionary.
+
+    Automatically detects whether the private key is RSA, ECDSA, or PQC (ML-DSA)
+    and returns the appropriate Signer instance.
+
+    Args:
+        info (Mapping[str, str]): Service account info dictionary.
+
+    Returns:
+        google.auth.crypt.Signer: The constructed signer.
+    """
+    private_key = info.get("private_key")
+    key_id = info.get("private_key_id")
+    if not private_key:
+        raise ValueError("The private_key field is missing from service account info.")
+
+    try:
+        return RSASigner.from_service_account_info(info)
+    except (ValueError, TypeError):
+        pass
+
+    try:
+        return EsSigner.from_string(private_key, key_id=key_id)
+    except (ValueError, TypeError):
+        pass
+
+    return PqcSigner.from_string(private_key, key_id=key_id)
+
+
 def verify_signature(message, signature, certs, verifier_cls=rsa.RSAVerifier):
     """Verify an RSA or ECDSA cryptographic signature.
 
@@ -89,8 +123,10 @@ class to use for verification. This can be used to select different
     "EsVerifier",
     "ES256Signer",
     "ES256Verifier",
+    "PqcSigner",
     "RSASigner",
     "RSAVerifier",
     "Signer",
     "Verifier",
+    "from_service_account_info",
 ]
@@ -101,6 +101,11 @@ def __init__(self, private_key, key_id=None):
         self._key = private_key
         self._key_id = key_id
 
+    @property
+    def algorithm(self):
+        """Name of the algorithm used to sign messages."""
+        return "RS256"
+
     @property  # type: ignore
     @_helpers.copy_docstring(base.Signer)
     def key_id(self):

@@ -151,6 +151,11 @@ def __init__(self, private_key, key_id=None):
         self._key = private_key
         self._key_id = key_id
 
+    @property
+    def algorithm(self):
+        """Name of the algorithm used to sign messages."""
+        return "RS256"
+
     @property  # type: ignore
     @_helpers.copy_docstring(base.Signer)
     def key_id(self):

@@ -52,6 +52,11 @@ def key_id(self):
         """Optional[str]: The key ID used to identify this private key."""
         raise NotImplementedError("Key id must be implemented")
 
+    @property
+    def algorithm(self):
+        """Optional[str]: Name of the algorithm used by this signer (e.g. 'RS256', 'ES256', 'ML-DSA-65')."""
+        return None
+
     @abc.abstractmethod
     def sign(self, message):
         """Signs a message.

@@ -0,0 +1,125 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""PQC (ML-DSA) verifier and signer that use the ``cryptography`` library."""
+
+from typing import Any, Optional, Union
+
+from cryptography.hazmat import backends
+
+try:
+    from cryptography.hazmat.primitives import serialization
+except ImportError:  # pragma: NO COVER
+    pass
+
+try:
+    from cryptography.hazmat.primitives.asymmetric import mldsa
+except ImportError:  # pragma: NO COVER
+    mldsa = None  # type: ignore
+
+from google.auth import _helpers
+from google.auth.crypt import base
+
+_BACKEND = backends.default_backend()
+
+
+class PqcSigner(base.Signer, base.FromServiceAccountMixin):
+    """Signs messages with a Post-Quantum Cryptography (ML-DSA) private key.
+
+    Args:
+        private_key: The ML-DSA private key object from cryptography.
+        key_id (str): Optional key ID used to identify this private key.
+    """
+
+    def __init__(self, private_key: Any, key_id: Optional[str] = None) -> None:
+        self._key = private_key
+        self._key_id = key_id
+
+    @property
+    def algorithm(self) -> str:
+        """Name of the algorithm used to sign messages.
+
+        Returns:
+            str: The algorithm name (e.g. 'ML-DSA-65' or 'ML-DSA-87').
+        """
+        if mldsa and isinstance(
+            self._key, getattr(mldsa, "MLDSA87PrivateKey", type(None))
+        ):
+            return "ML-DSA-87"
+        return "ML-DSA-65"
+
+    @property  # type: ignore
+    @_helpers.copy_docstring(base.Signer)
+    def key_id(self) -> Optional[str]:
+        return self._key_id
+
+    @_helpers.copy_docstring(base.Signer)
+    def sign(self, message: Union[str, bytes]) -> bytes:
+        message = _helpers.to_bytes(message)
+        return self._key.sign(message)
+
+    @classmethod
+    def from_string(  # type: ignore[override]
+        cls, key: Union[bytes, str], key_id: Optional[str] = None
+    ) -> "PqcSigner":
+        """Construct a PqcSigner from a private key in PEM format.
+
+        Args:
+            key (Union[bytes, str]): Private key in PEM format.
+            key_id (str): An optional key id used to identify the private key.
+
+        Returns:
+            PqcSigner: The constructed signer.
+        """
+        key_bytes = _helpers.to_bytes(key)
+        private_key = serialization.load_pem_private_key(
+            key_bytes, password=None, backend=_BACKEND
+        )
+        return cls(private_key, key_id=key_id)
+
+
+def is_pqc_disabled() -> bool:
+    """Checks whether Post-Quantum Cryptography (PQC) TLS key exchange is disabled via environment variable.
+
+    Returns:
+        bool: True if GOOGLE_CLOUD_DISABLE_PQC is set to '1', 'true', or 'yes'.
+    """
+    import os
+    from google.auth import environment_vars
+
+    val = os.environ.get(environment_vars.GOOGLE_CLOUD_DISABLE_PQC, "").lower()
+    return val in ("1", "true", "yes")
+
+
+def configure_ssl_context_pqc(ssl_context: Any) -> Any:
+    """Configures an ssl.SSLContext object based on PQC environment settings.
+
+    If PQC is disabled via GOOGLE_CLOUD_DISABLE_PQC, this function restricts the
+    ECDH / key-exchange curves to classical algorithms (e.g. prime256v1:secp384r1),
+    disabling hybrid PQC key-exchange (X25519MLKEM768).
+
+    Args:
+        ssl_context: An ssl.SSLContext instance.
+
+    Returns:
+        The configured ssl.SSLContext instance.
+    """
+    if is_pqc_disabled():
+        if hasattr(ssl_context, "set_ecdh_curve"):
+            try:
+                ssl_context.set_ecdh_curve("prime256v1:secp384r1")
+            except Exception:  # pragma: NO COVER
+                pass
+    return ssl_context
+
@@ -103,6 +103,11 @@ def __init__(self, private_key, key_id=None):
             raise ValueError(f"unrecognized private key type: {type(private_key)}")
         self._impl = impl_lib.RSASigner(private_key, key_id=key_id)
 
+    @property
+    def algorithm(self):
+        """Name of the algorithm used to sign messages."""
+        return self._impl.algorithm
+
     @property  # type: ignore
     @_helpers.copy_docstring(base.Signer)
     def key_id(self):

@@ -133,3 +133,7 @@
     "GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES"
 )
 """Environment variable to prevent agent token sharing for GCP services."""
+
+GOOGLE_CLOUD_DISABLE_PQC = "GOOGLE_CLOUD_DISABLE_PQC"
+"""Environment variable to disable Post-Quantum Cryptography (PQC) key exchange negotiation in TLS."""
+
@@ -91,13 +91,16 @@ def encode(signer, payload, header=None, key_id=None):
         header = {}
 
     if key_id is None:
-        key_id = signer.key_id
+        key_id = getattr(signer, "key_id", None)
+        if not isinstance(key_id, str):
+            key_id = None
 
     header.update({"typ": "JWT"})
 
     if "alg" not in header:
-        if es is not None and isinstance(signer, es.EsSigner):
-            header.update({"alg": signer.algorithm})
+        alg = getattr(signer, "algorithm", None)
+        if isinstance(alg, str):
+            header.update({"alg": alg})
         else:
             header.update({"alg": "RS256"})
 

@@ -0,0 +1,61 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest import mock
+
+from google.auth.crypt import pqc
+
+
+def test_pqc_signer_algorithm():
+    mock_key = mock.MagicMock()
+    signer = pqc.PqcSigner(mock_key, key_id="key-123")
+    assert signer.key_id == "key-123"
+    assert signer.algorithm == "ML-DSA-65"
+
+
+def test_pqc_signer_sign():
+    mock_key = mock.MagicMock()
+    mock_key.sign.return_value = b"mock_pqc_signature"
+    signer = pqc.PqcSigner(mock_key, key_id="key-123")
+
+    sig = signer.sign("hello")
+    assert sig == b"mock_pqc_signature"
+    mock_key.sign.assert_called_once_with(b"hello")
+
+
+def test_is_pqc_disabled(monkeypatch):
+    assert not pqc.is_pqc_disabled()
+    monkeypatch.setenv("GOOGLE_CLOUD_DISABLE_PQC", "1")
+    assert pqc.is_pqc_disabled()
+    monkeypatch.setenv("GOOGLE_CLOUD_DISABLE_PQC", "true")
+    assert pqc.is_pqc_disabled()
+    monkeypatch.setenv("GOOGLE_CLOUD_DISABLE_PQC", "0")
+    assert not pqc.is_pqc_disabled()
+
+
+def test_configure_ssl_context_pqc_enabled(monkeypatch):
+    monkeypatch.delenv("GOOGLE_CLOUD_DISABLE_PQC", raising=False)
+    mock_ctx = mock.MagicMock()
+    result = pqc.configure_ssl_context_pqc(mock_ctx)
+    assert result == mock_ctx
+    mock_ctx.set_ecdh_curve.assert_not_called()
+
+
+def test_configure_ssl_context_pqc_disabled(monkeypatch):
+    monkeypatch.setenv("GOOGLE_CLOUD_DISABLE_PQC", "1")
+    mock_ctx = mock.MagicMock()
+    result = pqc.configure_ssl_context_pqc(mock_ctx)
+    assert result == mock_ctx
+    mock_ctx.set_ecdh_curve.assert_called_once_with("prime256v1:secp384r1")
+
@@ -120,6 +120,7 @@ def test_init_with_cryptography_key(self, cryptography_private_key):
         assert isinstance(signer._impl, _cryptography_rsa.RSASigner)
         assert signer._impl._key == cryptography_private_key
         assert signer._impl.key_id == "123"
+        assert signer.algorithm == "RS256"
 
     @pytest.mark.skipif(not rsa_lib, reason="rsa library not installed")
     def test_init_with_rsa_key(self, rsa_private_key):

@@ -142,6 +142,19 @@ def factory(
     return factory
 
 
+def test_encode_mock_signer_algorithm():
+    # Verify that passing a mock object as signer (where getattr(signer, "algorithm", None)
+    # returns another mock object) correctly falls back to "RS256" without raising JSON errors.
+    mock_signer = mock.MagicMock()
+    mock_signer.sign.return_value = b"mock_sig"
+
+    token = jwt.encode(mock_signer, {"sub": "user@example.com"})
+    header_b64 = token.split(b".")[0]
+    header = json.loads(_helpers.padded_urlsafe_b64decode(header_b64).decode("utf-8"))
+
+    assert header["alg"] == "RS256"
+
+
 def test_decode_valid(token_factory):
     payload = jwt.decode(token_factory(), certs=PUBLIC_CERT_BYTES)
     assert payload["aud"] == "audience@example.com"