"""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 @pytest.mark.asyncio async def test_resolve_bearer_missing_header_returns_none(): assert await resolve_bearer(None) is None @pytest.mark.asyncio async def test_resolve_bearer_malformed_header_returns_none(): assert await resolve_bearer("Token abc") is None assert await resolve_bearer("Bearer") is None assert await resolve_bearer("Bearer ") is None assert await resolve_bearer("") 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("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 fake_key.scope = "write" with patch( "scribe.mcp.auth.lookup_key", AsyncMock(return_value=fake_key), ): uid, scope = await resolve_bearer("Bearer fmcp_validkey") assert (uid, scope) == (42, "write") @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 fake_key.scope = "write" mock_lookup = AsyncMock(return_value=fake_key) with patch("scribe.mcp.auth.lookup_key", mock_lookup): await resolve_bearer("Bearer fmcp_abc123 ") mock_lookup.assert_awaited_once_with("fmcp_abc123") # ── 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 _registered_tool_names() -> set[str]: """What the server actually mounts — not a glob of function names, which would count helpers and miss anything registered another way.""" from scribe.mcp.server import build_mcp_server return {tool.name for tool in build_mcp_server()._tool_manager.list_tools()} def test_every_registered_tool_is_classified_exactly_once(): """Every tool must be declared a read, a write, or a read-shaped write. `_READ_ONLY_TOOLS` is hand-maintained, and default-deny means a read omitted from it fails CLOSED — safe, but silent. That is how a read key ended up able to `get_note` and not `get_snippet` (#2496), and how `rules_due_for_verification` and `rule_history` sat denied (#3191). The fix keeps the DECISION explicit and derives the CANDIDATES. Deriving the decision would make a security boundary follow a naming convention, so any future `get_*` grants itself access. This test used to derive candidates from names too — tools starting `get_`, `list_`, `search`… — and that narrowing is exactly what let the two #3191 reads through: neither name looked like a read. So the candidates are now EVERY registered tool. """ from scribe.mcp.server import ( _DELIBERATELY_WRITE_SCOPED, _READ_ONLY_TOOLS, _WRITE_TOOLS, ) registered = _registered_tool_names() assert len(registered) > 50, "found almost no tools — registration moved" sets = { "_READ_ONLY_TOOLS": _READ_ONLY_TOOLS, "_WRITE_TOOLS": _WRITE_TOOLS, "_DELIBERATELY_WRITE_SCOPED": _DELIBERATELY_WRITE_SCOPED, } unclassified = sorted(registered - set().union(*sets.values())) assert not unclassified, ( f"these tools are classified by no set: {unclassified}. A read key is " f"silently denied them. Add each to _READ_ONLY_TOOLS if it mutates " f"nothing, otherwise to _WRITE_TOOLS." ) names = list(sets) for i, first in enumerate(names): for second in names[i + 1:]: both = sorted(sets[first] & sets[second]) assert not both, f"in both {first} and {second}: {both}" # The reverse: a classified name that is not a tool is a rename or a # deletion, and a stale grant is worth surfacing even though it grants # access to nothing. phantom = sorted(set().union(*sets.values()) - registered) 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." ) def test_the_completeness_check_can_fail(): """A guard that cannot fail is indistinguishable from one that is broken (rule 167). Drop a real tool from its set and the check must notice.""" from scribe.mcp.server import _READ_ONLY_TOOLS, _WRITE_TOOLS registered = _registered_tool_names() assert "rule_history" in _READ_ONLY_TOOLS without = (_READ_ONLY_TOOLS - {"rule_history"}) | _WRITE_TOOLS assert "rule_history" in registered - without