"""Tests for MCP auth: bearer-token validation that reuses api_keys.""" from unittest.mock import AsyncMock, MagicMock, patch import pytest from scribe.mcp.auth import resolve_bearer, 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( "scribe.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( "scribe.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("scribe.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(): fake_key = MagicMock() fake_key.user_id = 9 fake_key.scope = "read" with patch("scribe.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("scribe.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 scribe.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 def test_every_read_shaped_tool_is_explicitly_classified(): """A read-shaped tool must be classified, not left to default-deny. `_READ_ONLY_TOOLS` is hand-maintained, and default-deny means a getter omitted from it fails CLOSED — safe, but silent. That is how a read key ended up able to `get_note` and not `get_snippet`, both pure reads of the same table, while design systems were unreachable entirely (#2496). The `find_duplicate_snippets` entry was the tell: someone classified the report and missed the getters beside it. This is the same shape as #2476 (record_pulled on three of four getters) — a hand-written enumeration that missed the members added after it. The fix there and here is the same: derive the CANDIDATES, keep the DECISION explicit. Deriving the decision itself would be worse than a stale list — it would make a security boundary follow a naming convention, so any future `get_*` grants itself access. So: every tool whose name reads like a read must appear in one of the two sets. Adding a getter then forces a choice at review time. """ import ast import pathlib from scribe.mcp.server import _DELIBERATELY_WRITE_SCOPED, _READ_ONLY_TOOLS tools_dir = (pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe" / "mcp" / "tools") read_shaped = { node.name for path in tools_dir.glob("*.py") if path.name != "__init__.py" for node in ast.parse(path.read_text()).body if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)) and node.name.startswith(("get_", "list_", "search", "resolve_", "check_", "find_")) } assert read_shaped, "found no read-shaped tools — the tools package moved" unclassified = sorted(read_shaped - _READ_ONLY_TOOLS - _DELIBERATELY_WRITE_SCOPED) assert not unclassified, ( f"these read-shaped tools are classified by neither set: {unclassified}. " f"They currently fail closed for read-only keys, silently. Add each to " f"_READ_ONLY_TOOLS if it mutates nothing, or to " f"_DELIBERATELY_WRITE_SCOPED with a comment saying what it writes." ) # The reverse: a name in either set that no longer exists is a rename or a # deletion, and a stale grant is worth surfacing even though it grants # access to nothing. `enter_project` is the one read tool without a read # prefix, so it is checked against the full tool set, not `read_shaped`. all_tools = { node.name for path in tools_dir.glob("*.py") if path.name != "__init__.py" for node in ast.parse(path.read_text()).body if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)) and not node.name.startswith("_") and node.name != "register" } phantom = sorted((_READ_ONLY_TOOLS | _DELIBERATELY_WRITE_SCOPED) - all_tools) assert not phantom, ( f"these names are classified but are not tools: {phantom}. They were " f"renamed or removed — drop them, and check whatever replaced them got " f"classified." )