chore(generator): add post-quantum cryptography (PQC) integration tests

packages/gapic-generator/noxfile.py

@@ -476,6 +476,24 @@ def showcase_mtls(
         )
 
 
+# TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
+# Remove showcase_pqc once grpcio >= 1.83.0 is enforced by default.
+@nox.session(python=NEWEST_PYTHON)
+def showcase_pqc(
+    session,
+    templates="DEFAULT",
+    other_opts: typing.Iterable[str] = (),
+    env: typing.Optional[typing.Dict[str, str]] = {},
+):
+    """Run the Showcase PQC verification test suite against grpcio 1.83+ over standard TLS."""
+    with showcase_library(session, templates=templates, other_opts=other_opts):
+        session.install("pytest", "pytest-asyncio")
+        # TODO(https://github.com/googleapis/google-cloud-python/issues/17751):
+        # Update the version below to `1.83.0` once released, and remove `--pre`.
+        session.install("--pre", "--upgrade", "grpcio>=1.83.0rc0", "grpcio-status>=1.83.0rc0")
+        session.run("py.test", "--quiet", "--tls", *(session.posargs or ["tests/system/test_pqc.py"]), env=env)
+
+
 def run_showcase_unit_tests(session, fail_under=100, rest_async_io_enabled=False):
     session.install(
         "coverage",

packages/gapic-generator/tests/system/conftest.py

@@ -18,6 +18,7 @@
 import os
 import pytest
 import pytest_asyncio
+from requests.adapters import HTTPAdapter
 
 from typing import Sequence, Tuple
 
@@ -113,15 +114,18 @@ def async_identity(use_mtls, request, event_loop):
         )
 
 
-dir = os.path.dirname(__file__)
-with open(os.path.join(dir, "../cert/mtls.crt"), "rb") as fh:
+base_dir = os.path.dirname(__file__)
+CERT_PATH = os.path.join(base_dir, "../cert/mtls.crt")
+KEY_PATH = os.path.join(base_dir, "../cert/mtls.key")
+with open(CERT_PATH, "rb") as fh:
     cert = fh.read()
-with open(os.path.join(dir, "../cert/mtls.key"), "rb") as fh:
+with open(KEY_PATH, "rb") as fh:
     key = fh.read()
 
 ssl_credentials = grpc.ssl_channel_credentials(
     root_certificates=cert, certificate_chain=cert, private_key=key
 )
+tls_credentials = grpc.ssl_channel_credentials(root_certificates=cert)
 
 
 def callback():
@@ -136,6 +140,9 @@ def pytest_addoption(parser):
     parser.addoption(
         "--mtls", action="store_true", help="Run system test with mutual TLS channel"
     )
+    parser.addoption(
+        "--tls", action="store_true", help="Run system test with standard one-way TLS channel"
+    )
 
 
 # TODO: Need to test  without passing in a transport class
@@ -189,6 +196,11 @@ def use_mtls(request):
     return request.config.getoption("--mtls")
 
 
+@pytest.fixture
+def use_tls(request):
+    return request.config.getoption("--tls")
+
+
 @pytest.fixture
 def parametrized_echo(
     use_mtls,
@@ -328,7 +340,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
 
@@ -387,7 +399,7 @@ async def _add_request_metadata(self, client_call_details):
     async def intercept_unary_unary(self, continuation, client_call_details, request):
         await self._add_request_metadata(client_call_details)
         response = await continuation(client_call_details, request)
-        metadata = [(k, str(v)) for k, v in await response.trailing_metadata()]
+        metadata = [(k, str(v)) for k, v in await response.initial_metadata()] + [(k, str(v)) for k, v in await response.trailing_metadata()]
         self.response_metadata = metadata
         return response
 
@@ -412,7 +424,7 @@ async def intercept_stream_stream(
 
 
 @pytest.fixture
-def intercepted_echo_grpc(use_mtls):
+def intercepted_echo_grpc(use_tls):
     # The interceptor adds 'showcase-trailer' client metadata. Showcase server
     # echoes any metadata with key 'showcase-trailer', so the same metadata
     # should appear as trailing metadata in the response.
@@ -422,8 +434,8 @@ def intercepted_echo_grpc(use_mtls):
     )
     host = "localhost:7469"
     channel = (
-        grpc.secure_channel(host, ssl_credentials)
-        if use_mtls
+        grpc.secure_channel(host, tls_credentials)
+        if use_tls
         else grpc.insecure_channel(host)
     )
     intercept_channel = grpc.intercept_channel(channel, interceptor)
@@ -435,7 +447,7 @@ def intercepted_echo_grpc(use_mtls):
 
 
 @pytest_asyncio.fixture
-async def intercepted_echo_grpc_async():
+async def intercepted_echo_grpc_async(use_tls):
     # The interceptor adds 'showcase-trailer' client metadata. Showcase server
     # echoes any metadata with key 'showcase-trailer', so the same metadata
     # should appear as trailing metadata in the response.
@@ -444,28 +456,42 @@ async def intercepted_echo_grpc_async():
         "intercepted",
     )
     host = "localhost:7469"
