Update package dependencies and pylint/mypy hints thereof

Fixes #104

Makefile

@@ -6,7 +6,7 @@ setup:
 
 .PHONY: format
 format:
-	poetry run isort -y
+	poetry run isort .
 	poetry run autopep8 -r --in-place .
 
 .PHONY: pylint

authl/__init__.py

@@ -13,6 +13,7 @@
 import collections
 import logging
 import typing
+from typing import Optional
 
 import expiringdict
 from bs4 import BeautifulSoup
@@ -30,7 +31,7 @@ class Authl:
 
     """
 
-    def __init__(self, cfg_handlers: typing.List[handlers.Handler] = None):
+    def __init__(self, cfg_handlers: Optional[typing.List[handlers.Handler]] = None):
         """ Initialize an Authl library instance. """
         self._handlers: typing.Dict[str, handlers.Handler] = collections.OrderedDict()
         if cfg_handlers:
@@ -117,8 +118,8 @@ def handlers(self):
 
 
 def from_config(config: typing.Dict[str, typing.Any],
-                state_storage: dict = None,
-                token_storage: tokens.TokenStore = None) -> Authl:
+                state_storage: Optional[dict] = None,
+                token_storage: Optional[tokens.TokenStore] = None) -> Authl:
     """ Generate an Authl handler set from provided configuration directives.
 
     :param dict config: a configuration dictionary. See the individual handlers'

authl/disposition.py

@@ -10,6 +10,7 @@
 
 
 from abc import ABC
+from typing import Optional
 
 
 class Disposition(ABC):
@@ -56,7 +57,7 @@ class Verified(Disposition):
 
     """
 
-    def __init__(self, identity: str, redir: str, profile: dict = None):
+    def __init__(self, identity: str, redir: str, profile: Optional[dict] = None):
         self.identity = identity
         self.redir = redir
         self.profile = profile or {}

authl/flask.py

@@ -59,6 +59,7 @@ def logout():
 import os
 import typing
 import urllib.parse
+from typing import Optional
 
 import flask
 import werkzeug.exceptions as http_error
@@ -239,16 +240,16 @@ def __init__(self,
                  callback_name: str = 'authl.callback',
                  callback_path: str = '/cb',
                  tester_name: str = 'authl.test',
-                 tester_path: str = None,
-                 login_render_func: typing.Callable = None,
-                 notify_render_func: typing.Callable = None,
+                 tester_path: Optional[str] = None,
+                 login_render_func: Optional[typing.Callable] = None,
+                 notify_render_func: Optional[typing.Callable] = None,
                  session_auth_name: typing.Optional[str] = 'me',
                  force_https: bool = False,
-                 stylesheet: typing.Union[str, typing.Callable] = None,
-                 on_verified: typing.Callable = None,
+                 stylesheet: Optional[typing.Union[str, typing.Callable]] = None,
+                 on_verified: Optional[typing.Callable] = None,
                  make_permanent: bool = True,
-                 state_storage: typing.Dict = None,
-                 token_storage: tokens.TokenStore = None,
+                 state_storage: Optional[typing.Dict] = None,
+                 token_storage: Optional[tokens.TokenStore] = None,
                  session_namespace='_authl',
                  ):
         # pylint:disable=too-many-arguments,too-many-locals,too-many-statements

authl/handlers/__init__.py

@@ -45,7 +45,7 @@ def handles_url(self, url: str) -> typing.Optional[str]:
         Whatever value this returns will be passed back in to initiate_auth, so
         if that value matters, return a reasonable URL.
         """
-        # pylint:disable=no-self-use,unused-argument
+        # pylint:disable=unused-argument
         return None
 
     def handles_page(self, url: str, headers, content, links) -> bool:
@@ -64,7 +64,7 @@ def handles_page(self, url: str, headers, content, links) -> bool:
         .. _Requests: https://requests.readthedocs.io/
         .. _BeautifulSoup4: https://pypi.org/project/beautifulsoup4/
         """
-        # pylint:disable=no-self-use,unused-argument
+        # pylint:disable=unused-argument
         return False
 
     @property

authl/handlers/email_addr.py

@@ -16,6 +16,7 @@
 import math
 import time
 import urllib.parse
+from typing import Optional
 
 import expiringdict
 import validate_email
@@ -86,8 +87,8 @@ def __init__(self,
                  sendmail,
                  notify_cdata,
                  token_store: tokens.TokenStore,
-                 expires_time: int = None,
-                 pending_storage: dict = None,
+                 expires_time: Optional[int] = None,
+                 pending_storage: Optional[dict] = None,
                  email_template_text: str = DEFAULT_TEMPLATE_TEXT,
                  ):
         # pylint:disable=too-many-arguments

authl/handlers/fediverse.py

@@ -67,9 +67,10 @@ def __init__(self, name: str,
         self._homepage = homepage
         self._token_store = token_store
         self._timeout = timeout or 600
+        self._http_timeout = 30
 
     @staticmethod
-    def _get_instance(url) -> typing.Optional[str]:
+    def _get_instance(url, timeout: int) -> typing.Optional[str]:
         parsed = urllib.parse.urlparse(url)
         if not parsed.netloc:
             parsed = urllib.parse.urlparse('https://' + url)
@@ -79,7 +80,7 @@ def _get_instance(url) -> typing.Optional[str]:
 
         try:
             LOGGER.debug("Trying Fediverse instance: %s", instance)
-            request = requests.get(instance + '/api/v1/instance')
+            request = requests.get(instance + '/api/v1/instance', timeout=timeout)
             if request.status_code != 200:
                 LOGGER.debug("Instance endpoint returned error %d", request.status_code)
                 return None
@@ -104,7 +105,7 @@ def handles_url(self, url):
         """
         LOGGER.info("Checking URL %s", url)
 
