mirror of
https://github.com/simonw/datasette.git
synced 2026-09-24 19:04:08 +02:00
Deny SQLite statistics table access through a default hook
This commit is contained in:
parent
628cec8f0c
commit
d334539a1e
5 changed files with 152 additions and 4 deletions
|
|
@ -894,10 +894,7 @@ async def test_hidden_sqlite_stat1_table():
|
|||
await db.execute_write("analyze")
|
||||
data = (await ds.client.get("/db.json?_show_hidden=1")).json()
|
||||
tables = [(t["name"], t["hidden"]) for t in data["tables"]]
|
||||
assert tables in (
|
||||
[("normal", False), ("sqlite_stat1", True)],
|
||||
[("normal", False), ("sqlite_stat1", True), ("sqlite_stat4", True)],
|
||||
)
|
||||
assert tables == [("normal", False)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
113
tests/test_pr76_statistics_policy.py
Normal file
113
tests/test_pr76_statistics_policy.py
Normal file
|
|
@ -0,0 +1,113 @@
|
|||
"""Statistics access policy and plugin replacement coverage for PR #76."""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from datasette import hookimpl
|
||||
from datasette.app import Datasette
|
||||
from datasette.permissions import PermissionSQL
|
||||
from datasette.resources import TableResource
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("scope", [None, "global", "database", "table", "root"])
|
||||
async def test_statistics_denied_despite_allow_rules(scope):
|
||||
config = {"databases": {"data": {"tables": {"sqlite_stat1": {}}}}}
|
||||
grant = {"view-table": True}
|
||||
if scope == "global":
|
||||
config["permissions"] = grant
|
||||
elif scope == "database":
|
||||
config["databases"]["data"]["permissions"] = grant
|
||||
elif scope == "table":
|
||||
config["databases"]["data"]["tables"]["sqlite_stat1"]["permissions"] = grant
|
||||
ds = Datasette(memory=True, config=config)
|
||||
ds.root_enabled = scope == "root"
|
||||
actor = {"id": "root"} if scope == "root" else {"id": "reader"}
|
||||
db = ds.add_memory_database(uuid.uuid4().hex, name="data")
|
||||
await db.execute_write("create table items(value text)")
|
||||
await db.execute_write("create index items_value on items(value)")
|
||||
await db.execute_write("insert into items values ('example')")
|
||||
await db.execute_write("analyze")
|
||||
await ds.invoke_startup()
|
||||
try:
|
||||
assert "view-sqlite-statistics" not in ds.actions
|
||||
for name in ("sqlite_stat1", "SQLITE_STAT1"):
|
||||
assert not await ds.allowed(
|
||||
action="view-table", resource=TableResource("data", name), actor=actor
|
||||
)
|
||||
for suffix in ("", ".json", ".csv"):
|
||||
assert (
|
||||
await ds.client.get(f"/data/sqlite_stat1{suffix}", actor=actor)
|
||||
).status_code == 403
|
||||
resources = await ds.allowed_resources("view-table", parent="data", actor=actor)
|
||||
assert "sqlite_stat1" not in {r.child for r in resources.resources}
|
||||
assert "items" in {r.child for r in resources.resources}
|
||||
finally:
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"table", ["sqlite_stat1", "sqlite_stat2", "sqlite_stat3", "sqlite_stat4"]
|
||||
)
|
||||
@pytest.mark.parametrize("default_deny", [False, True])
|
||||
async def test_statistics_names_denied(table, default_deny):
|
||||
ds = Datasette(memory=True, default_deny=default_deny)
|
||||
ds.root_enabled = True
|
||||
await ds.invoke_startup()
|
||||
try:
|
||||
for name in (table, table.upper()):
|
||||
assert not await ds.allowed(
|
||||
action="view-table",
|
||||
resource=TableResource("_memory", name),
|
||||
actor={"id": "root"},
|
||||
)
|
||||
finally:
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_plugin_can_replace_statistics_policy():
|
||||
class ReplacementPolicy:
|
||||
@hookimpl
|
||||
def permission_resources_sql(self, action, actor):
|
||||
if action == "view-table":
|
||||
return PermissionSQL(
|
||||
sql="SELECT 'data' AS parent, 'sqlite_stat1' AS child, :statistics_allowed AS allow, 'custom statistics policy' AS reason",
|
||||
params={"statistics_allowed": int(actor == {"id": "reader"})},
|
||||
)
|
||||
|
||||
ds = Datasette(memory=True)
|
||||
db = ds.add_memory_database(uuid.uuid4().hex, name="data")
|
||||
await db.execute_write("create table items(value text)")
|
||||
await db.execute_write("analyze")
|
||||
await ds.invoke_startup()
|
||||
name = "datasette.default_permissions.sqlite_statistics"
|
||||
original = ds.pm.unregister(name=name)
|
||||
assert original is not None
|
||||
replacement = ReplacementPolicy()
|
||||
ds.pm.register(replacement, name="test-replacement-statistics-policy")
|
||||
try:
|
||||
actor = {"id": "reader"}
|
||||
assert await ds.allowed(
|
||||
action="view-table",
|
||||
resource=TableResource("data", "sqlite_stat1"),
|
||||
actor=actor,
|
||||
)
|
||||
assert not await ds.allowed(
|
||||
action="view-table", resource=TableResource("data", "sqlite_stat1")
|
||||
)
|
||||
resources = await ds.allowed_resources(
|
||||
"view-table", parent="data", actor=actor, include_is_private=True
|
||||
)
|
||||
stats = [r for r in resources.resources if r.child == "sqlite_stat1"]
|
||||
assert len(stats) == 1 and stats[0].private
|
||||
assert (
|
||||
await ds.client.get("/data/sqlite_stat1.json", actor=actor)
|
||||
).status_code == 200
|
||||
assert (await ds.client.get("/data/sqlite_stat1.json")).status_code == 403
|
||||
finally:
|
||||
ds.pm.unregister(replacement)
|
||||
ds.pm.register(original, name=name)
|
||||
ds.close()
|
||||
Loading…
Add table
Add a link
Reference in a new issue