Python 远程连接服务器，用它就够了

想用一把趁手的“瑞士军刀”去连服务器、跑命令、传文件、开隧道？不绕弯，用 paramiko 就能把大多数日常需求打包搞定。下面这篇就是从零到一，给你一套能直接落地的方案：装库、跑通、避坑、封装一个小工具，最后再给出扩展方向。

选谁？为什么是 paramiko

paramiko 是纯 Python 的 SSHv2 实现，能做三件关键事：命令执行、SFTP 传输、端口转发/隧道。跨平台、稳定，依赖少；和系统 ssh 命令比，它在代码里可控程度更高（密钥、超时、跳板、重试都能自己掌握）。

安装就一行：

pip install paramiko

先跑通：最小可用示例

下面三段小代码，覆盖最常用的三件事：执行命令、上传、下载。

import paramiko
host = "10.0.0.8"
user = "ubuntu"
pkey_path = "/home/me/.ssh/id_rsa"# 也可以用 password="..." 明文密码（不建议）
key = paramiko.RSAKey.from_private_key_file(pkey_path)
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname=host, username=user, pkey=key, timeout=10)
# 1) 执行命令
stdin, stdout, stderr = ssh.exec_command("uname -a && whoami", timeout=20)
print(stdout.read().decode())
print(stderr.read().decode())
# 2) 上传文件
sftp = ssh.open_sftp()
sftp.put("local.txt", "/tmp/remote.txt")
# 3) 下载文件
sftp.get("/var/log/syslog", "syslog.backup")
sftp.close()
ssh.close()

批量执行 & 并发

如果要在多台机器上跑同一条命令，用线程池就够了（IO 密集，线程更简单）。示例里带了超时、返回码、错误打印。

from concurrent.futures import ThreadPoolExecutor, as_completed
import paramiko
defrun_cmd(host, user, key_path, cmd, timeout=20):
    key = paramiko.RSAKey.from_private_key_file(key_path)
    cli = paramiko.SSHClient(); cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
        cli.connect(hostname=host, username=user, pkey=key, timeout=10)
        _, stdout, stderr = cli.exec_command(cmd, timeout=timeout)
        code = stdout.channel.recv_exit_status()
return host, code, stdout.read().decode(), stderr.read().decode()
except Exception as e:
return host, -1, "", str(e)
finally:
try: cli.close()
except: pass
hosts = ["10.0.0.8","10.0.0.9","10.0.0.10"]
with ThreadPoolExecutor(max_workers=8) as ex:
    futs = [ex.submit(run_cmd, h, "ubuntu", "/home/me/.ssh/id_rsa", "uptime") for h in hosts]
for f in as_completed(futs):
        h, code, out, err = f.result()
        print(h, code, out.strip() or err.strip())

跳板机（堡垒机）/ ProxyJump

有时候只能先连跳板，再从跳板到内网机器。思路是：先连跳板，拿到一个 Channel，把它当成“套接字”给二跳 SSH 用（类似 ssh -J）。

import paramiko, socket
defconnect_via_bastion(bastion, target, user, key_path):
    key = paramiko.RSAKey.from_private_key_file(key_path)
# 1) 先连跳板
    bastion_cli = paramiko.SSHClient()
    bastion_cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    bastion_cli.connect(bastion, username=user, pkey=key, timeout=10)
# 2) 通过跳板打开到目标主机的“直连通道”
    transport = bastion_cli.get_transport()
    dest_addr = (target, 22)
    src_addr = ("127.0.0.1", 0)
    channel = transport.open_channel("direct-tcpip", dest_addr, src_addr)
# 3) 用这个 channel 建立二跳 SSH
    target_cli = paramiko.SSHClient()
    target_cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    sock = channel  # 把 channel 当 socket
    target_cli.connect(target, username=user, pkey=key, sock=sock, timeout=10)
return bastion_cli, target_cli
bastion_cli, target_cli = connect_via_bastion("1.2.3.4", "10.0.0.8", "ubuntu", "/home/me/.ssh/id_rsa")
_, out, err = target_cli.exec_command("hostname && ip a")
print(out.read().decode() or err.read().decode())
target_cli.close(); bastion_cli.close()

本地端口转发（例如远程看数据库）

想把远端 10.0.0.8:5432 映射到本地 localhost:15432，这样本地客户端就能直接连接。paramiko 自带工具函数能写一个简单的 forwarder：

import select, socket, threading, paramiko
defforward_tunnel(local_port, remote_host, remote_port, transport):
# 接收本地连接 -> 经由 SSH transport 转发到远端
    server = socket.socket(); server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("127.0.0.1", local_port)); server.listen(100)
defhandler(client_sock):
try:
            chan = transport.open_channel("direct-tcpip", (remote_host, remote_port), client_sock.getsockname())
except Exception:
            client_sock.close(); return
whileTrue:
            r, _, _ = select.select([client_sock, chan], [], [])
if client_sock in r:
                data = client_sock.recv(1024)
ifnot data: break
                chan.send(data)
if chan in r:
                data = chan.recv(1024)
ifnot data: break
                client_sock.send(data)
        chan.close(); client_sock.close()
defloop():
whileTrue:
            client, _ = server.accept()
            threading.Thread(target=handler, args=(client,), daemon=True).start()
    threading.Thread(target=loop, daemon=True).start()
ssh = paramiko.SSHClient(); ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("10.0.0.8", username="ubuntu", key_filename="/home/me/.ssh/id_rsa")
forward_tunnel(local_port=15432, remote_host="127.0.0.1", remote_port=5432, transport=ssh.get_transport())
input("本地已转发到 15432，按回车退出…\n")

