From 8729b56dd58d646555ab73c8d0a809f8ec93c30a Mon Sep 17 00:00:00 2001 From: Amp Date: Fri, 14 Aug 2026 04:09:46 +0000 Subject: [PATCH 1/4] Reject unsafe PythonInterpreter mounts Amp-Thread-ID: https://ampcode.com/threads/T-019fcce1-bc8b-753e-886f-9b77c0ccb376 Co-authored-by: Isaac Miller --- dspy/primitives/python_interpreter.py | 75 ++++++++++++++----- tests/primitives/test_python_interpreter.py | 79 +++++++++++++++++++++ 2 files changed, 138 insertions(+), 16 deletions(-) diff --git a/dspy/primitives/python_interpreter.py b/dspy/primitives/python_interpreter.py index 3b1a63d072..a1c36dc152 100644 --- a/dspy/primitives/python_interpreter.py +++ b/dspy/primitives/python_interpreter.py @@ -75,6 +75,14 @@ def _canonicalize_path(path: PathLike | str) -> str: return os.path.realpath(os.path.expanduser(os.fspath(path))) +def _is_same_or_descendant(path: str, directory: str) -> bool: + """Return whether path is directory or lies beneath it.""" + try: + return os.path.commonpath([path, directory]) == directory + except ValueError: + return False + + def _find_deno_executable() -> str: """Prefer the Deno binary managed by the optional Python package.""" try: @@ -274,6 +282,7 @@ def __init__( self.output_fields = output_fields self.callbacks = list(callbacks or []) self._tools_registered = False + self._mounts = self._canonical_mounts() # TODO later on add enable_run (--allow-run) by proxying subprocess.run through Deno.run() to fix 'emscripten does not support processes' error self._uses_default_deno_command = not deno_command @@ -291,6 +300,7 @@ def __init__( # Also allow reading Deno's cache directory so Pyodide can load its files deno_dir = self._get_deno_dir(deno_executable) + self._validate_write_paths(deno_dir) raw_read_paths = [ self._get_runner_path(), *([deno_dir] if deno_dir else []), @@ -399,38 +409,71 @@ def _get_runner_path(self) -> str: current_dir = os.path.dirname(os.path.abspath(__file__)) return os.path.join(current_dir, "runner.js") + def _canonical_mounts(self) -> list[tuple[str, str, str]]: + """Build unique host-to-guest mounts and reject guest-name aliases.""" + mounts = [] + canonical_to_virtual = {} + virtual_to_canonical = {} + for path in [*self.enable_read_paths, *self.enable_write_paths]: + if not path: + continue + canonical = _canonicalize_path(path) + if canonical in canonical_to_virtual: + continue + virtual = f"/sandbox/{os.path.basename(str(path))}" + other = virtual_to_canonical.get(virtual) + if other is not None and other != canonical: + raise CodeInterpreterError( + f"Ambiguous sandbox mount {virtual!r}: {other!r} and {canonical!r} have the same basename." + ) + canonical_to_virtual[canonical] = virtual + virtual_to_canonical[virtual] = canonical + mounts.append((os.fspath(path), canonical, virtual)) + self._mount_virtual_paths = canonical_to_virtual + return mounts + + def _validate_write_paths(self, deno_dir: str | None) -> None: + """Keep the default runner and Deno cache outside user write grants.""" + runner = _canonicalize_path(self._get_runner_path()) + cache = _canonicalize_path(deno_dir) if deno_dir else None + for path in self.enable_write_paths: + if not path: + continue + canonical = _canonicalize_path(path) + protects_runner = _is_same_or_descendant(runner, canonical) + intersects_cache = cache and ( + _is_same_or_descendant(cache, canonical) or _is_same_or_descendant(canonical, cache) + ) + if protects_runner or intersects_cache: + raise CodeInterpreterError( + f"Write path {canonical!r} overlaps protected PythonInterpreter runtime files." + ) + def _mount_files(self): if self._mounted_files: return - paths_to_mount = [] - if self.enable_read_paths: - paths_to_mount.extend(self.enable_read_paths) - if self.enable_write_paths: - paths_to_mount.extend(self.enable_write_paths) - if not paths_to_mount: + if not self._mounts: return - for path in paths_to_mount: - if not path: - continue + write_paths = {_canonicalize_path(path) for path in self.enable_write_paths if path} + for path, host_path, virtual_path in self._mounts: if not os.path.exists(path): - if self.enable_write_paths and path in self.enable_write_paths: + if host_path in write_paths: open(path, "a").close() else: raise FileNotFoundError(f"Cannot mount non-existent file: {path}") - # Virtual path keeps the user's basename so sandbox code refers to the - # file by the name passed in; host_path is realpath'd so Deno's - # permission check matches the canonical entries in --allow-read. - virtual_path = f"/sandbox/{os.path.basename(str(path))}" - host_path = _canonicalize_path(path) self._send_request("mount_file", {"host_path": host_path, "virtual_path": virtual_path}, f"mounting {path}") self._mounted_files = True def _sync_files(self): if not self.enable_write_paths or not self.sync_files: return + synced = set() for path in self.enable_write_paths: - virtual_path = f"/sandbox/{os.path.basename(str(path))}" host_path = _canonicalize_path(path) + if host_path in synced: + continue + synced.add(host_path) + virtual_path = self._mount_virtual_paths[host_path] sync_msg = _jsonrpc_notification("sync_file", {"virtual_path": virtual_path, "host_path": host_path}) self._write_message(sync_msg, "while syncing files") diff --git a/tests/primitives/test_python_interpreter.py b/tests/primitives/test_python_interpreter.py index 65adcc52db..b45073cd0d 100644 --- a/tests/primitives/test_python_interpreter.py +++ b/tests/primitives/test_python_interpreter.py @@ -413,6 +413,85 @@ def test_custom_deno_command_is_unchanged(): assert interpreter.deno_command is not command +def test_rejects_distinct_mounts_with_same_basename(tmp_path): + first = tmp_path / "first" / "shared.txt" + second = tmp_path / "second" / "shared.txt" + first.parent.mkdir() + second.parent.mkdir() + first.write_text("first") + second.write_text("second") + + with pytest.raises(CodeInterpreterError, match=r"Ambiguous sandbox mount.*shared.txt"): + PythonInterpreter( + deno_command=["custom-deno"], + enable_read_paths=[first], + enable_write_paths=[second], + ) + + +def test_deduplicates_same_canonical_read_and_write_mount(tmp_path): + mounted = tmp_path / "shared.txt" + mounted.write_text("content") + interpreter = PythonInterpreter( + deno_command=["custom-deno"], + enable_read_paths=[mounted], + enable_write_paths=[mounted], + ) + requests = [] + interpreter._send_request = lambda method, params, context: requests.append((method, params)) + + interpreter._mount_files() + + assert requests == [ + ( + "mount_file", + {"host_path": str(mounted.resolve()), "virtual_path": "/sandbox/shared.txt"}, + ) + ] + + +@pytest.mark.parametrize("protected_path", ["runner", "runner_parent", "cache", "cache_parent", "cache_child"]) +def test_default_command_rejects_write_access_to_runtime_resources(monkeypatch, tmp_path, protected_path): + runtime = tmp_path / "runtime" + runner = runtime / "lib" / "runner.js" + cache = runtime / "cache" + runner.parent.mkdir(parents=True) + runner.write_text("runner") + cache.mkdir() + paths = { + "runner": runner, + "runner_parent": runner.parent, + "cache": cache, + "cache_parent": runtime, + "cache_child": cache / "packages" / "module", + } + monkeypatch.setattr(PythonInterpreter, "_get_runner_path", lambda self: str(runner)) + monkeypatch.setattr(PythonInterpreter, "_get_deno_dir", staticmethod(lambda executable: str(cache))) + + with pytest.raises(CodeInterpreterError, match="overlaps protected PythonInterpreter runtime files"): + PythonInterpreter(enable_write_paths=[paths[protected_path]]) + + +def test_runtime_write_validation_resolves_symlinks(monkeypatch, tmp_path): + cache = tmp_path / "cache" + cache.mkdir() + cache_link = tmp_path / "cache-link" + cache_link.symlink_to(cache, target_is_directory=True) + monkeypatch.setattr(PythonInterpreter, "_get_deno_dir", staticmethod(lambda executable: str(cache))) + + with pytest.raises(CodeInterpreterError, match="overlaps protected PythonInterpreter runtime files"): + PythonInterpreter(enable_write_paths=[cache_link / "package"]) + + +def test_custom_command_does_not_infer_protected_runtime_files(tmp_path): + interpreter = PythonInterpreter( + deno_command=["custom-deno", "custom-runner.js"], + enable_write_paths=[tmp_path], + ) + + assert interpreter.deno_command == ["custom-deno", "custom-runner.js"] + + def test_custom_deno_command_preserves_environment(monkeypatch): monkeypatch.setenv("DENO_NO_PACKAGE_JSON", "0") captured = {} From 575ed599ab9c0262139b4737b8ed43ff3b33483d Mon Sep 17 00:00:00 2001 From: Amp Date: Fri, 14 Aug 2026 10:42:57 +0000 Subject: [PATCH 2/4] Revert "Reject unsafe PythonInterpreter mounts" This reverts commit 8729b56dd58d646555ab73c8d0a809f8ec93c30a. Co-authored-by: Isaac Miller --- dspy/primitives/python_interpreter.py | 75 +++++-------------- tests/primitives/test_python_interpreter.py | 79 --------------------- 2 files changed, 16 insertions(+), 138 deletions(-) diff --git a/dspy/primitives/python_interpreter.py b/dspy/primitives/python_interpreter.py index a1c36dc152..3b1a63d072 100644 --- a/dspy/primitives/python_interpreter.py +++ b/dspy/primitives/python_interpreter.py @@ -75,14 +75,6 @@ def _canonicalize_path(path: PathLike | str) -> str: return os.path.realpath(os.path.expanduser(os.fspath(path))) -def _is_same_or_descendant(path: str, directory: str) -> bool: - """Return whether path is directory or lies beneath it.""" - try: - return os.path.commonpath([path, directory]) == directory - except ValueError: - return False - - def _find_deno_executable() -> str: """Prefer the Deno binary managed by the optional Python package.""" try: @@ -282,7 +274,6 @@ def __init__( self.output_fields = output_fields self.callbacks = list(callbacks or []) self._tools_registered = False - self._mounts = self._canonical_mounts() # TODO later on add enable_run (--allow-run) by proxying subprocess.run through Deno.run() to fix 'emscripten does not support processes' error self._uses_default_deno_command = not deno_command @@ -300,7 +291,6 @@ def __init__( # Also allow reading Deno's cache directory so Pyodide can load its files deno_dir = self._get_deno_dir(deno_executable) - self._validate_write_paths(deno_dir) raw_read_paths = [ self._get_runner_path(), *([deno_dir] if deno_dir else []), @@ -409,71 +399,38 @@ def _get_runner_path(self) -> str: current_dir = os.path.dirname(os.path.abspath(__file__)) return os.path.join(current_dir, "runner.js") - def _canonical_mounts(self) -> list[tuple[str, str, str]]: - """Build unique host-to-guest mounts and reject guest-name aliases.""" - mounts = [] - canonical_to_virtual = {} - virtual_to_canonical = {} - for path in [*self.enable_read_paths, *self.enable_write_paths]: - if not path: - continue - canonical = _canonicalize_path(path) - if canonical in canonical_to_virtual: - continue - virtual = f"/sandbox/{os.path.basename(str(path))}" - other = virtual_to_canonical.get(virtual) - if other is not None and other != canonical: - raise CodeInterpreterError( - f"Ambiguous sandbox mount {virtual!r}: {other!r} and {canonical!r} have the same basename." - ) - canonical_to_virtual[canonical] = virtual - virtual_to_canonical[virtual] = canonical - mounts.append((os.fspath(path), canonical, virtual)) - self._mount_virtual_paths = canonical_to_virtual - return mounts - - def _validate_write_paths(self, deno_dir: str | None) -> None: - """Keep the default runner and Deno cache outside user write grants.""" - runner = _canonicalize_path(self._get_runner_path()) - cache = _canonicalize_path(deno_dir) if deno_dir else None - for path in self.enable_write_paths: - if not path: - continue - canonical = _canonicalize_path(path) - protects_runner = _is_same_or_descendant(runner, canonical) - intersects_cache = cache and ( - _is_same_or_descendant(cache, canonical) or _is_same_or_descendant(canonical, cache) - ) - if protects_runner or intersects_cache: - raise CodeInterpreterError( - f"Write path {canonical!r} overlaps protected PythonInterpreter runtime files." - ) - def _mount_files(self): if self._mounted_files: return - if not self._mounts: + paths_to_mount = [] + if self.enable_read_paths: + paths_to_mount.extend(self.enable_read_paths) + if self.enable_write_paths: + paths_to_mount.extend(self.enable_write_paths) + if not paths_to_mount: return - write_paths = {_canonicalize_path(path) for path in self.enable_write_paths if path} - for path, host_path, virtual_path in self._mounts: + for path in paths_to_mount: + if not path: + continue if not os.path.exists(path): - if host_path in write_paths: + if self.enable_write_paths and path in self.enable_write_paths: open(path, "a").close() else: raise FileNotFoundError(f"Cannot mount non-existent file: {path}") + # Virtual path keeps the user's basename so sandbox code refers to the + # file by the name passed in; host_path is realpath'd so Deno's + # permission check matches the canonical entries in --allow-read. + virtual_path = f"/sandbox/{os.path.basename(str(path))}" + host_path = _canonicalize_path(path) self._send_request("mount_file", {"host_path": host_path, "virtual_path": virtual_path}, f"mounting {path}") self._mounted_files = True def _sync_files(self): if not self.enable_write_paths or not self.sync_files: return - synced = set() for path in self.enable_write_paths: + virtual_path = f"/sandbox/{os.path.basename(str(path))}" host_path = _canonicalize_path(path) - if host_path in synced: - continue - synced.add(host_path) - virtual_path = self._mount_virtual_paths[host_path] sync_msg = _jsonrpc_notification("sync_file", {"virtual_path": virtual_path, "host_path": host_path}) self._write_message(sync_msg, "while syncing files") diff --git a/tests/primitives/test_python_interpreter.py b/tests/primitives/test_python_interpreter.py index b45073cd0d..65adcc52db 100644 --- a/tests/primitives/test_python_interpreter.py +++ b/tests/primitives/test_python_interpreter.py @@ -413,85 +413,6 @@ def test_custom_deno_command_is_unchanged(): assert interpreter.deno_command is not command -def test_rejects_distinct_mounts_with_same_basename(tmp_path): - first = tmp_path / "first" / "shared.txt" - second = tmp_path / "second" / "shared.txt" - first.parent.mkdir() - second.parent.mkdir() - first.write_text("first") - second.write_text("second") - - with pytest.raises(CodeInterpreterError, match=r"Ambiguous sandbox mount.*shared.txt"): - PythonInterpreter( - deno_command=["custom-deno"], - enable_read_paths=[first], - enable_write_paths=[second], - ) - - -def test_deduplicates_same_canonical_read_and_write_mount(tmp_path): - mounted = tmp_path / "shared.txt" - mounted.write_text("content") - interpreter = PythonInterpreter( - deno_command=["custom-deno"], - enable_read_paths=[mounted], - enable_write_paths=[mounted], - ) - requests = [] - interpreter._send_request = lambda method, params, context: requests.append((method, params)) - - interpreter._mount_files() - - assert requests == [ - ( - "mount_file", - {"host_path": str(mounted.resolve()), "virtual_path": "/sandbox/shared.txt"}, - ) - ] - - -@pytest.mark.parametrize("protected_path", ["runner", "runner_parent", "cache", "cache_parent", "cache_child"]) -def test_default_command_rejects_write_access_to_runtime_resources(monkeypatch, tmp_path, protected_path): - runtime = tmp_path / "runtime" - runner = runtime / "lib" / "runner.js" - cache = runtime / "cache" - runner.parent.mkdir(parents=True) - runner.write_text("runner") - cache.mkdir() - paths = { - "runner": runner, - "runner_parent": runner.parent, - "cache": cache, - "cache_parent": runtime, - "cache_child": cache / "packages" / "module", - } - monkeypatch.setattr(PythonInterpreter, "_get_runner_path", lambda self: str(runner)) - monkeypatch.setattr(PythonInterpreter, "_get_deno_dir", staticmethod(lambda executable: str(cache))) - - with pytest.raises(CodeInterpreterError, match="overlaps protected PythonInterpreter runtime files"): - PythonInterpreter(enable_write_paths=[paths[protected_path]]) - - -def test_runtime_write_validation_resolves_symlinks(monkeypatch, tmp_path): - cache = tmp_path / "cache" - cache.mkdir() - cache_link = tmp_path / "cache-link" - cache_link.symlink_to(cache, target_is_directory=True) - monkeypatch.setattr(PythonInterpreter, "_get_deno_dir", staticmethod(lambda executable: str(cache))) - - with pytest.raises(CodeInterpreterError, match="overlaps protected PythonInterpreter runtime files"): - PythonInterpreter(enable_write_paths=[cache_link / "package"]) - - -def test_custom_command_does_not_infer_protected_runtime_files(tmp_path): - interpreter = PythonInterpreter( - deno_command=["custom-deno", "custom-runner.js"], - enable_write_paths=[tmp_path], - ) - - assert interpreter.deno_command == ["custom-deno", "custom-runner.js"] - - def test_custom_deno_command_preserves_environment(monkeypatch): monkeypatch.setenv("DENO_NO_PACKAGE_JSON", "0") captured = {} From 3df6c095fff9455a5761c14b3c90437cf80226f5 Mon Sep 17 00:00:00 2001 From: Amp Date: Fri, 14 Aug 2026 10:49:04 +0000 Subject: [PATCH 3/4] Reject colliding PythonInterpreter mounts Amp-Thread-ID: https://ampcode.com/threads/T-019fcce1-bc8b-753e-886f-9b77c0ccb376 Co-authored-by: Isaac Miller --- dspy/primitives/python_interpreter.py | 7 +++++++ tests/primitives/test_python_interpreter.py | 16 ++++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/dspy/primitives/python_interpreter.py b/dspy/primitives/python_interpreter.py index 3b1a63d072..b980a0531e 100644 --- a/dspy/primitives/python_interpreter.py +++ b/dspy/primitives/python_interpreter.py @@ -267,6 +267,13 @@ def __init__( self.enable_read_paths = enable_read_paths or [] self.enable_write_paths = enable_write_paths or [] + mounts = { + _canonicalize_path(path): os.path.basename(os.fspath(path)) + for path in [*self.enable_read_paths, *self.enable_write_paths] + if path + } + if len(mounts) != len(set(mounts.values())): + raise CodeInterpreterError("Mounted files must have unique basenames inside the sandbox.") self.enable_env_vars = enable_env_vars or [] self.enable_network_access = enable_network_access or [] self.sync_files = sync_files diff --git a/tests/primitives/test_python_interpreter.py b/tests/primitives/test_python_interpreter.py index 65adcc52db..ccccc6219f 100644 --- a/tests/primitives/test_python_interpreter.py +++ b/tests/primitives/test_python_interpreter.py @@ -413,6 +413,22 @@ def test_custom_deno_command_is_unchanged(): assert interpreter.deno_command is not command +def test_rejects_mounts_with_the_same_guest_basename(tmp_path): + first = tmp_path / "first" / "shared.txt" + second = tmp_path / "second" / "shared.txt" + first.parent.mkdir() + second.parent.mkdir() + + with pytest.raises(CodeInterpreterError, match="unique basenames"): + PythonInterpreter(deno_command=["deno"], enable_read_paths=[first], enable_write_paths=[second]) + + +def test_allows_same_canonical_file_as_read_and_write_mount(tmp_path): + path = tmp_path / "shared.txt" + + PythonInterpreter(deno_command=["deno"], enable_read_paths=[path], enable_write_paths=[path]) + + def test_custom_deno_command_preserves_environment(monkeypatch): monkeypatch.setenv("DENO_NO_PACKAGE_JSON", "0") captured = {} From 263967beaa4e8b43acd54ee9115f7c6b8427d9d1 Mon Sep 17 00:00:00 2001 From: Amp Date: Fri, 14 Aug 2026 11:03:02 +0000 Subject: [PATCH 4/4] Handle aliased mount collisions Amp-Thread-ID: https://ampcode.com/threads/T-019fcce1-bc8b-753e-886f-9b77c0ccb376 Co-authored-by: Isaac Miller --- dspy/primitives/python_interpreter.py | 15 ++++++++------- tests/primitives/test_python_interpreter.py | 12 ++++++++++++ 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/dspy/primitives/python_interpreter.py b/dspy/primitives/python_interpreter.py index b980a0531e..2f881d4c05 100644 --- a/dspy/primitives/python_interpreter.py +++ b/dspy/primitives/python_interpreter.py @@ -267,13 +267,14 @@ def __init__( self.enable_read_paths = enable_read_paths or [] self.enable_write_paths = enable_write_paths or [] - mounts = { - _canonicalize_path(path): os.path.basename(os.fspath(path)) - for path in [*self.enable_read_paths, *self.enable_write_paths] - if path - } - if len(mounts) != len(set(mounts.values())): - raise CodeInterpreterError("Mounted files must have unique basenames inside the sandbox.") + mounts = {} + for path in [*self.enable_read_paths, *self.enable_write_paths]: + if path: + virtual_path = os.path.basename(os.fspath(path)) + host_path = _canonicalize_path(path) + if virtual_path in mounts and mounts[virtual_path] != host_path: + raise CodeInterpreterError("Mounted files must have unique basenames inside the sandbox.") + mounts[virtual_path] = host_path self.enable_env_vars = enable_env_vars or [] self.enable_network_access = enable_network_access or [] self.sync_files = sync_files diff --git a/tests/primitives/test_python_interpreter.py b/tests/primitives/test_python_interpreter.py index ccccc6219f..8b0b8d16e3 100644 --- a/tests/primitives/test_python_interpreter.py +++ b/tests/primitives/test_python_interpreter.py @@ -429,6 +429,18 @@ def test_allows_same_canonical_file_as_read_and_write_mount(tmp_path): PythonInterpreter(deno_command=["deno"], enable_read_paths=[path], enable_write_paths=[path]) +def test_rejects_alias_basename_colliding_with_another_file(tmp_path): + first = tmp_path / "first.txt" + second = tmp_path / "second" / "shared.txt" + alias = tmp_path / "alias" / "shared.txt" + second.parent.mkdir() + alias.parent.mkdir() + alias.symlink_to(first) + + with pytest.raises(CodeInterpreterError, match="unique basenames"): + PythonInterpreter(deno_command=["deno"], enable_read_paths=[first, alias, second]) + + def test_custom_deno_command_preserves_environment(monkeypatch): monkeypatch.setenv("DENO_NO_PACKAGE_JSON", "0") captured = {}