1313# limitations under the License.
1414
1515import pytest
16- import requests
1716from google import showcase
1817
1918
@@ -23,14 +22,17 @@ def run_pqc_test(use_mtls):
2322 pytest .skip ("PQC integration test requires mTLS (--mtls flag) to be enabled." )
2423
2524
26- @ pytest . mark . parametrize (
27- "transport_fixture" ,
28- [ "intercepted_echo_grpc" , "intercepted_echo_rest" ]
29- )
30- def test_pqc_negotiated_group ( run_pqc_test , request , transport_fixture ):
31- """Verifies that the generated client library negotiates PQC (X25519MLKEM768) with Showcase server."""
32- client , interceptor = request . getfixturevalue ( transport_fixture )
25+ def _require_pqc_support ( transport_name ):
26+ # TODO(Phase 3): Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
27+ if transport_name == "grpc" :
28+ import grpc
29+ from packaging . version import Version
30+ if Version ( grpc . __version__ ) < Version ( "1.83.0rc0" ):
31+ pytest . skip ( f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: { grpc . __version__ } )" )
3332
33+
34+ def _verify_pqc_negotiated_group (client , interceptor , transport_name ):
35+ _require_pqc_support (transport_name )
3436 # Make secure call using standard GAPIC client library fixture
3537 response = client .echo (request = showcase .EchoRequest (content = "Verify PQC connection." ))
3638 assert response .content == "Verify PQC connection."
@@ -47,52 +49,22 @@ def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture):
4749 assert negotiated_group is not None , "Failed: Showcase server did not return negotiated TLS group header."
4850 assert supported_groups is not None , "Failed: Showcase server did not return client advertised supported groups."
4951
50- print (f"\n [PQC Verification] ({ transport_fixture } ) Negotiated TLS Group: { negotiated_group } " )
51- print (f"[PQC Verification] ({ transport_fixture } ) Client Advertised Supported Groups: { supported_groups } " )
52+ print (f"\n [PQC Verification] ({ transport_name } ) Negotiated TLS Group: { negotiated_group } " )
53+ print (f"[PQC Verification] ({ transport_name } ) Client Advertised Supported Groups: { supported_groups } " )
5254
5355 # Enforce PQC compliance (X25519MLKEM768 or Kyber)
5456 assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group , \
55- f"Failed: { transport_fixture } Connection is NOT PQC-compliant! Negotiated: { negotiated_group } "
56-
57-
58- def test_google_auth_transport_pqc (run_pqc_test ):
59- """Verifies that google-auth transport adapter negotiates PQC (X25519MLKEM768) with Showcase server."""
60- import google .auth .transport .requests
61- from conftest import HostNameIgnoringAdapter
62-
63- # 1. Initialize requests Session with mTLS certs
64- session = requests .Session ()
65- cert_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.crt"
66- key_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.key"
67-
68- session .verify = cert_path
69- session .cert = (cert_path , key_path )
70- session .mount ("https://" , HostNameIgnoringAdapter ())
57+ f"Failed: { transport_name } Connection is NOT PQC-compliant! Negotiated: { negotiated_group } "
7158
72- # 2. Wrap session in google-auth transport adapter
73- auth_transport = google .auth .transport .requests .Request (session = session )
7459
75- # 3. Serialize request body
76- req = showcase .EchoRequest (content = "Verify google-auth transport PQC connection." )
77- body = showcase .EchoRequest .to_json (req , including_default_value_fields = False ).encode ("utf-8" )
60+ def test_pqc_grpc (run_pqc_test , intercepted_echo_grpc ):
61+ """Verifies that the gRPC client library negotiates PQC (X25519MLKEM768) with Showcase server."""
62+ client , interceptor = intercepted_echo_grpc
63+ _verify_pqc_negotiated_group (client , interceptor , "grpc" )
7864
79- # 4. Execute request through google-auth's transport layer
80- url = "https://localhost:7469/v1beta1/echo:echo"
81- headers = {
82- "Content-Type" : "application/json" ,
83- "x-goog-api-client" : "gapic/1.0 rest/1.0" ,
84- }
8565
86- response = auth_transport (url = url , method = "POST" , body = body , headers = headers )
87- assert response .status == 200 , f"Failed: status={ response .status } , body={ response .data } "
66+ def test_pqc_rest (run_pqc_test , intercepted_echo_rest ):
67+ """Verifies that the REST client library negotiates PQC (X25519MLKEM768) with Showcase server."""
68+ client , interceptor = intercepted_echo_rest
69+ _verify_pqc_negotiated_group (client , interceptor , "rest" )
8870
89- # 5. Extract and verify negotiated TLS group returned by Showcase
90- negotiated_group = response .headers .get ("x-showcase-tls-group" )
91- supported_groups = response .headers .get ("x-showcase-tls-client-supported-groups" )
92-
93- assert negotiated_group is not None , "Failed: Showcase server did not return negotiated TLS group header."
94- print (f"\n [google-auth Transport PQC] Negotiated TLS Group: { negotiated_group } " )
95- print (f"[google-auth Transport PQC] Client Advertised Supported Groups: { supported_groups } " )
96-
97- assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group , \
98- f"Failed: google-auth Transport is NOT PQC-compliant! Negotiated: { negotiated_group } "
0 commit comments