在做 CI 流水线、运维平台和 AI 编程助手的过程中我处理过不少因为脚本命令不规范而引发的事故。最常见的问题不是开发者不认真而是人工 review shell 命令的上限太低几百行脚本里很难快速判断一条组合命令到底会不会删库、会不会泄露敏感信息、会不会被中间人利用。后来我把命令审查流程改成了自动完成先用 AST 解析命令的语法结构再用多个 subagent 做专项风险检测最后统一输出风险报告。这套方案在不少项目里都能直接落地也是现在 AI 编程助手和 Agent 工具链中很常见的“Auto Review 主从 agent”设计思路。本文会完整拆解这套 Auto Review 工具的工程实现覆盖 AST 解析、规则引擎、subagent 调度、误报漏报处理以及生产落地的注意事项。内容对做运维平台、CI 安全、AI Agent 工具链的后端工程师和安全同学比较友好也希望帮助想理解 subagent 主从架构的 AI 应用开发者建立整体认识。文章中的演示命令需要放在本地测试环境验证不要直接在生产环境执行。1. 为什么 Shell 命令需要 Auto Review1.1 人工 Review 的极限先聊一个实际场景。你在代码评审里看到这样的命令rm -rf /var/lib/app/data service app restart如果这条命令出现在高权限运维脚本中评审人需要确认/var/lib/app/data是否是可删除目录、目录名是否拼写正确、有没有备份机制、执行用户是否具备相应权限。单个命令还能靠人观察但真实场景往往是几十行 shell 混在一起tar -czf backup.tar.gz /opt/app/data rsync -av --delete /opt/app/ backup.tar.gz /backup/ curl -s http://example.com/install.sh | bash这时候靠肉眼找出风险点就非常吃力。curl | bash这类命令本身可能是合法的自动化安装流程也可能被中间人篡改rm -rf后面跟的路径如果被错误变量展开很可能直接变成删根。人工 review 的效率和准确率都不够支撑这种检查频率较高的场景。Auto Review 要解决的就是这个问题把命令作为程序输入自动分析它的语法结构、依赖关系、权限影响和敏感信息输出结构化的风险报告让人只做最终决策。1.2 正则黑名单方案的不足很多早期方案会用正则做黑名单匹配比如“包含rm -rf就告警”。这种方案的问题非常明显。第一是误报高。脚本注释里可能写了# 不要执行 rm -rf正则照样命中字符串里可能只是把某条命令写入日志并不会真正执行。第二是漏报严重。攻击者可以用变量拼接、反引号替换、转义符号、heredoc 等方式绕过简单正则。rrm ${r} -rf /tmp/test上面这段命令的意图用正则很难精确识别但是人一看就知道它在删除目录。正则在处理这类语法关系时没有上下文能力而 AST 解析可以。1.3 新方案AST 解析 Subagent 双引擎ASTAbstract Syntax Tree抽象语法树解析把命令文本转换成语法树。在语法树里命令名和参数是结构化节点注释、字符串、管道、重定向、子命令都有了明确的父子关系。规则引擎可以基于这些节点做精确检测这是正则难以做到的。但 AST 只能解决“这条命令是什么”的问题不能完全解决“这条命令是否值得警惕”的问题。因为风险评估往往需要多个维度命令本身是否危险、是否包含硬编码密钥、是否在管道中被直接执行、是否修改了关键权限。每个维度单独维护一套规则会越来越臃肿。Subagent 在这里就派上了用场。它把不同的审查维度拆成独立的 agent主 agent 负责任务调度和汇总subagent 各管一块。换一个角度看subagent 本质上就是另一种形式的高级“工具”工具返回结构化结果subagent 同样返回结构化结果区别只是 subagent 内部可以做比单次工具调用更复杂的推理和上下文判断。2. 核心概念AST、Auto Review 与 Subagent2.1 Auto Review 是什么Auto Review自动审查指用程序替代人工完成代码、脚本、命令、配置文件等对象的审查动作。它的输入通常是待审查文本输出是一份问题清单每一项包括风险等级、问题位置、原因说明和修改建议。在 shell 命令审查这个场景里Auto Review 要回答的核心问题包括这条命令是否包含删除、格式化、关机等高危操作。是否使用了未经验证的远程脚本执行方式。是否涉及敏感信息的硬编码。是否对文件权限、属主做了不安全的变更。是否存在可以被注入或利用的变量拼接过。和常见的静态代码扫描类似Auto Review 追求的是“确定性优先”。它能通过规则确定的部分直接给出结论不能确定的部分再交给更复杂的推理模块也就是 subagent 或大模型去补充判断。2.2 AST 解析能解决什么问题AST 解析的核心价值是“结构”。以echo rm -rf /为例正则看到字符串rm -rf /会直接告警但 AST 解析后我们知道它只是一个 echo 命令的字符串参数不会被真正执行告警等级应该大幅降低甚至忽略。以curl http://x.com/a.sh | bash为例AST 解析后能识别出这里有两个命令节点curl和bash并通过管道节点连接。规则引擎可以针对“管道后面的 bash/sh 执行了前面命令的输出”这一模式做专项检测这种检测在纯文本正则里实现起来很别扭。再来看重定向。echo root:newpass /etc/shadow这条命令在 AST 里体现为“echo 命令 重定向到 /etc/shadow 文件”我们可以精确地定位被写入的路径再结合路径敏感度判断风险而不是把所有包含/etc/shadow的命令都一刀切。AST 解析的意义不是让规则变复杂而是让规则变得可以被理解、被解释、被维护。2.3 Subagent 与主从架构另类的“工具”调用近年来多 Agent 系统逐渐从“一个 Agent 干所有事”演进到“主从模式”也就是一个主 Agentsupervisor负责任务规划、调度、结果汇总多个 subagent 分别执行某个具体子任务。一个经常被忽略的认知是subagent 本质上就是另一种 tool 调用方式。传统上的 tool 是函数调用输入参数、输出固定结构的 JSONsubagent 虽然有独立的推理过程和上下文但在主 Agent 视角来看它同样遵守“输入任务、输出结构化结果”的协议。把它当工具设计反而更容易控制边界。按工作方式划分subagent 大体有几种类型类型特点典型使用场景任务委派型主 agent 把子任务交给 subagent拿回结果代码审查、文档总结工具封装型subagent 内部封装多个 tool对外只暴露统一接口数据库查询、命令执行编排流程型多个 subagent 按固定流程串联或并行执行自动化测试流水线高并发分摊型把大批量任务分给多个同构 subagent 并行处理日志分析、批量审查本文的 Auto Review 工具主要使用“任务委派型 高并发分摊型”的组合主 agent 负责把完整命令拆成多个审查维度每个 subagent 负责一个专项检测最后统一汇总。3. 环境准备与项目结构3.1 运行环境本文示例使用 Python 编写主要依赖 bashlex 库做 AST 解析。运行环境建议如下操作系统Linux 或 macOSWindows 下建议通过 WSL 运行因为 bashlex 解析依赖 bash 语法特性。Python 版本3.9 及以上。命令行工具pip 用于安装依赖python 直接运行脚本。bashlex 是一个把 bash 命令解析为 AST 的第三方库不是官方库所以需要注意版本兼容性。本文示例不写死版本你安装时以实际环境中pip install bashlex得到的版本为准。如果解析复杂命令时出现兼容问题可以固定在较新的版本上再验证。3.2 安装依赖创建一个专门的项目目录然后在虚拟环境中安装 bashlexmkdir auto-review cd auto-review python3 -m venv venv source venv/bin/activate pip install bashlex安装完成后可以用一个最简单的命令验证库是否可用python -c import bashlex; print(bashlex.parse(echo hello))如果能够输出解析结果对象说明环境正常。3.3 项目目录设计为了让代码结构清晰我们把它拆成 4 个模块auto-review/ ├── main.py # 主入口命令行交互 ├── ast_parser.py # AST 解析与命令节点提取 ├── rules.py # 风险规则与 Finding 数据结构 ├── agents.py # Subagent 定义 └── orchestrator.py # 主 Agent 调度与报告汇总当然小规模工具也可以把所有代码写进一个文件。拆开主要是为了体现分层思路方便后续扩展新规则或新 subagent。4. 第一层实现基于 AST 的 Shell 命令解析4.1 解析一个简单命令先写一个最小验证看看 bashlex 如何工作import bashlex command echo hello rm -rf /tmp/test parts bashlex.parse(command) for node in parts: print(node.kind, node)运行后输出中可以看到根节点可能是list里面包含command和operator等节点。对这条命令来说echo hello是一个 command 节点是 operator 节点rm -rf /tmp/test是另一个 command 节点。这个结构就是 AST。有了它我们就能按节点类型做规则检测。4.2 遍历 AST 并提取命令节点不同的 bash 命令结构不同比如if语句、for循环、管道、子命令替换。如果我们写死只处理list和command很容易漏掉复杂命令。所以更好的方式是写一个通用的遍历函数尽可能覆盖常见节点属性。# 文件路径auto-review/ast_parser.py import bashlex def walk(node): 递归遍历 AST 节点逐个 yield 出来。 yield node attrs (parts, list, command, then, else, items) for attr in attrs: child getattr(node, attr, None) if child is None: continue if isinstance(child, list): for c in child: yield from walk(c) else: yield from walk(child) def extract_command_parts(node): 从 command 节点中提取命令名和参数。 返回 (command_name, arguments) 元组。 如果节点不是 command 类型返回 (None, [])。 if getattr(node, kind, None) ! command: return None, [] command_name None arguments [] for part in getattr(node, parts, []) or []: if getattr(part, kind, None) ! word: continue word_text getattr(part, word, ) if command_name is None: command_name word_text else: arguments.append(word_text) return command_name, arguments这段代码里要注意几点walk函数尝试从parts、list、command等属性中递归寻找子节点。bashlex 不同版本对节点内部属性的命名可能有差异所以用getattr兜底。extract_command_parts只处理kind command的节点并且只把kind word的子节点当作命令名和参数忽略重定向、赋值等结构。变量赋值如DATABASE_URLxxx在 bash 里不算 command 节点所以不会被误识别成一条执行命令。4.3 为什么不直接用字符串切分你可能会说把命令按空格切分不就能拿到命令名和参数了吗对于简单的echo hello确实可以但遇到引号、转义、管道、重定向之后就不行了。echo hello world | tr \n按空格切分会把hello和world当成两个独立词实际上它们是同一个字符串参数。AST 能正确识别引号包裹的完整 word这样规则引擎判断参数时就不容易误报。再比如sh -c rm -rf /tmp/app字符串切分会把rm当作sh的第一个参数但 AST 能识别出这是sh -c后面的命令字符串从而进一步解析字符串内部的命令。虽然第一层 AST 只会把整个字符串当成一个 wordsubagent 层可以补充“是否需要展开子命令再审查”的逻辑。5. 第二层实现规则引擎与 Subagent 专项审查5.1 定义统一的 Subagent 调用协议前面说过subagent 本质上可以当作一种“高级工具”来调用。给所有 subagent 定义一个统一的调用协议是主从架构里最容易踩坑的地方也是最重要的一部分。我们定义一个最小协议# 文件路径auto-review/agents.py class SubAgent: 所有 subagent 的基类。 name base-agent def invoke(self, command, command_nodes): 所有 subagent 对外暴露的统一入口。 command: 原始命令文本。 command_nodes: 已提取的 (命令名, 参数列表) 节点列表。 返回一个 list[Finding]。 return [] def __repr__(self): return fSubAgent {self.name}主 agent 只依赖invoke方法不需要知道 subagent 内部用的是规则、模型还是其他工具。这样一来以后想换掉某个检测逻辑只需要替换对应的 subagent 实例主调度器不用改。5.2 风险命令检测 Subagent第一个 subagent 负责检测已知的危险命令。它遍历所有 command 节点针对命令名做判断然后结合参数上下文给出风险结论。# 文件路径auto-review/rules.py from dataclasses import dataclass from typing import Optional dataclass class Finding: severity: str # high / medium / low rule: str # 规则名称 command: str # 触发风险的命令名 message: str # 风险说明 suggestion: str # 修改建议 # 危险命令及需要重点关注的参数模式 DANGEROUS_COMMANDS { rm: {high: [-rf], medium: [-r, -f]}, mkfs: {high: []}, dd: {high: [of/dev/]}, shutdown: {high: [now], medium: [-h]}, reboot: {high: []}, chmod: {high: [777], medium: [666, aw]}, chown: {medium: []}, curl: {medium: []}, wget: {medium: []}, eval: {high: []}, exec: {medium: []}, } REMOTE_PIPELINE_COMMANDS {curl, wget}这个规则的思路不是“看到rm就告警”而是根据参数组合判断。rm file.txt可能只是删除普通文件风险中等rm -rf /var/lib/app风险就很高。dd本身是磁盘工具但只有出现of/dev/这种向设备文件写入的用法时才需要高等级告警。对应的 subagent 实现# 文件路径auto-review/agents.py from rules import DANGEROUS_COMMANDS, REMOTE_PIPELINE_COMMANDS, Finding class DangerousCommandAgent(SubAgent): 检测危险命令及其危险参数组合。 name dangerous-command-agent def invoke(self, command, command_nodes): findings [] for cmd_name, args in command_nodes: if cmd_name not in DANGEROUS_COMMANDS: continue rule_config DANGEROUS_COMMANDS[cmd_name] joined_args .join(args) # 高等级命中 high 参数模式 for high_part in rule_config.get(high, []): if high_part and high_part in joined_args: findings.append( Finding( severityhigh, ruledangerous-command-high, commandcmd_name, messagef命令 {cmd_name} 使用了高危参数 {high_part}, suggestionf确认 {cmd_name} 的操作范围补充备份和审批流程, ) ) break else: # 中等级命中 medium 参数模式 for medium_part in rule_config.get(medium, []): if medium_part and medium_part in joined_args: findings.append( Finding( severitymedium, ruledangerous-command-medium, commandcmd_name, messagef命令 {cmd_name} 使用了需要注意的参数 {medium_part}, suggestion确认操作目标是否可恢复, ) ) break return findings这段代码说明了一个核心设计规则引擎只负责“确定性判断”不对命令做语义理解之外的过度推断。命中高风险参数就报 high否则降级为 medium 或忽略。5.3 敏感信息检测 Subagent第二个 subagent 专门扫描参数中的敏感信息。这类检测基于正则但作用于 AST 解析后的参数文本而不是原始字符串。这样做的好处是注释里的AKIA...不会被误报只有真正作为参数或赋值内容的字符串才会被检查。# 文件路径auto-review/agents.py import re class SensitiveInfoAgent(SubAgent): 检测硬编码的密钥、密码和 Token。 name sensitive-info-agent PATTERNS [ (aws-access-key, re.compile(rAKIA[0-9A-Z]{16})), (private-key, re.compile(r-----BEGIN (RSA |EC |OPENSSH )?PRIVATE KEY-----)), (password-pair, re.compile(r(password|passwd|pwd)\s*[:]\s*\S, re.IGNORECASE)), (generic-token, re.compile(r(token|api[_-]?key|secret)\s*[:]\s*\S, re.IGNORECASE)), ] def invoke(self, command, command_nodes): findings [] for cmd_name, args in command_nodes: for arg in args: for rule_name, pattern in self.PATTERNS: if pattern.search(arg): findings.append( Finding( severityhigh, rulerule_name, commandcmd_name, messagef检测到疑似敏感信息: {rule_name}, suggestion不要在命令行直接传递密钥改用环境变量或密钥管理服务, ) ) return findings注意这里为了可演示没有把 “检测到敏感信息” 的原始内容直接输出避免在日志里再次泄露密钥。工程实践中也应该这样处理只输出规则类型和位置不回显敏感内容本身。5.4 管道与组合命令风险 Subagent第三个 subagent 检测管道组合命令风险比如curl | bash、wget | sh这种模式在自动化安装中常用但同时也给了远程命令执行攻击机会。如果 URL 不可信风险等级应该很高。由于第一层 AST 只提取了命令名和参数丢失了管道连接关系我们需要在 subagent 里重新遍历 AST 节点并识别管道结构。bashlex 中管道节点通常是pipe类型命令节点之间有明确的顺序关系。# 文件路径auto-review/agents.py import bashlex from ast_parser import walk class PipelineRiskAgent(SubAgent): 检测 curl/wget 输出直接交给 bash/sh 执行的风险。 这类模式被称为 remote script execution。 name pipeline-risk-agent def invoke(self, command, command_nodes): findings [] findings.extend(self._detect_pipeline(command)) return findings def _detect_pipeline(self, command): findings [] try: nodes list(bashlex.parse(command)) except Exception: return findings pipeline_results [] for root_node in nodes: for node in walk(root_node): if getattr(node, kind, None) ! pipe: continue left_cmd, right_cmd self._extract_pipeline_sides(node) if left_cmd and right_cmd: pipeline_results.append((left_cmd, right_cmd)) for left_cmd, right_cmd in pipeline_results: if left_cmd in REMOTE_PIPELINE_COMMANDS and right_cmd in {bash, sh, zsh, dash}: findings.append( Finding( severityhigh, ruleremote-pipeline-execute, commandf{left_cmd} | {right_cmd}, message远程下载内容被直接交给 shell 执行存在供应链攻击风险, suggestion优先使用签名校验或改为先下载校验哈希后再执行, ) ) return findings def _extract_pipeline_sides(self, pipe_node): 从 pipe 节点中提取左右两边的命令名。 parts getattr(pipe_node, parts, []) or [] cmd_names [] for part in parts: if getattr(part, kind, None) command: words [ p.word for p in getattr(part, parts, []) or [] if getattr(p, kind, None) word ] cmd_names.append(words[0] if words else None) elif getattr(part, kind, None) in (word, operator): continue else: continue if len(cmd_names) 2: return cmd_names[0], cmd_names[-1] return None, None这个 subagent 展示了主从架构的一个关键优点主 agent 不需要理解管道逻辑只需要把完整的 AST 上下文传给管道 subagentsubagent 自己在内部完成解析和判断。这符合“subagent 是带上下文的工具”这一设计思想。5.5 主 Agent 调度与报告汇总主 agent 不让 subagent 之间互相感知它只负责三件事解析原始命令提取 command 节点。把同一份命令和节点列表分发给所有 subagent。汇总每个 subagent 返回的 Finding按风险等级排序输出最终报告。# 文件路径auto-review/orchestrator.py from ast_parser import extract_command_parts, walk from rules import Finding class ReviewOrchestrator: 主 Agent负责调度所有 subagent并汇总最终风险报告。 def __init__(self, subagents): self.subagents subagents def review(self, command): # 1. 解析并提取命令节点 command_nodes self._parse_command_nodes(command) # 2. 并发或顺序调用所有 subagent all_findings [] for agent in self.subagents: findings agent.invoke(command, command_nodes) all_findings.extend(findings) # 3. 排序高 中 低 severity_order {high: 0, medium: 1, low: 2} all_findings.sort(keylambda f: severity_order.get(f.severity, 3)) return all_findings def _parse_command_nodes(self, command): import bashlex command_nodes [] try: nodes list(bashlex.parse(command)) except Exception: # 解析失败时返回空列表由 subagent 各自处理 return command_nodes for root_node in nodes: for node in walk(root_node): cmd_name, args extract_command_parts(node) if cmd_name: command_nodes.append((cmd_name, args)) return command_nodes这段代码有几个值得注意的设计review方法是主 agent 对外开放的唯一接口。_parse_command_nodes负责把 AST 解析成统一的(cmd_name, args)列表。解析失败时不直接抛异常而是返回空列表让 subagent 在内部决定是忽略还是告警。这样可以避免一条无法解析的命令阻断整个审查流程。真实生产环境中多个 subagent 可以并行执行这里为了演示保持顺序调用。如果 subagent 内部接入大模型调用耗时较长改成线程池或异步任务队列会更合适。6. 完整实战整合 Auto Review 工具6.1 完整代码现在把前面所有模块整合到一起。为了便于复制运行这里给出一个完整可运行的main.py包含所有类定义。如果你的项目想拆文件可以按照第 3.3 节的目录拆分。# 文件路径auto-review/main.py Shell 命令 Auto Review 工具 用法 python main.py import re from dataclasses import dataclass from typing import List, Tuple import bashlex # --------------------------------------------------------------- # 1. 数据模型 # --------------------------------------------------------------- dataclass class Finding: severity: str rule: str command: str message: str suggestion: str # --------------------------------------------------------------- # 2. AST 解析层 # --------------------------------------------------------------- def walk(node): yield node attrs (parts, list, command, then, else, items) for attr in attrs: child getattr(node, attr, None) if child is None: continue if isinstance(child, list): for c in child: yield from walk(c) else: yield from walk(child) def extract_command_parts(node): if getattr(node, kind, None) ! command: return None, [] command_name None arguments [] for part in getattr(node, parts, []) or []: if getattr(part, kind, None) ! word: continue word_text getattr(part, word, ) if command_name is None: command_name word_text else: arguments.append(word_text) return command_name, arguments # --------------------------------------------------------------- # 3. 规则与 Subagent # --------------------------------------------------------------- DANGEROUS_COMMANDS { rm: {high: [-rf], medium: [-r, -f]}, mkfs: {high: []}, dd: {high: [of/dev/]}, shutdown: {high: [now], medium: [-h]}, reboot: {high: []}, chmod: {high: [777], medium: [666, aw]}, chown: {medium: []}, curl: {medium: []}, wget: {medium: []}, eval: {high: []}, exec: {medium: []}, } REMOTE_PIPELINE_COMMANDS {curl, wget} class SubAgent: name base-agent def invoke(self, command: str, command_nodes: List[Tuple[str, List[str]]]) - List[Finding]: return [] class DangerousCommandAgent(SubAgent): name dangerous-command-agent def invoke(self, command, command_nodes): findings [] for cmd_name, args in command_nodes: if cmd_name not in DANGEROUS_COMMANDS: continue rule_config DANGEROUS_COMMANDS[cmd_name] joined_args .join(args) matched False for high_part in rule_config.get(high, []): if high_part and high_part in joined_args: findings.append( Finding( severityhigh, ruledangerous-command-high, commandcmd_name, messagef命令 {cmd_name} 使用了高危参数 {high_part}, suggestionf确认 {cmd_name} 的操作范围补充备份和审批流程, ) ) matched True break if matched: continue for medium_part in rule_config.get(medium, []): if medium_part and medium_part in joined_args: findings.append( Finding( severitymedium, ruledangerous-command-medium, commandcmd_name, messagef命令 {cmd_name} 使用了需要注意的参数 {medium_part}, suggestion确认操作目标是否可恢复, ) ) break return findings class SensitiveInfoAgent(SubAgent): name sensitive-info-agent PATTERNS [ (aws-access-key, re.compile(rAKIA[0-9A-Z]{16})), (private-key, re.compile(r-----BEGIN (RSA |EC |OPENSSH )?PRIVATE KEY-----)), (password-pair, re.compile(r(password|passwd|pwd)\s*[:]\s*\S, re.IGNORECASE)), (generic-token, re.compile(r(token|api[_-]?key|secret)\s*[:]\s*\S, re.IGNORECASE)), ] def invoke(self, command, command_nodes): findings [] for cmd_name, args in command_nodes: for arg in args: for rule_name, pattern in self.PATTERNS: if pattern.search(arg): findings.append( Finding( severityhigh, rulerule_name, commandcmd_name, messagef检测到疑似敏感信息: {rule_name}, suggestion不要在命令行直接传递密钥改用环境变量或密钥管理服务, ) ) return findings class PipelineRiskAgent(SubAgent): name pipeline-risk-agent def invoke(self, command, command_nodes): findings [] try: nodes list(bashlex.parse(command)) except Exception: return findings for root_node in nodes: for node in walk(root_node): if getattr(node, kind, None) ! pipe: continue pipeline_sides self._extract_pipeline_sides(node) if len(pipeline_sides) ! 2: continue left_cmd, right_cmd pipeline_sides if left_cmd in REMOTE_PIPELINE_COMMANDS and right_cmd in {bash, sh, zsh, dash}: findings.append( Finding( severityhigh, ruleremote-pipeline-execute, commandf{left_cmd} | {right_cmd}, message远程下载内容被直接交给 shell 执行存在供应链攻击风险, suggestion优先使用签名校验或改为先下载校验哈希后再执行, ) ) return findings def _extract_pipeline_sides(self, pipe_node): parts getattr(pipe_node, parts, []) or [] cmd_names [] for part in parts: if getattr(part, kind, None) command: words [ p.word for p in getattr(part, parts, []) or [] if getattr(p, kind, None) word ] cmd_names.append(words[0] if words else None) return cmd_names # --------------------------------------------------------------- # 4. 主 Agent 调度器 # --------------------------------------------------------------- class ReviewOrchestrator: def __init__(self, subagents): self.subagents subagents def review(self, command): command_nodes self._parse_command_nodes(command) all_findings [] for agent in self.subagents: findings agent.invoke(command, command_nodes) all_findings.extend(findings) severity_order {high: 0, medium: 1, low: 2} all_findings.sort(keylambda f: severity_order.get(f.severity, 3)) return all_findings def _parse_command_nodes(self, command): command_nodes [] try: nodes list(bashlex.parse(command)) except Exception: return command_nodes for root_node in nodes: for node in walk(root_node): cmd_name, args extract_command_parts(node) if cmd_name: command_nodes.append((cmd_name, args)) return command_nodes # --------------------------------------------------------------- # 5. 主入口 # --------------------------------------------------------------- def print_report(command, findings): print( * 60) print(待审查命令) print(command) print( * 60) if not findings: print(未发现明显风险。) return for idx, finding in enumerate(findings, start1): print(f[{idx}] 等级: {finding.severity.upper()}) print(f 规则: {finding.rule}) print(f 命令: {finding.command}) print(f 说明: {finding.message}) print(f 建议: {finding.suggestion}) high_count sum(1 for f in findings if f.severity high) print(- * 60) print(f共发现 {len(findings)} 个问题其中高危 {high_count} 个。) def main(): subagents [ DangerousCommandAgent(), SensitiveInfoAgent(), PipelineRiskAgent(), ] orchestrator ReviewOrchestrator(subagents) test_command sudo rm -rf /var/lib/app/data curl -s http://example.com/install.sh | bash echo AWS_KEYAKIAIOSFODNN7EXAMPLE .env chmod 777 /opt/data findings orchestrator.review(test_command) print_report(test_command, findings) if __name__ __main__: main()注意代码中的AKIAIOSFODNN7EXAMPLE是 AWS 官方文档使用的示例访问密钥占位符不代表真实密钥。真实项目中如果用密钥检测规则需要在测试环境用自己生成的模拟值验证。6.2 运行效果演示保存文件后在项目目录执行python main.py预期输出大致如下 待审查命令 sudo rm -rf /var/lib/app/data curl -s http://example.com/install.sh | bash echo AWS_KEYAKIAIOSFODNN7EXAMPLE .env chmod 777 /opt/data [1] 等级: HIGH 规则: dangerous-command-high 命令: rm 说明: 命令 rm 使用了高危参数 -rf 建议: 确认 rm 的操作范围补充备份和审批流程 [2] 等级: HIGH 规则: remote-pipeline-execute 命令: curl | bash 说明: 远程下载内容被直接交给 shell 执行存在供应链攻击风险 建议: 优先使用签名校验或改为先下载校验哈希后再执行 [3] 等级: HIGH 规则: aws-access-key 命令: echo 说明: 检测到疑似敏感信息: aws-access-key 建议: 不要在命令行直接传递密钥改用环境变量或密钥管理服务 [4] 等级: HIGH 规则: dangerous-command-high 命令: chmod 说明: 命令 chmod 使用了高危参数 777 建议: 确认 chmod 的操作范围补充备份和审批流程 ------------------------------------------------------------ 共发现 4 个问题其中高危 4 个。当然你本机实际输出可能因为 bashlex 版本和 Python 版本不同有细微差别。如果curl后面的 URL 没有触发告警先检查 URL 是否被 AST 正确识别为curl的参数。6.3 结果说明与扩展方向从运行结果可以看到4 条命令里有 4 个高危问题且每个问题都有明确的规则名和修改建议。这就是 Auto Review 工具的最终效果把模糊的“感觉不安全”变成结构化的风险清单交给人和流程去决策。扩展方向可以从两个维度考虑新增规则比如检测mv移动系统文件、kill终止关键进程、systemctl重启服务等。增强 subagent 能力在 subagent 内部接入大模型推理让它可以理解命令语义比如判断find / -name *.log -exec rm {} \\;这类组合命令是否危险。规则引擎处理不了的场景交给大模型 subagent 兜底。7. 常见问题与排查思路7.1 bashlex 解析失败或解析不完整问题现象常见原因解决思路bashlex.parse抛出异常命令语法不完整或包含 bashlex 不支持的语法用 try-except 包裹解析解析失败时降级为字符串检查复杂 if/case 命令提取不到子命令bashlex 版本旧节点属性命名不一致升级 bashlex并检查walk里的属性列表是否覆盖了新结构命令中包含中文或特殊编码终端编码问题统一使用 UTF-8必要时先import sys; sys.stdout.reconfigure(encodingutf-8)7.2 误报过高正常安全命令也被标记误报在高风险检查场景中很常见。需要优先检查是不是规则粒度太粗。rm命令本身会被标记为 medium因为rm file.txt也有一定风险。如果你们团队明确规定某些目录rm -rf是安全操作可以考虑在规则层增加路径白名单比如允许删除/tmp/myapp但不允许删除/var/lib或/etc。另一个做法是把规则输出从“告警”改成“提示”减少告警疲劳。真正的高危规则只保留少数几种。7.3 漏报绕过变量拼接或编码绕过AST 可以处理一部分变量展开问题但无法执行真正的 bash 展开。比如cmdrm $cmd -rf /tmp/test第一层 AST 只看到$cmd这个 word无法知道它等于什么。此时需要增加一层“变量展开”subagent在 AST 上下文里做常量传播分析。这个功能把简单规则引擎变成了数据流分析工具复杂度会明显上升但对抗绕过攻击时往往需要它。7.4 Subagent 执行阻塞或超时如果 subagent 内部调用大模型网络延迟或模型超时会拖慢整个 review 流程。在主从架构中建议给每个 subagent 设置独立的超时时间并用线程池或异步任务并发调用避免一个慢 subagent 阻塞所有审查。from concurrent.futures import ThreadPoolExecutor, as_completed with ThreadPoolExecutor(max_workerslen(subagents)) as executor: futures {executor.submit(agent.invoke, command, command_nodes): agent for agent in subagents} for future in as_completed(futures): findings future.result(timeout10) all_findings.extend(findings)超时时间需要结合你的 subagent 实际耗时调整不能盲目设置。8. 最佳实践与工程建议8.1 规则引擎优先大模型兜底Auto Review 工具最容易犯的错误是“处处都想用大模型”。大模型推理能力强但有延迟、有不确定性和成本。工程上更稳妥的做法是两层结构第一层用确定性规则引擎处理所有能确定判断的场景比如危险命令组合、已知密钥格式、远程管道执行。第二层用大模型或复杂推理 subagent 处理规则引擎覆盖不到的语义场景比如跨命令的数据流风险、模糊命令意图判断。这样既能保证大多数扫描在毫秒级完成又能保留处理长尾场景的能力。8.2 Subagent 划分粒度subagent 的划分不是越细越好。每个 subagent 都会增加调度开销和上下文传递成本。按照以下原则划分比较合适按审查维度划分例如“危险命令”“敏感信息”“管道风险”“权限变更”而不是按rm、mkfs、dd每个命令建一个 agent。每个 subagent 对外暴露统一的invoke接口内部实现可以自由替换。新规则先放进已有规则引擎规则复杂度超过一定阈值再考虑拆成新 subagent。本质上subagent 是“另类的 tool”这个观点在这里非常实用把它当工具管理控制输入输出协议才能让主从架构长期可维护。8.3 安全边界与权限控制Auto Review 工具通常运行在 CI 或运维平台上它自身的权限不应该过高。以下几点需要特别注意review 进程不要使用 root 权限执行建议使用独立低权限账号运行。如果要解析执行类命令只做解析不实际执行任何 shell 命令。报告输出中不要回显检测到的密钥原文防止日志泄露。对规则配置的变更要走版本管理修改危险命令表前先评审。对权限变更类命令chmod、chown的审查结果要结合目标路径的敏感度判断。生产环境路径和用户目录的风险等级应该不同这个映射关系建议单独维护配置文件。8.4 可观测性与性能优化生产落地时Auto Review 应该像其他服务一样具备可观测性记录每次审查的命令来源、耗时、命中规则数量。用指标统计高危命令比例观察团队整体的命令规范度变化。高危告警接入 IM 或审批流而不是只输出在日志里。批量审查场景使用线程池或消息队列避免单条命令阻塞流水线。性能方面AST 解析本身很快主要瓶颈在 subagent。如果 subagent 全部是规则引擎单条命令审查可以控制在几毫秒内。如果接入了大模型就要考虑缓存、批量聚合和超时降级策略。8.5 哪些命令不能只靠规则引擎判断最后说一个容易被忽略的点。规则引擎适合处理已知模式但find ... -exec rm、while read循环配合mv、echo $PASSWORD | base64 -d这类组合命令单纯靠第一层 AST 提取命令节点是不够的。它们往往需要跨节点分析甚至需要模拟变量传播。这类问题在工程里不要急着用复杂算法解决可以先用“风险命中提示 人工二次确认”的方式兜底。等项目积累到足够多案例之后再把高频场景固化成新的规则或 subagent。Auto Review 工具的演进过程本质上就是规则从简单到复杂、subagent 从单薄到健壮的过程。动手在自己项目的测试环境里跑一遍这个示例把几条自己的常用脚本丢进去看看结果再逐步补充规则和 subagent会比直接照搬代码更有收获。 Shell命令自动审查：AST解析与Subagent主从架构实战
在做 CI 流水线、运维平台和 AI 编程助手的过程中我处理过不少因为脚本命令不规范而引发的事故。最常见的问题不是开发者不认真而是人工 review shell 命令的上限太低几百行脚本里很难快速判断一条组合命令到底会不会删库、会不会泄露敏感信息、会不会被中间人利用。后来我把命令审查流程改成了自动完成先用 AST 解析命令的语法结构再用多个 subagent 做专项风险检测最后统一输出风险报告。这套方案在不少项目里都能直接落地也是现在 AI 编程助手和 Agent 工具链中很常见的“Auto Review 主从 agent”设计思路。本文会完整拆解这套 Auto Review 工具的工程实现覆盖 AST 解析、规则引擎、subagent 调度、误报漏报处理以及生产落地的注意事项。内容对做运维平台、CI 安全、AI Agent 工具链的后端工程师和安全同学比较友好也希望帮助想理解 subagent 主从架构的 AI 应用开发者建立整体认识。文章中的演示命令需要放在本地测试环境验证不要直接在生产环境执行。1. 为什么 Shell 命令需要 Auto Review1.1 人工 Review 的极限先聊一个实际场景。你在代码评审里看到这样的命令rm -rf /var/lib/app/data service app restart如果这条命令出现在高权限运维脚本中评审人需要确认/var/lib/app/data是否是可删除目录、目录名是否拼写正确、有没有备份机制、执行用户是否具备相应权限。单个命令还能靠人观察但真实场景往往是几十行 shell 混在一起tar -czf backup.tar.gz /opt/app/data rsync -av --delete /opt/app/ backup.tar.gz /backup/ curl -s http://example.com/install.sh | bash这时候靠肉眼找出风险点就非常吃力。curl | bash这类命令本身可能是合法的自动化安装流程也可能被中间人篡改rm -rf后面跟的路径如果被错误变量展开很可能直接变成删根。人工 review 的效率和准确率都不够支撑这种检查频率较高的场景。Auto Review 要解决的就是这个问题把命令作为程序输入自动分析它的语法结构、依赖关系、权限影响和敏感信息输出结构化的风险报告让人只做最终决策。1.2 正则黑名单方案的不足很多早期方案会用正则做黑名单匹配比如“包含rm -rf就告警”。这种方案的问题非常明显。第一是误报高。脚本注释里可能写了# 不要执行 rm -rf正则照样命中字符串里可能只是把某条命令写入日志并不会真正执行。第二是漏报严重。攻击者可以用变量拼接、反引号替换、转义符号、heredoc 等方式绕过简单正则。rrm ${r} -rf /tmp/test上面这段命令的意图用正则很难精确识别但是人一看就知道它在删除目录。正则在处理这类语法关系时没有上下文能力而 AST 解析可以。1.3 新方案AST 解析 Subagent 双引擎ASTAbstract Syntax Tree抽象语法树解析把命令文本转换成语法树。在语法树里命令名和参数是结构化节点注释、字符串、管道、重定向、子命令都有了明确的父子关系。规则引擎可以基于这些节点做精确检测这是正则难以做到的。但 AST 只能解决“这条命令是什么”的问题不能完全解决“这条命令是否值得警惕”的问题。因为风险评估往往需要多个维度命令本身是否危险、是否包含硬编码密钥、是否在管道中被直接执行、是否修改了关键权限。每个维度单独维护一套规则会越来越臃肿。Subagent 在这里就派上了用场。它把不同的审查维度拆成独立的 agent主 agent 负责任务调度和汇总subagent 各管一块。换一个角度看subagent 本质上就是另一种形式的高级“工具”工具返回结构化结果subagent 同样返回结构化结果区别只是 subagent 内部可以做比单次工具调用更复杂的推理和上下文判断。2. 核心概念AST、Auto Review 与 Subagent2.1 Auto Review 是什么Auto Review自动审查指用程序替代人工完成代码、脚本、命令、配置文件等对象的审查动作。它的输入通常是待审查文本输出是一份问题清单每一项包括风险等级、问题位置、原因说明和修改建议。在 shell 命令审查这个场景里Auto Review 要回答的核心问题包括这条命令是否包含删除、格式化、关机等高危操作。是否使用了未经验证的远程脚本执行方式。是否涉及敏感信息的硬编码。是否对文件权限、属主做了不安全的变更。是否存在可以被注入或利用的变量拼接过。和常见的静态代码扫描类似Auto Review 追求的是“确定性优先”。它能通过规则确定的部分直接给出结论不能确定的部分再交给更复杂的推理模块也就是 subagent 或大模型去补充判断。2.2 AST 解析能解决什么问题AST 解析的核心价值是“结构”。以echo rm -rf /为例正则看到字符串rm -rf /会直接告警但 AST 解析后我们知道它只是一个 echo 命令的字符串参数不会被真正执行告警等级应该大幅降低甚至忽略。以curl http://x.com/a.sh | bash为例AST 解析后能识别出这里有两个命令节点curl和bash并通过管道节点连接。规则引擎可以针对“管道后面的 bash/sh 执行了前面命令的输出”这一模式做专项检测这种检测在纯文本正则里实现起来很别扭。再来看重定向。echo root:newpass /etc/shadow这条命令在 AST 里体现为“echo 命令 重定向到 /etc/shadow 文件”我们可以精确地定位被写入的路径再结合路径敏感度判断风险而不是把所有包含/etc/shadow的命令都一刀切。AST 解析的意义不是让规则变复杂而是让规则变得可以被理解、被解释、被维护。2.3 Subagent 与主从架构另类的“工具”调用近年来多 Agent 系统逐渐从“一个 Agent 干所有事”演进到“主从模式”也就是一个主 Agentsupervisor负责任务规划、调度、结果汇总多个 subagent 分别执行某个具体子任务。一个经常被忽略的认知是subagent 本质上就是另一种 tool 调用方式。传统上的 tool 是函数调用输入参数、输出固定结构的 JSONsubagent 虽然有独立的推理过程和上下文但在主 Agent 视角来看它同样遵守“输入任务、输出结构化结果”的协议。把它当工具设计反而更容易控制边界。按工作方式划分subagent 大体有几种类型类型特点典型使用场景任务委派型主 agent 把子任务交给 subagent拿回结果代码审查、文档总结工具封装型subagent 内部封装多个 tool对外只暴露统一接口数据库查询、命令执行编排流程型多个 subagent 按固定流程串联或并行执行自动化测试流水线高并发分摊型把大批量任务分给多个同构 subagent 并行处理日志分析、批量审查本文的 Auto Review 工具主要使用“任务委派型 高并发分摊型”的组合主 agent 负责把完整命令拆成多个审查维度每个 subagent 负责一个专项检测最后统一汇总。3. 环境准备与项目结构3.1 运行环境本文示例使用 Python 编写主要依赖 bashlex 库做 AST 解析。运行环境建议如下操作系统Linux 或 macOSWindows 下建议通过 WSL 运行因为 bashlex 解析依赖 bash 语法特性。Python 版本3.9 及以上。命令行工具pip 用于安装依赖python 直接运行脚本。bashlex 是一个把 bash 命令解析为 AST 的第三方库不是官方库所以需要注意版本兼容性。本文示例不写死版本你安装时以实际环境中pip install bashlex得到的版本为准。如果解析复杂命令时出现兼容问题可以固定在较新的版本上再验证。3.2 安装依赖创建一个专门的项目目录然后在虚拟环境中安装 bashlexmkdir auto-review cd auto-review python3 -m venv venv source venv/bin/activate pip install bashlex安装完成后可以用一个最简单的命令验证库是否可用python -c import bashlex; print(bashlex.parse(echo hello))如果能够输出解析结果对象说明环境正常。3.3 项目目录设计为了让代码结构清晰我们把它拆成 4 个模块auto-review/ ├── main.py # 主入口命令行交互 ├── ast_parser.py # AST 解析与命令节点提取 ├── rules.py # 风险规则与 Finding 数据结构 ├── agents.py # Subagent 定义 └── orchestrator.py # 主 Agent 调度与报告汇总当然小规模工具也可以把所有代码写进一个文件。拆开主要是为了体现分层思路方便后续扩展新规则或新 subagent。4. 第一层实现基于 AST 的 Shell 命令解析4.1 解析一个简单命令先写一个最小验证看看 bashlex 如何工作import bashlex command echo hello rm -rf /tmp/test parts bashlex.parse(command) for node in parts: print(node.kind, node)运行后输出中可以看到根节点可能是list里面包含command和operator等节点。对这条命令来说echo hello是一个 command 节点是 operator 节点rm -rf /tmp/test是另一个 command 节点。这个结构就是 AST。有了它我们就能按节点类型做规则检测。4.2 遍历 AST 并提取命令节点不同的 bash 命令结构不同比如if语句、for循环、管道、子命令替换。如果我们写死只处理list和command很容易漏掉复杂命令。所以更好的方式是写一个通用的遍历函数尽可能覆盖常见节点属性。# 文件路径auto-review/ast_parser.py import bashlex def walk(node): 递归遍历 AST 节点逐个 yield 出来。 yield node attrs (parts, list, command, then, else, items) for attr in attrs: child getattr(node, attr, None) if child is None: continue if isinstance(child, list): for c in child: yield from walk(c) else: yield from walk(child) def extract_command_parts(node): 从 command 节点中提取命令名和参数。 返回 (command_name, arguments) 元组。 如果节点不是 command 类型返回 (None, [])。 if getattr(node, kind, None) ! command: return None, [] command_name None arguments [] for part in getattr(node, parts, []) or []: if getattr(part, kind, None) ! word: continue word_text getattr(part, word, ) if command_name is None: command_name word_text else: arguments.append(word_text) return command_name, arguments这段代码里要注意几点walk函数尝试从parts、list、command等属性中递归寻找子节点。bashlex 不同版本对节点内部属性的命名可能有差异所以用getattr兜底。extract_command_parts只处理kind command的节点并且只把kind word的子节点当作命令名和参数忽略重定向、赋值等结构。变量赋值如DATABASE_URLxxx在 bash 里不算 command 节点所以不会被误识别成一条执行命令。4.3 为什么不直接用字符串切分你可能会说把命令按空格切分不就能拿到命令名和参数了吗对于简单的echo hello确实可以但遇到引号、转义、管道、重定向之后就不行了。echo hello world | tr \n按空格切分会把hello和world当成两个独立词实际上它们是同一个字符串参数。AST 能正确识别引号包裹的完整 word这样规则引擎判断参数时就不容易误报。再比如sh -c rm -rf /tmp/app字符串切分会把rm当作sh的第一个参数但 AST 能识别出这是sh -c后面的命令字符串从而进一步解析字符串内部的命令。虽然第一层 AST 只会把整个字符串当成一个 wordsubagent 层可以补充“是否需要展开子命令再审查”的逻辑。5. 第二层实现规则引擎与 Subagent 专项审查5.1 定义统一的 Subagent 调用协议前面说过subagent 本质上可以当作一种“高级工具”来调用。给所有 subagent 定义一个统一的调用协议是主从架构里最容易踩坑的地方也是最重要的一部分。我们定义一个最小协议# 文件路径auto-review/agents.py class SubAgent: 所有 subagent 的基类。 name base-agent def invoke(self, command, command_nodes): 所有 subagent 对外暴露的统一入口。 command: 原始命令文本。 command_nodes: 已提取的 (命令名, 参数列表) 节点列表。 返回一个 list[Finding]。 return [] def __repr__(self): return fSubAgent {self.name}主 agent 只依赖invoke方法不需要知道 subagent 内部用的是规则、模型还是其他工具。这样一来以后想换掉某个检测逻辑只需要替换对应的 subagent 实例主调度器不用改。5.2 风险命令检测 Subagent第一个 subagent 负责检测已知的危险命令。它遍历所有 command 节点针对命令名做判断然后结合参数上下文给出风险结论。# 文件路径auto-review/rules.py from dataclasses import dataclass from typing import Optional dataclass class Finding: severity: str # high / medium / low rule: str # 规则名称 command: str # 触发风险的命令名 message: str # 风险说明 suggestion: str # 修改建议 # 危险命令及需要重点关注的参数模式 DANGEROUS_COMMANDS { rm: {high: [-rf], medium: [-r, -f]}, mkfs: {high: []}, dd: {high: [of/dev/]}, shutdown: {high: [now], medium: [-h]}, reboot: {high: []}, chmod: {high: [777], medium: [666, aw]}, chown: {medium: []}, curl: {medium: []}, wget: {medium: []}, eval: {high: []}, exec: {medium: []}, } REMOTE_PIPELINE_COMMANDS {curl, wget}这个规则的思路不是“看到rm就告警”而是根据参数组合判断。rm file.txt可能只是删除普通文件风险中等rm -rf /var/lib/app风险就很高。dd本身是磁盘工具但只有出现of/dev/这种向设备文件写入的用法时才需要高等级告警。对应的 subagent 实现# 文件路径auto-review/agents.py from rules import DANGEROUS_COMMANDS, REMOTE_PIPELINE_COMMANDS, Finding class DangerousCommandAgent(SubAgent): 检测危险命令及其危险参数组合。 name dangerous-command-agent def invoke(self, command, command_nodes): findings [] for cmd_name, args in command_nodes: if cmd_name not in DANGEROUS_COMMANDS: continue rule_config DANGEROUS_COMMANDS[cmd_name] joined_args .join(args) # 高等级命中 high 参数模式 for high_part in rule_config.get(high, []): if high_part and high_part in joined_args: findings.append( Finding( severityhigh, ruledangerous-command-high, commandcmd_name, messagef命令 {cmd_name} 使用了高危参数 {high_part}, suggestionf确认 {cmd_name} 的操作范围补充备份和审批流程, ) ) break else: # 中等级命中 medium 参数模式 for medium_part in rule_config.get(medium, []): if medium_part and medium_part in joined_args: findings.append( Finding( severitymedium, ruledangerous-command-medium, commandcmd_name, messagef命令 {cmd_name} 使用了需要注意的参数 {medium_part}, suggestion确认操作目标是否可恢复, ) ) break return findings这段代码说明了一个核心设计规则引擎只负责“确定性判断”不对命令做语义理解之外的过度推断。命中高风险参数就报 high否则降级为 medium 或忽略。5.3 敏感信息检测 Subagent第二个 subagent 专门扫描参数中的敏感信息。这类检测基于正则但作用于 AST 解析后的参数文本而不是原始字符串。这样做的好处是注释里的AKIA...不会被误报只有真正作为参数或赋值内容的字符串才会被检查。# 文件路径auto-review/agents.py import re class SensitiveInfoAgent(SubAgent): 检测硬编码的密钥、密码和 Token。 name sensitive-info-agent PATTERNS [ (aws-access-key, re.compile(rAKIA[0-9A-Z]{16})), (private-key, re.compile(r-----BEGIN (RSA |EC |OPENSSH )?PRIVATE KEY-----)), (password-pair, re.compile(r(password|passwd|pwd)\s*[:]\s*\S, re.IGNORECASE)), (generic-token, re.compile(r(token|api[_-]?key|secret)\s*[:]\s*\S, re.IGNORECASE)), ] def invoke(self, command, command_nodes): findings [] for cmd_name, args in command_nodes: for arg in args: for rule_name, pattern in self.PATTERNS: if pattern.search(arg): findings.append( Finding( severityhigh, rulerule_name, commandcmd_name, messagef检测到疑似敏感信息: {rule_name}, suggestion不要在命令行直接传递密钥改用环境变量或密钥管理服务, ) ) return findings注意这里为了可演示没有把 “检测到敏感信息” 的原始内容直接输出避免在日志里再次泄露密钥。工程实践中也应该这样处理只输出规则类型和位置不回显敏感内容本身。5.4 管道与组合命令风险 Subagent第三个 subagent 检测管道组合命令风险比如curl | bash、wget | sh这种模式在自动化安装中常用但同时也给了远程命令执行攻击机会。如果 URL 不可信风险等级应该很高。由于第一层 AST 只提取了命令名和参数丢失了管道连接关系我们需要在 subagent 里重新遍历 AST 节点并识别管道结构。bashlex 中管道节点通常是pipe类型命令节点之间有明确的顺序关系。# 文件路径auto-review/agents.py import bashlex from ast_parser import walk class PipelineRiskAgent(SubAgent): 检测 curl/wget 输出直接交给 bash/sh 执行的风险。 这类模式被称为 remote script execution。 name pipeline-risk-agent def invoke(self, command, command_nodes): findings [] findings.extend(self._detect_pipeline(command)) return findings def _detect_pipeline(self, command): findings [] try: nodes list(bashlex.parse(command)) except Exception: return findings pipeline_results [] for root_node in nodes: for node in walk(root_node): if getattr(node, kind, None) ! pipe: continue left_cmd, right_cmd self._extract_pipeline_sides(node) if left_cmd and right_cmd: pipeline_results.append((left_cmd, right_cmd)) for left_cmd, right_cmd in pipeline_results: if left_cmd in REMOTE_PIPELINE_COMMANDS and right_cmd in {bash, sh, zsh, dash}: findings.append( Finding( severityhigh, ruleremote-pipeline-execute, commandf{left_cmd} | {right_cmd}, message远程下载内容被直接交给 shell 执行存在供应链攻击风险, suggestion优先使用签名校验或改为先下载校验哈希后再执行, ) ) return findings def _extract_pipeline_sides(self, pipe_node): 从 pipe 节点中提取左右两边的命令名。 parts getattr(pipe_node, parts, []) or [] cmd_names [] for part in parts: if getattr(part, kind, None) command: words [ p.word for p in getattr(part, parts, []) or [] if getattr(p, kind, None) word ] cmd_names.append(words[0] if words else None) elif getattr(part, kind, None) in (word, operator): continue else: continue if len(cmd_names) 2: return cmd_names[0], cmd_names[-1] return None, None这个 subagent 展示了主从架构的一个关键优点主 agent 不需要理解管道逻辑只需要把完整的 AST 上下文传给管道 subagentsubagent 自己在内部完成解析和判断。这符合“subagent 是带上下文的工具”这一设计思想。5.5 主 Agent 调度与报告汇总主 agent 不让 subagent 之间互相感知它只负责三件事解析原始命令提取 command 节点。把同一份命令和节点列表分发给所有 subagent。汇总每个 subagent 返回的 Finding按风险等级排序输出最终报告。# 文件路径auto-review/orchestrator.py from ast_parser import extract_command_parts, walk from rules import Finding class ReviewOrchestrator: 主 Agent负责调度所有 subagent并汇总最终风险报告。 def __init__(self, subagents): self.subagents subagents def review(self, command): # 1. 解析并提取命令节点 command_nodes self._parse_command_nodes(command) # 2. 并发或顺序调用所有 subagent all_findings [] for agent in self.subagents: findings agent.invoke(command, command_nodes) all_findings.extend(findings) # 3. 排序高 中 低 severity_order {high: 0, medium: 1, low: 2} all_findings.sort(keylambda f: severity_order.get(f.severity, 3)) return all_findings def _parse_command_nodes(self, command): import bashlex command_nodes [] try: nodes list(bashlex.parse(command)) except Exception: # 解析失败时返回空列表由 subagent 各自处理 return command_nodes for root_node in nodes: for node in walk(root_node): cmd_name, args extract_command_parts(node) if cmd_name: command_nodes.append((cmd_name, args)) return command_nodes这段代码有几个值得注意的设计review方法是主 agent 对外开放的唯一接口。_parse_command_nodes负责把 AST 解析成统一的(cmd_name, args)列表。解析失败时不直接抛异常而是返回空列表让 subagent 在内部决定是忽略还是告警。这样可以避免一条无法解析的命令阻断整个审查流程。真实生产环境中多个 subagent 可以并行执行这里为了演示保持顺序调用。如果 subagent 内部接入大模型调用耗时较长改成线程池或异步任务队列会更合适。6. 完整实战整合 Auto Review 工具6.1 完整代码现在把前面所有模块整合到一起。为了便于复制运行这里给出一个完整可运行的main.py包含所有类定义。如果你的项目想拆文件可以按照第 3.3 节的目录拆分。# 文件路径auto-review/main.py Shell 命令 Auto Review 工具 用法 python main.py import re from dataclasses import dataclass from typing import List, Tuple import bashlex # --------------------------------------------------------------- # 1. 数据模型 # --------------------------------------------------------------- dataclass class Finding: severity: str rule: str command: str message: str suggestion: str # --------------------------------------------------------------- # 2. AST 解析层 # --------------------------------------------------------------- def walk(node): yield node attrs (parts, list, command, then, else, items) for attr in attrs: child getattr(node, attr, None) if child is None: continue if isinstance(child, list): for c in child: yield from walk(c) else: yield from walk(child) def extract_command_parts(node): if getattr(node, kind, None) ! command: return None, [] command_name None arguments [] for part in getattr(node, parts, []) or []: if getattr(part, kind, None) ! word: continue word_text getattr(part, word, ) if command_name is None: command_name word_text else: arguments.append(word_text) return command_name, arguments # --------------------------------------------------------------- # 3. 规则与 Subagent # --------------------------------------------------------------- DANGEROUS_COMMANDS { rm: {high: [-rf], medium: [-r, -f]}, mkfs: {high: []}, dd: {high: [of/dev/]}, shutdown: {high: [now], medium: [-h]}, reboot: {high: []}, chmod: {high: [777], medium: [666, aw]}, chown: {medium: []}, curl: {medium: []}, wget: {medium: []}, eval: {high: []}, exec: {medium: []}, } REMOTE_PIPELINE_COMMANDS {curl, wget} class SubAgent: name base-agent def invoke(self, command: str, command_nodes: List[Tuple[str, List[str]]]) - List[Finding]: return [] class DangerousCommandAgent(SubAgent): name dangerous-command-agent def invoke(self, command, command_nodes): findings [] for cmd_name, args in command_nodes: if cmd_name not in DANGEROUS_COMMANDS: continue rule_config DANGEROUS_COMMANDS[cmd_name] joined_args .join(args) matched False for high_part in rule_config.get(high, []): if high_part and high_part in joined_args: findings.append( Finding( severityhigh, ruledangerous-command-high, commandcmd_name, messagef命令 {cmd_name} 使用了高危参数 {high_part}, suggestionf确认 {cmd_name} 的操作范围补充备份和审批流程, ) ) matched True break if matched: continue for medium_part in rule_config.get(medium, []): if medium_part and medium_part in joined_args: findings.append( Finding( severitymedium, ruledangerous-command-medium, commandcmd_name, messagef命令 {cmd_name} 使用了需要注意的参数 {medium_part}, suggestion确认操作目标是否可恢复, ) ) break return findings class SensitiveInfoAgent(SubAgent): name sensitive-info-agent PATTERNS [ (aws-access-key, re.compile(rAKIA[0-9A-Z]{16})), (private-key, re.compile(r-----BEGIN (RSA |EC |OPENSSH )?PRIVATE KEY-----)), (password-pair, re.compile(r(password|passwd|pwd)\s*[:]\s*\S, re.IGNORECASE)), (generic-token, re.compile(r(token|api[_-]?key|secret)\s*[:]\s*\S, re.IGNORECASE)), ] def invoke(self, command, command_nodes): findings [] for cmd_name, args in command_nodes: for arg in args: for rule_name, pattern in self.PATTERNS: if pattern.search(arg): findings.append( Finding( severityhigh, rulerule_name, commandcmd_name, messagef检测到疑似敏感信息: {rule_name}, suggestion不要在命令行直接传递密钥改用环境变量或密钥管理服务, ) ) return findings class PipelineRiskAgent(SubAgent): name pipeline-risk-agent def invoke(self, command, command_nodes): findings [] try: nodes list(bashlex.parse(command)) except Exception: return findings for root_node in nodes: for node in walk(root_node): if getattr(node, kind, None) ! pipe: continue pipeline_sides self._extract_pipeline_sides(node) if len(pipeline_sides) ! 2: continue left_cmd, right_cmd pipeline_sides if left_cmd in REMOTE_PIPELINE_COMMANDS and right_cmd in {bash, sh, zsh, dash}: findings.append( Finding( severityhigh, ruleremote-pipeline-execute, commandf{left_cmd} | {right_cmd}, message远程下载内容被直接交给 shell 执行存在供应链攻击风险, suggestion优先使用签名校验或改为先下载校验哈希后再执行, ) ) return findings def _extract_pipeline_sides(self, pipe_node): parts getattr(pipe_node, parts, []) or [] cmd_names [] for part in parts: if getattr(part, kind, None) command: words [ p.word for p in getattr(part, parts, []) or [] if getattr(p, kind, None) word ] cmd_names.append(words[0] if words else None) return cmd_names # --------------------------------------------------------------- # 4. 主 Agent 调度器 # --------------------------------------------------------------- class ReviewOrchestrator: def __init__(self, subagents): self.subagents subagents def review(self, command): command_nodes self._parse_command_nodes(command) all_findings [] for agent in self.subagents: findings agent.invoke(command, command_nodes) all_findings.extend(findings) severity_order {high: 0, medium: 1, low: 2} all_findings.sort(keylambda f: severity_order.get(f.severity, 3)) return all_findings def _parse_command_nodes(self, command): command_nodes [] try: nodes list(bashlex.parse(command)) except Exception: return command_nodes for root_node in nodes: for node in walk(root_node): cmd_name, args extract_command_parts(node) if cmd_name: command_nodes.append((cmd_name, args)) return command_nodes # --------------------------------------------------------------- # 5. 主入口 # --------------------------------------------------------------- def print_report(command, findings): print( * 60) print(待审查命令) print(command) print( * 60) if not findings: print(未发现明显风险。) return for idx, finding in enumerate(findings, start1): print(f[{idx}] 等级: {finding.severity.upper()}) print(f 规则: {finding.rule}) print(f 命令: {finding.command}) print(f 说明: {finding.message}) print(f 建议: {finding.suggestion}) high_count sum(1 for f in findings if f.severity high) print(- * 60) print(f共发现 {len(findings)} 个问题其中高危 {high_count} 个。) def main(): subagents [ DangerousCommandAgent(), SensitiveInfoAgent(), PipelineRiskAgent(), ] orchestrator ReviewOrchestrator(subagents) test_command sudo rm -rf /var/lib/app/data curl -s http://example.com/install.sh | bash echo AWS_KEYAKIAIOSFODNN7EXAMPLE .env chmod 777 /opt/data findings orchestrator.review(test_command) print_report(test_command, findings) if __name__ __main__: main()注意代码中的AKIAIOSFODNN7EXAMPLE是 AWS 官方文档使用的示例访问密钥占位符不代表真实密钥。真实项目中如果用密钥检测规则需要在测试环境用自己生成的模拟值验证。6.2 运行效果演示保存文件后在项目目录执行python main.py预期输出大致如下 待审查命令 sudo rm -rf /var/lib/app/data curl -s http://example.com/install.sh | bash echo AWS_KEYAKIAIOSFODNN7EXAMPLE .env chmod 777 /opt/data [1] 等级: HIGH 规则: dangerous-command-high 命令: rm 说明: 命令 rm 使用了高危参数 -rf 建议: 确认 rm 的操作范围补充备份和审批流程 [2] 等级: HIGH 规则: remote-pipeline-execute 命令: curl | bash 说明: 远程下载内容被直接交给 shell 执行存在供应链攻击风险 建议: 优先使用签名校验或改为先下载校验哈希后再执行 [3] 等级: HIGH 规则: aws-access-key 命令: echo 说明: 检测到疑似敏感信息: aws-access-key 建议: 不要在命令行直接传递密钥改用环境变量或密钥管理服务 [4] 等级: HIGH 规则: dangerous-command-high 命令: chmod 说明: 命令 chmod 使用了高危参数 777 建议: 确认 chmod 的操作范围补充备份和审批流程 ------------------------------------------------------------ 共发现 4 个问题其中高危 4 个。当然你本机实际输出可能因为 bashlex 版本和 Python 版本不同有细微差别。如果curl后面的 URL 没有触发告警先检查 URL 是否被 AST 正确识别为curl的参数。6.3 结果说明与扩展方向从运行结果可以看到4 条命令里有 4 个高危问题且每个问题都有明确的规则名和修改建议。这就是 Auto Review 工具的最终效果把模糊的“感觉不安全”变成结构化的风险清单交给人和流程去决策。扩展方向可以从两个维度考虑新增规则比如检测mv移动系统文件、kill终止关键进程、systemctl重启服务等。增强 subagent 能力在 subagent 内部接入大模型推理让它可以理解命令语义比如判断find / -name *.log -exec rm {} \\;这类组合命令是否危险。规则引擎处理不了的场景交给大模型 subagent 兜底。7. 常见问题与排查思路7.1 bashlex 解析失败或解析不完整问题现象常见原因解决思路bashlex.parse抛出异常命令语法不完整或包含 bashlex 不支持的语法用 try-except 包裹解析解析失败时降级为字符串检查复杂 if/case 命令提取不到子命令bashlex 版本旧节点属性命名不一致升级 bashlex并检查walk里的属性列表是否覆盖了新结构命令中包含中文或特殊编码终端编码问题统一使用 UTF-8必要时先import sys; sys.stdout.reconfigure(encodingutf-8)7.2 误报过高正常安全命令也被标记误报在高风险检查场景中很常见。需要优先检查是不是规则粒度太粗。rm命令本身会被标记为 medium因为rm file.txt也有一定风险。如果你们团队明确规定某些目录rm -rf是安全操作可以考虑在规则层增加路径白名单比如允许删除/tmp/myapp但不允许删除/var/lib或/etc。另一个做法是把规则输出从“告警”改成“提示”减少告警疲劳。真正的高危规则只保留少数几种。7.3 漏报绕过变量拼接或编码绕过AST 可以处理一部分变量展开问题但无法执行真正的 bash 展开。比如cmdrm $cmd -rf /tmp/test第一层 AST 只看到$cmd这个 word无法知道它等于什么。此时需要增加一层“变量展开”subagent在 AST 上下文里做常量传播分析。这个功能把简单规则引擎变成了数据流分析工具复杂度会明显上升但对抗绕过攻击时往往需要它。7.4 Subagent 执行阻塞或超时如果 subagent 内部调用大模型网络延迟或模型超时会拖慢整个 review 流程。在主从架构中建议给每个 subagent 设置独立的超时时间并用线程池或异步任务并发调用避免一个慢 subagent 阻塞所有审查。from concurrent.futures import ThreadPoolExecutor, as_completed with ThreadPoolExecutor(max_workerslen(subagents)) as executor: futures {executor.submit(agent.invoke, command, command_nodes): agent for agent in subagents} for future in as_completed(futures): findings future.result(timeout10) all_findings.extend(findings)超时时间需要结合你的 subagent 实际耗时调整不能盲目设置。8. 最佳实践与工程建议8.1 规则引擎优先大模型兜底Auto Review 工具最容易犯的错误是“处处都想用大模型”。大模型推理能力强但有延迟、有不确定性和成本。工程上更稳妥的做法是两层结构第一层用确定性规则引擎处理所有能确定判断的场景比如危险命令组合、已知密钥格式、远程管道执行。第二层用大模型或复杂推理 subagent 处理规则引擎覆盖不到的语义场景比如跨命令的数据流风险、模糊命令意图判断。这样既能保证大多数扫描在毫秒级完成又能保留处理长尾场景的能力。8.2 Subagent 划分粒度subagent 的划分不是越细越好。每个 subagent 都会增加调度开销和上下文传递成本。按照以下原则划分比较合适按审查维度划分例如“危险命令”“敏感信息”“管道风险”“权限变更”而不是按rm、mkfs、dd每个命令建一个 agent。每个 subagent 对外暴露统一的invoke接口内部实现可以自由替换。新规则先放进已有规则引擎规则复杂度超过一定阈值再考虑拆成新 subagent。本质上subagent 是“另类的 tool”这个观点在这里非常实用把它当工具管理控制输入输出协议才能让主从架构长期可维护。8.3 安全边界与权限控制Auto Review 工具通常运行在 CI 或运维平台上它自身的权限不应该过高。以下几点需要特别注意review 进程不要使用 root 权限执行建议使用独立低权限账号运行。如果要解析执行类命令只做解析不实际执行任何 shell 命令。报告输出中不要回显检测到的密钥原文防止日志泄露。对规则配置的变更要走版本管理修改危险命令表前先评审。对权限变更类命令chmod、chown的审查结果要结合目标路径的敏感度判断。生产环境路径和用户目录的风险等级应该不同这个映射关系建议单独维护配置文件。8.4 可观测性与性能优化生产落地时Auto Review 应该像其他服务一样具备可观测性记录每次审查的命令来源、耗时、命中规则数量。用指标统计高危命令比例观察团队整体的命令规范度变化。高危告警接入 IM 或审批流而不是只输出在日志里。批量审查场景使用线程池或消息队列避免单条命令阻塞流水线。性能方面AST 解析本身很快主要瓶颈在 subagent。如果 subagent 全部是规则引擎单条命令审查可以控制在几毫秒内。如果接入了大模型就要考虑缓存、批量聚合和超时降级策略。8.5 哪些命令不能只靠规则引擎判断最后说一个容易被忽略的点。规则引擎适合处理已知模式但find ... -exec rm、while read循环配合mv、echo $PASSWORD | base64 -d这类组合命令单纯靠第一层 AST 提取命令节点是不够的。它们往往需要跨节点分析甚至需要模拟变量传播。这类问题在工程里不要急着用复杂算法解决可以先用“风险命中提示 人工二次确认”的方式兜底。等项目积累到足够多案例之后再把高频场景固化成新的规则或 subagent。Auto Review 工具的演进过程本质上就是规则从简单到复杂、subagent 从单薄到健壮的过程。动手在自己项目的测试环境里跑一遍这个示例把几条自己的常用脚本丢进去看看结果再逐步补充规则和 subagent会比直接照搬代码更有收获。