build: replace unmaintained oauth2 with oauthlib

evernote_backup/evernote_client_oauth.py

@@ -1,10 +1,9 @@
 import threading
 import time
 from http.server import BaseHTTPRequestHandler, HTTPServer
-from typing import Optional
-from urllib.parse import parse_qsl, quote, urlparse
 
-import oauth2
+from requests_oauthlib import OAuth1Session
+from requests_oauthlib.oauth1_session import TokenMissing, TokenRequestDenied
 
 from evernote_backup.cli_app_util import is_inside_docker
 from evernote_backup.evernote_client import EvernoteClientBase
@@ -21,14 +20,12 @@ class CallbackHandler(BaseHTTPRequestHandler):
     }
 
     def do_GET(self) -> None:
-        response = urlparse(self.path)
-
-        if response.path != "/oauth_callback":
+        if not self.path.startswith("/oauth_callback?"):
             self.send_response(self.http_codes["NOT FOUND"])
             self.end_headers()
             return
 
-        self.server.callback_response = dict(parse_qsl(response.query))  # type: ignore
+        self.server.callback_response = self.path
 
         self.send_response(self.http_codes["OK"])
         self.end_headers()
@@ -45,7 +42,7 @@ class StoppableHTTPServer(HTTPServer):
     def __init__(self, *args, **kwargs) -> None:  # type: ignore
         super().__init__(*args, **kwargs)
 
-        self.callback_response: dict = {}
+        self.callback_response: str = ""
 
     def run(self) -> None:
         try:  # noqa: WPS501
