refactor(constants): share the /proc reader between the cgroup and root-mount probes

_root_mount_has_marker duplicated _proc_file_has_marker's open/OSError shell;
both now go through _read_proc(), and the root-mount scan checks every "/"
line instead of returning on the first. The regression test exercises the
helper on both arms (host running containers -> False, containerd rootfs ->
True, missing file -> False) instead of patching builtins.open globally.
This commit is contained in:
kshitijk4poor
2026-09-08 02:14:08 +05:30
committed by kshitij
parent 9160c4e27e
commit 7b29f19146
2 changed files with 29 additions and 46 deletions
+10 -12
View File
@@ -1042,12 +1042,16 @@ def is_container() -> bool:
return _container_detected
def _proc_file_has_marker(path: str, markers: tuple[str, ...]) -> bool:
def _read_proc(path: str) -> str:
try:
with open(path, "r", encoding="utf-8") as f:
content = f.read()
return f.read()
except OSError:
return False
return ""
def _proc_file_has_marker(path: str, markers: tuple[str, ...]) -> bool:
content = _read_proc(path)
return any(marker in content for marker in markers)
@@ -1067,15 +1071,9 @@ def _detect_container() -> bool:
def _root_mount_has_marker(path: str, markers: tuple[str, ...]) -> bool:
try:
with open(path, "r", encoding="utf-8") as f:
for line in f:
fields = line.split() # mountinfo field 5 (index 4) is the mount point
if len(fields) >= 5 and fields[4] == "/":
return any(marker in line for marker in markers)
except OSError:
pass
return False
"""mountinfo field 5 (index 4) is the mount point; only the root ("/") line is the process's own rootfs."""
root_lines = [line for line in _read_proc(path).splitlines() if len(f := line.split()) >= 5 and f[4] == "/"]
return any(marker in line for line in root_lines for marker in markers)
def get_config_path() -> Path:
+19 -34
View File
@@ -363,43 +363,28 @@ class TestIsContainer:
def test_host_running_containers_not_false_positive(self, monkeypatch, tmp_path):
"""A host that merely RUNS containers must not be classified as one.
def test_cgroup_v2_fallback_inspects_only_the_root_mount(self, tmp_path):
"""#58135: a host that merely RUNS containers exposes each container's overlay lowerdir
(``lowerdir=/var/lib/containerd/...``) at non-root mount points; only the root ('/') line
says whether *this* process lives in a runtime overlay."""
from hermes_constants import _root_mount_has_marker
Regression for NousResearch/hermes-agent#58135: on a cgroup-v2 host
with Docker's containerd image store, each running container adds an
overlay mount whose option string contains
``lowerdir=/var/lib/containerd/...``. The marker appears only in
non-root mount lines, so scanning the whole file produced a false
positive. Only the root ('/') mount line should be inspected.
"""
import builtins
self._reset_cache(monkeypatch)
monkeypatch.delenv("KUBERNETES_SERVICE_HOST", raising=False)
monkeypatch.setattr(os.path, "exists", lambda p: False)
cgroup_file = tmp_path / "cgroup"
cgroup_file.write_text("0::/\n") # cgroup v2 — no runtime marker
mountinfo_file = tmp_path / "mountinfo"
mountinfo_file.write_text(
# Root is a real block device on the host.
markers = ("kubepods", "containerd", "crio")
host = tmp_path / "host"
host.write_text(
"25 1 259:2 / / rw,relatime shared:1 - ext4 /dev/nvme0n1p2 rw\n"
# A running container's overlay rootfs mounted elsewhere — its
# lowerdir references containerd but must NOT flip the host.
"469 554 0:94 / /var/lib/docker/rootfs/overlayfs/7dda83 rw,relatime "
"shared:247 - overlay overlay rw,lowerdir=/var/lib/containerd/"
"io.containerd.snapshotter.v1.overlayfs/snapshots/33509/fs\n"
"469 554 0:94 / /var/lib/docker/rootfs/overlayfs/7dda83 rw,relatime shared:247 - overlay overlay "
"rw,lowerdir=/var/lib/containerd/io.containerd.snapshotter.v1.overlayfs/snapshots/33509/fs\n"
)
_real_open = builtins.open
def _fake_open(p, *a, **kw):
if p == "/proc/1/cgroup":
return _real_open(str(cgroup_file), *a, **kw)
if p == "/proc/self/mountinfo":
return _real_open(str(mountinfo_file), *a, **kw)
return _real_open(p, *a, **kw)
monkeypatch.setattr("builtins.open", _fake_open)
assert is_container() is False
container = tmp_path / "container"
container.write_text(
"1 0 0:50 / / rw,relatime - overlay overlay "
"rw,lowerdir=/var/lib/containerd/io.containerd.snapshotter.v1.overlayfs/snapshots/9/fs\n"
"2 1 0:51 / /proc rw,nosuid - proc proc rw\n"
)
assert _root_mount_has_marker(str(host), markers) is False
assert _root_mount_has_marker(str(container), markers) is True
assert _root_mount_has_marker(str(tmp_path / "missing"), markers) is False
def test_caches_result(self, monkeypatch):
"""Second call uses cached value without re-probing."""