hermes-agent/tests/tools/test_env_passthrough.py

"""Tests for tools.env_passthrough — skill and config env var passthrough."""

import os
import pytest
import hermes_yaml as yaml

from agent import secret_scope as ss
import tools.env_passthrough as _ep_mod
from tools.env_passthrough import (
    clear_env_passthrough,
    get_all_passthrough,
    is_env_passthrough,
    register_env_passthrough,
    resolve_passthrough_value,
)


@pytest.fixture(autouse=True)
def _clean_passthrough():
    """Ensure a clean passthrough state for every test."""
    clear_env_passthrough()
    _ep_mod._config_passthrough.clear()
    ss.set_multiplex_active(False)
    yield
    clear_env_passthrough()
    _ep_mod._config_passthrough.clear()
    ss.set_multiplex_active(False)


class TestSkillScopedPassthrough:


    def test_skips_empty(self):
        register_env_passthrough(["", "  ", "VALID_KEY"])
        assert is_env_passthrough("VALID_KEY")
        assert not is_env_passthrough("")


class TestConfigPassthrough:
    def test_reads_from_config(self, tmp_path, monkeypatch):
        config = {"terminal": {"env_passthrough": ["MY_CUSTOM_KEY", "ANOTHER_TOKEN"]}}
        config_path = tmp_path / "config.yaml"
        config_path.write_text(yaml.safe_dump(config), encoding="utf-8")
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        _ep_mod._config_passthrough.clear()

        assert is_env_passthrough("MY_CUSTOM_KEY")
        assert is_env_passthrough("ANOTHER_TOKEN")
        assert not is_env_passthrough("UNRELATED_VAR")


    def test_union_of_skill_and_config(self, tmp_path, monkeypatch):
        config = {"terminal": {"env_passthrough": ["CONFIG_KEY"]}}
        config_path = tmp_path / "config.yaml"
        config_path.write_text(yaml.safe_dump(config), encoding="utf-8")
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        _ep_mod._config_passthrough.clear()

        register_env_passthrough(["SKILL_KEY"])
        all_pt = get_all_passthrough()
        assert "CONFIG_KEY" in all_pt
        assert "SKILL_KEY" in all_pt


class TestProfileScopedResolution:
    def test_active_scope_overrides_process_fallback(self):
        ss.set_multiplex_active(True)
        token = ss.set_secret_scope({"SERVICE_TOKEN": "profile-b"})
        try:
            assert resolve_passthrough_value("SERVICE_TOKEN", "profile-a") == "profile-b"
        finally:
            ss.reset_secret_scope(token)

    def test_active_scope_does_not_fall_back_to_another_profile(self):
        ss.set_multiplex_active(True)
        token = ss.set_secret_scope({})
        try:
            assert resolve_passthrough_value("SERVICE_TOKEN", "profile-a") is None
        finally:
            ss.reset_secret_scope(token)

    def test_unscoped_multiplex_read_fails_closed(self):
        ss.set_multiplex_active(True)
        with pytest.raises(ss.UnscopedSecretError):
            resolve_passthrough_value("SERVICE_TOKEN", "profile-a")

    def test_single_profile_keeps_callers_fallback(self):
        assert resolve_passthrough_value("SERVICE_TOKEN", "profile-a") == "profile-a"

    def test_active_scope_keeps_explicit_global_override(self, monkeypatch):
        """Global terminal settings still honor a caller-provided override."""
        monkeypatch.setenv("TERMINAL_CWD", "/default")
        ss.set_multiplex_active(True)
        token = ss.set_secret_scope({})
        try:
            assert resolve_passthrough_value("TERMINAL_CWD", "/explicit") == "/explicit"
        finally:
            ss.reset_secret_scope(token)


