Merge pull request #4449 from PrefectHQ/modernize/path-security

Route skill file access through SDK path-security primitives

docs/servers/providers/skills.mdx

@@ -111,6 +111,10 @@ skill://pdf-processing/reference.md
 skill://pdf-processing/examples/sample.pdf
 ```
 
+<Note>
+Supporting-file access is confined to the skill directory. Requested paths are validated before any filesystem access: attempts to traverse out with `..`, inject an absolute path, or smuggle a null byte are rejected with a clear error, and symlinks that resolve outside the skill directory are refused.
+</Note>
+
 ## Provider Architecture
 
 The Skills Provider uses a two-layer architecture to handle both single skills and skill directories.

fastmcp_slim/fastmcp/server/providers/skills/skill_provider.py

@@ -8,6 +8,7 @@ from collections.abc import Sequence
 from pathlib import Path
 from typing import Any, Literal, cast
 
+from mcp.shared.path_security import PathEscapeError, safe_join
 from pydantic import AnyUrl
 
 from fastmcp.resources.base import Resource, ResourceResult
@@ -75,14 +76,12 @@ class SkillFileTemplate(ResourceTemplate):
     async def read(self, arguments: dict[str, Any]) -> str | bytes | ResourceResult:
         """Read a file from the skill directory."""
         file_path = arguments.get("path", "")
-        full_path = self.skill_info.path / file_path
 
-        # Security: ensure path doesn't escape skill directory
+        # Security: reject traversal, absolute-path injection, null bytes, and
+        # symlink escapes before touching the filesystem.
         try:
-            full_path = full_path.resolve()
-            if not full_path.is_relative_to(self.skill_info.path):
-                raise ValueError(f"Path {file_path} escapes skill directory")
-        except ValueError as e:
+            full_path = safe_join(self.skill_info.path, file_path)
+        except PathEscapeError as e:
             raise ValueError(f"Invalid path: {e}") from e
 
         if not full_path.exists():
@@ -119,11 +118,13 @@ class SkillFileTemplate(ResourceTemplate):
         Provided for compatibility with the ResourceTemplate interface.
         """
         file_path = params.get("path", "")
-        full_path = (self.skill_info.path / file_path).resolve()
 
-        # Security: ensure path doesn't escape skill directory
-        if not full_path.is_relative_to(self.skill_info.path):
-            raise ValueError(f"Path {file_path} escapes skill directory")
+        # Security: reject traversal, absolute-path injection, null bytes, and
+        # symlink escapes before touching the filesystem.
+        try:
+            full_path = safe_join(self.skill_info.path, file_path)
+        except PathEscapeError as e:
+            raise ValueError(f"Invalid path: {e}") from e
 
         mime_type, _ = mimetypes.guess_type(str(full_path))
 
@@ -154,12 +155,12 @@ class SkillFileResource(Resource):
 
     async def read(self) -> str | bytes | ResourceResult:
         """Read the file content."""
-        full_path = self.skill_info.path / self.file_path
-
-        # Security check
-        full_path = full_path.resolve()
-        if not full_path.is_relative_to(self.skill_info.path):
-            raise ValueError(f"Path {self.file_path} escapes skill directory")
+        # Security: reject traversal, absolute-path injection, null bytes, and
+        # symlink escapes before touching the filesystem.
+        try:
+            full_path = safe_join(self.skill_info.path, self.file_path)
+        except PathEscapeError as e:
+            raise ValueError(f"Invalid path: {e}") from e
 
         if not full_path.exists():
             raise FileNotFoundError(f"File not found: {self.file_path}")

tests/server/providers/test_skills_provider.py

@@ -15,6 +15,7 @@ from fastmcp.server.providers.skills import (
     SkillsProvider,
 )
 from fastmcp.server.providers.skills._common import parse_frontmatter
+from fastmcp.server.providers.skills.skill_provider import SkillFileResource
 
 
 class TestParseFrontmatter:
@@ -660,6 +661,112 @@ class TestPathTraversalPrevention:
                 )
 
 
+# Attack corpus mirroring the shapes exercised by the SDK's
+# mcp.shared.path_security tests: dot-dot traversal (bare, nested,
+# trailing), absolute-path injection (POSIX and Windows drive forms),
+# and null-byte injection. Each must be rejected before any filesystem
+# access, regardless of which skill surface receives it.
+SKILL_PATH_ESCAPES = [
+    "..",
+    "../secret.txt",
+    "../../../etc/passwd",
+    "sub/../../secret.txt",
+    "nested/../../outside.txt",
+    "/etc/passwd",
+    "/absolute/injection.txt",
+    "C:\\Windows\\system32",
+    "C:relative.txt",
+    "good\x00/../../../etc/passwd",
+    "file\x00.txt",
+]
+
+
+class TestPathSafetyAttackCorpus:
+    """Pin path-safety guards against the SDK's attack-shape corpus.
+
+    Every skill file surface routes user-supplied path parameters through
+    the SDK's ``safe_join``. These tests assert the whole corpus is
+    rejected with a clear error before the filesystem is touched, and
+    that legitimate nested paths continue to resolve.
+    """
+
+    @pytest.fixture
+    def skill_with_secret(self, tmp_path: Path) -> Path:
+        """A skill dir containing a nested file, with a secret one level up."""
+        skill_dir = tmp_path / "corpus-skill"
+        skill_dir.mkdir()
+        (skill_dir / "SKILL.md").write_text("# Corpus\n\nContent")
+        (skill_dir / "docs").mkdir()
+        (skill_dir / "docs" / "nested.txt").write_text("NESTED OK")
+        (tmp_path / "secret.txt").write_text("SECRET DATA")
+        return skill_dir
+
+    async def _template(self, skill_dir: Path):
+        provider = SkillProvider(skill_path=skill_dir)
+        templates = await provider.list_resource_templates()
+        return templates[0]
+
+    @pytest.mark.parametrize("attack", SKILL_PATH_ESCAPES)
+    async def test_template_read_rejects_escape(
+        self, skill_with_secret: Path, attack: str
+    ):
+        template = await self._template(skill_with_secret)
+        with pytest.raises(ValueError, match="Invalid path"):
+            await template.read(arguments={"path": attack})
+
+    @pytest.mark.parametrize("attack", SKILL_PATH_ESCAPES)
+    async def test_template_create_resource_rejects_escape(
+        self, skill_with_secret: Path, attack: str
+    ):
+        template = await self._template(skill_with_secret)
+        with pytest.raises(ValueError, match="Invalid path"):
+            await template.create_resource(
+                uri=f"skill://corpus-skill/{attack}", params={"path": attack}
+            )
+
+    @pytest.mark.parametrize("attack", SKILL_PATH_ESCAPES)
+    async def test_file_resource_read_rejects_escape(
+        self, skill_with_secret: Path, attack: str
+    ):
+        provider = SkillProvider(
+            skill_path=skill_with_secret, supporting_files="resources"
+        )
+        resource = SkillFileResource(
+            uri=AnyUrl("skill://corpus-skill/x"),
+            name="corpus-skill/x",
+            mime_type="text/plain",
+            skill_info=provider.skill_info,
+            file_path=attack,
+        )
+        with pytest.raises(ValueError, match="Invalid path"):
+            await resource.read()
+
+    async def test_template_read_allows_nested_path(self, skill_with_secret: Path):
+        template = await self._template(skill_with_secret)
+        result = await template.read(arguments={"path": "docs/nested.txt"})
+        assert result == "NESTED OK"
+
+    async def test_template_read_allows_within_bounds_dotdot(
+        self, skill_with_secret: Path
+    ):
+        template = await self._template(skill_with_secret)
+        result = await template.read(arguments={"path": "docs/../docs/nested.txt"})
+        assert result == "NESTED OK"
+
+    async def test_file_resource_read_allows_nested_path(self, skill_with_secret: Path):
+        provider = SkillProvider(
+            skill_path=skill_with_secret, supporting_files="resources"
+        )
+        resource = SkillFileResource(
+            uri=AnyUrl("skill://corpus-skill/docs/nested.txt"),
+            name="corpus-skill/docs/nested.txt",
+            mime_type="text/plain",
+            skill_info=provider.skill_info,
+            file_path="docs/nested.txt",
+        )
+        assert await resource.read() == "NESTED OK"
+
+
 async def test_skill_provider_loads_and_serves_utf8_skill_md(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None: