diff --git a/packages/gapic-generator/noxfile.py b/packages/gapic-generator/noxfile.py index 6fac5c48853d..343626eab4e4 100644 --- a/packages/gapic-generator/noxfile.py +++ b/packages/gapic-generator/noxfile.py @@ -457,7 +457,7 @@ def showcase_mtls( """Run the Showcase mtls test suite.""" with showcase_library(session, templates=templates, other_opts=other_opts): - session.install("pytest", "pytest-asyncio") + session.install("pytest", "pytest-asyncio", "pyopenssl") test_directory = Path("tests", "system") ignore_file = env.get("IGNORE_FILE") pytest_command = [ diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py index 180e48b8d59a..5c00c99a5976 100644 --- a/packages/gapic-generator/tests/system/conftest.py +++ b/packages/gapic-generator/tests/system/conftest.py @@ -18,6 +18,8 @@ import os import pytest import pytest_asyncio +from requests.adapters import HTTPAdapter +from urllib3.poolmanager import PoolManager from typing import Sequence, Tuple @@ -328,7 +330,7 @@ def _read_response_metadata_stream(self): def intercept_unary_unary(self, continuation, client_call_details, request): self._add_request_metadata(client_call_details) response = continuation(client_call_details, request) - metadata = [(k, str(v)) for k, v in response.trailing_metadata()] + metadata = [(k, str(v)) for k, v in response.initial_metadata()] + [(k, str(v)) for k, v in response.trailing_metadata()] self.response_metadata = metadata return response @@ -453,24 +455,44 @@ async def intercepted_echo_grpc_async(): return EchoAsyncClient(transport=transport), interceptor +class HostNameIgnoringAdapter(HTTPAdapter): + """Custom HTTPAdapter that disables hostname verification for local self-signed certs.""" + def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs): + self.poolmanager = PoolManager( + num_pools=connections, + maxsize=maxsize, + block=block, + assert_hostname=False, + **pool_kwargs + ) + + @pytest.fixture -def intercepted_echo_rest(): +def intercepted_echo_rest(use_mtls): transport_name = "rest" transport_cls = EchoClient.get_transport_class(transport_name) interceptor = EchoMetadataClientRestInterceptor() - # The custom host explicitly bypasses https. + url_scheme = "https" if use_mtls else "http" transport = transport_cls( credentials=ga_credentials.AnonymousCredentials(), host="localhost:7469", - url_scheme="http", + url_scheme=url_scheme, interceptor=interceptor, ) + if use_mtls: + dir = os.path.dirname(__file__) + cert_path = os.path.join(dir, "../cert/mtls.crt") + key_path = os.path.join(dir, "../cert/mtls.key") + transport._session.verify = cert_path + transport._session.cert = (cert_path, key_path) + transport._session.mount("https://", HostNameIgnoringAdapter()) + return EchoClient(transport=transport), interceptor @pytest.fixture -def intercepted_echo_rest_async(): +def intercepted_echo_rest_async(use_mtls): if not HAS_ASYNC_REST_ECHO_TRANSPORT: pytest.skip("Skipping test with async rest.") @@ -478,11 +500,19 @@ def intercepted_echo_rest_async(): transport_cls = EchoAsyncClient.get_transport_class(transport_name) interceptor = EchoMetadataClientRestAsyncInterceptor() - # The custom host explicitly bypasses https. + url_scheme = "https" if use_mtls else "http" transport = transport_cls( credentials=async_anonymous_credentials(), host="localhost:7469", - url_scheme="http", + url_scheme=url_scheme, interceptor=interceptor, ) + if use_mtls: + dir = os.path.dirname(__file__) + cert_path = os.path.join(dir, "../cert/mtls.crt") + key_path = os.path.join(dir, "../cert/mtls.key") + transport._session.verify = cert_path + transport._session.cert = (cert_path, key_path) + transport._session.mount("https://", HostNameIgnoringAdapter()) + return EchoAsyncClient(transport=transport), interceptor diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py new file mode 100644 index 000000000000..6b6876272c65 --- /dev/null +++ b/packages/gapic-generator/tests/system/test_pqc.py @@ -0,0 +1,98 @@ +# 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 +# +# https://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. + +import pytest +import requests +from google import showcase + + +@pytest.fixture +def run_pqc_test(use_mtls): + if not use_mtls: + pytest.skip("PQC integration test requires mTLS (--mtls flag) to be enabled.") + + +@pytest.mark.parametrize( + "transport_fixture", + ["intercepted_echo_grpc", "intercepted_echo_rest"] +) +def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture): + """Verifies that the generated client library negotiates PQC (X25519MLKEM768) with Showcase server.""" + client, interceptor = request.getfixturevalue(transport_fixture) + + # Make secure call using standard GAPIC client library fixture + response = client.echo(request=showcase.EchoRequest(content="Verify PQC connection.")) + assert response.content == "Verify PQC connection." + + # Extract negotiated group and supported groups from response headers + negotiated_group = None + supported_groups = None + for key, value in interceptor.response_metadata: + if key.lower() == "x-showcase-tls-group": + negotiated_group = value + elif key.lower() == "x-showcase-tls-client-supported-groups": + supported_groups = value + + assert negotiated_group is not None, "Failed: Showcase server did not return negotiated TLS group header." + assert supported_groups is not None, "Failed: Showcase server did not return client advertised supported groups." + + print(f"\n[PQC Verification] ({transport_fixture}) Negotiated TLS Group: {negotiated_group}") + print(f"[PQC Verification] ({transport_fixture}) Client Advertised Supported Groups: {supported_groups}") + + # Enforce PQC compliance (X25519MLKEM768 or Kyber) + assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \ + f"Failed: {transport_fixture} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}" + + +def test_google_auth_transport_pqc(run_pqc_test): + """Verifies that google-auth transport adapter negotiates PQC (X25519MLKEM768) with Showcase server.""" + import google.auth.transport.requests + from conftest import HostNameIgnoringAdapter + + # 1. Initialize requests Session with mTLS certs + session = requests.Session() + cert_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.crt" + key_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.key" + + session.verify = cert_path + session.cert = (cert_path, key_path) + session.mount("https://", HostNameIgnoringAdapter()) + + # 2. Wrap session in google-auth transport adapter + auth_transport = google.auth.transport.requests.Request(session=session) + + # 3. Serialize request body + req = showcase.EchoRequest(content="Verify google-auth transport PQC connection.") + body = showcase.EchoRequest.to_json(req, including_default_value_fields=False).encode("utf-8") + + # 4. Execute request through google-auth's transport layer + url = "https://localhost:7469/v1beta1/echo:echo" + headers = { + "Content-Type": "application/json", + "x-goog-api-client": "gapic/1.0 rest/1.0", + } + + response = auth_transport(url=url, method="POST", body=body, headers=headers) + assert response.status == 200, f"Failed: status={response.status}, body={response.data}" + + # 5. Extract and verify negotiated TLS group returned by Showcase + negotiated_group = response.headers.get("x-showcase-tls-group") + supported_groups = response.headers.get("x-showcase-tls-client-supported-groups") + + assert negotiated_group is not None, "Failed: Showcase server did not return negotiated TLS group header." + print(f"\n[google-auth Transport PQC] Negotiated TLS Group: {negotiated_group}") + print(f"[google-auth Transport PQC] Client Advertised Supported Groups: {supported_groups}") + + assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \ + f"Failed: google-auth Transport is NOT PQC-compliant! Negotiated: {negotiated_group}" 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..5486f47967f1 --- /dev/null +++ b/packages/google-auth/tests/crypt/test_pqc.py @@ -0,0 +1,44 @@ +# 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 +# +# https://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. + +import cryptography +import pytest +from packaging.version import Version + + +def test_cryptography_pqc_mlkem_support(): + """Verifies that the cryptography package installed in google-auth supports Post-Quantum Cryptography (ML-KEM / FIPS 203).""" + # Attempt to import ML-KEM (FIPS 203) introduced in cryptography 44.0.0 + try: + from cryptography.hazmat.primitives.asymmetric import mlkem + except ImportError: + pytest.fail( + f"PQC Compliance Failure: cryptography version {cryptography.__version__} does not support ML-KEM (FIPS 203). " + f"google-auth requires cryptography >= 44.0.0 for PQC support." + ) + + # 1. Generate ML-KEM-768 (FIPS 203) private key and derive public key + private_key = mlkem.MLKEM768PrivateKey.generate() + public_key = private_key.public_key() + assert isinstance(private_key, mlkem.MLKEM768PrivateKey) + assert isinstance(public_key, mlkem.MLKEM768PublicKey) + + # 2. Perform Key Encapsulation (sender generates shared secret + ciphertext) + shared_secret_sender, ciphertext = public_key.encapsulate() + assert len(ciphertext) > 0 + assert len(shared_secret_sender) > 0 + + # 3. Perform Key Decapsulation (receiver recovers shared secret from ciphertext) + shared_secret_receiver = private_key.decapsulate(ciphertext) + assert shared_secret_receiver == shared_secret_sender