From fa7e5085b03f78359715381827d9426e253830b5 Mon Sep 17 00:00:00 2001 From: Zhang Handuo Date: Tue, 6 Oct 2026 20:37:10 +0800 Subject: [PATCH 1/2] fix(bash-policy): screen shell nesting past the recursion limit instead of dropping it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the gap recorded in #61 and tracked upstream as ApodexHarness#632. `_parse_commands` stopped recursing at `_MAX_NEST` (4) and stopped silently, so every level below was assessed by nobody. Measured in `off` mode — the default, and what most deployments run: $($($($($($(sudo id)))))) -> allow (6 levels) $($($($($($(ssh h id)))))) -> allow $($($($($($(pkill -f x)))))) -> allow against Layer 1.5 rules that are supposed to bind in every mode. `enforce` denied them, but only because the masked sentinel left in executable position is off the allowlist — incidental, not the rule that should have applied. At the limit the remaining text is now screened by its WORDS in one linear pass (`_expansion_words`): expansion punctuation is blanked and each word becomes a candidate executable, so `sudo` falls out of any depth. Two wrong turns on the way, both worth recording since each looked right: - Removing the cap and recursing on gave `RecursionError` at depth 2,000 — a crash instead of a verdict, which is worse than the bypass. - Peeling the chain one layer at a time cost a scan per layer: 143ms at depth 500 against main's 19ms, 11.6s at 5,000 — and at any step bound it lost the payload again, restoring the very bypass. The word screen is 19.7ms at 500 and 770ms at 20,000. The screen is deliberately conservative past the limit: it cannot tell a command name from a word that looks like one, so `$(...$(echo ssh)...)` nested 8 deep is refused on the strength of `ssh` appearing in it. That applies only past a depth no real command reaches, and it can only add refusals. Differential check over the nesting corpus: 9 verdicts change, all of them `off`-mode allow -> deny with the correct group; no benign form changes at any depth, and no `enforce` verdict moves. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 9 +++ plugins/tools/_bash_policy.py | 52 ++++++++++++++- tests/test_bash_policy.py | 115 ++++++++++++++++++++++++++++++++++ 3 files changed, 173 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 044ceb9..5733db6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,15 @@ Initial open-source release of FrontierAgent. ### Fixed +- **Shell nesting past the parser's recursion limit no longer bypasses the + group denials.** `_parse_commands` stopped recursing at four levels and + stopped *silently*, so `$($($($($($(sudo id))))))` assessed as `allow` in + `off` mode — the mode most deployments run — against rules that are supposed + to bind unconditionally. At the limit the remaining text is now screened by + its words in a single pass, so any depth is refused and the verdict still + names the group. Recursing further instead would raise `RecursionError` + (a crash, not a verdict), and peeling the chain layer by layer cost a scan + per layer (11s at depth 5,000). Benign nesting is unchanged at every depth. - Keep bash protection on resolved local system aliases such as macOS `/private/etc`, while exempting the run's own writable mounts. Recursive deletion and mutation refuse ancestors and wildcard selections of read-only diff --git a/plugins/tools/_bash_policy.py b/plugins/tools/_bash_policy.py index 4daaaf0..f646cbc 100644 --- a/plugins/tools/_bash_policy.py +++ b/plugins/tools/_bash_policy.py @@ -2546,6 +2546,35 @@ def _argv_group_deny(commands: list[list[str]]) -> tuple[str, str] | None: return None +#: Expansion punctuation, blanked in one pass by :func:`_expansion_words`. +_EXPANSION_PUNCT_RE = re.compile(r"\$\(\(|\$\(|<\(|>\(|[`()]") + + +def _expansion_words(bodies: list[str]) -> list[str]: + """Every word inside ``bodies``, with expansion punctuation removed. + + Deliberately a WORD screen rather than a parse. Peeling a chain layer by + layer costs a scan per layer, so a 5,000-deep string took 11s — and it + still lost the payload once it hit any step bound, which is the bypass this + is here to close. Blanking ``$(``, backticks and parentheses in one linear + pass drops ``sudo id`` out of any depth for the price of one scan. + + Conservative by construction: it cannot tell a command name from a word + that merely looks like one, so ``$($($($($($(echo ssh))))))`` is refused on + the strength of ``ssh`` appearing in it. That only applies past + ``_MAX_NEST``, which no real command reaches, and it can only ADD + refusals — the alternative is assessing those levels not at all. + """ + words: list[str] = [] + for body in bodies: + flat = _EXPANSION_PUNCT_RE.sub(" ", body) + try: + words.extend(tokenize_shell_segment(flat)) + except ValueError: + words.extend(flat.split()) + return words + + def _parse_commands(command: str, depth: int = 0) -> list[list[str]]: """Parse into a list of argv lists (one per simple command), recursively including commands nested in ``$(...)`` / backticks, in the code argument of @@ -2581,9 +2610,6 @@ def _parse_commands(command: str, depth: int = 0) -> list[list[str]]: for argv in list(argvs): argvs.extend(_find_exec_payloads(argv)) - if depth >= _MAX_NEST: - return argvs - nested = _extract_nested_shell(stripped) + list(heredoc_bodies) for raw_argv in list(argvs): # ``_shell_code_args`` unwraps prefixes first, so ``env bash -c …`` / @@ -2596,6 +2622,26 @@ def _parse_commands(command: str, depth: int = 0) -> list[list[str]]: # (Layer 1.5) and the allowlist never saw ``watch 'sudo id'``. nested.extend(_evaluator_payloads(raw_argv)[0]) nested.extend(_env_split_payloads(raw_argv)) + if depth >= _MAX_NEST: + # The recursion stops here — and stopping SILENTLY is the one direction + # a policy must not fail in: every level below was assessed by nobody, + # so a group denial that is supposed to bind in EVERY mode could be + # walked around with enough parentheses. Measured in ``off`` mode + # against a limit of 4: + # $($($($($($(sudo id)))))) -> allow + # ``enforce`` denied it, but only because the masked sentinel left in + # executable position is off the allowlist — incidental, not the rule + # that should have applied. + # + # The remaining text is screened by its WORDS instead (see + # ``_expansion_words``). Recursing on would trade the bypass for a + # ``RecursionError`` — a crash rather than a verdict — and peeling the + # layers one at a time costs a scan per layer. + # One argv per word, so every one of them is a candidate executable + # for the group check rather than only the first. + argvs.extend([word] for word in _expansion_words(nested) if word.strip()) + return argvs + for sub in nested: if sub.strip(): # Unparseable nested code — the outer parse already recorded it. diff --git a/tests/test_bash_policy.py b/tests/test_bash_policy.py index bfc9f0a..b9c959a 100644 --- a/tests/test_bash_policy.py +++ b/tests/test_bash_policy.py @@ -596,3 +596,118 @@ def test_escapes_inside_backticks_are_decoded_once() -> None: def test_an_unterminated_backtick_still_assesses_its_body() -> None: assert assess_bash_command("echo `sudo id", mode="off").level == "deny" + + +# ── nesting past the recursion limit fails closed (Harness #632) ───────── +# +# ``_parse_commands`` stops recursing at ``_MAX_NEST``, and stopped SILENTLY: +# the levels below were assessed by nobody, so a group denial that is supposed +# to bind in every mode could be walked around with enough parentheses. +# Measured in ``off`` mode before the fix, against a limit of 4: +# $($($($($($(sudo id)))))) -> allow +# ``enforce`` denied it, but only because the masked sentinel left in +# executable position is not on the allowlist — incidental, not the rule that +# should have applied. + + +def _nest(body: str, depth: int) -> str: + return "$(" * depth + body + ")" * depth + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("depth", [1, 3, 4, 5, 6, 8, 12, 40]) +@pytest.mark.parametrize(("body", "group"), [ + ("sudo id", "priv_esc"), + ("ssh h 'cat ~/.aws/credentials'", "exfil"), + ("pkill -f python3", "process_kill"), +]) +def test_group_denials_survive_any_nesting_depth( + body: str, group: str, depth: int, mode: str, +) -> None: + result = assess_bash_command(_nest(body, depth), mode=mode) + assert result.level == "deny" + # Past the limit the whole leftover text is screened as one unit, so the + # group is still named rather than the denial coming from somewhere else. + assert result.group == group + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("depth", [5, 9, 20]) +@pytest.mark.parametrize("body", ["halt", "mkfs.ext4 /dev/sda", "reboot"]) +def test_word_screens_survive_any_nesting_depth( + body: str, depth: int, mode: str, +) -> None: + assert assess_bash_command(_nest(body, depth), mode=mode).level == "deny" + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("wrapper", [ + "bash <<'EOF'\n{code}\nEOF", + "bash -c '{code}'", + "watch -n 1 '{code}'", + "env -S '{code}'", +]) +def test_depth_is_not_reset_by_another_nesting_channel(wrapper: str, mode: str) -> None: + # Heredocs, shell -c payloads and evaluator payloads all recurse through + # the same counter, so stacking them cannot buy extra depth. + command = wrapper.format(code=_nest("sudo id", 8)) + assert assess_bash_command(command, mode=mode).level == "deny" + + +@pytest.mark.parametrize("depth", [1, 4, 6, 10, 30]) +def test_benign_nesting_is_unaffected(depth: int) -> None: + # The leftover screen can only ADD refusals, so a deep but harmless chain + # must read exactly as it did before. + assert assess_bash_command(_nest("date", depth), mode="off").level == "allow" + + +@pytest.mark.parametrize("command", [ + "echo $(echo $(echo $(echo $(echo $(date)))))", + "v=$(python3 -V); echo $v", + "echo $(( $(echo 1) + 1 ))", + "for f in $(ls $(pwd)); do echo $f; done", + "echo $(command -v $(echo python3))", + "diff <(sort $(echo a)) <(sort b)", +]) +def test_deeply_nested_benign_commands_stay_allowed(command: str) -> None: + assert assess_bash_command(command, mode="off").level == "allow" + + +@pytest.mark.parametrize("depth", [200, 2_000, 20_000]) +def test_pathological_depth_is_bounded_work_not_a_crash(depth: int) -> None: + """Three failure modes this has to avoid at once: losing the payload, + raising ``RecursionError`` instead of returning a verdict, and taking so + long that the assessment itself is the denial of service. + + An earlier attempt peeled the chain one layer at a time, which cost a scan + per layer: 11s at depth 5,000, and it lost the payload again once it hit + its step bound. The word screen is one pass. + """ + import time + + command = _nest("sudo id", depth) + started = time.perf_counter() + result = assess_bash_command(command, mode="off") + elapsed = time.perf_counter() - started + + assert result.level == "deny" + assert result.group == "priv_esc" + # Generous enough not to be flaky on a loaded machine, tight enough to + # catch a return to per-layer scanning (which was ~60x this at 20k). + assert elapsed < 5.0, f"depth {depth} took {elapsed:.1f}s" + + +def test_an_unterminated_deep_chain_still_fails_closed() -> None: + # The span scanner treats an unterminated ``$(`` as running to end-of-text. + assert assess_bash_command("$(" * 200 + "sudo id", mode="off").level == "deny" + + +def test_the_word_screen_is_conservative_past_the_limit() -> None: + """Past ``_MAX_NEST`` the screen cannot tell a command name from a word + that looks like one, so a denied name appearing as DATA is refused. Worth + pinning as deliberate: it only applies at a depth no real command reaches, + and the alternative is not assessing those levels at all.""" + quoted_data = _nest("echo ssh", 8) + assert assess_bash_command(quoted_data, mode="off").level == "deny" + # Below the limit the same text is read properly and allowed. + assert assess_bash_command(_nest("echo ssh", 2), mode="off").level == "allow" From 10109d795a574ec76ff85198837ed12b09f41fa8 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Wed, 7 Oct 2026 10:05:52 +0800 Subject: [PATCH 2/2] fix(bash-policy): inspect deep separators and quoted payloads --- CHANGELOG.md | 12 ++-- plugins/tools/_bash_policy.py | 96 ++++++++++++++++++------------ tests/test_bash_policy.py | 108 ++++++++++++++++++++++++++++++++++ 3 files changed, 173 insertions(+), 43 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5733db6..ebf2804 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -69,11 +69,13 @@ Initial open-source release of FrontierAgent. group denials.** `_parse_commands` stopped recursing at four levels and stopped *silently*, so `$($($($($($(sudo id))))))` assessed as `allow` in `off` mode — the mode most deployments run — against rules that are supposed - to bind unconditionally. At the limit the remaining text is now screened by - its words in a single pass, so any depth is refused and the verdict still - names the group. Recursing further instead would raise `RecursionError` - (a crash, not a verdict), and peeling the chain layer by layer cost a scan - per layer (11s at depth 5,000). Benign nesting is unchanged at every depth. + to bind unconditionally. At the limit a bounded screen now checks shell + separators, quoted evaluator payloads, and full argument lists as well as + candidate executable names. Unparseable or excessively quoted residual code + is refused rather than silently skipped. Plain substitution chains are + screened in linear time without additional parser recursion. The fallback + deliberately treats code-looking data conservatively past the limit. + - Keep bash protection on resolved local system aliases such as macOS `/private/etc`, while exempting the run's own writable mounts. Recursive deletion and mutation refuse ancestors and wildcard selections of read-only diff --git a/plugins/tools/_bash_policy.py b/plugins/tools/_bash_policy.py index f646cbc..3ddea3a 100644 --- a/plugins/tools/_bash_policy.py +++ b/plugins/tools/_bash_policy.py @@ -757,6 +757,8 @@ def _argv_hard_deny(commands: list[list[str]]) -> str | None: """Robust dangerous-op detection on parsed argvs. Complements the raw regex — order-, flag-combination-, prefix- and quote-independent. """ + if any(_EXPANSION_SCREEN_FAILURE in argv for argv in commands): + return "Cannot safely inspect shell code past the nesting limit; simplify the command." cd_into_protected = False for raw_argv in commands: argv = strip_command_prefixes(raw_argv) @@ -2546,33 +2548,64 @@ def _argv_group_deny(commands: list[list[str]]) -> tuple[str, str] | None: return None -#: Expansion punctuation, blanked in one pass by :func:`_expansion_words`. +#: Expansion punctuation, blanked before the bounded fallback screen. _EXPANSION_PUNCT_RE = re.compile(r"\$\(\(|\$\(|<\(|>\(|[`()]") +_EXPANSION_SCREEN_FAILURE = "__FA_EXPANSION_SCREEN_FAILURE__" +_MAX_SCREEN_PASSES = 8 +_RESIDUAL_SHELL_SYNTAX_RE = re.compile(r"[\s;|&'\"\\`()]|\$'") -def _expansion_words(bodies: list[str]) -> list[str]: - """Every word inside ``bodies``, with expansion punctuation removed. +def _expansion_commands(bodies: list[str]) -> list[list[str]]: + """Conservatively screen code past the parser's nesting limit. - Deliberately a WORD screen rather than a parse. Peeling a chain layer by - layer costs a scan per layer, so a 5,000-deep string took 11s — and it - still lost the payload once it hit any step bound, which is the bypass this - is here to close. Blanking ``$(``, backticks and parentheses in one linear - pass drops ``sudo id`` out of any depth for the price of one scan. + Flatten expansion punctuation once per pass, split shell separators, and + retain both complete argvs (for argument-sensitive hard denials) and each + word as a candidate executable. Quoted code is decoded by tokenization; + words still containing shell syntax are screened again, without recursion. + This intentionally treats code-looking data as code at excessive depth. - Conservative by construction: it cannot tell a command name from a word - that merely looks like one, so ``$($($($($($(echo ssh))))))`` is refused on - the strength of ``ssh`` appearing in it. That only applies past - ``_MAX_NEST``, which no real command reaches, and it can only ADD - refusals — the alternative is assessing those levels not at all. + Both the pass count and total input scanned are bounded. If quoting cannot + be resolved within that budget, or tokenization fails, record a hard denial + rather than dropping the remaining code. A plain substitution chain of any + depth needs just one pass, so its cost stays linear in the input size. """ - words: list[str] = [] - for body in bodies: - flat = _EXPANSION_PUNCT_RE.sub(" ", body) - try: - words.extend(tokenize_shell_segment(flat)) - except ValueError: - words.extend(flat.split()) - return words + commands: list[list[str]] = [] + candidates: list[list[str]] = [] + pending = bodies + remaining = max(1_024, _MAX_SCREEN_PASSES * sum(map(len, bodies))) + for _ in range(_MAX_SCREEN_PASSES): + next_pass: list[str] = [] + for body in pending: + remaining -= len(body) + if remaining < 0: + return [*commands, *candidates, [_EXPANSION_SCREEN_FAILURE]] + flat = _EXPANSION_PUNCT_RE.sub(" ", _normalize_ansi_c_quotes(body)) + for segment in _split_top_level(flat): + try: + words = tokenize_shell_segment(segment) + except ValueError: + return [*commands, *candidates, [_EXPANSION_SCREEN_FAILURE]] + if not words: + continue + keyword = _leading_shell_keyword(segment) + if keyword in _CONTROL_LEADERS and words[0] == keyword: + words[0] = _SHELL_SYNTAX_TOKEN + commands.append(words) + commands.extend(_find_exec_payloads(strip_command_prefixes(words))) + candidates.extend([word] for word in words if word.strip()) + for word in words: + # Encoded redirection syntax is already checked in the + # complete argv; feeding its sentinel back to shlex would + # turn an ordinary redirect into a parse failure. + if redirection_token(word) is None and _RESIDUAL_SHELL_SYNTAX_RE.search(word): + next_pass.append(word) + if not next_pass: + # Keep complete commands in execution order. Interspersing a + # singleton `cd` would reset the protected-directory state before + # a following relative deletion could be checked. + return [*commands, *candidates] + pending = next_pass + return [*commands, *candidates, [_EXPANSION_SCREEN_FAILURE]] def _parse_commands(command: str, depth: int = 0) -> list[list[str]]: @@ -2623,23 +2656,10 @@ def _parse_commands(command: str, depth: int = 0) -> list[list[str]]: nested.extend(_evaluator_payloads(raw_argv)[0]) nested.extend(_env_split_payloads(raw_argv)) if depth >= _MAX_NEST: - # The recursion stops here — and stopping SILENTLY is the one direction - # a policy must not fail in: every level below was assessed by nobody, - # so a group denial that is supposed to bind in EVERY mode could be - # walked around with enough parentheses. Measured in ``off`` mode - # against a limit of 4: - # $($($($($($(sudo id)))))) -> allow - # ``enforce`` denied it, but only because the masked sentinel left in - # executable position is off the allowlist — incidental, not the rule - # that should have applied. - # - # The remaining text is screened by its WORDS instead (see - # ``_expansion_words``). Recursing on would trade the bypass for a - # ``RecursionError`` — a crash rather than a verdict — and peeling the - # layers one at a time costs a scan per layer. - # One argv per word, so every one of them is a candidate executable - # for the group check rather than only the first. - argvs.extend([word] for word in _expansion_words(nested) if word.strip()) + # Deep code is screened without further parser recursion. The fallback + # must retain separators, quoted payloads and argument-sensitive rules; + # anything it cannot inspect within bounded work fails closed. + argvs.extend(_expansion_commands(nested)) return argvs for sub in nested: diff --git a/tests/test_bash_policy.py b/tests/test_bash_policy.py index b9c959a..b6f13b1 100644 --- a/tests/test_bash_policy.py +++ b/tests/test_bash_policy.py @@ -711,3 +711,111 @@ def test_the_word_screen_is_conservative_past_the_limit() -> None: assert assess_bash_command(quoted_data, mode="off").level == "deny" # Below the limit the same text is read properly and allowed. assert assess_bash_command(_nest("echo ssh", 2), mode="off").level == "allow" + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("depth", [4, 5, 6, 12, 40]) +@pytest.mark.parametrize(("body", "group"), [ + ("date;sudo id", "priv_esc"), + ("date|ssh h id", "exfil"), + ("date&&pkill -f x", "process_kill"), + ("date||sudo id", "priv_esc"), + ("date&sudo id", "priv_esc"), + ("date\nsudo id", "priv_esc"), +]) +def test_deep_screen_splits_shell_command_separators( + body: str, group: str, depth: int, mode: str, +) -> None: + result = assess_bash_command(_nest(body, depth), mode=mode) + assert result.level == "deny" + assert result.group == group + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("depth", [4, 5, 8, 40]) +@pytest.mark.parametrize(("body", "group"), [ + ("bash -c 'sudo id'", "priv_esc"), + ('bash -c "ssh h id"', "exfil"), + ("watch 'pkill -f x'", "process_kill"), + ("env -S 'sudo id'", "priv_esc"), + ("bash -c 'date;sudo id'", "priv_esc"), + ('bash -c "su\\\"do\\\" id"', "priv_esc"), + (r"bash -c $'\x73udo id'", "priv_esc"), + (r"bash -c 's\udo id'", "priv_esc"), +]) +def test_deep_screen_decodes_quoted_and_escaped_code( + body: str, group: str, depth: int, mode: str, +) -> None: + result = assess_bash_command(_nest(body, depth), mode=mode) + assert result.level == "deny" + assert result.group == group + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("body", [ + "rm -rf /etc", + "env rm -rf /etc", + "chmod -R 777 /etc", + "chown -R root /etc", + "find /etc -delete", + "date;rm -rf /etc", + "bash -c 'rm -rf /etc'", + "bash -c 'cd /etc; rm -rf .'", + "if rm -rf /etc; then date; fi", + "find /tmp -exec rm -rf /etc \\;", +]) +def test_deep_screen_preserves_argument_sensitive_hard_denials(body: str, mode: str) -> None: + result = assess_bash_command(_nest(body, 8), mode=mode, interactive=True) + assert result.level == "deny" + assert not result.group + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("body", ["date;sudo id", "bash -c 'sudo id'"]) +def test_deep_group_denials_still_require_human_confirmation(body: str, mode: str) -> None: + result = assess_bash_command(_nest(body, 8), mode=mode, interactive=True) + assert result.level == "confirm" + assert result.group == "priv_esc" + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("interactive", [False, True]) +def test_excessive_quoting_fails_closed_with_bounded_work(mode: str, interactive: bool) -> None: + import shlex + import time + + body = "sudo id" + for _ in range(9): + body = shlex.quote(body) + started = time.perf_counter() + result = assess_bash_command(_nest("eval " + body, 6), mode=mode, interactive=interactive) + assert result.level == "deny" + assert "nesting limit" in result.reason + assert time.perf_counter() - started < 5.0 + + +def test_exhausted_screen_budget_cannot_silently_drop_pending_code( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(policy, "_MAX_SCREEN_PASSES", 1) + result = assess_bash_command(_nest("bash -c 'sudo id'", 8), mode="off") + assert result.level == "deny" + assert "nesting limit" in result.reason + + +def test_malformed_residual_code_fails_closed() -> None: + commands = policy._expansion_commands(["bash -c 'sudo id"]) + assert policy._argv_hard_deny(commands) is not None + + +@pytest.mark.parametrize("body", [ + "date;pwd", + "date|cat", + "bash -c 'date;pwd'", + "watch 'date'", + "env -S 'date'", + "rm -rf /tmp/frontier-policy-test", + "echo ok > /tmp/frontier-policy-test", +]) +def test_benign_deep_code_survives_bounded_screen(body: str) -> None: + assert assess_bash_command(_nest(body, 8), mode="off").level == "allow"