Merge pull request #99 from PlaidWeb/feature/92-pkce-support

Add PKCE support

Makefile

@@ -19,7 +19,7 @@ flake8:
 
 .PHONY: mypy
 mypy:
-	poetry run mypy -p authl -m test --ignore-missing-imports
+	poetry run mypy -p authl -m test_app --ignore-missing-imports
 
 .PHONY: preflight
 preflight:

authl/flask.py

@@ -216,9 +216,7 @@ class AuthlFlask:
         browser window closing
 
     :param tokens.TokenStore token_storage: Storage for token data for
-        methods which use it. Defaults to :py:class:`tokens.Serializer` using
-        the Flask ``app.secret_key``; this is suitable for load-balancing
-        scenarios as long as all service nodes use the same secret_key.
+        methods which use it. Uses the same default as :py:func:`authl.from_config`.
 
         Note that if the default is used, the ``app.secret_key`` **MUST** be set
         before this class is initialized.
@@ -258,10 +256,6 @@ def __init__(self,
         if state_storage is None:
             state_storage = flask.session
 
-        if token_storage is None:
-            assert app.secret_key, "app.secret_key must be set to use default token_storage"
-            token_storage = tokens.Serializer(app.secret_key)
-
         self.authl = from_config(
             config,
             state_storage,

authl/handlers/indieauth.py

@@ -18,6 +18,7 @@
 """
 
 import logging
+import secrets
 import time
 import typing
 import urllib.parse
@@ -254,15 +255,7 @@ def verify_id(request_id: str, response_id: str) -> str:
 
 
 class IndieAuth(Handler):
-    """ Supports login via IndieAuth.
-
-    **SECURITY NOTE:** When used with tokens.Serializer, this is subject to certain
-    classes of replay attack; for example, if the user endpoint uses irrevocable
-    signed tokens for the code grant (which is done in many endpoints, e.g.
-    SelfAuth), an attacker can replay a transaction that it intercepts. As such
-    it is very important that the timeout be configured to a reasonably short
-    time to mitigate this.
-    """
+    """ Supports login via IndieAuth. """
 
     @property
     def service_name(self):
@@ -292,7 +285,12 @@ def __init__(self, client_id: typing.Union[str, typing.Callable[..., str]],
         :param client_id: The client_id to send to the remote IndieAuth
             provider. Can be a string or a function that returns a string.
 
-        :param token_store: Storage for the tokens
+        :param token_store: Storage for the tokens.
+
+            ***Security note:*** :py:class:`tokens.Serializer` is not supported,
+            as it makes this handler subject to replay attacks when used with
+            many common IndieAuth servers, and also prevents PKCE from being
+            effective.
 
         :param int timeout: Maximum time to wait for login to complete
             (default: 600)
@@ -303,6 +301,11 @@ def __init__(self, client_id: typing.Union[str, typing.Callable[..., str]],
         self._token_store = token_store
         self._timeout = timeout or 600
 
+        if isinstance(token_store, tokens.Serializer):
+            LOGGER.error(
+                "Cannot use tokens.Serializer with IndieAuth due to security considerations")
+            raise ValueError("Cannot use tokens.Serializer with IndieAuth")
+
     def handles_url(self, url):
         """
         If this page is already known to have an IndieAuth endpoint, we reuse
@@ -322,7 +325,9 @@ def initiate_auth(self, id_url, callback_uri, redir):
         if not endpoint:
             return disposition.Error("Failed to get IndieAuth endpoint", redir)
 
-        state = self._token_store.put((id_url, endpoint, callback_uri, time.time(), redir))
+        verifier = secrets.token_urlsafe()
+        state = self._token_store.put(
+            (id_url, endpoint, callback_uri, verifier, time.time(), redir))
 
         client_id = utils.resolve_value(self._client_id)
         LOGGER.debug("Using client_id %s", client_id)
@@ -331,20 +336,22 @@ def initiate_auth(self, id_url, callback_uri, redir):
             'redirect_uri': callback_uri,
             'client_id': client_id,
             'state': state,
+            'code_challenge': utils.pkce_challenge(verifier),
+            'code_challenge_method': 'S256',
             'response_type': 'code',
             'scope': 'profile email',
             'me': id_url})
         return disposition.Redirect(url)
 
     def check_callback(self, url, get, data):
-        # pylint:disable=too-many-return-statements
+        # pylint:disable=too-many-return-statements,too-many-locals
 
         state = get.get('state')
         if not state:
             return disposition.Error("No transaction provided", '')
 
         try:
