feat: integrate UriTemplate into MCPServer resource templates

Refactors the internal `ResourceTemplate` to use the RFC 6570
`UriTemplate` engine for matching, and adds a configurable
`ResourceSecurity` policy for path-safety checks on extracted
parameters.

`ResourceTemplate.matches()` now:
- Delegates to `UriTemplate.match()` for full RFC 6570 Level 1-3
  support (plus path-style explode). `{+path}` can match
  multi-segment paths.
- Enforces structural integrity: `%2F` smuggled into a simple
  `{var}` is rejected.
- Applies `ResourceSecurity` policy: path traversal (`..` components)
  and absolute paths rejected by default, with per-parameter
  exemption available.

The `@mcp.resource()` decorator now parses the template once at
decoration time via `UriTemplate.parse()`, replacing the regex-based
param extraction that couldn't handle operators like `{+path}`.
Malformed templates surface immediately with a clear
`InvalidUriTemplate` including position info.

Also fixes the pre-existing bug where template literals were not
regex-escaped (a `.` in the template acted as a wildcard).
This commit is contained in:
Max Isbey
2026-03-26 16:40:14 +00:00
parent e5ecf50e64
commit 0018eea38f
5 changed files with 198 additions and 20 deletions
@@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Any
from pydantic import AnyUrl
from mcp.server.mcpserver.resources.base import Resource
from mcp.server.mcpserver.resources.templates import ResourceTemplate
from mcp.server.mcpserver.resources.templates import DEFAULT_RESOURCE_SECURITY, ResourceSecurity, ResourceTemplate
from mcp.server.mcpserver.utilities.logging import get_logger
from mcp.types import Annotations, Icon
@@ -64,6 +64,7 @@ class ResourceManager:
icons: list[Icon] | None = None,
annotations: Annotations | None = None,
meta: dict[str, Any] | None = None,
security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
) -> ResourceTemplate:
"""Add a template from a function."""
template = ResourceTemplate.from_function(
@@ -76,6 +77,7 @@ class ResourceManager:
icons=icons,
annotations=annotations,
meta=meta,
security=security,
)
self._templates[template.uri_template] = template
return template
+91 -14
View File
@@ -3,16 +3,17 @@
from __future__ import annotations
import inspect
import re
from collections.abc import Callable
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
from urllib.parse import unquote
from pydantic import BaseModel, Field, validate_call
from mcp.server.mcpserver.resources.types import FunctionResource, Resource
from mcp.server.mcpserver.utilities.context_injection import find_context_parameter, inject_context
from mcp.server.mcpserver.utilities.func_metadata import func_metadata
from mcp.shared.path_security import contains_path_traversal, is_absolute_path
from mcp.shared.uri_template import UriTemplate
from mcp.types import Annotations, Icon
if TYPE_CHECKING:
@@ -20,6 +21,64 @@ if TYPE_CHECKING:
from mcp.server.mcpserver.context import Context
@dataclass(frozen=True)
class ResourceSecurity:
"""Security policy applied to extracted resource template parameters.
These checks run **after** :meth:`~mcp.shared.uri_template.UriTemplate.match`
has already enforced structural integrity (e.g., rejected ``%2F`` in
simple ``{var}``). They catch semantic attacks that structural checks
cannot: ``..`` traversal and absolute-path injection work even with
perfectly-formed URI components.
Example::
# Opt out for a parameter that legitimately contains ..
@mcp.resource(
"git://diff/{+range}",
security=ResourceSecurity(exempt_params=frozenset({"range"})),
)
def git_diff(range: str) -> str: ...
"""
reject_path_traversal: bool = True
"""Reject values containing ``..`` as a path component."""
reject_absolute_paths: bool = True
"""Reject values that look like absolute filesystem paths."""
exempt_params: frozenset[str] = field(default_factory=frozenset[str])
"""Parameter names to skip all checks for."""
def validate(self, params: Mapping[str, str | list[str]]) -> bool:
"""Check all parameter values against the configured policy.
Args:
params: Extracted template parameters. List values (from
explode variables) are checked element-wise.
Returns:
``True`` if all values pass; ``False`` on first violation.
"""
for name, value in params.items():
if name in self.exempt_params:
continue
values = value if isinstance(value, list) else [value]
for v in values:
if self.reject_path_traversal and contains_path_traversal(v):
return False
if self.reject_absolute_paths and is_absolute_path(v):
return False
return True
DEFAULT_RESOURCE_SECURITY = ResourceSecurity()
"""Secure-by-default policy: traversal and absolute paths rejected."""
UNSAFE_RESOURCE_SECURITY = ResourceSecurity(reject_path_traversal=False, reject_absolute_paths=False)
"""No path checks. Use only when parameters are never used as filesystem paths."""
class ResourceTemplate(BaseModel):
"""A template for dynamically creating resources."""
@@ -34,6 +93,8 @@ class ResourceTemplate(BaseModel):
fn: Callable[..., Any] = Field(exclude=True)
parameters: dict[str, Any] = Field(description="JSON schema for function parameters")
context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context")
parsed_template: UriTemplate = Field(exclude=True, description="Parsed RFC 6570 template")
security: ResourceSecurity = Field(exclude=True, description="Path-safety policy for extracted parameters")
@classmethod
def from_function(
@@ -48,12 +109,20 @@ class ResourceTemplate(BaseModel):
annotations: Annotations | None = None,
meta: dict[str, Any] | None = None,
context_kwarg: str | None = None,
security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
) -> ResourceTemplate:
"""Create a template from a function."""
"""Create a template from a function.
Raises:
InvalidUriTemplate: If ``uri_template`` is malformed or uses
unsupported RFC 6570 features.
"""
func_name = name or fn.__name__
if func_name == "<lambda>":
raise ValueError("You must provide a name for lambda functions") # pragma: no cover
parsed = UriTemplate.parse(uri_template)
# Find context parameter if it exists
if context_kwarg is None: # pragma: no branch
context_kwarg = find_context_parameter(fn)
@@ -80,20 +149,28 @@ class ResourceTemplate(BaseModel):
fn=fn,
parameters=parameters,
context_kwarg=context_kwarg,
parsed_template=parsed,
security=security,
)
def matches(self, uri: str) -> dict[str, Any] | None:
"""Check if URI matches template and extract parameters.
def matches(self, uri: str) -> dict[str, str | list[str]] | None:
"""Check if a URI matches this template and extract parameters.
Extracted parameters are URL-decoded to handle percent-encoded characters.
Delegates to :meth:`UriTemplate.match` for RFC 6570 matching
with structural integrity (``%2F`` smuggling rejected for simple
vars), then applies this template's :class:`ResourceSecurity`
policy (path traversal, absolute paths).
Returns:
Extracted parameters on success, or ``None`` if the URI
doesn't match or a parameter fails security validation.
"""
# Convert template to regex pattern
pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
match = re.match(f"^{pattern}$", uri)
if match:
# URL-decode all extracted parameter values
return {key: unquote(value) for key, value in match.groupdict().items()}
return None
params = self.parsed_template.match(uri)
if params is None:
return None
if not self.security.validate(params):
return None
return params
async def create_resource(
self,
+15 -5
View File
@@ -5,7 +5,6 @@ from __future__ import annotations
import base64
import inspect
import json
import re
from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from typing import Any, Generic, Literal, TypeVar, overload
@@ -43,6 +42,7 @@ from mcp.server.streamable_http import EventStore
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.server.transport_security import TransportSecuritySettings
from mcp.shared.exceptions import MCPError
from mcp.shared.uri_template import UriTemplate
from mcp.types import (
Annotations,
BlobResourceContents,
@@ -668,6 +668,13 @@ class MCPServer(Generic[LifespanResultT]):
data = await fetch_weather(city)
return f"Weather for {city}: {data}"
```
Raises:
InvalidUriTemplate: If ``uri`` is not a valid RFC 6570 template.
ValueError: If URI template parameters don't match the
function's parameters.
TypeError: If the decorator is applied without being called
(``@resource`` instead of ``@resource("uri")``).
"""
# Check if user passed function directly instead of calling decorator
if callable(uri):
@@ -676,18 +683,21 @@ class MCPServer(Generic[LifespanResultT]):
"Did you forget to call it? Use @resource('uri') instead of @resource"
)
# Parse once, early — surfaces malformed-template errors at
# decoration time with a clear position, and gives us correct
# variable names for all RFC 6570 operators.
parsed = UriTemplate.parse(uri)
uri_params = set(parsed.variable_names)
def decorator(fn: _CallableT) -> _CallableT:
# Check if this should be a template
sig = inspect.signature(fn)
has_uri_params = "{" in uri and "}" in uri
has_func_params = bool(sig.parameters)
if has_uri_params or has_func_params:
if uri_params or has_func_params:
# Check for Context parameter to exclude from validation
context_param = find_context_parameter(fn)
# Validate that URI params match function params (excluding context)
uri_params = set(re.findall(r"{(\w+)}", uri))
# We need to remove the context_param from the resource function if
# there is any.
func_params = {p for p in sig.parameters.keys() if p != context_param}
@@ -6,9 +6,80 @@ from pydantic import BaseModel
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.resources import FunctionResource, ResourceTemplate
from mcp.server.mcpserver.resources.templates import (
DEFAULT_RESOURCE_SECURITY,
UNSAFE_RESOURCE_SECURITY,
ResourceSecurity,
)
from mcp.types import Annotations
def _make(uri_template: str, security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY) -> ResourceTemplate:
def handler(**kwargs: Any) -> str:
return "ok"
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_in_simple_var():
# Path traversal via encoded slash: %2F smuggled into a simple {var}
t = _make("file://docs/{name}")
assert t.matches("file://docs/..%2F..%2Fetc%2Fpasswd") is None
def test_matches_rejects_path_traversal_by_default():
t = _make("file://docs/{name}")
assert t.matches("file://docs/..") is None
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}")
assert t.matches("file://docs/../../etc/passwd") is None
def test_matches_rejects_absolute_path():
t = _make("file://docs/{+path}")
assert t.matches("file://docs//etc/passwd") is None
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=frozenset({"range"}))
t = _make("git://diff/{+range}", security=policy)
assert t.matches("git://diff/../foo") == {"range": "../foo"}
def test_matches_unsafe_policy_disables_checks():
t = _make("file://docs/{name}", security=UNSAFE_RESOURCE_SECURITY)
assert t.matches("file://docs/..") == {"name": ".."}
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
assert t.matches("api/a/../c") is None
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:
"""Test ResourceTemplate functionality."""
+18
View File
@@ -19,6 +19,7 @@ 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
from mcp.types import (
AudioContent,
BlobResourceContents,
@@ -141,6 +142,23 @@ class TestServer:
def get_data(x: str) -> str: # pragma: no cover
return f"Data: {x}"
async def test_resource_decorator_rfc6570_reserved_expansion(self):
# 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(self):
mcp = MCPServer()
with pytest.raises(InvalidUriTemplate, match="Unclosed expression"):
mcp.resource("file://{name")
class TestDnsRebindingProtection:
"""Tests for automatic DNS rebinding protection on localhost.