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#
16from collections import OrderedDict
17from http import HTTPStatus
18import json
19import logging as std_logging
20import os
21import re
22from typing import (
23 Callable,
24 Dict,
25 Mapping,
26 MutableMapping,
27 MutableSequence,
28 Optional,
29 Sequence,
30 Tuple,
31 Type,
32 Union,
33 cast,
34)
35import warnings
37from google.api_core import client_options as client_options_lib
38from google.api_core import exceptions as core_exceptions
39from google.api_core import gapic_v1
40from google.api_core import retry as retries
41from google.auth import credentials as ga_credentials # type: ignore
42from google.auth.exceptions import MutualTLSChannelError # type: ignore
43from google.auth.transport import mtls # type: ignore
44from google.auth.transport.grpc import SslCredentials # type: ignore
45from google.oauth2 import service_account # type: ignore
46import google.protobuf
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__)
64from google.protobuf import duration_pb2 # type: ignore
65from google.protobuf import 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 = (
84 OrderedDict()
85 ) # type: Dict[str, Type[IAMCredentialsTransport]]
86 _transport_registry["grpc"] = IAMCredentialsGrpcTransport
87 _transport_registry["grpc_asyncio"] = IAMCredentialsGrpcAsyncIOTransport
88 _transport_registry["rest"] = IAMCredentialsRestTransport
90 def get_transport_class(
91 cls,
92 label: Optional[str] = None,
93 ) -> Type[IAMCredentialsTransport]:
94 """Returns an appropriate transport class.
96 Args:
97 label: The name of the desired transport. If none is
98 provided, then the first transport in the registry is used.
100 Returns:
101 The transport class to use.
102 """
103 # If a specific transport is requested, return that one.
104 if label:
105 return cls._transport_registry[label]
107 # No transport is requested; return the default (that is, the first one
108 # in the dictionary).
109 return next(iter(cls._transport_registry.values()))
112class IAMCredentialsClient(metaclass=IAMCredentialsClientMeta):
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.
123 """
125 @staticmethod
126 def _get_default_mtls_endpoint(api_endpoint):
127 """Converts api endpoint to mTLS endpoint.
129 Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
130 "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
131 Args:
132 api_endpoint (Optional[str]): the api endpoint to convert.
133 Returns:
134 str: converted mTLS api endpoint.
135 """
136 if not api_endpoint:
137 return api_endpoint
139 mtls_endpoint_re = re.compile(
140 r"(?P<name>[^.]+)(?P<mtls>\.mtls)?(?P<sandbox>\.sandbox)?(?P<googledomain>\.googleapis\.com)?"
141 )
143 m = mtls_endpoint_re.match(api_endpoint)
144 name, mtls, sandbox, googledomain = m.groups()
145 if mtls or not googledomain:
146 return api_endpoint
148 if sandbox:
149 return api_endpoint.replace(
150 "sandbox.googleapis.com", "mtls.sandbox.googleapis.com"
151 )
153 return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com")
155 # Note: DEFAULT_ENDPOINT is deprecated. Use _DEFAULT_ENDPOINT_TEMPLATE instead.
156 DEFAULT_ENDPOINT = "iamcredentials.googleapis.com"
157 DEFAULT_MTLS_ENDPOINT = _get_default_mtls_endpoint.__func__( # type: ignore
158 DEFAULT_ENDPOINT
159 )
161 _DEFAULT_ENDPOINT_TEMPLATE = "iamcredentials.{UNIVERSE_DOMAIN}"
162 _DEFAULT_UNIVERSE = "googleapis.com"
164 @staticmethod
165 def _use_client_cert_effective():
166 """Returns whether client certificate should be used for mTLS if the
167 google-auth version supports should_use_client_cert automatic mTLS enablement.
169 Alternatively, read from the GOOGLE_API_USE_CLIENT_CERTIFICATE env var.
171 Returns:
172 bool: whether client certificate should be used for mTLS
173 Raises:
174 ValueError: (If using a version of google-auth without should_use_client_cert and
175 GOOGLE_API_USE_CLIENT_CERTIFICATE is set to an unexpected value.)
176 """
177 # check if google-auth version supports should_use_client_cert for automatic mTLS enablement
178 if hasattr(mtls, "should_use_client_cert"): # pragma: NO COVER
179 return mtls.should_use_client_cert()
180 else: # pragma: NO COVER
181 # if unsupported, fallback to reading from env var
182 use_client_cert_str = os.getenv(
183 "GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"
184 ).lower()
185 if use_client_cert_str not in ("true", "false"):
186 raise ValueError(
187 "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be"
188 " either `true` or `false`"
189 )
190 return use_client_cert_str == "true"
192 @classmethod
193 def from_service_account_info(cls, info: dict, *args, **kwargs):
194 """Creates an instance of this client using the provided credentials
195 info.
197 Args:
198 info (dict): The service account private key info.
199 args: Additional arguments to pass to the constructor.
200 kwargs: Additional arguments to pass to the constructor.
202 Returns:
203 IAMCredentialsClient: The constructed client.
204 """
205 credentials = service_account.Credentials.from_service_account_info(info)
206 kwargs["credentials"] = credentials
207 return cls(*args, **kwargs)
209 @classmethod
210 def from_service_account_file(cls, filename: str, *args, **kwargs):
211 """Creates an instance of this client using the provided credentials
212 file.
214 Args:
215 filename (str): The path to the service account private key json
216 file.
217 args: Additional arguments to pass to the constructor.
218 kwargs: Additional arguments to pass to the constructor.
220 Returns:
221 IAMCredentialsClient: The constructed client.
222 """
223 credentials = service_account.Credentials.from_service_account_file(filename)
224 kwargs["credentials"] = credentials
225 return cls(*args, **kwargs)
227 from_service_account_json = from_service_account_file
229 @property
230 def transport(self) -> IAMCredentialsTransport:
231 """Returns the transport used by the client instance.
233 Returns:
234 IAMCredentialsTransport: The transport used by the client
235 instance.
236 """
237 return self._transport
239 @staticmethod
240 def service_account_path(
241 project: str,
242 service_account: str,
243 ) -> str:
244 """Returns a fully-qualified service_account string."""
245 return "projects/{project}/serviceAccounts/{service_account}".format(
246 project=project,
247 service_account=service_account,
248 )
250 @staticmethod
251 def parse_service_account_path(path: str) -> Dict[str, str]:
252 """Parses a service_account path into its component segments."""
253 m = re.match(
254 r"^projects/(?P<project>.+?)/serviceAccounts/(?P<service_account>.+?)$",
255 path,
256 )
257 return m.groupdict() if m else {}
259 @staticmethod
260 def common_billing_account_path(
261 billing_account: str,
262 ) -> str:
263 """Returns a fully-qualified billing_account string."""
264 return "billingAccounts/{billing_account}".format(
265 billing_account=billing_account,
266 )
268 @staticmethod
269 def parse_common_billing_account_path(path: str) -> Dict[str, str]:
270 """Parse a billing_account path into its component segments."""
271 m = re.match(r"^billingAccounts/(?P<billing_account>.+?)$", path)
272 return m.groupdict() if m else {}
274 @staticmethod
275 def common_folder_path(
276 folder: str,
277 ) -> str:
278 """Returns a fully-qualified folder string."""
279 return "folders/{folder}".format(
280 folder=folder,
281 )
283 @staticmethod
284 def parse_common_folder_path(path: str) -> Dict[str, str]:
285 """Parse a folder path into its component segments."""
286 m = re.match(r"^folders/(?P<folder>.+?)$", path)
287 return m.groupdict() if m else {}
289 @staticmethod
290 def common_organization_path(
291 organization: str,
292 ) -> str:
293 """Returns a fully-qualified organization string."""
294 return "organizations/{organization}".format(
295 organization=organization,
296 )
298 @staticmethod
299 def parse_common_organization_path(path: str) -> Dict[str, str]:
300 """Parse a organization path into its component segments."""
301 m = re.match(r"^organizations/(?P<organization>.+?)$", path)
302 return m.groupdict() if m else {}
304 @staticmethod
305 def common_project_path(
306 project: str,
307 ) -> str:
308 """Returns a fully-qualified project string."""
309 return "projects/{project}".format(
310 project=project,
311 )
313 @staticmethod
314 def parse_common_project_path(path: str) -> Dict[str, str]:
315 """Parse a project path into its component segments."""
316 m = re.match(r"^projects/(?P<project>.+?)$", path)
317 return m.groupdict() if m else {}
319 @staticmethod
320 def common_location_path(
321 project: str,
322 location: str,
323 ) -> str:
324 """Returns a fully-qualified location string."""
325 return "projects/{project}/locations/{location}".format(
326 project=project,
327 location=location,
328 )
330 @staticmethod
331 def parse_common_location_path(path: str) -> Dict[str, str]:
332 """Parse a location path into its component segments."""
333 m = re.match(r"^projects/(?P<project>.+?)/locations/(?P<location>.+?)$", path)
334 return m.groupdict() if m else {}
336 @classmethod
337 def get_mtls_endpoint_and_cert_source(
338 cls, client_options: Optional[client_options_lib.ClientOptions] = None
339 ):
340 """Deprecated. Return the API endpoint and client cert source for mutual TLS.
342 The client cert source is determined in the following order:
343 (1) if `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is not "true", the
344 client cert source is None.
345 (2) if `client_options.client_cert_source` is provided, use the provided one; if the
346 default client cert source exists, use the default one; otherwise the client cert
347 source is None.
349 The API endpoint is determined in the following order:
350 (1) if `client_options.api_endpoint` if provided, use the provided one.
351 (2) if `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is "always", use the
352 default mTLS endpoint; if the environment variable is "never", use the default API
353 endpoint; otherwise if client cert source exists, use the default mTLS endpoint, otherwise
354 use the default API endpoint.
356 More details can be found at https://google.aip.dev/auth/4114.
358 Args:
359 client_options (google.api_core.client_options.ClientOptions): Custom options for the
360 client. Only the `api_endpoint` and `client_cert_source` properties may be used
361 in this method.
363 Returns:
364 Tuple[str, Callable[[], Tuple[bytes, bytes]]]: returns the API endpoint and the
365 client cert source to use.
367 Raises:
368 google.auth.exceptions.MutualTLSChannelError: If any errors happen.
369 """
371 warnings.warn(
372 "get_mtls_endpoint_and_cert_source is deprecated. Use the api_endpoint property instead.",
373 DeprecationWarning,
374 )
375 if client_options is None:
376 client_options = client_options_lib.ClientOptions()
377 use_client_cert = IAMCredentialsClient._use_client_cert_effective()
378 use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto")
379 if use_mtls_endpoint not in ("auto", "never", "always"):
380 raise MutualTLSChannelError(
381 "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`, `auto` or `always`"
382 )
384 # Figure out the client cert source to use.
385 client_cert_source = None
386 if use_client_cert:
387 if client_options.client_cert_source:
388 client_cert_source = client_options.client_cert_source
389 elif mtls.has_default_client_cert_source():
390 client_cert_source = mtls.default_client_cert_source()
392 # Figure out which api endpoint to use.
393 if client_options.api_endpoint is not None:
394 api_endpoint = client_options.api_endpoint
395 elif use_mtls_endpoint == "always" or (
396 use_mtls_endpoint == "auto" and client_cert_source
397 ):
398 api_endpoint = cls.DEFAULT_MTLS_ENDPOINT
399 else:
400 api_endpoint = cls.DEFAULT_ENDPOINT
402 return api_endpoint, client_cert_source
404 @staticmethod
405 def _read_environment_variables():
406 """Returns the environment variables used by the client.
408 Returns:
409 Tuple[bool, str, str]: returns the GOOGLE_API_USE_CLIENT_CERTIFICATE,
410 GOOGLE_API_USE_MTLS_ENDPOINT, and GOOGLE_CLOUD_UNIVERSE_DOMAIN environment variables.
412 Raises:
413 ValueError: If GOOGLE_API_USE_CLIENT_CERTIFICATE is not
414 any of ["true", "false"].
415 google.auth.exceptions.MutualTLSChannelError: If GOOGLE_API_USE_MTLS_ENDPOINT
416 is not any of ["auto", "never", "always"].
417 """
418 use_client_cert = IAMCredentialsClient._use_client_cert_effective()
419 use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower()
420 universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN")
421 if use_mtls_endpoint not in ("auto", "never", "always"):
422 raise MutualTLSChannelError(
423 "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`, `auto` or `always`"
424 )
425 return use_client_cert, use_mtls_endpoint, universe_domain_env
427 @staticmethod
428 def _get_client_cert_source(provided_cert_source, use_cert_flag):
429 """Return the client cert source to be used by the client.
431 Args:
432 provided_cert_source (bytes): The client certificate source provided.
433 use_cert_flag (bool): A flag indicating whether to use the client certificate.
435 Returns:
436 bytes or None: The client cert source to be used by the client.
437 """
438 client_cert_source = None
439 if use_cert_flag:
440 if provided_cert_source:
441 client_cert_source = provided_cert_source
442 elif mtls.has_default_client_cert_source():
443 client_cert_source = mtls.default_client_cert_source()
444 return client_cert_source
446 @staticmethod
447 def _get_api_endpoint(
448 api_override, client_cert_source, universe_domain, use_mtls_endpoint
449 ):
450 """Return the API endpoint used by the client.
452 Args:
453 api_override (str): The API endpoint override. If specified, this is always
454 the return value of this function and the other arguments are not used.
455 client_cert_source (bytes): The client certificate source used by the client.
456 universe_domain (str): The universe domain used by the client.
457 use_mtls_endpoint (str): How to use the mTLS endpoint, which depends also on the other parameters.
458 Possible values are "always", "auto", or "never".
460 Returns:
461 str: The API endpoint to be used by the client.
462 """
463 if api_override is not None:
464 api_endpoint = api_override
465 elif use_mtls_endpoint == "always" or (
466 use_mtls_endpoint == "auto" and client_cert_source
467 ):
468 _default_universe = IAMCredentialsClient._DEFAULT_UNIVERSE
469 if universe_domain != _default_universe:
470 raise MutualTLSChannelError(
471 f"mTLS is not supported in any universe other than {_default_universe}."
472 )
473 api_endpoint = IAMCredentialsClient.DEFAULT_MTLS_ENDPOINT
474 else:
475 api_endpoint = IAMCredentialsClient._DEFAULT_ENDPOINT_TEMPLATE.format(
476 UNIVERSE_DOMAIN=universe_domain
477 )
478 return api_endpoint
480 @staticmethod
481 def _get_universe_domain(
482 client_universe_domain: Optional[str], universe_domain_env: Optional[str]
483 ) -> str:
484 """Return the universe domain used by the client.
486 Args:
487 client_universe_domain (Optional[str]): The universe domain configured via the client options.
488 universe_domain_env (Optional[str]): The universe domain configured via the "GOOGLE_CLOUD_UNIVERSE_DOMAIN" environment variable.
490 Returns:
491 str: The universe domain to be used by the client.
493 Raises:
494 ValueError: If the universe domain is an empty string.
495 """
496 universe_domain = IAMCredentialsClient._DEFAULT_UNIVERSE
497 if client_universe_domain is not None:
498 universe_domain = client_universe_domain
499 elif universe_domain_env is not None:
500 universe_domain = universe_domain_env
501 if len(universe_domain.strip()) == 0:
502 raise ValueError("Universe Domain cannot be an empty string.")
503 return universe_domain
505 def _validate_universe_domain(self):
506 """Validates client's and credentials' universe domains are consistent.
508 Returns:
509 bool: True iff the configured universe domain is valid.
511 Raises:
512 ValueError: If the configured universe domain is not valid.
513 """
515 # NOTE (b/349488459): universe validation is disabled until further notice.
516 return True
518 def _add_cred_info_for_auth_errors(
519 self, error: core_exceptions.GoogleAPICallError
520 ) -> None:
521 """Adds credential info string to error details for 401/403/404 errors.
523 Args:
524 error (google.api_core.exceptions.GoogleAPICallError): The error to add the cred info.
525 """
526 if error.code not in [
527 HTTPStatus.UNAUTHORIZED,
528 HTTPStatus.FORBIDDEN,
529 HTTPStatus.NOT_FOUND,
530 ]:
531 return
533 cred = self._transport._credentials
535 # get_cred_info is only available in google-auth>=2.35.0
536 if not hasattr(cred, "get_cred_info"):
537 return
539 # ignore the type check since pypy test fails when get_cred_info
540 # is not available
541 cred_info = cred.get_cred_info() # type: ignore
542 if cred_info and hasattr(error._details, "append"):
543 error._details.append(json.dumps(cred_info))
545 @property
546 def api_endpoint(self):
547 """Return the API endpoint used by the client instance.
549 Returns:
550 str: The API endpoint used by the client instance.
551 """
552 return self._api_endpoint
554 @property
555 def universe_domain(self) -> str:
556 """Return the universe domain used by the client instance.
558 Returns:
559 str: The universe domain used by the client instance.
560 """
561 return self._universe_domain
563 def __init__(
564 self,
565 *,
566 credentials: Optional[ga_credentials.Credentials] = None,
567 transport: Optional[
568 Union[str, IAMCredentialsTransport, Callable[..., IAMCredentialsTransport]]
569 ] = None,
570 client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
571 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
572 ) -> None:
573 """Instantiates the iam credentials client.
575 Args:
576 credentials (Optional[google.auth.credentials.Credentials]): The
577 authorization credentials to attach to requests. These
578 credentials identify the application to the service; if none
579 are specified, the client will attempt to ascertain the
580 credentials from the environment.
581 transport (Optional[Union[str,IAMCredentialsTransport,Callable[..., IAMCredentialsTransport]]]):
582 The transport to use, or a Callable that constructs and returns a new transport.
583 If a Callable is given, it will be called with the same set of initialization
584 arguments as used in the IAMCredentialsTransport constructor.
585 If set to None, a transport is chosen automatically.
586 client_options (Optional[Union[google.api_core.client_options.ClientOptions, dict]]):
587 Custom options for the client.
589 1. The ``api_endpoint`` property can be used to override the
590 default endpoint provided by the client when ``transport`` is
591 not explicitly provided. Only if this property is not set and
592 ``transport`` was not explicitly provided, the endpoint is
593 determined by the GOOGLE_API_USE_MTLS_ENDPOINT environment
594 variable, which have one of the following values:
595 "always" (always use the default mTLS endpoint), "never" (always
596 use the default regular endpoint) and "auto" (auto-switch to the
597 default mTLS endpoint if client certificate is present; this is
598 the default value).
600 2. If the GOOGLE_API_USE_CLIENT_CERTIFICATE environment variable
601 is "true", then the ``client_cert_source`` property can be used
602 to provide a client certificate for mTLS transport. If
603 not provided, the default SSL client certificate will be used if
604 present. If GOOGLE_API_USE_CLIENT_CERTIFICATE is "false" or not
605 set, no client certificate will be used.
607 3. The ``universe_domain`` property can be used to override the
608 default "googleapis.com" universe. Note that the ``api_endpoint``
609 property still takes precedence; and ``universe_domain`` is
610 currently not supported for mTLS.
612 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
613 The client info used to send a user-agent string along with
614 API requests. If ``None``, then default info will be used.
615 Generally, you only need to set this if you're developing
616 your own client library.
618 Raises:
619 google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
620 creation failed for any reason.
621 """
622 self._client_options = client_options
623 if isinstance(self._client_options, dict):
624 self._client_options = client_options_lib.from_dict(self._client_options)
625 if self._client_options is None:
626 self._client_options = client_options_lib.ClientOptions()
627 self._client_options = cast(
628 client_options_lib.ClientOptions, self._client_options
629 )
631 universe_domain_opt = getattr(self._client_options, "universe_domain", None)
633 (
634 self._use_client_cert,
635 self._use_mtls_endpoint,
636 self._universe_domain_env,
637 ) = IAMCredentialsClient._read_environment_variables()
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 = None # 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 "
673 "directly."
674 )
675 self._transport = cast(IAMCredentialsTransport, transport)
676 self._api_endpoint = self._transport.host
678 self._api_endpoint = (
679 self._api_endpoint
680 or IAMCredentialsClient._get_api_endpoint(
681 self._client_options.api_endpoint,
682 self._client_cert_source,
683 self._universe_domain,
684 self._use_mtls_endpoint,
685 )
686 )
688 if not transport_provided:
689 import google.auth._default # type: ignore
691 if api_key_value and hasattr(
692 google.auth._default, "get_api_key_credentials"
693 ):
694 credentials = google.auth._default.get_api_key_credentials(
695 api_key_value
696 )
698 transport_init: Union[
699 Type[IAMCredentialsTransport], Callable[..., IAMCredentialsTransport]
700 ] = (
701 IAMCredentialsClient.get_transport_class(transport)
702 if isinstance(transport, str) or transport is None
703 else cast(Callable[..., IAMCredentialsTransport], transport)
704 )
705 # initialize with the provided callable or the passed in class
706 self._transport = transport_init(
707 credentials=credentials,
708 credentials_file=self._client_options.credentials_file,
709 host=self._api_endpoint,
710 scopes=self._client_options.scopes,
711 client_cert_source_for_mtls=self._client_cert_source,
712 quota_project_id=self._client_options.quota_project_id,
713 client_info=client_info,
714 always_use_jwt_access=True,
715 api_audience=self._client_options.api_audience,
716 )
718 if "async" not in str(self._transport):
719 if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
720 std_logging.DEBUG
721 ): # pragma: NO COVER
722 _LOGGER.debug(
723 "Created client `google.iam.credentials_v1.IAMCredentialsClient`.",
724 extra={
725 "serviceName": "google.iam.credentials.v1.IAMCredentials",
726 "universeDomain": getattr(
727 self._transport._credentials, "universe_domain", ""
728 ),
729 "credentialsType": f"{type(self._transport._credentials).__module__}.{type(self._transport._credentials).__qualname__}",
730 "credentialsInfo": getattr(
731 self.transport._credentials, "get_cred_info", lambda: None
732 )(),
733 }
734 if hasattr(self._transport, "_credentials")
735 else {
736 "serviceName": "google.iam.credentials.v1.IAMCredentials",
737 "credentialsType": None,
738 },
739 )
741 def generate_access_token(
742 self,
743 request: Optional[Union[common.GenerateAccessTokenRequest, dict]] = None,
744 *,
745 name: Optional[str] = None,
746 delegates: Optional[MutableSequence[str]] = None,
747 scope: Optional[MutableSequence[str]] = None,
748 lifetime: Optional[duration_pb2.Duration] = None,
749 retry: OptionalRetry = gapic_v1.method.DEFAULT,
750 timeout: Union[float, object] = gapic_v1.method.DEFAULT,
751 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
752 ) -> common.GenerateAccessTokenResponse:
753 r"""Generates an OAuth 2.0 access token for a service
754 account.
756 .. code-block:: python
758 # This snippet has been automatically generated and should be regarded as a
759 # code template only.
760 # It will require modifications to work:
761 # - It may require correct/in-range values for request initialization.
762 # - It may require specifying regional endpoints when creating the service
763 # client as shown in:
764 # https://googleapis.dev/python/google-api-core/latest/client_options.html
765 from google.cloud import iam_credentials_v1
767 def sample_generate_access_token():
768 # Create a client
769 client = iam_credentials_v1.IAMCredentialsClient()
771 # Initialize request argument(s)
772 request = iam_credentials_v1.GenerateAccessTokenRequest(
773 name="name_value",
774 scope=['scope_value1', 'scope_value2'],
775 )
777 # Make the request
778 response = client.generate_access_token(request=request)
780 # Handle the response
781 print(response)
783 Args:
784 request (Union[google.cloud.iam_credentials_v1.types.GenerateAccessTokenRequest, dict]):
785 The request object.
786 name (str):
787 Required. The resource name of the service account for
788 which the credentials are requested, in the following
789 format:
790 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
791 The ``-`` wildcard character is required; replacing it
792 with a project ID is invalid.
794 This corresponds to the ``name`` field
795 on the ``request`` instance; if ``request`` is provided, this
796 should not be set.
797 delegates (MutableSequence[str]):
798 The sequence of service accounts in a delegation chain.
799 Each service account must be granted the
800 ``roles/iam.serviceAccountTokenCreator`` role on its
801 next service account in the chain. The last service
802 account in the chain must be granted the
803 ``roles/iam.serviceAccountTokenCreator`` role on the
804 service account that is specified in the ``name`` field
805 of the request.
807 The delegates must have the following format:
808 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
809 The ``-`` wildcard character is required; replacing it
810 with a project ID is invalid.
812 This corresponds to the ``delegates`` field
813 on the ``request`` instance; if ``request`` is provided, this
814 should not be set.
815 scope (MutableSequence[str]):
816 Required. Code to identify the scopes
817 to be included in the OAuth 2.0 access
818 token. See
819 https://developers.google.com/identity/protocols/googlescopes
820 for more information.
821 At least one value required.
823 This corresponds to the ``scope`` field
824 on the ``request`` instance; if ``request`` is provided, this
825 should not be set.
826 lifetime (google.protobuf.duration_pb2.Duration):
827 The desired lifetime duration of the
828 access token in seconds. Must be set to
829 a value less than or equal to 3600 (1
830 hour). If a value is not specified, the
831 token's lifetime will be set to a
832 default value of one hour.
834 This corresponds to the ``lifetime`` field
835 on the ``request`` instance; if ``request`` is provided, this
836 should not be set.
837 retry (google.api_core.retry.Retry): Designation of what errors, if any,
838 should be retried.
839 timeout (float): The timeout for this request.
840 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
841 sent along with the request as metadata. Normally, each value must be of type `str`,
842 but for metadata keys ending with the suffix `-bin`, the corresponding values must
843 be of type `bytes`.
845 Returns:
846 google.cloud.iam_credentials_v1.types.GenerateAccessTokenResponse:
848 """
849 # Create or coerce a protobuf request object.
850 # - Quick check: If we got a request object, we should *not* have
851 # gotten any keyword arguments that map to the request.
852 flattened_params = [name, delegates, scope, lifetime]
853 has_flattened_params = (
854 len([param for param in flattened_params if param is not None]) > 0
855 )
856 if request is not None and has_flattened_params:
857 raise ValueError(
858 "If the `request` argument is set, then none of "
859 "the individual field arguments should be set."
860 )
862 # - Use the request object if provided (there's no risk of modifying the input as
863 # there are no flattened fields), or create one.
864 if not isinstance(request, common.GenerateAccessTokenRequest):
865 request = common.GenerateAccessTokenRequest(request)
866 # If we have keyword arguments corresponding to fields on the
867 # request, apply these.
868 if name is not None:
869 request.name = name
870 if delegates is not None:
871 request.delegates = delegates
872 if scope is not None:
873 request.scope = scope
874 if lifetime is not None:
875 request.lifetime = lifetime
877 # Wrap the RPC method; this adds retry and timeout information,
878 # and friendly error handling.
879 rpc = self._transport._wrapped_methods[self._transport.generate_access_token]
881 # Certain fields should be provided within the metadata header;
882 # add these here.
883 metadata = tuple(metadata) + (
884 gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
885 )
887 # Validate the universe domain.
888 self._validate_universe_domain()
890 # Send the request.
891 response = rpc(
892 request,
893 retry=retry,
894 timeout=timeout,
895 metadata=metadata,
896 )
898 # Done; return the response.
899 return response
901 def generate_id_token(
902 self,
903 request: Optional[Union[common.GenerateIdTokenRequest, dict]] = None,
904 *,
905 name: Optional[str] = None,
906 delegates: Optional[MutableSequence[str]] = None,
907 audience: Optional[str] = None,
908 include_email: Optional[bool] = None,
909 retry: OptionalRetry = gapic_v1.method.DEFAULT,
910 timeout: Union[float, object] = gapic_v1.method.DEFAULT,
911 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
912 ) -> common.GenerateIdTokenResponse:
913 r"""Generates an OpenID Connect ID token for a service
914 account.
916 .. code-block:: python
918 # This snippet has been automatically generated and should be regarded as a
919 # code template only.
920 # It will require modifications to work:
921 # - It may require correct/in-range values for request initialization.
922 # - It may require specifying regional endpoints when creating the service
923 # client as shown in:
924 # https://googleapis.dev/python/google-api-core/latest/client_options.html
925 from google.cloud import iam_credentials_v1
927 def sample_generate_id_token():
928 # Create a client
929 client = iam_credentials_v1.IAMCredentialsClient()
931 # Initialize request argument(s)
932 request = iam_credentials_v1.GenerateIdTokenRequest(
933 name="name_value",
934 audience="audience_value",
935 )
937 # Make the request
938 response = client.generate_id_token(request=request)
940 # Handle the response
941 print(response)
943 Args:
944 request (Union[google.cloud.iam_credentials_v1.types.GenerateIdTokenRequest, dict]):
945 The request object.
946 name (str):
947 Required. The resource name of the service account for
948 which the credentials are requested, in the following
949 format:
950 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
951 The ``-`` wildcard character is required; replacing it
952 with a project ID is invalid.
954 This corresponds to the ``name`` field
955 on the ``request`` instance; if ``request`` is provided, this
956 should not be set.
957 delegates (MutableSequence[str]):
958 The sequence of service accounts in a delegation chain.
959 Each service account must be granted the
960 ``roles/iam.serviceAccountTokenCreator`` role on its
961 next service account in the chain. The last service
962 account in the chain must be granted the
963 ``roles/iam.serviceAccountTokenCreator`` role on the
964 service account that is specified in the ``name`` field
965 of the request.
967 The delegates must have the following format:
968 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
969 The ``-`` wildcard character is required; replacing it
970 with a project ID is invalid.
972 This corresponds to the ``delegates`` field
973 on the ``request`` instance; if ``request`` is provided, this
974 should not be set.
975 audience (str):
976 Required. The audience for the token,
977 such as the API or account that this
978 token grants access to.
980 This corresponds to the ``audience`` field
981 on the ``request`` instance; if ``request`` is provided, this
982 should not be set.
983 include_email (bool):
984 Include the service account email in the token. If set
985 to ``true``, the token will contain ``email`` and
986 ``email_verified`` claims.
988 This corresponds to the ``include_email`` field
989 on the ``request`` instance; if ``request`` is provided, this
990 should not be set.
991 retry (google.api_core.retry.Retry): Designation of what errors, if any,
992 should be retried.
993 timeout (float): The timeout for this request.
994 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
995 sent along with the request as metadata. Normally, each value must be of type `str`,
996 but for metadata keys ending with the suffix `-bin`, the corresponding values must
997 be of type `bytes`.
999 Returns:
1000 google.cloud.iam_credentials_v1.types.GenerateIdTokenResponse:
1002 """
1003 # Create or coerce a protobuf request object.
1004 # - Quick check: If we got a request object, we should *not* have
1005 # gotten any keyword arguments that map to the request.
1006 flattened_params = [name, delegates, audience, include_email]
1007 has_flattened_params = (
1008 len([param for param in flattened_params if param is not None]) > 0
1009 )
1010 if request is not None and has_flattened_params:
1011 raise ValueError(
1012 "If the `request` argument is set, then none of "
1013 "the individual field arguments should be set."
1014 )
1016 # - Use the request object if provided (there's no risk of modifying the input as
1017 # there are no flattened fields), or create one.
1018 if not isinstance(request, common.GenerateIdTokenRequest):
1019 request = common.GenerateIdTokenRequest(request)
1020 # If we have keyword arguments corresponding to fields on the
1021 # request, apply these.
1022 if name is not None:
1023 request.name = name
1024 if delegates is not None:
1025 request.delegates = delegates
1026 if audience is not None:
1027 request.audience = audience
1028 if include_email is not None:
1029 request.include_email = include_email
1031 # Wrap the RPC method; this adds retry and timeout information,
1032 # and friendly error handling.
1033 rpc = self._transport._wrapped_methods[self._transport.generate_id_token]
1035 # Certain fields should be provided within the metadata header;
1036 # add these here.
1037 metadata = tuple(metadata) + (
1038 gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
1039 )
1041 # Validate the universe domain.
1042 self._validate_universe_domain()
1044 # Send the request.
1045 response = rpc(
1046 request,
1047 retry=retry,
1048 timeout=timeout,
1049 metadata=metadata,
1050 )
1052 # Done; return the response.
1053 return response
1055 def sign_blob(
1056 self,
1057 request: Optional[Union[common.SignBlobRequest, dict]] = None,
1058 *,
1059 name: Optional[str] = None,
1060 delegates: Optional[MutableSequence[str]] = None,
1061 payload: Optional[bytes] = None,
1062 retry: OptionalRetry = gapic_v1.method.DEFAULT,
1063 timeout: Union[float, object] = gapic_v1.method.DEFAULT,
1064 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
1065 ) -> common.SignBlobResponse:
1066 r"""Signs a blob using a service account's system-managed
1067 private key.
1069 .. code-block:: python
1071 # This snippet has been automatically generated and should be regarded as a
1072 # code template only.
1073 # It will require modifications to work:
1074 # - It may require correct/in-range values for request initialization.
1075 # - It may require specifying regional endpoints when creating the service
1076 # client as shown in:
1077 # https://googleapis.dev/python/google-api-core/latest/client_options.html
1078 from google.cloud import iam_credentials_v1
1080 def sample_sign_blob():
1081 # Create a client
1082 client = iam_credentials_v1.IAMCredentialsClient()
1084 # Initialize request argument(s)
1085 request = iam_credentials_v1.SignBlobRequest(
1086 name="name_value",
1087 payload=b'payload_blob',
1088 )
1090 # Make the request
1091 response = client.sign_blob(request=request)
1093 # Handle the response
1094 print(response)
1096 Args:
1097 request (Union[google.cloud.iam_credentials_v1.types.SignBlobRequest, dict]):
1098 The request object.
1099 name (str):
1100 Required. The resource name of the service account for
1101 which the credentials are requested, in the following
1102 format:
1103 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
1104 The ``-`` wildcard character is required; replacing it
1105 with a project ID is invalid.
1107 This corresponds to the ``name`` field
1108 on the ``request`` instance; if ``request`` is provided, this
1109 should not be set.
1110 delegates (MutableSequence[str]):
1111 The sequence of service accounts in a delegation chain.
1112 Each service account must be granted the
1113 ``roles/iam.serviceAccountTokenCreator`` role on its
1114 next service account in the chain. The last service
1115 account in the chain must be granted the
1116 ``roles/iam.serviceAccountTokenCreator`` role on the
1117 service account that is specified in the ``name`` field
1118 of the request.
1120 The delegates must have the following format:
1121 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
1122 The ``-`` wildcard character is required; replacing it
1123 with a project ID is invalid.
1125 This corresponds to the ``delegates`` field
1126 on the ``request`` instance; if ``request`` is provided, this
1127 should not be set.
1128 payload (bytes):
1129 Required. The bytes to sign.
1130 This corresponds to the ``payload`` field
1131 on the ``request`` instance; if ``request`` is provided, this
1132 should not be set.
1133 retry (google.api_core.retry.Retry): Designation of what errors, if any,
1134 should be retried.
1135 timeout (float): The timeout for this request.
1136 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
1137 sent along with the request as metadata. Normally, each value must be of type `str`,
1138 but for metadata keys ending with the suffix `-bin`, the corresponding values must
1139 be of type `bytes`.
1141 Returns:
1142 google.cloud.iam_credentials_v1.types.SignBlobResponse:
1144 """
1145 # Create or coerce a protobuf request object.
1146 # - Quick check: If we got a request object, we should *not* have
1147 # gotten any keyword arguments that map to the request.
1148 flattened_params = [name, delegates, payload]
1149 has_flattened_params = (
1150 len([param for param in flattened_params if param is not None]) > 0
1151 )
1152 if request is not None and has_flattened_params:
1153 raise ValueError(
1154 "If the `request` argument is set, then none of "
1155 "the individual field arguments should be set."
1156 )
1158 # - Use the request object if provided (there's no risk of modifying the input as
1159 # there are no flattened fields), or create one.
1160 if not isinstance(request, common.SignBlobRequest):
1161 request = common.SignBlobRequest(request)
1162 # If we have keyword arguments corresponding to fields on the
1163 # request, apply these.
1164 if name is not None:
1165 request.name = name
1166 if delegates is not None:
1167 request.delegates = delegates
1168 if payload is not None:
1169 request.payload = payload
1171 # Wrap the RPC method; this adds retry and timeout information,
1172 # and friendly error handling.
1173 rpc = self._transport._wrapped_methods[self._transport.sign_blob]
1175 # Certain fields should be provided within the metadata header;
1176 # add these here.
1177 metadata = tuple(metadata) + (
1178 gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
1179 )
1181 # Validate the universe domain.
1182 self._validate_universe_domain()
1184 # Send the request.
1185 response = rpc(
1186 request,
1187 retry=retry,
1188 timeout=timeout,
1189 metadata=metadata,
1190 )
1192 # Done; return the response.
1193 return response
1195 def sign_jwt(
1196 self,
1197 request: Optional[Union[common.SignJwtRequest, dict]] = None,
1198 *,
1199 name: Optional[str] = None,
1200 delegates: Optional[MutableSequence[str]] = None,
1201 payload: Optional[str] = None,
1202 retry: OptionalRetry = gapic_v1.method.DEFAULT,
1203 timeout: Union[float, object] = gapic_v1.method.DEFAULT,
1204 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
1205 ) -> common.SignJwtResponse:
1206 r"""Signs a JWT using a service account's system-managed
1207 private key.
1209 .. code-block:: python
1211 # This snippet has been automatically generated and should be regarded as a
1212 # code template only.
1213 # It will require modifications to work:
1214 # - It may require correct/in-range values for request initialization.
1215 # - It may require specifying regional endpoints when creating the service
1216 # client as shown in:
1217 # https://googleapis.dev/python/google-api-core/latest/client_options.html
1218 from google.cloud import iam_credentials_v1
1220 def sample_sign_jwt():
1221 # Create a client
1222 client = iam_credentials_v1.IAMCredentialsClient()
1224 # Initialize request argument(s)
1225 request = iam_credentials_v1.SignJwtRequest(
1226 name="name_value",
1227 payload="payload_value",
1228 )
1230 # Make the request
1231 response = client.sign_jwt(request=request)
1233 # Handle the response
1234 print(response)
1236 Args:
1237 request (Union[google.cloud.iam_credentials_v1.types.SignJwtRequest, dict]):
1238 The request object.
1239 name (str):
1240 Required. The resource name of the service account for
1241 which the credentials are requested, in the following
1242 format:
1243 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
1244 The ``-`` wildcard character is required; replacing it
1245 with a project ID is invalid.
1247 This corresponds to the ``name`` field
1248 on the ``request`` instance; if ``request`` is provided, this
1249 should not be set.
1250 delegates (MutableSequence[str]):
1251 The sequence of service accounts in a delegation chain.
1252 Each service account must be granted the
1253 ``roles/iam.serviceAccountTokenCreator`` role on its
1254 next service account in the chain. The last service
1255 account in the chain must be granted the
1256 ``roles/iam.serviceAccountTokenCreator`` role on the
1257 service account that is specified in the ``name`` field
1258 of the request.
1260 The delegates must have the following format:
1261 ``projects/-/serviceAccounts/{ACCOUNT_EMAIL_OR_UNIQUEID}``.
1262 The ``-`` wildcard character is required; replacing it
1263 with a project ID is invalid.
1265 This corresponds to the ``delegates`` field
1266 on the ``request`` instance; if ``request`` is provided, this
1267 should not be set.
1268 payload (str):
1269 Required. The JWT payload to sign: a
1270 JSON object that contains a JWT Claims
1271 Set.
1273 This corresponds to the ``payload`` field
1274 on the ``request`` instance; if ``request`` is provided, this
1275 should not be set.
1276 retry (google.api_core.retry.Retry): Designation of what errors, if any,
1277 should be retried.
1278 timeout (float): The timeout for this request.
1279 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
1280 sent along with the request as metadata. Normally, each value must be of type `str`,
1281 but for metadata keys ending with the suffix `-bin`, the corresponding values must
1282 be of type `bytes`.
1284 Returns:
1285 google.cloud.iam_credentials_v1.types.SignJwtResponse:
1287 """
1288 # Create or coerce a protobuf request object.
1289 # - Quick check: If we got a request object, we should *not* have
1290 # gotten any keyword arguments that map to the request.
1291 flattened_params = [name, delegates, payload]
1292 has_flattened_params = (
1293 len([param for param in flattened_params if param is not None]) > 0
1294 )
1295 if request is not None and has_flattened_params:
1296 raise ValueError(
1297 "If the `request` argument is set, then none of "
1298 "the individual field arguments should be set."
1299 )
1301 # - Use the request object if provided (there's no risk of modifying the input as
1302 # there are no flattened fields), or create one.
1303 if not isinstance(request, common.SignJwtRequest):
1304 request = common.SignJwtRequest(request)
1305 # If we have keyword arguments corresponding to fields on the
1306 # request, apply these.
1307 if name is not None:
1308 request.name = name
1309 if delegates is not None:
1310 request.delegates = delegates
1311 if payload is not None:
1312 request.payload = payload
1314 # Wrap the RPC method; this adds retry and timeout information,
1315 # and friendly error handling.
1316 rpc = self._transport._wrapped_methods[self._transport.sign_jwt]
1318 # Certain fields should be provided within the metadata header;
1319 # add these here.
1320 metadata = tuple(metadata) + (
1321 gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
1322 )
1324 # Validate the universe domain.
1325 self._validate_universe_domain()
1327 # Send the request.
1328 response = rpc(
1329 request,
1330 retry=retry,
1331 timeout=timeout,
1332 metadata=metadata,
1333 )
1335 # Done; return the response.
1336 return response
1338 def __enter__(self) -> "IAMCredentialsClient":
1339 return self
1341 def __exit__(self, type, value, traceback):
1342 """Releases underlying transport's resources.
1344 .. warning::
1345 ONLY use as a context manager if the transport is NOT shared
1346 with other clients! Exiting the with block will CLOSE the transport
1347 and may cause errors in other clients!
1348 """
1349 self.transport.close()
1352DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
1353 gapic_version=package_version.__version__
1354)
1356if hasattr(DEFAULT_CLIENT_INFO, "protobuf_runtime_version"): # pragma: NO COVER
1357 DEFAULT_CLIENT_INFO.protobuf_runtime_version = google.protobuf.__version__
1359__all__ = ("IAMCredentialsClient",)