-            id_url, endpoint, callback_uri, when, redir = self._token_store.pop(state)
+            id_url, endpoint, callback_uri, verifier, when, redir = self._token_store.pop(state)
         except (KeyError, ValueError):
             return disposition.Error("Invalid token", '')
 
@@ -356,7 +363,8 @@ def check_callback(self, url, get, data):
             request = requests.post(endpoint, data={
                 'code': get['code'],
                 'client_id': utils.resolve_value(self._client_id),
-                'redirect_uri': callback_uri
+                'redirect_uri': callback_uri,
+                'code_verifier': verifier,
             }, headers={'accept': 'application/json'})
 
             if request.status_code != 200:

authl/tokens.py

@@ -113,6 +113,9 @@ class Serializer(TokenStore):
     need to share the same secret_key.
 
     Also note that tokens stored in this way cannot be revoked individually.
+
+    Additionally, this token storage mechanism may limit the security of some
+    of the identity providers.
     """
 
     def __init__(self, secret_key):

authl/utils.py

@@ -1,5 +1,7 @@
 """ Utility functions """
 
+import base64
+import hashlib
 import logging
 import os.path
 import typing
@@ -63,3 +65,19 @@ def normalize(url):
 
     # Last history item was a permanent redirect, or there was no history
     return normalize(response.url)
+
+
+def pkce_challenge(verifier: str, method: str = 'S256') -> str:
+    """ Convert a PKCE verifier string to a challenge string
+
+    See RFC 7636 """
+
+    if method == 'plain':
+        return verifier
+
+    if method == 'S256':
+        hashed = hashlib.sha256(verifier.encode()).digest()
+        encoded = base64.urlsafe_b64encode(hashed)
+        return encoded.decode().strip('=')
+
+    raise ValueError(f'Unknown PKCE method {method}')

pyproject.toml

@@ -32,6 +32,7 @@ pytest = "^5.4.3"
 requests-mock = "^1.8.0"
 sphinx = "^3.1.2"
 pytest-mock = "^3.2.0"
+gunicorn = "^20.1.0"
 
 [build-system]
 requires = ["poetry>=0.12"]

test.sh

@@ -1,4 +1,4 @@
 #!/bin/sh
 
 poetry install
-FLASK_DEBUG=1 FLASK_APP=test.py poetry run flask run
+FLASK_DEBUG=1 FLASK_APP=test_app.py poetry run flask run

tests/handlers/test_indieauth.py

@@ -9,7 +9,7 @@
 import requests
 from bs4 import BeautifulSoup
 
-from authl import disposition, tokens
+from authl import disposition, tokens, utils
 from authl.handlers import indieauth
 
 from . import parse_args
@@ -224,6 +224,8 @@ def test_handler_success(requests_mock):
     assert user_get['state'] in store
     assert user_get['response_type'] == 'code'
     assert 'me' in user_get
+    challenge = user_get['code_challenge']
+    assert user_get['code_challenge_method'] == 'S256'
 
     # fake the verification response
     def verify_callback(request, _):
@@ -232,6 +234,8 @@ def verify_callback(request, _):
         assert args['code'] == ['asdf']
         assert args['client_id'] == ['http://client/']
         assert 'redirect_uri' in args
+        verifier = args['code_verifier'][0]
+        assert utils.pkce_challenge(verifier) == challenge
         return json.dumps({
             'me': 'https://example.user/bob',
             'profile': {
@@ -558,3 +562,8 @@ def test_profile_caching(requests_mock):
     assert profile['email'] == 'qwer@poiu.fojar'
 
     assert profile_mock.call_count == 1
+
+
+def test_security_requirements():
+    with pytest.raises(Exception):
+        indieauth.IndieAuth('foobarbaz', tokens.Serializer('qwerpoiu'))

tests/test_utils.py

@@ -2,6 +2,7 @@
 # pylint:disable=missing-docstring
 
 
+import pytest
 import requests
 
 from authl import utils
@@ -82,3 +83,10 @@ def test_permanent_url(requests_mock):
     assert req.url == 'https://four/'
     assert utils.permanent_url(req) == 'https://four/'
     assert req.text == 'done'
+
+
+def test_pkce_challenge():
+    assert utils.pkce_challenge('asdf', 'plain') == 'asdf'
+    assert utils.pkce_challenge('foo', 'S256') == 'LCa0a2j_xo_5m0U8HTBBNBNCLXBkg7-g-YpeiGJm564'
+    with pytest.raises(Exception):
+        utils.pkce_challenge('moo', 'plap')