From b3e0896356d723e70f07a464d811f37cebbd6b39 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Tue, 16 Jun 2026 18:58:31 +0000 Subject: [PATCH] fix(workspace): harden remote POSIX workspace trust boundaries (#3664) (#4273) Normalize both the candidate and terminal.cwd before the remote-POSIX containment check so `{cwd}/../other/...` lexical escapes are rejected, and restore the blocked-system-root gate BELOW the home-directory allow-list in both resolve_trusted_workspace() and validate_workspace_to_add() so valid systemd-homed paths under /var/home/... stay allowed. Also normalize POSIX shapes at the source (_as_posix_path) so every downstream block probe is shape-consistent (/etc/../home/user -> /home/user). Co-authored-by: rodboev --- api/workspace.py | 169 +++++++++++++++++--- tests/test_remote_terminal_workspace.py | 24 +++ tests/test_workspace_blocked_roots_macos.py | 14 +- 3 files changed, 180 insertions(+), 27 deletions(-) diff --git a/api/workspace.py b/api/workspace.py index a1c9b44e..4b7117c9 100644 --- a/api/workspace.py +++ b/api/workspace.py @@ -11,11 +11,12 @@ import hashlib import json import logging import os +import posixpath import shutil import stat import subprocess import concurrent.futures -from pathlib import Path +from pathlib import Path, PurePosixPath logger = logging.getLogger(__name__) @@ -58,6 +59,65 @@ def _last_workspace_file() -> Path: return _profile_state_dir() / 'last_workspace.txt' +def _expanduser_path(path: str | Path) -> Path: + """Return *path* after shell-style home expansion. + + ``Path.expanduser()`` on Windows does not consistently honor a monkeypatched + ``HOME`` in tests, which makes host-native replay diverge from the repo's + portability expectations. Use ``os.path.expanduser`` as the expansion source, + then wrap it back into ``Path``. + """ + raw = str(path) + if raw.startswith('~'): + home = ( + os.environ.get('HOME') + or os.environ.get('USERPROFILE') + or ( + (os.environ.get('HOMEDRIVE') or '') + (os.environ.get('HOMEPATH') or '') + ) + or str(Path.home()) + ) + if raw in ('~', '~/', '~\\'): + return Path(home) + if raw.startswith('~/') or raw.startswith('~\\'): + return Path(home) / raw[2:] + return Path(raw) + + +def _resolve_path(path: str | Path) -> Path: + """Resolve *path* after env-aware home expansion, without raising.""" + return _safe_resolve(_expanduser_path(path)) + + +def _home_path() -> Path: + """Return the current effective home directory with env-aware expansion.""" + return _resolve_path("~") + + +def _as_posix_path(path: str | Path | None) -> PurePosixPath | None: + if path in (None, ""): + return None + raw = _strip_surrounding_quotes(str(path)).strip().replace('\\', '/') + if not raw.startswith('/'): + return None + return PurePosixPath(posixpath.normpath(raw)) + + +def _posix_is_within(path: PurePosixPath, root: PurePosixPath) -> bool: + try: + path.relative_to(root) + return True + except ValueError: + return False + + +def _normalize_posix_path(path: str | Path | None) -> str | None: + candidate = _as_posix_path(path) + if candidate is None: + return None + return candidate.as_posix() + + def _is_remote_terminal_backend(terminal_cfg: dict | None) -> bool: """Return True when the active terminal backend runs outside this WebUI host.""" if not isinstance(terminal_cfg, dict): @@ -91,8 +151,18 @@ def _remote_terminal_workspace_candidate(path: str | Path) -> Path | None: raw = _strip_surrounding_quotes(str(path)).strip() if not raw: return None - candidate = Path(raw).expanduser().resolve() - base = Path(cwd).expanduser().resolve() + normalized_raw = _normalize_posix_path(raw) + normalized_cwd = _normalize_posix_path(cwd) + if normalized_raw is not None and normalized_cwd is not None: + posix_candidate = PurePosixPath(normalized_raw) + posix_base = PurePosixPath(normalized_cwd) + if _is_blocked_workspace_path(Path(normalized_raw), normalized_raw) or _is_blocked_workspace_path(Path(normalized_cwd), normalized_cwd): + return None + if posix_candidate == posix_base or _posix_is_within(posix_candidate, posix_base): + return _resolve_path(normalized_raw) + return None + candidate = _resolve_path(raw) + base = _resolve_path(cwd) if _is_blocked_workspace_path(candidate, raw) or _is_blocked_workspace_path(base, cwd): return None if candidate == base or _is_within(candidate, base): @@ -124,14 +194,18 @@ def _profile_default_workspace() -> str: for key in ('workspace', 'default_workspace'): ws = cfg.get(key) if ws: - p = Path(str(ws)).expanduser().resolve() + if remote_terminal: + return str(ws).strip() + p = _resolve_path(str(ws)) if remote_terminal or p.is_dir(): return str(p) # Fall through to terminal.cwd — the agent's configured working directory if isinstance(terminal_cfg, dict): cwd = terminal_cfg.get('cwd', '') if cwd and str(cwd) not in ('.', ''): - p = Path(str(cwd)).expanduser().resolve() + if remote_terminal: + return str(cwd).strip() + p = _resolve_path(str(cwd)) if remote_terminal or p.is_dir(): return str(p) except (ImportError, Exception): @@ -139,9 +213,9 @@ def _profile_default_workspace() -> str: try: from api.config import DEFAULT_WORKSPACE as _LIVE_DEFAULT_WORKSPACE - return str(Path(_LIVE_DEFAULT_WORKSPACE).expanduser().resolve()) + return str(_resolve_path(_LIVE_DEFAULT_WORKSPACE)) except Exception: - return str(Path(_BOOT_DEFAULT_WORKSPACE).expanduser().resolve()) + return str(_resolve_path(_BOOT_DEFAULT_WORKSPACE)) # ── Public API ────────────────────────────────────────────────────────────── @@ -156,14 +230,14 @@ def _clean_workspace_list(workspaces: list) -> list: confusion with the 'default' profile name). Returns the cleaned list (may be empty). """ - hermes_profiles = (Path.home() / '.hermes' / 'profiles').resolve() + hermes_profiles = (_home_path() / '.hermes' / 'profiles').resolve() result = [] for w in workspaces: path = w.get('path', '') name = w.get('name', '') if not path: continue - p = _safe_resolve(Path(path).expanduser()) + p = _safe_resolve(_expanduser_path(path)) # Skip paths inside a DIFFERENT profile's directory (cross-profile leak). # Allow paths inside the CURRENT profile's own directory (e.g. test workspaces # created under ~/.hermes/profiles/webui/webui-mvp-test/). @@ -382,6 +456,46 @@ def _workspace_blocked_roots() -> tuple[Path, ...]: return tuple(_out) +def _is_blocked_posix_workspace_path(raw_path: str | Path | None) -> bool: + """Detect blocked POSIX-style system roots even on non-POSIX hosts.""" + candidate = _as_posix_path(raw_path) + if candidate is None: + return False + if candidate == PurePosixPath('/'): + return True + carveouts = ( + PurePosixPath('/var/folders'), + PurePosixPath('/private/var/folders'), + PurePosixPath('/var/tmp'), + PurePosixPath('/private/var/tmp'), + ) + for tmp in carveouts: + if _posix_is_within(candidate, tmp): + return False + blocked_roots = ( + PurePosixPath('/etc'), + PurePosixPath('/usr'), + PurePosixPath('/var'), + PurePosixPath('/bin'), + PurePosixPath('/sbin'), + PurePosixPath('/boot'), + PurePosixPath('/proc'), + PurePosixPath('/sys'), + PurePosixPath('/dev'), + PurePosixPath('/lib'), + PurePosixPath('/lib64'), + PurePosixPath('/opt/homebrew'), + PurePosixPath('/System'), + PurePosixPath('/Library'), + PurePosixPath('/private/etc'), + PurePosixPath('/private/var'), + ) + for blocked in blocked_roots: + if _posix_is_within(candidate, blocked): + return True + return False + + def _is_blocked_system_path(candidate: Path) -> bool: """Return True if *candidate* falls under a blocked system root. @@ -435,10 +549,15 @@ def _is_blocked_workspace_path(candidate: Path, raw_path: str | Path | None = No raw = None if raw_path not in (None, ""): try: - raw = Path(raw_path).expanduser() + normalized_posix = _normalize_posix_path(raw_path) + raw = Path(normalized_posix) if normalized_posix is not None else _expanduser_path(raw_path) except Exception: raw = None + posix_probe = raw_path if raw_path not in (None, "") else candidate.as_posix() + if _is_blocked_posix_workspace_path(posix_probe): + return True + exact = _workspace_blocked_exact_roots() if candidate in exact or (raw is not None and raw in _workspace_blocked_roots()): return True @@ -487,7 +606,7 @@ def _trusted_workspace_roots() -> list[Path]: if candidate in (None, ""): return try: - p = Path(candidate).expanduser().resolve() + p = _resolve_path(candidate) except Exception: return if not p.exists() or not p.is_dir(): @@ -497,7 +616,7 @@ def _trusted_workspace_roots() -> list[Path]: if p not in roots: roots.append(p) - add(Path.home()) + add(_home_path()) add(_BOOT_DEFAULT_WORKSPACE) for w in load_workspaces(): add(w.get("path")) @@ -525,14 +644,14 @@ def list_workspace_suggestions(prefix: str = "", limit: int = 12) -> list[str]: return [str(p) for p in roots[:limit]] if raw.startswith("~"): - target = Path(raw).expanduser() + target = _expanduser_path(raw) elif Path(raw).is_absolute(): target = Path(raw) else: - target = Path.home() / raw + target = _home_path() / raw try: - match_target = target.expanduser().resolve() + match_target = _resolve_path(target) except Exception: match_target = target @@ -542,7 +661,7 @@ def list_workspace_suggestions(prefix: str = "", limit: int = 12) -> list[str]: home_root: Path | None = None if preserve_tilde: try: - home_root = Path.home().expanduser().resolve() + home_root = _home_path() except Exception: home_root = None suggestions: list[str] = [] @@ -584,7 +703,7 @@ def list_workspace_suggestions(prefix: str = "", limit: int = 12) -> list[str]: show_hidden = leaf.startswith('.') try: - parent_resolved = parent.expanduser().resolve() + parent_resolved = _resolve_path(parent) except Exception: return suggestions[:limit] @@ -634,9 +753,9 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path: trusted (it was validated at server startup). """ if path in (None, ""): - return Path(_BOOT_DEFAULT_WORKSPACE).expanduser().resolve() + return _resolve_path(_BOOT_DEFAULT_WORKSPACE) - candidate = Path(path).expanduser().resolve() + candidate = _resolve_path(path) access_error = _workspace_access_error(candidate) remote_candidate = _remote_terminal_workspace_candidate(path) @@ -653,7 +772,7 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path: # (A) Trusted if under the user's home directory — cross-platform via Path.home() # Must be checked before system roots to allow symlinks like /var/home. - _home = Path.home().resolve() + _home = _home_path() if _home != Path("/"): try: candidate.relative_to(_home) @@ -661,14 +780,13 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path: except ValueError: pass - # Block known system roots and their children. if _is_blocked_workspace_path(candidate, path): raise ValueError(f"Path points to a system directory: {candidate}") # (B) Trusted if already in the saved workspace list — covers non-home installs try: saved = load_workspaces() - saved_paths = {Path(w["path"]).resolve() for w in saved if w.get("path")} + saved_paths = {_resolve_path(w["path"]) for w in saved if w.get("path")} if candidate in saved_paths: return candidate except Exception: @@ -680,7 +798,7 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path: # was already validated at server startup, so any sub-path of it is safe # without requiring the user to add it to the workspace list manually. try: - boot_default = Path(_BOOT_DEFAULT_WORKSPACE).expanduser().resolve() + boot_default = _resolve_path(_BOOT_DEFAULT_WORKSPACE) candidate.relative_to(boot_default) return candidate except ValueError: @@ -729,7 +847,7 @@ def validate_workspace_to_add(path: str) -> Path: and users routinely paste those into the Add Space input. """ path = _strip_surrounding_quotes(path) - candidate = Path(path).expanduser().resolve() + candidate = _resolve_path(path) access_error = _workspace_access_error(candidate) remote_candidate = _remote_terminal_workspace_candidate(path) @@ -745,11 +863,10 @@ def validate_workspace_to_add(path: str) -> Path: # Home directory is always trusted regardless of where it lives on disk # (e.g. /var/home/... on systemd-homed Fedora/RHEL). - _home = Path.home().resolve() + _home = _home_path() if _home != Path("/") and _is_within(candidate, _home): return candidate - # Block known system roots and their immediate children. if _is_blocked_workspace_path(candidate, path): raise ValueError(f"Path points to a system directory: {candidate}") diff --git a/tests/test_remote_terminal_workspace.py b/tests/test_remote_terminal_workspace.py index ecd0ff44..d3d99202 100644 --- a/tests/test_remote_terminal_workspace.py +++ b/tests/test_remote_terminal_workspace.py @@ -57,6 +57,18 @@ def test_remote_terminal_workspace_paths_outside_cwd_still_reject(monkeypatch): workspace.resolve_trusted_workspace("/Users/other/projects/demo") +def test_remote_terminal_workspace_paths_with_parent_escape_still_reject(monkeypatch): + monkeypatch.setattr(api_config, "get_config", lambda: _remote_config()) + + escaped = f"{REMOTE_CWD}/../other/projects/demo" + + with pytest.raises(ValueError, match="Path does not exist"): + workspace.validate_workspace_to_add(escaped) + + with pytest.raises(ValueError, match="Path does not exist"): + workspace.resolve_trusted_workspace(escaped) + + @pytest.mark.parametrize("workspace_path", ["/etc", "/etc/ssh"]) def test_remote_terminal_workspace_system_roots_still_reject(monkeypatch, workspace_path): monkeypatch.setattr(api_config, "get_config", lambda: _remote_config(terminal={"backend": "ssh", "cwd": "/etc"})) @@ -66,3 +78,15 @@ def test_remote_terminal_workspace_system_roots_still_reject(monkeypatch, worksp with pytest.raises(ValueError, match="Path points to a system directory"): workspace.resolve_trusted_workspace(workspace_path) + + +@pytest.mark.parametrize("validator", [workspace.resolve_trusted_workspace, workspace.validate_workspace_to_add]) +def test_var_home_workspaces_stay_allowed_before_system_root_blocklist(monkeypatch, validator): + home = Path("/var/home/joeyshiue") + candidate = home / "projects/demo" + + monkeypatch.setattr(workspace, "_resolve_path", lambda raw: candidate if str(raw) == str(candidate) else Path(raw)) + monkeypatch.setattr(workspace, "_home_path", lambda: home) + monkeypatch.setattr(workspace, "_workspace_access_error", lambda _candidate: None) + + assert validator(str(candidate)) == candidate diff --git a/tests/test_workspace_blocked_roots_macos.py b/tests/test_workspace_blocked_roots_macos.py index ae12fc9d..dfa7726d 100644 --- a/tests/test_workspace_blocked_roots_macos.py +++ b/tests/test_workspace_blocked_roots_macos.py @@ -124,7 +124,7 @@ class TestUserTmpPrefixes: """Carve-outs must not let /etc or other strict roots through.""" for tmp in _USER_TMP_PREFIXES: # tmp paths are under /var or /private/var, never under /etc, /usr, /bin, etc. - assert str(tmp).startswith('/var/') or str(tmp).startswith('/private/var/') + assert tmp.as_posix().startswith('/var/') or tmp.as_posix().startswith('/private/var/') # ── Other roots: literal == resolved on both platforms ───────────────────── @@ -173,3 +173,15 @@ class TestMacOSSystemAndLibraryBlocked: """Paths under /Library and /System must be blocked regardless of platform.""" from api.workspace import _is_blocked_workspace_path assert _is_blocked_workspace_path(Path(path)) + + +class TestPosixShapeNormalization: + def test_etc_parent_escape_is_not_falsely_blocked(self): + from api.workspace import _is_blocked_workspace_path + + assert not _is_blocked_workspace_path(Path('/home/user'), '/etc/../home/user') + + def test_dot_prefixed_etc_path_is_still_blocked(self): + from api.workspace import _is_blocked_workspace_path + + assert _is_blocked_workspace_path(Path('/etc/ssh'), '/./etc/ssh')