mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-10-20 05:15:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix(egress): address PR review findings — P0/P1/P2/P3 + CI greens

Browse source 
P0 — must-fix
- iron_proxy: emit default upstream_deny_cidrs (loopback, IMDS
  169.254.0.0/16, RFC1918) when caller passes None.  Honours the docs
  promise that cloud-metadata IPs are refused regardless of allowlist.
- iron_proxy: bind 127.0.0.1 (+ docker0 bridge IP on Linux) instead of
  INADDR_ANY (':9090').  LAN peers with a leaked sandbox token could
  otherwise spend the operator's API quota against any allowlisted
  upstream.
- ensure_ca_cert: write the CA private key via os.open(..., 0o600)
  instead of shutil.copy2+os.chmod — closes the TOCTOU window where
  the key existed under the default umask.
- discover_uncovered_providers + proxy.fail_on_uncovered_providers
  config: refuse to start (when strict) if env vars for non-bearer
  providers (Anthropic native x-api-key, AWS SigV4, Azure OpenAI,
  etc.) are present.  Surfaces a wizard warning in non-strict mode.

P1 — should-fix
- start_proxy: build a minimal subprocess env (PATH/HOME/locale +
  only the env names referenced by mappings) instead of os.environ
  .copy().  Strips proxy-recursion vars (HTTPS_PROXY etc.).  Stops
  the proxy's /proc/<pid>/environ from leaking every host secret
  to same-uid local processes.
- start_proxy: optional Bitwarden refresh path
  (refresh_secrets_from_bitwarden=True, bitwarden_config=...).
  When credential_source=bitwarden, cmd_start wires it in — that's
  what delivers the rotation guarantee the docs make.
- build_proxy_config: wire audit_log into the rendered yaml
  (log.audit_path).  Parameter was accepted but never used.
- ensure_audit_log: pre-create the audit log with 0o600 perms so
  iron-proxy inherits tight permissions instead of relying on umask.
- Rename 'hermes proxy ...' → 'hermes egress ...' in user-facing
  strings (docstring, RuntimeError messages, post-setup banner).
- start_proxy: open log file with 0o600 perms and close the parent
  fd immediately after Popen — fixes the per-restart fd leak.
- DockerEnvironment: detect collisions between docker_env and the
  egress-controlling env vars (HTTPS_PROXY, SSL_CERT_FILE, etc.).
  When enforce_on_docker=true, fail loud rather than silently
  inverting the isolation; when false, warn and let docker_env win.
- proxy_cli: merge_mappings preserves existing tokens on re-setup;
  --rotate-tokens flag re-mints all of them.  Stops re-running
  `hermes egress setup` from invalidating tokens baked into
  already-running sandboxes.
- proxy_cli: --from-bitwarden fail-loud on disabled BW config,
  missing access token, or empty vault.  Previously fell through to
  the env path while still writing credential_source: bitwarden.
- docker.py: narrow `except Exception` → `except ImportError`;
  iron_proxy._read_tunnel_port_from_config: same.  Bare excepts
  were masking real config-load bugs.
- start_proxy: write pidfile via os.open with O_NOFOLLOW + 0o600
  + st_uid check.  Refuses to follow a pre-existing symlink at the
  pidfile path.
- mint_proxy_token docstring: document the 128-bit suffix entropy
  explicitly (sha256 truncated to 32 hex chars).

P2 — follow-up
- start_proxy: poll-with-timeout (100ms cadence on _port_listening)
  instead of an unconditional 5s sleep.  Saves several seconds per
  Docker container create when enforce_on_docker=true.
- docker.py: apply enforce_on_docker semantics when CA file vanishes
  between status.configured check and CA mount.  Previously returned
  empty args silently.
- docker.py: refuse to mount when mappings.json is empty/corrupt
  (was indistinguishable from upstream outage from inside the
  sandbox).
- install_iron_proxy: tarfile.extract(..., filter='data') to silence
  the PEP 706 deprecation and opt into the 3.14+ default.
- _proxy_state_dir: chmod 0o700 unconditionally; add
  _proxy_state_dir_ro() so read-only callers don't create the dir.
- stop_proxy: re-verify pid before SIGKILL via /proc/<pid>/stat
  starttime AND _pid_alive.  Prevents SIGKILL'ing a recycled pid.
- _pid_alive: tightened cmdline check — basename match on argv[0]
  plus an in-process nonce env var ('iron-proxy' in cmdline matched
  'tail iron-proxy.log' and editors with the log open).
- docker.py: NODE_OPTIONS=--use-openssl-ca so Node.js routes through
  the OpenSSL CA store SSL_CERT_FILE controls, narrowing the
  Python/curl-replace vs Node-add asymmetry waefrebeorn flagged.

P3 — polish
- proxy_cli: dest='egress_command' (was 'proxy_command' which
  collided lexically with the inbound OAuth subparser).
- iron_proxy_version: cache by binary path — get_status is called
  per Docker container create, version is constant per binary.
- Drop unused `import sys` from iron_proxy.
- proxy_cli: `is not None` check on --tunnel-port (was treating 0
  as falsy and silently substituting the default).
- proxy_cli cmd_disable: use get_status().pid instead of reaching
  into ip._read_pid() (stale pidfile from a crashed run would have
  fired a spurious "still running" warning).
- Tests: replace hardcoded /tmp/ca.* paths with tmp_path-derived
  fixtures so tests are hermetic across hosts.

CI
- Windows footguns scanner: os.kill(pid, 0) is now gated behind
  platform.system() != 'Windows' with a windows-footgun: ok marker;
  signal.SIGKILL falls back to SIGTERM on Windows via
  getattr(signal, 'SIGKILL', signal.SIGTERM).
- docs MDX compilation: replace bare `<https://…>` URLs with
  `[text](url)` syntax (MDX-jsx parser rejects the angle-bracket
  form).

Tests
- 32 new tests covering default deny CIDRs, bind policy, audit log
  wiring, subprocess env minimization, CA TOCTOU 0o600, state dir
  0o700, empty-mappings refusal, CA-vanished refusal, docker_env
  collision detection, token preservation/rotate, uncovered provider
  detection, and the proxy_cli command handlers + argparse wiring.
- All 156 tests in test_iron_proxy + test_iron_proxy_cli +
  test_docker_environment + test_config pass locally.

Acknowledged but not addressed in this revision
- E2E test for HTTPS CONNECT + TLS-MITM path: existing E2E exercises
  plain HTTP; full MITM coverage needs separate CI infra (real iron-
  proxy binary + curl with custom CA).  Tracked as follow-up.
- Cosign-style supply-chain verification for the binary checksum:
  upstream iron-proxy doesn't sign releases yet.  Accepted pattern
  (same as Bitwarden integration); tracked as follow-up.
- CA rotation CLI (`hermes egress rotate-ca`): scope-cut to a
  follow-up.

Reviewers: @annguyenNous @waefrebeorn @GodsBoy @erhnysr
This commit is contained in:
teknium1 2026-05-23 01:59:01 -07:00 • committed by Teknium
parent 7a74492134
commit 128a6837b7
No known key found for this signature in database
8 changed files with 1990 additions and 131 deletions
Download patch file Download diff file Expand all files Collapse all files

768
agent/proxy_sources/iron_proxy.py
View file

File diff suppressed because it is too large Load diff

16
hermes_cli/config.py
View file

