Python SSH连接案例如何远程连接服务器

wen python案例 2026-06-29 236

本文目录导读：

使用 paramiko 库（最常用）
使用 asyncssh（异步版本）
使用 fabric（高级封装）
完整示例：自动部署脚本
注意事项

我来介绍几种Python连接SSH服务器的常用方法：

使用 paramiko 库（最常用）

安装

pip install paramiko

基础连接示例

import paramiko
# 创建SSH客户端
ssh = paramiko.SSHClient()
# 自动添加主机密钥（生产环境建议手动添加）
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
    # 连接服务器
    ssh.connect(
        hostname='192.168.1.100',  # 服务器IP或域名
        port=22,                    # SSH端口
        username='root',            # 用户名
        password='your_password'    # 密码
    )
    # 执行命令
    stdin, stdout, stderr = ssh.exec_command('ls -la')
    # 获取输出
    print(stdout.read().decode('utf-8'))
    print(stderr.read().decode('utf-8'))
finally:
    # 关闭连接
    ssh.close()

使用密钥认证

import paramiko
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 使用密钥文件连接
private_key = paramiko.RSAKey.from_private_key_file('/path/to/private_key.pem')
ssh.connect(
    hostname='192.168.1.100',
    port=22,
    username='root',
    pkey=private_key
)
# 执行多个命令
commands = [
    'uname -a',
    'df -h',
    'free -m',
    'uptime'
]
for cmd in commands:
    stdin, stdout, stderr = ssh.exec_command(cmd)
    print(f"执行命令: {cmd}")
    print(stdout.read().decode('utf-8'))
    print("-" * 50)
ssh.close()

文件传输（SFTP）

import paramiko
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('192.168.1.100', 22, 'root', 'password')
# 创建SFTP客户端
sftp = ssh.open_sftp()
# 上传文件
sftp.put('local_file.txt', '/remote/path/file.txt')
# 下载文件
sftp.get('/remote/path/file.txt', 'local_file.txt')
# 列出目录
files = sftp.listdir('/remote/path')
print(files)
sftp.close()
ssh.close()

使用 asyncssh（异步版本）

安装

pip install asyncssh

异步连接示例

import asyncio
import asyncssh
async def run_commands():
    async with asyncssh.connect(
        '192.168.1.100',
        username='root',
        password='password'
    ) as conn:
        # 执行单个命令
        result = await conn.run('uname -a')
        print(result.stdout)
        # 执行多个命令
        commands = ['ls -la', 'pwd', 'whoami']
        results = await asyncio.gather(
            *[conn.run(cmd) for cmd in commands]
        )
        for cmd, result in zip(commands, results):
            print(f"{cmd}: {result.stdout}")
asyncio.run(run_commands())

使用 fabric（高级封装）

安装

pip install fabric

示例

from fabric import Connection
# 连接服务器
with Connection(
    host='192.168.1.100',
    user='root',
    connect_kwargs={
        'password': 'password'
        # 或者使用密钥
        # 'key_filename': '/path/to/key.pem'
    }
) as conn:
    # 执行命令
    result = conn.run('uname -a', hide=True)
    print(result.stdout)
    # 在远程目录执行命令
    with conn.cd('/var/log'):
        conn.run('ls -la')
    # 上传文件
    conn.put('local_file.txt', '/remote/path/')
    # 下载文件
    conn.get('/remote/path/file.txt', 'local_file.txt')

完整示例：自动部署脚本

import paramiko
import time
import os
class SSHManager:
    def __init__(self, hostname, port=22, username='root', password=None, key_file=None):
        self.hostname = hostname
        self.port = port
        self.username = username
        self.password = password
        self.key_file = key_file
        self.client = None
        self.sftp = None
    def connect(self):
        """建立SSH连接"""
        try:
            self.client = paramiko.SSHClient()
            self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            if self.key_file:
                private_key = paramiko.RSAKey.from_private_key_file(self.key_file)
                self.client.connect(
                    hostname=self.hostname,
                    port=self.port,
                    username=self.username,
                    pkey=private_key
                )
            else:
                self.client.connect(
                    hostname=self.hostname,
                    port=self.port,
                    username=self.username,
                    password=self.password
                )
            self.sftp = self.client.open_sftp()
            print(f"成功连接到 {self.hostname}")
            return True
        except Exception as e:
            print(f"连接失败: {e}")
            return False
    def execute_command(self, command, timeout=10):
        """执行远程命令"""
        try:
            stdin, stdout, stderr = self.client.exec_command(command, timeout=timeout)
            output = stdout.read().decode('utf-8')
            error = stderr.read().decode('utf-8')
            if error:
                print(f"命令执行错误: {error}")
            return output, error
        except Exception as e:
            print(f"命令执行失败: {e}")
            return None, str(e)
    def upload_file(self, local_path, remote_path):
        """上传文件"""
        try:
            self.sftp.put(local_path, remote_path)
            print(f"文件上传成功: {local_path} -> {remote_path}")
        except Exception as e:
            print(f"文件上传失败: {e}")
    def download_file(self, remote_path, local_path):
        """下载文件"""
        try:
            self.sftp.get(remote_path, local_path)
            print(f"文件下载成功: {remote_path} -> {local_path}")
        except Exception as e:
            print(f"文件下载失败: {e}")
    def close(self):
        """关闭连接"""
        if self.sftp:
            self.sftp.close()
        if self.client:
            self.client.close()
        print("连接已关闭")
# 使用示例
if __name__ == "__main__":
    # 创建SSH管理器
    ssh_manager = SSHManager(
        hostname='192.168.1.100',
        port=22,
        username='root',
        password='your_password'
    )
    # 连接服务器
    if ssh_manager.connect():
        # 执行系统信息检查
        commands = [
            'uname -a',
            'df -h /',
            'free -h',
            'uptime'
        ]
        for cmd in commands:
            print(f"\n执行命令: {cmd}")
            output, error = ssh_manager.execute_command(cmd)
            if output:
                print(output)
        # 上传文件
        if os.path.exists('deploy.sh'):
            ssh_manager.upload_file('deploy.sh', '/tmp/deploy.sh')
        # 关闭连接
        ssh_manager.close()