hermes-agent/hermes_cli/gateway_migrate_guards.py

"""Boundaries the AUTOMATIC multiplex migration (``hermes update``) must not cross, and the opt-out.

The multiplexer replaces a kernel-enforced boundary (separate UNIX users, separate service domains,
separate HERMES_HOME trees) with in-process isolation. An operator may choose that with
``hermes gateway migrate --multiplex``; an unattended update hook must not choose it for them.
``build_migration_plan`` records the same findings as NOTICES so a dry run shows them; only
:func:`maybe_auto_migrate_after_update` treats them as blockers (#109954).
"""

from __future__ import annotations

import contextlib
import os
import subprocess
from pathlib import Path
from typing import TYPE_CHECKING, Callable, Optional

if TYPE_CHECKING:
    from hermes_cli.gateway_migrate import MigrationPlan, ProfileGateway


# --------------------------------------------------------------------------- identity resolution


def _pid_uid(pid: int) -> Optional[int]:
    """Owner uid of a live process: ``/proc`` where it exists, ``ps`` on macOS; None when unknown."""
    with contextlib.suppress(OSError):
        return os.stat(f"/proc/{pid}").st_uid
    from hermes_cli.gateway import is_macos
    if not is_macos():
        return None
    with contextlib.suppress(OSError, ValueError, subprocess.SubprocessError):
        result = subprocess.run(["ps", "-o", "uid=", "-p", str(pid)], capture_output=True, text=True, encoding="utf-8",
                                check=False, timeout=2)
        if result.returncode == 0 and result.stdout.strip():
            return int(result.stdout.strip())
    return None


def _system_unit_uid(unit_path: Path) -> Optional[int]:
    """uid a system unit runs as: its ``User=`` (root when absent); None when the name is unknown."""
    from hermes_cli.gateway import _read_systemd_user_from_unit
    user = _read_systemd_user_from_unit(unit_path)
    if user is None:
        return 0
    import pwd
    with contextlib.suppress(KeyError):
        return pwd.getpwnam(user).pw_uid
    return None


def gateway_identity(home: Path, pid: Optional[int], service: Optional[tuple[str, bool]]) -> tuple[Optional[int], Path]:
    """``(uid, runtime_home)`` of the gateway that serves ``home``.

    uid: the live process owner, else the system unit's ``User=``, else the owner of the profile
    directory (user-scope systemd / launchd / detached gateways run as the account that owns it).
    None means unknown — never a different user. runtime_home: the HERMES_HOME the installed unit
    pins, which is where the gateway really runs; ``home`` when there is no unit or no pin.
    """
    from hermes_cli.gateway import _hermes_home_pinned_by_unit, get_systemd_unit_path
    from hermes_cli.gateway_migrate import _home_env

    uid: Optional[int] = _pid_uid(pid) if pid is not None else None
    runtime_home = home
    if service is not None and service[0] == "systemd":
        with _home_env(home):
            unit_path = get_systemd_unit_path(system=service[1])
        pinned = _hermes_home_pinned_by_unit(unit_path)
        if pinned:
            runtime_home = Path(pinned).expanduser()
        if uid is None and service[1]:
            uid = _system_unit_uid(unit_path)
    if uid is None:
        with contextlib.suppress(OSError):
            uid = home.stat().st_uid
    return uid, runtime_home


# --------------------------------------------------------------------------- guards


def _service_label(profile: ProfileGateway) -> str:
    return profile.service_label() if profile.service is not None else "no service manager (detached)"


def _guard_service_domain(plan: MigrationPlan, profile: ProfileGateway) -> Optional[str]:
    """Different manager or scope than the default gateway (system vs user systemd, launchd vs systemd,
    or any service when the default is detached: the auto path never elects a secondary's manager)."""
    if profile.service == plan.default.service:
        return None
    return (f"Profile '{profile.name}' runs under {_service_label(profile)} while the default gateway "
            f"runs under {_service_label(plan.default)}: a different service domain is not folded automatically.")


def _guard_unix_user(plan: MigrationPlan, profile: ProfileGateway) -> Optional[str]:
    default_uid = plan.default.uid
    if default_uid is None or profile.uid is None or profile.uid == default_uid:
        return None
    return (f"Profile '{profile.name}' runs as uid {profile.uid} while the default gateway runs as uid "
            f"{default_uid}: a UNIX privilege boundary is not folded automatically.")


def _guard_home_tree(plan: MigrationPlan, profile: ProfileGateway) -> Optional[str]:
    profiles_root = (plan.default_home / "profiles").resolve()
    runtime_home = (profile.runtime_home or profile.home).resolve()
    if runtime_home.is_relative_to(profiles_root):
        return None
    return (f"Profile '{profile.name}' runs with HERMES_HOME={runtime_home}, outside {profiles_root}: "
            f"the multiplexer would serve {profile.home} instead of the live home.")


_AUTO_MIGRATION_GUARDS: tuple[Callable[[MigrationPlan, ProfileGateway], Optional[str]], ...] = (
    _guard_service_domain,
    _guard_unix_user,
    _guard_home_tree,
)


def auto_migration_blockers(plan: MigrationPlan) -> list[str]:
    """Every boundary a standalone secondary sits behind; empty when the fleet is one user, one service
    domain, one profiles/ tree — the only shape ``hermes update`` may fold on its own."""
    return [
        finding
        for profile in plan.standalone_secondaries
        for guard in _AUTO_MIGRATION_GUARDS
        if (finding := guard(plan, profile)) is not None
    ]


# --------------------------------------------------------------------------- opt-out


def auto_migration_opted_out(default_home: Path) -> bool:
    """``gateway.auto_multiplex_migration: false`` in the DEFAULT profile's config.yaml. Absent means
    opted in (the ``DEFAULT_CONFIG`` value); only the nested key counts, there is no top-level alias."""
    cfg_path = default_home / "config.yaml"
    if not cfg_path.exists():
        return False
    from hermes_cli.config import read_user_config_raw
    cfg = read_user_config_raw(cfg_path) or {}
    gateway_section = cfg.get("gateway")
    if not isinstance(gateway_section, dict):
        return False
    value = gateway_section.get("auto_multiplex_migration")
    return value is not None and not bool(value)