Implemented standalone IndieAuth

Still needs more tests and docs.

Closes #10, Closes #11, refs #2

Author: simonw

datasette_indieauth/__init__.py

@@ -1,49 +1,79 @@
 from datasette import hookimpl
-from .utils import display_url
+from .utils import (
+    build_authorization_url,
+    canonicalize_url,
+    discover_endpoints,
+    display_url,
+    verify_profile_url,
+)
 import httpx
+import itsdangerous
+from jinja2 import escape
 import urllib
 
+DATASETTE_INDIEAUTH_STATE = "datasette-indieauth-state"
+DATASETTE_INDIEAUTH_COOKIE = "datasette-indieauth-cookie"
+
 
 async def indieauth(request, datasette):
     return await indieauth_page(request, datasette)
 
 
-async def indieauth_page(request, datasette, initial_status=200):
+async def indieauth_page(request, datasette, status=200, error=None):
     from datasette.utils.asgi import Response
 
-    client_id = datasette.absolute_url(request, datasette.urls.instance())
-    redirect_uri = datasette.absolute_url(request, request.path)
+    urls = Urls(request, datasette)
+
+    if request.method == "POST":
+        post = await request.post_vars()
+        me = post.get("me")
+        if me:
+            me = canonicalize_url(me)
 
-    error = None
-    status = initial_status
+        if not me or not verify_profile_url(me):
+            datasette.add_message(
+                request, "Invalid IndieAuth identifier", message_type=datasette.ERROR
+            )
+            return Response.redirect(urls.login)
 
-    if request.args.get("code") and request.args.get("me"):
-        ok, extra = await verify_code(request.args["code"], client_id, redirect_uri)
-        if ok:
-            response = Response.redirect(datasette.urls.instance())
+        # Start the auth process
+        authorization_endpoint, token_endpoint = await discover_endpoints(me)
+        if not authorization_endpoint:
+            # Redirect to IndieAuth.com as a fallback
+            # TODO: Only do this if rel=me detected
+            # TODO: make this a configurable preference
+            indieauth_url = "https://indieauth.com/auth?" + urllib.parse.urlencode(
+                {
+                    "me": me,
+                    "client_id": urls.client_id,
+                    "redirect_uri": urls.indie_auth_com_redirect_uri,
+                }
+            )
+            return Response.redirect(indieauth_url)
+        else:
+            authorization_url, state, verifier = build_authorization_url(
+                authorization_endpoint=authorization_endpoint,
+                client_id=urls.client_id,
+                redirect_uri=urls.redirect_uri,
+                me=me,
+                signing_function=lambda x: datasette.sign(x, DATASETTE_INDIEAUTH_STATE),
+            )
+            response = Response.redirect(authorization_url)
             response.set_cookie(
-                "ds_actor",
+                "datasette_indieauth",
                 datasette.sign(
                     {
-                        "a": {
-                            "me": extra,
-                            "display": display_url(extra),
-                        }
+                        "v": verifier,
                     },
-                    "actor",
+                    DATASETTE_INDIEAUTH_COOKIE,
                 ),
             )
             return response
