diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 5b2ad0fd9270..eb0bba730df7 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2383,10 +2383,16 @@ def _ensure_hermes_home_managed(home: Path): # cron_mode — what to do when a cron job hits a dangerous command: # deny — block the command and let the agent find another way (default, safe) # approve — auto-approve all dangerous commands in cron jobs + # + # kanban_mode — same choice, for kanban-dispatched worker subprocesses + # (also unattended — no user present to approve): + # deny — block the command and let the worker find another way (default, safe) + # approve — auto-approve all dangerous commands in kanban workers "approvals": { "mode": "manual", "timeout": 60, "cron_mode": "deny", + "kanban_mode": "deny", # When true, /reload-mcp asks the user to confirm before rebuilding # the MCP tool set for the active session. Reloading invalidates # the provider prompt cache (tool schemas are baked into the system diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 6150b141537b..38b85f888551 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -7709,6 +7709,12 @@ def _default_spawn( if task.tenant: env["HERMES_TENANT"] = task.tenant env["HERMES_KANBAN_TASK"] = task.id + # Kanban workers run unattended (no user present to approve a dangerous + # command) -- flag the context so tools/approval.py applies the same + # deny-by-default policy cron jobs already get via HERMES_CRON_SESSION, + # instead of silently falling through to the bare non-interactive + # auto-approve branch. + env["HERMES_KANBAN_SESSION"] = "1" env["HERMES_KANBAN_WORKSPACE"] = workspace # Pin TERMINAL_CWD to the task's workspace so the worker's file tools and # context-file loader anchor on the workspace, not whatever cwd the diff --git a/tests/conftest.py b/tests/conftest.py index 5606300e5dc1..cf39fe5017a5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -183,6 +183,7 @@ def _looks_like_credential(name: str) -> bool: "HERMES_SESSION_KEY", "HERMES_GATEWAY_SESSION", "HERMES_CRON_SESSION", + "HERMES_KANBAN_SESSION", "_HERMES_GATEWAY", "HERMES_PLATFORM", "HERMES_MODEL", diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index 876a3d1489e4..c830f60b8d71 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -1773,6 +1773,7 @@ def setup_method(self): self._saved_env = { k: os.environ.get(k) for k in ("HERMES_GATEWAY_SESSION", "HERMES_CRON_SESSION", + "HERMES_KANBAN_SESSION", "HERMES_YOLO_MODE", "HERMES_SESSION_KEY", "HERMES_INTERACTIVE") } @@ -1782,6 +1783,8 @@ def setup_method(self): # _is_gateway_approval_context(); a leaked value from a parent cron # process would force the cron path and break these gateway tests. os.environ.pop("HERMES_CRON_SESSION", None) + # Same reasoning for a leaked HERMES_KANBAN_SESSION. + os.environ.pop("HERMES_KANBAN_SESSION", None) os.environ["HERMES_GATEWAY_SESSION"] = "1" os.environ["HERMES_SESSION_KEY"] = self.SESSION_KEY diff --git a/tests/tools/test_execute_code_approval_cluster.py b/tests/tools/test_execute_code_approval_cluster.py index c5d7f3fb78c9..6d4b86117409 100644 --- a/tests/tools/test_execute_code_approval_cluster.py +++ b/tests/tools/test_execute_code_approval_cluster.py @@ -167,6 +167,30 @@ def test_guard_cron_deny_blocks(monkeypatch): assert res["outcome"] == "blocked" +def test_guard_kanban_deny_blocks(monkeypatch): + """Kanban worker subprocesses are the same unattended-execution context + cron jobs are -- same #30882 fail-closed contract, via HERMES_KANBAN_SESSION + + approvals.kanban_mode instead of the cron equivalents.""" + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(A, "_get_kanban_approval_mode", lambda: "deny") + res = A.check_execute_code_guard("import os", "local") + assert res["approved"] is False + assert res["outcome"] == "blocked" + + +def test_guard_kanban_approve_allows(monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(A, "_get_kanban_approval_mode", lambda: "approve") + res = A.check_execute_code_guard("import os", "local") + assert res["approved"] is True + + def test_guard_gateway_user_approves_is_one_shot(gw_session): _register_resolver(gw_session, "once") res = A.check_execute_code_guard("import os; print(1)", "local") diff --git a/tests/tools/test_kanban_approval_mode.py b/tests/tools/test_kanban_approval_mode.py new file mode 100644 index 000000000000..37b0b1f02218 --- /dev/null +++ b/tests/tools/test_kanban_approval_mode.py @@ -0,0 +1,271 @@ +"""Tests for approvals.kanban_mode — configurable approval behavior for +kanban-dispatched worker subprocesses. + +Mirrors test_cron_approval_mode.py: kanban workers are the same kind of +unattended-execution context cron jobs are (no user present to approve), and +previously fell through to the bare non-interactive auto-approve branch +because no approval-context flag was ever set in their environment. +""" + +import pytest + +import tools.approval as approval_module +from tools.approval import ( + _get_kanban_approval_mode, + check_all_command_guards, + check_dangerous_command, + detect_dangerous_command, +) + + +@pytest.fixture(autouse=True) +def _clear_approval_state(): + approval_module._permanent_approved.clear() + approval_module.clear_session("default") + approval_module.clear_session("test-session") + yield + approval_module._permanent_approved.clear() + approval_module.clear_session("default") + approval_module.clear_session("test-session") + + +# --------------------------------------------------------------------------- +# _get_kanban_approval_mode() config parsing +# --------------------------------------------------------------------------- + +class TestKanbanApprovalModeParsing: + def test_default_is_deny(self): + """When no config is set, kanban_mode defaults to 'deny'.""" + from unittest.mock import patch as mock_patch + with mock_patch("hermes_cli.config.load_config", return_value={"approvals": {}}): + assert _get_kanban_approval_mode() == "deny" + + def test_explicit_deny(self): + from unittest.mock import patch as mock_patch + with mock_patch("hermes_cli.config.load_config", return_value={"approvals": {"kanban_mode": "deny"}}): + assert _get_kanban_approval_mode() == "deny" + + def test_explicit_approve(self): + from unittest.mock import patch as mock_patch + with mock_patch("hermes_cli.config.load_config", return_value={"approvals": {"kanban_mode": "approve"}}): + assert _get_kanban_approval_mode() == "approve" + + def test_off_maps_to_approve(self): + """'off' is an alias for 'approve' (matches --yolo semantics).""" + from unittest.mock import patch as mock_patch + with mock_patch("hermes_cli.config.load_config", return_value={"approvals": {"kanban_mode": "off"}}): + assert _get_kanban_approval_mode() == "approve" + + def test_unknown_value_defaults_to_deny(self): + from unittest.mock import patch as mock_patch + with mock_patch("hermes_cli.config.load_config", return_value={"approvals": {"kanban_mode": "maybe"}}): + assert _get_kanban_approval_mode() == "deny" + + def test_config_load_failure_defaults_to_deny(self): + """If config loading fails entirely, default to deny (safe).""" + from unittest.mock import patch as mock_patch + with mock_patch("hermes_cli.config.load_config", side_effect=RuntimeError("config broken")): + assert _get_kanban_approval_mode() == "deny" + + def test_yaml_boolean_false_maps_to_deny(self): + """YAML 1.1 parses bare 'off' as False. Ensure it maps to deny.""" + from unittest.mock import patch as mock_patch + with mock_patch("hermes_cli.config.load_config", return_value={"approvals": {"kanban_mode": False}}): + assert _get_kanban_approval_mode() == "deny" + + +# --------------------------------------------------------------------------- +# check_dangerous_command() with kanban worker session +# --------------------------------------------------------------------------- + +class TestKanbanDenyMode: + """When HERMES_KANBAN_SESSION is set and kanban_mode=deny, dangerous commands are blocked.""" + + def test_dangerous_command_blocked_in_kanban_deny_mode(self, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_kanban_approval_mode", return_value="deny"): + result = check_dangerous_command("rm -rf /tmp/stuff", "local") + assert not result["approved"] + assert "BLOCKED" in result["message"] + assert "kanban_mode" in result["message"] + + def test_safe_command_allowed_in_kanban_deny_mode(self, monkeypatch): + """Non-dangerous commands still work even with kanban_mode=deny.""" + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_kanban_approval_mode", return_value="deny"): + result = check_dangerous_command("ls -la", "local") + assert result["approved"] + + def test_world_writable_chmod_blocked(self, monkeypatch): + """The exact pattern from the live repro: chmod 777 on a kanban worker.""" + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_kanban_approval_mode", return_value="deny"): + result = check_dangerous_command("chmod 777 /tmp/some-file.txt", "local") + assert not result["approved"] + assert "BLOCKED" in result["message"] + + def test_block_message_includes_description(self, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_kanban_approval_mode", return_value="deny"): + result = check_dangerous_command("rm -rf /tmp/stuff", "local") + assert not result["approved"] + assert "dangerous" in result["message"].lower() or "delete" in result["message"].lower() + + +class TestKanbanApproveMode: + """When HERMES_KANBAN_SESSION is set and kanban_mode=approve, dangerous commands pass through.""" + + def test_dangerous_command_allowed_in_kanban_approve_mode(self, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_kanban_approval_mode", return_value="approve"): + result = check_dangerous_command("rm -rf /tmp/stuff", "local") + assert result["approved"] + + +# --------------------------------------------------------------------------- +# check_all_command_guards() with kanban worker session +# --------------------------------------------------------------------------- + +class TestKanbanDenyModeAllGuards: + """The combined guard function also respects kanban_mode.""" + + def test_dangerous_command_blocked_in_combined_guard(self, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_kanban_approval_mode", return_value="deny"): + result = check_all_command_guards("rm -rf /tmp/stuff", "local") + assert not result["approved"] + assert "BLOCKED" in result["message"] + + def test_safe_command_allowed_in_combined_guard(self, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_kanban_approval_mode", return_value="deny"): + result = check_all_command_guards("echo hello", "local") + assert result["approved"] + + def test_combined_guard_approve_mode(self, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_kanban_approval_mode", return_value="approve"): + result = check_all_command_guards("rm -rf /tmp/stuff", "local") + assert result["approved"] + + +# --------------------------------------------------------------------------- +# Edge cases: kanban mode interaction with other approval mechanisms +# --------------------------------------------------------------------------- + +class TestKanbanModeInteractions: + """Kanban mode should NOT interfere with other approval bypass mechanisms, + and must not regress the pre-existing cron/CLI/gateway/non-interactive + behavior this gap was originally found alongside. + """ + + def test_container_env_still_auto_approves(self, monkeypatch): + """Docker/sandbox environments bypass approvals regardless of kanban_mode.""" + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_kanban_approval_mode", return_value="deny"): + result = check_dangerous_command("rm -rf /", "docker") + assert result["approved"] + + def test_yolo_overrides_kanban_deny(self, monkeypatch): + """--yolo still bypasses kanban_mode=deny for dangerous (non-hardline) commands.""" + monkeypatch.setenv("HERMES_KANBAN_SESSION", "1") + monkeypatch.setenv("HERMES_YOLO_MODE", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + from unittest.mock import patch as mock_patch + import tools.approval + with ( + mock_patch.object(tools.approval, "_YOLO_MODE_FROZEN", True), + mock_patch("tools.approval._get_kanban_approval_mode", return_value="deny"), + ): + result = check_dangerous_command("rm -rf /tmp/stuff", "local") + assert result["approved"] + + def test_non_kanban_non_interactive_still_auto_approves(self, monkeypatch): + """Non-kanban, non-cron, non-interactive sessions (e.g. scripted usage) + still auto-approve -- this fix narrows the gap to kanban workers only, + it does not change the pre-existing scripted-usage contract. + """ + monkeypatch.delenv("HERMES_KANBAN_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + result = check_dangerous_command("rm -rf /tmp/stuff", "local") + assert result["approved"] + + def test_cron_session_unaffected_by_kanban_changes(self, monkeypatch): + """Sanity check: adding the kanban branch must not disturb the + pre-existing, already-tested cron_mode behavior. + """ + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + monkeypatch.delenv("HERMES_KANBAN_SESSION", raising=False) + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + result = check_dangerous_command("rm -rf /tmp/stuff", "local") + assert not result["approved"] + assert "cron_mode" in result["message"] diff --git a/tools/approval.py b/tools/approval.py index ae8c82e19989..2ae2621cd543 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -1208,6 +1208,19 @@ def _get_cron_approval_mode() -> str: return "deny" +def _get_kanban_approval_mode() -> str: + """Read the kanban-worker approval mode from config. Returns 'deny' or 'approve'.""" + try: + from hermes_cli.config import load_config + config = load_config() + mode = str(cfg_get(config, "approvals", "kanban_mode", default="deny")).lower().strip() + if mode in {"approve", "off", "allow", "yes"}: + return "approve" + return "deny" + except Exception: + return "deny" + + def _strip_shell_comments(command: str) -> str: """Strip shell-style comments from a command before LLM assessment. @@ -1410,6 +1423,20 @@ def check_dangerous_command(command: str, env_type: str, "approvals.cron_mode: approve in config.yaml." ), } + # Kanban worker subprocesses: same unattended-execution problem as + # cron — no user present to approve. Respect kanban_mode config. + if env_var_enabled("HERMES_KANBAN_SESSION"): + if _get_kanban_approval_mode() == "deny": + return { + "approved": False, + "message": ( + f"BLOCKED: Command flagged as dangerous ({description}) " + "but kanban workers run without a user present to approve it. " + "Find an alternative approach that avoids this command. " + "To allow dangerous commands in kanban workers, set " + "approvals.kanban_mode: approve in config.yaml." + ), + } logger.warning( "AUTO-APPROVED dangerous command in non-interactive non-gateway context " "(pattern: %s): %s — set HERMES_INTERACTIVE or HERMES_GATEWAY_SESSION to require approval.", @@ -1676,6 +1703,22 @@ def check_all_command_guards(command: str, env_type: str, "approvals.cron_mode: approve in config.yaml." ), } + # Kanban worker subprocesses: same unattended-execution problem as + # cron — no user present to approve. Respect kanban_mode config. + if env_var_enabled("HERMES_KANBAN_SESSION"): + if _get_kanban_approval_mode() == "deny": + is_dangerous, _pk, description = detect_dangerous_command(command) + if is_dangerous: + return { + "approved": False, + "message": ( + f"BLOCKED: Command flagged as dangerous ({description}) " + "but kanban workers run without a user present to approve it. " + "Find an alternative approach that avoids this command. " + "To allow dangerous commands in kanban workers, set " + "approvals.kanban_mode: approve in config.yaml." + ), + } return {"approved": True, "message": None} # --- Phase 1: Gather findings from both checks --- @@ -2007,6 +2050,26 @@ def check_execute_code_guard(code: str, env_type: str, } return {"approved": True, "message": None} + # Kanban worker subprocesses: same unattended-execution problem as cron. + if env_var_enabled("HERMES_KANBAN_SESSION"): + if _get_kanban_approval_mode() == "deny": + return { + "approved": False, + "message": ( + "BLOCKED: execute_code runs arbitrary local Python " + "(including subprocess calls that bypass shell-string " + "approval checks). Kanban workers run without a user " + "present to approve it. Use normal tools instead, or set " + "approvals.kanban_mode: approve only if this kanban " + "profile is intentionally trusted." + ), + "pattern_key": pattern_key, + "description": description, + "outcome": "blocked", + "user_consent": False, + } + return {"approved": True, "message": None} + # Only gateway/ask contexts get the one-shot whole-script approval. # * CLI interactive: the script's terminal() calls are guarded per-call # (context now propagates into the RPC thread, #33057); a whole-script