fix(egress): close three P0 security gaps in iron-proxy integration

P0-1: iron_proxy_version() leaked full host env to subprocess
- The version probe ('--version') is a one-shot subprocess call but
  inherited os.environ including OPENAI_API_KEY, ANTHROPIC_API_KEY,
  BWS_ACCESS_TOKEN, etc.
- Build a minimal env from _PROXY_SUBPROCESS_ENV_ALLOWLIST (PATH, HOME,
  locale) and pass it explicitly.  The S603 comment is updated to
  acknowledge the PATH-fallback risk is real but mitigated by the
  scrubbed env.

P0-2: Bitwarden ImportError branch never checked allow_env_fallback
- The sibling branches (missing-secret, empty-token) honor the flag;
  the ImportError branch silently fell through to host env regardless.
- Mirror the sibling behavior: raise unless allow_env_fallback is set.
  A wizard-time check cannot catch a dependency that goes missing
  between setup and a later restart.

P0-3: Container reuse could cross egress on -> off boundary
- When egress_label is 'off', _find_reusable_container skipped the
  egress label filter entirely, matching containers by (task, profile)
  alone.  An operator running 'hermes egress disable' could end up
  reusing a container created under egress=on with baked-in proxy env
  vars and CA mounts.
- Add a post-filter: when egress=off, parse the hermes-egress label
  from the docker ps output (3-field format) and reject containers
  whose label is present and not 'off'.

Tests: 3 new tests
- test_bitwarden_importerror_raise_without_fallback
- test_bitwarden_importerror_honor_allow_env_fallback
- test_reuse_off_rejects_non_off_egress_container
- Updated _mock_subprocess_run_with_reuse and inline mocks to 3-field
- All 191 iron_proxy + docker tests pass (0 regressions)

Review: @maxpetrusenko

agent/proxy_sources/iron_proxy.py

@@ -553,11 +553,26 @@ def iron_proxy_version(binary: Path) -> str:
         return cached
 
     try:
