87d7ad5358d3bd4d01d56cfc4738ebfa40b373b5-PR-128

Build # 87d7ad5358d3bd4d01d56cfc4738ebfa40b373b5-PR-128

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/praw-dev/prawcore/commit/<a class=hub.com/praw-dev/prawcore/commit/<a class="double-link" href="https://git"><a class=hub.com/praw-dev/prawcore/commit/<a class="double-link" href="https://git"><a class=hub.com/praw-dev/prawcore/commit/<a class="double-link" href="https://git"><a class=hub.com/praw-dev/prawcore/commit/87d7ad5358d3bd4d01d56cfc4738ebfa40b373b5">87d7ad535">&lt;a href=&quot;https://github.com/praw-dev/prawcore/commit/</a><a class="double-link" href="https://github.com/praw-dev/prawcore/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/praw-dev/prawcore/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/praw-dev/prawcore/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/praw-dev/prawcore/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/praw-dev/prawcore/commit/87d7ad5358d3bd4d01d56cfc4738ebfa40b373b5">87d7ad535</a><a href="https://github.com/praw-dev/prawcore/commit/87d7ad5358d3bd4d01d56cfc4738ebfa40b373b5">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/praw-dev/prawcore/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/praw-dev/prawcore/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/praw-dev/prawcore/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/praw-dev/prawcore/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/praw-dev/prawcore/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/praw-dev/prawcore/commit/87d7ad5358d3bd4d01d56cfc4738ebfa40b373b5&quot;&gt;87d7ad535&lt;/a&gt;&lt;a href=&quot;https://github.com/praw-dev/prawcore/commit/87d7ad5358d3bd4d01d56cfc4738ebfa40b373b5&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/praw-dev/prawcore/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/praw-dev/prawcore/commit/&amp;amp;lt;a ... (continued)

Pull Request Pull Request #128: Add local WSGI server authorizer

Run Details

17 of 75 new or added lines in 1 file covered. (22.67%)

413 of 471 relevant lines covered (87.69%)

3.51 hits per line

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

73.39

/prawcore/auth.py

"""Provides Authentication and Authorization classes."""
import time
from typing import Any, Dict, Optional

from uuid import uuid4
from urllib.parse import parse_qs, urlparse
from wsgiref.simple_server import make_server

from requests import Request
from requests.status_codes import codes

from . import const
from .exceptions import InvalidInvocation, OAuthException, ResponseException


class BaseAuthenticator(object):
    """Provide the base authenticator object that stores OAuth2 credentials."""

    def __init__(self, requestor, client_id, redirect_uri=None):
        """Represent a single authentication to Reddit's API.

        :param requestor: An instance of :class:`Requestor`.
        :param client_id: The OAuth2 client ID to use with the session.
        :param redirect_uri: (optional) The redirect URI exactly as specified in your
            OAuth application settings on Reddit. This parameter is required if you want
            to use the ``authorize_url`` method, or the ``authorize`` method of the
            ``Authorizer`` class.

        """
        self._requestor = requestor
        self.client_id = client_id
        self.redirect_uri = redirect_uri

    def _post(self, url, success_status=codes["ok"], **data):
        response = self._requestor.request(
            "post",
            url,
            auth=self._auth(),
            data=sorted(data.items()),
            headers={"Connection": "close"},
        )
        if response.status_code != success_status:
            raise ResponseException(response)
        return response

    def authorize_url(self, duration, scopes, state, implicit=False):
        """Return the URL used out-of-band to grant access to your application.

        :param duration: Either ``permanent`` or ``temporary``. ``temporary``
            authorizations generate access tokens that last only 1 hour. ``permanent``
            authorizations additionally generate a refresh token that can be
            indefinitely used to generate new hour-long access tokens. Only
            ``temporary`` can be specified if ``implicit`` is set to ``True``.
        :param scopes: A list of OAuth scopes to request authorization for.
        :param state: A string that will be reflected in the callback to
            ``redirect_uri``. Elements must be printable ASCII characters in the range
            0x20 through 0x7E inclusive. This value should be temporarily unique to the
            client for whom the URL was generated.
        :param implicit: (optional) Use the implicit grant flow (default: False). This
            flow is only available for UntrustedAuthenticators.

        """
        if self.redirect_uri is None:
            raise InvalidInvocation("redirect URI not provided")
        if implicit and not isinstance(self, UntrustedAuthenticator):
            raise InvalidInvocation(
                "Only UntrustedAuthenticator instances can "
                "use the implicit grant flow."
            )
        if implicit and duration != "temporary":
            raise InvalidInvocation(
                "The implicit grant flow only supports "
                "temporary access tokens."
            )

        params = {
            "client_id": self.client_id,
            "duration": duration,
            "redirect_uri": self.redirect_uri,
            "response_type": "token" if implicit else "code",
            "scope": " ".join(scopes),
            "state": state,
        }
        url = self._requestor.reddit_url + const.AUTHORIZATION_PATH
        request = Request("GET", url, params=params)
        return request.prepare().url

    def revoke_token(self, token, token_type=None):
        """Ask Reddit to revoke the provided token.

        :param token: The access or refresh token to revoke.
        :param token_type: (Optional) When provided, hint to Reddit what the token type
            is for a possible efficiency gain. The value can be either ``access_token``
            or ``refresh_token``.

        """
        data = {"token": token}
        if token_type is not None:
            data["token_type_hint"] = token_type
        url = self._requestor.reddit_url + const.REVOKE_TOKEN_PATH
        self._post(url, **data)


