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 pid_is_other_users(pid: int) -> bool:
    """True when *pid* is owned by another Unix user than this non-root caller.

    Gateway process scans match argv host-wide, so without this a second account's gateway (users
    isolate credentials by UID) became a ``restart --all`` SIGTERM target and "unfolded profile"
    advice. Root keeps the host-wide view (``sudo … --system`` over a ``User=`` unit), as does
    Windows (no ``geteuid``); an unknown owner is not called foreign.
    """
    my_uid = os.geteuid() if hasattr(os, "geteuid") else 0
    owner = _pid_uid(pid) if my_uid else None
    return owner is not None and owner != my_uid


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], services: list[tuple[str, bool]]) -> tuple[Optional[int], Path]:
    """``(uid, runtime_home)`` of the gateway that serves ``home``.

    uid: the live process owner; else, for a system unit, its ``User=`` — and ONLY that: a system
    unit's principal is whatever systemd will run, so an unresolvable ``User=`` stays None rather than
    borrowing the profile directory's owner (a stopped unit pinned to an absent NSS user is not the
    account that owns the files). Without a system unit (user-scope systemd / launchd / detached), the
    profile directory owner is the account the gateway runs as. None means unknown. runtime_home: the
    HERMES_HOME an installed unit pins, which is where the gateway really runs; ``home`` otherwise.
    """
    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
    has_system_unit = False
    for kind, system in services:
        if kind != "systemd":
            continue
        with _home_env(home):
            unit_path = get_systemd_unit_path(system=system)
        pinned = _hermes_home_pinned_by_unit(unit_path)
        if pinned and runtime_home == home:
            runtime_home = Path(pinned).expanduser()
        if system:
            has_system_unit = True
            if uid is None:
                uid = _system_unit_uid(unit_path)
    if uid is None and not has_system_unit:
        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.services else "no service manager (detached)"


def _guard_service_domain(plan: MigrationPlan, profile: ProfileGateway) -> Optional[str]:
    """Different manager or scope than the one the fleet converges on (system vs user systemd, launchd vs
    systemd). The reference is the default's own unit when it has one, else the manager
    ``target_service_kind()`` elects from the secondaries: a default that never had a gateway unit is not a
    service domain of its own, and refusing every secondary against it left the common upgrade fleet (N
    launchd profiles, unit-less default, #118097) printing blockers instead of folding. Two managers
    among the secondaries still refuse — the ones not elected differ from the target. Two units on one
    profile is an ambiguous topology the unattended path does not resolve either."""
    if len(profile.services) > 1:
        return (f"Profile '{profile.name}' has more than one installed service ({profile.service_label()}): "
                f"an ambiguous service topology is not folded automatically.")
    target = plan.target_service_kind()
    reference = plan.default.services or ([target] if target is not None else [])
    if set(profile.services) == set(reference):
        return None
    from hermes_cli.gateway_migrate import _service_label as _kind_label
    against = (f"the default gateway runs under {_service_label(plan.default)}" if plan.default.services
               else f"the fleet converges on {_kind_label(target)}")
    return (f"Profile '{profile.name}' runs under {_service_label(profile)} while {against}: "
            f"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 and plan.default.has_system_unit:
        # Consolidating INTO a principal this host cannot identify is the same unknown boundary
        # from the other side: the default's system unit names an account NSS does not resolve.
        return ("The default gateway runs a system unit whose User= cannot be resolved on this host: "
                "an unknown service principal is not folded into automatically.")
    if profile.uid is None and profile.has_system_unit:
        # Unknown principal is not "same user": the unit names an account this host cannot resolve.
        return (f"Profile '{profile.name}' runs a system unit whose User= cannot be resolved on this host: "
                f"an unknown service principal is not folded automatically.")
    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."""
    findings = [
        finding
        for profile in plan.standalone_secondaries
        for guard in _AUTO_MIGRATION_GUARDS
        if (finding := guard(plan, profile)) is not None
    ]
    return list(dict.fromkeys(findings))  # a default-side finding repeats per secondary


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


def masked_default_unit(default_home: Path) -> Optional[Path]:
    """The default profile's systemd unit when the operator MASKED it (``systemctl mask`` links it to
    ``/dev/null``), else None. A masked survivor cannot be enabled at boot, and masking is a deliberate
    "never start this", so the update hook reads it as an opt-out instead of refusing (#139324)."""
    from hermes_cli import gateway as gw
    from hermes_cli.gateway_migrate import _home_env
    with _home_env(default_home):
        units = [gw.get_systemd_unit_path(system=system) for system in (False, True)]
    for unit in units:
        with contextlib.suppress(OSError):
            if unit.is_symlink() and os.path.realpath(unit) == os.devnull:
                return unit
    return None


def auto_migration_opted_out(default_home: Path) -> bool:
    """``gateway.auto_multiplex_migration: false`` in the DEFAULT profile's EFFECTIVE config: the same
    ``load_config`` the rest of the CLI reads (``DEFAULT_CONFIG`` + config.yaml + the managed overlay), so
    an administrator's managed ``false`` wins over a user's ``true`` and a YAML string ``"false"`` is
    false, not truthy. Only the nested key counts, there is no top-level alias."""
    from hermes_cli.config import load_config_readonly
    from hermes_cli.gateway_migrate import _home_env
    from utils import is_truthy_value
    with _home_env(default_home):
        gateway_section = load_config_readonly().get("gateway")
    if not isinstance(gateway_section, dict):
        return False
    return not is_truthy_value(gateway_section.get("auto_multiplex_migration"), default=True)