Filter incoming foreign-key relationships by view permission

datasette/views/row.py

@@ -578,7 +578,7 @@ class RowView(BaseView):
                 "private": private,
                 "columns": reordered_columns,
                 "foreign_key_tables": await self.foreign_key_tables(
-                    database, table, pk_values
+                    database, table, pk_values, actor=request.actor
                 ),
                 "database_color": db.color,
                 "display_columns": display_columns,
@@ -655,12 +655,23 @@ class RowView(BaseView):
             ),
         )
 
-    async def foreign_key_tables(self, database, table, pk_values):
+    async def foreign_key_tables(self, database, table, pk_values, *, actor):
         if len(pk_values) != 1:
             return []
         db = self.ds.databases[database]
         all_foreign_keys = await db.get_all_foreign_keys()
-        foreign_keys = all_foreign_keys[table]["incoming"]
+        foreign_keys = []
+        table_permissions = {}
+        for fk in all_foreign_keys[table]["incoming"]:
+            other_table = fk["other_table"]
+            if other_table not in table_permissions:
+                table_permissions[other_table] = await self.ds.allowed(
+                    action="view-table",
+                    resource=TableResource(database=database, table=other_table),
+                    actor=actor,
+                )
+            if table_permissions[other_table]:
+                foreign_keys.append(fk)
         if len(foreign_keys) == 0:
             return []
 

datasette/views/table_extras.py

@@ -1206,7 +1206,10 @@ class ForeignKeyTablesExtra(Extra):
 
     async def resolve(self, context):
         return await context.foreign_key_tables(
-            context.database_name, context.table_name, context.pk_values
+            context.database_name,
+            context.table_name,
+            context.pk_values,
+            actor=context.request.actor,
         )
 
 

tests/test_api.py

@@ -5,6 +5,7 @@ import pytest
 
 from datasette.app import Datasette
 from datasette.plugins import DEFAULT_PLUGINS
+from datasette.resources import DatabaseResource, TableResource
 from datasette.utils import UNSTABLE_API_MESSAGE, escape_sqlite, tilde_encode
 from datasette.utils.sqlite import sqlite_version
 from datasette.version import __version__
@@ -458,6 +459,67 @@ async def test_row_foreign_key_tables(ds_client):
     ]
 
 
+@pytest.mark.asyncio
+async def test_row_foreign_key_tables_omit_denied_tables(request):
+    actor = {"id": "reader"}
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "tables": {
+                        "parents": {"permissions": {"view-table": True}},
+                        "private_children": {"permissions": {"view-table": False}},
+                    }
+                }
+            }
+        },
+    )
+    request.addfinalizer(ds.close)
+    db = ds.add_memory_database("fk_count_leak", name="data")
+    await db.execute_write("create table parents (id integer primary key, name text)")
+    await db.execute_write("""
+        create table private_children (
+            id integer primary key,
+            parent_id integer references parents(id)
+        )
+    """)
+    await db.execute_write("insert into parents values (1, 'Public parent')")
+    await db.execute_write("""
+        insert into private_children (id, parent_id) values
+            (1, 1),
+            (2, 1),
+            (3, 1)
+    """)
+    await ds.invoke_startup()
+
+    parent = TableResource(database="data", table="parents")
+    private_children = TableResource(database="data", table="private_children")
+    assert await ds.allowed(action="view-table", resource=parent, actor=actor)
+    assert not await ds.allowed(
+        action="view-table", resource=private_children, actor=actor
+    )
+    assert not await ds.allowed(
+        action="execute-sql",
+        resource=DatabaseResource(database="data"),
+        actor=actor,
+    )
+
+    direct_child = await ds.client.get("/data/private_children.json", actor=actor)
+    assert direct_child.status_code == 403
+    parent_response = await ds.client.get(
+        "/data/parents/1.json?_extra=foreign_key_tables", actor=actor
+    )
+    assert parent_response.status_code == 200
+
+    foreign_key_tables = parent_response.json().get("foreign_key_tables", [])
+    assert foreign_key_tables == [], (
+        "denied child table name, foreign-key column, and row count disclosed: "
+        f"{foreign_key_tables}"
+    )
+
+
 @pytest.mark.asyncio
 async def test_row_extras(ds_client):
     response = await ds_client.get(