class TestExecuteCodeIntegration:
    """Verify that the passthrough is checked in execute_code's env filtering."""



    def test_execute_code_uses_active_profile_for_passthrough(self, monkeypatch):
        """The execute_code child must receive the routed profile's value."""
        from tools.code_execution_env import _scrub_child_env

        register_env_passthrough(["SERVICE_TOKEN"])
        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
        ss.set_multiplex_active(True)
        token = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-routed-profile"})
        try:
            child_env = _scrub_child_env({"SERVICE_TOKEN": "token-for-default"})
        finally:
            ss.reset_secret_scope(token)
            ss.set_multiplex_active(False)

        assert child_env["SERVICE_TOKEN"] == "token-for-routed-profile"

    def test_execute_code_omits_missing_scoped_passthrough(self, monkeypatch):
        """A missing routed secret must not leak into the execute_code child."""
        from tools.code_execution_env import _scrub_child_env

        register_env_passthrough(["SERVICE_TOKEN"])
        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
        ss.set_multiplex_active(True)
        token = ss.set_secret_scope({})
        try:
            child_env = _scrub_child_env({"SERVICE_TOKEN": "token-for-default"})
        finally:
            ss.reset_secret_scope(token)
            ss.set_multiplex_active(False)

        assert "SERVICE_TOKEN" not in child_env

    def test_execute_code_strips_buzz_vars(self):
        """BUZZ_* credentials must stay out of the execute_code child even
        though they pass through to terminal children (issue #78026): the
        carve-out is terminal-only.

        - BUZZ_PRIVATE_KEY matches the KEY secret substring.
        - BUZZ_AUTH_TAG matches the AUTH secret substring.
        - BUZZ_RELAY_URL matches no secret substring but is not on the safe
          prefix allowlist, so it is dropped too.
        """
        from tools.code_execution_env import _scrub_child_env

        buzz_vars = {
            "BUZZ_PRIVATE_KEY": "nsec1fake",
            "BUZZ_AUTH_TAG": '["tag","data","kind","sig"]',
            "BUZZ_RELAY_URL": "https://mycommunity.communities.buzz.xyz",
            "PATH": "/usr/bin",
            "HOME": "/home/user",
        }
        child_env = _scrub_child_env(buzz_vars)

        assert "BUZZ_PRIVATE_KEY" not in child_env
        assert "BUZZ_AUTH_TAG" not in child_env
        assert "BUZZ_RELAY_URL" not in child_env
        assert child_env["PATH"] == "/usr/bin"
        assert child_env["HOME"] == "/home/user"