class TrustedAuthenticator(BaseAuthenticator):
    """Store OAuth2 authentication credentials for web, or script type apps."""

    RESPONSE_TYPE = "code"

    def __init__(self, requestor, client_id, client_secret, redirect_uri=None):
        """Represent a single authentication to Reddit's API.

        :param requestor: An instance of :class:`Requestor`.
        :param client_id: The OAuth2 client ID to use with the session.
        :param client_secret: The OAuth2 client secret to use with the session.
        :param redirect_uri: (optional) The redirect URI exactly as specified in your
            OAuth application settings on Reddit. This parameter is required if you want
            to use the ``authorize_url`` method, or the ``authorize`` method of the
            ``Authorizer`` class.

        """
        super(TrustedAuthenticator, self).__init__(
            requestor, client_id, redirect_uri
        )
        self.client_secret = client_secret

    def _auth(self):
        return self.client_id, self.client_secret


class UntrustedAuthenticator(BaseAuthenticator):
    """Store OAuth2 authentication credentials for installed applications."""

    def _auth(self):
        return self.client_id, ""


class BaseAuthorizer(object):
    """Superclass for OAuth2 authorization tokens and scopes."""

    def __init__(self, authenticator):
        """Represent a single authorization to Reddit's API.

        :param authenticator: An instance of :class:`BaseAuthenticator`.

        """
        self._authenticator = authenticator
        self._clear_access_token()
        self._validate_authenticator()

    def _clear_access_token(self):
        self._expiration_timestamp = None
        self.access_token = None
        self.scopes = None

    def _request_token(self, **data):
        url = (
            self._authenticator._requestor.reddit_url + const.ACCESS_TOKEN_PATH
        )
        pre_request_time = time.time()
        response = self._authenticator._post(url, **data)
        payload = response.json()
        if "error" in payload:  # Why are these OKAY responses?
            raise OAuthException(
                response, payload["error"], payload.get("error_description")
            )

        self._expiration_timestamp = (
            pre_request_time - 10 + payload["expires_in"]
        )
        self.access_token = payload["access_token"]
        if "refresh_token" in payload:
            self.refresh_token = payload["refresh_token"]
        self.scopes = set(payload["scope"].split(" "))

    def _validate_authenticator(self):
        if not isinstance(self._authenticator, self.AUTHENTICATOR_CLASS):
            raise InvalidInvocation(
                "Must use a authenticator of type"
                f" {self.AUTHENTICATOR_CLASS.__name__}."
            )

    def is_valid(self):
        """Return whether or not the Authorizer is ready to authorize requests.

        A ``True`` return value does not guarantee that the access_token is actually
        valid on the server side.

        """
        return (
            self.access_token is not None
            and time.time() < self._expiration_timestamp
        )

    def revoke(self):
        """Revoke the current Authorization."""
        if self.access_token is None:
            raise InvalidInvocation("no token available to revoke")

        self._authenticator.revoke_token(self.access_token, "access_token")
        self._clear_access_token()


class Authorizer(BaseAuthorizer):
    """Manages OAuth2 authorization tokens and scopes."""

    AUTHENTICATOR_CLASS = BaseAuthenticator

    def __init__(
        self,
        authenticator,
        *,
        post_refresh_callback=None,
        pre_refresh_callback=None,
        refresh_token=None,
    ):
        """Represent a single authorization to Reddit's API.

        :param authenticator: An instance of a subclass of :class:`BaseAuthenticator`.
        :param post_refresh_callback: (Optional) When a single-argument
            function is passed, the function will be called prior to refreshing
            the access and refresh tokens. The argument to the callback is the
            :class:`Authorizer` instance. This callback can be used to inspect
            and modify the attributes of the :class:`Authorizer`.
        :param pre_refresh_callback: (Optional) When a single-argument function
            is passed, the function will be called after refreshing the access
            and refresh tokens. The argument to the callback is the
            :class:`Authorizer` instance. This callback can be used to inspect
            and modify the attributes of the :class:`Authorizer`.
        :param refresh_token: (Optional) Enables the ability to refresh the
            authorization.

        """
        super(Authorizer, self).__init__(authenticator)
        self._post_refresh_callback = post_refresh_callback
        self._pre_refresh_callback = pre_refresh_callback
        self.refresh_token = refresh_token

    def authorize(self, code):
        """Obtain and set authorization tokens based on ``code``.

        :param code: The code obtained by an out-of-band authorization request to
            Reddit.

        """
        if self._authenticator.redirect_uri is None:
            raise InvalidInvocation("redirect URI not provided")
        self._request_token(
            code=code,
            grant_type="authorization_code",
            redirect_uri=self._authenticator.redirect_uri,
        )

    def refresh(self):
        """Obtain a new access token from the refresh_token."""
        if self._pre_refresh_callback:
            self._pre_refresh_callback(self)
        if self.refresh_token is None:
            raise InvalidInvocation("refresh token not provided")
        self._request_token(
            grant_type="refresh_token", refresh_token=self.refresh_token
        )
        if self._post_refresh_callback:
            self._post_refresh_callback(self)

    def revoke(self, only_access=False):
        """Revoke the current Authorization.

        :param only_access: (Optional) When explicitly set to True, do not evict the
            refresh token if one is set.

        Revoking a refresh token will in-turn revoke all access tokens associated with
        that authorization.

        """
        if only_access or self.refresh_token is None:
            super(Authorizer, self).revoke()
        else:
            self._authenticator.revoke_token(
                self.refresh_token, "refresh_token"
            )
            self._clear_access_token()
            self.refresh_token = None