@@ -62,28 +59,16 @@ def __init__(
 
         self.server_host = server_host
         self.server_port = oauth_port
-        self.oauth_token: dict = {}
 
     def get_oauth_url(self) -> str:
-        self.oauth_token = self.client.get_request_token(
+        return self.client.get_authorize_url(
             f"http://{self.server_host}:{self.server_port}/oauth_callback"
         )
 
-        return self.client.get_authorize_url(self.oauth_token)
-
     def wait_for_token(self) -> str:
-        callback = self._wait_for_callback()
-
-        if "oauth_verifier" not in callback:
-            raise OAuthDeclinedError
+        return self.client.get_access_token(self._wait_for_callback())
 
-        return self.client.get_access_token(
-            oauth_token=callback["oauth_token"],
-            oauth_verifier=callback["oauth_verifier"],
-            oauth_token_secret=self.oauth_token["oauth_token_secret"],
-        )
-
-    def _wait_for_callback(self) -> dict:
+    def _wait_for_callback(self) -> str:
         if is_inside_docker():
             server_param = ("0.0.0.0", self.server_port)  # noqa: S104
         else:
@@ -113,43 +98,31 @@ def __init__(
     ) -> None:
         super().__init__(backend=backend)
 
-        self.consumer_key = consumer_key
-        self.consumer_secret = consumer_secret
-
-    def get_authorize_url(self, request_token: dict) -> str:
-        return "{0}?oauth_token={1}".format(
-            self._get_endpoint("OAuth.action"),
-            quote(request_token["oauth_token"]),
-        )
+        self.client_key = consumer_key
+        self.client_secret = consumer_secret
 
-    def get_request_token(self, callback_url: str) -> dict:
-        client = self._get_oauth_client()
+        self._session = None
 
-        request_url = "{0}?oauth_callback={1}".format(
-            self._get_endpoint("oauth"), quote(callback_url)
+    def get_authorize_url(self, callback_url: str) -> str:
+        self._session = OAuth1Session(
+            client_key=self.client_key,
+            client_secret=self.client_secret,
+            callback_uri=callback_url,
         )
 
-        _, response_content = client.request(request_url, "GET")
-
-        return dict(parse_qsl(response_content.decode("utf-8")))
-
-    def get_access_token(
-        self,
-        oauth_token: str,
-        oauth_token_secret: str,
-        oauth_verifier: str,
-    ) -> str:
-        token = oauth2.Token(oauth_token, oauth_token_secret)
-        token.set_verifier(oauth_verifier)
+        self._session.fetch_request_token(self._get_endpoint("oauth"))
 
-        client = self._get_oauth_client(token)
+        return self._session.authorization_url(self._get_endpoint("OAuth.action"))
 
-        _, response_content = client.request(self._get_endpoint("oauth"), "POST")
-        access_token_dict = dict(parse_qsl(response_content.decode("utf-8")))
-
-        return access_token_dict["oauth_token"]
+    def get_access_token(self, callback_response_raw: str) -> str:
+        try:
+            self._session.parse_authorization_response(callback_response_raw)
+        except TokenMissing:
+            raise OAuthDeclinedError
 
-    def _get_oauth_client(self, token: Optional[oauth2.Token] = None) -> oauth2.Client:
-        consumer = oauth2.Consumer(self.consumer_key, self.consumer_secret)
+        try:
+            access_token = self._session.fetch_access_token(self._get_endpoint("oauth"))
+        except TokenRequestDenied:
+            raise OAuthDeclinedError
 
-        return oauth2.Client(consumer, token) if token else oauth2.Client(consumer)
+        return access_token["oauth_token"]

pyproject.toml

@@ -37,8 +37,8 @@ python = "^3.8"
 evernote3 = "^1.25.14"
 xmltodict = "^0.13.0"
 click = "^8.1.7"
-oauth2 = "^1.9.0"
 click-option-group = "^0.5.6"
+requests-oauthlib = "^1.3.1"
 
 [tool.poetry.group.test]
 optional = true

tests/conftest.py

@@ -11,7 +11,9 @@
     EDAMSystemException,
     EDAMUserException,
 )
+from requests_oauthlib.oauth1_session import TokenRequestDenied
 
+import evernote_backup
 from evernote_backup import cli_app, note_storage
 from evernote_backup.cli import cli
 from evernote_backup.token_util import get_token_shard
@@ -293,35 +295,41 @@ def fake_token():
 
 @pytest.fixture
 def mock_oauth_client(mocker):
-    oauth_mock = mocker.patch("evernote_backup.evernote_client_oauth.oauth2")
+    def fake_request(self, url, **request_kwargs):
+        if self._client.client.resource_owner_key is None:
+            token = {
+                "oauth_token": oauth_mock.fake_oauth_token_id,
+                "oauth_token_secret": oauth_mock.fake_oauth_secret,
+                "oauth_callback_confirmed": "true",
+            }
+        else:
+            if oauth_mock.fake_bad_response:
+                raise TokenRequestDenied(None, None)
+
+            token = {
+                "oauth_token": oauth_mock.fake_token,
+                "oauth_verifier": "FFF2",
+                "sandbox_lnb": "false",
+            }
+
+        self._populate_attributes(token)
+        self.token = token
+        return token
+
+    oauth_mock = mocker.patch.object(
+        evernote_backup.evernote_client_oauth.OAuth1Session,
+        "_fetch_token",
+        fake_request,
+    )
 
     oauth_mock.fake_oauth_token_id = "fake_app.FFF"
     oauth_mock.fake_oauth_secret = "FFF1"
-    oauth_mock.fake_request_url = (
-        f"oauth_token={oauth_mock.fake_oauth_token_id}&"
-        f"oauth_token_secret={oauth_mock.fake_oauth_secret}&"
-        f"oauth_callback_confirmed=true"
-    ).encode()
-
-    oauth_mock.fake_callback_response = {
-        "oauth_token": oauth_mock.fake_oauth_token_id,
-        "oauth_verifier": "FFF2",
-        "sandbox_lnb": "false",
-    }
-
-    oauth_mock.fake_token = "S=s100:U=fff:E=ffff:C=ffff:P=100:A=appname:V=2:H=ffffff"
 
-    def fake_request(url, method):
-        if method == "POST":
-            response = urllib.parse.urlencode(
-                {"oauth_token": oauth_mock.fake_token}
-            ).encode()
-        else:
-            response = oauth_mock.fake_request_url
+    oauth_mock.fake_callback_response = f"/?oauth_token={oauth_mock.fake_oauth_token_id}&oauth_verifier=FFF2&sandbox_lnb=false"
 
-        return None, response
+    oauth_mock.fake_token = "S=s100:U=fff:E=ffff:C=ffff:P=100:A=appname:V=2:H=ffffff"
 
-    oauth_mock.Client().request.side_effect = fake_request
+    oauth_mock.fake_bad_response = False
 
     return oauth_mock
 

tests/test_evernote_client_oauth.py

@@ -78,7 +78,22 @@ def test_get_auth_token_url(mock_oauth_client, mock_evernote_oauth_client):
 
 @pytest.mark.usefixtures("mock_oauth_http_server")
 def test_get_auth_token_declined(mock_oauth_client, mock_evernote_oauth_client):
-    del mock_oauth_client.fake_callback_response["oauth_verifier"]
+    mock_oauth_client.fake_callback_response = "/"
+
+    oauth_handler = EvernoteOAuthCallbackHandler(
+        mock_evernote_oauth_client, FAKE_OAUTH_PORT, FAKE_OAUTH_HOST
+    )
+    oauth_handler.get_oauth_url()
+
+    with pytest.raises(OAuthDeclinedError):
+        oauth_handler.wait_for_token()
+
+
+@pytest.mark.usefixtures("mock_oauth_http_server")
+def test_get_auth_token_declined_bad_response(
+    mock_oauth_client, mock_evernote_oauth_client
+):
+    mock_oauth_client.fake_bad_response = True
 
     oauth_handler = EvernoteOAuthCallbackHandler(
         mock_evernote_oauth_client, FAKE_OAUTH_PORT, FAKE_OAUTH_HOST
@@ -131,7 +146,7 @@ def test_callback_handler(mocker):
 
     CallbackHandler.do_GET(mock_instance)
 
-    assert mock_instance.server.callback_response == {"test_param": "test"}
+    assert mock_instance.server.callback_response == mock_instance.path
     mock_instance.send_response.assert_called_once_with(
         CallbackHandler.http_codes["OK"]
     )

tests/test_op_reauth.py

@@ -285,7 +285,7 @@ def test_oauth_login_custom_port(
 def test_oauth_login_declined_error(
     cli_invoker, fake_storage, mock_evernote_client, mock_oauth_client, mocker
 ):
-    del mock_oauth_client.fake_callback_response["oauth_verifier"]
+    mock_oauth_client.fake_callback_response = "/"
 
     mocker.patch("evernote_backup.cli_app_util.click.echo")
     mock_launch = mocker.patch("evernote_backup.cli_app_util.click.launch")