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]): Deprecated. 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 This argument will be removed in the next major version of this library.
166 scopes (Optional(Sequence[str])): A list of scopes. This argument is
167 ignored if a ``channel`` instance is provided.
168 channel (Optional[Union[grpc.Channel, Callable[..., grpc.Channel]]]):
169 A ``Channel`` instance through which to make calls, or a Callable
170 that constructs and returns one. If set to None, ``self.create_channel``
171 is used to create the channel. If a Callable is given, it will be called
172 with the same arguments as used in ``self.create_channel``.
173 api_mtls_endpoint (Optional[str]): Deprecated. The mutual TLS endpoint.
174 If provided, it overrides the ``host`` argument and tries to create
175 a mutual TLS channel with client SSL credentials from
176 ``client_cert_source`` or application default SSL credentials.
177 client_cert_source (Optional[Callable[[], Tuple[bytes, bytes]]]):
178 Deprecated. A callback to provide client SSL certificate bytes and
179 private key bytes, both in PEM format. It is ignored if
180 ``api_mtls_endpoint`` is None.
181 ssl_channel_credentials (grpc.ChannelCredentials): SSL credentials
182 for the grpc channel. It is ignored if a ``channel`` instance is provided.
183 client_cert_source_for_mtls (Optional[Callable[[], Tuple[bytes, bytes]]]):
184 A callback to provide client certificate bytes and private key bytes,
185 both in PEM format. It is used to configure a mutual TLS channel. It is
186 ignored if a ``channel`` instance or ``ssl_channel_credentials`` is provided.
187 quota_project_id (Optional[str]): An optional project to use for billing
188 and quota.
189 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
190 The client info used to send a user-agent string along with
191 API requests. If ``None``, then default info will be used.
192 Generally, you only need to set this if you're developing
193 your own client library.
194 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
195 be used for service account credentials.
197 Raises:
198 google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
199 creation failed for any reason.
200 google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
201 and ``credentials_file`` are passed.
202 """
203 self._grpc_channel = None
204 self._ssl_channel_credentials = ssl_channel_credentials
205 self._stubs: Dict[str, Callable] = {}
207 if api_mtls_endpoint:
208 warnings.warn("api_mtls_endpoint is deprecated", DeprecationWarning)
209 if client_cert_source:
210 warnings.warn("client_cert_source is deprecated", DeprecationWarning)
212 if isinstance(channel, grpc.Channel):
213 # Ignore credentials if a channel was passed.
214 credentials = None
215 self._ignore_credentials = True
216 # If a channel was explicitly provided, set it.
217 self._grpc_channel = channel
218 self._ssl_channel_credentials = None
220 else:
221 if api_mtls_endpoint:
222 host = api_mtls_endpoint
224 # Create SSL credentials with client_cert_source or application
225 # default SSL credentials.
226 if client_cert_source:
227 cert, key = client_cert_source()
228 self._ssl_channel_credentials = grpc.ssl_channel_credentials(
229 certificate_chain=cert, private_key=key
230 )
231 else:
232 self._ssl_channel_credentials = SslCredentials().ssl_credentials
234 else:
235 if client_cert_source_for_mtls and not ssl_channel_credentials:
236 cert, key = client_cert_source_for_mtls()
237 self._ssl_channel_credentials = grpc.ssl_channel_credentials(
238 certificate_chain=cert, private_key=key
239 )
241 # The base transport sets the host, credentials and scopes
242 super().__init__(
243 host=host,
244 credentials=credentials,
245 credentials_file=credentials_file,
246 scopes=scopes,
247 quota_project_id=quota_project_id,
248 client_info=client_info,
249 always_use_jwt_access=always_use_jwt_access,
250 api_audience=api_audience,
251 )
253 if not self._grpc_channel:
254 # initialize with the provided callable or the default channel
255 channel_init = channel or type(self).create_channel
256 self._grpc_channel = channel_init(
257 self._host,
258 # use the credentials which are saved
259 credentials=self._credentials,
260 # Set ``credentials_file`` to ``None`` here as
261 # the credentials that we saved earlier should be used.
262 credentials_file=None,
263 scopes=self._scopes,
264 ssl_credentials=self._ssl_channel_credentials,
265 quota_project_id=quota_project_id,
266 options=[
267 ("grpc.max_send_message_length", -1),
268 ("grpc.max_receive_message_length", -1),
269 ],
270 )
272 self._interceptor = _LoggingClientInterceptor()
273 self._logged_channel = grpc.intercept_channel(
274 self._grpc_channel, self._interceptor
275 )
277 # Wrap messages. This must be done after self._logged_channel exists
278 self._prep_wrapped_messages(client_info)
280 @classmethod
281 def create_channel(
282 cls,
283 host: str = "iamcredentials.googleapis.com",
284 credentials: Optional[ga_credentials.Credentials] = None,
285 credentials_file: Optional[str] = None,
286 scopes: Optional[Sequence[str]] = None,
287 quota_project_id: Optional[str] = None,
288 **kwargs,
289 ) -> grpc.Channel:
290 """Create and return a gRPC channel object.
291 Args:
292 host (Optional[str]): The host for the channel to use.
293 credentials (Optional[~.Credentials]): The
294 authorization credentials to attach to requests. These
295 credentials identify this application to the service. If
296 none are specified, the client will attempt to ascertain
297 the credentials from the environment.
298 credentials_file (Optional[str]): Deprecated. A file with credentials that can
299 be loaded with :func:`google.auth.load_credentials_from_file`.
300 This argument is mutually exclusive with credentials. This argument will be
301 removed in the next major version of this library.
302 scopes (Optional[Sequence[str]]): A optional list of scopes needed for this
303 service. These are only used when credentials are not specified and
304 are passed to :func:`google.auth.default`.
305 quota_project_id (Optional[str]): An optional project to use for billing
306 and quota.
307 kwargs (Optional[dict]): Keyword arguments, which are passed to the
308 channel creation.
309 Returns:
310 grpc.Channel: A gRPC channel object.
312 Raises:
313 google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
314 and ``credentials_file`` are passed.
315 """
317 return grpc_helpers.create_channel(
318 host,
319 credentials=credentials,
320 credentials_file=credentials_file,
321 quota_project_id=quota_project_id,
322 default_scopes=cls.AUTH_SCOPES,
323 scopes=scopes,
324 default_host=cls.DEFAULT_HOST,
325 **kwargs,
326 )
328 @property
329 def grpc_channel(self) -> grpc.Channel:
330 """Return the channel designed to connect to this service."""
331 return self._grpc_channel
333 @property
334 def generate_access_token(
335 self,
336 ) -> Callable[
337 [common.GenerateAccessTokenRequest], common.GenerateAccessTokenResponse
338 ]:
339 r"""Return a callable for the generate access token method over gRPC.
341 Generates an OAuth 2.0 access token for a service
342 account.
344 Returns:
345 Callable[[~.GenerateAccessTokenRequest],
346 ~.GenerateAccessTokenResponse]:
347 A function that, when called, will call the underlying RPC
348 on the server.
349 """
350 # Generate a "stub function" on-the-fly which will actually make
351 # the request.
352 # gRPC handles serialization and deserialization, so we just need
353 # to pass in the functions for each.
354 if "generate_access_token" not in self._stubs:
355 self._stubs["generate_access_token"] = self._logged_channel.unary_unary(
356 "/google.iam.credentials.v1.IAMCredentials/GenerateAccessToken",
357 request_serializer=common.GenerateAccessTokenRequest.serialize,
358 response_deserializer=common.GenerateAccessTokenResponse.deserialize,
359 )
360 return self._stubs["generate_access_token"]
362 @property
363 def generate_id_token(
364 self,
365 ) -> Callable[[common.GenerateIdTokenRequest], common.GenerateIdTokenResponse]:
366 r"""Return a callable for the generate id token method over gRPC.
368 Generates an OpenID Connect ID token for a service
369 account.
371 Returns:
372 Callable[[~.GenerateIdTokenRequest],
373 ~.GenerateIdTokenResponse]:
374 A function that, when called, will call the underlying RPC
375 on the server.
376 """
377 # Generate a "stub function" on-the-fly which will actually make
378 # the request.
379 # gRPC handles serialization and deserialization, so we just need
380 # to pass in the functions for each.
381 if "generate_id_token" not in self._stubs:
382 self._stubs["generate_id_token"] = self._logged_channel.unary_unary(
383 "/google.iam.credentials.v1.IAMCredentials/GenerateIdToken",
384 request_serializer=common.GenerateIdTokenRequest.serialize,
385 response_deserializer=common.GenerateIdTokenResponse.deserialize,
386 )
387 return self._stubs["generate_id_token"]
389 @property
390 def sign_blob(self) -> Callable[[common.SignBlobRequest], common.SignBlobResponse]:
391 r"""Return a callable for the sign blob method over gRPC.
393 Signs a blob using a service account's system-managed
394 private key.
396 Returns:
397 Callable[[~.SignBlobRequest],
398 ~.SignBlobResponse]:
399 A function that, when called, will call the underlying RPC
400 on the server.
401 """
402 # Generate a "stub function" on-the-fly which will actually make
403 # the request.
404 # gRPC handles serialization and deserialization, so we just need
405 # to pass in the functions for each.
406 if "sign_blob" not in self._stubs:
407 self._stubs["sign_blob"] = self._logged_channel.unary_unary(
408 "/google.iam.credentials.v1.IAMCredentials/SignBlob",
409 request_serializer=common.SignBlobRequest.serialize,
410 response_deserializer=common.SignBlobResponse.deserialize,
411 )
412 return self._stubs["sign_blob"]
414 @property
415 def sign_jwt(self) -> Callable[[common.SignJwtRequest], common.SignJwtResponse]:
416 r"""Return a callable for the sign jwt method over gRPC.
418 Signs a JWT using a service account's system-managed
419 private key.
421 Returns:
422 Callable[[~.SignJwtRequest],
423 ~.SignJwtResponse]:
424 A function that, when called, will call the underlying RPC
425 on the server.
426 """
427 # Generate a "stub function" on-the-fly which will actually make
428 # the request.
429 # gRPC handles serialization and deserialization, so we just need
430 # to pass in the functions for each.
431 if "sign_jwt" not in self._stubs:
432 self._stubs["sign_jwt"] = self._logged_channel.unary_unary(
433 "/google.iam.credentials.v1.IAMCredentials/SignJwt",
434 request_serializer=common.SignJwtRequest.serialize,
435 response_deserializer=common.SignJwtResponse.deserialize,
436 )
437 return self._stubs["sign_jwt"]
439 def close(self):
440 self._logged_channel.close()
442 @property
443 def kind(self) -> str:
444 return "grpc"
447__all__ = ("IAMCredentialsGrpcTransport",)