企业官网建设流程全解析

1. 这不是代码问题，是 exec 签名在替你背锅

“Agent 执行命令为什么总卡死”——这句话我去年在三个不同客户现场都听过。第一次是在某金融风控平台的灰度环境里，Agent 调用subprocess.run(['adb', 'shell', 'ps'], timeout=5)，90% 的概率卡在Running状态，超时后抛出TimeoutExpired；第二次是车载系统 OTA 升级 Agent，在执行sh /data/local/tmp/install.sh时，进程状态永远停在D（uninterruptible sleep），strace一看卡在wait4()；第三次最诡异：同一段 Python 代码，在本地 macOS 上秒级返回，在 Ubuntu 容器里却稳定卡住 30 秒以上，ps auxf显示子进程明明已退出，父进程就是不收尸。

所有这些现象，表面看是“Agent 卡死”，但真正罪魁祸首，是那一行看似无害的exec函数签名：

subprocess.run(args, *, stdin=None, stdout=None, stderr=None, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None, text=None, env=None, universal_newlines=None, **other_kwargs)

这行签名里藏着八个被绝大多数 Agent 开发者忽略的隐性契约。它们不报错、不崩溃、不抛异常，只默默让进程挂起、让管道阻塞、让信号失灵、让资源泄漏——直到你在凌晨三点盯着ps aux | grep python发现一堆Z（zombie）进程，才意识到：不是你的 Agent 逻辑错了，是你对exec的理解，还停留在教科书第一行。

这八个坑，和shell=True无关，和sudo无关，甚至和pty无关。它们深埋在fork-exec-wait这一操作系统原语的毛细血管里，是 Linux 进程模型与 Python 封装层之间那层薄如蝉翼、却坚不可摧的语义鸿沟。今天这篇，不讲原理图，不画流程框，就用真实复现的案例、strace截图、/proc/pid/status输出，带你一层层剥开这行签名背后的八重地狱。

提示：本文所有案例均基于 Linux 5.15+ 内核 + Python 3.10+ 复现，macOS 因其launchd机制差异较大，Windows 则因CreateProcess语义完全不同，不在讨论范围内。请勿直接套用到非 Linux 场景。

2. 坑一：stdin=PIPE 不等于“我接管了输入”，而是“我制造了死锁温床”

几乎所有初学 Agent 的人都会这么写：