长时间任务、sudo 与伪终端

exec_command 默认不是交互式终端。需要 sudo 的命令，常见做法是：要么免密 sudo，要么申请一个伪终端（PTY）并写入密码（生产环境更推荐免密+最小权限）。

import paramiko, time
ssh = paramiko.SSHClient(); ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("10.0.0.8", username="ubuntu", key_filename="/home/me/.ssh/id_rsa")
chan = ssh.invoke_shell()  # 交互式 shell，带 pty
time.sleep(0.2); chan.recv(4096)
chan.send("sudo systemctl restart nginx\n")
time.sleep(0.2)
chan.send("your_password_here\n")  # 更推荐免密 sudo，不在脚本里写密码
time.sleep(1)
print(chan.recv(65535).decode())
chan.close(); ssh.close()

跑长任务（比如备份、抓日志）可以用 nohup … & 或 tmux/screen，避免 SSH 掉线中断。

cmd = "nohup bash -c 'sleep 30; echo done >> /tmp/job.log' >/tmp/job.out 2>&1 &"
ssh.exec_command(cmd)

常见坑怎么躲

• known_hosts：默认严格校验主机指纹。上面示例用了 AutoAddPolicy() 直连方便开发，但生产建议自己维护 known_hosts，或手工比对指纹。
• 编码：stdout.read().decode(errors="ignore") 可避免偶发编码错误。
• 超时与重试：connect(timeout=...)、exec_command(timeout=...) 都要设；网络抖动就重试。
• 大输出：非常大的输出建议分块读，或把结果重定向到远端文件再 SFTP 回来。
• 权限：SFTP 上传目录权限不够是高频问题；chown/chmod 常配套使用。

一个可复用的小工具（支持：命令/上传/下载/跳板/重试）

把上面的点揉成一个简单类，直接塞进你的项目即可。

import time, paramiko
from typing import Optional, Tuple
classSSH:
def__init__(self, host: str, user: str, port: int = 22,
                 key_path: Optional[str] = None, password: Optional[str] = None,
                 bastion: Optional[str] = None, retries: int = 2, timeout: int = 10):
        self.host, self.user, self.port = host, user, port
        self.key_path, self.password = key_path, password
        self.bastion, self.retries, self.timeout = bastion, retries, timeout
        self._bcli = None# bastion client
        self._cli = None# target client
def_connect_once(self):
        pkey = None
if self.key_path:
try:
                pkey = paramiko.RSAKey.from_private_key_file(self.key_path)
except paramiko.PasswordRequiredException:
# 如果你的私钥有口令，这里可调用 from_private_key_file(path, password=...)
raise
        cli = paramiko.SSHClient(); cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        sock = None
if self.bastion:
            bcli = paramiko.SSHClient(); bcli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            bcli.connect(self.bastion, username=self.user, pkey=pkey, password=self.password, timeout=self.timeout)
            self._bcli = bcli
            chan = bcli.get_transport().open_channel(
"direct-tcpip", (self.host, self.port), ("127.0.0.1", 0)
            )
            sock = chan
        cli.connect(self.host, port=self.port, username=self.user,
                    pkey=pkey, password=self.password, sock=sock, timeout=self.timeout)
        self._cli = cli
defconnect(self):
        last = None
for i in range(self.retries + 1):
try:
                self._connect_once(); return self
except Exception as e:
                last = e; time.sleep(min(1 + i, 5))
raise last
defrun(self, cmd: str, timeout: Optional[int] = None) -> Tuple[int, str, str]:
ifnot self._cli: self.connect()
        _, out, err = self._cli.exec_command(cmd, timeout=timeout or self.timeout)
        code = out.channel.recv_exit_status()
return code, out.read().decode(errors="ignore"), err.read().decode(errors="ignore")
defupload(self, local: str, remote: str):
ifnot self._cli: self.connect()
        sftp = self._cli.open_sftp()
try: sftp.put(local, remote)
finally: sftp.close()
defdownload(self, remote: str, local: str):
ifnot self._cli: self.connect()
        sftp = self._cli.open_sftp()
try: sftp.get(remote, local)
finally: sftp.close()
defclose(self):
try:
if self._cli: self._cli.close()
finally:
try:
if self._bcli: self._bcli.close()
finally:
                self._cli = self._bcli = None
# 用法演示
if __name__ == "__main__":
    ssh = SSH(host="10.0.0.8", user="ubuntu", key_path="/home/me/.ssh/id_rsa",
              bastion=None, retries=2, timeout=10).connect()
    code, out, err = ssh.run("df -h && free -m")
    print(code, out or err)
    ssh.upload("app.tar.gz", "/tmp/app.tar.gz")
    ssh.download("/var/log/syslog", "syslog.backup")
    ssh.close()

何时考虑替代/补充

• 想要更“命令式”的语法：Fabric（基于 paramiko，写起来更像脚本），适合成体系的部署流程。
• 需要异步并发：asyncssh（asyncio 风格），成百上千台并发时更轻盈。
• Windows 主机：用 pywinrm 走 WinRM，比“在 Windows 上用 SSH 再 sudo”靠谱。

到这，一般日常“连上去把事儿办了”的诉求——无论单机、多机、跳板、传文件、开端口——一个 paramiko 就能打包覆盖。把上面的工具类丢进你的代码库，后面项目里基本就是三句话：connect()、run()、upload()/download()。剩下的，就是把超时、重试、权限这些小细节按你的环境稳稳拧紧。