8.7 KiB
以下是为提供的 Python 代码编写的 Markdown 格式技术文档，涵盖了模块功能、类说明、方法描述及使用示例。
SSH 连接管理模块技术文档
概述
本模块基于 asyncssh 实现了异步 SSH 客户端连接能力，支持通过跳板机（Jump Server）链式连接远程主机，并提供文件传输、命令执行和交互式 Shell 等高级功能。适用于批量运维操作、自动化部署与安全网络环境下的远程控制。
主要特性：
- 异步非阻塞 I/O，高并发处理多台服务器
- 支持密码或密钥认证
- 支持多级跳板机穿透
- 提供 SCP 文件上传下载接口
- 支持交互式命令执行（如 Bash）
- 封装简洁易用的高层 API
依赖库：
pip install asyncssh shlex
模块导入说明
import os
import sys
import time
import shlex
from traceback import format_exc
from contextlib import asynccontextmanager
from functools import partial
from threading import Thread
from appPublic.myTE import tmpTml
from appPublic.log import debug, exception
import asyncio
import asyncssh
注意：
appPublic.*是项目自定义工具包，包含日志记录与模板引擎等功能。
核心类说明
1. SSHServer
表示一个可通过跳板机访问的目标 SSH 服务器。
初始化参数
| 参数 | 类型 | 说明 |
|---|---|---|
server |
dict | 目标服务器配置，格式见下表 |
jumpservers |
list[dict] | 跳板机列表（可选），每个元素结构同 server |
server 字典字段说明：
| 键名 | 类型 | 默认值 | 说明 |
|---|---|---|---|
host |
str | 必填 | 主机地址（IP 或域名） |
username |
str | 'root' |
登录用户名 |
port |
int | 22 |
SSH 端口 |
password |
str | None |
登录密码（优先级低于密钥） |
client_keys |
list[str] | [] |
私钥路径列表（如 ['~/.ssh/id_rsa']） |
passphrase |
str | None |
密钥加密口令 |
jumpservers |
list[dict] | None |
内嵌跳板机配置（若未传构造函数参数） |
方法
_connect_server(server: dict, refconn=None) -> asyncssh.SSHClientConnection
内部方法，建立到指定服务器的连接。
- 若
refconn存在，则通过已有连接隧道建立新连接（用于跳板穿透） - 支持密钥或密码认证自动选择
get_connector() -> Async Context Manager
上下文管理器，返回目标主机连接对象，自动处理连接建立与释放。
示例：
async with ssh_server.get_connector() as conn:
result = await conn.run('ls -l')
print(result.stdout)
自动关闭所有跳板机连接，异常时捕获并记录日志。
2. SSHNode
增强版 SSH 节点，封装完整生命周期管理与多种操作接口。
初始化参数
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
host |
str | 必填 | 目标主机地址 |
username |
str | 'root' |
用户名 |
port |
int | 22 |
SSH 端口 |
password |
str | None |
登录密码 |
client_keys |
list[str] | [] |
客户端私钥路径 |
passphrase |
str | None |
私钥解密口令 |
jumpers |
list[dict] | [] |
跳板机列表 |
属性
conn: 当前节点主连接 (asyncssh.SSHClientConnection)jumper_conns: 所有跳板连接列表batch_cmds: 待执行命令队列（暂未完全实现）
方法
| 方法 | 功能 |
|---|---|
info() → dict |
返回当前节点信息快照 |
asjumper() → list[dict] |
返回可用于其他节点跳板的配置列表 |
set_jumpers(jumpers) |
动态设置跳板机 |
connect() |
建立完整连接链（跳板 + 主机） |
close() |
关闭所有连接资源 |
run(cmd, ...) |
执行单条命令（自动连接/关闭） |
_cmd, _run, _process, _l2r, _r2l |
内部操作封装 |
run(cmd, input=None, stdin=None, stdout=None)
执行远程命令，支持以下特殊语法：
l2r <local_path> <remote_path>：本地 → 远程复制（SCP）r2l <remote_path> <local_path>：远程 → 本地复制（SCP）
否则调用 conn.run() 执行普通命令。
_xcmd(cmd, xmsgs=[], ns={}, show_input=None, show_stdout=None)
执行交互式命令，支持定时输入模拟。
xmsgs: 输入消息列表，格式[ (delay_sec, template_string), ... ]ns: 模板变量命名空间show_input,show_stdout: 回调函数，用于输出监控
示例：自动登录 CLI 工具、数据库等需要逐步输入的场景。
3. SSHNodes
批量管理多个 SSHNode 的容器类，支持并行执行。
初始化参数
| 参数 | 类型 | 说明 |
|---|---|---|
nodes |
list[str] | 主机名/IP 列表 |
username |
str | 统一用户名 |
port |
int | 统一端口 |
jumpers |
list[dict] | 共享跳板机配置 |
方法
| 方法 | 功能 |
|---|---|
append_cmd(cmd, stdin, stdout) |
添加待批处理命令 |
run(cmd, ...) |
并发运行命令于所有节点 |
exe_batch() |
执行预设批处理命令队列（注意：当前存在 bug） |
使用
asyncio.gather(..., return_exceptions=True)避免单个失败中断整体流程。
示例
hosts = ['192.168.1.10', '192.168.1.11']
jump = {"host": "jumper.example.com", "username": "jumpuser", "port": 22}
cluster = SSHNodes(hosts, jumpers=[jump])
results = await cluster.run("uptime")
for r in results:
if isinstance(r, Exception):
print("Error:", r)
else:
print(r.stdout.strip())
4. SSHBash
实现交互式远程 Bash Shell 会话，支持实时输入输出转发。
构造函数
SSHBash(node: SSHNode, loop=None)
node: 已配置的SSHNode实例loop: 可选事件循环（默认使用当前）
内部启动独立线程运行子事件循环，避免阻塞主线程。
核心方法
run(read_co, write_co)
启动交互式 Bash 会话。
read_co: 异步可调用对象，用于读取用户输入（例如从stdin）write_co: 异步回调，用于写入输出内容（如打印到终端）
底层使用
create_process('bash', term_type='xterm-256color')
feed_stdin(f)
将输入数据推送到远程进程 stdin。
exit()
清理资源：关闭连接、停止子线程事件循环。
使用示例
async def read_input():
return os.read(sys.stdin.fileno(), 65535)
async def write_output(data):
sys.stdout.write(data.decode('utf-8'))
sys.stdout.flush()
bash = SSHBash(my_ssh_node)
await bash.run(read_input, write_output)
5. SshConnector
轻量级连接包装器，提供更直观的操作接口。
⚠️ Bug 提示：
l2r方法中recurse=Tree应为recurse=True
方法
| 方法 | 功能 |
|---|---|
r2l(rf, lf) |
下载文件：远程 → 本地 |
l2r(lf, rf) |
上传文件：本地 → 远程 |
run_process(*args, **kw) |
创建远程进程 |
run(cmdline, ...) |
执行命令并返回结果 |
辅助函数与入口点
main()
命令行交互入口，演示如何动态输入命令对多个主机批量执行。
python script.py "df -h" host1 host2 ...
当前版本进入无限循环等待用户输入。
test_sshbash()
测试交互式 Bash 功能的示例函数。
已知问题与改进建议
| 问题 | 描述 | 建议修复 |
|---|---|---|
SSHNodes.__init__ 拼写错误 |
usernmae → username |
更正拼写 |
exe_batch() 中 return_excetion=True |
拼写错误导致异常不被捕获 | 改为 return_exceptions=True |
exe_batch() 在 gather 后仍有代码 |
return 后不会执行后续行 |
移动 for 循环至 gather 之后 |
SshConnector.l2r 中 self.comm |
应为 self.conn |
修正属性引用 |
show_result() 中变量 e 未定义 |
print('Exception:', e) 报错 |
改为 print('Exception:', x) |
recurse=Tree |
应为布尔值 True |
修改为 recurse=True |
总结
该模块构建了一个灵活高效的异步 SSH 操作框架，特别适合在受限网络环境中通过跳板机批量管理服务器。结合 asyncssh 的强大功能，实现了文件传输、命令执行、交互式 Shell 等核心运维需求。
建议进一步完善单元测试、增加连接池、超时控制与重试机制以提升稳定性。
📌 维护建议：
- 添加类型注解（Type Hints）
- 补充 docstring 文档字符串
- 分离配置加载逻辑
- 增加连接健康检查机制
✅ 适用场景：
- 自动化部署系统
- 多数据中心运维脚本
- DevOps 流水线中的远程操作环节
- 内网服务器集中管理平台
文档版本：v1.0
最后更新：2025年4月5日