Codex
31b0e8dba2
Add Streamable HTTP transport for the source-file MCP server with bearer-token auth, configurable HTTP runtime options, and user-level systemd service helpers. Document remote client setup and cover auth/service behavior with unit tests.
123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
import asyncio
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from obsidian_wiki import mcp_server
|
|
|
|
|
|
class FakeApp:
|
|
def __init__(self):
|
|
self.called = False
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
self.called = True
|
|
await send({"type": "http.response.start", "status": 200, "headers": []})
|
|
await send({"type": "http.response.body", "body": b"ok"})
|
|
|
|
|
|
def run_asgi_app(app, *, headers=None, scope_type="http"):
|
|
messages = []
|
|
scope = {"type": scope_type, "headers": headers or []}
|
|
|
|
async def receive():
|
|
return {"type": "http.request", "body": b"", "more_body": False}
|
|
|
|
async def send(message):
|
|
messages.append(message)
|
|
|
|
asyncio.run(app(scope, receive, send))
|
|
return messages
|
|
|
|
|
|
class RemoteMcpServerTests(unittest.TestCase):
|
|
def test_bearer_token_middleware_accepts_matching_token(self):
|
|
inner = FakeApp()
|
|
middleware = mcp_server.BearerTokenAuthMiddleware(inner, "secret-token")
|
|
|
|
messages = run_asgi_app(
|
|
middleware,
|
|
headers=[(b"authorization", b"Bearer secret-token")],
|
|
)
|
|
|
|
self.assertTrue(inner.called)
|
|
self.assertEqual(messages[0]["status"], 200)
|
|
|
|
def test_bearer_token_middleware_rejects_missing_or_wrong_token(self):
|
|
inner = FakeApp()
|
|
middleware = mcp_server.BearerTokenAuthMiddleware(inner, "secret-token")
|
|
|
|
messages = run_asgi_app(
|
|
middleware,
|
|
headers=[(b"authorization", b"Bearer wrong-token")],
|
|
)
|
|
|
|
self.assertFalse(inner.called)
|
|
self.assertEqual(messages[0]["status"], 401)
|
|
self.assertIn((b"www-authenticate", b"Bearer"), messages[0]["headers"])
|
|
|
|
def test_bearer_token_middleware_requires_non_empty_token(self):
|
|
with self.assertRaises(ValueError):
|
|
mcp_server.BearerTokenAuthMiddleware(FakeApp(), "")
|
|
|
|
def test_read_token_from_env_trims_and_requires_value(self):
|
|
with patch.dict(os.environ, {"TEST_MCP_TOKEN": " secret-token "}, clear=False):
|
|
self.assertEqual(mcp_server.read_token_from_env("TEST_MCP_TOKEN"), "secret-token")
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaises(RuntimeError):
|
|
mcp_server.read_token_from_env("TEST_MCP_TOKEN")
|
|
|
|
def test_runtime_config_reads_environment_with_defaults(self):
|
|
with patch.dict(os.environ, {"OBSIDIAN_MCP_HOST": " 0.0.0.0 ", "OBSIDIAN_MCP_PORT": "9876"}, clear=True):
|
|
self.assertEqual(mcp_server.runtime_config_value("OBSIDIAN_MCP_HOST", "127.0.0.1"), "0.0.0.0")
|
|
self.assertEqual(mcp_server.runtime_config_int("OBSIDIAN_MCP_PORT", 8765), 9876)
|
|
|
|
with patch.dict(os.environ, {"OBSIDIAN_MCP_PORT": "not-an-int"}, clear=True):
|
|
self.assertEqual(mcp_server.runtime_config_int("OBSIDIAN_MCP_PORT", 8765), 8765)
|
|
|
|
def test_create_http_app_wraps_streamable_http_app_with_auth(self):
|
|
class FakeServer:
|
|
def __init__(self):
|
|
self.inner = FakeApp()
|
|
|
|
def streamable_http_app(self):
|
|
return self.inner
|
|
|
|
fake_server = FakeServer()
|
|
app = mcp_server.create_http_app(fake_server, "secret-token")
|
|
|
|
messages = run_asgi_app(
|
|
app,
|
|
headers=[(b"authorization", b"Bearer secret-token")],
|
|
)
|
|
|
|
self.assertTrue(fake_server.inner.called)
|
|
self.assertEqual(messages[0]["status"], 200)
|
|
|
|
def test_run_http_server_fails_before_server_creation_without_token(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch.object(mcp_server, "create_server") as create_server:
|
|
with self.assertRaises(RuntimeError):
|
|
mcp_server.run_http_server(
|
|
vault_path=Path(tmp),
|
|
db_path=None,
|
|
profile_name="bjd-general-bot",
|
|
max_file_bytes=1024,
|
|
expose_paths=False,
|
|
host="127.0.0.1",
|
|
port=8765,
|
|
token_env="TEST_MCP_TOKEN",
|
|
streamable_http_path="/mcp",
|
|
allowed_hosts=None,
|
|
allowed_origins=None,
|
|
)
|
|
create_server.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|