From 0bc657a682b7436b96b8b1a6d9698becc94d8644 Mon Sep 17 00:00:00 2001 From: ohmayr Date: Thu, 30 May 2024 21:16:26 +0000 Subject: [PATCH] add support for async rest --- google/auth/aio/_credentials_async.py | 171 ++++++++++++ google/auth/aio/timeout_guard.py | 46 +++ google/auth/aio/transport/__init__.py | 103 +++++++ google/auth/aio/transport/requests.py | 384 ++++++++++++++++++++++++++ 4 files changed, 704 insertions(+) create mode 100644 google/auth/aio/_credentials_async.py create mode 100644 google/auth/aio/timeout_guard.py create mode 100644 google/auth/aio/transport/__init__.py create mode 100644 google/auth/aio/transport/requests.py diff --git a/google/auth/aio/_credentials_async.py b/google/auth/aio/_credentials_async.py new file mode 100644 index 000000000..760758d85 --- /dev/null +++ b/google/auth/aio/_credentials_async.py @@ -0,0 +1,171 @@ +# Copyright 2020 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +"""Interfaces for credentials.""" + +import abc +import inspect + +from google.auth import credentials + + +class Credentials(credentials.Credentials, metaclass=abc.ABCMeta): + """Async inherited credentials class from google.auth.credentials. + The added functionality is the before_request call which requires + async/await syntax. + All credentials have a :attr:`token` that is used for authentication and + may also optionally set an :attr:`expiry` to indicate when the token will + no longer be valid. + + Most credentials will be :attr:`invalid` until :meth:`refresh` is called. + Credentials can do this automatically before the first HTTP request in + :meth:`before_request`. + + Although the token and expiration will change as the credentials are + :meth:`refreshed ` and used, credentials should be considered + immutable. Various credentials will accept configuration such as private + keys, scopes, and other options. These options are not changeable after + construction. Some classes will provide mechanisms to copy the credentials + with modifications such as :meth:`ScopedCredentials.with_scopes`. + """ + + async def before_request(self, request, method, url, headers): + """Performs credential-specific before request logic. + + Refreshes the credentials if necessary, then calls :meth:`apply` to + apply the token to the authentication header. + + Args: + request (google.auth.transport.Request): The object used to make + HTTP requests. + method (str): The request's HTTP method or the RPC method being + invoked. + url (str): The request's URI or the RPC service's URI. + headers (Mapping): The request's headers. + """ + # pylint: disable=unused-argument + # (Subclasses may use these arguments to ascertain information about + # the http request.) + + if not self.valid: + if inspect.iscoroutinefunction(self.refresh): + await self.refresh(request) + else: + self.refresh(request) + self.apply(headers) + + +class CredentialsWithQuotaProject(credentials.CredentialsWithQuotaProject): + """Abstract base for credentials supporting ``with_quota_project`` factory""" + + +class AnonymousCredentials(credentials.AnonymousCredentials, Credentials): + """Credentials that do not provide any authentication information. + + These are useful in the case of services that support anonymous access or + local service emulators that do not use credentials. This class inherits + from the sync anonymous credentials file, but is kept if async credentials + is initialized and we would like anonymous credentials. + """ + + +class ReadOnlyScoped(credentials.ReadOnlyScoped, metaclass=abc.ABCMeta): + """Interface for credentials whose scopes can be queried. + + OAuth 2.0-based credentials allow limiting access using scopes as described + in `RFC6749 Section 3.3`_. + If a credential class implements this interface then the credentials either + use scopes in their implementation. + + Some credentials require scopes in order to obtain a token. You can check + if scoping is necessary with :attr:`requires_scopes`:: + + if credentials.requires_scopes: + # Scoping is required. + credentials = _credentials_async.with_scopes(scopes=['one', 'two']) + + Credentials that require scopes must either be constructed with scopes:: + + credentials = SomeScopedCredentials(scopes=['one', 'two']) + + Or must copy an existing instance using :meth:`with_scopes`:: + + scoped_credentials = _credentials_async.with_scopes(scopes=['one', 'two']) + + Some credentials have scopes but do not allow or require scopes to be set, + these credentials can be used as-is. + + .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 + """ + + +class Scoped(credentials.Scoped): + """Interface for credentials whose scopes can be replaced while copying. + + OAuth 2.0-based credentials allow limiting access using scopes as described + in `RFC6749 Section 3.3`_. + If a credential class implements this interface then the credentials either + use scopes in their implementation. + + Some credentials require scopes in order to obtain a token. You can check + if scoping is necessary with :attr:`requires_scopes`:: + + if credentials.requires_scopes: + # Scoping is required. + credentials = _credentials_async.create_scoped(['one', 'two']) + + Credentials that require scopes must either be constructed with scopes:: + + credentials = SomeScopedCredentials(scopes=['one', 'two']) + + Or must copy an existing instance using :meth:`with_scopes`:: + + scoped_credentials = credentials.with_scopes(scopes=['one', 'two']) + + Some credentials have scopes but do not allow or require scopes to be set, + these credentials can be used as-is. + + .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 + """ + + +def with_scopes_if_required(credentials, scopes): + """Creates a copy of the credentials with scopes if scoping is required. + + This helper function is useful when you do not know (or care to know) the + specific type of credentials you are using (such as when you use + :func:`google.auth.default`). This function will call + :meth:`Scoped.with_scopes` if the credentials are scoped credentials and if + the credentials require scoping. Otherwise, it will return the credentials + as-is. + + Args: + credentials (google.auth.credentials.Credentials): The credentials to + scope if necessary. + scopes (Sequence[str]): The list of scopes to use. + + Returns: + google.auth._credentials_async.Credentials: Either a new set of scoped + credentials, or the passed in credentials instance if no scoping + was required. + """ + if isinstance(credentials, Scoped) and credentials.requires_scopes: + return credentials.with_scopes(scopes) + else: + return credentials + + +class Signing(credentials.Signing, metaclass=abc.ABCMeta): + """Interface for credentials that can cryptographically sign messages.""" diff --git a/google/auth/aio/timeout_guard.py b/google/auth/aio/timeout_guard.py new file mode 100644 index 000000000..82c78930a --- /dev/null +++ b/google/auth/aio/timeout_guard.py @@ -0,0 +1,46 @@ +import time +import numbers +import requests.exceptions # pylint: disable=ungrouped-imports + +class TimeoutGuard(object): + """A context manager raising an error if the suite execution took too long. + + Args: + timeout (Union[None, Union[float, Tuple[float, float]]]): + The maximum number of seconds a suite can run without the context + manager raising a timeout exception on exit. If passed as a tuple, + the smaller of the values is taken as a timeout. If ``None``, a + timeout error is never raised. + timeout_error_type (Optional[Exception]): + The type of the error to raise on timeout. Defaults to + :class:`requests.exceptions.Timeout`. + """ + + def __init__(self, timeout, timeout_error_type=requests.exceptions.Timeout): + self._timeout = timeout + self.remaining_timeout = timeout + self._timeout_error_type = timeout_error_type + + def __enter__(self): + self._start = time.time() + return self + + def __exit__(self, exc_type, exc_value, traceback): + if exc_value: + return # let the error bubble up automatically + + if self._timeout is None: + return # nothing to do, the timeout was not specified + + elapsed = time.time() - self._start + deadline_hit = False + + if isinstance(self._timeout, numbers.Number): + self.remaining_timeout = self._timeout - elapsed + deadline_hit = self.remaining_timeout <= 0 + else: + self.remaining_timeout = tuple(x - elapsed for x in self._timeout) + deadline_hit = min(self.remaining_timeout) <= 0 + + if deadline_hit: + raise self._timeout_error_type() \ No newline at end of file diff --git a/google/auth/aio/transport/__init__.py b/google/auth/aio/transport/__init__.py new file mode 100644 index 000000000..724568e58 --- /dev/null +++ b/google/auth/aio/transport/__init__.py @@ -0,0 +1,103 @@ +# Copyright 2016 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Transport - HTTP client library support. + +:mod:`google.auth` is designed to work with various HTTP client libraries such +as urllib3 and requests. In order to work across these libraries with different +interfaces some abstraction is needed. + +This module provides two interfaces that are implemented by transport adapters +to support HTTP libraries. :class:`Request` defines the interface expected by +:mod:`google.auth` to make requests. :class:`Response` defines the interface +for the return value of :class:`Request`. +""" + +import abc +import http.client as http_client + +DEFAULT_RETRYABLE_STATUS_CODES = ( + http_client.INTERNAL_SERVER_ERROR, + http_client.SERVICE_UNAVAILABLE, + http_client.REQUEST_TIMEOUT, + http_client.TOO_MANY_REQUESTS, +) +"""Sequence[int]: HTTP status codes indicating a request can be retried. +""" + + +DEFAULT_REFRESH_STATUS_CODES = (http_client.UNAUTHORIZED,) +"""Sequence[int]: Which HTTP status code indicate that credentials should be +refreshed. +""" + +DEFAULT_MAX_REFRESH_ATTEMPTS = 2 +"""int: How many times to refresh the credentials and retry a request.""" + + +class Response(metaclass=abc.ABCMeta): + """HTTP Response data.""" + + @abc.abstractproperty + def status(self): + """int: The HTTP status code.""" + raise NotImplementedError("status must be implemented.") + + @abc.abstractproperty + def headers(self): + """Mapping[str, str]: The HTTP response headers.""" + raise NotImplementedError("headers must be implemented.") + + @abc.abstractproperty + def data(self): + """bytes: The response body.""" + raise NotImplementedError("data must be implemented.") + + +class Request(metaclass=abc.ABCMeta): + """Interface for a callable that makes HTTP requests. + + Specific transport implementations should provide an implementation of + this that adapts their specific request / response API. + + .. automethod:: __call__ + """ + + @abc.abstractmethod + def __call__( + self, url, method="GET", body=None, headers=None, timeout=None, **kwargs + ): + """Make an HTTP request. + + Args: + url (str): The URI to be requested. + method (str): The HTTP method to use for the request. Defaults + to 'GET'. + body (bytes): The payload / body in HTTP request. + headers (Mapping[str, str]): Request headers. + timeout (Optional[int]): The number of seconds to wait for a + response from the server. If not specified or if None, the + transport-specific default timeout will be used. + kwargs: Additionally arguments passed on to the transport's + request method. + + Returns: + Response: The HTTP response. + + Raises: + google.auth.exceptions.TransportError: If any exception occurred. + """ + # pylint: disable=redundant-returns-doc, missing-raises-doc + # (pylint doesn't play well with abstract docstrings.) + raise NotImplementedError("__call__ must be implemented.") diff --git a/google/auth/aio/transport/requests.py b/google/auth/aio/transport/requests.py new file mode 100644 index 000000000..71e9d0c82 --- /dev/null +++ b/google/auth/aio/transport/requests.py @@ -0,0 +1,384 @@ +# Copyright 2020 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Transport adapter for Async HTTP (aiohttp). + +NOTE: This async support is experimental and marked internal. This surface may +change in minor releases. +""" + +from __future__ import absolute_import + +import asyncio +import functools +import logging + +try: + import aiohttp +except ImportError as caught_exc: + raise ImportError( + "The aiohttp library is not installed from please install the aiohttp package to use the requests transport." + ) from caught_exc +import urllib3 # type: ignore + +from google.auth import exceptions +from google.auth.aio import transport +from google.auth.aio import timeout_guard + +# Timeout can be re-defined depending on async requirement. Currently made 60s more than +# sync timeout. +_DEFAULT_TIMEOUT = 180 # in seconds + +_LOGGER = logging.getLogger(__name__) + +class _CombinedResponse(transport.Response): + """ + In order to more closely resemble the `requests` interface, where a raw + and deflated content could be accessed at once, this class lazily reads the + stream in `transport.Response` so both return forms can be used. + + The gzip and deflate transfer-encodings are automatically decoded for you + because the default parameter for autodecompress into the ClientSession is set + to False, and therefore we add this class to act as a wrapper for a user to be + able to access both the raw and decoded response bodies - mirroring the sync + implementation. + """ + + def __init__(self, response): + self._response = response + self._raw_content = None + + def _is_compressed(self): + headers = self._response.headers + return "Content-Encoding" in headers and ( + headers["Content-Encoding"] == "gzip" + or headers["Content-Encoding"] == "deflate" + ) + + @property + def status(self): + return self._response.status + + @property + def headers(self): + return self._response.headers + + @property + def data(self): + return self._response.content + + @property + def request(self): + return self._response.request_info + + async def raw_content(self): + if self._raw_content is None: + self._raw_content = await self._response.content.read() + return self._raw_content + + async def content(self): + # Load raw_content if necessary + await self.raw_content() + if self._is_compressed(): + decoder = urllib3.response.MultiDecoder( + self._response.headers["Content-Encoding"] + ) + decompressed = decoder.decompress(self._raw_content) + return decompressed + + return self._raw_content + + + +class Request(transport.Request): + """Requests request adapter. + + This class is used internally for making requests using asyncio transports + in a consistent way. If you use :class:`AuthorizedSession` you do not need + to construct or use this class directly. + + This class can be useful if you want to manually refresh a + :class:`~google.auth.credentials.Credentials` instance:: + + import google.auth.transport.aiohttp_requests + + request = google.auth.transport.aiohttp_requests.Request() + + credentials.refresh(request) + + Args: + session (aiohttp.ClientSession): An instance :class:`aiohttp.ClientSession` used + to make HTTP requests. If not specified, a session will be created. + + .. automethod:: __call__ + """ + + def __init__(self, session=None): + # TODO: Use auto_decompress property for aiohttp 3.7+ + if session is not None and session._auto_decompress: + raise exceptions.InvalidOperation( + "Client sessions with auto_decompress=True are not supported." + ) + self.session = session + + async def __call__( + self, + url, + method="GET", + body=None, + headers=None, + timeout=_DEFAULT_TIMEOUT, + **kwargs, + ): + """ + Make an HTTP request using aiohttp. + + Args: + url (str): The URL to be requested. + method (Optional[str]): + The HTTP method to use for the request. Defaults to 'GET'. + body (Optional[bytes]): + The payload or body in HTTP request. + headers (Optional[Mapping[str, str]]): + Request headers. + timeout (Optional[int]): The number of seconds to wait for a + response from the server. If not specified or if None, the + requests default timeout will be used. + kwargs: Additional arguments passed through to the underlying + requests :meth:`requests.Session.request` method. + + Returns: + google.auth.transport.Response: The HTTP response. + + Raises: + google.auth.exceptions.TransportError: If any exception occurred. + """ + + try: + if self.session is None: # pragma: NO COVER + self.session = aiohttp.ClientSession( + auto_decompress=False + ) # pragma: NO COVER + _LOGGER.debug("Making request: %s %s", method, url) + response = await self.session.request( + method, url, data=body, headers=headers, timeout=timeout, **kwargs + ) + return _CombinedResponse(response) + + except aiohttp.ClientError as caught_exc: + new_exc = exceptions.TransportError(caught_exc) + raise new_exc from caught_exc + + except asyncio.TimeoutError as caught_exc: + new_exc = exceptions.TransportError(caught_exc) + raise new_exc from caught_exc + + +class AuthorizedSession(aiohttp.ClientSession): + """This is an async implementation of the Authorized Session class. We utilize an + aiohttp transport instance, and the interface mirrors the google.auth.transport.requests + Authorized Session class, except for the change in the transport used in the async use case. + + A Requests Session class with credentials. + + This class is used to perform requests to API endpoints that require + authorization:: + + from google.auth.transport import aiohttp_requests + + async with aiohttp_requests.AuthorizedSession(credentials) as authed_session: + response = await authed_session.request( + 'GET', 'https://www.googleapis.com/storage/v1/b') + + The underlying :meth:`request` implementation handles adding the + credentials' headers to the request and refreshing credentials as needed. + + Args: + credentials (google.auth._credentials_async.Credentials): + The credentials to add to the request. + refresh_status_codes (Sequence[int]): Which HTTP status codes indicate + that credentials should be refreshed and the request should be + retried. + max_refresh_attempts (int): The maximum number of times to attempt to + refresh the credentials and retry the request. + refresh_timeout (Optional[int]): The timeout value in seconds for + credential refresh HTTP requests. + auth_request (google.auth.transport.aiohttp_requests.Request): + (Optional) An instance of + :class:`~google.auth.transport.aiohttp_requests.Request` used when + refreshing credentials. If not passed, + an instance of :class:`~google.auth.transport.aiohttp_requests.Request` + is created. + kwargs: Additional arguments passed through to the underlying + ClientSession :meth:`aiohttp.ClientSession` object. + """ + + def __init__( + self, + credentials, + refresh_status_codes=transport.DEFAULT_REFRESH_STATUS_CODES, + max_refresh_attempts=transport.DEFAULT_MAX_REFRESH_ATTEMPTS, + refresh_timeout=None, + auth_request=None, + auto_decompress=False, + **kwargs, + ): + super(AuthorizedSession, self).__init__(**kwargs) + self.credentials = credentials + self._refresh_status_codes = refresh_status_codes + self._max_refresh_attempts = max_refresh_attempts + self._refresh_timeout = refresh_timeout + self._is_mtls = False + self._auth_request = auth_request + self._auth_request_session = None + self._loop = asyncio.get_event_loop() + self._refresh_lock = asyncio.Lock() + self._auto_decompress = auto_decompress + + async def request( + self, + method, + url, + data=None, + headers=None, + max_allowed_time=None, + timeout=_DEFAULT_TIMEOUT, + auto_decompress=False, + **kwargs, + ): + + """Implementation of Authorized Session aiohttp request. + + Args: + method (str): + The http request method used (e.g. GET, PUT, DELETE) + url (str): + The url at which the http request is sent. + data (Optional[dict]): Dictionary, list of tuples, bytes, or file-like + object to send in the body of the Request. + headers (Optional[dict]): Dictionary of HTTP Headers to send with the + Request. + timeout (Optional[Union[float, aiohttp.ClientTimeout]]): + The amount of time in seconds to wait for the server response + with each individual request. Can also be passed as an + ``aiohttp.ClientTimeout`` object. + max_allowed_time (Optional[float]): + If the method runs longer than this, a ``Timeout`` exception is + automatically raised. Unlike the ``timeout`` parameter, this + value applies to the total method execution time, even if + multiple requests are made under the hood. + + Mind that it is not guaranteed that the timeout error is raised + at ``max_allowed_time``. It might take longer, for example, if + an underlying request takes a lot of time, but the request + itself does not timeout, e.g. if a large file is being + transmitted. The timout error will be raised after such + request completes. + """ + # Headers come in as bytes which isn't expected behavior, the resumable + # media libraries in some cases expect a str type for the header values, + # but sometimes the operations return these in bytes types. + if headers: + for key in headers.keys(): + if type(headers[key]) is bytes: + headers[key] = headers[key].decode("utf-8") + + async with aiohttp.ClientSession( + auto_decompress=self._auto_decompress + ) as self._auth_request_session: + auth_request = Request(self._auth_request_session) + self._auth_request = auth_request + + # Use a kwarg for this instead of an attribute to maintain + # thread-safety. + _credential_refresh_attempt = kwargs.pop("_credential_refresh_attempt", 0) + # Make a copy of the headers. They will be modified by the credentials + # and we want to pass the original headers if we recurse. + request_headers = headers.copy() if headers is not None else {} + + # Do not apply the timeout unconditionally in order to not override the + # _auth_request's default timeout. + auth_request = ( + self._auth_request + if timeout is None + else functools.partial(self._auth_request, timeout=timeout) + ) + + remaining_time = max_allowed_time + + with timeout_guard.TimeoutGuard(remaining_time, asyncio.TimeoutError) as guard: + if asyncio.iscoroutinefunction(self.credentials.before_request): + await self.credentials.before_request( + auth_request, method, url, request_headers + ) + + with timeout_guard.TimeoutGuard(remaining_time, asyncio.TimeoutError) as guard: + response = await super(AuthorizedSession, self).request( + method, + url, + data=data, + headers=request_headers, + timeout=timeout, + **kwargs, + ) + + remaining_time = guard.remaining_timeout + + if ( + response.status in self._refresh_status_codes + and _credential_refresh_attempt < self._max_refresh_attempts + ): + + _LOGGER.info( + "Refreshing credentials due to a %s response. Attempt %s/%s.", + response.status, + _credential_refresh_attempt + 1, + self._max_refresh_attempts, + ) + + # Do not apply the timeout unconditionally in order to not override the + # _auth_request's default timeout. + auth_request = ( + self._auth_request + if timeout is None + else functools.partial(self._auth_request, timeout=timeout) + ) + + with timeout_guard.TimeoutGuard( + remaining_time, asyncio.TimeoutError + ) as guard: + async with self._refresh_lock: + await self._loop.run_in_executor( + None, self.credentials.refresh, auth_request + ) + + remaining_time = guard.remaining_timeout + + return await self.request( + method, + url, + data=data, + headers=headers, + max_allowed_time=remaining_time, + timeout=timeout, + _credential_refresh_attempt=_credential_refresh_attempt + 1, + **kwargs, + ) + + return _CombinedResponse(response) + + async def close(self): + if self._auth_request_session is not None: + await self._auth_request_session.close() + await super(AuthorizedSession, self).close()