☰
用 Python sh 库编程式交互进程:以自动化 SSH 密码登录为例
2026/9/29 3:06:30 网站建设 项目流程
  • 开发工具

【免费下载链接】sh

Python process launching

项目地址:https://gitcode.com/gh_mirrors/sh/sh
点击查看免费下载

本文基于 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 默认值表 中有明确定义与默认值):

参数默认值作用本文中的用法
_outNone重定向 STDOUT;传可调用对象即进入交互回调模式(回调自动获得 Queue 型 stdin)交互回调
_out_bufsize1(行缓冲)0=无缓冲(逐字符)、1=行缓冲、其他正整数=定长块0,以捕获无换行的密码提示符
_tty_inFalse为 STDIN 创建伪终端True,满足 SSH 对终端的校验
_tty_outTrue为 STDOUT 创建伪终端保持默认(_unify_ttys要求其为 True)
_unify_ttysFalse让 STDIN/STDOUT 共用同一个 PTYTrue,SSH 正常工作前提

交互式进程自动化的通用四步法:

  1. 给_out挂回调,第二参数接收 Queue 型 stdin,向其中put数据即为写入子进程输入(参考测试 test_stdout_callback_with_input 与 test_manual_stdin_queue);
  2. 按提示符是否有换行选择_out_bufsize(0最稳,但回调会收到碎片,需自行聚合);
  3. 需要“回车确认”的输入记得补\n;
  4. 程序校验终端身份时(SSH、sudo等),开启_tty_in(必要时_unify_ttys)。注意_tty_in与_unify_ttys组合存在强校验约束,参数搭配错误会直接报错。

需要说明的适用前提:以上行为基于本仓库当前版本的 Linux 实现(伪终端依赖pty模块,且_unify_ttys仅在tty_in与tty_out同时为真时合法);同时,交互式输密码仍建议仅用于学习与可控场景,生产环境应优先采用密钥认证(第六节的方案)。

  • 开发工具

【免费下载链接】sh

Python process launching

项目地址:https://gitcode.com/gh_mirrors/sh/sh
点击查看免费下载

相关推荐

上一篇:网盘直链下载助手终极指南:告别限速,实现全平台高速下载
下一篇:7天掌握Minecraft地图编辑:MCEdit 2.0终极免费专业工具完全指南

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询