fix(security): close kanban worker auto-approve gap for dangerous commands

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

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

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

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
 

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

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