From bdad0534e4c13800505c0eed5b46dc9cff03d45f Mon Sep 17 00:00:00 2001 From: ohmayr Date: Fri, 26 Jun 2026 20:00:30 +0000 Subject: [PATCH 1/5] feat: add post-quantum cryptography (PQC) integration tests and fixtures --- .../gapic-generator/tests/system/conftest.py | 44 +++++++++++++--- .../gapic-generator/tests/system/test_pqc.py | 52 +++++++++++++++++++ 2 files changed, 89 insertions(+), 7 deletions(-) create mode 100644 packages/gapic-generator/tests/system/test_pqc.py 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..127c2f9edc74 --- /dev/null +++ b/packages/gapic-generator/tests/system/test_pqc.py @@ -0,0 +1,52 @@ +# 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 +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 with the Showcase server.""" + client, interceptor = request.getfixturevalue(transport_fixture) + + # Make secure call using the standard 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 (this will fail if not using MLKEM/Kyber) + assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \ + f"Failed: {transport_fixture} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}" From 6c043e60edea5e5482b01299d6060304153e7410 Mon Sep 17 00:00:00 2001 From: ohmayr Date: Mon, 29 Jun 2026 20:09:11 +0000 Subject: [PATCH 2/5] update --- packages/gapic-generator/noxfile.py | 2 +- .../gapic-generator/tests/system/test_pqc.py | 45 +++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) 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/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py index 127c2f9edc74..0db865f5a39f 100644 --- a/packages/gapic-generator/tests/system/test_pqc.py +++ b/packages/gapic-generator/tests/system/test_pqc.py @@ -12,7 +12,9 @@ # See the License for the specific language governing permissions and # limitations under the License. +import os import pytest +import requests from google import showcase @pytest.fixture @@ -50,3 +52,46 @@ def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture): # Enforce PQC compliance (this will fail if not using MLKEM/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 the google-auth HTTP transport adapter negotiates PQC with the Showcase server.""" + import google.auth.transport.requests + from conftest import HostNameIgnoringAdapter + + # 1. Initialize a standard 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) + + # Bypass localhost hostname mismatch + session.mount("https://", HostNameIgnoringAdapter()) + + # 2. Wrap it in google-auth's Transport Request adapter + # This is the exact object google-auth uses to execute its own HTTP/REST requests. + auth_transport = google.auth.transport.requests.Request(session=session) + + # 3. Make secure call using the google-auth transport adapter + url = "https://localhost:7469/v1beta1/echo:echo" + method = "POST" + headers = {"Content-Type": "application/json"} + body = b'{"content": "Verify google-auth transport PQC connection."}' + + # Execute the request through google-auth's transport layer + response = auth_transport(url=url, method=method, body=body, headers=headers) + assert response.status == 200 + + # 4. Extract TLS group from response headers 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 PQC compliance + assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \ + f"Failed: google-auth Transport is NOT PQC-compliant! Negotiated: {negotiated_group}" From 449f2e6342fffff99328e2f80ff827f4200d9d73 Mon Sep 17 00:00:00 2001 From: ohmayr Date: Mon, 29 Jun 2026 20:12:29 +0000 Subject: [PATCH 3/5] debug: disable verification in google-auth test --- packages/gapic-generator/tests/system/test_pqc.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py index 0db865f5a39f..59f47a65d946 100644 --- a/packages/gapic-generator/tests/system/test_pqc.py +++ b/packages/gapic-generator/tests/system/test_pqc.py @@ -64,7 +64,8 @@ def test_google_auth_transport_pqc(run_pqc_test): 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 + # Disable verification temporarily to isolate hostname/cert validation errors + session.verify = False session.cert = (cert_path, key_path) # Bypass localhost hostname mismatch From f47628385730a8882293c29d308d782749237e98 Mon Sep 17 00:00:00 2001 From: ohmayr Date: Mon, 29 Jun 2026 20:13:57 +0000 Subject: [PATCH 4/5] feat: use MessageToJson for google-auth test payload --- packages/gapic-generator/tests/system/test_pqc.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py index 59f47a65d946..f5b5686bcb86 100644 --- a/packages/gapic-generator/tests/system/test_pqc.py +++ b/packages/gapic-generator/tests/system/test_pqc.py @@ -57,6 +57,7 @@ def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture): def test_google_auth_transport_pqc(run_pqc_test): """Verifies that the google-auth HTTP transport adapter negotiates PQC with the Showcase server.""" import google.auth.transport.requests + from google.protobuf.json_format import MessageToJson from conftest import HostNameIgnoringAdapter # 1. Initialize a standard requests Session with mTLS certs @@ -64,28 +65,29 @@ def test_google_auth_transport_pqc(run_pqc_test): 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" - # Disable verification temporarily to isolate hostname/cert validation errors - session.verify = False + session.verify = cert_path session.cert = (cert_path, key_path) # Bypass localhost hostname mismatch session.mount("https://", HostNameIgnoringAdapter()) # 2. Wrap it in google-auth's Transport Request adapter - # This is the exact object google-auth uses to execute its own HTTP/REST requests. auth_transport = google.auth.transport.requests.Request(session=session) - # 3. Make secure call using the google-auth transport adapter + # 3. Serialize the request body using the official protobuf JSON serializer + req = showcase.EchoRequest(content="Verify google-auth transport PQC connection.") + body = MessageToJson(req, including_default_value_fields=True).encode("utf-8") + + # 4. Make secure call using the google-auth transport adapter url = "https://localhost:7469/v1beta1/echo:echo" method = "POST" headers = {"Content-Type": "application/json"} - body = b'{"content": "Verify google-auth transport PQC connection."}' # Execute the request through google-auth's transport layer response = auth_transport(url=url, method=method, body=body, headers=headers) assert response.status == 200 - # 4. Extract TLS group from response headers returned by Showcase + # 5. Extract TLS group from response headers returned by Showcase negotiated_group = response.headers.get("x-showcase-tls-group") supported_groups = response.headers.get("x-showcase-tls-client-supported-groups") From 2d0fe09e354b3898463eb9a84554b6fe2408bde6 Mon Sep 17 00:00:00 2001 From: ohmayr Date: Tue, 7 Jul 2026 06:10:37 +0000 Subject: [PATCH 5/5] wip --- .../gapic-generator/tests/system/test_pqc.py | 48 +++++++++---------- packages/google-auth/tests/crypt/test_pqc.py | 44 +++++++++++++++++ 2 files changed, 67 insertions(+), 25 deletions(-) create mode 100644 packages/google-auth/tests/crypt/test_pqc.py diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py index f5b5686bcb86..6b6876272c65 100644 --- a/packages/gapic-generator/tests/system/test_pqc.py +++ b/packages/gapic-generator/tests/system/test_pqc.py @@ -12,28 +12,29 @@ # See the License for the specific language governing permissions and # limitations under the License. -import os 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 with the Showcase server.""" + """Verifies that the generated client library negotiates PQC (X25519MLKEM768) with Showcase server.""" client, interceptor = request.getfixturevalue(transport_fixture) - - # Make secure call using the standard client library 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 @@ -42,52 +43,50 @@ def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture): 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 (this will fail if not using MLKEM/Kyber) + + # 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 the google-auth HTTP transport adapter negotiates PQC with the Showcase server.""" + """Verifies that google-auth transport adapter negotiates PQC (X25519MLKEM768) with Showcase server.""" import google.auth.transport.requests - from google.protobuf.json_format import MessageToJson from conftest import HostNameIgnoringAdapter - # 1. Initialize a standard requests Session with mTLS certs + # 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) - - # Bypass localhost hostname mismatch session.mount("https://", HostNameIgnoringAdapter()) - # 2. Wrap it in google-auth's Transport Request adapter + # 2. Wrap session in google-auth transport adapter auth_transport = google.auth.transport.requests.Request(session=session) - # 3. Serialize the request body using the official protobuf JSON serializer + # 3. Serialize request body req = showcase.EchoRequest(content="Verify google-auth transport PQC connection.") - body = MessageToJson(req, including_default_value_fields=True).encode("utf-8") + body = showcase.EchoRequest.to_json(req, including_default_value_fields=False).encode("utf-8") - # 4. Make secure call using the google-auth transport adapter + # 4. Execute request through google-auth's transport layer url = "https://localhost:7469/v1beta1/echo:echo" - method = "POST" - headers = {"Content-Type": "application/json"} + headers = { + "Content-Type": "application/json", + "x-goog-api-client": "gapic/1.0 rest/1.0", + } - # Execute the request through google-auth's transport layer - response = auth_transport(url=url, method=method, body=body, headers=headers) - assert response.status == 200 + 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 TLS group from response headers returned by Showcase + # 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") @@ -95,6 +94,5 @@ def test_google_auth_transport_pqc(run_pqc_test): 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 PQC compliance 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