12.3 可重复的网络诊断实验
这一课只在本机回环接口上制造故障。目标不是记住某台机器的输出,而是练习如何定义阶段、设置 deadline、保留证据并写出有限结论。
实验边界
- 只使用
127.0.0.1和本机临时端口; - 不扫描公网或未经授权的主机;
- 不修改防火墙、路由或系统 DNS;
- 不使用
curl -k掩盖证书错误; - 每个后台进程都记录 PID 并在结束时停止。
不同系统命令选项可能不同。下文使用 Linux、bash、GNU timeout 与 Python 3.11+。先打开专用 bash 会话,在新建的临时目录内保存实验脚本和日志;不要在日常工作目录直接启动文件服务。启动失败时,代码会退出这个会话,退出钩子会停止它创建的服务。实验结束后正常执行清理命令。
实验一:区分拒绝、连接成功与响应超时
创建 stall_server.py:
import socket
import time
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind(("127.0.0.1", 0))
listener.listen()
host, port = listener.getsockname()
print(port, flush=True)
connection, peer = listener.accept()
with connection:
print(f"accepted={peer}", flush=True)
time.sleep(10)运行它并读取操作系统分配的端口:
: >stall.log
python3 stall_server.py >stall.log 2>&1 &
server_pid=$!
trap 'kill "$server_pid" 2>/dev/null || true' EXIT
for _ in {1..50}; do
server_port=$(sed -n '1p' stall.log)
test -n "$server_port" && break
sleep 0.1
done
if ! [[ "$server_port" =~ ^[0-9]+$ ]]; then
cat stall.log >&2
exit 1
fi先验证监听:
ss -lnt "sport = :$server_port"再请求这个“接受连接但不返回 HTTP”的服务:
curl --noproxy '*' --verbose --connect-timeout 1 --max-time 2 \
"http://127.0.0.1:$server_port/"预期:TCP 很快连接成功,HTTP 响应在总 deadline 内没有到达,curl 返回 28。这里的 timeout 发生在连接之后。--noproxy '*' 让请求直接走回环接口,避免代理环境变量改变实验路径。
停止服务,再请求同一端口:
kill "$server_pid"
wait "$server_pid" 2>/dev/null || true
trap - EXIT
curl --noproxy '*' --verbose --connect-timeout 1 --max-time 2 \
"http://127.0.0.1:$server_port/"在普通本机 TCP 栈中,未监听端口通常很快返回 Connection refused,curl 返回 7。它与前一次“已连接、等不到响应”的阶段不同。
这个实验没有模拟 SYN 被静默丢弃。连接超时依赖路由、防火墙和重传计时,不能安全地用任意公网地址稳定复现。
实验二:用抓包验证错误阶段
先重新启动实验一的服务,再开始抓包,最后依次重做两次 curl 请求和停止服务的步骤。把本次启动的端口值复制到抓包终端的 server_port 变量中;两个终端不会自动共享 shell 变量。终端 A 在回环接口采集,只抓该临时端口:
sudo timeout --signal=INT 8s tcpdump -i lo -nn -s 0 \
"tcp port $server_port" -c 20 -w loopback-lab.pcap看到 tcpdump 的 listening on lo 后,终端 B 执行请求与停止步骤。抓够 20 个包或经过 8 秒就结束;GNU timeout 到时返回 124 是预期停止原因,仍需确认 tcpdump 没有报告采集错误。读取摘要:
tcpdump -nn -r loopback-lab.pcap回答这些问题:
- 两次请求是否都发出了 SYN?
- 哪一次完成了三次握手?
- 拒绝场景中是否出现 RST?
- 响应超时场景中,客户端是否发送了 HTTP 请求字节?
- FIN 或 RST 由哪一端发出?这与
curl的 deadline 有何关系?
回环接口上的包形态和普通以太网接口不完全相同,但 TCP 状态转换仍可用于本实验。
实验三:测量一个本地 HTTP 请求
把下面程序保存为 http_server.py。它只返回固定文本,不会把当前目录变成文件服务器:
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
body = b"ok\n"
self.send_response(200)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
with ThreadingHTTPServer(("127.0.0.1", 0), Handler) as server:
print(server.server_port, flush=True)
server.serve_forever()启动服务并读取端口:
: >http.log
python3 http_server.py >http.log 2>&1 &
http_pid=$!
trap 'kill "$http_pid" 2>/dev/null || true' EXIT
for _ in {1..50}; do
http_port=$(sed -n '1p' http.log)
test -n "$http_port" && break
sleep 0.1
done
if ! [[ "$http_port" =~ ^[0-9]+$ ]]; then
cat http.log >&2
exit 1
fi执行五次请求:
for run in {1..5}; do
curl --noproxy '*' --silent --show-error --output /dev/null \
--connect-timeout 1 --max-time 2 \
--write-out "run=$run code=%{http_code} connect=%{time_connect} starttransfer=%{time_starttransfer} total=%{time_total}\n" \
"http://127.0.0.1:$http_port/"
done在这个本地 HTTP/1.x 实验里,DNS 与 TLS 不参与。即使如此,首次请求、线程创建、调度和服务日志也会带来波动。不要用五个样本宣称某项优化普遍有效;本实验只训练测量字段和阶段解释。
结束时清理:
kill "$http_pid"
wait "$http_pid" 2>/dev/null || true
trap - EXITTCP 与 TLS 共用 deadline 的 Python 诊断器
下面程序先单独测量 DNS,再让 TCP 连接与 TLS 握手共享一个剩余预算。上一阶段花掉的时间不会在下一阶段“充值”。TLS 继续校验证书链和主机名;可选 cafile 只用于显式信任测试 CA,不会关闭验证。异常携带失败阶段和已完成的测量。保存为 diagnose.py,显式传入目标;本地测试应使用证书包含相应 IP 或主机名的临时 TLS 服务。
from __future__ import annotations
import argparse
import math
import socket
import ssl
import sys
import time
from dataclasses import dataclass
@dataclass(frozen=True)
class Stage:
name: str
milliseconds: float
detail: str
class DiagnosisError(RuntimeError):
def __init__(self, stage: str, completed: list[Stage], cause: OSError):
super().__init__(f"{stage}: {cause}")
self.stage = stage
self.completed = tuple(completed)
def diagnose(host: str, port: int = 443, timeout: float = 3.0,
*, cafile: str | None = None) -> list[Stage]:
if not isinstance(host, str) or not host.strip():
raise ValueError("host must be a nonempty string")
if type(port) is not int or not 1 <= port <= 65535:
raise ValueError("port must be an integer in 1..65535")
if type(timeout) not in (int, float) or not 0 < timeout < math.inf:
raise ValueError("timeout must be positive and finite")
context = ssl.create_default_context(cafile=cafile)
context.set_alpn_protocols(["http/1.1"])
stages: list[Stage] = []
phase = "dns"
try:
started = time.monotonic()
addresses = socket.getaddrinfo(
host, port, type=socket.SOCK_STREAM, proto=socket.IPPROTO_TCP,
)
if not addresses:
raise OSError("resolver returned no addresses")
stages.append(Stage("dns", (time.monotonic() - started) * 1000,
f"answers={len(addresses)}"))
deadline = time.monotonic() + timeout
def remaining() -> float:
seconds = deadline - time.monotonic()
if seconds <= 0:
raise TimeoutError("TCP/TLS deadline exceeded")
return seconds
family, socktype, proto, _, sockaddr = addresses[0]
phase = "tcp"
started = time.monotonic()
with socket.socket(family, socktype, proto) as raw:
raw.settimeout(remaining())
raw.connect(sockaddr)
stages.append(Stage("tcp", (time.monotonic() - started) * 1000,
f"peer={sockaddr}"))
phase = "tls"
started = time.monotonic()
with context.wrap_socket(raw, server_hostname=host,
do_handshake_on_connect=False) as tls:
tls.settimeout(remaining())
tls.do_handshake()
stages.append(Stage(
"tls", (time.monotonic() - started) * 1000,
f"version={tls.version()} alpn={tls.selected_alpn_protocol()}",
))
except OSError as error:
raise DiagnosisError(phase, stages, error) from error
return stages
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("host")
parser.add_argument("--port", type=int, default=443)
parser.add_argument("--timeout", type=float, default=3.0)
parser.add_argument("--cafile")
args = parser.parse_args()
try:
completed = diagnose(args.host, args.port, args.timeout,
cafile=args.cafile)
except DiagnosisError as error:
for stage in error.completed:
print(stage)
print(error, file=sys.stderr)
raise SystemExit(1)
for stage in completed:
print(stage)它仍有明确限制:
- 只尝试
getaddrinfo()返回的第一个地址,没有实现 Happy Eyeballs; - 同步
getaddrinfo()没有这里的硬 deadline;TLS 上下文准备也在预算之外,因此整个函数不保证在timeout内返回; - 没有发送 HTTP 请求;
- 一次样本不能代表长期性能;
alpn=None只说明未协商 ALPN;本例仅提出http/1.1,不能据此判断服务是否支持 HTTP/2;- 参数、CA 文件等配置错误应先修正,再开展测量。
教学程序应主动说明这些边界,而不是因为“运行成功”就把它包装成生产级探针。
关于端口扫描
nmap 对资产清点和故障核查很有用,也可能触发告警、违反授权范围或给脆弱服务造成压力。只扫描明确归你管理或书面授权的目标,并约定源地址、目标网段、端口范围、时间窗和速率。
即使扫描结果写着 open、closed 或 filtered,它也是基于探测响应的分类,不是应用健康证明。生产服务还需用协议级请求、身份校验和服务端证据确认。
验收清单
- [ ] 能把 connect timeout 与 response timeout 分开;
- [ ] 能用
ss证明本地监听地址,而不是笼统说“端口开了”; - [ ] 能从本地 pcap 指出 SYN、SYN-ACK、ACK、RST/FIN;
- [ ] 能说明一次抓包的接口、namespace、时间窗和过滤器;
- [ ] 能解释 curl 各时间点是累计值;
- [ ] 能为每个结论写出至少一个尚未排除的替代解释。
本章总结
第 12 章建立了一条完整诊断链:先定义症状与阶段,再用名称解析、路由、传输、TLS 和 HTTP 证据逐层收敛;需要深入时,在正确观测点做有界抓包;最后用本地实验验证自己是否真的能区分错误阶段。
下一章进入 IPv6。届时同样的诊断思路仍然适用,但地址选择、邻居发现、ICMPv6 和双栈回退会带来新的变量。