-    channel = grpc.aio.insecure_channel(host, interceptors=[interceptor])
-    # intercept_channel = grpc.aio.intercept_channel(channel, interceptor)
+    channel = (
+        grpc.aio.secure_channel(host, tls_credentials, interceptors=[interceptor])
+        if use_tls
+        else grpc.aio.insecure_channel(host, interceptors=[interceptor])
+    )
     transport = EchoAsyncClient.get_transport_class("grpc_asyncio")(
         credentials=ga_credentials.AnonymousCredentials(),
         channel=channel,
     )
     return EchoAsyncClient(transport=transport), interceptor
 
 
+class HostNameIgnoringAdapter(HTTPAdapter):
+    """Custom HTTPAdapter that disables hostname verification for local self-signed certs."""
+    def cert_verify(self, conn, url, verify, cert):
+        super().cert_verify(conn, url, verify, cert)
+        conn.assert_hostname = False
+
+
 @pytest.fixture
-def intercepted_echo_rest():
+def intercepted_echo_rest(use_tls):
     transport_name = "rest"
     transport_cls = EchoClient.get_transport_class(transport_name)
     interceptor = EchoMetadataClientRestInterceptor()
 
-    # The custom host explicitly bypasses https.
+    url_scheme = "https" if use_tls else "http"
     transport = transport_cls(
         credentials=ga_credentials.AnonymousCredentials(),
         host="localhost:7469",
-        url_scheme="http",
+        url_scheme=url_scheme,
         interceptor=interceptor,
     )
+    if use_tls:
+        transport._session.verify = CERT_PATH
+        transport._session.mount("https://", HostNameIgnoringAdapter())
+
     return EchoClient(transport=transport), interceptor
 
 
@@ -478,11 +504,11 @@ def intercepted_echo_rest_async():
     transport_cls = EchoAsyncClient.get_transport_class(transport_name)
     interceptor = EchoMetadataClientRestAsyncInterceptor()
 
-    # The custom host explicitly bypasses https.
     transport = transport_cls(
         credentials=async_anonymous_credentials(),
         host="localhost:7469",
         url_scheme="http",
         interceptor=interceptor,
     )
+
     return EchoAsyncClient(transport=transport), interceptor

packages/gapic-generator/tests/system/test_pqc.py

@@ -0,0 +1,83 @@
+# 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 grpc
+from packaging.version import Version
+import pytest
+from google import showcase
+
+
+@pytest.fixture(autouse=True)
+def require_tls(use_tls):
+    if not use_tls:
+        pytest.skip("PQC integration test requires standard TLS (--tls flag) to be enabled.")
+
+
+def _verify_pqc_metadata(interceptor, transport_name):
+    """Extracts and verifies negotiated PQC group and supported groups from interceptor metadata."""
+    response_metadata = getattr(interceptor, "response_metadata", []) or []
+    headers = {key.lower(): value for key, value in response_metadata}
+    negotiated_group = headers.get("x-showcase-tls-group")
+    supported_groups = headers.get("x-showcase-tls-client-supported-groups")
+
+    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."
+
+    # Enforce PQC compliance (X25519MLKEM768)
+    assert (
+        "MLKEM" in negotiated_group
+    ), f"Failed: {transport_name} Connection did not negotiate X25519MLKEM768! Negotiated: {negotiated_group}"
+
+
+def test_pqc_grpc(intercepted_echo_grpc):
+    """Verifies that the gRPC client library negotiates PQC (X25519MLKEM768) with Showcase server."""
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
+    # Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
+    if Version(grpc.__version__) < Version("1.83.0rc0"):
+        # TODO(https://github.com/googleapis/google-cloud-python/issues/17751): 
+        # Update the version in the check above to `1.83.0` once released.
+        pytest.skip(f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: {grpc.__version__})")
+
+    client, interceptor = intercepted_echo_grpc
+    response = client.echo(request=showcase.EchoRequest(content="Verify PQC connection."))
+    assert response.content == "Verify PQC connection."
+    _verify_pqc_metadata(interceptor, "grpc")
+
+
+def test_pqc_rest(intercepted_echo_rest):
+    """Verifies that the REST client library negotiates PQC (X25519MLKEM768) with Showcase server."""
+    client, interceptor = intercepted_echo_rest
+    response = client.echo(request=showcase.EchoRequest(content="Verify PQC connection."))
+    assert response.content == "Verify PQC connection."
+    _verify_pqc_metadata(interceptor, "rest")
+
+
+@pytest.mark.asyncio
+async def test_pqc_grpc_async(intercepted_echo_grpc_async):
+    """Verifies that the async gRPC client library negotiates PQC (X25519MLKEM768) with Showcase server."""
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
+    # Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
+    if Version(grpc.__version__) < Version("1.83.0rc0"):
+        # TODO(https://github.com/googleapis/google-cloud-python/issues/17751):
+        # Update the version in the check above to `1.83.0` once released.
+        pytest.skip(
+            f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: {grpc.__version__})"
+        )
+
+    client, interceptor = intercepted_echo_grpc_async
+    response = await client.echo(
+        request=showcase.EchoRequest(content="Verify PQC connection.")
+    )
+    assert response.content == "Verify PQC connection."
+    _verify_pqc_metadata(interceptor, "grpc_asyncio")