remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-09-28 21:04:08 +02:00
Code Issues Projects Releases Packages Wiki Activity

Limit derived-table permissions to one source hop

Browse source 
Simplify the solution to 5de0c1724e - avoid contextvar.
This commit is contained in:
Simon Willison 2026-09-10 15:46:12 -07:00
commit 92c7d4b608
9 changed files with 235 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

7
tests/test_api.py
View file

@ -102,14 +102,11 @@ async def test_database_page(ds_client):
"tags",
}
# Expected hidden tables
# The external-content index is visible, but its shadow tables need a
# second dependency hop and are excluded by the one-hop permission policy.
expected_hidden_tables = {
"no_primary_key",
"searchable_fts",
"searchable_fts_config",
"searchable_fts_data",
"searchable_fts_docsize",
"searchable_fts_idx",
}
# Verify all expected tables exist

121
tests/test_fts_permissions.py
View file

@ -1,7 +1,121 @@
import pytest
from datasette import hookimpl
from datasette.app import Datasette
from datasette.permissions import Action, PermissionSQL, _permission_check_cache
from datasette.resources import DatabaseResource, TableResource
from datasette.utils.sqlite import sqlite3, sqlite_derived_table_dependencies
@pytest.mark.asyncio
@pytest.mark.parametrize("fts_module", ["fts4", "fts5"])
@pytest.mark.parametrize("actor", [None, {"id": "root"}], ids=["anonymous", "root"])
async def test_derived_permissions_allow_one_hop_but_deny_nested_sources(
fts_module, actor
):
class InspectPlugin:
@hookimpl
def register_actions(self):
return [
Action(
name="inspect-derived",
description="Inspect a table",
resource_class=TableResource,
also_requires="view-table",
)
]
@hookimpl
def permission_resources_sql(self, action):
if action == "inspect-derived":
return PermissionSQL(
sql="SELECT NULL AS parent, NULL AS child, 1 AS allow, 'inspect allowed' AS reason"
)
ds = Datasette(memory=True)
ds.pm.register(InspectPlugin(), name="inspect-derived-test")
db = ds.add_memory_database(
f"derived_one_hop_{fts_module}_{actor is not None}", name="data"
)
await db.execute_write("create table Documents (body text)")
await db.execute_write(
f"create virtual table Search using {fts_module}(body, content='Documents')"
)
await db.execute_write(
f"create virtual table Nested using {fts_module}(body, content='sEaRcH')"
)
await ds.invoke_startup()
token = _permission_check_cache.set({})
try:
# Both direct permissions are allowed, but a derived source makes its
# dependent unavailable even to an actor who can view the whole chain.
# Check and cache Search first so its cached grant cannot grant Nested.
for table, expected in (
("Documents", True),
("Search", True),
("Nested", False),
("Search_docsize", False),
):
for spelling in (table, table.upper(), table.lower()):
assert await ds.allowed_many(
actions=["view-table", "inspect-derived"],
resource=TableResource("data", spelling),
actor=actor,
) == {"view-table": expected, "inspect-derived": expected}
page = await ds.allowed_resources(
"view-table", actor, parent="data", include_is_private=True, limit=1000
)
allowed = {resource.child for resource in page.resources}
assert {"Documents", "Search"}.issubset(allowed)
assert "Nested" not in allowed
assert "Search_docsize" not in allowed
finally:
_permission_check_cache.reset(token)
ds.pm.unregister(name="inspect-derived-test")
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("listing", [False, True], ids=["individual", "listing"])
async def test_derived_permission_discovery_error_is_retried(monkeypatch, listing):
ds = Datasette(memory=True)
db = ds.add_memory_database(f"derived_discovery_error_{listing}", name="data")
await db.execute_write("create table documents (id integer primary key)")
await ds.invoke_startup()
class UnavailableSchema:
def execute(self, sql):
raise sqlite3.DatabaseError("schema temporarily unavailable")
async def check():
if listing:
return await ds.allowed_resources("view-table", parent="data")
return await ds.allowed(
action="view-table", resource=TableResource("data", "documents")
)
token = _permission_check_cache.set({})
try:
with monkeypatch.context() as patch:
patch.setattr(
"datasette.database.sqlite_derived_table_dependencies",
lambda conn: sqlite_derived_table_dependencies(UnavailableSchema()),
)
with pytest.raises(sqlite3.DatabaseError, match="schema temporarily"):
await check()
# Failed discovery must not cache an empty map or a permission grant.
assert db._cached_derived_table_dependencies is None
assert not _permission_check_cache.get()
result = await check()
if listing:
assert [resource.child for resource in result.resources] == ["documents"]
else:
assert result is True
assert db._cached_derived_table_dependencies is not None
finally:
_permission_check_cache.reset(token)
@pytest.mark.asyncio
@ -243,8 +357,11 @@ async def test_derived_tables_propagate_private_flag_and_route_permissions():
resource.child: resource for resource in actor_page.resources
}
derived_names = set(await db.derived_table_dependencies())
assert derived_names.issubset(actor_resources)
assert all(actor_resources[name].private for name in derived_names)
assert "secret_fts" in actor_resources
assert actor_resources["secret_fts"].private
# Shadow tables depend on the already-derived external-content FTS
# table, so they remain unavailable even to the permitted reader.
assert not (derived_names - {"secret_fts"}).intersection(actor_resources)
anonymous_page = await ds.allowed_resources(
"view-table", parent="data", limit=1000

4
tests/test_html.py
View file

@ -36,8 +36,10 @@ def test_homepage(app_client_two_attached_databases):
h2 = soup.select("h2")[0]
assert "extra database" == h2.text.strip()
counts_p, links_p = h2.find_all_next("p")[:2]
# Shadow tables of the external-content index are denied, so they do not
# contribute to the table or row totals.
assert (
"2 rows in 1 table, 5 rows in 4 hidden tables, 1 view" == counts_p.text.strip()
"2 rows in 1 table, 2 rows in 1 hidden table, 1 view" == counts_p.text.strip()
)
# We should only show visible, not hidden tables here:
table_links = [

31
tests/test_internals_database.py
View file

@ -20,7 +20,11 @@ from datasette.database import (
_deliver_write_result,
)
from datasette.utils import Column
from datasette.utils.sqlite import sqlite3, supports_returning
from datasette.utils.sqlite import (
sqlite3,
sqlite_derived_table_dependencies,
supports_returning,
)
requires_sqlite_returning = pytest.mark.skipif(
not supports_returning(), reason="SQLite does not support RETURNING"
@ -39,6 +43,31 @@ async def test_execute(db):
assert 15 == len(results)
@pytest.mark.asyncio
async def test_derived_dependency_cache_survives_failed_refresh(monkeypatch):
ds = Datasette(memory=True)
db = ds.add_memory_database(uuid.uuid4().hex, name="data")
await db.derived_table_dependencies()
previous_cache = db._cached_derived_table_dependencies
await db.execute_write("create table dependency_cache_refresh (id integer)")
class UnavailableSchema:
def execute(self, sql):
raise sqlite3.DatabaseError("schema temporarily unavailable")
with monkeypatch.context() as patch:
patch.setattr(
"datasette.database.sqlite_derived_table_dependencies",
lambda conn: sqlite_derived_table_dependencies(UnavailableSchema()),
)
with pytest.raises(sqlite3.DatabaseError, match="schema temporarily"):
await db.derived_table_dependencies()
assert db._cached_derived_table_dependencies == previous_cache
await db.derived_table_dependencies()
assert db._cached_derived_table_dependencies[0] != previous_cache[0]
@pytest.mark.asyncio
async def test_results_first(db):
assert None is (await db.execute("select * from facetable where pk > 100")).first()

16
tests/test_pr76_fts_policy.py
View file

@ -35,17 +35,20 @@ def test_vocabulary_dependency_identity(module, arguments, vocab_name):
@pytest.mark.asyncio
@pytest.mark.parametrize("module", ["fts5", "fts4"])
@pytest.mark.parametrize("external_content", [False, True], ids=["one-hop", "two-hop"])
@pytest.mark.parametrize(
"source_allowed,vocab_allowed", [(False, True), (True, False), (True, True)]
)
async def test_vocabulary_transitive_permissions(module, source_allowed, vocab_allowed):
async def test_vocabulary_immediate_source_permissions(
module, external_content, source_allowed, vocab_allowed
):
ds = Datasette(
memory=True,
config={
"databases": {
"data": {
"tables": {
"documents": {
"search": {
"permissions": {
"view-table": (
{"id": "reader"} if source_allowed else False
@ -60,9 +63,8 @@ async def test_vocabulary_transitive_permissions(module, source_allowed, vocab_a
)
db = ds.add_memory_database(uuid.uuid4().hex, name="data")
await db.execute_write("create table documents(body text)")
await db.execute_write(
f"create virtual table search using {module}(body, content='documents')"
)
options = "body, content='documents'" if external_content else "body"
await db.execute_write(f"create virtual table search using {module}({options})")
definition = (
"fts5vocab('SEARCH', 'row')" if module == "fts5" else "fts4aux('SEARCH')"
)
@ -70,7 +72,7 @@ async def test_vocabulary_transitive_permissions(module, source_allowed, vocab_a
await ds.invoke_startup()
try:
actor = {"id": "reader"}
expected = source_allowed and vocab_allowed
expected = source_allowed and vocab_allowed and not external_content
for name in ("words", "WORDS"):
assert (
await ds.allowed(
@ -112,7 +114,7 @@ def test_cross_schema_vocabulary_is_unresolved(module, definition):
conn.execute(f"create virtual table search using {module}(body)")
conn.execute(f"create virtual table temp.words using {definition}")
# Cross-schema ownership is not representable by the current map.
# The self-dependency invokes the permission layer's cycle denial.
# The source is itself derived, so the immediate-source policy denies it.
assert (
sqlite_derived_table_dependencies(conn, schema="temp")["words"] == "words"
)