Fixed auth to use AWS package and use all the newest packages

requirements.txt

@@ -1,10 +1,11 @@
-requests>=2.31.0,<3.0.0
-six>=1.0.0,<2.0.0
-wheel>=0.38.0
-boto>=2.48.0
-python-dateutil>=2.6.1,<3.0.0
-diskcache>=2.9.0,<3.0.0
-setuptools>=65.5.1
-attrs>=18.1.0,<19.0.0
-trans>=2.1.0,<3.0.0
-certifi>=2023.7.22 # not directly required, pinned by Snyk to avoid a vulnerability
+requests
+#six
+wheel
+boto
+python-dateutil
+diskcache
+setuptools
+attrs
+trans
+#certifi
+botocore

src/imdbpie/auth.py

@@ -2,84 +2,34 @@
 from __future__ import absolute_import, unicode_literals
 
 import json
-import requests
 import tempfile
 from datetime import datetime
-
-try:
-    from base64 import encodebytes
-except ImportError:
-    from base64 import encodestring as encodebytes
-
+from urllib.parse import urlparse, parse_qs
 import diskcache
-from dateutil.tz import tzutc
+import requests
+from botocore.auth import SigV4Auth
+from botocore.awsrequest import AWSRequest
+from botocore.credentials import Credentials
 from dateutil.parser import parse
-import boto.utils
-from six.moves.urllib.parse import urlparse, parse_qs, quote
-from boto import provider
-from boto.connection import HTTPRequest
-from boto.auth import HmacAuthV3HTTPHandler
+from dateutil.tz import tzutc
 
 from .constants import APP_KEY, HOST, USER_AGENT, BASE_URI
 
 
-class ZuluHmacAuthV3HTTPHandler(HmacAuthV3HTTPHandler):
-    def sign_string(self, string_to_sign):
-        new_hmac = self._get_hmac()
-        new_hmac.update(string_to_sign)
-        return encodebytes(new_hmac.digest()).decode('utf-8').strip()
-
-    def headers_to_sign(self, http_request):
-        headers_to_sign = {'Host': self.host}
-        for name, value in http_request.headers.items():
-            lname = name.lower()
-            if lname.startswith('x-amz'):
-                headers_to_sign[name] = value
-        return headers_to_sign
-
-    def canonical_query_string(self, http_request):
-        if http_request.method == 'POST':
-            return ''
-        qs_parts = []
-        for param in sorted(http_request.params):
-            value = boto.utils.get_utf8_value(http_request.params[param])
-            param_ = quote(param, safe='-_.~')
-            value_ = quote(value, safe='-_.~')
-            qs_parts.append('{0}={1}'.format(param_, value_))
-        return '&'.join(qs_parts)
-
-    def string_to_sign(self, http_request):
-        headers_to_sign = self.headers_to_sign(http_request)
-        canonical_qs = self.canonical_query_string(http_request)
-        canonical_headers = self.canonical_headers(headers_to_sign)
-        string_to_sign = '\n'.join(
-            (
-                http_request.method,
-                http_request.path,
-                canonical_qs,
-                canonical_headers,
-                '',
-                http_request.body,
-            )
-        )
-        return string_to_sign, headers_to_sign
-
-
 def _get_credentials():
     url = '{0}/authentication/credentials/temporary/ios82'.format(BASE_URI)
     response = requests.post(
-        url, json={'appKey': APP_KEY}, headers={'User-Agent': USER_AGENT}
+        url, json={"appKey": APP_KEY}, headers={'User-Agent': USER_AGENT}
     )
     response.raise_for_status()
     return json.loads(response.content.decode('utf8'))['resource']
 
 
 class Auth(object):
-
     SOON_EXPIRES_SECONDS = 60
     _CREDS_STORAGE_KEY = 'imdbpie-credentials'
 
-    def __init__(self, creds=None):
+    def __init__(self):
         self._cachedir = tempfile.gettempdir()
 
     def _get_creds(self):
@@ -114,33 +64,19 @@ def get_auth_headers(self, url_path):
         creds, soon_expires = self._creds_soon_expiring()
         if soon_expires:
             creds = self._set_creds(creds=_get_credentials())
