Python多机管理-paramiko模块基础（自动化运维）

Paramiko 详解：Python 远程运维核心库

1. Paramiko 概述

1.1 什么是 Paramiko

Paramiko 是 Python 中用于实现 SSH2 协议的第三方库，核心能力覆盖远程运维场景的关键需求，包括：

  • 远程命令执行（如批量执行 shell 命令）
  • 安全文件传输（基于 SFTP 协议）
  • 端口转发与隧道（实现跨网络服务访问）
  • RSA/DSA 密钥认证（替代密码登录，提升安全性）

1.2 核心功能模块

模块/类 功能描述
SSHClient 封装 SSH 会话，管理连接建立、命令执行流程
SFTPClient 提供 SFTP 协议接口，支持文件上传/下载/删除
Transport 底层连接管理，支持连接复用、协议扩展
Channel SSH 通道对象，实现交互式会话或端口转发
RSAKey/DSSKey 处理 RSA/DSS 密钥的生成、加载与验证

1.3 工作流程

  1. 建立连接：通过 SSHClient() 初始化客户端对象
  2. 认证方式：支持三种认证模式（密码认证、密钥认证、混合认证）
  3. 执行操作：根据需求调用命令执行、文件传输或端口转发接口
  4. 关闭连接：显式调用 close() 方法，或使用 with 语句自动释放连接

1.4 安装与依赖

Paramiko 依赖 cryptography 等加密库，直接通过 pip 安装即可（建议使用 Python 3.6+ 版本）：

1
pip install paramiko

2. 核心功能实践

2.1 SFTPClient 基础（文件传输）

SFTPClient 类似于 SFTP 工具，可实现远程服务器文件的上传与下载，核心参数：

  • 第一个参数：远程文件路径
  • 第二个参数：本地文件路径

2.1.1 下载文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import paramiko
import os

remote_ip = '192.168.20.130'
username = 'sftpuser'
password = '202019'
remote_file = 'flag'  # 远程文件（默认路径：/home/sftpuser/flag）
local_dir = '/root/Paramiko'  # 本地保存目录
local_file = os.path.join(local_dir, 'flag.txt')  # 本地文件完整路径

try:
    # 确保本地目录存在
    if not os.path.exists(local_dir):
        os.makedirs(local_dir)
        print(f"已创建本地目录：{local_dir}")

    # 建立 SFTP 连接
    transport = paramiko.Transport((remote_ip, 22))
    transport.connect(username=username, password=password)
    sftp = paramiko.SFTPClient.from_transport(transport)
    print("✅ SFTP 连接成功！")

    # 执行下载
    sftp.get(remote_file, local_file)
    print(f"✅ 下载成功！")
    print(f"远程文件：/home/sftpuser/flag")
    print(f"本地保存路径：{local_file}")

    # 验证文件完整性
    if os.path.exists(local_file):
        file_size = os.path.getsize(local_file)
        print(f"本地文件大小：{file_size} 字节（与远程一致）")

except Exception as e:
    print(f"❌ 下载失败：{e}")
finally:
    # 关闭连接
    if 'transport' in locals() and transport.is_active():
        transport.close()
        print("✅ SFTP 连接已关闭")

下载文件结果

2.1.2 上传文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import paramiko
import os

remote_ip = '192.168.20.130'
remote_port = 22
username = 'sftpuser'
password = '202019'

local_file = '/root/Paramiko/flag.txt'  # 本地待上传文件
remote_dir = 'upload'  # 远程目标目录
remote_file = f"{remote_dir}/flag_191.txt"  # 远程文件完整路径

try:
    # 验证本地文件存在
    if not os.path.exists(local_file):
        raise FileNotFoundError(f"本地文件不存在：{local_file}")
    print(f"✅ 本地文件验证通过：{local_file}")

    # 建立 SFTP 连接
    transport = paramiko.Transport((remote_ip, remote_port))
    transport.connect(username=username, password=password)
    sftp = paramiko.SFTPClient.from_transport(transport)
    print("✅ SFTP 连接成功！")

    # 确保远程目录存在
    try:
        sftp.stat(remote_dir)
    except FileNotFoundError:
        sftp.mkdir(remote_dir)
        print(f"✅ 已创建远程目录：{remote_dir}")

    # 执行上传
    print(f"\n开始上传：{local_file} -> {remote_file}")
    sftp.put(local_file, remote_file)
    print("✅ 上传完成！")

    # 验证远程文件
    remote_file_attr = sftp.stat(remote_file)
    print(f"\n远程文件验证：")
    print(f"路径：{remote_file}")
    print(f"大小：{remote_file_attr.st_size} 字节（与本地一致）")
    print(f"权限：{oct(remote_file_attr.st_mode)[-3:]}")

