8621812797

Committed 09 Apr 2024 08:22PM UTC coverage: 90.097%. Remained the same

Build # 8621812797

Build Type

Pull #538

github

Committed by

web-flow

Commit Message

Merge 59f3d6a0e into 45eca1a69

Pull Request Pull Request #538: Added secrets to actions to run test example

Run Details

464 of 515 relevant lines covered (90.1%)

5.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.75

/requests_oauthlib/oauth2_session.py

import logging

from oauthlib.common import generate_token, urldecode
from oauthlib.oauth2 import WebApplicationClient, InsecureTransportError
from oauthlib.oauth2 import LegacyApplicationClient
from oauthlib.oauth2 import TokenExpiredError, is_secure_transport
import requests

log = logging.getLogger(__name__)


class TokenUpdated(Warning):
    def __init__(self, token):
        super(TokenUpdated, self).__init__()
        self.token = token


class OAuth2Session(requests.Session):
    """Versatile OAuth 2 extension to :class:`requests.Session`.

    Supports any grant type adhering to :class:`oauthlib.oauth2.Client` spec
    including the four core OAuth 2 grants.

    Can be used to create authorization urls, fetch tokens and access protected
    resources using the :class:`requests.Session` interface you are used to.

    - :class:`oauthlib.oauth2.WebApplicationClient` (default): Authorization Code Grant
    - :class:`oauthlib.oauth2.MobileApplicationClient`: Implicit Grant
    - :class:`oauthlib.oauth2.LegacyApplicationClient`: Password Credentials Grant
    - :class:`oauthlib.oauth2.BackendApplicationClient`: Client Credentials Grant

    Note that the only time you will be using Implicit Grant from python is if
    you are driving a user agent able to obtain URL fragments.
    """

    def __init__(
        self,
        client_id=None,
        client=None,
        auto_refresh_url=None,
        auto_refresh_kwargs=None,
        scope=None,
        redirect_uri=None,
        token=None,
        state=None,
        token_updater=None,
        pkce=None,
        **kwargs
    ):
        """Construct a new OAuth 2 client session.

        :param client_id: Client id obtained during registration
        :param client: :class:`oauthlib.oauth2.Client` to be used. Default is
                       WebApplicationClient which is useful for any
                       hosted application but not mobile or desktop.
        :param scope: List of scopes you wish to request access to
        :param redirect_uri: Redirect URI you registered as callback
        :param token: Token dictionary, must include access_token
                      and token_type.
        :param state: State string used to prevent CSRF. This will be given
                      when creating the authorization url and must be supplied
                      when parsing the authorization response.
                      Can be either a string or a no argument callable.
        :auto_refresh_url: Refresh token endpoint URL, must be HTTPS. Supply
                           this if you wish the client to automatically refresh
                           your access tokens.
        :auto_refresh_kwargs: Extra arguments to pass to the refresh token
                              endpoint.
        :token_updater: Method with one argument, token, to be used to update
                        your token database on automatic token refresh. If not
                        set a TokenUpdated warning will be raised when a token
                        has been refreshed. This warning will carry the token
                        in its token argument.
        :param pkce: Set "S256" or "plain" to enable PKCE. Default is disabled.
        :param kwargs: Arguments to pass to the Session constructor.
        """
        super(OAuth2Session, self).__init__(**kwargs)
        self._client = client or WebApplicationClient(client_id, token=token)
        self.token = token or {}
        self._scope = scope
        self.redirect_uri = redirect_uri
        self.state = state or generate_token
        self._state = state
        self.auto_refresh_url = auto_refresh_url
        self.auto_refresh_kwargs = auto_refresh_kwargs or {}
        self.token_updater = token_updater
        self._pkce = pkce

        if self._pkce not in ["S256", "plain", None]:
            raise AttributeError("Wrong value for {}(.., pkce={})".format(self.__class__, self._pkce))

        # Ensure that requests doesn't do any automatic auth. See #278.
        # The default behavior can be re-enabled by setting auth to None.
        self.auth = lambda r: r

        # Allow customizations for non compliant providers through various
        # hooks to adjust requests and responses.
        self.compliance_hook = {
            "access_token_response": set(),
            "refresh_token_response": set(),
            "protected_request": set(),
            "refresh_token_request": set(),
            "access_token_request": set(),
        }

    @property
    def scope(self):
        """By default the scope from the client is used, except if overridden"""
        if self._scope is not None:
            return self._scope
        elif self._client is not None:
            return self._client.scope
        else:
            return None

    @scope.setter
    def scope(self, scope):
        self._scope = scope

    def new_state(self):
        """Generates a state string to be used in authorizations."""
        try:
            self._state = self.state()
            log.debug("Generated new state %s.", self._state)
        except TypeError:
            self._state = self.state
            log.debug("Re-using previously supplied state %s.", self._state)
        return self._state

    @property
    def client_id(self):
        return getattr(self._client, "client_id", None)

    @client_id.setter
    def client_id(self, value):
        self._client.client_id = value

    @client_id.deleter
    def client_id(self):
        del self._client.client_id

    @property
    def token(self):
        return getattr(self._client, "token", None)

    @token.setter
    def token(self, value):
        self._client.token = value
        self._client.populate_token_attributes(value)

    @property
    def access_token(self):
        return getattr(self._client, "access_token", None)

    @access_token.setter
    def access_token(self, value):
        self._client.access_token = value

    @access_token.deleter
    def access_token(self):
        del self._client.access_token

    @property
    def authorized(self):
        """Boolean that indicates whether this session has an OAuth token
        or not. If `self.authorized` is True, you can reasonably expect
        OAuth-protected requests to the resource to succeed. If
        `self.authorized` is False, you need the user to go through the OAuth
        authentication dance before OAuth-protected requests to the resource
        will succeed.
        """
        return bool(self.access_token)

    def authorization_url(self, url, state=None, **kwargs):
        """Form an authorization URL.

        :param url: Authorization endpoint url, must be HTTPS.
        :param state: An optional state string for CSRF protection. If not
                      given it will be generated for you.
        :param kwargs: Extra parameters to include.
        :return: authorization_url, state
        """
        state = state or self.new_state()
        if self._pkce:
            self._code_verifier = self._client.create_code_verifier(43)
            kwargs["code_challenge_method"] = self._pkce
            kwargs["code_challenge"] = self._client.create_code_challenge(
                code_verifier=self._code_verifier,
                code_challenge_method=self._pkce
            )
        return (
            self._client.prepare_request_uri(
                url,
                redirect_uri=self.redirect_uri,
                scope=self.scope,
                state=state,
                **kwargs
            ),
            state,
        )

    def fetch_token(
        self,
        token_url,
        code=None,
        authorization_response=None,
        body="",
        auth=None,
        username=None,
        password=None,
        method="POST",
        force_querystring=False,
        timeout=None,
        headers=None,
        verify=None,
        proxies=None,
        include_client_id=None,
        client_secret=None,
        cert=None,
        **kwargs
    ):
        """Generic method for fetching an access token from the token endpoint.

        If you are using the MobileApplicationClient you will want to use
        `token_from_fragment` instead of `fetch_token`.

        The current implementation enforces the RFC guidelines.

        :param token_url: Token endpoint URL, must use HTTPS.
        :param code: Authorization code (used by WebApplicationClients).
        :param authorization_response: Authorization response URL, the callback
                                       URL of the request back to you. Used by
                                       WebApplicationClients instead of code.
        :param body: Optional application/x-www-form-urlencoded body to add the
                     include in the token request. Prefer kwargs over body.
        :param auth: An auth tuple or method as accepted by `requests`.
        :param username: Username required by LegacyApplicationClients to appear
                         in the request body.
        :param password: Password required by LegacyApplicationClients to appear
                         in the request body.
        :param method: The HTTP method used to make the request. Defaults
                       to POST, but may also be GET. Other methods should
                       be added as needed.
        :param force_querystring: If True, force the request body to be sent
            in the querystring instead.
        :param timeout: Timeout of the request in seconds.
        :param headers: Dict to default request headers with.
        :param verify: Verify SSL certificate.
        :param proxies: The `proxies` argument is passed onto `requests`.
        :param include_client_id: Should the request body include the
                                  `client_id` parameter. Default is `None`,
                                  which will attempt to autodetect. This can be
                                  forced to always include (True) or never
                                  include (False).
        :param client_secret: The `client_secret` paired to the `client_id`.
                              This is generally required unless provided in the
                              `auth` tuple. If the value is `None`, it will be
                              omitted from the request, however if the value is
                              an empty string, an empty string will be sent.
        :param cert: Client certificate to send for OAuth 2.0 Mutual-TLS Client
                     Authentication (draft-ietf-oauth-mtls). Can either be the
                     path of a file containing the private key and certificate or
                     a tuple of two filenames for certificate and key.
        :param kwargs: Extra parameters to include in the token request.
        :return: A token dict
        """
        if not is_secure_transport(token_url):
            raise InsecureTransportError()

        if not code and authorization_response:
            self._client.parse_request_uri_response(
                authorization_response, state=self._state
            )
            code = self._client.code
        elif not code and isinstance(self._client, WebApplicationClient):
            code = self._client.code
            if not code:
                raise ValueError(
                    "Please supply either code or " "authorization_response parameters."
                )

        if self._pkce:
            if self._code_verifier is None:
                raise ValueError(
                    "Code verifier is not found, authorization URL must be generated before"
                )
            kwargs["code_verifier"] = self._code_verifier

        # Earlier versions of this library build an HTTPBasicAuth header out of
        # `username` and `password`. The RFC states, however these attributes
        # must be in the request body and not the header.
        # If an upstream server is not spec compliant and requires them to
        # appear as an Authorization header, supply an explicit `auth` header
        # to this function.
        # This check will allow for empty strings, but not `None`.
        #
        # References
        # 4.3.2 - Resource Owner Password Credentials Grant
        #         https://tools.ietf.org/html/rfc6749#section-4.3.2

        if isinstance(self._client, LegacyApplicationClient):
            if username is None:
                raise ValueError(
                    "`LegacyApplicationClient` requires both the "
                    "`username` and `password` parameters."
                )
            if password is None:
                raise ValueError(
                    "The required parameter `username` was supplied, "
                    "but `password` was not."
                )

        # merge username and password into kwargs for `prepare_request_body`
        if username is not None:
            kwargs["username"] = username
        if password is not None:
            kwargs["password"] = password

        # is an auth explicitly supplied?
        if auth is not None:
            # if we're dealing with the default of `include_client_id` (None):
            # we will assume the `auth` argument is for an RFC compliant server
            # and we should not send the `client_id` in the body.
            # This approach allows us to still force the client_id by submitting
            # `include_client_id=True` along with an `auth` object.
            if include_client_id is None:
                include_client_id = False

        # otherwise we may need to create an auth header
        else:
            # since we don't have an auth header, we MAY need to create one
            # it is possible that we want to send the `client_id` in the body
            # if so, `include_client_id` should be set to True
            # otherwise, we will generate an auth header
            if include_client_id is not True:
                client_id = self.client_id
                if client_id:
                    log.debug(
                        'Encoding `client_id` "%s" with `client_secret` '
                        "as Basic auth credentials.",
                        client_id,
                    )
                    client_secret = client_secret if client_secret is not None else ""
                    auth = requests.auth.HTTPBasicAuth(client_id, client_secret)

        if include_client_id:
            # this was pulled out of the params
            # it needs to be passed into prepare_request_body
            if client_secret is not None:
                kwargs["client_secret"] = client_secret

        body = self._client.prepare_request_body(
            code=code,
            body=body,
            redirect_uri=self.redirect_uri,
            include_client_id=include_client_id,
            **kwargs
        )

        headers = headers or {
            "Accept": "application/json",
            "Content-Type": "application/x-www-form-urlencoded",
        }
        self.token = {}
        request_kwargs = {}
        if method.upper() == "POST":
            request_kwargs["params" if force_querystring else "data"] = dict(
                urldecode(body)
            )
        elif method.upper() == "GET":
            request_kwargs["params"] = dict(urldecode(body))
        else:
            raise ValueError("The method kwarg must be POST or GET.")

        for hook in self.compliance_hook["access_token_request"]:
            log.debug("Invoking access_token_request hook %s.", hook)
            token_url, headers, request_kwargs = hook(
                token_url, headers, request_kwargs
            )

        r = self.request(
            method=method,
            url=token_url,
            timeout=timeout,
            headers=headers,
            auth=auth,
            verify=verify,
            proxies=proxies,
            cert=cert,
            **request_kwargs
        )

        log.debug("Request to fetch token completed with status %s.", r.status_code)
        log.debug("Request url was %s", r.request.url)
        log.debug("Request headers were %s", r.request.headers)
        log.debug("Request body was %s", r.request.body)
        log.debug("Response headers were %s and content %s.", r.headers, r.text)
        log.debug(
            "Invoking %d token response hooks.",
            len(self.compliance_hook["access_token_response"]),
        )
        for hook in self.compliance_hook["access_token_response"]:
            log.debug("Invoking hook %s.", hook)
            r = hook(r)

        self._client.parse_request_body_response(r.text, scope=self.scope)
        self.token = self._client.token
        log.debug("Obtained token %s.", self.token)
        return self.token

    def token_from_fragment(self, authorization_response):
        """Parse token from the URI fragment, used by MobileApplicationClients.

        :param authorization_response: The full URL of the redirect back to you
        :return: A token dict
        """
        self._client.parse_request_uri_response(
            authorization_response, state=self._state
        )
        self.token = self._client.token
        return self.token

    def refresh_token(
        self,
        token_url,
        refresh_token=None,
        body="",
        auth=None,
        timeout=None,
        headers=None,
        verify=None,
        proxies=None,
        **kwargs
    ):
        """Fetch a new access token using a refresh token.

        :param token_url: The token endpoint, must be HTTPS.
        :param refresh_token: The refresh_token to use.
        :param body: Optional application/x-www-form-urlencoded body to add the
                     include in the token request. Prefer kwargs over body.
        :param auth: An auth tuple or method as accepted by `requests`.
        :param timeout: Timeout of the request in seconds.
        :param headers: A dict of headers to be used by `requests`.
        :param verify: Verify SSL certificate.
        :param proxies: The `proxies` argument will be passed to `requests`.
        :param kwargs: Extra parameters to include in the token request.
        :return: A token dict
        """
        if not token_url:
            raise ValueError("No token endpoint set for auto_refresh.")

        if not is_secure_transport(token_url):
            raise InsecureTransportError()

        refresh_token = refresh_token or self.token.get("refresh_token")

        log.debug(
            "Adding auto refresh key word arguments %s.", self.auto_refresh_kwargs
        )
        kwargs.update(self.auto_refresh_kwargs)
        body = self._client.prepare_refresh_body(
            body=body, refresh_token=refresh_token, scope=self.scope, **kwargs
        )
        log.debug("Prepared refresh token request body %s", body)

        if headers is None:
            headers = {
                "Accept": "application/json",
                "Content-Type": ("application/x-www-form-urlencoded"),
            }

        for hook in self.compliance_hook["refresh_token_request"]:
            log.debug("Invoking refresh_token_request hook %s.", hook)
            token_url, headers, body = hook(token_url, headers, body)

        r = self.post(
            token_url,
            data=dict(urldecode(body)),
            auth=auth,
            timeout=timeout,
            headers=headers,
            verify=verify,
            withhold_token=True,
            proxies=proxies,
        )
        log.debug("Request to refresh token completed with status %s.", r.status_code)
        log.debug("Response headers were %s and content %s.", r.headers, r.text)
        log.debug(
            "Invoking %d token response hooks.",
            len(self.compliance_hook["refresh_token_response"]),
        )
        for hook in self.compliance_hook["refresh_token_response"]:
            log.debug("Invoking hook %s.", hook)
            r = hook(r)

        self.token = self._client.parse_request_body_response(r.text, scope=self.scope)
        if "refresh_token" not in self.token:
            log.debug("No new refresh token given. Re-using old.")
            self.token["refresh_token"] = refresh_token
        return self.token

    def request(
        self,
        method,
        url,
        data=None,
        headers=None,
        withhold_token=False,
        client_id=None,
        client_secret=None,
        files=None,
        **kwargs
    ):
        """Intercept all requests and add the OAuth 2 token if present."""
        if not is_secure_transport(url):
            raise InsecureTransportError()
        if self.token and not withhold_token:
            log.debug(
                "Invoking %d protected resource request hooks.",
                len(self.compliance_hook["protected_request"]),
            )
            for hook in self.compliance_hook["protected_request"]:
                log.debug("Invoking hook %s.", hook)
                url, headers, data = hook(url, headers, data)

            log.debug("Adding token %s to request.", self.token)
            try:
                url, headers, data = self._client.add_token(
                    url, http_method=method, body=data, headers=headers
                )
            # Attempt to retrieve and save new access token if expired
            except TokenExpiredError:
                if self.auto_refresh_url:
                    log.debug(
                        "Auto refresh is set, attempting to refresh at %s.",
                        self.auto_refresh_url,
                    )

                    # We mustn't pass auth twice.
                    auth = kwargs.pop("auth", None)
                    if client_id and client_secret and (auth is None):
                        log.debug(
                            'Encoding client_id "%s" with client_secret as Basic auth credentials.',
                            client_id,
                        )
                        auth = requests.auth.HTTPBasicAuth(client_id, client_secret)
                    token = self.refresh_token(
                        self.auto_refresh_url, auth=auth, **kwargs
                    )
                    if self.token_updater:
                        log.debug(
                            "Updating token to %s using %s.", token, self.token_updater
                        )
                        self.token_updater(token)
                        url, headers, data = self._client.add_token(
                            url, http_method=method, body=data, headers=headers
                        )
                    else:
                        raise TokenUpdated(token)
                else:
                    raise

        log.debug("Requesting url %s using method %s.", url, method)
        log.debug("Supplying headers %s and data %s", headers, data)
        log.debug("Passing through key word arguments %s.", kwargs)
        return super(OAuth2Session, self).request(
            method, url, headers=headers, data=data, files=files, **kwargs
        )

    def register_compliance_hook(self, hook_type, hook):
        """Register a hook for request/response tweaking.

        Available hooks are:
            access_token_response invoked before token parsing.
            refresh_token_response invoked before refresh token parsing.
            protected_request invoked before making a request.
            access_token_request invoked before making a token fetch request.
            refresh_token_request invoked before making a refresh request.

        If you find a new hook is needed please send a GitHub PR request
        or open an issue.
        """
        if hook_type not in self.compliance_hook:
            raise ValueError(
                "Hook type %s is not in %s.", hook_type, self.compliance_hook
            )
        self.compliance_hook[hook_type].add(hook)

