Check view-table as part of /-/schema and /db/-/schema

Refs GHSA-926p-cw2f-643h

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

datasette/views/special.py

@@ -1262,14 +1262,21 @@ class SchemaBaseView(BaseView):
 
     has_json_alternate = False
 
-    async def get_database_schema(self, database_name):
+    async def get_database_schema(self, database_name, actor):
         """Get schema SQL for a database."""
         db = self.ds.databases[database_name]
-        result = await db.execute(
-            "select group_concat(sql, ';' || CHAR(10)) as schema from sqlite_master where sql is not null"
+        allowed_tables_page = await self.ds.allowed_resources(
+            "view-table", actor, parent=database_name
+        )
+        allowed_table_names = {
+            resource.child async for resource in allowed_tables_page.all()
+        }
+        result = await db.execute(
+            "select tbl_name, sql from sqlite_master where sql is not null"
+        )
+        return ";\n".join(
+            row["sql"] for row in result.rows if row["tbl_name"] in allowed_table_names
         )
-        row = result.first()
-        return row["schema"] if row and row["schema"] else ""
 
     def format_json_response(self, data):
         """Format data as JSON response with CORS headers if needed."""
@@ -1331,7 +1338,7 @@ class InstanceSchemaView(SchemaBaseView):
         # Get schema for each database
         schemas = []
         for database_name in allowed_databases:
-            schema = await self.get_database_schema(database_name)
+            schema = await self.get_database_schema(database_name, request.actor)
             schemas.append({"database": database_name, "schema": schema})
 
         if format_ == "json":
@@ -1372,7 +1379,7 @@ class DatabaseSchemaView(SchemaBaseView):
         if database_name not in self.ds.databases:
             return self.format_error_response("Database not found", format_)
 
-        schema = await self.get_database_schema(database_name)
+        schema = await self.get_database_schema(database_name, request.actor)
 
         if format_ == "json":
             return self.format_json_response(

tests/test_schema_endpoints.py

@@ -246,3 +246,90 @@ async def test_table_not_exists(schema_ds):
     response = await schema_ds.client.get("/schema_public_db/nonexistent/-/schema.md")
     assert response.status_code == 404
     assert "not found" in response.text.lower()
+
+
+@pytest_asyncio.fixture(scope="module")
+async def schema_table_perms_ds():
+    """
+    A database that is viewable by anonymous users, but with one table
+    locked down using the documented per-table lockdown recipe:
+    a table-level allow block combined with allow_sql: false.
+    """
+    ds = Datasette(
+        config={
+            "databases": {
+                "schema_table_perms_db": {
+                    "allow_sql": False,
+                    "tables": {"employee_salaries": {"allow": {"id": "root"}}},
+                }
+            }
+        }
+    )
+    db = ds.add_memory_database("schema_table_perms_db")
+    await db.execute_write(
+        "CREATE TABLE IF NOT EXISTS public_posts (id INTEGER PRIMARY KEY, title TEXT)"
+    )
+    await db.execute_write(
+        "CREATE TABLE IF NOT EXISTS employee_salaries "
+        "(id INTEGER PRIMARY KEY, ssn TEXT, salary_usd INTEGER)"
+    )
+    await db.execute_write(
+        "CREATE INDEX IF NOT EXISTS idx_employee_salaries_ssn ON employee_salaries(ssn)"
+    )
+    await db.execute_write(
+        "CREATE TRIGGER IF NOT EXISTS trg_employee_salaries "
+        "AFTER INSERT ON employee_salaries BEGIN SELECT 1; END"
+    )
+    return ds
+
+
+@pytest.mark.asyncio
+async def test_schema_table_perms_controls(schema_table_perms_ds):
+    """Sanity check: the locked down table really is denied to anonymous users."""
+    ds = schema_table_perms_ds
+    for path in (
+        "/schema_table_perms_db/employee_salaries.json",
+        "/schema_table_perms_db/employee_salaries/-/schema.json",
+        "/schema_table_perms_db/-/query.json?sql=select+*+from+employee_salaries",
+    ):
+        response = await ds.client.get(path)
+        assert response.status_code == 403, path
+    response = await ds.client.get("/schema_table_perms_db.json")
+    assert response.status_code == 200
+    assert "employee_salaries" not in response.text
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "base_url",
+    ["/-/schema", "/schema_table_perms_db/-/schema"],
+)
+@pytest.mark.parametrize("format_ext", ["json", "md", ""])
+async def test_schema_parent_views_hide_denied_tables(
+    schema_table_perms_ds, base_url, format_ext
+):
+    """
+    GHSA-926p-cw2f-643h: /-/schema and /db/-/schema must not disclose the DDL
+    of tables the actor is denied view-table on, including indexes and
+    triggers that belong to those tables.
+    """
+    url = base_url + (f".{format_ext}" if format_ext else "")
+
+    # Anonymous: allowed table visible, denied table (and its columns,
+    # index and trigger) absent
+    response = await schema_table_perms_ds.client.get(url)
+    assert response.status_code == 200
+    assert "public_posts" in response.text
+    assert "employee_salaries" not in response.text
+    assert "ssn" not in response.text
+    assert "salary_usd" not in response.text
+    assert "idx_employee_salaries_ssn" not in response.text
+    assert "trg_employee_salaries" not in response.text
+
+    # root can see everything
+    response = await schema_table_perms_ds.client.get(url, actor={"id": "root"})
+    assert response.status_code == 200
+    assert "public_posts" in response.text
+    assert "CREATE TABLE employee_salaries" in response.text
+    assert "idx_employee_salaries_ssn" in response.text
+    assert "trg_employee_salaries" in response.text