Python subprocess 实战——调用系统命令、管道、超时控制、安全防护

适读人群：需要在 Python 中调用系统工具的开发者、运维工程师 | 阅读时长：约15分钟 | 核心价值：掌握 subprocess 的完整工程实践，避免安全漏洞

---

那次差点酿成大事故的命令注入

我见过一个真实的案例，一家小公司的内部工具，有个功能是根据用户输入的文件名执行压缩：

# 千万不要这样写！
import os
filename = request.get("filename")
os.system(f"zip output.zip {filename}")

有个安全意识比较强的同事在测试的时候，把 filename 填成了 file.txt; rm -rf /data，然后……你懂的。

幸好是测试环境，数据可以恢复，但这个教训让整个团队都后背发凉。

这件事之后，我在团队里做了一次关于 subprocess 安全使用的分享，今天把这些内容系统整理出来，既讲怎么用好，也讲怎么避开危险。

---

一、subprocess 核心 API——只用这几个

Python 的 subprocess 模块有很多 API，但工程实践中其实只需要掌握少数几个：

API	用途
subprocess.run()	主力，运行命令并等待结束（Python 3.5+）
subprocess.Popen()	需要流式读取输出、或更复杂控制时使用
subprocess.check_output()	简单获取输出（是 run 的老式封装）

推荐原则：新代码一律用 subprocess.run()，只有需要流式处理的场景用 Popen。

---

二、subprocess.run 完整用法

import subprocess
import sys
from pathlib import Path


# 基础用法：运行命令，捕获输出
def run_command(
    args: list[str],
    cwd: str = None,
    timeout: int = 30,
    env: dict = None,
    input_data: str = None,
) -> tuple[int, str, str]:
    """
    安全执行命令
    返回 (returncode, stdout, stderr)
    """
    try:
        result = subprocess.run(
            args,               # 必须是列表！不要用字符串+shell=True
            capture_output=True,
            text=True,          # 自动 decode 为字符串
            timeout=timeout,
            cwd=cwd,
            env=env,
            input=input_data,
            encoding="utf-8",
            errors="replace",   # 遇到无法解码的字节，替换而不是抛异常
        )
        return result.returncode, result.stdout, result.stderr

    except subprocess.TimeoutExpired:
        return -1, "", f"命令超时（{timeout}s）"
    except FileNotFoundError:
        return -2, "", f"命令不存在: {args[0]}"
    except PermissionError:
        return -3, "", f"没有执行权限: {args[0]}"


# 使用示例
code, out, err = run_command(["ls", "-la", "/tmp"])
if code == 0:
    print(out)
else:
    print(f"执行失败: {err}")

# Git 操作示例
code, out, err = run_command(
    ["git", "log", "--oneline", "-10"],
    cwd="/path/to/repo"
)

# ffmpeg 转码（带超时）
code, out, err = run_command(
    ["ffmpeg", "-i", "input.mp4", "-c:v", "libx264", "output.mp4"],
    timeout=300  # 5分钟超时
)

---

三、安全防护——命令注入是大坑

踩坑实录1：shell=True 导致命令注入

