« 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#
16from typing import Callable, Dict, Optional, Sequence, Tuple, Union
17import warnings
19from google.api_core import gapic_v1, grpc_helpers
20import google.auth # type: ignore
21from google.auth import credentials as ga_credentials # type: ignore
22from google.auth.transport.grpc import SslCredentials # type: ignore
23import grpc # type: ignore
25from google.cloud.iam_credentials_v1.types import common
27from .base import DEFAULT_CLIENT_INFO, IAMCredentialsTransport
30class IAMCredentialsGrpcTransport(IAMCredentialsTransport):
31 """gRPC backend transport for IAMCredentials.
33 A service account is a special type of Google account that
34 belongs to your application or a virtual machine (VM), instead
35 of to an individual end user. Your application assumes the
36 identity of the service account to call Google APIs, so that the
37 users aren't directly involved.
39 Service account credentials are used to temporarily assume the
40 identity of the service account. Supported credential types
41 include OAuth 2.0 access tokens, OpenID Connect ID tokens,
42 self-signed JSON Web Tokens (JWTs), and more.
44 This class defines the same methods as the primary client, so the
45 primary client can load the underlying transport implementation
46 and call it.
48 It sends protocol buffers over the wire using gRPC (which is built on
49 top of HTTP/2); the ``grpcio`` package must be installed.
50 """
52 _stubs: Dict[str, Callable]
54 def __init__(
55 self,
56 *,
57 host: str = "iamcredentials.googleapis.com",
58 credentials: Optional[ga_credentials.Credentials] = None,
59 credentials_file: Optional[str] = None,
60 scopes: Optional[Sequence[str]] = None,
61 channel: Optional[grpc.Channel] = None,
62 api_mtls_endpoint: Optional[str] = None,
63 client_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
64 ssl_channel_credentials: Optional[grpc.ChannelCredentials] = None,
65 client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
66 quota_project_id: Optional[str] = None,
67 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
68 always_use_jwt_access: Optional[bool] = False,
69 api_audience: Optional[str] = None,
70 ) -> None:
71 """Instantiate the transport.
73 Args:
74 host (Optional[str]):
75 The hostname to connect to.
76 credentials (Optional[google.auth.credentials.Credentials]): The
77 authorization credentials to attach to requests. These
78 credentials identify the application to the service; if none
79 are specified, the client will attempt to ascertain the
80 credentials from the environment.
81 This argument is ignored if ``channel`` is provided.
82 credentials_file (Optional[str]): A file with credentials that can
83 be loaded with :func:`google.auth.load_credentials_from_file`.
84 This argument is ignored if ``channel`` is provided.
85 scopes (Optional(Sequence[str])): A list of scopes. This argument is
86 ignored if ``channel`` is provided.
87 channel (Optional[grpc.Channel]): A ``Channel`` instance through
88 which to make calls.
89 api_mtls_endpoint (Optional[str]): Deprecated. The mutual TLS endpoint.
90 If provided, it overrides the ``host`` argument and tries to create
91 a mutual TLS channel with client SSL credentials from
92 ``client_cert_source`` or application default SSL credentials.
93 client_cert_source (Optional[Callable[[], Tuple[bytes, bytes]]]):
94 Deprecated. A callback to provide client SSL certificate bytes and
95 private key bytes, both in PEM format. It is ignored if
96 ``api_mtls_endpoint`` is None.
97 ssl_channel_credentials (grpc.ChannelCredentials): SSL credentials
98 for the grpc channel. It is ignored if ``channel`` is provided.
99 client_cert_source_for_mtls (Optional[Callable[[], Tuple[bytes, bytes]]]):
100 A callback to provide client certificate bytes and private key bytes,
101 both in PEM format. It is used to configure a mutual TLS channel. It is
102 ignored if ``channel`` or ``ssl_channel_credentials`` is provided.
103 quota_project_id (Optional[str]): An optional project to use for billing
104 and quota.
105 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
106 The client info used to send a user-agent string along with
107 API requests. If ``None``, then default info will be used.
108 Generally, you only need to set this if you're developing
109 your own client library.
110 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
111 be used for service account credentials.
113 Raises:
114 google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
115 creation failed for any reason.
116 google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
117 and ``credentials_file`` are passed.
118 """
119 self._grpc_channel = None
120 self._ssl_channel_credentials = ssl_channel_credentials
121 self._stubs: Dict[str, Callable] = {}
123 if api_mtls_endpoint:
124 warnings.warn("api_mtls_endpoint is deprecated", DeprecationWarning)
125 if client_cert_source:
126 warnings.warn("client_cert_source is deprecated", DeprecationWarning)
128 if channel:
129 # Ignore credentials if a channel was passed.
130 credentials = False
131 # If a channel was explicitly provided, set it.
132 self._grpc_channel = channel
133 self._ssl_channel_credentials = None
135 else:
136 if api_mtls_endpoint:
137 host = api_mtls_endpoint
139 # Create SSL credentials with client_cert_source or application
140 # default SSL credentials.
141 if client_cert_source:
142 cert, key = client_cert_source()
143 self._ssl_channel_credentials = grpc.ssl_channel_credentials(
144 certificate_chain=cert, private_key=key
145 )
146 else:
147 self._ssl_channel_credentials = SslCredentials().ssl_credentials
149 else:
150 if client_cert_source_for_mtls and not ssl_channel_credentials:
151 cert, key = client_cert_source_for_mtls()
152 self._ssl_channel_credentials = grpc.ssl_channel_credentials(
153 certificate_chain=cert, private_key=key
154 )
156 # The base transport sets the host, credentials and scopes
157 super().__init__(
158 host=host,
159 credentials=credentials,
160 credentials_file=credentials_file,
161 scopes=scopes,
162 quota_project_id=quota_project_id,
163 client_info=client_info,
164 always_use_jwt_access=always_use_jwt_access,
165 api_audience=api_audience,
166 )
168 if not self._grpc_channel:
169 self._grpc_channel = type(self).create_channel(
170 self._host,
171 # use the credentials which are saved
172 credentials=self._credentials,
173 # Set ``credentials_file`` to ``None`` here as
174 # the credentials that we saved earlier should be used.
175 credentials_file=None,
176 scopes=self._scopes,
177 ssl_credentials=self._ssl_channel_credentials,
178 quota_project_id=quota_project_id,
179 options=[
180 ("grpc.max_send_message_length", -1),
181 ("grpc.max_receive_message_length", -1),
182 ],
183 )
185 # Wrap messages. This must be done after self._grpc_channel exists
186 self._prep_wrapped_messages(client_info)
188 @classmethod
189 def create_channel(
190 cls,
191 host: str = "iamcredentials.googleapis.com",
192 credentials: Optional[ga_credentials.Credentials] = None,
193 credentials_file: Optional[str] = None,
194 scopes: Optional[Sequence[str]] = None,
195 quota_project_id: Optional[str] = None,
196 **kwargs,
197 ) -> grpc.Channel:
198 """Create and return a gRPC channel object.
199 Args:
200 host (Optional[str]): The host for the channel to use.
201 credentials (Optional[~.Credentials]): The
202 authorization credentials to attach to requests. These
203 credentials identify this application to the service. If
204 none are specified, the client will attempt to ascertain
205 the credentials from the environment.
206 credentials_file (Optional[str]): A file with credentials that can
207 be loaded with :func:`google.auth.load_credentials_from_file`.
208 This argument is mutually exclusive with credentials.
209 scopes (Optional[Sequence[str]]): A optional list of scopes needed for this
210 service. These are only used when credentials are not specified and
211 are passed to :func:`google.auth.default`.
212 quota_project_id (Optional[str]): An optional project to use for billing
213 and quota.
214 kwargs (Optional[dict]): Keyword arguments, which are passed to the
215 channel creation.
216 Returns:
217 grpc.Channel: A gRPC channel object.
219 Raises:
220 google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
221 and ``credentials_file`` are passed.
222 """
224 return grpc_helpers.create_channel(
225 host,
226 credentials=credentials,
227 credentials_file=credentials_file,
228 quota_project_id=quota_project_id,
229 default_scopes=cls.AUTH_SCOPES,
230 scopes=scopes,
231 default_host=cls.DEFAULT_HOST,
232 **kwargs,
233 )
235 @property
236 def grpc_channel(self) -> grpc.Channel:
237 """Return the channel designed to connect to this service."""
238 return self._grpc_channel
240 @property
241 def generate_access_token(
242 self,
243 ) -> Callable[
244 [common.GenerateAccessTokenRequest], common.GenerateAccessTokenResponse
245 ]:
246 r"""Return a callable for the generate access token method over gRPC.
248 Generates an OAuth 2.0 access token for a service
249 account.
251 Returns:
252 Callable[[~.GenerateAccessTokenRequest],
253 ~.GenerateAccessTokenResponse]:
254 A function that, when called, will call the underlying RPC
255 on the server.
256 """
257 # Generate a "stub function" on-the-fly which will actually make
258 # the request.
259 # gRPC handles serialization and deserialization, so we just need
260 # to pass in the functions for each.
261 if "generate_access_token" not in self._stubs:
262 self._stubs["generate_access_token"] = self.grpc_channel.unary_unary(
263 "/google.iam.credentials.v1.IAMCredentials/GenerateAccessToken",
264 request_serializer=common.GenerateAccessTokenRequest.serialize,
265 response_deserializer=common.GenerateAccessTokenResponse.deserialize,
266 )
267 return self._stubs["generate_access_token"]
269 @property
270 def generate_id_token(
271 self,
272 ) -> Callable[[common.GenerateIdTokenRequest], common.GenerateIdTokenResponse]:
273 r"""Return a callable for the generate id token method over gRPC.
275 Generates an OpenID Connect ID token for a service
276 account.
278 Returns:
279 Callable[[~.GenerateIdTokenRequest],
280 ~.GenerateIdTokenResponse]:
281 A function that, when called, will call the underlying RPC
282 on the server.
283 """
284 # Generate a "stub function" on-the-fly which will actually make
285 # the request.
286 # gRPC handles serialization and deserialization, so we just need
287 # to pass in the functions for each.
288 if "generate_id_token" not in self._stubs:
289 self._stubs["generate_id_token"] = self.grpc_channel.unary_unary(
290 "/google.iam.credentials.v1.IAMCredentials/GenerateIdToken",
291 request_serializer=common.GenerateIdTokenRequest.serialize,
292 response_deserializer=common.GenerateIdTokenResponse.deserialize,
293 )
294 return self._stubs["generate_id_token"]
296 @property
297 def sign_blob(self) -> Callable[[common.SignBlobRequest], common.SignBlobResponse]:
298 r"""Return a callable for the sign blob method over gRPC.
300 Signs a blob using a service account's system-managed
301 private key.
303 Returns:
304 Callable[[~.SignBlobRequest],
305 ~.SignBlobResponse]:
306 A function that, when called, will call the underlying RPC
307 on the server.
308 """
309 # Generate a "stub function" on-the-fly which will actually make
310 # the request.
311 # gRPC handles serialization and deserialization, so we just need
312 # to pass in the functions for each.
313 if "sign_blob" not in self._stubs:
314 self._stubs["sign_blob"] = self.grpc_channel.unary_unary(
315 "/google.iam.credentials.v1.IAMCredentials/SignBlob",
316 request_serializer=common.SignBlobRequest.serialize,
317 response_deserializer=common.SignBlobResponse.deserialize,
318 )
319 return self._stubs["sign_blob"]
321 @property
322 def sign_jwt(self) -> Callable[[common.SignJwtRequest], common.SignJwtResponse]:
323 r"""Return a callable for the sign jwt method over gRPC.
325 Signs a JWT using a service account's system-managed
326 private key.
328 Returns:
329 Callable[[~.SignJwtRequest],
330 ~.SignJwtResponse]:
331 A function that, when called, will call the underlying RPC
332 on the server.
333 """
334 # Generate a "stub function" on-the-fly which will actually make
335 # the request.
336 # gRPC handles serialization and deserialization, so we just need
337 # to pass in the functions for each.
338 if "sign_jwt" not in self._stubs:
339 self._stubs["sign_jwt"] = self.grpc_channel.unary_unary(
340 "/google.iam.credentials.v1.IAMCredentials/SignJwt",
341 request_serializer=common.SignJwtRequest.serialize,
342 response_deserializer=common.SignJwtResponse.deserialize,
343 )
344 return self._stubs["sign_jwt"]
346 def close(self):
347 self.grpc_channel.close()
349 @property
350 def kind(self) -> str:
351 return "grpc"
354__all__ = ("IAMCredentialsGrpcTransport",)