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"""Transport adapter for Requests."""
17from __future__ import absolute_import
19import collections
20import copy
21import functools
22import http.client as http_client
23import logging
24import numbers
25import threading
26import time
27import urllib.parse as urllib_parse
28from typing import Optional
30try:
31 import requests
32except ImportError as caught_exc: # pragma: NO COVER
33 raise ImportError(
34 "The requests library is not installed from please install the requests package to use the requests transport."
35 ) from caught_exc
36import requests.adapters # pylint: disable=ungrouped-imports
37import requests.exceptions # pylint: disable=ungrouped-imports
38from requests.packages.urllib3.util.ssl_ import ( # type: ignore
39 create_urllib3_context,
40) # pylint: disable=ungrouped-imports
42import google.auth.transport._mtls_helper
43from google.auth import _helpers, exceptions, transport
44from google.auth.transport import _mtls_helper
45from google.oauth2 import service_account
47_LOGGER = logging.getLogger(__name__)
49_DEFAULT_TIMEOUT = 120 # in seconds
52class _Response(transport.Response):
53 """Requests transport response adapter.
55 Args:
56 response (requests.Response): The raw Requests response.
57 """
59 def __init__(self, response):
60 self._response = response
62 @property
63 def status(self):
64 return self._response.status_code
66 @property
67 def headers(self):
68 return self._response.headers
70 @property
71 def data(self):
72 return self._response.content
75class TimeoutGuard(object):
76 """A context manager raising an error if the suite execution took too long.
78 Args:
79 timeout (Union[None, Union[float, Tuple[float, float]]]):
80 The maximum number of seconds a suite can run without the context
81 manager raising a timeout exception on exit. If passed as a tuple,
82 the smaller of the values is taken as a timeout. If ``None``, a
83 timeout error is never raised.
84 timeout_error_type (Optional[Exception]):
85 The type of the error to raise on timeout. Defaults to
86 :class:`requests.exceptions.Timeout`.
87 """
89 def __init__(self, timeout, timeout_error_type=requests.exceptions.Timeout):
90 self._timeout = timeout
91 self.remaining_timeout = timeout
92 self._timeout_error_type = timeout_error_type
94 def __enter__(self):
95 self._start = time.time()
96 return self
98 def __exit__(self, exc_type, exc_value, traceback):
99 if exc_value:
100 return # let the error bubble up automatically
102 if self._timeout is None:
103 return # nothing to do, the timeout was not specified
105 elapsed = time.time() - self._start
106 deadline_hit = False
108 if isinstance(self._timeout, numbers.Number):
109 self.remaining_timeout = self._timeout - elapsed
110 deadline_hit = self.remaining_timeout <= 0
111 else:
112 self.remaining_timeout = tuple(x - elapsed for x in self._timeout)
113 deadline_hit = min(self.remaining_timeout) <= 0
115 if deadline_hit:
116 raise self._timeout_error_type()
119def _replace_adapter(session, prefix, adapter):
120 """Mounts an adapter on a session without mutating its adapters in place.
122 ``adapter`` is mounted on ``session`` for ``prefix``.
123 :meth:`requests.Session.mount` mutates the ``session.adapters``
124 OrderedDict in place. If another thread is iterating over it in
125 :meth:`requests.Session.get_adapter` at the same time, that thread fails
126 with ``RuntimeError: OrderedDict mutated during iteration``. This builds
127 an updated copy and assigns it in one step instead, so threads that are
128 already iterating over the old dictionary are not affected.
130 Args:
131 session (requests.Session): The session to mount the adapter on.
132 prefix (str): The URL prefix to mount the adapter for.
133 adapter (requests.adapters.BaseAdapter): The adapter to mount.
134 """
135 new_adapters = session.adapters.copy()
136 new_adapters[prefix] = adapter
137 # Keep the longest prefixes first, as requests.Session.mount() does, so
138 # that get_adapter() matches the most specific prefix.
139 session.adapters = collections.OrderedDict(
140 sorted(new_adapters.items(), key=lambda item: len(item[0]), reverse=True)
141 )
144class Request(transport.Request):
145 """Requests request adapter.
147 This class is used internally for making requests using various transports
148 in a consistent way. If you use :class:`AuthorizedSession` you do not need
149 to construct or use this class directly.
151 This class can be useful if you want to manually refresh a
152 :class:`~google.auth.credentials.Credentials` instance::
154 import google.auth.transport.requests
155 import requests
157 request = google.auth.transport.requests.Request()
159 credentials.refresh(request)
161 Args:
162 session (requests.Session): An instance :class:`requests.Session` used
163 to make HTTP requests. If not specified, a session will be created.
165 .. automethod:: __call__
166 """
168 _mtls_adapter = None
169 _mtls_unavailable = False
171 def __init__(self, session: Optional[requests.Session] = None) -> None:
172 if not session:
173 session = requests.Session()
175 self.session = session
176 self._mtls_adapter = None
177 self._mtls_unavailable = False
178 self._mtls_lock = threading.Lock()
180 def __getstate__(self):
181 state = self.__dict__.copy()
182 state["_mtls_lock"] = None
183 return state
185 def __setstate__(self, state):
186 self.__dict__.update(state)
187 self._mtls_lock = threading.Lock()
188 if not isinstance(self._mtls_adapter, _MutualTlsAdapter):
189 self._mtls_adapter = None
191 def __del__(self):
192 try:
193 if hasattr(self, "session") and self.session is not None:
194 self.session.close()
195 except TypeError:
196 # NOTE: For certain Python binary built, the queue.Empty exception
197 # might not be considered a normal Python exception causing
198 # TypeError.
199 pass
201 def _configure_mtls_if_needed(self, url):
202 """Lazily mounts a mutual TLS adapter onto the session for mTLS endpoints.
204 Args:
205 url (str): The target request URL.
206 """
207 if not _mtls_helper.is_mtls_endpoint(url):
208 return
210 # Mount on the specific origin rather than globally on "https://" so client
211 # certificates are not sent to other hosts sharing this session. The trailing
212 # slash prevents requests' startswith() matching from matching lookalike domains.
213 parsed = urllib_parse.urlparse(url)
214 prefix = f"{parsed.scheme}://{parsed.netloc.lower()}/"
215 with self._mtls_lock:
216 session_adapter = self.session.get_adapter(url if parsed.path else prefix)
217 # Skip if _MutualTlsAdapter is already mounted or the caller mounted a custom adapter.
218 if (
219 type(session_adapter) is not requests.adapters.HTTPAdapter
220 or self.session.get_adapter(prefix) is not session_adapter
221 ):
222 return
224 if self._mtls_adapter is None:
225 if self._mtls_unavailable or not _mtls_helper.check_use_client_cert():
226 return
227 has_cert, cert, key = _mtls_helper.get_client_cert_and_key()
228 if not has_cert:
229 self._mtls_unavailable = True
230 return
232 self._mtls_adapter = _MutualTlsAdapter(
233 cert,
234 key,
235 max_retries=session_adapter.max_retries,
236 pool_connections=session_adapter._pool_connections,
237 pool_maxsize=session_adapter._pool_maxsize,
238 pool_block=session_adapter._pool_block,
239 )
241 _replace_adapter(self.session, prefix, self._mtls_adapter)
243 def __call__(
244 self,
245 url,
246 method="GET",
247 body=None,
248 headers=None,
249 timeout=_DEFAULT_TIMEOUT,
250 **kwargs,
251 ):
252 """Make an HTTP request using requests.
254 Args:
255 url (str): The URI to be requested.
256 method (str): The HTTP method to use for the request. Defaults
257 to 'GET'.
258 body (bytes): The payload or body in HTTP request.
259 headers (Mapping[str, str]): Request headers.
260 timeout (Optional[int]): The number of seconds to wait for a
261 response from the server. If not specified or if None, the
262 requests default timeout will be used.
263 kwargs: Additional arguments passed through to the underlying
264 requests :meth:`~requests.Session.request` method.
266 Returns:
267 google.auth.transport.Response: The HTTP response.
269 Raises:
270 google.auth.exceptions.TransportError: If any exception occurred.
271 """
272 if not kwargs.get("cert"):
273 try:
274 self._configure_mtls_if_needed(url)
275 except (
276 exceptions.MutualTLSChannelError,
277 exceptions.ClientCertError,
278 OSError,
279 ) as caught_exc:
280 new_exc = exceptions.TransportError(caught_exc)
281 raise new_exc from caught_exc
283 # Missing CA bundles or cert paths raised by requests must bubble up
284 # unwrapped to fail immediately rather than triggering client retries.
285 try:
286 _helpers.request_log(_LOGGER, method, url, body, headers)
287 response = self.session.request(
288 method, url, data=body, headers=headers, timeout=timeout, **kwargs
289 )
290 _helpers.response_log(_LOGGER, response)
291 return _Response(response)
292 except requests.exceptions.RequestException as caught_exc:
293 new_exc = exceptions.TransportError(caught_exc)
294 raise new_exc from caught_exc
297class _MutualTlsAdapter(requests.adapters.HTTPAdapter):
298 """
299 A TransportAdapter that enables mutual TLS.
301 Args:
302 cert (bytes): client certificate in PEM format
303 key (bytes): client private key in PEM format
305 Raises:
306 ImportError: if certifi is not installed
307 google.auth.exceptions.MutualTLSChannelError: If the cert or key is invalid.
308 """
310 def __init__(self, cert, key, **kwargs):
311 import ssl
313 import certifi
315 ctx_poolmanager = create_urllib3_context()
316 ctx_poolmanager.check_hostname = False
317 ctx_poolmanager.load_verify_locations(cafile=certifi.where())
319 ctx_proxymanager = create_urllib3_context()
320 ctx_proxymanager.check_hostname = False
321 ctx_proxymanager.load_verify_locations(cafile=certifi.where())
323 try:
324 with _mtls_helper.secure_cert_key_paths(cert, key) as (
325 cert_path,
326 key_path,
327 passphrase,
328 ):
329 password = passphrase
330 ctx_poolmanager.load_cert_chain(
331 certfile=cert_path,
332 keyfile=key_path,
333 password=password,
334 )
335 ctx_proxymanager.load_cert_chain(
336 certfile=cert_path,
337 keyfile=key_path,
338 password=password,
339 )
340 except (
341 ssl.SSLError,
342 OSError,
343 IOError,
344 ValueError,
345 RuntimeError,
346 TypeError,
347 ) as exc:
348 raise exceptions.MutualTLSChannelError(
349 "Failed to configure client certificate and key for mTLS."
350 ) from exc
352 self._ctx_poolmanager = ctx_poolmanager
353 self._ctx_proxymanager = ctx_proxymanager
355 super(_MutualTlsAdapter, self).__init__(**kwargs)
357 def __deepcopy__(self, memo):
358 new_adapter = object.__new__(type(self))
359 memo[id(self)] = new_adapter
360 new_adapter._ctx_poolmanager = self._ctx_poolmanager
361 new_adapter._ctx_proxymanager = self._ctx_proxymanager
362 new_adapter.__setstate__(copy.deepcopy(self.__getstate__(), memo))
363 return new_adapter
365 def __reduce_ex__(self, protocol):
366 return (requests.adapters.HTTPAdapter, (), self.__getstate__())
368 def init_poolmanager(self, *args, **kwargs):
369 kwargs["ssl_context"] = self._ctx_poolmanager
370 super(_MutualTlsAdapter, self).init_poolmanager(*args, **kwargs)
372 def proxy_manager_for(self, *args, **kwargs):
373 kwargs["ssl_context"] = self._ctx_proxymanager
374 return super(_MutualTlsAdapter, self).proxy_manager_for(*args, **kwargs)
377class _MutualTlsOffloadAdapter(requests.adapters.HTTPAdapter):
378 """
379 A TransportAdapter that enables mutual TLS and offloads the client side
380 signing operation to the signing library.
382 Args:
383 enterprise_cert_file_path (str): the path to a enterprise cert JSON
384 file. The file should contain the following field:
386 {
387 "libs": {
388 "signer_library": "...",
389 "offload_library": "..."
390 }
391 }
393 Raises:
394 ImportError: if certifi is not installed
395 google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
396 creation failed for any reason.
397 """
399 def __init__(self, enterprise_cert_file_path):
400 import certifi
402 from google.auth.transport import _custom_tls_signer
404 self.signer = _custom_tls_signer.CustomTlsSigner(enterprise_cert_file_path)
405 self.signer.load_libraries()
407 poolmanager = create_urllib3_context()
408 poolmanager.load_verify_locations(cafile=certifi.where())
409 self.signer.attach_to_ssl_context(poolmanager)
410 self._ctx_poolmanager = poolmanager
412 proxymanager = create_urllib3_context()
413 proxymanager.load_verify_locations(cafile=certifi.where())
414 self.signer.attach_to_ssl_context(proxymanager)
415 self._ctx_proxymanager = proxymanager
417 super(_MutualTlsOffloadAdapter, self).__init__()
419 def init_poolmanager(self, *args, **kwargs):
420 kwargs["ssl_context"] = self._ctx_poolmanager
421 super(_MutualTlsOffloadAdapter, self).init_poolmanager(*args, **kwargs)
423 def proxy_manager_for(self, *args, **kwargs):
424 kwargs["ssl_context"] = self._ctx_proxymanager
425 return super(_MutualTlsOffloadAdapter, self).proxy_manager_for(*args, **kwargs)
428class AuthorizedSession(requests.Session):
429 """A Requests Session class with credentials.
431 This class is used to perform requests to API endpoints that require
432 authorization::
434 from google.auth.transport.requests import AuthorizedSession
436 authed_session = AuthorizedSession(credentials)
438 response = authed_session.request(
439 'GET', 'https://www.googleapis.com/storage/v1/b')
442 The underlying :meth:`request` implementation handles adding the
443 credentials' headers to the request and refreshing credentials as needed.
445 This class also supports mutual TLS via :meth:`configure_mtls_channel`
446 method. In order to use this method, the `GOOGLE_API_USE_CLIENT_CERTIFICATE`
447 environment variable must be explicitly set to ``true``, otherwise it does
448 nothing. Assume the environment is set to ``true``, the method behaves in the
449 following manner:
451 If client_cert_callback is provided, client certificate and private
452 key are loaded using the callback; if client_cert_callback is None,
453 application default SSL credentials will be used. Exceptions are raised if
454 there are problems with the certificate, private key, or the loading process,
455 so it should be called within a try/except block.
457 First we set the environment variable to ``true``, then create an :class:`AuthorizedSession`
458 instance and specify the endpoints::
460 regular_endpoint = 'https://pubsub.googleapis.com/v1/projects/{my_project_id}/topics'
461 mtls_endpoint = 'https://pubsub.mtls.googleapis.com/v1/projects/{my_project_id}/topics'
463 authed_session = AuthorizedSession(credentials)
465 Now we can pass a callback to :meth:`configure_mtls_channel`::
467 def my_cert_callback():
468 # some code to load client cert bytes and private key bytes, both in
469 # PEM format.
470 some_code_to_load_client_cert_and_key()
471 if loaded:
472 return cert, key
473 raise MyClientCertFailureException()
475 # Always call configure_mtls_channel within a try/except block.
476 try:
477 authed_session.configure_mtls_channel(my_cert_callback)
478 except:
479 # handle exceptions.
481 if authed_session.is_mtls:
482 response = authed_session.request('GET', mtls_endpoint)
483 else:
484 response = authed_session.request('GET', regular_endpoint)
487 You can alternatively use application default SSL credentials like this::
489 try:
490 authed_session.configure_mtls_channel()
491 except:
492 # handle exceptions.
494 Args:
495 credentials (google.auth.credentials.Credentials): The credentials to
496 add to the request.
497 refresh_status_codes (Sequence[int]): Which HTTP status codes indicate
498 that credentials should be refreshed and the request should be
499 retried.
500 max_refresh_attempts (int): The maximum number of times to attempt to
501 refresh the credentials and retry the request.
502 refresh_timeout (Optional[int]): The timeout value in seconds for
503 credential refresh HTTP requests.
504 auth_request (google.auth.transport.requests.Request):
505 (Optional) An instance of
506 :class:`~google.auth.transport.requests.Request` used when
507 refreshing credentials. If not passed,
508 an instance of :class:`~google.auth.transport.requests.Request`
509 is created.
510 default_host (Optional[str]): A host like "pubsub.googleapis.com".
511 This is used when a self-signed JWT is created from service
512 account credentials.
513 """
515 def __init__(
516 self,
517 credentials,
518 refresh_status_codes=transport.DEFAULT_REFRESH_STATUS_CODES,
519 max_refresh_attempts=transport.DEFAULT_MAX_REFRESH_ATTEMPTS,
520 refresh_timeout=None,
521 auth_request=None,
522 default_host=None,
523 ):
524 super(AuthorizedSession, self).__init__()
525 self.credentials = credentials
526 self._refresh_status_codes = refresh_status_codes
527 self._max_refresh_attempts = max_refresh_attempts
528 self._refresh_timeout = refresh_timeout
529 self._is_mtls = False
530 self._default_host = default_host
531 self._mtls_reauth_lock = threading.Lock()
533 if auth_request is None:
534 self._auth_request_session = requests.Session()
536 # Using an adapter to make HTTP requests robust to network errors.
537 # This adapter retrys HTTP requests when network errors occur
538 # and the requests seems safely retryable.
539 retry_adapter = requests.adapters.HTTPAdapter(max_retries=3)
540 self._auth_request_session.mount("https://", retry_adapter)
542 # Do not pass `self` as the session here, as it can lead to
543 # infinite recursion.
544 auth_request = Request(self._auth_request_session)
545 else:
546 self._auth_request_session = None
548 # Request instance used by internal methods (for example,
549 # credentials.refresh).
550 self._auth_request = auth_request
552 # https://google.aip.dev/auth/4111
553 # Attempt to use self-signed JWTs when a service account is used.
554 if isinstance(self.credentials, service_account.Credentials):
555 self.credentials._create_self_signed_jwt(
556 "https://{}/".format(self._default_host) if self._default_host else None
557 )
559 def configure_mtls_channel(self, client_cert_callback=None):
560 """Configure the client certificate and key for SSL connection.
562 This method configures mTLS if client certificates are explicitly enabled
563 (via GOOGLE_API_USE_CLIENT_CERTIFICATE=true) or auto-enabled (when the env
564 variable is unset and workload certificates are discovered). In these cases,
565 if the client certificate and key are successfully obtained, a
566 :class:`_MutualTlsAdapter` instance will be mounted to the "https://" prefix.
568 Args:
569 client_cert_callback (Optional[Callable[[], (bytes, bytes)]]):
570 The optional callback returns the client certificate and private
571 key bytes both in PEM format.
572 If the callback is None, application default SSL credentials
573 will be used.
575 .. warning::
576 Calling this method replaces the underlying `requests.Session`
577 adapters and closes the previous ones. It is not thread-safe to call
578 this explicitly while other threads are making requests.
580 Raises:
581 google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
582 creation failed for any reason. The existing session state (such
583 as adapter mounts) remains unmodified if this error is raised.
584 """
585 use_client_cert = google.auth.transport._mtls_helper.check_use_client_cert()
586 if not use_client_cert:
587 return
589 try:
590 (
591 is_mtls,
592 cert,
593 key,
594 ) = google.auth.transport._mtls_helper.get_client_cert_and_key(
595 client_cert_callback
596 )
598 old_adapter = self.adapters.get("https://")
600 kwargs = {}
601 if old_adapter is not None:
602 kwargs["max_retries"] = getattr(old_adapter, "max_retries", 0)
603 kwargs["pool_connections"] = getattr(
604 old_adapter, "_pool_connections", requests.adapters.DEFAULT_POOLSIZE
605 )
606 kwargs["pool_maxsize"] = getattr(
607 old_adapter, "_pool_maxsize", requests.adapters.DEFAULT_POOLSIZE
608 )
609 kwargs["pool_block"] = getattr(
610 old_adapter, "_pool_block", requests.adapters.DEFAULT_POOLBLOCK
611 )
613 old_auth_adapter = None
614 auth_kwargs = {}
615 if self._auth_request_session is not None:
616 old_auth_adapter = self._auth_request_session.adapters.get("https://")
618 if old_auth_adapter is not None:
619 auth_kwargs["max_retries"] = getattr(
620 old_auth_adapter, "max_retries", 0
621 )
622 auth_kwargs["pool_connections"] = getattr(
623 old_auth_adapter,
624 "_pool_connections",
625 requests.adapters.DEFAULT_POOLSIZE,
626 )
627 auth_kwargs["pool_maxsize"] = getattr(
628 old_auth_adapter,
629 "_pool_maxsize",
630 requests.adapters.DEFAULT_POOLSIZE,
631 )
632 auth_kwargs["pool_block"] = getattr(
633 old_auth_adapter,
634 "_pool_block",
635 requests.adapters.DEFAULT_POOLBLOCK,
636 )
638 if is_mtls:
639 new_adapter = _MutualTlsAdapter(cert, key, **kwargs)
640 if self._auth_request_session is not None:
641 new_auth_adapter = _MutualTlsAdapter(cert, key, **auth_kwargs)
642 else:
643 new_auth_adapter = None
644 else:
645 new_adapter = requests.adapters.HTTPAdapter(**kwargs)
646 if self._auth_request_session is not None:
647 new_auth_adapter = requests.adapters.HTTPAdapter(**auth_kwargs)
648 else:
649 new_auth_adapter = None
650 except (
651 exceptions.ClientCertError,
652 ImportError,
653 OSError,
654 ValueError,
655 ) as caught_exc:
656 new_exc = exceptions.MutualTLSChannelError(caught_exc)
657 raise new_exc from caught_exc
659 _replace_adapter(self, "https://", new_adapter)
661 if old_adapter is not None and old_adapter is not new_adapter:
662 old_adapter.close()
664 if self._auth_request_session is not None and new_auth_adapter is not None:
665 _replace_adapter(self._auth_request_session, "https://", new_auth_adapter)
667 if (
668 old_auth_adapter is not None
669 and old_auth_adapter is not new_auth_adapter
670 ):
671 old_auth_adapter.close()
673 self._is_mtls = is_mtls
674 if is_mtls:
675 self._cached_cert = cert
676 else:
677 if hasattr(self, "_cached_cert"):
678 del self._cached_cert
680 def request(
681 self,
682 method,
683 url,
684 data=None,
685 headers=None,
686 max_allowed_time=None,
687 timeout=_DEFAULT_TIMEOUT,
688 **kwargs,
689 ):
690 """Implementation of Requests' request.
692 Args:
693 timeout (Optional[Union[float, Tuple[float, float]]]):
694 The amount of time in seconds to wait for the server response
695 with each individual request. Can also be passed as a tuple
696 ``(connect_timeout, read_timeout)``. See :meth:`requests.Session.request`
697 documentation for details.
698 max_allowed_time (Optional[float]):
699 If the method runs longer than this, a ``Timeout`` exception is
700 automatically raised. Unlike the ``timeout`` parameter, this
701 value applies to the total method execution time, even if
702 multiple requests are made under the hood.
704 Mind that it is not guaranteed that the timeout error is raised
705 at ``max_allowed_time``. It might take longer, for example, if
706 an underlying request takes a lot of time, but the request
707 itself does not timeout, e.g. if a large file is being
708 transmitted. The timeout error will be raised after such
709 request completes.
710 Raises:
711 google.auth.exceptions.MutualTLSChannelError: If mutual TLS
712 channel creation fails for any reason.
713 ValueError: If the client certificate is invalid.
714 """
715 # pylint: disable=arguments-differ
716 # Requests has a ton of arguments to request, but only two
717 # (method, url) are required. We pass through all of the other
718 # arguments to super, so no need to exhaustively list them here.
720 # Use a kwarg for this instead of an attribute to maintain
721 # thread-safety.
722 _credential_refresh_attempt = kwargs.pop("_credential_refresh_attempt", 0)
724 # Make a copy of the headers. They will be modified by the credentials
725 # and we want to pass the original headers if we recurse.
726 request_headers = headers.copy() if headers is not None else {}
728 # Do not apply the timeout unconditionally in order to not override the
729 # _auth_request's default timeout.
730 auth_request = (
731 self._auth_request
732 if timeout is None
733 else functools.partial(self._auth_request, timeout=timeout)
734 )
736 remaining_time = max_allowed_time
738 with TimeoutGuard(remaining_time) as guard:
739 self.credentials.before_request(auth_request, method, url, request_headers)
740 remaining_time = guard.remaining_timeout
742 # Remember the client certificate that this request is sent with. If
743 # the request receives a 401 response, this tells whether another
744 # thread has already rotated the certificate in the meantime.
745 cert_at_request = getattr(self, "_cached_cert", None)
747 with TimeoutGuard(remaining_time) as guard:
748 _helpers.request_log(_LOGGER, method, url, data, headers)
749 response = super(AuthorizedSession, self).request(
750 method,
751 url,
752 data=data,
753 headers=request_headers,
754 timeout=timeout,
755 **kwargs,
756 )
757 remaining_time = guard.remaining_timeout
759 # If the response indicated that the credentials needed to be
760 # refreshed, then refresh the credentials and re-attempt the
761 # request.
762 # A stored token may expire between the time it is retrieved and
763 # the time the request is made, so we may need to try twice.
764 if (
765 response.status_code in self._refresh_status_codes
766 and _credential_refresh_attempt < self._max_refresh_attempts
767 ):
768 # Handle unauthorized permission error(401 status code)
769 if response.status_code == http_client.UNAUTHORIZED:
770 use_mtls = self.is_mtls and _mtls_helper.is_mtls_endpoint(url)
771 if use_mtls:
772 with TimeoutGuard(remaining_time) as guard:
773 self._reconfigure_mtls_channel_if_cert_changed(cert_at_request)
774 remaining_time = guard.remaining_timeout
775 _LOGGER.info(
776 "Refreshing credentials due to a %s response. Attempt %s/%s.",
777 response.status_code,
778 _credential_refresh_attempt + 1,
779 self._max_refresh_attempts,
780 )
782 # Do not apply the timeout unconditionally in order to not override the
783 # _auth_request's default timeout.
784 auth_request = (
785 self._auth_request
786 if timeout is None
787 else functools.partial(self._auth_request, timeout=timeout)
788 )
790 with TimeoutGuard(remaining_time) as guard:
791 self.credentials.refresh(auth_request)
792 remaining_time = guard.remaining_timeout
794 # Recurse. Pass in the original headers, not our modified set, but
795 # do pass the adjusted max allowed time (i.e. the remaining total time).
796 return self.request(
797 method,
798 url,
799 data=data,
800 headers=headers,
801 max_allowed_time=remaining_time,
802 timeout=timeout,
803 _credential_refresh_attempt=_credential_refresh_attempt + 1,
804 **kwargs,
805 )
807 return response
809 def _reconfigure_mtls_channel_if_cert_changed(self, cert_at_request):
810 """Reconfigures the mTLS channel if the client certificate has changed.
812 This is called when a request to an mTLS endpoint receives a 401
813 response. The certificate check and the channel reconfiguration run
814 while holding ``_mtls_reauth_lock``. When several threads receive a 401
815 response at the same time, only the first one reads the current
816 certificate and remounts the adapters. The other threads see that the
817 channel has already been reconfigured and reuse it.
819 Args:
820 cert_at_request (Optional[bytes]): The cached client certificate at
821 the time the request that received the 401 response was sent.
823 Raises:
824 google.auth.exceptions.MutualTLSChannelError: If the client
825 certificate has changed but the mTLS channel could not be
826 reconfigured.
827 """
828 with self._mtls_reauth_lock:
829 if not self.is_mtls:
830 # mTLS was disabled by another thread while this one waited.
831 return
833 if getattr(self, "_cached_cert", None) != cert_at_request:
834 _LOGGER.info(
835 "Skipping reconfiguration of mTLS channel because it has already"
836 " been reconfigured with a new client certificate."
837 )
838 return
840 try:
841 (
842 call_cert_bytes,
843 call_key_bytes,
844 cached_fingerprint,
845 current_cert_fingerprint,
846 ) = _mtls_helper.check_parameters_for_unauthorized_response(
847 cert_at_request
848 )
849 except Exception as e:
850 # Failing to load the current certificate (for example, because
851 # the certificate file is missing or the certificate provider
852 # failed) must not prevent the token refresh that follows.
853 _LOGGER.warning(
854 "Failed to check whether the client certificate has changed: %s",
855 e,
856 )
857 return
859 if cached_fingerprint == current_cert_fingerprint:
860 _LOGGER.info(
861 "Skipping reconfiguration of mTLS channel because the client"
862 " certificate has not changed."
863 )
864 return
866 try:
867 _LOGGER.info(
868 "Client certificate has changed, reconfiguring mTLS channel."
869 )
870 self.configure_mtls_channel(lambda: (call_cert_bytes, call_key_bytes))
871 except Exception as e:
872 _LOGGER.error("Failed to reconfigure mTLS channel: %s", e)
873 raise exceptions.MutualTLSChannelError(
874 "Failed to reconfigure mTLS channel"
875 ) from e
877 @property
878 def is_mtls(self):
879 """Indicates if the created SSL channel is mutual TLS."""
880 return self._is_mtls
882 def close(self):
883 if self._auth_request_session is not None:
884 self._auth_request_session.close()
885 super(AuthorizedSession, self).close()