From 1a720421159e1314d049beec10ba91951f3086fd Mon Sep 17 00:00:00 2001 From: brooklyn! Date: Thu, 24 Sep 2026 08:02:48 -0500 Subject: [PATCH] fix(docker): bind a Windows workspace when /workspace is already claimed A volume that already owns /workspace skipped the configured working directory, so tools treated that host path as unmounted. Bind it at a second mount, or point tools at the volume that already has it, for any drive path. --- tests/tools/test_docker_windows_cwd_mount.py | 99 ++++++++++++++++ tools/environments/docker.py | 118 ++++++++++++++++--- tools/file_operations.py | 10 +- tools/image_source.py | 8 ++ tools/terminal_tool.py | 110 +++++++++++++++-- tools/terminal_tool_background.py | 1 + tools/terminal_tool_config.py | 53 +++++++++ tools/terminal_tool_guards.py | 1 + 8 files changed, 370 insertions(+), 30 deletions(-) create mode 100644 tests/tools/test_docker_windows_cwd_mount.py diff --git a/tests/tools/test_docker_windows_cwd_mount.py b/tests/tools/test_docker_windows_cwd_mount.py new file mode 100644 index 0000000000..37b6e57c75 --- /dev/null +++ b/tests/tools/test_docker_windows_cwd_mount.py @@ -0,0 +1,99 @@ +"""Docker must bind a configured Windows workspace even when /workspace is taken. + +A volume that already claims ``/workspace`` used to set +``workspace_explicitly_mounted`` and skip the configured working directory. +Tools then refused that host path because nothing in the container mapped it. +The bind has to land (at ``/workspace`` when free, otherwise a second mount) +and tools have to follow that mount — any drive, any directory, not one user. +""" + +import os + +import tools.terminal_tool as tt +from tools.environments.docker import DockerEnvironment + + +WIN_WS = r"D:\Work\proj" +WIN_FILE = r"D:\Work\proj\Downloads\clip.jpg" + + +def _mount(volumes, host_cwd, auto_mount=True): + env = object.__new__(DockerEnvironment) + env._persistent = False + volume_args, _writable = DockerEnvironment._mount_args( + env, volumes, host_cwd, auto_mount, "t") + return env, volume_args + + +def _preserve_drive_path(monkeypatch, existing: str): + """POSIX ``abspath`` would prefix the process cwd onto ``D:\\...`` and hide it.""" + real_abspath = os.path.abspath + real_isdir = os.path.isdir + + def abspath(path): + if isinstance(path, str) and len(path) >= 3 and path[1] == ":" and path[2] in "\\/": + return path + return real_abspath(path) + + def isdir(path): + return path == existing or real_isdir(path) + + monkeypatch.setattr(os.path, "abspath", abspath) + monkeypatch.setattr(os.path, "isdir", isdir) + + +class TestClaimedWorkspaceStillBindsConfiguredCwd: + def test_real_directory_is_bound_beside_an_existing_workspace_volume(self, tmp_path): + host = tmp_path / "proj" + host.mkdir() + other = tmp_path / "other" + other.mkdir() + env, volume_args = _mount([f"{other}:/workspace"], str(host), auto_mount=True) + + specs = [arg for arg in volume_args if arg != "-v"] + assert any(spec.startswith(f"{other}:/workspace") for spec in specs) + mount = env.host_cwd_mount + assert env.host_cwd == os.path.abspath(str(host)) + assert mount and mount != "/workspace" + assert any(spec == f"{os.path.abspath(str(host))}:{mount}" for spec in specs) + + def test_windows_drive_cwd_binds_when_workspace_is_claimed(self, monkeypatch): + _preserve_drive_path(monkeypatch, WIN_WS) + env, volume_args = _mount( + [r"E:\elsewhere:/workspace"], WIN_WS, auto_mount=False) + + specs = [arg for arg in volume_args if arg != "-v"] + mount = env.host_cwd_mount + assert env.host_cwd == WIN_WS + assert mount and mount != "/workspace" + assert f"{WIN_WS}:{mount}" in specs + # The volume that already owns /workspace stays; we do not steal it. + assert any(":/workspace" in spec and WIN_WS not in spec for spec in specs) + + def test_same_windows_directory_already_at_workspace_points_tools_there(self, monkeypatch): + _preserve_drive_path(monkeypatch, WIN_WS) + env, volume_args = _mount([f"{WIN_WS}:/workspace"], WIN_WS, auto_mount=False) + + specs = [arg for arg in volume_args if arg != "-v"] + assert specs.count(f"{WIN_WS}:/workspace") == 1 + assert env.host_cwd == WIN_WS + assert env.host_cwd_mount == "/workspace" + + +class TestToolsFollowTheMount: + def test_live_cwd_and_child_paths_use_the_second_mount(self): + mount = "/host-cwd" + env = type("E", (), {"env_type": "docker", "host_cwd": WIN_WS, "host_cwd_mount": mount})() + assert tt._sanitize_cwd_for_live_env(env, WIN_WS) == mount + assert tt._sanitize_cwd_for_live_env(env, WIN_FILE) == f"{mount}/Downloads/clip.jpg" + + def test_file_ops_rewrite_a_mounted_windows_path(self): + from tools.file_operations import ShellFileOperations + + env = type("E", (), { + "env_type": "docker", "cwd": "/host-cwd", + "host_cwd": WIN_WS, "host_cwd_mount": "/host-cwd", + })() + ops = ShellFileOperations(env, cwd="/host-cwd") + assert ops._expand_path(WIN_FILE) == "/host-cwd/Downloads/clip.jpg" + assert ops._expand_path("/workspace/keep") == "/workspace/keep" diff --git a/tools/environments/docker.py b/tools/environments/docker.py index 5515521cdc..4fed39fcc7 100644 --- a/tools/environments/docker.py +++ b/tools/environments/docker.py @@ -21,9 +21,11 @@ import uuid from pathlib import Path from typing import Optional -from tools.environments.base import BaseEnvironment, EnvironmentConnectionError +from tools.environments.base import BaseEnvironment, EnvironmentConnectionError, _SHELL_ENV_NAME_RE +from tools.terminal_tool_config import ( + _host_path_key, _is_windows_drive_path, cwd_follows_host_mount, +) from tools.environments.base_output import _popen_bash -from tools.environments.base_session_env import _SHELL_ENV_NAME_RE from tools.environments.docker_egress import ( _EGRESS_LABEL_KEY, _critical_egress_env_names, _egress_enforce_on_docker, _egress_proxy_args_for_docker, _egress_reuse_fingerprint, check_docker_env_collisions, check_extra_args_collisions, @@ -511,6 +513,57 @@ def _host_user_args(run_as_host_user: bool) -> list[str]: return [] +_VOLUME_SPEC_RE = re.compile(r"^(?P.+):(?P/[^:]+)(?::[^:]*)?$") +# Second mount when a user volume already owns /workspace. Not a username. +_HOST_CWD_FALLBACK_MOUNTS = ("/host-cwd", "/host-cwd-2", "/host-cwd-3") + + +def _split_volume_spec(spec: str) -> tuple[str, str] | None: + """``host:container[:mode]`` → ``(host, container)``. Drive-letter hosts keep their colon.""" + if not isinstance(spec, str): + return None + match = _VOLUME_SPEC_RE.match(spec.strip()) + if not match: + return None + return match.group("host"), match.group("container") + + +def _container_mount_taken(volume_args: list[str], mount: str) -> bool: + target = mount.rstrip("/") or "/" + for arg in volume_args: + parsed = _split_volume_spec(arg) + if parsed and (parsed[1].rstrip("/") or "/") == target: + return True + return False + + +def _existing_host_mount(volume_args: list[str], host_cwd_abs: str) -> str | None: + """Container path if a user volume already bind-mounts this host directory.""" + want = _host_path_key(host_cwd_abs) + if not want: + return None + for arg in volume_args: + parsed = _split_volume_spec(arg) + if parsed and _host_path_key(parsed[0]) == want: + return parsed[1] + return None + + +def _free_host_cwd_mount(volume_args: list[str]) -> str: + for candidate in _HOST_CWD_FALLBACK_MOUNTS: + if not _container_mount_taken(volume_args, candidate): + return candidate + return _HOST_CWD_FALLBACK_MOUNTS[-1] + + +def _abs_host_cwd(host_cwd: str) -> str: + """Absolute host path. A Windows drive path is not prefixed with the POSIX process cwd.""" + expanded = os.path.expanduser(host_cwd) + if _is_windows_drive_path(expanded) and os.name != "nt": + return expanded + return os.path.abspath(expanded) + + class DockerEnvironment(BaseEnvironment): """Hardened Docker container execution (caps dropped, no-new-privileges, PID limits, size-limited tmpfs). The container is the security boundary — its filesystem stays @@ -570,6 +623,13 @@ class DockerEnvironment(BaseEnvironment): resource_args = self._resource_args(image, cpu, memory, disk, network, shm_size, extra_args) volume_args, writable_args = self._mount_args(volumes, host_cwd, auto_mount_cwd, task_id) + mount = getattr(self, "host_cwd_mount", None) + if mount and cwd_follows_host_mount(cwd, mount): + logger.info( + "Container cwd follows configured host workspace at %s (requested %s)", + mount, cwd) + cwd = mount + self.cwd = mount volume_args.extend(_readonly_skill_mount_args()) egress_label, egress_volume_args, egress_host_args, env_args, validated_extra = ( self._egress_and_env_args(extra_args)) @@ -697,7 +757,15 @@ class DockerEnvironment(BaseEnvironment): def _mount_args(self, volumes, host_cwd, auto_mount_cwd, task_id) -> tuple[list[str], list[str]]: """``(volume_args, writable_args)`` for user volumes, host cwd and /workspace,/root. - Persistent mode bind-mounts from TERMINAL_SANDBOX_DIR (default ~/.hermes/sandboxes/).""" + + Persistent mode bind-mounts from TERMINAL_SANDBOX_DIR (default ~/.hermes/sandboxes/). + A configured host working directory is bound even when another volume already + claims ``/workspace``: at ``/workspace`` when that path is free, otherwise at + a second mount. ``host_cwd`` / ``host_cwd_mount`` tell tools which container + path is that directory. A Windows drive path is bound whenever it exists on + the host — it can never be a path inside the Linux container, and the check + is the drive shape, not a username. + """ volume_args: list[str] = [] for vol in (volumes or []): if not isinstance(vol, str): @@ -712,17 +780,33 @@ class DockerEnvironment(BaseEnvironment): volume_args.extend(["-v", vol]) workspace_explicitly_mounted = any(":/workspace" in v for v in volume_args) - host_cwd_abs = os.path.abspath(os.path.expanduser(host_cwd)) if host_cwd else "" - bind_host_cwd = ( - auto_mount_cwd and bool(host_cwd_abs) and os.path.isdir(host_cwd_abs) - and not workspace_explicitly_mounted) - if auto_mount_cwd and host_cwd and not os.path.isdir(host_cwd_abs): + host_cwd_abs = _abs_host_cwd(host_cwd) if host_cwd else "" + windows_cwd = _is_windows_drive_path(host_cwd or "") or _is_windows_drive_path(host_cwd_abs) + host_dir_exists = bool(host_cwd_abs) and os.path.isdir(host_cwd_abs) + should_bind = host_dir_exists and (auto_mount_cwd or windows_cwd) + if (auto_mount_cwd or windows_cwd) and host_cwd and not host_dir_exists: logger.debug("Skipping docker cwd mount: host_cwd is not a valid directory: %s", host_cwd) - # The host directory actually bound at /workspace, if any. Readers that - # only hold the env instance (cwd remapping on live envs) use it to - # recognize a session workspace registered as a raw host path. - self.host_cwd = host_cwd_abs if bind_host_cwd else None - mount_workspace = not bind_host_cwd and not workspace_explicitly_mounted + + existing_mount = _existing_host_mount(volume_args, host_cwd_abs) if should_bind else None + if existing_mount: + # Already bind-mounted (often the volume that claimed /workspace). Point + # tools at that container path instead of adding a second -v. + self.host_cwd = host_cwd_abs + self.host_cwd_mount = existing_mount + bind_target = None + elif should_bind: + bind_target = ( + "/workspace" if not workspace_explicitly_mounted + else _free_host_cwd_mount(volume_args)) + self.host_cwd = host_cwd_abs + self.host_cwd_mount = bind_target + else: + self.host_cwd = None + self.host_cwd_mount = None + bind_target = None + + bind_at_workspace = bind_target == "/workspace" + mount_workspace = not bind_at_workspace and not workspace_explicitly_mounted writable_args: list[str] = [] if self._persistent: @@ -741,10 +825,10 @@ class DockerEnvironment(BaseEnvironment): writable_args += ["--tmpfs", "/workspace:rw,exec,size=10g"] if mount_workspace else [] writable_args += ["--tmpfs", "/home:rw,exec,size=1g", "--tmpfs", "/root:rw,exec,size=1g"] - if bind_host_cwd: - logger.info("Mounting configured host cwd to /workspace: %s", host_cwd_abs) - volume_args = ["-v", f"{host_cwd_abs}:/workspace", *volume_args] - elif workspace_explicitly_mounted: + if bind_target: + logger.info("Mounting configured host cwd to %s: %s", bind_target, host_cwd_abs) + volume_args = ["-v", f"{host_cwd_abs}:{bind_target}", *volume_args] + elif workspace_explicitly_mounted and not existing_mount: logger.debug("Skipping docker cwd mount: /workspace already mounted by user config") return volume_args, writable_args diff --git a/tools/file_operations.py b/tools/file_operations.py index 2a9d3aab63..e2002e06a9 100644 --- a/tools/file_operations.py +++ b/tools/file_operations.py @@ -468,7 +468,15 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): def _expand_path(self, path: str) -> str: """Expand ``~`` / ``~user`` via the backend's shell (its HOME, not the - host's). Must run BEFORE shell escaping — ~ doesn't expand in quotes.""" + host's). A host path under the configured workspace mount is rewritten + to that container path first, so a Windows drive path is readable + inside Docker. Must run BEFORE shell escaping — ~ doesn't expand in quotes.""" + from tools.terminal_tool_config import translate_mounted_host_path + host_root = getattr(self.env, "host_cwd", None) + container_root = getattr(self.env, "host_cwd_mount", None) or "/workspace" + translated = translate_mounted_host_path(path, host_root or "", container_root) + if translated: + return translated if not path or not path.startswith('~'): return path result = self._exec("echo $HOME") diff --git a/tools/image_source.py b/tools/image_source.py index c780cf7ef7..deef22e6b8 100644 --- a/tools/image_source.py +++ b/tools/image_source.py @@ -234,6 +234,14 @@ async def _resolve_container_fallback( f"'{p}' is not reachable inside the sandbox and no active sandbox " f"session is available to read it", src=src, origin="container") + from tools.terminal_tool_config import translate_mounted_host_path + translated = translate_mounted_host_path( + str(p), + getattr(env, "host_cwd", None) or "", + getattr(env, "host_cwd_mount", None) or "/workspace", + ) + if translated: + p = Path(translated) # Bound the read INSIDE the sandbox: head -c caps at ingest-limit+1 (+1 distinguishes "at the # cap" from "over") so /dev/zero can't stream unbounded base64 into host memory. The input # redirect avoids argv (leading-dash paths); tr -d instead of GNU-only base64 -w0 (BusyBox). diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index d3b8226266..4a8b454cc7 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -44,9 +44,9 @@ from tools.terminal_tool_lifecycle import ( _evict_environment_for_task, cleanup_all_environments, ensure_task_env, ) from tools.terminal_tool_config import ( - _is_container_backend, _is_host_cwd, _is_mounted_host_cwd, _is_unusable_container_cwd, _parse_env_var, - coerce_ssh_remote_cwd, - _plugin_env_flag, _quiet, _safe_getcwd, _tenv, _tenv_bool, + _is_container_backend, _is_host_cwd, _is_mounted_host_cwd, _is_unusable_container_cwd, + _is_windows_drive_path, _parse_env_var, _plugin_env_flag, _quiet, _safe_getcwd, _tenv, _tenv_bool, + coerce_ssh_remote_cwd, translate_mounted_host_path, ) from tools.terminal_tool_backends import ( _REQUIREMENT_CHECKERS, _VERCEL_SANDBOX_DEFAULT_CWD, _check_plugin_requirements, @@ -285,8 +285,8 @@ def _sanitize_cwd_for_live_env(env: Any, new_cwd: str) -> Optional[str]: file operations with an unrelated ``cd:`` error. Prefix-shaped host paths are already rejected on the creation paths. This write classifies the directory mounted at ``/workspace`` as unusable before that prefix - heuristic, then remaps the match to ``/workspace`` instead of storing the - host path. Non-container backends apply the override + heuristic, then remaps the match (or a child of it) to its container mount + instead of storing the host path. Non-container backends apply the override verbatim (ACP project-root switching must keep working). """ env_type = getattr(env, "env_type", None) @@ -299,8 +299,14 @@ def _sanitize_cwd_for_live_env(env: Any, new_cwd: str) -> Optional[str]: # every later file-tools exec would `cd` to it (exit 126). if not _is_unusable_container_cwd(new_cwd, mounted_host=mounted): return new_cwd - if _is_mounted_host_cwd(new_cwd, mounted): - return "/workspace" + if mounted: + # The bind may sit at a fallback mount when /workspace is claimed. + container_mount = getattr(env, "host_cwd_mount", None) or "/workspace" + if _is_mounted_host_cwd(new_cwd, mounted): + return container_mount + translated = translate_mounted_host_path(new_cwd, mounted, container_mount) + if translated: + return translated return None @@ -555,7 +561,7 @@ def _lookup_active_env(effective_task_id: str, task_id: Optional[str]): def _resolve_task_host_cwd(config: Dict[str, Any], task_id: Optional[str]) -> Optional[str]: - """Host directory to bind-mount at ``/workspace`` for *task_id*'s container. + """Host directory to bind into *task_id*'s container. Single owner of the cwd-mount policy for every creation site. Shared- container mode: the ``TERMINAL_CWD``-derived ``config["host_cwd"]``. @@ -565,8 +571,27 @@ def _resolve_task_host_cwd(config: Dict[str, Any], task_id: Optional[str]) -> Op fresh session's mount from it would leak the previous session's directory. Overrides tagged ``cwd_source: "process"`` are refused for the same reason; ``cwd_source: "session"`` or untagged (ACP/RL) overrides mount. + A Windows drive path is still a mount source when the cwd-to-/workspace + flag is off: it cannot exist inside the Linux container. The container + path may be ``/workspace`` or a second mount when that path is taken. """ - if config.get("env_type") != "docker" or not config.get("docker_mount_cwd_to_workspace"): + if config.get("env_type") != "docker": + return None + if not config.get("docker_mount_cwd_to_workspace"): + # POSIX homes stay behind the opt-in flag (isolation). A Windows drive + # workspace cannot exist in the container, so it is still a mount source. + host = config.get("host_cwd") + if isinstance(host, str) and _is_windows_drive_path(host): + return host + overrides = resolve_task_overrides(task_id) if task_id else {} + candidate = overrides.get("cwd") if isinstance(overrides, dict) else None + if ( + isinstance(overrides, dict) + and overrides.get("cwd_source") != "process" + and isinstance(candidate, str) + and _is_windows_drive_path(candidate) + ): + return candidate return None # Top-level CLI parent ("default") is a single-session process — legacy behavior. if not _docker_session_isolation_enabled() or _resolve_container_task_id(task_id) == "default": @@ -660,6 +685,21 @@ def _resolve_config_cwd(env_type: str, mount_docker_cwd: bool) -> tuple: ): host_cwd = candidate cwd = "/workspace" + elif env_type == "docker" and _is_windows_drive_path(cwd): + # A Windows workspace cannot exist inside the Linux container. Bind it + # even when docker_mount_cwd_to_workspace is off; the env retargets cwd + # to the mount (which may not be /workspace). + candidate = os.path.expanduser(cwd) + if os.name == "nt": + candidate = os.path.abspath(candidate) + if os.path.isdir(candidate): + host_cwd = candidate + cwd = candidate + else: + logger.info("Ignoring TERMINAL_CWD=%r for %s backend " + "(host/relative path won't work in sandbox). Using %r instead.", + cwd, env_type, default_cwd) + cwd = default_cwd elif _is_container_backend(env_type) and cwd and _is_unusable_container_cwd(cwd) and cwd != default_cwd: logger.info("Ignoring TERMINAL_CWD=%r for %s backend " "(host/relative path won't work in sandbox). Using %r instead.", @@ -827,6 +867,47 @@ def _resolve_notification_flag_conflict(*, notify_on_complete: bool, watch_patte return watch_patterns, "" +def _rewrite_via_env_mount(path: str, env) -> str | None: + """Container path for *path* when *env* bind-mounted that host directory.""" + if env is None or not path: + return None + host = getattr(env, "host_cwd", None) + mount = getattr(env, "host_cwd_mount", None) + if not isinstance(host, str) or not host or not mount: + return None + return translate_mounted_host_path(path, host, mount) + + +def _mount_envs(env): + if env is not None: + return [env] + return list(_active_environments.values()) + + +def _container_visible_cwd(path: str, env_type: str | None, env=None) -> str: + if not path or not _is_container_backend(env_type or ""): + return path + for item in _mount_envs(env): + translated = _rewrite_via_env_mount(path, item) + if translated: + return translated + return path + + +def _container_visible_default(default_cwd: str, env_type: str | None, env=None) -> str: + """Point a planner ``/workspace`` assumption at the mount that actually holds the host cwd.""" + if not _is_container_backend(env_type or ""): + return default_cwd + for item in _mount_envs(env): + mount = getattr(item, "host_cwd_mount", None) + host = getattr(item, "host_cwd", None) + if not mount or not host or mount == default_cwd: + continue + if default_cwd == "/workspace" or _is_unusable_container_cwd(default_cwd): + return mount + return default_cwd + + def _resolve_command_cwd( *, workdir: Optional[str], @@ -834,6 +915,7 @@ def _resolve_command_cwd( session_key: Optional[str] = None, env_type: Optional[str] = None, mounted_host: Optional[str] = None, + env=None, ) -> str: """cwd for a command: explicit ``workdir`` > the session's own cwd record > ``default_cwd``. @@ -852,11 +934,14 @@ def _resolve_command_cwd( Same guard class as the env-creation sanitizers (#50636, #54447); this is the per-command sibling site. """ if workdir: - return coerce_ssh_remote_cwd(workdir, env_type) + return coerce_ssh_remote_cwd(_container_visible_cwd(workdir, env_type, env), env_type) recorded = get_session_cwd(session_key) if recorded and _is_container_backend(env_type) and _is_unusable_container_cwd( recorded, mounted_host=mounted_host ): + visible = _container_visible_cwd(recorded, env_type, env) + if visible != recorded: + return visible if _is_mounted_host_cwd(recorded, mounted_host): logger.info( "Remapping recorded session cwd %r for %s backend " @@ -869,8 +954,8 @@ def _resolve_command_cwd( "(host/relative path won't work in sandbox). Using %r instead.", recorded, env_type, default_cwd, ) - return default_cwd - return coerce_ssh_remote_cwd(recorded or default_cwd, env_type) + return _container_visible_default(default_cwd, env_type, env) + return recorded or coerce_ssh_remote_cwd(_container_visible_default(default_cwd, env_type, env), env_type) def _error_json(error: str, *, exit_code: int = -1, status: Optional[str] = None, **extra) -> str: @@ -1168,6 +1253,7 @@ def _run_foreground( command_cwd = _resolve_command_cwd( workdir=workdir, default_cwd=plan.cwd, session_key=session_key, env_type=env_type, mounted_host=getattr(env, "host_cwd", None) or plan.host_cwd, + env=env, ) # bounded_capture: model-facing output keeps a head/tail window # while streaming so a verbose command can't OOM the gateway; diff --git a/tools/terminal_tool_background.py b/tools/terminal_tool_background.py index a2146a6511..306b26a843 100644 --- a/tools/terminal_tool_background.py +++ b/tools/terminal_tool_background.py @@ -162,6 +162,7 @@ def spawn_background_process( effective_cwd = _resolve_command_cwd( workdir=workdir, default_cwd=cwd, session_key=session_key, env_type=env_type, mounted_host=mounted_host if mounted_host is not None else getattr(env, "host_cwd", None), + env=env, ) try: proc_session = _spawn( diff --git a/tools/terminal_tool_config.py b/tools/terminal_tool_config.py index 1505b8bc4f..d69aab1f2f 100644 --- a/tools/terminal_tool_config.py +++ b/tools/terminal_tool_config.py @@ -62,6 +62,59 @@ _WINDOWS_DRIVE_RE = re.compile(r"^[A-Za-z]:[\\/]") def _is_host_cwd(path: str) -> bool: return path.startswith(_HOST_CWD_PREFIXES) or bool(_WINDOWS_DRIVE_RE.match(path)) + +def _is_windows_drive_path(path: str) -> bool: + """True for any ``D:\\...`` / ``e:/...`` path. Not a username check.""" + return bool(path) and bool(_WINDOWS_DRIVE_RE.match(path)) + + +def _host_path_key(path: str) -> str: + """Compare host paths without caring about slash style or drive-letter case. + + A trailing slash is not significant, except we never collapse a drive root + (``C:/``) into a bare ``C:`` that would prefix-match every path on that drive. + """ + text = (path or "").replace("\\", "/") + if len(text) >= 2 and text[1] == ":": + text = text[0].lower() + text[1:] + if len(text) > 3: + text = text.rstrip("/") + return text + + +def translate_mounted_host_path(path: str, host_root: str, container_root: str) -> str | None: + """Map *path* onto *container_root* when it is *host_root* or a child of it. + + Returns None when *path* is not under that host directory. Slash style and + drive-letter case do not matter; a sibling directory (``proj`` vs ``proj-other``) + is not a child. + """ + if not path or not host_root or not container_root: + return None + key = _host_path_key(path) + root = _host_path_key(host_root) + if not key or not root: + return None + mount = container_root.rstrip("/") or "/" + if key == root: + return mount + prefix = root if root.endswith("/") else root + "/" + if not key.startswith(prefix): + return None + return f"{mount}/{key[len(prefix):]}" + + +def cwd_follows_host_mount(cwd: str, mount: str) -> bool: + """True when *cwd* was the host workspace (or the assumed ``/workspace`` view of it). + + An explicit in-container path other than that assumption is left alone. + """ + if not mount or not cwd or cwd == mount: + return False + if _is_unusable_container_cwd(cwd): + return True + return cwd == "/workspace" and mount != "/workspace" + _CONTAINER_BACKENDS = frozenset({"docker", "singularity", "modal", "daytona", "vercel_sandbox"}) _BUILTIN_BACKENDS = _CONTAINER_BACKENDS | {"local", "ssh", "managed_modal"} diff --git a/tools/terminal_tool_guards.py b/tools/terminal_tool_guards.py index 18db120f4d..9d8195d951 100644 --- a/tools/terminal_tool_guards.py +++ b/tools/terminal_tool_guards.py @@ -229,6 +229,7 @@ def gateway_lifecycle_block( guard_cwd = _resolve_command_cwd( workdir=workdir, default_cwd=guard_cwd_base, session_key=session_key, env_type=env_type, mounted_host=getattr(env, "host_cwd", None), + env=env, ) unsafe, refusal = scan_gateway_lifecycle( command,