From 506c4bb522e857501ffaf9be68305648c9a39255 Mon Sep 17 00:00:00 2001 From: Simon Willison Date: Wed, 9 Sep 2026 07:54:48 -0700 Subject: [PATCH] Match table permission identities using SQLite case semantics --- datasette/app.py | 8 +- datasette/default_permissions/config.py | 13 +- datasette/default_permissions/restrictions.py | 14 +- datasette/permissions.py | 18 + datasette/resources.py | 1 + datasette/utils/actions_sql.py | 33 +- docs/authentication.rst | 9 + tests/test_table_resource_identity.py | 420 ++++++++++++++++++ 8 files changed, 497 insertions(+), 19 deletions(-) create mode 100644 tests/test_table_resource_identity.py diff --git a/datasette/app.py b/datasette/app.py index da36c3f3..acce7a4a 100644 --- a/datasette/app.py +++ b/datasette/app.py @@ -321,7 +321,7 @@ def _permission_cache_key(actor, action, parent, child): actor_key = ( json.dumps(actor, sort_keys=True, default=repr) if actor is not None else None ) - return (actor_key, action, parent, child) + return (actor_key, action.name, parent, action.normalize_child(child)) async def favicon(request, send): @@ -2103,7 +2103,7 @@ ORDER BY allowed.parent, allowed.child to_check = [] for name in expanded: if cache is not None: - key = _permission_cache_key(actor, name, parent, child) + key = _permission_cache_key(actor, self.actions[name], parent, child) if key in cache: final[name] = cache[key] continue @@ -2164,7 +2164,9 @@ ORDER BY allowed.parent, allowed.child # Cache the freshly computed checks if cache is not None: for name in to_check: - cache[_permission_cache_key(actor, name, parent, child)] = final[name] + cache[ + _permission_cache_key(actor, self.actions[name], parent, child) + ] = final[name] # Log every check (including cache hits) for the debug page, # dependencies before the actions that required them diff --git a/datasette/default_permissions/config.py b/datasette/default_permissions/config.py index 4494f07f..a4f5a4de 100644 --- a/datasette/default_permissions/config.py +++ b/datasette/default_permissions/config.py @@ -92,6 +92,13 @@ class ConfigPermissionProcessor: # Tables implicitly reference their parent databases self.restricted_databases.update(db for db, _ in self.restricted_tables) + # Resolve identity keys once per action, rather than scanning the + # restriction allowlist for every configured table's allow block. + self.restricted_table_keys = { + (db, self.action_obj.normalize_child(table) if self.action_obj else table) + for db, table in self.restricted_tables + } + def evaluate_allow_block(self, allow_block: Any) -> bool | None: """Evaluate an allow block against the current actor.""" if allow_block is None: @@ -125,8 +132,10 @@ class ConfigPermissionProcessor: if parent: table_restrictions = (self.restrictions.get("r", {}) or {}).get(parent, {}) if child: - table_actions = table_restrictions.get(child, []) - if self.action_checks.intersection(table_actions): + child_key = ( + self.action_obj.normalize_child(child) if self.action_obj else child + ) + if (parent, child_key) in self.restricted_table_keys: return True else: # Parent query should proceed if any child in this database is allowlisted diff --git a/datasette/default_permissions/restrictions.py b/datasette/default_permissions/restrictions.py index 88e1d274..d30ebd3f 100644 --- a/datasette/default_permissions/restrictions.py +++ b/datasette/default_permissions/restrictions.py @@ -185,11 +185,15 @@ def restrictions_allow_action( # Check table/resource level if resource is not None and not isinstance(resource, str) and len(resource) == 2: database, table = resource - table_allowed = restrictions.get("r", {}).get(database, {}).get(table) - if table_allowed is not None: - assert isinstance(table_allowed, list) - if to_check.intersection(table_allowed): - return True + action_obj = datasette.actions.get(action) + normalize = action_obj.normalize_child if action_obj else lambda name: name + for table_name, table_allowed in ( + restrictions.get("r", {}).get(database, {}).items() + ): + if normalize(table_name) == normalize(table): + assert isinstance(table_allowed, list) + if to_check.intersection(table_allowed): + return True # This action is not explicitly allowed, so reject it return False diff --git a/datasette/permissions.py b/datasette/permissions.py index e03b065c..2d242560 100644 --- a/datasette/permissions.py +++ b/datasette/permissions.py @@ -3,6 +3,10 @@ from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Any, NamedTuple +_SQLITE_IDENTIFIER_CASE = str.maketrans( + "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz" +) + # Context variable to track when permission checks should be skipped _skip_permission_checks = contextvars.ContextVar( "skip_permission_checks", default=False @@ -49,6 +53,15 @@ class Resource(ABC): # Class-level metadata (subclasses must define these) name: str = None # e.g., "table", "database", "model" parent_class: type["Resource"] | None = None # e.g., DatabaseResource for tables + case_insensitive_child: bool = False + + @classmethod + def normalize_child(cls, child: str | None) -> str | None: + """Return a comparison key without changing the resource's display name.""" + if cls.case_insensitive_child and child is not None: + # Match SQLite NOCASE: fold ASCII only, not Unicode lower/casefold. + return child.translate(_SQLITE_IDENTIFIER_CASE) + return child # Instance-level optional extra attributes reasons: list[str] | None = None @@ -146,6 +159,11 @@ class Action: resource_class: type[Resource] | None = None also_requires: str | None = None # Optional action name that must also be allowed + def normalize_child(self, child: str | None) -> str | None: + if self.resource_class is None: + return child + return self.resource_class.normalize_child(child) + @property def takes_parent(self) -> bool: """ diff --git a/datasette/resources.py b/datasette/resources.py index ee2e6d98..29bf7b1e 100644 --- a/datasette/resources.py +++ b/datasette/resources.py @@ -25,6 +25,7 @@ class TableResource(Resource): name = "table" parent_class = DatabaseResource + case_insensitive_child = True def __init__(self, database: str, table: str): super().__init__(parent=database, child=table) diff --git a/datasette/utils/actions_sql.py b/datasette/utils/actions_sql.py index d767e391..297f5ae5 100644 --- a/datasette/utils/actions_sql.py +++ b/datasette/utils/actions_sql.py @@ -29,6 +29,15 @@ from datasette.utils.permissions import gather_permission_sql_from_hooks if TYPE_CHECKING: from datasette.app import Datasette + from datasette.permissions import Action + + +def _child_collation(action: "Action") -> str: + """Match resource identity without changing the spelling returned by SQL.""" + resource_class = action.resource_class + if resource_class is not None and resource_class.case_insensitive_child: + return "NOCASE" + return "BINARY" async def build_allowed_resources_sql( @@ -149,6 +158,7 @@ async def _build_single_action_sql( raise ValueError(f"Unknown action: {action}") # Get base resources SQL from the resource class + child_collation = _child_collation(action_obj) base_resources_sql = await action_obj.resource_class.resources_sql( datasette, actor=actor ) @@ -185,7 +195,7 @@ async def _build_single_action_sql( if permission_sql.sql is None: continue rule_sqls.append(f""" - SELECT parent, child, allow, reason, '{permission_sql.source}' AS source_plugin FROM ( + SELECT parent, child COLLATE {child_collation} AS child, allow, reason, '{permission_sql.source}' AS source_plugin FROM ( {permission_sql.sql} ) """.strip()) @@ -299,9 +309,9 @@ async def _build_single_action_sql( query_parts.extend( ["anon_child_agg AS ("] + _anon_agg( - "parent, child,", + f"parent, child COLLATE {child_collation} AS child,", "parent IS NOT NULL AND child IS NOT NULL", - "parent, child", + f"parent, child COLLATE {child_collation}", ) + ["),", "anon_parent_agg AS ("] + _anon_agg("parent,", "parent IS NOT NULL AND child IS NULL", "parent") @@ -382,7 +392,8 @@ async def _build_single_action_sql( # Wrap each restriction_sql in a subquery to avoid operator precedence issues # with UNION ALL inside the restriction SQL statements restriction_intersect = "\nINTERSECT\n".join( - f"SELECT * FROM ({sql})" for sql in restriction_sqls + f"SELECT parent, child COLLATE {child_collation} AS child FROM ({sql})" + for sql in restriction_sqls ) # Decompose by NULL-pattern so the final filter can use pure-equality # EXISTS lookups (satisfiable via automatic indexes) instead of a @@ -480,6 +491,7 @@ async def build_permission_rules_sql( union_parts = [] all_params = {} restriction_sqls = [] + child_collation = _child_collation(action_obj) for permission_sql in permission_sqls: all_params.update(permission_sql.params or {}) @@ -493,7 +505,7 @@ async def build_permission_rules_sql( continue union_parts.append(f""" - SELECT parent, child, allow, reason, '{permission_sql.source}' AS source_plugin FROM ( + SELECT parent, child COLLATE {child_collation} AS child, allow, reason, '{permission_sql.source}' AS source_plugin FROM ( {permission_sql.sql} ) """.strip()) @@ -564,6 +576,7 @@ async def check_permissions_for_actions( verdicts = {} for i, (action, permission_sqls) in enumerate(zip(unique_actions, gathered)): + child_collation = _child_collation(datasette.actions[action]) prefix = f"a{i}_" rule_parts = [] restriction_parts = [] @@ -589,7 +602,7 @@ async def check_permissions_for_actions( if sql is None: continue rule_parts.append( - f"SELECT parent, child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (\n{sql}\n)" + f"SELECT parent, child COLLATE {child_collation} AS child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (\n{sql}\n)" ) if not rule_parts: @@ -623,7 +636,8 @@ async def check_permissions_for_actions( if restriction_parts: # Database-level restrictions (parent, NULL) match all children restriction_intersect = "\nINTERSECT\n".join( - f"SELECT * FROM ({sql})" for sql in restriction_parts + f"SELECT parent, child COLLATE {child_collation} AS child FROM ({sql})" + for sql in restriction_parts ) ctes.append(f"a{i}_restriction AS (\n{restriction_intersect}\n)") verdict_sql = f"""({verdict_sql}) AND EXISTS ( @@ -770,6 +784,7 @@ async def _explain_single_action( db = datasette.get_internal_database() matched_rules = [] restrictions = [] + child_collation = _child_collation(datasette.actions[action]) for permission_sql in permission_sqls: params = dict(permission_sql.params or {}) @@ -784,7 +799,7 @@ async def _explain_single_action( SELECT parent, child, allow, reason FROM ({permission_sql.sql}) AS permission_rules WHERE (parent IS NULL OR parent = :{parent_param}) - AND (child IS NULL OR child = :{child_param}) + AND (child IS NULL OR child COLLATE {child_collation} = :{child_param}) """, params, ) @@ -811,7 +826,7 @@ async def _explain_single_action( SELECT EXISTS( SELECT 1 FROM ({permission_sql.restriction_sql}) AS restriction_rules WHERE (parent IS NULL OR parent = :{parent_param}) - AND (child IS NULL OR child = :{child_param}) + AND (child IS NULL OR child COLLATE {child_collation} = :{child_param}) ) AS resource_is_in_allowlist """, params, diff --git a/docs/authentication.rst b/docs/authentication.rst index 659c6989..c82cfa1d 100644 --- a/docs/authentication.rst +++ b/docs/authentication.rst @@ -158,6 +158,15 @@ Datasette resolves matching rules from most specific to least specific: This means a resource-level allow can provide an exception to a parent-level deny. It also means that two plugins which disagree at the same level resolve to deny. +For table and view permissions, resource names use SQLite's case-insensitive +identifier matching: ``Secret``, ``secret`` and ``SECRET`` identify the same +table. This applies to configuration rules, plugin rules and token restrictions. +Only ASCII letters are case-insensitive; non-ASCII characters remain distinct. +Conflicting rules for different spellings of the same name follow the usual +deny-wins rule at the same scope. Names retain their original spelling in +resource listings and permission explanations. Database names, stored query +names and other resource types remain case-sensitive. + .. list-table:: Permission rule examples :header-rows: 1 diff --git a/tests/test_table_resource_identity.py b/tests/test_table_resource_identity.py new file mode 100644 index 00000000..35b4dcc6 --- /dev/null +++ b/tests/test_table_resource_identity.py @@ -0,0 +1,420 @@ +"""Table permission identities must agree with SQLite identifier resolution.""" + +import uuid +from unittest.mock import AsyncMock + +import pytest + +from datasette import hookimpl +from datasette.app import Datasette +from datasette.default_permissions import restrictions_allow_action +from datasette.permissions import Action, PermissionSQL, _permission_check_cache +from datasette.resources import QueryResource, TableResource +from datasette.utils.actions_sql import explain_permission_for_resource +from datasette.utils.asgi import Forbidden +from datasette.utils.permissions import gather_permission_sql_from_hooks + + +@pytest.mark.asyncio +@pytest.mark.parametrize("kind", ["table", "view"]) +@pytest.mark.parametrize("spelling", ["Inventory", "inventory", "INVENTORY"]) +@pytest.mark.parametrize("allowed", [False, True]) +@pytest.mark.parametrize("rule_spelling", ["Inventory", "iNvEnToRy"]) +async def test_table_permission_identity( + kind, spelling, allowed, rule_spelling, monkeypatch +): + ds = Datasette( + config={ + "permissions": {"view-table": not allowed, "insert-row": not allowed}, + "databases": { + "data": { + "tables": { + rule_spelling: { + "permissions": { + "view-table": allowed, + "insert-row": allowed, + } + } + } + } + }, + } + ) + db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data") + cache_token = _permission_check_cache.set({}) + try: + await db.execute_write( + "create table Inventory (id integer primary key)" + if kind == "table" + else "create view Inventory as select 1 as id" + ) + await ds.invoke_startup() + # Identity matching needs no target-schema lookup. Derived-table + # permissions may still check the schema version. All spellings and + # API entry points should share the existing permission result cache. + target_execute = AsyncMock(wraps=db.execute) + monkeypatch.setattr(db, "execute", target_execute) + internal_execute = AsyncMock(wraps=ds.get_internal_database().execute) + monkeypatch.setattr(ds.get_internal_database(), "execute", internal_execute) + resource = TableResource("data", spelling) + assert await ds.allowed_many( + actions=["view-table", "insert-row"], resource=resource + ) == {"view-table": allowed, "insert-row": allowed} + assert await ds.allowed(action="view-table", resource=resource) is allowed + assert await ds.check_visibility(None, "view-table", resource) == ( + allowed, + False, + ) + if allowed: + await ds.ensure_permission(action="view-table", resource=resource) + else: + with pytest.raises(Forbidden): + await ds.ensure_permission(action="view-table", resource=resource) + assert resource.child == spelling # Do not mutate caller-owned resources. + for variant in ("Inventory", "inventory", "INVENTORY"): + assert ( + await ds.allowed( + action="view-table", resource=TableResource("data", variant) + ) + is allowed + ) + assert internal_execute.await_count == 1 + assert all( + call.args[0] == "PRAGMA schema_version" + for call in target_execute.await_args_list + ) + assert all(key[3] == "inventory" for key in _permission_check_cache.get()) + finally: + _permission_check_cache.reset(cache_token) + ds.close() + + +@pytest.mark.asyncio +async def test_other_permission_identities_are_preserved(): + ds = Datasette( + config={ + "databases": { + "data": { + "tables": { + "Äpfel": {"permissions": {"view-table": False}}, + "Future": {"permissions": {"view-table": False}}, + }, + "queries": { + "Report": { + "sql": "select 1", + "permissions": {"view-query": False}, + }, + "report": { + "sql": "select 1", + "permissions": {"view-query": True}, + }, + }, + } + } + } + ) + db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data") + try: + await db.execute_write('create table "Äpfel" (id integer primary key)') + await db.execute_write('create table "äpfel" (id integer primary key)') + await db.execute_write("create table Report (id integer primary key)") + await ds.invoke_startup() + # SQLite folds ASCII identifier casing, not Unicode casing. + for name, expected in [ + ("ÄPFEL", False), + ("äPFEL", True), + ("Future", False), + ("future", False), + ]: + assert ( + await ds.allowed( + action="view-table", resource=TableResource("data", name) + ) + is expected + ) + # Query names remain case-sensitive even when a table has the same name. + for name, expected in [("Report", False), ("report", True)]: + assert ( + await ds.allowed( + action="view-query", resource=QueryResource("data", name) + ) + is expected + ) + finally: + ds.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("allow", [True, False, {"id": "reader"}]) +async def test_table_listings_and_explanations(allow): + ds = Datasette( + config={ + "databases": { + "data": { + "tables": { + "inventory": {"permissions": {"view-table": allow}}, + } + } + } + } + ) + db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data") + try: + await db.execute_write("create table Inventory (id integer primary key)") + await db.execute_write("create view InventoryView as select id from Inventory") + await ds.invoke_startup() + for actor in (None, {"id": "reader"}): + expected = allow is True or (isinstance(allow, dict) and actor == allow) + explanation = await explain_permission_for_resource( + datasette=ds, + actor=actor, + action="view-table", + parent="data", + child="INVENTORY", + ) + assert explanation["allowed"] is expected + assert explanation["winning_scope"] == "resource" + assert any( + "data/inventory" in rule["reason"] + for rule in explanation["matched_rules"] + ) + page = await ds.allowed_resources( + "view-table", + actor, + parent="data", + include_is_private=True, + include_reasons=True, + limit=1, + ) + resources = [resource async for resource in page.all()] + matching = [r for r in resources if r.child == "Inventory"] + assert bool(matching) is expected + assert len(matching) <= 1 + if matching: + assert matching[0].private is isinstance(allow, dict) + assert any(r.child == "InventoryView" for r in resources) + finally: + ds.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("deny_first", [True, False]) +async def test_case_variant_rules_deny_wins(deny_first): + rules = [("inventory", False), ("INVENTORY", True)] + if not deny_first: + rules.reverse() + ds = Datasette( + config={ + "databases": { + "data": { + "tables": { + name: {"permissions": {"view-table": allow}} + for name, allow in rules + } + } + } + } + ) + db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data") + try: + await db.execute_write("create table Inventory (id integer primary key)") + await ds.invoke_startup() + assert not await ds.allowed( + action="view-table", resource=TableResource("data", "Inventory") + ) + assert not ( + await ds.allowed_resources( + "view-table", parent="data", include_is_private=True + ) + ).resources + explanation = await explain_permission_for_resource( + datasette=ds, + actor=None, + action="view-table", + parent="data", + child="Inventory", + ) + assert not explanation["allowed"] + assert any( + rule["effect"] == "allow" and not rule["decisive"] + for rule in explanation["matched_rules"] + ) + assert any( + rule["effect"] == "deny" and rule["decisive"] + for rule in explanation["matched_rules"] + ) + finally: + ds.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("config_style", ["allow", "permissions"]) +@pytest.mark.parametrize("allowed", [True, False]) +async def test_case_variant_token_restrictions(config_style, allowed): + table_config = ( + {"allow": allowed} + if config_style == "allow" + else {"permissions": {"view-table": allowed}} + ) + ds = Datasette( + config={"databases": {"data": {"tables": {"Inventory": table_config}}}} + ) + db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data") + actor = {"id": "reader", "_r": {"r": {"data": {"inventory": ["vt"]}}}} + try: + await db.execute_write("create table Inventory (id integer primary key)") + await ds.invoke_startup() + assert restrictions_allow_action( + ds, actor["_r"], "view-table", ("data", "INVENTORY") + ) + assert not restrictions_allow_action( + ds, actor["_r"], "view-table", ("Data", "Inventory") + ) + assert ( + await ds.allowed( + action="view-table", + resource=TableResource("data", "INVENTORY"), + actor=actor, + ) + is allowed + ) + page = await ds.allowed_resources("view-table", actor, parent="data") + assert [(r.parent, r.child) for r in page.resources] == ( + [("data", "Inventory")] if allowed else [] + ) + explanation = await explain_permission_for_resource( + datasette=ds, + actor=actor, + action="view-table", + parent="data", + child="Inventory", + ) + assert explanation["restriction_allowed"] + assert explanation["allowed"] is allowed + finally: + ds.close() + + +@pytest.mark.asyncio +async def test_plugin_restriction_intersection_and_dependencies(): + class Plugin: + @hookimpl + def register_actions(self, datasette): + return [ + Action( + name="inspect-inventory", + description="Inspect inventory", + resource_class=TableResource, + also_requires="view-table", + ) + ] + + @hookimpl + def permission_resources_sql(self, action): + if action not in ("view-table", "inspect-inventory"): + return None + return [ + PermissionSQL( + sql="SELECT 'data' AS parent, 'INVENTORY' AS child, 1 AS allow, 'inventory grant' AS reason", + restriction_sql="SELECT 'data' AS parent, 'inventory' AS child", + ), + PermissionSQL( + restriction_sql="SELECT 'data' AS parent, 'InVeNtOrY' AS child" + ), + ] + + ds = Datasette(default_deny=True) + ds.pm.register(Plugin(), name="identity-test") + db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data") + try: + await db.execute_write("create table Inventory (id integer primary key)") + await db.execute_write("create table Other (id integer primary key)") + await ds.invoke_startup() + for action in ("view-table", "inspect-inventory"): + assert await ds.allowed( + action=action, resource=TableResource("data", "Inventory") + ) + assert not await ds.allowed( + action=action, resource=TableResource("data", "Other") + ) + resources = ( + await ds.allowed_resources( + action, parent="data", include_is_private=True + ) + ).resources + assert [r.child for r in resources] == ["Inventory"] + explanation = await explain_permission_for_resource( + datasette=ds, + actor=None, + action=action, + parent="data", + child="Inventory", + ) + assert explanation["allowed"] + assert all(item["allowed"] for item in explanation["restrictions"]) + finally: + ds.pm.unregister(name="identity-test") + ds.close() + + +@pytest.mark.asyncio +async def test_shared_plugin_rule_keeps_query_identity_and_original_sql(): + shared = PermissionSQL( + sql="SELECT 'data' AS parent, 'Inventory' AS child, 0 AS allow, 'shared deny' AS reason" + ) + original_sql = shared.sql + + class Plugin: + @hookimpl + def permission_resources_sql(self, action): + if action in ("view-table", "view-query"): + return shared + + ds = Datasette( + config={ + "databases": { + "data": { + "queries": { + "Inventory": "select 1", + "inventory": "select 1", + } + } + } + } + ) + ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data") + ds.pm.register(Plugin(), name="identity-test") + try: + await ds.invoke_startup() + for _ in range(2): + await gather_permission_sql_from_hooks( + datasette=ds, actor=None, action="view-table" + ) + assert shared.sql == original_sql + assert not await ds.allowed( + action="view-table", resource=TableResource("data", "inventory") + ) + assert await ds.allowed( + action="view-query", resource=QueryResource("data", "inventory") + ) + assert not await ds.allowed( + action="view-query", resource=QueryResource("data", "Inventory") + ) + assert await ds.allowed( + action="view-table", resource=TableResource("Data", "Inventory") + ) + assert restrictions_allow_action( + ds, + {"r": {"data": {"Inventory": ["vq"]}}}, + "view-query", + ("data", "Inventory"), + ) + assert not restrictions_allow_action( + ds, + {"r": {"data": {"Inventory": ["vq"]}}}, + "view-query", + ("data", "inventory"), + ) + finally: + ds.pm.unregister(name="identity-test") + ds.close()