xiaoxue_admin
/
mempalace
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
mempalace/tests/test_mcp_http_transport.py

660 lines
24 KiB
Python
Raw Permalink Blame History

# tests/test_mcp_http_transport.py
"""
Tests for the opt-in HTTP transport added for #1801.
These exercise the *production* server built by
``mempalace.mcp_server._build_http_server`` over a real loopback socket on an
ephemeral port — the earlier version of this file reimplemented the endpoint in
Starlette and guarded on ``pytest.importorskip("starlette")``/``uvicorn``,
neither of which is a project dependency, so it was silently skipped in CI and
the real ``_serve_http`` handler had zero coverage.
Design constraints
------------------
* Real sockets, but bound to ``127.0.0.1:0`` (OS-assigned port) so there is no
port conflict on any CI runner.
* Pure stdlib (``http.client``, ``threading``) — no third-party deps.
* Server runs in a daemon thread and is shut down in fixture teardown.
"""
import http.client
import json
import logging
import os
import socketserver
import ssl
import threading
import pytest
from mempalace import mcp_server as mcp
def _post(port, path, body, headers=None, host_header=None):
"""Raw POST with full control over Host / Origin / Authorization headers."""
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
try:
raw = body if isinstance(body, (bytes, bytearray)) else json.dumps(body).encode("utf-8")
headers = headers or {}
conn.putrequest("POST", path, skip_host=(host_header is not None))
if host_header is not None:
conn.putheader("Host", host_header)
conn.putheader("Content-Type", "application/json")
# Let a caller override Content-Length (used to fake an oversized body)
# instead of emitting a second, conflicting header.
if not any(k.lower() == "content-length" for k in headers):
conn.putheader("Content-Length", str(len(raw)))
for k, v in headers.items():
conn.putheader(k, v)
conn.endheaders()
conn.send(raw)
resp = conn.getresponse()
return resp.status, resp.read()
finally:
conn.close()
def _get(port, path, headers=None):
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
try:
conn.request("GET", path, headers=headers or {})
resp = conn.getresponse()
return resp.status, resp.read()
finally:
conn.close()
@pytest.fixture
def http_server():
"""A running production MCP HTTP server on an ephemeral loopback port."""
httpd = mcp._build_http_server("127.0.0.1", 0)
port = httpd.server_address[1]
thread = threading.Thread(
target=httpd.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True
)
thread.start()
try:
yield port, httpd
finally:
httpd.shutdown()
httpd.server_close()
thread.join(timeout=5)
def test_post_dispatches_to_handle_request(http_server):
"""A real POST to /mcp reaches handle_request and returns its JSON-RPC reply."""
port, _ = http_server
status, body = _post(port, "/mcp", {"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
assert status == 200
payload = json.loads(body)
assert payload["id"] == 1
names = {t["name"] for t in payload["result"]["tools"]}
assert "mempalace_search" in names
def test_initialize_reports_server_info(http_server):
port, _ = http_server
status, body = _post(port, "/mcp", {"jsonrpc": "2.0", "id": 7, "method": "initialize"})
assert status == 200
assert json.loads(body)["result"]["serverInfo"]["name"] == "mempalace"
def test_healthz_ok(http_server):
port, _ = http_server
status, body = _get(port, "/healthz")
assert status == 200
assert body == b"ok\n"
def test_statusz_reports_machine_readable_server_and_client_state(http_server, monkeypatch):
monkeypatch.setattr(mcp, "_sqlite_integrity_payload", lambda: {"ok": True, "errors": []})
port, _ = http_server
assert _get(port, "/healthz", headers={"User-Agent": "codex-test"})[0] == 200
status, body = _get(port, "/statusz", headers={"User-Agent": "codex-test"})
assert status == 200
payload = json.loads(body)
assert payload["ok"] is True
assert payload["server"]["name"] == "mempalace"
assert payload["server"]["transport"] == "http"
assert payload["server"]["port"] == port
assert payload["requests"]["total"] >= 1
assert payload["requests"]["by_status"]["200"] >= 1
assert payload["clients"]["active_window_seconds"] == mcp._HTTP_ACTIVE_CLIENT_WINDOW_S
assert payload["clients"]["recent"]
first = payload["clients"]["recent"][0]
assert first["peer"] == "127.0.0.1"
assert first["peer_hint"] == "127.0.0.1"
assert first["user_agent"] == "codex-test"
assert first["last_path"] == "/healthz"
assert "Authorization" not in json.dumps(payload)
if mcp._config.palace_path:
assert mcp._config.palace_path not in json.dumps(payload)
def test_unknown_path_404(http_server):
port, _ = http_server
assert _post(port, "/nope", {"jsonrpc": "2.0", "id": 1, "method": "ping"})[0] == 404
assert _get(port, "/nope")[0] == 404
def test_invalid_json_returns_parse_error(http_server):
port, _ = http_server
status, body = _post(port, "/mcp", b"{not valid json")
assert status == 400
assert json.loads(body)["error"]["code"] == -32700
def test_oversized_request_rejected_413(http_server):
"""A declared Content-Length over the cap is rejected before the body is read."""
port, _ = http_server
# Lie about the length: the handler checks the header and returns 413 before
# reading the (tiny) body, so we never have to ship 16 MiB.
status, body = _post(
port,
"/mcp",
b"{}",
headers={"Content-Length": str(mcp._HTTP_MAX_REQUEST_BYTES + 1)},
)
assert status == 413
assert json.loads(body)["error"]["code"] == -32600
def test_notification_returns_202_no_body(http_server):
port, _ = http_server
status, body = _post(port, "/mcp", {"jsonrpc": "2.0", "method": "notifications/initialized"})
assert status == 202
assert body == b""
def test_rejects_foreign_host_header(http_server):
"""DNS-rebinding guard: a request carrying an attacker domain in Host is 403."""
port, _ = http_server
status, _ = _post(
port,
"/mcp",
{"jsonrpc": "2.0", "id": 1, "method": "ping"},
host_header="evil.example.com",
)
assert status == 403
def test_rejects_cross_origin(http_server):
"""A browser Origin from a non-loopback page is 403 (rebinding/SSRF guard)."""
port, _ = http_server
status, _ = _post(
port,
"/mcp",
{"jsonrpc": "2.0", "id": 1, "method": "ping"},
headers={"Origin": "https://evil.example"},
)
assert status == 403
def test_allows_loopback_origin(http_server):
port, _ = http_server
status, _ = _post(
port,
"/mcp",
{"jsonrpc": "2.0", "id": 1, "method": "ping"},
headers={"Origin": "http://localhost:5173"},
)
assert status == 200
def test_bearer_token_enforced_when_configured(monkeypatch):
"""With MEMPALACE_MCP_HTTP_TOKEN set, /mcp requires a matching bearer token."""
monkeypatch.setenv("MEMPALACE_MCP_HTTP_TOKEN", "s3cret")
monkeypatch.setattr(mcp, "_sqlite_integrity_payload", lambda: {"ok": True, "errors": []})
httpd = mcp._build_http_server("127.0.0.1", 0)
port = httpd.server_address[1]
thread = threading.Thread(
target=httpd.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True
)
thread.start()
try:
ping = {"jsonrpc": "2.0", "id": 1, "method": "ping"}
# No token → 401.
assert _post(port, "/mcp", ping)[0] == 401
# Wrong token → 401.
assert _post(port, "/mcp", ping, headers={"Authorization": "Bearer nope"})[0] == 401
# Correct token → 200.
assert _post(port, "/mcp", ping, headers={"Authorization": "Bearer s3cret"})[0] == 200
# /healthz never requires the token (orchestrator liveness probes).
assert _get(port, "/healthz")[0] == 200
# /statusz exposes server/client metadata, so it follows the auth policy.
assert _get(port, "/statusz")[0] == 401
status, body = _get(port, "/statusz", headers={"Authorization": "Bearer s3cret"})
assert status == 200
assert "recent" in json.loads(body)["clients"]
finally:
httpd.shutdown()
httpd.server_close()
thread.join(timeout=5)
def test_read_only_hides_and_refuses_mutating_tools(http_server, monkeypatch):
"""Read-only mode (#1877): the refused tools are hidden from tools/list AND
refused at dispatch with -32003, while read tools still work."""
monkeypatch.setattr(mcp, "_READ_ONLY", True)
port, _ = http_server
status, body = _post(port, "/mcp", {"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
assert status == 200
names = {t["name"] for t in json.loads(body)["result"]["tools"]}
assert "mempalace_search" in names # read tool stays
assert "mempalace_add_drawer" not in names # mutating tool hidden
assert names.isdisjoint(mcp._READ_ONLY_REFUSED_TOOLS)
status, body = _post(
port,
"/mcp",
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {"name": "mempalace_add_drawer", "arguments": {"content": "x"}},
},
)
assert status == 200
assert json.loads(body)["error"]["code"] == -32003
def test_read_only_off_exposes_mutating_tools(http_server):
"""Sanity: without read-only, mutating tools are present (guards the test above)."""
port, _ = http_server
status, body = _post(port, "/mcp", {"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
names = {t["name"] for t in json.loads(body)["result"]["tools"]}
assert "mempalace_add_drawer" in names
def test_writable_http_refuses_startup_without_writer_lease(monkeypatch):
monkeypatch.setattr(mcp, "_READ_ONLY", False)
monkeypatch.setattr(
mcp,
"_acquire_mcp_writer_lock",
lambda: (False, "another writer owns the palace"),
)
monkeypatch.setattr(
mcp,
"_serve_http",
lambda *args: pytest.fail("server must not bind without the writer lease"),
)
with pytest.raises(SystemExit) as exc_info:
mcp._run_http_loop()
assert exc_info.value.code == 2
def test_read_only_http_skips_writer_lease(monkeypatch):
calls = []
monkeypatch.setattr(mcp, "_READ_ONLY", True)
monkeypatch.setattr(
mcp,
"_acquire_mcp_writer_lock",
lambda: pytest.fail("read-only HTTP must not acquire the writer lease"),
)
monkeypatch.setattr(mcp, "_refresh_vector_disabled_flag", lambda: None)
monkeypatch.setattr(mcp, "_start_idle_exit_watchdog", lambda: None)
monkeypatch.setattr(mcp, "_serve_http", lambda host, port: calls.append((host, port)))
mcp._run_http_loop()
assert calls == [(mcp._args.host, mcp._args.port)]
def test_writable_http_releases_writer_lease_when_serving_ends(monkeypatch):
events = []
class DummyLease:
def __exit__(self, *exc):
events.append("lease-exit")
return False
lease = DummyLease()
def acquire_writer():
mcp._MCP_WRITER_LOCK_CM = lease
return True, ""
monkeypatch.setattr(mcp, "_READ_ONLY", False)
monkeypatch.setattr(mcp, "_MCP_WRITER_LOCK_CM", None)
monkeypatch.setattr(mcp, "_acquire_mcp_writer_lock", acquire_writer)
monkeypatch.setattr(mcp, "_discard_mcp_storage_handles", lambda: events.append("discard"))
monkeypatch.setattr(mcp, "_refresh_vector_disabled_flag", lambda: None)
monkeypatch.setattr(mcp, "_start_idle_exit_watchdog", lambda: None)
monkeypatch.setattr(mcp, "_serve_http", lambda host, port: events.append("serve"))
mcp._run_http_loop()
assert events == ["serve", "discard", "lease-exit"]
assert mcp._MCP_WRITER_LOCK_CM is None
def test_writable_http_releases_writer_lease_after_bind_failure(monkeypatch):
events = []
class DummyLease:
def __exit__(self, *exc):
events.append("lease-exit")
return False
lease = DummyLease()
def acquire_writer():
mcp._MCP_WRITER_LOCK_CM = lease
return True, ""
def fail_bind(host, port):
events.append("bind-failed")
raise SystemExit(1)
monkeypatch.setattr(mcp, "_READ_ONLY", False)
monkeypatch.setattr(mcp, "_MCP_WRITER_LOCK_CM", None)
monkeypatch.setattr(mcp, "_acquire_mcp_writer_lock", acquire_writer)
monkeypatch.setattr(mcp, "_discard_mcp_storage_handles", lambda: events.append("discard"))
monkeypatch.setattr(mcp, "_refresh_vector_disabled_flag", lambda: None)
monkeypatch.setattr(mcp, "_start_idle_exit_watchdog", lambda: None)
monkeypatch.setattr(mcp, "_serve_http", fail_bind)
with pytest.raises(SystemExit) as exc_info:
mcp._run_http_loop()
assert exc_info.value.code == 1
assert events == ["bind-failed", "discard", "lease-exit"]
assert mcp._MCP_WRITER_LOCK_CM is None
def _hook_settings_call(req_id):
return {
"jsonrpc": "2.0",
"id": req_id,
"method": "tools/call",
"params": {
"name": "mempalace_hook_settings",
"arguments": {"silent_save": False, "desktop_toast": True},
},
}
def test_read_only_refuses_the_hook_settings_config_write(http_server, monkeypatch, tmp_path):
"""mempalace_hook_settings writes the server's ~/.mempalace/config.json.
It touches no palace state, so it is correctly absent from _MUTATING_TOOLS,
the palace-write set the peer-writer lease arbitrates. Read-only gated on
that set, which let a read-only server persist a config change on behalf of
a client that is supposed to have no write access at all.
The first half is the control: it proves the write really does land here, so
the "unchanged" assertion in the second half cannot pass vacuously.
"""
home = tmp_path / "home"
(home / ".mempalace").mkdir(parents=True)
cfg_file = home / ".mempalace" / "config.json"
cfg_file.write_text(
json.dumps({"hooks": {"silent_save": True, "desktop_toast": False}}), encoding="utf-8"
)
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("USERPROFILE", str(home))
monkeypatch.setenv("HOMEDRIVE", os.path.splitdrive(str(home))[0] or "C:")
monkeypatch.setenv("HOMEPATH", os.path.splitdrive(str(home))[1] or str(home))
pristine = cfg_file.read_bytes()
port, _ = http_server
# Control: the gate is off, so the very same call rewrites config.json.
# _READ_ONLY is resolved at import from the environment, so pin it rather
# than inherit whatever the suite was started with.
monkeypatch.setattr(mcp, "_READ_ONLY", False)
status, body = _post(port, "/mcp", _hook_settings_call(1))
assert status == 200
# The handler reports its own failures inside `result` as {"success": false},
# not as a JSON-RPC error, so check the payload rather than just the envelope.
payload = json.loads(body)
assert "error" not in payload
assert json.loads(payload["result"]["content"][0]["text"])["success"] is True
assert cfg_file.read_bytes() != pristine
cfg_file.write_bytes(pristine)
# Gate on: hidden from tools/list, refused at dispatch, file left alone.
monkeypatch.setattr(mcp, "_READ_ONLY", True)
status, body = _post(port, "/mcp", {"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
names = {t["name"] for t in json.loads(body)["result"]["tools"]}
assert "mempalace_hook_settings" not in names
status, body = _post(port, "/mcp", _hook_settings_call(3))
assert status == 200
assert json.loads(body)["error"]["code"] == -32003
assert cfg_file.read_bytes() == pristine
def test_read_only_refuses_the_checkpoint_ack_delete(http_server, monkeypatch, tmp_path):
"""mempalace_memories_filed_away unlinks the Stop hook's checkpoint ack file.
Consuming that file is the contract of the tool, but it is still a delete of
state that outlives the process, done for a client with no write access. Same
two-phase shape as the config test: the control proves the delete lands, so
the survival assertion afterwards cannot pass vacuously.
"""
home = tmp_path / "home"
state_dir = home / ".mempalace" / "hook_state"
state_dir.mkdir(parents=True)
ack = state_dir / "last_checkpoint"
ack.write_text(json.dumps({"msgs": 7, "ts": "2026-01-01T00:00:00"}), encoding="utf-8")
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("USERPROFILE", str(home))
monkeypatch.setenv("HOMEDRIVE", os.path.splitdrive(str(home))[0] or "C:")
monkeypatch.setenv("HOMEPATH", os.path.splitdrive(str(home))[1] or str(home))
port, _ = http_server
call = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {"name": "mempalace_memories_filed_away", "arguments": {}},
}
# Control: the gate is off, so the call consumes the ack file.
monkeypatch.setattr(mcp, "_READ_ONLY", False)
status, body = _post(port, "/mcp", call)
assert status == 200
assert json.loads(json.loads(body)["result"]["content"][0]["text"])["count"] == 7
assert not ack.exists()
# Gate on: refused, and a fresh ack file survives untouched.
ack.write_text(json.dumps({"msgs": 7, "ts": "2026-01-01T00:00:00"}), encoding="utf-8")
pristine = ack.read_bytes()
monkeypatch.setattr(mcp, "_READ_ONLY", True)
status, body = _post(port, "/mcp", {"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
names = {t["name"] for t in json.loads(body)["result"]["tools"]}
assert "mempalace_memories_filed_away" not in names
status, body = _post(port, "/mcp", dict(call, id=3))
assert status == 200
assert json.loads(body)["error"]["code"] == -32003
assert ack.read_bytes() == pristine
@pytest.mark.parametrize(
"disconnect_exc",
[
ConnectionResetError(104, "connection reset by peer"),
BrokenPipeError(32, "broken pipe"),
ssl.SSLEOFError("unexpected eof while reading"),
],
ids=["connreset", "brokenpipe", "ssleof"],
)
def test_handle_error_quiets_client_disconnect(caplog, monkeypatch, disconnect_exc):
"""Regression for #2003: a client that hangs up mid-response makes the send
path raise ConnectionError (BrokenPipeError / ConnectionResetError), or
ssl.SSLEOFError on the TLS transport. The server must log that quietly at
DEBUG instead of routing it to the default handler's per-request traceback.
"""
httpd = mcp._build_http_server("127.0.0.1", 0)
try:
delegated = []
monkeypatch.setattr(
socketserver.BaseServer,
"handle_error",
lambda self, request, addr: delegated.append(addr),
)
addr = ("127.0.0.1", 51234)
with caplog.at_level(logging.DEBUG, logger="mempalace_mcp"):
try:
raise disconnect_exc
except type(disconnect_exc):
httpd.handle_error(None, addr)
assert delegated == [] # noisy default handler NOT invoked
rec = next(r for r in caplog.records if "disconnect" in r.getMessage().lower())
assert rec.levelno == logging.DEBUG
assert rec.name == "mempalace_mcp"
finally:
httpd.server_close()
def test_handle_error_delegates_real_errors(monkeypatch):
"""A genuine error is NOT misclassified as a disconnect: it reaches the
default handler, so its traceback is still surfaced.
"""
httpd = mcp._build_http_server("127.0.0.1", 0)
try:
delegated = []
monkeypatch.setattr(
socketserver.BaseServer,
"handle_error",
lambda self, request, addr: delegated.append(addr),
)
addr = ("127.0.0.1", 51234)
try:
raise ValueError("boom")
except ValueError:
httpd.handle_error(None, addr)
assert delegated == [addr]
finally:
httpd.server_close()
def _make_self_signed_cert(tmp_path):
"""Write a throwaway self-signed cert/key via openssl; skip if unavailable."""
import shutil
import subprocess
if shutil.which("openssl") is None:
pytest.skip("openssl not available to generate a test certificate")
cert = tmp_path / "cert.pem"
key = tmp_path / "key.pem"
subprocess.run(
[
"openssl",
"req",
"-x509",
"-newkey",
"rsa:2048",
"-keyout",
str(key),
"-out",
str(cert),
"-days",
"1",
"-nodes",
"-subj",
"/CN=localhost",
],
check=True,
capture_output=True,
)
return cert, key
def test_tls_serves_https(tmp_path, monkeypatch):
"""With --tls-cert/--tls-key (via env), the server speaks TLS: a plain HTTP
client cannot read it, and an HTTPS client trusting the cert can."""
import ssl
cert, key = _make_self_signed_cert(tmp_path)
monkeypatch.setenv("MEMPALACE_MCP_TLS_CERT", str(cert))
monkeypatch.setenv("MEMPALACE_MCP_TLS_KEY", str(key))
httpd = mcp._build_http_server("127.0.0.1", 0)
assert getattr(httpd, "scheme", "http") == "https"
port = httpd.server_address[1]
thread = threading.Thread(
target=httpd.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True
)
thread.start()
try:
# Full verification on: trust the self-signed cert as the CA and dial
# "localhost" (the cert CN, resolves to 127.0.0.1) so hostname checking
# passes without being disabled.
ctx = ssl.create_default_context(cafile=str(cert))
conn = http.client.HTTPSConnection("localhost", port, context=ctx, timeout=5)
try:
conn.request("GET", "/healthz")
resp = conn.getresponse()
assert resp.status == 200
assert resp.read() == b"ok\n"
finally:
conn.close()
# A plaintext HTTP client must NOT be able to talk to the TLS socket.
with pytest.raises(Exception):
plain = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
plain.request("GET", "/healthz")
plain.getresponse()
finally:
httpd.shutdown()
httpd.server_close()
thread.join(timeout=5)
def test_tls_requires_both_cert_and_key(tmp_path, monkeypatch):
"""A cert without a key (or vice versa) is a startup error, not a silent skip."""
cert, _key = _make_self_signed_cert(tmp_path)
monkeypatch.setenv("MEMPALACE_MCP_TLS_CERT", str(cert))
monkeypatch.delenv("MEMPALACE_MCP_TLS_KEY", raising=False)
with pytest.raises(ValueError, match="both"):
mcp._build_http_server("127.0.0.1", 0)
def test_loopback_and_origin_helpers():
assert mcp._http_is_loopback("127.0.0.1")
assert mcp._http_is_loopback("localhost")
assert not mcp._http_is_loopback("0.0.0.0")
assert not mcp._http_is_loopback("192.168.1.10")
assert mcp._http_origin_allowed("http://127.0.0.1:8765")
assert mcp._http_origin_allowed("http://localhost")
assert not mcp._http_origin_allowed("https://evil.example")
assert not mcp._http_origin_allowed("garbage")
allowed = mcp._http_allowed_host_values("127.0.0.1", 8765)
assert "127.0.0.1:8765" in allowed and "localhost" in allowed
def test_extra_allowed_hosts_extend_the_loopback_pin(monkeypatch):
"""A loopback-bound server behind a fronting proxy (tailscale serve, nginx)
receives the public name in Host; the operator allowlists it via env."""
monkeypatch.setenv(
"MEMPALACE_MCP_EXTRA_ALLOWED_HOSTS",
"mybox.tail1234.ts.net, Proxy.Example:8443",
)
allowed = mcp._http_allowed_host_values("127.0.0.1", 8765)
# Bare hostname matches with and without the bound port.
assert "mybox.tail1234.ts.net" in allowed
assert "mybox.tail1234.ts.net:8765" in allowed
# host:port entries match exactly (lowercased); no bound-port variant added.
assert "proxy.example:8443" in allowed
assert "proxy.example" not in allowed
# The loopback pin itself is unchanged.
assert "127.0.0.1:8765" in allowed
def test_extra_allowed_hosts_default_empty(monkeypatch):
monkeypatch.delenv("MEMPALACE_MCP_EXTRA_ALLOWED_HOSTS", raising=False)
allowed = mcp._http_allowed_host_values("127.0.0.1", 8765)
assert not any("ts.net" in v for v in allowed)
Reference in New Issue View Git Blame Copy Permalink