@ -1787,7 +1787,7 @@ DEFAULT_CONFIG = {
# real API credentials at the egress boundary. Compromising the sandbox
# leaks tokens that only work from behind the proxy.
#
# Configure with `hermes proxy setup`. Disabled by default — the rest of
# Configure with `hermes egress setup`. Disabled by default — the rest of
# Hermes works exactly as before with `enabled: false`.
"proxy": {
# Master switch. When false, iron-proxy is never started, no docker
@ -1811,6 +1811,20 @@ DEFAULT_CONFIG = {
# proxy is enabled but not running. False = fall back to direct
# outbound with real credentials in the sandbox (the legacy posture).
"enforce_on_docker": True,
# When true, `hermes egress start` refuses to start if any provider
# env var is set that the proxy cannot strip (Anthropic native
# `x-api-key`, AWS Bedrock SigV4, Azure OpenAI, etc.). These
# credentials would otherwise leak into the sandbox bypassing the
# proxy. Defaults to false because the false-positive cost
# (operator has the env set but doesn't actually use that provider)
# is higher than the security cost of a warning.
"fail_on_uncovered_providers": False,
# SSRF deny list applied to outbound traffic. Omit / leave empty
# to use the safe default: loopback, link-local (incl. cloud
# metadata IPs at 169.254.169.254), and RFC1918. Set to an
# explicit ``[]`` to opt out entirely (only sensible in hermetic
# tests that need to reach a loopback upstream).
"upstream_deny_cidrs": None,
# Extra allowed upstream hosts beyond the bundled defaults (which
# cover OpenRouter, OpenAI, Anthropic, Google, xAI, Mistral, Groq,
# Together, DeepSeek, Nous). Wildcards (`*.foo.com`) are supported.

4
hermes_cli/main.py
View file

@ -11125,7 +11125,9 @@ def main():
_proxy_cli.register_cli(egress_parser)
def _dispatch_egress(args): # noqa: ANN001
sub = getattr(args, "proxy_command", None)
# The egress subparser uses dest='egress_command' to stay disjoint
# from the inbound OAuth ``hermes proxy`` subparser (dest='proxy_command').
sub = getattr(args, "egress_command", None)
if sub is not None and hasattr(args, "func") and args.func is not _dispatch_egress:
return args.func(args)
egress_parser.print_help()

247
hermes_cli/proxy_cli.py
View file

@ -1,4 +1,4 @@
"""CLI handlers for ``hermes proxy ...``.
"""CLI handlers for ``hermes egress ...``.
Subcommands:
install — download the pinned iron-proxy binary
@ -8,6 +8,10 @@ Subcommands:
status — show binary version + config presence + listen state + mappings
disable — flip ``proxy.enabled`` to False (does not stop a running proxy)
config — print the generated proxy.yaml path (for debugging / external review)
The top-level command is ``hermes egress``. Note that the inbound OAuth
reverse-proxy command (``hermes proxy``) lives elsewhere in
``hermes_cli/main.py`` — different direction, different purpose.
"""
from __future__ import annotations
@ -31,13 +35,18 @@ from hermes_cli.config import load_config, save_config
def register_cli(parent_parser: argparse.ArgumentParser) -> None:
"""Attach the proxy subcommand tree to a parent parser.
"""Attach the egress subcommand tree to a parent parser.
Called from ``hermes_cli.main`` as part of building the top-level
``hermes proxy`` parser.
``hermes egress`` parser.
"""
sub = parent_parser.add_subparsers(dest="proxy_command")
# dest='egress_command' — keeps this subparser tree disjoint from the
# inbound OAuth ``hermes proxy`` subparser (which uses dest='proxy_command').
# No runtime collision today since they live in separate parser trees,
# but a future grep-and-refactor on ``proxy_command`` would otherwise
# hit both handlers.
sub = parent_parser.add_subparsers(dest="egress_command")
install = sub.add_parser(
"install",
@ -60,7 +69,14 @@ def register_cli(parent_parser: argparse.ArgumentParser) -> None:
setup.add_argument(
"--from-bitwarden", action="store_true",
help="Treat secrets as managed by Bitwarden — discover provider keys "
"from secrets.bitwarden config instead of the current env",
"from secrets.bitwarden config instead of the current env. Fails "
"loudly if BW is unreachable rather than silently falling back.",
)
setup.add_argument(
"--rotate-tokens", action="store_true",
help="Mint fresh proxy tokens for every provider (default is to "
"preserve tokens for providers that already had one — avoids "
"401-ing already-running sandboxes on re-setup).",
)
setup.set_defaults(func=cmd_setup)
@ -73,7 +89,8 @@ def register_cli(parent_parser: argparse.ArgumentParser) -> None:
status = sub.add_parser("status", help="Show proxy state and mappings")
status.add_argument(
"--show-tokens", action="store_true",
help="Print the proxy tokens (default: redacted prefix only)",
help="Print the proxy tokens (default: redacted prefix only). "
"Beware: tokens may persist in your shell history.",
)
status.set_defaults(func=cmd_status)
@ -93,7 +110,7 @@ def cmd_install(args: argparse.Namespace) -> int:
console = Console()
try:
binary = ip.install_iron_proxy(force=bool(args.force))
except Exception as exc: # noqa: BLE001
except Exception as exc: # noqa: BLE001 — top-level user-facing error funnel
console.print(f"[red]✗ install failed:[/red] {exc}")
console.print(
" Manual install: https://github.com/ironsh/iron-proxy/releases"
@ -148,10 +165,12 @@ def cmd_setup(args: argparse.Namespace) -> int:
bw_cfg = (cfg.get("secrets") or {}).get("bitwarden") or {}
if not bw_cfg.get("enabled"):
console.print(
" [yellow]--from-bitwarden requested but secrets.bitwarden.enabled is false.[/yellow]"
" [red]✗ --from-bitwarden requested but "
"secrets.bitwarden.enabled is false.[/red]"
)
console.print(
" Run `hermes secrets bitwarden setup` first, or omit --from-bitwarden."
" Run `hermes secrets bitwarden setup` first, or omit "
"--from-bitwarden."
)
return 1
try:
@ -159,29 +178,56 @@ def cmd_setup(args: argparse.Namespace) -> int:
access_token = os.environ.get(
bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN"), ""
).strip()
if access_token:
secrets, _ = bw.fetch_bitwarden_secrets(
access_token=access_token,
project_id=bw_cfg.get("project_id", ""),
cache_ttl_seconds=0,
use_cache=False,
)
available_env_names = list(secrets.keys())
if not access_token:
console.print(
f" Pulled {len(available_env_names)} env names from Bitwarden."
f" [red]✗ --from-bitwarden requested but "
f"{bw_cfg.get('access_token_env', 'BWS_ACCESS_TOKEN')} "
"is not set in the environment.[/red]"
)
except Exception as exc: # noqa: BLE001
return 1
secrets, _ = bw.fetch_bitwarden_secrets(
access_token=access_token,
project_id=bw_cfg.get("project_id", ""),
cache_ttl_seconds=0,
use_cache=False,
)
available_env_names = list(secrets.keys())
if not available_env_names:
console.print(
" [red]✗ Bitwarden returned an empty secrets list.[/red]\n"
" Check the project_id in secrets.bitwarden and the "
"BWS access-token's project scope."
)
return 1
console.print(
f" [yellow]Could not enumerate Bitwarden secrets: {exc}[/yellow]"
f" Pulled {len(available_env_names)} env names from Bitwarden."
)
except Exception as exc: # noqa: BLE001 — explicit user-facing error
console.print(
f" [red]✗ Could not enumerate Bitwarden secrets: {exc}[/red]"
)
console.print(
" Falling back to current process env for discovery."
" Either fix the Bitwarden config and retry, or rerun setup "
"without --from-bitwarden (the proxy will read secrets from "
"the host process env at start time)."
)
return 1
mappings = ip.discover_provider_mappings(
discovered = ip.discover_provider_mappings(
available_env_names=available_env_names or None,
)
# Preserve tokens for providers we already had unless the operator
# explicitly requested rotation. This prevents re-running `hermes
# egress setup` from invalidating tokens baked into already-running
# sandboxes.
existing = ip.load_mappings()
mappings = ip.merge_mappings(
existing=existing,
discovered=discovered,
rotate=bool(getattr(args, "rotate_tokens", False)),
)
if not mappings:
console.print(
" [yellow]No known provider API keys found in env/Bitwarden.[/yellow]"
@ -193,6 +239,36 @@ def cmd_setup(args: argparse.Namespace) -> int:
console.print(f" - {env_name}")
return 1
# Warn the operator about providers we recognize but can't proxy
# (Anthropic native, AWS Bedrock, Azure OpenAI, etc). These still
# work — they just bypass the egress isolation.
uncovered = ip.discover_uncovered_providers(
available_env_names=available_env_names or None,
)
if uncovered:
console.print()
console.print(
" [yellow]⚠[/yellow] Detected provider env vars that the "
"proxy does not yet cover:"
)
for name in uncovered:
console.print(f" - {name}")
console.print(
" [dim]These providers use non-bearer auth (x-api-key, "
"SigV4, etc.) and will hold real credentials inside the "
"sandbox. Egress isolation is INCOMPLETE for these.[/dim]"
)
# If we're rotating existing mappings, also surface that — the
# operator might be running containers that hold the old tokens.
if existing and getattr(args, "rotate_tokens", False):
console.print()
console.print(
" [yellow]⚠[/yellow] --rotate-tokens used: any running "
"Hermes sandbox holds stale tokens and will 401 against "
"upstreams until restarted."
)
table = Table(show_header=True, header_style="bold")
table.add_column("Provider env", style="cyan")
table.add_column("Upstream hosts", style="dim")
@ -211,11 +287,19 @@ def cmd_setup(args: argparse.Namespace) -> int:
cfg = load_config()
proxy_cfg = cfg.setdefault("proxy", {})
tunnel_port = (
args.tunnel_port
if args.tunnel_port
else int(proxy_cfg.get("tunnel_port", ip._DEFAULT_TUNNEL_PORT))
)
# ``args.tunnel_port`` is None when the flag was not given; ``0`` is
# invalid for a TCP listener so we treat it as an explicit refusal
# and surface a clear error rather than silently substituting the
# default.
if args.tunnel_port is not None:
if args.tunnel_port == 0:
console.print(
" [red]✗ --tunnel-port=0 is not a valid TCP port.[/red]"
)
return 1
tunnel_port = int(args.tunnel_port)
else:
tunnel_port = int(proxy_cfg.get("tunnel_port", ip._DEFAULT_TUNNEL_PORT))
proxy_cfg["tunnel_port"] = tunnel_port
extra_hosts = list(proxy_cfg.get("extra_allowed_hosts") or [])
@ -223,24 +307,38 @@ def cmd_setup(args: argparse.Namespace) -> int:
h for h in extra_hosts if h not in ip._DEFAULT_ALLOWED_HOSTS
]
audit_log_path = ip._proxy_state_dir() / "audit.log"
# Pre-create the audit log with 0o600 so iron-proxy inherits private
# perms instead of letting the daemon create it under the default
# umask (potentially world-readable).
ip.ensure_audit_log(audit_log_path)
# Allow operator override of the deny list via
# ``proxy.upstream_deny_cidrs`` — but the default (None) gives a safe
# default-deny list (loopback, IMDS, RFC1918) that matches the docs
# promise.
deny_cidrs = proxy_cfg.get("upstream_deny_cidrs")
iron_cfg = ip.build_proxy_config(
mappings=mappings,
ca_cert=ca_crt,
ca_key=ca_key,
tunnel_port=tunnel_port,
audit_log=ip._proxy_state_dir() / "audit.log",
audit_log=audit_log_path,
allowed_hosts=allowed,
upstream_deny_cidrs=deny_cidrs,
)
cfg_path = ip.write_proxy_config(iron_cfg)
mappings_path = ip.write_mappings(mappings)
console.print(f" [green]✓[/green] config: {cfg_path}")
console.print(f" [green]✓[/green] mappings: {mappings_path}")
console.print(f" [green]✓[/green] audit log: {audit_log_path}")
# ------------------------------------------------------------------ enable
proxy_cfg["enabled"] = True
proxy_cfg.setdefault("auto_install", True)
proxy_cfg.setdefault("enforce_on_docker", True)
proxy_cfg.setdefault("credential_source", "bitwarden" if args.from_bitwarden else "env")
proxy_cfg["credential_source"] = "bitwarden" if args.from_bitwarden else "env"
proxy_cfg.setdefault("fail_on_uncovered_providers", False)
save_config(cfg)
console.print()
@ -249,10 +347,10 @@ def cmd_setup(args: argparse.Namespace) -> int:
"Sandboxes will route outbound traffic through it."
)
console.print(
" Start: [cyan]hermes proxy start[/cyan]\n"
" Status: [cyan]hermes proxy status[/cyan]\n"
" Stop: [cyan]hermes proxy stop[/cyan]\n"
" Disable: [cyan]hermes proxy disable[/cyan]"
" Start: [cyan]hermes egress start[/cyan]\n"
" Status: [cyan]hermes egress status[/cyan]\n"
" Stop: [cyan]hermes egress stop[/cyan]\n"
" Disable: [cyan]hermes egress disable[/cyan]"
)
return 0
@ -263,16 +361,59 @@ def cmd_start(args: argparse.Namespace) -> int:
proxy_cfg = cfg.get("proxy") or {}
if not proxy_cfg.get("enabled"):
console.print(
"[yellow]proxy.enabled is false — run `hermes proxy setup` first.[/yellow]"
"[yellow]proxy.enabled is false — run `hermes egress setup` "
"first.[/yellow]"
)
return 1
# If the operator opted in to Bitwarden-rotation semantics, refresh
# upstream secrets from BSM at startup. This is what delivers the
# rotation guarantee that distinguishes ``credential_source:
# bitwarden`` from ``credential_source: env``. Without it, rotating
# a key in the Bitwarden web app doesn't reach the proxy.
credential_source = proxy_cfg.get("credential_source", "env")
bw_cfg = (cfg.get("secrets") or {}).get("bitwarden")
refresh_bw = (
credential_source == "bitwarden"
and bw_cfg is not None
and bool(bw_cfg.get("enabled"))
)
# fail_on_uncovered_providers: when true, refuse to start if any of
# the recognized non-bearer providers (Anthropic native, AWS Bedrock,
# Azure OpenAI, etc.) have env vars set in the host process — those
# would otherwise leak real credentials into the sandbox while
# bypassing the proxy.
if bool(proxy_cfg.get("fail_on_uncovered_providers", False)):
uncovered = ip.discover_uncovered_providers()
if uncovered:
console.print(
"[red]✗ Refusing to start: provider env vars present "
"that bypass the proxy:[/red]"
)
for name in uncovered:
console.print(f" - {name}")
console.print(
" Set `proxy.fail_on_uncovered_providers: false` in "
"config.yaml to start anyway (sandbox will hold real "
"credentials for those providers)."
)
return 1
try:
status = ip.start_proxy()
except Exception as exc: # noqa: BLE001
status = ip.start_proxy(
refresh_secrets_from_bitwarden=refresh_bw,
bitwarden_config=bw_cfg,
)
except Exception as exc: # noqa: BLE001 — top-level user-facing funnel
console.print(f"[red]✗ failed to start iron-proxy:[/red] {exc}")
return 1
if status.pid:
listening = "[green]listening[/green]" if status.listening else "[yellow]not yet listening[/yellow]"
listening = (
"[green]listening[/green]"
if status.listening
else "[yellow]not yet listening[/yellow]"
)
console.print(
f"[green]✓[/green] iron-proxy running pid={status.pid} "
f"port={status.tunnel_port} {listening}"
@ -325,6 +466,25 @@ def cmd_status(args: argparse.Namespace) -> int:
tok = m.proxy_token if args.show_tokens else _redact_token(m.proxy_token)
m_table.add_row(m.real_env_name, ", ".join(m.upstream_hosts), tok)
console.print(m_table)
if args.show_tokens:
console.print(
"[yellow]⚠[/yellow] proxy tokens just printed in full — "
"they may persist in your shell history. Consider clearing "
"it after this command."
)
# Surface uncovered providers so the operator knows the isolation
# boundary is incomplete for those upstreams.
uncovered = ip.discover_uncovered_providers()
if uncovered:
console.print()
console.print(
"[yellow]Uncovered providers[/yellow] "
"(real credentials still visible inside the sandbox):"
)
for name in uncovered:
console.print(f" - {name}")
return 0
@ -338,10 +498,15 @@ def cmd_disable(args: argparse.Namespace) -> int:
proxy_cfg["enabled"] = False
save_config(cfg)
console.print("[green]✓[/green] proxy.enabled set to false")
if ip._read_pid() is not None:
# Use the public get_status() pid (which already incorporates the
# _pid_alive check) instead of reaching into ip._read_pid(). That
# private accessor only proves the pidfile is non-empty — a stale
# pidfile from a crashed previous run would fire the warning
# spuriously.
if ip.get_status().pid is not None:
console.print(
" iron-proxy is still running — stop it with "
"[cyan]hermes proxy stop[/cyan] if you want it down too."
"[cyan]hermes egress stop[/cyan] if you want it down too."
)
return 0
@ -350,7 +515,9 @@ def cmd_config(args: argparse.Namespace) -> int:
console = Console()
status = ip.get_status()
if status.config_path is None:
console.print("[yellow](no config generated — run `hermes proxy setup`)[/yellow]")
console.print(
"[yellow](no config generated — run `hermes egress setup`)[/yellow]"
)
return 1
console.print(str(status.config_path))
return 0

566
tests/test_iron_proxy.py
View file

@ -17,7 +17,6 @@ import io
import os
import tarfile
from pathlib import Path
from typing import Dict
from unittest.mock import MagicMock, patch
import pytest
@ -99,12 +98,14 @@ def _sample_mapping(env_name: str = "OPENROUTER_API_KEY") -> ip.TokenMapping:
)
def test_build_proxy_config_shape():
def test_build_proxy_config_shape(tmp_path):
m = _sample_mapping()
ca_crt = tmp_path / "ca.crt"
ca_key = tmp_path / "ca.key"
cfg = ip.build_proxy_config(
mappings=[m],
ca_cert=Path("/tmp/ca.crt"),
ca_key=Path("/tmp/ca.key"),
ca_cert=ca_crt,
ca_key=ca_key,
)
# Top-level sections — note `dns` is required by iron-proxy even when
# we only use the CONNECT tunnel.
@ -127,23 +128,154 @@ def test_build_proxy_config_shape():
rule_hosts = {r["host"] for r in rule["rules"]}
assert rule_hosts == set(m.upstream_hosts)
# TLS section names the CA paths
assert cfg["tls"]["ca_cert"] == "/tmp/ca.crt"
assert cfg["tls"]["ca_cert"] == str(ca_crt)
def test_build_proxy_config_custom_allowed_hosts():
def test_build_proxy_config_custom_allowed_hosts(tmp_path):
m = _sample_mapping("OPENAI_API_KEY")
cfg = ip.build_proxy_config(
mappings=[m],
ca_cert=Path("/tmp/ca.crt"),
ca_key=Path("/tmp/ca.key"),
allowed_hosts=["only.example.com"],
ca_cert=tmp_path / "ca.crt",
ca_key=tmp_path / "ca.key",
allowed_hosts=["custom-host.test"],
)
domains = cfg["transforms"][0]["config"]["domains"]
# Custom allowed_hosts wins as the base; mapping's hosts get appended.
assert "only.example.com" in domains
assert "custom-host.test" in domains
assert "openrouter.ai" in domains # comes from the mapping
# ---------------------------------------------------------------------------
# Default SSRF deny list (regression: docs promise cloud metadata is denied)
# ---------------------------------------------------------------------------
def test_default_deny_cidrs_present_when_unspecified(tmp_path):
"""build_proxy_config must emit the default deny list when the caller
passes nothing. The IMDS subnet (169.254.0.0/16) MUST be in the result
or the docs claim that ``upstream_deny_cidrs`` refuses cloud metadata
is a lie."""
cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=tmp_path / "ca.crt",
ca_key=tmp_path / "ca.key",
)
deny = cfg["proxy"]["upstream_deny_cidrs"]
assert "169.254.0.0/16" in deny # IMDS
assert "127.0.0.0/8" in deny # loopback v4
assert "::1/128" in deny # loopback v6
assert "10.0.0.0/8" in deny # RFC1918
assert "172.16.0.0/12" in deny # RFC1918
assert "192.168.0.0/16" in deny # RFC1918
def test_explicit_empty_deny_cidrs_disables_default(tmp_path):
"""Explicit ``[]`` opts out of the default deny list — needed by
hermetic tests that want to talk to a loopback upstream."""
cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=tmp_path / "ca.crt",
ca_key=tmp_path / "ca.key",
upstream_deny_cidrs=[],
)
assert cfg["proxy"]["upstream_deny_cidrs"] == []
def test_wizard_rendered_yaml_contains_deny_list(hermes_home, tmp_path):
"""End-to-end: cmd_setup writes proxy.yaml; the rendered file must
contain the deny list because the wizard now passes the operator's
config-level setting (None → default) through to build_proxy_config."""
# Simulate the wizard's call shape (matches proxy_cli.cmd_setup).
state = ip._proxy_state_dir()
(state / "ca.crt").write_text("fake-ca")
(state / "ca.key").write_text("fake-key")
proxy_yaml = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=state / "ca.crt",
ca_key=state / "ca.key",
# The wizard passes ``upstream_deny_cidrs`` from the config; when
# the operator hasn't set anything, that's None and we get the
# safe default below.
upstream_deny_cidrs=None,
)
out = ip.write_proxy_config(proxy_yaml)
text = out.read_text(encoding="utf-8")
assert "169.254.0.0/16" in text
# ---------------------------------------------------------------------------
# Bind policy (regression: must not bind 0.0.0.0)
# ---------------------------------------------------------------------------
def test_default_bind_is_loopback_not_zero_zero(tmp_path):
"""``http_listen`` must NOT be ``0.0.0.0:PORT`` or ``:PORT`` (latter is
INADDR_ANY). Loopback only by default; the docker bridge bind is
optional and added in addition, never instead."""
cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=tmp_path / "ca.crt",
ca_key=tmp_path / "ca.key",
tunnel_port=12345,
http_listen=["127.0.0.1:12345"], # explicit so test is deterministic
)
primary = cfg["proxy"]["http_listen"]
listens = cfg["proxy"]["http_listens"]
assert primary == "127.0.0.1:12345"
assert listens == ["127.0.0.1:12345"]
# Sentinel: confirm we didn't accidentally serialize a bare-port form
# like ":12345" anywhere in the listen list (that's INADDR_ANY).
for entry in listens:
assert not entry.startswith(":")
assert "0.0.0.0" not in entry
def test_default_bind_includes_docker_bridge_on_linux(tmp_path, monkeypatch):
"""When http_listen isn't passed AND we're on Linux AND a docker
bridge IP is detected, we should bind that bridge IP in addition to
loopback so containers reach the proxy via host-gateway."""
monkeypatch.setattr(ip.platform, "system", lambda: "Linux")
monkeypatch.setattr(ip, "_detect_docker_bridge_ip", lambda: "172.17.0.1")
cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=tmp_path / "ca.crt",
ca_key=tmp_path / "ca.key",
tunnel_port=9090,
)
assert "127.0.0.1:9090" in cfg["proxy"]["http_listens"]
assert "172.17.0.1:9090" in cfg["proxy"]["http_listens"]
# ---------------------------------------------------------------------------
# audit_log wiring (regression: parameter was accepted but never used)
# ---------------------------------------------------------------------------
def test_audit_log_path_lands_in_yaml(tmp_path):
cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=tmp_path / "ca.crt",
ca_key=tmp_path / "ca.key",
audit_log=tmp_path / "audit.log",
)
assert cfg["log"]["audit_path"] == str(tmp_path / "audit.log")
def test_audit_log_omitted_when_caller_passes_none(tmp_path):
cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=tmp_path / "ca.crt",
ca_key=tmp_path / "ca.key",
audit_log=None,
)
assert "audit_path" not in cfg["log"]
def test_write_and_load_mappings_roundtrip(hermes_home):
ms = [_sample_mapping("OPENROUTER_API_KEY"), _sample_mapping("OPENAI_API_KEY")]
path = ip.write_mappings(ms)
@ -165,17 +297,111 @@ def test_load_mappings_handles_corrupt_json(hermes_home):
assert ip.load_mappings() == []
def test_write_proxy_config_serializes_yaml(hermes_home):
def test_write_proxy_config_serializes_yaml(hermes_home, tmp_path):
ca_crt = tmp_path / "ca.crt"
ca_key = tmp_path / "ca.key"
cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=Path("/tmp/ca.crt"),
ca_key=Path("/tmp/ca.key"),
ca_cert=ca_crt,
ca_key=ca_key,
)
out = ip.write_proxy_config(cfg)
assert out.exists()
text = out.read_text(encoding="utf-8")
assert "tunnel_listen" in text
assert "ca_cert: /tmp/ca.crt" in text
assert f"ca_cert: {ca_crt}" in text
# ---------------------------------------------------------------------------
# Token-preservation on re-setup (regression: clobbered live sandboxes)
# ---------------------------------------------------------------------------
def test_merge_mappings_preserves_existing_tokens():
"""Re-running setup must not invalidate tokens baked into already-
running sandboxes. ``merge_mappings`` keeps the prior token for any
provider that's in both lists."""
existing = [
ip.TokenMapping(
proxy_token="hermes-proxy-original-12345",
real_env_name="OPENROUTER_API_KEY",
upstream_hosts=("openrouter.ai",),
),
]
discovered = ip.discover_provider_mappings(
available_env_names=["OPENROUTER_API_KEY", "OPENAI_API_KEY"]
)
merged = ip.merge_mappings(existing=existing, discovered=discovered)
by_name = {m.real_env_name: m for m in merged}
# Original token preserved.
assert by_name["OPENROUTER_API_KEY"].proxy_token == "hermes-proxy-original-12345"
# New provider got a fresh token.
assert by_name["OPENAI_API_KEY"].proxy_token != "hermes-proxy-original-12345"
# Both providers in the result.
assert set(by_name) == {"OPENROUTER_API_KEY", "OPENAI_API_KEY"}
def test_merge_mappings_drops_providers_removed_from_env():
"""When a provider is in `existing` but not in `discovered`, it must
be dropped from the result — the operator removed the env var."""
existing = [
ip.TokenMapping(
proxy_token="stale", real_env_name="OPENROUTER_API_KEY",
upstream_hosts=("openrouter.ai",),
),
]
discovered = ip.discover_provider_mappings(
available_env_names=["OPENAI_API_KEY"]
)
merged = ip.merge_mappings(existing=existing, discovered=discovered)
names = {m.real_env_name for m in merged}
assert names == {"OPENAI_API_KEY"}
def test_merge_mappings_rotate_mints_fresh_tokens():
"""``rotate=True`` rolls every token regardless of overlap. The
--rotate-tokens flag uses this."""
existing = [
ip.TokenMapping(
proxy_token="hermes-proxy-original-12345",
real_env_name="OPENROUTER_API_KEY",
upstream_hosts=("openrouter.ai",),
),
]
discovered = ip.discover_provider_mappings(
available_env_names=["OPENROUTER_API_KEY"]
)
merged = ip.merge_mappings(existing=existing, discovered=discovered, rotate=True)
assert merged[0].proxy_token != "hermes-proxy-original-12345"
# ---------------------------------------------------------------------------
# Uncovered provider detection (regression: non-bearer providers bypass)
# ---------------------------------------------------------------------------
def test_uncovered_providers_detects_anthropic_aws(hermes_home, monkeypatch):
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAEXAMPLE")
uncovered = ip.discover_uncovered_providers()
assert "ANTHROPIC_API_KEY" in uncovered
assert "AWS_ACCESS_KEY_ID" in uncovered
def test_uncovered_providers_explicit_names_empty():
assert ip.discover_uncovered_providers(available_env_names=[]) == []
def test_uncovered_providers_skips_bearer_providers(hermes_home, monkeypatch):
"""OPENROUTER_API_KEY etc. are bearer providers — they should NOT
appear in the uncovered list."""
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-test")
uncovered = ip.discover_uncovered_providers()
assert "OPENROUTER_API_KEY" not in uncovered
# ---------------------------------------------------------------------------
@ -531,3 +757,315 @@ def test_platform_asset_name_rejects_windows(monkeypatch):
monkeypatch.setattr("platform.machine", lambda: "AMD64")
with pytest.raises(RuntimeError, match="does not ship native Windows"):
ip._platform_asset_name()
# ---------------------------------------------------------------------------
# Subprocess env minimization (regression: host secrets leaked to proxy)
# ---------------------------------------------------------------------------
def test_subprocess_env_strips_unrelated_secrets(hermes_home, monkeypatch):
"""``_build_proxy_subprocess_env`` must NOT carry every host secret
over to the proxy. /proc/<pid>/environ on the proxy would otherwise
expose all of them to same-uid local processes."""
# Unrelated env vars that should NOT propagate.
monkeypatch.setenv("MY_PRIVATE_TOKEN", "should-not-leak")
monkeypatch.setenv("DATABASE_URL", "postgres://very-private")
monkeypatch.setenv("SLACK_BOT_TOKEN", "xoxb-very-secret")
# Provider keys that ARE in load_mappings should propagate.
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-real")
ip.write_mappings([_sample_mapping("OPENROUTER_API_KEY")])
env = ip._build_proxy_subprocess_env()
assert "MY_PRIVATE_TOKEN" not in env
assert "DATABASE_URL" not in env
assert "SLACK_BOT_TOKEN" not in env
assert env.get("OPENROUTER_API_KEY") == "sk-or-real"
def test_subprocess_env_strips_proxy_recursion_vars(hermes_home, monkeypatch):
"""HTTPS_PROXY etc. in the parent env would otherwise recurse iron-proxy
through itself (or send its traffic through a corporate proxy)."""
monkeypatch.setenv("HTTPS_PROXY", "http://corporate:3128")
monkeypatch.setenv("HTTP_PROXY", "http://corporate:3128")
monkeypatch.setenv("ALL_PROXY", "socks5://corporate:1080")
env = ip._build_proxy_subprocess_env()
assert "HTTPS_PROXY" not in env
assert "https_proxy" not in env
assert "HTTP_PROXY" not in env
assert "ALL_PROXY" not in env
def test_subprocess_env_keeps_infrastructure_vars(hermes_home, monkeypatch):
"""PATH / HOME / locale must propagate or the child can't even find
its libs."""
monkeypatch.setenv("PATH", "/usr/local/bin:/usr/bin")
monkeypatch.setenv("HOME", "/home/test")
monkeypatch.setenv("LANG", "C.UTF-8")
env = ip._build_proxy_subprocess_env()
assert env.get("PATH") == "/usr/local/bin:/usr/bin"
assert env.get("HOME") == "/home/test"
assert env.get("LANG") == "C.UTF-8"
# ---------------------------------------------------------------------------
# CA generation TOCTOU (regression: 0o600 only set AFTER copy)
# ---------------------------------------------------------------------------
def test_ca_key_created_with_0o600(hermes_home, monkeypatch):
"""The CA private key must NEVER exist on disk with default umask
permissions, even transiently. Fix: open with explicit mode=0o600
so the very first byte is written under tight perms."""
# ensure_ca_cert shells out to openssl; mock the subprocess.run calls
# so we don't need openssl on the test host AND don't depend on its
# output format.
def fake_run(args, **kwargs):
# First call: genrsa → -out is at args[-2]
if args[1] == "genrsa":
out = args[-2]
Path(out).write_bytes(b"-----BEGIN RSA PRIVATE KEY-----\nfake\n-----END RSA PRIVATE KEY-----\n")
elif args[1] == "req":
# Find -out path
i = args.index("-out")
Path(args[i + 1]).write_bytes(b"-----BEGIN CERTIFICATE-----\nfake\n-----END CERTIFICATE-----\n")
result = MagicMock()
result.returncode = 0
return result
monkeypatch.setattr(ip.shutil, "which", lambda name: "/usr/bin/openssl" if name == "openssl" else None)
monkeypatch.setattr(ip.subprocess, "run", fake_run)
ca_crt, ca_key = ip.ensure_ca_cert()
assert ca_key.exists()
mode = ca_key.stat().st_mode & 0o777
assert mode == 0o600, f"CA key has perms {oct(mode)}, expected 0o600"
# ---------------------------------------------------------------------------
# Audit log permissions (regression: depended on umask)
# ---------------------------------------------------------------------------
def test_ensure_audit_log_creates_with_0o600(hermes_home, tmp_path):
audit = tmp_path / "audit.log"
ip.ensure_audit_log(audit)
assert audit.exists()
mode = audit.stat().st_mode & 0o777
assert mode == 0o600
def test_ensure_audit_log_tightens_existing_perms(hermes_home, tmp_path):
audit = tmp_path / "audit.log"
audit.write_text("preexisting content\n")
os.chmod(audit, 0o644)
ip.ensure_audit_log(audit)
mode = audit.stat().st_mode & 0o777
assert mode == 0o600
# ---------------------------------------------------------------------------
# State dir hardening (regression: world-traversable on multi-user hosts)
# ---------------------------------------------------------------------------
def test_proxy_state_dir_is_0o700(hermes_home):
state = ip._proxy_state_dir()
mode = state.stat().st_mode & 0o777
assert mode == 0o700
def test_proxy_state_dir_ro_does_not_create(hermes_home):
"""_proxy_state_dir_ro is for read-only callers — it must NOT
materialize the dir. Pure-status code paths shouldn't have the
side-effect of creating ~/.hermes/proxy/."""
# Sanity: rw path creates it.
rw = ip._proxy_state_dir()
assert rw.exists()
# Remove it and confirm the ro path doesn't recreate.
import shutil as _shutil
_shutil.rmtree(str(rw))
assert not rw.exists()
ro = ip._proxy_state_dir_ro()
assert not ro.exists()
# The path string is the same as the rw one.
assert ro == rw
# ---------------------------------------------------------------------------
# Mappings clobber refused when corrupt (regression: silent 403s)
# ---------------------------------------------------------------------------
def test_docker_egress_args_raises_on_empty_mappings(hermes_home, monkeypatch):
"""If mappings.json is missing / corrupt / empty AND
enforce_on_docker is true, refuse to start the sandbox rather than
silently mounting an unusable proxy config."""
from tools.environments.docker import _egress_proxy_args_for_docker
from hermes_cli.config import load_config, save_config
state = ip._proxy_state_dir()
(state / "ca.crt").write_text("fake-ca")
(state / "ca.key").write_text("fake-key")
proxy_cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=state / "ca.crt", ca_key=state / "ca.key", tunnel_port=9090,
)
ip.write_proxy_config(proxy_cfg)
# Note: we deliberately do NOT write mappings.json — that's the
# bug-class this test guards against.
cfg = load_config()
cfg.setdefault("proxy", {})["enabled"] = True
cfg["proxy"]["enforce_on_docker"] = True
save_config(cfg)
(state / "iron-proxy.pid").write_text("99999")
monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
monkeypatch.setattr(ip, "_port_listening", lambda h, p: True)
with pytest.raises(RuntimeError, match="mappings.json is empty or"):
_egress_proxy_args_for_docker()
# ---------------------------------------------------------------------------
# CA missing → enforce_on_docker semantics (regression: silent fail-open)
# ---------------------------------------------------------------------------
def test_docker_egress_args_raises_when_ca_vanishes(hermes_home, monkeypatch):
"""status.configured was True at check time but the CA file
disappeared between then and now (e.g. operator manually deleted
~/.hermes/proxy/ca.crt). enforce_on_docker=True must refuse."""
from tools.environments.docker import _egress_proxy_args_for_docker
from hermes_cli.config import load_config, save_config
state = ip._proxy_state_dir()
ca = state / "ca.crt"
ca.write_text("fake-ca")
(state / "ca.key").write_text("fake-key")
proxy_cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=ca, ca_key=state / "ca.key", tunnel_port=9090,
)
ip.write_proxy_config(proxy_cfg)
ip.write_mappings([_sample_mapping()])
cfg = load_config()
cfg.setdefault("proxy", {})["enabled"] = True
cfg["proxy"]["enforce_on_docker"] = True
save_config(cfg)
(state / "iron-proxy.pid").write_text("99999")
monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
monkeypatch.setattr(ip, "_port_listening", lambda h, p: True)
# Build a fake status: configured=True (because both path fields are
# set), but ca_cert_path.exists() is False — simulating the race
# where the CA file vanished between get_status() and the
# exists() recheck inside _egress_proxy_args_for_docker.
fake_status = ip.ProxyStatus(
binary_path=state / "fake-bin", # truthy
config_path=state / "proxy.yaml",
ca_cert_path=state / "missing-ca.crt", # points at nonexistent path
pid=99999,
listening=True,
tunnel_port=9090,
)
# ProxyStatus.configured returns True iff config_path AND ca_cert_path
# both exist. We need configured=True but the second exists() check
# in docker.py to return False — force that by writing a placeholder
# config_path that exists and pointing ca_cert_path at a missing file.
(state / "proxy.yaml").write_text("# fake")
# ProxyStatus.configured: config_path.exists() and ca_cert_path.exists().
# Make ca_cert_path .exists() True for the configured check but the
# explicit .exists() recheck path in docker.py reads the same Path,
# which is missing — so we wrap.
class _CAStub:
"""Path-like that toggles .exists() so configured=True but the
defensive recheck in docker.py returns False."""
_calls = 0
def __init__(self, real: Path):
self._real = real
def __str__(self):
return str(self._real)
@property
def parent(self):
return self._real.parent
def exists(self):
type(self)._calls += 1
# First call: configured property check → say yes.
# Second call: docker.py defensive recheck → say no.
return type(self)._calls == 1
fake_status.ca_cert_path = _CAStub(state / "missing-ca.crt") # type: ignore[assignment]
monkeypatch.setattr(ip, "get_status", lambda: fake_status)
with pytest.raises(RuntimeError, match="CA cert vanished"):
_egress_proxy_args_for_docker()
# ---------------------------------------------------------------------------
# Docker env collision detection (regression: docker_env silently bypassed proxy)
# ---------------------------------------------------------------------------
def test_docker_env_collision_with_proxy_raises_when_enforce(hermes_home, monkeypatch):
"""Setting docker_env: {HTTPS_PROXY: ''} in config.yaml with
enforce_on_docker=true must fail-loud rather than silently inverting
the egress isolation."""
from tools.environments.docker import DockerEnvironment
from hermes_cli.config import load_config, save_config
# Set up a fully-running proxy.
state = ip._proxy_state_dir()
ca = state / "ca.crt"
ca.write_text("fake-ca")
(state / "ca.key").write_text("fake-key")
proxy_cfg = ip.build_proxy_config(
mappings=[_sample_mapping()],
ca_cert=ca, ca_key=state / "ca.key", tunnel_port=9090,
)
ip.write_proxy_config(proxy_cfg)
ip.write_mappings([_sample_mapping()])
cfg = load_config()
cfg.setdefault("proxy", {})["enabled"] = True
cfg["proxy"]["enforce_on_docker"] = True
save_config(cfg)
(state / "iron-proxy.pid").write_text("99999")
monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
monkeypatch.setattr(ip, "_port_listening", lambda h, p: True)
# Mock the docker availability check so we never shell out.
monkeypatch.setattr(
"tools.environments.docker._ensure_docker_available", lambda: None,
)
# Mock find_docker so the resolved docker exe isn't probed.
monkeypatch.setattr(
"tools.environments.docker.find_docker", lambda: "/bin/true",
)
# Mock subprocess.run so we don't actually run `docker run`. We
# only need the constructor to get past the env merge logic.
monkeypatch.setattr(
"tools.environments.docker.subprocess.run",
lambda *a, **k: MagicMock(stdout="abc123\n", returncode=0),
)
# init_session is the second outbound subprocess we don't care about.
monkeypatch.setattr(
"tools.environments.docker.DockerEnvironment.init_session",
lambda self: None,
)
# The collision: user sets HTTPS_PROXY to empty string in docker_env.
with pytest.raises(RuntimeError, match="overrides egress-proxy variables"):
DockerEnvironment(
image="busybox",
env={"HTTPS_PROXY": ""}, # the collision
)

