Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# -*- coding: utf-8 -*-
2# Copyright 2025 Google LLC
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16import json
17import logging as std_logging
18import pickle
19from typing import Callable, Dict, Optional, Sequence, Tuple, Union
20import warnings
22from google.api_core import gapic_v1, grpc_helpers
23import google.auth # type: ignore
24from google.auth import credentials as ga_credentials # type: ignore
25from google.auth.transport.grpc import SslCredentials # type: ignore
26from google.protobuf.json_format import MessageToJson
27import google.protobuf.message
28import grpc # type: ignore
29import proto # type: ignore
31from google.cloud.iam_credentials_v1.types import common
33from .base import DEFAULT_CLIENT_INFO, IAMCredentialsTransport
35try:
36 from google.api_core import client_logging # type: ignore
38 CLIENT_LOGGING_SUPPORTED = True # pragma: NO COVER
39except ImportError: # pragma: NO COVER
40 CLIENT_LOGGING_SUPPORTED = False
42_LOGGER = std_logging.getLogger(__name__)
45class _LoggingClientInterceptor(grpc.UnaryUnaryClientInterceptor): # pragma: NO COVER
46 def intercept_unary_unary(self, continuation, client_call_details, request):
47 logging_enabled = CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
48 std_logging.DEBUG
49 )
50 if logging_enabled: # pragma: NO COVER
51 request_metadata = client_call_details.metadata
52 if isinstance(request, proto.Message):
53 request_payload = type(request).to_json(request)
54 elif isinstance(request, google.protobuf.message.Message):
55 request_payload = MessageToJson(request)
56 else:
57 request_payload = f"{type(request).__name__}: {pickle.dumps(request)}"
59 request_metadata = {
60 key: value.decode("utf-8") if isinstance(value, bytes) else value
61 for key, value in request_metadata
62 }
63 grpc_request = {
64 "payload": request_payload,
65 "requestMethod": "grpc",
66 "metadata": dict(request_metadata),
67 }
68 _LOGGER.debug(
69 f"Sending request for {client_call_details.method}",
70 extra={
71 "serviceName": "google.iam.credentials.v1.IAMCredentials",
72 "rpcName": str(client_call_details.method),
73 "request": grpc_request,
74 "metadata": grpc_request["metadata"],
75 },
76 )
77 response = continuation(client_call_details, request)
78 if logging_enabled: # pragma: NO COVER
79 response_metadata = response.trailing_metadata()
80 # Convert gRPC metadata `<class 'grpc.aio._metadata.Metadata'>` to list of tuples
81 metadata = (
82 dict([(k, str(v)) for k, v in response_metadata])
83 if response_metadata
84 else None
85 )
86 result = response.result()
87 if isinstance(result, proto.Message):
88 response_payload = type(result).to_json(result)
89 elif isinstance(result, google.protobuf.message.Message):
90 response_payload = MessageToJson(result)
91 else:
92 response_payload = f"{type(result).__name__}: {pickle.dumps(result)}"
93 grpc_response = {
94 "payload": response_payload,
95 "metadata": metadata,
96 "status": "OK",
97 }
98 _LOGGER.debug(
99 f"Received response for {client_call_details.method}.",
100 extra={
101 "serviceName": "google.iam.credentials.v1.IAMCredentials",
102 "rpcName": client_call_details.method,
103 "response": grpc_response,
104 "metadata": grpc_response["metadata"],
105 },
106 )
107 return response
110class IAMCredentialsGrpcTransport(IAMCredentialsTransport):
111 """gRPC backend transport for IAMCredentials.
113 A service account is a special type of Google account that
114 belongs to your application or a virtual machine (VM), instead
115 of to an individual end user. Your application assumes the
116 identity of the service account to call Google APIs, so that the
117 users aren't directly involved.
119 Service account credentials are used to temporarily assume the
120 identity of the service account. Supported credential types
121 include OAuth 2.0 access tokens, OpenID Connect ID tokens,
122 self-signed JSON Web Tokens (JWTs), and more.
124 This class defines the same methods as the primary client, so the
125 primary client can load the underlying transport implementation
126 and call it.
128 It sends protocol buffers over the wire using gRPC (which is built on
129 top of HTTP/2); the ``grpcio`` package must be installed.
130 """
132 _stubs: Dict[str, Callable]
134 def __init__(
135 self,
136 *,
137 host: str = "iamcredentials.googleapis.com",
138 credentials: Optional[ga_credentials.Credentials] = None,
139 credentials_file: Optional[str] = None,
140 scopes: Optional[Sequence[str]] = None,
141 channel: Optional[Union[grpc.Channel, Callable[..., grpc.Channel]]] = None,
142 api_mtls_endpoint: Optional[str] = None,
143 client_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
144 ssl_channel_credentials: Optional[grpc.ChannelCredentials] = None,
145 client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
146 quota_project_id: Optional[str] = None,
147 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
148 always_use_jwt_access: Optional[bool] = False,
149 api_audience: Optional[str] = None,
150 ) -> None:
151 """Instantiate the transport.
153 Args:
154 host (Optional[str]):
155 The hostname to connect to (default: 'iamcredentials.googleapis.com').
156 credentials (Optional[google.auth.credentials.Credentials]): The
157 authorization credentials to attach to requests. These
158 credentials identify the application to the service; if none
159 are specified, the client will attempt to ascertain the
160 credentials from the environment.
161 This argument is ignored if a ``channel`` instance is provided.
162 credentials_file (Optional[str]): A file with credentials that can
163 be loaded with :func:`google.auth.load_credentials_from_file`.
164 This argument is ignored if a ``channel`` instance is provided.
165 scopes (Optional(Sequence[str])): A list of scopes. This argument is
166 ignored if a ``channel`` instance is provided.
167 channel (Optional[Union[grpc.Channel, Callable[..., grpc.Channel]]]):
168 A ``Channel`` instance through which to make calls, or a Callable
169 that constructs and returns one. If set to None, ``self.create_channel``
170 is used to create the channel. If a Callable is given, it will be called
171 with the same arguments as used in ``self.create_channel``.
172 api_mtls_endpoint (Optional[str]): Deprecated. The mutual TLS endpoint.
173 If provided, it overrides the ``host`` argument and tries to create
174 a mutual TLS channel with client SSL credentials from
175 ``client_cert_source`` or application default SSL credentials.
176 client_cert_source (Optional[Callable[[], Tuple[bytes, bytes]]]):
177 Deprecated. A callback to provide client SSL certificate bytes and
178 private key bytes, both in PEM format. It is ignored if
179 ``api_mtls_endpoint`` is None.
180 ssl_channel_credentials (grpc.ChannelCredentials): SSL credentials
181 for the grpc channel. It is ignored if a ``channel`` instance is provided.
182 client_cert_source_for_mtls (Optional[Callable[[], Tuple[bytes, bytes]]]):
183 A callback to provide client certificate bytes and private key bytes,
184 both in PEM format. It is used to configure a mutual TLS channel. It is
185 ignored if a ``channel`` instance or ``ssl_channel_credentials`` is provided.
186 quota_project_id (Optional[str]): An optional project to use for billing
187 and quota.
188 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
189 The client info used to send a user-agent string along with
190 API requests. If ``None``, then default info will be used.
191 Generally, you only need to set this if you're developing
192 your own client library.
193 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
194 be used for service account credentials.
196 Raises:
197 google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
198 creation failed for any reason.
199 google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
200 and ``credentials_file`` are passed.
201 """
202 self._grpc_channel = None
203 self._ssl_channel_credentials = ssl_channel_credentials
204 self._stubs: Dict[str, Callable] = {}
206 if api_mtls_endpoint:
207 warnings.warn("api_mtls_endpoint is deprecated", DeprecationWarning)
208 if client_cert_source:
209 warnings.warn("client_cert_source is deprecated", DeprecationWarning)
211 if isinstance(channel, grpc.Channel):
212 # Ignore credentials if a channel was passed.
213 credentials = None
214 self._ignore_credentials = True
215 # If a channel was explicitly provided, set it.
216 self._grpc_channel = channel
217 self._ssl_channel_credentials = None
219 else:
220 if api_mtls_endpoint:
221 host = api_mtls_endpoint
223 # Create SSL credentials with client_cert_source or application
224 # default SSL credentials.
225 if client_cert_source:
226 cert, key = client_cert_source()
227 self._ssl_channel_credentials = grpc.ssl_channel_credentials(
228 certificate_chain=cert, private_key=key
229 )
230 else:
231 self._ssl_channel_credentials = SslCredentials().ssl_credentials
233 else:
234 if client_cert_source_for_mtls and not ssl_channel_credentials:
235 cert, key = client_cert_source_for_mtls()
236 self._ssl_channel_credentials = grpc.ssl_channel_credentials(
237 certificate_chain=cert, private_key=key
238 )
240 # The base transport sets the host, credentials and scopes
241 super().__init__(
242 host=host,
243 credentials=credentials,
244 credentials_file=credentials_file,
245 scopes=scopes,
246 quota_project_id=quota_project_id,
247 client_info=client_info,
248 always_use_jwt_access=always_use_jwt_access,
249 api_audience=api_audience,
250 )
252 if not self._grpc_channel:
253 # initialize with the provided callable or the default channel
254 channel_init = channel or type(self).create_channel
255 self._grpc_channel = channel_init(
256 self._host,
257 # use the credentials which are saved
258 credentials=self._credentials,
259 # Set ``credentials_file`` to ``None`` here as
260 # the credentials that we saved earlier should be used.
261 credentials_file=None,
262 scopes=self._scopes,
263 ssl_credentials=self._ssl_channel_credentials,
264 quota_project_id=quota_project_id,
265 options=[
266 ("grpc.max_send_message_length", -1),
267 ("grpc.max_receive_message_length", -1),
268 ],
269 )
271 self._interceptor = _LoggingClientInterceptor()
272 self._logged_channel = grpc.intercept_channel(
273 self._grpc_channel, self._interceptor
274 )
276 # Wrap messages. This must be done after self._logged_channel exists
277 self._prep_wrapped_messages(client_info)
279 @classmethod
280 def create_channel(
281 cls,
282 host: str = "iamcredentials.googleapis.com",
283 credentials: Optional[ga_credentials.Credentials] = None,
284 credentials_file: Optional[str] = None,
285 scopes: Optional[Sequence[str]] = None,
286 quota_project_id: Optional[str] = None,
287 **kwargs,
288 ) -> grpc.Channel:
289 """Create and return a gRPC channel object.
290 Args:
291 host (Optional[str]): The host for the channel to use.
292 credentials (Optional[~.Credentials]): The
293 authorization credentials to attach to requests. These
294 credentials identify this application to the service. If
295 none are specified, the client will attempt to ascertain
296 the credentials from the environment.
297 credentials_file (Optional[str]): A file with credentials that can
298 be loaded with :func:`google.auth.load_credentials_from_file`.
299 This argument is mutually exclusive with credentials.
300 scopes (Optional[Sequence[str]]): A optional list of scopes needed for this
301 service. These are only used when credentials are not specified and
302 are passed to :func:`google.auth.default`.
303 quota_project_id (Optional[str]): An optional project to use for billing
304 and quota.
305 kwargs (Optional[dict]): Keyword arguments, which are passed to the
306 channel creation.
307 Returns:
308 grpc.Channel: A gRPC channel object.
310 Raises:
311 google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
312 and ``credentials_file`` are passed.
313 """
315 return grpc_helpers.create_channel(
316 host,
317 credentials=credentials,
318 credentials_file=credentials_file,
319 quota_project_id=quota_project_id,
320 default_scopes=cls.AUTH_SCOPES,
321 scopes=scopes,
322 default_host=cls.DEFAULT_HOST,
323 **kwargs,
324 )
326 @property
327 def grpc_channel(self) -> grpc.Channel:
328 """Return the channel designed to connect to this service."""
329 return self._grpc_channel
331 @property
332 def generate_access_token(
333 self,
334 ) -> Callable[
335 [common.GenerateAccessTokenRequest], common.GenerateAccessTokenResponse
336 ]:
337 r"""Return a callable for the generate access token method over gRPC.
339 Generates an OAuth 2.0 access token for a service
340 account.
342 Returns:
343 Callable[[~.GenerateAccessTokenRequest],
344 ~.GenerateAccessTokenResponse]:
345 A function that, when called, will call the underlying RPC
346 on the server.
347 """
348 # Generate a "stub function" on-the-fly which will actually make
349 # the request.
350 # gRPC handles serialization and deserialization, so we just need
351 # to pass in the functions for each.
352 if "generate_access_token" not in self._stubs:
353 self._stubs["generate_access_token"] = self._logged_channel.unary_unary(
354 "/google.iam.credentials.v1.IAMCredentials/GenerateAccessToken",
355 request_serializer=common.GenerateAccessTokenRequest.serialize,
356 response_deserializer=common.GenerateAccessTokenResponse.deserialize,
357 )
358 return self._stubs["generate_access_token"]
360 @property
361 def generate_id_token(
362 self,
363 ) -> Callable[[common.GenerateIdTokenRequest], common.GenerateIdTokenResponse]:
364 r"""Return a callable for the generate id token method over gRPC.
366 Generates an OpenID Connect ID token for a service
367 account.
369 Returns:
370 Callable[[~.GenerateIdTokenRequest],
371 ~.GenerateIdTokenResponse]:
372 A function that, when called, will call the underlying RPC
373 on the server.
374 """
375 # Generate a "stub function" on-the-fly which will actually make
376 # the request.
377 # gRPC handles serialization and deserialization, so we just need
378 # to pass in the functions for each.
379 if "generate_id_token" not in self._stubs:
380 self._stubs["generate_id_token"] = self._logged_channel.unary_unary(
381 "/google.iam.credentials.v1.IAMCredentials/GenerateIdToken",
382 request_serializer=common.GenerateIdTokenRequest.serialize,
383 response_deserializer=common.GenerateIdTokenResponse.deserialize,
384 )
385 return self._stubs["generate_id_token"]
387 @property
388 def sign_blob(self) -> Callable[[common.SignBlobRequest], common.SignBlobResponse]:
389 r"""Return a callable for the sign blob method over gRPC.
391 Signs a blob using a service account's system-managed
392 private key.
394 Returns:
395 Callable[[~.SignBlobRequest],
396 ~.SignBlobResponse]:
397 A function that, when called, will call the underlying RPC
398 on the server.
399 """
400 # Generate a "stub function" on-the-fly which will actually make
401 # the request.
402 # gRPC handles serialization and deserialization, so we just need
403 # to pass in the functions for each.
404 if "sign_blob" not in self._stubs:
405 self._stubs["sign_blob"] = self._logged_channel.unary_unary(
406 "/google.iam.credentials.v1.IAMCredentials/SignBlob",
407 request_serializer=common.SignBlobRequest.serialize,
408 response_deserializer=common.SignBlobResponse.deserialize,
409 )
410 return self._stubs["sign_blob"]
412 @property
413 def sign_jwt(self) -> Callable[[common.SignJwtRequest], common.SignJwtResponse]:
414 r"""Return a callable for the sign jwt method over gRPC.
416 Signs a JWT using a service account's system-managed
417 private key.
419 Returns:
420 Callable[[~.SignJwtRequest],
421 ~.SignJwtResponse]:
422 A function that, when called, will call the underlying RPC
423 on the server.
424 """
425 # Generate a "stub function" on-the-fly which will actually make
426 # the request.
427 # gRPC handles serialization and deserialization, so we just need
428 # to pass in the functions for each.
429 if "sign_jwt" not in self._stubs:
430 self._stubs["sign_jwt"] = self._logged_channel.unary_unary(
431 "/google.iam.credentials.v1.IAMCredentials/SignJwt",
432 request_serializer=common.SignJwtRequest.serialize,
433 response_deserializer=common.SignJwtResponse.deserialize,
434 )
435 return self._stubs["sign_jwt"]
437 def close(self):
438 self._logged_channel.close()
440 @property
441 def kind(self) -> str:
442 return "grpc"
445__all__ = ("IAMCredentialsGrpcTransport",)