bvandeusen
aef5009fc2
Drift-audit Group 5 (high-severity contract drift): - MCP read-only keys could call every write tool: the Bearer resolver discarded api_key.scope and dispatch had no gate. Add resolve_bearer() (returns user_id + scope) and a scope gate in the /mcp ASGI wrapper that buffers the JSON-RPC body and rejects tools/call for any tool outside a read all-list when scope=='read' (default-deny for unknown/new tools). - Shared project notes/tasks panel was empty for non-owners: get_project_notes_route now queries notes/milestones with the project OWNER's uid (mirrors the already-fixed milestones route). - Shared editors couldn't save/delete shared NOTES (tasks worked): the three notes write routes now resolve via get_note_for_user, gate on can_write_note, and write as the owner — matching the tasks routes. - Event timezone drift: naive datetimes from the MCP date+time split are now localized to the user's tz at a single canonical service point (create_event /update_event), so MCP- and UI-created events agree. tz-aware inputs (REST/CalDAV) pass through untouched. - create_note validates status/priority (TaskStatus/TaskPriority), closing the MCP create_task path that let out-of-enum values persist (no DB CHECK). Tests cover resolve_bearer scope + the write-tool classifier. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
"""Tests for MCP auth: bearer-token validation that reuses api_keys."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from fabledassistant.mcp.auth import resolve_bearer_to_user_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_bearer_missing_header_returns_none():
|
|
assert await resolve_bearer_to_user_id(None) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_bearer_malformed_header_returns_none():
|
|
assert await resolve_bearer_to_user_id("Token abc") is None
|
|
assert await resolve_bearer_to_user_id("Bearer") is None
|
|
assert await resolve_bearer_to_user_id("Bearer ") is None
|
|
assert await resolve_bearer_to_user_id("") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_bearer_unknown_token_returns_none():
|
|
with patch(
|
|
"fabledassistant.mcp.auth.lookup_key",
|
|
AsyncMock(return_value=None),
|
|
):
|
|
assert await resolve_bearer_to_user_id("Bearer fmcp_doesnotexist") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_bearer_valid_token_returns_user_id():
|
|
fake_key = MagicMock()
|
|
fake_key.user_id = 42
|
|
with patch(
|
|
"fabledassistant.mcp.auth.lookup_key",
|
|
AsyncMock(return_value=fake_key),
|
|
):
|
|
uid = await resolve_bearer_to_user_id("Bearer fmcp_validkey")
|
|
assert uid == 42
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_bearer_calls_lookup_with_stripped_token():
|
|
"""The Bearer prefix and any trailing whitespace must be stripped before lookup."""
|
|
fake_key = MagicMock()
|
|
fake_key.user_id = 1
|
|
mock_lookup = AsyncMock(return_value=fake_key)
|
|
with patch("fabledassistant.mcp.auth.lookup_key", mock_lookup):
|
|
await resolve_bearer_to_user_id("Bearer fmcp_abc123 ")
|
|
mock_lookup.assert_awaited_once_with("fmcp_abc123")
|
|
|
|
|
|
# ── resolve_bearer (user_id + scope) ────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_bearer_returns_user_id_and_scope():
|
|
from fabledassistant.mcp.auth import resolve_bearer
|
|
fake_key = MagicMock()
|
|
fake_key.user_id = 9
|
|
fake_key.scope = "read"
|
|
with patch("fabledassistant.mcp.auth.lookup_key", AsyncMock(return_value=fake_key)):
|
|
assert await resolve_bearer("Bearer fmcp_x") == (9, "read")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_bearer_none_for_invalid():
|
|
with patch("fabledassistant.mcp.auth.lookup_key", AsyncMock(return_value=None)):
|
|
assert await resolve_bearer("Bearer nope") is None
|
|
assert await resolve_bearer(None) is None
|
|
|
|
|
|
# ── read-only scope gate ────────────────────────────────────────────────
|
|
|
|
def test_body_calls_write_tool_classifies_correctly():
|
|
import json
|
|
from fabledassistant.mcp.server import _body_calls_write_tool
|
|
|
|
def call(name):
|
|
return json.dumps({"jsonrpc": "2.0", "id": 1, "method": "tools/call",
|
|
"params": {"name": name, "arguments": {}}}).encode()
|
|
|
|
# Write-class tools are gated.
|
|
assert _body_calls_write_tool(call("create_note")) is True
|
|
assert _body_calls_write_tool(call("delete_project")) is True
|
|
assert _body_calls_write_tool(call("purge_trash")) is True
|
|
# An unknown/new tool defaults to write (default-deny for read keys).
|
|
assert _body_calls_write_tool(call("brand_new_tool")) is True
|
|
# Read tools and non-call methods pass.
|
|
assert _body_calls_write_tool(call("list_notes")) is False
|
|
assert _body_calls_write_tool(call("get_recent")) is False
|
|
assert _body_calls_write_tool(
|
|
json.dumps({"method": "tools/list"}).encode()
|
|
) is False
|
|
assert _body_calls_write_tool(b"not json") is False
|