From bffc8d91bb0641709f57a64dfa9582549ef2f89f Mon Sep 17 00:00:00 2001 From: Arijit2916 <178023477+Samurai007AK@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:10:13 +0530 Subject: [PATCH] fix(apodex): prevent saved Bash allow from bypassing typed confirm via substitution Root cause: PermissionStore._matches compared raw segments, ignoring unquoted command substitution, and assess_with_rules dropped the danger label on downgrade. Validation: repro script confirms git push --force and echo pip install payloads stay confirm with danger; 5 permission tests pass; ruff and pyright clean. Closes #39 --- apodex/agent_tools.py | 9 +++- apodex/permissions.py | 99 +++++++++++++++++++++++++++++++++-- apodex/tests/test_features.py | 38 ++++++++++++++ 3 files changed, 141 insertions(+), 5 deletions(-) diff --git a/apodex/agent_tools.py b/apodex/agent_tools.py index a6687a5..a959b3a 100644 --- a/apodex/agent_tools.py +++ b/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 diff --git a/apodex/permissions.py b/apodex/permissions.py index a593ab6..b106ab6 100644 --- a/apodex/permissions.py +++ b/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 diff --git a/apodex/tests/test_features.py b/apodex/tests/test_features.py index 6615908..47b515d 100644 --- a/apodex/tests/test_features.py +++ b/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")