fix(bash-policy): bind group denials in every mode and screen only executed text

CHANGELOG.md

@@ -62,6 +62,20 @@ Initial open-source release of FrontierAgent.
 
 ### Fixed
 
+- Bash policy: privilege escalation (`sudo`/`su`/…), remote/exfil clients
+  (`ssh`/`nc`/`rsync`/…) and signal senders (`kill`/`pkill`/`killall`) are now
+  refused in every allowlist mode, including the default `off`. The local CLI
+  sends them to the human as a typed confirmation that auto-approve, `auto_for_me`
+  and saved rules cannot answer.
+- Bash policy: command substitutions are located by one scanner for both
+  masking and extraction, so quoted parens, apostrophes in double quotes and
+  unterminated `$(` no longer hide a nested command; `$((…))` arithmetic is no
+  longer assessed as a command.
+- Bash policy: the host-shutdown/`mkfs`/fork-bomb word screens only see text the
+  shell executes, so quoted arguments and heredoc data mentioning `halt` or
+  `reboot` are no longer refused, while `bash -c`, shell heredocs, pipes into a
+  shell, evaluators (`watch`/`tmux`/…) and `systemctl` shutdown units still are.
+  `DROP TABLE` keeps screening the whole text.
 - Surface finalize-gate bypasses on the final turn: an answer delivered despite
   open task-board items now carries an unfinished-work note and a
   `finalize_gate_bypassed` marker instead of reading as a clean success.

apodex/agent_tools.py

@@ -150,6 +150,11 @@ class ToolRisk:
     # dangerous shell). Unlike kimi's cosmetic red banner, this is wired into
     # the decision: the gate demands a deliberate typed confirmation.
     danger: str = ""
+    # Only a human answering THIS prompt may approve it: auto-approve,
+    # ``auto_for_me`` and saved allow rules never do. Set for bash commands the
+    # shared policy denies outright everywhere else (privilege escalation,
+    # remote/exfil clients, signal senders).
+    must_ask: bool = False
 
 
 # Destructive patterns that warrant a SECOND (typed) confirmation, even though
@@ -387,13 +392,20 @@ def assess_tool_risk(name: str, args: dict, cwd: str) -> ToolRisk:
         cmd = str(args.get("command", "")).strip()
         if _assess_bash_command is not None:
             try:
-                a = _assess_bash_command(cmd)
+                # ``interactive``: the shared policy's always-denied groups
+                # (sudo / ssh / kill …) come back as a group-tagged confirm, so
+                # the person at this gate decides instead of a blanket deny.
+                a = _assess_bash_command(cmd, interactive=True)
             except Exception:
                 # Fail closed: refuse rather than silently downgrading a
                 # possibly destructive command to a confirmable one.
                 return ToolRisk(RISK_DENY, "could not assess bash command safety", cmd)
             if a.level == "deny":
                 return ToolRisk(RISK_DENY, a.reason, cmd)
+            if getattr(a, "group", ""):
+                return ToolRisk(
+                    RISK_CONFIRM, a.reason, cmd, danger=a.reason, must_ask=True,
+                )
         # Read-only inspection (ls/find/grep/tree/git status/…) runs without a
         # prompt so the agent isn't blocked on every harmless command. Anything
         # that could mutate state still requires confirmation.
