diff --git a/src/fabledassistant/mcp/auth.py b/src/fabledassistant/mcp/auth.py new file mode 100644 index 0000000..4cad787 --- /dev/null +++ b/src/fabledassistant/mcp/auth.py @@ -0,0 +1,19 @@ +"""MCP-side Bearer token resolution. Reuses the existing api_keys infrastructure.""" +from __future__ import annotations + +from fabledassistant.services.api_keys import lookup_key + + +async def resolve_bearer_to_user_id(auth_header: str | None) -> int | None: + """Parse an `Authorization: Bearer ` header and return the user_id. + + Returns None if the header is missing, malformed, or the token is invalid + or revoked. The underlying lookup_key already updates last_used_at on hit. + """ + if not auth_header or not auth_header.startswith("Bearer "): + return None + raw_token = auth_header[len("Bearer "):].strip() + if not raw_token: + return None + api_key = await lookup_key(raw_token) + return api_key.user_id if api_key else None diff --git a/tests/test_mcp_auth.py b/tests/test_mcp_auth.py new file mode 100644 index 0000000..8579058 --- /dev/null +++ b/tests/test_mcp_auth.py @@ -0,0 +1,51 @@ +"""Tests for MCP auth: bearer-token validation that reuses api_keys.""" +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from fabledassistant.mcp.auth import 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( + "fabledassistant.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( + "fabledassistant.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("fabledassistant.mcp.auth.lookup_key", mock_lookup): + await resolve_bearer_to_user_id("Bearer fmcp_abc123 ") + mock_lookup.assert_awaited_once_with("fmcp_abc123")