- 开发工具
【免费下载链接】sh
Python process launching
本文基于 sh 开源仓库的官方教程文档 interacting_with_processes.rst 展开,结合 sh 核心实现 与 测试用例 进行源码级佐证。sh 是 Python 进程启动(process launching)库,本文以 SSH 为例,完整讲解如何为子进程挂接 STDOUT 回调、通过
queue.Queue回写 STDIN、控制缓冲区(_out_bufsize)、启用伪终端(_tty_in/_unify_ttys),并最终给出用sh.contrib.ssh与密钥认证远程执行命令的实战方案。读完你将掌握一套通用的“交互式进程自动化”套路,可迁移到任何需要应答密码、菜单、确认提示符的命令行程序。
一、需求与总体思路:为什么需要“与进程交互”
当我们调用一个普通命令(如ls)时,sh 会启动子进程、收集全部输出并返回;但有一类程序——典型代表就是ssh——会中途向你询问信息(如密码),并且要求输入来自真实的终端。此时单纯的“启动→等待→拿输出”模型行不通,我们必须在进程运行过程中参与对话。
sh 提供的交互模式非常简单直观:
给
_out传一个可调用对象(callback),sh 就会把子进程的 STDOUT 数据逐段喂给这个回调;回调的第二个参数是一个类似queue.Queue的对象,我们往里面put数据,sh 就会把它写入子进程的 STDIN。
这条链路在源码中有明确的落点:当_out是 callable 且未显式指定_in时,RunningCommand 会自动创建一个Queue作为 stdin:
# this accounts for when _out is a callable that is passed stdin. in that # case, if stdin is unspecified, we must set it to a queue, so callbacks can # put things on it if callable(ca["out"]) and self.stdin is None: self.stdin = Queue()因此,交互式回调的通用签名就是:
def my_callback(chunk, stdin): # chunk: 从子进程 STDOUT 读到的数据(行或字符,取决于缓冲设置) # stdin: queue.Queue,向其中 put 的内容会作为子进程的 STDIN ...二、第一版尝试:给 SSH 挂上回调(并直面行缓冲问题)
先看一个“看起来正确”的版本——用sh.ssh连接服务器,并在回调里识别密码提示符、回填密码:
from sh import ssh def ssh_interact(line, stdin): line = line.strip() print(line) if line.endswith("password:"): stdin.put("correcthorsebatterystaple") ssh("10.10.10.100", _out=ssh_interact)(把10.10.10.100换成你可达的服务器 IP。)运行后你会发现:回调里什么都没打印。问题出在 STDOUT 缓冲上。
sh 对 STDOUT 默认采用行缓冲(line-buffering),这体现在 Command._call_args 的默认值 中:
# stdin buffer size # 1 for line, 0 for unbuffered, any other number for that amount "in_bufsize": 0, # stdout buffer size, same values as above "out_bufsize": 1, "err_bufsize": 1,out_bufsize=1意味着回调只有在收到换行符后才会被触发一次。而 SSH 的密码提示符恰恰没有换行:
amoffat@10.10.10.100's password:光标停在行尾等待输入,永远等不到\n,于是ssh_interact永远收不到数据。测试 test_stdout_callback_line_buffered 验证了行缓冲行为:子进程打印 5 行herpderp,行缓冲模式下回调恰好被调用 5 次。
解决:把缓冲改成 0(完全无缓冲),使用 特殊关键字参数_out_bufsize:
from sh import ssh def ssh_interact(line, stdin): line = line.strip() print(line) if line.endswith("password:"): stdin.put("correcthorsebatterystaple") ssh("10.10.10.100", _out=ssh_interact, _out_bufsize=0)三、第二版尝试:从逐字符到聚合文本
改成_out_bufsize=0后,回调倒是被调用了,但输出变成了这样:
a m o f f a t @ 1 0 . 1 0 . 1 0 . 1 0 0 ' s p a s s w o r d :这正是“无缓冲”的本意:每次从 OS 读到多少数据就原样交给回调。SSH 逐字符写提示符,于是回调收到的就是单个字符(对应测试 test_stdout_callback_line_unbuffered,它断言无缓冲模式下回调收到的片段数量等于字符数)。顺带一提,_out_bufsize也可设成任意正整数来指定定长块缓冲,如_out_bufsize=4(见 test_stdout_callback_buffered)。
所以我们需要把逐字符的片段聚合起来,再判断是否出现了password:模式。为了演示简单,官方教程用全局变量聚合:
from sh import ssh import sys aggregated = "" def ssh_interact(char, stdin): global aggregated sys.stdout.write(char.encode()) sys.stdout.flush() aggregated += char if aggregated.endswith("password: "): stdin.put("correcthorsebatterystaple") ssh("10.10.10.100", _out=ssh_interact, _out_bufsize=0)生产代码里,更合适的方式是把聚合状态封装进闭包或类(sh 官方在
sh.contrib.ssh中正是用SessionContent类来维护会话状态,见下文第五节)。这里用全局变量只是为了突出核心机制。
四、第三版尝试:补上换行与 TTY(_tty_in+_unify_ttys)
运行上面的版本,依然失败,原因有二:
原因一:密码必须带换行。你往stdin.put(...)的内容会被写入子进程 STDIN;SSH 不知道密码有多长,它对 STDIN 做行缓冲,只有收到\n才认为输入结束(如同你在真实终端里输完密码再按回车)。因此要改成:
stdin.put("correcthorsebatterystaple\n")原因二:SSH 需要一个 TTY。更深层的问题在于 SSH 的安全机制——它要求 STDIN 是一个真实的终端(TTY),以此确认自己在跟真人交互。sh 为此提供两个特殊关键字参数:
_tty_in=True:给子进程的 STDIN 挂一个伪终端(PTY)。从 OProc 构造逻辑 可见,tty_in=True时走pty.openpty()分支,而不是默认的os.pipe()。_unify_ttys=True:要求 STDOUT 与 STDIN 来自同一个伪终端。这是 SSH 的硬性要求,实现位于 single_tty 分支:通过pty.openpty()创建一对主/从 fd,再对同一组 fd 做os.dup,使子进程的 stdin、stdout 全部指向同一终端——这正是“man 7 pty”里描述的终端行为,也是让 SSH 输出正常的前提。
源码中还提供了配套校验:若_unify_ttys=True而_tty_in/_tty_out没有同时为True,会直接抛错(见 tty_in_validator)。
组合全部修正后的最终版:
from sh import ssh import sys aggregated = "" def ssh_interact(char, stdin): global aggregated sys.stdout.write(char.encode()) sys.stdout.flush() aggregated += char if aggregated.endswith("password: "): stdin.put("correcthorsebatterystaple\n") ssh("10.10.10.100", _out=ssh_interact, _out_bufsize=0, _tty_in=True, _unify_ttys=True)运行成功后的会话输出:
amoffat@10.10.10.100's password: Linux 10.10.10.100 testhost #1 SMP Tue Jun 21 10:29:24 EDT 2011 i686 GNU/Linux Ubuntu 10.04.2 LTS Welcome to Ubuntu! * Documentation: https://help.ubuntu.com/ 66 packages can be updated. 53 updates are security updates. Ubuntu 10.04.2 LTS Welcome to Ubuntu! * Documentation: https://help.ubuntu.com/ You have new mail. Last login: Thu Sep 13 03:53:00 2012 from some.ip.address amoffat@10.10.10.100:~$补充:_tty_in到底改变了什么?
从测试 test_tty_input 可以直观看到区别:一个 Python 子进程用os.isatty(sys.stdin.fileno())探测自己的 STDIN,默认管道启动时输出no tty attached!;加上_tty_in=True后它报告自己是 TTY,能正常进行“提示 → 输入 → 回显掩码”的完整交互。同理,test_tty_output 验证了_tty_out控制 STDOUT 是否为 TTY(注意_tty_out=True恰恰是默认值,见 默认参数表)。这一点对 SSH 尤为重要:默认的 STDOUT 已经是 PTY,但 STDIN 不是,必须靠_tty_in=True补齐。
五、进阶:用sh.contrib.ssh一行完成密码登录
上面四个步骤的手工拼装(_out_bufsize=0+_tty_in=True+_unify_ttys=True+ 聚合状态机)其实非常机械。sh 官方把它封装成了contrib 命令sh.contrib.ssh,源码 显示它本质就是对ssh做了bake:
cmd = orig.bake( _out_bufsize=0, _tty_in=True, _unify_ttys=True, _arg_preprocess=process )其中_arg_preprocess=process负责把password、interact、prompt、login_success等业务参数翻译成底层特殊的 kwargs 和回调对象;SSHInteract回调内部用SessionContent(维护chars/lines/cur_char/cur_line/last_line的滚动状态)代替了手工全局变量。使用方式:
from sh.contrib import ssh def ssh_interact(content, stdin): sys.stdout.write(content.cur_char) sys.stdout.flush() # 自动完成密码登录,并把登录后的会话内容逐字符交给 ssh_interact ssh("10.10.10.100", password="correcthorsebatterystaple", interact=ssh_interact)SessionContent的完整字段与含义(见 源码):
| 字段 | 说明 |
|---|---|
chars | 已收到的全部字符(deque,上限 50000) |
lines | 已收到的完整行(deque,上限 5000) |
last_line | 最近一行的内容 |
cur_char | 当前这个字符 |
cur_line | 正在累积的当前行 |
sshcontrib 还支持自定义触发与成功判定:password(密码,缺省则用getpass交互式询问)、prompt(自定义匹配密码提示符的回调,默认匹配password:结尾)、login_success(自定义登录成功判定,默认直接返回True)。细节见 contrib.rst 中的 SSH contrib 小节。
六、回归正途:真正该用的 SSH 自动化方式
从工程角度看,程序化输密码通常不是最优解——官方文档也明确建议优先用密钥认证。推荐的完整链路如下:
第一步:ssh-copy-id安装公钥。
ssh-copy-id yourservername首次会要求输入一次密码,之后即可免密登录。
第二步:利用 SSH “直接执行远程命令”的能力。SSH 允许把命令作为参数直接交给远端执行,不必进入远端 shell:
ssh amoffat@10.10.10.100 ifconfig在 sh 中对应的就是一次普通调用:
import sh print(sh.ssh("amoffat@10.10.10.100", "ifconfig"))第三步:用 baking 固化服务器地址,得到可复用的命令对象。
import sh my_server = sh.ssh.bake("amoffat@10.10.10.100") print(my_server("ifconfig")) print(my_server("whoami"))bake会把参数“预绑定”到命令对象上(底层记录在_partial_baked_args,见 Command 构造),每次调用只传变化的部分。
第四步:结合 subcommands 把远程命令变成属性访问。sh 会把未匹配到系统可执行文件的属性访问当作子命令参数展开:
import sh my_server = sh.ssh.bake("amoffat@10.10.10.100") print(my_server.ifconfig()) print(my_server.whoami())这样便得到一组近乎“本地化”的远程运维句柄:my_server.ifconfig()、my_server.whoami()……无需任何密码与交互逻辑。
七、要点回顾与适用边界
本文涉及的 sh 特殊关键字参数(均在 Command._call_args 默认值表 中有明确定义与默认值):
| 参数 | 默认值 | 作用 | 本文中的用法 |
|---|---|---|---|
_out | None | 重定向 STDOUT;传可调用对象即进入交互回调模式(回调自动获得 Queue 型 stdin) | 交互回调 |
_out_bufsize | 1(行缓冲) | 0=无缓冲(逐字符)、1=行缓冲、其他正整数=定长块 | 0,以捕获无换行的密码提示符 |
_tty_in | False | 为 STDIN 创建伪终端 | True,满足 SSH 对终端的校验 |
_tty_out | True | 为 STDOUT 创建伪终端 | 保持默认(_unify_ttys要求其为 True) |
_unify_ttys | False | 让 STDIN/STDOUT 共用同一个 PTY | True,SSH 正常工作前提 |
交互式进程自动化的通用四步法:
- 给
_out挂回调,第二参数接收 Queue 型 stdin,向其中put数据即为写入子进程输入(参考测试 test_stdout_callback_with_input 与 test_manual_stdin_queue); - 按提示符是否有换行选择
_out_bufsize(0最稳,但回调会收到碎片,需自行聚合); - 需要“回车确认”的输入记得补
\n; - 程序校验终端身份时(SSH、
sudo等),开启_tty_in(必要时_unify_ttys)。注意_tty_in与_unify_ttys组合存在强校验约束,参数搭配错误会直接报错。
需要说明的适用前提:以上行为基于本仓库当前版本的 Linux 实现(伪终端依赖pty模块,且_unify_ttys仅在tty_in与tty_out同时为真时合法);同时,交互式输密码仍建议仅用于学习与可控场景,生产环境应优先采用密钥认证(第六节的方案)。
- 开发工具
【免费下载链接】sh
Python process launching
相关推荐
使用sh库实现自动化SSH密码输入与远程命令执行
使用sh库实现自动化SSH密码输入与远程命令执行 前言 在自动化运维和脚本开发中,经常需要通过SSH连接到远程服务器执行命令。Python的sh库提供了强大的子
开发工具如何用Llamafile打造专属LLM应用:从单文件部署到自动化文档生成
如何用Llamafile打造专属LLM应用:从单文件部署到自动化文档生成 Llamafile是一个革命性的开源项目,它将复杂的大语言模型 LLM 打包成单个可执
人工智能大模型本地部署推理引擎如何在Audacity中实现专业级AI音频处理:OpenVINO插件的5个强大功能
如何在Audacity中实现专业级AI音频处理:OpenVINO插件的5个强大功能 还在为音频处理中的复杂操作而烦恼吗?音乐分离、语音转文字、智能降噪这些专业功
人工智能AI 应用音频语音音乐生成本地部署
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考