2553 lines · 103.6 KB · Python
RawDownload
"""Hermes Web UI -- Profile state management.Wraps hermes_cli.profiles to provide profile switching for the web UI. The web UI maintains a process-level "active profile" that determines whichHERMES_HOME directory is used for config, skills, memory, cron, and API keys.Profile switches update os.environ['HERMES_HOME'] and monkey-patch module-levelcached paths in hermes-agent modules (skills_tool, skill_manager_tool,cron/jobs) that snapshot HERMES_HOME at import time."""import jsonimport loggingimport osimport reimport shutilimport sysimport threadingfrom contextlib import contextmanagerfrom pathlib import Pathfrom typing import Optional import yaml from api.session_events import publish_session_list_changed logger = logging.getLogger(__name__) # ── Constants (match hermes_cli.profiles upstream) ─────────────────────────_PROFILE_ID_RE = re.compile(r'^[a-z0-9][a-z0-9_-]{0,63}$')_PROFILE_DIRS = [    'memories', 'sessions', 'skills', 'skins',    'logs', 'plans', 'workspace', 'cron',]_CLONE_CONFIG_FILES = ['config.yaml', '.env', 'SOUL.md'] # ── Snapshot startup env before profile init / dotenv reload mutates it ───────# _is_isolated_profile_mode() needs startup HERMES_HOME, not the value after# init_profile_state() rewrites it. The opt-in flag is also an operator-level# startup control: a pinned profile's .env may be loaded into live os.environ# later, but must not be able to change whether the process is isolated._INITIAL_HERMES_HOME = os.getenv('HERMES_HOME', '').strip()_INITIAL_ISOLATED_PROFILE_OPT_IN = os.getenv('HERMES_WEBUI_ISOLATED_PROFILE', '').strip().lower()_ISOLATED_SYMLINK_WARNING_EMITTED = False_ISOLATED_PROFILE_SHAPE_WITHOUT_OPT_IN_WARNING_EMITTED = False_ISOLATED_PROFILE_TRUTHY_VALUES = frozenset({'1', 'true', 'yes', 'on'}) # ── Module state ────────────────────────────────────────────────────────────_active_profile = 'default'_profile_lock = threading.Lock()_loaded_profile_env_keys: set[str] = set() # Thread-local profile context: set per-request by server.py, cleared after.# Enables per-client profile isolation (issue #798) — each HTTP request thread# reads its own profile from the hermes_profile cookie instead of the# process-global _active_profile._tls = threading.local() _SKILL_HOME_MODULES = ("tools.skills_tool", "tools.skill_manager_tool")  def snapshot_skill_home_modules() -> dict[str, dict[str, object]]:    """Snapshot imported skill-module path globals before a temporary patch."""    snapshot: dict[str, dict[str, object]] = {}    for module_name in _SKILL_HOME_MODULES:        module = sys.modules.get(module_name)        if module is None:            snapshot[module_name] = {"module_present": False}            continue        snapshot[module_name] = {            "module_present": True,            "has_HERMES_HOME": hasattr(module, "HERMES_HOME"),            "HERMES_HOME": getattr(module, "HERMES_HOME", None),            "has_SKILLS_DIR": hasattr(module, "SKILLS_DIR"),            "SKILLS_DIR": getattr(module, "SKILLS_DIR", None),        }    return snapshot  def patch_skill_home_modules(home: Path) -> None:    """Patch imported skill modules that cache HERMES_HOME at import time."""    for module_name in _SKILL_HOME_MODULES:        module = sys.modules.get(module_name)        if module is None:            continue        try:            module.HERMES_HOME = home            module.SKILLS_DIR = home / "skills"        except AttributeError:            logger.debug("Failed to patch %s module", module_name)  def restore_skill_home_modules(snapshot: dict[str, dict[str, object]]) -> None:    """Restore skill-module globals captured by snapshot_skill_home_modules()."""    for module_name, values in snapshot.items():        module = sys.modules.get(module_name)        if not values.get("module_present"):            if module is not None:                sys.modules.pop(module_name, None)                parent_name, _, child_name = module_name.rpartition(".")                parent = sys.modules.get(parent_name)                if parent is not None:                    try:                        delattr(parent, child_name)                    except AttributeError:                        pass            continue        if module is None:            continue        for attr in ("HERMES_HOME", "SKILLS_DIR"):            has_attr = bool(values.get(f"has_{attr}"))            try:                if has_attr:                    setattr(module, attr, values.get(attr))                else:                    try:                        delattr(module, attr)                    except AttributeError:                        pass            except AttributeError:                logger.debug("Failed to restore %s.%s", module_name, attr)  def _unwrap_profile_home_to_base(home: Path) -> Path:    """Return the base Hermes home when *home* is already a named profile dir."""    if home.parent.name == 'profiles':        return home.parent.parent    return home  # Env keys a pinned profile's .env may NOT override via _reload_dotenv() — these# are operator/deployment-level postures, not per-profile toggles. Letting a# profile .env set HERMES_WEBUI_ISOLATED_PROFILE=0 would let a contained user# escape isolation (#4589)._PROTECTED_ENV_KEYS = frozenset({'HERMES_WEBUI_ISOLATED_PROFILE'})  def _isolated_profile_opt_in() -> bool:    """Return True only when isolated single-profile mode is EXPLICITLY enabled.    Isolated mode is an intentional multi-user deployment posture (each user is    pinned to one profile and cross-profile operations are rejected). It must be    opted into with ``HERMES_WEBUI_ISOLATED_PROFILE`` — it is NEVER inferred from    the ``HERMES_HOME`` shape alone, because a normal single-user who runs under a    named profile produces the byte-identical ``*/profiles/<name>`` shape (the    Hermes Agent launcher exports ``HERMES_HOME=~/.hermes/profiles/<name>`` for any    active named profile). Keying isolation off the shape alone therefore breaks    profile switching for ordinary single-user deployments (#4586).     Accepts the usual truthy values; default (unset/empty/falsey) is OFF.     Security: this reads the startup snapshot, not live ``os.environ``. A pinned    profile's ``.env`` is loaded after import, so live env can be profile-owned;    the opt-in must remain the operator/launcher posture captured at process    start (#4590). ``_reload_dotenv()`` and the runtime env paths still filter the    key as defense-in-depth, but detection does not depend on that filtering.    """    return _INITIAL_ISOLATED_PROFILE_OPT_IN in _ISOLATED_PROFILE_TRUTHY_VALUES  def _warn_if_profile_shape_without_isolated_opt_in() -> None:    """Log once when HERMES_HOME looks pinned but startup opt-in is absent."""    global _ISOLATED_PROFILE_SHAPE_WITHOUT_OPT_IN_WARNING_EMITTED    if _ISOLATED_PROFILE_SHAPE_WITHOUT_OPT_IN_WARNING_EMITTED:        return    hermes_home = _INITIAL_HERMES_HOME    if not hermes_home:        return    p = Path(hermes_home).expanduser()    if p.parent.name != 'profiles' or not p.name:        return    logger.warning(        "HERMES_HOME points at a profile directory (%s), but "        "HERMES_WEBUI_ISOLATED_PROFILE was not enabled at startup; isolated "        "profile mode stays off and normal multi-profile switching remains enabled.",        p,    )    _ISOLATED_PROFILE_SHAPE_WITHOUT_OPT_IN_WARNING_EMITTED = True  def _is_isolated_profile_mode() -> bool:    """Detect isolated single-profile mode.     Returns True only when BOTH conditions hold:      1. ``HERMES_WEBUI_ISOLATED_PROFILE`` is explicitly enabled (the PRIMARY         gate — see _isolated_profile_opt_in), AND      2. HERMES_HOME at startup points at a concrete profile subdirectory         (e.g., ~/.hermes/profiles/user1) rather than the base home.     Why the explicit flag is required (#4586 regression fix): the    ``*/profiles/<name>`` shape alone CANNOT distinguish an intentional    multi-user isolation deployment from an ordinary single-user running under a    named profile — the Hermes Agent launcher sets    ``HERMES_HOME=~/.hermes/profiles/<name>`` for any active named profile, so the    two cases are byte-identical at the env-var level. Inferring isolation from    the shape alone (the v0.51.528 behaviour from #2698) wrongly pinned ordinary    single-user deployments to one profile and disabled profile switching. The    multi-user wrapper that genuinely wants isolation now sets the explicit flag;    everyone else is never caught. The shape stays as a secondary requirement so    a stray flag without a profile-shaped HERMES_HOME does not engage isolation.     Uses _INITIAL_HERMES_HOME (snapshotted at import time) to detect the shape,    not the current os.environ value. init_profile_state() overwrites HERMES_HOME    at startup, which would disable detection if we read it here.    """    # PRIMARY gate: explicit startup opt-in. Default OFF → a normal named-profile    # launch is never treated as isolated, so profile switching keeps working    # (#4586). Read the snapshot, not live os.environ, so profile .env reloads    # cannot silently flip the deployment posture (#4590).    if not _isolated_profile_opt_in():        _warn_if_profile_shape_without_isolated_opt_in()        return False     hermes_home = _INITIAL_HERMES_HOME    if not hermes_home:        return False     p = Path(hermes_home).expanduser()    # SECONDARY requirement: HERMES_HOME must look like ~/.hermes/profiles/<name>    # i.e., parent dir is named 'profiles' and grandparent exists.    if p.parent.name == 'profiles' and p.parent.parent.exists():        return True    if p.is_symlink():        global _ISOLATED_SYMLINK_WARNING_EMITTED        if not _ISOLATED_SYMLINK_WARNING_EMITTED:            logger.warning(                "HERMES_WEBUI_ISOLATED_PROFILE is set but HERMES_HOME %s does not "                "literally match */profiles/<name>; isolated profile mode stays off "                "unless the literal profile path is used.",                p,            )            _ISOLATED_SYMLINK_WARNING_EMITTED = True    return False  def _isolated_profile_name() -> str:    """Return the profile directory name from _INITIAL_HERMES_HOME."""    return Path(_INITIAL_HERMES_HOME).expanduser().name  def _resolve_base_hermes_home() -> Path:    """Return the BASE ~/.hermes directory — the root that contains profiles/.     This is intentionally distinct from HERMES_HOME, which tracks the *active    profile's* home and changes on every profile switch.  The base dir must    always point to the top-level .hermes regardless of which profile is active.     Resolution order:      1. HERMES_BASE_HOME env var (set explicitly, highest priority)      2. HERMES_HOME env var — but only if it does NOT look like a profile subdir         (i.e. its parent is not named 'profiles').  This handles test isolation         where HERMES_HOME is set to an isolated test state dir.      3. ~/.hermes (always-correct default)     The bug this prevents: if HERMES_HOME has already been mutated to    /home/user/.hermes/profiles/webui (by init_profile_state at startup),    reading it here would make _DEFAULT_HERMES_HOME point to that subdir,    causing switch_profile('webui') to look for    /home/user/.hermes/profiles/webui/profiles/webui — which doesn't exist.     HERMES_BASE_HOME normally points at the base home already, but isolated    single-profile WebUI deployments can provide /base/profiles/<name> there as    well.  Normalize both env vars through the same helper so active-profile    and per-request resolution share one base-root contract (#749).    """    # Explicit override for tests or unusual setups    base_override = os.getenv('HERMES_BASE_HOME', '').strip()    if base_override:        return _unwrap_profile_home_to_base(Path(base_override).expanduser())     hermes_home = os.getenv('HERMES_HOME', '').strip()    if hermes_home:        p = Path(hermes_home).expanduser()        # If HERMES_HOME points to a profiles/ subdir, walk up two levels to the base        return _unwrap_profile_home_to_base(p)     # Platform default. On Windows this includes the #2905 migration-safety    # fallback (prefer the populated legacy %USERPROFILE%\.hermes over an    # empty %LOCALAPPDATA%\hermes). Import the shared path helper directly    # instead of importing api.config here; api.config imports profiles during    # startup, so going through config creates a partial-module circular import    # when api.profiles is imported first.    from api.paths import _platform_default_hermes_home     return _platform_default_hermes_home() _DEFAULT_HERMES_HOME = _resolve_base_hermes_home()  def _read_active_profile_file() -> str:    """Read the sticky active profile from ~/.hermes/active_profile."""    ap_file = _DEFAULT_HERMES_HOME / 'active_profile'    if ap_file.exists():        try:            name = ap_file.read_text(encoding="utf-8").strip()            if name:                return name        except Exception:            logger.debug("Failed to read active profile file")    return 'default'  # ── Public API ────────────────────────────────────────────────────────────── # ── Root-profile resolution (#1612) ────────────────────────────────────────## Hermes Agent allows the root/default profile (~/.hermes itself) to have a# display name other than the legacy literal 'default'.  When that happens,# WebUI must NOT resolve the display name as ~/.hermes/profiles/<name> — that# directory doesn't exist, and every site that does `if name == 'default':`# will fall through to the wrong filesystem path.## `_is_root_profile(name)` answers "does this name resolve to ~/.hermes?" and# is the canonical replacement for scattered `if name == 'default':` checks# in switch_profile, get_active_hermes_home, _validate_profile_name, etc.## Cost note: list_profiles_api() shells out via hermes_cli (non-trivial), so# we memoize the lookup. The cache is invalidated whenever profiles are# created, deleted, renamed, or cloned — i.e. on every mutation site we# control._root_profile_name_cache: set[str] = {'default'}_root_profile_name_cache_lock = threading.Lock()_root_profile_name_cache_loaded = False  def _invalidate_root_profile_cache() -> None:    """Drop the memoized root-profile-name set.     Called whenever profile metadata might have changed: create, clone,    delete, rename. The next _is_root_profile() call repopulates from    list_profiles_api().    """    global _root_profile_name_cache_loaded    with _root_profile_name_cache_lock:        _root_profile_name_cache.clear()        _root_profile_name_cache.add('default')        _root_profile_name_cache_loaded = False  def _is_root_profile(name: str) -> bool:    """True if *name* resolves to the Hermes Agent root profile (~/.hermes).     Matches the legacy 'default' alias plus any name where list_profiles_api()    reports is_default=True. Memoized; call _invalidate_root_profile_cache()    after mutating profile metadata.    """    global _root_profile_name_cache_loaded    if not name:        return False    if name == 'default':        return True    with _root_profile_name_cache_lock:        if _root_profile_name_cache_loaded:            return name in _root_profile_name_cache    # Cache miss — populate from list_profiles_api(). Done outside the lock to    # avoid holding it across a hermes_cli subprocess call.    try:        infos = list_profiles_api()    except Exception:        logger.debug("Failed to list profiles for root-profile lookup", exc_info=True)        return False    with _root_profile_name_cache_lock:        _root_profile_name_cache.clear()        _root_profile_name_cache.add('default')        for p in infos:            try:                if p.get('is_default') and p.get('name'):                    _root_profile_name_cache.add(p['name'])            except (AttributeError, TypeError):                continue        _root_profile_name_cache_loaded = True        return name in _root_profile_name_cache  def _profiles_match(row_profile, active_profile) -> bool:    """Return True if a session/project row's profile matches the active profile.     Treats both the literal alias 'default' and any renamed-root display name    (per _is_root_profile) as equivalent, so legacy rows tagged 'default'    still surface when the user has renamed the root profile to e.g. 'kinni',    and vice versa.     A row with no profile (`None` or empty string) is treated as belonging to    the root profile — that's the convention used by the legacy backfill at    api/models.py::all_sessions, and matches the default seen in    `static/sessions.js` (`S.activeProfile||'default'`).     Originally lived in api/routes.py; relocated here so both routes.py and    out-of-process consumers (mcp_server.py) can import the canonical helper    instead of duplicating the body. See #1614 for the visibility model.    """    row = row_profile or 'default'    active = active_profile or 'default'    if row == active:        return True    # Cross-alias the renamed root.    if _is_root_profile(row) and _is_root_profile(active):        return True    return False  def get_active_profile_name() -> str:    """Return the currently active profile name.     Priority:      1. Isolated-profile deployment name from the configured HERMES_HOME path      2. Thread-local (set per-request from hermes_profile cookie) — issue #798      3. Process-level default (_active_profile)    """    if _is_isolated_profile_mode():        return _isolated_profile_name()    tls_name = getattr(_tls, 'profile', None)    if tls_name is not None:        return tls_name    return _active_profile  def set_request_profile(name: str) -> None:    """Set the per-request profile context for this thread.     Called by server.py at the start of each request when a hermes_profile    cookie is present.  Always paired with clear_request_profile() in a    finally block so the thread-local is released after the request.    """    _tls.profile = name  def clear_request_profile() -> None:    """Clear the per-request profile context for this thread.     Called by server.py in the finally block of do_GET / do_POST.    Safe to call even if set_request_profile() was never called.    """    _tls.profile = None  def _resolve_profile_home_for_name(name: str) -> Path:    """Resolve a logical profile name to its Hermes home path.     Root/default aliases resolve to _DEFAULT_HERMES_HOME.  Valid named profiles    resolve to _DEFAULT_HERMES_HOME/profiles/<name> even when the directory has    not been created yet; the agent layer may create it on first use.  Invalid    names fall back to the base home so traversal-shaped cookie values cannot    influence filesystem paths.    """    # In isolated mode, every logical profile lookup clamps to the configured    # startup HERMES_HOME so callers cannot resolve a foreign profile path.    if _is_isolated_profile_mode():        isolated_name = _isolated_profile_name()        isolated_home = Path(_INITIAL_HERMES_HOME).expanduser()        if name and not _profiles_match(name, isolated_name):            logger.warning(                "Ignoring profile lookup %r in isolated profile mode; using pinned profile %r",                name, isolated_name,            )        return isolated_home    if not name or _is_root_profile(name):        return _DEFAULT_HERMES_HOME    if not _PROFILE_ID_RE.fullmatch(name):        return _DEFAULT_HERMES_HOME    return _resolve_named_profile_home(name)  def get_active_hermes_home() -> Path:    """Return the HERMES_HOME path for the currently active profile.     Uses get_active_profile_name() so per-request TLS context (issue #798)    is respected, not just the process-level global.    """    if _is_isolated_profile_mode():        return Path(_INITIAL_HERMES_HOME).expanduser()    return _resolve_profile_home_for_name(get_active_profile_name())   # ── Cron-call profile isolation (issue: Scheduled jobs ignored active profile) ─# `cron.jobs` reads HERMES_HOME from os.environ (process-global) at function-# call time. That bypasses our per-request thread-local profile, so the# `/api/crons*` endpoints always returned the process-default profile's jobs.# This context manager swaps HERMES_HOME (and the cached module-level constants# in cron.jobs) for the duration of a cron call, serialized by a lock so# concurrent requests from different profiles don't race on the global env var.## Thread-safety note on os.environ mutation:# CPython's os.environ assignment is GIL-protected at the bytecode level, but# multi-step read-modify-write sequences (snapshot prev → assign new → restore# on exit) are NOT atomic without explicit serialization. The _cron_env_lock# below makes the entire context-manager body run-to-completion serially, so# all webui access to HERMES_HOME goes through one thread at a time. Any# subprocess.Popen() call inside `run_job` inherits the env at fork time,# which is also under the lock — so child processes always see a consistent# (own-profile) HERMES_HOME, never a half-swapped state._cron_env_lock = threading.Lock()  def _cron_profile_context_depth() -> int:    return int(getattr(_tls, 'cron_profile_depth', 0) or 0)  def _push_cron_profile_context_depth() -> None:    _tls.cron_profile_depth = _cron_profile_context_depth() + 1 

Lines 1–500 of 2553.