weichen/Agent-System-Dev
1
0
Fork 0
Code Issues 1 Pull Requests 2 Actions 2 Packages Projects Releases Wiki Activity

agent-exec transport 7909cc2e3bc15bc12eb30244170e2eac/9723af6b7b77

Browse Source
This commit is contained in:
agent-exec-transport
2026-10-21 02:42:12 +00:00
parent 4e484a547d
commit 22dd37524c
1 changed files with 164 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_shell_pipe_hazard.py
+164 -25
View File
@@ -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'",