SSH 远程自动化（paramiko）

批量重启、分发配置、远程执行巡检，离不开 SSH。Python 常用 paramiko 建立会话、执行命令、SFTP 传文件。密钥认证优于密码；生产建议配合堡垒机与 sudo 审计。

---

安装

cd ~/python-for-ops
uv add paramiko

---

连接与执行命令

#!/usr/bin/env python3
"""使用密钥连接远程主机并执行只读巡检命令。"""
from __future__ import annotations

import os
import socket
from dataclasses import dataclass

import paramiko

@dataclass
class RemoteResult:
    host: str
    exit_code: int
    stdout: str
    stderr: str

def connect(
    host: str,
    username: str,
    key_path: str | None = None,
    port: int = 22,
    timeout: float = 10.0,
) -> paramiko.SSHClient:
    client = paramiko.SSHClient()
    known_hosts = os.environ.get("OPS_KNOWN_HOSTS", os.path.expanduser("~/.ssh/known_hosts"))
    if not os.path.exists(known_hosts):
        raise FileNotFoundError(f"known_hosts 不存在: {known_hosts}")
    # 先加载受管的主机公钥，再拒绝未知主机；不能用 AutoAddPolicy 绕过校验。
    client.load_host_keys(known_hosts)
    client.set_missing_host_key_policy(paramiko.RejectPolicy())
    key_path = key_path or os.path.expanduser("~/.ssh/id_ed25519")
    client.connect(
        hostname=host,
        port=port,
        username=username,
        # key_filename 支持 Ed25519、RSA 等多种密钥类型；不把私钥内容加载进业务代码。
        key_filename=key_path,
        timeout=timeout,
        banner_timeout=timeout,
        auth_timeout=timeout,
    )
    return client

def run_remote(client: paramiko.SSHClient, host: str, command: str, timeout: int = 60) -> RemoteResult:
    stdin, stdout, stderr = client.exec_command(command, timeout=timeout)
    exit_code = stdout.channel.recv_exit_status()
    return RemoteResult(
        host=host,
        exit_code=exit_code,
        stdout=stdout.read().decode(errors="replace"),
        stderr=stderr.read().decode(errors="replace"),
    )

def inspect_host(host: str, user: str = "ops") -> RemoteResult:
    cmd = "df -h / && free -m | head -2 && uptime"
    client = connect(host, user)
    try:
        return run_remote(client, host, cmd)
    finally:
        client.close()

if __name__ == "__main__":
    target = os.environ.get("TARGET_HOST", "192.168.1.10")
    try:
        res = inspect_host(target)
        print(res.stdout)
        if res.exit_code != 0:
            print("stderr:", res.stderr)
    except (paramiko.SSHException, socket.error) as e:
        raise SystemExit(f"SSH 失败 {target}: {e}") from e

---

SFTP 上传文件

def upload_file(host: str, user: str, local_path: str, remote_path: str) -> None:
    client = connect(host, user)
    try:
        sftp = client.open_sftp()
        sftp.put(local_path, remote_path)
        sftp.close()
    finally:
        client.close()

---

批量主机（线程池）

from concurrent.futures import ThreadPoolExecutor, as_completed

HOSTS = ["10.0.0.11", "10.0.0.12", "10.0.0.13"]

def batch_uptime(hosts: list[str], user: str = "ops") -> list[RemoteResult]:
    results = []
    with ThreadPoolExecutor(max_workers=10) as pool:
        futures = {pool.submit(inspect_host, h, user): h for h in hosts}
        for fut in as_completed(futures):
            try:
                results.append(fut.result())
            except Exception as e:
                h = futures[fut]
                results.append(RemoteResult(h, -1, "", str(e)))
    return results

---

安全与实践

项	说明
Host Key	首次可用 KnownHostsFile，勿长期 AutoAddPolicy
命令	远程命令避免 shell 拼接用户输入
权限	最小账号 + sudo 限定命令
超时	exec_command(..., timeout=) 防 hang
密钥	chmod 600，不入库

首次接入主机前，应通过 CMDB、控制台或带外渠道核验指纹后再写入受管 known_hosts；不要在自动化脚本执行时自动接受新指纹。可通过 OPS_KNOWN_HOSTS 指定项目专用文件，方便审计和轮换。

提示

大规模编排可交给 Ansible；paramiko 适合定制逻辑（交互式步骤、特殊 SFTP、与 Python 数据处理无缝衔接）。