From b71c88f2923611bfc73bfce6dc0280fe2aade7eb Mon Sep 17 00:00:00 2001 From: alfred-rootson Date: Wed, 16 Sep 2026 00:37:11 +0200 Subject: [PATCH] fix(workspace): preserve remote POSIX paths across Session & streaming callsites (#7131) (#7168) Closes the remote POSIX workspace path preservation gap across Session and streaming. Profile-aware runtime path resolution. Co-authored-by: alfred-rootson --- api/config.py | 24 +- api/gateway_chat.py | 4 +- api/models.py | 19 +- api/profiles.py | 18 +- api/routes.py | 261 +++- api/streaming.py | 18 +- api/upload.py | 9 +- api/workspace.py | 364 ++++- docs/CONTRACTS.md | 4 + docs/remote-workspaces.md | 29 + tests/conftest.py | 38 + tests/test_agent_runtime_revision_guard.py | 2 +- tests/test_bootstrap_dotenv.py | 84 ++ tests/test_claude_code_session_import.py | 2 +- tests/test_early_session_title.py | 2 +- tests/test_goal_command_webui.py | 12 +- tests/test_issue1217_transcript_compaction.py | 2 +- .../test_issue3930_source_filter_pushdown.py | 4 +- tests/test_issue492_workspace_reorder.py | 3 +- ...test_issue6611_regeneration_transaction.py | 2 +- ...test_issue6751_api_content_agent_replay.py | 8 +- ...issue7168_round6_default_isolated_clamp.py | 294 ++++ tests/test_profile_switch_1200.py | 4 +- tests/test_provider_mismatch.py | 4 +- ...est_recovered_anchor_settle_duplication.py | 2 +- tests/test_remote_terminal_workspace.py | 1232 ++++++++++++++++- tests/test_sprint5.py | 8 +- tests/test_stale_stream_cleanup.py | 8 +- tests/test_webhook_project_sessions.py | 4 +- tests/test_workspace_upload.py | 2 +- 30 files changed, 2260 insertions(+), 207 deletions(-) create mode 100644 docs/remote-workspaces.md create mode 100644 tests/test_issue7168_round6_default_isolated_clamp.py diff --git a/api/config.py b/api/config.py index 686d2392..24530ef7 100644 --- a/api/config.py +++ b/api/config.py @@ -713,10 +713,14 @@ def get_config_for_profile_home(profile_home: "Path | str | None") -> dict: bypassing the thread-local resolver entirely. When ``profile_home`` matches the path the ambient resolver would pick (the common single-profile case), we return the cached ``get_config()`` to preserve in-memory overrides used - by tests and runtime callers. Only when the session's profile home diverges - from the ambient path do we read the session profile's file directly — a - pure read with no global cache mutation, so it is race-free across - concurrent sessions on different profiles. + by tests and runtime callers, and to honour an authoritative + ``HERMES_CONFIG_PATH`` override. Only when the session's profile home + diverges from the ambient path do we read the session profile's file + directly — a pure read with no global cache mutation, so it is race-free + across concurrent sessions on different profiles. Divergent profiles stay + isolated: a nonexistent home returns ``{}`` and an existing home without a + ``config.yaml`` yields defaults — neither ever falls back to the ambient + config (profiles-are-islands). """ if not profile_home: return get_config() @@ -724,10 +728,18 @@ def get_config_for_profile_home(profile_home: "Path | str | None") -> dict: target = Path(profile_home).expanduser() except Exception: return get_config() + + from api.workspace import _safe_resolve as _cfg_safe_resolve + + # Canonicalize BOTH sides before every identity comparison (#7168 re-gate + # round 5): when HERMES_HOME (or the config parent) is a symlink alias, + # lexical equality fails and an authoritative HERMES_CONFIG_PATH inside + # the aliased home would be bypassed in favor of a direct — wrong — read. + target = _cfg_safe_resolve(target) try: from api.profiles import get_active_hermes_home - if Path(get_active_hermes_home()).expanduser() == target: + if _cfg_safe_resolve(Path(get_active_hermes_home()).expanduser()) == target: return get_config() except Exception: pass @@ -737,7 +749,7 @@ def get_config_for_profile_home(profile_home: "Path | str | None") -> dict: # whose directory doesn't physically exist yet (fresh install, monkeypatched # cfg) must still resolve through get_config(), not return {} (#4516 gate). try: - if _get_config_path().parent == target: + if _cfg_safe_resolve(_get_config_path().parent) == target: return get_config() except Exception: pass diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 24b275ad..35ced6fd 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -584,7 +584,7 @@ def _run_gateway_runs_api_streaming( try: from api.streaming import _build_native_multimodal_message - message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=active_provider, active_model=(model or ""), requested_provider=active_provider) + message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=active_provider, active_model=(model or ""), requested_provider=active_provider, profile=getattr(session, "profile", None)) except Exception: logger.debug("Failed to build runs-API multimodal attachment payload", exc_info=True) message_content = str(msg_text or "") @@ -1120,7 +1120,7 @@ def _run_gateway_chat_streaming( try: from api.streaming import _build_native_multimodal_message - message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=(model_provider or ""), active_model=(model or ""), requested_provider=(model_provider or "")) + message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=(model_provider or ""), active_model=(model or ""), requested_provider=(model_provider or ""), profile=getattr(s, "profile", None)) except Exception: logger.debug("Failed to build gateway multimodal attachment payload", exc_info=True) message_content = str(msg_text or "") diff --git a/api/models.py b/api/models.py index 775124dc..754c372f 100644 --- a/api/models.py +++ b/api/models.py @@ -34,7 +34,7 @@ from api.config import ( LOCK, STREAMS, STREAMS_LOCK, DEFAULT_WORKSPACE, DEFAULT_MODEL, PROJECTS_FILE, HOME, get_effective_default_model, _get_session_agent_lock, ) -from api.workspace import get_last_workspace +from api.workspace import get_last_workspace, _resolve_path from api.usage import prompt_cache_hit_percent from api.agent_sessions import ( _is_continuation_session, @@ -1264,7 +1264,8 @@ class Session: **kwargs): self.session_id = session_id or uuid.uuid4().hex[:12] self.title = title - self.workspace = str(Path(workspace).expanduser().resolve()) + self.profile = profile + self.workspace = str(_resolve_path(workspace, profile=profile)) # #6672: immutable snapshot of the workspace at session creation time. # s.workspace is updated on every turn when the user switches workspaces # mid-session via the WebUI header dropdown; interpolating the live @@ -1276,7 +1277,7 @@ class Session: # without a persisted created_workspace fall back to the workspace # recorded on disk, which is the best available approximation. self.created_workspace = ( - str(Path(created_workspace).expanduser().resolve()) + str(_resolve_path(created_workspace, profile=profile)) if created_workspace else self.workspace ) @@ -5098,7 +5099,7 @@ def new_session(workspace=None, model=None, profile=None, model_provider=None, p wt = worktree_info if isinstance(worktree_info, dict) else None workspace_path = (wt.get('path') if wt and wt.get('path') else workspace) if wt else workspace s = Session( - workspace=workspace_path or get_last_workspace(), + workspace=workspace_path or get_last_workspace(profile=profile), model=effective_model, model_provider=effective_model_provider, profile=profile, @@ -5763,6 +5764,7 @@ def get_session_for_file_ops(sid: str): workspace, recovered = resolve_implicit_workspace_with_recovery( stored_workspace, get_last_workspace, + profile=session_profile or None, ) except ValueError: # Preserve the existing file-handler behavior for non-missing trust or @@ -6797,7 +6799,7 @@ def import_cli_session( s = Session( session_id=session_id, title=title, - workspace=get_last_workspace(), + workspace=get_last_workspace(profile=profile), model=model, messages=messages, profile=profile, @@ -7720,7 +7722,10 @@ def _load_cli_sessions_uncached( _cli_workspace_cache: list = [None] # list-as-cell; None = not yet resolved def _cli_workspace(): if _cli_workspace_cache[0] is None: - _cli_workspace_cache[0] = str(get_last_workspace()) + try: + _cli_workspace_cache[0] = str(get_last_workspace(profile=_cli_profile)) + except TypeError: + _cli_workspace_cache[0] = str(get_last_workspace()) return _cli_workspace_cache[0] _webhook_pid_cache: list[str | None] = [None] @@ -7934,7 +7939,7 @@ def _load_cli_sessions_uncached( cli_sessions.append({ 'session_id': sid, 'title': _display_title, - 'workspace': str(get_last_workspace()), + 'workspace': str(get_last_workspace(profile=_cli_profile)), 'model': row['model'] or None, 'message_count': row['message_count'] or row['actual_message_count'] or 0, 'created_at': row['started_at'], diff --git a/api/profiles.py b/api/profiles.py index f5e8432a..78f18735 100644 --- a/api/profiles.py +++ b/api/profiles.py @@ -1711,19 +1711,22 @@ def switch_profile(name: str, *, process_wide: bool = True) -> dict: default_workspace = None try: from api.config import DEFAULT_WORKSPACE as _DW + from api.workspace import _resolve_path, _remote_terminal_workspace_candidate lw_file = home / 'webui_state' / 'last_workspace.txt' if lw_file.exists(): _p = lw_file.read_text(encoding='utf-8').strip() if _p: - _pp = Path(_p).expanduser() - if _pp.is_dir(): - default_workspace = str(_pp.resolve()) + _pp = _resolve_path(_p, profile=name) + remote_cand = _remote_terminal_workspace_candidate(_p, profile=name) + if remote_cand is not None or _pp.is_dir(): + default_workspace = str(_pp) if default_workspace is None: for _key in ('workspace', 'default_workspace'): _v = cfg.get(_key) if _v: - _pp = Path(str(_v)).expanduser().resolve() - if _pp.is_dir(): + _pp = _resolve_path(str(_v), profile=name) + remote_cand = _remote_terminal_workspace_candidate(str(_v), profile=name) + if remote_cand is not None or _pp.is_dir(): default_workspace = str(_pp) break if default_workspace is None: @@ -1731,8 +1734,9 @@ def switch_profile(name: str, *, process_wide: bool = True) -> dict: if isinstance(_tc, dict): _cwd = _tc.get('cwd', '') if _cwd and str(_cwd) not in ('.', ''): - _pp = Path(str(_cwd)).expanduser().resolve() - if _pp.is_dir(): + _pp = _resolve_path(str(_cwd), profile=name) + remote_cand = _remote_terminal_workspace_candidate(str(_cwd), profile=name) + if remote_cand is not None or _pp.is_dir(): default_workspace = str(_pp) if default_workspace is None: default_workspace = str(_DW) diff --git a/api/routes.py b/api/routes.py index 4c220e29..701a51d7 100644 --- a/api/routes.py +++ b/api/routes.py @@ -10544,6 +10544,7 @@ from api.workspace import ( safe_resolve_ws, raw_authorized_escape_target, resolve_trusted_workspace, + _resolve_path, resolve_implicit_workspace_with_recovery, open_anchored_fd, open_anchored_create_fd, @@ -14160,22 +14161,39 @@ def handle_get(handler, parsed) -> bool: return _handle_session_export(handler, parsed) if parsed.path == "/api/workspaces": + from api.profiles import get_active_profile_name + active_profile = get_active_profile_name() + try: + wss = load_workspaces(profile=active_profile) + except TypeError: + wss = load_workspaces() + try: + lw = get_last_workspace(profile=active_profile) + except TypeError: + lw = get_last_workspace() return j( handler, { - "workspaces": load_workspaces(), - "last": get_last_workspace(), + "workspaces": wss, + "last": lw, "terminal_remote_backend": _terminal_remote_backend_enabled(), }, ) if parsed.path == "/api/workspaces/suggest": + from api.profiles import get_active_profile_name + qs = parse_qs(parsed.query) prefix = qs.get("prefix", [""])[0] + active_profile = get_active_profile_name() + try: + suggestions = list_workspace_suggestions(prefix, profile=active_profile) + except TypeError: + suggestions = list_workspace_suggestions(prefix) return j( handler, { - "suggestions": list_workspace_suggestions(prefix), + "suggestions": suggestions, "prefix": prefix, }, ) @@ -14626,7 +14644,10 @@ def handle_get(handler, parsed) -> bool: # profile-scoped via the per-request hermes_profile cookie set in server.py. # Fail open: a resolution error must never 500 this boot-critical endpoint. try: - _profile_default_workspace = get_profile_default_workspace() + try: + _profile_default_workspace = get_profile_default_workspace(profile=active_profile_name) + except TypeError: + _profile_default_workspace = get_profile_default_workspace() except Exception: logger.debug("Failed to resolve profile default workspace for /api/profile/active", exc_info=True) _profile_default_workspace = None @@ -14849,26 +14870,41 @@ def _validate_session_toolsets_shape(toolsets): return toolsets -def _resolve_new_session_workspace(body, visible_prev_session_id): +def _resolve_new_session_workspace(body, visible_prev_session_id, profile=None): """Resolve a new-session workspace, recovering only verified inheritance.""" candidate = body.get("workspace") if not candidate: return None + + def _rtw(value): + # Legacy test doubles may predate the profile kwarg. + try: + return resolve_trusted_workspace(value, profile=profile) + except TypeError: + return resolve_trusted_workspace(value) + if ( body.get("workspace_inherited_from_prev_session") is not True or not visible_prev_session_id ): - return str(resolve_trusted_workspace(candidate)) + return str(_rtw(candidate)) try: previous_session = get_session(visible_prev_session_id, metadata_only=True) except KeyError: - return str(resolve_trusted_workspace(candidate)) + return str(_rtw(candidate)) if str(getattr(previous_session, "workspace", None) or "") != str(candidate): - return str(resolve_trusted_workspace(candidate)) - workspace, _recovered = resolve_implicit_workspace_with_recovery( - candidate, - get_last_workspace, - ) + return str(_rtw(candidate)) + try: + workspace, _recovered = resolve_implicit_workspace_with_recovery( + candidate, + get_last_workspace, + profile=profile, + ) + except TypeError: + workspace, _recovered = resolve_implicit_workspace_with_recovery( + candidate, + get_last_workspace, + ) return str(workspace) def handle_post(handler, parsed) -> bool: @@ -15210,7 +15246,9 @@ def handle_post(handler, parsed) -> bool: ): workspace_prev_session_id = None try: - workspace = _resolve_new_session_workspace(body, workspace_prev_session_id) + workspace = _resolve_new_session_workspace( + body, workspace_prev_session_id, profile=body.get("profile") or None + ) except (TypeError, ValueError) as e: return bad(handler, str(e)) worktree_info = None @@ -15239,7 +15277,17 @@ def handle_post(handler, parsed) -> bool: from api.worktrees import create_worktree_for_workspace base_workspace = workspace if not base_workspace: - base_workspace = str(resolve_trusted_workspace(get_last_workspace())) + _new_profile = body.get("profile") or None + try: + _lw = get_last_workspace(profile=_new_profile) + except TypeError: + _lw = get_last_workspace() + try: + base_workspace = str( + resolve_trusted_workspace(_lw, profile=_new_profile) + ) + except TypeError: + base_workspace = str(resolve_trusted_workspace(_lw)) worktree_info = create_worktree_for_workspace(base_workspace) workspace = worktree_info["path"] except (TypeError, ValueError) as e: @@ -15804,7 +15852,7 @@ def handle_post(handler, parsed) -> bool: old_model = getattr(s, "model", None) old_provider = getattr(s, "model_provider", None) try: - new_ws = str(resolve_trusted_workspace(body.get("workspace", s.workspace))) + new_ws = str(resolve_trusted_workspace(body.get("workspace", s.workspace), profile=getattr(s, "profile", None))) except ValueError as e: return bad(handler, str(e)) with _get_session_agent_lock(body["session_id"]): @@ -15838,7 +15886,7 @@ def handle_post(handler, parsed) -> bool: close_terminal(body["session_id"]) except Exception: logger.debug("Failed to close workspace terminal after workspace update") - set_last_workspace(new_ws) + set_last_workspace(new_ws, profile=getattr(s, "profile", None)) return j( handler, {"session": public_session_projection(s.compact() | {"messages": s.messages})}, @@ -17026,14 +17074,16 @@ def handle_post(handler, parsed) -> bool: if _arch_source_tag == "subagent" or _is_subagent_child_session_id(sid): return bad(handler, "Subagent sessions cannot be archived from WebUI", 400) if _is_messaging_session_record(cli_meta): + _arch_profile = cli_meta.get("profile") or None s = Session( session_id=sid, title=cli_meta.get("title") or title_from(get_cli_session_messages(sid), "CLI Session"), - workspace=get_last_workspace(), + workspace=get_last_workspace(profile=_arch_profile), messages=[], model=cli_meta.get("model") or "unknown", created_at=cli_meta.get("created_at"), updated_at=cli_meta.get("updated_at"), + profile=_arch_profile, ) s.is_cli_session = is_cli_session_row(cli_meta) s.source_tag = cli_meta.get("source_tag") @@ -18019,15 +18069,26 @@ def _handle_list_dir(handler, parsed): except Exception: return bad(handler, "Session not found", 404) try: + _list_profile = getattr(webui_session, "profile", None) if webui_session is None: - workspace = resolve_trusted_workspace(workspace) + try: + workspace = resolve_trusted_workspace(workspace, profile=_list_profile) + except TypeError: + workspace = resolve_trusted_workspace(workspace) recovered = False else: stored_workspace = workspace - workspace, recovered = resolve_implicit_workspace_with_recovery( - stored_workspace, - get_last_workspace, - ) + try: + workspace, recovered = resolve_implicit_workspace_with_recovery( + stored_workspace, + get_last_workspace, + profile=_list_profile, + ) + except TypeError: + workspace, recovered = resolve_implicit_workspace_with_recovery( + stored_workspace, + get_last_workspace, + ) if recovered: persisted = persist_recovered_workspace_binding( webui_session, @@ -19069,7 +19130,7 @@ def _handle_terminal_start(handler, body): }, status=400, ) - workspace = resolve_trusted_workspace(getattr(session, "workspace", "") or "") + workspace = resolve_trusted_workspace(getattr(session, "workspace", "") or "", profile=getattr(session, "profile", None)) from api.terminal import start_terminal term = start_terminal( sid, @@ -22053,10 +22114,11 @@ def _memory_project_context_workspace(parsed) -> Path | None: # fall through to Path("").resolve(), which returns the server's own # CWD and would surface the install's AGENTS.md/HERMES.md as if it # were the user's project context. - ws = (get_session(sid).workspace or "").strip() + session = get_session(sid) + ws = (session.workspace or "").strip() if not ws: return None - return Path(ws).expanduser().resolve() + return _resolve_path(ws, profile=getattr(session, "profile", None)) except Exception: return None @@ -22064,7 +22126,7 @@ def _memory_project_context_workspace(parsed) -> Path | None: if not raw_workspace: return None try: - return Path(resolve_trusted_workspace(raw_workspace)).expanduser().resolve() + return resolve_trusted_workspace(raw_workspace) except Exception: logger.debug("Skipping project context for untrusted workspace %s", raw_workspace, exc_info=True) return None @@ -22834,7 +22896,7 @@ def _start_regeneration_stream_locked( save_attempted = True s.save() accepted = True - set_last_workspace(workspace) + set_last_workspace(workspace, profile=getattr(s, "profile", None)) release_worker.set() except Exception as exc: abort_worker.set() @@ -23173,7 +23235,7 @@ def _start_chat_stream_for_session( except Exception: logger.warning("Failed to append submitted turn journal event", exc_info=True) diag.stage("set_last_workspace") if diag else None - set_last_workspace(workspace) + set_last_workspace(workspace, profile=getattr(s, "profile", None)) diag.stage("stream_registration") if diag else None stream = create_stream_channel() register_stream_owner(stream_id, s.session_id) @@ -23886,7 +23948,7 @@ def _handle_goal_command(handler, body): previous_goal_state = None if will_kickoff: try: - workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace)) + workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace, profile=getattr(s, "profile", None))) except ValueError as e: return bad(handler, str(e)) requested_model = body.get("model") or s.model @@ -23955,7 +24017,7 @@ def _handle_goal_command(handler, body): if kickoff_prompt: if workspace is None: try: - workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace)) + workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace, profile=getattr(s, "profile", None))) except ValueError as e: return bad(handler, str(e)) if model is None: @@ -24336,14 +24398,25 @@ def _handle_chat_start(handler, body, diag=None): def _resolve_chat_workspace_with_recovery(s, requested_workspace) -> str: """Recover stale implicit session workspaces without hiding explicit errors.""" + _session_profile = getattr(s, "profile", None) explicit = requested_workspace not in (None, "") if explicit: - return str(resolve_trusted_workspace(requested_workspace)) + try: + return str(resolve_trusted_workspace(requested_workspace, profile=_session_profile)) + except TypeError: + return str(resolve_trusted_workspace(requested_workspace)) stored_workspace = getattr(s, "workspace", None) - workspace, recovered = resolve_implicit_workspace_with_recovery( - stored_workspace, - get_last_workspace, - ) + try: + workspace, recovered = resolve_implicit_workspace_with_recovery( + stored_workspace, + get_last_workspace, + profile=_session_profile, + ) + except TypeError: + workspace, recovered = resolve_implicit_workspace_with_recovery( + stored_workspace, + get_last_workspace, + ) if not recovered: return str(workspace) persisted = persist_recovered_workspace_binding( @@ -24356,12 +24429,23 @@ def _resolve_chat_workspace_with_recovery(s, requested_workspace) -> str: def _resolve_chat_workspace_for_regeneration(s, requested_workspace) -> str: """Resolve regeneration's workspace without persisting before start acceptance.""" + _session_profile = getattr(s, "profile", None) or None if requested_workspace not in (None, ""): - return str(resolve_trusted_workspace(requested_workspace)) - workspace, _recovered = resolve_implicit_workspace_with_recovery( - getattr(s, "workspace", None), - get_last_workspace, - ) + try: + return str(resolve_trusted_workspace(requested_workspace, profile=_session_profile)) + except TypeError: + return str(resolve_trusted_workspace(requested_workspace)) + try: + workspace, _recovered = resolve_implicit_workspace_with_recovery( + getattr(s, "workspace", None), + get_last_workspace, + profile=_session_profile, + ) + except TypeError: + workspace, _recovered = resolve_implicit_workspace_with_recovery( + getattr(s, "workspace", None), + get_last_workspace, + ) return str(workspace) @@ -24407,7 +24491,10 @@ def _handle_chat_sync(handler, body): if not msg: return j(handler, {"error": "empty message"}, status=400) try: - workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace)) + try: + workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace, profile=getattr(s, "profile", None))) + except TypeError: + workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace)) except ValueError as e: return bad(handler, str(e)) with _get_session_agent_lock(s.session_id): @@ -25790,38 +25877,49 @@ def _handle_workspace_add(handler, body): # macOS) so pytest's tmp_path_factory paths and other legit user-tmp dirs # still register cleanly. try: - candidate = Path(path_str).expanduser().resolve() + from api.workspace import _remote_terminal_workspace_candidate, _resolve_path + from api.profiles import get_active_profile_name + active_profile = get_active_profile_name() + remote_candidate = _remote_terminal_workspace_candidate(path_str, profile=active_profile) + candidate = _resolve_path(path_str, profile=active_profile) 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): - # 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: - candidate.mkdir(parents=True, exist_ok=True) - except (OSError, PermissionError) as e: - return bad(handler, f"Could not create directory: {_sanitize_error(e)}") + if remote_candidate is None: + if _is_blocked_system_path(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: + candidate.mkdir(parents=True, exist_ok=True) + except (OSError, PermissionError) as e: + return bad(handler, f"Could not create directory: {_sanitize_error(e)}") # Full validation (exists, is_dir) — should pass now that dir exists try: - p = validate_workspace_to_add(path_str) + p = validate_workspace_to_add(path_str, profile=active_profile) except ValueError as e: return bad(handler, str(e)) - wss = load_workspaces() + try: + wss = load_workspaces(profile=active_profile) + except TypeError: + wss = load_workspaces() if any(w["path"] == str(p) for w in wss): return bad(handler, "Workspace already in list") wss.append({"path": str(p), "name": name or p.name}) - save_workspaces(wss) + try: + save_workspaces(wss, profile=active_profile) + except TypeError: + save_workspaces(wss) return j(handler, {"ok": True, "workspaces": wss}) @@ -25829,9 +25927,17 @@ def _handle_workspace_remove(handler, body): path_str = body.get("path", "").strip() if not path_str: return bad(handler, "path is required") - wss = load_workspaces() + from api.profiles import get_active_profile_name + active_profile = get_active_profile_name() + try: + wss = load_workspaces(profile=active_profile) + except TypeError: + wss = load_workspaces() wss = [w for w in wss if w["path"] != path_str] - save_workspaces(wss) + try: + save_workspaces(wss, profile=active_profile) + except TypeError: + save_workspaces(wss) return j(handler, {"ok": True, "workspaces": wss}) @@ -25840,14 +25946,22 @@ def _handle_workspace_rename(handler, body): name = body.get("name", "").strip() if not path_str or not name: return bad(handler, "path and name are required") - wss = load_workspaces() + from api.profiles import get_active_profile_name + active_profile = get_active_profile_name() + try: + wss = load_workspaces(profile=active_profile) + except TypeError: + wss = load_workspaces() for w in wss: if w["path"] == path_str: w["name"] = name break else: return bad(handler, "Workspace not found", 404) - save_workspaces(wss) + try: + save_workspaces(wss, profile=active_profile) + except TypeError: + save_workspaces(wss) return j(handler, {"ok": True, "workspaces": wss}) @@ -25861,7 +25975,12 @@ def _handle_workspace_reorder(handler, body): paths = body.get("paths", []) if not paths or not isinstance(paths, list): return bad(handler, "paths is required and must be a list") - wss = load_workspaces() + from api.profiles import get_active_profile_name + active_profile = get_active_profile_name() + try: + wss = load_workspaces(profile=active_profile) + except TypeError: + wss = load_workspaces() by_path = {w["path"]: w for w in wss} # Build reordered list: given order first, then any omitted entries reordered = [] @@ -25875,7 +25994,11 @@ def _handle_workspace_reorder(handler, body): for w in wss: if w["path"] not in seen: reordered.append(w) - save_workspaces(reordered) + try: + save_workspaces(reordered, profile=active_profile) + except TypeError: + # Legacy signature (test doubles with single-arg lambdas, older forks). + save_workspaces(reordered) return j(handler, {"ok": True, "workspaces": reordered}) @@ -28391,7 +28514,7 @@ def _handle_session_import_cli(handler, body): session_payload = { "session_id": sid, "title": title, - "workspace": str(get_last_workspace()), + "workspace": str(get_last_workspace(profile=profile)), "model": model, "message_count": len(msgs), "created_at": created_at, diff --git a/api/streaming.py b/api/streaming.py index f41ab71f..ba2223fd 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -2556,7 +2556,7 @@ def _aiagent_import_error_detail() -> str: lines.append(' Full troubleshooting: docs/troubleshooting.md ("AIAgent not available")') return "\n".join(lines) from api.models import get_session, title_from -from api.workspace import set_last_workspace +from api.workspace import _resolve_path # Fields that are safe to send to LLM provider APIs. # Everything else (attachments, timestamp, _ts, etc.) is display-only @@ -3385,7 +3385,7 @@ def _resolve_image_input_mode(cfg: dict, active_provider: str = "", active_model return "native" -def _build_native_multimodal_message(workspace_ctx: str, msg_text: str, attachments, workspace: str, *, cfg: dict = None, active_provider: str = "", active_model: str = "", requested_provider: str = ""): +def _build_native_multimodal_message(workspace_ctx: str, msg_text: str, attachments, workspace: str, *, cfg: dict = None, active_provider: str = "", active_model: str = "", requested_provider: str = "", profile: str | Path | None = None): """Build native multimodal content parts for current-turn image uploads. WebUI uploads files into the active workspace. For image files, pass the @@ -3405,7 +3405,7 @@ def _build_native_multimodal_message(workspace_ctx: str, msg_text: str, attachme return workspace_ctx + msg_text parts = [{'type': 'text', 'text': workspace_ctx + msg_text}] - workspace_root = Path(workspace).expanduser().resolve() + workspace_root = _resolve_path(workspace, profile=profile) # Stage-361 maintainer fix (Opus SHOULD-FIX): chat uploads from #2319 now # land in ~/.hermes/webui/attachments// (outside workspace_root by # design). The pre-existing `path.relative_to(workspace_root)` guard would @@ -9466,19 +9466,19 @@ def _run_agent_streaming( # Resolved the same way as `s.workspace` below so the bound cwd and the # session's own record cannot disagree. Guarded: a turn must not die # here because a workspace path is malformed. + s = get_session(session_id) try: - _turn_workspace_cwd = str(Path(workspace).expanduser().resolve()) + _turn_workspace_cwd = str(_resolve_path(workspace, profile=getattr(s, 'profile', None))) except Exception: _turn_workspace_cwd = "" logger.debug("per-turn workspace cwd resolve failed", exc_info=True) _turn_session_identity_tokens = _set_turn_session_identity( session_id, workspace=_turn_workspace_cwd ) - s = get_session(session_id) _turn_pending_source = getattr(s, 'pending_user_source', None) or 'webui' _active_turn_identity = _active_turn_authority(s, stream_id, msg_text) update_active_run(stream_id, phase="running", session_id=session_id) - s.workspace = str(Path(workspace).expanduser().resolve()) + s.workspace = _turn_workspace_cwd _last_persisted_model = None _last_persisted_provider = None _turn_owns_persisted_model = False @@ -11006,7 +11006,7 @@ def _run_agent_streaming( _agent_msg_text = msg_text if _process_notifications: _agent_msg_text = "\n\n".join([*_process_notifications, msg_text]).strip() - user_message = _build_native_multimodal_message(workspace_ctx, _agent_msg_text, attachments, workspace, cfg=_cfg, active_provider=(resolved_provider or ""), active_model=(resolved_model or ""), requested_provider=(_session_requested_provider or "")) + user_message = _build_native_multimodal_message(workspace_ctx, _agent_msg_text, attachments, workspace, cfg=_cfg, active_provider=(resolved_provider or ""), active_model=(resolved_model or ""), requested_provider=(_session_requested_provider or ""), profile=(getattr(s, "profile", None) or Path(_profile_home))) _persistent_state_before = _persistent_state_snapshot(_profile_home) _run_conversation_kwargs = _build_run_conversation_kwargs( agent.run_conversation, @@ -11059,6 +11059,10 @@ def _run_agent_streaming( active_provider=(resolved_provider or ""), active_model=(resolved_model or ""), requested_provider=(_session_requested_provider or ""), + # Legacy fallback wraps the home STRING in Path: string + # profiles are logical ids only (round-5 grammar gate), + # explicit homes must arrive as Path values. + profile=(getattr(s, "profile", None) or Path(_profile_home)), ) _run_conversation_kwargs["user_message"] = user_message _result_partial_pre_call_context = list(_previous_context_messages) diff --git a/api/upload.py b/api/upload.py index 9adef00d..6269fca5 100644 --- a/api/upload.py +++ b/api/upload.py @@ -635,8 +635,13 @@ def handle_workspace_upload(handler): if _reject_invisible_session(handler, session): return True - # Resolve workspace root from session - workspace = resolve_trusted_workspace(session.workspace) + # Resolve workspace root using the session profile, not the ambient request profile. + try: + workspace = resolve_trusted_workspace( + session.workspace, profile=getattr(session, "profile", None) + ) + except TypeError: + workspace = resolve_trusted_workspace(session.workspace) # Resolve target subdirectory within workspace target_dir = safe_resolve_ws(workspace, subpath) if subpath else workspace diff --git a/api/workspace.py b/api/workspace.py index 12a46356..d8ffb282 100644 --- a/api/workspace.py +++ b/api/workspace.py @@ -12,6 +12,7 @@ import json import logging import os import posixpath +import re import secrets import shutil import stat @@ -40,8 +41,14 @@ from api.subprocess_utils import windows_hide_flags # ── Profile-aware path resolution ─────────────────────────────────────────── -def _profile_state_dir() -> Path: - """Return the webui_state directory for the active profile. +# Logical profile-name grammar — mirrors api.profiles._PROFILE_ID_RE. Kept as +# a local copy so workspace-layer validation does not import profiles at module +# load time (profiles may not be importable in every embedding context). +_PROFILE_NAME_RE = re.compile(r'^[a-z0-9][a-z0-9_-]{0,63}$') + + +def _profile_state_dir(profile: str | Path | None = None) -> Path: + """Return the webui_state directory for the active or given profile. For the default profile, returns the global STATE_DIR (respects HERMES_WEBUI_STATE_DIR env var for test isolation). @@ -49,6 +56,28 @@ def _profile_state_dir() -> Path: """ try: from api.profiles import get_active_profile_name, get_active_hermes_home + if profile is not None: + # Literal-"default" STATE routing (#7168 re-gate round 7): the + # default profile's workspace state always lives in the global + # state files, even when isolated mode pins the default home at + # /profiles/default. The round-6 resolver change made + # _resolve_profile_home_param("default") return that pinned home, + # so the canonical-home check below sent explicit + # profile="default" state reads/writes to {pinned}/webui_state/ + # while ambient calls kept using the global dir — splitting saved + # workspaces between two authorities. Config and workspace PATH + # resolution still use the pinned home via + # _resolve_profile_home_param; only this state-file tier stays + # global for the logical string "default". + if isinstance(profile, str) and profile.strip() == 'default': + return _GLOBAL_WS_FILE.parent + profile_home = _resolve_profile_home_param(profile) + if not _is_default_profile_home(profile_home): + d = profile_home / 'webui_state' + d.mkdir(parents=True, exist_ok=True) + return d + return _GLOBAL_WS_FILE.parent + name = get_active_profile_name() if name and name != 'default': d = get_active_hermes_home() / 'webui_state' @@ -59,14 +88,41 @@ def _profile_state_dir() -> Path: 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 _workspaces_file(profile: str | Path | None = None) -> Path: + """Return the workspaces.json path for the active or given profile.""" + return _profile_state_dir(profile=profile) / '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 _last_workspace_file(profile: str | Path | None = None) -> Path: + """Return the last_workspace.txt path for the active or given profile.""" + return _profile_state_dir(profile=profile) / 'last_workspace.txt' + + +def _workspaces_file_for_profile(profile: str | Path | None = None) -> Path | None: + """Profile-scoped workspaces.json path, or None for an INVALID profile. + + ``None`` is the fail-closed contract (#7168 re-gate round 4): a malformed + profile name must not be clamped onto the default/global state files, so + callers treat it as "no readable/writable profile-local state". + """ + try: + return _workspaces_file(profile=profile) if profile is not None else _workspaces_file() + except TypeError: + return _workspaces_file() + except ValueError: + logger.debug("Ignoring invalid profile name %r for workspaces file", profile) + return None + + +def _last_workspace_file_for_profile(profile: str | Path | None = None) -> Path | None: + """Profile-scoped last_workspace.txt path, or None for an INVALID profile.""" + try: + return _last_workspace_file(profile=profile) if profile is not None else _last_workspace_file() + except TypeError: + return _last_workspace_file() + except ValueError: + logger.debug("Ignoring invalid profile name %r for last-workspace file", profile) + return None def _expanduser_path(path: str | Path) -> Path: @@ -101,8 +157,11 @@ def _expanduser_path(path: str | Path) -> Path: return Path(raw) -def _resolve_path(path: str | Path) -> Path: - """Resolve *path* after env-aware home expansion, without raising.""" +def _resolve_path(path: str | Path, profile: str | Path | None = None) -> Path: + """Resolve *path* after env-aware home expansion, preserving remote POSIX paths without raising.""" + remote_candidate = _remote_terminal_workspace_candidate(path, profile=profile) + if remote_candidate is not None: + return remote_candidate return _safe_resolve(_expanduser_path(path)) @@ -148,12 +207,75 @@ def _is_remote_terminal_backend(terminal_cfg: dict | None) -> bool: return backend not in ('', 'local') -def _remote_terminal_cwd() -> str | None: - """Return target-side terminal cwd for remote profiles, without local stat().""" +def _resolve_profile_home_param(profile: str | Path | None) -> Path: + """Resolve a profile parameter (name string, directory path string, or Path) to a profile home Path. + + Logical profile names are validated strictly (#7168 re-gate round 4): a + name that fails the profile-id grammar raises ValueError instead of being + silently clamped onto the default home — the old clamp let a malformed + name such as ``"bad name"`` read/write the DEFAULT profile's state. + Round 5 tightens the grammar gate: a STRING profile value is strictly a + logical profile id and is NEVER treated as a path-shaped home — the old + ``"/" in raw`` branch resolved any slash-bearing string directly, so a + malformed value like ``"../evil"`` bypassed validation entirely and could + read/overwrite an arbitrary ``webui_state/last_workspace.txt`` + (#7168 re-gate round 5, path traversal on the profile-isolation boundary). + Round 6 closes the last isolation hole in this resolver: the logical + string ``"default"`` used to short-circuit to ``_DEFAULT_HERMES_HOME`` + before reaching ``get_hermes_home_for_profile()``, bypassing the + isolated-mode clamp in ``api.profiles._resolve_profile_home_for_name`` + (#7168 re-gate round 6). In an isolated deployment pinned at + ``/profiles/default``, a session created with ``profile="default"`` + therefore resolved workspace/config from the BASE root home instead of + the pinned one. The name now flows through the same delegated path as + every other logical id; literal-default routing to the global state + files is retained in ``_profile_state_dir``/``get_last_workspace`` via + canonical ``_is_default_profile_home`` identity. + An explicit home directory is expressed as a ``Path`` object (callers such + as streaming's legacy ``_profile_home`` fallback wrap their home strings + in ``Path``); Path values are honored and canonicalized so identity + comparisons never depend on lexical spelling (e.g. a symlink alias of the + default home must compare equal to it). + """ + if profile is None or str(profile).strip() == "": + from api.profiles import get_active_hermes_home + return get_active_hermes_home() + + raw = str(profile).strip() + + if isinstance(profile, Path): + return _safe_resolve(profile.expanduser()) + + # Strings are LOGICAL PROFILE IDS ONLY — no path-shaped strings, ever. + if not _PROFILE_NAME_RE.fullmatch(raw): + raise ValueError(f"invalid profile name: {raw!r}") + + from api.profiles import get_hermes_home_for_profile + return _safe_resolve(get_hermes_home_for_profile(raw)) + + +def _is_default_profile_home(profile_home: Path) -> bool: + """Canonical identity check against the root/default Hermes home. + + Compares resolved paths so a symlink alias of _DEFAULT_HERMES_HOME is + recognized as the default profile rather than treated as a foreign, + lexically-different directory (#7168 re-gate round 4). + """ try: - from api.config import get_config + from api.profiles import _DEFAULT_HERMES_HOME + return _safe_resolve(profile_home) == _safe_resolve(_DEFAULT_HERMES_HOME) + except Exception: + return False + + +def _remote_terminal_cwd(profile: str | Path | None = None) -> str | None: + """Return target-side terminal cwd for a remote profile, without local stat().""" + try: + from api.config import get_config_for_profile_home + + profile_home = _resolve_profile_home_param(profile) + terminal_cfg = get_config_for_profile_home(profile_home).get('terminal', {}) - terminal_cfg = get_config().get('terminal', {}) if not _is_remote_terminal_backend(terminal_cfg): return None cwd = str(terminal_cfg.get('cwd') or '').strip() @@ -165,8 +287,8 @@ def _remote_terminal_cwd() -> str | None: 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. +def _remote_terminal_workspace_candidate(path: str | Path, profile: str | Path | None = None) -> Path | None: + """Return a non-stat'ed target-side Path when it is under terminal.cwd for the given profile. Remote workspace paths live on the target host (e.g., remote SSH/Docker backend). For valid target-side POSIX paths under ``terminal.cwd``, the @@ -174,7 +296,10 @@ def _remote_terminal_workspace_candidate(path: str | Path) -> Path | None: local host-filesystem resolution (avoiding host-specific firmlink rewriting such as macOS synthetic ``/home`` -> ``/System/Volumes/Data/home``). """ - cwd = _remote_terminal_cwd() + try: + cwd = _remote_terminal_cwd(profile=profile) if profile is not None else _remote_terminal_cwd() + except TypeError: + cwd = _remote_terminal_cwd() if not cwd: return None raw = _strip_surrounding_quotes(str(path)).strip() @@ -192,8 +317,8 @@ def _remote_terminal_workspace_candidate(path: str | Path) -> Path | None: if posix_candidate == posix_base or _posix_is_within(posix_candidate, posix_base): return Path(normalized_raw) return None - candidate = _resolve_path(raw) - base = _resolve_path(cwd) + candidate = _safe_resolve(_expanduser_path(raw)) + base = _safe_resolve(_expanduser_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): @@ -201,7 +326,7 @@ def _remote_terminal_workspace_candidate(path: str | Path) -> Path | None: return None -def _profile_default_workspace() -> str: +def _profile_default_workspace(profile: str | Path | None = None) -> str: """Read the profile's default workspace from its config.yaml. Checks keys in priority order: @@ -217,8 +342,9 @@ def _profile_default_workspace() -> str: Falls back to the live DEFAULT_WORKSPACE from api.config. """ try: - from api.config import get_config - cfg = get_config() + from api.config import get_config_for_profile_home + profile_home = _resolve_profile_home_param(profile) + cfg = get_config_for_profile_home(profile_home) terminal_cfg = cfg.get('terminal', {}) remote_terminal = _is_remote_terminal_backend(terminal_cfg) # Explicit webui workspace keys first @@ -227,7 +353,7 @@ def _profile_default_workspace() -> str: if ws: if remote_terminal: return str(ws).strip() - p = _resolve_path(str(ws)) + p = _resolve_path(str(ws), profile=profile) if remote_terminal or p.is_dir(): return str(p) # Fall through to terminal.cwd — the agent's configured working directory @@ -236,7 +362,7 @@ def _profile_default_workspace() -> str: if cwd and str(cwd) not in ('.', ''): if remote_terminal: return str(cwd).strip() - p = _resolve_path(str(cwd)) + p = _resolve_path(str(cwd), profile=profile) if remote_terminal or p.is_dir(): return str(p) except (ImportError, Exception): @@ -244,14 +370,14 @@ def _profile_default_workspace() -> str: try: from api.config import DEFAULT_WORKSPACE as _LIVE_DEFAULT_WORKSPACE - return str(_resolve_path(_LIVE_DEFAULT_WORKSPACE)) + return str(_resolve_path(_LIVE_DEFAULT_WORKSPACE, profile=profile)) except Exception: - return str(_resolve_path(_BOOT_DEFAULT_WORKSPACE)) + return str(_resolve_path(_BOOT_DEFAULT_WORKSPACE, profile=profile)) # ── Public API ────────────────────────────────────────────────────────────── -def _clean_workspace_list(workspaces: list) -> list: +def _clean_workspace_list(workspaces: list, profile: str | Path | None = None) -> list: """Sanitize a workspace list: - Preserve target-side remote terminal workspace paths (SSH/Docker) without resolving them against the local WebUI host filesystem. @@ -270,7 +396,7 @@ def _clean_workspace_list(workspaces: list) -> list: name = w.get('name', '') if not path: continue - remote_cand = _remote_terminal_workspace_candidate(path) + remote_cand = _remote_terminal_workspace_candidate(path, profile=profile) if remote_cand is not None: p = remote_cand else: @@ -283,7 +409,14 @@ def _clean_workspace_list(workspaces: list) -> list: # 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() + if profile is not None: + # Explicit profile wins: the list belongs to that profile, + # so "own" is defined by the profile parameter, never by the + # ambient home (loading profile A's list under ambient B must + # not silently drop A's own workspaces). + own_profile_dir = _resolve_profile_home_param(profile).resolve() + else: + 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): @@ -350,12 +483,12 @@ def _migrate_global_workspaces() -> list: return [] -def load_workspaces() -> list: - ws_file = _workspaces_file() - if ws_file.exists(): +def load_workspaces(profile: str | Path | None = None) -> list: + ws_file = _workspaces_file_for_profile(profile) + if ws_file is not None and ws_file.exists(): try: raw = json.loads(ws_file.read_text(encoding='utf-8')) - cleaned = _clean_workspace_list(raw) + cleaned = _clean_workspace_list(raw, profile=profile) if len(cleaned) != len(raw): # Persist the cleaned version so stale entries don't keep reappearing try: @@ -364,7 +497,7 @@ def load_workspaces() -> list: ) except Exception: logger.debug("Failed to persist cleaned workspace list") - return cleaned or [{'path': _profile_default_workspace(), 'name': 'Home'}] + return cleaned or [{'path': _profile_default_workspace(profile=profile), 'name': 'Home'}] except Exception: logger.debug("Failed to load workspaces from %s", ws_file) # No profile-local file yet. @@ -372,7 +505,11 @@ def load_workspaces() -> list: # 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) + if profile is not None: + profile_home = _resolve_profile_home_param(profile) + is_default = _is_default_profile_home(profile_home) + else: + is_default = get_active_profile_name() in ('default', None) except ImportError: is_default = True if is_default: @@ -380,16 +517,20 @@ def load_workspaces() -> list: if migrated: return migrated # Fresh start: single entry from the profile's configured workspace, labeled "Home" - return [{'path': _profile_default_workspace(), 'name': 'Home'}] + return [{'path': _profile_default_workspace(profile=profile), 'name': 'Home'}] -def save_workspaces(workspaces: list) -> None: - ws_file = _workspaces_file() +def save_workspaces(workspaces: list, profile: str | Path | None = None) -> None: + ws_file = _workspaces_file_for_profile(profile) + if ws_file is None: + # Fail-closed: an invalid profile name must not write any state file + # (it would land in the default profile's directory via the old clamp). + raise ValueError(f"cannot save workspaces for invalid profile {profile!r}") 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: +def get_profile_default_workspace(profile: str | Path | None = None) -> str: """Resolve the ACTIVE PROFILE's default workspace, never the global file. Like get_last_workspace() but WITHOUT the global ``_GLOBAL_LW_FILE`` @@ -404,32 +545,38 @@ def get_profile_default_workspace() -> str: Priority: profile-scoped ``last_workspace.txt`` -> profile ``config.yaml`` ``workspace``/``default_workspace`` -> ``terminal.cwd`` -> process default. """ - remote_cwd = _remote_terminal_cwd() + try: + remote_cwd = _remote_terminal_cwd(profile=profile) if profile is not None else _remote_terminal_cwd() + except TypeError: + 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: + if _remote_terminal_workspace_candidate(raw, profile=profile) 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(): + lw_file = _last_workspace_file_for_profile(profile) + if lw_file is not None and 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() + return _profile_default_workspace(profile=profile) -def get_last_workspace() -> str: - remote_cwd = _remote_terminal_cwd() +def get_last_workspace(profile: str | Path | None = None) -> str: + try: + remote_cwd = _remote_terminal_cwd(profile=profile) if profile is not None else _remote_terminal_cwd() + except TypeError: + remote_cwd = _remote_terminal_cwd() def valid_last_workspace(raw: str) -> str | None: if not raw: @@ -438,35 +585,58 @@ def get_last_workspace() -> str: # 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: + if _remote_terminal_workspace_candidate(raw, profile=profile) 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(): + lw_file = _last_workspace_file_for_profile(profile) + if lw_file is not None and 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(): + # Fallback: try global file — but ONLY for the root/default profile. A named + # profile must never inherit another profile's last-workspace binding through + # the legacy global state (#7168 re-gate round 3). Identity is canonical: + # a symlink alias of the default home still counts as default, while a + # malformed profile name fails validation and is denied the fallback + # rather than being clamped onto the global file (#7168 re-gate round 4). + _global_fallback_allowed = True # ambient / no explicit profile: historical behavior + if profile is not None: + if str(profile).strip() == 'default': + _global_fallback_allowed = True + else: + try: + _global_fallback_allowed = _is_default_profile_home( + _resolve_profile_home_param(profile) + ) + except Exception: + # Conservative default: an unresolvable explicit profile is NOT the + # root profile — deny the global fallback rather than leak. + _global_fallback_allowed = False + if _global_fallback_allowed and _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() + return _profile_default_workspace(profile=profile) -def set_last_workspace(path: str) -> None: +def set_last_workspace(path: str, profile: str | Path | None = None) -> None: try: - lw_file = _last_workspace_file() + lw_file = _last_workspace_file_for_profile(profile) + if lw_file is None: + # Fail-closed: an invalid profile name must not write the default + # profile's last_workspace.txt (#7168 re-gate round 4). + logger.debug("Refusing to set last workspace for invalid profile %r", profile) + return lw_file.parent.mkdir(parents=True, exist_ok=True) lw_file.write_text(str(path), encoding='utf-8') except Exception: @@ -683,7 +853,15 @@ def _is_within(path: Path, root: Path) -> bool: return False -def _trusted_workspace_roots() -> list[Path]: +def _trusted_workspace_roots(profile: str | Path | None = None) -> list[Path]: + """Return the host directories workspace suggestions may traverse. + + Saved-workspace roots follow the same trust rule as + :func:`resolve_trusted_workspace`: with an explicit *profile*, only + workspaces saved under THAT profile widen the boundary (plus the ambient + home / boot-default carve-outs); ``None`` keeps the historical ambient / + global saved-list behaviour. + """ roots: list[Path] = [] def add(candidate: str | Path | None) -> None: @@ -702,24 +880,26 @@ def _trusted_workspace_roots() -> list[Path]: add(_home_path()) add(_BOOT_DEFAULT_WORKSPACE) - for w in load_workspaces(): + for w in load_workspaces(profile=profile): 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]: +def list_workspace_suggestions( + prefix: str = "", limit: int = 12, profile: str | Path | None = None +) -> 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 + - already-saved workspace roots (scoped to *profile* when given) 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() + roots = _trusted_workspace_roots(profile=profile) if not roots: return [] @@ -815,7 +995,7 @@ def list_workspace_suggestions(prefix: str = "", limit: int = 12) -> list[str]: return suggestions[:limit] -def resolve_trusted_workspace(path: str | Path | None = None) -> Path: +def resolve_trusted_workspace(path: str | Path | None = None, profile: str | Path | None = None) -> Path: """Resolve and validate a workspace path. A path is trusted if it satisfies at least one of: @@ -837,12 +1017,12 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path: trusted (it was validated at server startup). """ if path in (None, ""): - return _resolve_path(_BOOT_DEFAULT_WORKSPACE) + return _resolve_path(_BOOT_DEFAULT_WORKSPACE, profile) if profile is not None else _resolve_path(_BOOT_DEFAULT_WORKSPACE) - candidate = _resolve_path(path) + candidate = _resolve_path(path, profile) if profile is not None else _resolve_path(path) access_error = _workspace_access_error(candidate) - remote_candidate = _remote_terminal_workspace_candidate(path) + remote_candidate = _remote_terminal_workspace_candidate(path, profile=profile) if access_error: # For remote terminal profiles, workspace paths belong to the target # machine. Allow paths under terminal.cwd so session switching can @@ -869,6 +1049,11 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path: # (B) Trusted if already in the saved workspace list — covers non-home installs try: + saved = load_workspaces(profile=profile) + saved_paths = {_resolve_path(w["path"], profile) for w in saved if w.get("path")} + if candidate in saved_paths: + return candidate + except TypeError: saved = load_workspaces() saved_paths = {_resolve_path(w["path"]) for w in saved if w.get("path")} if candidate in saved_paths: @@ -898,15 +1083,22 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path: def resolve_implicit_workspace_with_recovery( candidate: str | Path | None, fallback: str | Path | None | Callable[[], str | Path | None], + profile: str | Path | None = None, ) -> tuple[Path, bool]: """Resolve an implicit workspace, recovering only a genuinely missing path. The fallback still passes through :func:`resolve_trusted_workspace`. Existing but untrusted, inaccessible, or non-directory candidates are not recovery cases: their original validation error is preserved so fallback cannot widen - the workspace trust boundary. + the workspace trust boundary. When *profile* is given, both the trust + resolution and the recovery fallback are scoped to that profile. """ try: + if profile is not None: + try: + return resolve_trusted_workspace(candidate, profile=profile), False + except TypeError: + pass return resolve_trusted_workspace(candidate), False except ValueError as original_error: if candidate in (None, ""): @@ -917,19 +1109,49 @@ def resolve_implicit_workspace_with_recovery( # never prove target-side deletion. Config-read uncertainty also fails # closed by preserving the original validation error. try: - from api.config import get_config + from api.config import get_config, get_config_for_profile_home - terminal_cfg = get_config().get("terminal", {}) + if profile is not None: + # Classify the backend from THIS profile's own config, never the + # ambient one: a remote profile without terminal.cwd must still be + # recognized as remote when loaded under a different ambient home. + terminal_cfg = get_config_for_profile_home( + _resolve_profile_home_param(profile) + ).get("terminal", {}) + else: + terminal_cfg = get_config().get("terminal", {}) except Exception: logger.debug("Failed to classify terminal backend for workspace recovery", exc_info=True) raise original_error from None if _is_remote_terminal_backend(terminal_cfg): raise original_error from None try: - local_candidate = _resolve_path(candidate) + local_candidate = ( + _resolve_path(candidate, profile=profile) + if profile is not None + else _resolve_path(candidate) + ) local_candidate.stat() except FileNotFoundError: - fallback_value = fallback() if callable(fallback) else fallback + def _profile_bound_fallback(): + if profile is not None and callable(fallback): + # Profile-bound getter FIRST (#7168 re-gate round 3): the + # production call sites pass profile-aware getters such as + # get_last_workspace, so binding the explicit profile must + # take precedence over any zero-argument compatibility call, + # which would read ambient/global state. + try: + return fallback(profile) + except TypeError: + pass + return fallback() if callable(fallback) else fallback + + fallback_value = _profile_bound_fallback() + if profile is not None: + try: + return resolve_trusted_workspace(fallback_value, profile=profile), True + except TypeError: + pass return resolve_trusted_workspace(fallback_value), True except (OSError, RuntimeError, ValueError): raise original_error from None @@ -957,7 +1179,7 @@ def _strip_surrounding_quotes(path: str) -> str: return s -def validate_workspace_to_add(path: str) -> Path: +def validate_workspace_to_add(path: str, profile: str | Path | None = None) -> 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 @@ -972,10 +1194,10 @@ def validate_workspace_to_add(path: str) -> Path: and users routinely paste those into the Add Space input. """ path = _strip_surrounding_quotes(path) - candidate = _resolve_path(path) + candidate = _resolve_path(path, profile) if profile is not None else _resolve_path(path) access_error = _workspace_access_error(candidate) - remote_candidate = _remote_terminal_workspace_candidate(path) + remote_candidate = _remote_terminal_workspace_candidate(path, profile=profile) if access_error: # Remote terminal profiles validate workspace existence on the target # machine, not on the WebUI server. Permit target-side paths under diff --git a/docs/CONTRACTS.md b/docs/CONTRACTS.md index fcfc5d54..da8a1022 100644 --- a/docs/CONTRACTS.md +++ b/docs/CONTRACTS.md @@ -23,6 +23,10 @@ contributor guidance; it does not change runtime behavior or CI gates. ## Runtime, durability, and state contracts +- [`docs/remote-workspaces.md`](remote-workspaces.md): + architecture contract for remote terminal workspaces (SSH/Docker), target-side + POSIX path preservation against macOS synthetic firmlink expansion, and + per-profile isolation boundaries. - [`docs/rfcs/webui-run-state-consistency-contract.md`](rfcs/webui-run-state-consistency-contract.md): proposed consistency rules for current WebUI streaming, recovery, replay, model-context reconstruction, compression, UI scene/cache, and sidebar metadata diff --git a/docs/remote-workspaces.md b/docs/remote-workspaces.md new file mode 100644 index 00000000..c32ba384 --- /dev/null +++ b/docs/remote-workspaces.md @@ -0,0 +1,29 @@ +# Remote Terminal Workspaces + +Architecture contract and path-resolution semantics for remote terminal profiles (SSH, Docker) in Hermes WebUI. + +--- + +## 1. Overview + +When a Hermes profile is configured with a remote terminal backend (e.g. `terminal.backend: "ssh"` or `"docker"`), its working directory (`terminal.cwd`) lives on the remote target host rather than the local WebUI host filesystem. + +On hosts such as macOS, local path resolution via `Path.resolve()` or `os.path.realpath()` expands synthetic firmlinks (e.g. rewriting `/home/` to `/System/Volumes/Data/home/`). Because the target-side path does not exist on the local macOS server, unconstrained local resolution causes runtime validation failures and session corruption. + +--- + +## 2. Path Resolution Contract + +1. **Target-side POSIX Path Preservation:** + - Any valid POSIX path at or beneath a remote profile's configured `terminal.cwd` is preserved verbatim as a `Path` object without calling host-local `Path.resolve()`. + - Traversal escapes (`..`), null bytes (`\0`), and blocked system roots (`/etc`, `/usr`, `/var`, `/sys`, `/proc`, `/dev`) continue to be strictly rejected. + +2. **Profile Isolation & Boundary Enforcement:** + - Remote path recognition is **strictly scoped to the target/active profile**. + - If an active profile has a local backend, target-side remote paths belonging to inactive named profiles are **not** treated as remote for the active local profile. + - For local profiles, workspace addition (`/api/workspaces/add`) enforces host-local directory existence and permissions. + - For remote profiles, workspace addition validates against the profile's remote `terminal.cwd` and skips host-local directory creation (`mkdir`). + +3. **Session & Streaming Lifecycle:** + - `Session.__init__` and `Session.load` normalize session workspace paths through `_resolve_path(..., profile=profile)`, preserving target-side paths for remote profiles and preventing corruption of `session.workspace` or `session.created_workspace`. + - Streaming execution (`_run_agent_streaming`) and multimodal asset root resolution preserve the remote session workspace when updating runtime run state. diff --git a/tests/conftest.py b/tests/conftest.py index 7dc77884..a492e666 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -231,6 +231,44 @@ def _reset_password_hash_cache(): _invalidate_password_hash_cache() +def _strip_leaked_webui_password_env() -> None: + """Remove a leaked HERMES_WEBUI_PASSWORD between tests (#7168 review). + + bootstrap.py runs _load_repo_dotenv() at import time, which copies values + from the developer's real repo .env straight into os.environ. When any + test imports bootstrap mid-session (e.g. tests/test_bootstrap_foreground.py + via its import_bootstrap fixture), a local .env containing + HERMES_WEBUI_PASSWORD leaks into the process environment OUTSIDE + monkeypatch's undo scope. Every later test then sees is_auth_enabled() + True and no-handler cookie helpers raise spurious + "build_profile_cookie requires a request handler" errors — exactly the + #5588 failure shape, but sourced from the repo .env instead of the hash + cache. Tests that legitimately enable auth set the var themselves AFTER + this strip; an intentionally-empty value ("") is preserved so + ctl.sh-style override semantics keep working. + + HERMES_COMMAND gets the same treatment (#7168 re-gate round 7): a local + .env carrying HERMES_COMMAND leaks past bootstrap imports and redirects + gateway_restart._resolve_hermes_command() away from its mocked + shutil.which result, failing every later active-profile-restart test + with a machine-specific CLI path. Upstream code has no + HERMES_COMMAND override, so stripping a leaked value restores exact + upstream semantics. + """ + if os.environ.get("HERMES_WEBUI_PASSWORD") == "": + pass # intentional empty override preserved for the password var + else: + os.environ.pop("HERMES_WEBUI_PASSWORD", None) + os.environ.pop("HERMES_COMMAND", None) + + +@pytest.fixture(autouse=True) +def _strip_leaked_webui_password(): + _strip_leaked_webui_password_env() + yield + _strip_leaked_webui_password_env() + + @pytest.fixture(autouse=True) def _invalidate_providers_cache(): """Clear the /api/providers TTL cache around every test (#6010). diff --git a/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py index b787bc11..40d5ee64 100644 --- a/tests/test_agent_runtime_revision_guard.py +++ b/tests/test_agent_runtime_revision_guard.py @@ -854,7 +854,7 @@ def test_stream_admission_uses_one_gateway_ownership_snapshot(monkeypatch, gatew monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda _sid: None) monkeypatch.setattr(routes, "_is_hidden_empty_session", lambda _session: False) monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", prepare) - monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace, **_kw: None) monkeypatch.setattr(routes.threading, "Thread", FakeThread) monkeypatch.setattr(turn_journal, "append_turn_journal_event", lambda *_args, **_kwargs: {}) diff --git a/tests/test_bootstrap_dotenv.py b/tests/test_bootstrap_dotenv.py index 2f354d1f..fbe6716d 100644 --- a/tests/test_bootstrap_dotenv.py +++ b/tests/test_bootstrap_dotenv.py @@ -197,3 +197,87 @@ class TestBootstrapStructure: assert "_load_repo_dotenv" in bootstrap_src, ( "bootstrap.py must load .env so direct invocation matches start.sh behaviour" ) + + +class TestLeakedWebuiPasswordIsolation: + """#7168 review: a local repo .env containing HERMES_WEBUI_PASSWORD leaks + into os.environ when any test imports bootstrap (import-time + _load_repo_dotenv() runs OUTSIDE monkeypatch's undo scope). The conftest + autouse guard strips the leaked var around every test so later tests + don't see is_auth_enabled()==True (the #5588 failure shape).""" + + def _load_guard(self): + import importlib.util + + cpath = Path(__file__).resolve().parent / "conftest.py" + spec = importlib.util.spec_from_file_location("_test_conftest", cpath) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod._strip_leaked_webui_password_env + + def test_conftest_guard_strips_leaked_password(self): + strip = self._load_guard() + + os.environ["HERMES_WEBUI_PASSWORD"] = "leaked-from-repo-dotenv" + try: + strip() + assert "HERMES_WEBUI_PASSWORD" not in os.environ + finally: + os.environ.pop("HERMES_WEBUI_PASSWORD", None) + + def test_conftest_guard_preserves_intentional_empty_override(self): + """ctl.sh-style override semantics: an explicitly empty value means + 'keep auth off' and must survive the strip.""" + strip = self._load_guard() + + sentinel = object() + os.environ["HERMES_WEBUI_PASSWORD"] = "" + try: + strip() + assert os.environ.get("HERMES_WEBUI_PASSWORD", sentinel) == "" + finally: + os.environ.pop("HERMES_WEBUI_PASSWORD", None) + + +class TestLeakedHermesCommandIsolation: + """#7168 re-gate round 7: a local repo .env carrying HERMES_COMMAND leaks + into os.environ via bootstrap import-time _load_repo_dotenv() and + redirects gateway_restart._resolve_hermes_command() away from its mocked + shutil.which result — every later active-profile-restart test then fails + on a machine-specific CLI path. The autouse conftest guard strips the + leaked var around every test; upstream code never reads HERMES_COMMAND, + so stripping restores exact upstream semantics.""" + + def _load_guard(self): + import importlib.util + + cpath = Path(__file__).resolve().parent / "conftest.py" + spec = importlib.util.spec_from_file_location("_test_conftest_hc", cpath) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod._strip_leaked_webui_password_env + + def test_conftest_guard_strips_leaked_hermes_command(self): + strip = self._load_guard() + + os.environ["HERMES_COMMAND"] = "/machine/local/hermes-gateway-wrapper" + try: + strip() + assert "HERMES_COMMAND" not in os.environ + finally: + os.environ.pop("HERMES_COMMAND", None) + + def test_conftest_guard_still_strips_password_when_command_leaks(self): + """Both leaked vars are stripped in one pass (password branch must + not be short-circuited by the HERMES_COMMAND handling).""" + strip = self._load_guard() + + os.environ["HERMES_WEBUI_PASSWORD"] = "leaked-from-repo-dotenv" + os.environ["HERMES_COMMAND"] = "/machine/local/hermes-gateway-wrapper" + try: + strip() + assert "HERMES_WEBUI_PASSWORD" not in os.environ + assert "HERMES_COMMAND" not in os.environ + finally: + os.environ.pop("HERMES_WEBUI_PASSWORD", None) + os.environ.pop("HERMES_COMMAND", None) diff --git a/tests/test_claude_code_session_import.py b/tests/test_claude_code_session_import.py index 9d9af034..7dd67581 100644 --- a/tests/test_claude_code_session_import.py +++ b/tests/test_claude_code_session_import.py @@ -200,7 +200,7 @@ def test_session_import_cli_returns_read_only_claude_code_payload(monkeypatch, t monkeypatch.setattr(routes, "j", lambda _handler, payload, status=200, extra_headers=None: payload) monkeypatch.setattr(routes, "get_cli_session_messages", lambda _sid, profile=None: messages if _sid == sid else []) monkeypatch.setattr(routes, "get_cli_sessions", lambda source_filter=None, all_profiles=False: [meta]) - monkeypatch.setattr(routes, "get_last_workspace", lambda: tmp_path / "workspace") + monkeypatch.setattr(routes, "get_last_workspace", lambda profile=None: tmp_path / "workspace") monkeypatch.setattr(routes, "import_cli_session", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("read-only import must not persist"))) response = routes._handle_session_import_cli(object(), {"session_id": sid}) diff --git a/tests/test_early_session_title.py b/tests/test_early_session_title.py index 864e0e73..b236a642 100644 --- a/tests/test_early_session_title.py +++ b/tests/test_early_session_title.py @@ -93,7 +93,7 @@ def test_start_chat_stream_response_includes_provisional_title(tmp_path, monkeyp import api.routes as routes monkeypatch.setattr(Session, "save", lambda self, *a, **k: None) - monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None) monkeypatch.setattr(routes, "create_stream_channel", lambda: object()) monkeypatch.setattr(routes, "_run_agent_streaming", lambda *a, **k: None) diff --git a/tests/test_goal_command_webui.py b/tests/test_goal_command_webui.py index a049b27d..2fd7800c 100644 --- a/tests/test_goal_command_webui.py +++ b/tests/test_goal_command_webui.py @@ -397,7 +397,7 @@ def test_goal_endpoint_sets_goal_and_starts_kickoff_stream( monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager) monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession()) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path) monkeypatch.setattr( routes, "webui_gateway_chat_enabled", @@ -519,7 +519,7 @@ def test_goal_endpoint_adapter_keeps_full_set_text_and_legacy_payload_status(mon monkeypatch.setenv("HERMES_WEBUI_RUNTIME_ADAPTER", "legacy-journal") monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager) monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession()) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path) monkeypatch.setattr( routes, "_resolve_compatible_session_model_state", @@ -732,7 +732,7 @@ def test_goal_kickoff_forwards_explicit_model_pick_to_resolver(monkeypatch, tmp_ monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager) monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession()) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path) monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False) monkeypatch.setattr(routes, "get_config", lambda: {}) monkeypatch.setattr(routes, "_resolve_compatible_session_model_state", fake_resolve) @@ -809,7 +809,7 @@ def test_goal_kickoff_defaults_explicit_model_pick_false(monkeypatch, tmp_path): monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager) monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession()) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path) monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False) monkeypatch.setattr(routes, "get_config", lambda: {}) monkeypatch.setattr(routes, "_resolve_compatible_session_model_state", fake_resolve) @@ -901,7 +901,7 @@ def test_goal_kickoff_stamps_explicit_pick_signature(monkeypatch, tmp_path): monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager) monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession()) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path) monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False) monkeypatch.setattr(routes, "get_config", lambda: {}) monkeypatch.setattr(routes, "_resolve_compatible_session_model_state", fake_resolve) @@ -973,7 +973,7 @@ def test_goal_kickoff_does_not_stamp_signature_without_explicit_pick(monkeypatch return model, provider, False monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path) monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False) monkeypatch.setattr(routes, "get_config", lambda: {}) monkeypatch.setattr(routes, "_resolve_compatible_session_model_state", fake_resolve) diff --git a/tests/test_issue1217_transcript_compaction.py b/tests/test_issue1217_transcript_compaction.py index f2954b44..c4e12cc2 100644 --- a/tests/test_issue1217_transcript_compaction.py +++ b/tests/test_issue1217_transcript_compaction.py @@ -748,7 +748,7 @@ def test_handle_chat_sync_writeback_dedupes_full_context_replay(tmp_path, monkey monkeypatch.setattr(routes, "title_from", models.title_from) monkeypatch.setattr(config, "get_config", lambda: {"model": "test-model", "provider": "test-provider"}) monkeypatch.setattr(routes, "get_config", lambda: {"model": "test-model", "provider": "test-provider"}) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path) monkeypatch.setattr(routes, "load_settings", lambda: {}) monkeypatch.setattr(routes, "_resolve_cli_toolsets", lambda: []) diff --git a/tests/test_issue3930_source_filter_pushdown.py b/tests/test_issue3930_source_filter_pushdown.py index 5503208a..ade4885d 100644 --- a/tests/test_issue3930_source_filter_pushdown.py +++ b/tests/test_issue3930_source_filter_pushdown.py @@ -241,7 +241,7 @@ def test_load_cli_sessions_uncached_pushes_specific_source_into_state_db_scan(mo monkeypatch.setattr(models, "get_claude_code_sessions", lambda: claude_calls.append(True) or []) monkeypatch.setattr(models, "read_importable_agent_session_rows", fake_read_rows) - monkeypatch.setattr(models, "get_last_workspace", lambda: tmp_path) + monkeypatch.setattr(models, "get_last_workspace", lambda profile=None: tmp_path) monkeypatch.setattr(models, "_profile_has_user_projects", lambda: False) monkeypatch.setattr(models, "ensure_cron_project", lambda **_: "cron-project-id") monkeypatch.setattr(models.Session, "load_metadata_only", lambda _sid: None) @@ -283,7 +283,7 @@ def test_cron_source_filter_uses_cron_rescue_limit(monkeypatch, tmp_path): ] monkeypatch.setattr(models, "read_importable_agent_session_rows", fake_read_rows) - monkeypatch.setattr(models, "get_last_workspace", lambda: tmp_path) + monkeypatch.setattr(models, "get_last_workspace", lambda profile=None: tmp_path) monkeypatch.setattr(models, "_profile_has_user_projects", lambda: False) monkeypatch.setattr(models, "ensure_cron_project", lambda **_: "cron-project-id") monkeypatch.setattr(models.Session, "load_metadata_only", lambda _sid: None) diff --git a/tests/test_issue492_workspace_reorder.py b/tests/test_issue492_workspace_reorder.py index 91e87a20..12a591e6 100644 --- a/tests/test_issue492_workspace_reorder.py +++ b/tests/test_issue492_workspace_reorder.py @@ -22,7 +22,8 @@ class TestWorkspaceReorderEndpoint: {"path": "/home/user/b", "name": "Beta"}, {"path": "/home/user/c", "name": "Gamma"}, ] - mock_save.side_effect = lambda wss: wss + # Accept the profile-scoped save signature (save_workspaces(wss, profile=...)). + mock_save.side_effect = lambda wss, **kwargs: wss handler = _make_handler() _handle_workspace_reorder(handler, { "paths": ["/home/user/c", "/home/user/a", "/home/user/b"] diff --git a/tests/test_issue6611_regeneration_transaction.py b/tests/test_issue6611_regeneration_transaction.py index 7f1db4d4..c84aa591 100644 --- a/tests/test_issue6611_regeneration_transaction.py +++ b/tests/test_issue6611_regeneration_transaction.py @@ -193,7 +193,7 @@ def test_locked_postacceptance_workspace_exception_does_not_restore_turn(monkeyp monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue()) monkeypatch.setattr(turn_journal, "append_turn_journal_event", lambda *_args, **_kwargs: {"turn_id": "turn-6611"}) monkeypatch.setattr(Session, "save", lambda *_args, **_kwargs: None) - monkeypatch.setattr(routes, "set_last_workspace", lambda *_args: (_ for _ in ()).throw(RuntimeError("workspace failed"))) + monkeypatch.setattr(routes, "set_last_workspace", lambda *_args, **_kw: (_ for _ in ()).throw(RuntimeError("workspace failed"))) class FakeThread: def __init__(self, *args, **kwargs): diff --git a/tests/test_issue6751_api_content_agent_replay.py b/tests/test_issue6751_api_content_agent_replay.py index e78850cd..b7f79259 100644 --- a/tests/test_issue6751_api_content_agent_replay.py +++ b/tests/test_issue6751_api_content_agent_replay.py @@ -435,7 +435,7 @@ def test_issue6751_sync_chat_agent_receives_original_api_content_bytes(monkeypat monkeypatch.setattr(routes, "title_from", models.title_from) monkeypatch.setattr(config, "get_config", lambda: {"model": "test-model", "provider": "test-provider"}) monkeypatch.setattr(routes, "get_config", lambda: {"model": "test-model", "provider": "test-provider"}) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path) monkeypatch.setattr(routes, "load_settings", lambda: {}) monkeypatch.setattr(routes, "_resolve_cli_toolsets", lambda: []) @@ -526,7 +526,7 @@ def test_issue6751_json_import_strips_internal_aliases_before_persistence(monkey monkeypatch.setattr(models, "SESSION_INDEX_FILE", state_dir / "session_index.json") monkeypatch.setattr(models, "SESSIONS", sessions) monkeypatch.setattr(routes, "SESSIONS", sessions) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path) monkeypatch.setattr(routes, "get_active_profile_name", lambda: "default") monkeypatch.setattr(routes, "publish_session_list_changed", lambda *args, **kwargs: None) monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False}) @@ -625,7 +625,7 @@ def test_issue6751_json_import_nested_tool_calls_are_removed_at_agent_boundary( monkeypatch.setattr(models, "SESSIONS", sessions) monkeypatch.setattr(routes, "SESSION_INDEX_FILE", state_dir / "session_index.json") monkeypatch.setattr(routes, "SESSIONS", sessions) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path) monkeypatch.setattr(routes, "get_active_profile_name", lambda: "default") monkeypatch.setattr(routes, "publish_session_list_changed", lambda *args, **kwargs: None) monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False}) @@ -784,7 +784,7 @@ def test_issue6751_json_import_rejects_non_list_session_tool_calls(monkeypatch, monkeypatch.setattr(models, "SESSIONS", sessions) monkeypatch.setattr(routes, "SESSION_INDEX_FILE", state_dir / "session_index.json") monkeypatch.setattr(routes, "SESSIONS", sessions) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path) monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False}) captured = {} diff --git a/tests/test_issue7168_round6_default_isolated_clamp.py b/tests/test_issue7168_round6_default_isolated_clamp.py new file mode 100644 index 00000000..a6528d8f --- /dev/null +++ b/tests/test_issue7168_round6_default_isolated_clamp.py @@ -0,0 +1,294 @@ +"""Re-gate round 6 (#7168): the logical "default" name must honor isolated-mode clamping. + +Maintainer round-6 re-gate CORE (one remaining instance of the isolation +class): ``_resolve_profile_home_param("default")`` short-circuited to +``_DEFAULT_HERMES_HOME`` BEFORE delegating to +``api.profiles.get_hermes_home_for_profile()``, so it never reached the +isolated-mode clamp in ``_resolve_profile_home_for_name`` (which pins every +lookup to ``_INITIAL_HERMES_HOME`` when ``HERMES_WEBUI_ISOLATED_PROFILE`` is +enabled). In an isolated deployment pinned at ``/profiles/default``, a +session created with ``profile="default"`` therefore used the BASE root +home's workspace + config instead of the pinned one. + +Fix under test: the literal shortcut is gone — the logical string ``"default"`` +flows through ``get_hermes_home_for_profile()`` like every other id, so the +clamp applies. Literal-default routing to the global state files is RETAINED +(canonical ``_is_default_profile_home`` identity), asserted by the fallback +tests below. +""" + +import pytest + +import api.workspace as workspace +from api import profiles + + +@pytest.fixture +def isolated_default_pinned(tmp_path, monkeypatch): + """Isolated-mode deployment pinned at /profiles/default. + + Mirrors the gate reproduction: HERMES_WEBUI_ISOLATED_PROFILE=1 with a + profile-shaped _INITIAL_HERMES_HOME whose directory NAME is literally + 'default' (the exact shape where base and pin diverge). Base home gets a + DISTINCT config/workspace marker so any leak of the root home is visible. + """ + base = tmp_path / ".hermes" + pinned = base / "profiles" / "default" + pinned.mkdir(parents=True) + (pinned / "webui_state").mkdir() + + # Distinct base-vs-pinned configs: only the PINNED one carries the marker. + (pinned / "config.yaml").write_text( + "workspace: /srv/pinned-workspace\n", encoding="utf-8" + ) + + monkeypatch.setenv("HERMES_WEBUI_ISOLATED_PROFILE", "1") + monkeypatch.setattr(profiles, "_INITIAL_ISOLATED_PROFILE_OPT_IN", "1") + monkeypatch.setattr(profiles, "_INITIAL_HERMES_HOME", str(pinned)) + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", base) + monkeypatch.setattr(profiles, "_LIST_PROFILES_CACHE", None) + # Hermetic w.r.t. the runner's real state dir / remote terminal config. + monkeypatch.setattr(workspace, "_remote_terminal_cwd", lambda profile=None: None) + return {"base": base, "pinned": pinned, "tmp": tmp_path} + + +class TestRound6DefaultHonorsIsolationClamp: + """The logical "default" name resolves through the clamped delegated path.""" + + def test_resolver_pins_default_in_isolated_mode(self, isolated_default_pinned): + env = isolated_default_pinned + assert profiles._is_isolated_profile_mode() is True + assert ( + profiles.get_hermes_home_for_profile("default") == env["pinned"] + ) # pre-existing correct behavior (the clamp) + got = workspace._resolve_profile_home_param("default") + assert got == env["pinned"].resolve(), ( + f"_resolve_profile_home_param('default') must reach the isolated-mode " + f"clamp and return the PINNED home {env['pinned']}, got {got}" + ) + assert got != env["base"], "must not resolve to the base/root home" + + def test_get_last_workspace_uses_global_state_not_base(self, isolated_default_pinned, monkeypatch): + """Round 7 contract: 'default' STATE lives in the GLOBAL files only. + + Round-6 follow-up: once the resolver pins config/path resolution at + the isolated home, the canonical-home check in _profile_state_dir + split explicit profile="default" state I/O onto {pinned}/webui_state/ + while ambient calls kept the global dir. Round 7 restores ONE state + authority: literal "default" reads/writes the global files, so this + test now asserts the global binding WINS and the BASE home's + webui_state is never consulted. + """ + env = isolated_default_pinned + # The authoritative GLOBAL binding... + global_ws = env["tmp"] / "srv" / "global-ws" + global_ws.mkdir(parents=True) + global_lw = env["tmp"] / "global-state" + global_lw.mkdir(parents=True) + # Both global state files must move TOGETHER: in production + # _GLOBAL_WS_FILE and _GLOBAL_LW_FILE share one state dir. + monkeypatch.setattr(workspace, "_GLOBAL_WS_FILE", global_lw / "workspaces.json") + monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_lw / "last_workspace.txt") + (global_lw / "last_workspace.txt").write_text(str(global_ws), encoding="utf-8") + + # ...and a POISONED base binding that must NOT win. Point the base + # home's own webui_state at a foreign path to prove no base read. + poisoned_ws = env["tmp"] / "srv" / "poisoned-base-ws" + poisoned_ws.mkdir(parents=True) + base_state = env["base"] / "webui_state" + base_state.mkdir(parents=True) + (base_state / "last_workspace.txt").write_text(str(poisoned_ws), encoding="utf-8") + + got = workspace.get_last_workspace(profile="default") + assert str(got) == str(global_ws.resolve()), ( + f"profile='default' in isolated mode must read the GLOBAL state " + f"authority ({global_lw / 'last_workspace.txt'}), got {got!r}" + ) + + def test_get_last_workspace_global_fallback_only_when_canonically_default( + self, isolated_default_pinned, monkeypatch + ): + """No profile-local state + non-pinned layout → legacy global fallback. + + Literal-default routing to _GLOBAL_LW_FILE is retained OUTSIDE isolated + mode: when the resolved default home IS canonically the default home, + get_last_workspace(profile="default") still reads the global file + (historical behavior, re-gate round 3 contract). + """ + env = isolated_default_pinned + # Disable isolation so 'default' resolves canonically to the base home. + monkeypatch.delenv("HERMES_WEBUI_ISOLATED_PROFILE", raising=False) + monkeypatch.setattr(profiles, "_INITIAL_ISOLATED_PROFILE_OPT_IN", "") + monkeypatch.setattr(profiles, "_INITIAL_HERMES_HOME", str(env["base"])) + global_lw = env["tmp"] / "global-state" + global_lw.mkdir(parents=True) + monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_lw / "last_workspace.txt") + + shared_ws = env["tmp"] / "srv" / "shared-ws" + shared_ws.mkdir(parents=True) + (global_lw / "last_workspace.txt").write_text(str(shared_ws), encoding="utf-8") + + # The conftest autouse cleanup fixture writes TEST_STATE_DIR / + # last_workspace.txt at every teardown; for a canonically-default + # profile that IS the profile-local tier and would shadow the global + # file under test. Pin the profile-local tier to an absent path so the + # GLOBAL fallback is what actually gets exercised. + monkeypatch.setattr( + workspace, + "_last_workspace_file_for_profile", + lambda p=None: env["tmp"] / "absent-webui_state" / "last_workspace.txt", + ) + + got = workspace.get_last_workspace(profile="default") + assert str(got) == str(shared_ws), ( + "outside isolated mode, profile='default' keeps the legacy global " + "fallback (canonical identity retained after the round-6 fix)" + ) + + def test_new_session_binds_global_state_workspace(self, isolated_default_pinned, monkeypatch): + """Session creation with profile='default' honors the global state + authority and never leaks onto the base home (round-7 contract).""" + import api.models as models + + env = isolated_default_pinned + bound_ws = env["tmp"] / "srv" / "bound-session-ws" + bound_ws.mkdir(parents=True) + global_lw = env["tmp"] / "global-state" + global_lw.mkdir(parents=True) + # Both global state files must move TOGETHER: in production + # _GLOBAL_WS_FILE and _GLOBAL_LW_FILE share one state dir. + monkeypatch.setattr(workspace, "_GLOBAL_WS_FILE", global_lw / "workspaces.json") + monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_lw / "last_workspace.txt") + (global_lw / "last_workspace.txt").write_text(str(bound_ws), encoding="utf-8") + + # POISONED base-home binding must not win. + poisoned_ws = env["tmp"] / "srv" / "poisoned-base-ws" + poisoned_ws.mkdir(parents=True) + base_state = env["base"] / "webui_state" + base_state.mkdir(parents=True) + (base_state / "last_workspace.txt").write_text(str(poisoned_ws), encoding="utf-8") + + s = models.new_session(profile="default") + assert s.profile == "default" + assert str(s.workspace) == str(bound_ws.resolve()), ( + f"new_session(profile='default') in isolated mode must bind the " + f"GLOBAL state authority's workspace, got {s.workspace!r}" + ) + assert str(s.workspace) != str(poisoned_ws.resolve()), ( + "must never leak onto the BASE home's webui_state binding" + ) + + def test_session_config_reads_pinned_config( + self, isolated_default_pinned, monkeypatch + ): + """Config for the resolved 'default' home comes from the PINNED home. + + In isolated mode the pinned home IS the active home, so + get_config_for_profile_home takes the ambient-match branch — which is + exactly the authority we must verify: the ambient resolver (and its + HERMES_CONFIG_PATH) must be anchored at the PINNED home, never at the + base/root one. Point the authoritative config INSIDE the pinned home; + had the resolver leaked to the base home, this config would not win. + """ + from api import config as cfg_mod + from api.config import get_config_for_profile_home + + env = isolated_default_pinned + monkeypatch.setenv("HERMES_CONFIG_PATH", str(env["pinned"] / "config.yaml")) + cfg_mod.reload_config() + + home = workspace._resolve_profile_home_param("default") + assert home == env["pinned"].resolve() + got = get_config_for_profile_home(home) + assert got.get("workspace") == "/srv/pinned-workspace", ( + "config for the resolved 'default' home must be anchored at the " + "PINNED profile's home/config, never the base root's" + ) + # The BASE root has no config.yaml — a leak onto the base home would + # have produced the defaults-only dict (no 'workspace' key). + + +class TestRound7DefaultStateRouting: + """Explicit and ambient "default" STATE I/O share ONE authority (round 7). + + Round-6 CORE follow-up: routing the literal "default" name through the + delegated resolver made the canonical-home check in + ``_profile_state_dir`` send EXPLICIT ``profile="default"`` state + reads/writes to ``{pinned}/webui_state/`` while AMBIENT calls kept using + the global state dir — splitting saved workspaces between two + authorities and hiding the pre-upgrade list. Fix under test: + literal-"default" STATE routing stays on the global files, while config + and workspace PATH resolution keep the pinned home. + """ + + def _seed_global_state(self, env, monkeypatch): + """Point both global state files at an isolated tmp dir.""" + g = env["tmp"] / "global-state" + g.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(workspace, "_GLOBAL_WS_FILE", g / "workspaces.json") + monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", g / "last_workspace.txt") + return g + + def test_explicit_default_state_dir_is_global(self, isolated_default_pinned, monkeypatch): + env = isolated_default_pinned + g = self._seed_global_state(env, monkeypatch) + assert workspace._profile_state_dir(profile="default") == g, ( + "explicit profile='default' must route STATE to the global dir " + "(same authority as ambient), not {pinned}/webui_state/" + ) + assert workspace._workspaces_file("default") == g / "workspaces.json" + assert workspace._last_workspace_file("default") == g / "last_workspace.txt" + + def test_explicit_and_ambient_share_workspaces_authority( + self, isolated_default_pinned, monkeypatch + ): + import json + + env = isolated_default_pinned + g = self._seed_global_state(env, monkeypatch) + + # Pre-upgrade saved list lives ONLY in the global file. + pre_dir = env["tmp"] / "srv" / "pre-upgrade" + pre_dir.mkdir(parents=True) + pre = [{"path": str(pre_dir), "name": "Pre"}] + (g / "workspaces.json").write_text( + json.dumps(pre, ensure_ascii=False), encoding="utf-8" + ) + + # Explicit profile="default" reads the pre-upgrade list... + assert workspace.load_workspaces(profile="default") == pre + + # ...and explicit writes land in the SAME global file. + added = env["tmp"] / "srv" / "added" + added.mkdir(parents=True) + updated = pre + [{"path": str(added), "name": "Added"}] + workspace.save_workspaces(updated, profile="default") + assert json.loads((g / "workspaces.json").read_text(encoding="utf-8")) == updated + + # Ambient calls see the identical authority — no split. + assert workspace._profile_state_dir(None) == g + assert workspace.load_workspaces() == updated + + # No divergent webui_state authority may appear under the pinned home. + assert not (env["pinned"] / "webui_state" / "workspaces.json").exists() + assert not (env["pinned"] / "webui_state" / "last_workspace.txt").exists() + + def test_last_workspace_round_trip_shared_explicit_ambient( + self, isolated_default_pinned, monkeypatch + ): + env = isolated_default_pinned + g = self._seed_global_state(env, monkeypatch) + + ws = env["tmp"] / "srv" / "bound-ws" + ws.mkdir(parents=True) + + workspace.set_last_workspace(str(ws), profile="default") + assert (g / "last_workspace.txt").exists(), ( + "set_last_workspace('default') must write the GLOBAL file" + ) + assert not (env["pinned"] / "webui_state" / "last_workspace.txt").exists() + + assert workspace.get_last_workspace(profile="default") == str(ws) + # Ambient (isolated mode's active profile is literally "default") + # resolves to the same binding. + assert workspace.get_last_workspace() == str(ws) diff --git a/tests/test_profile_switch_1200.py b/tests/test_profile_switch_1200.py index 9c3f3801..e1c70e9c 100644 --- a/tests/test_profile_switch_1200.py +++ b/tests/test_profile_switch_1200.py @@ -696,7 +696,7 @@ def test_chat_start_retags_empty_session_to_request_profile(monkeypatch, tmp_pat "_resolve_compatible_session_model_state", lambda model, provider, **_: (model, provider, False), ) - monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None) monkeypatch.setattr(routes, "create_stream_channel", lambda: object()) started_threads = [] @@ -768,7 +768,7 @@ def test_chat_start_does_not_retag_non_empty_session(monkeypatch, tmp_path): "_resolve_compatible_session_model_state", lambda model, provider, **_: (model, provider, False), ) - monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None) monkeypatch.setattr(routes, "create_stream_channel", lambda: object()) class FakeThread: diff --git a/tests/test_provider_mismatch.py b/tests/test_provider_mismatch.py index 27e7a2e1..d22be3b9 100644 --- a/tests/test_provider_mismatch.py +++ b/tests/test_provider_mismatch.py @@ -1059,9 +1059,9 @@ def test_issue1734_chat_start_persists_repaired_codex_provider(monkeypatch): "_resolve_chat_workspace_with_recovery", lambda current, _requested: current.workspace, ) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: value) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: value) monkeypatch.setattr(routes, "_get_session_agent_lock", lambda sid: contextlib.nullcontext()) - monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None) monkeypatch.setattr(routes, "create_stream_channel", lambda: object()) monkeypatch.setattr(routes.threading, "Thread", FakeThread) diff --git a/tests/test_recovered_anchor_settle_duplication.py b/tests/test_recovered_anchor_settle_duplication.py index 10590a44..5dd7253c 100644 --- a/tests/test_recovered_anchor_settle_duplication.py +++ b/tests/test_recovered_anchor_settle_duplication.py @@ -537,7 +537,7 @@ def test_handle_chat_sync_passes_result_turn_authority_to_settlement(tmp_path, m monkeypatch.setattr(routes, "get_session", models.get_session) monkeypatch.setattr(routes, "title_from", models.title_from) monkeypatch.setattr(routes, "get_config", lambda: {"model": "m", "provider": "p"}) - monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path) + monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path) monkeypatch.setattr(routes, "load_settings", lambda: {}) monkeypatch.setattr(routes, "_resolve_cli_toolsets", lambda: []) monkeypatch.setattr(routes, "_agent_runtime_barrier_response", lambda **_k: None) diff --git a/tests/test_remote_terminal_workspace.py b/tests/test_remote_terminal_workspace.py index 2b896ce5..01475fb1 100644 --- a/tests/test_remote_terminal_workspace.py +++ b/tests/test_remote_terminal_workspace.py @@ -1,4 +1,6 @@ from pathlib import Path +import json +import pathlib import pytest @@ -155,7 +157,7 @@ def test_remote_terminal_linux_home_preserves_path_without_macos_synthetic_resol lambda: _remote_config(terminal={"backend": "ssh", "cwd": "/home/developer"}), ) - real_resolve = workspace._resolve_path + real_resolve = workspace._safe_resolve def fake_resolve(p): p_str = str(p) @@ -163,12 +165,1238 @@ def test_remote_terminal_linux_home_preserves_path_without_macos_synthetic_resol return Path(f"/System/Volumes/Data{p_str}") return real_resolve(p) - monkeypatch.setattr(workspace, "_resolve_path", fake_resolve) + monkeypatch.setattr(workspace, "_safe_resolve", fake_resolve) assert workspace.validate_workspace_to_add("/home/developer") == Path("/home/developer") assert workspace.resolve_trusted_workspace("/home/developer") == Path("/home/developer") assert workspace.get_profile_default_workspace() == "/home/developer" + assert workspace._resolve_path("/home/developer") == Path("/home/developer") assert workspace._clean_workspace_list([{"path": "/home/developer", "name": "Dev"}]) == [ {"path": "/home/developer", "name": "Dev"} ] + +def test_session_init_and_created_workspace_preserve_remote_posix_path(monkeypatch): + """Session initialization must not corrupt remote POSIX paths to macOS synthetic firmlinks.""" + monkeypatch.setattr( + api_config, + "get_config", + lambda: _remote_config(terminal={"backend": "ssh", "cwd": "/home/rootson"}), + ) + + from api.models import Session + + s = Session(workspace="/home/rootson", created_workspace="/home/rootson") + assert s.workspace == "/home/rootson" + assert s.created_workspace == "/home/rootson" + + +def test_named_profile_remote_terminal_workspace_candidate_isolated(monkeypatch, tmp_path): + """Remote paths must resolve per profile: remote for named remote profile, not for active local profile.""" + # Active profile is local + monkeypatch.setattr(api_config, "get_config", lambda: {"terminal": {"backend": "local", "cwd": "/Users/local"}}) + + # Named profile 'optiplex' under base home + profiles_dir = tmp_path / "profiles" / "optiplex" + profiles_dir.mkdir(parents=True) + (profiles_dir / "config.yaml").write_text( + "terminal:\n backend: ssh\n cwd: /home/rootson\n", encoding="utf-8" + ) + + from api import profiles + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path) + + # Scoped to named remote profile: candidate is recognized as remote + cand_optiplex = workspace._remote_terminal_workspace_candidate( + "/home/rootson/projects/app", profile="optiplex" + ) + assert cand_optiplex == Path("/home/rootson/projects/app") + + # Scoped to active local profile: candidate is NOT recognized as remote (prevents cross-profile bypass) + cand_local = workspace._remote_terminal_workspace_candidate( + "/home/rootson/projects/app", profile=None + ) + assert cand_local is None + + # Local validation fails on nonexistent path when active profile is local + with pytest.raises(ValueError, match="Path does not exist"): + workspace.validate_workspace_to_add("/home/rootson/projects/app", profile=None) + + # Remote validation succeeds when profile is optiplex + assert workspace.validate_workspace_to_add( + "/home/rootson/projects/app", profile="optiplex" + ) == Path("/home/rootson/projects/app") + + +def test_build_native_multimodal_message_preserves_remote_workspace(monkeypatch, tmp_path): + """Multimodal message builder must scope workspace resolution to the session profile.""" + # Active ambient profile is local + monkeypatch.setattr(api_config, "get_config", lambda: {"terminal": {"backend": "local", "cwd": "/Users/local"}}) + + profiles_dir = tmp_path / "profiles" / "optiplex" + profiles_dir.mkdir(parents=True) + (profiles_dir / "config.yaml").write_text( + "terminal:\n backend: ssh\n cwd: /home/rootson\n", encoding="utf-8" + ) + + from api import profiles, streaming + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path) + + # 1. Profile as logical name string + msg1 = streaming._build_native_multimodal_message( + "[Workspace::v1: /home/rootson]\n", + "hello", + attachments=[], + workspace="/home/rootson", + profile="optiplex", + ) + assert "[Workspace::v1: /home/rootson]" in msg1 + + # 2. Profile as filesystem path string (_profile_home) + msg2 = streaming._build_native_multimodal_message( + "[Workspace::v1: /home/rootson]\n", + "hello", + attachments=[], + workspace="/home/rootson", + profile=str(profiles_dir), + ) + assert "[Workspace::v1: /home/rootson]" in msg2 + + # 3. Profile as Path object + msg3 = streaming._build_native_multimodal_message( + "[Workspace::v1: /home/rootson]\n", + "hello", + attachments=[], + workspace="/home/rootson", + profile=profiles_dir, + ) + assert "[Workspace::v1: /home/rootson]" in msg3 + + +def test_resolve_profile_home_param_formats(tmp_path): + """_resolve_profile_home_param handles None, default, profile name, and Path. + + Round 5 (#7168): path-shaped STRINGS are no longer accepted as explicit + homes — a string is strictly a logical profile id (grammar-checked), so an + explicit home must be expressed as a Path value. + """ + from api.profiles import _DEFAULT_HERMES_HOME + + assert workspace._resolve_profile_home_param(None) == _DEFAULT_HERMES_HOME + assert workspace._resolve_profile_home_param("default") == _DEFAULT_HERMES_HOME + # Path-shaped string: rejected by the logical-id grammar. + with pytest.raises(ValueError): + workspace._resolve_profile_home_param(str(tmp_path / "profiles/custom")) + # Same location as a Path value: honored and canonicalized. + assert workspace._resolve_profile_home_param(tmp_path / "profiles/custom") == ( + tmp_path / "profiles/custom" + ) + + +def test_remote_profile_a_cannot_use_remote_profile_b_cwd(monkeypatch, tmp_path): + """Active remote profile Alice (/srv/remote-alice) cannot validate paths under remote profile Bob (/srv/remote-bob).""" + profiles_alice = tmp_path / "profiles" / "alice" + profiles_alice.mkdir(parents=True) + (profiles_alice / "config.yaml").write_text("terminal:\n backend: ssh\n cwd: /srv/remote-alice\n", encoding="utf-8") + + profiles_bob = tmp_path / "profiles" / "bob" + profiles_bob.mkdir(parents=True) + (profiles_bob / "config.yaml").write_text("terminal:\n backend: ssh\n cwd: /srv/remote-bob\n", encoding="utf-8") + + from api import profiles + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path) + + # Scoped to Alice: Bob's path must be rejected + with pytest.raises(ValueError, match="Path does not exist"): + workspace.validate_workspace_to_add("/srv/remote-bob/project", profile="alice") + + with pytest.raises(ValueError, match="Path does not exist"): + workspace.resolve_trusted_workspace("/srv/remote-bob/project", profile="alice") + + # Scoped to Alice: Alice's own path succeeds + assert workspace.validate_workspace_to_add("/srv/remote-alice/project", profile="alice") == Path("/srv/remote-alice/project") + assert workspace.resolve_trusted_workspace("/srv/remote-alice/project", profile="alice") == Path("/srv/remote-alice/project") + + +def test_local_symlink_resolving_to_system_path_is_rejected(monkeypatch, tmp_path): + """Local symlink pointing to a system directory (/etc) is strictly rejected.""" + # Active profile is local + monkeypatch.setattr(api_config, "get_config", lambda: {"terminal": {"backend": "local", "cwd": str(tmp_path)}}) + monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path / "home") + + symlink_to_etc = tmp_path / "symlink_etc" + try: + symlink_to_etc.symlink_to("/etc") + except OSError: + pytest.skip("Symlink creation requires permissions") + + with pytest.raises(ValueError, match="Path points to a system directory"): + workspace.validate_workspace_to_add(str(symlink_to_etc), profile=None) + + with pytest.raises(ValueError, match="Path points to a system directory"): + workspace.resolve_trusted_workspace(str(symlink_to_etc), profile=None) + + +def test_local_profile_add_workspace_auto_create_on_remote_path_collision(monkeypatch, tmp_path): + """Active local profile adding a path beneath an inactive remote profile's cwd still creates local directory.""" + # Active profile is local + monkeypatch.setattr(api_config, "get_config", lambda: {"terminal": {"backend": "local", "cwd": str(tmp_path)}}) + + # Inactive remote profile 'optiplex' has cwd /tmp/remote-test + profiles_dir = tmp_path / "profiles" / "optiplex" + profiles_dir.mkdir(parents=True) + (profiles_dir / "config.yaml").write_text("terminal:\n backend: ssh\n cwd: /tmp/remote-test\n", encoding="utf-8") + + from api import profiles + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path) + + local_target = tmp_path / "remote-test" / "subproject" + assert not local_target.exists() + + from api.routes import _handle_workspace_add + + class MockHandler: + def __init__(self): + self.response = None + def send_response(self, *args): pass + def send_header(self, *args): pass + def end_headers(self): pass + @property + def wfile(self): + class W: + def write(self, b): pass + return W() + + handler = MockHandler() + monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "default") + monkeypatch.setattr(workspace, "_workspaces_file", lambda: tmp_path / "workspaces.json") + monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path) + + _handle_workspace_add(handler, {"path": str(local_target), "create": True}) + assert local_target.is_dir() + + +def test_detached_streaming_worker_preserves_session_profile_workspace(monkeypatch, tmp_path): + """Detached streaming worker preserves session profile Alice (/srv/remote-alice) even with ambient local profile.""" + # Ambient process profile is local + monkeypatch.setattr(api_config, "get_config", lambda: {"terminal": {"backend": "local", "cwd": "/Users/local"}}) + + profiles_dir = tmp_path / "profiles" / "alice" + profiles_dir.mkdir(parents=True) + (profiles_dir / "config.yaml").write_text("terminal:\n backend: ssh\n cwd: /srv/remote-alice\n", encoding="utf-8") + + from api import profiles, models + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path) + + # Simulate macOS firmlink expansion on host resolve + real_resolve = workspace._safe_resolve + def fake_resolve(p): + p_str = str(p) + if p_str == "/srv/remote-alice" or p_str.startswith("/srv/remote-alice/"): + return Path(f"/System/Volumes/Data{p_str}") + return real_resolve(p) + + monkeypatch.setattr(workspace, "_safe_resolve", fake_resolve) + + s = models.Session(session_id="test1234", workspace="/srv/remote-alice", profile="alice") + assert s.workspace == "/srv/remote-alice" + assert s.created_workspace == "/srv/remote-alice" + + # Streaming workspace update + resolved_ws = workspace._resolve_path("/srv/remote-alice", profile=s.profile) + assert str(resolved_ws) == "/srv/remote-alice" + + from api import streaming + turn_ws = streaming._resolve_path("/srv/remote-alice", profile=getattr(s, "profile", None)) + assert str(turn_ws) == "/srv/remote-alice" + + pytest.importorskip("agent.runtime_cwd", reason="hermes-agent not installed") + tokens = streaming._set_turn_session_identity(s.session_id, workspace=str(turn_ws)) + try: + from agent.runtime_cwd import _SESSION_CWD + assert _SESSION_CWD.get() == "/srv/remote-alice" + finally: + streaming._reset_turn_session_identity(tokens) + + +def test_gateway_multimodal_message_preserves_remote_profile_workspace(monkeypatch, tmp_path): + """Gateway chat multimodal payload builder preserves remote profile workspace containment.""" + # Ambient process profile is local + monkeypatch.setattr(api_config, "get_config", lambda: {"terminal": {"backend": "local", "cwd": "/Users/local"}}) + + profiles_dir = tmp_path / "profiles" / "optiplex" + profiles_dir.mkdir(parents=True) + (profiles_dir / "config.yaml").write_text("terminal:\n backend: ssh\n cwd: /home/rootson\n", encoding="utf-8") + + from api import profiles, streaming, models + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path) + + s = models.Session(session_id="gw_test", workspace="/home/rootson", profile="optiplex") + + msg = streaming._build_native_multimodal_message( + "", + "hello", + attachments=[], + workspace=str(s.workspace), + profile=getattr(s, "profile", None), + ) + assert msg == "hello" + + +def test_isolated_profile_config_read_does_not_mutate_or_read_shared_cache(monkeypatch, tmp_path): + """Explicit profile config lookup must not leak into or read from mutable global config cache.""" + alice_home = tmp_path / "profiles" / "alice" + alice_home.mkdir(parents=True) + (alice_home / "config.yaml").write_text("terminal:\n backend: ssh\n cwd: /srv/remote-alice\n", encoding="utf-8") + + bob_home = tmp_path / "profiles" / "bob" + bob_home.mkdir(parents=True) + (bob_home / "config.yaml").write_text("terminal:\n backend: local\n cwd: /Users/bob\n", encoding="utf-8") + + from api import profiles + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path) + + # NO ambient TLS profile: simulates a detached worker resolving an explicit + # session profile while the process-global active profile is 'default' + # (local). Alice diverges from the ambient target, so her on-disk + # config.yaml is authoritative — regardless of any concurrent reload. + monkeypatch.setattr(api_config, "get_config", lambda: {"terminal": {"backend": "local", "cwd": "/Users/local"}}) + + # 1. alice resolves to remote cwd (divergent profile reads its own disk file) + assert workspace._remote_terminal_cwd("alice") == "/srv/remote-alice" + + # 2. bob resolves to None (local backend) + assert workspace._remote_terminal_cwd("bob") is None + + # 3. Simulate concurrent reload on global _cfg_cache pointing to dirty state + monkeypatch.setattr(api_config, "_cfg_cache", {"terminal": {"backend": "local", "cwd": "/Users/polluted"}}) + + # Explicit lookup of alice still resolves to alice's on-disk terminal cwd, not polluted global cache + assert workspace._remote_terminal_cwd("alice") == "/srv/remote-alice" + assert workspace.validate_workspace_to_add("/srv/remote-alice/sub", profile="alice") == Path("/srv/remote-alice/sub") + + +def test_workspace_routes_profile_isolation(monkeypatch, tmp_path): + """Workspace route actions (add, remove, rename, reorder) must be fully isolated per profile.""" + from api import profiles + from api.routes import ( + _handle_workspace_add, + ) + + alice_home = tmp_path / "profiles" / "alice" + alice_home.mkdir(parents=True) + (alice_home / "config.yaml").write_text("terminal:\n backend: ssh\n cwd: /srv/remote-alice\n", encoding="utf-8") + + bob_home = tmp_path / "profiles" / "bob" + bob_home.mkdir(parents=True) + (bob_home / "config.yaml").write_text("terminal:\n backend: ssh\n cwd: /srv/remote-bob\n", encoding="utf-8") + + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path) + monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path) + + # Ambient get_config() follows the per-request TLS profile exactly as + # production does (server.py sets the cookie context; the config loader + # resolves the active home). With the restored upstream authority rules, + # an ambient-target lookup delegates here — so each simulated request + # must see ITS OWN profile's terminal block. + def _ambient_cfg(): + name = profiles.get_active_profile_name() + if name == "bob": + return {"terminal": {"backend": "ssh", "cwd": "/srv/remote-bob"}} + if name == "alice": + return {"terminal": {"backend": "ssh", "cwd": "/srv/remote-alice"}} + return {"terminal": {"backend": "local", "cwd": str(tmp_path)}} + + monkeypatch.setattr(api_config, "get_config", _ambient_cfg) + + class MockHandler: + def __init__(self): + self.status = 200 + self.data = None + def send_response(self, code): self.status = code + def send_header(self, *args): pass + def end_headers(self): pass + @property + def wfile(self): + class W: + def write(inner_self, b): pass + return W() + + # Add workspace under Alice — TLS profile is alice, so the route's + # get_active_profile_name() == 'alice' and the add writes into alice's + # profile-scoped workspaces.json. + profiles.set_request_profile("alice") + try: + h1 = MockHandler() + _handle_workspace_add(h1, {"path": "/srv/remote-alice/project1", "name": "Project 1"}) + assert len(workspace.load_workspaces(profile="alice")) == 2 + assert any(w["path"] == "/srv/remote-alice/project1" for w in workspace.load_workspaces(profile="alice")) + finally: + profiles.clear_request_profile() + + # Bob's workspaces must be clean and not contain Alice's workspace + profiles.set_request_profile("bob") + try: + h2 = MockHandler() + assert len(workspace.load_workspaces(profile="bob")) == 1 + assert workspace.load_workspaces(profile="bob")[0]["path"] == "/srv/remote-bob" # Default workspace for bob + + _handle_workspace_add(h2, {"path": "/srv/remote-bob/app", "name": "Bob App"}) + bob_ws = workspace.load_workspaces(profile="bob") + assert any(w["path"] == "/srv/remote-bob/app" for w in bob_ws) + assert not any("remote-alice" in w["path"] for w in bob_ws) + finally: + profiles.clear_request_profile() + + # Re-verify Alice's workspaces did not get Bob's workspace + profiles.set_request_profile("alice") + try: + alice_ws = workspace.load_workspaces(profile="alice") + assert len(alice_ws) == 2 + assert any(w["path"] == "/srv/remote-alice/project1" for w in alice_ws) + assert not any("remote-bob" in w["path"] for w in alice_ws) + finally: + profiles.clear_request_profile() + + +def test_saved_workspace_trust_scoped_to_session_profile(monkeypatch, tmp_path): + """resolve_trusted_workspace's saved-workspace branch (B) must honor the explicit profile. + + Greptile #3838260381: when a session profile differs from the ambient + profile, the saved-workspace authorization branch called load_workspaces() + and _resolve_path() WITHOUT the session profile — so a session-owned + external workspace was rejected and a path saved only by the ambient + profile was wrongly trusted instead. + """ + from api import profiles + + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path) + monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path / "home") + (tmp_path / "home").mkdir(parents=True) + + # Session profile 'carol': remote SSH with cwd /srv/remote-carol. + carol_home = tmp_path / "profiles" / "carol" + carol_home.mkdir(parents=True) + (carol_home / "config.yaml").write_text( + "terminal:\n backend: ssh\n cwd: /srv/remote-carol\n", encoding="utf-8" + ) + # Ambient/default: local backend so nothing is remotely classified. + monkeypatch.setattr(api_config, "get_config", lambda: {"terminal": {"backend": "local", "cwd": str(tmp_path)}}) + + # Carol's saved list includes an external (non-home, non-cwd) workspace dir. + external_dir = tmp_path / "data" / "carol-projects" + external_dir.mkdir(parents=True) + ws_file = tmp_path / "state" + monkeypatch.setattr(workspace, "_workspaces_file_for_profile", lambda p=None: ws_file) + ws_file.parent.mkdir(parents=True, exist_ok=True) + ws_file.write_text( + json.dumps([{"path": str(external_dir), "name": "Carol External"}]), + encoding="utf-8", + ) + + # With the session profile passed explicitly, the saved branch trusts it. + resolved = workspace.resolve_trusted_workspace(str(external_dir), profile="carol") + assert resolved == Path(str(external_dir)).resolve() + + # Without the profile (ambient local), the same path is NOT trusted: + # it lives outside home/boot-default and the ambient profile has no such save. + monkeypatch.setattr(workspace, "_workspaces_file_for_profile", lambda p=None: tmp_path / "missing-workspaces.json") + with pytest.raises(ValueError, match="outside the user home directory"): + workspace.resolve_trusted_workspace(str(external_dir)) + + +def test_session_init_simulated_macos_resolution_is_red_on_master_shape(monkeypatch): + """Session.__init__ must preserve /home/ even when host resolve rewrites it. + + Maintainer gate r4 item 3: this test simulates macOS synthetic firmlink + expansion on the HOST resolver. On origin/master's shape + (Path(workspace).expanduser().resolve()) the assertion fails because the + stored workspace becomes /System/Volumes/Data/home/rootson; on this head + (_resolve_path with profile-scoped remote preservation) it passes — i.e. + genuinely RED -> GREEN across the fix boundary. + """ + monkeypatch.setattr( + api_config, + "get_config", + lambda: _remote_config(terminal={"backend": "ssh", "cwd": "/home/rootson"}), + ) + + real_resolve = workspace._safe_resolve + + def fake_resolve(p): + p_str = str(p) + if p_str == "/home/rootson" or p_str.startswith("/home/rootson/"): + return Path(f"/System/Volumes/Data{p_str}") + return real_resolve(p) + + monkeypatch.setattr(workspace, "_safe_resolve", fake_resolve) + + from api.models import Session + + s = Session(workspace="/home/rootson", created_workspace="/home/rootson") + assert s.workspace == "/home/rootson" + assert s.created_workspace == "/home/rootson" + + +def test_local_tilde_and_relative_paths_still_host_resolve(monkeypatch, tmp_path): + """Control: local ~/ paths keep host resolution while remotes are preserved. + + Guards the discriminator BOTH ways (maintainer gate r4): the remote-path + preservation must not swallow ordinary local path semantics — tilde + expansion and symlink normalization still run for local profiles. + """ + real_home = tmp_path / "users" / "local" + projects = real_home / "projects" + projects.mkdir(parents=True) + # A symlink inside home pointing elsewhere-under-home must normalize. + linked = real_home / "linked-projects" + try: + linked.symlink_to(projects) + except OSError: + pytest.skip("Symlink creation requires permissions") + + monkeypatch.setenv("HOME", str(real_home)) + monkeypatch.setattr(api_config, "get_config", lambda: {"terminal": {"backend": "local", "cwd": str(projects)}}) + + # Tilde expansion still resolves against the LOCAL home. + resolved = workspace._resolve_path("~/projects") + assert resolved == projects.resolve() + + # Symlinks under home still collapse to their real target. + resolved_link = workspace._resolve_path("~/linked-projects") + assert resolved_link == projects.resolve() + + +def test_config_authority_active_home_with_external_override(monkeypatch, tmp_path): + """Maintainer blocker 1: an authoritative HERMES_CONFIG_PATH override wins + over target/config.yaml when resolving the ACTIVE profile home.""" + from api import config as cfg, profiles as profiles_mod + import yaml + + active_home = tmp_path / "active-home" + active_home.mkdir() + (active_home / "config.yaml").write_text( + yaml.safe_dump({"terminal": {"backend": "ssh", "cwd": "/srv/from-active-home"}}, sort_keys=False), + encoding="utf-8", + ) + override_dir = tmp_path / "override-dir" + override_dir.mkdir() + override_cfg = override_dir / "config.yaml" + override_cfg.write_text( + yaml.safe_dump({"terminal": {"backend": "docker", "cwd": "/srv/from-override"}}, sort_keys=False), + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_CONFIG_PATH", str(override_cfg)) + monkeypatch.setattr(profiles_mod, "get_active_hermes_home", lambda: active_home) + cfg.reload_config() + + result = cfg.get_config_for_profile_home(active_home) + assert result.get("terminal", {}).get("cwd") == "/srv/from-override" + # NOTE: no manual delenv/reload here — monkeypatch restores + # HERMES_CONFIG_PATH at teardown, and the stale (_cfg_path vs pinned) + # check invalidates the cache on the next reader. A trailing + # reload_config() would bake this test's override into the global cache + # and pollute later tests (#7168 review triage). + + +def test_divergent_profile_without_config_yaml_stays_isolated(monkeypatch, tmp_path): + """Maintainer blocker 2: an existing divergent named-profile directory with + NO config.yaml must NOT inherit the ambient default's terminal config.""" + from api import config as cfg, profiles as profiles_mod + + default_home = tmp_path / "default-home" + default_home.mkdir(parents=True) + (default_home / "config.yaml").write_text( + "terminal:\n backend: ssh\n cwd: /srv/remote-default\n", + encoding="utf-8", + ) + # Fresh non-cloned named profile: exists on disk, no config.yaml yet. + fresh_profile_home = tmp_path / "profiles" / "fresh" + fresh_profile_home.mkdir(parents=True) + + monkeypatch.setenv("HERMES_CONFIG_PATH", str(default_home / "config.yaml")) + monkeypatch.setattr(profiles_mod, "get_active_hermes_home", lambda: default_home) + cfg.reload_config() + + result = cfg.get_config_for_profile_home(fresh_profile_home) + # Defaults applied, but crucially NOT the ambient default profile's terminal block. + assert result.get("terminal") != {"backend": "ssh", "cwd": "/srv/remote-default"} + assert not result.get("terminal", {}).get("cwd") + + # And a divergent home that doesn't exist at all still returns {}. + assert cfg.get_config_for_profile_home(tmp_path / "profiles" / "ghost") == {} + # NOTE: no manual delenv/reload here (see test_config_authority_active_home_with_external_override) — + # monkeypatch teardown + path-change invalidation keep the global cache clean. + + +def test_suggestions_scoped_to_explicit_profile_saved_roots(tmp_path): + """Sweep regression (#7168 review): list_workspace_suggestions with an + explicit profile must widen trust only via THAT profile's saved + workspaces — a workspace saved under a different profile stays out.""" + other_dir = tmp_path / "other-profile-dir" + other_dir.mkdir(parents=True) + mine = tmp_path / "mine" + mine.mkdir() + + ws_file = workspace._workspaces_file_for_profile("alice") + ws_file.parent.mkdir(parents=True, exist_ok=True) + ws_file.write_text(json.dumps([{"path": str(other_dir)}]), encoding="utf-8") + + prefix = str(tmp_path) + "/" + suggestions = workspace.list_workspace_suggestions(prefix, profile="alice") + assert any(s.endswith("other-profile-dir") for s in suggestions) + + # A different profile's saved list does NOT include alice's external dir. + suggestions_bob = workspace.list_workspace_suggestions(prefix, profile="bob") + assert not any(s.endswith("other-profile-dir") for s in suggestions_bob) + assert mine.exists() # control: fixture dir is real, absence is scoping + + +def test_recovery_helper_threads_profile_to_trust_and_fallback(tmp_path): + """Sweep regression (#7168 review): resolve_implicit_workspace_with_recovery + passes the profile into both trust resolution and the recovery fallback, + so a deleted session workspace recovers to a fallback trusted for THAT + profile — never via another profile's saved list.""" + external = tmp_path / "external-carol" + external.mkdir() + stale = tmp_path / "stale-workspace" + fallback_dir = tmp_path / "fallback" + fallback_dir.mkdir() + + ws_file = workspace._workspaces_file_for_profile("carol") + ws_file.parent.mkdir(parents=True, exist_ok=True) + ws_file.write_text( + json.dumps([{"path": str(external)}, {"path": str(fallback_dir)}]), + encoding="utf-8", + ) + + resolved, recovered = workspace.resolve_implicit_workspace_with_recovery( + str(stale), + lambda: str(fallback_dir), + profile="carol", + ) + assert recovered is True + assert resolved == fallback_dir.resolve() + + # Under a DIFFERENT profile neither the stale candidate nor the recovery + # fallback may gain trust via carol's saved list. + with pytest.raises(ValueError): + workspace.resolve_implicit_workspace_with_recovery( + str(stale), + lambda: str(fallback_dir), + profile="not-carol", + ) + + + +def test_clean_workspace_list_explicit_profile_keeps_own_entries(monkeypatch, tmp_path): + """_clean_workspace_list must define 'own profile dir' by the explicit profile. + + Maintainer re-gate 2026-08-25 (CORE / data-loss): with an explicit + profile="alice", own_profile_dir was still derived from the AMBIENT home. + Loading Alice's saved list while ambient Bob was active dropped Alice's own + workspace (it sat under Alice's home, not Bob's) and load_workspaces() + persisted the emptied list back to disk — silent destruction of a + profile's saved workspaces. + + Regression: Alice request / Bob ambient — Alice's own entry survives and + is persisted intact; a genuinely foreign profile path is still removed. + """ + from api import profiles + + # _clean_workspace_list derives the profiles root as _home_path()/'.hermes'/profiles. + monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path) + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path / ".hermes") + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path / ".hermes") + (tmp_path / ".hermes" / "profiles").mkdir(parents=True) + + alice_home = tmp_path / ".hermes" / "profiles" / "alice" + alice_home.mkdir(parents=True) + bob_home = tmp_path / ".hermes" / "profiles" / "bob" + bob_home.mkdir(parents=True) + + # Alice owns a workspace that lives inside her OWN profile directory. + alice_own = alice_home / "projects" + alice_own.mkdir(parents=True) + # A foreign entry pointing into BOB's profile dir must still be pruned. + foreign = bob_home / "secret" + foreign.mkdir(parents=True) + + raw = [ + {"path": str(alice_own), "name": "AliceProjects"}, + {"path": str(foreign), "name": "BobLeak"}, + ] + + cleaned = workspace._clean_workspace_list(raw, profile="alice") + + paths = [w["path"] for w in cleaned] + assert str(alice_own) in paths, ( + "explicit-profile cleaning must keep the target profile's OWN entry" + ) + assert not any("bob" in p for p in paths), "cross-profile leak must still be pruned" + + # End-to-end through the loader: with ambient Bob active, loading ALICE's + # list must neither drop nor rewrite-away her own entry. + monkeypatch.setattr( + profiles, "get_active_profile_name", lambda: "bob", raising=False + ) + ws_file = tmp_path / "state" / "workspaces.json" + monkeypatch.setattr(workspace, "_workspaces_file_for_profile", lambda p=None: ws_file) + ws_file.parent.mkdir(parents=True, exist_ok=True) + ws_file.write_text(json.dumps(raw), encoding="utf-8") + + loaded = workspace.load_workspaces(profile="alice") + loaded_paths = [w["path"] for w in loaded] + assert str(alice_own) in loaded_paths + on_disk = json.loads(ws_file.read_text(encoding="utf-8")) + assert any(w["path"] == str(alice_own) for w in on_disk), ( + "persisted list must retain Alice's own workspace" + ) + + +def test_stale_workspace_recovery_scoped_to_explicit_profile(monkeypatch, tmp_path): + """Recovery classification/probe/fallback must honor the explicit profile. + + Maintainer re-gate 2026-08-25 (CORE): resolve_implicit_workspace_with_recovery + classified the backend via ambient get_config(), probed candidate paths + without profile=, and fell back to an unbound getter. With profile="alice" + under ambient Bob it returned BOB's last workspace as recovered=True, and an + Alice remote backend without terminal.cwd was misclassified as local. + + Regression A (remote misclassification): alice is ssh WITHOUT cwd, ambient + is local — recovery must fail closed preserving the original error. + Regression B (cross-profile bind): alice local, stored path missing, + fallback=get_last_workspace — must never return Bob's path nor claim + recovered=True for it. + """ + from api import profiles + + monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path) + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path / ".hermes") + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path / ".hermes") + (tmp_path / ".hermes" / "profiles").mkdir(parents=True) + + alice_home = tmp_path / ".hermes" / "profiles" / "alice" + alice_home.mkdir(parents=True) + bob_home = tmp_path / ".hermes" / "profiles" / "bob" + bob_home.mkdir(parents=True) + + def _ambient_cfg(): + # Ambient context belongs to Bob: LOCAL backend. + return {"terminal": {"backend": "local"}} + + monkeypatch.setattr(api_config, "get_config", _ambient_cfg) + + # ── A: remote backend without terminal.cwd fails closed ── + (alice_home / "config.yaml").write_text( + "terminal:\n backend: ssh\n", encoding="utf-8" + ) + missing_remote = tmp_path / "gone" / "remote-ws" + + with pytest.raises(ValueError): + workspace.resolve_implicit_workspace_with_recovery( + str(missing_remote), + None, + profile="alice", + ) + + # ── B: missing-path fallback can never bind another profile's workspace ── + (alice_home / "config.yaml").write_text( + "terminal:\n backend: local\n", encoding="utf-8" + ) + stored_missing = tmp_path / "data" / "vanished" + bob_ws = tmp_path / "srv" / "bobs-place" + bob_ws.mkdir(parents=True) + alice_ws = tmp_path / "srv" / "alice-own" + alice_ws.mkdir(parents=True) + + # Alice's SAVED list contains her own external workspace (so it is trusted), + # while Bob's list contains his. The saved-workspace trust branch is itself + # profile-scoped (earlier fix), so each name resolves only its own list. + alice_list = tmp_path / "state-alice" / "workspaces.json" + bob_list = tmp_path / "state-bob" / "workspaces.json" + alice_list.parent.mkdir(parents=True, exist_ok=True) + bob_list.parent.mkdir(parents=True, exist_ok=True) + alice_list.write_text(json.dumps([{"path": str(alice_ws), "name": "AliceOwn"}]), encoding="utf-8") + bob_list.write_text(json.dumps([{"path": str(bob_ws), "name": "BobWs"}]), encoding="utf-8") + + def _ws_file_for(p=None): + return alice_list if p is not None and str(p) == "alice" else bob_list + + monkeypatch.setattr(workspace, "_workspaces_file_for_profile", _ws_file_for) + + def _last_for(p=None): + if p is not None: + # Profile-aware getter: only ever returns THAT profile's state. + if str(p) in ("alice",): + return str(alice_ws) + return str(bob_ws) + return str(bob_ws) # unscoped/ambient getter leaks Bob's path + + resolved, recovered = workspace.resolve_implicit_workspace_with_recovery( + str(stored_missing), + _last_for, + profile="alice", + ) + assert recovered is True + assert resolved == alice_ws.resolve(), ( + "recovery under profile='alice' must bind Alice's own workspace" + ) + assert "bobs-place" not in str(resolved), ( + "recovery under profile='alice' must never return Bob's workspace" + ) + +def test_get_last_workspace_named_profile_never_reads_global_file(monkeypatch, tmp_path): + """get_last_workspace(profile=...) must not fall back to the global file. + + Maintainer re-gate round 3 (CORE): for a named profile the getter still + consulted _GLOBAL_LW_FILE after missing the profile-local file, so a direct + probe returned Bob's globally-recorded workspace for profile="alice". + The legacy global fallback is now allowed ONLY for the root/default + profile. Uses the REAL getter with a real global-file fixture. + """ + from api import profiles + + monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path) + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path / ".hermes") + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path / ".hermes") + (tmp_path / ".hermes" / "profiles").mkdir(parents=True) + (tmp_path / ".hermes" / "profiles" / "alice").mkdir(parents=True) + + bobs_dir = tmp_path / "srv" / "bobs-global" + bobs_dir.mkdir(parents=True) + + # The legacy GLOBAL last-workspace file records BOB's workspace. + global_lw = tmp_path / "global-state" + global_lw.mkdir(parents=True) + (global_lw / "last_workspace.txt").write_text(str(bobs_dir), encoding="utf-8") + monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_lw / "last_workspace.txt") + + # Alice has NO profile-local last-workspace file at all. + alice_state = tmp_path / ".hermes" / "profiles" / "alice" / "webui_state" + + def _lw_file_for(p=None): + if p is not None and str(p) == "alice": + return alice_state / "last_workspace.txt" + return global_lw / "last_workspace.txt" + + monkeypatch.setattr(workspace, "_last_workspace_file_for_profile", _lw_file_for) + + # Direct probe: named profile must NOT inherit Bob's global binding. + got = workspace.get_last_workspace(profile="alice") + assert "bobs-global" not in str(got), ( + "named profile must never fall back to the global last-workspace file" + ) + + # Control: the root/default profile KEEPS the historical global fallback. + monkeypatch.setattr( + profiles, + "_resolve_profile_home_param", + lambda p: tmp_path / ".hermes" if p is None or str(p) in ("default",) else _lw_file_for(p).parent.parent.parent, + raising=False, + ) + got_default = workspace.get_last_workspace(profile="default") + assert str(got_default) == str(bobs_dir), ( + "root/default profile retains the global last-workspace fallback" + ) + + +def test_new_session_binds_explicit_profile_not_ambient_last_workspace(monkeypatch, tmp_path): + """new_session()/import_cli_session() must scope their fallback getter. + + Maintainer re-gate round 3 (CORE): both Session constructions fell back to + get_last_workspace() WITHOUT the profile even though profile=profile was + set on the same Session — an Alice session created under ambient Bob bound + to BOB's workspace. Uses the real getters and real state files. + """ + import api.models as models + from api import profiles + + monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path) + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", tmp_path / ".hermes") + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: tmp_path / ".hermes") + (tmp_path / ".hermes" / "profiles").mkdir(parents=True) + (tmp_path / ".hermes" / "profiles" / "alice").mkdir(parents=True) + + bobs_ws = tmp_path / "srv" / "bobs-workspace" + bobs_ws.mkdir(parents=True) + alice_ws = tmp_path / "srv" / "alice-workspace" + alice_ws.mkdir(parents=True) + + # Ambient/global last-workspace points at BOB's directory. + global_lw = tmp_path / "global-state" + global_lw.mkdir(parents=True) + (global_lw / "last_workspace.txt").write_text(str(bobs_ws), encoding="utf-8") + monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_lw / "last_workspace.txt") + + # Alice's OWN profile-local last-workspace points at her own directory. + alice_state = tmp_path / ".hermes" / "profiles" / "alice" / "webui_state" + alice_state.mkdir(parents=True) + (alice_state / "last_workspace.txt").write_text(str(alice_ws), encoding="utf-8") + + def _lw_file_for(p=None): + if p is not None and str(p) == "alice": + return alice_state / "last_workspace.txt" + return global_lw / "last_workspace.txt" + + monkeypatch.setattr(workspace, "_last_workspace_file_for_profile", _lw_file_for) + monkeypatch.setattr(models, "get_last_workspace", workspace.get_last_workspace) + + s = models.new_session(profile="alice") + assert str(s.workspace) == str(alice_ws), ( + "new_session under profile='alice' must bind ALICE's last workspace, " + f"got {s.workspace!r}" + ) + + imported = models.import_cli_session( + session_id="cli-import-alice-1", + title="t", + messages=[], + profile="alice", + ) + assert str(imported.workspace) == str(alice_ws), ( + "import_cli_session under profile='alice' must bind ALICE's last " + f"workspace, got {imported.workspace!r}" + ) + + +# ── Re-gate round 4 (#7168): malformed names, aliases, unwired boundaries ──── + + +class TestRound4ProfileIsolation: + """Re-gate round 4: the four remaining profile-isolation instances.""" + + def test_invalid_profile_name_never_reads_default_state(self, monkeypatch, tmp_path): + """A malformed name must NOT be clamped onto the default home. + + Old behavior: _resolve_profile_home_param('bad name') returned + _DEFAULT_HERMES_HOME, so get_last_workspace(profile='bad name') read + (and could fall back to) the DEFAULT profile's global state file. + """ + import api.models as models # noqa: F401 (import parity with R3 tests) + from api import profiles + + default_home = tmp_path / ".hermes" + default_home.mkdir(parents=True) + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", default_home) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: default_home) + + global_lw = tmp_path / "global-state" + global_lw.mkdir(parents=True) + poisoned = tmp_path / "poisoned-workspace" + poisoned.mkdir() + (global_lw / "last_workspace.txt").write_text(str(poisoned), encoding="utf-8") + monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_lw / "last_workspace.txt") + + # The invalid name resolves to NOTHING state-like and raises on resolve. + with pytest.raises(ValueError): + workspace._resolve_profile_home_param("bad name") + + # Getter: never returns the poisoned default/global binding. + got = workspace.get_last_workspace(profile="bad name") + assert str(got) != str(poisoned) + + # Setter: writes no file anywhere under the default home. + workspace.set_last_workspace(str(poisoned), profile="bad name") + assert not (default_home / "webui_state").exists() + + # save_workspaces refuses outright. + with pytest.raises(ValueError): + workspace.save_workspaces([{"path": str(poisoned)}], profile="bad name") + + def test_symlink_alias_of_default_home_keeps_legacy_fallback(self, monkeypatch, tmp_path): + """A symlink alias of _DEFAULT_HERMES_HOME is canonically the DEFAULT. + + Old behavior compared lexically: alias != _DEFAULT_HERMES_HOME, so the + default profile accessed through its alias wrongly LOST the legacy + global fallback. With canonical identity it keeps it. + """ + from api import profiles + + real_home = tmp_path / "real-hermes" + real_home.mkdir(parents=True) + alias = tmp_path / "alias-hermes" + try: + alias.symlink_to(real_home, target_is_directory=True) + except OSError: + pytest.skip("symlinks unavailable in this environment") + + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", real_home) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: real_home) + + shared_ws = tmp_path / "shared-workspace" + shared_ws.mkdir() + # The default profile's state dir IS the global STATE_DIR — pin both + # the legacy global fallback file and the profile-scoped file path so + # no real ~/.hermes state leaks into the assertion. + global_state = tmp_path / "global-state" + global_state.mkdir(parents=True) + (global_state / "last_workspace.txt").write_text(str(shared_ws), encoding="utf-8") + monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_state / "last_workspace.txt") + monkeypatch.setattr( + workspace, + "_GLOBAL_WS_FILE", + global_state / "workspaces.json", + ) + monkeypatch.setattr(workspace, "_STATE_DIR_OVERRIDE", str(global_state), raising=False) + monkeypatch.setattr( + workspace, + "_last_workspace_file_for_profile", + lambda p=None: ( + global_state / "last_workspace.txt" + if p is None or workspace._is_default_profile_home( + workspace._resolve_profile_home_param(p) + ) + else tmp_path / "other" / "webui_state" / "last_workspace.txt" + ), + raising=False, + ) + + assert workspace._is_default_profile_home(alias), ( + "symlink alias must compare canonically equal to the default home" + ) + + # No profile-local last_workspace.txt exists → legacy global fallback applies. + # Explicit home is expressed as a Path value (#7168 re-gate round 5: + # path-shaped STRINGS are rejected profile ids). + got = workspace.get_last_workspace(profile=alias) + assert str(got) == str(shared_ws) + + def test_set_last_workspace_threads_session_profile(self, monkeypatch, tmp_path): + """set_last_workspace call sites write the SESSION's profile state file. + + Repro of review finding #1: a server-initiated turn for named-profile + session Alice wrote only the GLOBAL file, so Alice fell back to her + configured workspace while default incorrectly read Alice's selection. + """ + alice_state = tmp_path / "profiles" / "alice" / "webui_state" + alice_state.mkdir(parents=True) + + lw_paths = {} + real_set = workspace.set_last_workspace + + def spy(path, profile=None): + lw = workspace._last_workspace_file_for_profile(profile) + lw_paths["profile"] = profile + lw_paths["file"] = str(lw) + return real_set(path, profile=profile) + + # Pin the resolver so 'alice' maps into the fixture tree (the product + # code path under test is the routes.py call shape, not profiles lookup). + monkeypatch.setattr( + workspace, + "_resolve_profile_home_param", + lambda p: tmp_path / "profiles" / (p or "default"), + raising=False, + ) + + # Simulate exactly what the three routes.py sites now do: + s_like = type("S", (), {"profile": "alice"})() + spy("/srv/alice-ws", profile=getattr(s_like, "profile", None)) + + assert lw_paths["profile"] == "alice" + assert pathlib.PurePath(lw_paths["file"]).name == "last_workspace.txt" + assert "alice" in lw_paths["file"], ( + f"session-profile write must land in ALICE's webui_state, got {lw_paths['file']!r}" + ) + assert not (tmp_path / "global").exists() + + def test_cli_projection_uses_own_cli_profile(self, monkeypatch, tmp_path): + """The all-profile CLI sidebar projection scopes its workspace probe. + + Review finding #3: _load_cli_sessions_uncached had _cli_profile in + scope but called get_last_workspace() ambiently, producing + profile='alice', workspace=''. + """ + import api.models as models + + calls = [] + + def fake_getlw(profile=None): + calls.append(profile) + return f"workspace-for:{profile}" + + monkeypatch.setattr(models, "get_last_workspace", fake_getlw) + monkeypatch.setattr(models, "get_claude_code_sessions", lambda: []) + monkeypatch.setattr(models, "read_importable_agent_session_rows", lambda *a, **k: [ + { + "id": "tui-row-alice", + "title": "TUI Session", + "model": "test-model", + "source": "tui", + "raw_source": "tui", + "message_count": 1, + "actual_message_count": 1, + "actual_user_message_count": 1, + "last_activity": 10.0, + "started_at": 9.0, + } + ]) + monkeypatch.setattr(models, "_profile_has_user_projects", lambda: False) + monkeypatch.setattr(models, "ensure_cron_project", lambda **_: None) + monkeypatch.setattr(models.Session, "load_metadata_only", lambda _sid: None) + + db = tmp_path / "state.db" + db.write_text("", encoding="utf-8") + rows = models._load_cli_sessions_uncached( + tmp_path, db, _cli_profile="alice" + ) + assert any(r.get("session_id") == "tui-row-alice" for r in rows), rows + assert calls and calls[0] == "alice", ( + f"_cli_workspace() must consult get_last_workspace(profile=_cli_profile); " + f"profiles seen: {calls}" + ) + alice_rows = [r for r in rows if r.get("workspace") == "workspace-for:alice"] + assert alice_rows, ( + f"projected row must carry the ALICE-scoped workspace; got " + f"{[r.get('workspace') for r in rows]}" + ) + + +# ── Re-gate round 5 (#7168): traversal gate + symlinked config authority ───── + + +class TestRound5TraversalAndSymlinkAuthority: + """Round 5: path-shaped profile STRINGS are rejected ids; config authority + survives a symlinked HERMES_HOME.""" + + def test_path_shaped_profile_string_rejected_no_traversal_write( + self, monkeypatch, tmp_path + ): + """'../evil' as a STRING profile must be rejected by the id grammar. + + Repro of review finding #1 (CORE / security): the old resolver treated + any string containing '/' or '\\' as an explicit home and resolved it + directly BEFORE the grammar check, so a malformed profile value like + ``../evil`` bypassed validation entirely and could read AND overwrite + an arbitrary ``webui_state/last_workspace.txt`` outside the profile + boundary. Real getters + real state files throughout. + """ + from api import profiles + + default_home = tmp_path / ".hermes" + default_home.mkdir(parents=True) + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", default_home) + monkeypatch.setattr(profiles, "_resolve_base_hermes_home", lambda: default_home) + + # The file the traversal WOULD reach if it escaped the boundary. + evil_dir = tmp_path / "evil" + evil_state = evil_dir / "webui_state" + evil_state.mkdir(parents=True) + victim = evil_state / "last_workspace.txt" + victim.write_text("/original-victim-binding", encoding="utf-8") + + traversal = "../evil" + + # Grammar gate: a path-shaped STRING is not a valid logical id. + with pytest.raises(ValueError): + workspace._resolve_profile_home_param(traversal) + + # Getter: reads NOTHING from the traversal target. + got = workspace.get_last_workspace(profile=traversal) + assert got != "/original-victim-binding" + + # Setter: writes NO file anywhere along the traversal. + workspace.set_last_workspace("/attacker-chosen", profile=traversal) + assert victim.read_text(encoding="utf-8") == "/original-victim-binding" + assert sorted(p.name for p in evil_state.iterdir()) == ["last_workspace.txt"], ( + "traversal profile must not create or modify any state file" + ) + + # Saved-list writer refuses outright too. + with pytest.raises(ValueError): + workspace.save_workspaces([{"path": str(evil_dir)}], profile=traversal) + + # Same location delivered as a REAL Path is still honored — explicit + # homes move to Path values; they are canonicalized, not banned. + assert workspace._resolve_profile_home_param(Path(traversal)) == Path( + traversal + ).expanduser().resolve() + + def test_symlinked_home_with_authoritative_config_path_keeps_authority( + self, monkeypatch, tmp_path + ): + """HERMES_HOME via symlink + authoritative HERMES_CONFIG_PATH. + + Repro of review finding #2 (CORE): workspace canonicalization hands + get_config_for_profile_home the RESOLVED alias target, but the config + authority compared the active home and the config parent LEXICALLY. + With HERMES_HOME reached through a symlink alias, both identity checks + failed and the function performed a DIRECT read of the symlink + target's own (wrong) config.yaml instead of deferring to the + authoritative HERMES_CONFIG_PATH — master trusts the remote + terminal.cwd from the override; the lexical head rejected it. + Canonicalize all three sides before comparing. + """ + import os + + import yaml + + from api import profiles + + real_home = tmp_path / "real-hermes" + real_home.mkdir(parents=True) + alias_home = tmp_path / "alias-hermes" + try: + alias_home.symlink_to(real_home, target_is_directory=True) + except OSError: + pytest.skip("symlinks unavailable in this environment") + + # The symlink target's OWN config is the WRONG one a naive direct + # read picks up (no remote cwd — exactly what "rejects the remote + # path" looks like downstream). + (real_home / "config.yaml").write_text( + yaml.safe_dump( + {"terminal": {"backend": "local", "cwd": "/srv/from-symlink-target"}}, + sort_keys=False, + ), + encoding="utf-8", + ) + # The AUTHORITATIVE config lives OUTSIDE the home, behind + # HERMES_CONFIG_PATH (maintainer-blocker-1 semantics). + override_dir = tmp_path / "override-dir" + override_dir.mkdir() + override_cfg = override_dir / "config.yaml" + override_cfg.write_text( + yaml.safe_dump( + {"terminal": {"backend": "ssh", "cwd": "/srv/from-override"}}, + sort_keys=False, + ), + encoding="utf-8", + ) + + # Ambient resolver reports the home under its ALIAS spelling (as an + # env-provided HERMES_HOME would); callers pass the canonicalized + # target — the exact divergence the lexical comparison mishandled. + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: alias_home) + monkeypatch.setenv("HERMES_CONFIG_PATH", str(override_cfg)) + api_config.reload_config() + + got = api_config.get_config_for_profile_home(real_home) + assert got.get("terminal", {}).get("cwd") == "/srv/from-override", ( + "canonicalized alias target must preserve HERMES_CONFIG_PATH " + f"authority; got {got.get('terminal')!r}" + ) + assert got.get("terminal", {}).get("backend") == "ssh" + + # The un-canonicalized alias spelling converges to the SAME answer — + # identity is by resolution, never by lexical spelling. + got_alias = api_config.get_config_for_profile_home(alias_home) + assert got_alias.get("terminal", {}).get("cwd") == "/srv/from-override" + + # Sanity: the fixture genuinely exercises alias-vs-resolved-target + # comparison rather than passing trivially. + assert os.path.realpath(alias_home) == str(real_home) + assert os.path.realpath(real_home) != str(alias_home) + # NOTE: no manual delenv/reload here (see + # test_config_authority_active_home_with_external_override) — + # monkeypatch teardown + stale-path check invalidate the cache. diff --git a/tests/test_sprint5.py b/tests/test_sprint5.py index a28f774f..a54ef090 100644 --- a/tests/test_sprint5.py +++ b/tests/test_sprint5.py @@ -117,7 +117,7 @@ def test_workspace_suggest_preserves_tilde_prefix(monkeypatch, tmp_path): child = home / "Projects" child.mkdir(parents=True) monkeypatch.setenv("HOME", str(home)) - monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda: [home.resolve()]) + monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda *a, **kw: [home.resolve()]) suggestions = workspace.list_workspace_suggestions("~/") @@ -132,7 +132,7 @@ def test_workspace_suggest_preserves_tilde_prefix_for_partial_child(monkeypatch, child = home / "Projects" child.mkdir(parents=True) monkeypatch.setenv("HOME", str(home)) - monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda: [home.resolve()]) + monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda *a, **kw: [home.resolve()]) suggestions = workspace.list_workspace_suggestions("~/Pro") @@ -152,7 +152,7 @@ def test_workspace_suggest_expands_tilde_when_home_is_symlink(monkeypatch, tmp_p pytest.skip(f"symlinks are not available in this environment: {exc}") monkeypatch.setenv("HOME", str(link_home)) - monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda: [actual_home.resolve()]) + monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda *a, **kw: [actual_home.resolve()]) suggestions = workspace.list_workspace_suggestions("~/Doc") @@ -166,7 +166,7 @@ def test_workspace_suggest_keeps_absolute_prefix_absolute(monkeypatch, tmp_path) child = home / "Projects" child.mkdir(parents=True) monkeypatch.setenv("HOME", str(home)) - monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda: [home.resolve()]) + monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda *a, **kw: [home.resolve()]) suggestions = workspace.list_workspace_suggestions(str(home) + "/Pro") diff --git a/tests/test_stale_stream_cleanup.py b/tests/test_stale_stream_cleanup.py index 2bd34e40..a41664f3 100644 --- a/tests/test_stale_stream_cleanup.py +++ b/tests/test_stale_stream_cleanup.py @@ -134,7 +134,7 @@ def test_chat_start_rechecks_active_stream_under_session_lock(monkeypatch, tmp_p monkeypatch.setattr(routes, "_get_session_agent_lock", lambda sid: MutatingSessionLock()) monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})()) - monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None) monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue()) monkeypatch.setattr(routes.threading, "Thread", NoopThread) @@ -199,7 +199,7 @@ def test_chat_start_blocks_same_session_active_run_after_cancel_clears_stream_id return None monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})()) - monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None) monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue()) monkeypatch.setattr(routes.threading, "Thread", NoopThread) @@ -257,7 +257,7 @@ def test_chat_start_allows_same_session_after_active_run_unregisters(monkeypatch return None monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})()) - monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None) monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue()) monkeypatch.setattr(routes.threading, "Thread", NoopThread) @@ -332,7 +332,7 @@ def test_chat_start_not_permanently_blocked_by_stale_active_run(monkeypatch, tmp return None monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})()) - monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None) monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue()) monkeypatch.setattr(routes.threading, "Thread", NoopThread) diff --git a/tests/test_webhook_project_sessions.py b/tests/test_webhook_project_sessions.py index fc60b6e3..0370e059 100644 --- a/tests/test_webhook_project_sessions.py +++ b/tests/test_webhook_project_sessions.py @@ -108,7 +108,7 @@ def test_webhook_rows_get_webhook_project_id(monkeypatch, tmp_path): db.write_text("", encoding="utf-8") monkeypatch.setattr(models, "read_importable_agent_session_rows", lambda *_a, **_kw: [_agent_row()]) - monkeypatch.setattr(models, "get_last_workspace", lambda: tmp_path) + monkeypatch.setattr(models, "get_last_workspace", lambda profile=None: tmp_path) monkeypatch.setattr(models, "ensure_cron_project", lambda: "cron-project-id") monkeypatch.setattr(models, "ensure_webhook_project", lambda: "webhook-project-id", raising=False) monkeypatch.setattr(models.Session, "load_metadata_only", lambda _sid: None) @@ -137,7 +137,7 @@ def test_webhook_second_pass_keeps_older_project_rows_available(monkeypatch, tmp return [] monkeypatch.setattr(models, "read_importable_agent_session_rows", fake_read_rows) - monkeypatch.setattr(models, "get_last_workspace", lambda: tmp_path) + monkeypatch.setattr(models, "get_last_workspace", lambda profile=None: tmp_path) monkeypatch.setattr(models, "ensure_cron_project", lambda: "cron-project-id") monkeypatch.setattr(models, "ensure_webhook_project", lambda: "webhook-project-id", raising=False) monkeypatch.setattr(models.Session, "load_metadata_only", lambda _sid: None) diff --git a/tests/test_workspace_upload.py b/tests/test_workspace_upload.py index b71790ff..731b9893 100644 --- a/tests/test_workspace_upload.py +++ b/tests/test_workspace_upload.py @@ -136,7 +136,7 @@ def _configure_direct_office_upload(monkeypatch, tmp_path): monkeypatch.setattr(upload, "get_session", lambda _sid: session) monkeypatch.setattr(upload, "_reject_invisible_session", lambda *_args: False) - monkeypatch.setattr(upload, "resolve_trusted_workspace", lambda path: path) + monkeypatch.setattr(upload, "resolve_trusted_workspace", lambda path, **_kw: path) return upload, office_documents, workspace