From 22dd37524cf6c6d4ed6ec2bfae4d451820db7095 Mon Sep 17 00:00:00 2001 From: agent-exec-transport Date: Wed, 21 Oct 2026 02:42:12 +0000 Subject: [PATCH] agent-exec transport 7909cc2e3bc15bc12eb30244170e2eac/9723af6b7b77 --- tests/test_shell_pipe_hazard.py | 189 +++++++++++++++++++++++++++----- 1 file changed, 164 insertions(+), 25 deletions(-) diff --git a/tests/test_shell_pipe_hazard.py b/tests/test_shell_pipe_hazard.py index 8abb3b24..4452ffd4 100644 --- a/tests/test_shell_pipe_hazard.py +++ b/tests/test_shell_pipe_hazard.py @@ -1,4 +1,5 @@ -"""托管 shell 脚本：管道任一读取级不得是提前退出的读取器（grep -q/-m、head、sed q、awk exit、read）。 +"""托管 shell 脚本：管道任一读取级不得是提前退出的读取器（grep -q/-m、head、sed q、awk exit/nextfile、read、 +带 break 的 while/until 循环；含被子壳或花括组包裹、带绝对路径或 exec 等前缀的写法）。 pipefail 下读取器先退出，上游生产者写满 64 KiB 管道缓冲后收到 SIGPIPE，管道状态报 141， `if producer | grep -q PAT` 与 `if ! ...` 的判定被静默翻转；读取器位于管道中段 @@ -6,7 +7,9 @@ pipefail 下读取器先退出，上游生产者写满 64 KiB 管道缓冲后收 读取器的结论才不会丢。 """ import re +import shutil import subprocess +import tempfile import unittest from pathlib import Path @@ -19,8 +22,11 @@ EXEMPT = {} HEREDOC = re.compile(r"<<(-?)\s*(['\"]?)([A-Za-z_][A-Za-z0-9_]*)\2") SED_QUIT = re.compile(r"(?:^|[\s;{}'\"/$0-9])[qQ][0-9]*\s*(?:$|[\s;}'\"])") AWK_END = re.compile(r"\bEND\s*\{") -AWK_EXIT = re.compile(r"\bexit\b") +AWK_EXIT = re.compile(r"\b(?:exit|nextfile)\b") ASSIGNMENT = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=") +WORD = re.compile(r"(\S+)\s*") +BREAK_ARG = re.compile(r"[ \t]+([0-9]+)") +PREFIXES = {"!", "time", "command", "builtin", "env", "exec"} GREP_ARG_OPTS = set("efABCdDX") BOUNDARY = set(" \t\n;&|()<>") @@ -35,15 +41,20 @@ def managed_scripts(): class _Frame: - """一层命令上下文（顶层、$(...)、(...)、<(...)）内正在累积的管道。""" + """一层命令上下文（顶层、$(...)、(...)、<(...)、管道级的 { ...; }、while/until 循环）内正在累积的管道。""" - def __init__(self, line): + def __init__(self, line, depth, closer=None): self.stages, self.stage, self.start, self.cases = [], [], line, [] + self.depth, self.closer, self.dos, self.broken = depth, closer, 0, False -def pipelines(src): - """词法扫描 bash 源码，返回每条多级管道各级的 [(起始行号, 文本), ...]；跳过注释与 heredoc 正文。""" - found, frames, modes, heredocs = [], [_Frame(1)], ["cmd"], [] +def _lex(src): + """词法扫描 bash 源码；跳过注释与 heredoc 正文。 + + 返回 (lists, loops)：lists 为每个命令列表的 (帧深度, [(起始行号, 级文本), ...])， + loops 为每个 while/until 循环的 (帧深度, 循环体是否 break 出本循环)。 + """ + found, loops, frames, modes, heredocs = [], [], [_Frame(1, 0)], ["cmd"], [] line, word, i, n = 1, "", 0, len(src) def emit(s): @@ -51,27 +62,77 @@ def pipelines(src): f.stage.append(s) def finish(f): - if f.stages: - found.append(f.stages + [(f.start, "".join(f.stage).strip())]) + text = "".join(f.stage).strip() + if f.stages or text: + found.append((f.depth, f.stages + [(f.start, text)])) f.stages, f.stage, f.start = [], [], line + def push(closer): + modes.append("cmd") + frames.append(_Frame(line, len(frames), closer)) + + def pop(): + finish(frames[-1]) + f = frames.pop() + modes.pop() + if f.closer == "done": + loops.append((f.depth, f.broken)) + def case_state(): cases = frames[-1].cases return cases[-1] if cases else None + def command_position(w): + j = i - len(w) + while j > 0 and src[j - 1] in " \t": + j -= 1 + if j == 0 or src[j - 1] in ";\n&|(": + return True + k = j + while k > 0 and src[k - 1] not in BOUNDARY: + k -= 1 + return src[k:j] in ("do", "then", "else", "{", "!") + + def break_out(levels): + for f in reversed(frames): + nested = f.dos - (f.closer == "done" and f.dos > 0) + if levels <= nested: + return + levels -= nested + if f.closer == "done": + f.broken = True + levels -= 1 + if levels <= 0: + return + elif f.closer != "}": + return + def keyword(w): - cases = frames[-1].cases + top = frames[-1] + cases = top.cases if w == "case": cases.append("in") elif w == "in" and case_state() == "in": cases[-1] = "pattern" elif w == "esac" and cases: cases.pop() + elif case_state() == "pattern": + pass + elif w in ("while", "until") and "".join(top.stage).strip() == w: + push("done") + elif w == "do" and command_position(w): + top.dos += 1 + elif w == "done" and command_position(w): + top.dos -= 1 + if top.closer == "done" and top.dos <= 0: + pop() + elif w == "break" and command_position(w): + m = BREAK_ARG.match(src, i) + break_out(int(m.group(1)) if m else 1) - def push_cmd(opener): + def push_cmd(opener, closer=")"): emit(opener) - modes.append("cmd") - frames.append(_Frame(line)) + push(closer) while i < n: c, nxt = src[i], src[i + 1 : i + 2] @@ -159,10 +220,15 @@ def pipelines(src): elif c == ")": if case_state() == "pattern": top.cases[-1] = "body" - elif len(frames) > 1: - finish(top) - frames.pop() - modes.pop() + elif top.closer == ")": + pop() + emit(c) + i += 1 + elif c == "{" and not word and nxt in (" ", "\t", "\n") and top.stages and not "".join(top.stage).strip(): + push_cmd(c, "}") + i += 1 + elif c == "}" and not word and top.closer == "}" and (nxt == "" or nxt in BOUNDARY) and command_position(""): + pop() emit(c) i += 1 elif c == "|": @@ -231,10 +297,17 @@ def pipelines(src): word += c i += 1 + if word: + keyword(word) if len(frames) != 1 or modes != ["cmd"]: raise ValueError(f"unbalanced shell lexing at EOF: modes={modes}") finish(frames[0]) - return found + return found, loops + + +def pipelines(src): + """返回 bash 源码中每条多级管道各级的 [(起始行号, 文本), ...]。""" + return [stages for _, stages in _lex(src)[0] if len(stages) > 1] def _grep_exits_early(words): @@ -255,7 +328,7 @@ def _grep_exits_early(words): def _awk_exits_early(prog): - """awk 程序在 END 块之外含 exit（含 BEGIN 内），即可能在读完输入前退出。""" + """awk 程序在 END 块之外含 exit 或 nextfile（含 BEGIN 内），即可能在读完输入前停止读取。""" while m := AWK_END.search(prog): depth, j = 0, m.end() - 1 while j < len(prog): @@ -267,14 +340,36 @@ def _awk_exits_early(prog): return bool(AWK_EXIT.search(prog)) +def _group_exits_early(group): + """( ... ) 或 { ...; } 内任一命令列表的首级（读组 stdin 者）是提前退出的读取器。""" + end = group.rfind(")" if group[0] == "(" else "}") + inner = group[1:end] if end > 0 else group[1:] + return any( + early_exit_reader(stages[0][1].lstrip(";&| \t\n")) for depth, stages in _lex(inner)[0] if depth == 0 + ) + + def early_exit_reader(stage): - """读取级读到结论后是否会提前关闭 stdin（grep -q/-m、head、sed q、awk exit、read）。""" - words = stage.split() - while words and (words[0] in ("{", "!", "time", "command", "env") or ASSIGNMENT.match(words[0])): - words.pop(0) - if not words: + """读取级读到结论后是否会提前关闭 stdin（grep -q/-m、head、sed q、awk exit/nextfile、read、 + 循环体 break 出本循环的 while/until）。 + + 先剥去 VAR=value 与 !/time/command/builtin/env/exec 前缀、取命令基名；子壳或花括组展开后逐条判定。 + """ + s = stage.strip() + while s: + if s[0] in "({": + return _group_exits_early(s) + m = WORD.match(s) + cmd = m.group(1).rsplit("/", 1)[-1] + if not (ASSIGNMENT.match(m.group(1)) or cmd in PREFIXES): + break + s = s[m.end() :] + if not s: return False - cmd, rest = words[0], " ".join(words[1:]) + if cmd in ("while", "until"): + return any(broken for depth, broken in _lex(s)[1] if depth == 1) + words = [cmd, *s[m.end() :].split()] + rest = " ".join(words[1:]) if cmd in ("grep", "egrep", "fgrep"): return _grep_exits_early(words) if cmd in ("head", "read"): @@ -330,6 +425,42 @@ class ShellPipeHazardTests(unittest.TestCase): with self.subTest(snippet=snippet): self.assertEqual(hazards(snippet + "\n"), [(1, stage)], snippet) + def test_wrapped_and_prefixed_readers_flagged_and_harmful(self): + """子壳/花括组、awk nextfile、带 break 的 while read、绝对路径与 exec 前缀：检测器命中，且 pipefail 下确实报 141。""" + grep = shutil.which("grep") or "/usr/bin/grep" + shapes = ( + ("seq 1 200000 | (head -1)", "(head -1)"), + ("seq 1 200000 | ( head -1 )", "( head -1 )"), + ("seq 1 200000 | { head -1; }", "{ head -1; }"), + ("seq 1 200000 | awk '$1 == 7 { print; nextfile }'", "awk '$1 == 7 { print; nextfile }'"), + ("seq 1 200000 | while read -r l; do break; done", "while read -r l; do break; done"), + (f"seq 1 200000 | {grep} -q 1", f"{grep} -q 1"), + ("seq 1 200000 | exec grep -q 1", "exec grep -q 1"), + ) + with tempfile.TemporaryDirectory(dir="/tmp") as tmp: + for snippet, stage in shapes: + with self.subTest(snippet=snippet): + self.assertEqual(hazards(snippet + "\n"), [(1, stage)], snippet) + script = Path(tmp) / "shape.sh" + script.write_text(f"set -euo pipefail\n{snippet}\necho survived\n", encoding="utf-8") + run = subprocess.run(["bash", str(script)], capture_output=True, text=True) + self.assertEqual(run.returncode, 141, run) + self.assertNotIn("survived", run.stdout) + + def test_detector_flags_group_loop_and_prefix_variants(self): + for snippet, stage in ( + ("seq 1 200000 | {\n head -1\n} | cat", "{\n head -1\n}"), + ("if seq 1 200000 | (grep -q 1); then :; fi", "(grep -q 1)"), + ("seq 1 200000 | (cd /tmp && head -1)", "(cd /tmp && head -1)"), + ("seq 1 200000 | while read -r l\ndo\n [ \"$l\" = 7 ] && break\ndone", "while read -r l\ndo\n [ \"$l\" = 7 ] && break\ndone"), + ("seq 1 200000 | until ! read -r l; do break; done", "until ! read -r l; do break; done"), + ("seq 1 200000 | while read -r l; do for x in a; do break 2; done; done", "while read -r l; do for x in a; do break 2; done; done"), + ("seq 1 200000 | /usr/bin/env LC_ALL=C grep -q 1", "/usr/bin/env LC_ALL=C grep -q 1"), + ("seq 1 200000 | exec /usr/bin/head -1", "exec /usr/bin/head -1"), + ): + with self.subTest(snippet=snippet): + self.assertEqual(hazards(snippet + "\n"), [(1, stage)], snippet) + def test_detector_accepts_safe_forms(self): src = "\n".join(( "set -euo pipefail", @@ -345,6 +476,14 @@ class ShellPipeHazardTests(unittest.TestCase): "seq 1 200000 | awk '{n++} END {exit !n}'", "seq 1 200000 | awk '$1 == 7 { f = 1 } END { exit !f }' && echo hit", "seq 1 200000 | while read -r l; do :; done", + "seq 1 200000 | while read -r l; do [ \"$l\" = 7 ] && continue; done", + "seq 1 200000 | while read -r l; do for x in a b; do break; done; done", + "while read -r l; do break; done < <(seq 1 200000)", + "seq 1 200000 | awk '{n++} END { nextfile }'", + "seq 1 200000 | (cat)", + "seq 1 200000 | { cat; }", + "seq 1 200000 | env FOO=1 command cat", + "seq 1 200000 | time builtin cat >/dev/null", "seq 1 200000 | LC_ALL=C grep -c 1", "case \"$w\" in --auth|--config) grep -q x <<<\"$w\" ;; esac", "cat <<'EOF'",