Revert "Fix commits"

app/api/mal.py

@@ -1,5 +1,6 @@
 import os
 import secrets
+from typing import Tuple
 import requests
 
 from urllib.parse import urlencode
@@ -10,54 +11,49 @@
 N_BYTES = 96
 QUERY_LIMIT = 30
 TIMEOUT = 30
+CODE_CHALLENGE_METHOD = 'plain'
 
+REDIRECT_URI = f'{Config.PROTOCOL}://{Config.REDIRECT_URL}/callback'
+CLIENT_ID = os.environ.get('MAL_ID')
+CLIENT_SECRET = os.environ.get('MAL_SECRET')
 
 class MyAnimeListAPI:
     """
     MyAnimeList API wrapper
     """
-
-    def __init__(self):
-        """
-        Initialize the MyAnimeList API wrapper
-        """
-
-        self.redirect_uri = f'{Config.PROTOCOL}://{Config.REDIRECT_URL}/callback'
-        self.client_id = os.environ.get('MAL_ID')
-        self.client_secret = os.environ.get('MAL_SECRET')
-
-        self.code_challenge_method = 'plain'
-        self.code_verifier, self.code_challenge = (
-            MyAnimeListAPI.__generate_verifier_challenger_pair(128, method=self.code_challenge_method))
-
-    def get_auth(self):
+    @staticmethod
+    def get_auth() -> Tuple[str, str]:
         """
         Get the authorization URL for MyAnimeList API
         :return: Authorization URL
         """
         state = secrets.token_urlsafe(N_BYTES)[:16]
+        code_verifier, code_challenge = (
+            MyAnimeListAPI.__generate_verifier_challenger_pair(128, method=CODE_CHALLENGE_METHOD))
+
         query_params = (f'response_type=code'
-                        f'&client_id={self.client_id}'
+                        f'&client_id={CLIENT_ID}'
                         f'&state={state}'
-                        f'&code_challenge={self.code_challenge}'
-                        f'&code_challenge_method={self.code_challenge_method}'
-                        f'&redirect_uri={self.redirect_uri}')
+                        f'&code_challenge={code_challenge}'
+                        f'&code_challenge_method={CODE_CHALLENGE_METHOD}'
+                        f'&redirect_uri={REDIRECT_URI}')
 
-        return f'{AUTH_URL}/oauth2/authorize?{query_params}'
+        return f'{AUTH_URL}/oauth2/authorize?{query_params}', code_verifier
 
-    def get_access_token(self, authorization_code: str):
+    @staticmethod
+    def get_access_token(authorization_code: str, code_verifier: str):
         """
         Get the access token for MyAnimeList
         :param authorization_code: Authorization Code from MyAnimeList
         :return: Access Token
         """
         url = f'{AUTH_URL}/oauth2/token'
-        data = {'client_id': self.client_id,
-                'client_secret': self.client_secret,
+        data = {'client_id': CLIENT_ID,
+                'client_secret': CLIENT_SECRET,
                 'grant_type': 'authorization_code',
                 'code': authorization_code,
-                'code_verifier': self.code_verifier,
-                'redirect_uri': self.redirect_uri}
+                'code_verifier': code_verifier,
+                'redirect_uri': REDIRECT_URI}
 
         resp = requests.post(url=url, data=data, timeout=TIMEOUT)
         resp.raise_for_status()
@@ -69,15 +65,16 @@ def get_access_token(self, authorization_code: str):
                 'access_token': resp_json['access_token'],
                 'refresh_token': resp_json['refresh_token']}
 
-    def refresh_token(self, refresh_token: str):
+    @staticmethod
+    def refresh_token(refresh_token: str):
         """
         Refresh the access token for MyAnimeList
         :param refresh_token: Refresh Token
         :return: Access Token
         """
         url = f'{AUTH_URL}/oauth2/token'
