适用场景

本文适用于 Python 3.11 及以上版本。业务程序需要调用数据库备份工具、硬件厂商 CLI、Windows 遗留程序或内部运维命令，但这些程序的输出编码不是 UTF-8，并且可能出现长时间无响应、非零退出、输出管道堵塞或参数注入风险。

目标不是再封装一层 subprocess.run()，而是建立一个可验收的命令执行边界：参数不经过 shell 拼接；输出先按字节接收，再按配置解码；每次执行都有超时和退出码检查；异常携带必要上下文，但日志不泄露密码、令牌和原始敏感输出。

常见现象

生产现场通常会同时出现几类问题：

  • 开发机输出正常，部署到中文 Windows 主机后出现 UnicodeDecodeError 或乱码；
  • 调用方只使用 subprocess.run(command, shell=True)，用户输入中的 &、|、; 等字符被 shell 当成新命令；
  • 只调用 Popen.wait()，子进程写满 stdout=PIPE 或 stderr=PIPE 后互相等待；
  • 没有设置超时，定时任务一直占用工作线程，后续任务全部排队；
  • 进程返回非零退出码，但代码仍把不完整输出当成成功结果；
  • 为了排障把完整命令行和标准错误写入日志，意外记录了密码、连接串或访问令牌。

这些问题不能靠统一加上 text=True 解决。text=True 会使用指定编码或运行环境默认编码创建文本流；如果外部程序实际输出 GBK，而容器或主机默认使用 UTF-8，错误只是从“乱码”变成“解码异常”。

先定义执行契约

在写代码前，先为每个外部命令明确以下约束：

  1. 允许执行哪些可执行文件，路径是否固定；
  2. 参数来自配置、数据库还是用户请求，分别允许什么格式和长度；
  3. 标准输出和标准错误使用 UTF-8、GBK 还是 GB18030；
  4. 正常耗时上限是多少，超时后能否安全重试；
  5. 哪些退出码表示成功，哪些表示可重试或永久失败；
  6. 输出规模是否有上限，是否可能持续产生大量数据；
  7. 日志允许记录命令名、退出码和耗时，但哪些参数与输出必须脱敏。

编码不确定时，不要在运行时轮流尝试多种编码并把第一个“能解码”的结果当真。很多字节序列在错误编码下也能成功解码，只是内容已经被误解释。更可靠的做法是把 output_encoding 作为命令适配器的显式配置；中文 Windows 遗留工具通常可按其文档配置为 gbk 或覆盖字符更完整的 gb18030。

一个可复用的同步执行器

下面的实现面向输出量有明确上限的短命令。它始终以二进制模式捕获输出，并使用参数列表执行，不让 shell 参与解析。

from __future__ import annotations

import subprocess
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Sequence


class CommandError(RuntimeError):
    """外部命令执行失败的基类。"""


class CommandStartError(CommandError):
    """外部命令无法启动。"""


class CommandDecodeError(CommandError):
    """外部命令输出无法按约定编码解码。"""


class CommandTimeoutError(CommandError):
    """外部命令超过执行时限。"""

    def __init__(self, command_name: str, timeout_seconds: float) -> None:
        self.command_name = command_name
        self.timeout_seconds = timeout_seconds
        super().__init__(
            f"外部命令执行超时: command_name={command_name}, "
            f"timeout_seconds={timeout_seconds}"
        )


class CommandExitError(CommandError):
    """外部命令以非零状态退出。"""

    def __init__(self, command_name: str, return_code: int) -> None:
        self.command_name = command_name
        self.return_code = return_code
        super().__init__(
            f"外部命令执行失败: command_name={command_name}, "
            f"return_code={return_code}"
        )


@dataclass(frozen=True, slots=True)
class CommandResult:
    """外部命令成功结果。"""

    command_name: str
    return_code: int
    stdout: str
    stderr: str
    duration_ms: int
    output_encoding: str


def _decode_output(
    data: bytes,
    *,
    command_name: str,
    stream_name: str,
    output_encoding: str,
) -> str:
    """按明确约定解码输出，禁止静默替换非法字节。"""
    try:
        return data.decode(output_encoding, errors="strict")
    except LookupError as exc:
        raise CommandDecodeError(
            f"外部命令输出编码配置无效: command_name={command_name}, "
            f"stream_name={stream_name}, output_encoding={output_encoding}"
        ) from exc
    except UnicodeDecodeError as exc:
        raise CommandDecodeError(
            f"外部命令输出解码失败: command_name={command_name}, "
            f"stream_name={stream_name}, output_encoding={output_encoding}, "
            f"byte_offset={exc.start}"
        ) from exc


