hermes-agent/tests/agent/test_iron_proxy.py

"""Hermetic tests for the iron-proxy egress integration.

Covers the pure-function surface (token mint, mapping discovery, config build,
config + mappings I/O), the subprocess lifecycle
(spawn / PID / pid_alive / stop, with subprocess.Popen mocked), and the
docker backend's egress arg builder.

PM acquisition has real loopback/worker coverage in test_security_consumers.
Live network and the real ``iron-proxy`` binary are NEVER touched.  See
``tests/agent/test_iron_proxy_e2e.py`` (gated behind a marker) for the real-binary
smoke test.
"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from agent.proxy_sources import iron_proxy as ip


# ---------------------------------------------------------------------------
# Per-test isolation
# ---------------------------------------------------------------------------


@pytest.fixture
def hermes_home(tmp_path, monkeypatch):
    """Point HERMES_HOME at a temp dir so install paths don't touch the real $HOME."""

    home = tmp_path / "hermes"
    home.mkdir()
    monkeypatch.setenv("HERMES_HOME", str(home))
    # Make sure no stale provider keys influence discovery.
    for key in list(os.environ):
        if key.endswith("_API_KEY"):
            monkeypatch.delenv(key, raising=False)
    return home


# ---------------------------------------------------------------------------
# Token mint + mapping discovery
# ---------------------------------------------------------------------------


def test_mint_proxy_token_has_prefix_and_length():
    t = ip.mint_proxy_token("alpha")
    assert t.startswith("alpha-")
    assert len(t) >= len("alpha-") + 32


def test_management_token_path_is_single_authority(hermes_home):
    """One token path: <hermes_home>/proxy/management.token, shared by mint, reuse and readers."""
    assert ip._management_token_path() == ip._proxy_state_dir_ro() / "management.token"
    assert not (hermes_home / "proxy").exists()

    token = ip.ensure_management_token()
    assert token.startswith("hermes-mgmt-")
    p = ip._management_token_path()
    assert p.is_file()
    assert p.read_text(encoding="utf-8-sig").strip() == token
    # 0600-style private write: reuse without minting a second token.
    assert ip.ensure_management_token() == token
    # Forced rotation mints a new token at the same single path.
    rotated = ip.ensure_management_token(force=True)
    assert rotated != token
    assert ip._management_token_path().read_text(encoding="utf-8-sig").strip() == rotated


    # Unknown providers (no entry in _BEARER_PROVIDERS) are skipped, not warned.


# ---------------------------------------------------------------------------
# Config / mapping serialization
# ---------------------------------------------------------------------------


def _sample_mapping(env_name: str = "OPENROUTER_API_KEY") -> ip.TokenMapping:
    return ip.TokenMapping(
        proxy_token=ip.mint_proxy_token("test"),
        real_env_name=env_name,
        upstream_hosts=("openrouter.ai", "*.openrouter.ai"),
    )


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=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 "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)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Bind policy (regression: must not bind 0.0.0.0)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# audit_log file pre-creation (parameter still accepted; v0.39 doesn't
# wire it into the binary config but ensure_audit_log() still creates
# the file at 0o600 as a logrotate / monitoring sentinel)
# ---------------------------------------------------------------------------


