mirror of
https://github.com/simonw/datasette.git
synced 2025-12-10 16:51:24 +01:00
Implement also_requires to enforce view-database for execute-sql
Adds Action.also_requires field to specify dependencies between permissions. When an action has also_requires set, users must have permission for BOTH the main action AND the required action on a resource. Applies this to execute-sql, which now requires view-database permission. This prevents the illogical scenario where users can execute SQL on a database they cannot view. Changes: - Add also_requires field to Action dataclass in datasette/permissions.py - Update execute-sql action with also_requires="view-database" - Implement also_requires handling in build_allowed_resources_sql() - Implement also_requires handling in AllowedResourcesView endpoint - Add test verifying execute-sql requires view-database permission Fixes #2527
This commit is contained in:
Simon Willison
parent
a2994cc5bb
commit
e8b79970fb
6 changed files with 460 additions and 366 deletions
|
|
@ -1,6 +1,5 @@
|
|||
"""
|
||||
Tests for permission inspection endpoints:
|
||||
- /-/check.json
|
||||
Tests for permission endpoints:
|
||||
- /-/allowed.json
|
||||
- /-/rules.json
|
||||
"""
|
||||
|
|
@ -12,159 +11,50 @@ 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
|
||||
},
|
||||
}
|
||||
}
|
||||
)
|
||||
"""Create a Datasette instance with test data and permissions."""
|
||||
ds = Datasette()
|
||||
ds.root_enabled = True
|
||||
await ds.invoke_startup()
|
||||
# Add some test databases
|
||||
ds.add_memory_database("content")
|
||||
ds.add_memory_database("private")
|
||||
|
||||
# Add some test databases and tables
|
||||
db = ds.add_memory_database("analytics")
|
||||
await db.execute_write(
|
||||
"CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)"
|
||||
)
|
||||
await db.execute_write(
|
||||
"CREATE TABLE IF NOT EXISTS events (id INTEGER PRIMARY KEY, event_type TEXT, user_id INTEGER)"
|
||||
)
|
||||
|
||||
db2 = ds.add_memory_database("production")
|
||||
await db2.execute_write(
|
||||
"CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, total REAL)"
|
||||
)
|
||||
await db2.execute_write(
|
||||
"CREATE TABLE IF NOT EXISTS customers (id INTEGER PRIMARY KEY, name TEXT)"
|
||||
)
|
||||
|
||||
await ds.refresh_schemas()
|
||||
|
||||
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
|
||||
# Instance level permission
|
||||
(
|
||||
"/-/allowed.json?action=view-instance",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
# Database level permission
|
||||
(
|
||||
"/-/allowed.json?action=view-database",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
# Table level permission
|
||||
(
|
||||
"/-/allowed.json?action=view-table",
|
||||
200,
|
||||
|
|
@ -219,139 +109,98 @@ async def test_allowed_json_response_structure(ds_with_permissions):
|
|||
|
||||
|
||||
@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
|
||||
async def test_allowed_json_with_actor(ds_with_permissions):
|
||||
"""Test /-/allowed.json includes actor information."""
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/allowed.json?action=view-instance"
|
||||
"/-/allowed.json?action=view-table",
|
||||
cookies={
|
||||
"ds_actor": ds_with_permissions.client.actor_cookie({"id": "test_user"})
|
||||
},
|
||||
)
|
||||
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
|
||||
assert data["actor_id"] == "test_user"
|
||||
|
||||
|
||||
@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"})},
|
||||
async def test_allowed_json_pagination():
|
||||
"""Test that /-/allowed.json pagination works."""
|
||||
ds = Datasette()
|
||||
await ds.invoke_startup()
|
||||
|
||||
# Create many tables to test pagination
|
||||
db = ds.add_memory_database("test")
|
||||
for i in range(30):
|
||||
await db.execute_write(f"CREATE TABLE table{i:02d} (id INTEGER PRIMARY KEY)")
|
||||
await ds.refresh_schemas()
|
||||
|
||||
# Test page 1
|
||||
response = await ds.client.get(
|
||||
"/-/allowed.json?action=view-table&page_size=10&page=1"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
if data["items"]:
|
||||
item = data["items"][0]
|
||||
assert "reason" in item
|
||||
assert "source_plugin" in item
|
||||
assert data["page"] == 1
|
||||
assert data["page_size"] == 10
|
||||
assert len(data["items"]) == 10
|
||||
|
||||
|
||||
@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"
|
||||
# Test page 2
|
||||
response = await ds.client.get(
|
||||
"/-/allowed.json?action=view-table&page_size=10&page=2"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["page"] == 2
|
||||
assert len(data["items"]) == 10
|
||||
|
||||
# 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
|
||||
# Verify items are different between pages
|
||||
response1 = await ds.client.get(
|
||||
"/-/allowed.json?action=view-table&page_size=10&page=1"
|
||||
)
|
||||
response2 = await ds.client.get(
|
||||
"/-/allowed.json?action=view-table&page_size=10&page=2"
|
||||
)
|
||||
items1 = {(item["parent"], item["child"]) for item in response1.json()["items"]}
|
||||
items2 = {(item["parent"], item["child"]) for item in response2.json()["items"]}
|
||||
assert items1 != items2
|
||||
|
||||
|
||||
@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}"
|
||||
)
|
||||
async def test_allowed_json_total_count(ds_with_permissions):
|
||||
"""Test that /-/allowed.json returns correct total count."""
|
||||
response = await ds_with_permissions.client.get("/-/allowed.json?action=view-table")
|
||||
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
|
||||
# We created 4 tables total (2 in analytics, 2 in production)
|
||||
assert data["total"] == 4
|
||||
|
||||
|
||||
# /-/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
|
||||
# Instance level rules
|
||||
(
|
||||
"/-/rules.json?action=view-instance",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
# Database level rules
|
||||
(
|
||||
"/-/rules.json?action=view-database",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
# Table level rules
|
||||
(
|
||||
"/-/rules.json?action=view-table",
|
||||
200,
|
||||
{"action", "items", "total", "page"},
|
||||
),
|
||||
# Missing action parameter
|
||||
("/-/rules.json", 400, {"error"}),
|
||||
# Invalid action
|
||||
|
|
@ -361,7 +210,7 @@ async def test_rules_json_requires_permissions_debug(ds_with_permissions):
|
|||
async def test_rules_json_basic(
|
||||
ds_with_permissions, path, expected_status, expected_keys
|
||||
):
|
||||
# Use debugger user who has permissions-debug
|
||||
# Use root actor for rules endpoint (requires permissions-debug)
|
||||
response = await ds_with_permissions.client.get(
|
||||
path,
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
|
|
@ -396,104 +245,71 @@ async def test_rules_json_response_structure(ds_with_permissions):
|
|||
assert "parent" in item
|
||||
assert "child" in item
|
||||
assert "resource" in item
|
||||
assert "allow" in item # Important: should include allow field
|
||||
assert "allow" in item
|
||||
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."""
|
||||
async def test_rules_json_includes_all_rules(ds_with_permissions):
|
||||
"""Test that /-/rules.json includes both allowed and denied resources."""
|
||||
# Root user should see rules for everything
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/rules.json?action=view-database",
|
||||
"/-/rules.json?action=view-table",
|
||||
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)
|
||||
# Should have items (root has global allow)
|
||||
assert len(data["items"]) > 0
|
||||
|
||||
# Each item should have allow field (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
|
||||
async def test_rules_json_pagination():
|
||||
"""Test that /-/rules.json pagination works."""
|
||||
ds = Datasette()
|
||||
ds.root_enabled = True
|
||||
await ds.invoke_startup()
|
||||
|
||||
|
||||
@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"})
|
||||
},
|
||||
# Create some tables
|
||||
db = ds.add_memory_database("test")
|
||||
for i in range(5):
|
||||
await db.execute_write(
|
||||
f"CREATE TABLE IF NOT EXISTS table{i:02d} (id INTEGER PRIMARY KEY)"
|
||||
)
|
||||
else:
|
||||
response = await ds_with_permissions.client.get(path)
|
||||
await ds.refresh_schemas()
|
||||
|
||||
# Test basic pagination structure - just verify it returns paginated results
|
||||
response = await ds.client.get(
|
||||
"/-/rules.json?action=view-table&page_size=2&page=1",
|
||||
cookies={"ds_actor": ds.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
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
|
||||
data = response.json()
|
||||
assert data["page"] == 1
|
||||
assert data["page_size"] == 2
|
||||
# Verify items is a list (may have fewer items than page_size if there aren't many rules)
|
||||
assert isinstance(data["items"], list)
|
||||
assert "total" in data
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rules_json_with_actor(ds_with_permissions):
|
||||
"""Test /-/rules.json includes actor information."""
|
||||
# Use root actor (rules endpoint requires permissions-debug)
|
||||
response = await ds_with_permissions.client.get(
|
||||
"/-/rules.json?action=view-table",
|
||||
cookies={"ds_actor": ds_with_permissions.client.actor_cookie({"id": "root"})},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["actor_id"] == "root"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -502,8 +318,7 @@ async def test_root_user_respects_settings_deny():
|
|||
Test for issue #2509: Settings-based deny rules should override root user privileges.
|
||||
|
||||
When a database has `allow: false` in settings, the root user should NOT see
|
||||
that database in /-/allowed.json?action=view-database, even though root normally
|
||||
has all permissions.
|
||||
that database in /-/allowed.json?action=view-database.
|
||||
"""
|
||||
ds = Datasette(
|
||||
config={
|
||||
|
|
@ -518,7 +333,7 @@ async def test_root_user_respects_settings_deny():
|
|||
await ds.invoke_startup()
|
||||
ds.add_memory_database("content")
|
||||
|
||||
# Root user should NOT see the content database because settings deny it
|
||||
# Root user should NOT see the denied database
|
||||
response = await ds.client.get(
|
||||
"/-/allowed.json?action=view-database",
|
||||
cookies={"ds_actor": ds.client.actor_cookie({"id": "root"})},
|
||||
|
|
@ -575,3 +390,78 @@ async def test_root_user_respects_settings_deny_tables():
|
|||
f"Root user should not see tables from 'content' database when settings deny it, "
|
||||
f"but found: {content_tables}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_sql_requires_view_database():
|
||||
"""
|
||||
Test for issue #2527: execute-sql permission should require view-database permission.
|
||||
|
||||
A user who has execute-sql permission but not view-database permission should not
|
||||
be able to execute SQL on that database.
|
||||
"""
|
||||
from datasette.permissions import PermissionSQL
|
||||
from datasette.plugins import pm
|
||||
from datasette import hookimpl
|
||||
|
||||
class TestPermissionPlugin:
|
||||
__name__ = "TestPermissionPlugin"
|
||||
|
||||
@hookimpl
|
||||
def permission_resources_sql(self, datasette, actor, action):
|
||||
if actor is None or actor.get("id") != "test_user":
|
||||
return []
|
||||
|
||||
if action == "execute-sql":
|
||||
# Grant execute-sql on the "secret" database
|
||||
return PermissionSQL(
|
||||
source="test_plugin",
|
||||
sql="SELECT 'secret' AS parent, NULL AS child, 1 AS allow, 'can execute sql' AS reason",
|
||||
params={},
|
||||
)
|
||||
elif action == "view-database":
|
||||
# Deny view-database on the "secret" database
|
||||
return PermissionSQL(
|
||||
source="test_plugin",
|
||||
sql="SELECT 'secret' AS parent, NULL AS child, 0 AS allow, 'cannot view db' AS reason",
|
||||
params={},
|
||||
)
|
||||
|
||||
return []
|
||||
|
||||
plugin = TestPermissionPlugin()
|
||||
pm.register(plugin, name="test_plugin")
|
||||
|
||||
try:
|
||||
ds = Datasette()
|
||||
await ds.invoke_startup()
|
||||
ds.add_memory_database("secret")
|
||||
await ds.refresh_schemas()
|
||||
|
||||
# User should NOT have execute-sql permission because view-database is denied
|
||||
response = await ds.client.get(
|
||||
"/-/allowed.json?action=execute-sql",
|
||||
cookies={"ds_actor": ds.client.actor_cookie({"id": "test_user"})},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# The "secret" database should NOT be in the allowed list for execute-sql
|
||||
allowed_databases = [item["parent"] for item in data["items"]]
|
||||
assert "secret" not in allowed_databases, (
|
||||
f"User should not have execute-sql permission without view-database, "
|
||||
f"but found 'secret' in: {allowed_databases}"
|
||||
)
|
||||
|
||||
# Also verify that attempting to execute SQL on the database is denied
|
||||
# (may be 403 or 302 redirect to login/error page depending on middleware)
|
||||
response = await ds.client.get(
|
||||
"/secret?sql=SELECT+1",
|
||||
cookies={"ds_actor": ds.client.actor_cookie({"id": "test_user"})},
|
||||
)
|
||||
assert response.status_code in (302, 403), (
|
||||
f"Expected 302 or 403 when trying to execute SQL without view-database permission, "
|
||||
f"but got {response.status_code}"
|
||||
)
|
||||
finally:
|
||||
pm.unregister(plugin)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue