Fix for security vulnerability caused by trusting "me" values

See GHSA-mjcr-rqjg-rhg3

datasette_indieauth/__init__.py

@@ -5,6 +5,7 @@
     discover_endpoints,
     display_url,
     verify_profile_url,
+    verify_same_domain,
 )
 import httpx
 import itsdangerous
@@ -59,6 +60,7 @@ async def indieauth_page(request, datasette, status=200, error=None):
                 datasette.sign(
                     {
                         "v": verifier,
+                        "m": me,
                     },
                     DATASETTE_INDIEAUTH_COOKIE,
                 ),
@@ -98,15 +100,17 @@ async def indieauth_done(request, datasette):
 
     # code_verifier should be in a signed cookie
     code_verifier = None
+    original_me = None
     if "ds_indieauth" in request.cookies:
         try:
             cookie_bits = datasette.unsign(
                 request.cookies["ds_indieauth"], DATASETTE_INDIEAUTH_COOKIE
             )
             code_verifier = cookie_bits["v"]
-        except itsdangerous.BadSignature:
+            original_me = cookie_bits["m"]
+        except (itsdangerous.BadSignature, KeyError):
             pass
-    if not code_verifier:
+    if not code_verifier or not original_me:
         return await indieauth_page(
             request, datasette, error="Invalid ds_indieauth cookie"
         )
@@ -134,6 +138,15 @@ async def indieauth_done(request, datasette):
                 error="Invalid authorization_code response from authorization server",
             )
         me = info["me"]
+
+        # Returned me must be on the same domain as original_me
+        if not verify_same_domain(me, original_me):
+            return await indieauth_page(
+                request,
+                datasette,
+                error='"me" value returned by authorization server had a domain that did not match the initial URL',
+            )
+
         actor = {
             "me": me,
             "display": display_url(me),

datasette_indieauth/utils.py

@@ -195,3 +195,9 @@ def build_authorization_url(
         args["scope"] = scope
     url = authorization_endpoint + "?" + urlencode(args)
     return url, state, verifier
+
+
+def verify_same_domain(url, other_url):
+    url_bits = urlparse(url)
+    other_url_bits = urlparse(other_url)
+    return url_bits.netloc == other_url_bits.netloc

tests/test_indieauth.py

@@ -125,6 +125,14 @@ async def test_h_app(title):
             None,
             "Invalid authorization_code response from authorization server",
         ),
+        # Security issue: returned me value must wrap original domain
+        (
+            "https://indieauth.simonwillison.net/",
+            200,
+            "me=https%3A%2F%2Findieauth.simonwillison.com%2F&scope=email",
+            None,
+            ""me" value returned by authorization server had a domain that did not match the initial URL",
+        ),
     ),
 )
 async def test_indieauth_flow(
@@ -165,7 +173,9 @@ async def test_indieauth_flow(
         assert post_response.status_code == 302
         assert "ds_indieauth" in post_response.cookies
         ds_indieauth = post_response.cookies["ds_indieauth"]
-        verifier = datasette.unsign(ds_indieauth, "datasette-indieauth-cookie")["v"]
+        ds_indieauth_bits = datasette.unsign(ds_indieauth, "datasette-indieauth-cookie")
+        verifier = ds_indieauth_bits["v"]
+        assert ds_indieauth_bits["m"] == me
         # Verify the location is in the right shape
         location = post_response.headers["location"]
         assert location.startswith("https://indieauth.simonwillison.net/auth?")
@@ -270,18 +280,25 @@ async def test_indieauth_errors(httpx_mock, me, bodies, expected_error):
 
 
 @pytest.mark.asyncio
-async def test_invalid_ds_indieauth_cookie():
+@pytest.mark.parametrize(
+    "bad_cookie", ["this-is-bad", {"v": "blah"}, {"m": "blah"}, {"s": "blah"}]
+)
+async def test_invalid_ds_indieauth_cookie(bad_cookie):
     datasette = Datasette([], memory=True)
     app = datasette.app()
     state = datasette.sign({"a": "auth-url"}, "datasette-indieauth-state")
+    if isinstance(bad_cookie, dict):
+        ds_indieauth = datasette.sign(bad_cookie, "datasette-indieauth-cookie")
+    else:
+        ds_indieauth = bad_cookie
     async with httpx.AsyncClient(app=app) as client:
         response = await client.get(
             "http://localhost/-/indieauth/done",
             params={
                 "state": state,
                 "code": "123",
             },
-            cookies={"ds_indieauth": "this-is-bad"},
+            cookies={"ds_indieauth": ds_indieauth},
             allow_redirects=False,
         )
     assert '<p class="message-error">Invalid ds_indieauth cookie' in response.text

tests/test_utils.py

@@ -216,3 +216,16 @@ def test_build_authorization_url():
     assert hashlib.sha256(verifier.encode("utf-8")).digest() == utils.decode_challenge(
         bits["code_challenge"]
     )
+
+
+@pytest.mark.parametrize(
+    "url,other_url,expected",
+    [
+        ("https://www.example.com/", "https://www.example.com/", True),
+        ("https://www.example.com/foo", "https://www.example.com/bar", True),
+        ("https://www.example.com/", "https://www.example.org/", False),
+        ("https://foo.example.com/", "https://www.example.com/", False),
+    ],
+)
+def test_verify_same_domain(url, other_url, expected):
+    assert utils.verify_same_domain(url, other_url) is expected