fix(constants): one home for managed, container and HERMES_UID policy; drop the config twins

Review follow-up: get_managed_system, the container/chmod-skip check and
the HERMES_UID/GID chown (_resolve_hermes_uid_gid/_chown_to_hermes_uid) now
live only in hermes_constants — the import-safe module apply_secure_dir_policy
already needed them in — and hermes_cli.config re-exports them, so there are
no 'keep in sync' copies and _secure_file skips on the same canonical
_detect_container signal as _secure_dir/get_scratch_dir. The dead config
copies are deleted and test_ensure_hermes_home_uid.py drives the new
symbols; one invariant test pins the single implementation.
This commit is contained in:
teknium1
2026-09-20 15:31:22 -07:00
committed by Teknium
parent 9218f751de
commit 891c1d79c6
4 changed files with 95 additions and 118 deletions

View File

@@ -37,7 +37,12 @@ from hermes_cli.colors import Colors, color
from hermes_cli import managed_scope
from hermes_cli.default_soul import DEFAULT_SOUL_MD, is_legacy_template_soul
from hermes_cli.secret_prompt import masked_secret_prompt
from hermes_constants import apply_secure_dir_policy
# Managed-mode, container and HERMES_UID/GID policy live in hermes_constants (import-safe);
# re-exported here so existing callers/patch targets keep working.
from hermes_constants import ( # noqa: F401
_IGNORED_MANAGED_VALUES, _LEGACY_MANAGED_SYSTEM, _MANAGED_FALSE_VALUES, _MANAGED_TRUE_VALUES,
_chown_to_hermes_uid, _container_or_chmod_skipped, _resolve_hermes_uid_gid,
apply_secure_dir_policy, get_managed_system)
# Re-export from hermes_constants — canonical definition lives there.
from hermes_constants import get_hermes_home, get_process_hermes_home # noqa: F401
from utils import atomic_replace, atomic_yaml_write, fast_safe_load, file_signature
@@ -287,37 +292,10 @@ _EXTRA_ENV_KEYS = frozenset({
# ---- Managed mode (NixOS declarative config) ----
_MANAGED_TRUE_VALUES = ("true", "1", "yes")
_NIX_MANAGED_SYSTEMS = {"nixos", "home-manager"}
# Only the NixOS module ever wrote a bare "true" or an empty marker.
_LEGACY_MANAGED_SYSTEM = "nixos"
# Nix store root; identifies `nix run` / `nix profile install` installs (which don't set
# HERMES_MANAGED). Module-level so tests can patch it without touching /nix/store.
_NIX_STORE = Path("/nix/store")
# Homebrew is no longer a supported distribution: these markers fall through to git/unknown
# detection instead of blocking config writes.
_IGNORED_MANAGED_VALUES = frozenset({"brew", "homebrew"})
# Explicit opt-out (``HERMES_MANAGED=false``): without this a bool-shaped value became a package
# manager literally named "false" and is_managed() blocked `hermes update` (#12864).
_MANAGED_FALSE_VALUES = frozenset({"false", "0", "no", "off"})
def get_managed_system() -> Optional[str]:
"""Return the package manager owning this install, if any.
Signals: HERMES_MANAGED env var (systemd service) or a ``.managed`` marker file in
HERMES_HOME (NixOS activation script — interactive shells don't see the service env)."""
marker = os.getenv("HERMES_MANAGED", "").strip().lower() or None
managed_marker = get_hermes_home() / ".managed"
if marker is None and managed_marker.exists():
try:
marker = managed_marker.read_text(encoding="utf-8", errors="replace").strip().lower()
except OSError:
marker = ""
if marker is None or marker in _IGNORED_MANAGED_VALUES or marker in _MANAGED_FALSE_VALUES:
return None
if marker == "" or marker in _MANAGED_TRUE_VALUES:
return _LEGACY_MANAGED_SYSTEM
return marker
def is_managed() -> bool:
@@ -568,42 +546,6 @@ def get_project_root() -> Path:
return Path(__file__).parent.parent.resolve()
def _resolve_hermes_uid_gid() -> tuple[Optional[int], Optional[int]]:
"""Read HERMES_UID / HERMES_GID (set by Docker deployments); (None, None) if unset/invalid/Windows.
The entrypoint chowns HERMES_HOME once, but subdirs created at runtime (``profiles/<name>/``)
need the same chown or they land root:root and block later uid-mapped workers.
Docker containers running Hermes commonly set these to map the in-container user to a host user so
volume-mounted state files end up with the right ownership. See #34107.
"""
if sys.platform == "win32":
return None, None
def _env_int(name: str) -> Optional[int]:
try:
return int(os.environ.get(name, "").strip() or None)
except (TypeError, ValueError):
return None
return _env_int("HERMES_UID"), _env_int("HERMES_GID")
def _chown_to_hermes_uid(path) -> None:
"""Chown ``path`` to ``HERMES_UID:HERMES_GID`` when set; EPERM/ENOENT are non-fatal (the
entrypoint's startup chown -R fixes ownership on the next restart).
Used by :func:`_secure_dir` to keep ownership consistent across all directories created by
:func:`ensure_hermes_home` on Docker deployments. See #34107.
"""
uid, gid = _resolve_hermes_uid_gid()
if uid is None and gid is None:
return
try:
os.chown(path, uid if uid is not None else -1, gid if gid is not None else -1)
except (OSError, AttributeError, NotImplementedError):
pass
def _secure_dir(path):
"""chmod a directory owner-only (0700) and apply HERMES_UID/GID ownership. No-op when managed;
in a container only an explicit HERMES_HOME_MODE is applied. HERMES_HOME_MODE (e.g. 0701)
@@ -620,25 +562,10 @@ def _secure_dir(path):
return apply_secure_dir_policy(path)
def _is_container() -> bool:
"""Detect Docker/Podman/LXC (or HERMES_CONTAINER / HERMES_SKIP_CHMOD opt-out).
Volume-mounted config is not forced to 0o600 in containers: gateway and dashboard may run
as different UIDs, or the mount itself needs broader permissions."""
if (os.environ.get("HERMES_CONTAINER") or os.environ.get("HERMES_SKIP_CHMOD")
or os.path.exists("/.dockerenv")):
return True
try:
with open("/proc/1/cgroup", "r", encoding="utf-8") as f:
cgroup_content = f.read()
return any(marker in cgroup_content for marker in ("docker", "lxc", "kubepods"))
except (OSError, IOError):
return False
def _secure_file(path):
"""chmod a file 0600. Skipped when managed (activation sets 0640 group-readable) or in a
container (mounts often need broader permissions)."""
if is_managed() or _is_container():
if is_managed() or _container_or_chmod_skipped():
return
try:
if os.path.exists(str(path)):

View File

@@ -1004,27 +1004,38 @@ def socket_safe_tmpdir() -> str:
return "/tmp" # no-tmp: ok — AF_UNIX 108-byte socket path limit on Linux
def _is_managed_home() -> bool:
"""Managed-install signal for the active home: the ``HERMES_MANAGED`` env var (set by the
systemd service) or a ``.managed`` marker file in the home (NixOS activation script).
# ---- Managed mode (NixOS declarative config) ----
# Canonical home of "is this install package-manager managed": ``hermes_cli.config`` re-exports
# these, and :func:`apply_secure_dir_policy` below reads them. Lives here because constants
# must stay import-safe from the CLI.
_MANAGED_TRUE_VALUES = ("true", "1", "yes")
# Only the NixOS module ever wrote a bare "true" or an empty marker.
_LEGACY_MANAGED_SYSTEM = "nixos"
# Homebrew is no longer a supported distribution: these markers fall through to git/unknown
# detection instead of blocking config writes.
_IGNORED_MANAGED_VALUES = frozenset({"brew", "homebrew"})
# Explicit opt-out (``HERMES_MANAGED=false``): without this a bool-shaped value became a package
# manager literally named "false" and is_managed() blocked `hermes update` (#12864).
_MANAGED_FALSE_VALUES = frozenset({"false", "0", "no", "off"})
Same values and precedence as ``hermes_cli.config.is_managed`` (an unreadable or empty
marker file still counts as managed; brew/homebrew and explicit false values do not) —
keep the two in sync. Lives here because constants must stay import-safe from the CLI.
"""
def get_managed_system() -> str | None:
"""Return the package manager owning this install, if any.
Signals: HERMES_MANAGED env var (systemd service) or a ``.managed`` marker file in
HERMES_HOME (NixOS activation script — interactive shells don't see the service env).
An unreadable or empty marker still counts as managed (the legacy NixOS shape)."""
marker = os.getenv("HERMES_MANAGED", "").strip().lower() or None
if marker is None:
marker_file = get_hermes_home() / ".managed"
if marker_file.exists():
try:
marker = marker_file.read_text(encoding="utf-8", errors="replace").strip().lower()
except OSError:
marker = ""
if marker is None or marker in ("brew", "homebrew", "false", "0", "no", "off"):
return False
# An empty or unreadable marker ("" — including the OSError fallback above) is the legacy
# NixOS shape and still counts as managed, matching hermes_cli.config.get_managed_system.
return True
managed_marker = get_hermes_home() / ".managed"
if marker is None and managed_marker.exists():
try:
marker = managed_marker.read_text(encoding="utf-8", errors="replace").strip().lower()
except OSError:
marker = ""
if marker is None or marker in _IGNORED_MANAGED_VALUES or marker in _MANAGED_FALSE_VALUES:
return None
if marker == "" or marker in _MANAGED_TRUE_VALUES:
return _LEGACY_MANAGED_SYSTEM
return marker
def _container_or_chmod_skipped() -> bool:
@@ -1032,27 +1043,42 @@ def _container_or_chmod_skipped() -> bool:
overrides on top of the canonical :func:`_detect_container` signals (same breadth as
:func:`is_container` — Docker/Podman/LXC/Kubernetes, cgroup and root-mountinfo). The cached
:func:`is_container` itself is deliberately avoided: it ignores these env overrides and is
computed only once per process, so tests could not flip it."""
computed only once per process, so tests could not flip it. Volume-mounted config is not
forced to owner-only in containers: gateway and dashboard may run as different UIDs, or
the mount itself needs broader permissions."""
if os.environ.get("HERMES_CONTAINER") or os.environ.get("HERMES_SKIP_CHMOD"):
return True
return _detect_container()
def _chown_dir_to_hermes_uid(path) -> None:
"""Chown *path* to ``HERMES_UID:HERMES_GID`` when set; EPERM/ENOENT are non-fatal.
def _resolve_hermes_uid_gid() -> tuple[int | None, int | None]:
"""Read HERMES_UID / HERMES_GID (set by Docker deployments); (None, None) if unset/invalid/Windows.
The entrypoint chowns HERMES_HOME once, but subdirs created at runtime (``profiles/<name>/``)
need the same chown or they land root:root and block later uid-mapped workers.
Used by :func:`apply_secure_dir_policy` so Docker deployments keep directory ownership
consistent. Unlike ``hermes_cli.config._chown_to_hermes_uid`` there is no ``win32``
early return here — on Windows the ``AttributeError`` from the missing ``os.chown``
below is what makes this a no-op.
Docker containers running Hermes commonly set these to map the in-container user to a host user so
volume-mounted state files end up with the right ownership. See #34107.
"""
def env_int(name: str):
if sys.platform == "win32":
return None, None
def _env_int(name: str) -> int | None:
try:
return int(os.environ.get(name, "").strip() or None)
except (TypeError, ValueError):
return None
uid, gid = env_int("HERMES_UID"), env_int("HERMES_GID")
return _env_int("HERMES_UID"), _env_int("HERMES_GID")
def _chown_to_hermes_uid(path) -> None:
"""Chown ``path`` to ``HERMES_UID:HERMES_GID`` when set; EPERM/ENOENT are non-fatal (the
entrypoint's startup chown -R fixes ownership on the next restart).
Used by :func:`apply_secure_dir_policy` to keep ownership consistent across all directories
created by ``ensure_hermes_home`` on Docker deployments. See #34107.
"""
uid, gid = _resolve_hermes_uid_gid()
if uid is None and gid is None:
return
try:
@@ -1074,11 +1100,11 @@ def apply_secure_dir_policy(path) -> None:
Import-safe twin of ``hermes_cli.config._secure_dir`` (which delegates here), so callers
outside the CLI package — like :func:`get_scratch_dir` — share one policy implementation.
"""
if _is_managed_home():
if get_managed_system() is not None:
return
explicit_mode = os.environ.get("HERMES_HOME_MODE", "").strip()
if _container_or_chmod_skipped() and not explicit_mode:
_chown_dir_to_hermes_uid(path)
_chown_to_hermes_uid(path)
return
try:
mode = int(explicit_mode or "700", 8)
@@ -1088,7 +1114,7 @@ def apply_secure_dir_policy(path) -> None:
os.chmod(path, mode)
except (OSError, NotImplementedError):
pass
_chown_dir_to_hermes_uid(path)
_chown_to_hermes_uid(path)
def get_scratch_dir(home: str | Path | None = None, *, prune: bool = True) -> Path:

View File

@@ -7,8 +7,9 @@ for profile namespaces under ``profiles/<name>/`` spawned by kanban
workers — were landing as ``root:root`` and blocking subsequent
uid-mapped worker invocations with ``PermissionError [Errno 13]``.
The fix is a ``_chown_to_hermes_uid`` helper that reads the env vars and
applies chown after ``mkdir``, invoked from ``_secure_dir`` (which already
The fix is a ``_chown_to_hermes_uid`` helper (``hermes_constants``, the single home of the
managed/container/HERMES_UID policy) that reads the env vars and applies chown after
``mkdir``, invoked from ``_secure_dir`` via ``apply_secure_dir_policy`` (which already
runs after every directory creation in the home-init path).
"""
from __future__ import annotations
@@ -30,7 +31,7 @@ class TestResolveHermesUidGid:
def test_returns_parsed_values_when_both_set(self, monkeypatch):
monkeypatch.setenv("HERMES_UID", "1000")
monkeypatch.setenv("HERMES_GID", "911")
from hermes_cli.config import _resolve_hermes_uid_gid
from hermes_constants import _resolve_hermes_uid_gid
uid, gid = _resolve_hermes_uid_gid()
assert uid == 1000
assert gid == 911
@@ -44,7 +45,7 @@ class TestResolveHermesUidGid:
def test_windows_returns_none_none(self, monkeypatch):
monkeypatch.setenv("HERMES_UID", "1000")
monkeypatch.setenv("HERMES_GID", "911")
from hermes_cli.config import _resolve_hermes_uid_gid
from hermes_constants import _resolve_hermes_uid_gid
uid, gid = _resolve_hermes_uid_gid()
assert uid is None
assert gid is None
@@ -59,7 +60,7 @@ class TestChownToHermesUid:
def test_calls_os_chown_when_both_set(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_UID", "1000")
monkeypatch.setenv("HERMES_GID", "911")
from hermes_cli import config as cfg
import hermes_constants as cfg
d = tmp_path / "subdir"
d.mkdir()
@@ -76,7 +77,7 @@ class TestChownToHermesUid:
user anyway."""
monkeypatch.setenv("HERMES_UID", "1000")
monkeypatch.setenv("HERMES_GID", "911")
from hermes_cli import config as cfg
import hermes_constants as cfg
d = tmp_path / "subdir"
d.mkdir()
@@ -93,7 +94,7 @@ class TestChownToHermesUid:
the helper portable."""
monkeypatch.setenv("HERMES_UID", "1000")
monkeypatch.setenv("HERMES_GID", "911")
from hermes_cli import config as cfg
import hermes_constants as cfg
d = tmp_path / "subdir"
d.mkdir()

View File

@@ -163,3 +163,26 @@ class TestScratchDirPermissionPolicy:
with patch.object(os, "chown") as mock_chown:
get_scratch_dir(tmp_path, prune=False)
mock_chown.assert_called_once_with(tmp_path / "cache" / "scratch", 1000, 911)
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX file modes")
def test_config_and_constants_share_one_policy_implementation(tmp_path, monkeypatch):
"""hermes_constants is the single home of managed / container / HERMES_UID policy: config
re-exports it (no keep-in-sync twins), so _secure_file skips on the same canonical container
signal that apply_secure_dir_policy / get_scratch_dir already honor."""
import hermes_constants
from hermes_cli import config
assert config.get_managed_system is hermes_constants.get_managed_system
assert config._chown_to_hermes_uid is hermes_constants._chown_to_hermes_uid
assert not hasattr(config, "_is_container")
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))
for var in ("HERMES_MANAGED", "HERMES_CONTAINER", "HERMES_SKIP_CHMOD"):
monkeypatch.delenv(var, raising=False)
monkeypatch.setattr("hermes_constants._detect_container", lambda: True)
f = tmp_path / "config.yaml"
f.write_text("", encoding="utf-8")
os.chmod(f, 0o640)
config._secure_file(f)
assert stat.S_IMODE(os.stat(f).st_mode) == 0o640