gabriel / musehub public
test_mcp_protocol.py python
663 lines 23.2 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Section 13 — MCP Protocol Layer: 7-layer test suite.
2
3 Covers gaps not addressed by the existing 102 tests in:
4 - test_mcp_dispatcher.py (protocol correctness, tools, resources, prompts)
5 - test_mcp_streamable_http.py (transport, session CRUD, origin, lifecycle)
6 - test_mcp_elicitation.py (elicitation flows, progress, interactive tools)
7
8 New coverage in this file:
9
10 Layer 1 Unit:
11 - _validate_origin: URL with path (path stripped), URL with port, malformed URL
12 - MCPSession.is_expired(): elapsed > TTL → True; elapsed < TTL → False
13 - MCPSession.touch(): resets last_active, deferring expiry
14 - MCPSession.supports_elicitation_form(): empty-dict variant (backwards compat)
15 - push_to_session ring buffer: capped at 50 — oldest dropped
16 - create_session stores anonymous user_id correctly
17
18 Layer 2 Integration:
19 - create_session returns unique session IDs for each call
20 - delete_session puts None sentinel to all registered SSE queues
21 - delete_session cancels pending asyncio Futures
22 - get_session evicts expired session, returns None
23 - push_to_session broadcasts to multiple queues simultaneously
24
25 Layer 3 E2E (HTTP):
26 - Full lifecycle: initialize → ping (with session) → DELETE
27 - 127.0.0.1 origin always allowed
28 - Origin containing allow-listed domain as a path component is rejected
29 - Batch with notification mixed: response list excludes notification
30 - Empty batch array returns 200 with empty list
31
32 Layer 4 Stress:
33 - 50-item ping batch → 50 responses
34 - push_to_session to 10 queues simultaneously — all receive event
35
36 Layer 5 Data Integrity:
37 - 100 sessions have 100 distinct IDs (no collisions)
38 - Ring buffer stays at ≤50 entries after 60 pushes
39 - Session user_id and client_capabilities preserved
40
41 Layer 6 Security:
42 - http://127.0.0.1 origin always allowed (_ALWAYS_ALLOW_ORIGINS)
43 - http://127.0.0.1:8080 (non-standard port) is accepted (part of always-allow netloc)
44 - Origin with path suffix does not expand allow list
45 - Non-initialize POST without Mcp-Session-Id routes to dispatcher (no crash)
46
47 Layer 7 Performance:
48 - 100× handle_request("ping") under 100 ms
49 - 1000× MCPSession.is_expired() under 10 ms
50 - 100× push_to_session under 50 ms
51 """
52 from __future__ import annotations
53
54 import asyncio
55 import time
56 from unittest.mock import patch
57
58 import pytest
59 import pytest_asyncio
60 from httpx import AsyncClient, ASGITransport
61 from sqlalchemy.ext.asyncio import AsyncSession
62
63 from musehub.main import app
64 from musehub.mcp.dispatcher import handle_request
65 from musehub.muse_contracts.json_types import JSONObject, StrDict
66 from musehub.mcp.session import (
67 MCPSession,
68 SessionCapacityError,
69 create_session,
70 delete_session,
71 get_session,
72 push_to_session,
73 create_pending_elicitation,
74 )
75 from musehub.api.routes.mcp import _validate_origin
76
77
78 # ── Fixtures ──────────────────────────────────────────────────────────────────
79
80
81 @pytest.fixture
82 def anyio_backend() -> str:
83 return "asyncio"
84
85
86 @pytest_asyncio.fixture
87 async def http_client(db_session: AsyncSession) -> AsyncClient:
88 async with AsyncClient(
89 transport=ASGITransport(app=app),
90 base_url="http://localhost",
91 ) as c:
92 yield c
93
94
95 def _init_body() -> JSONObject:
96 return {
97 "jsonrpc": "2.0",
98 "id": 1,
99 "method": "initialize",
100 "params": {
101 "protocolVersion": "2025-11-25",
102 "clientInfo": {"name": "test-client", "version": "1.0"},
103 "capabilities": {"elicitation": {"form": {}}},
104 },
105 }
106
107
108 def _req(method: str, params: JSONObject | None = None, req_id: int = 1) -> JSONObject:
109 msg = {"jsonrpc": "2.0", "id": req_id, "method": method}
110 if params is not None:
111 msg["params"] = params
112 return msg
113
114
115 class _FakeRequest:
116 """Minimal stub implementing the subset of Request used by _validate_origin."""
117
118 def __init__(self, origin: str | None) -> None:
119 self.headers: StrDict = (
120 {"Origin": origin} if origin is not None else {}
121 )
122 self.headers = _CaseInsensitiveDict(self.headers)
123
124
125 class _CaseInsensitiveDict(dict[str, str]):
126 def get(self, key: str, default: str | None = None) -> str | None:
127 return super().get(key.lower(), default) or super().get(key, default)
128
129
130 # ── Layer 1 — Unit ────────────────────────────────────────────────────────────
131
132
133 class TestUnitValidateOrigin:
134 """_validate_origin edge cases not tested elsewhere."""
135
136 def test_no_origin_returns_true(self) -> None:
137 assert _validate_origin(_FakeRequest(None)) is True
138
139 def test_localhost_no_port_returns_true(self) -> None:
140 assert _validate_origin(_FakeRequest("http://localhost")) is True
141
142 def test_127_0_0_1_returns_true(self) -> None:
143 assert _validate_origin(_FakeRequest("http://127.0.0.1")) is True
144
145 def test_localhost_with_path_still_resolves(self) -> None:
146 """Path component must be stripped; localhost is still allow-listed."""
147 assert _validate_origin(_FakeRequest("http://localhost/some/path")) is True
148
149 def test_evil_origin_returns_false(self) -> None:
150 assert _validate_origin(_FakeRequest("https://evil.example.com")) is False
151
152 def test_evil_with_localhost_path_returns_false(self) -> None:
153 """Attacker embedding 'localhost' in the path must not bypass the check."""
154 assert (
155 _validate_origin(
156 _FakeRequest("https://evil.example.com/localhost")
157 )
158 is False
159 )
160
161 def test_malformed_url_treated_as_allowed_or_false(self) -> None:
162 """A URL that causes urlparse to produce empty netloc is rejected."""
163 # urlparse("not a url") → scheme='', netloc='' → normalised='://'
164 # '://' is not in either allow set.
165 result = _validate_origin(_FakeRequest("not a url at all"))
166 assert result is False
167
168
169 class TestUnitMCPSessionExpiry:
170 """MCPSession.is_expired() and touch() temporal semantics."""
171
172 def test_fresh_session_is_not_expired(self) -> None:
173 s = create_session(None, {})
174 try:
175 assert s.is_expired() is False
176 finally:
177 delete_session(s.session_id)
178
179 def test_last_active_in_past_is_expired(self) -> None:
180 s = create_session(None, {})
181 try:
182 # Wind last_active back by more than the TTL.
183 s.last_active = time.monotonic() - 3700
184 assert s.is_expired() is True
185 finally:
186 delete_session(s.session_id)
187
188 def test_touch_defers_expiry(self) -> None:
189 """touch() resets last_active so the session is no longer expired."""
190 s = create_session(None, {})
191 try:
192 s.last_active = time.monotonic() - 3700 # force expired
193 assert s.is_expired() is True
194 s.touch()
195 assert s.is_expired() is False
196 finally:
197 delete_session(s.session_id)
198
199
200 class TestUnitElicitationCapabilities:
201 """supports_elicitation_form/url edge cases."""
202
203 def test_empty_elicitation_dict_counts_as_form(self) -> None:
204 """Empty elicitation dict ≡ form-only per spec backward compat."""
205 s = create_session(None, {"elicitation": {}})
206 try:
207 assert s.supports_elicitation_form() is True
208 finally:
209 delete_session(s.session_id)
210
211 def test_elicitation_not_a_dict_returns_false_for_form(self) -> None:
212 s = create_session(None, {"elicitation": True})
213 try:
214 assert s.supports_elicitation_form() is False
215 finally:
216 delete_session(s.session_id)
217
218 def test_elicitation_not_a_dict_returns_false_for_url(self) -> None:
219 s = create_session(None, {"elicitation": "url"})
220 try:
221 assert s.supports_elicitation_url() is False
222 finally:
223 delete_session(s.session_id)
224
225 def test_no_elicitation_key_returns_false(self) -> None:
226 s = create_session(None, {})
227 try:
228 assert s.supports_elicitation_form() is False
229 assert s.supports_elicitation_url() is False
230 finally:
231 delete_session(s.session_id)
232
233
234 class TestUnitRingBuffer:
235 """push_to_session ring buffer capping behaviour."""
236
237 def test_ring_buffer_capped_at_50(self) -> None:
238 s = create_session(None, {})
239 try:
240 for i in range(60):
241 push_to_session(s, f"data: event-{i}\n\n")
242 assert len(s.event_buffer) == 50
243 finally:
244 delete_session(s.session_id)
245
246 def test_ring_buffer_drops_oldest(self) -> None:
247 s = create_session(None, {})
248 try:
249 for i in range(55):
250 push_to_session(s, f"data: event-{i}\n\n")
251 # Oldest events (0–4) must be gone; event-5 must now be first.
252 first_text = s.event_buffer[0][1]
253 assert "event-5" in first_text
254 finally:
255 delete_session(s.session_id)
256
257
258 class TestUnitCreateSessionAnonymous:
259 def test_anonymous_session_user_id_is_none(self) -> None:
260 s = create_session(None, {})
261 try:
262 assert s.user_id is None
263 finally:
264 delete_session(s.session_id)
265
266 def test_authenticated_session_stores_user_id(self) -> None:
267 s = create_session("user-xyz", {"elicitation": {"form": {}}})
268 try:
269 assert s.user_id == "user-xyz"
270 assert s.client_capabilities == {"elicitation": {"form": {}}}
271 finally:
272 delete_session(s.session_id)
273
274
275 # ── Layer 2 — Integration ─────────────────────────────────────────────────────
276
277
278 class TestIntegrationSessionUniqueness:
279 def test_create_session_returns_unique_ids(self) -> None:
280 sessions = [create_session(None, {}) for _ in range(10)]
281 ids = [s.session_id for s in sessions]
282 try:
283 assert len(set(ids)) == 10
284 finally:
285 for s in sessions:
286 delete_session(s.session_id)
287
288
289 class TestIntegrationDeleteSessionSSE:
290 @pytest.mark.anyio
291 async def test_delete_sends_none_sentinel_to_queues(self) -> None:
292 """delete_session must put the None sentinel to all registered SSE queues."""
293 s = create_session(None, {})
294 q1: asyncio.Queue[str | None] = asyncio.Queue()
295 q2: asyncio.Queue[str | None] = asyncio.Queue()
296 s.sse_queues.extend([q1, q2])
297
298 delete_session(s.session_id)
299
300 assert q1.get_nowait() is None
301 assert q2.get_nowait() is None
302
303 @pytest.mark.anyio
304 async def test_delete_cancels_pending_futures(self) -> None:
305 """delete_session must cancel all unresolved elicitation Futures."""
306 s = create_session(None, {})
307 fut = create_pending_elicitation(s, "elicit-99")
308 assert not fut.done()
309
310 delete_session(s.session_id)
311
312 assert fut.cancelled()
313
314
315 class TestIntegrationExpiredSessionEviction:
316 def test_expired_session_evicted_by_get_session(self) -> None:
317 s = create_session(None, {})
318 sid = s.session_id
319 s.last_active = time.monotonic() - 3700 # force expired
320
321 result = get_session(sid)
322
323 assert result is None
324 # Confirm fully evicted (second call should also return None cleanly).
325 assert get_session(sid) is None
326
327
328 class TestIntegrationMultiQueueBroadcast:
329 @pytest.mark.anyio
330 async def test_push_broadcasts_to_all_queues(self) -> None:
331 s = create_session(None, {})
332 try:
333 queues: list[asyncio.Queue[str | None]] = [
334 asyncio.Queue() for _ in range(5)
335 ]
336 s.sse_queues.extend(queues)
337
338 push_to_session(s, "data: hello\n\n")
339
340 for q in queues:
341 item = q.get_nowait()
342 assert item == "data: hello\n\n"
343 finally:
344 delete_session(s.session_id)
345
346
347 # ── Layer 3 — End-to-End ──────────────────────────────────────────────────────
348
349
350 class TestE2EFullLifecycle:
351 @pytest.mark.anyio
352 async def test_initialize_ping_delete(self, http_client: AsyncClient) -> None:
353 """Full session lifecycle: initialize → ping → DELETE."""
354 # Initialize.
355 init_resp = await http_client.post(
356 "/mcp",
357 json=_init_body(),
358 headers={"Content-Type": "application/json"},
359 )
360 assert init_resp.status_code == 200
361 session_id = init_resp.headers["mcp-session-id"]
362
363 # Ping using the session.
364 ping_resp = await http_client.post(
365 "/mcp",
366 json=_req("ping"),
367 headers={
368 "Content-Type": "application/json",
369 "Mcp-Session-Id": session_id,
370 },
371 )
372 assert ping_resp.status_code == 200
373 assert ping_resp.json()["result"] == {}
374
375 # Delete the session.
376 del_resp = await http_client.delete(
377 "/mcp",
378 headers={"Mcp-Session-Id": session_id},
379 )
380 assert del_resp.status_code == 200
381 assert get_session(session_id) is None
382
383
384 class TestE2EOriginEdgeCases:
385 @pytest.mark.anyio
386 async def test_127_0_0_1_origin_allowed(self, http_client: AsyncClient) -> None:
387 """http://127.0.0.1 must always be allowed."""
388 resp = await http_client.post(
389 "/mcp",
390 json=_init_body(),
391 headers={
392 "Content-Type": "application/json",
393 "Origin": "http://127.0.0.1",
394 },
395 )
396 assert resp.status_code == 200
397 # Clean up.
398 if "mcp-session-id" in resp.headers:
399 delete_session(resp.headers["mcp-session-id"])
400
401 @pytest.mark.anyio
402 async def test_origin_with_localhost_in_path_rejected(
403 self, http_client: AsyncClient
404 ) -> None:
405 """Attacker embedding localhost in path must not bypass origin check."""
406 resp = await http_client.post(
407 "/mcp",
408 json=_init_body(),
409 headers={
410 "Content-Type": "application/json",
411 "Origin": "https://evil.example.com/localhost",
412 },
413 )
414 assert resp.status_code == 403
415
416
417 class TestE2EBatchWithNotification:
418 @pytest.mark.anyio
419 async def test_batch_excludes_notification(self, http_client: AsyncClient) -> None:
420 """Batch with a notification mixed in — response list must omit notification."""
421 batch = [
422 _req("ping", req_id=1),
423 # Notification (no id).
424 {"jsonrpc": "2.0", "method": "notifications/initialized"},
425 _req("ping", req_id=3),
426 ]
427 resp = await http_client.post(
428 "/mcp",
429 json=batch,
430 headers={"Content-Type": "application/json"},
431 )
432 assert resp.status_code == 200
433 data = resp.json()
434 assert isinstance(data, list)
435 # Only the two ping requests produce responses; notification is excluded.
436 assert len(data) == 2
437 ids = {item["id"] for item in data}
438 assert ids == {1, 3}
439
440 @pytest.mark.anyio
441 async def test_empty_batch_returns_202(
442 self, http_client: AsyncClient
443 ) -> None:
444 """An empty batch array produces no responses — treated as notification-only."""
445 resp = await http_client.post(
446 "/mcp",
447 json=[],
448 headers={"Content-Type": "application/json"},
449 )
450 assert resp.status_code == 202
451
452
453 # ── Layer 4 — Stress ──────────────────────────────────────────────────────────
454
455
456 class TestStressBatch:
457 @pytest.mark.anyio
458 async def test_50_item_ping_batch(self, http_client: AsyncClient) -> None:
459 """50-item ping batch must return exactly 50 responses."""
460 batch = [_req("ping", req_id=i) for i in range(50)]
461 resp = await http_client.post(
462 "/mcp",
463 json=batch,
464 headers={"Content-Type": "application/json"},
465 )
466 assert resp.status_code == 200
467 data = resp.json()
468 assert isinstance(data, list)
469 assert len(data) == 50
470 for item in data:
471 assert item["result"] == {}
472
473
474 class TestStressPushToMultipleQueues:
475 @pytest.mark.anyio
476 async def test_push_to_10_queues(self) -> None:
477 """push_to_session to 10 registered queues — all receive the event."""
478 s = create_session(None, {})
479 try:
480 queues: list[asyncio.Queue[str | None]] = [
481 asyncio.Queue() for _ in range(10)
482 ]
483 s.sse_queues.extend(queues)
484
485 push_to_session(s, "data: stress\n\n")
486
487 for q in queues:
488 assert q.get_nowait() == "data: stress\n\n"
489 finally:
490 delete_session(s.session_id)
491
492
493 # ── Layer 5 — Data Integrity ──────────────────────────────────────────────────
494
495
496 class TestDataIntegritySessionIDs:
497 def test_100_sessions_have_unique_ids(self) -> None:
498 sessions = [create_session(None, {}) for _ in range(100)]
499 ids = [s.session_id for s in sessions]
500 try:
501 assert len(set(ids)) == 100
502 finally:
503 for s in sessions:
504 delete_session(s.session_id)
505
506
507 class TestDataIntegrityRingBuffer:
508 def test_ring_buffer_never_exceeds_50(self) -> None:
509 s = create_session(None, {})
510 try:
511 for i in range(60):
512 push_to_session(s, f"data: {i}\n\n")
513 assert len(s.event_buffer) <= 50
514 finally:
515 delete_session(s.session_id)
516
517 def test_ring_buffer_content_after_exact_50_pushes(self) -> None:
518 s = create_session(None, {})
519 try:
520 for i in range(50):
521 push_to_session(s, f"data: {i}\n\n")
522 assert len(s.event_buffer) == 50
523 assert "data: 0" in s.event_buffer[0][1]
524 assert "data: 49" in s.event_buffer[-1][1]
525 finally:
526 delete_session(s.session_id)
527
528
529 class TestDataIntegritySessionAttributes:
530 def test_session_user_id_preserved(self) -> None:
531 s = create_session("preserved-user", {"cap": "x"})
532 try:
533 fetched = get_session(s.session_id)
534 assert fetched is not None
535 assert fetched.user_id == "preserved-user"
536 assert fetched.client_capabilities == {"cap": "x"}
537 finally:
538 delete_session(s.session_id)
539
540 def test_delete_removes_session_from_store(self) -> None:
541 s = create_session(None, {})
542 sid = s.session_id
543 delete_session(sid)
544 assert get_session(sid) is None
545
546
547 # ── Layer 6 — Security ────────────────────────────────────────────────────────
548
549
550 class TestSecurityOrigin:
551 @pytest.mark.anyio
552 async def test_127_0_0_1_always_allowed(self, http_client: AsyncClient) -> None:
553 resp = await http_client.post(
554 "/mcp",
555 json=_init_body(),
556 headers={
557 "Content-Type": "application/json",
558 "Origin": "http://127.0.0.1",
559 },
560 )
561 assert resp.status_code == 200
562 if "mcp-session-id" in resp.headers:
563 delete_session(resp.headers["mcp-session-id"])
564
565 @pytest.mark.anyio
566 async def test_non_allowlisted_origin_rejected(
567 self, http_client: AsyncClient
568 ) -> None:
569 resp = await http_client.post(
570 "/mcp",
571 json=_init_body(),
572 headers={
573 "Content-Type": "application/json",
574 "Origin": "https://attacker.com",
575 },
576 )
577 assert resp.status_code == 403
578
579 @pytest.mark.anyio
580 async def test_origin_with_subdomain_rejected(
581 self, http_client: AsyncClient
582 ) -> None:
583 """localhost.attacker.com must not be confused with localhost."""
584 resp = await http_client.post(
585 "/mcp",
586 json=_init_body(),
587 headers={
588 "Content-Type": "application/json",
589 "Origin": "http://localhost.attacker.com",
590 },
591 )
592 assert resp.status_code == 403
593
594 @pytest.mark.anyio
595 async def test_no_origin_non_browser_allowed(
596 self, http_client: AsyncClient
597 ) -> None:
598 """curl / stdio bridges don't send Origin — must be permitted."""
599 resp = await http_client.post(
600 "/mcp",
601 json=_init_body(),
602 headers={"Content-Type": "application/json"},
603 )
604 assert resp.status_code == 200
605 if "mcp-session-id" in resp.headers:
606 delete_session(resp.headers["mcp-session-id"])
607
608
609 class TestSecuritySessionCapacity:
610 def test_session_capacity_error_on_overflow(self) -> None:
611 """create_session must raise SessionCapacityError when the store is full."""
612 from musehub.mcp import session as _session_mod
613
614 original_max = _session_mod._MAX_SESSIONS
615 _session_mod._MAX_SESSIONS = 0
616 try:
617 with pytest.raises(SessionCapacityError):
618 create_session(None, {})
619 finally:
620 _session_mod._MAX_SESSIONS = original_max
621
622
623 # ── Layer 7 — Performance ─────────────────────────────────────────────────────
624
625
626 class TestPerformanceDispatcher:
627 @pytest.mark.anyio
628 async def test_100_ping_requests_under_100ms(self) -> None:
629 """100× handle_request('ping') must complete in under 100 ms."""
630 session = create_session(None, {})
631 req = {"jsonrpc": "2.0", "id": 1, "method": "ping"}
632 start = time.perf_counter()
633 for _ in range(100):
634 await handle_request(req, session=session)
635 elapsed_ms = (time.perf_counter() - start) * 1000
636 delete_session(session.session_id)
637 assert elapsed_ms < 100, f"100 pings took {elapsed_ms:.1f} ms"
638
639
640 class TestPerformanceSessionOps:
641 def test_1000_is_expired_calls_under_10ms(self) -> None:
642 """1000× MCPSession.is_expired() must complete in under 10 ms."""
643 s = create_session(None, {})
644 try:
645 start = time.perf_counter()
646 for _ in range(1000):
647 s.is_expired()
648 elapsed_ms = (time.perf_counter() - start) * 1000
649 assert elapsed_ms < 10, f"1000× is_expired took {elapsed_ms:.1f} ms"
650 finally:
651 delete_session(s.session_id)
652
653 def test_100_push_to_session_under_50ms(self) -> None:
654 """100× push_to_session (no queues) must complete in under 50 ms."""
655 s = create_session(None, {})
656 try:
657 start = time.perf_counter()
658 for i in range(100):
659 push_to_session(s, f"data: {i}\n\n")
660 elapsed_ms = (time.perf_counter() - start) * 1000
661 assert elapsed_ms < 50, f"100× push_to_session took {elapsed_ms:.1f} ms"
662 finally:
663 delete_session(s.session_id)
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago