diff --git a/packages/google-auth/google/auth/crypt/__init__.py b/packages/google-auth/google/auth/crypt/__init__.py index e56bc7b82df7..67fffca5937f 100644 --- a/packages/google-auth/google/auth/crypt/__init__.py +++ b/packages/google-auth/google/auth/crypt/__init__.py @@ -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", ] diff --git a/packages/google-auth/google/auth/crypt/_cryptography_rsa.py b/packages/google-auth/google/auth/crypt/_cryptography_rsa.py index 1a3e9ff52c66..80f7b185970f 100644 --- a/packages/google-auth/google/auth/crypt/_cryptography_rsa.py +++ b/packages/google-auth/google/auth/crypt/_cryptography_rsa.py @@ -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): diff --git a/packages/google-auth/google/auth/crypt/_python_rsa.py b/packages/google-auth/google/auth/crypt/_python_rsa.py index d9305e835dc9..62a828611544 100644 --- a/packages/google-auth/google/auth/crypt/_python_rsa.py +++ b/packages/google-auth/google/auth/crypt/_python_rsa.py @@ -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): diff --git a/packages/google-auth/google/auth/crypt/base.py b/packages/google-auth/google/auth/crypt/base.py index ad871c311566..1e3c2dcb68ee 100644 --- a/packages/google-auth/google/auth/crypt/base.py +++ b/packages/google-auth/google/auth/crypt/base.py @@ -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. diff --git a/packages/google-auth/google/auth/crypt/pqc.py b/packages/google-auth/google/auth/crypt/pqc.py new file mode 100644 index 000000000000..ed307560601a --- /dev/null +++ b/packages/google-auth/google/auth/crypt/pqc.py @@ -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 + diff --git a/packages/google-auth/google/auth/crypt/rsa.py b/packages/google-auth/google/auth/crypt/rsa.py index 639be9069549..e7e873c1f025 100644 --- a/packages/google-auth/google/auth/crypt/rsa.py +++ b/packages/google-auth/google/auth/crypt/rsa.py @@ -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): diff --git a/packages/google-auth/google/auth/environment_vars.py b/packages/google-auth/google/auth/environment_vars.py index b7ff66c8b54a..3c2d50301640 100644 --- a/packages/google-auth/google/auth/environment_vars.py +++ b/packages/google-auth/google/auth/environment_vars.py @@ -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.""" + diff --git a/packages/google-auth/google/auth/jwt.py b/packages/google-auth/google/auth/jwt.py index 1241aee70121..9d768c3ce975 100644 --- a/packages/google-auth/google/auth/jwt.py +++ b/packages/google-auth/google/auth/jwt.py @@ -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"}) diff --git a/packages/google-auth/tests/crypt/test_pqc.py b/packages/google-auth/tests/crypt/test_pqc.py new file mode 100644 index 000000000000..c912398bb1fc --- /dev/null +++ b/packages/google-auth/tests/crypt/test_pqc.py @@ -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") + diff --git a/packages/google-auth/tests/crypt/test_rsa.py b/packages/google-auth/tests/crypt/test_rsa.py index a54beb7632cf..50e7a9bd1b51 100644 --- a/packages/google-auth/tests/crypt/test_rsa.py +++ b/packages/google-auth/tests/crypt/test_rsa.py @@ -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): diff --git a/packages/google-auth/tests/test_jwt.py b/packages/google-auth/tests/test_jwt.py index 27b951b8b7bc..8a89505dfc62 100644 --- a/packages/google-auth/tests/test_jwt.py +++ b/packages/google-auth/tests/test_jwt.py @@ -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"