diff --git a/CHANGELOG.md b/CHANGELOG.md index 1431b07f..06e64811 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,12 @@ ## [Unreleased] +## [v0.51.463] — 2026-06-16 — Release PX (tighter remote workspace boundaries) + +### Fixed + +- **Remote POSIX workspace trust boundaries are hardened (#3664).** The remote-terminal workspace containment check now normalizes both the candidate path and `terminal.cwd` before comparing, so a lexical `..` escape (e.g. `{cwd}/../other/...`) no longer slips past the containment check. The blocked-system-root gate was also restored below the home-directory allow-list in both workspace validators, so legitimate systemd-homed paths under `/var/home//...` stay allowed while system directories remain blocked. POSIX path shapes are normalized at the source so every downstream block probe is consistent (`/etc/../home/user` → `/home/user`, `/./etc/ssh` → `/etc/ssh`). Workspace paths containing an embedded null byte now fail closed with a clean validation error instead of surfacing a 500 (at both the workspace validators and the `/api/workspaces/add` route preflight), and the add-workspace route now applies the same home-directory carve-out as the validators so systemd-homed workspaces under `/var/home//...` register correctly. Thanks @rodboev. + ## [v0.51.462] — 2026-06-16 — Release PW (sidebar survives slow loads) ### Fixed diff --git a/api/routes.py b/api/routes.py index dcfd9db4..05185107 100644 --- a/api/routes.py +++ b/api/routes.py @@ -4829,6 +4829,8 @@ from api.workspace import ( make_anchored_dir, validate_workspace_to_add, _is_blocked_system_path, + _home_path, + _is_within, _strip_surrounding_quotes, _is_remote_terminal_backend, _workspace_blocked_roots, @@ -15522,9 +15524,23 @@ def _handle_workspace_add(handler, body): # _is_blocked_system_path honours user-tmp carve-outs (e.g. /var/folders on # macOS) so pytest's tmp_path_factory paths and other legit user-tmp dirs # still register cleanly. - candidate = Path(path_str).expanduser().resolve() + try: + candidate = Path(path_str).expanduser().resolve() + except (ValueError, OSError, RuntimeError) as e: + # Invalid path (e.g. embedded null byte) — fail closed with a clean 400 + # instead of letting .resolve() raise an uncaught 500. + return bad(handler, f"Invalid path: {_sanitize_error(e)}") if _is_blocked_system_path(candidate): - return bad(handler, f"Path points to a system directory: {candidate}") + # Home-directory carve-out, mirroring the validators + # (resolve_trusted_workspace / validate_workspace_to_add): a workspace + # at or under the active user's home must stay allowed even when that + # home lives under an otherwise-blocked root (e.g. systemd-homed + # /var/home//...). Without this the route rejects valid + # /var/home workspaces before validate_workspace_to_add()'s carve-out + # can run. + _home = _home_path() + if not (_home != Path("/") and (candidate == _home or _is_within(candidate, _home))): + return bad(handler, f"Path points to a system directory: {candidate}") # Now safe to create the directory if requested if auto_create: try: diff --git a/api/workspace.py b/api/workspace.py index a1c9b44e..74d1c53c 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,77 @@ 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:] + # NOTE: ``~user`` / ``~root`` forms are intentionally NOT expanded here. + # Master deliberately does not block ``/root`` (#510/#521 — Hermes commonly + # runs as root, where ``/root`` is the legitimate home and is allowed via + # the home carve-out). Expanding ``~root`` -> ``/root`` for a NON-root + # deployment would let it register root's home; leaving the literal form + # (resolved relative to cwd, then rejected if it escapes) is the safer + # behavior and matches the current security model. + 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('\\', '/') + # Reject embedded null bytes here rather than letting them survive normpath + # and crash later at .resolve() with an uncaught ValueError (surfaces as a + # 500). Fail-closed: treat as an invalid path. + if '\x00' in raw: + return None + 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 +163,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 +206,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 +225,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 +242,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/). @@ -199,6 +285,11 @@ def _workspace_access_error(candidate: Path, *, missing_label: str = "Path does st = candidate.stat() except FileNotFoundError: return f"{missing_label}: {candidate}" + except ValueError as exc: + # Embedded null byte (or similar invalid path) — .stat() raises ValueError, + # not OSError. Report as an access error rather than letting it surface as + # an uncaught 500. + return f"Cannot access path: {candidate!r}. Invalid path ({exc})." except PermissionError as exc: return ( f"Cannot access path: {candidate}. The server process could not inspect " @@ -322,7 +413,10 @@ def _safe_resolve(p: Path) -> Path: """Path.resolve() that never raises — falls back to the input path on error.""" try: return p.resolve() - except (OSError, RuntimeError): + except (OSError, RuntimeError, ValueError): + # ValueError covers embedded-null-byte paths, which .resolve() raises on + # — fail-closed to the raw path so the downstream block-list gate rejects + # it cleanly instead of surfacing a 500. return p @@ -382,6 +476,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 +569,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 +626,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 +636,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 +664,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 +681,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 +723,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 +773,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 +792,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 +800,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 +818,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 +867,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 +883,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_4273_expanduser_and_nullbyte.py b/tests/test_4273_expanduser_and_nullbyte.py new file mode 100644 index 00000000..389b2c1e --- /dev/null +++ b/tests/test_4273_expanduser_and_nullbyte.py @@ -0,0 +1,56 @@ +"""Regression tests for the #4273 ship-gate follow-up fixes (null-byte handling). + +Scope note: an earlier gate round suggested restoring ~user/~root expansion and +blocking /root. Both were REVERTED after the full suite surfaced +tests/test_batch_fixes.py::TestRootWorkspaceUnblocked — /root is DELIBERATELY +not blocked (#510/#521: Hermes commonly runs as root, where /root is the +legitimate home, allowed via the home carve-out). Expanding ~root for a non-root +deployment would then register root's home, so the PR's original behavior +(leave ~user literal) is the correct, safer choice. Only the null-byte +fail-closed hardening is kept here — it has no contradiction. +""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import api.workspace as ws # noqa: E402 + + +def test_as_posix_path_rejects_null_byte(): + assert ws._as_posix_path("/home/u/p/\x00x") is None + + +def test_safe_resolve_fail_closed_on_null_byte(): + # Must not raise; falls back to the raw path so the block gate can reject it. + out = ws._safe_resolve(Path("/home/u/\x00x")) + assert isinstance(out, Path) + + +def test_validate_workspace_to_add_null_byte_is_clean_validation_error(): + import pytest + with pytest.raises(ValueError) as ei: + ws.validate_workspace_to_add("/home/user/project/\x00x") + # Clean, fail-closed message — not the raw "embedded null byte" crash bubbling + # uncaught (the route catches ValueError and returns a 400). + assert "Invalid path" in str(ei.value) or "null" in str(ei.value).lower() + + +def test_root_remains_unblocked_for_root_deployments(): + # Guard the deliberate #510/#521 decision: /root must stay registrable + # (Hermes-as-root deployments). This pins that our null-byte work did not + # accidentally re-block it. + src = (Path(__file__).resolve().parent.parent / "api" / "workspace.py").read_text(encoding="utf-8") + assert "'/root'" not in src and "PurePosixPath('/root')" not in src, ( + "/root must not be blocked — breaks Hermes-as-root deployments (#510/#521)" + ) + + +def test_workspaces_add_route_has_home_carveout_before_block(): + # The /api/workspaces/add preflight must apply the home carve-out before + # rejecting a blocked-system-root path, so systemd-homed /var/home/ + # workspaces register (matching the validators). Pin the carve-out exists. + src = (Path(__file__).resolve().parent.parent / "api" / "routes.py").read_text(encoding="utf-8") + assert "_home_path()" in src and "_is_within(candidate, _home)" in src, ( + "route preflight must apply the home carve-out before _is_blocked_system_path rejection" + ) 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')