638 lines
26 KiB
Python
638 lines
26 KiB
Python
"""托管 shell 脚本：管道任一读取级不得是提前退出的读取器（READER_FAMILIES 声明的各族：grep -q/-m、head、sed q、
|
||
awk exit/nextfile、read、mapfile/readarray -n、带 break 的 while/until 循环；含被子壳或花括组包裹、带绝对路径或
|
||
exec 等前缀的写法）。守护的承诺是“已声明的各族在发现到的脚本中处处被拦截”，而非“理解任意 bash 命令”。
|
||
|
||
pipefail 下读取器先退出，上游生产者写满 64 KiB 管道缓冲后收到 SIGPIPE，管道状态报 141，
|
||
`if producer | grep -q PAT` 与 `if ! ...` 的判定被静默翻转；读取器位于管道中段
|
||
（`producer | head -1 | sed ...`）时同理。改用 here-string、进程替换或 sed 打印即退出，
|
||
读取器的结论才不会丢。
|
||
"""
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
SCRIPT_GLOBS = (":(glob)scripts/**/*.sh", ":(glob)scripts/hooks/*")
|
||
EXTRA_SCRIPTS = ("docker/verify-env.sh", "tests/load/collaboration/run_pgbench.sh")
|
||
# 发现到但检测器无法正确解析的脚本：路径 -> 理由。条目须仍被发现，否则视为过期。
|
||
EXEMPT = {}
|
||
|
||
READER_FAMILIES = (
|
||
("grep", "-q/--quiet/--silent、-m/--max-count（含 egrep/fgrep）", "得出匹配结论或读满 N 条匹配即退出"),
|
||
("head", "任意用法", "输出前 N 行/字节后即退出"),
|
||
("sed", "q/Q 命令", "执行 q/Q 即停止读取"),
|
||
("awk", "END 块之外的 exit/nextfile（含 gawk/mawk/nawk）", "exit 立即结束，nextfile 放弃当前输入剩余部分"),
|
||
("read", "任意用法（含 -n N/--chars N）", "只读一行，-n N 读满 N 个字符即返回"),
|
||
("mapfile", "-n/-n<N>/-n <N>/--n/--numlines（含 readarray）", "读满 N 行即停止；长选项被 bash 拒绝时一字节不读即退出"),
|
||
("while", "循环体 break 出本循环（含 until）", "break 后不再读取剩余输入"),
|
||
)
|
||
"""守护覆盖的提前退出读取器族：(命令, 触发选项, 提前退出的原因)。
|
||
|
||
检测器保证这些族（含子壳/花括组包裹、VAR= 与 env/exec 等前缀、绝对路径写法）位于管道读取级时处处被拦截；
|
||
不带触发选项的 mapfile/readarray、无 break 的 while read 读完全部输入，不在拦截之列。
|
||
"""
|
||
|
||
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"\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")
|
||
MAPFILE_ARG_OPTS = set("dOsuCc")
|
||
BOUNDARY = set(" \t\n;&|()<>")
|
||
|
||
|
||
def managed_scripts():
|
||
"""git 跟踪的托管 shell 脚本；未跟踪副本（如 runtime/ 下的陈旧克隆）不在其中。"""
|
||
out = subprocess.run(
|
||
["git", "-C", str(ROOT), "ls-files", "-z", "--", *SCRIPT_GLOBS, *EXTRA_SCRIPTS],
|
||
check=True, capture_output=True, text=True,
|
||
).stdout
|
||
return sorted(p for p in out.split("\0") if p)
|
||
|
||
|
||
class _Frame:
|
||
"""一层命令上下文（顶层、$(...)、(...)、<(...)、管道级的 { ...; }、while/until 循环）内正在累积的管道。"""
|
||
|
||
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 _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):
|
||
for f in frames:
|
||
f.stage.append(s)
|
||
|
||
def finish(f):
|
||
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):
|
||
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, closer=")"):
|
||
emit(opener)
|
||
push(closer)
|
||
|
||
while i < n:
|
||
c, nxt = src[i], src[i + 1 : i + 2]
|
||
mode = modes[-1]
|
||
if mode in ("sq", "ansi"):
|
||
if c == "\\" and mode == "ansi":
|
||
emit(src[i : i + 2])
|
||
i += 2
|
||
continue
|
||
emit(c)
|
||
line += c == "\n"
|
||
if c == "'":
|
||
modes.pop()
|
||
i += 1
|
||
continue
|
||
if mode in ("dq", "brace"):
|
||
if c == "\\":
|
||
emit(src[i : i + 2])
|
||
line += nxt == "\n"
|
||
i += 2
|
||
elif c == "$" and nxt == "(":
|
||
push_cmd("$(")
|
||
i += 2
|
||
elif c == "$" and nxt == "{":
|
||
emit("${")
|
||
modes.append("brace")
|
||
i += 2
|
||
elif c == '"':
|
||
emit(c)
|
||
if mode == "dq":
|
||
modes.pop()
|
||
else:
|
||
modes.append("dq")
|
||
i += 1
|
||
elif c == "}" and mode == "brace":
|
||
emit(c)
|
||
modes.pop()
|
||
i += 1
|
||
else:
|
||
emit(c)
|
||
line += c == "\n"
|
||
i += 1
|
||
continue
|
||
|
||
if c in BOUNDARY and word:
|
||
keyword(word)
|
||
word = ""
|
||
top = frames[-1]
|
||
if c == "\\":
|
||
if nxt == "\n":
|
||
emit(" ")
|
||
line += 1
|
||
else:
|
||
emit(src[i : i + 2])
|
||
word += "\0"
|
||
i += 2
|
||
elif c == "#" and (i == 0 or src[i - 1] in " \t\n;&|("):
|
||
j = src.find("\n", i)
|
||
i = n if j < 0 else j
|
||
elif c == "'" or (c == "$" and nxt == "'"):
|
||
emit(src[i : i + 2] if c == "$" else c)
|
||
modes.append("ansi" if c == "$" else "sq")
|
||
word += "\0"
|
||
i += 2 if c == "$" else 1
|
||
elif c == '"':
|
||
emit(c)
|
||
modes.append("dq")
|
||
word += "\0"
|
||
i += 1
|
||
elif c == "$" and nxt == "{":
|
||
emit("${")
|
||
modes.append("brace")
|
||
word += "\0"
|
||
i += 2
|
||
elif c == "$" and nxt == "(":
|
||
push_cmd("$(")
|
||
word += "\0"
|
||
i += 2
|
||
elif c == "(":
|
||
if case_state() == "pattern":
|
||
emit(c)
|
||
else:
|
||
push_cmd(c)
|
||
i += 1
|
||
elif c == ")":
|
||
if case_state() == "pattern":
|
||
top.cases[-1] = "body"
|
||
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 == "|":
|
||
if nxt == "|":
|
||
finish(top)
|
||
emit("||")
|
||
i += 2
|
||
elif case_state() == "pattern":
|
||
emit(c)
|
||
i += 1
|
||
else:
|
||
top.stages.append((top.start, "".join(top.stage).strip()))
|
||
emit(c)
|
||
top.stage, top.start = [], line
|
||
i += 2 if nxt == "&" else 1
|
||
elif c == ";":
|
||
if src.startswith(";;", i) or src.startswith(";&", i):
|
||
if case_state() == "body":
|
||
top.cases[-1] = "pattern"
|
||
k = 3 if src.startswith(";;&", i) else 2
|
||
else:
|
||
k = 1
|
||
finish(top)
|
||
emit(src[i : i + k])
|
||
i += k
|
||
elif c == "&":
|
||
if nxt == "&":
|
||
finish(top)
|
||
emit("&&")
|
||
i += 2
|
||
elif nxt == ">" or (i > 0 and src[i - 1] in "<>"):
|
||
emit(c)
|
||
i += 1
|
||
else:
|
||
finish(top)
|
||
emit(c)
|
||
i += 1
|
||
elif c == "<" and src.startswith("<<", i):
|
||
m = None if src.startswith("<<<", i) else HEREDOC.match(src, i)
|
||
if m:
|
||
heredocs.append((m.group(3), m.group(1) == "-"))
|
||
emit(m.group(0))
|
||
i = m.end()
|
||
else:
|
||
k = 3 if src.startswith("<<<", i) else 2
|
||
emit(src[i : i + k])
|
||
i += k
|
||
elif c == "\n":
|
||
line += 1
|
||
i += 1
|
||
for term, dash in heredocs:
|
||
while i < n:
|
||
j = src.find("\n", i)
|
||
body = src[i : n if j < 0 else j]
|
||
i = n if j < 0 else j + 1
|
||
line += 1
|
||
if (body.lstrip("\t") if dash else body) == term:
|
||
break
|
||
heredocs = []
|
||
if not (top.stages and not "".join(top.stage).strip()):
|
||
finish(top)
|
||
emit("\n")
|
||
else:
|
||
emit(c)
|
||
if c not in BOUNDARY:
|
||
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, loops
|
||
|
||
|
||
def pipelines(src):
|
||
"""返回 bash 源码中每条多级管道各级的 [(起始行号, 文本), ...]。"""
|
||
return [stages for _, stages in _lex(src)[0] if len(stages) > 1]
|
||
|
||
|
||
def _grep_exits_early(words):
|
||
for w in words[1:]:
|
||
w = w.strip("'\"")
|
||
if w == "--":
|
||
break
|
||
if w.startswith("--"):
|
||
if w.split("=")[0] in ("--quiet", "--silent", "--max-count"):
|
||
return True
|
||
elif w.startswith("-") and len(w) > 1:
|
||
for ch in w[1:]:
|
||
if ch in "qm":
|
||
return True
|
||
if ch in GREP_ARG_OPTS:
|
||
break
|
||
return False
|
||
|
||
|
||
def _mapfile_exits_early(words):
|
||
"""mapfile/readarray 带 -n（-n<N>、-n <N>、选项簇内的 n）或 --n/--numlines 时只读 N 行即停止。"""
|
||
skip = False
|
||
for w in words[1:]:
|
||
w = w.strip("'\"")
|
||
if skip:
|
||
skip = False
|
||
continue
|
||
if w == "--" or not w.startswith("-") or len(w) == 1:
|
||
break
|
||
if w.startswith("--"):
|
||
if w.split("=")[0] in ("--n", "--numlines"):
|
||
return True
|
||
continue
|
||
for k, ch in enumerate(w[1:], 1):
|
||
if ch == "n":
|
||
return True
|
||
if ch in MAPFILE_ARG_OPTS:
|
||
skip = k == len(w) - 1
|
||
break
|
||
return False
|
||
|
||
|
||
def _awk_exits_early(prog):
|
||
"""awk 程序在 END 块之外含 exit 或 nextfile（含 BEGIN 内），即可能在读完输入前停止读取。"""
|
||
while m := AWK_END.search(prog):
|
||
depth, j = 0, m.end() - 1
|
||
while j < len(prog):
|
||
depth += {"{": 1, "}": -1}.get(prog[j], 0)
|
||
j += 1
|
||
if depth == 0:
|
||
break
|
||
prog = prog[: m.start()] + prog[j:]
|
||
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（READER_FAMILIES：grep -q/-m、head、sed q、awk exit/nextfile、read、
|
||
mapfile/readarray -n、循环体 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
|
||
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"):
|
||
return True
|
||
if cmd in ("mapfile", "readarray"):
|
||
return _mapfile_exits_early(words)
|
||
if cmd == "sed":
|
||
return bool(SED_QUIT.search(rest))
|
||
if cmd in ("awk", "gawk", "mawk", "nawk"):
|
||
return _awk_exits_early(rest)
|
||
return False
|
||
|
||
|
||
def hazards(src):
|
||
"""各管道第二级起（读 pipe 的各级）中提前退出的读取器，返回 (行号, 该级文本)。"""
|
||
return [
|
||
(line, text)
|
||
for stages in pipelines(src)
|
||
for line, text in stages[1:]
|
||
if early_exit_reader(text)
|
||
]
|
||
|
||
|
||
class ShellPipeHazardTests(unittest.TestCase):
|
||
def test_detector_flags_synthetic_hazard(self):
|
||
found = hazards("set -euo pipefail\nseq 1 200000 | grep -q 1\n")
|
||
self.assertEqual([line for line, _ in found], [2], found)
|
||
|
||
def test_detector_flags_hazard_shapes(self):
|
||
for snippet in (
|
||
'if ! seq 1 200000 | grep -q 1; then echo miss; fi',
|
||
'x="$(sed -n \'s/^k=//p\' "$f" | head -1)"',
|
||
'if ! "$bin" serve -h 2>&1 | grep -Eq -- "(^|[[:space:]])-{1,2}config"; then :; fi',
|
||
'echo "${r:+($(printf \'%s\' "$r" | head -c 120))}"',
|
||
'find . -type f \\\n | grep -m1 x',
|
||
"seq 1 200000 | sed -n '/1/{p;q;}'",
|
||
'seq 1 200000 | grep -qiE "a|b" || exit 1',
|
||
):
|
||
with self.subTest(snippet=snippet):
|
||
self.assertTrue(hazards(snippet + "\n"), snippet)
|
||
|
||
def test_detector_flags_newly_covered_shapes(self):
|
||
for snippet, stage in (
|
||
("seq 1 200000 | head -1 | sed 's/^/x/'", "head -1"),
|
||
("x=\"$(seq 1 200000 | head -n 3 | tr -d ' ')\"", "head -n 3"),
|
||
("seq 1 200000 | awk '{print $1; exit}'", "awk '{print $1; exit}'"),
|
||
("seq 1 200000 | awk 'BEGIN { exit 1 }'", "awk 'BEGIN { exit 1 }'"),
|
||
("seq 1 200000 | awk '/7/{f=1; exit} END{print f}' | cat", "awk '/7/{f=1; exit} END{print f}'"),
|
||
("seq 1 200000 | read -r first", "read -r first"),
|
||
("seq 1 200000 | IFS= read -r first", "IFS= read -r first"),
|
||
("if seq 1 200000 | LC_ALL=C grep -q 1; then :; fi", "LC_ALL=C grep -q 1"),
|
||
("seq 1 200000 | env LC_ALL=C grep -m1 1 | wc -l", "env LC_ALL=C grep -m1 1"),
|
||
("seq 1 200000 | grep -q 1 | cat", "grep -q 1"),
|
||
):
|
||
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_counted_readers_flagged_and_harmful(self):
|
||
"""mapfile/readarray -n 与 read -n：检测器命中，且 pipefail 下脚本中止（读满 N 后退出报 141，长选项被拒报非零）。"""
|
||
shapes = (
|
||
("seq 1 200000 | mapfile -n 1 arr", "mapfile -n 1 arr", 141),
|
||
("seq 1 200000 | readarray --numlines 1 arr", "readarray --numlines 1 arr", None),
|
||
("seq 1 200000 | read -n 5 -r line", "read -n 5 -r line", 141),
|
||
)
|
||
with tempfile.TemporaryDirectory(dir="/tmp") as tmp:
|
||
for snippet, stage, rc 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.assertNotEqual(run.returncode, 0, run)
|
||
if rc is not None:
|
||
self.assertEqual(run.returncode, rc, run)
|
||
self.assertNotIn("survived", run.stdout)
|
||
script = Path(tmp) / "safe.sh"
|
||
script.write_text(
|
||
"set -euo pipefail\nseq 1 200000 > in.txt\nmapfile arr < in.txt\nread -r line < in.txt\n"
|
||
"seq 1 200000 | mapfile -t arr\necho \"survived ${#arr[@]} $line\"\n",
|
||
encoding="utf-8",
|
||
)
|
||
run = subprocess.run(["bash", str(script)], capture_output=True, text=True, cwd=tmp)
|
||
self.assertEqual((run.returncode, run.stdout), (0, "survived 200000 1\n"), run)
|
||
self.assertEqual(hazards(script.read_text(encoding="utf-8")), [])
|
||
|
||
def test_detector_flags_counted_reader_variants(self):
|
||
for snippet, stage in (
|
||
("seq 1 200000 | mapfile -n1 arr", "mapfile -n1 arr"),
|
||
("seq 1 200000 | mapfile -t -n 3 arr", "mapfile -t -n 3 arr"),
|
||
("seq 1 200000 | readarray -tn2 arr", "readarray -tn2 arr"),
|
||
("seq 1 200000 | mapfile -u 0 -n 1 arr", "mapfile -u 0 -n 1 arr"),
|
||
("seq 1 200000 | readarray --n 1 arr", "readarray --n 1 arr"),
|
||
("seq 1 200000 | builtin mapfile -n 1 arr", "builtin mapfile -n 1 arr"),
|
||
("seq 1 200000 | { mapfile -n 1 arr; }", "{ mapfile -n 1 arr; }"),
|
||
("seq 1 200000 | read --chars 5 line", "read --chars 5 line"),
|
||
):
|
||
with self.subTest(snippet=snippet):
|
||
self.assertEqual(hazards(snippet + "\n"), [(1, stage)], snippet)
|
||
|
||
def test_declared_reader_families_each_flagged(self):
|
||
"""READER_FAMILIES 是守护的契约：非空，且每个声明族至少有一个被检测器命中的合成用例。"""
|
||
cases = {
|
||
"grep": "seq 1 200000 | grep -q 1",
|
||
"head": "seq 1 200000 | head -1",
|
||
"sed": "seq 1 200000 | sed -n '/1/{p;q;}'",
|
||
"awk": "seq 1 200000 | awk '{print; exit}'",
|
||
"read": "seq 1 200000 | read -n 5 -r line",
|
||
"mapfile": "seq 1 200000 | mapfile -n 1 arr",
|
||
"while": "seq 1 200000 | while read -r l; do break; done",
|
||
}
|
||
self.assertTrue(READER_FAMILIES)
|
||
for family in READER_FAMILIES:
|
||
command = family[0]
|
||
with self.subTest(family=command):
|
||
self.assertEqual(len(family), 3)
|
||
self.assertTrue(all(family), family)
|
||
self.assertIn(command, cases, "declared family lacks a synthetic case")
|
||
self.assertTrue(hazards(cases[command] + "\n"), cases[command])
|
||
|
||
def test_detector_accepts_safe_forms(self):
|
||
src = "\n".join((
|
||
"set -euo pipefail",
|
||
"# seq 1 200000 | grep -q 1",
|
||
"grep -q 1 < <(seq 1 200000)",
|
||
'grep -q x <<<"$v"',
|
||
"x=\"$(sed -n '/^k=/{s///p;q;}' \"$f\")\"",
|
||
"x=\"$(sed -n '/^k=/{s///p;q;}' < <(seq 1 200000))\"",
|
||
"awk '{print $1; exit}' < <(seq 1 200000)",
|
||
"head -c 160 < <(seq 1 200000)",
|
||
"ps -eo comm | sed -n '1,12p'",
|
||
"seq 1 200000 | sed -n '1,12p' | tr -d ' '",
|
||
"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",
|
||
"mapfile arr < file",
|
||
"read -r line",
|
||
"read -n 5 -r line < file",
|
||
"seq 1 200000 | mapfile arr",
|
||
"seq 1 200000 | mapfile -t arr",
|
||
"seq 1 200000 | readarray -u 0 -t arr",
|
||
"seq 1 200000 | mapfile -d '' arr",
|
||
"seq 1 200000 | mapfile -t -- arr",
|
||
"case \"$w\" in --auth|--config) grep -q x <<<\"$w\" ;; esac",
|
||
"cat <<'EOF'",
|
||
"seq 1 200000 | grep -q 1",
|
||
"EOF",
|
||
"",
|
||
))
|
||
self.assertEqual(hazards(src), [])
|
||
|
||
def test_discovery_covers_managed_scripts(self):
|
||
found = managed_scripts()
|
||
for rel in (
|
||
"scripts/svc.sh",
|
||
"scripts/task.sh",
|
||
"scripts/e2e.sh",
|
||
"scripts/e2e-duties.sh",
|
||
"scripts/lint-boundaries.sh",
|
||
"scripts/hooks/pre-commit",
|
||
*EXTRA_SCRIPTS,
|
||
):
|
||
self.assertIn(rel, found)
|
||
self.assertFalse([p for p in found if p.startswith("runtime/")], found)
|
||
self.assertFalse(set(EXEMPT) - set(found), "stale EXEMPT entries")
|
||
|
||
def test_managed_scripts_have_no_early_exit_pipe_reader(self):
|
||
for rel in managed_scripts():
|
||
if rel in EXEMPT:
|
||
continue
|
||
with self.subTest(script=rel):
|
||
path = ROOT / rel
|
||
self.assertTrue(path.is_file(), f"{rel} missing")
|
||
found = hazards(path.read_text(encoding="utf-8"))
|
||
self.assertEqual(
|
||
found, [], "\n".join(f"{rel}:{line}: | {text}" for line, text in found)
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|