"""Hermes Web UI -- Workspace and file system helpers. Workspace lists and last-used workspace are stored per-profile so eachprofile 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_DIRpaths are used as fallback when no profile module is available."""import hashlibimport jsonimport loggingimport osimport posixpathimport secretsimport shutilimport statimport subprocessimport concurrent.futuresimport threadingimport timefrom pathlib import Path, PurePosixPath logger = logging.getLogger(__name__) _ESCAPE_AUTH_TTL_SECONDS = 300_ESCAPE_AUTH_LOCK = threading.Lock()_ESCAPE_AUTH_TOKENS: dict[str, dict[str, str | int | float]] = {} 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_profile_default_workspace() -> str:    """Resolve the ACTIVE PROFILE's default workspace, never the global file.     Like get_last_workspace() but WITHOUT the global ``_GLOBAL_LW_FILE``    fallback: for a named profile that has not yet written its own    profile-scoped ``last_workspace.txt``, that global fallback would leak the    *global* last-workspace instead of the profile's configured workspace —    which is exactly the #5169 bug (the composer chip on a blank new-chat page    showing the wrong/global workspace for a named profile). Used by    ``GET /api/profile/active`` so a cold boot under a profile cookie reflects    the profile's own configured working directory.     Priority: profile-scoped ``last_workspace.txt`` -> profile ``config.yaml``    ``workspace``/``default_workspace`` -> ``terminal.cwd`` -> process default.    """    remote_cwd = _remote_terminal_cwd()     def _valid(raw: str) -> str | None:        if not raw:            return None        if remote_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(lw_file.read_text(encoding='utf-8').strip())            if p:                return p        except Exception:            logger.debug("Failed to read profile last workspace from %s", lw_file)    return _profile_default_workspace()  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/<hash>/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 = (

Lines 1–500 of 1730. Load more