fastmcp/tests/resources/test_resource_security.py

"""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.
    """

    @pytest.mark.subprocess_heavy
    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"