From 7ae9dd572d0e7d7ef6aba55e8a7bb55315629d4a Mon Sep 17 00:00:00 2001 From: Simon Willison Date: Thu, 10 Sep 2026 16:29:28 -0700 Subject: [PATCH] Match table permissions using SQLite case-insensitive names --- datasette/app.py | 86 +++++++- datasette/default_permissions.py | 7 +- tests/test_table_permissions.py | 360 +++++++++++++++++++++++++++++++ 3 files changed, 440 insertions(+), 13 deletions(-) create mode 100644 tests/test_table_permissions.py diff --git a/datasette/app.py b/datasette/app.py index f256f2c8..047128e3 100644 --- a/datasette/app.py +++ b/datasette/app.py @@ -2,6 +2,7 @@ import asyncio from typing import Sequence, Union, Tuple, Optional import asgi_csrf import collections +import copy import datetime import functools import glob @@ -87,6 +88,9 @@ app_root = Path(__file__).parent.parent # https://github.com/simonw/datasette/issues/283#issuecomment-781591015 SQLITE_LIMIT_ATTACHED = 10 +_SQLITE_IDENTIFIER_CASE = str.maketrans( + "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz" +) Setting = collections.namedtuple("Setting", ("name", "default", "help")) SETTINGS = ( @@ -474,6 +478,65 @@ class Datasette: orig[key] = upd_value return orig + def _metadata_sources(self, key, database, table): + yield from pm.hook.get_metadata( + datasette=self, key=key, database=database, table=table + ) + # Local configuration takes precedence over plugin metadata. + yield self._metadata_local + + def _table_permission_allows(self, database, table): + def normalize_tables(source, inherited): + if not isinstance(source, dict): + return source + source = dict(source) + if isinstance(source.get("tables"), dict): + tables = {} + for name, config in source["tables"].items(): + name = name.translate(_SQLITE_IDENTIFIER_CASE) + previous = ((inherited or {}).get("tables") or {}).get(name) + previous_configs = previous["configs"] if previous else [{}] + # Keep source aliases separate so none can overwrite a denial. + configs = tables.setdefault(name, {"configs": []})["configs"] + for previous_config in previous_configs: + if isinstance(previous_config, dict) and isinstance( + config, dict + ): + merged = self._metadata_recursive_update( + copy.deepcopy(previous_config), config + ) + else: + merged = config + if merged not in configs: + configs.append(merged) + source["tables"] = tables + return source + + metadata = {} + for source in self._metadata_sources("tables", database, None): + source = normalize_tables(copy.deepcopy(source), metadata) + if isinstance(source, dict) and isinstance(source.get("databases"), dict): + databases = dict(source["databases"]) + if database in databases: + databases[database] = normalize_tables( + databases[database], + (metadata.get("databases") or {}).get(database), + ) + source["databases"] = databases + metadata = self._metadata_recursive_update(metadata, source) + + database_metadata = (metadata.get("databases") or {}).get(database) or {} + tables = database_metadata.get("tables", metadata.get("tables")) or {} + configs = (tables.get(table.translate(_SQLITE_IDENTIFIER_CASE)) or {}).get( + "configs", [] + ) + # Same-source aliases are rules for one resource: an explicit denial wins. + return [ + config["allow"] + for config in configs + if config and config.get("allow") is not None + ] + def metadata(self, key=None, database=None, table=None, fallback=True): """ Looks up metadata, cascading backwards from specified level. @@ -484,17 +547,9 @@ class Datasette: ), "Cannot call metadata() with table= specified but not database=" metadata = {} - for hook_dbs in pm.hook.get_metadata( - datasette=self, key=key, database=database, table=table - ): + for hook_dbs in self._metadata_sources(key, database, table): metadata = self._metadata_recursive_update(metadata, hook_dbs) - # security precaution!! don't allow anything in the local config - # to be overwritten. this is a temporary measure, not sure if this - # is a good idea long term or maybe if it should just be a concern - # of the plugin's implemtnation - metadata = self._metadata_recursive_update(metadata, self._metadata_local) - databases = metadata.get("databases") or {} search_list = [] @@ -689,6 +744,19 @@ class Datasette: async def permission_allowed(self, actor, action, resource=None, default=False): """Check permissions using the permissions_allowed plugin hook""" + if action == "view-table" and resource is not None: + database, table = resource + db = self.databases.get(database) + if db is not None: + # Use SQLite's spelling for both table and view permission hooks. + # NOCASE folds ASCII only, unlike str.lower() or str.casefold(). + result = await db.execute( + "select name from sqlite_master " + "where type in ('table', 'view') and name = ? collate nocase", + (table,), + ) + if result.rows: + resource = (database, result.rows[0][0]) result = None for check in pm.hook.permission_allowed( datasette=self, diff --git a/datasette/default_permissions.py b/datasette/default_permissions.py index a0681e83..7908c88e 100644 --- a/datasette/default_permissions.py +++ b/datasette/default_permissions.py @@ -21,11 +21,10 @@ def permission_allowed(datasette, actor, action, resource): return actor_matches_allow(actor, database_allow) elif action == "view-table": database, table = resource - tables = datasette.metadata("tables", database=database) or {} - table_allow = (tables.get(table) or {}).get("allow") - if table_allow is None: + table_allows = datasette._table_permission_allows(database, table) + if not table_allows: return None - return actor_matches_allow(actor, table_allow) + return all(actor_matches_allow(actor, allow) for allow in table_allows) elif action == "view-query": # Check if this query has a "allow" block in metadata database, query_name = resource diff --git a/tests/test_table_permissions.py b/tests/test_table_permissions.py new file mode 100644 index 00000000..2b82d4c2 --- /dev/null +++ b/tests/test_table_permissions.py @@ -0,0 +1,360 @@ +import copy +import sqlite3 + +import pytest + +from datasette import hookimpl +from datasette.app import Datasette +from datasette.plugins import pm + + +@pytest.fixture +def ds(tmp_path): + path = tmp_path / "catalog.db" + with sqlite3.connect(path) as connection: + connection.executescript( + "create table Items (id integer primary key, label text);" + "insert into Items values (1, 'Example');" + "create view Report as select id, label from Items;" + 'create table "Ärea" (id integer primary key);' + 'create table "ärea" (id integer primary key);' + ) + datasette = Datasette([path]) + yield datasette + datasette.executor.shutdown(wait=True) + + +@pytest.fixture +def register_plugin(): + plugins = [] + + def register(plugin): + pm.register(plugin) + plugins.append(plugin) + return plugin + + yield register + for plugin in reversed(plugins): + pm.unregister(plugin) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "requested,canonical", + [ + ("Items", "Items"), + ("items", "Items"), + ("ITEMS", "Items"), + ("ItEmS", "Items"), + ("Report", "Report"), + ("report", "Report"), + ("REPORT", "Report"), + ("ÄREA", "Ärea"), + ("äREA", "ärea"), + ], +) +async def test_table_permission_hook_and_log_use_schema_spelling( + ds, register_plugin, requested, canonical +): + calls = [] + + class Observer: + @hookimpl + def permission_allowed(self, action, resource): + calls.append((action, resource)) + + register_plugin(Observer()) + assert await ds.permission_allowed( + None, "view-table", ("catalog", requested), default=True + ) + assert calls == [("view-table", ("catalog", canonical))] + assert ds._permission_checks[-1]["resource"] == ("catalog", canonical) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("name", ["items", "ITEMS", "report", "REPORT"]) +async def test_table_allow_and_visibility_use_ascii_identity(ds, name): + ds._metadata_local = { + "databases": { + "catalog": { + "tables": { + "iTeMs": {"allow": {"id": "reader"}}, + "rEpOrT": {"allow": {"id": "reader"}}, + } + } + } + } + assert await ds.check_visibility( + {"id": "reader"}, "view-table", ("catalog", name) + ) == (True, True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("name,actor_id", [("ÄREA", "first"), ("äREA", "second")]) +async def test_non_ascii_table_identities_remain_distinct(ds, name, actor_id): + ds._metadata_local = { + "databases": { + "catalog": { + "tables": { + "Ärea": {"allow": {"id": "first"}}, + "ärea": {"allow": {"id": "second"}}, + } + } + } + } + assert ( + await ds.permission_allowed( + {"id": actor_id}, "view-table", ("catalog", name), default=None + ) + is True + ) + assert ds._table_permission_allows("catalog", name) == [{"id": actor_id}] + + +@pytest.mark.parametrize( + "plugin_name,local_name", [("Items", "ITEMS"), ("ITEMS", "Items")] +) +def test_local_metadata_precedence_across_table_spellings( + ds, register_plugin, plugin_name, local_name +): + plugin_metadata = { + "databases": { + "catalog": { + "tables": { + plugin_name: {"allow": {"id": "plugin"}, "description": "Plugin"} + } + } + } + } + ds._metadata_local = { + "databases": {"catalog": {"tables": {local_name: {"allow": {"id": "local"}}}}} + } + original_plugin = copy.deepcopy(plugin_metadata) + original_local = copy.deepcopy(ds._metadata_local) + calls = [] + + class Metadata: + @hookimpl + def get_metadata(self, key, database, table): + calls.append((key, database, table)) + return plugin_metadata + + register_plugin(Metadata()) + assert ds._table_permission_allows("catalog", "items") == [{"id": "local"}] + assert calls == [("tables", "catalog", None)] + assert plugin_metadata == original_plugin + assert ds._metadata_local == original_local + # The ordinary public metadata API keeps the configured spellings. + public_tables = ds.metadata("tables", database="catalog") + assert public_tables[plugin_name]["allow"] == {"id": "plugin"} + assert public_tables[local_name]["allow"] == {"id": "local"} + + +@pytest.mark.parametrize( + "local_tables,expected", + [ + ({}, {"id": "reader"}), + ({"ITEMS": {"description": "Local"}}, {"id": "reader"}), + ({"ITEMS": {"allow": None}}, None), + ({"ITEMS": {"allow": {}}}, {"id": "reader"}), + (None, None), + ], +) +def test_legacy_nested_metadata_merge_is_preserved( + ds, register_plugin, local_tables, expected +): + class Metadata: + @hookimpl + def get_metadata(self): + return { + "databases": { + "catalog": {"tables": {"Items": {"allow": {"id": "reader"}}}} + } + } + + register_plugin(Metadata()) + ds._metadata_local = {"databases": {"catalog": {"tables": local_tables}}} + assert ds._table_permission_allows("catalog", "Items") == ( + [expected] if expected is not None else [] + ) + + +@pytest.mark.parametrize( + "database_metadata,expected", + [ + ({}, {"id": "fallback"}), + ({"tables": {}}, None), + ({"tables": None}, None), + ({"tables": {"items": {"allow": {"id": "database"}}}}, {"id": "database"}), + ], +) +def test_existing_instance_table_metadata_fallback(ds, database_metadata, expected): + ds._metadata_local = { + "tables": {"ITEMS": {"allow": {"id": "fallback"}}}, + "databases": {"catalog": database_metadata}, + } + assert ds._table_permission_allows("catalog", "Items") == ( + [expected] if expected is not None else [] + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("reverse", [False, True]) +@pytest.mark.parametrize("source", ["local", "plugin"]) +@pytest.mark.parametrize("requested", ["Items", "ITEMS"]) +async def test_same_source_case_variant_rules_use_deny_precedence( + ds, register_plugin, reverse, source, requested +): + entries = [("Items", {"allow": {"id": "reader"}}), ("ITEMS", {"allow": {}})] + if reverse: + entries.reverse() + metadata = {"databases": {"catalog": {"tables": dict(entries)}}} + if source == "local": + ds._metadata_local = metadata + else: + + class Metadata: + @hookimpl + def get_metadata(self): + return metadata + + register_plugin(Metadata()) + original = copy.deepcopy(metadata) + assert ( + await ds.permission_allowed( + {"id": "reader"}, "view-table", ("catalog", requested), default=None + ) + is False + ) + assert metadata == original + + +@pytest.mark.asyncio +@pytest.mark.parametrize("reverse", [False, True]) +async def test_local_allow_overlays_same_source_plugin_aliases( + ds, register_plugin, reverse +): + entries = [ + ("Items", {"allow": {"id": "first"}}), + ("ITEMS", {"allow": {"id": "second"}}), + ] + if reverse: + entries.reverse() + + class Metadata: + @hookimpl + def get_metadata(self): + return {"databases": {"catalog": {"tables": dict(entries)}}} + + register_plugin(Metadata()) + ds._metadata_local = { + "databases": {"catalog": {"tables": {"items": {"allow": {"id": "local"}}}}} + } + assert ds._table_permission_allows("catalog", "Items") == [{"id": "local"}] + assert ( + await ds.permission_allowed( + {"id": "local"}, "view-table", ("catalog", "Items"), default=None + ) + is True + ) + + +@pytest.mark.asyncio +async def test_plugin_permission_verdict_still_overrides_metadata(ds, register_plugin): + ds._metadata_local = { + "databases": {"catalog": {"tables": {"Items": {"allow": {"id": "reader"}}}}} + } + + class Policy: + @hookimpl + def permission_allowed(self, action, resource): + if action == "view-table": + assert resource == ("catalog", "Items") + return False + + register_plugin(Policy()) + assert ( + await ds.permission_allowed( + {"id": "reader"}, "view-table", ("catalog", "Items") + ) + is False + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "action,resource", + [ + ("view-database", "Catalog"), + ("execute-sql", "Catalog"), + ("no-match", ("catalog", "ITEMS")), + ("view-table", ("missing", "ITEMS")), + ("view-table", ("catalog", "FutureTable")), + ], +) +async def test_other_and_hypothetical_resources_keep_identity( + ds, register_plugin, action, resource +): + calls = [] + + class Observer: + @hookimpl + def permission_allowed(self, action, resource): + calls.append((action, resource)) + + register_plugin(Observer()) + assert await ds.permission_allowed(None, action, resource, default=True) + assert calls == [(action, resource)] + + +@pytest.mark.asyncio +async def test_canned_query_names_remain_case_sensitive(ds, register_plugin): + ds._metadata_local = { + "databases": { + "catalog": { + "queries": { + "Report": {"sql": "select 1", "allow": {"id": "first"}}, + "report": {"sql": "select 2", "allow": {"id": "second"}}, + } + } + } + } + for name, actor_id in (("Report", "first"), ("report", "second")): + assert ( + await ds.permission_allowed( + {"id": actor_id}, "view-query", ("catalog", name), default=None + ) + is True + ) + assert ds._permission_checks[-1]["resource"] == ("catalog", name) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "path", + [ + "/catalog", + "/catalog/Items.json", + "/catalog/Report.json", + "/catalog/Items/1.json", + ], +) +async def test_ordinary_requests_retain_table_permission_hooks( + ds, register_plugin, path +): + calls = [] + + class Observer: + @hookimpl + def permission_allowed(self, action, resource): + if action == "view-table": + calls.append(resource) + + register_plugin(Observer()) + response = await ds.client.get(path) + assert response.status_code == 200 + assert calls + assert all( + database == "catalog" and name in {"Items", "Report", "Ärea", "ärea"} + for database, name in calls + )