mirror of
https://github.com/simonw/datasette.git
synced 2025-12-10 16:51:24 +01:00
New allowed_resources_sql plugin hook and debug tools (#2505)
* allowed_resources_sql plugin hook and infrastructure * New methods for checking permissions with the new system * New /-/allowed and /-/check and /-/rules special endpoints Still needs to be integrated more deeply into Datasette, especially for listing visible tables. Refs: #2502 --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Simon Willison
GitHub
parent
85da8474d4
commit
27084caa04
20 changed files with 3381 additions and 27 deletions
118
tests/test_config_permission_rules.py
Normal file
118
tests/test_config_permission_rules.py
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
import pytest
|
||||
|
||||
from datasette.app import Datasette
|
||||
from datasette.database import Database
|
||||
|
||||
|
||||
async def setup_datasette(config=None, databases=None):
|
||||
ds = Datasette(memory=True, config=config)
|
||||
for name in databases or []:
|
||||
ds.add_database(Database(ds, memory_name=f"{name}_memory"), name=name)
|
||||
await ds.invoke_startup()
|
||||
await ds.refresh_schemas()
|
||||
return ds
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_root_permissions_allow():
|
||||
config = {"permissions": {"execute-sql": {"id": "alice"}}}
|
||||
ds = await setup_datasette(config=config, databases=["content"])
|
||||
|
||||
assert await ds.permission_allowed_2({"id": "alice"}, "execute-sql", "content")
|
||||
assert not await ds.permission_allowed_2({"id": "bob"}, "execute-sql", "content")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_database_permission():
|
||||
config = {
|
||||
"databases": {
|
||||
"content": {
|
||||
"permissions": {
|
||||
"insert-row": {"id": "alice"},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
ds = await setup_datasette(config=config, databases=["content"])
|
||||
|
||||
assert await ds.permission_allowed_2(
|
||||
{"id": "alice"}, "insert-row", ("content", "repos")
|
||||
)
|
||||
assert not await ds.permission_allowed_2(
|
||||
{"id": "bob"}, "insert-row", ("content", "repos")
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_table_permission():
|
||||
config = {
|
||||
"databases": {
|
||||
"content": {
|
||||
"tables": {"repos": {"permissions": {"delete-row": {"id": "alice"}}}}
|
||||
}
|
||||
}
|
||||
}
|
||||
ds = await setup_datasette(config=config, databases=["content"])
|
||||
|
||||
assert await ds.permission_allowed_2(
|
||||
{"id": "alice"}, "delete-row", ("content", "repos")
|
||||
)
|
||||
assert not await ds.permission_allowed_2(
|
||||
{"id": "bob"}, "delete-row", ("content", "repos")
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_view_table_allow_block():
|
||||
config = {
|
||||
"databases": {"content": {"tables": {"repos": {"allow": {"id": "alice"}}}}}
|
||||
}
|
||||
ds = await setup_datasette(config=config, databases=["content"])
|
||||
|
||||
assert await ds.permission_allowed_2(
|
||||
{"id": "alice"}, "view-table", ("content", "repos")
|
||||
)
|
||||
assert not await ds.permission_allowed_2(
|
||||
{"id": "bob"}, "view-table", ("content", "repos")
|
||||
)
|
||||
assert await ds.permission_allowed_2(
|
||||
{"id": "bob"}, "view-table", ("content", "other")
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_view_table_allow_false_blocks():
|
||||
config = {"databases": {"content": {"tables": {"repos": {"allow": False}}}}}
|
||||
ds = await setup_datasette(config=config, databases=["content"])
|
||||
|
||||
assert not await ds.permission_allowed_2(
|
||||
{"id": "alice"}, "view-table", ("content", "repos")
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_allow_sql_blocks():
|
||||
config = {"allow_sql": {"id": "alice"}}
|
||||
ds = await setup_datasette(config=config, databases=["content"])
|
||||
|
||||
assert await ds.permission_allowed_2({"id": "alice"}, "execute-sql", "content")
|
||||
assert not await ds.permission_allowed_2({"id": "bob"}, "execute-sql", "content")
|
||||
|
||||
config = {"databases": {"content": {"allow_sql": {"id": "bob"}}}}
|
||||
ds = await setup_datasette(config=config, databases=["content"])
|
||||
|
||||
assert await ds.permission_allowed_2({"id": "bob"}, "execute-sql", "content")
|
||||
assert not await ds.permission_allowed_2({"id": "alice"}, "execute-sql", "content")
|
||||
|
||||
config = {"allow_sql": False}
|
||||
ds = await setup_datasette(config=config, databases=["content"])
|
||||
assert not await ds.permission_allowed_2({"id": "alice"}, "execute-sql", "content")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_view_instance_allow_block():
|
||||
config = {"allow": {"id": "alice"}}
|
||||
ds = await setup_datasette(config=config)
|
||||
|
||||
assert await ds.permission_allowed_2({"id": "alice"}, "view-instance")
|
||||
assert not await ds.permission_allowed_2({"id": "bob"}, "view-instance")
|
||||
495
tests/test_permission_endpoints.py
Normal file
495
tests/test_permission_endpoints.py
Normal file
|
|
@ -0,0 +1,495 @@
|
|||
"""
|
||||
Tests for permission inspection endpoints:
|
||||
- /-/check.json
|
||||
- /-/allowed.json
|
||||
- /-/rules.json
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from datasette.app import Datasette
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def ds_with_permissions():
|
||||
"""Create a Datasette instance with some permission rules configured."""
|
||||
ds = Datasette(
|
||||
config={
|
||||
"databases": {
|
||||
"content": {
|
||||
"allow": {"id": "*"}, # Allow all authenticated users
|
||||
"tables": {
|
||||
"articles": {
|
||||
"allow": {"id": "editor"}, # Only editor can view
|
||||
}
|
||||
},
|
||||
},
|
||||
"private": {
|
||||
"allow": False, # Deny everyone
|
||||
},
|
||||
}
|
||||
}
|
||||
)
|
||||
await ds.invoke_startup()
|
||||
# Add some test databases
|
||||
ds.add_memory_database("content")
|
||||
ds.add_memory_database("private")
|
||||
return ds
|
||||
|
||||
|
||||
# /-/check.json tests
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"path,expected_status,expected_keys",
|
||||
[
|
||||
# Valid request
|
||||
(
|
||||
"/-/check.json?action=view-instance",
|
||||
200,
|
||||
{"action", "allowed", "resource"},
|
||||
),
|
||||
# Missing action parameter
|
||||
("/-/check.json", 400, {"error"}),
|
||||
# Invalid action
|
||||
("/-/check.json?action=nonexistent", 404, {"error"}),
|
||||
# With parent parameter
|
||||
(
|
||||
"/-/check.json?action=view-database&parent=content",
|
||||
200,
|
||||
{"action", "allowed", "resource"},
|
||||
),
|
||||
# With parent and child parameters
|
||||
(
|
||||
"/-/check.json?action=view-table&parent=content&child=articles",
|
||||
200,
|
||||
{"action", "allowed", "resource"},
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_check_json_basic(
|
||||
ds_with_permissions, path, expected_status, expected_keys
|
||||
):
|
||||
response = await ds_with_permissions.client.get(path)
|
||||
assert response.status_code == expected_status
|
||||
data = response.json()
|
||||
assert expected_keys.issubset(data.keys())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_json_response_structure(ds_with_permissions):
|
||||
"""Test that /-/check.json returns the expected structure."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/check.json?action=view-instance"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# Check required fields
|
||||
assert "action" in data
|
||||
assert "allowed" in data
|
||||
assert "resource" in data
|
||||
|
||||
# Check resource structure
|
||||
assert "parent" in data["resource"]
|
||||
assert "child" in data["resource"]
|
||||
assert "path" in data["resource"]
|
||||
|
||||
# Check allowed is boolean
|
||||
assert isinstance(data["allowed"], bool)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_json_redacts_sensitive_fields_without_debug_permission(
|
||||
ds_with_permissions,
|
||||
):
|
||||
"""Test that /-/check.json redacts reason and source_plugin without permissions-debug."""
|
||||
# Anonymous user should not see sensitive fields
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/check.json?action=view-instance"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
# Sensitive fields should not be present
|
||||
assert "reason" not in data
|
||||
assert "source_plugin" not in data
|
||||
# But these non-sensitive fields should be present
|
||||
assert "used_default" in data
|
||||
assert "depth" in data
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_json_shows_sensitive_fields_with_debug_permission(
|
||||
ds_with_permissions,
|
||||
):
|
||||
"""Test that /-/check.json shows reason and source_plugin with permissions-debug."""
|
||||
# User with permissions-debug should see sensitive fields
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/check.json?action=view-instance",
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
# Sensitive fields should be present
|
||||
assert "reason" in data
|
||||
assert "source_plugin" in data
|
||||
assert "used_default" in data
|
||||
assert "depth" in data
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_json_child_requires_parent(ds_with_permissions):
|
||||
"""Test that child parameter requires parent parameter."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/check.json?action=view-table&child=articles"
|
||||
)
|
||||
assert response.status_code == 400
|
||||
data = response.json()
|
||||
assert "error" in data
|
||||
assert "parent" in data["error"].lower()
|
||||
|
||||
|
||||
# /-/allowed.json tests
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"path,expected_status,expected_keys",
|
||||
[
|
||||
# Valid supported actions
|
||||
(
|
||||
"/-/allowed.json?action=view-instance",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
(
|
||||
"/-/allowed.json?action=view-database",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
(
|
||||
"/-/allowed.json?action=view-table",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
(
|
||||
"/-/allowed.json?action=execute-sql",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
# Missing action parameter
|
||||
("/-/allowed.json", 400, {"error"}),
|
||||
# Invalid action
|
||||
("/-/allowed.json?action=nonexistent", 404, {"error"}),
|
||||
# Unsupported action (valid but not in CANDIDATE_SQL)
|
||||
("/-/allowed.json?action=insert-row", 400, {"error"}),
|
||||
],
|
||||
)
|
||||
async def test_allowed_json_basic(
|
||||
ds_with_permissions, path, expected_status, expected_keys
|
||||
):
|
||||
response = await ds_with_permissions.client.get(path)
|
||||
assert response.status_code == expected_status
|
||||
data = response.json()
|
||||
assert expected_keys.issubset(data.keys())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_allowed_json_response_structure(ds_with_permissions):
|
||||
"""Test that /-/allowed.json returns the expected structure."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/allowed.json?action=view-instance"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# Check required fields
|
||||
assert "action" in data
|
||||
assert "actor_id" in data
|
||||
assert "page" in data
|
||||
assert "page_size" in data
|
||||
assert "total" in data
|
||||
assert "items" in data
|
||||
|
||||
# Check items structure
|
||||
assert isinstance(data["items"], list)
|
||||
if data["items"]:
|
||||
item = data["items"][0]
|
||||
assert "parent" in item
|
||||
assert "child" in item
|
||||
assert "resource" in item
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_allowed_json_redacts_sensitive_fields_without_debug_permission(
|
||||
ds_with_permissions,
|
||||
):
|
||||
"""Test that /-/allowed.json redacts reason and source_plugin without permissions-debug."""
|
||||
# Anonymous user should not see sensitive fields
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/allowed.json?action=view-instance"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
if data["items"]:
|
||||
item = data["items"][0]
|
||||
assert "reason" not in item
|
||||
assert "source_plugin" not in item
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_allowed_json_shows_sensitive_fields_with_debug_permission(
|
||||
ds_with_permissions,
|
||||
):
|
||||
"""Test that /-/allowed.json shows reason and source_plugin with permissions-debug."""
|
||||
# User with permissions-debug should see sensitive fields
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/allowed.json?action=view-instance",
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
if data["items"]:
|
||||
item = data["items"][0]
|
||||
assert "reason" in item
|
||||
assert "source_plugin" in item
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_allowed_json_only_shows_allowed_resources(ds_with_permissions):
|
||||
"""Test that /-/allowed.json only shows resources with allow=1."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/allowed.json?action=view-instance"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# All items should have allow implicitly set to 1 (not in response but verified by the endpoint logic)
|
||||
# The endpoint filters to only show allowed resources
|
||||
assert isinstance(data["items"], list)
|
||||
assert data["total"] >= 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"page,page_size",
|
||||
[
|
||||
(1, 10),
|
||||
(2, 50),
|
||||
(1, 200), # max page size
|
||||
],
|
||||
)
|
||||
async def test_allowed_json_pagination(ds_with_permissions, page, page_size):
|
||||
"""Test pagination parameters."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
f"/-/allowed.json?action=view-instance&page={page}&page_size={page_size}"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["page"] == page
|
||||
assert data["page_size"] == min(page_size, 200) # Capped at 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"params,expected_status",
|
||||
[
|
||||
("page=0", 400), # page must be >= 1
|
||||
("page=-1", 400),
|
||||
("page_size=0", 400), # page_size must be >= 1
|
||||
("page_size=-1", 400),
|
||||
("page=abc", 400), # page must be integer
|
||||
("page_size=xyz", 400), # page_size must be integer
|
||||
],
|
||||
)
|
||||
async def test_allowed_json_pagination_errors(
|
||||
ds_with_permissions, params, expected_status
|
||||
):
|
||||
"""Test pagination error handling."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
f"/-/allowed.json?action=view-instance&{params}"
|
||||
)
|
||||
assert response.status_code == expected_status
|
||||
|
||||
|
||||
# /-/rules.json tests
|
||||
@pytest.mark.asyncio
|
||||
async def test_rules_json_requires_permissions_debug(ds_with_permissions):
|
||||
"""Test that /-/rules.json requires permissions-debug permission."""
|
||||
# Anonymous user should be denied
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/rules.json?action=view-instance"
|
||||
)
|
||||
assert response.status_code == 403
|
||||
|
||||
# Regular authenticated user should also be denied
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/rules.json?action=view-instance",
|
||||
cookies={
|
||||
"ds_actor": ds_with_permissions.client.actor_cookie({"id": "regular-user"})
|
||||
},
|
||||
)
|
||||
assert response.status_code == 403
|
||||
|
||||
# User with permissions-debug should be allowed
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/rules.json?action=view-instance",
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"path,expected_status,expected_keys",
|
||||
[
|
||||
# Valid request
|
||||
(
|
||||
"/-/rules.json?action=view-instance",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
(
|
||||
"/-/rules.json?action=view-database",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
# Missing action parameter
|
||||
("/-/rules.json", 400, {"error"}),
|
||||
# Invalid action
|
||||
("/-/rules.json?action=nonexistent", 404, {"error"}),
|
||||
],
|
||||
)
|
||||
async def test_rules_json_basic(
|
||||
ds_with_permissions, path, expected_status, expected_keys
|
||||
):
|
||||
# Use debugger user who has permissions-debug
|
||||
response = await ds_with_permissions.client.get(
|
||||
path,
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
assert response.status_code == expected_status
|
||||
data = response.json()
|
||||
assert expected_keys.issubset(data.keys())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rules_json_response_structure(ds_with_permissions):
|
||||
"""Test that /-/rules.json returns the expected structure."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/rules.json?action=view-instance",
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# Check required fields
|
||||
assert "action" in data
|
||||
assert "actor_id" in data
|
||||
assert "page" in data
|
||||
assert "page_size" in data
|
||||
assert "total" in data
|
||||
assert "items" in data
|
||||
|
||||
# Check items structure
|
||||
assert isinstance(data["items"], list)
|
||||
if data["items"]:
|
||||
item = data["items"][0]
|
||||
assert "parent" in item
|
||||
assert "child" in item
|
||||
assert "resource" in item
|
||||
assert "allow" in item # Important: should include allow field
|
||||
assert "reason" in item
|
||||
assert "source_plugin" in item
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rules_json_includes_both_allow_and_deny(ds_with_permissions):
|
||||
"""Test that /-/rules.json includes both allow and deny rules."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/rules.json?action=view-database",
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# Check that items have the allow field
|
||||
assert isinstance(data["items"], list)
|
||||
if data["items"]:
|
||||
# Verify allow field exists and is 0 or 1
|
||||
for item in data["items"]:
|
||||
assert "allow" in item
|
||||
assert item["allow"] in (0, 1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"page,page_size",
|
||||
[
|
||||
(1, 10),
|
||||
(2, 50),
|
||||
(1, 200), # max page size
|
||||
],
|
||||
)
|
||||
async def test_rules_json_pagination(ds_with_permissions, page, page_size):
|
||||
"""Test pagination parameters."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
f"/-/rules.json?action=view-instance&page={page}&page_size={page_size}",
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["page"] == page
|
||||
assert data["page_size"] == min(page_size, 200) # Capped at 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"params,expected_status",
|
||||
[
|
||||
("page=0", 400), # page must be >= 1
|
||||
("page=-1", 400),
|
||||
("page_size=0", 400), # page_size must be >= 1
|
||||
("page_size=-1", 400),
|
||||
("page=abc", 400), # page must be integer
|
||||
("page_size=xyz", 400), # page_size must be integer
|
||||
],
|
||||
)
|
||||
async def test_rules_json_pagination_errors(
|
||||
ds_with_permissions, params, expected_status
|
||||
):
|
||||
"""Test pagination error handling."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
f"/-/rules.json?action=view-instance&{params}",
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
assert response.status_code == expected_status
|
||||
|
||||
|
||||
# Test that HTML endpoints return HTML (not JSON) when accessed without .json
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"path,needs_debug",
|
||||
[
|
||||
("/-/check", False),
|
||||
("/-/check?action=view-instance", False),
|
||||
("/-/allowed", False),
|
||||
("/-/allowed?action=view-instance", False),
|
||||
("/-/rules", True),
|
||||
("/-/rules?action=view-instance", True),
|
||||
],
|
||||
)
|
||||
async def test_html_endpoints_return_html(ds_with_permissions, path, needs_debug):
|
||||
"""Test that endpoints without .json extension return HTML."""
|
||||
if needs_debug:
|
||||
# Rules endpoint requires permissions-debug
|
||||
response = await ds_with_permissions.client.get(
|
||||
path,
|
||||
cookies={
|
||||
"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})
|
||||
},
|
||||
)
|
||||
else:
|
||||
response = await ds_with_permissions.client.get(path)
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers["content-type"]
|
||||
# Check for HTML structure
|
||||
text = response.text
|
||||
assert "<!DOCTYPE html>" in text or "<html" in text
|
||||
|
|
@ -12,8 +12,9 @@ from datasette.app import Datasette
|
|||
from datasette import cli, hookimpl, Permission
|
||||
from datasette.filters import FilterArguments
|
||||
from datasette.plugins import get_plugins, DEFAULT_PLUGINS, pm
|
||||
from datasette.utils.permissions import PluginSQL
|
||||
from datasette.utils.sqlite import sqlite3
|
||||
from datasette.utils import StartupError
|
||||
from datasette.utils import StartupError, await_me_maybe
|
||||
from jinja2 import ChoiceLoader, FileSystemLoader
|
||||
import base64
|
||||
import datetime
|
||||
|
|
@ -701,6 +702,29 @@ async def test_hook_permission_allowed(action, expected):
|
|||
pm.unregister(name="undo_register_extras")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_hook_permission_resources_sql():
|
||||
ds = Datasette()
|
||||
await ds.invoke_startup()
|
||||
|
||||
collected = []
|
||||
for block in pm.hook.permission_resources_sql(
|
||||
datasette=ds,
|
||||
actor={"id": "alice"},
|
||||
action="view-table",
|
||||
):
|
||||
block = await await_me_maybe(block)
|
||||
if block is None:
|
||||
continue
|
||||
if isinstance(block, (list, tuple)):
|
||||
collected.extend(block)
|
||||
else:
|
||||
collected.append(block)
|
||||
|
||||
assert collected
|
||||
assert all(isinstance(item, PluginSQL) for item in collected)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_actor_json(ds_client):
|
||||
assert (await ds_client.get("/-/actor.json")).json() == {"actor": None}
|
||||
|
|
|
|||
440
tests/test_utils_permissions.py
Normal file
440
tests/test_utils_permissions.py
Normal file
|
|
@ -0,0 +1,440 @@
|
|||
import pytest
|
||||
from datasette.app import Datasette
|
||||
from datasette.utils.permissions import (
|
||||
PluginSQL,
|
||||
PluginProvider,
|
||||
resolve_permissions_from_catalog,
|
||||
)
|
||||
from typing import List
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db():
|
||||
ds = Datasette()
|
||||
import tempfile
|
||||
from datasette.database import Database
|
||||
|
||||
path = tempfile.mktemp(suffix="demo.db")
|
||||
db = ds.add_database(Database(ds, path=path))
|
||||
print(path)
|
||||
return db
|
||||
|
||||
|
||||
NO_RULES_SQL = (
|
||||
"SELECT NULL AS parent, NULL AS child, NULL AS allow, NULL AS reason WHERE 0"
|
||||
)
|
||||
|
||||
|
||||
def plugin_allow_all_for_user(user: str) -> PluginProvider:
|
||||
def provider(action: str) -> PluginSQL:
|
||||
return PluginSQL(
|
||||
"allow_all",
|
||||
"""
|
||||
SELECT NULL AS parent, NULL AS child, 1 AS allow,
|
||||
'global allow for ' || :user || ' on ' || :action AS reason
|
||||
WHERE :actor = :user
|
||||
""",
|
||||
{"user": user, "action": action},
|
||||
)
|
||||
|
||||
return provider
|
||||
|
||||
|
||||
def plugin_deny_specific_table(user: str, parent: str, child: str) -> PluginProvider:
|
||||
def provider(action: str) -> PluginSQL:
|
||||
return PluginSQL(
|
||||
"deny_specific_table",
|
||||
"""
|
||||
SELECT :parent AS parent, :child AS child, 0 AS allow,
|
||||
'deny ' || :parent || '/' || :child || ' for ' || :user || ' on ' || :action AS reason
|
||||
WHERE :actor = :user
|
||||
""",
|
||||
{"parent": parent, "child": child, "user": user, "action": action},
|
||||
)
|
||||
|
||||
return provider
|
||||
|
||||
|
||||
def plugin_org_policy_deny_parent(parent: str) -> PluginProvider:
|
||||
def provider(action: str) -> PluginSQL:
|
||||
return PluginSQL(
|
||||
"org_policy_parent_deny",
|
||||
"""
|
||||
SELECT :parent AS parent, NULL AS child, 0 AS allow,
|
||||
'org policy: parent ' || :parent || ' denied on ' || :action AS reason
|
||||
""",
|
||||
{"parent": parent, "action": action},
|
||||
)
|
||||
|
||||
return provider
|
||||
|
||||
|
||||
def plugin_allow_parent_for_user(user: str, parent: str) -> PluginProvider:
|
||||
def provider(action: str) -> PluginSQL:
|
||||
return PluginSQL(
|
||||
"allow_parent",
|
||||
"""
|
||||
SELECT :parent AS parent, NULL AS child, 1 AS allow,
|
||||
'allow full parent for ' || :user || ' on ' || :action AS reason
|
||||
WHERE :actor = :user
|
||||
""",
|
||||
{"parent": parent, "user": user, "action": action},
|
||||
)
|
||||
|
||||
return provider
|
||||
|
||||
|
||||
def plugin_child_allow_for_user(user: str, parent: str, child: str) -> PluginProvider:
|
||||
def provider(action: str) -> PluginSQL:
|
||||
return PluginSQL(
|
||||
"allow_child",
|
||||
"""
|
||||
SELECT :parent AS parent, :child AS child, 1 AS allow,
|
||||
'allow child for ' || :user || ' on ' || :action AS reason
|
||||
WHERE :actor = :user
|
||||
""",
|
||||
{"parent": parent, "child": child, "user": user, "action": action},
|
||||
)
|
||||
|
||||
return provider
|
||||
|
||||
|
||||
def plugin_root_deny_for_all() -> PluginProvider:
|
||||
def provider(action: str) -> PluginSQL:
|
||||
return PluginSQL(
|
||||
"root_deny",
|
||||
"""
|
||||
SELECT NULL AS parent, NULL AS child, 0 AS allow, 'root deny for all on ' || :action AS reason
|
||||
""",
|
||||
{"action": action},
|
||||
)
|
||||
|
||||
return provider
|
||||
|
||||
|
||||
def plugin_conflicting_same_child_rules(
|
||||
user: str, parent: str, child: str
|
||||
) -> List[PluginProvider]:
|
||||
def allow_provider(action: str) -> PluginSQL:
|
||||
return PluginSQL(
|
||||
"conflict_child_allow",
|
||||
"""
|
||||
SELECT :parent AS parent, :child AS child, 1 AS allow,
|
||||
'team grant at child for ' || :user || ' on ' || :action AS reason
|
||||
WHERE :actor = :user
|
||||
""",
|
||||
{"parent": parent, "child": child, "user": user, "action": action},
|
||||
)
|
||||
|
||||
def deny_provider(action: str) -> PluginSQL:
|
||||
return PluginSQL(
|
||||
"conflict_child_deny",
|
||||
"""
|
||||
SELECT :parent AS parent, :child AS child, 0 AS allow,
|
||||
'exception deny at child for ' || :user || ' on ' || :action AS reason
|
||||
WHERE :actor = :user
|
||||
""",
|
||||
{"parent": parent, "child": child, "user": user, "action": action},
|
||||
)
|
||||
|
||||
return [allow_provider, deny_provider]
|
||||
|
||||
|
||||
def plugin_allow_all_for_action(user: str, allowed_action: str) -> PluginProvider:
|
||||
def provider(action: str) -> PluginSQL:
|
||||
if action != allowed_action:
|
||||
return PluginSQL(
|
||||
f"allow_all_{allowed_action}_noop",
|
||||
NO_RULES_SQL,
|
||||
{},
|
||||
)
|
||||
return PluginSQL(
|
||||
f"allow_all_{allowed_action}",
|
||||
"""
|
||||
SELECT NULL AS parent, NULL AS child, 1 AS allow,
|
||||
'global allow for ' || :user || ' on ' || :action AS reason
|
||||
WHERE :actor = :user
|
||||
""",
|
||||
{"user": user, "action": action},
|
||||
)
|
||||
|
||||
return provider
|
||||
|
||||
|
||||
VIEW_TABLE = "view-table"
|
||||
|
||||
|
||||
# ---------- Catalog DDL (from your schema) ----------
|
||||
CATALOG_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS catalog_databases (
|
||||
database_name TEXT PRIMARY KEY,
|
||||
path TEXT,
|
||||
is_memory INTEGER,
|
||||
schema_version INTEGER
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS catalog_tables (
|
||||
database_name TEXT,
|
||||
table_name TEXT,
|
||||
rootpage INTEGER,
|
||||
sql TEXT,
|
||||
PRIMARY KEY (database_name, table_name),
|
||||
FOREIGN KEY (database_name) REFERENCES catalog_databases(database_name)
|
||||
);
|
||||
"""
|
||||
|
||||
PARENTS = ["accounting", "hr", "analytics"]
|
||||
SPECIALS = {"accounting": ["sales"], "analytics": ["secret"], "hr": []}
|
||||
|
||||
TABLE_CANDIDATES_SQL = (
|
||||
"SELECT database_name AS parent, table_name AS child FROM catalog_tables"
|
||||
)
|
||||
PARENT_CANDIDATES_SQL = (
|
||||
"SELECT database_name AS parent, NULL AS child FROM catalog_databases"
|
||||
)
|
||||
|
||||
|
||||
# ---------- Helpers ----------
|
||||
async def seed_catalog(db, per_parent: int = 10) -> None:
|
||||
await db.execute_write_script(CATALOG_DDL)
|
||||
# databases
|
||||
db_rows = [(p, f"/{p}.db", 0, 1) for p in PARENTS]
|
||||
await db.execute_write_many(
|
||||
"INSERT OR REPLACE INTO catalog_databases(database_name, path, is_memory, schema_version) VALUES (?,?,?,?)",
|
||||
db_rows,
|
||||
)
|
||||
|
||||
# tables
|
||||
def tables_for(parent: str, n: int):
|
||||
base = [f"table{i:02d}" for i in range(1, n + 1)]
|
||||
for s in SPECIALS.get(parent, []):
|
||||
if s not in base:
|
||||
base[0] = s
|
||||
return base
|
||||
|
||||
table_rows = []
|
||||
for p in PARENTS:
|
||||
for t in tables_for(p, per_parent):
|
||||
table_rows.append((p, t, 0, f"CREATE TABLE {t} (id INTEGER PRIMARY KEY)"))
|
||||
await db.execute_write_many(
|
||||
"INSERT OR REPLACE INTO catalog_tables(database_name, table_name, rootpage, sql) VALUES (?,?,?,?)",
|
||||
table_rows,
|
||||
)
|
||||
|
||||
|
||||
def res_allowed(rows, parent=None):
|
||||
return sorted(
|
||||
r["resource"]
|
||||
for r in rows
|
||||
if r["allow"] == 1 and (parent is None or r["parent"] == parent)
|
||||
)
|
||||
|
||||
|
||||
def res_denied(rows, parent=None):
|
||||
return sorted(
|
||||
r["resource"]
|
||||
for r in rows
|
||||
if r["allow"] == 0 and (parent is None or r["parent"] == parent)
|
||||
)
|
||||
|
||||
|
||||
# ---------- Tests ----------
|
||||
@pytest.mark.asyncio
|
||||
async def test_alice_global_allow_with_specific_denies_catalog(db):
|
||||
await seed_catalog(db)
|
||||
plugins = [
|
||||
plugin_allow_all_for_user("alice"),
|
||||
plugin_deny_specific_table("alice", "accounting", "sales"),
|
||||
plugin_org_policy_deny_parent("hr"),
|
||||
]
|
||||
rows = await resolve_permissions_from_catalog(
|
||||
db, "alice", plugins, VIEW_TABLE, TABLE_CANDIDATES_SQL, implicit_deny=True
|
||||
)
|
||||
# Alice can see everything except accounting/sales and hr/*
|
||||
assert "/accounting/sales" in res_denied(rows)
|
||||
for r in rows:
|
||||
if r["parent"] == "hr":
|
||||
assert r["allow"] == 0
|
||||
elif r["resource"] == "/accounting/sales":
|
||||
assert r["allow"] == 0
|
||||
else:
|
||||
assert r["allow"] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_carol_parent_allow_but_child_conflict_deny_wins_catalog(db):
|
||||
await seed_catalog(db)
|
||||
plugins = [
|
||||
plugin_org_policy_deny_parent("hr"),
|
||||
plugin_allow_parent_for_user("carol", "analytics"),
|
||||
*plugin_conflicting_same_child_rules("carol", "analytics", "secret"),
|
||||
]
|
||||
rows = await resolve_permissions_from_catalog(
|
||||
db, "carol", plugins, VIEW_TABLE, TABLE_CANDIDATES_SQL, implicit_deny=True
|
||||
)
|
||||
allowed_analytics = res_allowed(rows, parent="analytics")
|
||||
denied_analytics = res_denied(rows, parent="analytics")
|
||||
|
||||
assert "/analytics/secret" in denied_analytics
|
||||
# 10 analytics children total, 1 denied
|
||||
assert len(allowed_analytics) == 9
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_specificity_child_allow_overrides_parent_deny_catalog(db):
|
||||
await seed_catalog(db)
|
||||
plugins = [
|
||||
plugin_allow_all_for_user("alice"),
|
||||
plugin_org_policy_deny_parent("analytics"), # parent-level deny
|
||||
plugin_child_allow_for_user(
|
||||
"alice", "analytics", "table02"
|
||||
), # child allow beats parent deny
|
||||
]
|
||||
rows = await resolve_permissions_from_catalog(
|
||||
db, "alice", plugins, VIEW_TABLE, TABLE_CANDIDATES_SQL, implicit_deny=True
|
||||
)
|
||||
|
||||
# table02 allowed, other analytics tables denied
|
||||
assert any(r["resource"] == "/analytics/table02" and r["allow"] == 1 for r in rows)
|
||||
assert all(
|
||||
(r["parent"] != "analytics" or r["child"] == "table02" or r["allow"] == 0)
|
||||
for r in rows
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_root_deny_all_but_parent_allow_rescues_specific_parent_catalog(db):
|
||||
await seed_catalog(db)
|
||||
plugins = [
|
||||
plugin_root_deny_for_all(), # root deny
|
||||
plugin_allow_parent_for_user(
|
||||
"bob", "accounting"
|
||||
), # parent allow (more specific)
|
||||
]
|
||||
rows = await resolve_permissions_from_catalog(
|
||||
db, "bob", plugins, VIEW_TABLE, TABLE_CANDIDATES_SQL, implicit_deny=True
|
||||
)
|
||||
for r in rows:
|
||||
if r["parent"] == "accounting":
|
||||
assert r["allow"] == 1
|
||||
else:
|
||||
assert r["allow"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parent_scoped_candidates(db):
|
||||
await seed_catalog(db)
|
||||
plugins = [
|
||||
plugin_org_policy_deny_parent("hr"),
|
||||
plugin_allow_parent_for_user("carol", "analytics"),
|
||||
]
|
||||
rows = await resolve_permissions_from_catalog(
|
||||
db, "carol", plugins, VIEW_TABLE, PARENT_CANDIDATES_SQL, implicit_deny=True
|
||||
)
|
||||
d = {r["resource"]: r["allow"] for r in rows}
|
||||
assert d["/analytics"] == 1
|
||||
assert d["/hr"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_implicit_deny_behavior(db):
|
||||
await seed_catalog(db)
|
||||
plugins = [] # no rules at all
|
||||
|
||||
# implicit_deny=True -> everything denied with reason 'implicit deny'
|
||||
rows = await resolve_permissions_from_catalog(
|
||||
db, "erin", plugins, VIEW_TABLE, TABLE_CANDIDATES_SQL, implicit_deny=True
|
||||
)
|
||||
assert all(r["allow"] == 0 and r["reason"] == "implicit deny" for r in rows)
|
||||
|
||||
# implicit_deny=False -> no winner => allow is None, reason is None
|
||||
rows2 = await resolve_permissions_from_catalog(
|
||||
db, "erin", plugins, VIEW_TABLE, TABLE_CANDIDATES_SQL, implicit_deny=False
|
||||
)
|
||||
assert all(r["allow"] is None and r["reason"] is None for r in rows2)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_candidate_filters_via_params(db):
|
||||
await seed_catalog(db)
|
||||
# Add some metadata to test filtering
|
||||
# Mark 'hr' as is_memory=1 and increment analytics schema_version
|
||||
await db.execute_write(
|
||||
"UPDATE catalog_databases SET is_memory=1 WHERE database_name='hr'"
|
||||
)
|
||||
await db.execute_write(
|
||||
"UPDATE catalog_databases SET schema_version=2 WHERE database_name='analytics'"
|
||||
)
|
||||
|
||||
# Candidate SQL that filters by db metadata via params
|
||||
candidate_sql = """
|
||||
SELECT t.database_name AS parent, t.table_name AS child
|
||||
FROM catalog_tables t
|
||||
JOIN catalog_databases d ON d.database_name = t.database_name
|
||||
WHERE (:exclude_memory = 1 AND d.is_memory = 1) IS NOT 1
|
||||
AND (:min_schema_version IS NULL OR d.schema_version >= :min_schema_version)
|
||||
"""
|
||||
|
||||
plugins = [
|
||||
plugin_root_deny_for_all(),
|
||||
plugin_allow_parent_for_user(
|
||||
"dev", "analytics"
|
||||
), # analytics rescued if included by candidates
|
||||
]
|
||||
|
||||
# Case 1: exclude memory dbs, require schema_version >= 2 -> only analytics appear, and thus are allowed
|
||||
rows = await resolve_permissions_from_catalog(
|
||||
db,
|
||||
"dev",
|
||||
plugins,
|
||||
VIEW_TABLE,
|
||||
candidate_sql,
|
||||
candidate_params={"exclude_memory": 1, "min_schema_version": 2},
|
||||
implicit_deny=True,
|
||||
)
|
||||
assert rows and all(r["parent"] == "analytics" for r in rows)
|
||||
assert all(r["allow"] == 1 for r in rows)
|
||||
|
||||
# Case 2: include memory dbs, min_schema_version = None -> accounting/hr/analytics appear,
|
||||
# but root deny wins except where specifically allowed (none except analytics parent allow doesn’t apply to table depth if candidate includes children; still fine—policy is explicit).
|
||||
rows2 = await resolve_permissions_from_catalog(
|
||||
db,
|
||||
"dev",
|
||||
plugins,
|
||||
VIEW_TABLE,
|
||||
candidate_sql,
|
||||
candidate_params={"exclude_memory": 0, "min_schema_version": None},
|
||||
implicit_deny=True,
|
||||
)
|
||||
assert any(r["parent"] == "accounting" for r in rows2)
|
||||
assert any(r["parent"] == "hr" for r in rows2)
|
||||
# For table-scoped candidates, the parent-level allow does not override root deny unless you have child-level rules
|
||||
assert all(r["allow"] in (0, 1) for r in rows2)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_action_specific_rules(db):
|
||||
await seed_catalog(db)
|
||||
plugins = [plugin_allow_all_for_action("dana", VIEW_TABLE)]
|
||||
|
||||
view_rows = await resolve_permissions_from_catalog(
|
||||
db,
|
||||
"dana",
|
||||
plugins,
|
||||
VIEW_TABLE,
|
||||
TABLE_CANDIDATES_SQL,
|
||||
implicit_deny=True,
|
||||
)
|
||||
assert view_rows and all(r["allow"] == 1 for r in view_rows)
|
||||
assert all(r["action"] == VIEW_TABLE for r in view_rows)
|
||||
|
||||
insert_rows = await resolve_permissions_from_catalog(
|
||||
db,
|
||||
"dana",
|
||||
plugins,
|
||||
"insert-row",
|
||||
TABLE_CANDIDATES_SQL,
|
||||
implicit_deny=True,
|
||||
)
|
||||
assert insert_rows and all(r["allow"] == 0 for r in insert_rows)
|
||||
assert all(r["reason"] == "implicit deny" for r in insert_rows)
|
||||
assert all(r["action"] == "insert-row" for r in insert_rows)
|
||||
Loading…
Add table
Add a link
Reference in a new issue