网络与远程操作

运维脚本的网络活就两类：调 HTTP 接口（健康检查、查 CMDB、触发 webhook）和 SSH 到机器上干活。这篇笔记按「requests 调接口 → socket 测端口 → paramiko 单兵 SSH → fabric 人性化封装」的顺序记录，示例都是可直接照抄运行的完整脚本。示例环境：macOS / Linux，Python 3.12+。

依赖安装

文中第三方库统一用 pip 安装，建议在 venv 或 pipx 环境里玩：

python3 -m pip install requests paramiko fabric

requests：HTTP 一把梭

python3 -m pip install requests

GET 与 POST，超时必带

import requests

# GET：查一个接口
resp = requests.get("https://api.example.com/v1/nodes", params={"env": "prod"}, timeout=5)
resp.raise_for_status()          # 状态码非 2xx 直接抛异常，别静默吞掉
data = resp.json()               # 返回 JSON 时一把解析成 dict

# POST：提交 JSON
resp = requests.post(
    "https://api.example.com/v1/deploy",
    json={"app": "web", "version": "1.2.3"},   # json= 自动序列化并带 Content-Type
    headers={"Authorization": "Bearer <token>"},
    timeout=5,
)

timeout=5 表示连接+读取总等待 5 秒。requests 默认永不超时，线上接口挂住时你的脚本就跟着挂一辈子，这是新手脚本半夜卡死的最常见原因。想分开控制就传元组 timeout=(3, 10)（连接 3 秒、读取 10 秒）。

Session：复用连接

循环调同一个服务的接口时，用 Session 复用 TCP 连接和公共请求头，比每次 requests.get 快一截：

import requests

s = requests.Session()
s.headers.update({"Authorization": "Bearer <token>"})

for node in ["web1", "web2", "web3"]:
    r = s.get(f"http://{node}:8080/health", timeout=3)
    print(node, r.status_code)

重试：HTTPAdapter + Retry

网络抖动时失败一次就报警太神经质，给 Session 挂个重试策略：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

s = requests.Session()
retry = Retry(
    total=3,                    # 总共重试 3 次
    backoff_factor=0.5,         # 间隔 0.5s、1s、2s 指数退避
    status_forcelist=[502, 503, 504],  # 这些状态码才重试
    allowed_methods=["GET"],    # 只对幂等的 GET 重试，POST 重试要想清楚
)
s.mount("http://", HTTPAdapter(max_retries=retry))
s.mount("https://", HTTPAdapter(max_retries=retry))

POST 别盲目重试

创建工单、触发部署这类非幂等接口，重试可能产生重复记录。要么只在连接错误时重试，要么让服务端用幂等键去重。

证书校验 verify

verify 默认 True，校验 HTTPS 证书。公司内网自签证书两种处理：

# 方式一：指定内部 CA 证书（推荐，校验仍然生效）
requests.get("https://internal.example.com", verify="/etc/pki/internal-ca.pem", timeout=5)

# 方式二：关掉校验（会刷 InsecureRequestWarning 警告，仅限临时调试）
requests.get("https://internal.example.com", verify=False, timeout=5)

verify=False 等于放弃了防中间人攻击，脚本能跑通后第一时间换回 CA 证书方式。

实战：URL 存活探测脚本

巡检一批 URL，输出状态码和耗时，非 200 或超时的标出来——这是监控系统的穷人版：

#!/usr/bin/env python3
"""check_url.py：批量探测 URL 存活，输出状态码与耗时"""
import time

import requests

URLS = [
    "https://www.example.com",
    "http://web1.internal:8080/health",
    "http://web2.internal:8080/health",
]

def probe(url: str) -> str:
    start = time.monotonic()
    try:
        r = requests.get(url, timeout=3)
        cost = (time.monotonic() - start) * 1000
        mark = "OK  " if r.status_code == 200 else "FAIL"
        return f"{mark} {r.status_code} {cost:6.0f}ms  {url}"
    except requests.RequestException as e:
        cost = (time.monotonic() - start) * 1000
        return f"FAIL ERR  {cost:6.0f}ms  {url}  ({type(e).__name__})"

if __name__ == "__main__":
    for line in map(probe, URLS):
        print(line)

输出长这样：

OK   200    182ms  https://www.example.com
FAIL ERR    21ms  http://web1.internal:8080/health  (ConnectTimeout)
OK   200     35ms  http://web2.internal:8080/health

接 crontab 每分钟跑一次，非 OK 的发邮件或 webhook，一套简陋但够用的拨测就有了。

调 REST API 拿 JSON

以调 Prometheus 查询接口为例，看看「拿 JSON → 取字段」的标准姿势：

import requests

r = requests.get(
    "http://prometheus:9090/api/v1/query",
    params={"query": 'up{job="node"}'},
    timeout=5,
)
r.raise_for_status()
for item in r.json()["data"]["result"]:
    instance = item["metric"]["instance"]
    value = item["value"][1]          # ["1729814400", "1"] 里取第二个元素
    print(instance, "UP" if value == "1" else "DOWN")

要点就一个：先 raise_for_status() 再 .json()，否则 502 返回的 HTML 错误页会让 .json() 抛一个莫名其妙的解码异常，排障时绕远路。

socket：最底层的端口探测

标准库自带，不用装任何东西。connect_ex 尝试建立 TCP 连接，返回 0 表示通，非 0 是错误码——比脚本里调 telnet / nc 干净得多，不用解析命令输出，也没有额外依赖。

import socket

def port_open(host: str, port: int, timeout: float = 2.0) -> bool:
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(timeout)             # 不设超时会阻塞到系统超时（可能两分钟）
    try:
        return s.connect_ex((host, port)) == 0
    finally:
        s.close()

print(port_open("192.168.1.11", 22))    # True
print(port_open("192.168.1.11", 3306))  # False

20 行的多端口扫描

上线前验证「这台机器该开的端口都开了、不该开的都没开」：

#!/usr/bin/env python3
"""port_scan.py：扫一台机器的一批端口"""
import socket
from concurrent.futures import ThreadPoolExecutor

HOST = "192.168.1.11"
PORTS = [22, 80, 443, 3306, 6379, 8080]

def check(port: int) -> str:
    with socket.socket() as s:        # with 语句自动 close
        s.settimeout(1.5)
        state = "open  " if s.connect_ex((HOST, port)) == 0 else "closed"
    return f"{HOST}:{port:<6} {state}"

with ThreadPoolExecutor(max_workers=20) as pool:
    for line in pool.map(check, PORTS):
        print(line)

192.168.1.11:22     open
192.168.1.11:80     open
192.168.1.11:443    open
192.168.1.11:3306   closed
192.168.1.11:6379   closed
192.168.1.11:8080   open

超时必须设

socket 默认阻塞模式，connect_ex 对不可达地址会等到内核 TCP 重试耗尽（一两分钟）才返回。任何网络探测代码，第一行先 settimeout。

paramiko：SSH 到任何机器

socket 只能回答「端口通不通」，真要在远端执行命令得走 SSH。paramiko 是纯 Python 的 SSHv2 库，是这块事实上的标准。

python3 -m pip install paramiko

连接：密钥优先，密码兜底

import paramiko

client = paramiko.SSHClient()
client.load_system_host_keys()                      # 读 ~/.ssh/known_hosts
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())  # 首次连接的指纹自动收下

# 方式一：密钥登录（推荐，会顺藤摸瓜找 ~/.ssh/ 下的私钥和 ssh-agent）
client.connect("192.168.1.11", port=22, username="ops", timeout=10)

# 方式二：密码登录
client.connect("192.168.1.11", port=22, username="ops", password="S3cret", timeout=10)

timeout=10 是连接超时，别省。密钥方式显式指定私钥文件就加 key_filename="/home/ops/.ssh/id_ed25519"。

exec_command：执行命令拿三件套

stdin, stdout, stderr = client.exec_command("df -h /", timeout=15)
rc = stdout.channel.recv_exit_status()   # 返回码要先拿，它会等命令执行完
out = stdout.read().decode().strip()
err = stderr.read().decode().strip()
print(f"rc={rc}\n{out}")
if rc != 0:
    print(f"命令失败: {err}")

两个容易踩的点：

• exec_command 本身只是「发出命令」，recv_exit_status() 或 read() 才会阻塞等结果。
• exec_command 的 timeout 是「命令执行多久算超时」，和 connect 的 timeout 是两回事，两个都要设。

SFTP：上传下载文件

sftp = client.open_sftp()
sftp.put("local/app.tar.gz", "/tmp/app.tar.gz")      # 上传
sftp.get("/var/log/app/error.log", "logs/error.log") # 下载
sftp.close()

AutoAddPolicy 与 known_hosts 的取舍

AutoAddPolicy 的意思是「没见过的指纹就自动相信」，图省事但理论上给中间人攻击留了门。按场景选：

场景	策略
内网临时脚本、机器清单经常变	AutoAddPolicy，能接受
生产长期运行的自动化	提前用 ssh-keyscan 收集指纹写进 known_hosts，不配 missing policy（未知指纹直接报错）

# 提前把一批机器的指纹收进 known_hosts
ssh-keyscan -t ed25519 192.168.1.{11,12,13} >> ~/.ssh/known_hosts

实战：三台机器批量 uptime 并汇总

把上面拼起来，加上并发，就是一个真正能用的批量巡检脚本：

#!/usr/bin/env python3
"""batch_uptime.py：批量执行 uptime 并汇总输出"""
import paramiko
from concurrent.futures import ThreadPoolExecutor

