hermes-agent: Harden quick commands and import checks

agent/file_safety.py

@@ -27,6 +27,7 @@ def build_write_denied_paths(home: str) -> set[str]:
             os.path.join(home, ".ssh", "id_ed25519"),
             os.path.join(home, ".ssh", "config"),
             str(hermes_home / ".env"),
+            str(hermes_home / "config.yaml"),
             os.path.join(home, ".bashrc"),
             os.path.join(home, ".zshrc"),
             os.path.join(home, ".profile"),

cli.py

@@ -6174,9 +6174,17 @@ def process_command(self, command: str) -> bool:
                 qcmd = quick_commands[base_cmd.lstrip("/")]
                 if qcmd.get("type") == "exec":
                     import subprocess
+                    from tools.approval import detect_dangerous_command
                     exec_cmd = qcmd.get("command", "")
                     if exec_cmd:
                         try:
+                            is_dangerous, _, desc = detect_dangerous_command(exec_cmd)
+                            if is_dangerous:
+                                self._console_print(
+                                    f"[bold red]Quick command blocked: {desc}. "
+                                    "Use the agent or terminal approval flow instead.[/]"
+                                )
+                                return True
                             result = subprocess.run(
                                 exec_cmd, shell=True, capture_output=True,
                                 text=True, timeout=30

gateway/run.py

@@ -3739,6 +3739,14 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                     exec_cmd = qcmd.get("command", "")
                     if exec_cmd:
                         try:
+                            from tools.approval import detect_dangerous_command
+
+                            is_dangerous, _, desc = detect_dangerous_command(exec_cmd)
+                            if is_dangerous:
+                                return (
+                                    f"Quick command '/{command}' blocked: {desc}. "
+                                    "Use the normal agent approval flow instead."
+                                )
                             proc = await asyncio.create_subprocess_shell(
                                 exec_cmd,
                                 stdout=asyncio.subprocess.PIPE,

hermes_cli/main.py

@@ -5003,6 +5003,7 @@ def _update_via_zip(args):
     """
     import tempfile
     import zipfile
+    import stat
     from urllib.request import urlretrieve
 
     branch = "main"
@@ -5018,7 +5019,7 @@ def _update_via_zip(args):
 
         print("→ Extracting...")
         with zipfile.ZipFile(zip_path, "r") as zf:
-            # Validate paths to prevent zip-slip (path traversal)
+            # Validate paths to prevent zip-slip and reject symlink members.
             tmp_dir_real = os.path.realpath(tmp_dir)
             for member in zf.infolist():
                 member_path = os.path.realpath(os.path.join(tmp_dir, member.filename))
@@ -5029,7 +5030,22 @@ def _update_via_zip(args):
                     raise ValueError(
                         f"Zip-slip detected: {member.filename} escapes extraction directory"
                     )
-            zf.extractall(tmp_dir)
+                mode = (member.external_attr >> 16) & 0o170000
+                if stat.S_ISLNK(mode):
+                    raise ValueError(
+                        f"ZIP contains unsupported symlink member: {member.filename}"
+                    )
+
+            for member in zf.infolist():
+                target = os.path.join(tmp_dir, member.filename)
+                if member.is_dir():
+                    os.makedirs(target, exist_ok=True)
+                    continue
+                parent = os.path.dirname(target)
+                if parent:
+                    os.makedirs(parent, exist_ok=True)
+                with zf.open(member, "r") as src, open(target, "wb") as dst:
+                    shutil.copyfileobj(src, dst)
 
         # GitHub ZIPs extract to hermes-agent-<branch>/
         extracted = os.path.join(tmp_dir, f"hermes-agent-{branch}")

hermes_cli/memory_setup.py

@@ -9,6 +9,7 @@
 
 import getpass
 import os
+import shlex
 import sys
 from pathlib import Path
 
@@ -131,15 +132,45 @@ def _install_dependencies(provider_name: str) -> None:
         dep_name = dep.get("name", "")
         check_cmd = dep.get("check", "")
         install_cmd = dep.get("install", "")
-        if check_cmd:
-            try:
-                subprocess.run(
-                    check_cmd, shell=True, capture_output=True, timeout=5
-                )
-            except Exception:
-                if install_cmd:
-                    print(f"\n  ⚠ '{dep_name}' not found. Install with:")
-                    print(f"    {install_cmd}")
+        if check_cmd and not _external_dependency_available(check_cmd):
+            if install_cmd:
+                print(f"\n  ⚠ '{dep_name}' not found. Install with:")
+                print(f"    {install_cmd}")
+
+
+def _external_dependency_available(check_cmd) -> bool:
+    """Return True when an external dependency probe succeeds.
+
+    ``plugin.yaml`` metadata is treated as data, not an arbitrary shell
+    script. Accept either a list/tuple argv or a plain string that can be
+    tokenized with ``shlex.split()``. Shell metacharacters are not
+    interpreted, which prevents plugin metadata from smuggling extra
+    commands via ``shell=True``.
+    """
+    import subprocess
+
+    argv: list[str]
+    if isinstance(check_cmd, (list, tuple)):
+        argv = [str(part).strip() for part in check_cmd if str(part).strip()]
+    elif isinstance(check_cmd, str):
+        raw = check_cmd.strip()
+        if not raw:
+            return True
+        try:
+            argv = shlex.split(raw)
+        except ValueError:
+            return False
+    else:
+        return False
+
+    if not argv:
+        return True
+
+    try:
+        result = subprocess.run(argv, capture_output=True, timeout=5)
+    except (OSError, ValueError, subprocess.TimeoutExpired):
+        return False
+    return result.returncode == 0
 
 
 def _get_available_providers() -> list:

tests/cli/test_quick_commands.py

@@ -127,6 +127,18 @@ def test_timeout_shows_error(self):
         args = cli.console.print.call_args[0][0]
         assert "timed out" in args.lower()
 
+    def test_dangerous_exec_command_is_blocked(self):
+        cli = self._make_cli({"nuke": {"type": "exec", "command": "rm -rf /tmp/demo"}})
+
+        with patch("subprocess.run") as run:
+            result = cli.process_command("/nuke")
+
+        assert result is True
+        run.assert_not_called()
+        printed = self._printed_plain(cli.console.print.call_args[0][0])
+        assert "blocked" in printed.lower()
+        assert "recursive delete" in printed.lower()
+
 
 # ── Gateway tests ──────────────────────────────────────────────────────────
 
@@ -205,3 +217,21 @@ async def test_gateway_config_object_supports_quick_commands(self):
         event = self._make_event("limits")
         result = await runner._handle_message(event)
         assert result == "ok"
+
+    @pytest.mark.asyncio
+    async def test_gateway_dangerous_exec_command_is_blocked(self):
+        from gateway.run import GatewayRunner
+
+        runner = GatewayRunner.__new__(GatewayRunner)
+        runner.config = {"quick_commands": {"nuke": {"type": "exec", "command": "rm -rf /tmp/demo"}}}
+        runner._running_agents = {}
+        runner._pending_messages = {}
+        runner._is_user_authorized = MagicMock(return_value=True)
+
+        event = self._make_event("nuke")
+        with patch("asyncio.create_subprocess_shell") as spawn:
+            result = await runner._handle_message(event)
+
+        spawn.assert_not_called()
+        assert result is not None
+        assert "blocked" in result.lower()

tests/hermes_cli/test_memory_setup.py

@@ -0,0 +1,51 @@
+import subprocess
+
+from hermes_cli.memory_setup import _external_dependency_available
+
+
+def test_external_dependency_available_accepts_argv_list(monkeypatch):
+    captured = {}
+
+    def fake_run(argv, capture_output, timeout):
+        captured["argv"] = argv
+        return type("Result", (), {"returncode": 0})()
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+
+    assert _external_dependency_available(["docker", "--version"]) is True
+    assert captured["argv"] == ["docker", "--version"]
+
+
+def test_external_dependency_available_rejects_nonzero(monkeypatch):
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        lambda argv, capture_output, timeout: type("Result", (), {"returncode": 1})(),
+    )
+
+    assert _external_dependency_available("docker --version") is False
+
+
+def test_external_dependency_available_does_not_invoke_shell_fragments(monkeypatch):
+    captured = {}
+
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        lambda argv, capture_output, timeout: (
+            captured.setdefault("argv", argv),
+            type("Result", (), {"returncode": 0})(),
+        )[1],
+    )
+
+    assert _external_dependency_available("docker --version && touch /tmp/pwned") is True
+    assert captured["argv"] == ["docker", "--version", "&&", "touch", "/tmp/pwned"]
+
+
+def test_external_dependency_available_reports_timeouts(monkeypatch):
+    def raise_timeout(argv, capture_output, timeout):
+        raise subprocess.TimeoutExpired(cmd=argv, timeout=timeout)
+
+    monkeypatch.setattr(subprocess, "run", raise_timeout)
+
+    assert _external_dependency_available("docker --version") is False

tests/test_tui_gateway_server.py

@@ -860,6 +860,26 @@ def test_command_dispatch_exec_nonzero_surfaces_error(monkeypatch):
     assert "failed" in resp["error"]["message"]
 
 
+def test_command_dispatch_exec_blocks_dangerous_quick_command(monkeypatch):
+    monkeypatch.setattr(
+        server,
+        "_load_cfg",
+        lambda: {"quick_commands": {"nuke": {"type": "exec", "command": "rm -rf /tmp/demo"}}},
+    )
+    monkeypatch.setattr(
+        server.subprocess,
+        "run",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not run")),
+    )
+
+    resp = server.handle_request(
+        {"id": "1", "method": "command.dispatch", "params": {"name": "nuke"}}
+    )
+
+    assert "error" in resp
+    assert "blocked" in resp["error"]["message"]
+
+
 def test_plugins_list_surfaces_loader_error(monkeypatch):
     with patch("hermes_cli.plugins.get_plugin_manager", side_effect=Exception("boom")):
         resp = server.handle_request(

tests/tools/test_env_passthrough.py

@@ -105,6 +105,19 @@ def test_union_of_skill_and_config(self, tmp_path, monkeypatch):
         assert "CONFIG_KEY" in all_pt
         assert "SKILL_KEY" in all_pt
 
+    def test_config_passthrough_cannot_allow_provider_credential(self, tmp_path, monkeypatch):
+        from tools.environments.local import _HERMES_PROVIDER_ENV_BLOCKLIST
+
+        blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
+        config = {"terminal": {"env_passthrough": [blocked_var, "CONFIG_KEY"]}}
+        config_path = tmp_path / "config.yaml"
+        config_path.write_text(yaml.dump(config))
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        _ep_mod._config_passthrough = None
+
+        assert not is_env_passthrough(blocked_var)
+        assert is_env_passthrough("CONFIG_KEY")
+
 
 class TestExecuteCodeIntegration:
     """Verify that the passthrough is checked in execute_code's env filtering."""
@@ -195,6 +208,26 @@ def test_passthrough_cannot_override_provider_blocklist(self):
         assert blocked_var not in result
         assert "PATH" in result
 
+    def test_config_passthrough_cannot_override_provider_blocklist(self, tmp_path, monkeypatch):
+        """Config allowlists must not re-expose Hermes-managed provider creds."""
+        from tools.environments.local import (
+            _sanitize_subprocess_env,
+            _HERMES_PROVIDER_ENV_BLOCKLIST,
+        )
+
+        blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
+        config = {"terminal": {"env_passthrough": [blocked_var]}}
+        (tmp_path / "config.yaml").write_text(yaml.dump(config))
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        _ep_mod._config_passthrough = None
+
+        assert not is_env_passthrough(blocked_var)
+
+        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_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."""

tests/tools/test_file_write_safety.py

@@ -24,6 +24,12 @@ def test_ssh_key_is_denied(self):
     def test_etc_shadow_is_denied(self):
         assert _is_write_denied("/etc/shadow") is True
 
+    def test_hermes_config_yaml_is_denied(self, tmp_path: Path, monkeypatch):
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+        assert _is_write_denied(str(hermes_home / "config.yaml")) is True
+
 
 class TestSafeWriteRoot:
     """HERMES_WRITE_SAFE_ROOT should sandbox writes to a specific subtree."""

tools/env_passthrough.py

@@ -100,7 +100,14 @@ def register_env_passthrough(var_names: Iterable[str]) -> None:
 
 
 def _load_config_passthrough() -> frozenset[str]:
-    """Load ``tools.env_passthrough`` from config.yaml (cached)."""
+    """Load ``tools.env_passthrough`` from config.yaml (cached).
+
+    Config-based passthrough is intentionally subject to the same
+    Hermes-provider credential blocklist as skill-declared passthrough.
+    Otherwise ``terminal.env_passthrough`` could re-expose the exact
+    provider secrets that ``execute_code`` and local terminal backends
+    deliberately scrub from child processes.
+    """
     global _config_passthrough
     if _config_passthrough is not None:
         return _config_passthrough
@@ -113,7 +120,15 @@ def _load_config_passthrough() -> frozenset[str]:
         if isinstance(passthrough, list):
             for item in passthrough:
                 if isinstance(item, str) and item.strip():
-                    result.add(item.strip())
+                    name = item.strip()
+                    if _is_hermes_provider_credential(name):
+                        logger.warning(
+                            "env passthrough: refusing config allowlist entry %r "
+                            "because it is a Hermes-managed provider credential.",
+                            name,
+                        )
+                        continue
+                    result.add(name)
     except Exception as e:
         logger.debug("Could not read tools.env_passthrough from config: %s", e)
 
@@ -141,4 +156,3 @@ def clear_env_passthrough() -> None:
     """Reset the skill-scoped allowlist (e.g. on session reset)."""
     _get_allowed().clear()
 
-

tui_gateway/server.py

@@ -3266,8 +3266,18 @@ def _(rid, params: dict) -> dict:
     if name in qcmds:
         qc = qcmds[name]
         if qc.get("type") == "exec":
+            from tools.approval import detect_dangerous_command
+
+            exec_cmd = qc.get("command", "")
+            is_dangerous, _, desc = detect_dangerous_command(exec_cmd)
+            if is_dangerous:
+                return _err(
+                    rid,
+                    4017,
+                    f"quick command blocked: {desc}. use the agent approval flow instead.",
+                )
             r = subprocess.run(
-                qc.get("command", ""),
+                exec_cmd,
                 shell=True,
                 capture_output=True,
                 text=True,