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 os
19import re
20import warnings
21from collections import OrderedDict
22from http import HTTPStatus
23from typing import (
24 Callable,
25 Dict,
26 Mapping,
27 MutableMapping,
28 MutableSequence,
29 Optional,
30 Sequence,
31 Tuple,
32 Type,
33 Union,
34 cast,
35)
37import google.protobuf
38from google.api_core import client_options as client_options_lib
39from google.api_core import exceptions as core_exceptions
40from google.api_core import gapic_v1
41from google.api_core import retry as retries
42from google.auth import credentials as ga_credentials # type: ignore
43from google.auth.exceptions import MutualTLSChannelError # type: ignore
44from google.auth.transport import mtls # type: ignore
45from google.auth.transport.grpc import SslCredentials # type: ignore
46from google.oauth2 import service_account # type: ignore
48from google.cloud.iam_credentials_v1 import gapic_version as package_version
50try:
51 OptionalRetry = Union[retries.Retry, gapic_v1.method._MethodDefault, None]
52except AttributeError: # pragma: NO COVER
53 OptionalRetry = Union[retries.Retry, object, None] # type: ignore
55try:
56 from google.api_core import client_logging # type: ignore
58 CLIENT_LOGGING_SUPPORTED = True # pragma: NO COVER
59except ImportError: # pragma: NO COVER
60 CLIENT_LOGGING_SUPPORTED = False
62_LOGGER = std_logging.getLogger(__name__)
64import google.protobuf.duration_pb2 as duration_pb2 # type: ignore
65import google.protobuf.timestamp_pb2 as timestamp_pb2 # type: ignore
67from google.cloud.iam_credentials_v1.types import common
69from .transports.base import DEFAULT_CLIENT_INFO, IAMCredentialsTransport
70from .transports.grpc import IAMCredentialsGrpcTransport
71from .transports.grpc_asyncio import IAMCredentialsGrpcAsyncIOTransport
72from .transports.rest import IAMCredentialsRestTransport
75class IAMCredentialsClientMeta(type):
76 """Metaclass for the IAMCredentials client.
78 This provides class-level methods for building and retrieving
79 support objects (e.g. transport) without polluting the client instance
80 objects.
81 """
83 _transport_registry = OrderedDict() # type: Dict[str, Type[IAMCredentialsTransport]]
84 _transport_registry["grpc"] = IAMCredentialsGrpcTransport
85 _transport_registry["grpc_asyncio"] = IAMCredentialsGrpcAsyncIOTransport
86 _transport_registry["rest"] = IAMCredentialsRestTransport
88 def get_transport_class(
89 cls,
90 label: Optional[str] = None,
91 ) -> Type[IAMCredentialsTransport]:
92 """Returns an appropriate transport class.
94 Args:
95 label: The name of the desired transport. If none is
96 provided, then the first transport in the registry is used.
98 Returns:
99 The transport class to use.
100 """
101 # If a specific transport is requested, return that one.
102 if label:
103 return cls._transport_registry[label]
105 # No transport is requested; return the default (that is, the first one
106 # in the dictionary).
107 return next(iter(cls._transport_registry.values()))
110class IAMCredentialsClient(metaclass=IAMCredentialsClientMeta):
111 """A service account is a special type of Google account that
112 belongs to your application or a virtual machine (VM), instead
113 of to an individual end user. Your application assumes the
114 identity of the service account to call Google APIs, so that the
115 users aren't directly involved.
117 Service account credentials are used to temporarily assume the
118 identity of the service account. Supported credential types
119 include OAuth 2.0 access tokens, OpenID Connect ID tokens,
120 self-signed JSON Web Tokens (JWTs), and more.
121 """
123 @staticmethod
124 def _get_default_mtls_endpoint(api_endpoint) -> Optional[str]:
125 """Converts api endpoint to mTLS endpoint.
127 Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
128 "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
129 Args:
130 api_endpoint (Optional[str]): the api endpoint to convert.
131 Returns:
132 Optional[str]: converted mTLS api endpoint.
133 """
134 if not api_endpoint:
135 return api_endpoint
137 mtls_endpoint_re = re.compile(
138 r"(?P<name>[^.]+)(?P<mtls>\.mtls)?(?P<sandbox>\.sandbox)?(?P<googledomain>\.googleapis\.com)?"
139 )
141 m = mtls_endpoint_re.match(api_endpoint)
142 if m is None:
143 # Could not parse api_endpoint; return as-is.
144 return api_endpoint
146 name, mtls, sandbox, googledomain = m.groups()
147 if mtls or not googledomain:
148 return api_endpoint
150 if sandbox:
151 return api_endpoint.replace(
152 "sandbox.googleapis.com", "mtls.sandbox.googleapis.com"
153 )
155 return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com")
157 # Note: DEFAULT_ENDPOINT is deprecated. Use _DEFAULT_ENDPOINT_TEMPLATE instead.
158 DEFAULT_ENDPOINT = "iamcredentials.googleapis.com"
159 DEFAULT_MTLS_ENDPOINT = _get_default_mtls_endpoint.__func__( # type: ignore
160 DEFAULT_ENDPOINT
161 )
163 _DEFAULT_ENDPOINT_TEMPLATE = "iamcredentials.{UNIVERSE_DOMAIN}"
164 _DEFAULT_UNIVERSE = "googleapis.com"
166 @staticmethod
167 def _use_client_cert_effective():
168 """Returns whether client certificate should be used for mTLS if the
169 google-auth version supports should_use_client_cert automatic mTLS enablement.
171 Alternatively, read from the GOOGLE_API_USE_CLIENT_CERTIFICATE env var.
173 Returns:
174 bool: whether client certificate should be used for mTLS
175 Raises:
176 ValueError: (If using a version of google-auth without should_use_client_cert and
177 GOOGLE_API_USE_CLIENT_CERTIFICATE is set to an unexpected value.)
178 """
179 # check if google-auth version supports should_use_client_cert for automatic mTLS enablement
180 if hasattr(mtls, "should_use_client_cert"): # pragma: NO COVER
181 return mtls.should_use_client_cert()
182 else: # pragma: NO COVER
183 # if unsupported, fallback to reading from env var
184 use_client_cert_str = os.getenv(
185 "GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"
186 ).lower()
187 if use_client_cert_str not in ("true", "false"):
188 raise ValueError(
189 "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be"
190 " either `true` or `false`"
191 )
192 return use_client_cert_str == "true"
194 @classmethod
195 def from_service_account_info(cls, info: dict, *args, **kwargs):
196 """Creates an instance of this client using the provided credentials
197 info.
199 Args:
200 info (dict): The service account private key info.
201 args: Additional arguments to pass to the constructor.
202 kwargs: Additional arguments to pass to the constructor.
204 Returns:
205 IAMCredentialsClient: The constructed client.
206 """
207 credentials = service_account.Credentials.from_service_account_info(info)
208 kwargs["credentials"] = credentials
209 return cls(*args, **kwargs)
211 @classmethod
212 def from_service_account_file(cls, filename: str, *args, **kwargs):
213 """Creates an instance of this client using the provided credentials
214 file.
216 Args:
217 filename (str): The path to the service account private key json
218 file.
219 args: Additional arguments to pass to the constructor.
220 kwargs: Additional arguments to pass to the constructor.
222 Returns:
223 IAMCredentialsClient: The constructed client.
224 """
225 credentials = service_account.Credentials.from_service_account_file(filename)
226 kwargs["credentials"] = credentials
227 return cls(*args, **kwargs)
229 from_service_account_json = from_service_account_file
231 @property
232 def transport(self) -> IAMCredentialsTransport:
233 """Returns the transport used by the client instance.
235 Returns:
236 IAMCredentialsTransport: The transport used by the client
237 instance.
238 """
239 return self._transport
241 @staticmethod
242 def service_account_path(
243 project: str,
244 service_account: str,
245 ) -> str:
246 """Returns a fully-qualified service_account string."""
247 return "projects/{project}/serviceAccounts/{service_account}".format(
248 project=project,
249 service_account=service_account,
250 )
252 @staticmethod
253 def parse_service_account_path(path: str) -> Dict[str, str]:
254 """Parses a service_account path into its component segments."""
255 m = re.match(
256 r"^projects/(?P<project>.+?)/serviceAccounts/(?P<service_account>.+?)$",
257 path,
258 )
259 return m.groupdict() if m else {}
261 @staticmethod
262 def common_billing_account_path(
263 billing_account: str,
264 ) -> str:
265 """Returns a fully-qualified billing_account string."""
266 return "billingAccounts/{billing_account}".format(
267 billing_account=billing_account,
268 )
270 @staticmethod
271 def parse_common_billing_account_path(path: str) -> Dict[str, str]:
272 """Parse a billing_account path into its component segments."""
273 m = re.match(r"^billingAccounts/(?P<billing_account>.+?)$", path)
274 return m.groupdict() if m else {}
276 @staticmethod
277 def common_folder_path(
278 folder: str,
279 ) -> str:
280 """Returns a fully-qualified folder string."""
281 return "folders/{folder}".format(
282 folder=folder,
283 )
285 @staticmethod
286 def parse_common_folder_path(path: str) -> Dict[str, str]:
287 """Parse a folder path into its component segments."""
288 m = re.match(r"^folders/(?P<folder>.+?)$", path)
289 return m.groupdict() if m else {}
291 @staticmethod
292 def common_organization_path(
293 organization: str,
294 ) -> str:
295 """Returns a fully-qualified organization string."""
296 return "organizations/{organization}".format(
297 organization=organization,
298 )
300 @staticmethod
301 def parse_common_organization_path(path: str) -> Dict[str, str]:
302 """Parse a organization path into its component segments."""
303 m = re.match(r"^organizations/(?P<organization>.+?)$", path)
304 return m.groupdict() if m else {}
306 @staticmethod
307 def common_project_path(
308 project: str,
309 ) -> str:
310 """Returns a fully-qualified project string."""
311 return "projects/{project}".format(
312 project=project,
313 )
315 @staticmethod
316 def parse_common_project_path(path: str) -> Dict[str, str]:
317 """Parse a project path into its component segments."""
318 m = re.match(r"^projects/(?P<project>.+?)$", path)
319 return m.groupdict() if m else {}
321 @staticmethod
322 def common_location_path(
323 project: str,
324 location: str,
325 ) -> str:
326 """Returns a fully-qualified location string."""
327 return "projects/{project}/locations/{location}".format(
328 project=project,
329 location=location,
330 )
332 @staticmethod
333 def parse_common_location_path(path: str) -> Dict[str, str]:
334 """Parse a location path into its component segments."""
335 m = re.match(r"^projects/(?P<project>.+?)/locations/(?P<location>.+?)$", path)
336 return m.groupdict() if m else {}
338 @classmethod
339 def get_mtls_endpoint_and_cert_source(
340 cls, client_options: Optional[client_options_lib.ClientOptions] = None
341 ):
342 """Deprecated. Return the API endpoint and client cert source for mutual TLS.
344 The client cert source is determined in the following order:
345 (1) if `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is not "true", the
346 client cert source is None.
347 (2) if `client_options.client_cert_source` is provided, use the provided one; if the
348 default client cert source exists, use the default one; otherwise the client cert
349 source is None.
351 The API endpoint is determined in the following order:
352 (1) if `client_options.api_endpoint` if provided, use the provided one.
353 (2) if `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is "always", use the
354 default mTLS endpoint; if the environment variable is "never", use the default API
355 endpoint; otherwise if client cert source exists, use the default mTLS endpoint, otherwise
356 use the default API endpoint.
358 More details can be found at https://google.aip.dev/auth/4114.
360 Args:
361 client_options (google.api_core.client_options.ClientOptions): Custom options for the
362 client. Only the `api_endpoint` and `client_cert_source` properties may be used
363 in this method.
365 Returns:
366 Tuple[str, Callable[[], Tuple[bytes, bytes]]]: returns the API endpoint and the
367 client cert source to use.
369 Raises:
370 google.auth.exceptions.MutualTLSChannelError: If any errors happen.
371 """
373 warnings.warn(
374 "get_mtls_endpoint_and_cert_source is deprecated. Use the api_endpoint property instead.",
375 DeprecationWarning,
376 )
377 if client_options is None:
378 client_options = client_options_lib.ClientOptions()
379 use_client_cert = IAMCredentialsClient._use_client_cert_effective()
380 use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto")
381 if use_mtls_endpoint not in ("auto", "never", "always"):
382 raise MutualTLSChannelError(
383 "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`, `auto` or `always`"
384 )
386 # Figure out the client cert source to use.
387 client_cert_source = None
388 if use_client_cert:
389 if client_options.client_cert_source:
390 client_cert_source = client_options.client_cert_source
391 elif mtls.has_default_client_cert_source():
392 client_cert_source = mtls.default_client_cert_source()
394 # Figure out which api endpoint to use.
395 if client_options.api_endpoint is not None:
396 api_endpoint = client_options.api_endpoint
397 elif use_mtls_endpoint == "always" or (
398 use_mtls_endpoint == "auto" and client_cert_source
399 ):
400 api_endpoint = cls.DEFAULT_MTLS_ENDPOINT
401 else:
402 api_endpoint = cls.DEFAULT_ENDPOINT
404 return api_endpoint, client_cert_source
406 @staticmethod
407 def _read_environment_variables():
408 """Returns the environment variables used by the client.
410 Returns:
411 Tuple[bool, str, str]: returns the GOOGLE_API_USE_CLIENT_CERTIFICATE,
412 GOOGLE_API_USE_MTLS_ENDPOINT, and GOOGLE_CLOUD_UNIVERSE_DOMAIN environment variables.
414 Raises:
415 ValueError: If GOOGLE_API_USE_CLIENT_CERTIFICATE is not
416 any of ["true", "false"].
417 google.auth.exceptions.MutualTLSChannelError: If GOOGLE_API_USE_MTLS_ENDPOINT
418 is not any of ["auto", "never", "always"].
419 """
420 use_client_cert = IAMCredentialsClient._use_client_cert_effective()
421 use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower()
422 universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN")
423 if use_mtls_endpoint not in ("auto", "never", "always"):
424 raise MutualTLSChannelError(
425 "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`, `auto` or `always`"
426 )
427 return use_client_cert, use_mtls_endpoint, universe_domain_env
429 @staticmethod
430 def _get_client_cert_source(provided_cert_source, use_cert_flag):
431 """Return the client cert source to be used by the client.
433 Args:
434 provided_cert_source (bytes): The client certificate source provided.
435 use_cert_flag (bool): A flag indicating whether to use the client certificate.
437 Returns:
438 bytes or None: The client cert source to be used by the client.
439 """
440 client_cert_source = None
441 if use_cert_flag:
442 if provided_cert_source:
443 client_cert_source = provided_cert_source
444 elif mtls.has_default_client_cert_source():
445 client_cert_source = mtls.default_client_cert_source()
446 return client_cert_source
448 @staticmethod
449 def _get_api_endpoint(
450 api_override, client_cert_source, universe_domain, use_mtls_endpoint
451 ) -> str:
452 """Return the API endpoint used by the client.
454 Args:
455 api_override (str): The API endpoint override. If specified, this is always
456 the return value of this function and the other arguments are not used.
457 client_cert_source (bytes): The client certificate source used by the client.
458 universe_domain (str): The universe domain used by the client.
459 use_mtls_endpoint (str): How to use the mTLS endpoint, which depends also on the other parameters.
460 Possible values are "always", "auto", or "never".
462 Returns:
463 str: The API endpoint to be used by the client.
464 """
465 if api_override is not None:
466 api_endpoint = api_override
467 elif use_mtls_endpoint == "always" or (
468 use_mtls_endpoint == "auto" and client_cert_source
469 ):
470 _default_universe = IAMCredentialsClient._DEFAULT_UNIVERSE
471 if universe_domain != _default_universe:
472 raise MutualTLSChannelError(
473 f"mTLS is not supported in any universe other than {_default_universe}."
474 )
475 api_endpoint = IAMCredentialsClient.DEFAULT_MTLS_ENDPOINT
476 else:
477 api_endpoint = IAMCredentialsClient._DEFAULT_ENDPOINT_TEMPLATE.format(
478 UNIVERSE_DOMAIN=universe_domain
479 )
480 return api_endpoint
482 @staticmethod
483 def _get_universe_domain(
484 client_universe_domain: Optional[str], universe_domain_env: Optional[str]
485 ) -> str:
486 """Return the universe domain used by the client.
488 Args:
489 client_universe_domain (Optional[str]): The universe domain configured via the client options.
490 universe_domain_env (Optional[str]): The universe domain configured via the "GOOGLE_CLOUD_UNIVERSE_DOMAIN" environment variable.
492 Returns:
493 str: The universe domain to be used by the client.
495 Raises:
496 ValueError: If the universe domain is an empty string.
497 """
498 universe_domain = IAMCredentialsClient._DEFAULT_UNIVERSE
499 if client_universe_domain is not None:
500 universe_domain = client_universe_domain
501 elif universe_domain_env is not None:
502 universe_domain = universe_domain_env
503 if len(universe_domain.strip()) == 0:
504 raise ValueError("Universe Domain cannot be an empty string.")
505 return universe_domain
507 def _validate_universe_domain(self):
508 """Validates client's and credentials' universe domains are consistent.
510 Returns:
511 bool: True iff the configured universe domain is valid.
513 Raises:
514 ValueError: If the configured universe domain is not valid.
515 """
517 # NOTE (b/349488459): universe validation is disabled until further notice.
518 return True
520 def _add_cred_info_for_auth_errors(
521 self, error: core_exceptions.GoogleAPICallError
522 ) -> None:
523 """Adds credential info string to error details for 401/403/404 errors.
525 Args:
526 error (google.api_core.exceptions.GoogleAPICallError): The error to add the cred info.
527 """
528 if error.code not in [
529 HTTPStatus.UNAUTHORIZED,
530 HTTPStatus.FORBIDDEN,
531 HTTPStatus.NOT_FOUND,
532 ]:
533 return
535 cred = self._transport._credentials
537 # get_cred_info is only available in google-auth>=2.35.0
538 if not hasattr(cred, "get_cred_info"):
539 return
541 # ignore the type check since pypy test fails when get_cred_info
542 # is not available
543 cred_info = cred.get_cred_info() # type: ignore
544 if cred_info and hasattr(error._details, "append"):
545 error._details.append(json.dumps(cred_info))
547 @property
548 def api_endpoint(self) -> str:
549 """Return the API endpoint used by the client instance.
551 Returns:
552 str: The API endpoint used by the client instance.
553 """
554 return self._api_endpoint
556 @property
557 def universe_domain(self) -> str:
558 """Return the universe domain used by the client instance.
560 Returns:
561 str: The universe domain used by the client instance.
562 """
563 return self._universe_domain
565 def __init__(
566 self,
567 *,
568 credentials: Optional[ga_credentials.Credentials] = None,
569 transport: Optional[
570 Union[str, IAMCredentialsTransport, Callable[..., IAMCredentialsTransport]]
571 ] = None,
572 client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
573 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
574 ) -> None:
575 """Instantiates the iam credentials client.
577 Args:
578 credentials (Optional[google.auth.credentials.Credentials]): The
579 authorization credentials to attach to requests. These
580 credentials identify the application to the service; if none
581 are specified, the client will attempt to ascertain the
582 credentials from the environment.
583 transport (Optional[Union[str,IAMCredentialsTransport,Callable[..., IAMCredentialsTransport]]]):
584 The transport to use, or a Callable that constructs and returns a new transport.
585 If a Callable is given, it will be called with the same set of initialization
586 arguments as used in the IAMCredentialsTransport constructor.
587 If set to None, a transport is chosen automatically.
588 client_options (Optional[Union[google.api_core.client_options.ClientOptions, dict]]):
589 Custom options for the client.
591 1. The ``api_endpoint`` property can be used to override the
592 default endpoint provided by the client when ``transport`` is
593 not explicitly provided. Only if this property is not set and
594 ``transport`` was not explicitly provided, the endpoint is
595 determined by the GOOGLE_API_USE_MTLS_ENDPOINT environment
596 variable, which have one of the following values:
597 "always" (always use the default mTLS endpoint), "never" (always
598 use the default regular endpoint) and "auto" (auto-switch to the
599 default mTLS endpoint if client certificate is present; this is
600 the default value).
602 2. If the GOOGLE_API_USE_CLIENT_CERTIFICATE environment variable
603 is "true", then the ``client_cert_source`` property can be used
604 to provide a client certificate for mTLS transport. If
605 not provided, the default SSL client certificate will be used if
606 present. If GOOGLE_API_USE_CLIENT_CERTIFICATE is "false" or not
607 set, no client certificate will be used.
609 3. The ``universe_domain`` property can be used to override the
610 default "googleapis.com" universe. Note that the ``api_endpoint``
611 property still takes precedence; and ``universe_domain`` is
612 currently not supported for mTLS.
614 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
615 The client info used to send a user-agent string along with
616 API requests. If ``None``, then default info will be used.
617 Generally, you only need to set this if you're developing
618 your own client library.
620 Raises:
621 google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
622 creation failed for any reason.
623 """
624 self._client_options = client_options
625 if isinstance(self._client_options, dict):
626 self._client_options = client_options_lib.from_dict(self._client_options)
627 if self._client_options is None:
628 self._client_options = client_options_lib.ClientOptions()
629 self._client_options = cast(
630 client_options_lib.ClientOptions, self._client_options
631 )
633 universe_domain_opt = getattr(self._client_options, "universe_domain", None)
635 self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = (
636 IAMCredentialsClient._read_environment_variables()
637 )
638 self._client_cert_source = IAMCredentialsClient._get_client_cert_source(
639 self._client_options.client_cert_source, self._use_client_cert
640 )
641 self._universe_domain = IAMCredentialsClient._get_universe_domain(
642 universe_domain_opt, self._universe_domain_env
643 )
644 self._api_endpoint: str = "" # updated below, depending on `transport`
646 # Initialize the universe domain validation.
647 self._is_universe_domain_valid = False
649 if CLIENT_LOGGING_SUPPORTED: # pragma: NO COVER
650 # Setup logging.
651 client_logging.initialize_logging()
653 api_key_value = getattr(self._client_options, "api_key", None)
654 if api_key_value and credentials:
655 raise ValueError(
656 "client_options.api_key and credentials are mutually exclusive"
657 )
659 # Save or instantiate the transport.
660 # Ordinarily, we provide the transport, but allowing a custom transport
661 # instance provides an extensibility point for unusual situations.
662 transport_provided = isinstance(transport, IAMCredentialsTransport)
663 if transport_provided:
664 # transport is a IAMCredentialsTransport instance.
665 if credentials or self._client_options.credentials_file or api_key_value:
666 raise ValueError(
667 "When providing a transport instance, "
668 "provide its credentials directly."
669 )
670 if self._client_options.scopes:
671 raise ValueError(
672 "When providing a transport instance, provide its scopes directly."
673 )
674 self._transport = cast(IAMCredentialsTransport, transport)
675 self._api_endpoint = self._transport.host
677 self._api_endpoint = (
678 self._api_endpoint
679 or IAMCredentialsClient._get_api_endpoint(
680 self._client_options.api_endpoint,
681 self._client_cert_source,
682 self._universe_domain,
683 self._use_mtls_endpoint,
684 )
685 )
687 if not transport_provided:
688 import google.auth._default # type: ignore
690 if api_key_value and hasattr(
691 google.auth._default, "get_api_key_credentials"
692 ):
693 credentials = google.auth._default.get_api_key_credentials(
694 api_key_value
695 )
697 transport_init: Union[
698 Type[IAMCredentialsTransport], Callable[..., IAMCredentialsTransport]
699 ] = (
700 IAMCredentialsClient.get_transport_class(transport)
701 if isinstance(transport, str) or transport is None
702 else cast(Callable[..., IAMCredentialsTransport], transport)
703 )
704 # initialize with the provided callable or the passed in class
705 self._transport = transport_init(
706 credentials=credentials,
707 credentials_file=self._client_options.credentials_file,
708 host=self._api_endpoint,
709 scopes=self._client_options.scopes,
710 client_cert_source_for_mtls=self._client_cert_source,
711 quota_project_id=self._client_options.quota_project_id,
712 client_info=client_info,
713 always_use_jwt_access=True,
714 api_audience=self._client_options.api_audience,
715 )
717 if "async" not in str(self._transport):
718 if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
719 std_logging.DEBUG
720 ): # pragma: NO COVER
721 _LOGGER.debug(
722 "Created client `google.iam.credentials_v1.IAMCredentialsClient`.",
723 extra={
724 "serviceName": "google.iam.credentials.v1.IAMCredentials",
725 "universeDomain": getattr(
726 self._transport._credentials, "universe_domain", ""
727 ),
728 "credentialsType": f"{type(self._transport._credentials).__module__}.{type(self._transport._credentials).__qualname__}",
729 "credentialsInfo": getattr(
730 self.transport._credentials, "get_cred_info", lambda: None
731 )(),
732 }
733 if hasattr(self._transport, "_credentials")
734 else {
735 "serviceName": "google.iam.credentials.v1.IAMCredentials",
736 "credentialsType": None,
737 },
738 )
740 def generate_access_token(
741 self,
742 request: Optional[Union[common.GenerateAccessTokenRequest, dict]] = None,
743 *,
744 name: Optional[str] = None,
745 delegates: Optional[MutableSequence[str]] = None,
746 scope: Optional[MutableSequence[str]] = None,
747 lifetime: Optional[duration_pb2.Duration] = None,
748 retry: OptionalRetry = gapic_v1.method.DEFAULT,
749 timeout: Union[float, object] = gapic_v1.method.DEFAULT,
750 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
751 ) -> common.GenerateAccessTokenResponse:
752 r"""Generates an OAuth 2.0 access token for a service
753 account.
755 .. code-block:: python
757 # This snippet has been automatically generated and should be regarded as a
758 # code template only.
759 # It will require modifications to work:
760 # - It may require correct/in-range values for request initialization.
761 # - It may require specifying regional endpoints when creating the service
762 # client as shown in:
763 # https://googleapis.dev/python/google-api-core/latest/client_options.html
764 from google.cloud import iam_credentials_v1
766 def sample_generate_access_token():
767 # Create a client
768 client = iam_credentials_v1.IAMCredentialsClient()
770 # Initialize request argument(s)
771 request = iam_credentials_v1.GenerateAccessTokenRequest(
772 name="name_value",
773 scope=['scope_value1', 'scope_value2'],
774 )
776 # Make the request
777 response = client.generate_access_token(request=request)
779 # Handle the response
780 print(response)
782 Args:
783 request (Union[google.cloud.iam_credentials_v1.types.GenerateAccessTokenRequest, dict]):
784 The request object.
785 name (str):
786 Required. The resource name of the service account for
787 which the credentials are requested, in the following
788 format:
789 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
790 The ``-`` wildcard character is required; replacing it
791 with a project ID is invalid.
793 This corresponds to the ``name`` field
794 on the ``request`` instance; if ``request`` is provided, this
795 should not be set.
796 delegates (MutableSequence[str]):
797 The sequence of service accounts in a delegation chain.
798 Each service account must be granted the
799 ``roles/iam.serviceAccountTokenCreator`` role on its
800 next service account in the chain. The last service
801 account in the chain must be granted the
802 ``roles/iam.serviceAccountTokenCreator`` role on the
803 service account that is specified in the ``name`` field
804 of the request.
806 The delegates must have the following format:
807 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
808 The ``-`` wildcard character is required; replacing it
809 with a project ID is invalid.
811 This corresponds to the ``delegates`` field
812 on the ``request`` instance; if ``request`` is provided, this
813 should not be set.
814 scope (MutableSequence[str]):
815 Required. Code to identify the scopes
816 to be included in the OAuth 2.0 access
817 token. See
818 https://developers.google.com/identity/protocols/googlescopes
819 for more information.
820 At least one value required.
822 This corresponds to the ``scope`` field
823 on the ``request`` instance; if ``request`` is provided, this
824 should not be set.
825 lifetime (google.protobuf.duration_pb2.Duration):
826 The desired lifetime duration of the
827 access token in seconds. Must be set to
828 a value less than or equal to 3600 (1
829 hour). If a value is not specified, the
830 token's lifetime will be set to a
831 default value of one hour.
833 This corresponds to the ``lifetime`` field
834 on the ``request`` instance; if ``request`` is provided, this
835 should not be set.
836 retry (google.api_core.retry.Retry): Designation of what errors, if any,
837 should be retried.
838 timeout (float): The timeout for this request.
839 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
840 sent along with the request as metadata. Normally, each value must be of type `str`,
841 but for metadata keys ending with the suffix `-bin`, the corresponding values must
842 be of type `bytes`.
844 Returns:
845 google.cloud.iam_credentials_v1.types.GenerateAccessTokenResponse:
847 """
848 # Create or coerce a protobuf request object.
849 # - Quick check: If we got a request object, we should *not* have
850 # gotten any keyword arguments that map to the request.
851 flattened_params = [name, delegates, scope, lifetime]
852 has_flattened_params = (
853 len([param for param in flattened_params if param is not None]) > 0
854 )
855 if request is not None and has_flattened_params:
856 raise ValueError(
857 "If the `request` argument is set, then none of "
858 "the individual field arguments should be set."
859 )
861 # - Use the request object if provided (there's no risk of modifying the input as
862 # there are no flattened fields), or create one.
863 if not isinstance(request, common.GenerateAccessTokenRequest):
864 request = common.GenerateAccessTokenRequest(request)
865 # If we have keyword arguments corresponding to fields on the
866 # request, apply these.
867 if name is not None:
868 request.name = name
869 if delegates is not None:
870 request.delegates = delegates
871 if scope is not None:
872 request.scope = scope
873 if lifetime is not None:
874 request.lifetime = lifetime
876 # Wrap the RPC method; this adds retry and timeout information,
877 # and friendly error handling.
878 rpc = self._transport._wrapped_methods[self._transport.generate_access_token]
880 # Certain fields should be provided within the metadata header;
881 # add these here.
882 metadata = tuple(metadata) + (
883 gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
884 )
886 # Validate the universe domain.
887 self._validate_universe_domain()
889 # Send the request.
890 response = rpc(
891 request,
892 retry=retry,
893 timeout=timeout,
894 metadata=metadata,
895 )
897 # Done; return the response.
898 return response
900 def generate_id_token(
901 self,
902 request: Optional[Union[common.GenerateIdTokenRequest, dict]] = None,
903 *,
904 name: Optional[str] = None,
905 delegates: Optional[MutableSequence[str]] = None,
906 audience: Optional[str] = None,
907 include_email: Optional[bool] = None,
908 retry: OptionalRetry = gapic_v1.method.DEFAULT,
909 timeout: Union[float, object] = gapic_v1.method.DEFAULT,
910 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
911 ) -> common.GenerateIdTokenResponse:
912 r"""Generates an OpenID Connect ID token for a service
913 account.
915 .. code-block:: python
917 # This snippet has been automatically generated and should be regarded as a
918 # code template only.
919 # It will require modifications to work:
920 # - It may require correct/in-range values for request initialization.
921 # - It may require specifying regional endpoints when creating the service
922 # client as shown in:
923 # https://googleapis.dev/python/google-api-core/latest/client_options.html
924 from google.cloud import iam_credentials_v1
926 def sample_generate_id_token():
927 # Create a client
928 client = iam_credentials_v1.IAMCredentialsClient()
930 # Initialize request argument(s)
931 request = iam_credentials_v1.GenerateIdTokenRequest(
932 name="name_value",
933 audience="audience_value",
934 )
936 # Make the request
937 response = client.generate_id_token(request=request)
939 # Handle the response
940 print(response)
942 Args:
943 request (Union[google.cloud.iam_credentials_v1.types.GenerateIdTokenRequest, dict]):
944 The request object.
945 name (str):
946 Required. The resource name of the service account for
947 which the credentials are requested, in the following
948 format:
949 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
950 The ``-`` wildcard character is required; replacing it
951 with a project ID is invalid.
953 This corresponds to the ``name`` field
954 on the ``request`` instance; if ``request`` is provided, this
955 should not be set.
956 delegates (MutableSequence[str]):
957 The sequence of service accounts in a delegation chain.
958 Each service account must be granted the
959 ``roles/iam.serviceAccountTokenCreator`` role on its
960 next service account in the chain. The last service
961 account in the chain must be granted the
962 ``roles/iam.serviceAccountTokenCreator`` role on the
963 service account that is specified in the ``name`` field
964 of the request.
966 The delegates must have the following format:
967 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
968 The ``-`` wildcard character is required; replacing it
969 with a project ID is invalid.
971 This corresponds to the ``delegates`` field
972 on the ``request`` instance; if ``request`` is provided, this
973 should not be set.
974 audience (str):
975 Required. The audience for the token,
976 such as the API or account that this
977 token grants access to.
979 This corresponds to the ``audience`` field
980 on the ``request`` instance; if ``request`` is provided, this
981 should not be set.
982 include_email (bool):
983 Include the service account email in the token. If set
984 to ``true``, the token will contain ``email`` and
985 ``email_verified`` claims.
987 This corresponds to the ``include_email`` field
988 on the ``request`` instance; if ``request`` is provided, this
989 should not be set.
990 retry (google.api_core.retry.Retry): Designation of what errors, if any,
991 should be retried.
992 timeout (float): The timeout for this request.
993 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
994 sent along with the request as metadata. Normally, each value must be of type `str`,
995 but for metadata keys ending with the suffix `-bin`, the corresponding values must
996 be of type `bytes`.
998 Returns:
999 google.cloud.iam_credentials_v1.types.GenerateIdTokenResponse:
1001 """
1002 # Create or coerce a protobuf request object.
1003 # - Quick check: If we got a request object, we should *not* have
1004 # gotten any keyword arguments that map to the request.
1005 flattened_params = [name, delegates, audience, include_email]
1006 has_flattened_params = (
1007 len([param for param in flattened_params if param is not None]) > 0
1008 )
1009 if request is not None and has_flattened_params:
1010 raise ValueError(
1011 "If the `request` argument is set, then none of "
1012 "the individual field arguments should be set."
1013 )
1015 # - Use the request object if provided (there's no risk of modifying the input as
1016 # there are no flattened fields), or create one.
1017 if not isinstance(request, common.GenerateIdTokenRequest):
1018 request = common.GenerateIdTokenRequest(request)
1019 # If we have keyword arguments corresponding to fields on the
1020 # request, apply these.
1021 if name is not None:
1022 request.name = name
1023 if delegates is not None:
1024 request.delegates = delegates
1025 if audience is not None:
1026 request.audience = audience
1027 if include_email is not None:
1028 request.include_email = include_email
1030 # Wrap the RPC method; this adds retry and timeout information,
1031 # and friendly error handling.
1032 rpc = self._transport._wrapped_methods[self._transport.generate_id_token]
1034 # Certain fields should be provided within the metadata header;
1035 # add these here.
1036 metadata = tuple(metadata) + (
1037 gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
1038 )
1040 # Validate the universe domain.
1041 self._validate_universe_domain()
1043 # Send the request.
1044 response = rpc(
1045 request,
1046 retry=retry,
1047 timeout=timeout,
1048 metadata=metadata,
1049 )
1051 # Done; return the response.
1052 return response
1054 def sign_blob(
1055 self,
1056 request: Optional[Union[common.SignBlobRequest, dict]] = None,
1057 *,
1058 name: Optional[str] = None,
1059 delegates: Optional[MutableSequence[str]] = None,
1060 payload: Optional[bytes] = None,
1061 retry: OptionalRetry = gapic_v1.method.DEFAULT,
1062 timeout: Union[float, object] = gapic_v1.method.DEFAULT,
1063 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
1064 ) -> common.SignBlobResponse:
1065 r"""Signs a blob using a service account's system-managed
1066 private key.
1068 .. code-block:: python
1070 # This snippet has been automatically generated and should be regarded as a
1071 # code template only.
1072 # It will require modifications to work:
1073 # - It may require correct/in-range values for request initialization.
1074 # - It may require specifying regional endpoints when creating the service
1075 # client as shown in:
1076 # https://googleapis.dev/python/google-api-core/latest/client_options.html
1077 from google.cloud import iam_credentials_v1
1079 def sample_sign_blob():
1080 # Create a client
1081 client = iam_credentials_v1.IAMCredentialsClient()
1083 # Initialize request argument(s)
1084 request = iam_credentials_v1.SignBlobRequest(
1085 name="name_value",
1086 payload=b'payload_blob',
1087 )
1089 # Make the request
1090 response = client.sign_blob(request=request)
1092 # Handle the response
1093 print(response)
1095 Args:
1096 request (Union[google.cloud.iam_credentials_v1.types.SignBlobRequest, dict]):
1097 The request object.
1098 name (str):
1099 Required. The resource name of the service account for
1100 which the credentials are requested, in the following
1101 format:
1102 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
1103 The ``-`` wildcard character is required; replacing it
1104 with a project ID is invalid.
1106 This corresponds to the ``name`` field
1107 on the ``request`` instance; if ``request`` is provided, this
1108 should not be set.
1109 delegates (MutableSequence[str]):
1110 The sequence of service accounts in a delegation chain.
1111 Each service account must be granted the
1112 ``roles/iam.serviceAccountTokenCreator`` role on its
1113 next service account in the chain. The last service
1114 account in the chain must be granted the
1115 ``roles/iam.serviceAccountTokenCreator`` role on the
1116 service account that is specified in the ``name`` field
1117 of the request.
1119 The delegates must have the following format:
1120 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
1121 The ``-`` wildcard character is required; replacing it
1122 with a project ID is invalid.
1124 This corresponds to the ``delegates`` field
1125 on the ``request`` instance; if ``request`` is provided, this
1126 should not be set.
1127 payload (bytes):
1128 Required. The bytes to sign.
1129 This corresponds to the ``payload`` field
1130 on the ``request`` instance; if ``request`` is provided, this
1131 should not be set.
1132 retry (google.api_core.retry.Retry): Designation of what errors, if any,
1133 should be retried.
1134 timeout (float): The timeout for this request.
1135 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
1136 sent along with the request as metadata. Normally, each value must be of type `str`,
1137 but for metadata keys ending with the suffix `-bin`, the corresponding values must
1138 be of type `bytes`.
1140 Returns:
1141 google.cloud.iam_credentials_v1.types.SignBlobResponse:
1143 """
1144 # Create or coerce a protobuf request object.
1145 # - Quick check: If we got a request object, we should *not* have
1146 # gotten any keyword arguments that map to the request.
1147 flattened_params = [name, delegates, payload]
1148 has_flattened_params = (
1149 len([param for param in flattened_params if param is not None]) > 0
1150 )
1151 if request is not None and has_flattened_params:
1152 raise ValueError(
1153 "If the `request` argument is set, then none of "
1154 "the individual field arguments should be set."
1155 )
1157 # - Use the request object if provided (there's no risk of modifying the input as
1158 # there are no flattened fields), or create one.
1159 if not isinstance(request, common.SignBlobRequest):
1160 request = common.SignBlobRequest(request)
1161 # If we have keyword arguments corresponding to fields on the
1162 # request, apply these.
1163 if name is not None:
1164 request.name = name
1165 if delegates is not None:
1166 request.delegates = delegates
1167 if payload is not None:
1168 request.payload = payload
1170 # Wrap the RPC method; this adds retry and timeout information,
1171 # and friendly error handling.
1172 rpc = self._transport._wrapped_methods[self._transport.sign_blob]
1174 # Certain fields should be provided within the metadata header;
1175 # add these here.
1176 metadata = tuple(metadata) + (
1177 gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
1178 )
1180 # Validate the universe domain.
1181 self._validate_universe_domain()
1183 # Send the request.
1184 response = rpc(
1185 request,
1186 retry=retry,
1187 timeout=timeout,
1188 metadata=metadata,
1189 )
1191 # Done; return the response.
1192 return response
1194 def sign_jwt(
1195 self,
1196 request: Optional[Union[common.SignJwtRequest, dict]] = None,
1197 *,
1198 name: Optional[str] = None,
1199 delegates: Optional[MutableSequence[str]] = None,
1200 payload: Optional[str] = None,
1201 retry: OptionalRetry = gapic_v1.method.DEFAULT,
1202 timeout: Union[float, object] = gapic_v1.method.DEFAULT,
1203 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
1204 ) -> common.SignJwtResponse:
1205 r"""Signs a JWT using a service account's system-managed
1206 private key.
1208 .. code-block:: python
1210 # This snippet has been automatically generated and should be regarded as a
1211 # code template only.
1212 # It will require modifications to work:
1213 # - It may require correct/in-range values for request initialization.
1214 # - It may require specifying regional endpoints when creating the service
1215 # client as shown in:
1216 # https://googleapis.dev/python/google-api-core/latest/client_options.html
1217 from google.cloud import iam_credentials_v1
1219 def sample_sign_jwt():
1220 # Create a client
1221 client = iam_credentials_v1.IAMCredentialsClient()
1223 # Initialize request argument(s)
1224 request = iam_credentials_v1.SignJwtRequest(
1225 name="name_value",
1226 payload="payload_value",
1227 )
1229 # Make the request
1230 response = client.sign_jwt(request=request)
1232 # Handle the response
1233 print(response)
1235 Args:
1236 request (Union[google.cloud.iam_credentials_v1.types.SignJwtRequest, dict]):
1237 The request object.
1238 name (str):
1239 Required. The resource name of the service account for
1240 which the credentials are requested, in the following
1241 format:
1242 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
1243 The ``-`` wildcard character is required; replacing it
1244 with a project ID is invalid.
1246 This corresponds to the ``name`` field
1247 on the ``request`` instance; if ``request`` is provided, this
1248 should not be set.
1249 delegates (MutableSequence[str]):
1250 The sequence of service accounts in a delegation chain.
1251 Each service account must be granted the
1252 ``roles/iam.serviceAccountTokenCreator`` role on its
1253 next service account in the chain. The last service
1254 account in the chain must be granted the
1255 ``roles/iam.serviceAccountTokenCreator`` role on the
1256 service account that is specified in the ``name`` field
1257 of the request.
1259 The delegates must have the following format:
1260 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
1261 The ``-`` wildcard character is required; replacing it
1262 with a project ID is invalid.
1264 This corresponds to the ``delegates`` field
1265 on the ``request`` instance; if ``request`` is provided, this
1266 should not be set.
1267 payload (str):
1268 Required. The JWT payload to sign: a
1269 JSON object that contains a JWT Claims
1270 Set.
1272 This corresponds to the ``payload`` field
1273 on the ``request`` instance; if ``request`` is provided, this
1274 should not be set.
1275 retry (google.api_core.retry.Retry): Designation of what errors, if any,
1276 should be retried.
1277 timeout (float): The timeout for this request.
1278 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
1279 sent along with the request as metadata. Normally, each value must be of type `str`,
1280 but for metadata keys ending with the suffix `-bin`, the corresponding values must
1281 be of type `bytes`.
1283 Returns:
1284 google.cloud.iam_credentials_v1.types.SignJwtResponse:
1286 """
1287 # Create or coerce a protobuf request object.
1288 # - Quick check: If we got a request object, we should *not* have
1289 # gotten any keyword arguments that map to the request.
1290 flattened_params = [name, delegates, payload]
1291 has_flattened_params = (
1292 len([param for param in flattened_params if param is not None]) > 0
1293 )
1294 if request is not None and has_flattened_params:
1295 raise ValueError(
1296 "If the `request` argument is set, then none of "
1297 "the individual field arguments should be set."
1298 )
1300 # - Use the request object if provided (there's no risk of modifying the input as
1301 # there are no flattened fields), or create one.
1302 if not isinstance(request, common.SignJwtRequest):
1303 request = common.SignJwtRequest(request)
1304 # If we have keyword arguments corresponding to fields on the
1305 # request, apply these.
1306 if name is not None:
1307 request.name = name
1308 if delegates is not None:
1309 request.delegates = delegates
1310 if payload is not None:
1311 request.payload = payload
1313 # Wrap the RPC method; this adds retry and timeout information,
1314 # and friendly error handling.
1315 rpc = self._transport._wrapped_methods[self._transport.sign_jwt]
1317 # Certain fields should be provided within the metadata header;
1318 # add these here.
1319 metadata = tuple(metadata) + (
1320 gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
1321 )
1323 # Validate the universe domain.
1324 self._validate_universe_domain()
1326 # Send the request.
1327 response = rpc(
1328 request,
1329 retry=retry,
1330 timeout=timeout,
1331 metadata=metadata,
1332 )
1334 # Done; return the response.
1335 return response
1337 def __enter__(self) -> "IAMCredentialsClient":
1338 return self
1340 def __exit__(self, type, value, traceback):
1341 """Releases underlying transport's resources.
1343 .. warning::
1344 ONLY use as a context manager if the transport is NOT shared
1345 with other clients! Exiting the with block will CLOSE the transport
1346 and may cause errors in other clients!
1347 """
1348 self.transport.close()
1351DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
1352 gapic_version=package_version.__version__
1353)
1355if hasattr(DEFAULT_CLIENT_INFO, "protobuf_runtime_version"): # pragma: NO COVER
1356 DEFAULT_CLIENT_INFO.protobuf_runtime_version = google.protobuf.__version__
1358__all__ = ("IAMCredentialsClient",)