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..a3a8bc1 --- /dev/null +++ b/django_sql_dashboard/mcp.py @@ -0,0 +1,517 @@ +""" +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 +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. +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 +""" + +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 +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" + +SERVER_INFO = { + "name": "django-sql-dashboard", + "version": "1.2", +} + +SERVER_INSTRUCTIONS = ( + "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, + 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(user): + return {"tables": [table for table, _ in _tables_and_columns()]} + + +def tool_get_schema(user): + 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 _execute_query(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, + } + + +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": {}, + "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, + "requires_execute_sql": True, + "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, + "requires_execute_sql": True, + "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"], + }, + }, + { + "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} + + +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(user, params): + return { + "protocolVersion": MCP_PROTOCOL_VERSION, + "capabilities": {"tools": {}}, + "serverInfo": SERVER_INFO, + "instructions": SERVER_INSTRUCTIONS, + } + + +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": [ + { + "name": tool["name"], + "description": tool["description"], + "inputSchema": tool["inputSchema"], + "outputSchema": tool["outputSchema"], + "annotations": {"readOnlyHint": True, "openWorldHint": False}, + } + for tool in _available_tools(user) + ] + } + + +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: + 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)}], + "isError": True, + } + return { + "content": [ + {"type": "text", "text": json.dumps(structured, default=str)}, + ], + "structuredContent": structured, + "isError": False, + } + + +METHODS = { + "initialize": _handle_initialize, + "ping": lambda user, params: {}, + "tools/list": _handle_tools_list, + "tools/call": _handle_tools_call, +} + + +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. 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 + return request.user, None + + +@csrf_exempt +def mcp_endpoint(request): + user, error_response = _authenticate(request) + if error_response is not None: + return error_response + 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(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/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/django_sql_dashboard/views.py b/django_sql_dashboard/views.py index d9ae2e0..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 = [ { @@ -390,32 +388,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/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..7bde167 --- /dev/null +++ b/docs/mcp.md @@ -0,0 +1,102 @@ +# 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 + +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. +- `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. + +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 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. +- 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. + +## 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 +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. + +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/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_mcp.py b/test_project/test_mcp.py new file mode 100644 index 0000000..55b0de7 --- /dev/null +++ b/test_project/test_mcp.py @@ -0,0 +1,460 @@ +import json + +import pytest +from django.db import connections +from django.test.client import Client + +from django_sql_dashboard.models import Dashboard + +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, 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", + headers=headers, + ) + + +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 tool_names(client): + response = rpc(client, "tools/list") + 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_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) + # 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) + client.force_login(user) + 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( + 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"} + headers = {"authorization": "Bearer correct-token"} + response = rpc( + client, + "tools/call", + {"name": "execute_sql", "arguments": {"sql": "select 1"}}, + headers=headers, + ) + 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): + # 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) + 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", + "list_dashboards", + "execute_dashboard", + ] + 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 + 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 + + +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]] 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")