FabledScribe/tests/test_mcp_auth.py

"""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 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."
    )