自动化运维-Python-paramiko模块基础

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. 实例 1：使用密码认证（含文件下载）

2.1 场景说明

通过「用户名+密码」登录远程服务器，执行基础命令（如 pwd/ls -l），并下载远程文件（示例：/root/flag.txt）到本地。

注意：建议在原生 CMD/PowerShell 中运行脚本，避免虚拟环境依赖冲突；敏感信息（如密码）实际使用时建议通过环境变量或配置文件管理。

2.2 配置参数说明

参数 示例值 说明
username root 远程服务器用户名
password your_remote_password 远程服务器登录密码
timeout 10 连接超时时间（单位：秒）
host 192.168.xxx.xxx 远程服务器 IP 地址
remote_file /root/flag.txt 远程文件路径
local_file C:/Users/ZhuanZ/Desktop/flag.txt 本地保存路径

2.3 完整代码

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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
""" 使用密码连接远程服务器，并支持命令执行与文件下载 """
import paramiko
import os
from dotenv import load_dotenv

def AutoPassConn(host, username, password, port=22, timeout=10):
    """
    建立 SSH 密码认证连接
    :param host: 远程服务器 IP/域名
    :param username: 登录用户名
    :param password: 登录密码
    :param port: SSH 端口（默认 22）
    :param timeout: 连接超时时间（默认 10 秒）
    :return: 成功返回 SSHClient 对象，失败返回 None
    """
    try:
        # 初始化 SSH 客户端
        client = paramiko.SSHClient()
        # 自动添加未知主机密钥（避免首次连接时的确认提示）
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        print(f"正在连接 {host}:{port}...")
        client.connect(
            hostname=host,
            port=port,
            username=username,
            password=password,
            timeout=timeout
        )
        print(f"成功连接到主机: {host}")
        return client
        
    except paramiko.AuthenticationException:
        print("❌ 认证失败，请检查用户名和密码")
        return None
    except paramiko.SSHException as ssh_ex:
        print(f"❌ SSH 连接错误: {str(ssh_ex)}")
        return None
    except Exception as ex:
        print(f"❌ 未知错误: {str(ex)}")
        return None

def ExecuteCommand(client, command):
    """
    执行远程命令并返回结果
    :param client: 已连接的 SSHClient 对象
    :param command: 待执行的 shell 命令（字符串）
    :return: 字典格式结果（status: 退出状态码, stdout: 标准输出, stderr: 错误输出）
    """
    try:
        print(f"\n📝 执行命令: {command}")
        # 执行命令（stdin 为输入流，stdout 为输出流，stderr 为错误流）
        stdin, stdout, stderr = client.exec_command(command)
        
        # 获取命令退出状态码（0 表示成功，非 0 表示失败）
        exit_status = stdout.channel.recv_exit_status()
        # 读取输出内容（需解码为 UTF-8 字符串）
        output = stdout.read().decode('utf-8')
        error = stderr.read().decode('utf-8')
        
        print(f"✅ 命令执行完毕，退出状态: {exit_status}")
        if output:
            print("📤 标准输出:")
            print(output.strip())
        if error:
            print("⚠️  错误输出:")
            print(error.strip())
            
        return {
            'status': exit_status,
            'stdout': output,
            'stderr': error
        }
    except Exception as ex:
        print(f"❌ 执行命令时出错: {str(ex)}")
        return None

def FileGet(client, remote_path, local_path):
    """
    从远程服务器下载文件到本地
    :param client: 已连接的 SSHClient 对象
    :param remote_path: 远程文件的完整路径（如 /root/flag.txt）
    :param local_path: 本地保存的完整路径（如 C:/Desktop/flag.txt）
    :return: 下载成功返回 True，失败返回 False
    """
    try:
        # 基于已有的 SSH 连接创建 SFTP 客户端（with 语句自动关闭 SFTP 连接）
        with client.open_sftp() as sftp:
            print(f"\n📥 正在下载文件: {remote_path} -> {local_path}")
            
            # 先检查远程文件是否存在
            try:
                sftp.stat(remote_path)
            except FileNotFoundError:
                print(f"❌ 错误: 远程文件 {remote_path} 不存在")
                return False
                
            # 检查本地目录是否存在，不存在则创建
            local_dir = os.path.dirname(local_path)
            if not os.path.exists(local_dir):
                os.makedirs(local_dir, exist_ok=True)
                print(f"📁 已创建本地目录: {local_dir}")
                
            # 执行下载
            sftp.get(remote_path, local_path)
            print(f"✅ 文件下载成功: {local_path}")
            return True
            
    except Exception as ex:
        print(f"❌ 文件下载失败: {str(ex)}")
        return False

if __name__ == "__main__":
    # 加载环境变量（若使用 .env 文件存储敏感信息，需确保文件存在）
    load_dotenv()
    
    # 配置参数（实际使用时建议替换为自己的服务器信息）
    config = {
        'host': '192.168.xxx.xxx',          # 替换为远程服务器 IP
        'username': 'root',                 # 替换为登录用户名
        'password': 'your_remote_password', # 替换为登录密码
        'remote_file': '/root/flag.txt',    # 替换为远程文件路径
        'local_file': 'C:/Users/ZhuanZ/Desktop/python脚本/Parmiko_OS/flag.txt' # 替换为本地保存路径
    }
    
    # 建立 SSH 连接
    ssh_client = AutoPassConn(
        config['host'],
        config['username'],
        config['password']
    )
    
    if ssh_client:
        try:
            # 执行远程命令（示例：查看当前路径 + 列出文件详情）
            ExecuteCommand(ssh_client, 'pwd;ls -l')
            
            # 下载远程文件
            if FileGet(ssh_client, config['remote_file'], config['local_file']):
                print("\n🎉 文件下载任务完成☆*: .｡. o(≧▽≦)o .｡.:*☆")
            else:
                print("\n❌ 文件下载任务失败")
                
        finally:
            # 无论执行结果如何，均关闭 SSH 连接
            ssh_client.close()
            print("\n🔌 SSH 连接已关闭")

2.4 执行效果

密码认证执行效果

3. 实例 2：使用密钥认证（更安全）

3.1 场景说明

密码认证存在泄露风险，密钥认证通过非对称加密（公钥+私钥）实现登录，更适合生产环境。需先完成「本地私钥 -> 远程服务器公钥」的配置，再通过脚本实现密钥登录。

3.2 前置准备（密钥配置）

  1. 获取本地私钥：
    本地私钥默认路径为 C:\Users\<你的用户名>\.ssh\id_rsa（Windows）或 ~/.ssh/id_rsa（Linux/Mac）。若用记事本打开乱码，需确认文件编码为「UTF-8」（无 BOM）。

  2. 上传公钥到远程服务器：
    将本地公钥（id_rsa.pub 内容）追加到远程服务器的 ~/.ssh/authorized_keys 文件中：

    1
    2
    3
    # 远程服务器执行（需替换为你的公钥内容）
    vim /root/.ssh/authorized_keys
    # 粘贴本地 id_rsa.pub 的内容，保存退出

  3. 验证密钥登录：
    本地通过终端测试登录，若无需输入密码则配置成功：

    1
    ssh root@192.168.xxx.xxx

3.3 完整代码

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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
""" 使用 RSA 密钥连接远程服务器，并执行命令 """
import paramiko
from paramiko.ssh_exception import SSHException

def connect_with_key(host, username, key_path, port=22, passphrase=None, timeout=10):
    """
    建立 SSH 密钥认证连接
    :param host: 远程服务器 IP/域名
    :param username: 登录用户名
    :param key_path: 本地私钥文件路径（如 C:/Users/ZhuanZ/.ssh/id_rsa）
    :param port: SSH 端口（默认 22）
    :param passphrase: 私钥密码（若无则为 None）
    :param timeout: 连接超时时间（默认 10 秒）
    :return: 成功返回 SSHClient 对象，失败返回 None
    """
    try:
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        # 加载本地私钥
        private_key = paramiko.RSAKey.from_private_key_file(
            filename=key_path, 
            password=passphrase  # 若私钥有密码，需传入该参数
        )
        
        # 建立密钥认证连接
        client.connect(
            hostname=host,
            port=port,
            username=username,
            pkey=private_key,
            timeout=timeout
        )
        
        print(f"✅ 成功使用密钥连接到 {host}:{port}")
        return client
    
    except FileNotFoundError:
        print(f"❌ 错误：私钥文件 {key_path} 不存在")
    except paramiko.PasswordRequiredException:
        print("❌ 错误：私钥已加密，需提供 passphrase 参数")
    except paramiko.AuthenticationException:
        print("❌ 认证失败：请检查私钥有效性或用户名")
    except SSHException as e:
        print(f"❌ SSH 连接错误：{str(e)}")
    except Exception as e:
        print(f"❌ 未知错误：{str(e)}")
    
    return None

if __name__ == "__main__":
    # 配置参数（替换为自己的服务器和密钥路径）
    config = {
        "host": "192.168.xxx.xxx",                  # 远程服务器 IP
        "username": "root",                         # 登录用户名
        "key_path": "C:/Users/ZhuanZ/.ssh/id_rsa",  # 本地私钥路径
        "passphrase": None                           # 私钥无密码则为 None
    }
    
    # 建立密钥连接
    client = connect_with_key(**config)
    
    if client:
        try:
            # 执行远程命令（示例：读取 flag.txt 内容）
            command = "cat /root/flag.txt"
            print(f"\n📝 执行命令: {command}")
            stdin, stdout, stderr = client.exec_command(command)
            
            # 获取命令结果
            exit_status = stdout.channel.recv_exit_status()
            output = stdout.read().decode("utf-8")
            error = stderr.read().decode("utf-8")
            
            # 输出结果
            if exit_status == 0:
                print("✅ 命令执行成功，输出内容：")
                print(output.strip())
            else:
                print(f"❌ 命令执行失败，状态码：{exit_status}")
                print(f"⚠️  错误信息：{error.strip()}")
        
        finally:
            # 关闭连接
            client.close()
            print(f"\n🔌 SSH 连接已关闭")

3.4 执行效果

密钥认证执行效果

4. 总结

  1. 核心用途：Paramiko 是 Python 自动化运维的核心库，解决「远程命令执行」和「安全文件传输」两大核心需求。
  2. 认证选择：
    • 测试环境：密码认证（配置简单，适合快速验证）
    • 生产环境：密钥认证（安全性高，避免密码泄露风险）
  3. 最佳实践：
    • 连接使用 with 语句或 finally 块确保关闭，避免资源泄漏
    • 敏感信息（密码、私钥路径）通过环境变量或配置文件管理，不硬编码
    • 执行命令后需检查 exit_status，避免忽略执行失败的情况