""" Hermes Web UI -- Workspace and file system helpers. Workspace lists and last-used workspace are stored per-profile so each profile has its own workspace configuration. State files live at ``{profile_home}/webui_state/workspaces.json`` and ``{profile_home}/webui_state/last_workspace.txt``. The global STATE_DIR paths are used as fallback when no profile module is available. """ import hashlib import json import logging import os import posixpath import shutil import stat import subprocess import concurrent.futures from pathlib import Path, PurePosixPath logger = logging.getLogger(__name__) from api.config import ( WORKSPACES_FILE as _GLOBAL_WS_FILE, LAST_WORKSPACE_FILE as _GLOBAL_LW_FILE, DEFAULT_WORKSPACE as _BOOT_DEFAULT_WORKSPACE, MAX_FILE_BYTES, IMAGE_EXTS, MD_EXTS ) # ── Profile-aware path resolution ─────────────────────────────────────────── def _profile_state_dir() -> Path: """Return the webui_state directory for the active profile. For the default profile, returns the global STATE_DIR (respects HERMES_WEBUI_STATE_DIR env var for test isolation). For named profiles, returns {profile_home}/webui_state/. """ try: from api.profiles import get_active_profile_name, get_active_hermes_home name = get_active_profile_name() if name and name != 'default': d = get_active_hermes_home() / 'webui_state' d.mkdir(parents=True, exist_ok=True) return d except ImportError: logger.debug("Failed to import profiles module, using global state dir") return _GLOBAL_WS_FILE.parent def _workspaces_file() -> Path: """Return the workspaces.json path for the active profile.""" return _profile_state_dir() / 'workspaces.json' def _last_workspace_file() -> Path: """Return the last_workspace.txt path for the active profile.""" 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): return False backend = str(terminal_cfg.get('backend') or '').strip().lower() return backend not in ('', 'local') def _remote_terminal_cwd() -> str | None: """Return target-side terminal cwd for remote profiles, without local stat().""" try: from api.config import get_config terminal_cfg = get_config().get('terminal', {}) if not _is_remote_terminal_backend(terminal_cfg): return None cwd = str(terminal_cfg.get('cwd') or '').strip() if not cwd or cwd == '.': return None return cwd except Exception: logger.debug("Failed to read remote terminal cwd", exc_info=True) return None def _remote_terminal_workspace_candidate(path: str | Path) -> Path | None: """Return a non-stat'ed target-side Path when it is under terminal.cwd.""" cwd = _remote_terminal_cwd() if not cwd: return None raw = _strip_surrounding_quotes(str(path)).strip() if not raw: return None if '\x00' in raw or '\x00' in cwd: return None 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): return candidate return None def _profile_default_workspace() -> str: """Read the profile's default workspace from its config.yaml. Checks keys in priority order: 1. 'workspace' — explicit webui workspace key 2. 'default_workspace' — alternate explicit key 3. 'terminal.cwd' — hermes-agent terminal working dir (most common) For remote/SSH terminal profiles, ``terminal.cwd`` lives on the target machine, not on the WebUI server. In that case return it without a server-local existence check so WebUI can send the correct workspace hint to the agent/tool backend. Falls back to the live DEFAULT_WORKSPACE from api.config. """ try: from api.config import get_config cfg = get_config() terminal_cfg = cfg.get('terminal', {}) remote_terminal = _is_remote_terminal_backend(terminal_cfg) # Explicit webui workspace keys first for key in ('workspace', 'default_workspace'): ws = cfg.get(key) if ws: 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 ('.', ''): 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): logger.debug("Failed to load profile default workspace config") try: from api.config import DEFAULT_WORKSPACE as _LIVE_DEFAULT_WORKSPACE return str(_resolve_path(_LIVE_DEFAULT_WORKSPACE)) except Exception: return str(_resolve_path(_BOOT_DEFAULT_WORKSPACE)) # ── Public API ────────────────────────────────────────────────────────────── def _clean_workspace_list(workspaces: list) -> list: """Sanitize a workspace list: - Preserve saved paths even when they are currently missing or inaccessible; picker state must not be destroyed by a transient stat/permission failure. - Remove entries whose paths live inside another profile's directory (e.g. ~/.hermes/profiles/X/... should not appear on a different profile). - Rename any entry whose name is literally 'default' to 'Home' (avoids confusion with the 'default' profile name). Returns the cleaned list (may be empty). """ 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(_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/). try: p.relative_to(hermes_profiles) # p is under ~/.hermes/profiles/ — only skip if it's under a DIFFERENT profile try: from api.profiles import get_active_hermes_home own_profile_dir = get_active_hermes_home().resolve() p.relative_to(own_profile_dir) # p is under our own profile dir — keep it except (ValueError, Exception): continue # under profiles/ but not our own — cross-profile leak, skip except ValueError: pass # not under profiles/ at all — keep it # Rename confusing 'default' label to 'Home' if name.lower() == 'default': name = 'Home' result.append({'path': str(p), 'name': name}) return result def _workspace_access_error(candidate: Path, *, missing_label: str = "Path does not exist") -> str | None: """Return a user-facing validation error for an unusable workspace path. ``Path.exists()`` can collapse permission/stat failures into a generic falsey result on some Python/OS combinations, which produced misleading "does not exist" messages for macOS/TCC-denied directories. Probe with ``stat()`` so missing paths, non-directories, and permission-denied paths can be reported separately. """ try: 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 " f"this directory ({exc}). On macOS, grant Full Disk Access or Files and " f"Folders permission to the Hermes/WebUI app or server process, then try again." ) except OSError as exc: return f"Cannot access path: {candidate}. The server process could not inspect this path ({exc})." if not stat.S_ISDIR(st.st_mode): return f"Path is not a directory: {candidate}" return None def _migrate_global_workspaces() -> list: """Read the legacy global workspaces.json, clean it, and return the result. This is the migration path for users upgrading from a pre-profile version: their global file may contain cross-profile entries, test artifacts, and stale paths accumulated over time. We clean it in-place and rewrite it. """ if not _GLOBAL_WS_FILE.exists(): return [] try: raw = json.loads(_GLOBAL_WS_FILE.read_text(encoding='utf-8')) cleaned = _clean_workspace_list(raw) if len(cleaned) != len(raw): # Rewrite the cleaned version so future reads are already clean _GLOBAL_WS_FILE.write_text( json.dumps(cleaned, ensure_ascii=False, indent=2), encoding='utf-8' ) return cleaned except Exception: return [] def load_workspaces() -> list: ws_file = _workspaces_file() if ws_file.exists(): try: raw = json.loads(ws_file.read_text(encoding='utf-8')) cleaned = _clean_workspace_list(raw) if len(cleaned) != len(raw): # Persist the cleaned version so stale entries don't keep reappearing try: ws_file.write_text( json.dumps(cleaned, ensure_ascii=False, indent=2), encoding='utf-8' ) except Exception: logger.debug("Failed to persist cleaned workspace list") return cleaned or [{'path': _profile_default_workspace(), 'name': 'Home'}] except Exception: logger.debug("Failed to load workspaces from %s", ws_file) # No profile-local file yet. # For the DEFAULT profile: migrate from the legacy global file (one-time cleanup). # For NAMED profiles: always start clean with just their own workspace. try: from api.profiles import get_active_profile_name is_default = get_active_profile_name() in ('default', None) except ImportError: is_default = True if is_default: migrated = _migrate_global_workspaces() if migrated: return migrated # Fresh start: single entry from the profile's configured workspace, labeled "Home" return [{'path': _profile_default_workspace(), 'name': 'Home'}] def save_workspaces(workspaces: list) -> None: ws_file = _workspaces_file() ws_file.parent.mkdir(parents=True, exist_ok=True) ws_file.write_text(json.dumps(workspaces, ensure_ascii=False, indent=2), encoding='utf-8') def get_last_workspace() -> str: remote_cwd = _remote_terminal_cwd() def valid_last_workspace(raw: str) -> str | None: if not raw: return None if remote_cwd: # For remote/SSH profiles, last_workspace is target-side state. Do # not accept stale server-local paths merely because they exist on # the WebUI host; require the value to stay under terminal.cwd. if _remote_terminal_workspace_candidate(raw) is not None: return raw return None if Path(raw).is_dir(): return raw return None lw_file = _last_workspace_file() if lw_file.exists(): try: p = valid_last_workspace(lw_file.read_text(encoding='utf-8').strip()) if p: return p except Exception: logger.debug("Failed to read last workspace from %s", lw_file) # Fallback: try global file if _GLOBAL_LW_FILE.exists(): try: p = valid_last_workspace(_GLOBAL_LW_FILE.read_text(encoding='utf-8').strip()) if p: return p except Exception: logger.debug("Failed to read global last workspace") return _profile_default_workspace() def set_last_workspace(path: str) -> None: try: lw_file = _last_workspace_file() lw_file.parent.mkdir(parents=True, exist_ok=True) lw_file.write_text(str(path), encoding='utf-8') except Exception: logger.debug("Failed to set last workspace") 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, 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 # Per-user temp directories that sit nominally under a "system" prefix but are # actually user-writable scratch space. Workspaces registered here (e.g. by # pytest's ``tmp_path_factory`` on macOS, which uses ``/var/folders//T/``) # must remain accepted even though their parent (``/var``) is blocked. These # carve-outs apply to BOTH workspace registration and runtime file ops so a # symlink target inside the carve-out is also reachable. _USER_TMP_PREFIXES: tuple[Path, ...] = ( Path('/var/folders'), # macOS per-user tmp (literal form) Path('/private/var/folders'), # macOS per-user tmp (resolved form) Path('/var/tmp'), # Linux/macOS system-wide tmp (user-writable) Path('/private/var/tmp'), # macOS resolved form ) def _workspace_blocked_roots() -> tuple[Path, ...]: """System roots that must never be accepted as workspace candidates. Returns both the literal path and its symlink-resolved canonical form, deduped. This matters on macOS where ``/etc``, ``/var``, and ``/tmp`` are symlinks to ``/private/etc`` etc. Without the resolved forms, callers that pass a ``.resolve()``-d candidate (every caller does) would compare ``/private/etc`` against literal ``Path('/etc')`` and the ``relative_to`` check would miss — letting ``/etc`` through as a registered workspace on macOS. Carve-outs for legitimate user-tmp paths nominally under these roots (e.g. ``/var/folders/.../T/`` on macOS) are handled by :func:`_is_blocked_system_path`, not by exclusion from this list. """ _raw = ( # Linux / macOS '/etc', '/usr', '/var', '/bin', '/sbin', '/boot', '/proc', '/sys', '/dev', '/lib', '/lib64', '/opt/homebrew', '/System', '/Library', ) _seen: set[Path] = set() _out: list[Path] = [] for _p in _raw: for _form in (Path(_p), _safe_resolve(Path(_p))): if _form not in _seen: _seen.add(_form) _out.append(_form) 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. Honours :data:`_USER_TMP_PREFIXES` carve-outs so per-user tmp directories nominally under ``/var`` (``/var/folders`` on macOS, ``/var/tmp`` on Linux/macOS) remain valid workspace candidates and reachable file targets. """ for tmp in _USER_TMP_PREFIXES: if _is_within(candidate, tmp): return False for blocked in _workspace_blocked_roots(): if _is_within(candidate, blocked): return True return False def _workspace_blocked_resolved_subtrees() -> tuple[Path, ...]: roots = list(_workspace_blocked_roots()) + [Path('/private/etc')] resolved: list[Path] = [] for root in roots: try: p = root.expanduser().resolve() except Exception: p = root if p not in resolved: resolved.append(p) return tuple(resolved) def _workspace_blocked_exact_roots() -> tuple[Path, ...]: roots = [Path('/'), Path('/private/var')] for root in _workspace_blocked_roots(): try: roots.append(root.expanduser().resolve()) except Exception: roots.append(root) unique: list[Path] = [] for root in roots: if root not in unique: unique.append(root) return tuple(unique) def _is_blocked_workspace_path(candidate: Path, raw_path: str | Path | None = None) -> bool: """Return True when candidate points at a known OS/system directory. Compare both the original spelling and the resolved path. This closes the macOS /etc -> /private/etc bypass without globally banning temporary pytest paths under /private/var/folders. """ raw = None if raw_path not in (None, ""): try: 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 for tmp in _USER_TMP_PREFIXES: if _is_within(candidate, tmp) or (raw is not None and _is_within(raw, tmp)): return False # Raw paths under literal roots (e.g. /etc/ssh, /var/db) are always blocked. if raw is not None: for blocked in _workspace_blocked_roots(): if _is_within(raw, blocked): return True # Resolved subtree checks catch symlink aliases such as /private/etc. The # macOS temp root /private/var/folders is intentionally allowed for pytest # and per-user temporary workspaces; other direct /private/var system data # such as /private/var/db and /private/var/log remains blocked. allowed_private_var = (Path('/private/var/folders'), Path('/private/var/tmp')) for blocked in _workspace_blocked_resolved_subtrees(): if blocked == Path('/private/var'): if candidate == blocked: return True if any(_is_within(candidate, allowed) for allowed in allowed_private_var): continue if _is_within(candidate, blocked): return True continue if _is_within(candidate, blocked): return True return False def _is_within(path: Path, root: Path) -> bool: try: path.relative_to(root) return True except ValueError: return False def _trusted_workspace_roots() -> list[Path]: roots: list[Path] = [] def add(candidate: str | Path | None) -> None: if candidate in (None, ""): return try: p = _resolve_path(candidate) except Exception: return if not p.exists() or not p.is_dir(): return if _is_blocked_workspace_path(p, candidate): return if p not in roots: roots.append(p) add(_home_path()) add(_BOOT_DEFAULT_WORKSPACE) for w in load_workspaces(): add(w.get("path")) roots.sort(key=lambda p: len(str(p))) return roots def list_workspace_suggestions(prefix: str = "", limit: int = 12) -> list[str]: """Return workspace path suggestions under trusted roots only. Suggestions are limited to directories under one of: - Path.home() - the boot default workspace - already-saved workspace roots Arbitrary system prefixes return an empty list rather than an error so the UI can safely autocomplete while the user types. """ roots = _trusted_workspace_roots() if not roots: return [] raw = (prefix or "").strip() if not raw: return [str(p) for p in roots[:limit]] if raw.startswith("~"): target = _expanduser_path(raw) elif Path(raw).is_absolute(): target = Path(raw) else: target = _home_path() / raw try: match_target = _resolve_path(target) except Exception: match_target = target normalized = str(match_target) normalized_lower = normalized.lower() preserve_tilde = raw.startswith("~") home_root: Path | None = None if preserve_tilde: try: home_root = _home_path() except Exception: home_root = None suggestions: list[str] = [] def format_suggestion(path: Path) -> str: if preserve_tilde and home_root is not None: try: rel = path.resolve().relative_to(home_root) if str(rel) == ".": return "~" return "~/" + rel.as_posix() except (OSError, ValueError): pass return str(path) def add(path: Path) -> None: value = format_suggestion(path) if value not in suggestions: suggestions.append(value) # If the user is typing a partial trusted root like /Users/xuef..., suggest # the matching trusted roots without scanning arbitrary system parents. for root in roots: if str(root).lower().startswith(normalized_lower): add(root) in_root = [ root for root in roots if normalized == str(root) or normalized.startswith(str(root) + os.sep) ] if not in_root: return suggestions[:limit] anchor_root = max(in_root, key=lambda p: len(str(p))) ends_with_sep = raw.endswith(os.sep) or raw.endswith('/') parent = target if ends_with_sep else target.parent leaf = '' if ends_with_sep else target.name show_hidden = leaf.startswith('.') try: parent_resolved = _resolve_path(parent) except Exception: return suggestions[:limit] if not parent_resolved.exists() or not parent_resolved.is_dir(): return suggestions[:limit] if not _is_within(parent_resolved, anchor_root): return suggestions[:limit] leaf_lower = leaf.lower() try: children = sorted(parent_resolved.iterdir(), key=lambda p: p.name.lower()) except OSError: return suggestions[:limit] for child in children: if not child.is_dir(): continue if child.name.startswith('.') and not show_hidden: continue if leaf_lower and not child.name.lower().startswith(leaf_lower): continue add(child.resolve()) if len(suggestions) >= limit: break return suggestions[:limit] def resolve_trusted_workspace(path: str | Path | None = None) -> Path: """Resolve and validate a workspace path. A path is trusted if it satisfies at least one of: (A) It is under the user's home directory (Path.home()). Works cross-platform: ~/... on Linux/macOS, C:\\Users\\... on Windows. (B) It is already in the profile's saved workspace list. This covers self-hosted deployments where workspaces live outside home (e.g. /data/projects, /opt/workspace) — once a workspace is saved by an admin, it can be reused without re-validation. Additionally enforced regardless of (A)/(B): 1. The path must exist. 2. The path must be a directory. 3. The path must not be a known system root (/etc, /usr, /var, /bin, /sbin, /boot, /proc, /sys, /dev, /root on Linux/macOS; Windows system dirs). This prevents even admin-saved workspaces from pointing at OS internals. None/empty path falls back to the boot-time DEFAULT_WORKSPACE, which is always trusted (it was validated at server startup). """ if path in (None, ""): return _resolve_path(_BOOT_DEFAULT_WORKSPACE) candidate = _resolve_path(path) access_error = _workspace_access_error(candidate) remote_candidate = _remote_terminal_workspace_candidate(path) if access_error: # For remote terminal profiles, workspace paths belong to the target # machine. Allow paths under terminal.cwd so session switching can # update the workspace hint even though this WebUI host cannot stat # the target-side path. if remote_candidate is None: raise ValueError(access_error) if remote_candidate is not None: return remote_candidate # (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 = _home_path() if _home != Path("/"): try: candidate.relative_to(_home) return candidate except ValueError: pass 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 = {_resolve_path(w["path"]) for w in saved if w.get("path")} if candidate in saved_paths: return candidate except Exception: pass # (C) Trusted if it is equal to or under the boot-time DEFAULT_WORKSPACE. # In Docker deployments HERMES_WEBUI_DEFAULT_WORKSPACE is often set to a # volume mount outside the user's home (e.g. /data/workspace). That 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 = _resolve_path(_BOOT_DEFAULT_WORKSPACE) candidate.relative_to(boot_default) return candidate except ValueError: pass raise ValueError( f"Path is outside the user home directory, not in the saved workspace " f"list, and not under the default workspace: {candidate}. " f"Add it via Settings → Workspaces first." ) def _strip_surrounding_quotes(path: str) -> str: """Strip a single pair of surrounding single or double quotes from a path string. macOS Finder's "Copy as Pathname" (Cmd+Option+C) returns paths wrapped in single quotes, e.g. ``'/Users/x/Documents/foo'``. Other shells and OS file managers do similar things with double quotes. Users routinely paste these quoted strings into the Add Space input expecting them to "just work" — the only reason they didn't was a missing strip. Only paired quotes are stripped (matching opener and closer). One-sided quotes are preserved on the slim chance a path legitimately contains a literal quote character. """ s = path.strip() if len(s) >= 2 and s[0] == s[-1] and s[0] in ("'", '"'): return s[1:-1] return s def validate_workspace_to_add(path: str) -> Path: """Validate a path for *adding* to the workspace list (less restrictive than resolve_trusted_workspace). When a user explicitly adds a new workspace path, we trust their intent — they have console or filesystem access to that path and are consciously registering it. We only block: non-existent paths, non-directories, and known system roots. The stricter ``resolve_trusted_workspace`` is used when *using* an existing workspace (file reads/writes) to prevent path traversal after the list is built. Surrounding quotes (single or double) are stripped before validation — macOS Finder's "Copy as Pathname" wraps paths in single quotes by default, and users routinely paste those into the Add Space input. """ path = _strip_surrounding_quotes(path) candidate = _resolve_path(path) access_error = _workspace_access_error(candidate) remote_candidate = _remote_terminal_workspace_candidate(path) if access_error: # Remote terminal profiles validate workspace existence on the target # machine, not on the WebUI server. Permit target-side paths under # terminal.cwd. if remote_candidate is None: raise ValueError(access_error) if remote_candidate is not None: return remote_candidate # Home directory is always trusted regardless of where it lives on disk # (e.g. /var/home/... on systemd-homed Fedora/RHEL). _home = _home_path() if _home != Path("/") and _is_within(candidate, _home): return candidate if _is_blocked_workspace_path(candidate, path): raise ValueError(f"Path points to a system directory: {candidate}") return candidate def safe_resolve_ws(root: Path, requested: str) -> Path: """Resolve a relative path inside a workspace root, raising ValueError on traversal. Both raw ``..`` traversal and symlink escapes are blocked. Workspace file APIs can be reached by browser UI actions and agent/tool calls, so a symlink inside the workspace must not expand the trusted workspace boundary to an arbitrary host path. """ root_resolved = root.resolve() resolved = (root / requested).resolve() try: resolved.relative_to(root_resolved) except ValueError: raise ValueError(f"Path traversal blocked: {requested}") return resolved # ── Race-safe (TOCTOU) anchored open ───────────────────────────────────────── # safe_resolve_ws() validates a path, but if callers then re-open by pathname a # symlink swapped in AFTER the check could still escape the workspace. To close # that window we open the (already symlink-resolved) target component-by-component # from the workspace root using openat (dir_fd) + O_NOFOLLOW: every component must # be a real, non-symlink entry, so a component swapped to a symlink mid-flight is # refused. Legit in-workspace symlinks still work because safe_resolve_ws() has # already collapsed them to their real in-workspace target, and we walk that real # (symlink-free) path. Portable: uses os.supports_dir_fd where available (Linux, # macOS); on platforms without dir_fd support (Windows — where creating symlinks # also requires admin) we fall back to a plain pathname open, matching the prior # behaviour with no regression. _DIR_FD_OK = os.open in getattr(os, "supports_dir_fd", set()) _O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) _O_DIRECTORY = getattr(os, "O_DIRECTORY", 0) def open_anchored_fd(workspace: Path, target: Path, *, want_dir: bool) -> int: """Open ``target`` race-safely and return an owned file descriptor. ``target`` must be the symlink-resolved path returned by safe_resolve_ws() (i.e. already verified to live under the workspace). Raises FileNotFoundError if a component is missing / wrong-type, or ValueError if a component was swapped to a symlink (escape attempt). Caller owns and must close the fd. """ root_resolved = workspace.resolve() # Relative, symlink-free component list (resolve() already collapsed any links). try: rel_parts = target.relative_to(root_resolved).parts except ValueError: raise ValueError(f"Path traversal blocked: {target}") from None if not _DIR_FD_OK: # Windows / no openat: fall back to a plain pathname open. No new race # protection, but no regression vs the prior path-based behaviour, and # symlink creation needs admin on Windows anyway. flags = os.O_RDONLY | (_O_DIRECTORY if want_dir else 0) | _O_NOFOLLOW try: return os.open(str(target), flags) except OSError: raise FileNotFoundError(f"Not found: {target}") from None # Open the (trusted) workspace root. root_resolved is canonical (resolve() # collapsed any symlinks to REACH it, e.g. macOS /tmp -> /private/tmp), so its # final component is legitimately a real directory — O_NOFOLLOW here only fires # if the root itself was raced into a symlink after resolve() (escape attempt). fd = os.open(str(root_resolved), os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW) try: for i, part in enumerate(rel_parts): is_last = i == len(rel_parts) - 1 want_directory = (not is_last) or want_dir flags = os.O_RDONLY | _O_NOFOLLOW | (_O_DIRECTORY if want_directory else 0) try: nfd = os.open(part, flags, dir_fd=fd) except OSError: # ELOOP (component is a symlink — swapped in) or missing/wrong type. raise FileNotFoundError(f"Not found: {target}") from None os.close(fd) fd = nfd return fd except BaseException: try: os.close(fd) except OSError: pass raise def open_anchored_create_fd(root: Path, dest: Path) -> int: """Create ``dest`` for exclusive writing race-safely, anchored under ``root``. Walks from ``root`` via openat + O_NOFOLLOW (creating missing intermediate directories with mkdir(dir_fd=...)), then creates the leaf with O_CREAT|O_EXCL|O_NOFOLLOW so a symlink raced into any component cannot redirect the write outside ``root``. ``dest`` must be the resolved path and must not already exist (callers dedup first). Raises ValueError if ``dest`` is not under ``root``, FileExistsError if it exists, FileNotFoundError if a component was swapped to a symlink. Caller owns and must close the returned write fd. On platforms without dir_fd support (Windows) falls back to a plain exclusive create — no new race protection but no regression. """ root_resolved = root.resolve() try: rel_parts = dest.relative_to(root_resolved).parts except ValueError: raise ValueError(f"Path traversal blocked: {dest}") from None if not rel_parts: raise ValueError(f"Invalid destination: {dest}") if not _DIR_FD_OK: # Windows / no openat: create parent dirs then exclusively create the leaf. dest.parent.mkdir(parents=True, exist_ok=True) return os.open(str(dest), os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW, 0o644) fd = os.open(str(root_resolved), os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW) try: for part in rel_parts[:-1]: try: nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd) except FileNotFoundError: os.mkdir(part, 0o755, dir_fd=fd) nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd) except OSError: # ELOOP — component swapped to a symlink (escape attempt). raise FileNotFoundError(f"Not found: {dest}") from None os.close(fd) fd = nfd return os.open( rel_parts[-1], os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW, 0o644, dir_fd=fd, ) finally: try: os.close(fd) except OSError: pass def make_anchored_dir(root: Path, dest: Path) -> None: """Create directory ``dest`` (and any missing parents) race-safely under ``root``. Walks from ``root`` via openat + O_NOFOLLOW, creating each missing component with mkdir(dir_fd=...), so a symlink raced into any component cannot make the server create directories outside ``root``. Idempotent (existing dirs are fine). Raises ValueError if ``dest`` is not under ``root``, FileNotFoundError if a component was swapped to a symlink. On platforms without dir_fd support (Windows) falls back to a plain Path.mkdir — no regression. """ root_resolved = root.resolve() dest_resolved = dest.resolve() if dest_resolved == root_resolved: return try: rel_parts = dest_resolved.relative_to(root_resolved).parts except ValueError: raise ValueError(f"Path traversal blocked: {dest}") from None if not _DIR_FD_OK: dest.mkdir(parents=True, exist_ok=True) return fd = os.open(str(root_resolved), os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW) try: for part in rel_parts: try: nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd) except FileNotFoundError: os.mkdir(part, 0o755, dir_fd=fd) nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd) except OSError: # ELOOP — component swapped to a symlink (escape attempt). raise FileNotFoundError(f"Not found: {dest}") from None os.close(fd) fd = nfd finally: try: os.close(fd) except OSError: pass def open_anchored_write_fd(root: Path, target: Path) -> int: """Open existing ``target`` for truncating writes anchored under ``root``.""" root_resolved = root.resolve() target_resolved = target.resolve() try: rel_parts = target_resolved.relative_to(root_resolved).parts except ValueError: raise ValueError(f"Path traversal blocked: {target}") from None if not rel_parts: raise ValueError(f"Invalid target: {target}") flags = os.O_WRONLY | os.O_TRUNC | _O_NOFOLLOW if not _DIR_FD_OK: return os.open(str(target_resolved), flags) parent_fd = open_anchored_fd(root_resolved, target_resolved.parent, want_dir=True) try: return os.open(rel_parts[-1], flags, dir_fd=parent_fd) finally: os.close(parent_fd) def unlink_anchored(root: Path, target: Path) -> None: """Unlink an existing file anchored under ``root``.""" root_resolved = root.resolve() target_resolved = target.resolve() try: rel_parts = target_resolved.relative_to(root_resolved).parts except ValueError: raise ValueError(f"Path traversal blocked: {target}") from None if not rel_parts: raise ValueError(f"Invalid target: {target}") if not _DIR_FD_OK: target_resolved.unlink() return parent_fd = open_anchored_fd(root_resolved, target_resolved.parent, want_dir=True) try: os.unlink(rel_parts[-1], dir_fd=parent_fd) finally: os.close(parent_fd) def rmtree_anchored(root: Path, target: Path) -> None: """Remove a directory tree anchored under ``root`` without following symlink swaps.""" root_resolved = root.resolve() target_resolved = target.resolve() try: rel_parts = target_resolved.relative_to(root_resolved).parts except ValueError: raise ValueError(f"Path traversal blocked: {target}") from None if not rel_parts: raise ValueError(f"Invalid target: {target}") if not _DIR_FD_OK: shutil.rmtree(target_resolved) return parent_fd = open_anchored_fd(root_resolved, target_resolved.parent, want_dir=True) try: shutil.rmtree(rel_parts[-1], dir_fd=parent_fd) finally: os.close(parent_fd) def rename_anchored(root: Path, source: Path, dest: Path) -> None: """Rename ``source`` to ``dest`` using anchored parent directory fds.""" root_resolved = root.resolve() source_resolved = source.resolve() dest_parent_resolved = dest.parent.resolve() try: source_parts = source_resolved.relative_to(root_resolved).parts except ValueError: raise ValueError(f"Path traversal blocked: {source}") from None try: dest_parent_resolved.relative_to(root_resolved) except ValueError: raise ValueError(f"Path traversal blocked: {dest}") from None if not source_parts: raise ValueError(f"Invalid source: {source}") dest_leaf = dest.name if not dest_leaf: raise ValueError(f"Invalid destination: {dest}") if not _DIR_FD_OK: source_resolved.rename(dest) return src_parent_fd = open_anchored_fd(root_resolved, source_resolved.parent, want_dir=True) try: dst_parent_fd = open_anchored_fd(root_resolved, dest_parent_resolved, want_dir=True) try: try: os.stat(dest_leaf, dir_fd=dst_parent_fd, follow_symlinks=False) raise FileExistsError(dest_leaf) except FileNotFoundError: pass os.rename( source_parts[-1], dest_leaf, src_dir_fd=src_parent_fd, dst_dir_fd=dst_parent_fd, ) finally: os.close(dst_parent_fd) finally: os.close(src_parent_fd) def list_dir(workspace: Path, rel: str='.'): target = safe_resolve_ws(workspace, rel) if not target.is_dir(): raise FileNotFoundError(f"Not a directory: {rel}") ws_resolved = workspace.resolve() target_resolved = target.resolve() entries = [] def _process(name, is_symlink, raw_link, lstat_result, reachable): """Append one directory entry. ``raw_link`` is the os.readlink() result for symlinks (else None); ``lstat_result`` is an os.stat_result obtained with follow_symlinks=False (else None); ``reachable`` is False when a follow_symlinks=True stat raised (broken target or symlink loop).""" if is_symlink: if raw_link is None: return # A symlink whose follow-stat raised (ELOOP / broken target) can never # be opened — filter it. This catches mutual/self loops portably across # Python versions where Path.resolve() loop handling differs (3.11 # raises RuntimeError, 3.13 can return a path), so do not rely on # resolve() raising for cycle detection. if not reachable: return try: link_target = (target_resolved / raw_link).resolve() except (OSError, RuntimeError): return # Cycle detection: skip if symlink points back to current dir or root. if link_target == target_resolved or link_target == ws_resolved: return try: target_resolved.relative_to(link_target) return # target is under link_target — ancestor → cycle except ValueError: pass # Tag symlinks whose resolved target escapes the workspace root. # Previously silently dropped; now emitted with target_outside_workspace=True # so the workspace tree can show the link exists (display-only — the # read/list gate in safe_resolve_ws / open_anchored_fd still blocks # navigation through it). target_outside_workspace = False try: link_target.relative_to(ws_resolved) except ValueError: target_outside_workspace = True if _is_blocked_system_path(link_target): return display_path = name if rel and rel != '.': display_path = rel + '/' + display_path mtime_ns = lstat_result.st_mtime_ns if lstat_result is not None else None if target_outside_workspace: # #4581 hardening: a display-only escape-target symlink must NOT # disclose where it points. Emit ONLY display-safe fields — never # the resolved outside path, target-derived is_dir, or target size # (the row exists to show the link is present; navigation/read # through it stays blocked by safe_resolve_ws/open_anchored_fd). entry = { 'name': name, 'path': display_path, 'type': 'symlink', 'is_dir': False, 'target_outside_workspace': True, 'mtime_ns': mtime_ns, } entries.append(entry) else: is_dir = link_target.is_dir() entry = { 'name': name, 'path': display_path, 'type': 'symlink', 'target': str(link_target), 'is_dir': is_dir, 'target_outside_workspace': False, 'mtime_ns': mtime_ns, } if not is_dir: try: entry['size'] = link_target.stat().st_size except OSError: entry['size'] = None entries.append(entry) else: entry_path = name if rel and rel != '.': entry_path = rel + '/' + name if lstat_result is not None: is_file = stat.S_ISREG(lstat_result.st_mode) size = lstat_result.st_size if is_file else None mtime_ns = lstat_result.st_mtime_ns is_dir_entry = stat.S_ISDIR(lstat_result.st_mode) else: size = None mtime_ns = None is_dir_entry = False entries.append({ 'name': name, 'path': entry_path, 'type': 'dir' if is_dir_entry else 'file', 'size': size, 'mtime_ns': mtime_ns, }) if _DIR_FD_OK: # #3398 TOCTOU hardening (Linux/macOS): open the directory via an anchored # openat-walk (O_NOFOLLOW on every component) and enumerate via the verified # fd (os.scandir(fd) + fd-relative fstatat/readlinkat), so a path component # swapped to an escaping symlink after safe_resolve_ws() cannot redirect the # listing. def _sort_key_de(de): try: is_link = de.is_symlink() except OSError: is_link = False is_file = False if not is_link: try: is_file = de.is_file() except OSError: pass return (not is_link, is_file, de.name.lower()) dir_fd = open_anchored_fd(workspace, target, want_dir=True) try: st = os.fstat(dir_fd) if not stat.S_ISDIR(st.st_mode): raise FileNotFoundError(f"Not a directory: {rel}") with os.scandir(dir_fd) as scan: scandir_entries = sorted(scan, key=_sort_key_de) for de in scandir_entries: name = de.name is_symlink = de.is_symlink() raw_link = None if is_symlink: try: raw_link = os.readlink(name, dir_fd=dir_fd) except OSError: raw_link = None try: lst = os.stat(name, dir_fd=dir_fd, follow_symlinks=False) except OSError: lst = None # reachable: follow-stat succeeds (filters ELOOP/broken symlinks). reachable = True if is_symlink: try: os.stat(name, dir_fd=dir_fd, follow_symlinks=True) except OSError: reachable = False _process(name, is_symlink, raw_link, lst, reachable) if len(entries) >= 200: break finally: try: os.close(dir_fd) except OSError: pass else: # Portability fallback (Windows / no dir_fd): path-based enumeration after # safe_resolve_ws(). No anchored-fd race protection on these platforms, but # no regression vs the prior behaviour (creating symlinks on Windows needs # admin anyway), and safe_resolve_ws() still blocks the static escape. def _sort_key_p(p: Path): is_link = p.is_symlink() is_file = False if not is_link: try: is_file = p.is_file() except OSError: pass return (not is_link, is_file, p.name.lower()) for item in sorted(target.iterdir(), key=_sort_key_p): name = item.name is_symlink = item.is_symlink() raw_link = None if is_symlink: try: raw_link = os.readlink(str(item)) except OSError: raw_link = None try: lst = item.lstat() except OSError: lst = None # reachable: follow-stat succeeds (filters ELOOP/broken symlinks). reachable = True if is_symlink: try: os.stat(str(item), follow_symlinks=True) except OSError: reachable = False _process(name, is_symlink, raw_link, lst, reachable) if len(entries) >= 200: break return entries def dir_signature(workspace: Path, rel: str = '.', entries: list[dict] | None = None) -> str: """Return a cheap, stable signature for a listed workspace directory. The signature is based only on bounded directory-entry metadata already used by the workspace tree: names, displayed paths, entry type, file sizes, mtimes, and symlink targets. It intentionally does not read file contents. """ if entries is None: entries = list_dir(workspace, rel) payload = [] for entry in entries: payload.append({ 'name': entry.get('name'), 'path': entry.get('path'), 'type': entry.get('type'), 'is_dir': entry.get('is_dir'), 'size': entry.get('size'), 'mtime_ns': entry.get('mtime_ns'), 'target': entry.get('target'), 'target_outside_workspace': entry.get('target_outside_workspace'), }) raw = json.dumps(payload, sort_keys=True, separators=(',', ':'), ensure_ascii=False) return hashlib.sha256(raw.encode('utf-8')).hexdigest() def read_file_content(workspace: Path, rel: str) -> dict: target = safe_resolve_ws(workspace, rel) if not target.is_file(): raise FileNotFoundError(f"Not a file: {rel}") # #3398 TOCTOU hardening: open the resolved file via an anchored openat-walk # (O_NOFOLLOW on every component) so a path swapped to an escaping symlink # after safe_resolve_ws() cannot be followed, then read from the fd (not the # pathname) so the bytes returned are guaranteed to be the verified file. fd = open_anchored_fd(workspace, target, want_dir=False) with os.fdopen(fd, 'rb', closefd=True) as fh: st = os.fstat(fh.fileno()) if not stat.S_ISREG(st.st_mode): raise FileNotFoundError(f"Not a file: {rel}") if st.st_size > MAX_FILE_BYTES: raise ValueError(f"File too large ({st.st_size} bytes, max {MAX_FILE_BYTES})") raw = fh.read(MAX_FILE_BYTES + 1) content = raw.decode('utf-8', errors='replace') return {'path': rel, 'content': content, 'size': len(raw), 'lines': content.count('\n') + 1} # ── Git detection ────────────────────────────────────────────────────────── def _run_git(args, cwd, timeout=3): """Run a git command and return stdout, or None on failure.""" try: r = subprocess.run( ['git'] + args, cwd=str(cwd), capture_output=True, text=True, timeout=timeout, ) return r.stdout.strip() if r.returncode == 0 else None except (subprocess.TimeoutExpired, FileNotFoundError, OSError): return None def git_info_for_workspace(workspace: Path) -> dict: """Return git info for a workspace directory, or None if not a git repo.""" if not (workspace / '.git').exists(): return None branch = _run_git(['rev-parse', '--abbrev-ref', 'HEAD'], workspace) if branch is None: return None # Run the remaining git commands in parallel via threads — they are # independent subprocess calls and together can take 50-200ms when run # serially. Threading is safe here because each call blocks only on the # subprocess pipe, not on the GIL. def _ahead(): r = _run_git(['rev-list', '--count', '@{u}..HEAD'], workspace) return int(r) if r and r.isdigit() else 0 def _behind(): r = _run_git(['rev-list', '--count', 'HEAD..@{u}'], workspace) return int(r) if r and r.isdigit() else 0 def _status(): out = _run_git(['status', '--porcelain'], workspace) or '' lines = [l for l in out.splitlines() if l] modified = sum(1 for l in lines if len(l) >= 2 and (l[0] in 'MAR' or l[1] in 'MAR')) untracked = sum(1 for l in lines if l.startswith('??')) return len(lines), modified, untracked with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool: f_status = pool.submit(_status) f_ahead = pool.submit(_ahead) f_behind = pool.submit(_behind) dirty, modified, untracked = f_status.result() ahead = f_ahead.result() behind = f_behind.result() return { 'branch': branch, 'dirty': dirty, 'modified': modified, 'untracked': untracked, 'ahead': ahead, 'behind': behind, 'is_git': True, }