"""Per-turn terminal scope: profile-scoped TERMINAL_* policy. Multiplexed surfaces (gateway, dashboard/TUI, cron) serve several profiles from one process; mirroring terminal settings into ``os.environ`` let the first profile pin its backend onto everyone else (sandbox escape). Like ``agent/secret_scope.py``, a ContextVar holds the active profile's COMPLETE ``TERMINAL_*`` policy; while bound, ``terminal_env`` resolves ONLY from it (omitted keys -> defined default, never ambient env). If the policy cannot be resolved a *refusal* scope is installed and terminal execution raises :class:`TerminalPolicyUnavailable`. """ from __future__ import annotations import logging import os from contextlib import contextmanager from contextvars import ContextVar, Token from pathlib import Path from typing import Any, Dict, Iterator, Optional logger = logging.getLogger(__name__) # None = no scope bound (process-env behavior); dict = complete policy; Refusal = resolution failed. _terminal_scope_var: ContextVar = ContextVar("hermes_terminal_scope", default=None) # Keys whose default lives in terminal_tool.py, not DEFAULT_CONFIG (which wins on overlap); # without them the projection is not total. _TOOL_LEVEL_DEFAULTS: Dict[str, Any] = { "cwd": ".", "ssh_host": "", "ssh_user": "", "ssh_port": 22, "ssh_key": "", "docker_orphan_reaper": True, "docker_persist_across_processes": True, "sandbox_dir": "", "lifetime_seconds": 300, "docker_shared_container_key": "", "home_mode": "auto", } class TerminalPolicyUnavailable(Exception): """The routed profile's ``.env``/``config.yaml`` exists but cannot be read/parsed.""" class TerminalPolicyRefusal(Dict[str, str]): """Marker scope (empty dict subclass) installed when policy resolution failed.""" def __init__(self, reason: str) -> None: super().__init__() self.reason = reason def set_terminal_scope(mapping: Optional[Dict[str, str]]) -> Token: """Install *mapping* as the current context's terminal policy.""" return _terminal_scope_var.set(mapping) def reset_terminal_scope(token: Token) -> None: _terminal_scope_var.reset(token) def get_terminal_scope() -> Optional[Dict[str, str]]: """The active scope mapping/refusal, or ``None`` when no scope is bound.""" return _terminal_scope_var.get() def enforce_no_refusal() -> None: """Raise when the active scope is a refusal scope (fail closed). Execution paths (terminal tool, execute_code) call this before spawning anything: under a refusal scope the profile's terminal policy could not be resolved, and running with the launch process's ambient policy is exactly the authority leak this module closes (#68559 requires refusal, not fallback). Non-scoped and policy-scoped contexts pass silently. """ scope = _terminal_scope_var.get() if isinstance(scope, TerminalPolicyRefusal): raise TerminalPolicyUnavailable( f"terminal policy unavailable for this profile: {scope.reason}") def terminal_env(name: str, default: str = "") -> str: """Authoritative read of a ``TERMINAL_*`` variable. No scope: process env, then *default*. Refusal scope: raise. Policy scope: ONLY the policy; a missing key yields *default*, never os.environ. """ scope = _terminal_scope_var.get() if scope is None: return os.environ.get(name, default) enforce_no_refusal() value = scope.get(name) return default if value is None else str(value) def build_profile_terminal_scope( hermes_home: "Any", *, env_overlay: Optional[Dict[str, str]] = None) -> Dict[str, str]: """Build the COMPLETE effective ``TERMINAL_*`` policy for a profile home. Projection: ``DEFAULT_CONFIG['terminal']`` <- profile ``.env`` TERMINAL_* <- *env_overlay* <- profile ``config.yaml`` ``terminal:``. Total by construction, so a bound scope never widens back to ambient authority. Raises :class:`TerminalPolicyUnavailable` if a present file is unreadable. *env_overlay* is a TRUSTED ``TERMINAL_*`` mapping captured from the launch process before multiplexing began (``tui_gateway/launch_profile_policy.py``): the launch profile's env-only policy (``TERMINAL_ENV=ssh`` from systemd, ``op run``, a launcher bridge) has no file to rebuild it from, and reading live ``os.environ`` here is the leak this module closes. It sits where the process env sits in the standalone bridge — explicit YAML keys still win (``apply_terminal_config_to_env``). """ from hermes_cli.config import TERMINAL_CONFIG_ENV_MAP, _terminal_env_value from hermes_cli.config_defaults import DEFAULT_CONFIG home = Path(hermes_home) scope: Dict[str, str] = {} def _apply(mapping: Dict[str, Any]) -> None: for cfg_key, value in mapping.items(): # cwd placeholders are resolved per-surface later; not a policy value. if value is None or (cfg_key == "cwd" and str(value).strip() in {".", "auto", "cwd"}): continue env_var = TERMINAL_CONFIG_ENV_MAP.get(cfg_key) if env_var: # List/dict config values must be JSON (same contract as # apply_terminal_config_to_env). str() yields Python repr, which # json.loads in terminal_tool rejects. scope[env_var] = _terminal_env_value(value) _apply({**_TOOL_LEVEL_DEFAULTS, **(DEFAULT_CONFIG.get("terminal") or {})}) default_image = scope.get("TERMINAL_DOCKER_IMAGE") env_path = home / ".env" if env_path.exists(): # load_env_file swallows OSError by design (secret scope fails soft); an unreadable # profile .env must fail closed here. try: env_path.read_bytes() except Exception as exc: raise TerminalPolicyUnavailable(f"cannot read {env_path}: {exc}") from exc from agent.secret_scope import load_env_file profile_env = load_env_file(env_path) scope.update((k, str(v)) for k, v in profile_env.items() if k.startswith("TERMINAL_")) # Provenance, not value: an image WRITTEN in the profile's .env is the user's choice even when # it spells the default (same rule apply_terminal_config_to_env applies to a set env var). image_pinned = "TERMINAL_DOCKER_IMAGE" in profile_env else: image_pinned = False if env_overlay: scope.update((k, str(v)) for k, v in env_overlay.items() if k.startswith("TERMINAL_") and k != "TERMINAL_DOCKER_IMAGE_PINNED") # The overlay is the LAUNCHER's environment: its pin verdict is about the launcher's profile and # is never inherited, and the bridge backfills TERMINAL_DOCKER_IMAGE for defaults too, so only a # value that differs from the default can prove a choice made there. if env_overlay.get("TERMINAL_DOCKER_IMAGE") not in (None, default_image): image_pinned = True # Read config.yaml directly, not via read_raw_config() (which collapses "missing" and # "unparseable" into {}): present-but-unparseable must fail closed. config_path = home / "config.yaml" try: config_exists = config_path.exists() except Exception as exc: raise TerminalPolicyUnavailable(f"cannot resolve terminal config in {home}: {exc}") from exc if config_exists: from utils import load_yaml_file_readonly try: # Signature-cached: a scope is rebuilt per routed turn/poll, the file rarely changes. raw = load_yaml_file_readonly(config_path) except Exception as exc: raise TerminalPolicyUnavailable(f"cannot parse {config_path}: {exc}") from exc raw_terminal = raw.get("terminal") if isinstance(raw, dict) else None if isinstance(raw_terminal, dict): _apply(raw_terminal) image_pinned = image_pinned or "docker_image" in raw_terminal scope["TERMINAL_DOCKER_IMAGE_PINNED"] = "1" if image_pinned else "0" _resolve_scope_cwd_placeholder(scope) return scope def _resolve_scope_cwd_placeholder(scope: Dict[str, str]) -> None: """Give a scope with no explicit ``terminal.cwd`` the same resolved ``TERMINAL_CWD`` a standalone gateway computes at import (``gateway/run.py``: local backend → ``$HOME``; docker with the workspace mount → the host cwd signal; other backends → unset). Without it a routed turn's ``resolve_agent_cwd()`` falls back to the multiplexer PROCESS cwd (wherever ``hermes gateway`` was launched), so the system prompt, context-file discovery and the local terminal all run in a directory the profile's standalone gateway would never have used.""" if scope.get("TERMINAL_CWD"): return from gateway.cwd_placeholder import resolve_placeholder_terminal_cwd resolved = resolve_placeholder_terminal_cwd( configured_cwd="", terminal_backend=scope.get("TERMINAL_ENV", ""), messaging_cwd=None, docker_mount_cwd_to_workspace=scope.get( "TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE", "false").strip().lower() in {"true", "1", "yes"}, home_fallback=str(Path.home()), ) if resolved: scope["TERMINAL_CWD"] = resolved def install_profile_terminal_scope( hermes_home: "Any", *, env_overlay: Optional[Dict[str, str]] = None) -> Token: """Build AND install a profile's policy; on failure install the refusal scope. Never raises.""" try: return set_terminal_scope(build_profile_terminal_scope(hermes_home, env_overlay=env_overlay)) except TerminalPolicyUnavailable as exc: logger.warning("terminal policy unavailable: %s", exc) return _terminal_scope_var.set(TerminalPolicyRefusal(str(exc))) @contextmanager def install_and_reset_profile_terminal_scope(hermes_home: "Any") -> Iterator[None]: """Install the profile's terminal policy for a bounded turn/fire. Never raises.""" token = install_profile_terminal_scope(hermes_home) try: yield finally: reset_terminal_scope(token)