except FileNotFoundError as e:
    print(f"\n❌ 错误：{e}")
except PermissionError as e:
    print(f"\n❌ 错误：权限不足！{e}")
    print("请检查：1. 本地文件是否有读取权限；2. 远程目录是否有写入权限")
except Exception as e:
    print(f"\n❌ 上传失败：{e}")
finally:
    # 关闭连接
    if 'transport' in locals() and transport.is_active():
        transport.close()
        print("\n✅ SFTP 连接已关闭")

上传文件结果

2.2 SSHClient 基础（远程命令执行）

SSHClient 类似于 Linux 的 ssh 命令，可通过 SSH 协议对远程服务器执行 shell 命令。

2.2.1 密码连接

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import paramiko

# 初始化 SSH 客户端
ssh_client = paramiko.SSHClient()
# 自动添加未知主机密钥（避免首次连接确认）
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 建立密码认证连接
ssh_client.connect(
    hostname='192.168.20.130',
    port=22,
    username='root',
    password='202019'
)

# 执行命令（多个命令用分号分隔）
stdin, stdout, stderr = ssh_client.exec_command('pwd;ls -all')

# 输出结果（解码为 UTF-8 字符串）
print("标准输出：")
print(stdout.read().decode('utf-8'))
print("错误输出：")
print(stderr.read().decode('utf-8'))

密码连接执行效果

2.2.2 密钥连接

前置准备：生成并配置密钥
  1. 本地生成密钥对（以 ed25519 算法为例）：
    1
    ssh-keygen -t ed25519 -C "sftpuser@192.168.20.130"
    默认密钥路径：
  • 私钥：/root/.ssh/id_ed25519（本地保存，切勿泄露）
  • 公钥：/root/.ssh/id_ed25519.pub（需上传到远程服务器）

生成密钥

  1. 配置远程服务器公钥：
    将本地公钥内容复制到远程服务器的 ~/.ssh/authorized_keys 文件中：
    1
    2
    3
    # 远程服务器执行
    vim ~/.ssh/authorized_keys
    # 粘贴本地 id_ed25519.pub 内容，保存退出
密钥连接代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import paramiko
import os

remote_ip = '192.168.20.130'
remote_port = 22
username = 'root'
private_key_path = '/root/.ssh/id_ed25519'  # 本地私钥路径
local_file = '/root/Paramiko/upload.txt'    # 本地待上传文件
remote_file = 'upload/test_uploaded_by_key.txt'  # 远程目标路径

ssh = None
try:
    # 初始化 SSH 客户端
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    # 密钥认证连接
    ssh.connect(
        hostname=remote_ip,
        port=remote_port,
        username=username,
        key_filename=private_key_path,  # 指定私钥文件
        timeout=10,
        look_for_keys=False,  # 不自动查找其他密钥
        allow_agent=False     # 禁用 SSH 代理
    )
    print(f"✅ 已连接到 {username}@{remote_ip}（免密认证成功）")

    # 上传文件（基于 SSH 连接创建 SFTP 客户端）
    print(f"\n=== 开始上传文件：{local_file} ===")
    sftp = ssh.open_sftp()
    try:
        sftp.stat('upload')  # 检查远程目录是否存在
    except FileNotFoundError:
        sftp.mkdir('upload')  # 不存在则创建
        print(f"✅ 已创建远程目录：upload")
    
    sftp.put(local_file, remote_file)
    print(f"✅ 上传完成：{local_file} -> {remote_file}")
    
    # 验证文件完整性
    if os.path.getsize(local_file) == sftp.stat(remote_file).st_size:
        print("✅ 文件完整性验证通过")
    sftp.close()

except paramiko.AuthenticationException:
    print("❌ 认证失败！")
    print("当前使用的私钥：", private_key_path)
    print("请检查公钥是否已配置到远程服务器")
except Exception as e:
    print(f"❌ 操作失败：{type(e).__name__} -> {e}")
finally:
    # 关闭连接
    if ssh:
        transport = ssh.get_transport()
        if transport and transport.is_active():
            ssh.close()
            print("\n✅ 连接已关闭")

密钥连接执行效果