diff --git a/src/fastmcp/server/providers/filesystem_discovery.py b/src/fastmcp/server/providers/filesystem_discovery.py index 5dc15e11c..db8e0ce83 100644 --- a/src/fastmcp/server/providers/filesystem_discovery.py +++ b/src/fastmcp/server/providers/filesystem_discovery.py @@ -8,6 +8,8 @@ This module provides functions to: from __future__ import annotations +import contextlib +import hashlib import importlib.util import sys from dataclasses import dataclass, field @@ -68,10 +70,16 @@ def _is_package_dir(directory: Path) -> bool: return (directory / "__init__.py").exists() -def _find_package_root(file_path: Path) -> Path | None: +def _find_package_root(file_path: Path, stop_at: Path | None = None) -> Path | None: """Find the root of the package containing this file. - Walks up the directory tree until we find a directory without __init__.py. + Walks up the directory tree until we find a directory without __init__.py, + but never above stop_at (the provider root). This prevents escaping into + ancestor packages when the provider is nested inside a larger Python project. + + Args: + file_path: Path to the Python file. + stop_at: Do not walk above this directory. Typically the provider root. Returns: The package root directory, or None if not in a package. @@ -80,6 +88,8 @@ def _find_package_root(file_path: Path) -> Path | None: package_root = None while current != current.parent: # Stop at filesystem root + if stop_at is not None and current == stop_at.parent: + break # Don't escape above the provider root if _is_package_dir(current): package_root = current current = current.parent @@ -106,15 +116,22 @@ def _compute_module_name(file_path: Path, package_root: Path) -> str: return ".".join(parts) -def import_module_from_file(file_path: Path) -> ModuleType: +def import_module_from_file( + file_path: Path, provider_root: Path | None = None +) -> ModuleType: """Import a Python file as a module. If the file is part of a package (directory has __init__.py), imports it as a proper package member (relative imports work). Otherwise, imports directly using spec_from_file_location. + sys.path is modified only for the duration of the import and restored + immediately after, so no permanent pollution occurs. + Args: file_path: Path to the Python file. + provider_root: The provider's root directory. Prevents package root + discovery from walking above this boundary into ancestor packages. Returns: The imported module. @@ -123,22 +140,24 @@ def import_module_from_file(file_path: Path) -> ModuleType: ImportError: If the module cannot be imported. """ file_path = file_path.resolve() + if provider_root is not None: + provider_root = provider_root.resolve() # Check if this file is part of a package - package_root = _find_package_root(file_path) + package_root = _find_package_root(file_path, stop_at=provider_root) if package_root is not None: # Import as part of a package module_name = _compute_module_name(file_path, package_root) - # Ensure package root's parent is in sys.path + # Temporarily add package root's parent to sys.path for the import package_parent = str(package_root.parent) - if package_parent not in sys.path: + path_added = package_parent not in sys.path + if path_added: sys.path.insert(0, package_parent) - # Import using standard import machinery - # If already imported, reload to pick up changes (for reload mode) try: + # If already imported, reload to pick up changes (for reload mode) if module_name in sys.modules: return importlib.reload(sys.modules[module_name]) return importlib.import_module(module_name) @@ -146,30 +165,71 @@ def import_module_from_file(file_path: Path) -> ModuleType: raise ImportError( f"Failed to import {module_name} from {file_path}: {e}" ) from e + finally: + if path_added: + with contextlib.suppress(ValueError): + sys.path.remove(package_parent) else: # Import directly using spec_from_file_location - module_name = file_path.stem - - # Ensure parent directory is in sys.path for imports + stem = file_path.stem parent_dir = str(file_path.parent) - if parent_dir not in sys.path: + + # Determine the sys.modules key. Prefer the bare stem (so that sibling + # imports like `import helpers` resolve correctly), but fall back to a + # private collision-safe key if the bare stem is already claimed by + # something else (stdlib, a third-party package, or another provider file + # from a different directory). + existing = sys.modules.get(stem) + if existing is not None and getattr(existing, "__file__", None) != str( + file_path + ): + module_name = f"_fastmcp_{stem}_{hashlib.sha1(str(file_path).encode()).hexdigest()[:12]}" + else: + module_name = stem + + # Temporarily add parent to sys.path so module-level sibling imports resolve. + # Safe to remove after exec_module: all top-level imports are resolved by then, + # and sibling files imported as side effects are already in sys.modules. + path_added = parent_dir not in sys.path + if path_added: sys.path.insert(0, parent_dir) - spec = importlib.util.spec_from_file_location(module_name, file_path) - if spec is None or spec.loader is None: - raise ImportError(f"Cannot load spec for {file_path}") - - module = importlib.util.module_from_spec(spec) - sys.modules[module_name] = module - try: - spec.loader.exec_module(module) - except Exception as e: - # Clean up sys.modules on failure - sys.modules.pop(module_name, None) - raise ImportError(f"Failed to execute module {file_path}: {e}") from e + spec = importlib.util.spec_from_file_location(module_name, file_path) + if spec is None or spec.loader is None: + raise ImportError(f"Cannot load spec for {file_path}") - return module + existing = sys.modules.get(module_name) + if existing is not None: + # Re-exec in place rather than importlib.reload: reload() re-finds + # the module by name via sys.path, which fails for private keys + # (the file is tool.py, not _fastmcp_tool_xxx.py). + existing.__spec__ = spec + existing.__loader__ = spec.loader + existing.__file__ = str(file_path) + try: + spec.loader.exec_module(existing) + except Exception as e: + raise ImportError( + f"Failed to reload module {file_path}: {e}" + ) from e + return existing + + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + + try: + spec.loader.exec_module(module) + except Exception as e: + # Clean up sys.modules on failure + sys.modules.pop(module_name, None) + raise ImportError(f"Failed to execute module {file_path}: {e}") from e + + return module + finally: + if path_added: + with contextlib.suppress(ValueError): + sys.path.remove(parent_dir) def extract_components(module: ModuleType) -> list[FastMCPComponent]: @@ -316,10 +376,7 @@ def discover_and_import(root: Path) -> DiscoveryResult: for file_path in discover_files(root): try: - module = import_module_from_file(file_path) - except ImportError as e: - result.failed_files[file_path] = str(e) - continue + module = import_module_from_file(file_path, provider_root=root) except Exception as e: result.failed_files[file_path] = str(e) continue diff --git a/tests/fs/test_discovery.py b/tests/fs/test_discovery.py index 1b81911de..4f88bd34e 100644 --- a/tests/fs/test_discovery.py +++ b/tests/fs/test_discovery.py @@ -1,5 +1,6 @@ """Tests for filesystem discovery module.""" +import sys from pathlib import Path from fastmcp.prompts.base import Prompt @@ -485,3 +486,101 @@ def my_tool(x: str) -> str: assert components[0].version is None meta = components[0].get_meta() assert "version" not in meta["fastmcp"] + + +class TestImportMachineryFixes: + """Tests for import machinery correctness: sys.path cleanup, sys.modules safety, package root boundary.""" + + def test_syspath_not_polluted_after_import(self, tmp_path: Path): + """sys.path should not contain the file's parent after import_module_from_file returns.""" + (tmp_path / "mymod.py").write_text("VALUE = 1") + path_before = list(sys.path) + import_module_from_file(tmp_path / "mymod.py") + assert sys.path == path_before + + def test_syspath_not_polluted_after_package_import(self, tmp_path: Path): + """sys.path should not contain the package root's parent after a package import.""" + pkg = tmp_path / "mypkg_syspath" + pkg.mkdir() + (pkg / "__init__.py").write_text("") + (pkg / "mod.py").write_text("VALUE = 2") + path_before = list(sys.path) + import_module_from_file(pkg / "mod.py", provider_root=tmp_path) + assert sys.path == path_before + + def test_stdlib_not_shadowed_by_same_named_file(self, tmp_path: Path): + """A provider file named json.py must not overwrite sys.modules['json'].""" + import json as stdlib_json + + saved = sys.modules["json"] + try: + (tmp_path / "json.py").write_text( + "from fastmcp.tools import tool\n@tool\ndef parse(): return 'provider'" + ) + import_module_from_file(tmp_path / "json.py") + assert sys.modules.get("json") is stdlib_json + finally: + sys.modules["json"] = saved + + def test_same_stem_files_get_independent_modules(self, tmp_path: Path): + """Two files with the same stem in different directories must not collide in sys.modules. + + The first-imported file keeps the bare stem key; the second gets a private key. + Both modules must be independently accessible with correct content. + """ + dir_a = tmp_path / "a" + dir_b = tmp_path / "b" + dir_a.mkdir() + dir_b.mkdir() + (dir_a / "helpers.py").write_text("ORIGIN = 'a'") + (dir_b / "helpers.py").write_text("ORIGIN = 'b'") + + mod_a = import_module_from_file(dir_a / "helpers.py") + mod_b = import_module_from_file(dir_b / "helpers.py") + + assert mod_a.ORIGIN == "a" + assert mod_b.ORIGIN == "b" + # The first module retains the bare stem key; the second uses a private key. + # They must be distinct objects — the second import must not have clobbered the first. + assert mod_a is not mod_b + assert sys.modules.get("helpers") is not mod_b + + def test_package_root_bounded_by_provider_root(self, tmp_path: Path): + """When the provider root is nested inside a larger package, import_module_from_file + with provider_root must not escape into ancestor packages. + + The generated module name should be relative to the provider root (e.g. "myprovider.tools"), + not to an ancestor package (e.g. "myproject.myprovider.tools"), and tmp_path (the + ancestor's parent) must not be added to sys.path. + """ + # Use a name that won't collide with any installed package + project = tmp_path / "myproject" + project.mkdir() + (project / "__init__.py").write_text("") + provider = project / "myprovider" + provider.mkdir() + (provider / "__init__.py").write_text("") + (provider / "tools.py").write_text("VALUE = 42") + + path_before = set(sys.path) + mod = import_module_from_file(provider / "tools.py", provider_root=provider) + path_after = set(sys.path) + + # Module was correctly imported + assert mod.VALUE == 42 + # sys.path should not contain tmp_path (the ancestor's grandparent); + # that would only happen if the package root escaped past the provider boundary + assert str(tmp_path) not in (path_after - path_before) + # The module name is bounded to the provider root, not "myproject.myprovider.tools" + assert mod.__name__ == "myprovider.tools" + + def test_non_package_reload_returns_updated_content(self, tmp_path: Path): + """Re-importing a non-package file should reflect file changes (exec_module path).""" + f = tmp_path / "reloadable_np.py" + f.write_text("VALUE = 'original'") + mod = import_module_from_file(f) + assert mod.VALUE == "original" + + f.write_text("VALUE = 'updated'") + mod2 = import_module_from_file(f) + assert mod2.VALUE == "updated"