Python 学习笔记（第十三期）——运维自动化（下·前篇）：远程命令执行——paramiko基础篇

Python 学习笔记（第十三期）——运维自动化（下·前篇）：远程命令执行——paramiko基础篇

更多Python知识：Python基础入门

我的主页：AOwhisky，这里有更多运维系统性知识整理和其他有趣内容，欢迎与我一起探讨学习~

📖 引言：运维的“远程遥控器”

想象一下这样的场景：你管理着 50 台 Linux 服务器，需要在每台服务器上执行 df -h 检查磁盘空间，然后逐台分析结果。如果手动登录每一台服务器，输入命令，记录结果，等全部检查完，可能已经过去了一个小时，而最早检查的那台服务器可能又有了新的变化。

这就是运维工作中最常见的困境——重复性操作的效率瓶颈。

SSH（Secure Shell）是运维人员最常使用的远程管理协议，而 paramiko 是 Python 生态中最成熟的 SSH 客户端库。它让你能够用 Python 代码替代手工操作：批量执行命令、自动收集结果、集中分析数据，将运维效率提升一个数量级。

本期作为 paramiko 系列的第一篇，我们将从最基础、最常用的功能入手：建立 SSH 连接、执行远程命令、处理命令输出。学完本期，你将具备用 Python 批量管理服务器的基本能力。

— Compiled and Authored by Whisky — July 21
st, 2026

📑 本期目录

正文

一、paramiko 概述与环境部署

1.1 什么是 paramiko

paramiko 是一个基于 Python 实现的 SSHv2 协议客户端库，它提供了 SSH 连接、远程命令执行、文件传输（SFTP）、密钥认证等完整的 SSH 功能。

💡 白话理解：paramiko 就是一个“程序员的 SSH 客户端”——你用 PuTTY 或 ssh 命令能做的事情，paramiko 都能用 Python 代码来实现。而且因为它是程序化的，可以批量、自动、定时地执行操作。

paramiko vs 其他远程执行方式：

1.2 环境部署

paramiko 的安装依赖于一些底层库，下面分别介绍 Linux 和 Windows 下的部署方法。

Linux 环境下部署（CentOS / RHEL 系列）：

首先安装编译 Python3 所需的依赖包，然后编译安装 Python3，最后通过 pip 安装 paramiko。

# 1. 安装编译依赖
yum install -y openssl-devel bzip2-devel expat-devel gdbm-devel readline-devel \\
sqlite-devel libffi-devel gcc gcc-c++

# 2. 下载并编译安装 Python 3.13（可选，如果系统已有 Python3 可跳过）
tar zxvf Python-3.13.2.tgz
cd Python-3.13.2/
./configure –prefix=/usr/local/python3
make && make install

# 3. 创建软链接
ln -s /usr/local/python3/bin/python3 /usr/bin/python3
ln -s /usr/local/python3/bin/pip3 /usr/bin/pip

# 4. 验证 Python 版本
python3 –version

# 5. 配置 pip 国内镜像源（加速下载）
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple/

# 或手动编辑配置文件
vim /root/.config/pip/pip.conf

# 配置文件内容：
# [global]
# timeout = 40
# index-url = https://pypi.tuna.tsinghua.edu.cn/simple

# 6. 安装 paramiko
pip install paramiko

验证安装：

import paramiko
print(paramiko.__version__)

运行结果：

3.4.0

1.3 paramiko 的核心组件

paramiko 有两个最常用的核心类：

本期我们聚焦 SSHClient，下一期再深入 SFTPClient。

二、SSHClient 核心组件

2.1 创建 SSHClient 对象

import paramiko

# 创建 SSHClient 对象
ssh_client = paramiko.SSHClient()
print(f"SSHClient 对象：{ssh_client}")

2.2 connect() 方法详解

connect() 是 SSHClient 最核心的方法，用于建立 SSH 连接。它支持的参数非常丰富：

四种认证方式：

2.3 密码认证连接示例

import paramiko

hostname = "192.168.128.128"
username = "root"
password = "123123"

# 创建 SSHClient 对象
ssh_client = paramiko.SSHClient()

# 设置主机密钥策略（后面详细介绍）
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 建立连接
ssh_client.connect(
hostname=hostname,
port=22,
username=username,
password=password,
timeout=10,
allow_agent=False
)

print(f"✅ 成功连接到 {hostname}")

# 关闭连接
ssh_client.close()

运行结果：

✅ 成功连接到 192.168.128.128

2.4 私钥认证连接示例

import paramiko

hostname = "192.168.128.128"
username = "root"
key_filename = "/root/.ssh/id_rsa"

ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 私钥认证
ssh_client.connect(
hostname=hostname,
port=22,
username=username,
key_filename=key_filename,
timeout=10
)

print(f"✅ 使用私钥成功连接到 {hostname}")
ssh_client.close()

三、主机密钥策略（Host Key Policy）

3.1 什么是主机密钥策略

SSH 协议在首次连接时，会提示用户确认目标主机的指纹（fingerprint）：

The authenticity of host '192.168.128.128 (192.168.128.128)' can't be established.
ECDSA key fingerprint is SHA256:xxxxx.
Are you sure you want to continue connecting (yes/no)?

在自动化的脚本中，无法手动输入 yes，因此 paramiko 提供了三种自动处理策略：

3.2 三种策略的代码对比

import paramiko

# 策略1：AutoAddPolicy（自动添加）
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 效果：首次连接自动接受密钥，不提示

# 策略2：WarningPolicy（警告后接受）
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.WarningPolicy())
# 效果：日志中记录警告，但连接继续

# 策略3：RejectPolicy（拒绝，默认）
ssh = paramiko.SSHClient()
# 不设置策略时，默认使用 RejectPolicy
# 效果：首次连接会抛出 SSHException

⚠️ 生产环境安全建议：

• 在生产环境中，应使用 RejectPolicy（默认策略），并预先在 ~/.ssh/known_hosts 中配置所有目标主机的密钥
• 或者使用 WarningPolicy 并配合日志监控，及时发现异常
• AutoAddPolicy 虽然方便，但存在中间人攻击（MITM）风险，仅建议在隔离的测试环境中使用

3.3 预配置 known_hosts 的方法

如果希望在安全的前提下实现自动化，可以预先配置 known_hosts：

# 方法1：手动连接一次（输入 yes）
ssh root@192.168.128.128

# 方法2：使用 ssh-keyscan 批量添加
ssh-keyscan 192.168.128.128 >> ~/.ssh/known_hosts

# 批量添加多个主机
for ip in 192.168.128.128 192.168.128.129 192.168.128.130; do
ssh-keyscan $ip >> ~/.ssh/known_hosts
done

四、远程命令执行——exec_command()

4.1 基本用法

exec_command() 方法用于在远程主机上执行单条命令，返回三个文件对象：

stdin, stdout, stderr = ssh_client.exec_command(command)

4.2 执行简单命令

import paramiko

hostname = "192.168.128.128"
username = "root"
password = "123123"

ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

try:
ssh_client.connect(hostname=hostname, port=22, username=username, password=password)

# 执行命令
stdin, stdout, stderr = ssh_client.exec_command("free -m")

# 读取输出
print("命令执行结果：")
for line in stdout:
print(line.strip())

# 检查错误
error = stderr.read().decode('utf-8')
if error:
print(f"错误信息：{error}")

finally:
ssh_client.close()

运行结果：

命令执行结果：
total used free shared buff/cache available
Mem: 3869 534 587 25 2748 3026
Swap: 8191 0 8191

4.3 执行带参数的命令

# 使用列表参数（自动拼接）
stdin, stdout, stderr = ssh_client.exec_command("df -h /")

for line in stdout:
print(line.strip())

运行结果：

Filesystem Size Used Avail Use% Mounted on
/dev/sda2 50G 12G 35G 26% /

4.4 执行需要输入的命令

有些命令需要交互式输入（如 sudo、passwd）。这时可以通过 stdin 发送输入：

import paramiko
import time

ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(hostname="192.168.128.128", username="root", password="123123")

# 执行 sudo 命令（需要输入密码）
stdin, stdout, stderr = ssh_client.exec_command("sudo -S ls /root")

# 通过 stdin 发送密码
stdin.write("123123\\n")
stdin.flush() # 确保数据发送

# 读取输出
for line in stdout:
print(line.strip())

ssh_client.close()

运行结果：

anaconda-ks.cfg

⚠️ 注意：sudo -S 会从标准输入读取密码，-S 参数是必需的。

4.5 命令执行的状态检查

import paramiko

ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(hostname="192.168.128.128", username="root", password="123123")

# 执行一个可能失败的命令
stdin, stdout, stderr = ssh_client.exec_command("ls /non_existent_directory")

# 检查返回码（通过 channel）
# 注意：需要先读取输出，否则 channel 可能阻塞
stdout.read()
exit_code = stdout.channel.recv_exit_status()

if exit_code == 0:
print("✅ 命令执行成功")
else:
print(f"❌ 命令执行失败，返回码：{exit_code}")
error = stderr.read().decode('utf-8')
print(f"错误信息：{error}")

ssh_client.close()

运行结果：

❌ 命令执行失败，返回码：2
错误信息：ls: cannot access '/non_existent_directory': No such file or directory

五、命令执行结果处理

5.1 读取输出的方式

import paramiko

ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(hostname="192.168.128.128", username="root", password="123123")

stdin, stdout, stderr = ssh_client.exec_command("df -h")

# 方式1：逐行读取（适合大输出）
print("方式1：逐行读取")
for line in stdout:
print(f" {line.strip()}")

# 方式2：一次性读取全部（适合小输出）
print("方式2：一次性读取")
output = stdout.read().decode('utf-8')
print(output)

# 方式3：读取为列表
print("方式3：读取为列表")
lines = stdout.readlines()
for line in lines:
print(f" {line.strip()}")

ssh_client.close()

5.2 输出解码问题

stdout.read() 返回的是字节流（bytes），需要解码为字符串：

# ❌ 不推荐：直接使用字节流
data = stdout.read()
print(data) # b'total 4096\\\\n'

# ✅ 推荐：解码为字符串
text = stdout.read().decode('utf-8')
print(text) # total 4096

# ✅ 更推荐：使用 text=True 参数（需注意版本支持）
# Python 3.7+ 的 exec_command 支持 text 参数
stdin, stdout, stderr = ssh_client.exec_command("df -h", timeout=30)
# 但默认仍返回 bytes，建议显式解码

5.3 命令执行超时控制

exec_command() 支持 timeout 参数，防止命令卡死：

import paramiko

ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(hostname="192.168.128.128", username="root", password="123123")

try:
# 设置 5 秒超时
stdin, stdout, stderr = ssh_client.exec_command("sleep 10", timeout=5)

for line in stdout:
print(line.strip())

except paramiko.SSHException as e:
print(f"命令执行超时：{e}")

ssh_client.close()

运行结果：

命令执行超时：Timeout exceeded in channel operation

六、批量运维实战：多服务器内存巡检

需求描述：巡检 2 台及以上 Linux 服务器的内存使用情况，获取空闲内存（free）容量，按降序排序后写入文件。

设计思路：

• 将 SSHS 连接和命令执行封装为可复用的函数
• 循环连接所有目标服务器，收集内存空闲数据
• 对收集到的数据进行排序
• 将结果写入文件，每行一个值

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
多服务器内存巡检脚本
功能：批量获取多台服务器的空闲内存容量，排序后写入文件
作者：Whisky
"""

import paramiko
import time

def get_mem_free(host, port=22, user='root', pwd='123123'):
"""
连接目标服务器，获取空闲内存容量（MB）
返回：
success: True/False
mem_free: 空闲内存（MB），失败时返回 None
error: 错误信息
"""
ssh_client = None

try:
# 创建 SSH 客户端
ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 连接服务器
ssh_client.connect(
hostname=host,
port=port,
username=user,
password=pwd,
timeout=10,
allow_agent=False
)

# 执行获取内存空闲量的命令
# free -m | grep Mem | awk '{print $4}'
command = "free -m | grep Mem | awk '{print $4}'"
stdin, stdout, stderr = ssh_client.exec_command(command, timeout=10)

# 读取输出
output = stdout.read().decode('utf-8').strip()
error = stderr.read().decode('utf-8').strip()

if error:
return False, None, error

if not output:
return False, None, "命令无输出"

# 尝试转为整数
try:
mem_free = int(output)
return True, mem_free, None
except ValueError:
return False, None, f"无法解析输出：{output}"

except paramiko.AuthenticationException:
return False, None, "认证失败（用户名或密码错误）"
except paramiko.SSHException as e:
return False, None, f"SSH 连接异常：{e}"
except Exception as e:
return False, None, f"连接错误：{e}"
finally:
if ssh_client:
ssh_client.close()

def main():
"""主程序"""
print("=" * 60)
print(" 🔍 多服务器内存巡检工具 v1.0")
print("=" * 60)

# 定义待巡检的服务器列表
hosts = [
{"ip": "192.168.100.138", "user": "root", "pwd": "123123"},
{"ip": "192.168.100.140", "user": "root", "pwd": "123123"},
# 可继续添加更多服务器
]

# 存储结果
results = []
failed_hosts = []

print(f"开始巡检 {len(hosts)} 台服务器…\\n")

for idx, host_info in enumerate(hosts, 1):
host = host_info["ip"]
user = host_info["user"]
pwd = host_info["pwd"]

print(f"[{idx}/{len(hosts)}] 正在连接 {host} …")

success, mem_free, error = get_mem_free(host, user=user, pwd=pwd)

if success:
print(f" ✅ {host} 空闲内存：{mem_free} MB")
results.append({
"host": host,
"mem_free": mem_free
})
else:
print(f" ❌ {host} 采集失败：{error}")
failed_hosts.append({
"host": host,
"error": error
})

# 避免连接过快，稍作间隔
time.sleep(0.5)

# 输出汇总
print("\\n" + "=" * 60)
print(" 📊 采集结果汇总")
print("=" * 60)

if results:
# 按空闲内存从大到小排序
results.sort(key=lambda x: x["mem_free"], reverse=True)

print("\\n✅ 成功采集（按空闲内存降序）：")
print(" 服务器地址 空闲内存(MB)")
print(" " + "-" * 35)
for r in results:
print(f" {r['host']:20} {r['mem_free']:>6} MB")

# 写入文件
output_file = "memory_report.txt"
with open(output_file, "w", encoding="utf-8") as f:
f.write("服务器内存空闲报告\\n")
f.write("生成时间：" + time.strftime("%Y-%m-%d %H:%M:%S") + "\\n")
f.write("=" * 40 + "\\n")
for r in results:
f.write(f"{r['host']}: {r['mem_free']} MB\\n")

print(f"\\n💾 报告已保存到：{output_file}")

if failed_hosts:
print("\\n❌ 采集失败的服务器：")
for f in failed_hosts:
print(f" {f['host']}：{f['error']}")

print("\\n" + "=" * 60)
print(" 巡检完成！")
print("=" * 60)

if __name__ == "__main__":
main()

运行结果：

============================================================
🔍 多服务器内存巡检工具 v1.0
============================================================
开始巡检 2 台服务器…

[1/2] 正在连接 192.168.100.138 …
✅ 192.168.100.138 空闲内存：512 MB
[2/2] 正在连接 192.168.100.140 …
✅ 192.168.100.140 空闲内存：234 MB

============================================================
📊 采集结果汇总
============================================================

✅ 成功采集（按空闲内存降序）：
服务器地址 空闲内存(MB)
———————————–
192.168.100.138 512 MB
192.168.100.140 234 MB

💾 报告已保存到：memory_report.txt

============================================================
巡检完成！
============================================================

七、补充知识：日志记录

在生产环境中，运维脚本通常需要记录日志以便于审计和故障排查。paramiko 提供了日志功能：

import paramiko

# 记录 paramiko 日志到文件
paramiko.util.log_to_file('paramiko.log')

# 后续所有 paramiko 操作都会记录日志
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname="192.168.128.128", username="root", password="123123")

日志文件示例（paramiko.log）：

2026-07-20 10:23:45,123 [INFO] paramiko.transport: Connected to 192.168.128.128
2026-07-20 10:23:45,234 [INFO] paramiko.transport: Authentication successful
2026-07-20 10:23:45,345 [INFO] paramiko.channel: Channel opened

📝 总结与知识点一览表

核心 API 速查表

主机密钥策略对比

认证方式对比

常见错误排查指南

学习/生产最佳实践

🔜 下期预告

本期我们学习了 paramiko 的基础篇——从环境部署、SSHClient 的创建与连接认证、主机密钥策略，到远程命令执行（exec_command）和结果处理，最后通过“多服务器内存巡检”脚本将知识整合为实战工具。

下一期 ，我们将进入 paramiko 的进阶篇——远程文件传输。你将学习 SFTPClient 的使用，掌握如何通过 Python 代码实现文件的上传、下载、远程目录管理等操作，并通过“堡垒机场景下的文件分发”实战案例，将 SSH 命令执行与 SFTP 文件传输结合起来，构建更完整的自动化运维工具链。敬请期待！

本期（paramiko 基础篇）到此结束。 如果您在使用过程中发现任何问题或有改进建议，欢迎随时提出！

— Compiled and Authored by Whisky — July 21
st, 2026