result = subprocess.run( ["sh", "-c", "read -p 'Enter password: ' pw; echo $pw | sha256sum"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=10 )

逻辑很清晰：启动一个交互式 shell，读取密码，哈希后输出。但运行起来，100% 卡死在timeout。strace显示子进程卡在read(0, ...)，父进程卡在wait4()。

为什么？

因为stdin=PIPE创建了一个匿名管道，Python 自动为子进程dup2()了管道读端到fd 0。但关键点在于：这个管道的写端（fd 1）始终由 Python 进程持有，且从未写入任何数据。read(0, ...)在等待 EOF 或数据，而写端未关闭，内核永远不会发送 EOF；你又没往写端写东西，它就永远等下去。

这不是 bug，是 POSIX 的设计哲学：管道是同步通信信道，一端不关，另一端永不结束。

真实修复方案（三选一，按场景）：

• 场景 A：你确实需要向子进程喂数据

  # 必须显式写入并关闭写端 proc = subprocess.Popen( ["sh", "-c", "read pw; echo $pw | sha256sum"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) stdout, stderr = proc.communicate(input="mysecret\n") # communicate() 自动关闭写端并等待

• 场景 B：你只是想让子进程有 stdin 可读（比如避免isatty()返回 False）

  # 用 /dev/null 模拟空输入流，不创建管道 with open(os.devnull, "rb") as devnull: result = subprocess.run( ["sh", "-c", "if [ -t 0 ]; then echo 'tty'; else echo 'not tty'; fi"], stdin=devnull, stdout=subprocess.PIPE, text=True )

• 场景 C：你希望子进程继承当前终端的 stdin（最常见于调试）

  # 直接传 sys.stdin，不创建新管道 result = subprocess.run( ["sh", "-c", "read -p 'Input: ' x; echo $x"], stdin=sys.stdin, # 关键！ stdout=subprocess.PIPE, text=True )

注意：subprocess.run()的input=参数本质就是communicate(input=...)的语法糖，它内部做了stdin=PIPE+communicate()的封装。但很多 Agent 框架为了“统一接口”，硬把input=和stdin=混用，导致管道状态混乱。我的经验是：在 Agent 中，永远优先用Popen+communicate()，禁用run()的input=参数。因为communicate()是原子操作，而run()的input=在超时处理上存在竞态。

3. 坑二：stdout=PIPE 的真实代价——缓冲区溢出与 SIGPIPE 的双重绞杀

这是 Agent 卡死的头号原因，占比超过 40%。现象极其典型：执行ls -lR /usr或journalctl -n 10000时，进程卡住，top显示 CPU 0%，iostat显示磁盘空闲，strace显示父进程卡在wait4()，子进程卡在write(1, ...)。

根因解析：

当stdout=PIPE时，Python 创建一个管道，子进程dup2()其写端到fd 1。但管道内核缓冲区大小是有限的（Linux 默认 64KB）。一旦子进程输出超过此值，write()系统调用就会阻塞，等待父进程从管道读端read()。而父进程此时正在wait4()等子进程退出——一个等读，一个等写，完美死锁。

更致命的是SIGPIPE：如果父进程在子进程write()时提前关闭了管道读端（比如Popen对象被 gc 回收），子进程再write()就会收到SIGPIPE信号，默认终止。但很多 shell 命令（如grep,awk）会忽略SIGPIPE，继续尝试write()，陷入无限阻塞。

实测验证：

# 在终端 A 启动一个持续输出的进程 $ yes | head -c 1000000 > /dev/null & # 输出 1MB 到 /dev/null，很快结束 $ yes | head -c 1000000 | cat > /dev/null & # 输出 1MB 到管道，cat 会立即消费，也快 $ yes | head -c 1000000 | true > /dev/null & # 输出 1MB 到管道，true 不读，卡死！

true进程状态变为D，ps显示defunct，lsof -p <pid>显示管道写端被占用。

Agent 级解决方案：

• 永远不要用stdout=PIPE执行可能产生大量输出的命令。改用stdout=subprocess.DEVNULL或重定向到文件。
• 必须捕获大输出时，用iter(proc.stdout.readline, b'')流式读取：

  proc = subprocess.Popen( ["journalctl", "-n", "10000"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, # 行缓冲 universal_newlines=True ) for line in iter(proc.stdout.readline, ''): if "ERROR" in line: log_error(line) # 实时处理，不积压 proc.wait() # 此时 stdout 已读完，不会卡

• 终极保险：设置preexec_fn=os.setsid并用timeout+killpg强制清理：

  def kill_child_processes(): try: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) except ProcessLookupError: pass proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, preexec_fn=os.setsid # 创建新进程组 ) try: stdout, _ = proc.communicate(timeout=30) except subprocess.TimeoutExpired: kill_child_processes() raise

经验教训：我在某云厂商的 Agent 日志采集模块里见过最惨烈的案例——他们用stdout=PIPE执行find / -name "*.log" -exec ls -l {} \;，结果/var/log/journal下单个日志文件就 2GB，管道瞬间填满，整个 Agent 进程组被D状态锁死，必须重启宿主机。后来改成stdout=open("/tmp/agent_out.log", "w")，问题消失。

4. 坑三：shell=False 下的路径陷阱——execve() 的绝对信仰与相对幻觉

Agent 开发者常犯一个低级但致命的错误：认为shell=False就绝对安全，可以随便拼接字符串。于是写出这样的代码：

# 危险！ cmd = ["ls", "-l", "/home/" + username + "/.ssh/authorized_keys"] result = subprocess.run(cmd, shell=False, capture_output=True)

看起来天衣无缝。但当username = "alice; rm -rf /"时，cmd变成["ls", "-l", "/home/alice; rm -rf /"]，ls会试图列出一个叫/home/alice; rm -rf /的文件——它不会执行rm。这确实是安全的。

但真正的坑在路径本身：

execve()系统调用要求第一个参数（pathname）必须是可执行文件的绝对路径或相对于当前工作目录的路径。它不查$PATH。所以：

# 这行会失败！ subprocess.run(["ls", "-l"], shell=False) # FileNotFoundError: [Errno 2] No such file or directory: 'ls' # 这行成功 subprocess.run(["/bin/ls", "-l"], shell=False) # 绝对路径，ok # 这行也成功（如果 cwd 是 /usr/bin） subprocess.run(["ls", "-l"], cwd="/usr/bin", shell=False) # 相对路径，ok

Agent 往往运行在容器或受限环境中，$PATH可能被精简（如 Alpine 的/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin），而ls实际在/bin/ls。shell=False下，Python 不会帮你查$PATH，它直接调用execve("ls", ...), 内核找不到ls就报错。

为什么shell=True没这个问题？因为shell=True会先启动/bin/sh，再由 shell 解析ls并查$PATH。

Agent 安全实践：

• 永远用shutil.which()预检命令是否存在：

  import shutil ls_path = shutil.which("ls") if not ls_path: raise RuntimeError("Command 'ls' not found in PATH") result = subprocess.run([ls_path, "-l"], ...)

• 在容器化 Agent 中，强制使用绝对路径白名单：

  SAFE_COMMANDS = { "ls": "/bin/ls", "cat": "/bin/cat", "grep": "/bin/grep", "sh": "/bin/sh", "bash": "/bin/bash" } cmd = [SAFE_COMMANDS.get("ls"), "-l", "/tmp"]

• 禁用shell=True，除非你明确需要 shell 特性（如通配符、管道）。shell=True会引入sh解析，带来额外攻击面（如$(rm -rf /)）。

一次血泪教训：某 IoT 设备 Agent 用subprocess.run(["df", "-h"], shell=False)检查磁盘空间。设备固件升级后，/usr/bin/df被移除，/bin/df保留。Agent 启动失败，整台设备变砖。后来加了shutil.which("df")检查，问题解决。

5. 坑四：cwd 参数的幽灵行为——chdir() 的原子性与子进程的孤儿时刻

cwd参数看似简单：“指定子进程的工作目录”。但它的实现细节，足以让 Agent 在多线程环境下集体阵亡。

subprocess模块在fork()之后、execve()之前，调用os.chdir(cwd)。这是一个原子操作，但它发生在子进程上下文中。问题来了：如果 Agent 是多线程的，主线程调用os.chdir("/tmp")，然后另一个线程同时调用subprocess.run(..., cwd="/home")，会发生什么？

答案是：不确定。chdir()是进程级的，不是线程级的。fork()复制的是整个进程地址空间，包括当前工作目录。所以subprocess.run(..., cwd="/home")的chdir()会覆盖主线程刚设的/tmp，但这个覆盖只对子进程有效——父进程的工作目录不变。

听起来没问题？错。更大的坑在cwd为相对路径时：

# 当前目录是 /opt/agent os.chdir("/opt/agent") subprocess.run(["ls"], cwd="logs") # ok，等价于 chdir("/opt/agent/logs") # 但如果此时另一个线程执行了： os.chdir("/var/lib/agent") # 主线程工作目录变了！ # 那么上面的 cwd="logs" 就变成了 chdir("/var/lib/agent/logs") —— 目录不存在！

subprocess模块在chdir()前不会校验cwd路径是否存在或是否可访问。它直接调用chdir()，如果失败，execve()会因ENOENT或EACCES报错，但错误信息是FileNotFoundError，完全看不出是cwd导致的。

Agent 稳定性加固方案：

• 永远用绝对路径作为cwd：

  import os cwd_abs = os.path.abspath("/relative/path/to/logs") if not os.path.isdir(cwd_abs): raise ValueError(f"cwd not exists: {cwd_abs}") result = subprocess.run(["ls"], cwd=cwd_abs)

• 在Popen构造函数中，用preexec_fn做原子化chdir：

  def safe_chdir(path): try: os.chdir(path) except (OSError, FileNotFoundError) as e: # 子进程内无法抛异常给父进程，只能 exit os._exit(127) # bash 的 command not found 退出码 proc = subprocess.Popen( ["ls"], cwd="/tmp", # 这个 cwd 会被 preexec_fn 覆盖 preexec_fn=lambda: safe_chdir("/opt/agent/logs") )

• 对cwd做权限预检：

  def validate_cwd(path): if not os.path.exists(path): return False, f"Path does not exist: {path}" if not os.path.isdir(path): return False, f"Not a directory: {path}" if not os.access(path, os.R_OK | os.X_OK): # 需要读和执行（进入）权限 return False, f"No read/execute access: {path}" return True, "" ok, msg = validate_cwd("/opt/agent/logs") if not ok: raise PermissionError(msg)

生产环境建议：在 Agent 启动时，将所有可能用到的cwd路径加入白名单，并在配置加载阶段做一次validate_cwd()全检。这样错误发生在启动时，而不是半夜执行任务时。

6. 坑五：env 参数的隐形污染——父进程环境的“遗产税”

env参数让你可以自定义子进程的环境变量。但它的默认行为，是env=None，即继承父进程的全部环境变量。这对 Agent 来说，是巨大的安全隐患和稳定性雷区。

问题一：敏感信息泄露

Agent 进程可能设置了API_KEY=xxx,DB_PASSWORD=yyy等环境变量。env=None下，这些变量会完整传递给子进程。如果子进程是sh或bash，它就能echo $API_KEY。更糟的是，如果子进程崩溃，其环境变量会出现在core dump或systemd日志中。

问题二：PATH 污染

父进程的PATH可能包含/tmp,/home/user/bin等非标准路径。子进程执行ls时，可能优先找到/tmp/ls（一个恶意脚本），而非/bin/ls。

问题三：locale 与编码冲突

父进程LANG=C.UTF-8，子进程LANG=en_US.UTF-8，可能导致subprocess.run(..., text=True)解码失败，抛出UnicodeDecodeError。

正确做法：最小化env

import os # 基于系统默认环境，只添加必要变量 base_env = { "PATH": "/usr/local/bin:/usr/bin:/bin:/usr/local/sbin:/usr/sbin:/sbin", "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "HOME": "/root", # 或根据用户动态设置 "USER": "root", "SHELL": "/bin/sh" } # Agent 特定变量，显式添加 env = base_env.copy() env["AGENT_VERSION"] = "1.2.3" env["AGENT_CONFIG_PATH"] = "/etc/agent/config.yaml" result = subprocess.run(["ls", "-l"], env=env, text=True)

进阶技巧：用os.environ.copy()做减法（不推荐，仅作说明）

# 不推荐！因为无法控制父进程环境的初始状态 env = os.environ.copy() env.pop("API_KEY", None) env.pop("SECRET_TOKEN", None) env["PATH"] = "/usr/bin:/bin" # 覆盖 PATH

安全红线：在金融、政务类 Agent 中，env=None是审计必过项。我们团队的标准是：所有subprocess调用，env参数必须显式传入一个字典，且该字典必须经过base_env白名单过滤。env=None在 CI/CD 流水线中会被pylint插件直接拦截。

7. 坑六：timeout 的幻觉——它只管父进程，不管子进程树

timeout参数是 Agent 的救命稻草，但它的作用范围，远比你想象的窄。

subprocess.run(..., timeout=5)的含义是：如果run()函数本身（即wait()等待子进程退出）耗时超过 5 秒，则抛出TimeoutExpired。它不杀死子进程，也不杀死子进程的子进程（孙子进程）。

这就是为什么你经常看到：TimeoutExpired抛出了，但ps aux | grep your_cmd还能看到子进程在跑，甚至pstree -p $parent_pid显示一个完整的进程树。

根本原因：timeout是 Python 在用户态做的计时，wait4()系统调用本身没有超时机制。Python 只是wait4()时加了个select()或epoll轮询，超时就raise。子进程依然活着。

Agent 必须的清理逻辑：

import subprocess import os import signal import time def run_with_tree_timeout(cmd, timeout=30, kill_children=True): proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True # 关键！创建新会话，子进程成为会话首进程 ) try: stdout, stderr = proc.communicate(timeout=timeout) return proc.returncode, stdout, stderr except subprocess.TimeoutExpired: if kill_children: # 向整个进程组发送 SIGTERM try: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) except ProcessLookupError: pass # 等待 5 秒，不成就 SIGKILL try: proc.wait(timeout=5) except subprocess.TimeoutExpired: try: os.killpg(os.getpgid(proc.pid), signal.SIGKILL) except ProcessLookupError: pass proc.wait() raise # 使用 try: code, out, err = run_with_tree_timeout(["sh", "-c", "sleep 100"], timeout=5) except subprocess.TimeoutExpired: print("Killed by timeout")

为什么start_new_session=True如此重要？

• start_new_session=True调用setsid()，让子进程脱离父进程的会话和进程组。
• os.getpgid(proc.pid)能准确获取子进程的进程组 ID（PGID）。
• os.killpg(pgid, sig)向整个进程组发送信号，确保sleep 100及其可能 fork 的子进程（如sh启动的sleep）都被杀死。

没有start_new_session=True，os.getpgid(proc.pid)返回的可能是父进程的 PGID，killpg就会误杀 Agent 自身。

真实案例：某 Kubernetes 集群 Agent 用timeout=10执行kubectl get pods，超时后 Agent 认为任务失败，但kubectl进程还在，它又去拉取了 1000 个 pod 的详情，占满节点 CPU。加上start_new_session=True和killpg后，问题根治。

8. 坑七：text=True 的编码陷阱——UTF-8 的霸权与 locale 的沉默反抗

text=True（或universal_newlines=True）让subprocess自动帮你 decodestdout和stderr为str。但它的默认编码是locale.getpreferredencoding(False)，在大多数 Linux 系统上是UTF-8。问题在于：子进程的 locale 可能和父进程完全不同。

例如，Agent 运行在LANG=C环境下（纯 ASCII），但子进程sh启动时，locale是en_US.UTF-8。sh输出的ls结果包含中文文件名，用UTF-8解码正常。但如果子进程是sh启动的perl脚本，而perl的use utf8没生效，它可能用ISO-8859-1输出乱码，subprocess用UTF-8去 decode，就抛UnicodeDecodeError。

更隐蔽的是Clocale 下的printf '\xc3\xa9'（UTF-8 编码的é），在LANG=C下，printf会原样输出两个字节，subprocess用UTF-8decode 会失败，因为\xc3\xa9在Clocale 下不被视为合法 UTF-8。

解决方案分三层：

• 底层：强制指定encoding

  # 无论 locale 如何，都用 UTF-8 result = subprocess.run( ["ls", "-l"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", # 显式指定 errors="replace" # 解码失败时用 替代，不抛异常 )

• 中层：统一子进程 locale

  env = base_env.copy() env["LANG"] = "C.UTF-8" # 强制子进程用 C.UTF-8 env["LC_ALL"] = "C.UTF-8" result = subprocess.run(["ls", "-l"], env=env, text=True, encoding="utf-8")

• 上层：二进制处理，业务层解码

  # 最稳健：永远用 bytes，自己 decode result = subprocess.run( ["ls", "-l"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=False # 关键！返回 bytes ) try: stdout_str = result.stdout.decode("utf-8") except UnicodeDecodeError: # fallback to latin-1，它能 decode 任意 bytes stdout_str = result.stdout.decode("latin-1")

经验之谈：在 Agent 的核心执行函数中，我坚持text=False。因为bytes是无损的，decode 是业务逻辑的一部分。text=True把解码这个易错环节藏在了底层，一旦出错，堆栈信息指向subprocess.py，排查成本极高。

9. 坑八：check=True 的傲慢——它只检查 returncode，不检查世界是否崩塌

check=True是subprocess.run()的“自动异常开关”：如果returncode != 0，就抛CalledProcessError。它让代码更简洁：

# 简洁但危险 result = subprocess.run(["git", "commit", "-m", "init"], check=True)

但check=True的傲慢在于：它只关心returncode，对子进程的其他死亡方式视而不见。

• 如果子进程被SIGKILL（kill -9）杀死，returncode是-9，check=True会抛异常，这还好。
• 如果子进程被SIGSEGV（段错误）杀死，returncode是-11，也会抛异常。
• 但如果子进程因OOM Killer被干掉呢？OOM Killer会给进程发SIGKILL，returncode是-9，check=True会捕获。
• 最致命的是：子进程 fork 了一个后台 daemon，然后自己 exit(0)，但 daemon 在疯狂写磁盘，导致 Agent 所在磁盘爆满，整个系统卡死。check=True看到returncode=0，欢天喜地，完全不知道灾难已在酝酿。

check=True给了你一种虚假的安全感：只要returncode == 0，万事大吉。但 Agent 的世界里，0只代表“我按协议退出了”，不代表“我完成了任务”。

Agent 级别的健康检查必须超越returncode：

• 检查输出内容：

  result = subprocess.run(["curl", "-s", "http://localhost:8080/health"], capture_output=True, text=True) if result.returncode != 0: raise RuntimeError("curl failed") if "status\":\"UP\"" not in result.stdout: raise RuntimeError("Service health check failed")

• 检查资源消耗（需psutil）：

  import psutil proc = subprocess.Popen(["heavy_task"]) try: proc.wait(timeout=60) # 检查是否产生了巨型临时文件 if os.path.getsize("/tmp/heavy_task.out") > 100 * 1024 * 1024: # 100MB raise RuntimeError("Output too large") except subprocess.TimeoutExpired: proc.kill() raise

• 检查副作用（如文件、端口、进程）：

  # 执行 service restart subprocess.run(["systemctl", "restart", "nginx"], check=True) # 验证 nginx 进程是否存在 if not any("nginx:" in p.info["name"] for p in psutil.process_iter(["name"])): raise RuntimeError("nginx did not start after restart")

核心思想：check=True是一个语法糖，不是质量保证。在 Agent 开发中，returncode只是健康检查的第一步，后面必须跟上业务语义的验证。我把这叫做“returncode之后的三问”：输出对吗？资源稳吗？状态正吗？

10. 终极武器：一个生产就绪的 AgentCommand 执行器

把以上八个坑全部踩过一遍后，我提炼出了一个在多个高可用 Agent 项目中稳定运行三年的AgentCommand类。它不是万能的，但覆盖了 95% 的生产场景。

import subprocess import os import signal import time import shlex import logging from pathlib import Path from typing import List, Dict, Optional, Union, Tuple, Any logger = logging.getLogger(__name__) class AgentCommand: """ Production-ready subprocess executor for Agent. Solves the 8 pitfalls with explicit, auditable defaults. """ # 白名单 PATH，拒绝一切 PATH 污染 SAFE_PATH = "/usr/local/bin:/usr/bin:/bin:/usr/local/sbin:/usr/sbin:/sbin" # 安全基础环境 BASE_ENV = { "PATH": SAFE_PATH, "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "HOME": "/root", "USER": "root", "SHELL": "/bin/sh", "TZ": "UTC" } def __init__( self, cmd: Union[str, List[str]], cwd: Optional[Union[str, Path]] = None, env: Optional[Dict[str, str]] = None, timeout: int = 30, max_output_size: int = 10 * 1024 * 1024, # 10MB check_returncode: bool = True, kill_children_on_timeout: bool = True, encoding: str = "utf-8", errors: str = "replace" ): self.cmd = cmd self.cwd = self._resolve_cwd(cwd) self.env = self._build_env(env) self.timeout = timeout self.max_output_size = max_output_size self.check_returncode = check_returncode self.kill_children_on_timeout = kill_children_on_timeout self.encoding = encoding self.errors = errors # 日志记录命令（脱敏） self.log_cmd = self._sanitize_cmd(cmd) logger.debug(f"AgentCommand init: {self.log_cmd} | cwd={self.cwd}") def _resolve_cwd(self, cwd: Optional[Union[str, Path]]) -> str: if cwd is None: return "/" cwd_abs = os.path.abspath(str(cwd)) if not os.path.isdir(cwd_abs): raise ValueError(f"cwd not exists: {cwd_abs}") if not os.access(cwd_abs, os.R_OK | os.X_OK): raise PermissionError(f"No read/execute access to cwd: {cwd_abs}") return cwd_abs def _build_env(self, user_env: Optional[Dict[str, str]]) -> Dict[str, str]: env = self.BASE_ENV.copy() if user_env: # 只允许覆盖白名单中的 key，或添加新 key for k, v in user_env.items(): if k in self.BASE_ENV or k not in ["PATH", "LANG", "LC_ALL"]: env[k] = v else: logger.warning(f"Ignored unsafe env key: {k}") return env def _sanitize_cmd(self, cmd: Union[str, List[str]]) -> str: if isinstance(cmd, str): return shlex.quote(cmd) return " ".join(shlex.quote(str(x)) for x in cmd) def execute(self) -> Tuple[int, bytes, bytes]: """Execute and return (returncode, stdout_bytes, stderr_bytes)""" # 构建 Popen 参数 popen_kwargs = { "args": self.cmd, "cwd": self.cwd, "env": self.env, "stdout": subprocess.PIPE, "stderr": subprocess.PIPE, "start_new_session": True, # 关键！为 killpg 准备 "bufsize": -1, } proc = None try: proc = subprocess.Popen(**popen_kwargs) logger.debug(f"Started process {proc.pid} for {self.log_cmd}") # 流式读取，防止 stdout/stderr 缓冲区溢出 stdout_chunks = [] stderr_chunks = [] start_time = time.time() while proc.poll() is None: # 检查超时 if time.time() - start_time > self.timeout: raise subprocess.TimeoutExpired( self.cmd, self.timeout, output=b"", stderr=b"" ) # 非阻塞读取 stdout/stderr if proc.stdout: chunk = proc.stdout.read(8192) if chunk: stdout_chunks.append(chunk) if sum(len(c) for c in stdout_chunks) > self.max_output_size: raise RuntimeError("stdout size exceeded limit") if proc.stderr: chunk = proc.stderr.read(8192) if chunk: stderr_chunks.append(chunk) if sum(len(c) for c in stderr_chunks) > self.max_output_size: raise RuntimeError("stderr size exceeded limit") time.sleep(0.01) # 避免 busy loop # 读取剩余数据 if proc.stdout: chunk = proc.stdout