跳到内容

第 4 卷 / 第 12 章 / 第 3 课

12.3 可重复的网络诊断实验 ​

这一课只在本机回环接口上制造故障。目标不是记住某台机器的输出,而是练习如何定义阶段、设置 deadline、保留证据并写出有限结论。

实验边界 ​

  • 只使用 127.0.0.1 和本机临时端口;
  • 不扫描公网或未经授权的主机;
  • 不修改防火墙、路由或系统 DNS;
  • 不使用 curl -k 掩盖证书错误;
  • 每个后台进程都记录 PID 并在结束时停止。

不同系统命令选项可能不同。下文使用 Linux、bash、GNU timeout 与 Python 3.11+。先打开专用 bash 会话,在新建的临时目录内保存实验脚本和日志;不要在日常工作目录直接启动文件服务。启动失败时,代码会退出这个会话,退出钩子会停止它创建的服务。实验结束后正常执行清理命令。

实验一:区分拒绝、连接成功与响应超时 ​

创建 stall_server.py:

python
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)

运行它并读取操作系统分配的端口:

bash
: >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

先验证监听:

bash
ss -lnt "sport = :$server_port"

再请求这个“接受连接但不返回 HTTP”的服务:

bash
curl --noproxy '*' --verbose --connect-timeout 1 --max-time 2 \
  "http://127.0.0.1:$server_port/"

预期:TCP 很快连接成功,HTTP 响应在总 deadline 内没有到达,curl 返回 28。这里的 timeout 发生在连接之后。--noproxy '*' 让请求直接走回环接口,避免代理环境变量改变实验路径。

停止服务,再请求同一端口:

bash
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 在回环接口采集,只抓该临时端口:

bash
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 没有报告采集错误。读取摘要:

bash
tcpdump -nn -r loopback-lab.pcap

回答这些问题:

  1. 两次请求是否都发出了 SYN?
  2. 哪一次完成了三次握手?
  3. 拒绝场景中是否出现 RST?
  4. 响应超时场景中,客户端是否发送了 HTTP 请求字节?
  5. FIN 或 RST 由哪一端发出?这与 curl 的 deadline 有何关系?

回环接口上的包形态和普通以太网接口不完全相同,但 TCP 状态转换仍可用于本实验。

实验三:测量一个本地 HTTP 请求 ​

把下面程序保存为 http_server.py。它只返回固定文本,不会把当前目录变成文件服务器:

python
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()

启动服务并读取端口:

bash
: >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

执行五次请求:

bash
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 不参与。即使如此,首次请求、线程创建、调度和服务日志也会带来波动。不要用五个样本宣称某项优化普遍有效;本实验只训练测量字段和阶段解释。

结束时清理:

bash
kill "$http_pid"
wait "$http_pid" 2>/dev/null || true
trap - EXIT

TCP 与 TLS 共用 deadline 的 Python 诊断器 ​

下面程序先单独测量 DNS,再让 TCP 连接与 TLS 握手共享一个剩余预算。上一阶段花掉的时间不会在下一阶段“充值”。TLS 继续校验证书链和主机名;可选 cafile 只用于显式信任测试 CA,不会关闭验证。异常携带失败阶段和已完成的测量。保存为 diagnose.py,显式传入目标;本地测试应使用证书包含相应 IP 或主机名的临时 TLS 服务。

python
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 和双栈回退会带来新的变量。

实验参考 ​

Built with VitePress | Software Systems Atlas