-        data = {'client_id': self.client_id,
-                'client_secret': self.client_secret,
+        data = {'client_id': CLIENT_ID,
+                'client_secret': CLIENT_SECRET,
                 'grant_type': 'refresh_token',
                 'refresh_token': refresh_token}
 
@@ -240,4 +237,3 @@ def __generate_challenge(verifier, method):
         """
         if method == 'plain' or method is None:
             return verifier
-

app/db/db.py

@@ -1,6 +1,6 @@
 import re
 from functools import lru_cache
-from typing import Dict
+from typing import Dict, Tuple
 
 from pymongo import MongoClient
 
@@ -37,7 +37,7 @@ def store_user(user_details: Dict):
 
 
 @lru_cache(maxsize=10000)
-def get_kitsu_id_from_mal_id(mal_id) -> (bool, str):
+def get_kitsu_id_from_mal_id(mal_id) -> Tuple[bool, str]:
     """
     Get kitsu_id from mal_id from db
     :param mal_id: The MyAnimeList id of the anime
@@ -48,13 +48,15 @@ def get_kitsu_id_from_mal_id(mal_id) -> (bool, str):
         mal_id = int(mal_id)
         if res := anime_mapping.find_one({'mal_id': mal_id}):
             return True, res['kitsu_id']
+    except KeyError:
+        log_error(f"No Kitsu ID for: MAL:{mal_id}")
     except ValueError:
         log_error(f"Invalid MyAnimeList ID: {mal_id}")
     return False, None
 
 
 @lru_cache(maxsize=10000)
-def get_mal_id_from_kitsu_id(kitsu_id) -> (bool, str):
+def get_mal_id_from_kitsu_id(kitsu_id) -> Tuple[bool, str]:
     """
     Get mal_id from kitsu_id from db
     :param kitsu_id: The kitsu id of the anime
@@ -66,6 +68,8 @@ def get_mal_id_from_kitsu_id(kitsu_id) -> (bool, str):
         res = anime_mapping.find_one({'kitsu_id': kitsu_id})
         if res:
             return True, res['mal_id']
+    except KeyError:
+        log_error(f"No MAL ID for KITSU:{kitsu_id}")
     except ValueError:
         log_error(f"Invalid kitsu ID: {kitsu_id}")
     return False, None

app/routes/auth.py

@@ -38,7 +38,10 @@ def authorize_user():
     if 'user' in session:
         flash("You are already logged in.", "warning")
         return redirect(url_for('index'))
-    return redirect(mal_client.get_auth())
+
+    auth_url, code_verifier = mal_client.get_auth()
+    session['code_verifier'] = code_verifier
+    return redirect(auth_url)
 
 
 @auth_blueprint.route('/callback')
@@ -60,7 +63,13 @@ def callback():
         # exchange auth code for access token
         if not (auth_code := request.args.get('code', None)):
             return redirect(url_for('index'))
-        resp = mal_client.get_access_token(auth_code)
+
+        if 'code_verifier' not in session:
+            flash("Invalid callback request. First log in.", "warning")
+            return redirect(url_for('index'))
+
+        code_verifier = session['code_verifier']
+        resp = mal_client.get_access_token(auth_code, code_verifier)
 
         # get user details and append the access and refresh token the info
         user_details = mal_client.get_user_details(token=resp['access_token'])

app/routes/manifest.py

@@ -15,7 +15,7 @@
 
 MANIFEST = {
     'id': 'com.sagetendo.mal-stremio-addon',
-    'version': '3.0.0',
+    'version': '3.0.1',
     'name': 'MAL-Stremio Addon',
     'logo': 'https://i.imgur.com/zVYdffr.png',
     'description': 'Provides users with watchlist content from MyAnimeList within Stremio. '

tests/api/test_mal.py

@@ -1,7 +1,7 @@
 import unittest
 from unittest.mock import patch, MagicMock
 
-from app.api.mal import MyAnimeListAPI, TIMEOUT, QUERY_LIMIT
+from app.api.mal import CLIENT_ID, CLIENT_SECRET, REDIRECT_URI, MyAnimeListAPI, TIMEOUT, QUERY_LIMIT
 
 
 class TestMyAnimeListAPI(unittest.TestCase):
@@ -22,16 +22,17 @@ def test_init(self):
         """
         Test that the MyAnimeListAPI class is initialized correctly
         """
-        self.assertIsNotNone(self.mal_api.redirect_uri)
-        self.assertIsNotNone(self.mal_api.client_id)
-        self.assertIsNotNone(self.mal_api.client_secret)
+        self.assertIsNotNone(REDIRECT_URI)
+        self.assertIsNotNone(CLIENT_ID)
+        self.assertIsNotNone(CLIENT_SECRET)
 
     def test_get_auth(self):
         """
         Test that the get_auth function returns a valid auth URL
         """
-        auth_url = self.mal_api.get_auth()
+        auth_url, code_verifier = self.mal_api.get_auth()
         self.assertIsNotNone(auth_url)
+        self.assertIsNotNone(code_verifier)
 
     @patch('requests.post')
     def test_get_access_token(self, mock_post):
@@ -50,18 +51,19 @@ def test_get_access_token(self, mock_post):
         mock_post.return_value = mock_response
 
         authorization_code = "auth_code_from_redirect"
-        token_details = self.mal_api.get_access_token(authorization_code)
+        code_verifier = "code_verifier"
+        token_details = self.mal_api.get_access_token(authorization_code, code_verifier=code_verifier)
 
         # Assert that the access token details are correct
         self.assertEqual(token_details['access_token'], 'new_access_token')
         self.assertEqual(token_details['refresh_token'], 'refresh_token_value')
         mock_post.assert_called_once_with(url='https://myanimelist.net/v1/oauth2/token',
-                                          data={'client_id': self.mal_api.client_id,
-                                                'client_secret': self.mal_api.client_secret,
+                                          data={'client_id': CLIENT_ID,
+                                                'client_secret': CLIENT_SECRET,
                                                 'grant_type': 'authorization_code',
                                                 'code': authorization_code,
-                                                'code_verifier': self.mal_api.code_verifier,
-                                                'redirect_uri': self.mal_api.redirect_uri},
+                                                'code_verifier': code_verifier,
+                                                'redirect_uri': REDIRECT_URI},
                                           timeout=TIMEOUT)
 
     @patch('requests.post')
@@ -85,8 +87,8 @@ def test_refresh_token(self, mock_post):
         self.assertEqual(token_details['access_token'], 'new_access_token')
         self.assertEqual(token_details['refresh_token'], 'refresh_token_value')
         mock_post.assert_called_once_with(url='https://myanimelist.net/v1/oauth2/token',
-                                          data={'client_id': self.mal_api.client_id,
-                                                'client_secret': self.mal_api.client_secret,
+                                          data={'client_id': CLIENT_ID,
+                                                'client_secret': CLIENT_SECRET,
                                                 'grant_type': 'refresh_token',
                                                 'refresh_token': refresh_token},
                                           timeout=TIMEOUT)

tests/routes/test_auth.py

@@ -14,8 +14,9 @@ def setUp(self):
         """
         Set up the test class
         """
-        self.client = app.test_client()
+        app.config['SECRET'] = "Testing Secret"
         app.config['TESTING'] = True
+        self.client = app.test_client()
 
     def test_get_token(self):
         """
@@ -70,8 +71,9 @@ def test_callback(self, mock_store_user, mock_get_user_details, mock_get_access_
         mock_store_user.return_value = None
 
         # Simulate the callback with a successful authorization code
-        with self.client:
-            response = self.client.get('/callback?code=mocked_auth_code')
+        with self.client.session_transaction() as sess:
+            sess['code_verifier'] = 'mocked_verifier'
+        response = self.client.get('/callback?code=mocked_auth_code')
 
         # Assert that the user is redirected to the home page with a success message
         self.assertEqual(response.status_code, 302)