【Python 网络】网络变更怕翻车？5个Python脚本实现自动化验收
适用场景：网络变更后的自动化验收。覆盖端口连通性、路由验证、DNS解析、配置对比、端到端探测5个核心场景。
01
凌晨3点的验收噩梦
上周三凌晨，一个朋友发来消息：数据中心割接，400台服务器迁移到新网络，变更本身1个小时搞定，验收干了4个小时。
400台服务器，每台要验证管理口连通性、业务端口可达性、DNS解析正确性、路由路径正确性。他一个人在终端里 ping、telnet、dig、traceroute 轮着敲，敲到手指抽筋。
结果还是漏了3台服务器的DNS没更新，第二天早上业务方打电话过来才发现。
这种场景，做过网络变更的运维都经历过。变更本身不可怕，可怕的是变更后的验收——项目多、时间紧、容易遗漏、出了问题还不好追溯。
下面这5个Python脚本，分别覆盖网络验收的5个核心场景：端口扫描、路由验证、DNS检查、配置对比、端到端探测。每个脚本都能独立使用，也能串成一套完整的验收流水线。
| 脚本 |
场景 |
解决的问题 |
| port_checker |
端口连通性 |
批量验证防火墙规则是否生效 |
| route_validator |
路由验证 |
确认路由路径和下一跳正确 |
| dns_checker |
DNS解析 |
验证域名解析记录是否已更新 |
| config_diff |
配置对比 |
变更前后配置差异自动比对 |
| e2e_probe |
端到端探测 |
全链路连通性和延迟综合检测 |
02
脚本一：端口连通性批量扫描
网络变更后第一件事就是验证端口。防火墙规则加了没有？新交换机的ACL配了没有？负载均衡的后端端口通了没有？这些问题都需要逐一验证。
手动 telnet 或 nc 一个个试太慢。这个脚本读取一份端口清单（CSV格式），用多线程并发扫描，几十秒就能验证几百个端口。
#!/usr/bin/env python3
# port_checker.py — 端口连通性批量扫描
# 用法：python3 port_checker.py ports.csv --timeout 3
import socket, csv, concurrent.futures, sys, argparse
from datetime import datetime
def check_port(host, port, timeout=3):
"""检测单个端口是否可达"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
result = sock.connect_ex((host, int(port)))
sock.close()
return {"host": host, "port": port, "status": "OPEN" if result == 0 else "CLOSED"}
except Exception as e:
return {"host": host, "port": port, "status": "ERROR", "detail": str(e)}
def main():
parser = argparse.ArgumentParser(description="批量端口扫描")
parser.add_argument("csvfile", help="端口清单CSV: host,port,service")
parser.add_argument("--timeout", type=int, default=3)
parser.add_argument("--workers", type=int, default=50)
args = parser.parse_args()
tasks = []
with open(args.csvfile, "r") as f:
for row in csv.DictReader(f):
tasks.append((row["host"], row["port"]))
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool:
futures = {pool.submit(check_port, h, p, args.timeout): (h, p) for h, p in tasks}
for future in concurrent.futures.as_completed(futures):
results.append(future.result())
open_count = sum(1 for r in results if r["status"] == "OPEN")
closed = [r for r in results if r["status"] != "OPEN"]
print(f"扫描完成: {open_count}/{len(results)} 端口开放")
if closed:
print("未开放端口:")
for r in closed:
print(f" ✗ {r['host']}:{r['port']} → {r['status']}")
if __name__ == "__main__":
main()
CSV 文件格式示例：
# ports.csv
host,port,service
10.0.1.10,443,HTTPS
10.0.1.10,3389,RDP
10.0.1.20,1433,MSSQL
10.0.2.5,22,SSH
50个并发线程，扫描200个端口通常不超过10秒。扫描结束后，所有未开放的端口会被汇总输出，方便快速定位问题。
03
脚本二：路由路径验证
端口通了不代表路由对了。新网络的默认网关配了吗？静态路由加了吗？BGP邻居建起来了吗？流量走的是预期路径吗？
这个脚本通过解析系统路由表来验证关键网段的路由是否指向正确的下一跳。适合在割接后快速确认路由配置是否生效。
#!/usr/bin/env python3
# route_validator.py — 路由路径验证
# 用法：python3 route_validator.py routes.json
import subprocess, json, sys
def get_route_for(target_ip):
"""通过 ip route get 获取到达目标IP的实际路由"""
try:
out = subprocess.run(
["ip", "route", "get", target_ip],
capture_output=True, text=True, timeout=5
)
line = out.stdout.strip().split("\n")[0]
parts = line.split()
gateway = None
dev = None
for i, p in enumerate(parts):
if p == "via" and i + 1 < len(parts):
gateway = parts[i + 1]
if p == "dev" and i + 1 < len(parts):
dev = parts[i + 1]
return {"gateway": gateway, "dev": dev}
except Exception as e:
return {"error": str(e)}
def main():
with open(sys.argv[1], "r") as f:
expected = json.load(f)
fail_count = 0
for item in expected:
target = item["target"]
expect_gw = item.get("gateway")
expect_dev = item.get("dev")
actual = get_route_for(target)
if "error" in actual:
print(f"✗ {target}: 路由查询失败")
fail_count += 1
continue
mismatch = []
if expect_gw and actual.get("gateway") != expect_gw:
mismatch.append(f"网关: 期望{expect_gw} 实际{actual.get('gateway')}")
if expect_dev and actual.get("dev") != expect_dev:
mismatch.append(f"接口: 期望{expect_dev} 实际{actual.get('dev')}")
if mismatch:
print(f"✗ {target}: {' | '.join(mismatch)}")
fail_count += 1
else:
print(f"✓ {target}: 路由正确")
print(f"\n验证完成: {len(expected) - fail_count}/{len(expected)} 通过")
sys.exit(1 if fail_count > 0 else 0)
if __name__ == "__main__":
main()
routes.json 示例：
// routes.json — 预期路由配置
[
{"target": "10.0.1.10", "gateway": "10.0.0.1", "dev": "eth0"},
{"target": "172.16.0.0", "gateway": "10.0.0.254", "dev": "eth1"}
]
脚本退出码为0表示全部通过，非0表示有路由不匹配。可以直接嵌入CI/CD或变更流水线中作为门禁检查。
04
脚本三：DNS解析记录验证
DNS是网络变更后最容易被遗漏的环节。服务器IP换了，DNS A记录改了吗？CNAME指向对吗？MX记录要不要更新？
这个脚本用 dnspython 库直接查询DNS记录，和预期值做对比。支持A、AAAA、CNAME、MX、TXT等多种记录类型。
安装依赖：
pip3 install dnspython
dns_checker.py 核心代码：
#!/usr/bin/env python3
# dns_checker.py — DNS解析记录验证
# 用法：python3 dns_checker.py dns_records.json --server 8.8.8.8
import json, sys, argparse
import dns.resolver
def check_dns(domain, rtype, expected, server=None):
resolver = dns.resolver.Resolver()
if server:
resolver.nameservers = [server]
try:
answers = resolver.resolve(domain, rtype)
actual = sorted([str(r) for r in answers])
expect = sorted(expected) if isinstance(expected, list) else [expected]
if actual == expect:
return {"status": "OK", "actual": actual}
else:
return {"status": "MISMATCH", "expected": expect, "actual": actual}
except dns.resolver.NXDOMAIN:
return {"status": "NXDOMAIN"}
except Exception as e:
return {"status": "ERROR", "detail": str(e)}
def main():
parser = argparse.ArgumentParser(description="DNS记录验证")
parser.add_argument("jsonfile", help="DNS记录清单JSON")
parser.add_argument("--server", default=None, help="指定DNS服务器")
args = parser.parse_args()
with open(args.jsonfile) as f:
records = json.load(f)
fail = 0
for rec in records:
result = check_dns(rec["domain"], rec["type"], rec["expected"], args.server)
tag = "✓" if result["status"] == "OK" else "✗"
print(f"{tag} {rec['domain']} {rec['type']}: {result['status']}")
if result["status"] != "OK":
fail += 1
if "expected" in result:
print(f" 期望: {result['expected']}")
print(f" 实际: {result['actual']}")
print(f"\n验证完成: {len(records)-fail}/{len(records)} 通过")
sys.exit(1 if fail > 0 else 0)
if __name__ == "__main__":
main()
dns_records.json 示例：
// dns_records.json
[
{"domain": "api.example.com", "type": "A", "expected": ["10.0.1.10"]},
{"domain": "mail.example.com", "type": "MX", "expected": ["10 mail.example.com."]}
]
支持指定DNS服务器进行查询，可以验证内部DNS是否已更新，也可以对比公网DNS（如8.8.8.8）的解析结果。
05
脚本四：网络设备配置对比
变更前后，设备配置到底改了什么？手动 diff 几百行配置文件容易看花眼。这个脚本自动抓取变更前后配置，生成差异报告。
#!/usr/bin/env python3
# config_diff.py — 配置差异对比
# 用法：python3 config_diff.py before.conf after.conf
import difflib, sys
def diff_configs(before_path, after_path):
with open(before_path) as f:
before = f.readlines()
with open(after_path) as f:
after = f.readlines()
diff = difflib.unified_diff(before, after, fromfile="变更前", tofile="变更后", lineterm="")
changes = list(diff)
if not changes:
print("✓ 配置无变化")
return 0
added = sum(1 for line in changes if line.startswith("+") and not line.startswith("+++"))
removed = sum(1 for line in changes if line.startswith("-") and not line.startswith("---"))
print(f"发现 {added} 行新增, {removed} 行删除")
print("\n".join(changes))
return added + removed
if __name__ == "__main__":
count = diff_configs(sys.argv[1], sys.argv[2])
sys.exit(1 if count > 0 else 0)
这个脚本利用 Python 标准库 difflib 生成 unified diff 格式的差异报告。可以在变更前后分别导出配置文件，然后自动对比。
06
脚本五：端到端链路探测
最后一个脚本是端到端探测。它模拟真实的业务请求，从源端发起探测，验证整条链路的连通性和延迟。
#!/usr/bin/env python3
# e2e_probe.py — 端到端链路探测
# 用法：python3 e2e_probe.py endpoints.json
import socket, json, sys, time, statistics
def probe_endpoint(host, port, count=5):
latencies = []
for _ in range(count):
try:
start = time.time()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
sock.connect((host, port))
latency = (time.time() - start) * 1000
sock.close()
latencies.append(latency)
except:
return {"status": "FAIL"}
return {
"status": "OK",
"avg_ms": round(statistics.mean(latencies), 2),
"min_ms": round(min(latencies), 2),
"max_ms": round(max(latencies), 2)
}
def main():
with open(sys.argv[1]) as f:
endpoints = json.load(f)
fail = 0
for ep in endpoints:
result = probe_endpoint(ep["host"], ep["port"])
if result["status"] == "OK":
print(f"✓ {ep['name']}: avg={result['avg_ms']}ms")
else:
print(f"✗ {ep['name']}: 连接失败")
fail += 1
print(f"\n探测完成: {len(endpoints)-fail}/{len(endpoints)} 通过")
sys.exit(1 if fail > 0 else 0)
if __name__ == "__main__":
main()
网络变更后的验收不应该靠人肉。这5个脚本覆盖了端口、路由、DNS、配置、端到端5个核心场景，可以独立使用，也可以串成一套完整的验收流水线。把验收时间从4小时压缩到30分钟，关键是——不会遗漏。
— END —
