From 5d9a74f370bca53cc073e5970ce5ef1439938069 Mon Sep 17 00:00:00 2001 From: Simon Willison Date: Thu, 3 Sep 2026 14:36:03 -0700 Subject: [PATCH] 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 | 21 +++++--- tests/test_schema_endpoints.py | 87 ++++++++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 7 deletions(-) diff --git a/datasette/views/special.py b/datasette/views/special.py index b3bfeb5c..435d5583 100644 --- a/datasette/views/special.py +++ b/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( diff --git a/tests/test_schema_endpoints.py b/tests/test_schema_endpoints.py index 2ef78867..d3ed0477 100644 --- a/tests/test_schema_endpoints.py +++ b/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