fix(docker): keep root out of the hermes-owned gateway scandir

/run/hermes-service is owned by hermes, so a hermes process can rename
anything root creates there and leave a symlink in its place. Root
registration built `.gateway-<p>.tmp` inside it and then chmodded and
chowned every entry, so a swap mid-build handed any root-owned directory
to hermes. Root's s6-svc/s6-svwait/s6-svscanctl calls had the same
shape: they open control FIFOs and fifodirs in that tree and followed a
hermes-planted symlink (s6-svc wrote its command byte into a root 0600
file).

- Root registration builds each slot in /run/.hermes-service.staging
  (root, 0700, same tmpfs; every ancestor must be root-controlled) and
  publishes it with one rename(2), which never follows a symlink at the
  destination. Chowns use follow_symlinks=False.
- S6ServiceManager runs its s6 tools through `s6-setuidgid hermes` when
  called as root, so lifecycle control acts with hermes's own rights.
- Non-root registration is unchanged.

Tests: unit cases for the staging swap, the ancestor/staging trust
check and the s6 privilege drop; tests/docker/test_root_registration_
symlink_races.py plays the hermes process in a real container. Both
Docker cases fail with the previous service_manager.py copied in
(swap succeeds; victim file becomes "tRIGINAL") and pass now.

Reported-by: Adolanium

hermes_cli/service_manager.py

@@ -6,7 +6,9 @@ import os
 import re
 import shlex
 import shutil
+import stat
 import subprocess
+import tempfile
 import time
 from pathlib import Path
 from typing import Literal, Protocol, runtime_checkable
@@ -318,18 +320,61 @@ def s6_service_up(service_dir: Path) -> bool:
     return result.returncode == 0 and result.stdout.split() == ["true"]
 
 
-def _chown_hermes(path: Path) -> None:
-    """Root-created slot entries belong to the current, possibly remapped hermes account."""
+def _root_hermes_account():
+    """The current (possibly remapped) hermes account when running as root, else None."""
     if not hasattr(os, "geteuid") or os.geteuid() != 0:
-        return
+        return None
     import pwd
 
     try:
-        account = pwd.getpwnam("hermes")
+        return pwd.getpwnam("hermes")
     except KeyError:
         # Outside the image there may be no hermes account; retain the existing owner.
-        return
-    os.chown(path, account.pw_uid, account.pw_gid)
+        return None
+
+
+def _chown_hermes(path: Path) -> None:
+    """Root-created slot entries belong to the current, possibly remapped hermes account."""
+    account = _root_hermes_account()
+    if account is not None:
+        os.chown(path, account.pw_uid, account.pw_gid, follow_symlinks=False)
+
+
+def _require_root_controlled(path: Path, *, private: bool) -> os.stat_result:
+    """lstat ``path`` and refuse it unless only root can rename or modify it.
+
+    ``private`` additionally requires a 0700-style directory that nobody else can traverse.
+    """
+    st = os.lstat(path)
+    if not stat.S_ISDIR(st.st_mode) or st.st_uid != 0:
+        raise RuntimeError(f"{path} must be a root-owned directory, not a symlink")
+    if private and st.st_mode & 0o077:
+        raise RuntimeError(f"{path} must not be accessible to group or others")
+    if st.st_mode & 0o022 and not st.st_mode & stat.S_ISVTX:
+        raise RuntimeError(f"{path} must not be writable by group or others")
+    return st
+
+
+def _root_staging_dir(scandir: Path) -> Path:
+    """Root-private build area beside ``scandir``, on the same filesystem.
+
+    The gateway scandir is hermes-owned, so anything root builds inside it can be renamed and
+    swapped for a symlink mid-build, redirecting root's chmod/chown/writes. Root therefore builds
+    slots in ``.<scandir>.staging`` (root, 0700) and publishes them with one atomic rename. Every
+    ancestor must be root-controlled, or an unprivileged user could rename the staging dir away.
+    """
+    parent = scandir.parent
+    for ancestor in (parent, *parent.parents):
+        _require_root_controlled(ancestor, private=False)
+    staging = parent / f".{scandir.name}.staging"
+    try:
+        os.mkdir(staging, 0o700)
+    except FileExistsError:
+        pass
+    st = _require_root_controlled(staging, private=True)
+    if os.stat(scandir).st_dev != st.st_dev:
+        raise RuntimeError(f"{staging} and {scandir} must share a filesystem for an atomic publish")
+    return staging
 
 
 def _seed_supervise_skeleton(svc_dir: Path) -> None:
