Validate Mcp-Param-* headers server-side on the 2026-07-28 HTTP path (SEP-2243) (#3033)
This commit is contained in:
Max
GitHub
parent
4df609119f
commit
48ef569f7e
1 changed files with 971 additions and 37 deletions
|
|
@ -6,6 +6,7 @@ or protocol-version literals — all facts are imported from their one source.
|
|||
"""
|
||||
|
||||
import dataclasses
|
||||
from collections.abc import Iterator, Mapping
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
|
@ -39,8 +40,10 @@ from mcp.shared.inbound import (
|
|||
classify_inbound_request,
|
||||
decode_header_value,
|
||||
encode_header_value,
|
||||
find_duplicated_routing_header,
|
||||
find_invalid_x_mcp_header,
|
||||
mcp_param_headers,
|
||||
validate_mcp_param_headers,
|
||||
x_mcp_header_map,
|
||||
)
|
||||
|
||||
|
|
@ -574,3 +577,273 @@ def test_mcp_param_headers_omits_when_nested_path_is_broken() -> None:
|
|||
header_map = {("outer", "inner"): "Inner"}
|
||||
assert mcp_param_headers(header_map, {"outer": "not-a-mapping"}) == {}
|
||||
assert mcp_param_headers(header_map, {}) == {}
|
||||
|
||||
|
||||
# --- validate_mcp_param_headers --------------------------------------------
|
||||
|
||||
REGION_SCHEMA: dict[str, Any] = {
|
||||
"type": "object",
|
||||
"properties": {"region": {"type": "string", "x-mcp-header": "Region"}},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("argument", "header"),
|
||||
[
|
||||
pytest.param("Hello", "Hello", id="plain-literal"),
|
||||
pytest.param("Hello", "=?base64?SGVsbG8=?=", id="valid-sentinel"),
|
||||
pytest.param("", "=?base64??=", id="empty-sentinel"),
|
||||
pytest.param("SGVsbG8=", "SGVsbG8=", id="missing-prefix-is-literal"),
|
||||
pytest.param("=?base64?SGVsbG8=", "=?base64?SGVsbG8=", id="missing-suffix-is-literal"),
|
||||
],
|
||||
)
|
||||
def test_validate_mcp_param_headers_accepts_agreeing_header_and_argument(argument: str, header: str) -> None:
|
||||
"""Spec Value Encoding: a fully-wrapped sentinel decodes before comparison; anything else is a literal."""
|
||||
assert validate_mcp_param_headers(REGION_SCHEMA, {"region": argument}, {"Mcp-Param-Region": header}) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"header",
|
||||
[
|
||||
pytest.param("=?base64?SGVsbG8?=", id="missing-padding"),
|
||||
pytest.param("=?base64?SGVs!!!bG8=?=", id="non-alphabet-chars"),
|
||||
pytest.param("=?base64?SGVsbG9=?=", id="non-canonical-trailing-bits"),
|
||||
pytest.param("=?base64?gA==?=", id="invalid-utf8"),
|
||||
],
|
||||
)
|
||||
def test_validate_mcp_param_headers_rejects_malformed_sentinel(header: str) -> None:
|
||||
"""Spec: servers MUST reject a recognized header whose sentinel cannot be strictly decoded — not a literal."""
|
||||
rejection = assert_rejected(
|
||||
validate_mcp_param_headers(REGION_SCHEMA, {"region": "Hello"}, {"Mcp-Param-Region": header}),
|
||||
HEADER_MISMATCH,
|
||||
)
|
||||
assert "malformed base64" in rejection.message
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_rejects_missing_header_for_present_argument() -> None:
|
||||
"""Spec table: client omits the header but the value is in the body → server MUST reject."""
|
||||
rejection = assert_rejected(
|
||||
validate_mcp_param_headers(REGION_SCHEMA, {"region": "test-value"}, {}), HEADER_MISMATCH
|
||||
)
|
||||
assert "missing" in rejection.message
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"arguments",
|
||||
[pytest.param({}, id="absent"), pytest.param({"region": None}, id="null")],
|
||||
)
|
||||
def test_validate_mcp_param_headers_rejects_orphan_header_for_absent_or_null_argument(
|
||||
arguments: dict[str, Any],
|
||||
) -> None:
|
||||
"""SDK-defined posture on a spec gap: an orphan header is the routing-spoof case; go rejects too, ts skips."""
|
||||
rejection = assert_rejected(
|
||||
validate_mcp_param_headers(REGION_SCHEMA, arguments, {"Mcp-Param-Region": "eu"}), HEADER_MISMATCH
|
||||
)
|
||||
assert "absent" in rejection.message
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_rejects_value_mismatch() -> None:
|
||||
rejection = assert_rejected(
|
||||
validate_mcp_param_headers(REGION_SCHEMA, {"region": "us"}, {"Mcp-Param-Region": "eu"}), HEADER_MISMATCH
|
||||
)
|
||||
assert "does not match" in rejection.message
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_accepts_absent_argument_with_no_header() -> None:
|
||||
"""Spec table: parameter not in arguments / null → client MUST omit, server MUST NOT expect."""
|
||||
assert validate_mcp_param_headers(REGION_SCHEMA, {}, {}) is None
|
||||
assert validate_mcp_param_headers(REGION_SCHEMA, {"region": None}, {}) is None
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_matches_header_names_case_insensitively() -> None:
|
||||
"""Spec Case Sensitivity: header-name comparison MUST be case-insensitive."""
|
||||
assert validate_mcp_param_headers(REGION_SCHEMA, {"region": "eu"}, {"MCP-PARAM-REGION": "eu"}) is None
|
||||
rejection = validate_mcp_param_headers(REGION_SCHEMA, {"region": "us"}, {"MCP-PARAM-REGION": "eu"})
|
||||
assert_rejected(rejection, HEADER_MISMATCH)
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_ignores_undeclared_mcp_param_headers() -> None:
|
||||
"""Spec: an undeclared `Mcp-Param-*` header is unrecognized — forwarded and ignored, never a failure."""
|
||||
headers = {"Mcp-Param-Region": "eu", "Mcp-Param-Undeclared": "=?base64?not even base64?="}
|
||||
assert validate_mcp_param_headers(REGION_SCHEMA, {"region": "eu"}, headers) is None
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_validates_nothing_for_an_invalid_annotation_schema() -> None:
|
||||
"""Spec gives definition rejection to clients (drop the tool), so an invalid schema recognizes no headers."""
|
||||
invalid = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"region": {"type": "string", "x-mcp-header": "Region"},
|
||||
"dupe": {"type": "string", "x-mcp-header": "region"},
|
||||
},
|
||||
}
|
||||
assert find_invalid_x_mcp_header(invalid) is not None
|
||||
assert validate_mcp_param_headers(invalid, {"region": "us"}, {"Mcp-Param-Region": "eu"}) is None
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_reads_nested_argument_paths() -> None:
|
||||
"""A nested annotated property compares against the matching nested `arguments` path; a broken path is absent."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"outer": {"type": "object", "properties": {"inner": {"type": "string", "x-mcp-header": "Inner"}}}
|
||||
},
|
||||
}
|
||||
assert validate_mcp_param_headers(schema, {"outer": {"inner": "deep"}}, {"Mcp-Param-Inner": "deep"}) is None
|
||||
rejection = validate_mcp_param_headers(schema, {"outer": {"inner": "deep"}}, {"Mcp-Param-Inner": "other"})
|
||||
assert_rejected(rejection, HEADER_MISMATCH)
|
||||
assert validate_mcp_param_headers(schema, {"outer": "not-a-mapping"}, {}) is None
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_compares_booleans_against_true_false_rendering() -> None:
|
||||
"""Booleans compare against the lowercase `true`/`false` rendering the client emits."""
|
||||
schema = {"type": "object", "properties": {"flag": {"type": "boolean", "x-mcp-header": "Flag"}}}
|
||||
assert validate_mcp_param_headers(schema, {"flag": True}, {"Mcp-Param-Flag": "true"}) is None
|
||||
assert validate_mcp_param_headers(schema, {"flag": False}, {"Mcp-Param-Flag": "false"}) is None
|
||||
rejection = validate_mcp_param_headers(schema, {"flag": True}, {"Mcp-Param-Flag": "True"})
|
||||
assert_rejected(rejection, HEADER_MISMATCH)
|
||||
|
||||
|
||||
INTEGER_SCHEMA: dict[str, Any] = {
|
||||
"type": "object",
|
||||
"properties": {"n": {"type": "integer", "x-mcp-header": "N"}},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("body_value", "header", "matches"),
|
||||
[
|
||||
pytest.param(42, "42", True, id="exact"),
|
||||
pytest.param(42, "42.0", True, id="trailing-zero-fraction"),
|
||||
pytest.param(42, "42.000", True, id="long-zero-fraction"),
|
||||
pytest.param(42, "42.5", False, id="real-fraction"),
|
||||
pytest.param(42, "1e2", False, id="scientific-notation-never-numeric"),
|
||||
pytest.param(42, "43", False, id="different-value"),
|
||||
pytest.param(-7, "-7.0", True, id="negative"),
|
||||
pytest.param(9007199254740993, "9007199254740993", True, id="beyond-ieee754-safe-range-exact"),
|
||||
pytest.param(9007199254740993, "9007199254740992", False, id="beyond-ieee754-safe-range-off-by-one"),
|
||||
],
|
||||
)
|
||||
def test_validate_mcp_param_headers_compares_integers_numerically(body_value: int, header: str, matches: bool) -> None:
|
||||
"""Spec SHOULD: integers compare numerically (`42` == `42.0`) — gated to canonical decimals, compared exactly."""
|
||||
result = validate_mcp_param_headers(INTEGER_SCHEMA, {"n": body_value}, {"Mcp-Param-N": header})
|
||||
if matches:
|
||||
assert result is None
|
||||
else:
|
||||
assert_rejected(result, HEADER_MISMATCH)
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_non_primitive_body_value_rejects_only_when_a_header_claims_it() -> None:
|
||||
"""A header claiming a non-primitive argument is a mismatch; without one, rejection is params validation's job."""
|
||||
rejection = assert_rejected(
|
||||
validate_mcp_param_headers(REGION_SCHEMA, {"region": {"k": "v"}}, {"Mcp-Param-Region": "x"}),
|
||||
HEADER_MISMATCH,
|
||||
)
|
||||
assert "does not match" in rejection.message
|
||||
assert validate_mcp_param_headers(REGION_SCHEMA, {"region": {"k": "v"}}, {}) is None
|
||||
assert validate_mcp_param_headers(REGION_SCHEMA, {"region": [1, 2]}, {}) is None
|
||||
|
||||
|
||||
class _RepeatedHeaders(Mapping[str, str]):
|
||||
"""A header carrier whose `items()` yields duplicate names, like a raw HTTP header list."""
|
||||
|
||||
def __init__(self, pairs: list[tuple[str, str]]) -> None:
|
||||
self._pairs = pairs
|
||||
|
||||
def __getitem__(self, key: str) -> str:
|
||||
return next(value for name, value in self._pairs if name == key)
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return (name for name, _ in self._pairs)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._pairs)
|
||||
|
||||
def items(self) -> Any:
|
||||
return list(self._pairs)
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_rejects_a_recognized_header_supplied_more_than_once() -> None:
|
||||
"""Duplicate recognized headers reject even when one matches: first-copy readers and last-wins checks diverge."""
|
||||
headers = _RepeatedHeaders([("Mcp-Param-Region", "spoofed"), ("mcp-param-region", "eu")])
|
||||
# The carrier behaves like a raw header list: first-wins lookup, every line iterated.
|
||||
assert headers["Mcp-Param-Region"] == "spoofed"
|
||||
assert len(headers) == len(list(headers)) == 2
|
||||
rejection = assert_rejected(validate_mcp_param_headers(REGION_SCHEMA, {"region": "eu"}, headers), HEADER_MISMATCH)
|
||||
assert "more than once" in rejection.message
|
||||
noisy = _RepeatedHeaders([("Mcp-Param-Region", "eu"), ("Mcp-Param-Other", "a"), ("mcp-param-other", "b")])
|
||||
assert validate_mcp_param_headers(REGION_SCHEMA, {"region": "eu"}, noisy) is None
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_rejects_a_header_exceeding_the_int_conversion_limit() -> None:
|
||||
"""A canonical-decimal header beyond CPython's int-conversion digit limit is a clean mismatch, never an error."""
|
||||
rejection = validate_mcp_param_headers(INTEGER_SCHEMA, {"n": 1}, {"Mcp-Param-N": "1" * 5000})
|
||||
assert_rejected(rejection, HEADER_MISMATCH)
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_compares_integral_float_bodies_numerically() -> None:
|
||||
"""JSON Schema admits `42.0` as an integer; the numeric SHOULD applies in both directions."""
|
||||
assert validate_mcp_param_headers(INTEGER_SCHEMA, {"n": 42.0}, {"Mcp-Param-N": "42"}) is None
|
||||
assert validate_mcp_param_headers(INTEGER_SCHEMA, {"n": 42.0}, {"Mcp-Param-N": "42.0"}) is None
|
||||
assert_rejected(validate_mcp_param_headers(INTEGER_SCHEMA, {"n": 42.0}, {"Mcp-Param-N": "43"}), HEADER_MISMATCH)
|
||||
# A genuinely fractional body value falls back to the exact string rendering.
|
||||
assert validate_mcp_param_headers(INTEGER_SCHEMA, {"n": 42.5}, {"Mcp-Param-N": "42.5"}) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
[
|
||||
pytest.param("=?base64?SGVsbG9=?=", id="non-canonical-trailing-bits"),
|
||||
pytest.param("=?base64?SGVsbG8?=", id="missing-padding"),
|
||||
],
|
||||
)
|
||||
def test_decode_header_value_returns_none_for_non_canonical_base64(value: str) -> None:
|
||||
"""Canonical base64 only: a payload that decodes but does not re-encode byte-identically is malformed."""
|
||||
assert decode_header_value(value) is None
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_union_typed_annotation_invalidates_the_whole_tool() -> None:
|
||||
"""A union-typed annotation fails the integer/string/boolean-only rule, so the whole schema validates nothing."""
|
||||
union_schema = {
|
||||
"type": "object",
|
||||
"properties": {"n": {"type": ["integer", "null"], "x-mcp-header": "N"}},
|
||||
}
|
||||
assert find_invalid_x_mcp_header(union_schema) is not None
|
||||
assert validate_mcp_param_headers(union_schema, {"n": 42}, {"Mcp-Param-N": "999"}) is None
|
||||
assert validate_mcp_param_headers(union_schema, {"n": 42}, {}) is None
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_accepts_the_clients_own_rendering_of_large_integral_floats() -> None:
|
||||
"""A non-canonical-decimal header falls back to rendered comparison, so the client's own mirroring round-trips."""
|
||||
emitted = mcp_param_headers(x_mcp_header_map(INTEGER_SCHEMA), {"n": 1e16})
|
||||
assert emitted == {"Mcp-Param-N": "1e+16"}
|
||||
assert validate_mcp_param_headers(INTEGER_SCHEMA, {"n": 1e16}, emitted) is None
|
||||
assert_rejected(validate_mcp_param_headers(INTEGER_SCHEMA, {"n": 42}, {"Mcp-Param-N": "1e2"}), HEADER_MISMATCH)
|
||||
|
||||
|
||||
def test_validate_mcp_param_headers_handles_unrenderable_huge_integer_bodies_without_raising() -> None:
|
||||
"""An integer beyond CPython's int-to-str digit limit has no rendering: claimed → mismatch, unclaimed → fine."""
|
||||
huge = 10**5000
|
||||
rejection = validate_mcp_param_headers(REGION_SCHEMA, {"region": huge}, {"Mcp-Param-Region": "x"})
|
||||
assert_rejected(rejection, HEADER_MISMATCH)
|
||||
assert validate_mcp_param_headers(REGION_SCHEMA, {"region": huge}, {}) is None
|
||||
assert validate_mcp_param_headers(INTEGER_SCHEMA, {"n": huge}, {}) is None
|
||||
|
||||
|
||||
def test_mcp_param_headers_omits_values_with_no_scalar_rendering() -> None:
|
||||
"""Objects, arrays, and over-limit integers have no scalar rendering, so the client omits the header."""
|
||||
header_map = {("v",): "Val"}
|
||||
assert mcp_param_headers(header_map, {"v": {"k": 1}}) == {}
|
||||
assert mcp_param_headers(header_map, {"v": [1, 2]}) == {}
|
||||
assert mcp_param_headers(header_map, {"v": 10**5000}) == {}
|
||||
|
||||
|
||||
def test_find_duplicated_routing_header_detects_repeats_of_routing_headers_only() -> None:
|
||||
"""Repeated routing headers report case-insensitively; `Mcp-Param-*` or unrelated repeats are ignored."""
|
||||
assert find_duplicated_routing_header([("Mcp-Name", "a"), ("mcp-name", "b")]) == MCP_NAME_HEADER
|
||||
assert find_duplicated_routing_header([("MCP-Protocol-Version", "x"), ("mcp-protocol-version", "x")]) == (
|
||||
MCP_PROTOCOL_VERSION_HEADER
|
||||
)
|
||||
assert find_duplicated_routing_header([("Mcp-Method", "a"), ("Mcp-Method", "a")]) == MCP_METHOD_HEADER
|
||||
assert find_duplicated_routing_header([("Mcp-Name", "a"), ("Mcp-Param-X", "1"), ("Mcp-Param-X", "2")]) is None
|
||||
assert find_duplicated_routing_header([("accept", "a"), ("accept", "b")]) is None
|
||||
|
|
|
|||
Loading…
Reference in a new issue