« prev ^ index » next coverage.py v7.3.0, created at 2023-08-16 06:17 +0000
1# -*- coding: utf-8 -*-
2# Copyright 2023 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#
17import dataclasses
18import json # type: ignore
19import re
20from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
21import warnings
23from google.api_core import gapic_v1, path_template, rest_helpers, rest_streaming
24from google.api_core import exceptions as core_exceptions
25from google.api_core import retry as retries
26from google.auth import credentials as ga_credentials # type: ignore
27from google.auth.transport.grpc import SslCredentials # type: ignore
28from google.auth.transport.requests import AuthorizedSession # type: ignore
29from google.protobuf import json_format
30import grpc # type: ignore
31from requests import __version__ as requests_version
33try:
34 OptionalRetry = Union[retries.Retry, gapic_v1.method._MethodDefault]
35except AttributeError: # pragma: NO COVER
36 OptionalRetry = Union[retries.Retry, object] # type: ignore
39from google.cloud.iam_credentials_v1.types import common
41from .base import DEFAULT_CLIENT_INFO as BASE_DEFAULT_CLIENT_INFO
42from .base import IAMCredentialsTransport
44DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
45 gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
46 grpc_version=None,
47 rest_version=requests_version,
48)
51class IAMCredentialsRestInterceptor:
52 """Interceptor for IAMCredentials.
54 Interceptors are used to manipulate requests, request metadata, and responses
55 in arbitrary ways.
56 Example use cases include:
57 * Logging
58 * Verifying requests according to service or custom semantics
59 * Stripping extraneous information from responses
61 These use cases and more can be enabled by injecting an
62 instance of a custom subclass when constructing the IAMCredentialsRestTransport.
64 .. code-block:: python
65 class MyCustomIAMCredentialsInterceptor(IAMCredentialsRestInterceptor):
66 def pre_generate_access_token(self, request, metadata):
67 logging.log(f"Received request: {request}")
68 return request, metadata
70 def post_generate_access_token(self, response):
71 logging.log(f"Received response: {response}")
72 return response
74 def pre_generate_id_token(self, request, metadata):
75 logging.log(f"Received request: {request}")
76 return request, metadata
78 def post_generate_id_token(self, response):
79 logging.log(f"Received response: {response}")
80 return response
82 def pre_sign_blob(self, request, metadata):
83 logging.log(f"Received request: {request}")
84 return request, metadata
86 def post_sign_blob(self, response):
87 logging.log(f"Received response: {response}")
88 return response
90 def pre_sign_jwt(self, request, metadata):
91 logging.log(f"Received request: {request}")
92 return request, metadata
94 def post_sign_jwt(self, response):
95 logging.log(f"Received response: {response}")
96 return response
98 transport = IAMCredentialsRestTransport(interceptor=MyCustomIAMCredentialsInterceptor())
99 client = IAMCredentialsClient(transport=transport)
102 """
104 def pre_generate_access_token(
105 self,
106 request: common.GenerateAccessTokenRequest,
107 metadata: Sequence[Tuple[str, str]],
108 ) -> Tuple[common.GenerateAccessTokenRequest, Sequence[Tuple[str, str]]]:
109 """Pre-rpc interceptor for generate_access_token
111 Override in a subclass to manipulate the request or metadata
112 before they are sent to the IAMCredentials server.
113 """
114 return request, metadata
116 def post_generate_access_token(
117 self, response: common.GenerateAccessTokenResponse
118 ) -> common.GenerateAccessTokenResponse:
119 """Post-rpc interceptor for generate_access_token
121 Override in a subclass to manipulate the response
122 after it is returned by the IAMCredentials server but before
123 it is returned to user code.
124 """
125 return response
127 def pre_generate_id_token(
128 self,
129 request: common.GenerateIdTokenRequest,
130 metadata: Sequence[Tuple[str, str]],
131 ) -> Tuple[common.GenerateIdTokenRequest, Sequence[Tuple[str, str]]]:
132 """Pre-rpc interceptor for generate_id_token
134 Override in a subclass to manipulate the request or metadata
135 before they are sent to the IAMCredentials server.
136 """
137 return request, metadata
139 def post_generate_id_token(
140 self, response: common.GenerateIdTokenResponse
141 ) -> common.GenerateIdTokenResponse:
142 """Post-rpc interceptor for generate_id_token
144 Override in a subclass to manipulate the response
145 after it is returned by the IAMCredentials server but before
146 it is returned to user code.
147 """
148 return response
150 def pre_sign_blob(
151 self, request: common.SignBlobRequest, metadata: Sequence[Tuple[str, str]]
152 ) -> Tuple[common.SignBlobRequest, Sequence[Tuple[str, str]]]:
153 """Pre-rpc interceptor for sign_blob
155 Override in a subclass to manipulate the request or metadata
156 before they are sent to the IAMCredentials server.
157 """
158 return request, metadata
160 def post_sign_blob(
161 self, response: common.SignBlobResponse
162 ) -> common.SignBlobResponse:
163 """Post-rpc interceptor for sign_blob
165 Override in a subclass to manipulate the response
166 after it is returned by the IAMCredentials server but before
167 it is returned to user code.
168 """
169 return response
171 def pre_sign_jwt(
172 self, request: common.SignJwtRequest, metadata: Sequence[Tuple[str, str]]
173 ) -> Tuple[common.SignJwtRequest, Sequence[Tuple[str, str]]]:
174 """Pre-rpc interceptor for sign_jwt
176 Override in a subclass to manipulate the request or metadata
177 before they are sent to the IAMCredentials server.
178 """
179 return request, metadata
181 def post_sign_jwt(self, response: common.SignJwtResponse) -> common.SignJwtResponse:
182 """Post-rpc interceptor for sign_jwt
184 Override in a subclass to manipulate the response
185 after it is returned by the IAMCredentials server but before
186 it is returned to user code.
187 """
188 return response
191@dataclasses.dataclass
192class IAMCredentialsRestStub:
193 _session: AuthorizedSession
194 _host: str
195 _interceptor: IAMCredentialsRestInterceptor
198class IAMCredentialsRestTransport(IAMCredentialsTransport):
199 """REST backend transport for IAMCredentials.
201 A service account is a special type of Google account that
202 belongs to your application or a virtual machine (VM), instead
203 of to an individual end user. Your application assumes the
204 identity of the service account to call Google APIs, so that the
205 users aren't directly involved.
207 Service account credentials are used to temporarily assume the
208 identity of the service account. Supported credential types
209 include OAuth 2.0 access tokens, OpenID Connect ID tokens,
210 self-signed JSON Web Tokens (JWTs), and more.
212 This class defines the same methods as the primary client, so the
213 primary client can load the underlying transport implementation
214 and call it.
216 It sends JSON representations of protocol buffers over HTTP/1.1
218 """
220 def __init__(
221 self,
222 *,
223 host: str = "iamcredentials.googleapis.com",
224 credentials: Optional[ga_credentials.Credentials] = None,
225 credentials_file: Optional[str] = None,
226 scopes: Optional[Sequence[str]] = None,
227 client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
228 quota_project_id: Optional[str] = None,
229 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
230 always_use_jwt_access: Optional[bool] = False,
231 url_scheme: str = "https",
232 interceptor: Optional[IAMCredentialsRestInterceptor] = None,
233 api_audience: Optional[str] = None,
234 ) -> None:
235 """Instantiate the transport.
237 Args:
238 host (Optional[str]):
239 The hostname to connect to.
240 credentials (Optional[google.auth.credentials.Credentials]): The
241 authorization credentials to attach to requests. These
242 credentials identify the application to the service; if none
243 are specified, the client will attempt to ascertain the
244 credentials from the environment.
246 credentials_file (Optional[str]): A file with credentials that can
247 be loaded with :func:`google.auth.load_credentials_from_file`.
248 This argument is ignored if ``channel`` is provided.
249 scopes (Optional(Sequence[str])): A list of scopes. This argument is
250 ignored if ``channel`` is provided.
251 client_cert_source_for_mtls (Callable[[], Tuple[bytes, bytes]]): Client
252 certificate to configure mutual TLS HTTP channel. It is ignored
253 if ``channel`` is provided.
254 quota_project_id (Optional[str]): An optional project to use for billing
255 and quota.
256 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
257 The client info used to send a user-agent string along with
258 API requests. If ``None``, then default info will be used.
259 Generally, you only need to set this if you are developing
260 your own client library.
261 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
262 be used for service account credentials.
263 url_scheme: the protocol scheme for the API endpoint. Normally
264 "https", but for testing or local servers,
265 "http" can be specified.
266 """
267 # Run the base constructor
268 # TODO(yon-mg): resolve other ctor params i.e. scopes, quota, etc.
269 # TODO: When custom host (api_endpoint) is set, `scopes` must *also* be set on the
270 # credentials object
271 maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host)
272 if maybe_url_match is None:
273 raise ValueError(
274 f"Unexpected hostname structure: {host}"
275 ) # pragma: NO COVER
277 url_match_items = maybe_url_match.groupdict()
279 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
281 super().__init__(
282 host=host,
283 credentials=credentials,
284 client_info=client_info,
285 always_use_jwt_access=always_use_jwt_access,
286 api_audience=api_audience,
287 )
288 self._session = AuthorizedSession(
289 self._credentials, default_host=self.DEFAULT_HOST
290 )
291 if client_cert_source_for_mtls:
292 self._session.configure_mtls_channel(client_cert_source_for_mtls)
293 self._interceptor = interceptor or IAMCredentialsRestInterceptor()
294 self._prep_wrapped_messages(client_info)
296 class _GenerateAccessToken(IAMCredentialsRestStub):
297 def __hash__(self):
298 return hash("GenerateAccessToken")
300 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
302 @classmethod
303 def _get_unset_required_fields(cls, message_dict):
304 return {
305 k: v
306 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
307 if k not in message_dict
308 }
310 def __call__(
311 self,
312 request: common.GenerateAccessTokenRequest,
313 *,
314 retry: OptionalRetry = gapic_v1.method.DEFAULT,
315 timeout: Optional[float] = None,
316 metadata: Sequence[Tuple[str, str]] = (),
317 ) -> common.GenerateAccessTokenResponse:
318 r"""Call the generate access token method over HTTP.
320 Args:
321 request (~.common.GenerateAccessTokenRequest):
322 The request object.
323 retry (google.api_core.retry.Retry): Designation of what errors, if any,
324 should be retried.
325 timeout (float): The timeout for this request.
326 metadata (Sequence[Tuple[str, str]]): Strings which should be
327 sent along with the request as metadata.
329 Returns:
330 ~.common.GenerateAccessTokenResponse:
332 """
334 http_options: List[Dict[str, str]] = [
335 {
336 "method": "post",
337 "uri": "/v1/{name=projects/*/serviceAccounts/*}:generateAccessToken",
338 "body": "*",
339 },
340 ]
341 request, metadata = self._interceptor.pre_generate_access_token(
342 request, metadata
343 )
344 pb_request = common.GenerateAccessTokenRequest.pb(request)
345 transcoded_request = path_template.transcode(http_options, pb_request)
347 # Jsonify the request body
349 body = json_format.MessageToJson(
350 transcoded_request["body"],
351 including_default_value_fields=False,
352 use_integers_for_enums=True,
353 )
354 uri = transcoded_request["uri"]
355 method = transcoded_request["method"]
357 # Jsonify the query params
358 query_params = json.loads(
359 json_format.MessageToJson(
360 transcoded_request["query_params"],
361 including_default_value_fields=False,
362 use_integers_for_enums=True,
363 )
364 )
365 query_params.update(self._get_unset_required_fields(query_params))
367 query_params["$alt"] = "json;enum-encoding=int"
369 # Send the request
370 headers = dict(metadata)
371 headers["Content-Type"] = "application/json"
372 response = getattr(self._session, method)(
373 "{host}{uri}".format(host=self._host, uri=uri),
374 timeout=timeout,
375 headers=headers,
376 params=rest_helpers.flatten_query_params(query_params, strict=True),
377 data=body,
378 )
380 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
381 # subclass.
382 if response.status_code >= 400:
383 raise core_exceptions.from_http_response(response)
385 # Return the response
386 resp = common.GenerateAccessTokenResponse()
387 pb_resp = common.GenerateAccessTokenResponse.pb(resp)
389 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
390 resp = self._interceptor.post_generate_access_token(resp)
391 return resp
393 class _GenerateIdToken(IAMCredentialsRestStub):
394 def __hash__(self):
395 return hash("GenerateIdToken")
397 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
399 @classmethod
400 def _get_unset_required_fields(cls, message_dict):
401 return {
402 k: v
403 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
404 if k not in message_dict
405 }
407 def __call__(
408 self,
409 request: common.GenerateIdTokenRequest,
410 *,
411 retry: OptionalRetry = gapic_v1.method.DEFAULT,
412 timeout: Optional[float] = None,
413 metadata: Sequence[Tuple[str, str]] = (),
414 ) -> common.GenerateIdTokenResponse:
415 r"""Call the generate id token method over HTTP.
417 Args:
418 request (~.common.GenerateIdTokenRequest):
419 The request object.
420 retry (google.api_core.retry.Retry): Designation of what errors, if any,
421 should be retried.
422 timeout (float): The timeout for this request.
423 metadata (Sequence[Tuple[str, str]]): Strings which should be
424 sent along with the request as metadata.
426 Returns:
427 ~.common.GenerateIdTokenResponse:
429 """
431 http_options: List[Dict[str, str]] = [
432 {
433 "method": "post",
434 "uri": "/v1/{name=projects/*/serviceAccounts/*}:generateIdToken",
435 "body": "*",
436 },
437 ]
438 request, metadata = self._interceptor.pre_generate_id_token(
439 request, metadata
440 )
441 pb_request = common.GenerateIdTokenRequest.pb(request)
442 transcoded_request = path_template.transcode(http_options, pb_request)
444 # Jsonify the request body
446 body = json_format.MessageToJson(
447 transcoded_request["body"],
448 including_default_value_fields=False,
449 use_integers_for_enums=True,
450 )
451 uri = transcoded_request["uri"]
452 method = transcoded_request["method"]
454 # Jsonify the query params
455 query_params = json.loads(
456 json_format.MessageToJson(
457 transcoded_request["query_params"],
458 including_default_value_fields=False,
459 use_integers_for_enums=True,
460 )
461 )
462 query_params.update(self._get_unset_required_fields(query_params))
464 query_params["$alt"] = "json;enum-encoding=int"
466 # Send the request
467 headers = dict(metadata)
468 headers["Content-Type"] = "application/json"
469 response = getattr(self._session, method)(
470 "{host}{uri}".format(host=self._host, uri=uri),
471 timeout=timeout,
472 headers=headers,
473 params=rest_helpers.flatten_query_params(query_params, strict=True),
474 data=body,
475 )
477 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
478 # subclass.
479 if response.status_code >= 400:
480 raise core_exceptions.from_http_response(response)
482 # Return the response
483 resp = common.GenerateIdTokenResponse()
484 pb_resp = common.GenerateIdTokenResponse.pb(resp)
486 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
487 resp = self._interceptor.post_generate_id_token(resp)
488 return resp
490 class _SignBlob(IAMCredentialsRestStub):
491 def __hash__(self):
492 return hash("SignBlob")
494 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
496 @classmethod
497 def _get_unset_required_fields(cls, message_dict):
498 return {
499 k: v
500 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
501 if k not in message_dict
502 }
504 def __call__(
505 self,
506 request: common.SignBlobRequest,
507 *,
508 retry: OptionalRetry = gapic_v1.method.DEFAULT,
509 timeout: Optional[float] = None,
510 metadata: Sequence[Tuple[str, str]] = (),
511 ) -> common.SignBlobResponse:
512 r"""Call the sign blob method over HTTP.
514 Args:
515 request (~.common.SignBlobRequest):
516 The request object.
517 retry (google.api_core.retry.Retry): Designation of what errors, if any,
518 should be retried.
519 timeout (float): The timeout for this request.
520 metadata (Sequence[Tuple[str, str]]): Strings which should be
521 sent along with the request as metadata.
523 Returns:
524 ~.common.SignBlobResponse:
526 """
528 http_options: List[Dict[str, str]] = [
529 {
530 "method": "post",
531 "uri": "/v1/{name=projects/*/serviceAccounts/*}:signBlob",
532 "body": "*",
533 },
534 ]
535 request, metadata = self._interceptor.pre_sign_blob(request, metadata)
536 pb_request = common.SignBlobRequest.pb(request)
537 transcoded_request = path_template.transcode(http_options, pb_request)
539 # Jsonify the request body
541 body = json_format.MessageToJson(
542 transcoded_request["body"],
543 including_default_value_fields=False,
544 use_integers_for_enums=True,
545 )
546 uri = transcoded_request["uri"]
547 method = transcoded_request["method"]
549 # Jsonify the query params
550 query_params = json.loads(
551 json_format.MessageToJson(
552 transcoded_request["query_params"],
553 including_default_value_fields=False,
554 use_integers_for_enums=True,
555 )
556 )
557 query_params.update(self._get_unset_required_fields(query_params))
559 query_params["$alt"] = "json;enum-encoding=int"
561 # Send the request
562 headers = dict(metadata)
563 headers["Content-Type"] = "application/json"
564 response = getattr(self._session, method)(
565 "{host}{uri}".format(host=self._host, uri=uri),
566 timeout=timeout,
567 headers=headers,
568 params=rest_helpers.flatten_query_params(query_params, strict=True),
569 data=body,
570 )
572 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
573 # subclass.
574 if response.status_code >= 400:
575 raise core_exceptions.from_http_response(response)
577 # Return the response
578 resp = common.SignBlobResponse()
579 pb_resp = common.SignBlobResponse.pb(resp)
581 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
582 resp = self._interceptor.post_sign_blob(resp)
583 return resp
585 class _SignJwt(IAMCredentialsRestStub):
586 def __hash__(self):
587 return hash("SignJwt")
589 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
591 @classmethod
592 def _get_unset_required_fields(cls, message_dict):
593 return {
594 k: v
595 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
596 if k not in message_dict
597 }
599 def __call__(
600 self,
601 request: common.SignJwtRequest,
602 *,
603 retry: OptionalRetry = gapic_v1.method.DEFAULT,
604 timeout: Optional[float] = None,
605 metadata: Sequence[Tuple[str, str]] = (),
606 ) -> common.SignJwtResponse:
607 r"""Call the sign jwt method over HTTP.
609 Args:
610 request (~.common.SignJwtRequest):
611 The request object.
612 retry (google.api_core.retry.Retry): Designation of what errors, if any,
613 should be retried.
614 timeout (float): The timeout for this request.
615 metadata (Sequence[Tuple[str, str]]): Strings which should be
616 sent along with the request as metadata.
618 Returns:
619 ~.common.SignJwtResponse:
621 """
623 http_options: List[Dict[str, str]] = [
624 {
625 "method": "post",
626 "uri": "/v1/{name=projects/*/serviceAccounts/*}:signJwt",
627 "body": "*",
628 },
629 ]
630 request, metadata = self._interceptor.pre_sign_jwt(request, metadata)
631 pb_request = common.SignJwtRequest.pb(request)
632 transcoded_request = path_template.transcode(http_options, pb_request)
634 # Jsonify the request body
636 body = json_format.MessageToJson(
637 transcoded_request["body"],
638 including_default_value_fields=False,
639 use_integers_for_enums=True,
640 )
641 uri = transcoded_request["uri"]
642 method = transcoded_request["method"]
644 # Jsonify the query params
645 query_params = json.loads(
646 json_format.MessageToJson(
647 transcoded_request["query_params"],
648 including_default_value_fields=False,
649 use_integers_for_enums=True,
650 )
651 )
652 query_params.update(self._get_unset_required_fields(query_params))
654 query_params["$alt"] = "json;enum-encoding=int"
656 # Send the request
657 headers = dict(metadata)
658 headers["Content-Type"] = "application/json"
659 response = getattr(self._session, method)(
660 "{host}{uri}".format(host=self._host, uri=uri),
661 timeout=timeout,
662 headers=headers,
663 params=rest_helpers.flatten_query_params(query_params, strict=True),
664 data=body,
665 )
667 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
668 # subclass.
669 if response.status_code >= 400:
670 raise core_exceptions.from_http_response(response)
672 # Return the response
673 resp = common.SignJwtResponse()
674 pb_resp = common.SignJwtResponse.pb(resp)
676 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
677 resp = self._interceptor.post_sign_jwt(resp)
678 return resp
680 @property
681 def generate_access_token(
682 self,
683 ) -> Callable[
684 [common.GenerateAccessTokenRequest], common.GenerateAccessTokenResponse
685 ]:
686 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
687 # In C++ this would require a dynamic_cast
688 return self._GenerateAccessToken(self._session, self._host, self._interceptor) # type: ignore
690 @property
691 def generate_id_token(
692 self,
693 ) -> Callable[[common.GenerateIdTokenRequest], common.GenerateIdTokenResponse]:
694 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
695 # In C++ this would require a dynamic_cast
696 return self._GenerateIdToken(self._session, self._host, self._interceptor) # type: ignore
698 @property
699 def sign_blob(self) -> Callable[[common.SignBlobRequest], common.SignBlobResponse]:
700 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
701 # In C++ this would require a dynamic_cast
702 return self._SignBlob(self._session, self._host, self._interceptor) # type: ignore
704 @property
705 def sign_jwt(self) -> Callable[[common.SignJwtRequest], common.SignJwtResponse]:
706 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
707 # In C++ this would require a dynamic_cast
708 return self._SignJwt(self._session, self._host, self._interceptor) # type: ignore
710 @property
711 def kind(self) -> str:
712 return "rest"
714 def close(self):
715 self._session.close()
718__all__ = ("IAMCredentialsRestTransport",)