class _OAuth2ClientUserAuthApp(object):
    def __init__(self, authenticator, scopes, implicit=False, duration="permanent") -> None:
        super().__init__()
        self.__finished = False
        self.__state = str(uuid4())
        self.__auth_url = authenticator.authorize_url(duration, scopes, self.__state, implicit)
        self.__redirect_uri = authenticator.redirect_uri
        self.__callback_uri = urlparse(self.__redirect_uri).path
        self.__auth_exchange_data = {}

    def __call__(self, environ, start_resp):
        if not self.__finished:
            req_method = environ["REQUEST_METHOD"]
            req_uri = environ["PATH_INFO"]
            req_query = parse_qs(environ["QUERY_STRING"])

            if req_method == "GET":
                if req_uri == "/":
                    start_resp("200 OK", [("Content-type", "text/html; charset=utf-8")])
                    return ['<a href="/auth">Authenticate with reddit</a>'.encode("utf8")]

                elif req_uri == "/auth":
                    start_resp("302 Moved Temporarily", [("Location", self.__auth_url)])
                    return []

                elif req_uri == self.__callback_uri:
                    if "error" not in req_query:
                        req_state = req_query["state"][0]

                        if req_state == self.__state:
                            if "code" in req_query:
                                self.__auth_exchange_data.update({"code": req_query["code"][0]})

                            else:
                                self.__auth_exchange_data.update({
                                    "access_token": req_query["access_token"][0],
                                    "expires_in": req_query["expires_in"][0],
                                    "scope": req_query["scope"][0]
                                })

                            start_resp("200 OK", [])
                            received = "Authorization Code" if "code" in self.__auth_exchange_data else "User Access Token"
                            self.__finished = True
                            return [f"Received {received} Successfully! You may safely close this page.".encode("utf8")]

                        else:
                            start_resp("200 OK", [])
                            return [f"State mismatch!\nExpected State: {self.__state}\nReceived State: {req_state}".encode("utf8")]

                    else:
                        start_resp("200 OK", [])
                        err_val = req_query["error"]

                        if err_val == "access_denied":
                            return ["User denied permission!".encode("utf8")]

                        elif err_val == "unsupported_response_type":
                            return ["Invalid initial authorization response_type!".encode("utf8")]

                        elif err_val == "invalid_scope":
                            return ["Invalid authorization scope(s) requested!".encode("utf8")]

                        elif err_val == "invalid_request":
                            return ["Invalid authorization request!".encode("utf8")]

                        else:
                            return ["Unknown Error!" + f"\nERROR: {err_val}".encode("utf8")]

                else:
                    start_resp("404 Not Found", [])
                    return ["Unknown URI!".encode("utf8")]

            else:
                start_resp("405 Method Not Allowed", [])
                return ["Specified HTTP Request Method is not accepted by server!".encode("utf8")]

    @property
    def finished(self) -> bool:
        return self.__finished

    @property
    def auth_code(self) -> Optional[str]:
        return self.__auth_exchange_data["code"] if "code" in self.__auth_exchange_data else None

    @property
    def implicit_grant(self) -> Optional[Dict[str, Any]]:
        return self.__auth_exchange_data if "access_token" in self.__auth_exchange_data else None


class LocalWSGIServerAuthorizer(Authorizer):
    AUTHENTICATOR_CLASS = TrustedAuthenticator

    def __init__(
        self,
        authenticator,
        scopes,
        duration="permanent",
    ):
        super(LocalWSGIServerAuthorizer, self).__init__(authenticator)
        self.scopes = scopes
        self.duration = duration

    def authorize_local_server(self):
        auth_app = _OAuth2ClientUserAuthApp(
            self._authenticator,
            self.scopes,
            implicit=False,
            duration=self.duration,
        )

        redirect_uri = urlparse(self._authenticator.redirect_uri)
        auth_server = make_server(
            redirect_uri.netloc.split(":")[0],
            redirect_uri.netloc.split(":")[1] if ":" in redirect_uri.netloc else 80,
            auth_app
        )

        while not auth_app.finished:
            auth_server.handle_request()

        self.authorize(auth_app.auth_code)


class DeviceIDAuthorizer(BaseAuthorizer):
    """Manages app-only OAuth2 for 'installed' applications.

    While the '*' scope will be available, some endpoints simply will not work due to
    the lack of an associated Reddit account.

    """

    AUTHENTICATOR_CLASS = UntrustedAuthenticator

    def __init__(
        self, authenticator, device_id="DO_NOT_TRACK_THIS_DEVICE", scopes=None
    ):
        """Represent an app-only OAuth2 authorization for 'installed' apps.

        :param authenticator: An instance of :class:`UntrustedAuthenticator`.
        :param device_id: (optional) A unique ID (20-30 character ASCII string) (default
            DO_NOT_TRACK_THIS_DEVICE). For more information about this parameter, see:
            https://github.com/reddit/reddit/wiki/OAuth2#application-only-oauth
        :param scopes: (Optional) A list of OAuth scopes to request authorization for
            (default: None). The scope ``*`` is requested when the default argument is
            used.

        """
        super(DeviceIDAuthorizer, self).__init__(authenticator)
        self._device_id = device_id
        self._scopes = scopes

    def refresh(self):
        """Obtain a new access token."""
        additional_kwargs = {}
        if self._scopes:
            additional_kwargs["scope"] = " ".join(self._scopes)
        grant_type = "https://oauth.reddit.com/grants/installed_client"
        self._request_token(
            grant_type=grant_type,
            device_id=self._device_id,
            **additional_kwargs,
        )