def test_audit_log_kwarg_does_not_inject_audit_path_v039(tmp_path):
    """v0.39 of iron-proxy rejects ``log.audit_path`` (not a struct
    field).  build_proxy_config still accepts the audit_log kwarg for
    forward compatibility but MUST NOT emit it into the rendered yaml
    until the upstream binary supports it.  See the kwarg's docstring
    for the upgrade 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 "audit_path" not in cfg["log"], (
        "iron-proxy v0.39 has no log.audit_path field; emitting it "
        "causes 'field audit_path not found in type config.Log' at "
        "daemon start.  ensure_audit_log() still creates the file as "
        "an operator-facing logrotate target."
    )


def test_load_mappings_handles_corrupt_json(hermes_home):
    state = ip._proxy_state_dir()
    (state / "mappings.json").write_text("{not json", encoding="utf-8")
    assert ip.load_mappings() == []


# ---------------------------------------------------------------------------
# Token-preservation on re-setup (regression: clobbered live sandboxes)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Uncovered provider detection (regression: signature-auth providers bypass)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Binary discovery + lazy install
# ---------------------------------------------------------------------------


# ── GPG release-signature verification (maxpetrusenko P1) ────────────────────

def test_verify_checksums_signature_skips_without_gpg(hermes_home, monkeypatch, tmp_path):
    """No gpg on PATH → degrade gracefully (return False), do not raise."""
    monkeypatch.setattr(ip.shutil, "which", lambda name: None)
    cks = tmp_path / "checksums.txt"
    cks.write_text("abc  iron-proxy.tar.gz\n")
    assert ip._verify_checksums_signature(tmp_path, cks) is False


# ---------------------------------------------------------------------------
# Subprocess lifecycle
# ---------------------------------------------------------------------------


def test_start_proxy_idempotent_when_already_running(hermes_home, monkeypatch):
    state = ip._proxy_state_dir()
    pid_file = state / "iron-proxy.pid"
    pid_file.write_text("12345")
    monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
    monkeypatch.setattr(ip, "_port_listening", lambda h, p: True)
    monkeypatch.setattr(ip, "iron_proxy_version", lambda b: "test")
    # Materialize config so we get past that check (we shouldn't reach it,
    # but if the idempotent path regresses we want a clean failure mode).
    (state / "proxy.yaml").write_text("proxy: {}")
    # Sentinel: subprocess.Popen must NOT be called.
    with patch("subprocess.Popen", lambda *a, **k: pytest.fail("should not spawn")):
        status = ip.start_proxy()
    # Should return without spawning anything.
    assert status is not None


# ---------------------------------------------------------------------------
# Docker integration
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Platform asset name resolution
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# 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"


# ---------------------------------------------------------------------------
# CA generation TOCTOU (regression: 0o600 only set AFTER copy)
# ---------------------------------------------------------------------------


@pytest.mark.platforms("linux")
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)
# ---------------------------------------------------------------------------


@pytest.mark.platforms("linux")
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


@pytest.mark.platforms("linux")
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)
# ---------------------------------------------------------------------------


@pytest.mark.platforms("linux")
def test_proxy_state_dir_is_0o700(hermes_home):
    state = ip._proxy_state_dir()
    mode = state.stat().st_mode & 0o777
    assert mode == 0o700


# ---------------------------------------------------------------------------
# Mappings clobber refused when corrupt (regression: silent 403s)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# CA missing → enforce_on_docker semantics (regression: silent fail-open)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Docker env collision detection (regression: docker_env silently bypassed proxy)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# v3 round: bridge-IP parser hardening (P1 #1)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# v3: default deny-list adjacency (P2 IPv4-mapped-v6 + CGNAT)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Header-auth providers (x-api-key family) — match_headers + aliases
# ---------------------------------------------------------------------------


def test_mappings_roundtrip_preserves_headers_and_aliases(hermes_home):
    m = ip.TokenMapping(
        proxy_token=ip.mint_proxy_token("gemini"),
        real_env_name="GEMINI_API_KEY",
        upstream_hosts=("generativelanguage.googleapis.com",),
        match_headers=("x-goog-api-key",),
        alias_env_names=("GOOGLE_API_KEY",),
    )
    ip.write_mappings([m])
    loaded = ip.load_mappings()
    assert loaded[0].match_headers == ("x-goog-api-key",)
    assert loaded[0].alias_env_names == ("GOOGLE_API_KEY",)


# ---------------------------------------------------------------------------
# Management API (hot reload)
# ---------------------------------------------------------------------------


@pytest.mark.platforms("linux")
def test_management_token_is_private(hermes_home):
    ip.ensure_management_token()
    assert (ip._management_token_path().stat().st_mode & 0o777) == 0o600


def test_reload_proxy_refuses_when_not_running(hermes_home, monkeypatch):
    monkeypatch.setattr(ip, "_read_pid", lambda: None)
    with pytest.raises(RuntimeError, match="not running"):
        ip.reload_proxy()


def test_reload_proxy_posts_bearer_to_management_endpoint(hermes_home, monkeypatch):
    monkeypatch.setattr(ip, "_read_pid", lambda: 4242)
    monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
    monkeypatch.setattr(
        ip, "_read_management_listen_from_config",
        lambda config_path=None: ("127.0.0.1", 9092),
    )
    ip.ensure_management_token()

    captured = {}

    class _FakeResp:
        status = 200
        def __enter__(self):
            return self
        def __exit__(self, *a):
            return False

    def fake_urlopen(req, timeout=None):
        captured["url"] = req.full_url
        captured["method"] = req.get_method()
        captured["auth"] = req.get_header("Authorization")
        return _FakeResp()

    import urllib.request
    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)

    assert ip.reload_proxy() is True
    assert captured["url"] == "http://127.0.0.1:9092/v1/reload"
    assert captured["method"] == "POST"
    token = (ip._proxy_state_dir() / "management.token").read_text().strip()
    assert captured["auth"] == f"Bearer {token}"


@pytest.mark.platforms("linux")
def test_start_proxy_injects_management_key_env(hermes_home, monkeypatch):
    """When the generated config has a management listener, start_proxy
    must inject the bearer key env var — v0.39 refuses to start when
    api_key_env is empty."""

    cfg_path = ip._proxy_state_dir() / "proxy.yaml"
    cfg = ip.build_proxy_config(
        mappings=[_sample_mapping()],
        ca_cert=hermes_home / "ca.crt",
        ca_key=hermes_home / "ca.key",
        http_listen=["127.0.0.1:9090"],
    )
    ip.write_proxy_config(cfg)
    (hermes_home / "bin").mkdir(parents=True, exist_ok=True)
    fake_bin = hermes_home / "bin" / "iron-proxy"
    fake_bin.write_text("#!/bin/sh\nsleep 60\n")
    fake_bin.chmod(0o755)

    captured_env = {}

    class _FakeProc:
        pid = 99999
        def poll(self):
            return None

    def fake_popen(cmd, **kw):
        captured_env.update(kw.get("env") or {})
        return _FakeProc()

    monkeypatch.setattr(ip.subprocess, "Popen", fake_popen)
    monkeypatch.setattr(ip, "_write_pidfile_safely", lambda pf, pid: None)
    monkeypatch.setattr(ip, "_port_listening", lambda h, p: True)
    monkeypatch.setattr(ip, "get_status", lambda: ip.ProxyStatus(pid=99999, listening=True))

    ip.start_proxy(binary=fake_bin, config_path=cfg_path, install_if_missing=False)
    assert captured_env.get(ip._MGMT_API_KEY_ENV)
    assert captured_env[ip._MGMT_API_KEY_ENV].startswith("hermes-mgmt-")


# ---------------------------------------------------------------------------
# v3: _pid_proc_starttime parser (handles comm with parens, brackets)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# v3: stop_proxy SIGKILL suppression on pid recycle (P3 #5 coverage gap)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# v3: _reset_for_tests actually clears module state (P3 #1)
# ---------------------------------------------------------------------------




# ---------------------------------------------------------------------------
# v3: version cache doesn't poison on empty stdout (P2 _VERSION_CACHE bug B)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# v3: NODE_OPTIONS append-merge in docker env (arshkumarsingh #1)
# ---------------------------------------------------------------------------


def test_docker_egress_node_options_uses_sentinel(hermes_home, monkeypatch):
    """``_egress_proxy_args_for_docker`` should NOT put NODE_OPTIONS in
    env_overrides directly; it uses a sentinel key
    ``_HERMES_EGRESS_NODE_OPTIONS_APPEND`` so DockerEnvironment can
    append-merge with the operator's existing NODE_OPTIONS."""

    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")
    mapping = _sample_mapping("OPENROUTER_API_KEY")
    proxy_cfg = ip.build_proxy_config(
        mappings=[mapping], ca_cert=ca, ca_key=state / "ca.key", tunnel_port=9090,
    )
    ip.write_proxy_config(proxy_cfg)
    ip.write_mappings([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)

    _, env, _ = _egress_proxy_args_for_docker()
    # The egress dict should contain the sentinel, NOT a raw NODE_OPTIONS.
    assert env.get("_HERMES_EGRESS_NODE_OPTIONS_APPEND") == "--use-openssl-ca"
    assert "NODE_OPTIONS" not in env, (
        "NODE_OPTIONS in egress env_overrides would clobber the operator's "
        "docker_env NODE_OPTIONS — that's exactly the bug arshkumarsingh "
        "flagged."
    )


# ---------------------------------------------------------------------------
# v3: ensure_audit_log fails loud on OSError (P2 promise mismatch)
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# v3: persisted nonce roundtrip (stephenschoettler #3 cross-CLI defense)
# ---------------------------------------------------------------------------


def test_persisted_nonce_roundtrip(hermes_home, monkeypatch):
    """Write the nonce next to the pidfile (simulating one CLI invocation
    finishing start_proxy), then verify a fresh _read_persisted_nonce
    can pick it up — that's what cross-process _pid_alive uses."""

    nonce_path = ip._persisted_nonce_path()
    nonce_path.parent.mkdir(parents=True, exist_ok=True)
    nonce_path.write_text("test-nonce-abc123")
    assert ip._read_persisted_nonce() == "test-nonce-abc123"


# ---------------------------------------------------------------------------
# v4 round (GodsBoy follow-up): bind-host-aware liveness probes +
# allow_env_fallback on the partial-secret path
# ---------------------------------------------------------------------------


def test_get_status_probes_configured_bind_host(hermes_home, monkeypatch):
    """get_status must probe the configured bind host (e.g. the docker
    bridge IP), not loopback unconditionally."""

    state = ip._proxy_state_dir()
    (state / "proxy.yaml").write_text(
        "proxy:\n  http_listen: 172.17.0.1:9123\n", encoding="utf-8"
    )
    (state / "ca.crt").write_text("cert")
    ip._write_pidfile_safely(ip._pidfile(), 99999)
    monkeypatch.setattr(ip, "_pid_alive", lambda pid: True)
    monkeypatch.setattr(ip, "find_iron_proxy", lambda **kw: None)

    probed = {}

    def fake_probe(host, port):
        probed["host"] = host
        probed["port"] = port
        return True

    monkeypatch.setattr(ip, "_port_listening", fake_probe)
    status = ip.get_status()
    assert probed == {"host": "172.17.0.1", "port": 9123}
    assert status.listening is True
    assert status.tunnel_port == 9123


def test_partial_bitwarden_secrets_honor_allow_env_fallback(
    hermes_home, monkeypatch,
):
    """The missing-secret branch's own error message tells operators to
    set proxy.allow_env_fallback — so the flag must actually work there
    (previously only the empty-token branch honored it)."""

    ip.write_mappings([_sample_mapping("OPENROUTER_API_KEY")])
    monkeypatch.setenv("OPENROUTER_API_KEY", "sk-host-fallback")

    import agent.secret_sources.bitwarden as bw
    monkeypatch.setattr(
        bw, "fetch_bitwarden_secrets", lambda **kw: ({}, []),
    )
    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,
    )
    # Falls back to the host env value instead of raising.
    assert env.get("OPENROUTER_API_KEY") == "sk-host-fallback"


def test_partial_bitwarden_secrets_raise_without_fallback(
    hermes_home, monkeypatch,
):
    """Strict default: missing BWS secrets raise."""

    ip.write_mappings([_sample_mapping("OPENROUTER_API_KEY")])
    monkeypatch.setenv("OPENROUTER_API_KEY", "sk-host")

    import agent.secret_sources.bitwarden as bw
    monkeypatch.setattr(
        bw, "fetch_bitwarden_secrets", lambda **kw: ({}, []),
    )
    monkeypatch.setenv("BWS_ACCESS_TOKEN", "tok")
    bw_cfg = {"project_id": "proj", "access_token_env": "BWS_ACCESS_TOKEN"}

    with pytest.raises(RuntimeError, match="did not return secrets"):
        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,
        )