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..7168878 100644 --- a/apodex/permissions.py +++ b/apodex/permissions.py @@ -5,14 +5,20 @@ ``npm test``", "never allow ``git push``" — matched by command prefix, so the gate stays livable without being all-or-nothing. -Rules are strings: ``Bash(npm test)`` / ``Bash(git push)`` for shell (matched by -prefix across every ``&&``/``|``/``;`` segment, fail-safe), or a bare tool name -(``write_file``) for everything else. +Rules are strings: ``Bash(npm test)`` / ``Bash(git push)`` for shell, or a bare +tool name (``write_file``) for everything else. Shell rules match by prefix per +segment, where segments are split on ``&&``, ``||``, ``|``, ``;``, ``&`` and +newlines. An allow must cover every segment and a deny fires on any one +segment, so both fail safe. 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``. Every command the shell +would run from a ``$(...)``, backtick or ``<(...)``/``>(...)`` substitution +needs its own match against the saved allow prefixes. A deny prefix also fires +on a command nested inside one. A command carrying a ``danger`` label never +downgrades, so the typed-confirmation gate still fires. """ from __future__ import annotations @@ -30,12 +36,230 @@ "git", "npm", "pnpm", "yarn", "uv", "pip", "pip3", "cargo", "go", "docker", "poetry", "conda", "make", "apt", "apt-get", "brew", "kubectl", "gh", }) -_SEGMENT_SPLIT = re.compile(r"&&|\|\||\||;") +# Command separators: ``&&`` ``||`` ``|`` ``;``, a newline, and a single ``&`` +# (background). The ``&`` inside a redirection (``2>&1``, ``>&2``, ``&>log``) +# is not a separator, and the lookarounds skip it. +_SEGMENT_SPLIT = re.compile(r"&&|\|\||\||;|\n|(?])&(?![>&])") +# How many levels of nested ``$(...)`` the matcher follows. Real commands use +# one or two. Past this, allow fails closed and deny fires, instead of +# recursing until Python raises RecursionError. +_MAX_NEST_DEPTH = 16 _HELPER_CMDS = frozenset({ "cd", "pwd", "export", "set", "env", "echo", "mkdir", "clear", "true", "source", ".", }) +def _nested_shell_snippets(cmd: str) -> list[str]: + """Shell-code strings nested in ``$(...)``, backticks, ``<(...)`` and + ``>(...)``, which the shell runs as separate commands. + + Reuses :func:`plugins.tools._bash_policy._extract_nested_shell` (stdlib-only, + no import cycle). It reads quotes the way bash does. A single-quoted span is + skipped, so ``echo '$(rm -rf /)'`` is a harmless literal. A ``'`` inside + double quotes is an ordinary character, so ``echo "'$(rm -rf /)'"`` still + yields ``rm -rf /``. If the import fails it uses + :func:`_fallback_nested_shell`, so matching never throws and never silently + allows. ``test_nested_shell_extractors_agree`` checks the two give the same + results. + """ + try: + from plugins.tools._bash_policy import ( # type: ignore + _extract_nested_shell as _extract, + ) + + return list(_extract(cmd or "")) + except Exception: + return _fallback_nested_shell(cmd or "") + + +def _fallback_backtick_body(command: str, i: int) -> tuple[str, int]: + """Read a backtick body and apply bash's first-pass escape removal. + + Inside backticks, backslashes before $, `, a backslash or a newline are removed + before the body is parsed as shell code. Other backslashes are retained. + Return the decoded body and the closing backtick's index (len if absent). + """ + body: list[str] = [] + n = len(command) + while i < n: + c = command[i] + if c == "`": + break + if c == "\\" and i + 1 < n and command[i + 1] in "$`\\\n": + i += 1 + if command[i] != "\n": + body.append(command[i]) + else: + body.append(c) + i += 1 + return "".join(body), i + + +def _fallback_substitution_end(s: str, i: int) -> int: + """Index of the ``)`` closing a ``$(`` whose body starts at ``i`` (``len`` + when unterminated). Quoted or escaped parens don't count, and every nested + ``$(`` starts with its own quote state, as in bash.""" + n = len(s) + quotes: list[str | None] = [None] # quote state of each open $( level + depths = [1] # unquoted "(" nesting inside each open $( level + while i < n: + c = s[i] + quote = quotes[-1] + if quote == "'": + if c == "'": + quotes[-1] = None + elif c == "\\": + i += 1 + elif s.startswith("(", i + 1) and (c == "$" or (c in "<>" and quote is None)): + quotes.append(None) + depths.append(1) + i += 1 + elif c == "`": + i += 1 + while i < n and s[i] != "`": + i += 2 if s[i] == "\\" else 1 + elif c == '"': + quotes[-1] = None if quote else '"' + elif quote is None: + if c == "'": + quotes[-1] = "'" + elif c == "(": + depths[-1] += 1 + elif c == ")": + depths[-1] -= 1 + if depths[-1] == 0: + if len(depths) == 1: + return i + depths.pop() + quotes.pop() + i += 1 + return n + + +def _fallback_dup_redirect_word(s: str, i: int) -> str: + """Target word of a ``>&`` redirect starting at ``i``, quotes and + backslashes removed. bash expands it a second time after quote removal.""" + n = len(s) + while i < n and s[i] in " \t": + i += 1 + start = i + quote: str | None = None + while i < n: + c = s[i] + if quote == "'": + if c == "'": + quote = None + elif c == "\\": + i += 1 + elif s.startswith("$(", i): + i = _fallback_substitution_end(s, i + 2) + elif c == "`": + i += 1 + while i < n and s[i] != "`": + i += 2 if s[i] == "\\" else 1 + elif c == '"': + quote = None if quote else '"' + elif quote is None: + if c == "'": + quote = "'" + elif c.isspace() or c in ";&|<>()": + break + i += 1 + return re.sub(r"[\\'\"]", "", s[start:i]) + + +def _fallback_nested_shell(s: str) -> list[str]: + """Standalone copy of ``_extract_nested_shell`` for when its import fails. + + It tracks single quotes, double quotes and backslash escapes. An + unterminated substitution yields the rest of the string, which fails closed. + The target of ``>&`` is read with its quotes removed, because bash expands + it twice. + """ + out: list[str] = [] + i, n = 0, len(s) + quote: str | None = None + while i < n: + c = s[i] + if quote == "'": + if c == "'": + quote = None + i += 1 + continue + if c == "\\": + i += 2 + continue + if c == "'" and quote is None: + quote = "'" + elif c == '"': + quote = None if quote else '"' + elif s.startswith("(", i + 1) and (c == "$" or (c in "<>" and quote is None)): + end = _fallback_substitution_end(s, i + 2) + out.append(s[i + 2 : end]) + i = end + 1 + continue + elif c == "`": + body, j = _fallback_backtick_body(s, i + 1) + out.append(body) + i = j + 1 + continue + elif c == ">" and quote is None and s.startswith("&", i + 1): + out.extend(_fallback_nested_shell(_fallback_dup_redirect_word(s, i + 2))) + i += 1 + return out + + +def _nested_segments_authorized(nested: str, prefixes: set[str], depth: int = 0) -> bool: + """True when every segment 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, and so does nesting deeper than + ``_MAX_NEST_DEPTH``. + """ + if depth >= _MAX_NEST_DEPTH: + return False + segs = _segments(nested) + if not segs or not all(_seg_matches(seg, prefixes) for seg in segs): + return False + # Transitive: ``echo $(foo $(bar))`` needs ``bar`` authorized as well. The + # inner payloads come from the whole snippet, not from each piece. The + # split ignores quotes, so a quoted ``;`` would cut a string in half and + # could hide ``$(...)`` inside what then looks like a single-quoted span. + return all( + _nested_segments_authorized(inner, prefixes, depth + 1) + for inner in _nested_shell_snippets(nested) + ) + + +def _segments(cmd: str) -> list[str]: + return [s.strip() for s in _SEGMENT_SPLIT.split(cmd or "") if s.strip()] + + +def _seg_matches(seg: str, prefixes: set[str]) -> bool: + return any(seg == p or seg.startswith(p + " ") for p in prefixes) + + +def _denied_anywhere(cmd: str, prefixes: set[str], depth: int = 0) -> bool: + """True when any segment matches a deny prefix, at the top level or nested + in a substitution at any depth. + + This is the opposite of the allow check, which needs every segment to + match. A nested command that matches no rule never cancels a deny on its + parent, so ``Bash(echo)`` still denies ``echo $(touch x)``. Nesting deeper + than ``_MAX_NEST_DEPTH`` counts as denied, because the rest can't be checked. + """ + if depth >= _MAX_NEST_DEPTH: + return True + if any(_seg_matches(seg, prefixes) for seg in _segments(cmd)): + return True + return any( + _denied_anywhere(inner, prefixes, depth + 1) + for inner in _nested_shell_snippets(cmd) + ) + + def _extract_prefix_from_segment(seg: str) -> str: try: toks = shlex.split(seg) @@ -95,10 +319,37 @@ def save(self) -> None: pass def allows(self, name: str, args: dict) -> bool: - return self._matches(self.allow, name, args) + """True only when every segment and every nested payload matches.""" + if _name_matches(self.allow, name): + return True + prefixes = _bash_rule_prefixes(self.allow) if name == "bash" else set() + if not prefixes: + return False + cmd = str(args.get("command", "")) + # Issue #39. On the raw string ``echo $(pip install x)`` looks like a + # plain echo, so every payload the shell would run has to match a saved + # prefix on its own. Payloads come from the whole command before the + # helper filter runs. Otherwise ``echo $(touch x) && python -V`` would + # pass, because the filter drops the ``echo`` segment. + for nested in _nested_shell_snippets(cmd): + if not _nested_segments_authorized(nested, prefixes): + return False + segs = _segments(cmd) + # Filter out helper segments (e.g. 'cd /foo') unless all segments are helpers + non_helpers = [ + s for s in segs + # ``or [""]`` stops an empty-word segment (``"" && ...``) raising IndexError. + if (_extract_prefix_from_segment(s).split() or [""])[0] not in _HELPER_CMDS + ] + check_segs = non_helpers if non_helpers else segs + return bool(check_segs) and all(_seg_matches(s, prefixes) for s in check_segs) def denies(self, name: str, args: dict) -> bool: - return self._matches(self.deny, name, args) + """True when any segment, top-level or nested, matches a deny rule.""" + if _name_matches(self.deny, name): + return True + prefixes = _bash_rule_prefixes(self.deny) if name == "bash" else set() + return bool(prefixes) and _denied_anywhere(str(args.get("command", "")), prefixes) def add_allow(self, name: str, args: dict) -> str: """Persist 'always allow' for this call; returns the rule added.""" @@ -107,27 +358,16 @@ def add_allow(self, name: str, args: dict) -> str: self.save() return rule - @staticmethod - def _matches(rules: set[str], name: str, args: dict) -> bool: - if name in rules: - return True - if name == "bash": - if "bash" in rules or "Bash" in rules or "Bash(*)" in rules: - return True - prefixes = {r[5:-1] for r in rules if r.startswith("Bash(") and r.endswith(")")} - if not prefixes: - return False - segs = [s.strip() for s in _SEGMENT_SPLIT.split(str(args.get("command", ""))) if s.strip()] - # Filter out helper segments (e.g. 'cd /foo') unless all segments are helpers - non_helpers = [ - s for s in segs - 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 - ) - return False + +def _name_matches(rules: set[str], name: str) -> bool: + """A bare tool-name rule, or a whole-tool bash rule (``bash``/``Bash(*)``).""" + return name in rules or ( + name == "bash" and ("Bash" in rules or "Bash(*)" in rules) + ) + + +def _bash_rule_prefixes(rules: set[str]) -> set[str]: + return {r[5:-1] for r in rules if r.startswith("Bash(") and r.endswith(")")} __all__ = ["PermissionStore", "rule_for"] diff --git a/apodex/tests/test_features.py b/apodex/tests/test_features.py index 6615908..bdaaf4c 100644 --- a/apodex/tests/test_features.py +++ b/apodex/tests/test_features.py @@ -1290,6 +1290,181 @@ 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)'"}) + assert rules.allows("bash", {"command": r"echo \$(pip install x)"}) # escaped, so literal + + +def test_deny_rule_still_matches_parent_of_substitution(tmp_path): + """A nested command that matches no rule must not cancel a deny (PR #42 review).""" + from apodex.agent_tools import RISK_DENY, assess_with_rules + from apodex.permissions import PermissionStore + + cwd = str(tmp_path) + rules = PermissionStore(allow={"Bash(*)"}, deny={"Bash(echo)"}) + cmd = {"command": "echo $(touch /tmp/marker)"} + assert rules.denies("bash", cmd) + assert assess_with_rules("bash", cmd, cwd, rules).level == RISK_DENY + # A deny prefix also fires on a command nested inside a substitution. + nested = PermissionStore(allow={"Bash(*)"}, deny={"Bash(touch)"}) + assert assess_with_rules("bash", cmd, cwd, nested).level == RISK_DENY + # It also fires on any one top-level segment, not only when all of them match. + assert PermissionStore(deny={"Bash(git push)"}).denies( + "bash", {"command": "git status && git push origin main"} + ) + + +def test_helper_segment_substitution_needs_authorization(tmp_path): + """The helper filter must not hide a payload inside an echo segment (PR #42 review).""" + from apodex.agent_tools import RISK_CONFIRM, assess_with_rules + from apodex.permissions import PermissionStore + + rules = PermissionStore(allow={"Bash(python)"}) + cmd = {"command": "echo $(touch /tmp/marker) && python -V"} + assert not rules.allows("bash", cmd) + assert assess_with_rules("bash", cmd, str(tmp_path), rules).level == RISK_CONFIRM + assert rules.allows("bash", {"command": "echo hi && python -V"}) # a plain helper is still skipped + assert not rules.allows("bash", {"command": '"" && python -V'}) # empty word returns False, no IndexError + + +def test_double_quoted_substitution_is_not_literal(tmp_path): + """A ``'`` inside ``"..."`` is an ordinary character, so ``$(...)`` still runs (PR #42 review).""" + from apodex.agent_tools import RISK_CONFIRM, assess_with_rules + from apodex.permissions import PermissionStore + from plugins.tools._bash_policy import assess_bash_command + + rules = PermissionStore(allow={"Bash(echo)"}) + for cmd in ( + "echo \"'$(touch /tmp/marker)'\"", + r"echo \' $(touch /tmp/marker) \'", # an escaped ' does not start a quoted span + r'''echo "a\"'$(touch /tmp/marker)'"''', # an escaped " does not end the string + 'echo $(echo ")"; touch /tmp/marker)', # a quoted ")" does not end the substitution + ): + assert not rules.allows("bash", {"command": cmd}), cmd + assert assess_with_rules("bash", {"command": cmd}, str(tmp_path), rules).level == RISK_CONFIRM + # The sandbox bash policy uses the same extractor. + assert assess_bash_command("echo \"'$(foobarcmd)'\"", mode="enforce").level == "deny" + + +def test_nested_shell_extractors_agree(): + """The fallback scanner in permissions.py must match the shared extractor.""" + from apodex.permissions import _fallback_nested_shell + from plugins.tools._bash_policy import _extract_nested_shell + + corpus = { + "echo $(a)": ["a"], + "echo `b`": ["b"], + "echo '$(c)'": [], + "echo \"'$(d)'\"": ["d"], + r"echo \$(e)": [], + r"echo \' $(f) \'": ["f"], + r'''echo "a\"'$(g)'"''': ["g"], + 'echo $(echo ")"; h)': ['echo ")"; h'], + "echo $(i $(j))": ["i $(j)"], + "echo $(unterminated": ["unterminated"], + r"echo \`k\`": [], + """echo $(echo "$(echo ")'")" $(l))""": ["""echo "$(echo ")'")" $(l)"""], + "cat <(m) >(n)": ["m", "n"], + 'echo "<(o)"': [], + "echo x >&'$(p)'": ["p"], + "echo x > '$(q)'": [], + "python x.py 2>&1": [], + } + for cmd, want in corpus.items(): + assert _extract_nested_shell(cmd) == want, cmd + assert _fallback_nested_shell(cmd) == want, cmd + + +def test_nested_quotes_do_not_end_the_outer_substitution(tmp_path): + """Each nested ``$(`` has its own quote state (PR #42 second review).""" + from apodex.agent_tools import RISK_CONFIRM, assess_with_rules + from apodex.permissions import PermissionStore + + rules = PermissionStore(allow={"Bash(echo)"}) + for cmd in ( + """echo $(echo "$(echo ")'")" $(touch /tmp/marker))""", + # A quoted ";" must not split the snippet before its payloads are read. + """echo $(echo "a;echo '" $(touch /tmp/marker) "'")""", + ): + assert not rules.allows("bash", {"command": cmd}), cmd + assert assess_with_rules("bash", {"command": cmd}, str(tmp_path), rules).level == RISK_CONFIRM + assert rules.allows("bash", {"command": """echo $(echo "$(echo ")'")")"""}) + + +def test_separators_and_process_substitution_need_authorization(): + """``&``, newlines and ``<(...)``/``>(...)`` all run another command.""" + from apodex.permissions import PermissionStore + + rules = PermissionStore(allow={"Bash(echo)", "Bash(python)", "Bash(cat)"}) + for cmd in ("echo hi & touch /tmp/marker", "echo hi\ntouch /tmp/marker", + "cat <(touch /tmp/marker)", "echo >(touch /tmp/marker)"): + assert not rules.allows("bash", {"command": cmd}), cmd + # The "&" in a redirection is not a separator. + for cmd in ("python x.py 2>&1", "python x.py >&2", "python x.py &> out.log", + "python x.py &", "cat <(echo a)", 'echo "<(touch /tmp/marker)"'): + assert rules.allows("bash", {"command": cmd}), cmd + assert PermissionStore(deny={"Bash(touch)"}).denies( + "bash", {"command": "echo hi & touch /tmp/marker"} + ) + + +def test_dup_redirect_target_is_expanded_twice(tmp_path): + """bash expands a ``>&word`` target again after quote removal, so + ``echo x >&'$(touch m)'`` runs ``touch`` despite the single quotes.""" + from apodex.agent_tools import RISK_CONFIRM, assess_with_rules + from apodex.permissions import PermissionStore + from plugins.tools._bash_policy import assess_bash_command + + rules = PermissionStore(allow={"Bash(echo)"}) + cmd = {"command": "echo x >&'$(touch /tmp/marker)'"} + assert not rules.allows("bash", cmd) + assert assess_with_rules("bash", cmd, str(tmp_path), rules).level == RISK_CONFIRM + assert rules.allows("bash", {"command": "echo x > '$(touch /tmp/marker)'"}) # plain > is literal + assert assess_bash_command("echo x >&'$(foobarcmd)'", mode="enforce").level == "deny" + + +def test_deep_nesting_fails_closed_without_recursion_error(): + from apodex.permissions import PermissionStore + + cmd = {"command": "echo " + "$(echo " * 2000 + "x" + ")" * 2000} + assert not PermissionStore(allow={"Bash(echo)"}).allows("bash", cmd) + assert PermissionStore(allow={"Bash(*)"}, deny={"Bash(rm)"}).denies("bash", cmd) + + def test_user_settings_save_and_load(tmp_path): from apodex.config import UserSettings p = str(tmp_path / "settings.json") diff --git a/plugins/tools/_bash_policy.py b/plugins/tools/_bash_policy.py index a01ade0..a15c7f1 100644 --- a/plugins/tools/_bash_policy.py +++ b/plugins/tools/_bash_policy.py @@ -798,44 +798,157 @@ def _split_top_level(command: str) -> list[str]: return [s.strip() for s in segs if s.strip()] +def _backtick_body(command: str, i: int) -> tuple[str, int]: + """Read a backtick body and apply bash's first-pass escape removal. + + Inside backticks, backslashes before $, `, a backslash or a newline are removed + before the body is parsed as shell code. Other backslashes are retained. + Return the decoded body and the closing backtick's index (len if absent). + """ + body: list[str] = [] + n = len(command) + while i < n: + c = command[i] + if c == "`": + break + if c == "\\" and i + 1 < n and command[i + 1] in "$`\\\n": + i += 1 + if command[i] != "\n": + body.append(command[i]) + else: + body.append(c) + i += 1 + return "".join(body), i + + +def _substitution_end(command: str, i: int) -> int: + """Index of the ``)`` closing a ``$(`` whose body starts at ``i``, or + ``len(command)`` when unterminated. Quoted or escaped parens don't count, + so ``$(echo ")"; rm x)`` closes at the last ``)``, not inside the quotes. + + Like bash, every nested ``$(`` starts with its own quote state, so the + quotes in ``$(echo "$(echo ")'")" $(rm x))`` pair up inside the inner + substitution and the scan still reaches ``rm x``. The scan keeps the levels + on a list instead of the call stack, so deep nesting can't hit the + recursion limit. It skips a backtick span whole. + """ + n = len(command) + quotes: list[str | None] = [None] # quote state of each open $( level + depths = [1] # unquoted "(" nesting inside each open $( level + while i < n: + c = command[i] + quote = quotes[-1] + if quote == "'": + if c == "'": + quotes[-1] = None + elif c == "\\": + i += 1 + elif command.startswith("(", i + 1) and (c == "$" or (c in "<>" and quote is None)): + quotes.append(None) + depths.append(1) + i += 1 + elif c == "`": + i += 1 + while i < n and command[i] != "`": + i += 2 if command[i] == "\\" else 1 + elif c == '"': + quotes[-1] = None if quote else '"' + elif quote is None: + if c == "'": + quotes[-1] = "'" + elif c == "(": + depths[-1] += 1 + elif c == ")": + depths[-1] -= 1 + if depths[-1] == 0: + if len(depths) == 1: + return i + depths.pop() + quotes.pop() + i += 1 + return n + + +def _dup_redirect_word(command: str, i: int) -> str: + """The target word of a ``>&`` redirect that starts at ``i``, with its + quotes and backslashes removed. + + bash expands that word a second time after quote removal, so + ``echo x >&'$(id)'`` runs ``id``. The stripped text is roughly what the + second pass sees. Dropping every backslash can only expose more ``$(``, + never hide one. + """ + n = len(command) + while i < n and command[i] in " \t": + i += 1 + start = i + quote: str | None = None + while i < n: + c = command[i] + if quote == "'": + if c == "'": + quote = None + elif c == "\\": + i += 1 + elif command.startswith("$(", i): + i = _substitution_end(command, i + 2) + elif c == "`": + i += 1 + while i < n and command[i] != "`": + i += 2 if command[i] == "\\" else 1 + elif c == '"': + quote = None if quote else '"' + elif quote is None: + if c == "'": + quote = "'" + elif c.isspace() or c in ";&|<>()": + break + i += 1 + return re.sub(r"[\\'\"]", "", command[start:i]) + + def _extract_nested_shell(command: str) -> list[str]: - """Return shell-code strings nested in ``$(...)`` and backticks (which the - shell expands+executes). Single-quoted spans are skipped — the shell does - not expand them, so ``echo '$(rm -rf /)'`` is a harmless literal.""" + """Return shell-code strings nested in ``$(...)``, backticks and unquoted + process substitution ``<(...)``/``>(...)`` (which the shell executes). + + It reads quotes the way bash does. A single-quoted span is skipped, so + ``echo '$(rm -rf /)'`` is a harmless literal. Inside double quotes a ``'`` + is an ordinary character and substitution still runs, so + ``echo "'$(rm -rf /)'"`` yields ``rm -rf /``. A backslash outside single + quotes escapes the next character (``\\$(...)``, ``\\'``, ``\\"``). An + unterminated substitution yields the rest of the string, which fails closed. + The one exception to quoting is the target of ``>&``, which bash expands + twice (see :func:`_dup_redirect_word`). + """ out: list[str] = [] i, n = 0, len(command) - sq = False + quote: str | None = None while i < n: c = command[i] - if sq: + if quote == "'": if c == "'": - sq = False + quote = None i += 1 continue - if c == "'": - sq = True - i += 1 + if c == "\\": + i += 2 continue - if c == "$" and i + 1 < n and command[i + 1] == "(": - depth, j = 1, i + 2 - start = j - while j < n and depth: - if command[j] == "(": - depth += 1 - elif command[j] == ")": - depth -= 1 - j += 1 - if depth == 0: - out.append(command[start:j - 1]) - i = j + if c == "'" and quote is None: + quote = "'" + elif c == '"': + quote = None if quote else '"' + elif command.startswith("(", i + 1) and (c == "$" or (c in "<>" and quote is None)): + end = _substitution_end(command, i + 2) + out.append(command[i + 2:end]) + i = end + 1 continue - if c == "`": - j = i + 1 - while j < n and command[j] != "`": - j += 1 - out.append(command[i + 1:j]) + elif c == "`": + body, j = _backtick_body(command, i + 1) + out.append(body) i = j + 1 continue + elif c == ">" and quote is None and command.startswith("&", i + 1): + out.extend(_extract_nested_shell(_dup_redirect_word(command, i + 2))) i += 1 return out diff --git a/tests/test_permissions_shell_expansion.py b/tests/test_permissions_shell_expansion.py new file mode 100644 index 0000000..4d4cdf7 --- /dev/null +++ b/tests/test_permissions_shell_expansion.py @@ -0,0 +1,100 @@ +"""Compare saved shell permissions with actual bash expansion semantics.""" + +from __future__ import annotations + +import itertools +import shlex +import shutil +import subprocess + +import pytest + +from apodex.agent_tools import RISK_CONFIRM, RISK_DENY, assess_with_rules +from apodex.permissions import PermissionStore, _fallback_nested_shell +from plugins.tools._bash_policy import _extract_nested_shell, assess_bash_command + + +def _backtick(code: str) -> str: + """Encode a shell body for one legacy command-substitution level.""" + escaped = "".join("\\" + c if c in "\\$`" else c for c in code) + return "`" + escaped + "`" + + +@pytest.mark.parametrize("body,expected", [ + (r"echo \$(touch marker)", "echo $(touch marker)"), + (r"echo \`touch marker\`", "echo `touch marker`"), + (r"echo \\$(touch marker)", r"echo \$(touch marker)"), + (r"echo \q", r"echo \q"), + ("echo to\\\nuch", "echo touch"), +]) +def test_backtick_scanners_apply_bash_escape_removal(body, expected): + command = "echo `" + body + "`" + assert _extract_nested_shell(command) == [expected] + assert _fallback_nested_shell(command) == [expected] + + +@pytest.mark.parametrize("command", [ + r"echo `echo \$(foobarcmd)`", + r"echo `echo \`foobarcmd\``", + 'echo "`echo \\$(foobarcmd)`"', +]) +def test_backtick_escape_removal_reaches_enforced_policy(command): + assert assess_bash_command(command, mode="enforce").level == "deny" + + +def test_saved_permissions_match_real_bash_nested_expansions(tmp_path): + """Every generated command actually runs touch; none may inherit echo's allow. + + Mix both substitution syntaxes and double-quoted variants at each level. + Execute only harmless commands that write one marker in this test's folder. + """ + bash = shutil.which("bash") + if bash is None: + pytest.skip("bash is needed for the shell-semantics comparison") + marker = tmp_path / "marker" + marker_word = shlex.quote(marker.as_posix()) + allow = PermissionStore(allow={"Bash(echo)"}) + deny = PermissionStore(allow={"Bash(*)"}, deny={"Bash(touch)"}) + wrappers = ( + lambda code: "echo $(" + code + ")", + lambda code: 'echo "$(' + code + ')"', + lambda code: "echo " + _backtick(code), + lambda code: 'echo "' + _backtick(code) + '"', + ) + commands = [] + for depth in range(1, 4): + for sequence in itertools.product(wrappers, repeat=depth): + command = "touch " + marker_word + for wrap in sequence: + command = wrap(command) + commands.append(command) + # The concrete review examples also include a second evaluation level. + commands.extend([ + r"echo `echo \$(touch " + marker_word + r")`", + r"echo `echo \`touch " + marker_word + r"\``", + 'echo $(echo "$(echo ")\'")" $(touch ' + marker_word + '))', + ]) + for command in commands: + result = subprocess.run( + [bash, "--noprofile", "--norc", "-c", command], + cwd=tmp_path, capture_output=True, text=True, timeout=5, + ) + assert result.returncode == 0, (command, result.stderr) + assert marker.exists(), command + marker.unlink() + args = {"command": command} + assert assess_with_rules("bash", args, str(tmp_path), allow).level == RISK_CONFIRM, command + assert assess_with_rules("bash", args, str(tmp_path), deny).level == RISK_DENY, command + assert _extract_nested_shell(command) == _fallback_nested_shell(command), command + + +@pytest.mark.parametrize("command", [ + "echo '$(touch marker)'", + "echo '`touch marker`'", + r"echo \`touch marker\`", + r"echo `echo '\$(touch marker)'`", + "echo $(echo literal)", + "echo `echo literal`", +]) +def test_literal_and_authorized_backtick_payloads_stay_allowed(command): + assert PermissionStore(allow={"Bash(echo)"}).allows("bash", {"command": command})