`paramiko` 模块：SSH 协议自动化与文件传输

好的，各位观众，欢迎来到今天的Paramiko模块专场！今天咱们不搞虚的，直接上干货，目标只有一个：让你用Paramiko玩转SSH协议，实现自动化运维和文件传输，告别手动敲命令的苦逼生活！

一、Paramiko是啥？为啥要用它？

想象一下，你手头有几十台甚至几百台服务器，每天要登录上去执行各种命令，改配置，传文件……是不是感觉要崩溃？这时候，Paramiko就是你的救星！

Paramiko是一个Python模块，它实现了SSHv2协议，允许你通过Python代码安全地连接到远程服务器，执行命令，传输文件，就像你在本地终端操作一样。

为啥要用它？

• 自动化运维： 告别手动登录，批量执行命令，定时任务，简直是运维福音！
• 文件传输： 安全地上传下载文件，比FTP靠谱多了。
• 脚本化管理： 将复杂的运维操作封装成脚本，一键搞定，提高效率。
• 安全性： 基于SSH协议，数据加密传输，保证安全。

二、Paramiko安装与环境配置

首先，确保你的Python环境已经准备好。然后，打开你的终端，输入以下命令：

pip install paramiko

搞定！是不是很简单？

三、Paramiko核心概念与基本用法

Paramiko的核心在于SSHClient类和SFTPClient类。SSHClient用于建立SSH连接和执行命令，SFTPClient用于文件传输。

1. SSHClient：连接与命令执行

import paramiko

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

# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 连接远程服务器
try:
    ssh.connect('your_server_ip', port=22, username='your_username', password='your_password')

    # 执行命令
    stdin, stdout, stderr = ssh.exec_command('df -h')  # 磁盘空间查看命令

    # 获取命令执行结果
    output = stdout.read().decode()
    error = stderr.read().decode()

    # 打印结果
    print(output)
    if error:
        print("Error:", error)

except paramiko.AuthenticationException:
    print("Authentication failed.")
except paramiko.SSHException as e:
    print(f"SSH connection failed: {e}")
except Exception as e:
    print(f"An error occurred: {e}")
finally:
    # 关闭连接
    if ssh:
        ssh.close()

代码解释：

• paramiko.SSHClient(): 创建一个SSH客户端对象。
• ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()): 这是一个非常重要的设置。默认情况下，Paramiko会检查远程主机的公钥是否在know_hosts文件中。如果不在，连接会失败。AutoAddPolicy()会自动添加新的主机公钥到know_hosts文件中，方便我们快速连接。但是，在生产环境中，最好手动验证主机公钥，以防止中间人攻击。
• ssh.connect('your_server_ip', port=22, username='your_username', password='your_password'): 连接远程服务器。你需要替换成你自己的服务器IP、端口、用户名和密码。
• ssh.exec_command('df -h'): 执行远程命令。df -h是Linux下查看磁盘空间的命令。
• stdout.read().decode(): 读取命令执行的输出结果。
• stderr.read().decode(): 读取命令执行的错误信息。如果命令执行成功，stderr通常为空。
• ssh.close(): 关闭SSH连接，释放资源。

注意：

• 强烈建议使用SSH密钥认证，而不是密码认证，更加安全。
• 错误处理非常重要，要捕获各种可能的异常，例如认证失败、连接失败等。

2. SFTPClient：文件传输

import paramiko

# 创建SSHClient对象
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

try:
    ssh.connect('your_server_ip', port=22, username='your_username', password='your_password')

    # 创建SFTPClient对象
    sftp = ssh.open_sftp()

    # 上传文件
    sftp.put('/path/to/your/local/file.txt', '/path/to/remote/file.txt')

    # 下载文件
    sftp.get('/path/to/remote/file.txt', '/path/to/your/local/downloaded_file.txt')

    print("File transfer completed.")

except paramiko.AuthenticationException:
    print("Authentication failed.")
except paramiko.SSHException as e:
    print(f"SSH connection failed: {e}")
except Exception as e:
    print(f"An error occurred: {e}")
