feat(mcp): bearer-token auth resolver
Thin parser over the existing api_keys lookup. Strips the Bearer prefix, validates the token via services/api_keys.lookup_key (which already filters revoked keys and updates last_used_at), and returns the user_id for the in-flight MCP request. Tests follow the existing mock-async_session pattern in test_api_keys.py rather than introducing a real DB fixture. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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 <token>` 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
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user