-        res = subprocess.run(  # noqa: S603 — binary path is trusted
+        # Build a minimal env: only PATH, HOME, and locale vars.
+        # The version probe is a one-shot subprocess — forwarding
+        # the full host env (OPENAI_API_KEY, ANTHROPIC_API_KEY, etc.)
+        # to a PATH-resolved or unverified binary is an unnecessary
+        # credential leak.  Reuse the same allowlist the daemon
+        # subprocess uses (see _build_proxy_subprocess_env).
+        minimal_env: Dict[str, str] = {}
+        parent = os.environ
+        for name in _PROXY_SUBPROCESS_ENV_ALLOWLIST:
+            if name in parent:
+                minimal_env[name] = parent[name]
+        # The S603 warning is legitimate for the PATH-fallback case
+        # (find_iron_proxy → shutil.which), but --version with a
+        # scrubbed env is safe regardless of binary provenance.
+        res = subprocess.run(  # noqa: S603
             [str(binary), "--version"],
             capture_output=True,
             text=True,
             timeout=_RUN_TIMEOUT,
+            env=minimal_env,
         )
     except (OSError, subprocess.TimeoutExpired):
         return ""
@@ -1859,12 +1874,25 @@ def _build_proxy_subprocess_env(
                     "(allow_env_fallback=true).",
                 )
         except (ImportError,) as exc:
-            # Truly unrecoverable import failure: log + fall through to
-            # the env path so we don't completely block start.  Callers
-            # that want strict mode can pre-check at the wizard layer.
+            # The BWS module or one of its runtime deps isn't importable.
+            # Mirror the sibling branches: if allow_env_fallback isn't
+            # explicitly enabled, fail closed — credential_source=bitwarden
+            # with a unavailable module should not silently degrade to host
+            # env.  A wizard-time check can't catch a dependency that goes
+            # missing between setup and a later restart.
+            if not (bitwarden_config or {}).get("allow_env_fallback"):
+                raise RuntimeError(
+                    "Bitwarden refresh module unavailable at proxy start "
+                    "(credential_source=bitwarden with "
+                    "proxy.allow_env_fallback: false).  Either fix the "
+                    "import, switch to credential_source: env, or set "
+                    "`proxy.allow_env_fallback: true` to opt into the "
+                    "legacy fallback behaviour."
+                ) from exc
             logger.warning(
                 "Bitwarden refresh module unavailable at proxy start, "
-                "falling back to parent env: %s", exc,
+                "falling back to parent env (allow_env_fallback=true): %s",
+                exc,
             )
 
     # Caller-supplied overrides win.  This is intentionally last so the

tests/test_iron_proxy.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 
 import io
 import os
+import sys
 import tarfile
 from pathlib import Path
 from unittest.mock import MagicMock, patch
@@ -1620,3 +1621,79 @@ def test_partial_bitwarden_secrets_raise_without_fallback(
         ip._build_proxy_subprocess_env(
             refresh_from_bitwarden=True, bitwarden_config=bw_cfg,
         )
+
+
+def test_bitwarden_importerror_raise_without_fallback(
+    hermes_home, monkeypatch,
+):
+    """Strict default: ImportError on BWS module raises when
+    allow_env_fallback is unset, matching the sibling branches."""
+
+    ip.write_mappings([_sample_mapping("OPENROUTER_API_KEY")])
+    monkeypatch.setenv("OPENROUTER_API_KEY", "sk-host")
+
+    # Simulate the BWS SDK not being installed.  The lazy import
+    # ``from agent.secret_sources import bitwarden`` inside
+    # _build_proxy_subprocess_env resolves through the parent package's
+    # cached attribute; deleting both the sys.modules entry AND the
+    # parent-package attribute forces a real import that we intercept.
+    #
+    # In addition, block importlib.reload in case the test infra used it.
+    import agent.secret_sources as ss
+    monkeypatch.delitem(sys.modules, "agent.secret_sources.bitwarden", raising=False)
+    monkeypatch.delitem(sys.modules, "agent.secret_sources.bitwarden.bws", raising=False)
+    monkeypatch.delattr(ss, "bitwarden", raising=False)
+
+    # Now block the re-import.  ``from agent.secret_sources import
+    # bitwarden`` resolves to a submodule attribute; setting it to a
+    # sentinel that raises on attribute access is more reliable than
+    # trying to intercept __import__ at the C level.
+    class _MissingBWS:
+        """Sentinel: accessing any attribute raises ImportError."""
+        def __getattr__(self, _name):
+            raise ImportError("bws SDK not installed")
+        def __call__(self, *a, **kw):
+            raise ImportError("bws SDK not installed")
+    monkeypatch.setattr(ss, "bitwarden", _MissingBWS(), raising=False)
+
+    monkeypatch.setenv("BWS_ACCESS_TOKEN", "tok")
+    bw_cfg = {"project_id": "proj", "access_token_env": "BWS_ACCESS_TOKEN"}
+
+    with pytest.raises(RuntimeError, match="Bitwarden refresh module unavailable"):
+        ip._build_proxy_subprocess_env(
+            refresh_from_bitwarden=True, bitwarden_config=bw_cfg,
+        )
+
+
+def test_bitwarden_importerror_honor_allow_env_fallback(
+    hermes_home, monkeypatch,
+):
+    """With allow_env_fallback, an ImportError falls through to host env
+    instead of raising."""
+
+    ip.write_mappings([_sample_mapping("OPENROUTER_API_KEY")])
+    monkeypatch.setenv("OPENROUTER_API_KEY", "sk-host-fallback")
+
+    import agent.secret_sources as ss
+    monkeypatch.delitem(sys.modules, "agent.secret_sources.bitwarden", raising=False)
+    monkeypatch.delitem(sys.modules, "agent.secret_sources.bitwarden.bws", raising=False)
+    monkeypatch.delattr(ss, "bitwarden", raising=False)
+
+    class _MissingBWS:
+        def __getattr__(self, _name):
+            raise ImportError("bws SDK not installed")
+        def __call__(self, *a, **kw):
+            raise ImportError("bws SDK not installed")
+    monkeypatch.setattr(ss, "bitwarden", _MissingBWS(), raising=False)
+
+    monkeypatch.setenv("BWS_ACCESS_TOKEN", "tok")
+    bw_cfg = {
+        "project_id": "proj",
+        "access_token_env": "BWS_ACCESS_TOKEN",
+        "allow_env_fallback": True,
+    }
+
+    env = ip._build_proxy_subprocess_env(
+        refresh_from_bitwarden=True, bitwarden_config=bw_cfg,
+    )
+    assert env.get("OPENROUTER_API_KEY") == "sk-host-fallback"

tests/tools/test_docker_environment.py

@@ -707,8 +707,13 @@ def _mock_subprocess_run_with_reuse(monkeypatch, ps_state: str | None,
             if sub == "ps":
                 if ps_state is None:
                     return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
+                # 3-field format: ID, State, EgressLabel.  When egress_label
+                # is "off" the code parses all three fields; <no value> means
+                # the container has no egress label, which is acceptable.
                 return subprocess.CompletedProcess(
-                    cmd, 0, stdout=f"reused-cid\t{ps_state}\n", stderr="",
+                    cmd, 0,
+                    stdout=f"reused-cid\t{ps_state}\t<no value>\n",
+                    stderr="",
                 )
             if sub == "start":
                 if not start_succeeds:
@@ -994,10 +999,11 @@ def test_find_reusable_container_prefers_running_over_stopped(monkeypatch):
             if cmd[1] == "version":
                 return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")
             if cmd[1] == "ps":
-                # Two matches: stopped first, running second.
+                # Two matches: stopped first, running second.  3-field format
+                # with absent egress label for the "off" path.
                 return subprocess.CompletedProcess(
                     cmd, 0,
-                    stdout="stopped-cid\texited\nrunning-cid\trunning\n",
+                    stdout="stopped-cid\texited\t<no value>\nrunning-cid\trunning\t<no value>\n",
                     stderr="",
                 )
         return subprocess.CompletedProcess(cmd, 0, stdout="fresh-cid\n", stderr="")
@@ -1010,6 +1016,42 @@ def test_find_reusable_container_prefers_running_over_stopped(monkeypatch):
     )
 
 
+def test_reuse_off_rejects_non_off_egress_container(monkeypatch):
+    """When egress is off, a container that still has hermes-egress=on
+    (e.g. from before ``hermes egress disable``) must be rejected and a
+    fresh container created.  The post-filter protects against silently
+    reusing a container with baked-in proxy env and CA mounts."""
+
+    monkeypatch.setattr(docker_env, "find_docker", lambda: "/usr/bin/docker")
+    monkeypatch.setattr(docker_env, "_get_active_profile_name", lambda: "default")
+
+    def _run(cmd, **kwargs):
+        if isinstance(cmd, list) and len(cmd) >= 2:
+            if cmd[1] == "version":
+                return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")
+            if cmd[1] == "ps":
+                # Return a container with hermes-egress=on.  With egress=off
+                # the three-field format includes the label; the post-filter
+                # must skip this entry.
+                return subprocess.CompletedProcess(
+                    cmd, 0,
+                    stdout="stale-cid\trunning\ton\n",
+                    stderr="",
+                )
+            if cmd[1] == "run":
+                return subprocess.CompletedProcess(cmd, 0, stdout="fresh-cid\n", stderr="")
+        return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(docker_env.subprocess, "run", _run)
+
+    env = _make_dummy_env(task_id="egress-off-reject")
+    # Should fall through to fresh container because the stale one has
+    # hermes-egress=on.
+    assert env._container_id == "fresh-cid", (
+        f"expected fresh container, got {env._container_id!r}"
+    )
+
+
 # ── Cleanup correctness (issue #20561) ────────────────────────────
 
 

tools/environments/docker.py

@@ -1575,11 +1575,21 @@ class DockerEnvironment(BaseEnvironment):
             ]
             if egress_label != "off":
                 filters.extend(["--filter", f"label={_EGRESS_LABEL_KEY}={egress_label}"])
+                fmt = "{{.ID}}\t{{.State}}"
+            else:
+                # When egress is off, we widen the probe to find any
+                # task+profile container (regardless of egress label), then
+                # post-filter in Python: reject containers whose
+                # hermes-egress label is present and not "off".  Without
+                # this, a container created with egress=on can be silently
+                # reused after the operator runs "hermes egress disable",
+                # preserving baked-in proxy env and CA mounts.
+                fmt = '{{.ID}}\t{{.State}}\t{{.Label "' + _EGRESS_LABEL_KEY + '"}}'
             result = subprocess.run(
                 [
                     self._docker_exe, "ps", "-a",
                     *filters,
-                    "--format", "{{.ID}}\t{{.State}}",
+                    "--format", fmt,
                 ],
                 capture_output=True,
                 text=True,
@@ -1607,10 +1617,24 @@ class DockerEnvironment(BaseEnvironment):
         running = None
         first = None
         for ln in lines:
-            parts = ln.split("\t", 1)
-            if len(parts) != 2:
-                continue
-            cid, state = parts[0], parts[1].lower()
+            if egress_label == "off":
+                # Format: ID\tState\tEgressLabel — parse all three fields
+                # and reject containers with a non-off egress label.
+                parts = ln.split("\t", 2)
+                if len(parts) < 3:
+                    continue
+                cid, state, egress_val = parts[0], parts[1].lower(), parts[2]
+                if egress_val not in ("", "<no value>", "off"):
+                    logger.debug(
+                        "skipping container %s for egress=off reuse: "
+                        "label %s=%r", cid, _EGRESS_LABEL_KEY, egress_val,
+                    )
+                    continue
+            else:
+                parts = ln.split("\t", 1)
+                if len(parts) != 2:
+                    continue
+                cid, state = parts[0], parts[1].lower()
             if first is None:
                 first = (cid, state)
             if state == "running" and running is None: