Paramiko

1068 views

Python 的 Paramiko 是一个基于 SSH 协议的第三方库，用于实现 SSH 客户端功能，支持通过 SSH 连接远程服务器、执行命令、传输文件（SFTP）等操作。它广泛应用于自动化运维、远程服务器管理、批量任务执行等场景，无需手动登录服务器即可通过代码完成远程操作。

一、Paramiko 安装

Paramiko 是第三方库，需通过 pip 安装： 

pip install paramiko

二、核心功能：SSH 远程操作

Paramiko 的核心是通过 SSH 协议与远程服务器建立连接，主要功能包括：
- 执行远程命令（单条或批量）。
- 通过 SFTP（SSH File Transfer Protocol）传输文件（上传/下载）。
- 支持密码认证和密钥认证（更安全，推荐）。

1. 建立 SSH 连接并执行命令

通过 SSHClient 类创建 SSH 客户端，连接远程服务器后执行命令，获取输出结果。

（1）密码认证示例
import paramiko

# 1. 创建 SSH 客户端对象
ssh = paramiko.SSHClient()

# 2. 设置自动添加远程服务器的主机密钥（首次连接时避免因 unknown host 报错）
# 生产环境中建议手动管理 known_hosts 文件，而非自动添加
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

try:
    # 3. 连接远程服务器（密码认证）
    ssh.connect(
        hostname="192.168.1.100",  # 远程服务器 IP 或主机名
        port=22,                   # SSH 端口（默认 22）
        username="root",           # 登录用户名
        password="123456"          # 登录密码
    )
    print("SSH 连接成功")

    # 4. 执行远程命令（如查看当前目录文件）
    command = "ls -l"  # Linux 命令；Windows 服务器可改为 "dir"
    stdin, stdout, stderr = ssh.exec_command(command)

    # 5. 读取命令输出（stdout 为标准输出，stderr 为错误输出）
    # 注意：输出是 bytes 类型，需解码为字符串
    output = stdout.read().decode("utf-8")
    error = stderr.read().decode("utf-8")

    if output:
        print("命令输出：")
        print(output)
    if error:
        print("命令错误：")
        print(error)

except paramiko.AuthenticationException:
    print("认证失败：用户名或密码错误")
except paramiko.SSHException as e:
    print(f"SSH 连接失败：{e}")
finally:
    # 6. 关闭连接
    if ssh.get_transport() and ssh.get_transport().is_active():
        ssh.close()
        print("SSH 连接已关闭")
（2）密钥认证示例（更安全）

多数服务器禁用密码登录，仅允许 SSH 密钥认证（通过公钥/私钥对验证）。需提前在本地生成密钥对（如 id_rsa 私钥和 id_rsa.pub 公钥），并将公钥上传至服务器的 ~/.ssh/authorized_keys 中。 

import paramiko

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

try:
    # 加载本地私钥文件（如 ~/.ssh/id_rsa）
    private_key = paramiko.RSAKey.from_private_key_file("/home/user/.ssh/id_rsa")

    # 连接服务器（密钥认证，无需密码）
    ssh.connect(
        hostname="192.168.1.100",
        port=22,
        username="root",
        pkey=private_key  # 指定私钥
    )
    print("密钥认证成功")

    # 执行命令（示例：查看服务器内存使用情况）
    stdin, stdout, stderr = ssh.exec_command("free -m")
    print("内存使用情况：")
    print(stdout.read().decode("utf-8"))

except paramiko.AuthenticationException:
    print("密钥认证失败：私钥或公钥配置错误")
finally:
    if ssh.get_transport().is_active():
        ssh.close()

2. 执行长命令或交互式命令

exec_command 适合执行非交互式短命令（如 ls、pwd）。若需执行长命令（如持续输出日志）或交互式命令（如 sudo 输入密码），需通过 invoke_shell 创建交互式 shell。

（1）执行长命令（实时获取输出）
import paramiko
import time

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("192.168.1.100", username="root", password="123456")

# 创建交互式 shell
shell = ssh.invoke_shell()

# 执行长命令（如 ping 测试，持续输出）
command = "ping -c 5 www.baidu.com\n"  # Linux ping 5 次；Windows 用 "ping -n 5 www.baidu.com"
shell.send(command)

# 等待命令执行，实时读取输出
time.sleep(2)  # 适当等待（根据命令耗时调整）
while shell.recv_ready():
    output = shell.recv(4096).decode("utf-8")  # 每次读取 4096 字节
    print(output)

shell.close()
ssh.close()
（2）执行交互式命令（如 sudo）
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("192.168.1.100", username="user", password="user_pass")

shell = ssh.invoke_shell()

# 执行需要 sudo 的命令（如重启服务）
shell.send("sudo systemctl restart nginx\n")
time.sleep(1)  # 等待 sudo 提示输入密码

# 发送 sudo 密码（注意末尾加换行符）
shell.send("sudo_pass\n")  # 替换为实际 sudo 密码
time.sleep(2)

# 读取输出
output = shell.recv(4096).decode("utf-8")
print(output)

shell.close()
ssh.close()

3. SFTP 文件传输（上传/下载）

Paramiko 通过 SFTPClient 类支持 SFTP 协议，实现本地与远程服务器的文件传输。

（1）上传本地文件到远程服务器
import paramiko

# 建立 SSH 连接（同上，密码或密钥认证）
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("192.168.1.100", username="root", password="123456")

# 创建 SFTP 客户端
sftp = ssh.open_sftp()

try:
    # 本地文件路径（如当前目录的 test.txt）
    local_path = "test.txt"
    # 远程服务器路径（如 /tmp/test.txt）
    remote_path = "/tmp/test.txt"

    # 上传文件
    sftp.put(local_path, remote_path)
    print(f"文件 {local_path} 上传至 {remote_path} 成功")
except FileNotFoundError:
    print(f"本地文件 {local_path} 不存在")
except Exception as e:
    print(f"上传失败：{e}")
finally:
    # 关闭 SFTP 和 SSH 连接
    sftp.close()
    ssh.close()
（2）从远程服务器下载文件到本地
# 建立连接（同上）
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("192.168.1.100", username="root", password="123456")
sftp = ssh.open_sftp()

try:
    # 远程文件路径
    remote_path = "/var/log/syslog"
    # 本地保存路径
    local_path = "syslog_local.txt"

    # 下载文件
    sftp.get(remote_path, local_path)
    print(f"文件 {remote_path} 下载至 {local_path} 成功")
except FileNotFoundError:
    print(f"远程文件 {remote_path} 不存在")
finally:
    sftp.close()
    ssh.close()

三、关键注意事项

1.主机密钥验证：

  • 首次连接陌生服务器时，SSHClient 会因 known_hosts 中无记录而报错。AutoAddPolicy 可自动添加，但生产环境建议手动管理 known_hosts（通过 load_system_host_keys 加载），避免中间人攻击。 
# 加载系统默认的 known_hosts 文件（推荐）
ssh.load_system_host_keys()
# 若有自定义 known_hosts，可额外加载
ssh.load_host_keys("/path/to/custom_known_hosts")

2.连接安全性：

  • 避免硬编码密码/私钥路径，建议通过配置文件或环境变量读取。
  • 私钥文件权限需严格限制（如 Linux 上设置为 600，否则 Paramiko 可能拒绝使用）。

3.超时设置： 连接或执行命令时可设置超时，避免无限阻塞：

# 连接超时（5秒）
ssh.connect(..., timeout=5)

# 命令执行超时（10秒）
stdin, stdout, stderr = ssh.exec_command(command, timeout=10)

4.异常处理：

常见异常包括：

  • AuthenticationException：认证失败（密码/密钥错误）。
  • SSHException：SSH 协议错误（如连接被拒绝、协议不匹配）。
  • socket.timeout：连接超时。
  • FileNotFoundError：文件传输时路径不存在。

四、应用场景

Paramiko 适合以下场景：

  • 自动化运维：批量在多台服务器执行命令（如部署脚本、更新配置）。
  • 远程监控：定期从服务器拉取日志、性能数据（如 CPU/内存使用率）。
  • 文件同步：通过 SFTP 自动上传/下载备份文件、日志等。
  • CI/CD 流程：在持续集成/部署中远程执行构建、测试命令。

总结

Paramiko 是 Python 操作 SSH 协议的核心库，通过 SSHClient 实现远程命令执行，通过 SFTPClient 实现文件传输，支持密码和密钥两种认证方式。其优势在于：
- 功能全面：覆盖 SSH 客户端的核心需求（命令执行、文件传输）。
- 灵活性高：支持交互式命令和长命令执行，适配复杂场景。
- 跨平台：可连接 Linux、macOS、Windows（需开启 SSH 服务）等服务器。

掌握 Paramiko 能显著提升远程服务器管理的自动化效率，是运维脚本和自动化工具开发的必备技能。

什么是 Paramiko?

Paramiko 是一个 Python 实现的 SSH 协议库，提供了 SSH 客户端和 SSH 服务器的 API。它允许你通过 SSH 协议远程控制服务器，进行数据传输或在 Shell 中执行命令等操作。

如何安装 Paramiko?

Paramiko 可以使用 pip 安装，命令如下：

pip install paramiko

如何使用 Paramiko 连接 SSH 服务器? 使用 Paramiko 连接 SSH 服务器可以通过如下代码实现：

import paramiko
# SSH credentials
ssh_host = 'your_ssh_host'
ssh_user = 'your_ssh_username'
ssh_password = 'your_ssh_password'
# Establish SSH connection
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(ssh_host, username=ssh_user, password=ssh_password)
# Execute command
command = 'your_ssh_command'
stdin, stdout, stderr = ssh.exec_command(command)
# Print output
print(stdout.read().decode())
# Close SSH connection
ssh.close()

如何使用 Paramiko 上传和下载文件? 可以使用 Paramiko 的 SFTP API 上传和下载文件：

import paramiko
# SFTP credentials
sftp_host = 'your_sftp_host'
sftp_user = 'your_sftp_username'
sftp_password = 'your_sftp_password'
# Establish SFTP connection
transport = paramiko.Transport((sftp_host, 22))
transport.connect(username=sftp_user, password=sftp_password)
sftp = transport.open_sftp()
# Download a remote file
remote_file_path = '/path/to/remote/file'
local_file_path = '/path/to/local/file'
sftp.get(remote_file_path, local_file_path)
# Upload a local file
local_file_path = '/path/to/local/file'
remote_file_path = '/path/to/remote/file'
sftp.put(local_file_path, remote_file_path)
# Close SFTP connection
sftp.close()
transport.close()

如何使用 Paramiko 执行 sudo 命令? 可以使用 Paramiko 的 invoke_shell() 方法来模拟一个终端会话，然后执行 sudo 命令：

import paramiko
# SSH credentials
ssh_host = 'your_ssh_host'
ssh_user = 'your_ssh_username'
ssh_password = 'your_ssh_password'
# Establish SSH connection
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(ssh_host, username=ssh_user, password=ssh_password)
# Start a shell session
shell = ssh.invoke_shell()
shell.send('sudo your_command\n')
# Wait for the password prompt
while not shell.recv_ready():
    pass
shell.send('your_password\n')
# Execute command
output = shell.recv(1024)
# Print output
print(output.decode())
 # Close SSH connection
ssh.close()

** 增加异常处理 **

import paramiko
import json


# SSH连接信息
hostname = "1.1.1.1"
username = "root"
password = "123456"
# 命令
command = "kubectl get node -o json"
# 创建SSH客户端
client = paramiko.SSHClient()
# 自动添加主机密钥
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())


# 连接SSH服务器
try:
    client.connect(hostname, username=username, password=password)
except paramiko.AuthenticationException:
    print("Authentication failed, please verify your password.")
except paramiko.SSHException as sshException:
    print("Unable to establish SSH connection: %s" % sshException)
except paramiko.SSHException as e:
    print(e)


# 执行命令
try:
    stdin, stdout, stderr = client.exec_command(command, timeout=10)
    output = stdout.read().decode()
    print(output)
    result = json.loads(output)
    print(result["kind"])
except paramiko.SSHException as sshException:
    print("Unable to execute command: %s" % sshException)

# 关闭连接
client.close()