Robert Helewka
642268cec1
Both helpers were load-bearing during the Pallas<->Mnemosyne shakedown:
* _extract_tool_name: covers the current FastMCP shape
(context.message.name directly), the legacy .params.name fallback,
prefer-direct behaviour, and every None-producing path. Includes a
contract test against the real mcp.types.CallToolRequestParams which
skips if the mcp package isn't importable.
* _extract_token: covers Bearer/bearer schemes, Authorization/
authorization header casing, whitespace stripping, missing/empty/
non-Bearer headers, RuntimeError degrading to None (outside an
HTTP dispatch), and non-RuntimeError propagating loudly.
Uses SimpleTestCase (no DB) with unittest.mock.patch on
mcp_server.auth.get_http_request to avoid pulling in FastMCP internals.
Run as part of mnemosyne's mcp_server suite:
TEST_NEO4J_ENABLED=0 python manage.py test mcp_server \
--testrunner=test_db_manager.django_integration.PostgreSQLTestRunner
17 new tests, all green; total mcp_server suite 59 tests passing.
251 lines
10 KiB
Python
251 lines
10 KiB
Python
"""Unit tests for ``MCPAuthMiddleware._extract_tool_name`` / ``_extract_token``.
|
|
|
|
Both of these helpers turned out to be load-bearing during the
|
|
Pallas↔Mnemosyne shakedown — see ``pallas/docs/pallas.md`` §"Incidents
|
|
& Lessons Learned" for the full story:
|
|
|
|
* ``_extract_tool_name`` reads ``context.message.name`` directly.
|
|
FastMCP's ``Middleware.on_call_tool`` is typed as
|
|
``MiddlewareContext[CallToolRequestParams]``, so ``message`` *is* the
|
|
params object — there is no nested ``.params``. The legacy
|
|
``message.params.name`` access silently returned ``None`` and caused
|
|
the ``_PUBLIC_TOOLS`` bypass for ``get_health`` to stop working
|
|
(every call was treated as "no name known", which then short-circuited
|
|
the per-tool ACL to the fail-open branch in ``can_use_tool``).
|
|
|
|
* ``_extract_token`` relies on ``fastmcp.server.dependencies.get_http_request``
|
|
which raises ``RuntimeError`` outside of an active HTTP dispatch
|
|
(background tasks, pre-session init hooks). It must degrade to
|
|
``None`` in that case so the caller can decide whether to raise
|
|
``PermissionError``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from django.test import SimpleTestCase
|
|
|
|
from mcp_server.auth import MCPAuthMiddleware
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_tool_name
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ExtractToolNameTest(SimpleTestCase):
|
|
"""Exercise every branch of ``MCPAuthMiddleware._extract_tool_name``.
|
|
|
|
The helper is a pure staticmethod, so we feed it hand-built
|
|
``MiddlewareContext`` stand-ins rather than spinning up a real
|
|
FastMCP server.
|
|
"""
|
|
|
|
def _ctx(self, message):
|
|
"""Build a minimal stand-in for ``MiddlewareContext``.
|
|
|
|
Only ``.message`` is read by the helper, so a ``SimpleNamespace``
|
|
with that attribute is sufficient — keeping this test decoupled
|
|
from FastMCP's evolving class shape.
|
|
"""
|
|
return SimpleNamespace(message=message)
|
|
|
|
def test_reads_name_from_message_directly(self):
|
|
"""Current (correct) shape: ``message`` IS the CallToolRequestParams."""
|
|
# Shape matches ``mcp.types.CallToolRequestParams`` — verified via
|
|
# ``CallToolRequestParams(name='search', arguments={...}).name``.
|
|
msg = SimpleNamespace(name="search", arguments={"q": "hello"})
|
|
self.assertEqual(
|
|
MCPAuthMiddleware._extract_tool_name(self._ctx(msg)),
|
|
"search",
|
|
)
|
|
|
|
def test_falls_back_to_nested_params_name(self):
|
|
"""Legacy safety net for a hypothetical future FastMCP shape.
|
|
|
|
Older code read ``message.params.name``. We keep that path as a
|
|
fallback so a version bump that reverts the shape doesn't silently
|
|
re-introduce the public-tools bypass bug.
|
|
"""
|
|
msg = SimpleNamespace(
|
|
name=None,
|
|
params=SimpleNamespace(name="get_health"),
|
|
)
|
|
self.assertEqual(
|
|
MCPAuthMiddleware._extract_tool_name(self._ctx(msg)),
|
|
"get_health",
|
|
)
|
|
|
|
def test_prefers_direct_name_over_nested_params(self):
|
|
"""If both are populated, the direct attribute wins.
|
|
|
|
That's the current FastMCP reality and the shape we want to
|
|
exercise in production; the nested fallback must not shadow it.
|
|
"""
|
|
msg = SimpleNamespace(
|
|
name="search",
|
|
params=SimpleNamespace(name="wrong"),
|
|
)
|
|
self.assertEqual(
|
|
MCPAuthMiddleware._extract_tool_name(self._ctx(msg)),
|
|
"search",
|
|
)
|
|
|
|
def test_returns_none_when_message_missing(self):
|
|
"""A context with no ``.message`` at all (defensive)."""
|
|
ctx = SimpleNamespace() # no message attr
|
|
self.assertIsNone(MCPAuthMiddleware._extract_tool_name(ctx))
|
|
|
|
def test_returns_none_when_message_is_none(self):
|
|
self.assertIsNone(
|
|
MCPAuthMiddleware._extract_tool_name(self._ctx(None))
|
|
)
|
|
|
|
def test_returns_none_when_no_name_anywhere(self):
|
|
"""No direct ``name`` and no ``params`` — nothing to extract."""
|
|
msg = SimpleNamespace(name=None)
|
|
self.assertIsNone(
|
|
MCPAuthMiddleware._extract_tool_name(self._ctx(msg))
|
|
)
|
|
|
|
def test_returns_none_when_params_has_no_name(self):
|
|
msg = SimpleNamespace(name=None, params=SimpleNamespace())
|
|
self.assertIsNone(
|
|
MCPAuthMiddleware._extract_tool_name(self._ctx(msg))
|
|
)
|
|
|
|
def test_real_call_tool_request_params_shape(self):
|
|
"""Contract test against the actual ``mcp.types.CallToolRequestParams``.
|
|
|
|
Skips silently if the ``mcp`` package isn't importable in this
|
|
test environment — the rest of the suite uses hand-built
|
|
namespaces precisely so it stays runnable in minimal envs.
|
|
"""
|
|
try:
|
|
from mcp.types import CallToolRequestParams
|
|
except Exception: # pragma: no cover - env-dependent
|
|
self.skipTest("mcp package not installed in this environment")
|
|
|
|
params = CallToolRequestParams(name="list_libraries", arguments={})
|
|
self.assertEqual(
|
|
MCPAuthMiddleware._extract_tool_name(self._ctx(params)),
|
|
"list_libraries",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_token
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeHeaders:
|
|
"""Starlette-style headers with case-insensitive ``get`` and ``keys``."""
|
|
|
|
def __init__(self, mapping: dict[str, str]):
|
|
# Starlette itself is case-insensitive; emulate that closely enough
|
|
# for the helper without depending on Starlette in test deps.
|
|
self._store = {k.lower(): v for k, v in mapping.items()}
|
|
|
|
def get(self, key: str, default: str = "") -> str:
|
|
return self._store.get(key.lower(), default)
|
|
|
|
def keys(self):
|
|
return list(self._store.keys())
|
|
|
|
|
|
class _FakeRequest:
|
|
def __init__(self, headers: dict[str, str], url: str = "http://x/mcp"):
|
|
self.headers = _FakeHeaders(headers)
|
|
self.url = url
|
|
|
|
|
|
class ExtractTokenTest(SimpleTestCase):
|
|
"""Exercise ``MCPAuthMiddleware._extract_token``.
|
|
|
|
We patch ``mcp_server.auth.get_http_request`` directly rather than
|
|
reaching into FastMCP internals — the helper only cares about the
|
|
return value (or raised ``RuntimeError``) of that import-bound name.
|
|
"""
|
|
|
|
def test_returns_token_when_bearer_header_present(self):
|
|
request = _FakeRequest({"Authorization": "Bearer abc.def.ghi"})
|
|
with patch("mcp_server.auth.get_http_request", return_value=request):
|
|
self.assertEqual(
|
|
MCPAuthMiddleware._extract_token(),
|
|
"abc.def.ghi",
|
|
)
|
|
|
|
def test_accepts_lowercase_bearer_scheme(self):
|
|
"""Some clients emit ``bearer`` (lowercase) — accept it."""
|
|
request = _FakeRequest({"Authorization": "bearer xyz"})
|
|
with patch("mcp_server.auth.get_http_request", return_value=request):
|
|
self.assertEqual(MCPAuthMiddleware._extract_token(), "xyz")
|
|
|
|
def test_accepts_lowercase_header_name(self):
|
|
"""HTTP/2 normalizes header names to lowercase; some proxies follow.
|
|
|
|
Belt-and-braces lookup in ``_extract_token`` tries both forms.
|
|
"""
|
|
request = _FakeRequest({"authorization": "Bearer lower-case-hdr"})
|
|
with patch("mcp_server.auth.get_http_request", return_value=request):
|
|
self.assertEqual(
|
|
MCPAuthMiddleware._extract_token(),
|
|
"lower-case-hdr",
|
|
)
|
|
|
|
def test_strips_surrounding_whitespace(self):
|
|
request = _FakeRequest({"Authorization": "Bearer padded-token "})
|
|
with patch("mcp_server.auth.get_http_request", return_value=request):
|
|
self.assertEqual(
|
|
MCPAuthMiddleware._extract_token(),
|
|
"padded-token",
|
|
)
|
|
|
|
def test_returns_none_when_header_missing(self):
|
|
request = _FakeRequest({})
|
|
with patch("mcp_server.auth.get_http_request", return_value=request):
|
|
self.assertIsNone(MCPAuthMiddleware._extract_token())
|
|
|
|
def test_returns_none_when_header_is_empty_bearer(self):
|
|
"""``Authorization: Bearer `` with no value should read as missing."""
|
|
request = _FakeRequest({"Authorization": "Bearer "})
|
|
with patch("mcp_server.auth.get_http_request", return_value=request):
|
|
self.assertIsNone(MCPAuthMiddleware._extract_token())
|
|
|
|
def test_returns_none_when_scheme_is_not_bearer(self):
|
|
"""Non-Bearer schemes (Basic, Digest, etc.) are ignored."""
|
|
request = _FakeRequest({"Authorization": "Basic dXNlcjpwdw=="})
|
|
with patch("mcp_server.auth.get_http_request", return_value=request):
|
|
self.assertIsNone(MCPAuthMiddleware._extract_token())
|
|
|
|
def test_runtime_error_from_get_http_request_degrades_to_none(self):
|
|
"""Outside an HTTP dispatch, ``get_http_request`` raises.
|
|
|
|
The helper must swallow that and return ``None`` so the caller
|
|
(typically ``on_call_tool``) can decide whether to raise
|
|
``PermissionError`` based on ``MCP_REQUIRE_AUTH`` — rather than
|
|
letting a bare RuntimeError bubble out into the FastMCP layer
|
|
where it gets rewrapped into an opaque ``CallToolResult(isError=True)``.
|
|
"""
|
|
with patch(
|
|
"mcp_server.auth.get_http_request",
|
|
side_effect=RuntimeError("no active request"),
|
|
):
|
|
self.assertIsNone(MCPAuthMiddleware._extract_token())
|
|
|
|
def test_other_exceptions_propagate(self):
|
|
"""Only ``RuntimeError`` is caught — other errors must surface.
|
|
|
|
If ``get_http_request`` ever starts raising something unexpected
|
|
we want a loud failure in logs, not a silent ``None`` that masks
|
|
the bug.
|
|
"""
|
|
with patch(
|
|
"mcp_server.auth.get_http_request",
|
|
side_effect=ValueError("something else entirely"),
|
|
):
|
|
with self.assertRaises(ValueError):
|
|
MCPAuthMiddleware._extract_token()
|