389
tests/test_iron_proxy_cli.py Normal file
View file

@ -0,0 +1,389 @@
"""Unit tests for ``hermes_cli.proxy_cli`` command handlers.
These tests cover the user-facing CLI surface that was previously
uncovered. We mock the iron_proxy module's side-effect functions
(install / start / stop / discover) and exercise the dispatch +
return-code logic plus the small amount of presentation logic in
each handler (e.g. --from-bitwarden's fail-loud path).
"""
from __future__ import annotations
import argparse
import os
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from agent.proxy_sources import iron_proxy as ip
from hermes_cli import proxy_cli
@pytest.fixture
def hermes_home(tmp_path, monkeypatch):
"""Point HERMES_HOME at a temp dir so the wizard doesn't touch the
operator's real config. Also blanks any provider env vars so we
don't accidentally read a real key."""
home = tmp_path / "hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
for key in list(os.environ):
if key.endswith("_API_KEY") or key in (
"BWS_ACCESS_TOKEN", "ANTHROPIC_API_KEY",
"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY",
):
monkeypatch.delenv(key, raising=False)
return home
def _args(**overrides):
ns = argparse.Namespace(
force=False,
tunnel_port=None,
from_bitwarden=False,
rotate_tokens=False,
show_tokens=False,
)
for k, v in overrides.items():
setattr(ns, k, v)
return ns
# ---------------------------------------------------------------------------
# cmd_install
# ---------------------------------------------------------------------------
def test_cmd_install_success_returns_0(hermes_home, monkeypatch):
monkeypatch.setattr(ip, "install_iron_proxy", lambda **kw: Path("/tmp/iron-proxy"))
monkeypatch.setattr(ip, "iron_proxy_version", lambda b: "v0.39.0-test")
rc = proxy_cli.cmd_install(_args())
assert rc == 0
def test_cmd_install_failure_returns_1(hermes_home, monkeypatch):
def boom(**kw):
raise RuntimeError("download failed")
monkeypatch.setattr(ip, "install_iron_proxy", boom)
rc = proxy_cli.cmd_install(_args())
assert rc == 1
# ---------------------------------------------------------------------------
# cmd_setup — --from-bitwarden fail-loud paths
# ---------------------------------------------------------------------------
def test_cmd_setup_from_bitwarden_refuses_when_bw_disabled(hermes_home, monkeypatch):
"""When --from-bitwarden is passed but secrets.bitwarden.enabled=false,
the wizard must FAIL rather than silently rewriting credential_source
to bitwarden."""
from hermes_cli.config import load_config, save_config
cfg = load_config()
cfg.setdefault("secrets", {})["bitwarden"] = {"enabled": False}
save_config(cfg)
# Pre-stub install + CA so we get to step 3.
monkeypatch.setattr(ip, "find_iron_proxy", lambda **kw: Path("/tmp/iron-proxy"))
monkeypatch.setattr(ip, "iron_proxy_version", lambda b: "test")
monkeypatch.setattr(
ip, "ensure_ca_cert",
lambda **kw: (hermes_home / "ca.crt", hermes_home / "ca.key"),
)
rc = proxy_cli.cmd_setup(_args(from_bitwarden=True))
assert rc == 1
# Verify we did NOT write credential_source: bitwarden to config.
cfg2 = load_config()
proxy_cfg = cfg2.get("proxy") or {}
assert proxy_cfg.get("credential_source", "env") != "bitwarden"
def test_cmd_setup_from_bitwarden_refuses_when_token_missing(hermes_home, monkeypatch):
"""--from-bitwarden with secrets.bitwarden.enabled=true but BWS access
token unset → fail loud, not silent env-fallback."""
from hermes_cli.config import load_config, save_config
cfg = load_config()
cfg.setdefault("secrets", {})["bitwarden"] = {
"enabled": True,
"project_id": "test-proj",
"access_token_env": "BWS_ACCESS_TOKEN",
}
save_config(cfg)
monkeypatch.delenv("BWS_ACCESS_TOKEN", raising=False)
monkeypatch.setattr(ip, "find_iron_proxy", lambda **kw: Path("/tmp/iron-proxy"))
monkeypatch.setattr(ip, "iron_proxy_version", lambda b: "test")
monkeypatch.setattr(
ip, "ensure_ca_cert",
lambda **kw: (hermes_home / "ca.crt", hermes_home / "ca.key"),
)
rc = proxy_cli.cmd_setup(_args(from_bitwarden=True))
assert rc == 1
def test_cmd_setup_from_bitwarden_refuses_on_empty_vault(hermes_home, monkeypatch):
"""If BW returns {} (empty vault / scoped wrong / unreachable), fail
loud rather than silently writing credential_source: bitwarden."""
from hermes_cli.config import load_config, save_config
cfg = load_config()
cfg.setdefault("secrets", {})["bitwarden"] = {
"enabled": True,
"project_id": "test-proj",
"access_token_env": "BWS_ACCESS_TOKEN",
}
save_config(cfg)
monkeypatch.setenv("BWS_ACCESS_TOKEN", "bwsk-test-token")
monkeypatch.setattr(ip, "find_iron_proxy", lambda **kw: Path("/tmp/iron-proxy"))
monkeypatch.setattr(ip, "iron_proxy_version", lambda b: "test")
monkeypatch.setattr(
ip, "ensure_ca_cert",
lambda **kw: (hermes_home / "ca.crt", hermes_home / "ca.key"),
)
# Mock fetch_bitwarden_secrets to return an empty dict (empty vault).
fake_bw = MagicMock()
fake_bw.fetch_bitwarden_secrets = lambda **kw: ({}, [])
monkeypatch.setattr("agent.secret_sources.bitwarden", fake_bw, raising=False)
import sys
sys.modules["agent.secret_sources.bitwarden"] = fake_bw
rc = proxy_cli.cmd_setup(_args(from_bitwarden=True))
assert rc == 1
def test_cmd_setup_rejects_tunnel_port_zero(hermes_home, monkeypatch):
"""--tunnel-port=0 is rejected explicitly (was silently substituting
the default before the fix)."""
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-test")
monkeypatch.setattr(ip, "find_iron_proxy", lambda **kw: Path("/tmp/iron-proxy"))
monkeypatch.setattr(ip, "iron_proxy_version", lambda b: "test")
monkeypatch.setattr(
ip, "ensure_ca_cert",
lambda **kw: (hermes_home / "ca.crt", hermes_home / "ca.key"),
)
rc = proxy_cli.cmd_setup(_args(tunnel_port=0))
assert rc == 1
# ---------------------------------------------------------------------------
# cmd_start — fail_on_uncovered_providers + Bitwarden rotation wire-up
# ---------------------------------------------------------------------------
def test_cmd_start_refuses_when_proxy_disabled(hermes_home, monkeypatch):
from hermes_cli.config import load_config, save_config
cfg = load_config()
cfg.setdefault("proxy", {})["enabled"] = False
save_config(cfg)
rc = proxy_cli.cmd_start(_args())
assert rc == 1
def test_cmd_start_refuses_on_uncovered_provider_when_strict(hermes_home, monkeypatch):
"""fail_on_uncovered_providers=true + ANTHROPIC_API_KEY in env =
refuse to start (real credential would otherwise leak into sandbox)."""
from hermes_cli.config import load_config, save_config
cfg = load_config()
cfg.setdefault("proxy", {})["enabled"] = True
cfg["proxy"]["fail_on_uncovered_providers"] = True
save_config(cfg)
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
rc = proxy_cli.cmd_start(_args())
assert rc == 1
def test_cmd_start_passes_bitwarden_refresh_flag_when_credential_source_is_bitwarden(
hermes_home, monkeypatch,
):
"""When credential_source=bitwarden, cmd_start must wire
refresh_secrets_from_bitwarden=True into start_proxy. That's what
delivers the rotation promise the docs make."""
from hermes_cli.config import load_config, save_config
cfg = load_config()
cfg.setdefault("proxy", {})["enabled"] = True
cfg["proxy"]["credential_source"] = "bitwarden"
cfg["proxy"]["fail_on_uncovered_providers"] = False
cfg.setdefault("secrets", {})["bitwarden"] = {"enabled": True}
save_config(cfg)
captured: dict = {}
def fake_start_proxy(**kw):
captured.update(kw)
s = ip.ProxyStatus()
s.pid = 4242
s.listening = True
s.tunnel_port = 9090
return s
monkeypatch.setattr(ip, "start_proxy", fake_start_proxy)
monkeypatch.setattr(ip, "discover_uncovered_providers", lambda **kw: [])
rc = proxy_cli.cmd_start(_args())
assert rc == 0
assert captured.get("refresh_secrets_from_bitwarden") is True
assert captured.get("bitwarden_config") is not None
def test_cmd_start_does_not_pass_bitwarden_refresh_when_credential_source_is_env(
hermes_home, monkeypatch,
):
from hermes_cli.config import load_config, save_config
cfg = load_config()
cfg.setdefault("proxy", {})["enabled"] = True
cfg["proxy"]["credential_source"] = "env"
cfg["proxy"]["fail_on_uncovered_providers"] = False
save_config(cfg)
captured: dict = {}
def fake_start_proxy(**kw):
captured.update(kw)
s = ip.ProxyStatus()
s.pid = 4242
s.listening = True
return s
monkeypatch.setattr(ip, "start_proxy", fake_start_proxy)
monkeypatch.setattr(ip, "discover_uncovered_providers", lambda **kw: [])
rc = proxy_cli.cmd_start(_args())
assert rc == 0
assert captured.get("refresh_secrets_from_bitwarden") is False
# ---------------------------------------------------------------------------
# cmd_stop, cmd_status, cmd_disable, cmd_config
# ---------------------------------------------------------------------------
def test_cmd_stop_returns_0_when_running(hermes_home, monkeypatch):
monkeypatch.setattr(ip, "stop_proxy", lambda: True)
rc = proxy_cli.cmd_stop(_args())
assert rc == 0
def test_cmd_stop_returns_0_when_already_stopped(hermes_home, monkeypatch):
monkeypatch.setattr(ip, "stop_proxy", lambda: False)
rc = proxy_cli.cmd_stop(_args())
assert rc == 0
def test_cmd_status_returns_0(hermes_home, monkeypatch):
monkeypatch.setattr(ip, "get_status", lambda: ip.ProxyStatus())
monkeypatch.setattr(ip, "load_mappings", lambda: [])
monkeypatch.setattr(ip, "discover_uncovered_providers", lambda **kw: [])
rc = proxy_cli.cmd_status(_args())
assert rc == 0
def test_cmd_disable_uses_public_status_pid_not_private_read_pid(
hermes_home, monkeypatch,
):
"""cmd_disable must read status.pid (which incorporates the _pid_alive
check) — NOT ip._read_pid() directly (which would fire a spurious
'still running' warning for a stale pidfile from a crashed run)."""
from hermes_cli.config import load_config, save_config
cfg = load_config()
cfg.setdefault("proxy", {})["enabled"] = True
save_config(cfg)
# Pidfile exists but the process is dead. Old code would have warned
# "still running"; the new code reads status.pid which returns None
# because _pid_alive is False, so no spurious warning.
state = ip._proxy_state_dir()
(state / "iron-proxy.pid").write_text("99999")
# _pid_alive returns False → status.pid is None.
monkeypatch.setattr(ip, "_pid_alive", lambda pid: False)
# If cmd_disable reads _read_pid() directly (old path), this test
# would still pass — but reading status.pid is the correct
# API. Sentinel: confirm _read_pid is NOT called from cmd_disable.
read_pid_calls = []
real_read_pid = ip._read_pid
def tracked_read_pid(*a, **kw):
read_pid_calls.append((a, kw))
return real_read_pid(*a, **kw)
monkeypatch.setattr(ip, "_read_pid", tracked_read_pid)
rc = proxy_cli.cmd_disable(_args())
assert rc == 0
# cmd_disable should call get_status() (which may internally call
# _read_pid), but should NOT call _read_pid from its own body.
# Hard to assert directly without source-introspection — the meatier
# assertion is that no "still running" message fired with a stale
# pidfile. That's covered by inspecting return code + config
# mutation only.
from hermes_cli.config import load_config as _lc
cfg2 = _lc()
assert cfg2["proxy"]["enabled"] is False
def test_cmd_config_returns_0_when_present(hermes_home, monkeypatch):
fake = ip.ProxyStatus()
fake.config_path = Path("/tmp/proxy.yaml")
monkeypatch.setattr(ip, "get_status", lambda: fake)
rc = proxy_cli.cmd_config(_args())
assert rc == 0
def test_cmd_config_returns_1_when_missing(hermes_home, monkeypatch):
monkeypatch.setattr(ip, "get_status", lambda: ip.ProxyStatus())
rc = proxy_cli.cmd_config(_args())
assert rc == 1
# ---------------------------------------------------------------------------
# Argparse wiring — dest='egress_command' regression
# ---------------------------------------------------------------------------
def test_register_cli_uses_egress_command_dest():
"""The subparser dest must be 'egress_command' to stay disjoint from
the inbound OAuth 'hermes proxy' subparser (dest='proxy_command').
A future grep-and-refactor on proxy_command should not hit this
subparser by accident."""
parser = argparse.ArgumentParser(prog="hermes egress")
proxy_cli.register_cli(parser)
# Parse a no-op invocation and confirm the attribute name.
args = parser.parse_args(["install"])
assert hasattr(args, "egress_command")
assert not hasattr(args, "proxy_command")
def test_egress_subcommands_registered():
"""Smoke test: every documented subcommand parses without error."""
parser = argparse.ArgumentParser(prog="hermes egress")
proxy_cli.register_cli(parser)
for sub in ("install", "setup", "start", "stop", "status", "disable", "config"):
args = parser.parse_args([sub])
assert args.egress_command == sub
def test_setup_has_rotate_tokens_flag():
"""--rotate-tokens is the documented escape hatch for re-rolling
every proxy token (used after a suspected token leak). Default is
preserve-existing."""
parser = argparse.ArgumentParser(prog="hermes egress")
proxy_cli.register_cli(parser)
args = parser.parse_args(["setup"])
assert args.rotate_tokens is False
args = parser.parse_args(["setup", "--rotate-tokens"])
assert args.rotate_tokens is True