finally:
    # 关闭SFTP连接和SSH连接
    if 'sftp' in locals():
        sftp.close()
    if ssh:
        ssh.close()

代码解释：

• ssh.open_sftp(): 创建一个SFTP客户端对象。
• sftp.put('/path/to/your/local/file.txt', '/path/to/remote/file.txt'): 上传本地文件到远程服务器。
• sftp.get('/path/to/remote/file.txt', '/path/to/your/local/downloaded_file.txt'): 从远程服务器下载文件到本地。
• sftp.close(): 关闭SFTP连接。

四、更高级的用法：密钥认证、通道转发、交互式会话

1. 密钥认证

使用密钥认证可以避免每次都输入密码，更加安全方便。首先，你需要生成SSH密钥对（公钥和私钥）。然后，将公钥添加到远程服务器的~/.ssh/authorized_keys文件中。

import paramiko

# 创建SSHClient对象
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 使用密钥认证
try:
    private_key_path = '/path/to/your/private_key'  # 你的私钥文件路径
    key = paramiko.RSAKey.from_private_key_file(private_key_path)

    ssh.connect('your_server_ip', port=22, username='your_username', pkey=key)

    # 执行命令
    stdin, stdout, stderr = ssh.exec_command('whoami')
    output = stdout.read().decode()
    print(output)

except paramiko.AuthenticationException:
    print("Authentication failed.")
except paramiko.SSHException as e:
    print(f"SSH connection failed: {e}")
except Exception as e:
    print(f"An error occurred: {e}")
finally:
    # 关闭连接
    if ssh:
        ssh.close()

代码解释：

• paramiko.RSAKey.from_private_key_file(private_key_path): 从私钥文件加载私钥。
• ssh.connect('your_server_ip', port=22, username='your_username', pkey=key): 使用私钥进行认证。

2. 通道转发

通道转发可以将本地端口转发到远程服务器，或者将远程服务器的端口转发到本地。这在访问内网服务时非常有用。

import paramiko
import socket
import threading

# 转发端口
LOCAL_PORT = 8000
REMOTE_HOST = 'your_server_ip'
REMOTE_PORT = 3306  # 例如，MySQL端口

# 远程服务器信息
SSH_HOST = 'your_server_ip'
SSH_PORT = 22
SSH_USER = 'your_username'
SSH_PASSWORD = 'your_password'

def handler(chan, host, port):
    sock = socket.socket()
    try:
        sock.connect((host, port))
    except Exception as e:
        print(f"Forwarding request to {host}:{port} failed: {e}")
        return

    print(f"Connected! Tunnel open {chan} -> {host}:{port} -> {sock}")

    while True:
        try:
            r, w, x = select.select([sock, chan], [], [])
            if sock in r:
                data = sock.recv(1024)
                if len(data) == 0:
                    break
                chan.send(data)
            if chan in r:
                data = chan.recv(1024)
                if len(data) == 0:
                    break
                sock.send(data)
        except Exception as e:
            print(f"Channel Exception: {e}")
            break

    sock.close()
    chan.close()
    print("Tunnel closed.")

def reverse_forward_tunnel(local_port, remote_host, remote_port, transport):
    transport.request_port_forward('', local_port)
    while True:
        chan = transport.accept(1000)
        if chan is None:
            continue
        thr = threading.Thread(target=handler, args=(chan, remote_host, remote_port))
        thr.daemon = True
        thr.start()

try:
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    client.connect(SSH_HOST, SSH_PORT, username=SSH_USER, password=SSH_PASSWORD)

    transport = client.get_transport()
    transport.set_keepalive(30)

    reverse_forward_tunnel(LOCAL_PORT, REMOTE_HOST, REMOTE_PORT, transport)

except Exception as e:
    print(f"An error occurred: {e}")
    traceback.print_exc()
finally:
    if client:
        client.close()

3. 交互式会话

有时候，你需要与远程服务器进行交互，例如运行top命令或者编辑文件。Paramiko也支持交互式会话。

import paramiko
import select
import sys
import tty
import termios

# 创建SSHClient对象
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

try:
    ssh.connect('your_server_ip', port=22, username='your_username', password='your_password')

    # 获取终端大小
    rows, cols = os.popen('stty size', 'r').read().split()
    rows = int(rows)
    cols = int(cols)

    # 打开一个channel
    channel = ssh.invoke_shell(width=cols, height=rows)

    # 设置终端模式
    old_tty = termios.tcgetattr(sys.stdin)
    tty.setraw(sys.stdin.fileno())
    channel.settimeout(0)

    try:
        while True:
            r, w, e = select.select([channel, sys.stdin], [], [])
            if channel in r:
                try:
                    x = channel.recv(1024).decode()
                    if len(x) == 0:
                        print('rn*** EOFrn')
                        break
                    sys.stdout.write(x)
                    sys.stdout.flush()
                except Exception as e:
                    print(f"Channel Exception: {e}")
                    break
            if sys.stdin in r:
                x = sys.stdin.read(1)
                if len(x) == 0:
                    break
                channel.send(x)

    finally:
        # 恢复终端模式
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)

except paramiko.AuthenticationException:
    print("Authentication failed.")
except paramiko.SSHException as e:
    print(f"SSH connection failed: {e}")
except Exception as e:
    print(f"An error occurred: {e}")
finally:
    # 关闭连接
    if ssh:
        ssh.close()

五、Paramiko实战案例

案例1：批量执行命令

import paramiko

# 服务器列表
servers = [
    {'host': 'server1_ip', 'username': 'user1', 'password': 'password1'},
    {'host': 'server2_ip', 'username': 'user2', 'password': 'password2'},
    {'host': 'server3_ip', 'username': 'user3', 'password': 'password3'},
]

# 命令
command = 'uptime'

# 遍历服务器列表，执行命令
for server in servers:
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(server['host'], username=server['username'], password=server['password'])

        stdin, stdout, stderr = ssh.exec_command(command)
        output = stdout.read().decode()
        error = stderr.read().decode()

        print(f"Server: {server['host']}")
        print(output)
        if error:
            print("Error:", error)
        print("-" * 20)

    except paramiko.AuthenticationException:
        print(f"Authentication failed for server: {server['host']}")
    except paramiko.SSHException as e:
        print(f"SSH connection failed for server: {server['host']}: {e}")
    except Exception as e:
        print(f"An error occurred for server: {server['host']}: {e}")
    finally:
        if ssh:
            ssh.close()

案例2：定时备份文件

import paramiko
import datetime
import os
import time

# 服务器信息
HOST = 'your_server_ip'
USERNAME = 'your_username'
PASSWORD = 'your_password'

# 备份目录
REMOTE_DIR = '/path/to/remote/directory'
LOCAL_BACKUP_DIR = '/path/to/local/backup/directory'

# 备份频率（秒）
BACKUP_INTERVAL = 86400  # 每天备份一次

def backup_files():
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(HOST, username=USERNAME, password=PASSWORD)

        sftp = ssh.open_sftp()

        # 创建本地备份目录
        if not os.path.exists(LOCAL_BACKUP_DIR):
            os.makedirs(LOCAL_BACKUP_DIR)

        # 生成备份文件名
        timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
        backup_filename = f"backup_{timestamp}.tar.gz"
        local_backup_path = os.path.join(LOCAL_BACKUP_DIR, backup_filename)

        # 远程打包备份
        tar_command = f"tar -czvf /tmp/{backup_filename} {REMOTE_DIR}"
        stdin, stdout, stderr = ssh.exec_command(tar_command)
        exit_status = stdout.channel.recv_exit_status()
        if exit_status != 0:
            error = stderr.read().decode()
            print(f"Error creating tar archive: {error}")
            return

        # 下载备份文件
        sftp.get(f'/tmp/{backup_filename}', local_backup_path)

        # 删除远程备份文件
        remove_command = f"rm /tmp/{backup_filename}"
        stdin, stdout, stderr = ssh.exec_command(remove_command)
        exit_status = stdout.channel.recv_exit_status()
        if exit_status != 0:
            error = stderr.read().decode()
            print(f"Error removing temporary file: {error}")

        print(f"Backup completed: {local_backup_path}")

    except paramiko.AuthenticationException:
        print("Authentication failed.")
    except paramiko.SSHException as e:
        print(f"SSH connection failed: {e}")
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        if 'sftp' in locals():
            sftp.close()
        if ssh:
            ssh.close()

# 定时执行备份
while True:
    backup_files()
    time.sleep(BACKUP_INTERVAL)

六、常见问题与解决方案

问题	解决方案
AuthenticationException	检查用户名、密码是否正确。如果使用密钥认证，检查私钥路径是否正确，公钥是否已添加到authorized_keys文件中。
SSHException: Server refused to allocate pty	这是因为远程服务器拒绝分配伪终端。可以尝试在exec_command中添加get_pty=True参数，或者尝试使用交互式会话。
连接超时	检查网络连接是否正常。可以尝试增加connect方法的timeout参数，例如ssh.connect('your_server_ip', port=22, username='your_username', password='your_password', timeout=10)。
文件传输速度慢	检查网络带宽是否足够。可以尝试使用压缩算法，例如gzip，或者使用更高效的文件传输协议，例如scp。
乱码问题	确保本地和远程服务器的编码方式一致。可以在stdout.read().decode()中指定编码方式，例如stdout.read().decode('utf-8')。
socket.timeout: timed out	这是因为连接超时。检查网络连接是否正常。可以尝试增加connect方法的timeout参数，例如ssh.connect('your_server_ip', port=22, username='your_username', password='your_password', timeout=10)。 如果是在执行命令时出现，可能是命令执行时间过长，可以尝试设置channel.settimeout()来增加超时时间。
Host key verification failed.	这种情况是因为你连接的服务器的公钥不在你的known_hosts文件中。 解决方法有两种： 1. 手动将服务器的公钥添加到你的known_hosts文件中。 2. 使用ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())来自动添加。 但是，在生产环境中，建议手动验证公钥，以防止中间人攻击。
Channel closed	这种情况通常是因为远程服务器关闭了连接。 可能的原因包括： 1. 远程服务器资源不足。 2. 远程服务器配置了连接数限制。 3. 你的程序中存在bug，导致连接异常关闭。 你需要检查远程服务器的日志，以及你的程序代码，来找到问题的原因。
权限问题	检查你的用户名在远程服务器上是否有足够的权限执行相应的操作。 例如，如果你尝试上传文件到没有写入权限的目录，或者尝试执行需要root权限的命令，都会遇到权限问题。 你可以尝试使用sudo来提升权限，或者联系服务器管理员来修改你的用户权限。
如何处理需要交互的命令？	对于需要交互的命令，例如passwd，你需要使用invoke_shell()来创建一个交互式会话，然后通过channel.send()和channel.recv()来发送和接收数据。 具体可以参考上面“交互式会话”的例子。 注意，你需要模拟用户的输入，并处理服务器的输出，才能完成交互。
如何处理中文乱码？	确保你的Python脚本、本地终端和远程服务器都使用相同的编码方式，例如UTF-8。 你可以在stdout.read().decode()中指定编码方式，例如stdout.read().decode('utf-8')。 同时，你也需要设置远程服务器的环境变量，例如export LANG=en_US.UTF-8。

七、总结与展望

Paramiko是一个非常强大的SSH自动化工具，掌握它可以大大提高你的运维效率。希望今天的讲座能让你对Paramiko有一个更深入的了解。

当然，Paramiko的功能远不止这些，还有很多高级用法等你来探索。例如，你可以使用Paramiko来实现堡垒机功能，或者与其他工具集成，构建更复杂的自动化运维系统。

最后，祝大家使用Paramiko愉快，早日成为自动化运维大神！记住，遇到问题不要怕，多查文档，多尝试，你一定可以搞定的！ 下次再见！