fix(apodex): prevent saved Bash allow from bypassing typed confirm via substitution

apodex/agent_tools.py

@@ -404,7 +404,9 @@ def assess_with_rules(
     3. 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
-       ``RISK_CONFIRM`` to ``RISK_SAFE``.
+       ``RISK_CONFIRM`` to ``RISK_SAFE`` — unless the call carries a ``danger``
+       label (dep-install, force-push, delete, ...). A dangerous call never
+       downgrades: the typed-confirmation gate must still fire.
     """
     base = assess_tool_risk(name, args, cwd)
     if rules is not None and rules.denies(name, args):
@@ -413,6 +415,11 @@ def assess_with_rules(
         return base
     if auto_for_me:
         return ToolRisk(RISK_SAFE, "auto for me (docker/trusted env)", base.target)
+    if base.level == RISK_CONFIRM and base.danger:
+        # Saved allows only downgrade *plain* confirms. ``observers`` skips
+        # ``confirm()`` entirely when level is SAFE, so preserving ``danger``
+        # on a SAFE result would still bypass the typed-yes gate.
+        return base
     if base.level == RISK_CONFIRM and rules is not None and rules.allows(name, args):
         return ToolRisk(RISK_SAFE, "allowed by a saved rule", base.target)
     return base

apodex/permissions.py

@@ -12,7 +12,10 @@
 Safety contract: this store only ever *downgrades a plain confirm to safe*, or
 *forces a deny*. It is consulted in :func:`agent_tools.assess_tool_risk` AFTER
 danger detection and the hard denylist — so a saved ``Bash(git)`` allow can
-never green-light a dangerous ``git push --force``.
+never green-light a dangerous ``git push --force``. Unquoted ``$(...)`` and
+backtick substitutions must be separately authorized against the same saved
+prefixes, and a command carrying a ``danger`` label never downgrades (the
+typed-confirmation gate still fires).
 """
 
 from __future__ import annotations
@@ -36,6 +39,84 @@
 })
 
 
+def _nested_shell_snippets(cmd: str) -> list[str]:
+    """Shell-code strings nested in unquoted ``$(...)``/backticks.
+
+    Reuses :func:`plugins.tools._bash_policy._extract_nested_shell` (stdlib-only,
+    no import cycle). Single-quoted spans are skipped — the shell does not expand
+    them, so ``echo '$(rm -rf /)'`` is a harmless literal. Falls back to a small
+    self-contained scanner when the import fails so matching never throws and
+    never silently allows.
+    """
+    try:
+        from plugins.tools._bash_policy import (  # type: ignore
+            _extract_nested_shell as _extract,
+        )
+
+        return list(_extract(cmd or ""))
+    except Exception:
+        pass
+    out: list[str] = []
+    s = cmd or ""
+    n = len(s)
+    i = 0
+    sq = False
+    while i < n:
+        c = s[i]
+        if sq:
+            if c == "'":
+                sq = False
+            i += 1
+            continue
+        if c == "'":
+            sq = True
+            i += 1
+            continue
+        if c == "$" and i + 1 < n and s[i + 1] == "(":
+            depth, j = 1, i + 2
+            start = j
+            while j < n and depth:
+                if s[j] == "(":
+                    depth += 1
+                elif s[j] == ")":
+                    depth -= 1
+                j += 1
+            if depth == 0:
+                out.append(s[start : j - 1])
+            i = j
+            continue
+        if c == "`":
+            j = i + 1
+            while j < n and s[j] != "`":
+                j += 1
+            out.append(s[i + 1 : j])
+            i = j + 1
+            continue
+        i += 1
+    return out
+
+
+def _nested_segments_authorized(nested: str, prefixes: set[str]) -> bool:
+    """True when every ``&&``/``|``/``;`` piece of a nested snippet matches.
+
+    Each piece must itself satisfy the same ``seg == p or seg.startswith(p)``
+    prefix rule, transitively (a nested snippet containing further substitution
+    must have that inner payload authorized too). Fail-closed: empty or
+    unmatched pieces return False.
+    """
+    segs = [p.strip() for p in _SEGMENT_SPLIT.split(nested or "") if p.strip()]
+    if not segs:
+        return False
+    for seg in segs:
+        if not any(seg == p or seg.startswith(p + " ") for p in prefixes):
+            return False
+        # Transitive: ``echo $(foo $(bar))`` needs ``bar`` authorized as well.
+        for inner in _nested_shell_snippets(seg):
+            if not _nested_segments_authorized(inner, prefixes):
+                return False
+    return True
+
+
 def _extract_prefix_from_segment(seg: str) -> str:
     try:
         toks = shlex.split(seg)
@@ -124,9 +205,19 @@ def _matches(rules: set[str], name: str, args: dict) -> bool:
                 if _extract_prefix_from_segment(s).split()[0] not in _HELPER_CMDS
             ]
             check_segs = non_helpers if non_helpers else segs
-            return bool(check_segs) and all(
-                any(seg == p or seg.startswith(p + " ") for p in prefixes) for seg in check_segs
-            )
+            if not check_segs:
+                return False
+            for seg in check_segs:
+                if not any(seg == p or seg.startswith(p + " ") for p in prefixes):
+                    return False
+                # Nested-shell guard (issue #39): ``echo $(pip install x)`` is
+                # "just an echo" only on the raw string. Each unquoted nested
+                # payload must independently match a saved prefix, else the
+                # whole command is not authorized.
+                for nested in _nested_shell_snippets(seg):
+                    if not _nested_segments_authorized(nested, prefixes):
+                        return False
+            return True
         return False
 
 

apodex/tests/test_features.py

@@ -1290,6 +1290,44 @@ def test_assess_with_rules_layering(tmp_path):
     assert r2.level == RISK_SAFE
 
 
+def test_saved_allow_does_not_cover_substitution(tmp_path):
+    """Issue #39: ``Bash(echo)`` must not authorize ``echo $(pip install x)``."""
+    from apodex.agent_tools import RISK_CONFIRM, assess_with_rules
+    from apodex.permissions import PermissionStore
+
+    cwd = str(tmp_path)
+    rules = PermissionStore(allow={"Bash(echo)"})
+    assert not rules.allows("bash", {"command": "echo $(pip install evil-pkg)"})
+    assert not rules.allows("bash", {"command": "echo `pip install evil-pkg`"})
+    r = assess_with_rules(
+        "bash", {"command": "echo $(pip install evil-pkg)"}, cwd, rules
+    )
+    assert r.level == RISK_CONFIRM
+    assert r.danger == "installs dependencies"
+
+
+def test_saved_allow_does_not_cover_force_push(tmp_path):
+    """Issue #39: ``Bash(git push)`` must not downgrade a force-push confirm."""
+    from apodex.agent_tools import RISK_CONFIRM, assess_with_rules
+    from apodex.permissions import PermissionStore
+
+    cwd = str(tmp_path)
+    rules = PermissionStore(allow={"Bash(git push)"})
+    r = assess_with_rules(
+        "bash", {"command": "git push --force origin main"}, cwd, rules
+    )
+    assert r.level == RISK_CONFIRM
+    assert r.danger == "git force-push"
+
+
+def test_single_quoted_substitution_is_literal(tmp_path):
+    """Single-quoted ``$(...)`` is not expanded by the shell — still allowed."""
+    from apodex.permissions import PermissionStore
+
+    rules = PermissionStore(allow={"Bash(echo)"})
+    assert rules.allows("bash", {"command": "echo '$(pip install x)'"})
+
+
 def test_user_settings_save_and_load(tmp_path):
     from apodex.config import UserSettings
     p = str(tmp_path / "settings.json")