diff --git a/hermes_constants.py b/hermes_constants.py index a848a0df80a..a80e9763148 100644 --- a/hermes_constants.py +++ b/hermes_constants.py @@ -461,11 +461,21 @@ _container_detected: bool | None = None def is_container() -> bool: - """Return True when running inside a Docker/Podman container. + """Return True when running inside a container. - Checks ``/.dockerenv`` (Docker), ``/run/.containerenv`` (Podman), - and ``/proc/1/cgroup`` for container runtime markers. Result is - cached for the process lifetime. Import-safe — no heavy deps. + Recognizes Docker (``/.dockerenv``), Podman (``/run/.containerenv``), + and — via ``/proc/1/cgroup`` — the docker/podman/lxc cgroup-v1 markers. + + cgroup v2 collapses ``/proc/1/cgroup`` to a single ``0::/`` line with no + runtime marker, so containerd/CRI-O runtimes (the common case on + Kubernetes/k3s) were previously missed. To cover those, also check: + * ``KUBERNETES_SERVICE_HOST`` env var — set in every Kubernetes pod. + * ``kubepods`` / ``containerd`` / ``crio`` markers in ``/proc/1/cgroup``. + * the same markers in ``/proc/self/mountinfo`` (cgroup-v2 fallback). + + Result is cached for the process lifetime. Import-safe — no heavy deps. + + See: NousResearch/hermes-agent#47111 """ global _container_detected if _container_detected is not None: @@ -476,10 +486,26 @@ def is_container() -> bool: if os.path.exists("/run/.containerenv"): _container_detected = True return True + # Kubernetes always injects this into pod containers; absent on hosts. + if os.environ.get("KUBERNETES_SERVICE_HOST"): + _container_detected = True + return True + _CGROUP_MARKERS = ("docker", "podman", "/lxc/", "kubepods", "containerd", "crio") try: with open("/proc/1/cgroup", "r", encoding="utf-8") as f: cgroup = f.read() - if "docker" in cgroup or "podman" in cgroup or "/lxc/" in cgroup: + if any(marker in cgroup for marker in _CGROUP_MARKERS): + _container_detected = True + return True + except OSError: + pass + # cgroup v2: /proc/1/cgroup is just "0::/" with no marker. The container + # runtime still shows up in the mount table (overlay rootfs, runtime mount + # paths), so scan mountinfo as a last resort. + try: + with open("/proc/self/mountinfo", "r", encoding="utf-8") as f: + mountinfo = f.read() + if any(marker in mountinfo for marker in ("kubepods", "containerd", "crio")): _container_detected = True return True except OSError: diff --git a/tests/test_hermes_constants.py b/tests/test_hermes_constants.py index de347e23e08..0a9dcce3651 100644 --- a/tests/test_hermes_constants.py +++ b/tests/test_hermes_constants.py @@ -139,12 +139,66 @@ class TestIsContainer: """Returns False on a regular Linux host.""" 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("12:memory:/\n") + mountinfo_file = tmp_path / "mountinfo" + mountinfo_file.write_text("22 21 0:20 / /sys rw shared:7 - sysfs sysfs rw\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 + + def test_detects_kubernetes_env(self, monkeypatch): + """KUBERNETES_SERVICE_HOST env var triggers detection (k8s/k3s pod).""" + self._reset_cache(monkeypatch) + monkeypatch.setattr(os.path, "exists", lambda p: False) + monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.43.0.1") + assert is_container() is True + + def test_detects_cgroup_kubepods(self, monkeypatch, tmp_path): + """/proc/1/cgroup containing 'kubepods' triggers detection.""" + 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("12:memory:/kubepods/besteffort/podabc\n") _real_open = builtins.open monkeypatch.setattr("builtins.open", lambda p, *a, **kw: _real_open(str(cgroup_file), *a, **kw) if p == "/proc/1/cgroup" else _real_open(p, *a, **kw)) - assert is_container() is False + assert is_container() is True + + def test_detects_cgroup_v2_via_mountinfo(self, monkeypatch, tmp_path): + """cgroup v2 (0::/ only) falls back to containerd marker in mountinfo.""" + 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( + "1234 1233 0:42 /containerd/.../rootfs / rw - overlay overlay rw\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 True def test_caches_result(self, monkeypatch): """Second call uses cached value without re-probing."""