class TestTerminalIntegration:
    """Verify that the passthrough is checked in terminal's env sanitizers."""

    def test_background_terminal_uses_active_profile_for_passthrough(self, monkeypatch):
        """Background/PTY terminal children must use the routed profile value."""
        from tools.environments.local import _sanitize_subprocess_env

        register_env_passthrough(["SERVICE_TOKEN"])
        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
        ss.set_multiplex_active(True)
        token = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-routed-profile"})
        try:
            child_env = _sanitize_subprocess_env(
                {"SERVICE_TOKEN": "token-for-default"},
                {"SERVICE_TOKEN": "token-for-default"},
            )
        finally:
            ss.reset_secret_scope(token)
            ss.set_multiplex_active(False)

        assert child_env["SERVICE_TOKEN"] == "token-for-routed-profile"

    def test_background_terminal_omits_missing_scoped_passthrough(self, monkeypatch):
        """A missing routed secret must not leak into background terminal work."""
        from tools.environments.local import _sanitize_subprocess_env

        register_env_passthrough(["SERVICE_TOKEN"])
        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
        ss.set_multiplex_active(True)
        token = ss.set_secret_scope({})
        try:
            child_env = _sanitize_subprocess_env({"SERVICE_TOKEN": "token-for-default"})
        finally:
            ss.reset_secret_scope(token)
            ss.set_multiplex_active(False)

        assert "SERVICE_TOKEN" not in child_env

    def test_scope_only_declared_name_reaches_every_local_child(self, monkeypatch):
        """A routed profile's declared secret lives only in its scope (its .env never enters the
        process env), so it must be added from the scope on every local spawn surface; an
        undeclared scope entry stays out. No scope bound -> byte-identical single-profile env."""
        from tools.code_execution_env import _scrub_child_env
        from tools.environments.local import _sanitize_subprocess_env

        register_env_passthrough(["SERVICE_TOKEN"])
        monkeypatch.delenv("SERVICE_TOKEN", raising=False)
        base = {"PATH": "/usr/bin", "HOME": "/home/user"}
        ss.set_multiplex_active(True)
        token = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-routed-profile",
                                     "UNDECLARED_TOKEN": "never-forwarded"})
        try:
            terminal_env = _sanitize_subprocess_env(dict(base))
            sandbox_env = _scrub_child_env(dict(base))
        finally:
            ss.reset_secret_scope(token)
            ss.set_multiplex_active(False)

        for child_env in (terminal_env, sandbox_env):
            assert child_env["SERVICE_TOKEN"] == "token-for-routed-profile"
            assert "UNDECLARED_TOKEN" not in child_env
        assert "SERVICE_TOKEN" not in _sanitize_subprocess_env(dict(base))
        assert "SERVICE_TOKEN" not in _scrub_child_env(dict(base))

    def test_scope_overlay_failure_is_loud_on_both_local_surfaces(self, monkeypatch):
        """A scope/config failure while resolving declared scope-only names must raise, not be
        swallowed into a debug log that silently drops the declared secret again (#114209)."""
        import tools.env_passthrough as ep
        from tools.code_execution_env import _scrub_child_env
        from tools.environments.local import _sanitize_subprocess_env

        def _boom(_present):
            raise RuntimeError("scope lookup failed")

        monkeypatch.setattr(ep, "scoped_passthrough_additions", _boom)
        with pytest.raises(RuntimeError, match="scope lookup failed"):
            _sanitize_subprocess_env({"PATH": "/usr/bin"})
        with pytest.raises(RuntimeError, match="scope lookup failed"):
            _scrub_child_env({"PATH": "/usr/bin"})

    def test_shared_local_snapshot_re_resolves_current_profile(self, monkeypatch, tmp_path):
        """A persistent shell snapshot must not retain the previous profile's value."""
        from tools.environments.local import LocalEnvironment

        register_env_passthrough(["SERVICE_TOKEN"])
        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
        ss.set_multiplex_active(True)
        env = None
        token_b = None
        token_c = None
        try:
            token_a = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-profile-a"})
            try:
                env = LocalEnvironment(cwd=str(tmp_path))
                assert env.execute("printf '%s' \"$SERVICE_TOKEN\"")["output"] == "token-for-profile-a"
            finally:
                ss.reset_secret_scope(token_a)

            token_b = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-profile-b"})
            result = env.execute("printf '%s' \"$SERVICE_TOKEN\"")
            ss.reset_secret_scope(token_b)
            token_b = None

            token_c = ss.set_secret_scope({})
            missing = env.execute("printf '%s' \"${SERVICE_TOKEN-unset}\"")
        finally:
            if token_b is not None:
                ss.reset_secret_scope(token_b)
            if token_c is not None:
                ss.reset_secret_scope(token_c)
            ss.set_multiplex_active(False)
            if env is not None:
                env.cleanup()

        assert result["output"] == "token-for-profile-b"
        assert missing["output"] == "unset"

    def test_blocklisted_var_blocked_by_default(self):
        from tools.environments.local import _sanitize_subprocess_env
        from tools.environments.local_env_policy import _HERMES_PROVIDER_ENV_BLOCKLIST

        # Pick a var we know is in the blocklist
        blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
        env = {blocked_var: "secret_value", "PATH": "/usr/bin"}
        result = _sanitize_subprocess_env(env)
        assert blocked_var not in result
        assert "PATH" in result

    def test_passthrough_cannot_override_provider_blocklist(self):
        """GHSA-rhgp-j443-p4rf: register_env_passthrough must NOT accept
        Hermes provider credentials — that was the bypass where a skill
        could declare ANTHROPIC_TOKEN / OPENAI_API_KEY as passthrough and
        defeat the execute_code sandbox scrubbing."""
        from tools.environments.local import _sanitize_subprocess_env
        from tools.environments.local_env_policy import _HERMES_PROVIDER_ENV_BLOCKLIST

        blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
        # Attempt to register — must be silently refused (logged warning).
        register_env_passthrough([blocked_var])

        # is_env_passthrough must NOT report it as allowed
        assert not is_env_passthrough(blocked_var)

        # Sanitizer still strips the var from subprocess env
        env = {blocked_var: "secret_value", "PATH": "/usr/bin"}
        result = _sanitize_subprocess_env(env)
        assert blocked_var not in result
        assert "PATH" in result

    def test_passthrough_case_variant_of_blocklist_rejected(self):
        """A case-variant registration (``openai_api_key``) must be refused just
        like the canonical name: the remote-exec env builder resolves each
        registered name via ``os.getenv``, which is case-insensitive on Windows,
        so a variant would tunnel the real ``OPENAI_API_KEY`` value into
        SSH/Docker children: the same GHSA-rhgp-j443-p4rf primitive."""
        for var in ("openai_api_key", "OpenAi_Api_Key", "anthropic_api_key",
                    "Aws_Bearer_Token_Bedrock"):
            register_env_passthrough([var])
            assert not is_env_passthrough(var), (
                f"{var} should be refused passthrough registration")

    def test_passthrough_case_variant_via_config_rejected(self, tmp_path, monkeypatch):
        """The config-based allowlist (terminal.env_passthrough) must refuse
        case variants of provider credentials on the same filter."""
        config = {"terminal": {"env_passthrough": ["openai_api_key", "MY_OWN_KEY"]}}
        config_path = tmp_path / "config.yaml"
        config_path.write_text(yaml.safe_dump(config), encoding="utf-8")
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        _ep_mod._config_passthrough.clear()

        assert not is_env_passthrough("openai_api_key")
        assert is_env_passthrough("MY_OWN_KEY")

    def test_passthrough_case_variant_never_reaches_remote_exec_env(self, monkeypatch):
        """Even if a case-variant name were present in the registered set, the
        remote env builder must not resolve it: on Windows ``os.getenv`` is
        case-insensitive, so ``openai_api_key`` would carry the real
        ``OPENAI_API_KEY`` into SSH/Docker exec envs."""
        from tools.environments import remote_common

        register_env_passthrough(["openai_api_key"])
        exec_env, _unset = remote_common.resolve_passthrough_env(set())
        assert not any(k.lower() == "openai_api_key" for k in exec_env)

        # Defense in depth, load-bearing on POSIX too: force the variant into
        # the registered set and set a real env entry under that spelling, so
        # os.getenv() resolves it and only the folded blocklist drops it.
        monkeypatch.setenv("openai_api_key", "sk-variant-value")
        monkeypatch.setenv("MY_OWN_KEY", "own-value")
        monkeypatch.setattr(
            "tools.env_passthrough.get_all_passthrough",
            lambda: {"openai_api_key", "MY_OWN_KEY"})
        exec_env, _unset = remote_common.resolve_passthrough_env(set())
        assert "openai_api_key" not in exec_env
        assert exec_env.get("MY_OWN_KEY") == "own-value"

    def test_passthrough_case_variant_never_reaches_execute_code_env(self):
        """The GHSA-rhgp-j443-p4rf path end to end: the variant registration
        is refused, so is_env_passthrough cannot carry the name past the
        execute_code scrub."""
        from tools.code_execution_env import _scrub_child_env

        register_env_passthrough(["openai_api_key"])
        child_env = _scrub_child_env(
            {"openai_api_key": "sk-real", "PATH": "/usr/bin"},
            is_passthrough=is_env_passthrough, is_windows=False)
        assert "openai_api_key" not in child_env
        assert child_env["PATH"] == "/usr/bin"

    def test_passthrough_cannot_override_internal_dynamic_secret(self):
        """A skill must NOT be able to register dynamically-named Hermes
        secrets (AUXILIARY_*_API_KEY / _BASE_URL, GATEWAY_RELAY_* auth) as
        passthrough — they aren't in the static blocklist, so this is the
        defense-in-depth layer that keeps env_passthrough consistent with the
        unconditional strip in the sanitizers."""
        from tools.environments.local import _sanitize_subprocess_env

        for var in (
            "AUXILIARY_VISION_API_KEY",
            "AUXILIARY_VISION_BASE_URL",
            "GATEWAY_RELAY_SECRET",
            "GATEWAY_RELAY_DELIVERY_KEY",
        ):
            register_env_passthrough([var])
            assert not is_env_passthrough(var), (
                f"{var} should be refused passthrough registration"
            )
            result = _sanitize_subprocess_env({var: "secret", "PATH": "/usr/bin"})
            assert var not in result
            assert "PATH" in result

    def test_passthrough_cannot_register_buzz_vars(self, monkeypatch):
        """GHSA-rhgp-j443-p4rf seal stays intact for the BUZZ_* first-party
        platform credentials: even though they pass through to terminal
        children in a Buzz agent context (issue #78026), env_passthrough
        registration must still refuse them — the carve-out opens NO
        registration path, so a skill cannot expand BUZZ_* exposure to
        execute_code."""
        from tools.environments.local import _sanitize_subprocess_env

        monkeypatch.setenv("BUZZ_MANAGED_AGENT", "1")
        for var in (
            "BUZZ_PRIVATE_KEY",
            "BUZZ_AUTH_TAG",
            "BUZZ_RELAY_URL",
        ):
            register_env_passthrough([var])
            assert not is_env_passthrough(var), (
                f"{var} should be refused passthrough registration"
            )
            # Terminal sanitizer still passes BUZZ_* through to terminal
            # children by the first-party carve-out...
            result = _sanitize_subprocess_env({var: "value", "PATH": "/usr/bin"})
            assert result.get(var) == "value"
            # ...but the execute_code child never sees them.
            from tools.code_execution_env import _scrub_child_env

            child_env = _scrub_child_env({var: "value", "PATH": "/usr/bin"})
            assert var not in child_env

    def test_passthrough_allows_auxiliary_non_secret_routing(self):
        """AUXILIARY_*_PROVIDER / _MODEL and GATEWAY_RELAY routing hints are not
        secrets, so a skill may still register them (they're not protected)."""
        register_env_passthrough([
            "AUXILIARY_VISION_PROVIDER",
            "AUXILIARY_VISION_MODEL",
            "GATEWAY_RELAY_URL",
        ])
        assert is_env_passthrough("AUXILIARY_VISION_PROVIDER")
        assert is_env_passthrough("AUXILIARY_VISION_MODEL")
        assert is_env_passthrough("GATEWAY_RELAY_URL")

    def test_make_run_env_blocklist_override_rejected(self):
        """_make_run_env must NOT expose a blocklisted var to subprocess env
        even after a skill attempts to register it via passthrough."""
        from tools.environments.local import _make_run_env
        from tools.environments.local_env_policy import _HERMES_PROVIDER_ENV_BLOCKLIST

        blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
        os.environ[blocked_var] = "secret_value"
        try:
            # Without passthrough — blocked
            result_before = _make_run_env({})
            assert blocked_var not in result_before

            # Skill tries to register it — must be refused, so still blocked
            register_env_passthrough([blocked_var])
            result_after = _make_run_env({})
            assert blocked_var not in result_after
        finally:
            os.environ.pop(blocked_var, None)

    def test_non_hermes_api_key_still_registerable(self):
        """Third-party API keys (TENOR_API_KEY, NOTION_TOKEN, etc.) are NOT
        Hermes provider credentials and must still pass through — skills
        that legitimately wrap third-party APIs must keep working."""
        # TENOR_API_KEY is a real example — used by the gif-search skill
        register_env_passthrough(["TENOR_API_KEY"])
        assert is_env_passthrough("TENOR_API_KEY")

        # Arbitrary skill-specific var
        register_env_passthrough(["MY_SKILL_CUSTOM_CONFIG"])
        assert is_env_passthrough("MY_SKILL_CUSTOM_CONFIG")

    def test_provider_blocklist_import_failure_fails_closed(self, monkeypatch):
        """If the dynamic provider blocklist can't be imported, provider
        credentials must be treated as protected and refused passthrough —
        otherwise a skill could tunnel a Hermes credential into the
        execute_code child (regression for #37950 / GHSA-rhgp-j443-p4rf).

        Verifies the full path: _is_hermes_provider_credential returns True,
        register_env_passthrough refuses the var, and _scrub_child_env keeps
        it out of the child env. A non-Hermes key is also rejected here (the
        fallback is conservative: when we can't tell, we fail closed), which
        is the safe direction.
        """
        import builtins

        from tools.code_execution_env import _scrub_child_env

        real_import = builtins.__import__

        def fail_local_import(name, *args, **kwargs):
            if name == "tools.environments.local":
                raise ImportError("synthetic blocklist import failure")
            return real_import(name, *args, **kwargs)

        monkeypatch.setattr(builtins, "__import__", fail_local_import)

        # Every name is now treated as a protected provider credential.
        assert _ep_mod._is_hermes_provider_credential("OPENAI_API_KEY")
        assert _ep_mod._is_hermes_provider_credential("ANTHROPIC_API_KEY")
        assert _ep_mod._is_hermes_provider_credential("GH_TOKEN")

        # Registration is refused while the blocklist is unavailable.
        register_env_passthrough(["OPENAI_API_KEY", "ANTHROPIC_API_KEY"])
        assert not is_env_passthrough("OPENAI_API_KEY")
        assert not is_env_passthrough("ANTHROPIC_API_KEY")

        # And the credential never reaches the execute_code child.
        child_env = _scrub_child_env(
            {
                "OPENAI_API_KEY": "synthetic-secret",
                "ANTHROPIC_API_KEY": "synthetic-secret",
                "PATH": "/usr/bin",
            },
            is_passthrough=is_env_passthrough,
            is_windows=False,
        )
        assert "OPENAI_API_KEY" not in child_env
        assert "ANTHROPIC_API_KEY" not in child_env
        assert child_env["PATH"] == "/usr/bin"