@@ -426,6 +471,18 @@ class S6ServiceManager:
     def __init__(self, scandir: Path = S6_DYNAMIC_SCANDIR) -> None:
         self.scandir = scandir
 
+    @staticmethod
+    def _s6(cmd: str, *args: str, timeout: float = 5, check: bool = False):
+        """Run an s6 tool against the hermes-owned tree, as hermes when called as root.
+
+        s6-svc/s6-svscanctl/s6-svwait open FIFOs and fifodirs inside that tree; as root they would
+        follow a hermes-planted symlink and write to or create entries in any root-owned path.
+        """
+        if _root_hermes_account() is not None:
+            return _s6_run("s6-setuidgid", "hermes", f"{_S6_BIN_DIR}/{cmd}", *args,
+                           timeout=timeout, check=check)
+        return _s6_run(cmd, *args, timeout=timeout, check=check)
+
     def _service_dir(self, profile: str) -> Path:
         validate_profile_name(profile)
         return self.scandir / f"{S6_SERVICE_PREFIX}{profile}"
@@ -551,7 +608,7 @@ class S6ServiceManager:
         if not service_dir.is_dir():
             raise GatewayNotRegisteredError(_profile_from_service(name))
         try:
-            _s6_run("s6-svc", action_flag, str(service_dir), check=True)
+            self._s6("s6-svc", action_flag, str(service_dir), check=True)
         except subprocess.CalledProcessError as exc:
             raise S6CommandError(
                 service=name, action=action_label, returncode=exc.returncode, stderr=exc.stderr or ""
@@ -567,7 +624,7 @@ class S6ServiceManager:
         Never parse the human status line: s6 2.15 prints ``up (pid N pgid N) Ss``, which the old
         ``(pid N)`` regex never matched, so the planned-stop marker was silently skipped.
         """
-        result = _s6_run("s6-svstat", "-o", ",".join(fields), str(self.scandir / name))
+        result = self._s6("s6-svstat", "-o", ",".join(fields), str(self.scandir / name))
         if result.returncode != 0:
             return None
         values = result.stdout.split()
@@ -621,7 +678,7 @@ class S6ServiceManager:
         if ``s6-svscanctl`` fails.
         """
         svc_dir = self._service_dir(profile)
-        if svc_dir.exists():
+        if os.path.lexists(svc_dir):
             raise ValueError(f"profile gateway {profile!r} already registered at {svc_dir}")
         # Build atomically in a DOT-PREFIXED sibling (``.gateway-<profile>.tmp``) then rename:
         # s6-svscan skips dot entries, so a concurrent rescan (cont-init reconciler, sibling
@@ -629,10 +686,18 @@ class S6ServiceManager:
         # ROOT on the ``.tmp`` (it already has ``type``/``run``), mkdir ``supervise/`` root-owned
         # 0700, and our ``_seed_supervise_skeleton`` would EACCES on ``supervise/event`` — the
         # arm64-only CI flake on test_s6_unregister_removes_service_dir_in_live_container.
-        tmp_dir = svc_dir.with_name("." + svc_dir.name + ".tmp")
-        if tmp_dir.exists():
-            shutil.rmtree(tmp_dir, ignore_errors=True)
-        tmp_dir.mkdir(parents=True)
+        #
+        # Root never builds inside the hermes-owned scandir: hermes could swap the dot dir for a
+        # symlink mid-build and redirect root's writes, chmods and chowns. Root builds in a
+        # root-private staging dir instead and publishes with a single rename.
+        if _root_hermes_account() is not None:
+            tmp_dir = Path(tempfile.mkdtemp(prefix=f"{svc_dir.name}.",
+                                            dir=_root_staging_dir(self.scandir)))
+        else:
+            tmp_dir = svc_dir.with_name("." + svc_dir.name + ".tmp")
+            if tmp_dir.exists():
+                shutil.rmtree(tmp_dir, ignore_errors=True)
+            tmp_dir.mkdir(parents=True)
 
         def _write_script(path: Path, text: str) -> None:
             path.write_text(text, encoding="utf-8")
@@ -656,11 +721,12 @@ class S6ServiceManager:
                 if path.is_dir() and path in (tmp_dir, tmp_dir / "log"):
                     path.chmod(0o755)
                 _chown_hermes(path)
-            tmp_dir.rename(svc_dir)
+            # rename(2) never follows a symlink at svc_dir and refuses a non-empty directory.
+            os.rename(tmp_dir, svc_dir)
         except Exception:
             shutil.rmtree(tmp_dir, ignore_errors=True)
             raise
-        result = _s6_run("s6-svscanctl", "-a", str(self.scandir))
+        result = self._s6("s6-svscanctl", "-a", str(self.scandir))
         if result.returncode != 0:
             # No supervisor is watching it — leaving the directory would be confusing.
             shutil.rmtree(svc_dir, ignore_errors=True)
@@ -678,9 +744,9 @@ class S6ServiceManager:
         svc_dir = self._service_dir(profile)
         if not svc_dir.exists():
             return
-        _s6_run("s6-svc", "-d", str(svc_dir))
-        _s6_run("s6-svwait", "-D", "-t", "10000", str(svc_dir), timeout=15)
-        _s6_run("s6-svscanctl", "-an", str(self.scandir))
+        self._s6("s6-svc", "-d", str(svc_dir))
+        self._s6("s6-svwait", "-D", "-t", "10000", str(svc_dir), timeout=15)
+        self._s6("s6-svscanctl", "-an", str(self.scandir))
         # No synchronous "scan completed" handshake — -a/-n just set a flag s6-svscan reads on its
         # next loop; 200ms is comfortably above that resolution.
         time.sleep(0.2)

optional-skills/devops/hermes-s6-container-supervision/SKILL.md

@@ -158,6 +158,10 @@ The cont-init reconciler runs as hermes (`s6-setuidgid hermes` in `02-reconcile-
 
 `docker exec` defaults to root. Either pass `--user hermes` or rely on the stage2 chown sweep next reboot. Don't write files under `$HERMES_HOME/profiles/<name>/` as root manually — the next reconcile pass will sweep them but in-flight operations may hit perm errors.
 
+### Root code must not work inside `/run/hermes-service`
+
+The gateway scandir is hermes-owned, so hermes can rename anything in it and leave a symlink behind. Root writes, chmods and chowns there follow that symlink. Root registration therefore builds each slot in `/run/.hermes-service.staging` (root, 0700, same tmpfs) and publishes it with one `rename`; `S6ServiceManager` runs `s6-svc`, `s6-svwait` and `s6-svscanctl` through `s6-setuidgid hermes` when called as root. Keep new root-side code on those paths.
+
 ### Service slot exists but s6-svstat says "s6-supervise not running"
 
 The service directory is on tmpfs and was wiped on container restart. Either the cont-init reconciler hasn't run yet (give it a moment after `docker restart`) or it failed. Check `docker logs <c> | grep '02-reconcile'`.

tests/docker/test_root_registration_symlink_races.py

@@ -0,0 +1,92 @@
+"""Root acting on the hermes-owned gateway tree must not follow hermes-planted symlinks.
+
+``/run/hermes-service`` is owned by hermes, so a hermes process can rename anything root
+creates there and put a symlink in its place. These tests play that hermes process for real,
+inside the image, against root registration (``HERMES_DOCKER_EXEC_AS_ROOT=1`` or a direct
+Python call) and root lifecycle control.
+"""
+from __future__ import annotations
+
+from tests.docker.conftest import docker_exec, docker_exec_sh, poll_container, start_container
+
+SCANDIR = "/run/hermes-service"
+STAGING = "/run/.hermes-service.staging"
+
+# Runs as root. After the slot's supervise skeleton is seeded, the hermes account tries to
+# swap every half-built slot it can reach (in the scandir and in root's staging dir).
+_SWAP_DURING_REGISTRATION = f"""
+import os, subprocess
+from hermes_cli import service_manager as sm
+assert os.geteuid() == 0
+os.umask(0o077)
+os.mkdir('/opt/victim', 0o700)
+seed = sm._seed_supervise_skeleton
+attack = (
+    'for d in {SCANDIR}/.gateway-* {STAGING}/gateway-*; do '
+    '[ -e "$d" ] || continue; '
+    'mv "$d" "$d.moved" && ln -s /opt/victim "$d" && echo "swapped $d"; '
+    'done; true'
+)
+def seed_then_swap(svc_dir):
+    seed(svc_dir)
+    out = subprocess.run(['/command/s6-setuidgid', 'hermes', 'sh', '-c', attack],
+                         capture_output=True, text=True)
+    print('attacker:', repr(out.stdout), repr(out.stderr))
+sm._seed_supervise_skeleton = seed_then_swap
+sm.S6ServiceManager().register_profile_gateway('swap', start_now=False)
+"""
+
+
+def test_root_registration_survives_staging_swap(built_image: str, container_name: str) -> None:
+    start_container(built_image, container_name)
+    registered = docker_exec(container_name, "python3", "-c", _SWAP_DURING_REGISTRATION,
+                             user="root", extra_docker_args=("-e", "HERMES_DOCKER_EXEC_AS_ROOT=1"))
+    assert registered.returncode == 0, registered.stderr
+    # hermes could not reach root's half-built slot anywhere.
+    assert "swapped" not in registered.stdout, registered.stdout
+    victim = docker_exec_sh(container_name, "stat -c '%u:%g %a' /opt/victim", user="root")
+    assert victim.stdout.strip() == "0:0 700", victim.stdout
+    staging = docker_exec_sh(container_name, f"stat -c '%U:%G %a' {STAGING}; ls -A {STAGING}",
+                             user="root")
+    assert staging.stdout.splitlines() == ["root:root 700"], staging.stdout
+    slot = f"{SCANDIR}/gateway-swap"
+    published = docker_exec_sh(container_name, f"test -d {slot} && test ! -L {slot} && "
+                               f"stat -c '%U:%G' {slot} {slot}/run {slot}/supervise")
+    assert published.returncode == 0, published.stderr
+    assert published.stdout.splitlines() == ["hermes:hermes"] * 3
+    ok, state = poll_container(container_name, f"/command/s6-svstat {slot}", deadline_s=30)
+    assert ok, state
+
+
+def test_root_lifecycle_control_does_not_follow_control_symlink(
+    built_image: str, container_name: str,
+) -> None:
+    """s6-svc as root would write its command byte through a hermes-planted control symlink."""
+    start_container(built_image, container_name)
+    registered = docker_exec(container_name, "python3", "-c", (
+        "from hermes_cli.service_manager import S6ServiceManager; "
+        "S6ServiceManager().register_profile_gateway('ctl', start_now=False)"
+    ), user="root", extra_docker_args=("-e", "HERMES_DOCKER_EXEC_AS_ROOT=1"))
+    assert registered.returncode == 0, registered.stderr
+    slot = f"{SCANDIR}/gateway-ctl"
+    ok, state = poll_container(container_name, f"/command/s6-svstat {slot}", deadline_s=30)
+    assert ok, state
+    seeded = docker_exec_sh(container_name, (
+        "printf ORIGINAL > /opt/victim-file && chmod 0600 /opt/victim-file"
+    ), user="root")
+    assert seeded.returncode == 0, seeded.stderr
+    planted = docker_exec_sh(container_name, (
+        f"rm {slot}/supervise/control && ln -s /opt/victim-file {slot}/supervise/control"
+    ))
+    assert planted.returncode == 0, planted.stderr
+    control = docker_exec(container_name, "python3", "-c", (
+        "from hermes_cli.service_manager import S6ServiceManager, S6CommandError\n"
+        "try:\n"
+        "    S6ServiceManager().restart('gateway-ctl')\n"
+        "except S6CommandError as exc:\n"
+        "    print('refused', exc)\n"
+    ), user="root", extra_docker_args=("-e", "HERMES_DOCKER_EXEC_AS_ROOT=1"))
+    assert control.returncode == 0, control.stderr
+    victim = docker_exec_sh(container_name, "cat /opt/victim-file; echo; stat -c '%u %a' "
+                            "/opt/victim-file", user="root")
+    assert victim.stdout.splitlines() == ["ORIGINAL", "0 600"], victim.stdout

tests/hermes_cli/test_service_manager_ownership.py

@@ -1,4 +1,5 @@
 """Root registration follows the runtime hermes account, including UID remaps."""
+import stat
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import Mock
@@ -10,39 +11,146 @@ from hermes_cli import service_manager as sm
 pytestmark = pytest.mark.platforms("posix")
 
 
+def _as_root(monkeypatch, *, uid=1000, gid=1000):
+    import pwd
+
+    monkeypatch.setattr(sm.os, "geteuid", lambda: 0)
+    lookup = Mock(return_value=SimpleNamespace(pw_uid=uid, pw_gid=gid))
+    monkeypatch.setattr(pwd, "getpwnam", lookup)
+    return lookup
+
+
 @pytest.mark.parametrize("root", [True, False])
 def test_registration_uses_current_hermes_owner(tmp_path, monkeypatch, root):
     import pwd
 
+    scandir, staging = tmp_path / "scan", tmp_path / "staging"
+    scandir.mkdir()
+    staging.mkdir()
     monkeypatch.setattr(sm.os, "geteuid", lambda: 0 if root else 1000)
     lookup = Mock(return_value=SimpleNamespace(pw_uid=1000, pw_gid=1000))
     monkeypatch.setattr(pwd, "getpwnam", lookup)
+    staging_for = Mock(return_value=staging)
+    monkeypatch.setattr(sm, "_root_staging_dir", staging_for)
     chown = Mock()
     monkeypatch.setattr(sm.os, "chown", chown)
     monkeypatch.setattr(sm, "_s6_run", lambda *a, **kw: SimpleNamespace(returncode=0))
     # Even a root caller's restrictive umask must not hide scripts or slot directories.
     old_umask = sm.os.umask(0o077)
     try:
-        sm.S6ServiceManager(tmp_path).register_profile_gateway("remapped", start_now=False)
+        sm.S6ServiceManager(scandir).register_profile_gateway("remapped", start_now=False)
     finally:
         sm.os.umask(old_umask)
-    slot = tmp_path / "gateway-remapped"
+    slot = scandir / "gateway-remapped"
     if root:
-        owned = {Path(call.args[0]).relative_to(tmp_path) for call in chown.call_args_list}
-        expected = {Path(".gateway-remapped.tmp") / p.relative_to(slot)
-                    for p in [slot, *slot.rglob("*")]}
-        assert expected <= owned
+        # Built in the root-private staging dir, then published by rename.
+        staging_for.assert_called_once_with(scandir)
+        owned = {Path(*Path(call.args[0]).relative_to(staging).parts[1:])
+                 for call in chown.call_args_list}
+        assert {p.relative_to(slot) for p in [slot, *slot.rglob("*")]} | {Path()} <= owned
         assert {call.args[1:] for call in chown.call_args_list} == {(1000, 1000)}
+        assert {call.kwargs.get("follow_symlinks") for call in chown.call_args_list} == {False}
+        assert list(staging.iterdir()) == []
         lookup.assert_called_with("hermes")
     else:
+        staging_for.assert_not_called()
         chown.assert_not_called()
         lookup.assert_not_called()
+    assert not slot.is_symlink()
     for name in ("run", "finish", "log/run"):
         assert (slot / name).stat().st_mode & 0o555 == 0o555
     for name in (".", "log"):
         assert (slot / name).stat().st_mode & 0o555 == 0o555
 
 
+def test_root_registration_ignores_staging_swap_in_scandir(tmp_path, monkeypatch):
+    """hermes owns the scandir, so it can swap any dot dir root builds there for a symlink."""
+    scandir, staging, victim = tmp_path / "scan", tmp_path / "staging", tmp_path / "victim"
+    for path in (scandir, staging, victim):
+        path.mkdir()
+    victim.chmod(0o700)
+    _as_root(monkeypatch)
+    monkeypatch.setattr(sm, "_root_staging_dir", lambda _scandir: staging, raising=False)
+    chown = Mock()
+    monkeypatch.setattr(sm.os, "chown", chown)
+    monkeypatch.setattr(sm, "_s6_run", lambda *a, **kw: SimpleNamespace(returncode=0))
+    seed = sm._seed_supervise_skeleton
+
+    def seed_then_swap(svc_dir):
+        seed(svc_dir)
+        # The concurrent hermes process: rename the half-built slot, plant a symlink.
+        for entry in list(scandir.iterdir()):
+            if entry.name.startswith("."):
+                entry.rename(scandir / f"moved{entry.name}")
+                entry.symlink_to(victim)
+
+    monkeypatch.setattr(sm, "_seed_supervise_skeleton", seed_then_swap)
+    sm.S6ServiceManager(scandir).register_profile_gateway("swap", start_now=False)
+
+    assert victim.stat().st_mode & 0o777 == 0o700
+    assert list(victim.iterdir()) == []
+    touched = {Path(call.args[0]).resolve() for call in chown.call_args_list}
+    assert not {p for p in touched if p == victim or victim in p.parents}
+    slot = scandir / "gateway-swap"
+    assert slot.is_dir() and not slot.is_symlink()
+    assert (slot / "run").is_file()
+
+
+@pytest.mark.parametrize("st_mode, st_uid, private, ok", [
+    (stat.S_IFDIR | 0o755, 0, False, True),
+    (stat.S_IFDIR | 0o1777, 0, False, True),       # sticky: others cannot rename root's entries
+    (stat.S_IFDIR | 0o700, 0, True, True),
+    (stat.S_IFDIR | 0o755, 0, True, False),        # staging must be private
+    (stat.S_IFDIR | 0o775, 0, False, False),       # group-writable parent
+    (stat.S_IFDIR | 0o757, 0, False, False),       # world-writable parent
+    (stat.S_IFDIR | 0o755, 10000, False, False),   # hermes-owned
+    (stat.S_IFLNK | 0o777, 0, False, False),       # symlink
+])
+def test_require_root_controlled(monkeypatch, st_mode, st_uid, private, ok):
+    fake = SimpleNamespace(st_mode=st_mode, st_uid=st_uid)
+    monkeypatch.setattr(sm.os, "lstat", lambda path: fake)
+    if ok:
+        assert sm._require_root_controlled(Path("/x"), private=private) is fake
+    else:
+        with pytest.raises(RuntimeError):
+            sm._require_root_controlled(Path("/x"), private=private)
+
+
+def test_root_staging_refuses_unprivileged_ancestor(tmp_path, monkeypatch):
+    """Test dirs are owned by the test user, which stands in for an untrusted ancestor."""
+    monkeypatch.setattr(sm.os, "geteuid", lambda: 0)
+    scandir = tmp_path / "scan"
+    scandir.mkdir()
+    with pytest.raises(RuntimeError, match="root-owned"):
+        sm._root_staging_dir(scandir)
+    assert not (tmp_path / ".scan.staging").exists()
+
+
+@pytest.mark.parametrize("root", [True, False])
+def test_s6_tools_drop_to_hermes_when_root(tmp_path, monkeypatch, root):
+    """As root, s6 tools would follow hermes-planted symlinks for control FIFOs/fifodirs."""
+    if root:
+        _as_root(monkeypatch)
+    else:
+        monkeypatch.setattr(sm.os, "geteuid", lambda: 1000)
+    (tmp_path / "gateway-p").mkdir()
+    calls = []
+    monkeypatch.setattr(sm, "_s6_run", lambda *a, **kw: calls.append(a) or SimpleNamespace(
+        returncode=0, stdout="true\n", stderr=""))
+    monkeypatch.setattr(sm, "_write_gateway_desired_state", lambda *a: None)
+    mgr = sm.S6ServiceManager(tmp_path)
+    mgr.restart("gateway-p")
+    mgr.is_running("gateway-p")
+    mgr.unregister_profile_gateway("p")
+    assert calls
+    for call in calls:
+        if root:
+            assert call[:2] == ("s6-setuidgid", "hermes")
+            assert call[2].startswith(f"{sm._S6_BIN_DIR}/s6-")
+        else:
+            assert call[0].startswith("s6-") and call[0] != "s6-setuidgid"
+
+
 def test_seed_uses_remapped_account(tmp_path, monkeypatch):
     import pwd