From ce91a64cb788be44581c7c6401cf28d14faf77c9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 1 Aug 2026 02:14:29 +0000 Subject: [PATCH 1/4] Add MCP server endpoint at /-/mcp Implements a Model Context Protocol server for the dashboard, inspired by https://github.com/datasette/datasette-mcp - so LLM tools can explore the database and answer questions with read-only SQL. - New dependency-free stateless MCP Streamable HTTP endpoint (JSON responses) served at -/mcp under the dashboard URLs - Three read-only tools: list_tables, get_schema and execute_sql, with input/output schemas and readOnlyHint annotations - Requires an authenticated user with the execute_sql permission, and runs SQL through the same protected pattern as the dashboard views: dashboard database alias, rolled-back transaction, single statement only, DASHBOARD_ROW_LIMIT truncation and %(name)s parameter support - Documentation in docs/mcp.md - conftest.py now lets tests declare their own django_db marker, used by new transactional tests that verify read-only enforcement Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017ZRFpnuD9tZKWb19Ffg8iF --- conftest.py | 3 + django_sql_dashboard/mcp.py | 326 +++++++++++++++++++++++++++++++++++ django_sql_dashboard/urls.py | 2 + docs/index.md | 1 + docs/mcp.md | 50 ++++++ test_project/test_mcp.py | 237 +++++++++++++++++++++++++ 6 files changed, 619 insertions(+) create mode 100644 django_sql_dashboard/mcp.py create mode 100644 docs/mcp.md create mode 100644 test_project/test_mcp.py diff --git a/conftest.py b/conftest.py index bc0cd63..e2205f9 100644 --- a/conftest.py +++ b/conftest.py @@ -7,6 +7,9 @@ def pytest_collection_modifyitems(items): """Add django_db marker with databases to tests that need database access.""" for item in items: + if item.get_closest_marker("django_db") is not None: + # Test has its own explicit marker + continue fixturenames = getattr(item, "fixturenames", ()) # Tests using client fixtures or dashboard_db need both databases if any(f in fixturenames for f in ("admin_client", "client", "dashboard_db")): diff --git a/django_sql_dashboard/mcp.py b/django_sql_dashboard/mcp.py new file mode 100644 index 0000000..5061d58 --- /dev/null +++ b/django_sql_dashboard/mcp.py @@ -0,0 +1,326 @@ +""" +A Model Context Protocol (MCP) server for django-sql-dashboard. + +This implements the stateless variant of the MCP Streamable HTTP transport +as a single Django view, without any additional dependencies. It exposes +three read-only tools: + +- list_tables - list the tables visible to the dashboard connection +- get_schema - return the schema (tables, columns, types) for the database +- execute_sql - execute one read-only SQL query and return its results + +SQL execution uses the same protected path as the dashboard views: the +read-only "dashboard" database alias, a transaction that is rolled back, +the ``DASHBOARD_ROW_LIMIT`` row limit and the same named parameter support. +Callers must be authenticated and have the +``django_sql_dashboard.execute_sql`` permission. + +Inspired by https://github.com/datasette/datasette-mcp +""" + +import json + +from django.conf import settings +from django.db import connections +from django.db.utils import ProgrammingError +from django.http import HttpResponse, JsonResponse +from django.views.decorators.csrf import csrf_exempt + +from .utils import displayable_rows, extract_named_parameters + +MCP_PROTOCOL_VERSION = "2025-06-18" + +SERVER_INFO = { + "name": "django-sql-dashboard", + "version": "1.2", +} + +SERVER_INSTRUCTIONS = ( + "Call list_tables, then get_schema to see columns and types, " + "then execute_sql to answer questions using read-only PostgreSQL." +) + +TABLES_AND_COLUMNS_SQL = """ +select + information_schema.columns.table_name, + cast(column_name as text), + cast(data_type as text) +from + information_schema.columns +join + information_schema.tables on + information_schema.columns.table_name = information_schema.tables.table_name + and information_schema.tables.table_schema = 'public' +where + information_schema.columns.table_schema = 'public' +order by + information_schema.columns.table_name, + information_schema.columns.ordinal_position +""" + + +class ToolError(Exception): + "An expected, recoverable tool failure - returned as isError content" + + +def _dashboard_connection(): + alias = getattr(settings, "DASHBOARD_DB_ALIAS", "dashboard") + return connections[alias] + + +def _tables_and_columns(): + # Returns [(table_name, [(column, data_type), ...]), ...] + tables = [] + with _dashboard_connection().cursor() as cursor: + cursor.execute(TABLES_AND_COLUMNS_SQL) + for table_name, column, data_type in cursor.fetchall(): + if not tables or tables[-1][0] != table_name: + tables.append((table_name, [])) + tables[-1][1].append((column, data_type)) + return tables + + +def tool_list_tables(): + return {"tables": [table for table, _ in _tables_and_columns()]} + + +def tool_get_schema(): + blocks = [] + for table, columns in _tables_and_columns(): + column_lines = ",\n".join( + " {} {}".format(column, data_type) for column, data_type in columns + ) + blocks.append("{} (\n{}\n)".format(table, column_lines)) + return {"schema": "\n\n".join(blocks)} + + +def _serialize_cell(value): + if value is None or isinstance(value, (str, int, float, bool)): + return value + if isinstance(value, memoryview): + value = value.tobytes() + if isinstance(value, bytes): + return {"hex": value.hex()} + return str(value) + + +def tool_execute_sql(sql, parameters=None): + sql = sql.strip().rstrip(";") + if not sql: + raise ToolError("SQL query is required") + if ";" in sql: + raise ToolError("';' not allowed in SQL queries") + try: + extracted_parameters = extract_named_parameters(sql) + except ValueError: + raise ToolError(r"Invalid query - try escaping single '%' as double '%%'") + parameter_values = { + parameter: str((parameters or {}).get(parameter, "")) + for parameter in extracted_parameters + } + row_limit = getattr(settings, "DASHBOARD_ROW_LIMIT", None) or 100 + connection = _dashboard_connection() + with connection.cursor() as cursor: + try: + cursor.execute("BEGIN;") + # Running a SELECT prevents future SET TRANSACTION READ WRITE: + cursor.execute("SELECT 1;") + cursor.fetchall() + cursor.execute(sql, parameter_values) + try: + rows = list(cursor.fetchmany(row_limit + 1)) + columns = [c.name for c in cursor.description] + except ProgrammingError: + rows = [[str(cursor.statusmessage)]] + columns = ["statusmessage"] + except Exception as e: + raise ToolError(str(e)) + finally: + cursor.execute("ROLLBACK;") + return { + "columns": columns, + "rows": [ + [_serialize_cell(cell) for cell in row] + for row in displayable_rows(rows[:row_limit]) + ], + "truncated": len(rows) == row_limit + 1, + } + + +TOOLS = [ + { + "name": "list_tables", + "description": "List the tables available to SQL queries in this dashboard.", + "handler": tool_list_tables, + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + "outputSchema": { + "type": "object", + "properties": { + "tables": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["tables"], + }, + }, + { + "name": "get_schema", + "description": ( + "Return the schema for the dashboard database: every table with " + "its columns and their PostgreSQL types. Call this before " + "constructing a SQL query." + ), + "handler": tool_get_schema, + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + "outputSchema": { + "type": "object", + "properties": { + "schema": {"type": "string"}, + }, + "required": ["schema"], + }, + }, + { + "name": "execute_sql", + "description": ( + "Execute one read-only PostgreSQL SELECT query and return its " + "columns and rows. Use %(name)s placeholders in the SQL and pass " + "their values in the parameters argument. The ';' character is " + "not allowed. Results are truncated to the dashboard row limit." + ), + "handler": tool_execute_sql, + "inputSchema": { + "type": "object", + "properties": { + "sql": { + "type": "string", + "description": "The PostgreSQL SELECT query to execute", + }, + "parameters": { + "type": "object", + "description": ( + "Values for any %(name)s parameters used in the SQL" + ), + "additionalProperties": {"type": "string"}, + }, + }, + "required": ["sql"], + "additionalProperties": False, + }, + "outputSchema": { + "type": "object", + "properties": { + "columns": {"type": "array", "items": {"type": "string"}}, + "rows": {"type": "array", "items": {"type": "array"}}, + "truncated": {"type": "boolean"}, + }, + "required": ["columns", "rows", "truncated"], + }, + }, +] + +TOOLS_BY_NAME = {tool["name"]: tool for tool in TOOLS} + + +def _jsonrpc_response(id, result): + return JsonResponse({"jsonrpc": "2.0", "id": id, "result": result}) + + +def _jsonrpc_error(id, code, message, status=200): + return JsonResponse( + {"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}}, + status=status, + ) + + +def _handle_initialize(params): + return { + "protocolVersion": MCP_PROTOCOL_VERSION, + "capabilities": {"tools": {}}, + "serverInfo": SERVER_INFO, + "instructions": SERVER_INSTRUCTIONS, + } + + +def _handle_tools_list(params): + return { + "tools": [ + { + "name": tool["name"], + "description": tool["description"], + "inputSchema": tool["inputSchema"], + "outputSchema": tool["outputSchema"], + "annotations": {"readOnlyHint": True, "openWorldHint": False}, + } + for tool in TOOLS + ] + } + + +def _handle_tools_call(params): + name = params.get("name") + tool = TOOLS_BY_NAME.get(name) + if tool is None: + raise KeyError("Unknown tool: {}".format(name)) + arguments = params.get("arguments") or {} + try: + structured = tool["handler"](**arguments) + except ToolError as e: + return { + "content": [{"type": "text", "text": str(e)}], + "isError": True, + } + return { + "content": [ + {"type": "text", "text": json.dumps(structured, default=str)}, + ], + "structuredContent": structured, + "isError": False, + } + + +METHODS = { + "initialize": _handle_initialize, + "ping": lambda params: {}, + "tools/list": _handle_tools_list, + "tools/call": _handle_tools_call, +} + + +@csrf_exempt +def mcp_endpoint(request): + if not request.user.is_authenticated: + return JsonResponse({"error": "Authentication required"}, status=401) + if not request.user.has_perm("django_sql_dashboard.execute_sql"): + return JsonResponse( + {"error": "You do not have permission to execute SQL"}, status=403 + ) + if request.method != "POST": + response = JsonResponse({"error": "Method not allowed"}, status=405) + response["Allow"] = "POST" + return response + try: + message = json.loads(request.body) + except ValueError: + return _jsonrpc_error(None, -32700, "Parse error", status=400) + if not isinstance(message, dict): + return _jsonrpc_error(None, -32600, "Invalid request", status=400) + method = message.get("method") + if "id" not in message: + # A notification - accept it without a response + return HttpResponse(status=202) + id = message["id"] + handler = METHODS.get(method) + if handler is None: + return _jsonrpc_error(id, -32601, "Method not found: {}".format(method)) + try: + result = handler(message.get("params") or {}) + except (TypeError, KeyError) as e: + return _jsonrpc_error(id, -32602, "Invalid params: {}".format(e)) + return _jsonrpc_response(id, result) diff --git a/django_sql_dashboard/urls.py b/django_sql_dashboard/urls.py index 3bc2eec..b86db7c 100644 --- a/django_sql_dashboard/urls.py +++ b/django_sql_dashboard/urls.py @@ -1,9 +1,11 @@ from django.urls import path +from .mcp import mcp_endpoint from .views import dashboard, dashboard_json, dashboard_index urlpatterns = [ path("", dashboard_index, name="django_sql_dashboard-index"), + path("-/mcp", mcp_endpoint, name="django_sql_dashboard-mcp"), path("/", dashboard, name="django_sql_dashboard-dashboard"), path(".json", dashboard_json, name="django_sql_dashboard-dashboard_json"), ] diff --git a/docs/index.md b/docs/index.md index 1cf05c3..b4f9a94 100644 --- a/docs/index.md +++ b/docs/index.md @@ -9,6 +9,7 @@ setup sql saved-dashboards widgets +mcp security contributing ``` diff --git a/docs/mcp.md b/docs/mcp.md new file mode 100644 index 0000000..5c47429 --- /dev/null +++ b/docs/mcp.md @@ -0,0 +1,50 @@ +# MCP server + +Django SQL Dashboard includes a [Model Context Protocol](https://modelcontextprotocol.io/) +(MCP) server, so LLM-powered tools such as Claude can explore your database +and answer questions by executing read-only SQL queries. + +The server is enabled automatically when you include `django_sql_dashboard.urls` +in your URL configuration. If the dashboard is mounted at `/dashboard/` the MCP +endpoint is: + + https://your-site.example.com/dashboard/-/mcp + +The endpoint implements the stateless variant of the MCP Streamable HTTP +transport, using JSON responses. Configure an MCP client to connect to that +URL. Note that the URL has no trailing slash. + +## Tools + +Three read-only tools are exposed: + +- `list_tables` lists the tables that are visible to the dashboard's database + connection. +- `get_schema` returns every table with its columns and their PostgreSQL + types, so a model can construct correct queries. +- `execute_sql` executes a single read-only SQL query and returns its columns + and rows. Queries can use `%(name)s` style parameters, with values passed + in a separate `parameters` argument. Results are truncated to the + `DASHBOARD_ROW_LIMIT` setting (default 100 rows) and a `truncated` flag + indicates when this has happened. + +## Authentication and security + +The MCP endpoint applies the same rules as the rest of the dashboard: + +- The request must come from a logged-in user with the + `django_sql_dashboard.execute_sql` permission. Unauthenticated requests + receive a 401 response; authenticated users without the permission + receive a 403 response. MCP clients need to authenticate in the same way + as the user's browser, for example by passing a valid session cookie. +- SQL executes against the `DASHBOARD_DB_ALIAS` database connection, inside + a transaction that is always rolled back, using the same protective + pattern as the dashboard itself. +- Queries containing `;` are rejected, so only one statement can run at a + time. + +As with the rest of Django SQL Dashboard, you should configure the dashboard +database connection to use a read-only PostgreSQL role with a statement +timeout - see [Security](security.md) for details. The MCP server deliberately +provides no way to run write queries, but the read-only database role is the +real enforcement mechanism. diff --git a/test_project/test_mcp.py b/test_project/test_mcp.py new file mode 100644 index 0000000..b616088 --- /dev/null +++ b/test_project/test_mcp.py @@ -0,0 +1,237 @@ +import json + +import pytest +from django.db import connections +from django.test.client import Client + +MCP_PATH = "/dashboard/-/mcp" + + +@pytest.fixture +def writable_dashboard_db(): + # Reconnect the dashboard alias with no connection options, so the + # transactional test teardown flush can TRUNCATE tables afterwards + connection = connections["dashboard"] + connection.close() + connection.settings_dict.setdefault("OPTIONS", {}).pop("options", None) + yield + connection.close() + + +@pytest.fixture +def read_only_dashboard_db(writable_dashboard_db): + # Reconnect the dashboard alias with the read-only option applied, to + # match the recommended production configuration + connection = connections["dashboard"] + options = connection.settings_dict["OPTIONS"] + options["options"] = "-c default_transaction_read_only=on" + yield + connection.close() + options.pop("options", None) + + +def rpc(client, method, params=None, id=1): + message = {"jsonrpc": "2.0", "id": id, "method": method} + if params is not None: + message["params"] = params + return client.post( + MCP_PATH, data=json.dumps(message), content_type="application/json" + ) + + +def call_tool(client, name, arguments=None): + response = rpc(client, "tools/call", {"name": name, "arguments": arguments or {}}) + assert response.status_code == 200 + return response.json()["result"] + + +def test_mcp_requires_authentication(client, dashboard_db): + response = rpc(client, "tools/list") + assert response.status_code == 401 + + +def test_mcp_requires_execute_sql_permission( + client, dashboard_db, django_user_model, execute_sql_permission +): + user = django_user_model.objects.create(username="mcp_user") + client.force_login(user) + assert rpc(client, "tools/list").status_code == 403 + # Now grant the permission + user.user_permissions.add(execute_sql_permission) + user = django_user_model.objects.get(pk=user.pk) # to clear permission cache + client.force_login(user) + assert rpc(client, "tools/list").status_code == 200 + + +def test_mcp_is_csrf_exempt(admin_client, dashboard_db, admin_user): + csrf_client = Client(enforce_csrf_checks=True) + csrf_client.force_login(admin_user) + assert rpc(csrf_client, "ping").status_code == 200 + + +def test_initialize(admin_client, dashboard_db): + response = rpc(admin_client, "initialize") + assert response.status_code == 200 + result = response.json()["result"] + assert result["protocolVersion"] == "2025-06-18" + assert result["capabilities"] == {"tools": {}} + assert result["serverInfo"]["name"] == "django-sql-dashboard" + assert result["instructions"] + + +def test_notifications_are_accepted(admin_client, dashboard_db): + response = admin_client.post( + MCP_PATH, + data=json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}), + content_type="application/json", + ) + assert response.status_code == 202 + + +def test_unknown_method_returns_error(admin_client, dashboard_db): + response = rpc(admin_client, "resources/list") + assert response.status_code == 200 + error = response.json()["error"] + assert error["code"] == -32601 + + +def test_invalid_json_returns_parse_error(admin_client, dashboard_db): + response = admin_client.post( + MCP_PATH, data="this is not json", content_type="application/json" + ) + assert response.status_code == 400 + assert response.json()["error"]["code"] == -32700 + + +def test_get_is_not_allowed(admin_client, dashboard_db): + response = admin_client.get(MCP_PATH) + assert response.status_code == 405 + assert response["Allow"] == "POST" + + +def test_tools_list(admin_client, dashboard_db): + response = rpc(admin_client, "tools/list") + assert response.status_code == 200 + tools = response.json()["result"]["tools"] + assert [tool["name"] for tool in tools] == [ + "list_tables", + "get_schema", + "execute_sql", + ] + for tool in tools: + assert tool["description"] + assert tool["inputSchema"] + assert tool["outputSchema"] + assert tool["annotations"] == {"readOnlyHint": True, "openWorldHint": False} + + +def test_list_tables(admin_client, dashboard_db): + result = call_tool(admin_client, "list_tables") + assert result["isError"] is False + tables = result["structuredContent"]["tables"] + assert "django_sql_dashboard_dashboard" in tables + assert tables == sorted(tables) + + +def test_get_schema(admin_client, dashboard_db): + result = call_tool(admin_client, "get_schema") + assert result["isError"] is False + schema = result["structuredContent"]["schema"] + assert "django_sql_dashboard_dashboard (" in schema + assert "slug character varying" in schema + + +def test_execute_sql(admin_client, dashboard_db): + result = call_tool( + admin_client, "execute_sql", {"sql": "select 1 + 2 as total, 'hi' as greeting"} + ) + assert result["isError"] is False + assert result["structuredContent"] == { + "columns": ["total", "greeting"], + "rows": [[3, "hi"]], + "truncated": False, + } + # The text content should be the JSON-encoded structured content + assert json.loads(result["content"][0]["text"]) == result["structuredContent"] + + +def test_execute_sql_with_parameters(admin_client, dashboard_db): + result = call_tool( + admin_client, + "execute_sql", + { + "sql": "select %(name)s as name", + "parameters": {"name": "Cleo"}, + }, + ) + assert result["structuredContent"]["rows"] == [["Cleo"]] + + +def test_execute_sql_truncates_at_row_limit(admin_client, dashboard_db, settings): + settings.DASHBOARD_ROW_LIMIT = 5 + result = call_tool( + admin_client, "execute_sql", {"sql": "select * from generate_series(1, 10)"} + ) + assert result["structuredContent"]["rows"] == [[1], [2], [3], [4], [5]] + assert result["structuredContent"]["truncated"] is True + + +@pytest.mark.parametrize( + "sql,expected_error", + ( + ("select 1; select 2", "';' not allowed in SQL queries"), + ("", "SQL query is required"), + ("select 100% of results", "Invalid query"), + ("select * from no_such_table", 'relation "no_such_table" does not exist'), + ), +) +def test_execute_sql_errors(admin_client, dashboard_db, sql, expected_error): + result = call_tool(admin_client, "execute_sql", {"sql": sql}) + assert result["isError"] is True + assert expected_error in result["content"][0]["text"] + + +@pytest.mark.django_db(databases=["default", "dashboard"], transaction=True) +@pytest.mark.parametrize( + "sql", + ( + "create table forbidden (id integer)", + "delete from auth_user", + "update auth_user set username = 'hacked'", + ), +) +def test_execute_sql_rejects_writes_on_read_only_connection( + admin_client, read_only_dashboard_db, sql +): + result = call_tool(admin_client, "execute_sql", {"sql": sql}) + assert result["isError"] is True + assert "read-only transaction" in result["content"][0]["text"] + # A subsequent valid read still succeeds + result = call_tool(admin_client, "execute_sql", {"sql": "select 1 as value"}) + assert result["structuredContent"]["rows"] == [[1]] + + +@pytest.mark.django_db(databases=["default", "dashboard"], transaction=True) +def test_execute_sql_writes_are_rolled_back(admin_client, writable_dashboard_db): + # Even without a read-only connection, the wrapping transaction is + # always rolled back so writes never stick + from django_sql_dashboard.models import Dashboard + + Dashboard.objects.create(slug="rollback-test") + count_sql = "select count(*) from django_sql_dashboard_dashboard" + result = call_tool(admin_client, "execute_sql", {"sql": count_sql}) + count = result["structuredContent"]["rows"][0][0] + assert count == 1 + call_tool( + admin_client, + "execute_sql", + {"sql": "delete from django_sql_dashboard_dashboard"}, + ) + result = call_tool(admin_client, "execute_sql", {"sql": count_sql}) + assert result["structuredContent"]["rows"][0][0] == count + + +def test_unknown_tool_returns_invalid_params(admin_client, dashboard_db): + response = rpc(admin_client, "tools/call", {"name": "drop_tables", "arguments": {}}) + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32602 From 6f0549fd5d092deb8134d9611627df7b40f4c99f Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 1 Aug 2026 05:15:58 +0000 Subject: [PATCH 2/4] Add DASHBOARD_MCP_TOKENS for MCP bearer token authentication Headless MCP clients cannot easily obtain a Django session cookie. The new optional DASHBOARD_MCP_TOKENS setting maps secret tokens to usernames, letting clients authenticate to the /-/mcp endpoint with an Authorization: Bearer header. - Tokens are compared with secrets.compare_digest, resolved with get_by_natural_key so custom user models work, and rejected for missing or inactive users - The mapped user must still hold the execute_sql permission - A request carrying a Bearer header never falls back to session auth - Documented in docs/mcp.md and the settings list in docs/setup.md Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017ZRFpnuD9tZKWb19Ffg8iF --- django_sql_dashboard/mcp.py | 42 +++++++++++++++++-- docs/mcp.md | 34 +++++++++++++-- docs/setup.md | 1 + test_project/test_mcp.py | 83 ++++++++++++++++++++++++++++++++++++- 4 files changed, 152 insertions(+), 8 deletions(-) diff --git a/django_sql_dashboard/mcp.py b/django_sql_dashboard/mcp.py index 5061d58..bc2cf30 100644 --- a/django_sql_dashboard/mcp.py +++ b/django_sql_dashboard/mcp.py @@ -19,8 +19,10 @@ """ import json +import secrets from django.conf import settings +from django.contrib.auth import get_user_model from django.db import connections from django.db.utils import ProgrammingError from django.http import HttpResponse, JsonResponse @@ -293,11 +295,45 @@ def _handle_tools_call(params): } +def _user_for_bearer_token(token): + # Returns the user for a DASHBOARD_MCP_TOKENS token, or None + tokens = getattr(settings, "DASHBOARD_MCP_TOKENS", None) or {} + username = None + for configured_token, configured_username in tokens.items(): + # compare_digest to avoid leaking token prefixes via timing + if secrets.compare_digest(str(configured_token), token): + username = configured_username + if username is None: + return None + UserModel = get_user_model() + try: + user = UserModel._default_manager.get_by_natural_key(username) + except UserModel.DoesNotExist: + return None + if not user.is_active: + return None + return user + + +def _authenticate(request): + # Returns (user, error_response) - exactly one is not None + authorization = request.headers.get("Authorization", "") + if authorization.startswith("Bearer "): + user = _user_for_bearer_token(authorization[len("Bearer ") :].strip()) + if user is None: + return None, JsonResponse({"error": "Invalid token"}, status=401) + return user, None + if request.user.is_authenticated: + return request.user, None + return None, JsonResponse({"error": "Authentication required"}, status=401) + + @csrf_exempt def mcp_endpoint(request): - if not request.user.is_authenticated: - return JsonResponse({"error": "Authentication required"}, status=401) - if not request.user.has_perm("django_sql_dashboard.execute_sql"): + user, error_response = _authenticate(request) + if error_response is not None: + return error_response + if not user.has_perm("django_sql_dashboard.execute_sql"): return JsonResponse( {"error": "You do not have permission to execute SQL"}, status=403 ) diff --git a/docs/mcp.md b/docs/mcp.md index 5c47429..ad35707 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -32,11 +32,12 @@ Three read-only tools are exposed: The MCP endpoint applies the same rules as the rest of the dashboard: -- The request must come from a logged-in user with the +- The request must be authenticated as a user with the `django_sql_dashboard.execute_sql` permission. Unauthenticated requests receive a 401 response; authenticated users without the permission - receive a 403 response. MCP clients need to authenticate in the same way - as the user's browser, for example by passing a valid session cookie. + receive a 403 response. MCP clients can authenticate in the same way + as the user's browser, for example by passing a valid session cookie, + or using a token - see below. - SQL executes against the `DASHBOARD_DB_ALIAS` database connection, inside a transaction that is always rolled back, using the same protective pattern as the dashboard itself. @@ -48,3 +49,30 @@ database connection to use a read-only PostgreSQL role with a statement timeout - see [Security](security.md) for details. The MCP server deliberately provides no way to run write queries, but the read-only database role is the real enforcement mechanism. + +## Token authentication + +Headless MCP clients usually cannot log in through a browser to obtain a +session cookie. To support them, the optional `DASHBOARD_MCP_TOKENS` setting +maps secret tokens to usernames: + +```python +DASHBOARD_MCP_TOKENS = { + "your-secret-token": "username", +} +``` + +An MCP client can then authenticate by sending that token in an +`Authorization` header: + + Authorization: Bearer your-secret-token + +A request with a valid token is treated as coming from the corresponding +user, who must still have the `django_sql_dashboard.execute_sql` permission. +Tokens for missing or inactive users are rejected, and if a `Bearer` header +is present it must be valid - an invalid token is never ignored in favor of +the request's session cookie. + +Treat these tokens like passwords: each one grants the full dashboard SQL +access of its user. Keep them out of source control (for example by loading +them from environment variables) and rotate them by changing the setting. diff --git a/docs/setup.md b/docs/setup.md index c97f0ec..4fbe937 100644 --- a/docs/setup.md +++ b/docs/setup.md @@ -152,6 +152,7 @@ You can customize the following settings in Django's `settings.py` module: - `DASHBOARD_UPGRADE_OLD_BASE64_LINKS` - prior to version 0.8a0 SQL URLs used base64-encoded JSON. If you set this to `True` any hits that include those old URLs will be automatically redirected to the upgraded new version. Use this if you have an existing installation of `django-sql-dashboard` that people already have saved bookmarks for. - `DASHBOARD_ENABLE_FULL_EXPORT` - set this to `True` to enable the full results CSV/TSV export feature. It defaults to `False`. Enable this feature only if you are confident that the database alias you are using does not have write permissions to anything. - `DASHBOARD_DISABLE_JSON` - set to `True` to disable the feature where `/dashboard/name-of-dashboard.json` provides a JSON representation of the dashboard. This defaults to `False`. +- `DASHBOARD_MCP_TOKENS` - a dictionary mapping secret tokens to usernames, used to authenticate MCP clients that cannot use a session cookie. See [MCP server](mcp.md). ## Custom templates diff --git a/test_project/test_mcp.py b/test_project/test_mcp.py index b616088..1872af0 100644 --- a/test_project/test_mcp.py +++ b/test_project/test_mcp.py @@ -30,12 +30,15 @@ def read_only_dashboard_db(writable_dashboard_db): options.pop("options", None) -def rpc(client, method, params=None, id=1): +def rpc(client, method, params=None, id=1, headers=None): message = {"jsonrpc": "2.0", "id": id, "method": method} if params is not None: message["params"] = params return client.post( - MCP_PATH, data=json.dumps(message), content_type="application/json" + MCP_PATH, + data=json.dumps(message), + content_type="application/json", + headers=headers, ) @@ -63,6 +66,82 @@ def test_mcp_requires_execute_sql_permission( assert rpc(client, "tools/list").status_code == 200 +def test_mcp_token_authentication( + client, dashboard_db, settings, django_user_model, execute_sql_permission +): + user = django_user_model.objects.create(username="token_user") + user.user_permissions.add(execute_sql_permission) + settings.DASHBOARD_MCP_TOKENS = {"correct-token": "token_user"} + headers = {"authorization": "Bearer correct-token"} + response = rpc(client, "tools/list", headers=headers) + assert response.status_code == 200 + # And SQL can be executed + response = rpc( + client, + "tools/call", + {"name": "execute_sql", "arguments": {"sql": "select 1 as one"}}, + headers=headers, + ) + assert response.status_code == 200 + assert response.json()["result"]["structuredContent"]["rows"] == [[1]] + + +@pytest.mark.parametrize( + "authorization", + ( + "Bearer wrong-token", + "Bearer ", + "Bearer correct-token-with-suffix", + ), +) +def test_mcp_token_invalid_tokens_are_rejected( + client, dashboard_db, settings, django_user_model, authorization +): + django_user_model.objects.create(username="token_user") + settings.DASHBOARD_MCP_TOKENS = {"correct-token": "token_user"} + response = rpc(client, "tools/list", headers={"authorization": authorization}) + assert response.status_code == 401 + + +def test_mcp_token_for_missing_user_is_rejected(client, dashboard_db, settings): + settings.DASHBOARD_MCP_TOKENS = {"correct-token": "no_such_user"} + response = rpc( + client, "tools/list", headers={"authorization": "Bearer correct-token"} + ) + assert response.status_code == 401 + + +def test_mcp_token_for_inactive_user_is_rejected( + client, dashboard_db, settings, django_user_model, execute_sql_permission +): + user = django_user_model.objects.create(username="inactive_user", is_active=False) + user.user_permissions.add(execute_sql_permission) + settings.DASHBOARD_MCP_TOKENS = {"correct-token": "inactive_user"} + response = rpc( + client, "tools/list", headers={"authorization": "Bearer correct-token"} + ) + assert response.status_code == 401 + + +def test_mcp_token_user_still_needs_execute_sql_permission( + client, dashboard_db, settings, django_user_model +): + django_user_model.objects.create(username="powerless_user") + settings.DASHBOARD_MCP_TOKENS = {"correct-token": "powerless_user"} + response = rpc( + client, "tools/list", headers={"authorization": "Bearer correct-token"} + ) + assert response.status_code == 403 + + +def test_mcp_bearer_header_does_not_fall_back_to_session(admin_client, dashboard_db): + # A logged-in session with an invalid Bearer token is still rejected + response = rpc( + admin_client, "tools/list", headers={"authorization": "Bearer wrong-token"} + ) + assert response.status_code == 401 + + def test_mcp_is_csrf_exempt(admin_client, dashboard_db, admin_user): csrf_client = Client(enforce_csrf_checks=True) csrf_client.force_login(admin_user) From eab725883842354750e2de3d06d5bed6db8e340b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 1 Aug 2026 05:24:28 +0000 Subject: [PATCH 3/4] Add MCP dashboard tools with anonymous access to public dashboards Two new MCP tools that follow saved dashboard view policies instead of requiring the execute_sql permission: - list_dashboards returns the dashboards the current user can view. Unlisted dashboards are excluded, matching the dashboard index page. - execute_dashboard runs every query saved on a dashboard by slug and returns per-query results or errors, with %(name)s parameter support. Unlisted dashboards work for anyone who knows the slug; dashboards that are missing or not visible produce the same error so private slugs are not disclosed. This makes the MCP endpoint useful to anonymous clients with no credentials at all: they can list and execute public dashboards. The SQL tools (list_tables, get_schema, execute_sql) still require the execute_sql permission, now enforced per tool call, and tools/list only advertises the tools the current user is allowed to use. The dashboard view policy logic is extracted from the dashboard view into a new Dashboard.user_can_view() model method shared by both the web view and the MCP tools. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017ZRFpnuD9tZKWb19Ffg8iF --- django_sql_dashboard/mcp.py | 201 +++++++++++++++++++++++++++++---- django_sql_dashboard/models.py | 20 ++++ django_sql_dashboard/views.py | 30 +---- docs/mcp.md | 36 +++++- test_project/test_mcp.py | 164 +++++++++++++++++++++++++-- 5 files changed, 386 insertions(+), 65 deletions(-) diff --git a/django_sql_dashboard/mcp.py b/django_sql_dashboard/mcp.py index bc2cf30..a3a8bc1 100644 --- a/django_sql_dashboard/mcp.py +++ b/django_sql_dashboard/mcp.py @@ -3,17 +3,22 @@ This implements the stateless variant of the MCP Streamable HTTP transport as a single Django view, without any additional dependencies. It exposes -three read-only tools: +five read-only tools: - list_tables - list the tables visible to the dashboard connection - get_schema - return the schema (tables, columns, types) for the database - execute_sql - execute one read-only SQL query and return its results +- list_dashboards - list saved dashboards visible to the current user +- execute_dashboard - execute the queries saved on a dashboard SQL execution uses the same protected path as the dashboard views: the read-only "dashboard" database alias, a transaction that is rolled back, the ``DASHBOARD_ROW_LIMIT`` row limit and the same named parameter support. -Callers must be authenticated and have the -``django_sql_dashboard.execute_sql`` permission. +The first three tools require an authenticated user with the +``django_sql_dashboard.execute_sql`` permission. The dashboard tools follow +each dashboard's view policy instead, so anonymous MCP clients with no +credentials can list and execute public dashboards, and can execute +unlisted dashboards if they know the slug. Inspired by https://github.com/datasette/datasette-mcp """ @@ -28,6 +33,7 @@ from django.http import HttpResponse, JsonResponse from django.views.decorators.csrf import csrf_exempt +from .models import Dashboard from .utils import displayable_rows, extract_named_parameters MCP_PROTOCOL_VERSION = "2025-06-18" @@ -38,10 +44,14 @@ } SERVER_INSTRUCTIONS = ( - "Call list_tables, then get_schema to see columns and types, " - "then execute_sql to answer questions using read-only PostgreSQL." + "Call list_dashboards to see saved dashboards and execute_dashboard to " + "run one. If the SQL tools are available, call list_tables, then " + "get_schema to see columns and types, then execute_sql to answer " + "questions using read-only PostgreSQL." ) +PERMISSION_DENIED_ERROR = "You do not have permission to execute SQL" + TABLES_AND_COLUMNS_SQL = """ select information_schema.columns.table_name, @@ -82,11 +92,11 @@ def _tables_and_columns(): return tables -def tool_list_tables(): +def tool_list_tables(user): return {"tables": [table for table, _ in _tables_and_columns()]} -def tool_get_schema(): +def tool_get_schema(user): blocks = [] for table, columns in _tables_and_columns(): column_lines = ",\n".join( @@ -106,7 +116,7 @@ def _serialize_cell(value): return str(value) -def tool_execute_sql(sql, parameters=None): +def _execute_query(sql, parameters=None): sql = sql.strip().rstrip(";") if not sql: raise ToolError("SQL query is required") @@ -149,11 +159,59 @@ def tool_execute_sql(sql, parameters=None): } +def tool_execute_sql(user, sql, parameters=None): + return _execute_query(sql, parameters) + + +def tool_list_dashboards(user): + if user.is_authenticated: + dashboards = Dashboard.get_visible_to_user(user) + else: + dashboards = Dashboard.objects.filter( + view_policy=Dashboard.ViewPolicies.PUBLIC + ).order_by("slug") + return { + "dashboards": [ + { + "slug": dashboard.slug, + "title": dashboard.title, + "description": dashboard.description, + } + for dashboard in dashboards + ] + } + + +def tool_execute_dashboard(user, slug, parameters=None): + try: + dashboard = Dashboard.objects.get(slug=slug) + except Dashboard.DoesNotExist: + dashboard = None + if dashboard is None or not dashboard.user_can_view(user): + raise ToolError( + "Dashboard '{}' does not exist or is not available".format(slug) + ) + queries = [] + for query in dashboard.queries.all(): + try: + result = _execute_query(query.sql, parameters) + except ToolError as e: + result = {"error": str(e)} + queries.append(dict({"sql": query.sql}, **result)) + return { + "slug": dashboard.slug, + "title": dashboard.title, + "description": dashboard.description, + "queries": queries, + } + + TOOLS = [ { "name": "list_tables", "description": "List the tables available to SQL queries in this dashboard.", "handler": tool_list_tables, + "requires_execute_sql": True, "inputSchema": { "type": "object", "properties": {}, @@ -175,6 +233,7 @@ def tool_execute_sql(sql, parameters=None): "constructing a SQL query." ), "handler": tool_get_schema, + "requires_execute_sql": True, "inputSchema": { "type": "object", "properties": {}, @@ -197,6 +256,7 @@ def tool_execute_sql(sql, parameters=None): "not allowed. Results are truncated to the dashboard row limit." ), "handler": tool_execute_sql, + "requires_execute_sql": True, "inputSchema": { "type": "object", "properties": { @@ -225,6 +285,95 @@ def tool_execute_sql(sql, parameters=None): "required": ["columns", "rows", "truncated"], }, }, + { + "name": "list_dashboards", + "description": ( + "List the saved dashboards that are visible to the current user. " + "Unlisted dashboards are not included, but can still be executed " + "by slug with execute_dashboard." + ), + "handler": tool_list_dashboards, + "requires_execute_sql": False, + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + "outputSchema": { + "type": "object", + "properties": { + "dashboards": { + "type": "array", + "items": { + "type": "object", + "properties": { + "slug": {"type": "string"}, + "title": {"type": "string"}, + "description": {"type": "string"}, + }, + "required": ["slug", "title", "description"], + }, + }, + }, + "required": ["dashboards"], + }, + }, + { + "name": "execute_dashboard", + "description": ( + "Execute every SQL query saved on a dashboard and return their " + "results. Use the slug from list_dashboards, or a known slug for " + "an unlisted dashboard. Pass values for any %(name)s parameters " + "used by the dashboard's queries in the parameters argument." + ), + "handler": tool_execute_dashboard, + "requires_execute_sql": False, + "inputSchema": { + "type": "object", + "properties": { + "slug": { + "type": "string", + "description": "The slug identifying the dashboard", + }, + "parameters": { + "type": "object", + "description": ( + "Values for any %(name)s parameters used by the " + "dashboard's queries" + ), + "additionalProperties": {"type": "string"}, + }, + }, + "required": ["slug"], + "additionalProperties": False, + }, + "outputSchema": { + "type": "object", + "properties": { + "slug": {"type": "string"}, + "title": {"type": "string"}, + "description": {"type": "string"}, + "queries": { + "type": "array", + "items": { + "type": "object", + "properties": { + "sql": {"type": "string"}, + "columns": { + "type": "array", + "items": {"type": "string"}, + }, + "rows": {"type": "array", "items": {"type": "array"}}, + "truncated": {"type": "boolean"}, + "error": {"type": "string"}, + }, + "required": ["sql"], + }, + }, + }, + "required": ["slug", "title", "description", "queries"], + }, + }, ] TOOLS_BY_NAME = {tool["name"]: tool for tool in TOOLS} @@ -241,7 +390,7 @@ def _jsonrpc_error(id, code, message, status=200): ) -def _handle_initialize(params): +def _handle_initialize(user, params): return { "protocolVersion": MCP_PROTOCOL_VERSION, "capabilities": {"tools": {}}, @@ -250,7 +399,14 @@ def _handle_initialize(params): } -def _handle_tools_list(params): +def _available_tools(user): + can_execute_sql = user.has_perm("django_sql_dashboard.execute_sql") + return [ + tool for tool in TOOLS if can_execute_sql or not tool["requires_execute_sql"] + ] + + +def _handle_tools_list(user, params): return { "tools": [ { @@ -260,19 +416,23 @@ def _handle_tools_list(params): "outputSchema": tool["outputSchema"], "annotations": {"readOnlyHint": True, "openWorldHint": False}, } - for tool in TOOLS + for tool in _available_tools(user) ] } -def _handle_tools_call(params): +def _handle_tools_call(user, params): name = params.get("name") tool = TOOLS_BY_NAME.get(name) if tool is None: raise KeyError("Unknown tool: {}".format(name)) arguments = params.get("arguments") or {} try: - structured = tool["handler"](**arguments) + if tool["requires_execute_sql"] and not user.has_perm( + "django_sql_dashboard.execute_sql" + ): + raise ToolError(PERMISSION_DENIED_ERROR) + structured = tool["handler"](user, **arguments) except ToolError as e: return { "content": [{"type": "text", "text": str(e)}], @@ -289,7 +449,7 @@ def _handle_tools_call(params): METHODS = { "initialize": _handle_initialize, - "ping": lambda params: {}, + "ping": lambda user, params: {}, "tools/list": _handle_tools_list, "tools/call": _handle_tools_call, } @@ -316,16 +476,15 @@ def _user_for_bearer_token(token): def _authenticate(request): - # Returns (user, error_response) - exactly one is not None + # Returns (user, error_response) - exactly one is not None. The user + # may be anonymous: public dashboards work with no credentials at all. authorization = request.headers.get("Authorization", "") if authorization.startswith("Bearer "): user = _user_for_bearer_token(authorization[len("Bearer ") :].strip()) if user is None: return None, JsonResponse({"error": "Invalid token"}, status=401) return user, None - if request.user.is_authenticated: - return request.user, None - return None, JsonResponse({"error": "Authentication required"}, status=401) + return request.user, None @csrf_exempt @@ -333,10 +492,6 @@ def mcp_endpoint(request): user, error_response = _authenticate(request) if error_response is not None: return error_response - if not user.has_perm("django_sql_dashboard.execute_sql"): - return JsonResponse( - {"error": "You do not have permission to execute SQL"}, status=403 - ) if request.method != "POST": response = JsonResponse({"error": "Method not allowed"}, status=405) response["Allow"] = "POST" @@ -356,7 +511,7 @@ def mcp_endpoint(request): if handler is None: return _jsonrpc_error(id, -32601, "Method not found: {}".format(method)) try: - result = handler(message.get("params") or {}) + result = handler(user, message.get("params") or {}) except (TypeError, KeyError) as e: return _jsonrpc_error(id, -32602, "Invalid params: {}".format(e)) return _jsonrpc_response(id, result) diff --git a/django_sql_dashboard/models.py b/django_sql_dashboard/models.py index 3dd7007..195ade4 100644 --- a/django_sql_dashboard/models.py +++ b/django_sql_dashboard/models.py @@ -84,6 +84,26 @@ def get_edit_url(self): class Meta: permissions = [("execute_sql", "Can execute arbitrary SQL queries")] + def user_can_view(self, user): + policy = self.view_policy + if policy in (self.ViewPolicies.PUBLIC, self.ViewPolicies.UNLISTED): + return True + if policy == self.ViewPolicies.PRIVATE: + return user == self.owned_by + if not user.is_authenticated: + return False + if user == self.owned_by: + return True + if policy == self.ViewPolicies.LOGGEDIN: + return True + if policy == self.ViewPolicies.GROUP: + return user.groups.filter(pk=self.view_group_id).exists() + if policy == self.ViewPolicies.STAFF: + return user.is_staff + if policy == self.ViewPolicies.SUPERUSER: + return user.is_superuser + return False + def user_can_edit(self, user): if not user: return False diff --git a/django_sql_dashboard/views.py b/django_sql_dashboard/views.py index d9ae2e0..7719eb2 100644 --- a/django_sql_dashboard/views.py +++ b/django_sql_dashboard/views.py @@ -390,32 +390,10 @@ def dashboard_json(request, slug): def dashboard(request, slug, json_mode=False): dashboard = get_object_or_404(Dashboard, slug=slug) # Can current user see it, based on view_policy? - view_policy = dashboard.view_policy - owner = dashboard.owned_by - denied = HttpResponseForbidden("You cannot access this dashboard") - denied["cache-control"] = "private" - if view_policy == Dashboard.ViewPolicies.PRIVATE: - if request.user != owner: - return denied - elif view_policy == Dashboard.ViewPolicies.LOGGEDIN: - if not request.user.is_authenticated: - return denied - elif view_policy == Dashboard.ViewPolicies.GROUP: - if (not request.user.is_authenticated) or not ( - request.user == owner - or request.user.groups.filter(pk=dashboard.view_group_id).exists() - ): - return denied - elif view_policy == Dashboard.ViewPolicies.STAFF: - if (not request.user.is_authenticated) or ( - request.user != owner and not request.user.is_staff - ): - return denied - elif view_policy == Dashboard.ViewPolicies.SUPERUSER: - if (not request.user.is_authenticated) or ( - request.user != owner and not request.user.is_superuser - ): - return denied + if not dashboard.user_can_view(request.user): + denied = HttpResponseForbidden("You cannot access this dashboard") + denied["cache-control"] = "private" + return denied return _dashboard_index( request, sql_queries=[query.sql for query in dashboard.queries.all()], diff --git a/docs/mcp.md b/docs/mcp.md index ad35707..7bde167 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -16,7 +16,10 @@ URL. Note that the URL has no trailing slash. ## Tools -Three read-only tools are exposed: +Five read-only tools are exposed. + +Three SQL tools, available to users with the +`django_sql_dashboard.execute_sql` permission: - `list_tables` lists the tables that are visible to the dashboard's database connection. @@ -28,16 +31,36 @@ Three read-only tools are exposed: `DASHBOARD_ROW_LIMIT` setting (default 100 rows) and a `truncated` flag indicates when this has happened. +And two [saved dashboard](saved-dashboards.md) tools, which follow each +dashboard's view policy rather than requiring the `execute_sql` permission: + +- `list_dashboards` lists the dashboards the current user is allowed to + view: their slug, title and description. Unlisted dashboards are not + included, matching how they are hidden from the dashboard index page. +- `execute_dashboard` executes every query saved on a dashboard, identified + by its slug, and returns the results of each query - or a per-query error + message if one of them fails. Values for `%(name)s` parameters used by + the dashboard's queries can be passed in the `parameters` argument. + Unlisted dashboards can be executed by anyone who knows their slug. + +Because the dashboard tools follow view policies, an anonymous MCP client +with no credentials at all can connect and use dashboards that have been +made public. `tools/list` only advertises the tools the current user is +allowed to use, so anonymous clients see just the two dashboard tools. + ## Authentication and security The MCP endpoint applies the same rules as the rest of the dashboard: -- The request must be authenticated as a user with the - `django_sql_dashboard.execute_sql` permission. Unauthenticated requests - receive a 401 response; authenticated users without the permission - receive a 403 response. MCP clients can authenticate in the same way +- The SQL tools require the request to be authenticated as a user with the + `django_sql_dashboard.execute_sql` permission - calling them without it + returns a tool error. MCP clients can authenticate in the same way as the user's browser, for example by passing a valid session cookie, or using a token - see below. +- The dashboard tools check the view policy of each dashboard, exactly like + the dashboard pages themselves: a dashboard that is not visible to the + current user is reported as not existing, whether it is missing or merely + private. - SQL executes against the `DASHBOARD_DB_ALIAS` database connection, inside a transaction that is always rolled back, using the same protective pattern as the dashboard itself. @@ -68,7 +91,8 @@ An MCP client can then authenticate by sending that token in an Authorization: Bearer your-secret-token A request with a valid token is treated as coming from the corresponding -user, who must still have the `django_sql_dashboard.execute_sql` permission. +user, who must still have the `django_sql_dashboard.execute_sql` permission +to use the SQL tools. Tokens for missing or inactive users are rejected, and if a `Bearer` header is present it must be valid - an invalid token is never ignored in favor of the request's session cookie. diff --git a/test_project/test_mcp.py b/test_project/test_mcp.py index 1872af0..55b0de7 100644 --- a/test_project/test_mcp.py +++ b/test_project/test_mcp.py @@ -4,6 +4,8 @@ from django.db import connections from django.test.client import Client +from django_sql_dashboard.models import Dashboard + MCP_PATH = "/dashboard/-/mcp" @@ -48,22 +50,38 @@ def call_tool(client, name, arguments=None): return response.json()["result"] -def test_mcp_requires_authentication(client, dashboard_db): +def tool_names(client): response = rpc(client, "tools/list") - assert response.status_code == 401 + assert response.status_code == 200 + return [tool["name"] for tool in response.json()["result"]["tools"]] + +def test_anonymous_clients_only_see_dashboard_tools(client, dashboard_db): + assert tool_names(client) == ["list_dashboards", "execute_dashboard"] -def test_mcp_requires_execute_sql_permission( + +def test_sql_tools_require_execute_sql_permission( client, dashboard_db, django_user_model, execute_sql_permission ): user = django_user_model.objects.create(username="mcp_user") client.force_login(user) - assert rpc(client, "tools/list").status_code == 403 + # Without the permission: dashboard tools only, SQL calls are refused + assert tool_names(client) == ["list_dashboards", "execute_dashboard"] + result = call_tool(client, "execute_sql", {"sql": "select 1"}) + assert result["isError"] is True + assert "You do not have permission to execute SQL" in result["content"][0]["text"] # Now grant the permission user.user_permissions.add(execute_sql_permission) - user = django_user_model.objects.get(pk=user.pk) # to clear permission cache client.force_login(user) - assert rpc(client, "tools/list").status_code == 200 + assert tool_names(client) == [ + "list_tables", + "get_schema", + "execute_sql", + "list_dashboards", + "execute_dashboard", + ] + result = call_tool(client, "execute_sql", {"sql": "select 1 as one"}) + assert result["structuredContent"]["rows"] == [[1]] def test_mcp_token_authentication( @@ -128,10 +146,17 @@ def test_mcp_token_user_still_needs_execute_sql_permission( ): django_user_model.objects.create(username="powerless_user") settings.DASHBOARD_MCP_TOKENS = {"correct-token": "powerless_user"} + headers = {"authorization": "Bearer correct-token"} response = rpc( - client, "tools/list", headers={"authorization": "Bearer correct-token"} + client, + "tools/call", + {"name": "execute_sql", "arguments": {"sql": "select 1"}}, + headers=headers, ) - assert response.status_code == 403 + assert response.status_code == 200 + result = response.json()["result"] + assert result["isError"] is True + assert "You do not have permission to execute SQL" in result["content"][0]["text"] def test_mcp_bearer_header_does_not_fall_back_to_session(admin_client, dashboard_db): @@ -196,6 +221,8 @@ def test_tools_list(admin_client, dashboard_db): "list_tables", "get_schema", "execute_sql", + "list_dashboards", + "execute_dashboard", ] for tool in tools: assert tool["description"] @@ -294,8 +321,6 @@ def test_execute_sql_rejects_writes_on_read_only_connection( def test_execute_sql_writes_are_rolled_back(admin_client, writable_dashboard_db): # Even without a read-only connection, the wrapping transaction is # always rolled back so writes never stick - from django_sql_dashboard.models import Dashboard - Dashboard.objects.create(slug="rollback-test") count_sql = "select count(*) from django_sql_dashboard_dashboard" result = call_tool(admin_client, "execute_sql", {"sql": count_sql}) @@ -314,3 +339,122 @@ def test_unknown_tool_returns_invalid_params(admin_client, dashboard_db): response = rpc(admin_client, "tools/call", {"name": "drop_tables", "arguments": {}}) assert response.status_code == 200 assert response.json()["error"]["code"] == -32602 + + +def test_list_dashboards_anonymous_sees_only_public(client, dashboard_db): + Dashboard.objects.create(slug="public-one", title="Public", view_policy="public") + Dashboard.objects.create(slug="secret", view_policy="unlisted") + Dashboard.objects.create(slug="private-one", view_policy="private") + Dashboard.objects.create(slug="members", view_policy="loggedin") + result = call_tool(client, "list_dashboards") + assert result["isError"] is False + assert result["structuredContent"] == { + "dashboards": [ + {"slug": "public-one", "title": "Public", "description": ""}, + ] + } + + +def test_list_dashboards_logged_in(client, dashboard_db, django_user_model): + user = django_user_model.objects.create(username="viewer") + Dashboard.objects.create(slug="public-one", view_policy="public") + Dashboard.objects.create(slug="members", view_policy="loggedin") + Dashboard.objects.create(slug="secret", view_policy="unlisted") + Dashboard.objects.create(slug="mine", view_policy="private", owned_by=user) + client.force_login(user) + result = call_tool(client, "list_dashboards") + slugs = [d["slug"] for d in result["structuredContent"]["dashboards"]] + # Visible: loggedin, public and their own private - but never unlisted + assert sorted(slugs) == ["members", "mine", "public-one"] + + +def test_execute_dashboard_anonymous_public(client, dashboard_db, saved_dashboard): + result = call_tool(client, "execute_dashboard", {"slug": "test"}) + assert result["isError"] is False + assert result["structuredContent"] == { + "slug": "test", + "title": "Test dashboard", + "description": "This [supports markdown](http://example.com/)", + "queries": [ + { + "sql": "select 11 + 33", + "columns": ["?column?"], + "rows": [[44]], + "truncated": False, + }, + { + "sql": "select 22 + 55", + "columns": ["?column?"], + "rows": [[77]], + "truncated": False, + }, + ], + } + + +def test_execute_dashboard_unlisted_works_by_slug(client, dashboard_db): + dashboard = Dashboard.objects.create(slug="secret", view_policy="unlisted") + dashboard.queries.create(sql="select 1 as one") + result = call_tool(client, "execute_dashboard", {"slug": "secret"}) + assert result["isError"] is False + assert result["structuredContent"]["queries"][0]["rows"] == [[1]] + + +@pytest.mark.parametrize("slug", ("no-such-dashboard", "private-one", "members")) +def test_execute_dashboard_unavailable_dashboards_are_not_disclosed( + client, dashboard_db, slug +): + Dashboard.objects.create(slug="private-one", view_policy="private") + Dashboard.objects.create(slug="members", view_policy="loggedin") + result = call_tool(client, "execute_dashboard", {"slug": slug}) + assert result["isError"] is True + assert ( + "Dashboard '{}' does not exist or is not available".format(slug) + in result["content"][0]["text"] + ) + + +def test_execute_dashboard_owner_can_execute_private( + client, dashboard_db, django_user_model +): + user = django_user_model.objects.create(username="owner") + dashboard = Dashboard.objects.create( + slug="private-one", view_policy="private", owned_by=user + ) + dashboard.queries.create(sql="select 1 as one") + client.force_login(user) + result = call_tool(client, "execute_dashboard", {"slug": "private-one"}) + assert result["isError"] is False + assert result["structuredContent"]["queries"][0]["rows"] == [[1]] + + +def test_execute_dashboard_with_parameters(client, dashboard_db): + dashboard = Dashboard.objects.create(slug="params", view_policy="public") + dashboard.queries.create(sql="select %(name)s as name") + result = call_tool( + client, + "execute_dashboard", + {"slug": "params", "parameters": {"name": "Cleo"}}, + ) + assert result["structuredContent"]["queries"][0]["rows"] == [["Cleo"]] + + +def test_execute_dashboard_reports_per_query_errors(client, dashboard_db): + dashboard = Dashboard.objects.create(slug="mixed", view_policy="public") + dashboard.queries.create(sql="select * from no_such_table") + dashboard.queries.create(sql="select 1 as one") + result = call_tool(client, "execute_dashboard", {"slug": "mixed"}) + assert result["isError"] is False + queries = result["structuredContent"]["queries"] + assert 'relation "no_such_table" does not exist' in queries[0]["error"] + assert queries[1]["rows"] == [[1]] + + +def test_execute_dashboard_does_not_require_execute_sql_permission( + client, dashboard_db, django_user_model, saved_dashboard +): + user = django_user_model.objects.create(username="powerless_viewer") + client.force_login(user) + result = call_tool(client, "execute_dashboard", {"slug": "test"}) + assert result["isError"] is False + assert result["structuredContent"]["queries"][0]["rows"] == [[44]] From 40ffbf18e57c7959750181d44c36cd3c3e99ee9e Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 1 Aug 2026 05:57:25 +0000 Subject: [PATCH 4/4] Apply black Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017ZRFpnuD9tZKWb19Ffg8iF --- django_sql_dashboard/views.py | 6 ++---- test_project/config/urls.py | 1 - test_project/manage.py | 1 + test_project/test_widgets.py | 12 ++++-------- 4 files changed, 7 insertions(+), 13 deletions(-) diff --git a/django_sql_dashboard/views.py b/django_sql_dashboard/views.py index 7719eb2..2653e7b 100644 --- a/django_sql_dashboard/views.py +++ b/django_sql_dashboard/views.py @@ -146,8 +146,7 @@ def _dashboard_index( connection = connections[alias] reserved_words = postgresql_reserved_words(connection) with connection.cursor() as tables_cursor: - tables_cursor.execute( - """ + tables_cursor.execute(""" with visible_tables as ( select table_name from information_schema.tables @@ -173,8 +172,7 @@ def _dashboard_index( information_schema.columns.table_name order by information_schema.columns.table_name - """ - ) + """) fetched = tables_cursor.fetchall() available_tables = [ { diff --git a/test_project/config/urls.py b/test_project/config/urls.py index b781c44..3148385 100644 --- a/test_project/config/urls.py +++ b/test_project/config/urls.py @@ -4,7 +4,6 @@ import django_sql_dashboard - urlpatterns = [ path("dashboard/", include(django_sql_dashboard.urls)), path("admin/", admin.site.urls), diff --git a/test_project/manage.py b/test_project/manage.py index d28672e..aabb818 100755 --- a/test_project/manage.py +++ b/test_project/manage.py @@ -1,5 +1,6 @@ #!/usr/bin/env python """Django's command-line utility for administrative tasks.""" + import os import sys diff --git a/test_project/test_widgets.py b/test_project/test_widgets.py index 8957ef9..ab6fe08 100644 --- a/test_project/test_widgets.py +++ b/test_project/test_widgets.py @@ -10,12 +10,10 @@ def test_default_widget(admin_client, dashboard_db): response = admin_client.post( "/dashboard/", - { - "sql": """ + {"sql": """ SELECT * FROM ( VALUES (1, 'one', 4.5), (2, 'two', 3.6), (3, 'three', 4.1) - ) AS t (id, name, size)""" - }, + ) AS t (id, name, size)"""}, follow=True, ) html = response.content.decode("utf-8") @@ -34,11 +32,9 @@ def test_default_widget(admin_client, dashboard_db): def test_default_widget_pretty_prints_json(admin_client, dashboard_db): response = admin_client.post( "/dashboard/", - { - "sql": """ + {"sql": """ select json_build_object('hello', json_build_array(1, 2, 3)) as json - """ - }, + """}, follow=True, ) html = response.content.decode("utf-8")