def run_command(
    args: Sequence[str],
    *,
    timeout_seconds: float,
    output_encoding: str,
) -> CommandResult:
    """执行输出量受控的外部命令，并统一检查超时、编码和退出码。"""
    normalized_args = tuple(args)
    if not normalized_args or not normalized_args[0]:
        raise ValueError("外部命令不能为空")
    if timeout_seconds <= 0:
        raise ValueError("timeout_seconds 必须大于 0")
    if any("\x00" in arg for arg in normalized_args):
        raise ValueError("外部命令参数不能包含空字符")

    command_name = Path(normalized_args[0]).name
    started_at = time.monotonic()

    try:
        completed = subprocess.run(
            normalized_args,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=timeout_seconds,
            check=False,
            shell=False,
        )
    except FileNotFoundError as exc:
        raise CommandStartError(
            f"外部命令不存在: command_name={command_name}"
        ) from exc
    except OSError as exc:
        raise CommandStartError(
            f"外部命令无法启动: command_name={command_name}, "
            f"error_type={type(exc).__name__}"
        ) from exc
    except subprocess.TimeoutExpired as exc:
        raise CommandTimeoutError(command_name, timeout_seconds) from exc

    duration_ms = round((time.monotonic() - started_at) * 1000)
    stdout = _decode_output(
        completed.stdout,
        command_name=command_name,
        stream_name="stdout",
        output_encoding=output_encoding,
    )
    stderr = _decode_output(
        completed.stderr,
        command_name=command_name,
        stream_name="stderr",
        output_encoding=output_encoding,
    )

    if completed.returncode != 0:
        raise CommandExitError(command_name, completed.returncode)

    return CommandResult(
        command_name=command_name,
        return_code=completed.returncode,
        stdout=stdout,
        stderr=stderr,
        duration_ms=duration_ms,
        output_encoding=output_encoding,
    )

这里有几个有意为之的设计：

  • args 是字符串序列，而不是拼接后的命令行；shell=False 避免 shell 元字符被二次解释；
  • stdin=DEVNULL 防止无人值守任务意外等待交互输入；
  • stdout 与 stderr 同时由 subprocess.run() 内部的 communicate() 读取，避免只读一个管道造成死锁；
  • 先捕获字节，再使用明确的 output_encoding 严格解码；编码配置错误会显式失败，不会用替换字符掩盖数据问题；
  • 超时、启动失败和非零退出分别建模，调用方可以决定哪些错误允许重试；
  • 错误消息只包含命令文件名、退出码、编码和异常类型，不包含完整参数与原始输出。

subprocess.run() 的 timeout 超时后会终止它直接创建的子进程并等待回收，然后抛出 TimeoutExpired。但这不等于跨平台终止整个进程树：如果命令又启动了后台进程或守护进程，需要针对操作系统设计进程组、作业对象或工具自身的取消机制，不能假设杀掉父进程就完成清理。

在业务边界做允许列表校验

通用执行器只能保证“不经过 shell”，不能判断某个参数在业务上是否安全。不要让 HTTP 请求直接传入可执行文件路径和任意参数。应为具体工具建立窄接口：

from pathlib import Path


BACKUP_TOOL = Path(r"C:\Program Files\Vendor\backup.exe")
ALLOWED_DATABASES = {"orders", "inventory"}


def backup_database(database_name: str) -> CommandResult:
    """备份允许列表中的业务数据库。"""
    if database_name not in ALLOWED_DATABASES:
        raise ValueError("数据库名称不在允许列表中")

    return run_command(
        [str(BACKUP_TOOL), "backup", "--database", database_name],
        timeout_seconds=300,
        output_encoding="gb18030",
    )

密码不要作为命令行参数传递。命令行可能出现在进程列表、任务审计和异常采集中。优先使用权限受控的配置文件、标准输入或工具官方提供的凭据存储；即便通过环境变量传递，也要限制子进程环境，并确认监控系统不会采集该变量。

Windows 上的 .bat 和 .cmd 可能由操作系统通过命令解释器启动，参数解析规则与普通可执行文件不同。若参数来自不可信输入，优先调用真正的 .exe 或编写固定参数的受控适配器，不要把“已经设置 shell=False”当成批处理参数绝对安全的证明。

结构化日志只记录诊断字段

执行器负责返回结果或抛出异常，日志应在知道业务上下文的调用边界统一记录，避免每一层重复打印同一个错误：

import logging


logger = logging.getLogger(__name__)


def run_daily_backup(database_name: str) -> None:
    """执行每日备份并记录不含敏感数据的结果。"""
    try:
        result = backup_database(database_name)
    except CommandTimeoutError as exc:
        logger.error(
            "数据库备份命令执行超时",
            extra={
                "command_name": exc.command_name,
                "timeout_seconds": exc.timeout_seconds,
                "database_name": database_name,
            },
        )
        raise
    except CommandExitError as exc:
        logger.error(
            "数据库备份命令执行失败",
            extra={
                "command_name": exc.command_name,
                "return_code": exc.return_code,
                "database_name": database_name,
            },
        )
        raise

    logger.info(
        "数据库备份命令执行完成",
        extra={
            "command_name": result.command_name,
            "return_code": result.return_code,
            "duration_ms": result.duration_ms,
            "database_name": database_name,
        },
    )