HOSTS = [
    {"hostname": "192.168.1.11", "username": "ops"},
    {"hostname": "192.168.1.12", "username": "ops"},
    {"hostname": "192.168.1.13", "username": "ops"},
]

def run_uptime(h: dict) -> str:
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(**h, timeout=10)
        _, stdout, _ = client.exec_command("uptime", timeout=15)
        stdout.channel.recv_exit_status()
        return f"{h['hostname']:<16} {stdout.read().decode().strip()}"
    except Exception as e:
        return f"{h['hostname']:<16} ERROR: {e}"
    finally:
        client.close()

if __name__ == "__main__":
    with ThreadPoolExecutor(max_workers=10) as pool:
        for line in pool.map(run_uptime, HOSTS):
            print(line)

192.168.1.11      14:32:01 up 46 days,  3:12,  1 user,  load average: 0.42, 0.38, 0.35
192.168.1.12      14:32:01 up 46 days,  3:11,  0 users,  load average: 1.87, 1.62, 1.40
192.168.1.13      ERROR: [Errno 60] Operation timed out

为什么并发时每个线程独立 new 一个 SSHClient

paramiko 的 SSHClient 不是线程安全的，多线程共用一个连接会互相踩状态，出现诡异的挂死。正确姿势就是上面这样：每个线程（或每台机器）各开各的连接，用完各自 close()。

fabric：paramiko 的人性化封装

paramiko 写多了会发现样板代码一堆：建连接、收三件套、判返回码、关连接。fabric（2.x 版本）在 paramiko 之上封了一层，把这些都消化掉了。

python3 -m pip install fabric

from fabric import Connection

c = Connection("ops@192.168.1.11", connect_kwargs={"timeout": 10})

r = c.run("uptime")                       # 直接返回结果对象，stdout/return_code 都在上面
print(r.stdout.strip(), r.return_code)

c.run("df -h / | tail -1", hide=True)     # hide=True 不在终端回显命令输出

c.sudo("systemctl restart nginx")         # 提权执行，自动处理 sudo 密码提示

c.put("local/nginx.conf", "/tmp/nginx.conf")  # 上传文件

c.close()

对比 paramiko 版本，run 一个调用就替代了「exec_command + recv_exit_status + read」三连，脚本里逻辑密度高很多。Connection 也支持 for 循环批量：

from fabric import Connection

for host in ["ops@192.168.1.11", "ops@192.168.1.12", "ops@192.168.1.13"]:
    try:
        with Connection(host, connect_kwargs={"timeout": 10}) as c:
            r = c.run("uptime", hide=True)
            print(f"{host:<20} {r.stdout.strip()}")
    except Exception as e:
        print(f"{host:<20} ERROR: {e}")

fabric 还是 Ansible

两台三台机器、临时一次性任务、逻辑里要拼条件判断——fabric 写起来快，几十行 Python 搞定。机器几十台、任务要反复跑、要幂等要留档——那是声明式编排的地盘，请上 Ansible，两者的定位差异在 Ansible 核心概念与架构 里讲得很清楚。经验法则：第二次还要跑，就从 fabric 搬到 playbook。

fabric 不是 paramiko 的并发加速器

fabric 让单连接好用，但批量并发还是得自己开线程（或用 fabric.ThreadingGroup）。几十台机器以上、要带汇报和幂等的场景，别再堆 Python 并发代码了，上 Ansible。

常见坑

• paramiko 无超时挂死：connect() 和 exec_command() 的超时是分开的，漏掉任何一个，遇到半死机器（TCP 能连、SSH 握手卡死，或命令不退出）脚本就永远挂住。两个都设，10 秒和 15 秒是顺手的值。
• HostKey 未知被拒：不配 set_missing_host_key_policy、known_hosts 里又没有目标机器时，connect 直接抛 SSHException: Server ... not found in known_hosts。新机器进自动化清单前，要么 ssh-keyscan 预收指纹，要么脚本里显式用 AutoAddPolicy。
• 连接不 close 泄漏：脚本循环连几十台机器不关连接，本机文件描述符和远端 sshd 连接数都会被打爆（表现是 Too many open files 或 sshd 拒绝新连接）。用 with 或 try/finally 保证 close()。
• 并发时共享连接：SSHClient 不是线程安全的，多线程共用会随机挂死。每线程独立连接，别图省事共享。

小结

• 调 HTTP 接口用 requests：timeout 必带，Session 复用连接，HTTPAdapter + Retry 扛抖动，raise_for_status() 别省。
• 测端口用 socket.connect_ex，比 telnet/nc 适合脚本化；超时必须设。
• SSH 自动化用 paramiko：connect/exec_command 双超时，三件套（stdout/stderr/返回码）都要拿，用完 close()。
• 嫌 paramiko 啰嗦就上 fabric；任务变成「重复跑、要幂等、要编排」就该换 Ansible。
• 网络脚本的第一原则是所有调用都有超时——没超时的脚本，挂死只是时间问题。