fix(bash-policy): screen shell nesting past the recursion limit instead of dropping it

CHANGELOG.md

@@ -65,6 +65,17 @@ 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 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.
+
 - **A skipped sandbox-isolation test can no longer pass for verification.**
   The real-bubblewrap test's guard probed with `bwrap --ro-bind / / true`,
   which fails on a *working* host — bwrap is not setuid, so without

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,6 +2548,66 @@ def _argv_group_deny(commands: list[list[str]]) -> tuple[str, str] | None:
     return None
 
 
+#: 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_commands(bodies: list[str]) -> list[list[str]]:
+    """Conservatively screen code past the parser's nesting limit.
+
+    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.
+
+    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.
+    """
+    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]]:
     """Parse into a list of argv lists (one per simple command), recursively
     including commands nested in ``$(...)`` / backticks, in the code argument of
@@ -2581,9 +2643,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 +2655,13 @@ 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:
+        # 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:
         if sub.strip():
             # Unparseable nested code — the outer parse already recorded it.

tests/test_bash_policy.py

@@ -596,3 +596,226 @@ 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"
+
+
+@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"