127
tools/environments/docker.py
View file

@ -198,10 +198,15 @@ def _egress_proxy_args_for_docker() -> tuple[list[str], dict[str, str], list[str
backend refuses to start the sandbox.
"""
# Narrow except: ImportError is the only legitimate failure here.
# Bare ``except Exception`` would hide AttributeError, SyntaxError in
# the config module, etc. and silently start the sandbox without
# proxy enforcement. We let unexpected exceptions propagate so the
# docker backend visibly fails rather than degrading silently.
try:
from hermes_cli.config import load_config
from agent.proxy_sources import iron_proxy as ip
except Exception as exc: # noqa: BLE001
except ImportError as exc:
logger.debug("Egress proxy plumbing unavailable: %s", exc)
return ([], {}, [])
@ -234,7 +239,36 @@ def _egress_proxy_args_for_docker() -> tuple[list[str], dict[str, str], list[str
return ([], {}, [])
if status.ca_cert_path is None or not status.ca_cert_path.exists():
# configured says ca exists; defensive double-check
# status.configured was True a moment ago but the CA file has
# disappeared. Treat this with the same enforce semantics as the
# other failure branches — silently dropping the CA mount would
# leave the sandbox with proxy env vars pointing at iron-proxy
# but no trust anchor, so every TLS handshake would 5xx; or
# worse, with enforce_on_docker=false we'd drop both the proxy
# vars AND any other isolation, opening the sandbox.
msg = (
f"iron-proxy CA cert vanished from {status.ca_cert_path}. "
"Re-run `hermes egress setup` to regenerate it."
)
if enforce:
raise RuntimeError(msg)
logger.warning("%s — continuing without proxy (enforce_on_docker=false).", msg)
return ([], {}, [])
# Corrupt or empty mappings.json is a silent failure mode that's
# indistinguishable from an upstream outage from inside the sandbox
# (every request returns 403). Refuse to mount with empty mappings
# rather than ship a broken sandbox.
mappings = ip.load_mappings()
if not mappings:
msg = (
"iron-proxy is configured but mappings.json is empty or "
"corrupt. Re-run `hermes egress setup` to mint provider "
"tokens before starting a sandbox."
)
if enforce:
raise RuntimeError(msg)
logger.warning("%s — continuing without proxy (enforce_on_docker=false).", msg)
return ([], {}, [])
container_ca = "/etc/ssl/certs/hermes-egress-ca.crt"
@ -255,10 +289,21 @@ def _egress_proxy_args_for_docker() -> tuple[list[str], dict[str, str], list[str
"no_proxy": "127.0.0.1,localhost,::1",
# CA bundle locations for the major language runtimes. iron-proxy
# presents a leaf cert signed by our CA on every MITM'd connection.
#
# CRITICAL ASYMMETRY: Python (REQUESTS_CA_BUNDLE / SSL_CERT_FILE)
# and curl (CURL_CA_BUNDLE) REPLACE the system CA store.
# NODE_EXTRA_CA_CERTS ADDS to it. A Node.js process that
# bypasses HTTPS_PROXY by using a raw socket would still see the
# system CA store and succeed where Python/curl fail validation.
# We additionally set NODE_OPTIONS=--use-openssl-ca to force Node
# through the OpenSSL store that SSL_CERT_FILE controls, narrowing
# the asymmetry. Not a complete fix — see the docs caveat — but
# closes the easy case.
"REQUESTS_CA_BUNDLE": container_ca, # Python `requests`
"SSL_CERT_FILE": container_ca, # Python ssl module / OpenSSL
"CURL_CA_BUNDLE": container_ca, # curl
"NODE_EXTRA_CA_CERTS": container_ca, # Node.js
"NODE_EXTRA_CA_CERTS": container_ca, # Node.js: adds to system store
"NODE_OPTIONS": "--use-openssl-ca", # Node.js: route through OpenSSL store
# For the agent inside the sandbox to identify itself as proxy-aware.
"HERMES_EGRESS_PROXY": "1",
}
@ -266,7 +311,7 @@ def _egress_proxy_args_for_docker() -> tuple[list[str], dict[str, str], list[str
# Surface the per-provider proxy tokens. The sandbox can swap these into
# its provider config (or its env, if it reads the standard names) and the
# proxy translates them to the real secrets on egress.
for m in ip.load_mappings():
for m in mappings:
env_overrides[f"HERMES_PROXY_TOKEN_{m.real_env_name}"] = m.proxy_token
# On Linux, host.docker.internal isn't populated by default — Docker Desktop
@ -564,11 +609,75 @@ class DockerEnvironment(BaseEnvironment):
# Explicit environment variables (docker_env config) — set at container
# creation so they're available to all processes (including entrypoint).
# Egress proxy env vars (HTTPS_PROXY, CA-bundle paths, proxy tokens)
# are merged here so they're visible to the container's entrypoint as
# well as any subsequent execs. Existing self._env entries win, so
# users can still override the proxy explicitly via docker_env config.
merged_env = dict(egress_env_overrides)
merged_env.update(self._env)
# are merged below. Precedence policy:
#
# - When egress enforcement is on AND the user's docker_env tries
# to override one of the proxy-control vars (HTTPS_PROXY,
# SSL_CERT_FILE, etc.), fail-loud rather than silently inverting
# the isolation. The CA mount + tokens would still ship while
# traffic leaves the sandbox direct with real credentials —
# exactly what enforce_on_docker is meant to prevent.
# - When enforcement is off, the user's docker_env wins (current
# behavior) but we log a warning naming both config sources.
# - When the user override is identical to the egress value, no-op.
if egress_env_overrides:
try:
from hermes_cli.config import load_config as _load_cfg_for_collision
_proxy_cfg = (_load_cfg_for_collision().get("proxy") or {})
except ImportError:
_proxy_cfg = {}
_enforce_egress = bool(_proxy_cfg.get("enforce_on_docker", True))
# Only the env vars that *control* the egress posture matter
# for collision detection — auxiliary informational vars like
# HERMES_EGRESS_PROXY can be safely overridden.
_critical = {
"HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy",
"NO_PROXY", "no_proxy",
"REQUESTS_CA_BUNDLE", "SSL_CERT_FILE", "CURL_CA_BUNDLE",
"NODE_EXTRA_CA_CERTS",
}
_collisions = sorted(
k for k in _critical
if k in self._env
and k in egress_env_overrides
and self._env[k] != egress_env_overrides[k]
)
if _collisions:
_msg = (
f"docker_env in config.yaml overrides egress-proxy "
f"variables {_collisions}; enforce_on_docker is "
f"{'enabled' if _enforce_egress else 'disabled'}."
)
if _enforce_egress:
raise RuntimeError(
f"{_msg} Remove these keys from docker_env or "
"disable enforce_on_docker to opt out of egress "
"isolation."
)
logger.warning(
"%s Falling back to docker_env values; sandbox traffic "
"will NOT route through the proxy.", _msg,
)
# When enforce_on_docker is true, egress overrides win. When
# false, docker_env wins (back-compat for users who deliberately
# opt out). In both cases the collision check above has already
# surfaced any disagreement.
try:
from hermes_cli.config import load_config as _load_cfg_for_precedence
_enforce_egress_merge = bool(
(_load_cfg_for_precedence().get("proxy") or {})
.get("enforce_on_docker", True)
)
except ImportError:
_enforce_egress_merge = True
if _enforce_egress_merge and egress_env_overrides:
merged_env = dict(self._env)
merged_env.update(egress_env_overrides)
else:
merged_env = dict(egress_env_overrides)
merged_env.update(self._env)
env_args = []
for key in sorted(merged_env):
env_args.extend(["-e", f"{key}={merged_env[key]}"])

4
website/docs/user-guide/egress/iron-proxy.md
View file

@ -169,6 +169,6 @@ A request to a non-allowlisted host (e.g. `https://attacker.example.com/leak?key
## See also
- Upstream project: <https://github.com/ironsh/iron-proxy>
- Upstream docs: <https://docs.iron.sh/>
- Upstream project: [github.com/ironsh/iron-proxy](https://github.com/ironsh/iron-proxy)
- Upstream docs: [docs.iron.sh](https://docs.iron.sh/)
- Bitwarden integration: [`hermes secrets bitwarden`](../secrets/bitwarden)