+        credentials = Credentials(access_key=creds['accessKeyId'], secret_key=creds['secretAccessKey'],
+                        token=creds['sessionToken'])
+
 
-        handler = ZuluHmacAuthV3HTTPHandler(
-            host=HOST,
-            config={},
-            provider=provider.Provider(
-                name='aws',
-                access_key=creds['accessKeyId'],
-                secret_key=creds['secretAccessKey'],
-                security_token=creds['sessionToken'],
-            ),
-        )
         parsed_url = urlparse(url_path)
         params = {
             key: val[0] for key, val in parse_qs(parsed_url.query).items()
         }
-        request = HTTPRequest(
-            method='GET',
-            protocol='https',
-            host=HOST,
-            port=443,
-            path=parsed_url.path,
-            auth_path=None,
-            params=params,
-            headers={},
-            body='',
-        )
-        handler.add_auth(req=request)
-        headers = request.headers
-        headers['User-Agent'] = USER_AGENT
-        return headers
+        req = requests.Request('GET', f'https://{HOST}{parsed_url.path}', params=params, data='', headers={})
+        prepared_request = req.prepare()
+        prepared_request.headers['User-Agent'] = USER_AGENT
+        aws_request = AWSRequest(method=prepared_request.method, url=prepared_request.url, data=prepared_request.body,
+                                 headers=prepared_request.headers)
+        SigV4Auth(credentials, 'imdbapi', 'us-east-1').add_auth(aws_request)
+        return aws_request.prepare().headers
+

src/imdbpie/imdbpie.py

@@ -9,17 +9,15 @@
 
 from trans import trans
 import requests
-from six import text_type
-from six.moves import http_client as httplib
-from six.moves.urllib.parse import urlencode, urljoin, quote
+from urllib.parse import urlencode, urljoin, quote
+from http import client as httplib
 
 from .constants import BASE_URI, SEARCH_BASE_URI
 from .auth import Auth
 from .exceptions import ImdbAPIError
 
 logger = logging.getLogger(__name__)
 
-
 # client method name -> api path
 _SIMPLE_GET_ENDPOINTS = {
     'get_name_images': '/name/{imdb_id}/images',
@@ -50,9 +48,9 @@
     'get_name_filmography': '/name/{imdb_id}/filmography',
 }
 
-
 class Imdb(Auth):
     def __init__(self, locale=None, exclude_episodes=False, session=None):
+        super().__init__()
         self.locale = locale or 'en_US'
         self.region = self.locale.split('_')[-1].upper()
         self.exclude_episodes = exclude_episodes
@@ -191,7 +189,7 @@ def search_for_title(self, title):
                 continue
             result_item = {
                 'title': result['l'],
-                'year': text_type(result['y']) if result.get('y') else None,
+                'year': str(result['y']) if result.get('y') else None,
                 'imdb_id': result['id'],
                 'type': result.get('q'),
             }
@@ -341,15 +339,17 @@ def is_redirection_title(self, imdb_id):
         else:
             return False
 
-    def _query_first_alpha_num(self, query):
+    @staticmethod
+    def _query_first_alpha_num(query):
         for char in query.lower():
             if char.isalnum():
                 return char
         raise ValueError(
             'invalid query, does not contain any alphanumeric characters'
         )
 
-    def _title_not_found(self, msg=''):
+    @staticmethod
+    def _title_not_found(msg=''):
         if msg:
             msg = ' {0}'.format(msg)
-        raise LookupError('Title not found.{0}'.format(msg))
+        raise LookupError('Title not found.{0}'.format(msg))

test_requirements.txt

@@ -1,12 +1,12 @@
 -r requirements.txt
 pytest
-coverage==4.4.2
+coverage
 flake8
 pep8-naming
 mccabe
 freezegun
 pytest-cov
 codecov
-setuptools>=65.5.1 # not directly required, pinned by Snyk to avoid a vulnerability
-brunette>=0.2.8,<1.0.0
-wheel>=0.38.0 # not directly required, pinned by Snyk to avoid a vulnerability
+setuptools
+brunette
+wheel