mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 07:09:11 +02:00
Screen templated resource parameters for path traversal by default (#4482)
* Add ResourceSecurity screening for templated resources (defaults on)
* Add tests for resource path-security screening
* Document resource path-security; fix ty in tests
* Carry child template security policy through provider mount
Preserve a mounted template's explicit ResourceSecurity (per-param
exemptions or a deliberate opt-out) through FastMCPProviderResourceTemplate.wrap
so the parent read chokepoint honours it instead of the parent default.
* Defer mcp SDK import so fastmcp.resources loads without the [mcp] extra
* Make resource path-security docs examples self-contained and runnable
* Match exempt_params under both hyphen and underscore spellings
Template placeholders like {git-ref} extract as git_ref, so an exemption
written with the natural URI-template spelling never matched.
* Docs: describe net-depth traversal rule accurately; make example runnable
The screening only rejects .. segments that escape the starting depth
(foo/../bar passes) — saying any standalone .. is rejected overstated
the guarantee. Also define DOCS_ROOT so the example runs.
This commit is contained in:
Jeremiah Lowin
GitHub
parent
918b85f9b2
commit
d779414f8a
12 changed files with 824 additions and 11 deletions
|
|
@ -324,6 +324,14 @@ FastMCP keeps its own DCR redirect-URI hardening (PRs #4419, #4408) regardless o
|
|||
|
||||
*Verify:* recent commits `67527c1f` (block unsafe OAuth redirect schemes), `57a27992` (DNS rebinding), `cccb529f` (DCR redirect URI validation) on `main`.
|
||||
|
||||
### Templated resource parameters are path-screened by default — Breaking (behavior)
|
||||
|
||||
Every templated resource now has its extracted parameter values screened for path-traversal (`..` segments), absolute paths, and null bytes **before the handler runs** — on by default, at the server's read chokepoint, covering local and provider-sourced (mounted/proxied) templates alike. Previously these payloads reached handlers raw; a template whose parameter flowed into a filesystem path or upstream URL was exposed unless the author added their own check. A rejected read now surfaces a non-leaky "resource not found" error (`-32602`) and a debug log.
|
||||
|
||||
The check is component-based, matching the SDK's `contains_path_traversal`: only a standalone `..` segment is traversal, so values that merely contain dots (`HEAD~3..HEAD`, `file.tar.gz`) and dotfiles (`.env`) still pass. This can break a template that legitimately accepts `..`-bearing or absolute values — exempt the parameter with `ResourceSecurity(exempt_params={...})`, disable per-component with `security=None`, or set a server-wide default with `FastMCP(resource_security=...)`. See [Resources → Path Security](/servers/resources#path-security).
|
||||
|
||||
*Verify:* `fastmcp_slim/fastmcp/resources/security.py` (`ResourceSecurity`), the screening block in `FastMCP.read_resource` (`fastmcp_slim/fastmcp/server/server.py`), and `tests/resources/test_resource_security.py`.
|
||||
|
||||
## Removed in 4.0
|
||||
|
||||
Deprecations that warned in 3.x are removed in 4.0. Each entry below is a hard removal — the old surface raises `TypeError` / `AttributeError` rather than warning, unless noted otherwise.
|
||||
|
|
|
|||
|
|
@ -522,11 +522,85 @@ Wildcard parameters are useful when:
|
|||
|
||||
Note that like regular parameters, each wildcard parameter must still be a named parameter in your function signature, and all required function parameters must appear in the URI template.
|
||||
|
||||
#### Filesystem Path Safety
|
||||
#### Path Security
|
||||
|
||||
Template parameters are decoded before your function receives them. A standard `{filename}` parameter matches one URI segment before decoding, so a request like `files://a%2Fb` passes `filename="a/b"` to the handler. Treat template values as untrusted decoded URI data whenever they determine filesystem paths.
|
||||
Template parameters are extracted from the request URI and decoded before your function receives them, so a path-traversal payload like `../` or an absolute path can reach a handler that builds filesystem paths or upstream URLs. FastMCP screens every templated resource's parameter values **before the handler runs**, and this screening is **on by default**.
|
||||
|
||||
Validate the final resolved path against an allowed root before reading:
|
||||
By default, a parameter value is rejected if its `..` path segments would escape the value's own starting depth, if it looks like an absolute path, or if it contains a null byte. A rejected read surfaces a clean "resource not found" error to the client and logs the reason at debug level, so the failing parameter and policy are never revealed on the wire.
|
||||
|
||||
The traversal check is component-based and tracks net depth: `..` only counts against you when it climbs above where the value starts. `../secret`, a bare `..`, and `a/../../b` are rejected; `foo/../bar` is allowed because it never leaves the starting directory, and values that merely *contain* dots — `HEAD~3..HEAD`, `v1..v2`, `file.tar.gz`, dotfiles like `.env` — all pass. Screening runs on the decoded value, so `..%2F` is caught the same as a literal `../`. This bounds relative escapes; anchoring the *final* path inside a root directory is still your handler's job (for example with `safe_join`), since only the handler knows what the value is joined to.
|
||||
|
||||
```python
|
||||
from pathlib import Path
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP(name="DocsServer")
|
||||
|
||||
DOCS_ROOT = Path("/srv/docs")
|
||||
|
||||
|
||||
@mcp.resource("docs://{path*}")
|
||||
def read_doc(path: str) -> str:
|
||||
# A request for docs://../secret is rejected before this runs.
|
||||
return (DOCS_ROOT / path).read_text(encoding="utf-8")
|
||||
```
|
||||
|
||||
##### Exempting parameters
|
||||
|
||||
Some parameters legitimately carry values that look like traversal — a git ref, a version range, an opaque token. Exempt them by name with `ResourceSecurity`:
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.resources import ResourceSecurity
|
||||
|
||||
mcp = FastMCP(name="DocsServer")
|
||||
|
||||
|
||||
@mcp.resource(
|
||||
"git://diff/{ref}",
|
||||
security=ResourceSecurity(exempt_params={"ref"}),
|
||||
)
|
||||
def git_diff(ref: str) -> str:
|
||||
# ref="HEAD~3..HEAD" is allowed
|
||||
...
|
||||
```
|
||||
|
||||
##### Disabling screening
|
||||
|
||||
Pass `security=None` to turn screening off for a single component:
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP(name="DocsServer")
|
||||
|
||||
|
||||
@mcp.resource("raw://{value}", security=None)
|
||||
def raw(value: str) -> str: ...
|
||||
```
|
||||
|
||||
Or set a server-wide default with `resource_security`, which applies to every templated resource that does not set its own `security`:
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.resources import ResourceSecurity
|
||||
|
||||
# Relax one check across the whole server:
|
||||
relaxed = FastMCP(
|
||||
name="DocsServer",
|
||||
resource_security=ResourceSecurity(reject_absolute_paths=False),
|
||||
)
|
||||
|
||||
# Or disable screening entirely across the server:
|
||||
unscreened = FastMCP(name="DocsServer", resource_security=None)
|
||||
```
|
||||
|
||||
A per-component `security` always overrides the server default.
|
||||
|
||||
<Warning>
|
||||
Screening rejects the obvious injection shapes, but it does not know your filesystem root. When a parameter determines a real path, still resolve it against an allowed root and confirm containment before reading — screening and containment are complementary layers.
|
||||
</Warning>
|
||||
|
||||
```python
|
||||
from pathlib import Path
|
||||
|
|
@ -548,8 +622,6 @@ def read_doc(filename: str) -> str:
|
|||
return requested_path.read_text(encoding="utf-8")
|
||||
```
|
||||
|
||||
Use wildcard parameters (`{path*}`) for resources whose URI shape intentionally includes slashes, and apply the same containment check before accessing the filesystem.
|
||||
|
||||
#### Query Parameters
|
||||
|
||||
<VersionBadge version="2.13.0" />
|
||||
|
|
|
|||
|
|
@ -80,6 +80,17 @@ class DisabledError(Exception):
|
|||
"""Object is disabled."""
|
||||
|
||||
|
||||
class ResourceSecurityError(NotFoundError):
|
||||
"""A templated resource parameter failed path-security screening.
|
||||
|
||||
Subclasses ``NotFoundError`` so the read handler surfaces a
|
||||
non-leaky ``INVALID_PARAMS`` (-32602) "resource not found" error to
|
||||
the client — a traversal attempt is indistinguishable from a request
|
||||
for a resource that does not exist, and never reveals which parameter
|
||||
or policy tripped.
|
||||
"""
|
||||
|
||||
|
||||
class AuthorizationError(FastMCPError):
|
||||
"""Error when authorization check fails."""
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import sys
|
|||
|
||||
from .function_resource import FunctionResource, resource
|
||||
from .base import Resource, ResourceContent, ResourceResult
|
||||
from .security import ResourceSecurity
|
||||
from .template import ResourceTemplate
|
||||
from .types import (
|
||||
BinaryResource,
|
||||
|
|
@ -20,6 +21,7 @@ __all__ = [
|
|||
"Resource",
|
||||
"ResourceContent",
|
||||
"ResourceResult",
|
||||
"ResourceSecurity",
|
||||
"ResourceTemplate",
|
||||
"TextResource",
|
||||
"resource",
|
||||
|
|
|
|||
|
|
@ -22,6 +22,11 @@ from pydantic import AnyUrl
|
|||
from pydantic.json_schema import SkipJsonSchema
|
||||
|
||||
from fastmcp.resources.base import Resource, ResourceResult
|
||||
from fastmcp.resources.security import (
|
||||
INHERIT_SECURITY,
|
||||
InheritSecurity,
|
||||
ResourceSecurity,
|
||||
)
|
||||
from fastmcp.utilities.async_utils import (
|
||||
call_sync_fn_in_threadpool,
|
||||
is_coroutine_function,
|
||||
|
|
@ -64,6 +69,7 @@ class ResourceMeta:
|
|||
task: bool | TaskConfig | None = None
|
||||
auth: AuthCheck | list[AuthCheck] | None = None
|
||||
enabled: bool = True
|
||||
security: ResourceSecurity | None | InheritSecurity = INHERIT_SECURITY
|
||||
|
||||
|
||||
class FunctionResource(Resource):
|
||||
|
|
@ -255,6 +261,7 @@ def resource(
|
|||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
security: ResourceSecurity | None | InheritSecurity = INHERIT_SECURITY,
|
||||
) -> Callable[[F], F]:
|
||||
"""Standalone decorator to mark a function as an MCP resource.
|
||||
|
||||
|
|
@ -284,6 +291,7 @@ def resource(
|
|||
meta=meta,
|
||||
task=task,
|
||||
auth=auth,
|
||||
security=security,
|
||||
)
|
||||
target = fn.__func__ if isinstance(fn, staticmethod | MethodType) else fn
|
||||
cast(Any, target).__fastmcp__ = metadata
|
||||
|
|
|
|||
162
fastmcp_slim/fastmcp/resources/security.py
Normal file
162
fastmcp_slim/fastmcp/resources/security.py
Normal file
|
|
@ -0,0 +1,162 @@
|
|||
"""Path-safety policy for templated resource parameters.
|
||||
|
||||
Templated resources (`@mcp.resource("file:///{path}")`-style) extract
|
||||
parameter values straight out of the request URI and hand them to the
|
||||
resource function. When those values flow into filesystem or URI
|
||||
construction, a malicious client can smuggle path-traversal payloads
|
||||
(`../`, absolute paths, null bytes) through the template.
|
||||
|
||||
`ResourceSecurity` screens extracted parameter values *before* the
|
||||
resource handler runs. It is applied by default to every templated
|
||||
read, mirroring the posture of the underlying MCP SDK's
|
||||
`ResourceSecurity` (traversal, absolute paths, and null bytes rejected).
|
||||
|
||||
The screening reuses the SDK's component-based traversal check, so a
|
||||
value that merely *contains* dots (e.g. `HEAD~3..HEAD`, `v1..v2`,
|
||||
`file.tar.gz`) is not rejected — only an actual `..` path segment is.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable, Mapping, Set
|
||||
from dataclasses import dataclass, field
|
||||
from functools import cache
|
||||
from typing import Any
|
||||
|
||||
from pydantic import GetCoreSchemaHandler
|
||||
from pydantic_core import core_schema
|
||||
|
||||
__all__ = ["ResourceSecurity"]
|
||||
|
||||
|
||||
@cache
|
||||
def _path_checks() -> tuple[Callable[[str], bool], Callable[[str], bool]]:
|
||||
"""Lazily load the SDK's path-safety helpers.
|
||||
|
||||
The screening logic lives in the `mcp` SDK, which is an optional
|
||||
dependency of `fastmcp-slim`. Importing it at module top would make
|
||||
`from fastmcp.resources import Resource` require the SDK, so the
|
||||
import is deferred to the point of first use (and cached).
|
||||
"""
|
||||
from mcp.shared.path_security import (
|
||||
contains_path_traversal,
|
||||
is_absolute_path,
|
||||
)
|
||||
|
||||
return contains_path_traversal, is_absolute_path
|
||||
|
||||
|
||||
class InheritSecurity:
|
||||
"""Sentinel type: inherit the server-wide resource-security default.
|
||||
|
||||
Distinguishes "no per-component policy was set" (inherit whatever the
|
||||
server configured) from an explicit ``None`` (screening disabled for
|
||||
this component).
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover - debug aid
|
||||
return "INHERIT_SECURITY"
|
||||
|
||||
@classmethod
|
||||
def __get_pydantic_core_schema__(
|
||||
cls, source_type: Any, handler: GetCoreSchemaHandler
|
||||
) -> core_schema.CoreSchema:
|
||||
# Accept the singleton sentinel as-is; it is an internal, excluded
|
||||
# field value, so no serialization support is needed.
|
||||
return core_schema.is_instance_schema(cls)
|
||||
|
||||
|
||||
INHERIT_SECURITY = InheritSecurity()
|
||||
"""Sentinel instance signalling a template should inherit the server default."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ResourceSecurity:
|
||||
"""Security policy applied to extracted resource template parameters.
|
||||
|
||||
These checks run after a URI has matched a template and its
|
||||
parameter values have been extracted and percent-decoded. They catch
|
||||
path-traversal and absolute-path injection regardless of how the
|
||||
value was encoded in the URI (literal, `%2F`, `%5C`, `%2E%2E`).
|
||||
|
||||
All checks default on. Screen a value like `HEAD~3..HEAD` (dots
|
||||
inside a single segment) passes — only a standalone `..` segment is
|
||||
treated as traversal.
|
||||
|
||||
Example:
|
||||
Opt a parameter out of screening (e.g. a git ref that may
|
||||
legitimately contain `..`):
|
||||
|
||||
```python
|
||||
from fastmcp.resources import ResourceSecurity
|
||||
|
||||
@mcp.resource(
|
||||
"git://diff/{ref}",
|
||||
security=ResourceSecurity(exempt_params={"ref"}),
|
||||
)
|
||||
def git_diff(ref: 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."""
|
||||
|
||||
reject_null_bytes: bool = True
|
||||
"""Reject values containing NUL (`\\x00`). Null bytes defeat string
|
||||
comparisons (`"..\\x00" != ".."`) and can cause truncation in C
|
||||
extensions or subprocess calls."""
|
||||
|
||||
exempt_params: Set[str] = field(default_factory=frozenset)
|
||||
"""Parameter names to skip all checks for. Hyphenated URI-template
|
||||
spellings are accepted: `{git-ref}` is extracted as `git_ref`, and an
|
||||
exemption written either way matches it."""
|
||||
|
||||
def _exempt(self, name: str) -> bool:
|
||||
"""True if `name` is exempted under either its extracted or its
|
||||
URI-template spelling (hyphens normalize to underscores on
|
||||
extraction, so `exempt_params={"git-ref"}` must match `git_ref`)."""
|
||||
if name in self.exempt_params:
|
||||
return True
|
||||
return any(exempt.replace("-", "_") == name for exempt in self.exempt_params)
|
||||
|
||||
def validate(self, params: Mapping[str, object]) -> str | None:
|
||||
"""Check all parameter values against the configured policy.
|
||||
|
||||
String values (and lists of strings, from wildcard `{path*}`
|
||||
parameters that span multiple segments) are screened; non-string
|
||||
values are ignored, since traversal is a string-path concern.
|
||||
|
||||
Args:
|
||||
params: Extracted template parameters.
|
||||
|
||||
Returns:
|
||||
The name of the first parameter that fails, or `None` if all
|
||||
values pass.
|
||||
"""
|
||||
contains_path_traversal, is_absolute_path = _path_checks()
|
||||
for name, value in params.items():
|
||||
if self._exempt(name):
|
||||
continue
|
||||
if isinstance(value, str):
|
||||
candidates = [value]
|
||||
elif isinstance(value, (list, tuple)):
|
||||
candidates = [v for v in value if isinstance(v, str)]
|
||||
else:
|
||||
continue
|
||||
for candidate in candidates:
|
||||
if self.reject_null_bytes and "\0" in candidate:
|
||||
return name
|
||||
if self.reject_path_traversal and contains_path_traversal(candidate):
|
||||
return name
|
||||
if self.reject_absolute_paths and is_absolute_path(candidate):
|
||||
return name
|
||||
return None
|
||||
|
||||
|
||||
DEFAULT_RESOURCE_SECURITY = ResourceSecurity()
|
||||
"""Secure-by-default policy: traversal, absolute paths, and null bytes rejected."""
|
||||
|
|
@ -24,6 +24,11 @@ from pydantic import (
|
|||
)
|
||||
|
||||
from fastmcp.resources.base import Resource, ResourceResult
|
||||
from fastmcp.resources.security import (
|
||||
INHERIT_SECURITY,
|
||||
InheritSecurity,
|
||||
ResourceSecurity,
|
||||
)
|
||||
from fastmcp.utilities.authorization import AuthCheck
|
||||
from fastmcp.utilities.components import FastMCPComponent
|
||||
from fastmcp.utilities.json_schema import compress_schema
|
||||
|
|
@ -186,6 +191,29 @@ class ResourceTemplate(FastMCPComponent):
|
|||
description="Authorization checks for this resource template",
|
||||
exclude=True,
|
||||
)
|
||||
security: SkipJsonSchema[ResourceSecurity | None | InheritSecurity] = Field(
|
||||
default=INHERIT_SECURITY,
|
||||
description=(
|
||||
"Path-safety policy for extracted parameters. INHERIT_SECURITY "
|
||||
"(default) inherits the server-wide default; None disables "
|
||||
"screening; a ResourceSecurity instance applies that explicit "
|
||||
"policy."
|
||||
),
|
||||
exclude=True,
|
||||
)
|
||||
|
||||
def resolve_security(
|
||||
self, server_default: ResourceSecurity | None
|
||||
) -> ResourceSecurity | None:
|
||||
"""Resolve the effective security policy for this template.
|
||||
|
||||
A per-component ``security`` overrides the server default.
|
||||
``INHERIT_SECURITY`` (the field default) inherits ``server_default``;
|
||||
an explicit ``None`` disables screening for this template.
|
||||
"""
|
||||
if isinstance(self.security, InheritSecurity):
|
||||
return server_default
|
||||
return self.security
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__.__name__}(uri_template={self.uri_template!r}, name={self.name!r}, description={self.description!r}, tags={self.tags})"
|
||||
|
|
@ -205,6 +233,7 @@ class ResourceTemplate(FastMCPComponent):
|
|||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
security: ResourceSecurity | None | InheritSecurity = INHERIT_SECURITY,
|
||||
) -> FunctionResourceTemplate:
|
||||
return FunctionResourceTemplate.from_function(
|
||||
fn=fn,
|
||||
|
|
@ -220,6 +249,7 @@ class ResourceTemplate(FastMCPComponent):
|
|||
meta=meta,
|
||||
task=task,
|
||||
auth=auth,
|
||||
security=security,
|
||||
)
|
||||
|
||||
@field_validator("mime_type", mode="before")
|
||||
|
|
@ -544,6 +574,7 @@ class FunctionResourceTemplate(ResourceTemplate):
|
|||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
security: ResourceSecurity | None | InheritSecurity = INHERIT_SECURITY,
|
||||
) -> FunctionResourceTemplate:
|
||||
"""Create a template from a function."""
|
||||
|
||||
|
|
@ -683,4 +714,5 @@ class FunctionResourceTemplate(ResourceTemplate):
|
|||
meta=meta,
|
||||
task_config=task_config,
|
||||
auth=auth,
|
||||
security=security,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -363,6 +363,7 @@ class FastMCPProviderResourceTemplate(ResourceTemplate):
|
|||
meta=template.get_meta(),
|
||||
title=template.title,
|
||||
icons=template.icons,
|
||||
security=template.security,
|
||||
)
|
||||
|
||||
async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
|
||||
|
|
|
|||
|
|
@ -14,6 +14,11 @@ import mcp_types
|
|||
from mcp_types import Annotations
|
||||
|
||||
from fastmcp.resources.base import Resource
|
||||
from fastmcp.resources.security import (
|
||||
INHERIT_SECURITY,
|
||||
InheritSecurity,
|
||||
ResourceSecurity,
|
||||
)
|
||||
from fastmcp.resources.template import ResourceTemplate
|
||||
from fastmcp.server.auth.authorization import AuthCheck
|
||||
from fastmcp.server.tasks.config import TaskConfig
|
||||
|
|
@ -70,6 +75,7 @@ class ResourceDecoratorMixin:
|
|||
meta=meta.meta,
|
||||
task=resolved_task,
|
||||
auth=meta.auth,
|
||||
security=meta.security,
|
||||
)
|
||||
else:
|
||||
resource = Resource.from_function(
|
||||
|
|
@ -119,6 +125,7 @@ class ResourceDecoratorMixin:
|
|||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
security: ResourceSecurity | None | InheritSecurity = INHERIT_SECURITY,
|
||||
) -> Callable[[F], F]:
|
||||
"""Decorator to register a function as a resource.
|
||||
|
||||
|
|
@ -202,6 +209,7 @@ class ResourceDecoratorMixin:
|
|||
task=task,
|
||||
auth=auth,
|
||||
enabled=enabled,
|
||||
security=security,
|
||||
)
|
||||
target = fn.__func__ if hasattr(fn, "__func__") else fn
|
||||
target.__fastmcp__ = metadata # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
|
||||
|
|
|
|||
|
|
@ -49,6 +49,7 @@ from fastmcp.exceptions import (
|
|||
NotFoundError,
|
||||
PromptError,
|
||||
ResourceError,
|
||||
ResourceSecurityError,
|
||||
ToolError,
|
||||
ValidationError,
|
||||
)
|
||||
|
|
@ -57,6 +58,12 @@ from fastmcp.prompts import Prompt
|
|||
from fastmcp.prompts.base import PromptResult
|
||||
from fastmcp.prompts.function_prompt import FunctionPrompt
|
||||
from fastmcp.resources.base import Resource, ResourceResult
|
||||
from fastmcp.resources.security import (
|
||||
DEFAULT_RESOURCE_SECURITY,
|
||||
INHERIT_SECURITY,
|
||||
InheritSecurity,
|
||||
ResourceSecurity,
|
||||
)
|
||||
from fastmcp.resources.template import ResourceTemplate
|
||||
from fastmcp.server.auth import AuthCheck, AuthContext, AuthProvider, run_auth_checks
|
||||
from fastmcp.server.caching import build_cache_hints
|
||||
|
|
@ -330,6 +337,7 @@ class FastMCP(
|
|||
dereference_schemas: bool = True,
|
||||
strict_input_validation: bool | None = None,
|
||||
list_page_size: int | None = None,
|
||||
resource_security: ResourceSecurity | None = DEFAULT_RESOURCE_SECURITY,
|
||||
cache_ttl: int | None = None,
|
||||
cache_scope: Literal["public", "private"] | None = None,
|
||||
tasks: bool | None = None,
|
||||
|
|
@ -390,6 +398,13 @@ class FastMCP(
|
|||
raise ValueError("list_page_size must be a positive integer")
|
||||
self._list_page_size: int | None = list_page_size
|
||||
|
||||
# Server-wide default path-security policy for templated resources.
|
||||
# Applied before the handler runs to every templated read whose
|
||||
# component does not override it. DEFAULT_RESOURCE_SECURITY screens
|
||||
# traversal, absolute paths, and null bytes; None disables screening
|
||||
# server-wide.
|
||||
self._resource_security: ResourceSecurity | None = resource_security
|
||||
|
||||
# Server-level SEP-2549 cache hints, applied uniformly to every
|
||||
# SDK-cacheable result by the low-level server's runner (raises on
|
||||
# invalid ttl/scope).
|
||||
|
|
@ -1490,6 +1505,24 @@ class FastMCP(
|
|||
span.set_attributes(template.get_span_attributes())
|
||||
params = template.matches(uri)
|
||||
assert params is not None
|
||||
|
||||
# Path-security screening: reject traversal / absolute-path /
|
||||
# null-byte payloads in extracted parameter values BEFORE the
|
||||
# handler runs. This is the single chokepoint for every
|
||||
# templated read (local decorator and provider-sourced), so
|
||||
# enforcement lives here rather than in any decorator.
|
||||
security = template.resolve_security(self._resource_security)
|
||||
if security is not None:
|
||||
failed = security.validate(params)
|
||||
if failed is not None:
|
||||
logger.debug(
|
||||
"Rejected resource %r: parameter %r failed "
|
||||
"path-security screening",
|
||||
uri,
|
||||
failed,
|
||||
)
|
||||
raise ResourceSecurityError(f"Unknown resource: {uri!r}")
|
||||
|
||||
if task_meta is not None and task_meta.fn_key is None:
|
||||
task_meta = replace(task_meta, fn_key=template.key)
|
||||
try:
|
||||
|
|
@ -1824,6 +1857,7 @@ class FastMCP(
|
|||
app: AppConfig | dict[str, Any] | bool | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
auth: AuthCheck | list[AuthCheck] | None = None,
|
||||
security: ResourceSecurity | None | InheritSecurity = INHERIT_SECURITY,
|
||||
) -> Callable[[F], F]:
|
||||
"""Decorator to register a function as a resource.
|
||||
|
||||
|
|
@ -1923,6 +1957,7 @@ class FastMCP(
|
|||
meta=meta,
|
||||
task=task if task is not None else self._support_tasks_by_default,
|
||||
auth=auth,
|
||||
security=security,
|
||||
)
|
||||
|
||||
return inner_decorator
|
||||
|
|
|
|||
473
tests/resources/test_resource_security.py
Normal file
473
tests/resources/test_resource_security.py
Normal file
|
|
@ -0,0 +1,473 @@
|
|||
"""Tests for path-security screening of templated resource parameters.
|
||||
|
||||
Templated resources extract parameter values from request URIs and hand
|
||||
them to the handler. `ResourceSecurity` screens those values (traversal,
|
||||
absolute paths, null bytes) before the handler runs, defaults-on, at the
|
||||
server's read chokepoint.
|
||||
|
||||
The screening is applied to the *raw* URI string reaching the server
|
||||
(`FastMCP.read_resource(str)`), which is the path the JSON-RPC handler and
|
||||
internal callers use. Over the in-memory `Client`, URIs are wrapped in
|
||||
`AnyUrl`, which independently normalises many `..` payloads away before
|
||||
they reach the server — a separate layer of defense.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
import textwrap
|
||||
|
||||
import pytest
|
||||
|
||||
from fastmcp import Client, FastMCP
|
||||
from fastmcp.exceptions import ResourceSecurityError
|
||||
from fastmcp.resources.security import (
|
||||
DEFAULT_RESOURCE_SECURITY,
|
||||
INHERIT_SECURITY,
|
||||
ResourceSecurity,
|
||||
)
|
||||
from fastmcp.resources.template import ResourceTemplate
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ResourceSecurity model (unit)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestResourceSecurityModel:
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
[
|
||||
"../etc/passwd",
|
||||
"..",
|
||||
"a/../../b",
|
||||
"nested/../../outside",
|
||||
],
|
||||
)
|
||||
def test_rejects_traversal(self, value: str):
|
||||
assert ResourceSecurity().validate({"path": value}) == "path"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
[
|
||||
"/etc/passwd",
|
||||
"/absolute/injection",
|
||||
"C:\\Windows",
|
||||
"C:relative",
|
||||
"\\\\server\\share",
|
||||
],
|
||||
)
|
||||
def test_rejects_absolute(self, value: str):
|
||||
assert ResourceSecurity().validate({"path": value}) == "path"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
[
|
||||
"a\x00b",
|
||||
"good\x00/../../../etc/passwd",
|
||||
],
|
||||
)
|
||||
def test_rejects_null_bytes(self, value: str):
|
||||
assert ResourceSecurity().validate({"path": value}) == "path"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
[
|
||||
"HEAD~3..HEAD",
|
||||
"v1..v2",
|
||||
"a.b.c",
|
||||
"file.tar.gz",
|
||||
"1.0..2.0",
|
||||
".env",
|
||||
".git/config",
|
||||
"...",
|
||||
"docs/readme.txt",
|
||||
"foo/../bar", # net depth stays >= 0 -> not an escape (SDK semantics)
|
||||
"café/naïve",
|
||||
],
|
||||
)
|
||||
def test_allows_safe_values(self, value: str):
|
||||
"""Dots inside a segment, benign relative paths, and dotfiles pass.
|
||||
|
||||
This mirrors the SDK's component-based `contains_path_traversal`:
|
||||
only a standalone `..` segment counts as traversal. A leading-dot
|
||||
single segment (`.env`) is an ordinary name, not traversal, and
|
||||
passes default screening — filesystem exposure of such names is the
|
||||
handler's concern (e.g. via `safe_join` to a root), not this check.
|
||||
"""
|
||||
assert ResourceSecurity().validate({"path": value}) is None
|
||||
|
||||
def test_exempt_params_skipped(self):
|
||||
security = ResourceSecurity(exempt_params={"ref"})
|
||||
assert security.validate({"ref": "../anything"}) is None
|
||||
# A non-exempt param is still screened.
|
||||
assert security.validate({"path": "../x", "ref": "../y"}) == "path"
|
||||
|
||||
def test_hyphenated_exemption_matches_normalized_param(self):
|
||||
"""`{git-ref}` extracts as `git_ref`; an exemption written with the
|
||||
URI-template (hyphen) spelling must still match it."""
|
||||
security = ResourceSecurity(exempt_params={"git-ref"})
|
||||
assert security.validate({"git_ref": "HEAD~3../x"}) is None
|
||||
assert security.validate({"git_ref": "../x"}) is None
|
||||
# The underscore spelling keeps working too.
|
||||
assert (
|
||||
ResourceSecurity(exempt_params={"git_ref"}).validate({"git_ref": "../x"})
|
||||
is None
|
||||
)
|
||||
# An unrelated hyphenated exemption does not leak onto other params.
|
||||
assert security.validate({"path": "../x"}) == "path"
|
||||
|
||||
def test_wildcard_segments_screened_element_wise(self):
|
||||
"""List values (from wildcard {path*}) are screened per element."""
|
||||
assert ResourceSecurity().validate({"path": ["a", "..", "b"]}) == "path"
|
||||
assert ResourceSecurity().validate({"path": ["a", "b", "c"]}) is None
|
||||
|
||||
def test_non_string_values_ignored(self):
|
||||
assert ResourceSecurity().validate({"n": 5, "flag": True}) is None
|
||||
|
||||
def test_individual_checks_toggleable(self):
|
||||
no_traversal = ResourceSecurity(reject_path_traversal=False)
|
||||
assert no_traversal.validate({"path": "../x"}) is None
|
||||
# but absolute still rejected
|
||||
assert no_traversal.validate({"path": "/etc/passwd"}) == "path"
|
||||
|
||||
def test_returns_first_failing_param_name(self):
|
||||
# dict order preserved; first failing name returned
|
||||
result = ResourceSecurity().validate({"safe": "ok", "bad": ".."})
|
||||
assert result == "bad"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Bare-slim import: the module must not eagerly require the optional SDK
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBareSlimImport:
|
||||
"""`fastmcp-slim` installs the `mcp` SDK only under the `[mcp]` extra.
|
||||
|
||||
The path-safety helpers live in `mcp.shared.path_security`, so importing
|
||||
them at module top would make `from fastmcp.resources import Resource`
|
||||
require the SDK — regressing a previously dependency-free import path.
|
||||
The import must be deferred to the point of actual screening.
|
||||
"""
|
||||
|
||||
def test_resources_import_without_sdk(self):
|
||||
code = textwrap.dedent(
|
||||
"""
|
||||
import sys, builtins
|
||||
_real_import = builtins.__import__
|
||||
|
||||
def blocked_import(name, *args, **kwargs):
|
||||
if name == "mcp" or name.startswith("mcp."):
|
||||
raise ModuleNotFoundError(f"No module named '{name}'")
|
||||
return _real_import(name, *args, **kwargs)
|
||||
|
||||
builtins.__import__ = blocked_import
|
||||
for mod in list(sys.modules):
|
||||
if mod == "mcp" or mod.startswith("mcp."):
|
||||
del sys.modules[mod]
|
||||
|
||||
from fastmcp.resources import Resource, ResourceSecurity # noqa: F401
|
||||
import fastmcp.resources # noqa: F401
|
||||
print("OK")
|
||||
"""
|
||||
)
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", code],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert "OK" in result.stdout
|
||||
|
||||
def test_screening_still_works_with_sdk(self):
|
||||
# With the SDK present (the normal test environment), the deferred
|
||||
# import resolves and screening behaves exactly as before.
|
||||
assert ResourceSecurity().validate({"path": "../etc/passwd"}) == "path"
|
||||
assert ResourceSecurity().validate({"path": "/etc/passwd"}) == "path"
|
||||
assert ResourceSecurity().validate({"path": "safe/file.txt"}) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Enforcement at the server chokepoint (raw-string reads)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestChokepointEnforcement:
|
||||
@pytest.fixture
|
||||
def server(self) -> FastMCP:
|
||||
mcp = FastMCP("test")
|
||||
|
||||
@mcp.resource("file:///{path*}")
|
||||
def read_file(path: str) -> str:
|
||||
return f"content:{path}"
|
||||
|
||||
return mcp
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"uri",
|
||||
[
|
||||
"file:///../etc/passwd",
|
||||
"file:///a/../../b",
|
||||
"file:////etc/passwd", # -> path param '/etc/passwd' (absolute)
|
||||
"file:///a\x00b",
|
||||
],
|
||||
)
|
||||
async def test_traversal_rejected_by_default(self, server: FastMCP, uri: str):
|
||||
with pytest.raises(ResourceSecurityError):
|
||||
await server.read_resource(uri)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"uri",
|
||||
[
|
||||
"file:///docs/readme.txt",
|
||||
"file:///HEAD~3..HEAD",
|
||||
"file:///v1..v2",
|
||||
"file:///file.tar.gz",
|
||||
"file:///.env",
|
||||
],
|
||||
)
|
||||
async def test_safe_uris_pass_by_default(self, server: FastMCP, uri: str):
|
||||
result = await server.read_resource(uri)
|
||||
content = result.contents[0].content
|
||||
assert isinstance(content, str)
|
||||
assert content.startswith("content:")
|
||||
|
||||
|
||||
class TestServerDefaultConfiguration:
|
||||
async def test_server_default_disabled(self):
|
||||
mcp = FastMCP("test", resource_security=None)
|
||||
|
||||
@mcp.resource("file:///{path*}")
|
||||
def read_file(path: str) -> str:
|
||||
return f"content:{path}"
|
||||
|
||||
# Traversal passes when server-wide screening is disabled.
|
||||
result = await mcp.read_resource("file:///../etc/passwd")
|
||||
assert result.contents[0].content == "content:../etc/passwd"
|
||||
|
||||
async def test_server_default_custom_exemption(self):
|
||||
mcp = FastMCP(
|
||||
"test",
|
||||
resource_security=ResourceSecurity(exempt_params={"path"}),
|
||||
)
|
||||
|
||||
@mcp.resource("file:///{path*}")
|
||||
def read_file(path: str) -> str:
|
||||
return f"content:{path}"
|
||||
|
||||
result = await mcp.read_resource("file:///../etc/passwd")
|
||||
assert result.contents[0].content == "content:../etc/passwd"
|
||||
|
||||
async def test_server_default_applies_to_all_templates(self):
|
||||
"""A single server default screens every templated resource."""
|
||||
mcp = FastMCP("test")
|
||||
|
||||
@mcp.resource("a://{path*}")
|
||||
def read_a(path: str) -> str:
|
||||
return path
|
||||
|
||||
@mcp.resource("b://{path*}")
|
||||
def read_b(path: str) -> str:
|
||||
return path
|
||||
|
||||
for scheme in ("a", "b"):
|
||||
with pytest.raises(ResourceSecurityError):
|
||||
await mcp.read_resource(f"{scheme}://../escape")
|
||||
|
||||
|
||||
class TestPerComponentOverride:
|
||||
async def test_component_disable_overrides_server_default(self):
|
||||
mcp = FastMCP("test") # default: screening on
|
||||
|
||||
@mcp.resource("git://diff/{ref}", security=None)
|
||||
def git_diff(ref: str) -> str:
|
||||
return f"diff:{ref}"
|
||||
|
||||
# '..' in the ref is allowed because this component disabled screening.
|
||||
result = await mcp.read_resource("git://diff/HEAD~3..HEAD")
|
||||
assert result.contents[0].content == "diff:HEAD~3..HEAD"
|
||||
|
||||
async def test_component_exemption_overrides_server_default(self):
|
||||
mcp = FastMCP("test")
|
||||
|
||||
@mcp.resource(
|
||||
"git://diff/{ref}",
|
||||
security=ResourceSecurity(exempt_params={"ref"}),
|
||||
)
|
||||
def git_diff(ref: str) -> str:
|
||||
return f"diff:{ref}"
|
||||
|
||||
result = await mcp.read_resource("git://diff/..")
|
||||
assert result.contents[0].content == "diff:.."
|
||||
|
||||
async def test_component_enables_over_disabled_server_default(self):
|
||||
"""A per-component policy overrides a server default of None."""
|
||||
mcp = FastMCP("test", resource_security=None)
|
||||
|
||||
@mcp.resource("file:///{path*}", security=ResourceSecurity())
|
||||
def read_file(path: str) -> str:
|
||||
return path
|
||||
|
||||
with pytest.raises(ResourceSecurityError):
|
||||
await mcp.read_resource("file:///../etc/passwd")
|
||||
|
||||
def test_inherit_default_on_template(self):
|
||||
def read_file(path: str) -> str:
|
||||
return path
|
||||
|
||||
template = ResourceTemplate.from_function(read_file, "file:///{path*}")
|
||||
assert template.security is INHERIT_SECURITY
|
||||
assert template.resolve_security(DEFAULT_RESOURCE_SECURITY) is (
|
||||
DEFAULT_RESOURCE_SECURITY
|
||||
)
|
||||
|
||||
def test_explicit_none_disables(self):
|
||||
def read_file(path: str) -> str:
|
||||
return path
|
||||
|
||||
template = ResourceTemplate.from_function(
|
||||
read_file, "file:///{path*}", security=None
|
||||
)
|
||||
assert template.resolve_security(DEFAULT_RESOURCE_SECURITY) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# End-to-end through the in-memory Client
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEndToEndClient:
|
||||
async def test_traversal_read_gets_clean_not_found(self):
|
||||
"""A traversal attempt over the wire surfaces a non-leaky error.
|
||||
|
||||
`resource://..` survives `AnyUrl` normalisation (the `..` sits in
|
||||
the authority, not the path), so it reaches the server chokepoint
|
||||
and is rejected. The client sees a generic "resource not found"
|
||||
error that never reveals the screening reason.
|
||||
"""
|
||||
mcp = FastMCP("test")
|
||||
|
||||
@mcp.resource("resource://{path*}")
|
||||
def read(path: str) -> str:
|
||||
return path
|
||||
|
||||
async with Client(mcp) as client:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
await client.read_resource("resource://..")
|
||||
|
||||
message = str(exc_info.value)
|
||||
assert "not found" in message.lower()
|
||||
# Non-leaky: the error must not name the failing parameter or policy.
|
||||
assert "path" not in message.lower()
|
||||
assert "security" not in message.lower()
|
||||
|
||||
async def test_legit_read_succeeds(self):
|
||||
mcp = FastMCP("test")
|
||||
|
||||
@mcp.resource("file:///{path*}")
|
||||
def read(path: str) -> str:
|
||||
return f"content:{path}"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
result = await client.read_resource("file:///docs/readme.txt")
|
||||
|
||||
assert result[0].text == "content:docs/readme.txt"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Provider-sourced templates (mounted servers)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProviderSourcedTemplates:
|
||||
"""Templates surfaced by a provider route through the same chokepoint.
|
||||
|
||||
Enforcement lives at the server read chokepoint, not in the decorator,
|
||||
so a mounted server's templates inherit the *parent* server's default
|
||||
policy and are screened before the request is delegated.
|
||||
"""
|
||||
|
||||
async def test_mounted_template_screened_by_parent_default(self):
|
||||
child = FastMCP("child")
|
||||
|
||||
@child.resource("file:///{path*}")
|
||||
def read_file(path: str) -> str:
|
||||
return f"child:{path}"
|
||||
|
||||
parent = FastMCP("parent")
|
||||
parent.mount(child)
|
||||
|
||||
with pytest.raises(ResourceSecurityError):
|
||||
await parent.read_resource("file:///../escape")
|
||||
|
||||
async def test_mounted_template_safe_read_succeeds(self):
|
||||
child = FastMCP("child")
|
||||
|
||||
@child.resource("file:///{path*}")
|
||||
def read_file(path: str) -> str:
|
||||
return f"child:{path}"
|
||||
|
||||
parent = FastMCP("parent")
|
||||
parent.mount(child)
|
||||
|
||||
result = await parent.read_resource("file:///docs/ok.txt")
|
||||
assert result.contents[0].content == "child:docs/ok.txt"
|
||||
|
||||
async def test_parent_default_screens_even_if_child_disabled(self):
|
||||
"""The parent's policy applies even when the child disabled its own.
|
||||
|
||||
Screening runs at each server's chokepoint. A traversal is caught by
|
||||
the parent before delegation regardless of the child's configuration.
|
||||
"""
|
||||
child = FastMCP("child", resource_security=None)
|
||||
|
||||
@child.resource("file:///{path*}")
|
||||
def read_file(path: str) -> str:
|
||||
return f"child:{path}"
|
||||
|
||||
parent = FastMCP("parent") # default screening on
|
||||
parent.mount(child)
|
||||
|
||||
with pytest.raises(ResourceSecurityError):
|
||||
await parent.read_resource("file:///../escape")
|
||||
|
||||
async def test_mounted_template_exempt_param_preserved(self):
|
||||
"""A child template's explicit per-param exemption survives the mount.
|
||||
|
||||
The child opts one parameter out of screening. That policy must be
|
||||
carried through the provider-wrapped template so the parent's read
|
||||
chokepoint honours it instead of falling back to the parent default.
|
||||
"""
|
||||
child = FastMCP("child")
|
||||
|
||||
@child.resource(
|
||||
"git://diff/{ref}/{path*}",
|
||||
security=ResourceSecurity(exempt_params={"ref"}),
|
||||
)
|
||||
def git_diff(ref: str, path: str) -> str:
|
||||
return f"child:{ref}:{path}"
|
||||
|
||||
parent = FastMCP("parent") # default screening on
|
||||
parent.mount(child)
|
||||
|
||||
# `..` in the exempt `ref` param is allowed through the mount.
|
||||
result = await parent.read_resource("git://diff/../safe")
|
||||
assert result.contents[0].content == "child:..:safe"
|
||||
|
||||
# A traversal on the NON-exempt `path` param is still rejected.
|
||||
with pytest.raises(ResourceSecurityError):
|
||||
await parent.read_resource("git://diff/main/../escape")
|
||||
|
||||
async def test_mounted_template_disabled_security_preserved(self):
|
||||
"""A child template that explicitly disables screening keeps that
|
||||
opt-out through the mount rather than inheriting the parent default."""
|
||||
child = FastMCP("child")
|
||||
|
||||
@child.resource("git://raw/{path*}", security=None)
|
||||
def read_raw(path: str) -> str:
|
||||
return f"child:{path}"
|
||||
|
||||
parent = FastMCP("parent") # default screening on
|
||||
parent.mount(child)
|
||||
|
||||
result = await parent.read_resource("git://raw/../escape")
|
||||
assert result.contents[0].content == "child:../escape"
|
||||
|
|
@ -762,14 +762,15 @@ class TestResourceTemplateRequestBuilding:
|
|||
mcp.add_provider(provider)
|
||||
|
||||
async with Client(mcp) as mcp_client:
|
||||
await mcp_client.read_resource(
|
||||
"resource://get_user/..%2F..%2Fadmin%2Fsecret"
|
||||
)
|
||||
# Reserved characters (encoded slash + space) must be
|
||||
# re-encoded when building the outbound URL. A traversal
|
||||
# payload (`..%2F...`) would be rejected by the default
|
||||
# resource-security screening, so use a benign value that
|
||||
# still exercises reserved-character encoding.
|
||||
await mcp_client.read_resource("resource://get_user/a%2Fb%20c")
|
||||
|
||||
assert seen_urls == [
|
||||
httpx.URL(
|
||||
"https://api.example.com/api/v1/users/%2E%2E%2F%2E%2E%2Fadmin%2Fsecret"
|
||||
)
|
||||
httpx.URL("https://api.example.com/api/v1/users/a%2Fb%20c")
|
||||
]
|
||||
|
||||
async def test_resource_template_ignores_unmatched_query_string(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue