Python subprocess 实战:超时控制、管道传输与命令注入防护用 Python 调外部命令,十有八九是从os.system(ping host)开始的。它能跑,直到有一天线上 host 变量里混进了; rm -rf /,或者某个命令卡死把整个 worker 拖挂。subprocess才是正解,但它的坑也不少:shellTrue 的注入、卡死没超时、大输出撑爆内存、管道拼接写错。这篇把这几个真实痛点一次讲透。先看朴素写法为什么危险importos hostinput(host: )os.system(ping -c 1 host)# 危险如果用户输入baidu.com; cat /etc/passwd,两条命令都会执行。os.system把整串丢给 shell,分号、反引号、$()全都是 shell 元字符。这不是理论风险,是 OWASP 命令注入的标准案例。第一反应可能是「那我转义一下」,但手写转义几乎不可能覆盖所有 shell 语法。正确思路是:根本别让 shell 参与解析。正确写法:参数列表 不走 shellimportsubprocess hostinput(host: )# 传 list,shellFalse(默认),host 只会被当成一个参数,不解析元字符resultsubprocess.run([ping,-c,1,host],capture_outputTrue,textTrue,# 输出按文本解码,不用手动 decodetimeout5,# 关键:超时保护,见下节)print(result.returncode,result.stdout)把命令拆成列表,host里就算带分号也只是一个普通字符串参数,ping会报「无法解析主机」而不会执行第二条命令。记住一条铁律:参数来自外部输入时,永远用 list 形式,永远不加shellTrue。什么时候才需要shellTrue?只有当你确实要用 shell 特性(管道、通配符展开、环境变量插值)且命令是你自己写死的常量时。一旦命令里拼接了变量,shellTrue就是定时炸弹。超时:别让子进程把你拖死subprocess.run不传timeout,子进程卡住你就永远等下去。生产环境这会耗尽线程/连接。importsubprocesstry:resultsubprocess.run([sleep,10],capture_outputTrue,textTrue,timeout3,)exceptsubprocess.TimeoutExpiredase:# 超时后子进程已被 kill,但要注意它的子孙进程可能还在print(f命令超时,已跑了{e.timeout}s)有个容易忽略的坑:TimeoutExpired触发后,run会 kill 直接子进程,但如果子进程又 fork 了孙子进程(比如 shell 脚本里起了后台任务),那些不一定被清掉。需要彻底清干净时,用Popen 进程组:importsubprocess,os,signal procsubprocess.Popen([./long_task.sh],stdoutsubprocess.PIPE,textTrue,start_new_sessionTrue,# 让子进程成为新会话的组长,方便整组 kill)try:out,_proc.communicate(timeout3)exceptsubprocess.TimeoutExpired:# 杀掉整个进程组,连带孙子进程os.killpg(os.getpgid(proc.pid),signal.SIGKILL)proc.communicate()# 回收僵尸进程管道:别用 shell 拼|,用 stdin 串起来很多人想实现cat access.log | grep 404 | wc -l,直接shellTrue拼一个字符串。更安全、更可控的做法是把前一个进程的stdout接到后一个的stdin:importsubprocess# cat access.log | grep 404p1subprocess.Popen([cat,access.log],stdoutsubprocess.PIPE)p2subprocess.Popen([grep,404],stdinp1.stdout,stdoutsubprocess.PIPE,textTrue)# 关键:关掉父进程持有的 p1.stdout,否则 p1 结束时 p2 收不到 EOF 会挂住p1.stdout.close()out,_p2.communicate()print(out)p1.stdout.close()这行经常被漏掉。原理是:Popen创建 p2 时复制了一份 p1.stdout 的文件描述符,父进程自己也还留着一份。只有当所有副本都关闭,读端才会收到 EOF。父进程不关的话,p2 会一直等着以为还有数据,整条管道卡死。大输出别用 capture_output 一次性读capture_outputTrue会把全部 stdout 读进内存。如果命令输出几个 G(比如 dump 数据库),内存直接爆。要流式处理就逐行读:importsubprocesswithsubprocess.Popen([tail,-f,/var/log/app.log],stdoutsubprocess.PIPE,textTrue,bufsize1,# 行缓冲)asproc:forlineinproc.stdout:# 边产边读,内存恒定ifERRORinline:print(line.strip())还有个隐蔽死锁:如果你用proc.stdout.read()全读完再读 stderr,而子进程 stderr 缓冲区先被写满,子进程会阻塞在写 stderr 上、永远不结束,你也就永远读不完 stdout。同时要 stdout 和 stderr 时,别手动分开读,用communicate(),它内部用线程/select 同时排空两个管道。checkTrue:让失败不再被静默吞掉run默认不因非零退出码报错,returncode你不看就丢了。想让命令失败直接抛异常:importsubprocesstry:subprocess.run([git,push],checkTrue,capture_outputTrue,textTrue)exceptsubprocess.CalledProcessErrorase:# e.returncode / e.stderr 都在,方便定位print(fpush 失败(码{e.returncode}):{e.stderr})checkTruecapture_outputTrue是脚本里调命令的推荐默认组合:失败即抛、错误信息带在异常里。小结防注入:外部输入一律用 list 形式传参,不加shellTrue;shellTrue只配死命令常量用。防卡死:run一定带timeout;要清孙子进程用start_new_sessionTrueos.killpg。管道:用stdin前一个.stdout串联,记得p1.stdout.close()让下游收到 EOF。大输出:流式逐行读,别capture_output全塞内存;同时读两个流用communicate()防死锁。失败可见:加checkTrue,让非零退出码抛CalledProcessError。一句话记忆点:list 传参防注入,timeout 防卡死,communicate 防死锁——这三条守住,subprocess 就不会再半夜叫你起床。