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# Copyright 2016 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""Authorization support for gRPC."""
17from __future__ import absolute_import
19import logging
20import warnings
22from google.auth import exceptions
23from google.auth.transport import _mtls_helper, mtls
24from google.oauth2 import service_account
26try:
27 import grpc # type: ignore
28except ImportError as caught_exc: # pragma: NO COVER
29 raise ImportError(
30 "gRPC is not installed from please install the grpcio package to use the gRPC transport."
31 ) from caught_exc
34_grpc_ver_str = getattr(grpc, "__version__", None)
35if isinstance(_grpc_ver_str, str):
36 _parts = []
37 for _part in _grpc_ver_str.split("."):
38 try:
39 _parts.append(int(_part))
40 except ValueError:
41 break
42 if _parts and tuple(_parts) < (1, 83, 0):
43 warnings.warn(
44 "grpcio < 1.83.0 does not support Post-Quantum Cryptography (PQC). "
45 "Support for non-PQC environments is deprecated. In April 2027, "
46 "google-auth will raise its minimum requirements "
47 "to enforce grpcio >= 1.83.0. "
48 "For more details on Google Cloud's post-quantum security migration, visit: "
49 "https://cloud.google.com/security/resources/post-quantum-cryptography",
50 FutureWarning,
51 )
53_LOGGER = logging.getLogger(__name__)
56class AuthMetadataPlugin(grpc.AuthMetadataPlugin):
57 """A `gRPC AuthMetadataPlugin`_ that inserts the credentials into each
58 request.
60 .. _gRPC AuthMetadataPlugin:
61 http://www.grpc.io/grpc/python/grpc.html#grpc.AuthMetadataPlugin
63 Args:
64 credentials (google.auth.credentials.Credentials): The credentials to
65 add to requests.
66 request (google.auth.transport.Request): A HTTP transport request
67 object used to refresh credentials as needed.
68 default_host (Optional[str]): A host like "pubsub.googleapis.com".
69 This is used when a self-signed JWT is created from service
70 account credentials.
71 suppress_metrics_header (bool): When enabled, ``x-goog-api-client``
72 will be stripped from authorization headers.
73 """
75 def __init__(
76 self, credentials, request, default_host=None, *, suppress_metrics_header=False
77 ):
78 # pylint: disable=no-value-for-parameter
79 # pylint doesn't realize that the super method takes no arguments
80 # because this class is the same name as the superclass.
81 super(AuthMetadataPlugin, self).__init__()
82 self._credentials = credentials
83 self._request = request
84 self._default_host = default_host
85 self._suppress_metrics_header = suppress_metrics_header
87 def _get_authorization_headers(self, context):
88 """Gets the authorization headers for a request.
90 Returns:
91 Sequence[Tuple[str, str]]: A list of request headers (key, value)
92 to add to the request.
93 """
94 headers = {}
96 # https://google.aip.dev/auth/4111
97 # Attempt to use self-signed JWTs when a service account is used.
98 # A default host must be explicitly provided since it cannot always
99 # be determined from the context.service_url.
100 if isinstance(self._credentials, service_account.Credentials):
101 self._credentials._create_self_signed_jwt(
102 "https://{}/".format(self._default_host) if self._default_host else None
103 )
105 self._credentials.before_request(
106 self._request, context.method_name, context.service_url, headers
107 )
109 if self._suppress_metrics_header and "x-goog-api-client" in headers:
110 del headers["x-goog-api-client"]
112 return list(headers.items())
114 def __call__(self, context, callback):
115 """Passes authorization metadata into the given callback.
117 Args:
118 context (grpc.AuthMetadataContext): The RPC context.
119 callback (grpc.AuthMetadataPluginCallback): The callback that will
120 be invoked to pass in the authorization metadata.
121 """
122 callback(self._get_authorization_headers(context), None)
125def secure_authorized_channel(
126 credentials,
127 request,
128 target,
129 ssl_credentials=None,
130 client_cert_callback=None,
131 **kwargs,
132):
133 """Creates a secure authorized gRPC channel.
135 This creates a channel with SSL and :class:`AuthMetadataPlugin`. This
136 channel can be used to create a stub that can make authorized requests.
137 Users can configure client certificate or rely on device certificates to
138 establish a mutual TLS channel, if the `GOOGLE_API_USE_CLIENT_CERTIFICATE`
139 variable is explicitly set to `true`.
141 Example::
143 import google.auth
144 import google.auth.transport.grpc
145 import google.auth.transport.requests
146 from google.cloud.speech.v1 import cloud_speech_pb2
148 # Get credentials.
149 credentials, _ = google.auth.default()
151 # Get an HTTP request function to refresh credentials.
152 request = google.auth.transport.requests.Request()
154 # Create a channel.
155 channel = google.auth.transport.grpc.secure_authorized_channel(
156 credentials, regular_endpoint, request,
157 ssl_credentials=grpc.ssl_channel_credentials())
159 # Use the channel to create a stub.
160 cloud_speech.create_Speech_stub(channel)
162 Usage:
164 There are actually a couple of options to create a channel, depending on if
165 you want to create a regular or mutual TLS channel.
167 First let's list the endpoints (regular vs mutual TLS) to choose from::
169 regular_endpoint = 'speech.googleapis.com:443'
170 mtls_endpoint = 'speech.mtls.googleapis.com:443'
172 Option 1: create a regular (non-mutual) TLS channel by explicitly setting
173 the ssl_credentials::
175 regular_ssl_credentials = grpc.ssl_channel_credentials()
177 channel = google.auth.transport.grpc.secure_authorized_channel(
178 credentials, request, regular_endpoint,
179 ssl_credentials=regular_ssl_credentials)
181 Option 2: create a mutual TLS channel by calling a callback which returns
182 the client side certificate and the key (Note that
183 `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable must be explicitly
184 set to `true`)::
186 def my_client_cert_callback():
187 code_to_load_client_cert_and_key()
188 if loaded:
189 return (pem_cert_bytes, pem_key_bytes)
190 raise MyClientCertFailureException()
192 try:
193 channel = google.auth.transport.grpc.secure_authorized_channel(
194 credentials, request, mtls_endpoint,
195 client_cert_callback=my_client_cert_callback)
196 except MyClientCertFailureException:
197 # handle the exception
199 Option 3: use application default SSL credentials. It searches and uses
200 the command in a context aware metadata file, which is available on devices
201 with endpoint verification support (Note that
202 `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable must be explicitly
203 set to `true`).
204 See https://cloud.google.com/endpoint-verification/docs/overview::
206 try:
207 default_ssl_credentials = SslCredentials()
208 except:
209 # Exception can be raised if the context aware metadata is malformed.
210 # See :class:`SslCredentials` for the possible exceptions.
212 # Choose the endpoint based on the SSL credentials type.
213 if default_ssl_credentials.is_mtls:
214 endpoint_to_use = mtls_endpoint
215 else:
216 endpoint_to_use = regular_endpoint
217 channel = google.auth.transport.grpc.secure_authorized_channel(
218 credentials, request, endpoint_to_use,
219 ssl_credentials=default_ssl_credentials)
221 Option 4: not setting ssl_credentials and client_cert_callback. For devices
222 without endpoint verification support or `GOOGLE_API_USE_CLIENT_CERTIFICATE`
223 environment variable is not `true`, a regular TLS channel is created;
224 otherwise, a mutual TLS channel is created, however, the call should be
225 wrapped in a try/except block in case of malformed context aware metadata.
227 The following code uses regular_endpoint, it works the same no matter the
228 created channle is regular or mutual TLS. Regular endpoint ignores client
229 certificate and key::
231 channel = google.auth.transport.grpc.secure_authorized_channel(
232 credentials, request, regular_endpoint)
234 The following code uses mtls_endpoint, if the created channle is regular,
235 and API mtls_endpoint is confgured to require client SSL credentials, API
236 calls using this channel will be rejected::
238 channel = google.auth.transport.grpc.secure_authorized_channel(
239 credentials, request, mtls_endpoint)
241 Args:
242 credentials (google.auth.credentials.Credentials): The credentials to
243 add to requests.
244 request (google.auth.transport.Request): A HTTP transport request
245 object used to refresh credentials as needed. Even though gRPC
246 is a separate transport, there's no way to refresh the credentials
247 without using a standard http transport.
248 target (str): The host and port of the service.
249 ssl_credentials (grpc.ChannelCredentials): Optional SSL channel
250 credentials. This can be used to specify different certificates.
251 This argument is mutually exclusive with client_cert_callback;
252 providing both will raise an exception.
253 If ssl_credentials and client_cert_callback are None, application
254 default SSL credentials are used if `GOOGLE_API_USE_CLIENT_CERTIFICATE`
255 environment variable is explicitly set to `true`, otherwise one way TLS
256 SSL credentials are used.
257 client_cert_callback (Callable[[], (bytes, bytes)]): Optional
258 callback function to obtain client certicate and key for mutual TLS
259 connection. This argument is mutually exclusive with
260 ssl_credentials; providing both will raise an exception.
261 This argument does nothing unless `GOOGLE_API_USE_CLIENT_CERTIFICATE`
262 environment variable is explicitly set to `true`.
263 kwargs: Additional arguments to pass to :func:`grpc.secure_channel`.
265 Returns:
266 grpc.Channel: The created gRPC channel.
268 Raises:
269 google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
270 creation failed for any reason.
271 """
272 # Create the metadata plugin for inserting the authorization header.
273 metadata_plugin = AuthMetadataPlugin(credentials, request)
275 # Create a set of grpc.CallCredentials using the metadata plugin.
276 google_auth_credentials = grpc.metadata_call_credentials(metadata_plugin)
278 if ssl_credentials and client_cert_callback:
279 raise exceptions.MalformedError(
280 "Received both ssl_credentials and client_cert_callback; "
281 "these are mutually exclusive."
282 )
284 # If SSL credentials are not explicitly set, try client_cert_callback and ADC.
285 if not ssl_credentials:
286 use_client_cert = _mtls_helper.check_use_client_cert()
287 if use_client_cert and client_cert_callback:
288 # Use the callback if provided.
289 cert, key = client_cert_callback()
290 ssl_credentials = grpc.ssl_channel_credentials(
291 certificate_chain=cert, private_key=key
292 )
293 elif use_client_cert:
294 # Use application default SSL credentials.
295 adc_ssl_credentils = SslCredentials()
296 ssl_credentials = adc_ssl_credentils.ssl_credentials
297 else:
298 ssl_credentials = grpc.ssl_channel_credentials()
300 # Combine the ssl credentials and the authorization credentials.
301 composite_credentials = grpc.composite_channel_credentials(
302 ssl_credentials, google_auth_credentials
303 )
305 return grpc.secure_channel(target, composite_credentials, **kwargs)
308class SslCredentials:
309 """Class for application default SSL credentials.
311 Mutual TLS (mTLS) is enabled if either:
313 1. The `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is explicitly
314 set to `"true"`.
315 2. The `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is unset or empty,
316 but a valid workload certificate configuration is found (e.g., via the
317 `GOOGLE_API_CERTIFICATE_CONFIG` environment variable or the default gcloud config path).
319 See https://google.aip.dev/auth/4114 for client certificate discovery details.
321 If client certificate usage is enabled, then for devices with endpoint
322 verification support, a device certificate will be automatically loaded and
323 mutual TLS will be established.
324 See https://cloud.google.com/endpoint-verification/docs/overview.
325 """
327 def __init__(self):
328 use_client_cert = _mtls_helper.check_use_client_cert()
329 if not use_client_cert:
330 self._is_mtls = False
331 else:
332 self._is_mtls = mtls.has_default_client_cert_source()
334 @property
335 def ssl_credentials(self):
336 """Get the created SSL channel credentials.
338 For devices with endpoint verification support, if the device certificate
339 loading has any problems, corresponding exceptions will be raised. For
340 a device without endpoint verification support, no exceptions will be
341 raised.
343 Returns:
344 grpc.ChannelCredentials: The created grpc channel credentials.
346 Raises:
347 google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
348 creation failed for any reason.
349 """
350 if self._is_mtls:
351 try:
352 has_cert, cert, key, _ = _mtls_helper.get_client_ssl_credentials()
353 if has_cert:
354 self._ssl_credentials = grpc.ssl_channel_credentials(
355 certificate_chain=cert, private_key=key
356 )
357 else:
358 self._ssl_credentials = grpc.ssl_channel_credentials()
359 self._is_mtls = False
360 except (exceptions.ClientCertError, OSError) as caught_exc:
361 new_exc = exceptions.MutualTLSChannelError(caught_exc)
362 raise new_exc from caught_exc
363 else:
364 self._ssl_credentials = grpc.ssl_channel_credentials()
366 return self._ssl_credentials
368 @property
369 def is_mtls(self):
370 """Indicates if the created SSL channel credentials is mutual TLS."""
371 return self._is_mtls