Match table permission identities using SQLite case semantics

This commit is contained in:
Simon Willison 2026-09-09 07:54:48 -07:00
commit 506c4bb522
8 changed files with 497 additions and 19 deletions

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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:
"""

View file

@ -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)

View file

@ -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,

View file

@ -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

View file

@ -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()