现象：用户输入了恶意文件名，导致额外命令被执行。
原因：shell=True 把整个命令字符串传给 shell 解释，shell 的特殊字符（;、|、$、`）会被执行。
解法：永远用列表传参，永远不用 shell=True（除非你明确知道自己在做什么）。

import subprocess
import shlex


# 危险写法（绝对不要用）
user_input = "file.txt; rm -rf /"
# subprocess.run(f"cat {user_input}", shell=True)  # 命令注入！

# 安全写法1：直接用列表
subprocess.run(["cat", user_input], capture_output=True)
# user_input 会被当作 cat 的参数，不会被 shell 解释

# 安全写法2：如果真的需要 shell 字符串（比如管道），先用 shlex.split 解析
# 注意：即使用 shlex.split，也要确保输入来自可信来源
safe_cmd = "ls -la | head -10"
args = shlex.split(safe_cmd)
# 但这仍然无法完全防止注入，最好的方式是不用 shell=True


# 输入验证函数
import re


def validate_filename(filename: str) -> bool:
    """验证文件名安全性"""
    # 只允许字母、数字、点、下划线、连字符
    return bool(re.match(r'^[\w\-. ]+$', filename)) and ".." not in filename


def safe_zip(filename: str, output_dir: Path) -> bool:
    """安全的文件压缩"""
    if not validate_filename(filename):
        raise ValueError(f"不安全的文件名: {filename!r}")

    source = output_dir / filename
    if not source.exists():
        raise FileNotFoundError(f"文件不存在: {source}")

    output = output_dir / f"{Path(filename).stem}.zip"
    code, _, err = run_command(
        ["zip", str(output), str(source)],
        cwd=str(output_dir),
    )
    return code == 0

---

四、管道——多命令协同

import subprocess


def run_pipeline(commands: list[list[str]], timeout: int = 30) -> tuple[int, str, str]:
    """
    执行命令管道
    等价于: cmd1 | cmd2 | cmd3
    """
    processes = []
    try:
        for i, cmd in enumerate(commands):
            stdin = processes[-1].stdout if processes else None
            proc = subprocess.Popen(
                cmd,
                stdin=stdin,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            processes.append(proc)
            # 关闭前一个进程的 stdout，让它正确收到 SIGPIPE
            if len(processes) > 1:
                processes[-2].stdout.close()

        # 等待最后一个进程完成
        last_proc = processes[-1]
        stdout, stderr = last_proc.communicate(timeout=timeout)
        return last_proc.returncode, stdout.decode("utf-8", errors="replace"), stderr.decode()

    except subprocess.TimeoutExpired:
        for p in processes:
            p.kill()
        return -1, "", "管道执行超时"
    finally:
        for p in processes:
            if p.poll() is None:
                p.terminate()


# 示例：ps aux | grep python | grep -v grep
code, out, err = run_pipeline([
    ["ps", "aux"],
    ["grep", "python"],
    ["grep", "-v", "grep"],
])
print(out)

# 示例：统计代码行数
code, out, err = run_pipeline([
    ["find", ".", "-name", "*.py"],
    ["xargs", "wc", "-l"],
])

---

五、流式输出——实时显示长时间任务进度

import subprocess
import sys
import threading


def run_with_realtime_output(
    args: list[str],
    timeout: int = 300,
    on_stdout: callable = None,
    on_stderr: callable = None,
) -> int:
    """
    实时输出命令执行结果
    适用于：pip install、docker build 等长时间执行的命令
    """
    proc = subprocess.Popen(
        args,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        bufsize=1,  # 行缓冲
        universal_newlines=True,
    )

    def read_stream(stream, callback, stream_name):
        for line in iter(stream.readline, ""):
            line = line.rstrip("\n")
            if callback:
                callback(line)
            else:
                prefix = "" if stream_name == "stdout" else "[ERR] "
                print(f"{prefix}{line}", flush=True)
        stream.close()

    # 并发读取 stdout 和 stderr，防止管道缓冲区满导致死锁
    t_stdout = threading.Thread(
        target=read_stream,
        args=(proc.stdout, on_stdout, "stdout"),
        daemon=True,
    )
    t_stderr = threading.Thread(
        target=read_stream,
        args=(proc.stderr, on_stderr, "stderr"),
        daemon=True,
    )
    t_stdout.start()
    t_stderr.start()

    try:
        proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        return -1

    t_stdout.join()
    t_stderr.join()
    return proc.returncode


# 使用示例：实时显示 pip install 进度
code = run_with_realtime_output(
    [sys.executable, "-m", "pip", "install", "requests", "-v"]
)
print(f"退出码: {code}")

---

六、踩坑实录

踩坑实录2：communicate() 死锁

现象：调用 proc.communicate() 之后，程序卡死。
原因：subprocess 的 stdout 和 stderr 管道缓冲区满了，子进程写不进去，阻塞等待读取；父进程也在等 communicate，形成死锁。
解法：并发读取 stdout 和 stderr（见上面的 run_with_realtime_output），或者用 communicate() 替代 wait()——communicate() 内部其实已经处理了这个问题，但不要先 read() 再 wait()。

踩坑实录3：子进程变成僵尸进程

现象：大量子进程执行完后没有被回收，ps aux 看到一堆 <defunct> 进程。
原因：父进程没有调用 wait() 或 communicate() 来获取子进程退出状态。
解法：始终用 subprocess.run() 或确保在 Popen 使用完后调用 proc.wait()；或者在 Popen 时设置 close_fds=True。

---

七、选型建议

场景	推荐方案
简单执行命令	subprocess.run()
需要实时输出	subprocess.Popen() + 线程读取
纯 Python 能实现的功能	不要调 subprocess，用 Python 库
系统工具不存在的操作	Python 标准库（shutil、os、pathlib）

最后还是要强调：能用 Python 标准库实现的，尽量不要调用外部命令。 subprocess 最适合调用成熟的命令行工具（ffmpeg、git、docker等），不适合作为通用的"做什么事都 os.system"的偷懒方式。