feat: RFC 6570 URI templates with operator-aware security (#2356)

tests/docs_src/test_uri_templates.py

@@ -0,0 +1,214 @@
+"""`docs/advanced/uri-templates.md`: every claim the page makes, proved against the real SDK."""
+
+from pathlib import Path
+
+import pytest
+from inline_snapshot import snapshot
+from mcp_types import INVALID_PARAMS, ErrorData, ResourceTemplate, TextResourceContents
+
+from docs_src.uri_templates import tutorial001, tutorial002, tutorial003, tutorial004, tutorial005
+from mcp import Client, MCPError
+from mcp.server import MCPServer
+from mcp.shared.path_security import PathEscapeError, contains_path_traversal, safe_join
+from mcp.shared.uri_template import InvalidUriTemplate, UriTemplate
+
+# See test_index.py for why this is a per-module mark and not a conftest hook.
+pytestmark = [pytest.mark.anyio, pytest.mark.filterwarnings("error::mcp.MCPDeprecationWarning")]
+
+
+async def test_simple_expansion_maps_the_segment_to_the_argument() -> None:
+    """tutorial001: `books://{isbn}` reads `books://978-...` and the matched string is the argument."""
+    async with Client(tutorial001.mcp) as client:
+        (content,) = (await client.read_resource("books://978-0441172719")).contents
+        assert isinstance(content, TextResourceContents)
+        assert content.text == snapshot('{\n  "title": "Dune",\n  "author": "Frank Herbert"\n}')
+
+
+async def test_an_int_parameter_is_converted_from_the_uri_string() -> None:
+    """tutorial001: `order_id: int` receives `12345`, not `"12345"`, so `order_id + 1` is `12346`."""
+    async with Client(tutorial001.mcp) as client:
+        (content,) = (await client.read_resource("orders://12345")).contents
+        assert isinstance(content, TextResourceContents)
+        assert content.text == snapshot('{\n  "order_id": 12345,\n  "next_order": 12346,\n  "status": "shipped"\n}')
+
+
+async def test_plus_keeps_the_slashes_in_the_captured_value() -> None:
+    """tutorial001: `{+path}` matches `printing/setup.md` as one value; a plain `{path}` would not."""
+    async with Client(tutorial001.mcp) as client:
+        (content,) = (await client.read_resource("manuals://printing/setup.md")).contents
+        assert isinstance(content, TextResourceContents)
+        assert content.text == "# Printer setup\n\nLoad paper, then power on."
+
+
+async def test_omitted_query_params_fall_through_to_function_defaults() -> None:
+    """tutorial001: `{?limit,sort}` is lenient. No query string means `limit=10, sort="newest"`."""
+    async with Client(tutorial001.mcp) as client:
+        (content,) = (await client.read_resource("reviews://978-0441172719")).contents
+        assert isinstance(content, TextResourceContents)
+        assert content.text == "10 newest reviews of Dune"
+
+
+async def test_a_query_param_overrides_only_the_default_it_names() -> None:
+    """tutorial001: `?sort=top` sets `sort` and leaves `limit` at its default."""
+    async with Client(tutorial001.mcp) as client:
+        (content,) = (await client.read_resource("reviews://978-0441172719?sort=top")).contents
+        assert isinstance(content, TextResourceContents)
+        assert content.text == "10 top reviews of Dune"
+
+
+async def test_exploded_path_arrives_as_a_list_of_segments() -> None:
+    """tutorial001: `{/path*}` splits `/fiction/sci-fi` into `["fiction", "sci-fi"]`."""
+    async with Client(tutorial001.mcp) as client:
+        (content,) = (await client.read_resource("shelves://browse/fiction/sci-fi")).contents
+        assert isinstance(content, TextResourceContents)
+        assert content.text == "catalog > fiction > sci-fi"
+
+
+def test_two_adjacent_variables_are_rejected_at_parse_time() -> None:
+    """'What the parser rejects': nothing separates `path` from `ext`, so the template is refused."""
+    with pytest.raises(InvalidUriTemplate) as exc_info:
+        UriTemplate.parse("manuals://{+path}{ext}")
+    assert str(exc_info.value) == snapshot(
+        "Variables 'path' and 'ext' are adjacent with no literal separator; matching cannot "
+        "determine where one ends and the other begins. Add a literal between them or use a single variable."
+    )
+
+
+def test_a_self_delimiting_operator_supplies_the_separator() -> None:
+    """'What the parser rejects': `{.ext}` contributes the `.` itself, so `{+path}{.ext}` is accepted."""
+    template = UriTemplate.parse("manuals://{+path}{.ext}")
+    assert template.match("manuals://printing/setup.md") == {"path": "printing/setup", "ext": "md"}
+
+
+def test_a_second_multi_segment_variable_is_rejected_at_parse_time() -> None:
+    """'What the parser rejects': two `{+...}` are ambiguous about which one absorbs an extra segment."""
+    with pytest.raises(InvalidUriTemplate) as exc_info:
+        UriTemplate.parse("copy://{+source}/to/{+destination}")
+    assert str(exc_info.value) == snapshot(
+        "Template contains more than one multi-segment variable ({+var}, {#var}, or explode modifier); "
+        "matching would be ambiguous"
+    )
+
+
+def test_a_query_parameter_without_a_python_default_is_rejected_at_decoration_time() -> None:
+    """'What the parser rejects': a client may omit `{?limit}`, so the bound parameter must declare a default."""
+    strict = MCPServer("Bookshop")
+    with pytest.raises(ValueError) as exc_info:
+
+        @strict.resource("reviews://{isbn}{?limit}")
+        def list_reviews(isbn: str, limit: int) -> None:
+            """Reviews of a book."""
+
+    assert str(exc_info.value) == snapshot(
+        "Resource 'reviews://{isbn}{?limit}': query parameter(s) ['limit'] have no default value. "
+        "A client may omit a {?...}/{&...} query parameter, so the matching handler parameter "
+        "must declare a default."
+    )
+
+
+async def test_traversal_is_rejected_before_the_handler_runs() -> None:
+    """The `!!! check`: `../` triggers `-32602` "Unknown resource" and `read_manual` is never called."""
+    async with Client(tutorial001.mcp) as client:
+        with pytest.raises(MCPError) as exc_info:
+            await client.read_resource("manuals://../etc/passwd")
+    assert exc_info.value.error == snapshot(
+        ErrorData(
+            code=INVALID_PARAMS,
+            message="Unknown resource: manuals://../etc/passwd",
+            data={"uri": "manuals://../etc/passwd"},
+        )
+    )
+
+
+def test_dotdot_is_a_component_check_not_a_substring_scan() -> None:
+    """The page's prose: `v1.0..v2.0` passes because `..` is not a standalone path segment."""
+    assert contains_path_traversal("../etc") is True
+    assert contains_path_traversal("v1.0..v2.0") is False
+
+
+async def test_safe_join_serves_a_file_inside_the_base_directory(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """tutorial002: `safe_join(DOCS_ROOT, path).read_text()` returns the file under the base."""
+    (tmp_path / "printing").mkdir()
+    (tmp_path / "printing" / "setup.md").write_text("# Printer setup")
+    monkeypatch.setattr(tutorial002, "DOCS_ROOT", tmp_path)
+    async with Client(tutorial002.mcp) as client:
+        (content,) = (await client.read_resource("manuals://printing/setup.md")).contents
+        assert isinstance(content, TextResourceContents)
+        assert content.text == "# Printer setup"
+
+
+def test_safe_join_raises_when_the_resolved_path_escapes_the_base(tmp_path: Path) -> None:
+    """tutorial002: a path that climbs out of `DOCS_ROOT` raises `PathEscapeError`."""
+    with pytest.raises(PathEscapeError):
+        safe_join(tmp_path, "../etc/passwd")
+
+
+async def test_exempt_params_lets_an_absolute_path_through() -> None:
+    """tutorial003: `exempt_params={"source"}` skips the checks for that one parameter."""
+    async with Client(tutorial003.mcp) as client:
+        (content,) = (await client.read_resource("imports://preview//srv/incoming/catalog.csv")).contents
+        assert isinstance(content, TextResourceContents)
+        assert content.text == "Would import from /srv/incoming/catalog.csv"
+
+
+async def test_server_wide_resource_security_relaxes_every_resource() -> None:
+    """tutorial003: `resource_security=ResourceSecurity(reject_path_traversal=False)` exempts the whole server."""
+    async with Client(tutorial003.relaxed) as client:
+        (content,) = (await client.read_resource("imports://preview/../sibling/catalog.csv")).contents
+        assert isinstance(content, TextResourceContents)
+        assert content.text == "Would import from ../sibling/catalog.csv"
+
+
+async def test_lowlevel_static_dispatch_lists_and_reads_by_exact_uri() -> None:
+    """tutorial004: the registry is the listing, and a known URI returns its text."""
+    async with Client(tutorial004.server) as client:
+        listed = (await client.list_resources()).resources
+        assert [r.uri for r in listed] == ["config://shop", "status://health"]
+        (content,) = (await client.read_resource("status://health")).contents
+        assert content == TextResourceContents(uri="status://health", text="ok")
+
+
+async def test_lowlevel_unknown_uri_raises() -> None:
+    """tutorial004: a URI outside the registry raises and surfaces as a protocol error."""
+    async with Client(tutorial004.server) as client:
+        with pytest.raises(MCPError):
+            await client.read_resource("config://missing")
+
+
+def test_uritemplate_match_returns_a_dict_or_none() -> None:
+    """tutorial005: `match()` extracts decoded variables, or `None` when the URI doesn't fit."""
+    assert tutorial005.TEMPLATES["manuals"].match("manuals://printing/setup.md") == {"path": "printing/setup.md"}
+    assert tutorial005.TEMPLATES["books"].match("manuals://nope") is None
+
+
+async def test_lowlevel_match_routes_the_request_to_the_right_template() -> None:
+    """tutorial005: two templates, one handler. Each concrete URI lands in its own branch."""
+    async with Client(tutorial005.server) as client:
+        (manual,) = (await client.read_resource("manuals://printing/setup.md")).contents
+        assert manual == TextResourceContents(uri="manuals://printing/setup.md", text="# Printer setup")
+        (book,) = (await client.read_resource("books://978-0441172719")).contents
+        assert book == TextResourceContents(uri="books://978-0441172719", text="Dune by Frank Herbert")
+
+
+async def test_lowlevel_handler_applies_the_safety_checks_itself() -> None:
+    """tutorial005: there is no default policy down here; `read_manual_safely` is the gate."""
+    async with Client(tutorial005.server) as client:
+        with pytest.raises(MCPError):
+            await client.read_resource("manuals://../etc/passwd")
+        with pytest.raises(MCPError):
+            await client.read_resource("nothing://matches")
+
+
+async def test_str_of_a_template_round_trips_to_the_original_string() -> None:
+    """tutorial005: `str(template)` is the source string, so the listing reuses the parsed templates."""
+    assert str(tutorial005.TEMPLATES["manuals"]) == "manuals://{+path}"
+    async with Client(tutorial005.server) as client:
+        result = await client.list_resource_templates()
+        assert result.resource_templates == snapshot(
+            [
+                ResourceTemplate(name="manuals", uri_template="manuals://{+path}"),
+                ResourceTemplate(name="books", uri_template="books://{isbn}"),
+            ]
+        )

tests/server/mcpserver/resources/test_resource_template.py

@@ -9,6 +9,152 @@ from pydantic import BaseModel
 from mcp.server.mcpserver import Context, MCPServer
 from mcp.server.mcpserver.exceptions import ResourceError
 from mcp.server.mcpserver.resources import FunctionResource, ResourceTemplate
+from mcp.server.mcpserver.resources.templates import (
+    DEFAULT_RESOURCE_SECURITY,
+    ResourceSecurity,
+    ResourceSecurityError,
+)
+
+
+def _make(uri_template: str, security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY) -> ResourceTemplate:
+    def handler(**kwargs: Any) -> str:
+        raise NotImplementedError  # these tests only exercise matches()
+
+    return ResourceTemplate.from_function(fn=handler, uri_template=uri_template, security=security)
+
+
+def test_matches_rfc6570_reserved_expansion():
+    # {+path} allows / — the feature the old regex implementation couldn't support
+    t = _make("file://docs/{+path}")
+    assert t.matches("file://docs/src/main.py") == {"path": "src/main.py"}
+
+
+def test_matches_rejects_encoded_slash_traversal():
+    # %2F decodes to / in UriTemplate.match(), giving "../../etc/passwd".
+    # ResourceSecurity's traversal check then rejects the '..' components.
+    t = _make("file://docs/{name}")
+    with pytest.raises(ResourceSecurityError, match="'name'"):
+        t.matches("file://docs/..%2F..%2Fetc%2Fpasswd")
+
+
+def test_matches_rejects_path_traversal_by_default():
+    t = _make("file://docs/{name}")
+    with pytest.raises(ResourceSecurityError):
+        t.matches("file://docs/..")
+
+
+def test_matches_rejects_path_traversal_in_reserved_var():
+    # Even {+path} gets the traversal check — it's semantic, not structural
+    t = _make("file://docs/{+path}")
+    with pytest.raises(ResourceSecurityError):
+        t.matches("file://docs/../../etc/passwd")
+
+
+def test_matches_rejects_absolute_path():
+    t = _make("file://docs/{+path}")
+    with pytest.raises(ResourceSecurityError):
+        t.matches("file://docs//etc/passwd")
+
+
+def test_matches_allows_dotdot_as_substring():
+    # .. is only dangerous as a path component
+    t = _make("git://refs/{range}")
+    assert t.matches("git://refs/v1.0..v2.0") == {"range": "v1.0..v2.0"}
+
+
+def test_matches_exempt_params_skip_security():
+    policy = ResourceSecurity(exempt_params={"range"})
+    t = _make("git://diff/{+range}", security=policy)
+    assert t.matches("git://diff/../foo") == {"range": "../foo"}
+
+
+def test_matches_disabled_policy_allows_traversal():
+    policy = ResourceSecurity(reject_path_traversal=False, reject_absolute_paths=False)
+    t = _make("file://docs/{name}", security=policy)
+    assert t.matches("file://docs/..") == {"name": ".."}
+
+
+def test_matches_rejects_null_byte_by_default():
+    # %00 decodes to \x00 which defeats string comparisons
+    # ("..\x00" != "..") and can truncate in C extensions.
+    t = _make("file://docs/{name}")
+    with pytest.raises(ResourceSecurityError):
+        t.matches("file://docs/key%00.txt")
+    # Null byte also defeats the traversal check's component comparison
+    with pytest.raises(ResourceSecurityError):
+        t.matches("file://docs/..%00%2Fsecret")
+
+
+def test_matches_null_byte_check_can_be_disabled():
+    policy = ResourceSecurity(reject_null_bytes=False)
+    t = _make("file://docs/{name}", security=policy)
+    assert t.matches("file://docs/key%00.txt") == {"name": "key\x00.txt"}
+
+
+def test_security_rejection_does_not_fall_through_to_next_template():
+    # A strict template's security rejection must halt iteration, not
+    # fall through to a later permissive template. Previously matches()
+    # returned None for both "no match" and "security failed", making
+    # registration order security-critical.
+    strict = _make("file://docs/{name}")
+    lax = _make(
+        "file://docs/{+path}",
+        security=ResourceSecurity(exempt_params={"path"}),
+    )
+    uri = "file://docs/..%2Fsecrets"
+    # Strict matches structurally then fails security -> raises.
+    with pytest.raises(ResourceSecurityError) as exc:
+        strict.matches(uri)
+    assert exc.value.param == "name"
+    # If this raised, the resource manager never reaches the lax
+    # template. Verify the lax template WOULD have accepted it.
+    assert lax.matches(uri) == {"path": "../secrets"}
+
+
+def test_matches_explode_checks_each_segment():
+    t = _make("api{/parts*}")
+    assert t.matches("api/a/b/c") == {"parts": ["a", "b", "c"]}
+    # Any segment with traversal rejects the whole match
+    with pytest.raises(ResourceSecurityError):
+        t.matches("api/a/../c")
+
+
+def test_matches_encoded_backslash_caught_by_traversal_check():
+    # %5C decodes to '\\'. The traversal check normalizes '\\' to '/'
+    # and catches the '..' components.
+    t = _make("file://docs/{name}")
+    with pytest.raises(ResourceSecurityError):
+        t.matches("file://docs/..%5C..%5Csecret")
+
+
+def test_matches_encoded_dots_caught_by_traversal_check():
+    # %2E%2E decodes to '..' which the traversal check rejects.
+    t = _make("file://docs/{name}")
+    with pytest.raises(ResourceSecurityError):
+        t.matches("file://docs/%2E%2E")
+
+
+def test_matches_mixed_encoded_and_literal_slash():
+    # The literal '/' stops the simple-var regex, so the URI doesn't
+    # match the template at all.
+    t = _make("file://docs/{name}")
+    assert t.matches("file://docs/..%2F../etc") is None
+
+
+def test_matches_encoded_slash_without_traversal_allowed():
+    # %2F decoding to '/' is fine when there's no traversal involved.
+    # UriTemplate accepts it; ResourceSecurity only blocks '..' and
+    # absolute paths. Handlers that need single-segment should use
+    # safe_join or validate explicitly.
+    t = _make("file://docs/{name}")
+    assert t.matches("file://docs/sub%2Ffile.txt") == {"name": "sub/file.txt"}
+
+
+def test_matches_escapes_template_literals():
+    # Regression: old impl treated . as regex wildcard
+    t = _make("data://v1.0/{id}")
+    assert t.matches("data://v1.0/42") == {"id": "42"}
+    assert t.matches("data://v1X0/42") is None
 
 
 class TestResourceTemplate:

tests/server/mcpserver/test_server.py

@@ -43,13 +43,14 @@ from starlette.routing import Mount, Route
 
 from mcp.client import Client
 from mcp.server.context import ServerRequestContext
-from mcp.server.mcpserver import Context, MCPServer
+from mcp.server.mcpserver import Context, MCPServer, ResourceSecurity
 from mcp.server.mcpserver.exceptions import ResourceNotFoundError, ToolError
 from mcp.server.mcpserver.prompts.base import Message, UserMessage
 from mcp.server.mcpserver.resources import FileResource, FunctionResource
 from mcp.server.mcpserver.utilities.types import Audio, Image
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
+from mcp.shared.uri_template import InvalidUriTemplate
 
 pytestmark = pytest.mark.anyio
 
@@ -862,7 +863,7 @@ class TestServerResourceTemplates:
         parameters don't match"""
         mcp = MCPServer()
 
-        with pytest.raises(ValueError, match="Mismatch between URI parameters"):
+        with pytest.raises(ValueError, match="has no URI template variables"):
 
             @mcp.resource("resource://data")
             def get_data_fn(param: str) -> str:  # pragma: no cover
@@ -1489,6 +1490,258 @@ class TestServerPrompts:
                 await client.get_prompt("prompt_fn")
 
 
+async def test_resource_decorator_rfc6570_reserved_expansion():
+    # Regression: old regex-based param extraction couldn't see `path`
+    # in `{+path}` and failed with a confusing mismatch error.
+    mcp = MCPServer()
+
+    @mcp.resource("file://docs/{+path}")
+    def read_doc(path: str) -> str:
+        raise NotImplementedError
+
+    templates = await mcp.list_resource_templates()
+    assert [t.uri_template for t in templates] == ["file://docs/{+path}"]
+
+
+async def test_resource_decorator_rejects_malformed_template():
+    mcp = MCPServer()
+    with pytest.raises(InvalidUriTemplate, match="Unclosed expression"):
+        mcp.resource("file://{name")
+
+
+async def test_resource_optional_query_params_use_function_defaults():
+    """Omitted {?...} query params should fall through to the
+    handler's Python defaults. Partial and reordered params work."""
+    mcp = MCPServer()
+
+    @mcp.resource("logs://{service}{?since,level}")
+    def tail_logs(service: str, since: str = "1h", level: str = "info") -> str:
+        return f"{service}|{since}|{level}"
+
+    async with Client(mcp) as client:
+        # No query → all defaults
+        r = await client.read_resource("logs://api")
+        assert isinstance(r.contents[0], TextResourceContents)
+        assert r.contents[0].text == "api|1h|info"
+
+        # Partial query → one default
+        r = await client.read_resource("logs://api?since=15m")
+        assert isinstance(r.contents[0], TextResourceContents)
+        assert r.contents[0].text == "api|15m|info"
+
+        # Reordered, both present
+        r = await client.read_resource("logs://api?level=error&since=5m")
+        assert isinstance(r.contents[0], TextResourceContents)
+        assert r.contents[0].text == "api|5m|error"
+
+        # Extra param ignored
+        r = await client.read_resource("logs://api?since=2h&utm=x")
+        assert isinstance(r.contents[0], TextResourceContents)
+        assert r.contents[0].text == "api|2h|info"
+
+
+async def test_resource_query_param_without_default_rejected_at_decoration():
+    """A handler parameter bound to a {?...} query variable must have a
+    Python default: a client may omit a query parameter, so the handler has
+    to be callable without it. Omitting the default is an error when the
+    decorator runs, not on the first request that leaves the parameter out."""
+    mcp = MCPServer()
+
+    with pytest.raises(ValueError, match=r"logs://.*\['level'\].*must declare a default"):
+
+        @mcp.resource("logs://{service}{?level}")
+        def tail_logs(service: str, level: str) -> str:
+            raise NotImplementedError
+
+
+async def test_resource_path_param_without_default_accepted():
+    """The default requirement applies only to query-bound parameters.
+    A path variable is always present in a matching URI, so its handler
+    parameter may be required."""
+    mcp = MCPServer()
+
+    @mcp.resource("logs://{service}{?level}")
+    def tail_logs(service: str, level: str = "info") -> str:
+        raise NotImplementedError
+
+    templates = await mcp.list_resource_templates()
+    assert [t.uri_template for t in templates] == ["logs://{service}{?level}"]
+
+
+async def test_resource_security_default_rejects_traversal():
+    mcp = MCPServer()
+
+    @mcp.resource("data://items/{name}")
+    def get_item(name: str) -> str:
+        return f"item:{name}"
+
+    async with Client(mcp) as client:
+        # Safe value passes through to the handler
+        r = await client.read_resource("data://items/widget")
+        assert isinstance(r.contents[0], TextResourceContents)
+        assert r.contents[0].text == "item:widget"
+
+        # ".." as a path component is rejected by default policy
+        with pytest.raises(MCPError, match="Unknown resource"):
+            await client.read_resource("data://items/..")
+
+
+async def test_resource_template_non_match_is_unknown_resource():
+    """A URI that doesn't satisfy a registered template — including one
+    shorter than the template's literal segments — must surface as the
+    standard -32602 Unknown resource, not an internal error."""
+    mcp = MCPServer()
+
+    @mcp.resource("api://{+path}/{id}")
+    def get(path: str, id: str) -> str:
+        return f"{path}|{id}"
+
+    async with Client(mcp) as client:
+        with pytest.raises(MCPError) as exc_info:
+            await client.read_resource("api://foo")
+        assert exc_info.value.error.code == INVALID_PARAMS
+        assert exc_info.value.error.message == "Unknown resource: api://foo"
+
+        # And a satisfying URI still routes to the handler.
+        r = await client.read_resource("api://a/b/c")
+        assert isinstance(r.contents[0], TextResourceContents)
+        assert r.contents[0].text == "a/b|c"
+
+
+async def test_resource_security_rejection_indistinguishable_from_not_found():
+    """A path-safety rejection must produce the same wire error as a
+    genuinely-absent resource: same code, same message shape, no hint
+    about which check failed."""
+    mcp = MCPServer()
+
+    @mcp.resource("data://items/{name}")
+    def get_item(name: str) -> str:  # pragma: no cover - never reached
+        return name
+
+    async with Client(mcp) as client:
+        with pytest.raises(MCPError) as rejected:
+            await client.read_resource("data://items/..")
+        with pytest.raises(MCPError) as absent:
+            await client.read_resource("nosuch://thing")
+
+        assert rejected.value.error.code == absent.value.error.code == INVALID_PARAMS
+        # Message echoes the requested URI and nothing else; no
+        # reference to which validation step rejected it.
+        assert rejected.value.error.message == "Unknown resource: data://items/.."
+        assert absent.value.error.message == "Unknown resource: nosuch://thing"
+        assert rejected.value.error.data == {"uri": "data://items/.."}
+        assert absent.value.error.data == {"uri": "nosuch://thing"}
+
+
+async def test_resource_security_per_resource_override():
+    mcp = MCPServer()
+
+    @mcp.resource(
+        "git://diff/{+range}",
+        security=ResourceSecurity(exempt_params={"range"}),
+    )
+    def git_diff(range: str) -> str:
+        return f"diff:{range}"
+
+    async with Client(mcp) as client:
+        # "../foo" would be rejected by default, but "range" is exempt
+        result = await client.read_resource("git://diff/../foo")
+        assert isinstance(result.contents[0], TextResourceContents)
+        assert result.contents[0].text == "diff:../foo"
+
+
+async def test_resource_security_server_wide_override():
+    mcp = MCPServer(resource_security=ResourceSecurity(reject_path_traversal=False))
+
+    @mcp.resource("data://items/{name}")
+    def get_item(name: str) -> str:
+        return f"item:{name}"
+
+    async with Client(mcp) as client:
+        # Server-wide policy disabled traversal check; ".." now allowed
+        result = await client.read_resource("data://items/..")
+        assert isinstance(result.contents[0], TextResourceContents)
+        assert result.contents[0].text == "item:.."
+
+
+async def test_resource_security_namespaced_identifier_requires_exempt():
+    """Single-letter-colon values like ``x:y`` are flagged by the
+    default absolute-path check (they parse as Windows drive-relative,
+    which discards the join base). A non-filesystem parameter that
+    legitimately accepts such values opts out via ``exempt_params``."""
+    mcp = MCPServer()
+
+    @mcp.resource("data://items/{id}")
+    def get_item(id: str) -> str:  # pragma: no cover - rejected before call
+        return f"item:{id}"
+
+    async with Client(mcp) as client:
+        with pytest.raises(MCPError, match="Unknown resource") as exc:
+            await client.read_resource("data://items/x:y")
+        assert exc.value.error.code == INVALID_PARAMS
+
+    # Exempting the parameter lets the value through.
+    mcp = MCPServer()
+
+    @mcp.resource("data://items/{id}", security=ResourceSecurity(exempt_params={"id"}))
+    def get_item_exempt(id: str) -> str:
+        return f"item:{id}"
+
+    async with Client(mcp) as client:
+        r = await client.read_resource("data://items/x:y")
+        assert isinstance(r.contents[0], TextResourceContents)
+        assert r.contents[0].text == "item:x:y"
+
+
+async def test_resource_security_rejection_halts_template_iteration():
+    """A strict template's security rejection must surface as
+    not-found and stop; a later permissive template must not be
+    reached."""
+    mcp = MCPServer()
+
+    @mcp.resource("file://docs/{name}")
+    def strict(name: str) -> str:  # pragma: no cover - never reached
+        return name
+
+    @mcp.resource(
+        "file://docs/{+path}",
+        security=ResourceSecurity(exempt_params={"path"}),
+    )
+    def lax(path: str) -> str:  # pragma: no cover - must not be reached
+        raise AssertionError("permissive template reached after security rejection")
+
+    async with Client(mcp) as client:
+        with pytest.raises(MCPError) as exc:
+            await client.read_resource("file://docs/..%2Fsecrets")
+        assert exc.value.error.code == INVALID_PARAMS
+        assert "Unknown resource" in exc.value.error.message
+
+
+async def test_static_resource_with_context_param_errors():
+    """A non-template URI with a Context-only handler should error
+    at decoration time with a clear message, not silently register
+    an unreachable resource."""
+    mcp = MCPServer()
+
+    with pytest.raises(ValueError, match="Context injection for static resources is not supported"):
+
+        @mcp.resource("weather://current")
+        def current_weather(ctx: Context) -> str:
+            raise NotImplementedError
+
+
+async def test_static_resource_with_extra_params_errors():
+    """A non-template URI with non-Context params should error at
+    decoration time."""
+    mcp = MCPServer()
+
+    with pytest.raises(ValueError, match="has no URI template variables"):
+
+        @mcp.resource("data://fixed")
+        def get_data(name: str) -> str:
+            raise NotImplementedError
+
+
 async def test_completion_decorator() -> None:
     """Test that the completion decorator registers a working handler."""
     mcp = MCPServer()

tests/shared/test_path_security.py

@@ -0,0 +1,159 @@
+"""Tests for filesystem path safety primitives."""
+
+from pathlib import Path
+
+import pytest
+
+from mcp.shared.path_security import (
+    PathEscapeError,
+    contains_path_traversal,
+    is_absolute_path,
+    safe_join,
+)
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        # Safe: no traversal
+        ("a/b/c", False),
+        ("readme.txt", False),
+        ("", False),
+        (".", False),
+        ("./a/b", False),
+        # Safe: .. balanced by prior descent
+        ("a/../b", False),
+        ("a/b/../c", False),
+        ("a/b/../../c", False),
+        # Unsafe: net escape
+        ("..", True),
+        ("../etc", True),
+        ("../../etc/passwd", True),
+        ("a/../../b", True),
+        ("./../../etc", True),
+        # .. as substring, not component — safe
+        ("1.0..2.0", False),
+        ("foo..bar", False),
+        ("..foo", False),
+        ("foo..", False),
+        # Backslash separator
+        ("..\\etc", True),
+        ("a\\..\\..\\b", True),
+        ("a\\b\\c", False),
+        # Mixed separators
+        ("a/..\\..\\b", True),
+    ],
+)
+def test_contains_path_traversal(value: str, expected: bool):
+    assert contains_path_traversal(value) is expected
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        # Relative
+        ("relative/path", False),
+        ("file.txt", False),
+        ("", False),
+        (".", False),
+        ("..", False),
+        # POSIX absolute
+        ("/", True),
+        ("/etc/passwd", True),
+        ("/a", True),
+        # Windows drive
+        ("C:", True),
+        ("C:\\Windows", True),
+        ("c:/foo", True),
+        ("Z:\\", True),
+        # Windows UNC / backslash-absolute
+        ("\\\\server\\share", True),
+        ("\\foo", True),
+        # Windows drive-relative — discards the join base when drives differ
+        ("C:relative", True),
+        ("x:y", True),
+        ("a:debug", True),
+        # Not a drive: digit before colon
+        ("1:foo", False),
+        # Colon not in position 1
+        ("ab:c", False),
+        # Non-ASCII letter is not a drive letter
+        ("Ω:namespace", False),
+        ("é:foo", False),
+    ],
+)
+def test_is_absolute_path(value: str, expected: bool):
+    assert is_absolute_path(value) is expected
+
+
+def test_safe_join_simple(tmp_path: Path):
+    result = safe_join(tmp_path, "docs", "readme.txt")
+    assert result == tmp_path / "docs" / "readme.txt"
+
+
+def test_safe_join_resolves_relative_base(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.chdir(tmp_path)
+    result = safe_join(".", "file.txt")
+    assert result == tmp_path / "file.txt"
+
+
+def test_safe_join_rejects_dotdot_escape(tmp_path: Path):
+    with pytest.raises(PathEscapeError, match="escapes base"):
+        safe_join(tmp_path, "../../../etc/passwd")
+
+
+def test_safe_join_rejects_balanced_then_escape(tmp_path: Path):
+    with pytest.raises(PathEscapeError, match="escapes base"):
+        safe_join(tmp_path, "a/../../etc")
+
+
+def test_safe_join_allows_balanced_dotdot(tmp_path: Path):
+    result = safe_join(tmp_path, "a/../b")
+    assert result == tmp_path / "b"
+
+
+def test_safe_join_rejects_absolute_part(tmp_path: Path):
+    with pytest.raises(PathEscapeError, match="is absolute"):
+        safe_join(tmp_path, "/etc/passwd")
+
+
+def test_safe_join_rejects_absolute_in_later_part(tmp_path: Path):
+    with pytest.raises(PathEscapeError, match="is absolute"):
+        safe_join(tmp_path, "docs", "/etc/passwd")
+
+
+def test_safe_join_rejects_windows_drive(tmp_path: Path):
+    with pytest.raises(PathEscapeError, match="is absolute"):
+        safe_join(tmp_path, "C:\\Windows\\System32")
+
+
+def test_safe_join_rejects_null_byte(tmp_path: Path):
+    with pytest.raises(PathEscapeError, match="null byte"):
+        safe_join(tmp_path, "file\0.txt")
+
+
+def test_safe_join_rejects_null_byte_in_later_part(tmp_path: Path):
+    with pytest.raises(PathEscapeError, match="null byte"):
+        safe_join(tmp_path, "docs", "file\0.txt")
+
+
+def test_safe_join_rejects_symlink_escape(tmp_path: Path):
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    sandbox = tmp_path / "sandbox"
+    sandbox.mkdir()
+    (sandbox / "escape").symlink_to(outside)
+
+    with pytest.raises(PathEscapeError, match="escapes base"):
+        safe_join(sandbox, "escape", "secret.txt")
+
+
+def test_safe_join_base_equals_target(tmp_path: Path):
+    # Joining nothing (or ".") should return the base itself
+    assert safe_join(tmp_path) == tmp_path
+    assert safe_join(tmp_path, ".") == tmp_path
+
+
+def test_path_escape_error_is_value_error():
+    with pytest.raises(ValueError):
+        safe_join("/tmp", "/etc")