paramiko 是一个强大的 Python 库，用于实现 SSHv2 协议，它不仅可以让你执行远程命令，还可以进行文件传输（SFTP）和建立 SSH 隧道。
（图片来源网络，侵删）
“交互”这个词在 SSH 场景下通常有两种含义：
- 执行单条命令并获取其输出：这是最常见的用法，也是自动化运维的基础，你发送一个命令（如
ls -l），然后读取并打印服务器返回的结果。 - 模拟一个真实的终端会话：这更复杂，用于需要持续交互的场景，比如登录路由器、进入
sudo密码提示、使用top命令或运行交互式脚本（如vim或bash）。
下面我们将分别详细讲解这两种交互方式,并提供完整的代码示例。
环境准备
确保你已经安装了 paramiko 库，如果没有，可以通过 pip 安装：
pip install paramiko
执行单条命令并获取输出（最常用）
这是最直接、最常用的方式，它通过 SSHClient 对象的 exec_command() 方法实现，这个方法会返回三个对象：
（图片来源网络，侵删）
stdin: 标准输入流，你可以向它写入数据（在sudo时输入密码）。stdout: 标准输出流，你需要从这里读取命令的返回结果。stderr: 标准错误流，如果命令执行失败，错误信息会在这里。
基本示例
import paramiko
# --- 1. 创建 SSHClient 对象 ---
ssh = paramiko.SSHClient()
# --- 2. 自动添加主机密钥 (不推荐用于生产环境) ---
# 第一次连接时，会提示你是否信任服务器的 host key。
# 为了方便测试，我们可以自动添加。
# 生产环境中，应该使用 ssh.get_host_keys().add() 来手动添加。
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
# --- 3. 连接服务器 ---
# 请替换为你的服务器信息
hostname = "your_server_ip"
port = 22
username = "your_username"
password = "your_password" # 或者使用密钥认证
print(f"正在连接到 {hostname}...")
ssh.connect(hostname, port=port, username=username, password=password)
print("连接成功！")
# --- 4. 执行命令 ---
command = "ls -l /tmp"
print(f"正在执行命令: '{command}'")
# exec_command 会返回三个流
stdin, stdout, stderr = ssh.exec_command(command)
# --- 5. 获取输出 ---
# stdout.read() 会读取所有输出，并解码为字符串
output = stdout.read().decode('utf-8')
error = stderr.read().decode('utf-8')
if error:
print(f"命令执行出错:\n{error}")
else:
print("命令执行成功，输出如下:")
print(output)
except paramiko.AuthenticationException:
print("认证失败，请检查用户名和密码。")
except paramiko.SSHException as e:
print(f"SSH 连接或命令执行出错: {e}")
except Exception as e:
print(f"发生未知错误: {e}")
finally:
# --- 6. 关闭连接 ---
if ssh:
ssh.close()
print("连接已关闭。")
关键点说明
ssh.set_missing_host_key_policy(): 这是处理“未知主机密钥”问题的关键。AutoAddPolicy会自动接收并保存服务器的公钥，这会带来“中间人攻击”的风险，在生产环境中，你应该预先知道服务器的公钥指纹，并使用ssh.get_host_keys().add()来添加它。- 认证方式: 除了密码，
paramiko还支持更安全的密钥对认证。# 使用密钥对认证 private_key_path = '/path/to/your/private_key' key = paramiko.RSAKey.from_private_key_file(private_key_path) ssh.connect(hostname, username=username, pkey=key)
stdin,stdout,stderr: 这三个流必须在with语句中或手动关闭，以释放资源。exec_command在所有流关闭后才会返回，上面的例子中，read()方法会一直等待直到流关闭，所以无需手动关闭，但养成良好的习惯总是好的。
模拟真实终端交互（高级用法）
当你需要执行一个需要持续输入输出的命令时（sudo su、htop 或一个交互式安装脚本），exec_command 就无能为力了，这时，你需要使用 paramiko.Channel 来建立一个持久的交互式会话。
这种方式的核心是：
- 创建一个
Channel。 - 在一个循环中,持续地从
stdout读取服务器的输出。 - 根据输出内容,判断下一步该发送什么命令或输入到
stdin。 - 处理超时,避免程序无限等待。
交互式登录示例（sudo）
这个例子模拟了输入 sudo 命令后，等待密码提示，然后输入密码的过程。
import paramiko
import time
import select
def interactive_shell(hostname, username, password, sudo_password=None):
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
print(f"正在连接到 {hostname}...")
ssh.connect(hostname, port=22, username=username, password=password)
print("连接成功！")
# --- 1. 创建一个交互式 Channel ---
channel = ssh.invoke_shell()
print("已进入交互式 Shell。")
# 设置一个超时，避免无限等待
# 等待5秒内没有输出就认为当前命令执行完毕
timeout = 5
# --- 2. 交互式循环 ---
while True:
# 使用 select 来检查 channel 是否有数据可读
# 这比直接 channel.recv() 更高效，可以避免阻塞
if channel.recv_ready():
# 读取并打印服务器的输出
output = channel.recv(4096).decode('utf-8')
print(output, end='')
# 如果输出中包含密码提示，就输入密码
if sudo_password and "password for" in output.lower():
print("检测到 sudo 密码提示，正在输入密码...")
channel.send(sudo_password + '\n')
# 输入密码后，清空 sudo_password 变量，避免重复输入
sudo_password = None
# 检查 channel 是否关闭
if channel.exit_status_ready():
print("\n服务器已关闭连接。")
break
# 短暂休眠，避免 CPU 占用过高
time.sleep(0.1)
except paramiko.AuthenticationException:
print("认证失败，请检查用户名和密码。")
except paramiko.SSHException as e:
print(f"SSH 连接或命令执行出错: {e}")
except Exception as e:
print(f"发生未知错误: {e}")
finally:
if ssh:
ssh.close()
print("连接已关闭。")
# --- 使用示例 ---
if __name__ == "__main__":
# 请替换为你的信息
HOSTNAME = "your_server_ip"
USERNAME = "your_username"
PASSWORD = "your_password"
SUDO_PASSWORD = "your_sudo_password" # 如果不需要 sudo，可以设为 None
interactive_shell(HOSTNAME, USERNAME, PASSWORD, SUDO_PASSWORD)
交互式 Shell 示例 (top)
这个例子展示了如何启动一个持续输出的命令,并在一段时间后终止它。
import paramiko
import time
def run_interactive_command(hostname, username, password, command, duration=10):
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(hostname, port=22, username=username, password=password)
channel = ssh.invoke_shell()
print(f"正在执行命令: '{command}' 并运行 {duration} 秒...")
# 发送命令
channel.send(command + '\n')
start_time = time.time()
while True:
# 检查是否超时
if time.time() - start_time > duration:
print("\n时间到，正在终止命令...")
# 发送 Ctrl+C 来终止命令
channel.send('\x03') # \x03 是 EOT (End of Transmission) 字符，通常代表 Ctrl+C
break
# 读取并打印输出
if channel.recv_ready():
output = channel.recv(1024).decode('utf-8')
print(output, end='')
time.sleep(0.1)
# 等待一会儿，确保命令被终止
time.sleep(2)
print("\n命令执行结束。")
except Exception as e:
print(f"发生错误: {e}")
finally:
if ssh:
ssh.close()
print("连接已关闭。")
# --- 使用示例 ---
if __name__ == "__main__":
HOSTNAME = "your_server_ip"
USERNAME = "your_username"
PASSWORD = "your_password"
# 运行 top 命令 5 秒
run_interactive_command(HOSTNAME, USERNAME, PASSWORD, "top", duration=5)
总结与最佳实践
| 特性 | exec_command() |
invoke_shell() |
|---|---|---|
| 用途 | 执行简单的、一次性的命令 | 模拟真实终端，用于持续交互 |
| 返回值 | 三个独立的流 (stdin, stdout, stderr) |
一个 Channel 对象，通过 send() 和 recv() 交互 |
| 复杂度 | 简单，易于使用 | 复杂，需要处理循环、超时和输出解析 |
| 适用场景 | 文件操作 (ls, cp)、软件安装 (yum install)、系统信息查询 (df -h) |
sudo、su、top、htop、vim、交互式脚本 |
| 资源管理 | 命令执行完毕后自动关闭流 | 需要手动管理 Channel 的生命周期 |
最佳实践建议：
- 优先使用
exec_command()：对于绝大多数自动化任务，exec_command()是更简单、更安全、更高效的选择。 - 仅在必要时使用
invoke_shell()：只有当你明确需要模拟一个真实用户坐在终端前操作的场景时，才使用invoke_shell()。 - 处理异常和资源：始终使用
try...finally或with语句来确保SSHClient和Channel被正确关闭，防止资源泄露。 - 考虑安全性：避免在代码中硬编码密码，优先使用 SSH 密钥对进行认证，如果必须使用密码，可以考虑使用环境变量或配置文件来存储。
- 处理输出编码：服务器返回的输出编码可能是
utf-8，也可能是其他编码（如gbk），使用.decode('utf-8', errors='ignore')可以避免因编码问题导致的程序崩溃。