-        instance = self._get_instance(url)
+        instance = self._get_instance(url, self._http_timeout)
         if not instance:
             LOGGER.debug("Not a Fediverse instance: %s", url)
             return None
@@ -151,7 +152,7 @@ def _get_identity(instance, response, redir) -> disposition.Disposition:
 
     def initiate_auth(self, id_url, callback_uri, redir):
         try:
-            instance = self._get_instance(id_url)
+            instance = self._get_instance(id_url, self._http_timeout)
             client_id, client_secret = mastodon.Mastodon.create_app(
                 api_base_url=instance,
                 client_name=self._name,
@@ -176,19 +177,16 @@ def initiate_auth(self, id_url, callback_uri, redir):
             redir
         ))
 
-        # mastodon.py doesn't support a state parameter for some reason, so we
-        # have to add it ourselves
-        # pylint:disable=consider-using-f-string
-        url = '{}&{}'.format(
-            client.auth_request_url(
-                redirect_uris=callback_uri,
-                scopes=['read:accounts'],
-            ),
-            urllib.parse.urlencode({'state': state}))
+        url = client.auth_request_url(
+            redirect_uris=callback_uri,
+            scopes=['read:accounts'],
+            state=state
+        )
 
         return disposition.Redirect(url)
 
     def check_callback(self, url, get, data):
+        print(url, get)
         try:
             (
                 instance,
@@ -215,7 +213,7 @@ def check_callback(self, url, get, data):
         )
 
         try:
-            access_token = client.log_in(
+            client.log_in(
                 code=get['code'],
                 redirect_uri=url,
                 scopes=['read:accounts'],
@@ -228,22 +226,7 @@ def check_callback(self, url, get, data):
         result = self._get_identity(instance, client.me(), redir)
 
         # clean up after ourselves
-        # mastodon.py does not currently support the revoke endpoint; see
-        # https://github.com/halcy/Mastodon.py/issues/217
-        try:
-            request = requests.post(instance + '/oauth/revoke', data={
-                'client_id': client_id,
-                'client_secret': client_secret,
-                'token': access_token
-            }, headers={
-                'Authorization': f'Bearer {access_token}',
-                'User-Agent': utils.get_user_agent(self._homepage),
-            })
-            LOGGER.info("OAuth token revocation: %d %s",
-                        request.status_code,
-                        request.text)
-        except Exception as err:  # pylint:disable=broad-except
-            LOGGER.warning("Token revocation failed: %s", err)
+        client.revoke_access_token()
 
         return result
 

authl/handlers/indieauth.py

@@ -22,6 +22,7 @@
 import time
 import typing
 import urllib.parse
+from typing import Optional
 
 import expiringdict
 import mf2py
@@ -42,8 +43,8 @@
 
 
 def find_endpoint(id_url: str,
-                  links: typing.Dict = None,
-                  content: BeautifulSoup = None,
+                  links: Optional[typing.Dict] = None,
+                  content: Optional[BeautifulSoup] = None,
                   rel: str = "authorization_endpoint") -> typing.Tuple[typing.Optional[str],
                                                                        str]:
     """ Given an identity URL, get its IndieAuth endpoint
@@ -60,9 +61,9 @@ def find_endpoint(id_url: str,
 
 
 def find_endpoints(id_url: str,
-                   links: typing.Dict = None,
-                   content: BeautifulSoup = None) -> typing.Tuple[typing.Dict[str, str],
-                                                                  str]:
+                   links: Optional[typing.Dict] = None,
+                   content: Optional[BeautifulSoup] = None) -> typing.Tuple[typing.Dict[str, str],
+                                                                            str]:
     """ Given an identity URL, discover its IndieWeb endpoints
 
     :param str id_url: an identity URL to check
@@ -163,9 +164,9 @@ def get_url(prop, scheme=None) -> typing.Tuple[typing.Optional[str],
 
 
 def get_profile(id_url: str,
-                server_profile: dict = None,
+                server_profile: Optional[dict] = None,
                 links=None,
-                content: BeautifulSoup = None,
+                content: Optional[BeautifulSoup] = None,
                 endpoints=None) -> dict:
     """ Given an identity URL, try to parse out an Authl profile
 
@@ -280,7 +281,7 @@ def logo_html(self):
         return [(utils.read_icon('indieauth.svg'), 'IndieAuth')]
 
     def __init__(self, client_id: typing.Union[str, typing.Callable[..., str]],
-                 token_store: tokens.TokenStore, timeout: int = None):
+                 token_store: tokens.TokenStore, timeout: Optional[int] = None):
         """
         :param client_id: The client_id to send to the remote IndieAuth
             provider. Can be a string or a function that returns a string.
@@ -300,6 +301,7 @@ def __init__(self, client_id: typing.Union[str, typing.Callable[..., str]],
         self._client_id = client_id
         self._token_store = token_store
         self._timeout = timeout or 600
+        self._http_timeout = 5
 
         if isinstance(token_store, tokens.Serializer):
             LOGGER.error(
@@ -369,7 +371,7 @@ def check_callback(self, url, get, data):
             }, headers={
                 'accept': 'application/json',
                 'User-Agent': f'{utils.USER_AGENT} for {client_id}',
-            })
+            }, timeout=self._http_timeout)
 
             if request.status_code != 200:
                 LOGGER.error("Request returned code %d: %s", request.status_code, request.text)

authl/handlers/twitter.py

@@ -21,6 +21,7 @@
 import re
 import time
 import urllib.parse
+from typing import Optional
 
 import expiringdict
 import requests
@@ -68,21 +69,21 @@ def logo_html(self):
 
     def __init__(self, client_key: str,
                  client_secret: str,
-                 timeout: int = None,
-                 storage: dict = None):
+                 timeout: Optional[int] = None,
+                 storage: Optional[dict] = None):
         # pylint:disable=too-many-arguments
         self._client_key = client_key
         self._client_secret = client_secret
         self._timeout = timeout or 600
         self._pending = expiringdict.ExpiringDict(
             max_len=128,
             max_age_seconds=self._timeout) if storage is None else storage