class ImplicitAuthorizer(BaseAuthorizer):
    """Manages implicit installed-app type authorizations."""

    AUTHENTICATOR_CLASS = UntrustedAuthenticator

    def __init__(self, authenticator, access_token, expires_in, scope):
        """Represent a single implicit authorization to Reddit's API.

        :param authenticator: An instance of :class:`UntrustedAuthenticator`.
        :param access_token: The access_token obtained from Reddit via callback to the
            authenticator's redirect_uri.
        :param expires_in: The number of seconds the ``access_token`` is valid for. The
            origin of this value was returned from Reddit via callback to the
            authenticator's redirect uri. Note, you may need to subtract an offset
            before passing in this number to account for a delay between when Reddit
            prepared the response, and when you make this function call.
        :param scope: A space-delimited string of Reddit OAuth2 scope names as returned
            from Reddit in the callback to the authenticator's redirect uri.

        """
        super(ImplicitAuthorizer, self).__init__(authenticator)
        self._expiration_timestamp = time.time() + expires_in
        self.access_token = access_token
        self.scopes = set(scope.split(" "))


class ReadOnlyAuthorizer(Authorizer):
    """Manages authorizations that are not associated with a Reddit account.

    While the '*' scope will be available, some endpoints simply will not work due to
    the lack of an associated Reddit account.

    """

    AUTHENTICATOR_CLASS = TrustedAuthenticator

    def __init__(self, authenticator, scopes=None):
        """Represent a ReadOnly authorization to Reddit's API.

        :param scopes: (Optional) A list of OAuth scopes to request authorization for
            (default: None). The scope ``*`` is requested when the default argument is
            used.

        """
        super().__init__(authenticator)
        self._scopes = scopes

    def refresh(self):
        """Obtain a new ReadOnly access token."""
        additional_kwargs = {}
        if self._scopes:
            additional_kwargs["scope"] = " ".join(self._scopes)
        self._request_token(
            grant_type="client_credentials", **additional_kwargs
        )


class ScriptAuthorizer(Authorizer):
    """Manages personal-use script type authorizations.

    Only users who are listed as developers for the application will be granted access
    tokens.

    """

    AUTHENTICATOR_CLASS = TrustedAuthenticator

    def __init__(
        self,
        authenticator,
        username,
        password,
        two_factor_callback=None,
        scopes=None,
    ):
        """Represent a single personal-use authorization to Reddit's API.

        :param authenticator: An instance of :class:`TrustedAuthenticator`.
        :param username: The Reddit username of one of the application's developers.
        :param password: The password associated with ``username``.
        :param two_factor_callback: A function that returns OTPs (One-Time
            Passcodes), also known as 2FA auth codes. If this function is
            provided, prawcore will call it when authenticating.
        :param scopes: (Optional) A list of OAuth scopes to request authorization for
            (default: None). The scope ``*`` is requested when the default argument is
            used.

        """
        super(ScriptAuthorizer, self).__init__(authenticator)
        self._password = password
        self._scopes = scopes
        self._two_factor_callback = two_factor_callback
        self._username = username

    def refresh(self):
        """Obtain a new personal-use script type access token."""
        additional_kwargs = {}
        if self._scopes:
            additional_kwargs["scope"] = " ".join(self._scopes)
        two_factor_code = (
            self._two_factor_callback and self._two_factor_callback()
        )
        if two_factor_code:
            additional_kwargs["otp"] = two_factor_code
        self._request_token(
            grant_type="password",
            username=self._username,
            password=self._password,
            **additional_kwargs,
        )

1	"""Provides Authentication and Authorization classes."""
2	import time	4✔
3	from typing import Any, Dict, Optional	4✔
4
5	from uuid import uuid4	4✔
6	from urllib.parse import parse_qs, urlparse	4✔
7	from wsgiref.simple_server import make_server	4✔
8
9	from requests import Request	4✔
10	from requests.status_codes import codes	4✔
11
12	from . import const	4✔
13	from .exceptions import InvalidInvocation, OAuthException, ResponseException	4✔
14
15
16	class BaseAuthenticator(object):	4✔
17	"""Provide the base authenticator object that stores OAuth2 credentials."""
18
19	def __init__(self, requestor, client_id, redirect_uri=None):	4✔
20	"""Represent a single authentication to Reddit's API.
21
22	:param requestor: An instance of :class:`Requestor`.
23	:param client_id: The OAuth2 client ID to use with the session.
24	:param redirect_uri: (optional) The redirect URI exactly as specified in your
25	OAuth application settings on Reddit. This parameter is required if you want
26	to use the ``authorize_url`` method, or the ``authorize`` method of the
27	``Authorizer`` class.
28
29	"""
30	self._requestor = requestor	4✔
31	self.client_id = client_id	4✔
32	self.redirect_uri = redirect_uri	4✔
33
34	def _post(self, url, success_status=codes["ok"], **data):	4✔
35	response = self._requestor.request(	4✔
36	"post",
37	url,
38	auth=self._auth(),
39	data=sorted(data.items()),
40	headers={"Connection": "close"},
41	)
42	if response.status_code != success_status:	4✔
43	raise ResponseException(response)	4✔
44	return response	4✔
45
46	def authorize_url(self, duration, scopes, state, implicit=False):	4✔
47	"""Return the URL used out-of-band to grant access to your application.
48
49	:param duration: Either ``permanent`` or ``temporary``. ``temporary``
50	authorizations generate access tokens that last only 1 hour. ``permanent``
51	authorizations additionally generate a refresh token that can be
52	indefinitely used to generate new hour-long access tokens. Only
53	``temporary`` can be specified if ``implicit`` is set to ``True``.
54	:param scopes: A list of OAuth scopes to request authorization for.
55	:param state: A string that will be reflected in the callback to
56	``redirect_uri``. Elements must be printable ASCII characters in the range
57	0x20 through 0x7E inclusive. This value should be temporarily unique to the
58	client for whom the URL was generated.
59	:param implicit: (optional) Use the implicit grant flow (default: False). This
60	flow is only available for UntrustedAuthenticators.
61
62	"""
63	if self.redirect_uri is None:	4✔
64	raise InvalidInvocation("redirect URI not provided")	4✔
65	if implicit and not isinstance(self, UntrustedAuthenticator):	4✔
66	raise InvalidInvocation(	4✔
67	"Only UntrustedAuthenticator instances can "
68	"use the implicit grant flow."
69	)
70	if implicit and duration != "temporary":	4✔
71	raise InvalidInvocation(	4✔
72	"The implicit grant flow only supports "
73	"temporary access tokens."
74	)
75
76	params = {	4✔
77	"client_id": self.client_id,
78	"duration": duration,
79	"redirect_uri": self.redirect_uri,
80	"response_type": "token" if implicit else "code",
81	"scope": " ".join(scopes),
82	"state": state,
83	}
84	url = self._requestor.reddit_url + const.AUTHORIZATION_PATH	4✔
85	request = Request("GET", url, params=params)	4✔
86	return request.prepare().url	4✔
87
88	def revoke_token(self, token, token_type=None):	4✔
89	"""Ask Reddit to revoke the provided token.
90
91	:param token: The access or refresh token to revoke.
92	:param token_type: (Optional) When provided, hint to Reddit what the token type
93	is for a possible efficiency gain. The value can be either ``access_token``
94	or ``refresh_token``.
95
96	"""
97	data = {"token": token}	4✔
98	if token_type is not None:	4✔
99	data["token_type_hint"] = token_type	4✔
100	url = self._requestor.reddit_url + const.REVOKE_TOKEN_PATH	4✔
101	self._post(url, **data)	4✔
102
103
104	class TrustedAuthenticator(BaseAuthenticator):	4✔
105	"""Store OAuth2 authentication credentials for web, or script type apps."""
106
107	RESPONSE_TYPE = "code"	4✔
108
109	def __init__(self, requestor, client_id, client_secret, redirect_uri=None):	4✔
110	"""Represent a single authentication to Reddit's API.
111
112	:param requestor: An instance of :class:`Requestor`.
113	:param client_id: The OAuth2 client ID to use with the session.
114	:param client_secret: The OAuth2 client secret to use with the session.
115	:param redirect_uri: (optional) The redirect URI exactly as specified in your
116	OAuth application settings on Reddit. This parameter is required if you want
117	to use the ``authorize_url`` method, or the ``authorize`` method of the
118	``Authorizer`` class.
119
120	"""
121	super(TrustedAuthenticator, self).__init__(	4✔
122	requestor, client_id, redirect_uri
123	)
124	self.client_secret = client_secret	4✔
125
126	def _auth(self):	4✔
127	return self.client_id, self.client_secret	4✔
128
129
130	class UntrustedAuthenticator(BaseAuthenticator):	4✔
131	"""Store OAuth2 authentication credentials for installed applications."""
132
133	def _auth(self):	4✔
134	return self.client_id, ""	4✔
135
136
137	class BaseAuthorizer(object):	4✔
138	"""Superclass for OAuth2 authorization tokens and scopes."""
139
140	def __init__(self, authenticator):	4✔
141	"""Represent a single authorization to Reddit's API.
142
143	:param authenticator: An instance of :class:`BaseAuthenticator`.
144
145	"""
146	self._authenticator = authenticator	4✔
147	self._clear_access_token()	4✔
148	self._validate_authenticator()	4✔
149
150	def _clear_access_token(self):	4✔
151	self._expiration_timestamp = None	4✔
152	self.access_token = None	4✔
153	self.scopes = None	4✔
154
155	def _request_token(self, **data):	4✔
156	url = (	4✔
157	self._authenticator._requestor.reddit_url + const.ACCESS_TOKEN_PATH
158	)
159	pre_request_time = time.time()	4✔
160	response = self._authenticator._post(url, **data)	4✔
161	payload = response.json()	4✔
162	if "error" in payload: # Why are these OKAY responses?	4✔
163	raise OAuthException(	4✔
164	response, payload["error"], payload.get("error_description")
165	)
166
167	self._expiration_timestamp = (	4✔
168	pre_request_time - 10 + payload["expires_in"]
169	)
170	self.access_token = payload["access_token"]	4✔
171	if "refresh_token" in payload:	4✔
172	self.refresh_token = payload["refresh_token"]	4✔
173	self.scopes = set(payload["scope"].split(" "))	4✔
174
175	def _validate_authenticator(self):	4✔
176	if not isinstance(self._authenticator, self.AUTHENTICATOR_CLASS):	4✔
177	raise InvalidInvocation(	4✔
178	"Must use a authenticator of type"
179	f" {self.AUTHENTICATOR_CLASS.__name__}."
180	)
181
182	def is_valid(self):	4✔
183	"""Return whether or not the Authorizer is ready to authorize requests.
184
185	A ``True`` return value does not guarantee that the access_token is actually
186	valid on the server side.
187
188	"""
189	return (	4✔
190	self.access_token is not None
191	and time.time() < self._expiration_timestamp
192	)
193
194	def revoke(self):	4✔
195	"""Revoke the current Authorization."""
196	if self.access_token is None:	4✔
197	raise InvalidInvocation("no token available to revoke")	4✔
198
199	self._authenticator.revoke_token(self.access_token, "access_token")	4✔
200	self._clear_access_token()	4✔
201
202
203	class Authorizer(BaseAuthorizer):	4✔
204	"""Manages OAuth2 authorization tokens and scopes."""
205
206	AUTHENTICATOR_CLASS = BaseAuthenticator	4✔
207
208	def __init__(	4✔
209	self,
210	authenticator,
211	*,
212	post_refresh_callback=None,
213	pre_refresh_callback=None,
214	refresh_token=None,
215	):
216	"""Represent a single authorization to Reddit's API.
217
218	:param authenticator: An instance of a subclass of :class:`BaseAuthenticator`.
219	:param post_refresh_callback: (Optional) When a single-argument
220	function is passed, the function will be called prior to refreshing
221	the access and refresh tokens. The argument to the callback is the
222	:class:`Authorizer` instance. This callback can be used to inspect
223	and modify the attributes of the :class:`Authorizer`.
224	:param pre_refresh_callback: (Optional) When a single-argument function
225	is passed, the function will be called after refreshing the access
226	and refresh tokens. The argument to the callback is the
227	:class:`Authorizer` instance. This callback can be used to inspect
228	and modify the attributes of the :class:`Authorizer`.
229	:param refresh_token: (Optional) Enables the ability to refresh the
230	authorization.
231
232	"""
233	super(Authorizer, self).__init__(authenticator)	4✔
234	self._post_refresh_callback = post_refresh_callback	4✔
235	self._pre_refresh_callback = pre_refresh_callback	4✔
236	self.refresh_token = refresh_token	4✔
237
238	def authorize(self, code):	4✔
239	"""Obtain and set authorization tokens based on ``code``.
240
241	:param code: The code obtained by an out-of-band authorization request to
242	Reddit.
243
244	"""
245	if self._authenticator.redirect_uri is None:	4✔
246	raise InvalidInvocation("redirect URI not provided")	4✔
247	self._request_token(	4✔
248	code=code,
249	grant_type="authorization_code",
250	redirect_uri=self._authenticator.redirect_uri,
251	)
252
253	def refresh(self):	4✔
254	"""Obtain a new access token from the refresh_token."""
255	if self._pre_refresh_callback:	4✔
256	self._pre_refresh_callback(self)	4✔
257	if self.refresh_token is None:	4✔
258	raise InvalidInvocation("refresh token not provided")	4✔
259	self._request_token(	4✔
260	grant_type="refresh_token", refresh_token=self.refresh_token
261	)
262	if self._post_refresh_callback:	4✔
263	self._post_refresh_callback(self)	4✔
264
265	def revoke(self, only_access=False):	4✔
266	"""Revoke the current Authorization.
267
268	:param only_access: (Optional) When explicitly set to True, do not evict the
269	refresh token if one is set.
270
271	Revoking a refresh token will in-turn revoke all access tokens associated with
272	that authorization.
273
274	"""
275	if only_access or self.refresh_token is None:	4✔
276	super(Authorizer, self).revoke()	4✔
277	else:
278	self._authenticator.revoke_token(	4✔
279	self.refresh_token, "refresh_token"
280	)
281	self._clear_access_token()	4✔
282	self.refresh_token = None	4✔
283
284
285	class _OAuth2ClientUserAuthApp(object):	4✔
286	def __init__(self, authenticator, scopes, implicit=False, duration="permanent") -> None:	4✔
NEW 287	super().__init__()	×
NEW 288	self.__finished = False	×
NEW 289	self.__state = str(uuid4())	×
NEW 290	self.__auth_url = authenticator.authorize_url(duration, scopes, self.__state, implicit)	×
NEW 291	self.__redirect_uri = authenticator.redirect_uri	×
NEW 292	self.__callback_uri = urlparse(self.__redirect_uri).path	×
NEW 293	self.__auth_exchange_data = {}	×
294
295	def __call__(self, environ, start_resp):	4✔
NEW 296	if not self.__finished:	×
NEW 297	req_method = environ["REQUEST_METHOD"]	×
NEW 298	req_uri = environ["PATH_INFO"]	×
NEW 299	req_query = parse_qs(environ["QUERY_STRING"])	×
300
NEW 301	if req_method == "GET":	×
NEW 302	if req_uri == "/":	×
NEW 303	start_resp("200 OK", [("Content-type", "text/html; charset=utf-8")])	×
NEW 304	return ['<a href="/auth">Authenticate with reddit</a>'.encode("utf8")]	×
305
NEW 306	elif req_uri == "/auth":	×
NEW 307	start_resp("302 Moved Temporarily", [("Location", self.__auth_url)])	×
NEW 308	return []	×
309
NEW 310	elif req_uri == self.__callback_uri:	×
NEW 311	if "error" not in req_query:	×
NEW 312	req_state = req_query["state"][0]	×
313
NEW 314	if req_state == self.__state:	×
NEW 315	if "code" in req_query:	×
NEW 316	self.__auth_exchange_data.update({"code": req_query["code"][0]})	×
317
318	else:
NEW 319	self.__auth_exchange_data.update({	×
320	"access_token": req_query["access_token"][0],
321	"expires_in": req_query["expires_in"][0],
322	"scope": req_query["scope"][0]
323	})
324
NEW 325	start_resp("200 OK", [])	×
NEW 326	received = "Authorization Code" if "code" in self.__auth_exchange_data else "User Access Token"	×
NEW 327	self.__finished = True	×
NEW 328	return [f"Received {received} Successfully! You may safely close this page.".encode("utf8")]	×
329
330	else:
NEW 331	start_resp("200 OK", [])	×
NEW 332	return [f"State mismatch!\nExpected State: {self.__state}\nReceived State: {req_state}".encode("utf8")]	×
333
334	else:
NEW 335	start_resp("200 OK", [])	×
NEW 336	err_val = req_query["error"]	×
337
NEW 338	if err_val == "access_denied":	×
NEW 339	return ["User denied permission!".encode("utf8")]	×
340
NEW 341	elif err_val == "unsupported_response_type":	×
NEW 342	return ["Invalid initial authorization response_type!".encode("utf8")]	×
343
NEW 344	elif err_val == "invalid_scope":	×
NEW 345	return ["Invalid authorization scope(s) requested!".encode("utf8")]	×
346
NEW 347	elif err_val == "invalid_request":	×
NEW 348	return ["Invalid authorization request!".encode("utf8")]	×
349
350	else:
NEW 351	return ["Unknown Error!" + f"\nERROR: {err_val}".encode("utf8")]	×
352
353	else:
NEW 354	start_resp("404 Not Found", [])	×
NEW 355	return ["Unknown URI!".encode("utf8")]	×
356
357	else:
NEW 358	start_resp("405 Method Not Allowed", [])	×
NEW 359	return ["Specified HTTP Request Method is not accepted by server!".encode("utf8")]	×
360
361	@property	4✔
362	def finished(self) -> bool:	4✔
NEW 363	return self.__finished	×
364
365	@property	4✔
366	def auth_code(self) -> Optional[str]:	4✔
NEW 367	return self.__auth_exchange_data["code"] if "code" in self.__auth_exchange_data else None	×
368
369	@property	4✔
370	def implicit_grant(self) -> Optional[Dict[str, Any]]:	4✔
NEW 371	return self.__auth_exchange_data if "access_token" in self.__auth_exchange_data else None	×
372
373
374	class LocalWSGIServerAuthorizer(Authorizer):	4✔
375	AUTHENTICATOR_CLASS = TrustedAuthenticator	4✔
376
377	def __init__(	4✔
378	self,
379	authenticator,
380	scopes,
381	duration="permanent",
382	):
NEW 383	super(LocalWSGIServerAuthorizer, self).__init__(authenticator)	×
NEW 384	self.scopes = scopes	×
NEW 385	self.duration = duration	×
386
387	def authorize_local_server(self):	4✔
NEW 388	auth_app = _OAuth2ClientUserAuthApp(	×
389	self._authenticator,
390	self.scopes,
391	implicit=False,
392	duration=self.duration,
393	)
394
NEW 395	redirect_uri = urlparse(self._authenticator.redirect_uri)	×
NEW 396	auth_server = make_server(	×
397	redirect_uri.netloc.split(":")[0],
398	redirect_uri.netloc.split(":")[1] if ":" in redirect_uri.netloc else 80,
399	auth_app
400	)
401
NEW 402	while not auth_app.finished:	×
NEW 403	auth_server.handle_request()	×
404
NEW 405	self.authorize(auth_app.auth_code)	×
406
407
408	class DeviceIDAuthorizer(BaseAuthorizer):	4✔
409	"""Manages app-only OAuth2 for 'installed' applications.
410
411	While the '*' scope will be available, some endpoints simply will not work due to
412	the lack of an associated Reddit account.
413
414	"""
415
416	AUTHENTICATOR_CLASS = UntrustedAuthenticator	4✔
417
418	def __init__(	4✔
419	self, authenticator, device_id="DO_NOT_TRACK_THIS_DEVICE", scopes=None
420	):
421	"""Represent an app-only OAuth2 authorization for 'installed' apps.
422
423	:param authenticator: An instance of :class:`UntrustedAuthenticator`.
424	:param device_id: (optional) A unique ID (20-30 character ASCII string) (default
425	DO_NOT_TRACK_THIS_DEVICE). For more information about this parameter, see:
426	https://github.com/reddit/reddit/wiki/OAuth2#application-only-oauth
427	:param scopes: (Optional) A list of OAuth scopes to request authorization for
428	(default: None). The scope ``*`` is requested when the default argument is
429	used.
430
431	"""
432	super(DeviceIDAuthorizer, self).__init__(authenticator)	4✔
433	self._device_id = device_id	4✔
434	self._scopes = scopes	4✔
435
436	def refresh(self):	4✔
437	"""Obtain a new access token."""
438	additional_kwargs = {}	4✔
439	if self._scopes:	4✔
440	additional_kwargs["scope"] = " ".join(self._scopes)	4✔
441	grant_type = "https://oauth.reddit.com/grants/installed_client"	4✔
442	self._request_token(	4✔
443	grant_type=grant_type,
444	device_id=self._device_id,
445	**additional_kwargs,
446	)
447
448
449	class ImplicitAuthorizer(BaseAuthorizer):	4✔
450	"""Manages implicit installed-app type authorizations."""
451
452	AUTHENTICATOR_CLASS = UntrustedAuthenticator	4✔
453
454	def __init__(self, authenticator, access_token, expires_in, scope):	4✔
455	"""Represent a single implicit authorization to Reddit's API.
456
457	:param authenticator: An instance of :class:`UntrustedAuthenticator`.
458	:param access_token: The access_token obtained from Reddit via callback to the
459	authenticator's redirect_uri.
460	:param expires_in: The number of seconds the ``access_token`` is valid for. The
461	origin of this value was returned from Reddit via callback to the
462	authenticator's redirect uri. Note, you may need to subtract an offset
463	before passing in this number to account for a delay between when Reddit
464	prepared the response, and when you make this function call.
465	:param scope: A space-delimited string of Reddit OAuth2 scope names as returned
466	from Reddit in the callback to the authenticator's redirect uri.
467
468	"""
469	super(ImplicitAuthorizer, self).__init__(authenticator)	4✔
470	self._expiration_timestamp = time.time() + expires_in	4✔
471	self.access_token = access_token	4✔
472	self.scopes = set(scope.split(" "))	4✔
473
474
475	class ReadOnlyAuthorizer(Authorizer):	4✔
476	"""Manages authorizations that are not associated with a Reddit account.
477
478	While the '*' scope will be available, some endpoints simply will not work due to
479	the lack of an associated Reddit account.
480
481	"""
482
483	AUTHENTICATOR_CLASS = TrustedAuthenticator	4✔
484
485	def __init__(self, authenticator, scopes=None):	4✔
486	"""Represent a ReadOnly authorization to Reddit's API.
487
488	:param scopes: (Optional) A list of OAuth scopes to request authorization for
489	(default: None). The scope ``*`` is requested when the default argument is
490	used.
491
492	"""
493	super().__init__(authenticator)	4✔
494	self._scopes = scopes	4✔
495
496	def refresh(self):	4✔
497	"""Obtain a new ReadOnly access token."""
498	additional_kwargs = {}	4✔
499	if self._scopes:	4✔
500	additional_kwargs["scope"] = " ".join(self._scopes)	4✔
501	self._request_token(	4✔
502	grant_type="client_credentials", **additional_kwargs
503	)
504
505
506	class ScriptAuthorizer(Authorizer):	4✔
507	"""Manages personal-use script type authorizations.
508
509	Only users who are listed as developers for the application will be granted access
510	tokens.
511
512	"""
513
514	AUTHENTICATOR_CLASS = TrustedAuthenticator	4✔
515
516	def __init__(	4✔
517	self,
518	authenticator,
519	username,
520	password,
521	two_factor_callback=None,
522	scopes=None,
523	):
524	"""Represent a single personal-use authorization to Reddit's API.
525
526	:param authenticator: An instance of :class:`TrustedAuthenticator`.
527	:param username: The Reddit username of one of the application's developers.
528	:param password: The password associated with ``username``.
529	:param two_factor_callback: A function that returns OTPs (One-Time
530	Passcodes), also known as 2FA auth codes. If this function is
531	provided, prawcore will call it when authenticating.
532	:param scopes: (Optional) A list of OAuth scopes to request authorization for
533	(default: None). The scope ``*`` is requested when the default argument is
534	used.
535
536	"""
537	super(ScriptAuthorizer, self).__init__(authenticator)	4✔
538	self._password = password	4✔
539	self._scopes = scopes	4✔
540	self._two_factor_callback = two_factor_callback	4✔
541	self._username = username	4✔
542
543	def refresh(self):	4✔
544	"""Obtain a new personal-use script type access token."""
545	additional_kwargs = {}	4✔
546	if self._scopes:	4✔
547	additional_kwargs["scope"] = " ".join(self._scopes)	4✔
548	two_factor_code = (	4✔
549	self._two_factor_callback and self._two_factor_callback()
550	)
551	if two_factor_code:	4✔
552	additional_kwargs["otp"] = two_factor_code	4✔
553	self._request_token(	4✔
554	grant_type="password",
555	username=self._username,
556	password=self._password,
557	**additional_kwargs,
558	)

praw-dev / prawcore / 87d7ad5358d3bd4d01d56cfc4738ebfa40b373b5-PR-128

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