1	import logging	6✔
2
3	from oauthlib.common import generate_token, urldecode	6✔
4	from oauthlib.oauth2 import WebApplicationClient, InsecureTransportError	6✔
5	from oauthlib.oauth2 import LegacyApplicationClient	6✔
6	from oauthlib.oauth2 import TokenExpiredError, is_secure_transport	6✔
7	import requests	6✔
8
9	log = logging.getLogger(__name__)	6✔
10
11
12	class TokenUpdated(Warning):	6✔
13	def __init__(self, token):	6✔
14	super(TokenUpdated, self).__init__()	6✔
15	self.token = token	6✔
16
17
18	class OAuth2Session(requests.Session):	6✔
19	"""Versatile OAuth 2 extension to :class:`requests.Session`.
20
21	Supports any grant type adhering to :class:`oauthlib.oauth2.Client` spec
22	including the four core OAuth 2 grants.
23
24	Can be used to create authorization urls, fetch tokens and access protected
25	resources using the :class:`requests.Session` interface you are used to.
26
27	- :class:`oauthlib.oauth2.WebApplicationClient` (default): Authorization Code Grant
28	- :class:`oauthlib.oauth2.MobileApplicationClient`: Implicit Grant
29	- :class:`oauthlib.oauth2.LegacyApplicationClient`: Password Credentials Grant
30	- :class:`oauthlib.oauth2.BackendApplicationClient`: Client Credentials Grant
31
32	Note that the only time you will be using Implicit Grant from python is if
33	you are driving a user agent able to obtain URL fragments.
34	"""
35
36	def __init__(	6✔
37	self,
38	client_id=None,
39	client=None,
40	auto_refresh_url=None,
41	auto_refresh_kwargs=None,
42	scope=None,
43	redirect_uri=None,
44	token=None,
45	state=None,
46	token_updater=None,
47	pkce=None,
48	**kwargs
49	):
50	"""Construct a new OAuth 2 client session.
51
52	:param client_id: Client id obtained during registration
53	:param client: :class:`oauthlib.oauth2.Client` to be used. Default is
54	WebApplicationClient which is useful for any
55	hosted application but not mobile or desktop.
56	:param scope: List of scopes you wish to request access to
57	:param redirect_uri: Redirect URI you registered as callback
58	:param token: Token dictionary, must include access_token
59	and token_type.
60	:param state: State string used to prevent CSRF. This will be given
61	when creating the authorization url and must be supplied
62	when parsing the authorization response.
63	Can be either a string or a no argument callable.
64	:auto_refresh_url: Refresh token endpoint URL, must be HTTPS. Supply
65	this if you wish the client to automatically refresh
66	your access tokens.
67	:auto_refresh_kwargs: Extra arguments to pass to the refresh token
68	endpoint.
69	:token_updater: Method with one argument, token, to be used to update
70	your token database on automatic token refresh. If not
71	set a TokenUpdated warning will be raised when a token
72	has been refreshed. This warning will carry the token
73	in its token argument.
74	:param pkce: Set "S256" or "plain" to enable PKCE. Default is disabled.
75	:param kwargs: Arguments to pass to the Session constructor.
76	"""
77	super(OAuth2Session, self).__init__(**kwargs)	6✔
78	self._client = client or WebApplicationClient(client_id, token=token)	6✔
79	self.token = token or {}	6✔
80	self._scope = scope	6✔
81	self.redirect_uri = redirect_uri	6✔
82	self.state = state or generate_token	6✔
83	self._state = state	6✔
84	self.auto_refresh_url = auto_refresh_url	6✔
85	self.auto_refresh_kwargs = auto_refresh_kwargs or {}	6✔
86	self.token_updater = token_updater	6✔
87	self._pkce = pkce	6✔
88
89	if self._pkce not in ["S256", "plain", None]:	6✔
90	raise AttributeError("Wrong value for {}(.., pkce={})".format(self.__class__, self._pkce))	×
91
92	# Ensure that requests doesn't do any automatic auth. See #278.
93	# The default behavior can be re-enabled by setting auth to None.
94	self.auth = lambda r: r	6✔
95
96	# Allow customizations for non compliant providers through various
97	# hooks to adjust requests and responses.
98	self.compliance_hook = {	6✔
99	"access_token_response": set(),
100	"refresh_token_response": set(),
101	"protected_request": set(),
102	"refresh_token_request": set(),
103	"access_token_request": set(),
104	}
105
106	@property	6✔
107	def scope(self):	6✔
108	"""By default the scope from the client is used, except if overridden"""
109	if self._scope is not None:	6✔
110	return self._scope	×
111	elif self._client is not None:	6✔
112	return self._client.scope	6✔
113	else:
114	return None	×
115
116	@scope.setter	6✔
117	def scope(self, scope):	6✔
118	self._scope = scope	×
119
120	def new_state(self):	6✔
121	"""Generates a state string to be used in authorizations."""
122	try:	6✔
123	self._state = self.state()	6✔
124	log.debug("Generated new state %s.", self._state)	6✔
125	except TypeError:	×
126	self._state = self.state	×
127	log.debug("Re-using previously supplied state %s.", self._state)	×
128	return self._state	6✔
129
130	@property	6✔
131	def client_id(self):	6✔
132	return getattr(self._client, "client_id", None)	6✔
133
134	@client_id.setter	6✔
135	def client_id(self, value):	6✔
136	self._client.client_id = value	6✔
137
138	@client_id.deleter	6✔
139	def client_id(self):	6✔
140	del self._client.client_id	6✔
141
142	@property	6✔
143	def token(self):	6✔
144	return getattr(self._client, "token", None)	6✔
145
146	@token.setter	6✔
147	def token(self, value):	6✔
148	self._client.token = value	6✔
149	self._client.populate_token_attributes(value)	6✔
150
151	@property	6✔
152	def access_token(self):	6✔
153	return getattr(self._client, "access_token", None)	6✔
154
155	@access_token.setter	6✔
156	def access_token(self, value):	6✔
157	self._client.access_token = value	6✔
158
159	@access_token.deleter	6✔
160	def access_token(self):	6✔
161	del self._client.access_token	6✔
162
163	@property	6✔
164	def authorized(self):	6✔
165	"""Boolean that indicates whether this session has an OAuth token
166	or not. If `self.authorized` is True, you can reasonably expect
167	OAuth-protected requests to the resource to succeed. If
168	`self.authorized` is False, you need the user to go through the OAuth
169	authentication dance before OAuth-protected requests to the resource
170	will succeed.
171	"""
172	return bool(self.access_token)	6✔
173
174	def authorization_url(self, url, state=None, **kwargs):	6✔
175	"""Form an authorization URL.
176
177	:param url: Authorization endpoint url, must be HTTPS.
178	:param state: An optional state string for CSRF protection. If not
179	given it will be generated for you.
180	:param kwargs: Extra parameters to include.
181	:return: authorization_url, state
182	"""
183	state = state or self.new_state()	6✔
184	if self._pkce:	6✔
185	self._code_verifier = self._client.create_code_verifier(43)	6✔
186	kwargs["code_challenge_method"] = self._pkce	6✔
187	kwargs["code_challenge"] = self._client.create_code_challenge(	6✔
188	code_verifier=self._code_verifier,
189	code_challenge_method=self._pkce
190	)
191	return (	6✔
192	self._client.prepare_request_uri(
193	url,
194	redirect_uri=self.redirect_uri,
195	scope=self.scope,
196	state=state,
197	**kwargs
198	),
199	state,
200	)
201
202	def fetch_token(	6✔
203	self,
204	token_url,
205	code=None,
206	authorization_response=None,
207	body="",
208	auth=None,
209	username=None,
210	password=None,
211	method="POST",
212	force_querystring=False,
213	timeout=None,
214	headers=None,
215	verify=None,
216	proxies=None,
217	include_client_id=None,
218	client_secret=None,
219	cert=None,
220	**kwargs
221	):
222	"""Generic method for fetching an access token from the token endpoint.
223
224	If you are using the MobileApplicationClient you will want to use
225	`token_from_fragment` instead of `fetch_token`.
226
227	The current implementation enforces the RFC guidelines.
228
229	:param token_url: Token endpoint URL, must use HTTPS.
230	:param code: Authorization code (used by WebApplicationClients).
231	:param authorization_response: Authorization response URL, the callback
232	URL of the request back to you. Used by
233	WebApplicationClients instead of code.
234	:param body: Optional application/x-www-form-urlencoded body to add the
235	include in the token request. Prefer kwargs over body.
236	:param auth: An auth tuple or method as accepted by `requests`.
237	:param username: Username required by LegacyApplicationClients to appear
238	in the request body.
239	:param password: Password required by LegacyApplicationClients to appear
240	in the request body.
241	:param method: The HTTP method used to make the request. Defaults
242	to POST, but may also be GET. Other methods should
243	be added as needed.
244	:param force_querystring: If True, force the request body to be sent
245	in the querystring instead.
246	:param timeout: Timeout of the request in seconds.
247	:param headers: Dict to default request headers with.
248	:param verify: Verify SSL certificate.
249	:param proxies: The `proxies` argument is passed onto `requests`.
250	:param include_client_id: Should the request body include the
251	`client_id` parameter. Default is `None`,
252	which will attempt to autodetect. This can be
253	forced to always include (True) or never
254	include (False).
255	:param client_secret: The `client_secret` paired to the `client_id`.
256	This is generally required unless provided in the
257	`auth` tuple. If the value is `None`, it will be
258	omitted from the request, however if the value is
259	an empty string, an empty string will be sent.
260	:param cert: Client certificate to send for OAuth 2.0 Mutual-TLS Client
261	Authentication (draft-ietf-oauth-mtls). Can either be the
262	path of a file containing the private key and certificate or
263	a tuple of two filenames for certificate and key.
264	:param kwargs: Extra parameters to include in the token request.
265	:return: A token dict
266	"""
267	if not is_secure_transport(token_url):	6✔
268	raise InsecureTransportError()	×
269
270	if not code and authorization_response:	6✔
271	self._client.parse_request_uri_response(	6✔
272	authorization_response, state=self._state
273	)
274	code = self._client.code	6✔
275	elif not code and isinstance(self._client, WebApplicationClient):	6✔
276	code = self._client.code	6✔
277	if not code:	6✔
278	raise ValueError(	×
279	"Please supply either code or " "authorization_response parameters."
280	)
281
282	if self._pkce:	6✔
283	if self._code_verifier is None:	6✔
284	raise ValueError(	×
285	"Code verifier is not found, authorization URL must be generated before"
286	)
287	kwargs["code_verifier"] = self._code_verifier	6✔
288
289	# Earlier versions of this library build an HTTPBasicAuth header out of
290	# `username` and `password`. The RFC states, however these attributes
291	# must be in the request body and not the header.
292	# If an upstream server is not spec compliant and requires them to
293	# appear as an Authorization header, supply an explicit `auth` header
294	# to this function.
295	# This check will allow for empty strings, but not `None`.
296	#
297	# References
298	# 4.3.2 - Resource Owner Password Credentials Grant
299	# https://tools.ietf.org/html/rfc6749#section-4.3.2
300
301	if isinstance(self._client, LegacyApplicationClient):	6✔
302	if username is None:	6✔
303	raise ValueError(	6✔
304	"`LegacyApplicationClient` requires both the "
305	"`username` and `password` parameters."
306	)
307	if password is None:	6✔
308	raise ValueError(	6✔
309	"The required parameter `username` was supplied, "
310	"but `password` was not."
311	)
312
313	# merge username and password into kwargs for `prepare_request_body`
314	if username is not None:	6✔
315	kwargs["username"] = username	6✔
316	if password is not None:	6✔
317	kwargs["password"] = password	6✔
318
319	# is an auth explicitly supplied?
320	if auth is not None:	6✔
321	# if we're dealing with the default of `include_client_id` (None):
322	# we will assume the `auth` argument is for an RFC compliant server
323	# and we should not send the `client_id` in the body.
324	# This approach allows us to still force the client_id by submitting
325	# `include_client_id=True` along with an `auth` object.
326	if include_client_id is None:	6✔
327	include_client_id = False	6✔
328
329	# otherwise we may need to create an auth header
330	else:
331	# since we don't have an auth header, we MAY need to create one
332	# it is possible that we want to send the `client_id` in the body
333	# if so, `include_client_id` should be set to True
334	# otherwise, we will generate an auth header
335	if include_client_id is not True:	6✔
336	client_id = self.client_id	6✔
337	if client_id:	6✔
338	log.debug(	6✔
339	'Encoding `client_id` "%s" with `client_secret` '
340	"as Basic auth credentials.",
341	client_id,
342	)
343	client_secret = client_secret if client_secret is not None else ""	6✔
344	auth = requests.auth.HTTPBasicAuth(client_id, client_secret)	6✔
345
346	if include_client_id:	6✔
347	# this was pulled out of the params
348	# it needs to be passed into prepare_request_body
349	if client_secret is not None:	6✔
350	kwargs["client_secret"] = client_secret	6✔
351
352	body = self._client.prepare_request_body(	6✔
353	code=code,
354	body=body,
355	redirect_uri=self.redirect_uri,
356	include_client_id=include_client_id,
357	**kwargs
358	)
359
360	headers = headers or {	6✔
361	"Accept": "application/json",
362	"Content-Type": "application/x-www-form-urlencoded",
363	}
364	self.token = {}	6✔
365	request_kwargs = {}	6✔
366	if method.upper() == "POST":	6✔
367	request_kwargs["params" if force_querystring else "data"] = dict(	6✔
368	urldecode(body)
369	)
370	elif method.upper() == "GET":	×
371	request_kwargs["params"] = dict(urldecode(body))	×
372	else:
373	raise ValueError("The method kwarg must be POST or GET.")	×
374
375	for hook in self.compliance_hook["access_token_request"]:	6✔
376	log.debug("Invoking access_token_request hook %s.", hook)	6✔
377	token_url, headers, request_kwargs = hook(	6✔
378	token_url, headers, request_kwargs
379	)
380
381	r = self.request(	6✔
382	method=method,
383	url=token_url,
384	timeout=timeout,
385	headers=headers,
386	auth=auth,
387	verify=verify,
388	proxies=proxies,
389	cert=cert,
390	**request_kwargs
391	)
392
393	log.debug("Request to fetch token completed with status %s.", r.status_code)	6✔
394	log.debug("Request url was %s", r.request.url)	6✔
395	log.debug("Request headers were %s", r.request.headers)	6✔
396	log.debug("Request body was %s", r.request.body)	6✔
397	log.debug("Response headers were %s and content %s.", r.headers, r.text)	6✔
398	log.debug(	6✔
399	"Invoking %d token response hooks.",
400	len(self.compliance_hook["access_token_response"]),
401	)
402	for hook in self.compliance_hook["access_token_response"]:	6✔
403	log.debug("Invoking hook %s.", hook)	6✔
404	r = hook(r)	6✔
405
406	self._client.parse_request_body_response(r.text, scope=self.scope)	6✔
407	self.token = self._client.token	6✔
408	log.debug("Obtained token %s.", self.token)	6✔
409	return self.token	6✔
410
411	def token_from_fragment(self, authorization_response):	6✔
412	"""Parse token from the URI fragment, used by MobileApplicationClients.
413
414	:param authorization_response: The full URL of the redirect back to you
415	:return: A token dict
416	"""
417	self._client.parse_request_uri_response(	6✔
418	authorization_response, state=self._state
419	)
420	self.token = self._client.token	6✔
421	return self.token	6✔
422
423	def refresh_token(	6✔
424	self,
425	token_url,
426	refresh_token=None,
427	body="",
428	auth=None,
429	timeout=None,
430	headers=None,
431	verify=None,
432	proxies=None,
433	**kwargs
434	):
435	"""Fetch a new access token using a refresh token.
436
437	:param token_url: The token endpoint, must be HTTPS.
438	:param refresh_token: The refresh_token to use.
439	:param body: Optional application/x-www-form-urlencoded body to add the
440	include in the token request. Prefer kwargs over body.
441	:param auth: An auth tuple or method as accepted by `requests`.
442	:param timeout: Timeout of the request in seconds.
443	:param headers: A dict of headers to be used by `requests`.
444	:param verify: Verify SSL certificate.
445	:param proxies: The `proxies` argument will be passed to `requests`.
446	:param kwargs: Extra parameters to include in the token request.
447	:return: A token dict
448	"""
449	if not token_url:	6✔
450	raise ValueError("No token endpoint set for auto_refresh.")	×
451
452	if not is_secure_transport(token_url):	6✔
453	raise InsecureTransportError()	×
454
455	refresh_token = refresh_token or self.token.get("refresh_token")	6✔
456
457	log.debug(	6✔
458	"Adding auto refresh key word arguments %s.", self.auto_refresh_kwargs
459	)
460	kwargs.update(self.auto_refresh_kwargs)	6✔
461	body = self._client.prepare_refresh_body(	6✔
462	body=body, refresh_token=refresh_token, scope=self.scope, **kwargs
463	)
464	log.debug("Prepared refresh token request body %s", body)	6✔
465
466	if headers is None:	6✔
467	headers = {	6✔
468	"Accept": "application/json",
469	"Content-Type": ("application/x-www-form-urlencoded"),
470	}
471
472	for hook in self.compliance_hook["refresh_token_request"]:	6✔
473	log.debug("Invoking refresh_token_request hook %s.", hook)	6✔
474	token_url, headers, body = hook(token_url, headers, body)	6✔
475
476	r = self.post(	6✔
477	token_url,
478	data=dict(urldecode(body)),
479	auth=auth,
480	timeout=timeout,
481	headers=headers,
482	verify=verify,
483	withhold_token=True,
484	proxies=proxies,
485	)
486	log.debug("Request to refresh token completed with status %s.", r.status_code)	6✔
487	log.debug("Response headers were %s and content %s.", r.headers, r.text)	6✔
488	log.debug(	6✔
489	"Invoking %d token response hooks.",
490	len(self.compliance_hook["refresh_token_response"]),
491	)
492	for hook in self.compliance_hook["refresh_token_response"]:	6✔
493	log.debug("Invoking hook %s.", hook)	6✔
494	r = hook(r)	6✔
495
496	self.token = self._client.parse_request_body_response(r.text, scope=self.scope)	6✔
497	if "refresh_token" not in self.token:	6✔
498	log.debug("No new refresh token given. Re-using old.")	6✔
499	self.token["refresh_token"] = refresh_token	6✔
500	return self.token	6✔
501
502	def request(	6✔
503	self,
504	method,
505	url,
506	data=None,
507	headers=None,
508	withhold_token=False,
509	client_id=None,
510	client_secret=None,
511	files=None,
512	**kwargs
513	):
514	"""Intercept all requests and add the OAuth 2 token if present."""
515	if not is_secure_transport(url):	6✔
516	raise InsecureTransportError()	×
517	if self.token and not withhold_token:	6✔
518	log.debug(	6✔
519	"Invoking %d protected resource request hooks.",
520	len(self.compliance_hook["protected_request"]),
521	)
522	for hook in self.compliance_hook["protected_request"]:	6✔
523	log.debug("Invoking hook %s.", hook)	6✔
524	url, headers, data = hook(url, headers, data)	6✔
525
526	log.debug("Adding token %s to request.", self.token)	6✔
527	try:	6✔
528	url, headers, data = self._client.add_token(	6✔
529	url, http_method=method, body=data, headers=headers
530	)
531	# Attempt to retrieve and save new access token if expired
532	except TokenExpiredError:	6✔
533	if self.auto_refresh_url:	6✔
534	log.debug(	6✔
535	"Auto refresh is set, attempting to refresh at %s.",
536	self.auto_refresh_url,
537	)
538
539	# We mustn't pass auth twice.
540	auth = kwargs.pop("auth", None)	6✔
541	if client_id and client_secret and (auth is None):	6✔
542	log.debug(	6✔
543	'Encoding client_id "%s" with client_secret as Basic auth credentials.',
544	client_id,
545	)
546	auth = requests.auth.HTTPBasicAuth(client_id, client_secret)	6✔
547	token = self.refresh_token(	6✔
548	self.auto_refresh_url, auth=auth, **kwargs
549	)
550	if self.token_updater:	6✔
551	log.debug(	6✔
552	"Updating token to %s using %s.", token, self.token_updater
553	)
554	self.token_updater(token)	6✔
555	url, headers, data = self._client.add_token(	6✔
556	url, http_method=method, body=data, headers=headers
557	)
558	else:
559	raise TokenUpdated(token)	6✔
560	else:
561	raise	6✔
562
563	log.debug("Requesting url %s using method %s.", url, method)	6✔
564	log.debug("Supplying headers %s and data %s", headers, data)	6✔
565	log.debug("Passing through key word arguments %s.", kwargs)	6✔
566	return super(OAuth2Session, self).request(	6✔
567	method, url, headers=headers, data=data, files=files, **kwargs
568	)
569
570	def register_compliance_hook(self, hook_type, hook):	6✔
571	"""Register a hook for request/response tweaking.
572
573	Available hooks are:
574	access_token_response invoked before token parsing.
575	refresh_token_response invoked before refresh token parsing.
576	protected_request invoked before making a request.
577	access_token_request invoked before making a token fetch request.
578	refresh_token_request invoked before making a refresh request.
579
580	If you find a new hook is needed please send a GitHub PR request
581	or open an issue.
582	"""
583	if hook_type not in self.compliance_hook:	6✔
584	raise ValueError(	×
585	"Hook type %s is not in %s.", hook_type, self.compliance_hook
586	)
587	self.compliance_hook[hook_type].add(hook)	6✔

requests / requests-oauthlib / 8621812797

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous