Python Paramiko库自动化远程管理服务器

用户11081884

发布于 2026-07-20 18:43:11

250

Paramiko是一个强大的Python库，实现了SSHv2协议，用于安全地连接到远程服务器并执行命令或传输文件。在现代IT运维和自动化开发中，远程服务器管理是一项常见需求，Paramiko为Python开发者提供了简单而强大的工具来实现这一目标。

应用场景

1、批量服务器维护：在多台服务器上执行相同的维护命令。

2、自动化部署：将应用程序或配置文件推送到多台服务器。

3、日志收集：从多台服务器收集和分析日志文件。

4、监控系统：定期检查服务器状态和性能指标。

5、配置管理：统一修改多台服务器的配置文件。

安装与准备

pip install -U paramiko typer rich

验证：

import paramiko, typer, concurrent.futures, re, time, os, sys
from pathlib import Path
from typing import List, Optional
print(paramiko.__version__)   # ≥ 3.4

基本连接与命令执行

import paramiko

def execute_remote_command(host, port, username, password, command):
    # 创建SSH客户端实例
    ssh = paramiko.SSHClient()
    
    # 自动添加主机密钥（生产环境应考虑使用known_hosts）
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        # 建立SSH连接
        ssh.connect(host, port=port, username=username, password=password)
        
        # 执行远程命令
        stdin, stdout, stderr = ssh.exec_command(command)
        
        # 读取输出
        output = stdout.read().decode('utf-8')
        errors = stderr.read().decode('utf-8')
        
        return output, errors
        
    except paramiko.AuthenticationException:
        print("认证失败，请检查用户名和密码")
    except paramiko.SSHException as e:
        print(f"SSH连接错误: {str(e)}")
    except Exception as e:
        print(f"发生错误: {str(e)}")
    finally:
        # 确保连接关闭
        ssh.close()

# 使用示例
output, errors = execute_remote_command(
    host="192.168.1.100",
    port=22,
    username="admin",
    password="securepassword",
    command="df -h"
)

print("命令输出:")
print(output)
if errors:
    print("错误信息:")
    print(errors)

功能示例

使用密钥认证替代密码

def connect_with_key(host, port, username, key_path):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        # 加载私钥
        private_key = paramiko.RSAKey.from_private_key_file(key_path)
        
        # 使用密钥连接
        ssh.connect(host, port=port, username=username, pkey=private_key)
        
        # 执行命令示例
        stdin, stdout, stderr = ssh.exec_command('uname -a')
        print(stdout.read().decode())
        
    except Exception as e:
        print(f"连接失败: {str(e)}")
    finally:
        ssh.close()

# 使用示例
connect_with_key(
    host="example.com",
    port=22,
    username="ubuntu",
    key_path="/path/to/private_key.pem"
)

文件传输示例

def transfer_file(host, port, username, password, local_path, remote_path):
    transport = paramiko.Transport((host, port))
    
    try:
        transport.connect(username=username, password=password)
        sftp = paramiko.SFTPClient.from_transport(transport)
        
        # 上传文件
        sftp.put(local_path, remote_path)
        print(f"文件 {local_path} 已上传至 {remote_path}")
        
        # 下载文件示例
        # sftp.get(remote_path, local_path)
        
    except Exception as e:
        print(f"文件传输失败: {str(e)}")
    finally:
        transport.close()

# 使用示例
transfer_file(
    host="192.168.1.100",
    port=22,
    username="user",
    password="password",
    local_path="local_file.txt",
    remote_path="/remote/path/file.txt"
)

异常处理

1、使用上下文管理器：确保资源正确释放。

2、日志记录：记录所有操作和错误。

3、超时设置：避免长时间等待。

4、安全实践：避免硬编码密码，使用环境变量或配置文件。

import logging
from contextlib import contextmanager

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@contextmanager
def ssh_connection(host, port, username, password=None, pkey_path=None):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        if pkey_path:
            private_key = paramiko.RSAKey.from_private_key_file(pkey_path)
            ssh.connect(host, port=port, username=username, pkey=private_key, timeout=10)
        else:
            ssh.connect(host, port=port, username=username, password=password, timeout=10)
            
        logger.info(f"成功连接到 {host}")
        yield ssh
        
    except Exception as e:
        logger.error(f"连接 {host} 失败: {str(e)}")
        raise
    finally:
        ssh.close()
        logger.info(f"已关闭与 {host} 的连接")

# 使用示例
with ssh_connection(
    host="example.com",
    port=22,
    username="admin",
    pkey_path="/path/to/key.pem"
) as conn:
    stdin, stdout, stderr = conn.exec_command("hostname")
    print(stdout.read().decode())

【场景 1】单命令执行（密码 & 密钥）

# 01_run_command.py
import paramiko, os, logging

logging.basicConfig(level=logging.INFO)
LOG = logging.getLogger("paramiko")

def run_cmd(host: str, port: int = 22, user: str = "root",
            password: str|None = None, pkey: str|None = None,
            cmd: str = "hostname", timeout: int = 10) ->tuple[str, str]:
    """返回 (stdout, stderr)"""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        if pkey and Path(pkey).exists():
            key = paramiko.RSAKey.from_private_key_file(pkey)
            client.connect(host, port=port, username=user, pkey=key, timeout=timeout)
        else:
            client.connect(host, port=port, username=user, password=password, timeout=timeout)
        LOG.info(f"{host} 已连接")
        stdin, stdout, stderr = client.exec_command(cmd, timeout=timeout)
        out = stdout.read().decode()
        err = stderr.read().decode()
        return out, err
    finally:
        client.close()

if__name__ == "__main__":
    out, err = run_cmd("192.168.1.100", password="your_pass", cmd="df -h")
    print(out)

【场景 2】批量文件分发 / 日志采集

# 02_sftp_batch.py
from paramiko import Transport
from pathlib import Path

def sftp_put(host: str, port: int, user: str, password: str,
             local: Path, remote: str):
    with Transport((host, port)) as trans:
        trans.connect(username=user, password=password)
        sftp = paramiko.SFTPClient.from_transport(trans)
        sftp.put(local, remote)
        print(f"📤 {local} → {host}:{remote}")

def sftp_get(host: str, port: int, user: str, password: str,
             remote: str, local: Path):
    with Transport((host, port)) as trans:
        trans.connect(username=user, password=password)
        sftp = paramiko.SFTPClient.from_transport(trans)
        sftp.get(remote, local)
        print(f"{host}:{remote} → {local}")

# 示例：采集 100 台日志
hosts = [f"192.168.1.{i}"for i in range(1, 101)]
for h in hosts:
    sftp_get(h, 22, "root", "pass", "/var/log/nginx/access.log",
             Path(f"logs/{h}.log"))

【场景 3】并发执行 200 台主机

# 03_concurrent_runner.py
import concurrent.futures as fut
from rich.progress import track

CMD = "systemctl status nginx"

def task(host: str):
    out, err = run_cmd(host, pkey="/root/.ssh/id_rsa", cmd=CMD)
    return host, err == "", err

with fut.ThreadPoolExecutor(max_workers=50) as pool:
    results = list(track(pool.map(task, hosts), total=len(hosts)))

failed = [r for r in results if not r[1]]
print(f"失败数: {len(failed)}")

【场景 4】跳板机（ProxyJump）一键登录

# 04_jump_host.py
import paramiko

JUMP_IP = "jump.example.com"
JUMP_USER = "jumpuser"
REAL_IP = "10.0.0.5"
REAL_USER = "root"

def make_jump_channel() ->paramiko.SSHClient:
    jump = paramiko.SSHClient()
    jump.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    jump.connect(JUMP_IP, username=JUMP_USER, key_filename="/root/.ssh/jump.pem")
    transport = jump.get_transport()
    dest_addr = (REAL_IP, 22)
    local_addr = ('127.0.0.1', 22)
    channel = transport.open_channel("direct-tcpip", dest_addr, local_addr)
    return channel

def run_via_jump():
    channel = make_jump_channel()
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(REAL_IP, username=REAL_USER, key_filename="/root/.ssh/id_rsa", sock=channel)
    stdin, stdout, stderr = client.exec_command("hostname")
    print(stdout.read().decode())
    client.close()

if__name__ == "__main__":
    run_via_jump()

【场景 5】配置漂移检查（幂等比对）

# 05_drift_check.py
import hashlib, tempfile

def remote_md5(host: str, remote_path: str) ->str:
    out, _ = run_cmd(host, pkey="/root/.ssh/id_rsa", cmd=f"md5sum {remote_path}")
    returnout.split()[0]

def check_drift(hosts: List[str], remote: str, local: Path):
    local_hash = hashlib.md5(local.read_bytes()).hexdigest()
    for h in hosts:
        if remote_md5(h, remote) != local_hash:
            print(f"❌ {h} 配置已漂移")
            sftp_put(h, 22, "root", None, local, remote)
        else:
            print(f"{h} 一致")

【场景 6】交互式 Shell（长命令）

# 06_interactive.py
def interactive_shell(host: str):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(host, username="root", pkey="/root/.ssh/id_rsa")
    channel = client.invoke_shell()
    while True:
        cmd = input("$ ")
        if cmd.lower() == "exit":
            break
        channel.send(cmd+"\n")
        time.sleep(0.5)
        print(channel.recv(65535).decode())
    client.close()

【场景 7】Typer 命令行工具

# 07_cli.py
import typer
from rich.console import Console
from rich.table import Table

app = typer.Typer(help="Paramiko 批量运维 CLI")
console = Console()

@app.command()
def run(hosts: List[str], cmd: str):
    """在多台主机执行命令"""
    table = Table("Host", "OK", "Output")
    for h in track(hosts, description="Running..."):
        out, err = run_cmd(h, pkey="/root/.ssh/id_rsa", cmd=cmd)
        table.add_row(h, str(err == ""), out[:60])
    console.print(table)

@app.command()
def put(local: Path, remote: str, hosts: List[str]):
    """批量上传文件"""
    for h in hosts:
        sftp_put(h, 22, "root", None, local, remote)

if__name__ == "__main__":
    app()

使用：

python 07_cli.py run 192.168.1.{1..10} "uptime"
python 07_cli.py put ./app.jar /opt/app.jar 192.168.1.{1..10}

Paramiko为Python开发者提供了强大的远程服务器管理能力，通过SSH协议实现安全的命令执行和文件传输，帮助开发者构建可靠的自动化运维工具。在实际应用中，应根据具体需求和安全要求进行适当调整，确保系统的安全性和稳定性。通过合理利用Paramiko，可以显著提高服务器管理效率，减少重复性工作。

“无他，惟手熟尔”！有需要就用起来。

如果你觉得这篇文章有用，欢迎关注、点赞、转发、收藏、留言、推荐❤！

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-10-09，如有侵权请联系 cloudcommunity@tencent.com 删除

服务器