固定消息使用中文，字段名保持稳定的英文 snake_case。默认不要记录 args、stdout、stderr、环境变量或原始请求体。如果某个工具的错误输出经过审查、确定不含敏感信息，可以在专用适配器中做长度限制和脱敏后记录；不要在通用执行器里擅自打印。

用测试固定四条关键边界

下面的测试不依赖真实厂商工具，而是用当前 Python 解释器模拟不同外部行为：

import subprocess
import sys

import pytest

from command_runner import CommandExitError, CommandTimeoutError, run_command


def test_decodes_gb18030_output() -> None:
    """应按配置解码遗留程序输出。"""
    result = run_command(
        [
            sys.executable,
            "-c",
            "import sys; sys.stdout.buffer.write('检查完成'.encode('gb18030'))",
        ],
        timeout_seconds=2,
        output_encoding="gb18030",
    )

    assert result.stdout == "检查完成"
    assert result.return_code == 0


def test_rejects_non_zero_exit_code() -> None:
    """非零退出不能被误判为成功。"""
    with pytest.raises(CommandExitError) as error:
        run_command(
            [sys.executable, "-c", "raise SystemExit(7)"],
            timeout_seconds=2,
            output_encoding="utf-8",
        )

    assert error.value.return_code == 7


def test_times_out_and_returns_control() -> None:
    """超时后应终止直接子进程并返回控制权。"""
    with pytest.raises(CommandTimeoutError):
        run_command(
            [sys.executable, "-c", "import time; time.sleep(10)"],
            timeout_seconds=0.1,
            output_encoding="utf-8",
        )


def test_shell_metacharacters_are_plain_arguments() -> None:
    """参数中的 shell 元字符不得被解释为新命令。"""
    value = "report & echo injected"
    result = run_command(
        [
            sys.executable,
            "-c",
            "import sys; print(sys.argv[1])",
            value,
        ],
        timeout_seconds=2,
        output_encoding="utf-8",
    )

    assert result.stdout.strip() == value

验证时还应补充项目自己的边界用例：错误编码、命令不存在、空参数、超长参数、标准错误包含多行内容，以及工具约定的特殊成功退出码。若工具把 1 用作“发现差异但执行成功”，就在专用适配器中显式声明允许退出码，不要在通用层忽略所有非零状态。

大输出、异步服务与重试的边界

上述实现会把输出完整缓存在内存中，只适合输出规模受控的管理命令。如果外部程序可能持续输出日志、导出数 GB 数据或永不关闭管道，应将输出直接写入权限受控的文件，或用独立线程/异步流同时消费两个管道并施加总字节数限制。只在结果返回后截断字符串，不能降低执行期间的内存峰值。

异步 Web 服务不能直接在事件循环中调用这个同步函数。可以使用 asyncio.create_subprocess_exec() 建立异步适配器，或在受控线程池中调用同步执行器；无论采用哪种方式，都必须保留超时、退出码、双管道消费和取消后的进程回收。

超时不代表可以自动重试。备份、付款、发布和配置变更等命令可能已经完成副作用，只是调用方没有及时收到结果。只有命令本身幂等，或者提供了可查询的任务 ID、幂等键和完成状态，调用方才应自动重试。

上线检查清单

  • 可执行文件使用固定路径或允许列表，业务输入不能选择任意程序；
  • 参数使用列表传递，没有字符串拼接，也没有无必要的 shell=True；
  • 每个工具明确配置输出编码，并用严格模式解码；
  • 每次执行都有与业务相符的超时，超时后确认直接子进程已回收；
  • 非零退出码进入失败路径，特殊成功状态有明确白名单；
  • 日志记录 command_name、return_code、duration_ms 等诊断字段，不记录凭据和原始敏感输出；
  • 对输出规模无上限的命令改用文件或流式消费，不使用内存全量捕获；
  • 对有副作用的命令，在启用自动重试前先设计幂等与状态查询；
  • 在目标操作系统和目标工具版本上执行集成测试，不能只依赖模拟子进程。

总结

可靠的外部命令调用不是“能运行”就结束。乱码来自未明确的字节编码，卡死来自缺失的超时或错误的管道读取，误报成功来自忽略退出码，注入风险来自把参数交给 shell 重新解释。

把命令执行收敛到一个窄边界：参数列表执行、二进制捕获、显式编码、严格解码、超时回收、退出码检查和最小化日志；再由每个业务适配器补上允许列表、凭据传递和幂等策略。这样即使面对 GB18030 遗留工具，也能得到可测试、可诊断且不会泄露敏感信息的执行链路。

参考资料