FabledGrimoire/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 _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: a read key able to
    `get_note` and not `get_snippet`.

    So the DECISION stays explicit and the CANDIDATES are derived. Deriving
    the decision would make a security boundary follow a naming convention, so
    any future `get_*` grants itself access. Deriving the candidates from names
    — tools starting `get_`, `list_`, `search`… — misses reads whose names do
    not look like reads (`rules_due_for_verification`, `rule_history`). So the
    candidates are 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
    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