@@ -416,15 +428,17 @@ def assess_with_rules(
     1. A saved ``deny`` forces a block.
     2. Hard ``RISK_DENY`` (writes outside working directory, dangerous system blocks)
        is NEVER bypassed.
-    3. If ``auto_for_me`` is enabled (Docker / trusted env mode), any non-denied call
+    3. A ``must_ask`` call (privilege escalation, remote/exfil clients, signal
+       senders) always goes to the human; nothing below may downgrade it.
+    4. If ``auto_for_me`` is enabled (Docker / trusted env mode), any non-denied call
        is treated as safe.
-    4. If the user saved an explicit ``allow`` rule for this command/tool, downgrade
+    5. If the user saved an explicit ``allow`` rule for this command/tool, downgrade
        ``RISK_CONFIRM`` to ``RISK_SAFE``.
     """
     base = assess_tool_risk(name, args, cwd)
     if rules is not None and rules.denies(name, args):
         return ToolRisk(RISK_DENY, "denied by a saved rule", base.target)
-    if base.level == RISK_DENY:
+    if base.level == RISK_DENY or base.must_ask:
         return base
     if auto_for_me:
         return ToolRisk(RISK_SAFE, "auto for me (docker/trusted env)", base.target)

apodex/observers.py

@@ -494,6 +494,16 @@ async def on_tool_call(
             return self._skip_tool(
                 f"[blocked by safety policy: {risk.reason}]",
             )
+        if risk.must_ask and getattr(self.approver, "auto_approve", False):
+            # Auto-approve covers routine calls, not the ones the shared bash
+            # policy refuses everywhere else. Blocking (rather than silently
+            # prompting) keeps an unattended ``-y`` run from hanging.
+            self.r.note(f"✗ blocked: {risk.reason} (needs explicit approval)")
+            return self._skip_tool(
+                f"[blocked: {risk.reason} This needs explicit per-call approval, "
+                "which auto-approve does not give. Ask the user to run it "
+                "themselves or to turn auto-approve off.]",
+            )
         if risk.level != RISK_SAFE:  # confirm: ask the human
             decision = await self.approver.confirm(
                 name, risk.target, risk.reason, dangerous=risk.danger,

apodex/tests/test_features.py

@@ -1592,6 +1592,53 @@ def test_download_file_target_is_the_resolved_destination(monkeypatch, tmp_path)
     assert "renamed" in named                   # collisions rename it
 
 
+# ── shared bash policy: always-denied groups go to the human, never auto ─────
+
+
+@pytest.mark.parametrize("cmd", ["sudo systemctl restart x", "ssh host uptime", "pkill -f node"])
+def test_group_denied_bash_is_a_must_ask_confirm(tmp_path, cmd):
+    from apodex.agent_tools import RISK_CONFIRM, assess_tool_risk, assess_with_rules
+    from apodex.permissions import PermissionStore
+    cwd = str(tmp_path)
+    risk = assess_tool_risk("bash", {"command": cmd}, cwd)
+    assert risk.level == RISK_CONFIRM and risk.must_ask and risk.danger
+    # Neither auto_for_me nor a saved allow rule may answer it.
+    prefix = cmd.split()[0]
+    for kwargs in ({"auto_for_me": True}, {"rules": PermissionStore(allow={f"Bash({prefix})"})}):
+        assert assess_with_rules("bash", {"command": cmd}, cwd, **kwargs).level == RISK_CONFIRM
+
+
+def test_group_denied_bash_asks_the_human_with_typed_confirmation(tmp_path):
+    from apodex.observers import Decision, TerminalObserver
+
+    seen = {}
+
+    class _Human:
+        auto_approve = False
+        async def confirm(self, name, target, reason, **kw):
+            seen.update(kw)
+            return Decision(True)
+
+    obs = TerminalObserver(Renderer(theme="mono"), _Human(), str(tmp_path))
+    iv = asyncio.run(obs.on_tool_call(_turn_ctx(), {"name": "bash", "args": {"command": "sudo id"}}))
+    assert iv is None  # approved → runs
+    assert "Privilege escalation" in seen["dangerous"]
+
+
+def test_auto_approve_does_not_cover_group_denied_bash(tmp_path):
+    from apodex.observers import Approver, TerminalObserver
+
+    obs = TerminalObserver(Renderer(theme="mono"), Approver(auto_approve=True), str(tmp_path))
+    iv = asyncio.run(obs.on_tool_call(_turn_ctx(), {"name": "bash", "args": {"command": "sudo id"}}))
+    assert iv is not None and iv.skip_with_result
+    assert "auto-approve" in iv.skip_with_result
+
+
+def test_hard_denylist_still_blocks_under_the_human_gate(tmp_path):
+    from apodex.agent_tools import RISK_DENY, assess_tool_risk
+    assert assess_tool_risk("bash", {"command": "sudo rm -rf /"}, str(tmp_path)).level == RISK_DENY
+
+
 def test_native_workflow_uses_authoritative_loop_telemetry(
     tmp_path, monkeypatch, capsys,
 ):