Disallow ?_through= if user lacks view-table permission

Refs GHSA-53fc-rhfg-h7qp

Co-authored-by: Alex Garcia <15178711+asg017@users.noreply.github.com>

datasette/filters.py

@@ -2,7 +2,7 @@ import json
 from typing import ClassVar
 
 from datasette import hookimpl
-from datasette.resources import DatabaseResource
+from datasette.resources import DatabaseResource, TableResource
 from datasette.utils.asgi import BadRequest
 from datasette.views.base import DatasetteError
 
@@ -135,6 +135,11 @@ def through_filters(request, database, table, datasette):
                 through_table = through_data["table"]
                 other_column = through_data["column"]
                 value = through_data["value"]
+                await datasette.ensure_permission(
+                    action="view-table",
+                    resource=TableResource(database=database, table=through_table),
+                    actor=request.actor,
+                )
                 db = datasette.get_database(database)
                 outgoing_foreign_keys = await db.foreign_keys_for_table(through_table)
                 fk_to_us = next(

tests/test_table_api.py

@@ -1778,3 +1778,34 @@ async def test_next_url_included_by_default(ds_client):
         data = response.json()
     assert data["next"] is None
     assert data["next_url"] is None
+
+
+@pytest.mark.asyncio
+async def test_table_through_requires_view_table_on_through_table():
+    # GHSA-53fc-rhfg-h7qp issue 3: ?_through= runs a sub-select against the
+    # caller-supplied through table, so the actor must be allowed to view it.
+    # Otherwise it is an equality oracle over any column of a denied table.
+    from datasette.app import Datasette
+
+    ds = Datasette(
+        memory=True,
+        config={"databases": {"data": {"tables": {"salaries": {"allow": False}}}}},
+    )
+    db = ds.add_memory_database("table_through_denied", name="data")
+    await db.execute_write("create table people (id integer primary key, name text)")
+    await db.execute_write(
+        "create table salaries (id integer primary key, "
+        "person_id integer references people(id), note text)"
+    )
+    await db.execute_write("insert into people values (1, 'alice'), (2, 'bob')")
+    await db.execute_write("insert into salaries values (1, 1, 'TOPSECRET-A')")
+    await ds.invoke_startup()
+
+    # Sanity: anonymous cannot read salaries directly
+    assert (await ds.client.get("/data/salaries.json")).status_code == 403
+
+    response = await ds.client.get(
+        "/data/people.json?_shape=array"
+        '&_through={"table":"salaries","column":"note","value":"TOPSECRET-A"}'
+    )
+    assert response.status_code == 403, response.text