-        else:
-            error = extra
-            status = 403
 
     return Response.html(
         await datasette.render_template(
             "indieauth.html",
             {
-                "client_id": client_id,
-                "redirect_uri": redirect_uri,
                 "error": error,
             },
             request=request,
@@ -52,7 +82,138 @@ async def indieauth_page(request, datasette, initial_status=200):
     )
 
 
-async def verify_code(code, client_id, redirect_uri):
+async def indieauth_done(request, datasette):
+    from datasette.utils.asgi import Response
+
+    state = request.args.get("state")
+    code = request.args.get("code")
+    try:
+        state_bits = datasette.unsign(state, DATASETTE_INDIEAUTH_STATE)
+    except itsdangerous.BadSignature:
+        return await indieauth_page(
+            request, datasette, error="Invalid state returned by authorization server"
+        )
+    authorization_endpoint = state_bits["a"]
+
+    client_id = datasette.absolute_url(request, datasette.urls.instance())
+    redirect_path = datasette.urls.path("/-/indieauth/done")
+    redirect_uri = datasette.absolute_url(request, redirect_path)
+
+    # code_verifier should be in a signed cookie
+    code_verifier = None
+    if "datasette_indieauth" in request.cookies:
+        try:
+            cookie_bits = datasette.unsign(
+                request.cookies["datasette_indieauth"], DATASETTE_INDIEAUTH_COOKIE
+            )
+            code_verifier = cookie_bits["v"]
+        except itsdangerous.BadSignature:
+            pass
+    if not code_verifier:
+        return await indieauth_page(
+            request, datasette, error="Invalid datasette_indieauth cookie"
+        )
+
+    data = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "client_id": client_id,
+        "redirect_uri": redirect_uri,
+        "code_verifier": code_verifier,
+    }
+    async with httpx.AsyncClient() as client:
+        response = await client.post(authorization_endpoint, data=data)
+
+    if response.status_code == 200:
+        info = response.json()
+        if "me" not in info:
+            return await indieauth_page(
+                request,
+                datasette,
+                error="Invalid authorization_code response from authorization server",
+            )
+        me = info["me"]
+        actor = {
+            "me": me,
+            "display": display_url(me),
+        }
+        if "scope" in info:
+            actor["indieauth_scope"] = info["scope"]
+
+        if "profile" in info and isinstance(info["profile", dict]):
+            actor.update(info["profile"])
+        response = Response.redirect(datasette.urls.instance())
+        response.set_cookie(
+            "ds_actor",
+            datasette.sign(
+                {"a": actor},
+                "actor",
+            ),
+        )
+        return response
+    else:
+        return await indieauth_page(
+            request,
+            datasette,
+            error="Invalid authorization_code response from authorization server",
+        )
+
+
+async def indieauth_com_done(request, datasette):
+    from datasette.utils.asgi import Response
+
+    urls = Urls(request, datasette)
+
+    if not (request.args.get("code") and request.args.get("me")):
+        return Response.html("?code= and ?me= are required")
+    ok, extra = await verify_indieauth_com_code(
+        request.args["code"], urls.client_id, urls.indie_auth_com_redirect_uri
+    )
+    if ok:
+        response = Response.redirect(datasette.urls.instance())
+        response.set_cookie(
+            "ds_actor",
+            datasette.sign(
+                {
+                    "a": {
+                        "me": extra,
+                        "display": display_url(extra),
+                    }
+                },
+                "actor",
+            ),
+        )
+        return response
+    else:
+        return Response.html(escape(extra), status=403)
+
+
+class Urls:
+    def __init__(self, request, datasette):
+        self.request = request
+        self.datasette = datasette
+
+    def absolute(self, path):
+        return self.datasette.absolute_url(self.request, self.datasette.urls.path(path))
+
+    @property
+    def login(self):
+        return self.absolute("/-/indieauth")
+
+    @property
+    def client_id(self):
+        return self.absolute(self.datasette.urls.instance())
+
+    @property
+    def redirect_uri(self):
+        return self.absolute("/-/indieauth/done")
+
+    @property
+    def indie_auth_com_redirect_uri(self):
+        return self.absolute("/-/indieauth/indieauth-com-done")
+
+
+async def verify_indieauth_com_code(code, client_id, redirect_uri):
     async with httpx.AsyncClient() as client:
         response = await client.post(
             "https://indieauth.com/auth",
@@ -80,6 +241,8 @@ async def verify_code(code, client_id, redirect_uri):
 def register_routes():
     return [
         (r"^/-/indieauth$", indieauth),
+        (r"^/-/indieauth/done$", indieauth_done),
+        (r"^/-/indieauth/indieauth-com-done$", indieauth_com_done),
     ]
 
 

datasette_indieauth/templates/indieauth.html

@@ -6,11 +6,14 @@
 <h1>Sign in with IndieAuth</h1>
 
 {% if error %}<p class="message-error">{{ error }}</p>{% endif %}
-<form action="https://indieauth.com/auth" method="get">
+
+<form action="{{ urls.path("/-/indieauth") }}" method="post">
     <p><input type="text" name="me">
-        <input type="hidden" name="client_id" value="{{ client_id }}">
-        <input type="hidden" name="redirect_uri" value="{{ redirect_uri }}">
+        <input type="hidden" name="csrftoken" value="{{ csrftoken() }}">
         <input type="submit" value="Login">
     </p>
 </form>
-{% endblock %}
+
+<p>More about <a href="https://indieauth.net/">IndieAuth</a>.</p>
+
+{% endblock %}

datasette_indieauth/utils.py

@@ -109,22 +109,21 @@ def parse_link_rels(html):
 async def discover_endpoints(url):
     authorization_endpoint = None
     token_endpoint = None
+    chunk = None
     async with httpx.AsyncClient() as client:
-        async with client.stream("GET", url) as response:
-            # Check response.links first
-            if "authorization_endpoint" in response.links and response.links[
-                "authorization_endpoint"
-            ].get("url"):
-                authorization_endpoint = response.links["authorization_endpoint"]["url"]
-            if "token_endpoint" in response.links and response.links[
-                "token_endpoint"
-            ].get("url"):
-                token_endpoint = response.links["token_endpoint"]["url"]
-            if authorization_endpoint and token_endpoint:
-                # No need to consume any HTML at all
-                return authorization_endpoint, token_endpoint
-            # Just pull the first chunk - chunks are 64KB
-            chunk = next(response.iter_text())
+        response = await client.get(url)
+        # Check response.links first
+        if "authorization_endpoint" in response.links and response.links[
+            "authorization_endpoint"
+        ].get("url"):
+            authorization_endpoint = response.links["authorization_endpoint"]["url"]
+        if "token_endpoint" in response.links and response.links["token_endpoint"].get(
+            "url"
+        ):
+            token_endpoint = response.links["token_endpoint"]["url"]
+        if authorization_endpoint and token_endpoint:
+            return authorization_endpoint, token_endpoint
+        chunk = response.text
     rels = parse_link_rels(chunk)
     if authorization_endpoint is None:
         matches = [r["href"] for r in rels if r["rel"] == "authorization_endpoint"]
@@ -152,9 +151,7 @@ def challenge_verifier_pair(length=64):
     assert 43 <= length <= 128
     verifier = secrets.token_hex(length // 2)
     # challenge = BASE64URL-ENCODE(SHA256(ASCII(code_verifier)))
-    challenge = encode_challenge(
-        hashlib.sha256(verifier.encode("ascii")).digest()
-    )
+    challenge = encode_challenge(hashlib.sha256(verifier.encode("ascii")).digest())
     return challenge, verifier
 
 
@@ -174,12 +171,17 @@ def build_authorization_url(
     client_id,
     redirect_uri,
     me,
+    signing_function,
     scope=None,
     verifier_length=64
 ):
     "Returns (URL, state, verifier)"
     challenge, verifier = challenge_verifier_pair(verifier_length)
-    state = secrets.token_hex(16)
+    state = signing_function(
+        {
+            "a": authorization_endpoint,
+        }
+    )
     args = {
         "response_type": "code",
         "client_id": client_id,

tests/test_indieauth.py

@@ -19,15 +19,15 @@ async def test_plugin_is_installed():
 
 
 @pytest.mark.asyncio
-async def test_auth_succeeds(httpx_mock):
+async def test_indieauth_com_succeeds(httpx_mock):
     httpx_mock.add_response(
         url="https://indieauth.com/auth", data=b"me=https://simonwillison.net/"
     )
     datasette = Datasette([], memory=True)
     app = datasette.app()
     async with httpx.AsyncClient(app=app) as client:
         response = await client.get(
-            "http://localhost/-/indieauth?code=code&me=https://simonwillison.net/",
+            "http://localhost/-/indieauth/indieauth-com-done?code=code&me=https://simonwillison.net/",
             allow_redirects=False,
         )
         # Should set a cookie
@@ -38,7 +38,7 @@ async def test_auth_succeeds(httpx_mock):
 
 
 @pytest.mark.asyncio
-async def test_auth_fails(httpx_mock):
+async def test_indieauth_com_fails(httpx_mock):
     httpx_mock.add_response(
         url="https://indieauth.com/auth",
         status_code=404,
@@ -48,7 +48,7 @@ async def test_auth_fails(httpx_mock):
     app = datasette.app()
     async with httpx.AsyncClient(app=app) as client:
         response = await client.get(
-            "http://localhost/-/indieauth?code=code&me=example.com",
+            "http://localhost/-/indieauth/indieauth-com-done?code=code&me=example.com",
             allow_redirects=False,
         )
         # Should return error
@@ -77,12 +77,12 @@ async def test_restrict_access(httpx_mock):
         for path in paths:
             response = await client.get("http://localhost{}".format(path))
             assert response.status_code == 403
-            assert '<form action="https://indieauth.com/auth"' in response.text
+            assert '<form action="/-/indieauth" method="post">' in response.text
             assert "simonwillison.net" not in response.text
 
         # Now do the login and try again
         response2 = await client.get(
-            "http://localhost/-/indieauth?code=code&me=example.com",
+            "http://localhost/-/indieauth/indieauth-com-done?code=code&me=example.com",
             allow_redirects=False,
         )
         assert response2.status_code == 302