+        self._http_timeout = 30
 
     # regex to match a twitter URL and optionally extract the username
     twitter_regex = re.compile(r'(https?://)?[^/]*\.?twitter\.com/?@?([^?]*)')
 
-    @staticmethod
-    def handles_url(url):
+    def handles_url(self, url):
         match = Twitter.twitter_regex.match(url)
         if match:
             return 'https://twitter.com/' + match.group(2)
@@ -161,7 +162,7 @@ def check_callback(self, url, get, data):
 
             user_info = requests.get(
                 'https://api.twitter.com/1.1/account/verify_credentials.json?skip_status=1',
-                auth=auth).json()
+                auth=auth, timeout=self._http_timeout).json()
             LOGGER.log(logging.WARNING if 'errors' in user_info else logging.NOTSET,
                        "User profile showed error: %s", user_info.get('errors'))
             return disposition.Verified(
@@ -174,15 +175,14 @@ def check_callback(self, url, get, data):
             if auth:
                 # let's clean up after ourselves
                 request = requests.post('https://api.twitter.com/1.1/oauth/invalidate_token.json',
-                                        auth=auth)
+                                        auth=auth, timeout=self._http_timeout)
                 LOGGER.debug("Token revocation request: %d %s", request.status_code, request.text)
 
     @property
     def generic_url(self):
         return 'https://twitter.com/'
 
-    @staticmethod
-    def build_profile(user_info: dict) -> dict:
+    def build_profile(self, user_info: dict) -> dict:
         """ Convert a Twitter userinfo JSON into an Authl profile """
         entities = user_info.get('entities', {})
 
@@ -210,7 +210,8 @@ def expand_entities(name):
         # attempt to get a more suitable image
         if 'avatar' in profile:
             for rendition in ('_400x400', ''):
-                req = requests.head(profile['avatar'].replace('_normal', rendition))
+                req = requests.head(profile['avatar'].replace('_normal', rendition),
+                                    timeout=self._http_timeout)
                 if 200 <= req.status_code < 300:
                     LOGGER.info("Found better avatar rendition: %s", req.url)
                     profile['avatar'] = req.url

authl/tokens.py

@@ -15,6 +15,7 @@
 import typing
 import uuid
 from abc import ABC, abstractmethod
+from typing import Optional
 
 import expiringdict
 import itsdangerous
@@ -78,7 +79,7 @@ class DictStore(TokenStore):
     .. _expiringdict: https://pypi.org/project/expiringdict/
     """
 
-    def __init__(self, store: dict = None,
+    def __init__(self, store: Optional[dict] = None,
                  keygen: typing.Callable[..., str] = lambda _: str(uuid.uuid4())):
         """ Initialize the store """
         self._store: dict = expiringdict.ExpiringDict(