www.baidu.net手写实现3步搞定解析避坑

盯着屏幕上那串红色的 StackTrace 报错，是不是脑子瞬间一片空白？Connection refused、DNS resolution failed 还有那些看不懂的十六进制地址，堆在一起像天书一样。很多新手以为这是网络断了，其实多半是你对底层解析机制的误解。别慌，今天我们不背八股文，直接手写实现一个极简版的 DNS 解析器，把 www.baidu.net 的查询过程彻底拆开了看。只有懂了底层，报错时你才知道该查哪一行。

项目目标

我们不是要造轮子去替代系统自带的 DNS，而是为了理解 HTTP 请求背后那个看不见的“电话簿”。当你在浏览器输入 www.baidu.net 时，电脑并不知道这个域名对应哪个 IP。它需要向 DNS 服务器发起询问。这个过程遵循严格的协议标准，参考 RFC 1035 规范，DNS 报文有着固定的二进制结构。

本项目目标很明确：用 Python 从零构建一个 UDP 客户端，手动构造 DNS 查询包，发送给公共 DNS 服务器（如 8.8.8.8），然后解析返回的二进制响应包，提取出最终的 IP 地址。通过这个过程，你将亲眼看到数据是如何被打包、传输和解包的，彻底搞懂为什么有时候会报 Timeout，有时候又是 SERVFAIL。

目录结构

为了保持代码清晰可维护，我们将项目拆分为三个核心文件。这种结构在后续扩展支持 TCP 或 EDNS0 时也非常方便。

dns_scratch/
├── packet.py       # 负责 DNS 报文的构建与解析
├── client.py       # 负责 UDP 网络通信与主流程控制
├── main.py         # 入口文件，执行具体查询
└── requirements.txt # 依赖管理（本例几乎无依赖，仅用标准库）

packet.py 是核心中的核心，所有关于位操作、字节序转换的逻辑都在这。client.py 则处理网络 I/O，确保我们在不同操作系统下的兼容性。

核心代码实现

1. 构建查询报文

DNS 报文头包含 12 个字节的固定头部。我们需要手动填充事务 ID（TXID）、标志位、查询数量等字段。这里我们使用 Python 的 struct 模块来处理字节序，因为网络传输默认是大端序（Big-Endian），而 CPU 通常是小端序。

import struct
import randomdef build_dns_query(domain: str) -> bytes:"""构建 DNS 查询报文:param domain: 待解析的域名，如 www.baidu.net:return: 打包后的 bytes 对象"""# 1. 生成随机事务ID，用于匹配请求与响应，防止中间人攻击或乱序txid = random.randint(0, 65535)# 2. 构建头部 (12 bytes)# TXID: 2 bytes# Flags: 2 bytes (Recursion Desired = 1, Standard Query = 0x0100)# QDCOUNT: 1 (Query Count)# ANCOUNT, NSCOUNT, ARCOUNT: 0header = struct.pack(">HHHHHH", txid, 0x0100, 1, 0, 0, 0)# 3. 构建域名部分# DNS 域名是以标签分隔的，每个标签前有一个长度字节# 例如: www.baidu.net -> b'\x03www\x05baidu\x03net\x00'labels = domain.split('.')domain_bytes = b''for label in labels:encoded = label.encode('ascii')if len(encoded) > 63:raise ValueError("Label too long")domain_bytes += struct.pack("B", len(encoded)) + encodeddomain_bytes += b'\x00' # 根域结束符# 4. 构建查询类型和类# Type: A (0x01), Class: IN (0x01)qtype_qclass = struct.pack(">HH", 0x01, 0x01)return header + domain_bytes + qtype_qclass

注意 struct.pack 中的 > 符号，它代表网络字节序。如果你漏掉这个，解析出来的数字会完全错位，导致后续所有逻辑崩溃。

2. 解析响应报文

响应包比查询包复杂得多，因为答案区域（Answer Section）可能包含多个记录，且可能使用指针压缩（Pointer Compression）来节省空间。但在简单场景中，我们主要关注前 12 字节的头部和紧随其后的答案部分。

def parse_dns_response(data: bytes) -> dict:"""解析 DNS 响应报文:param data: 接收到的原始 bytes:return: 包含解析结果的字典"""if len(data) < 12:raise ValueError("Invalid DNS packet length")# 解析头部txid, flags, qdcount, ancount, nscount, arcount = struct.unpack(">HHHHHH", data[:12])# 简单起见，这里假设没有压缩指针，直接从第13字节开始解析# 实际生产环境中需要处理 0xC0 开头的指针offset = 12# 跳过查询域名部分 (这里简化处理，假设域名结构已知或重新解析)# 为了代码简洁，我们直接从 ancount 个答案记录中解析# 注意：真实情况需要先跳过 Question 部分answers = []# 遍历答案记录for _ in range(ancount):# 解析 Name (简化：假设直接是长度+字符，无指针)# 在实际代码中，这里需要复杂的递归解析逻辑# 此处为演示核心逻辑，假设 Name 占位或简单解析# 为了演示，我们假设 Name 部分已被跳过，直接读取后续字段# 这是一个简化的解析逻辑，实际需完整实现 Name 解析# 读取 Type, Class, TTL, RDLENGTH# 注意：在实际完整实现中，offset 需要准确指向答案记录的起始# 这里为了演示，我们假设 offset 已经指向了第一个答案的 Name 之后# 为了代码可读性，我们直接硬编码偏移量演示（不推荐生产使用）# 真实场景：需先解析 Question 中的域名长度pass # 以下是一个更完整的解析骨架，包含 Name 解析def parse_name(data, offset):labels = []while offset < len(data):length = data[offset]if length == 0:offset += 1breakelif (length & 0xC0) == 0xC0:# 指针压缩，这里简化处理，实际需递归pointer = struct.unpack(">H", data[offset:offset+2])[0]# 递归解析指针指向的位置labels.extend(parse_name(data, pointer))offset += 2breakelse:labels.append(data[offset+1:offset+1+length].decode('ascii'))offset += 1 + lengthreturn '.'.join(labels), offset# 重新定位到答案区开始# 1. 跳过 Question 区offset = 12_, offset = parse_name(data, offset)offset += 4 # 跳过 QType 和 QClass# 2. 解析 Answer 区for _ in range(ancount):name, offset = parse_name(data, offset)rtype, rclass, ttl, rdlength = struct.unpack(">HHIH", data[offset:offset+10])offset += 10rdata = data[offset:offset+rdlength]offset += rdlengthif rtype == 1: # A Recordip_address = '.'.join(str(b) for b in rdata)answers.append(ip_address)return {"txid": txid,"flags": flags,"answers": answers}

这段代码中的 parse_name 函数是关键难点。DNS 报文为了节省空间，允许使用 16 位指针来引用之前出现过的域名后缀。如果你的解析器不处理 0xC0 开头的指针，遇到压缩域名时会直接读取错误的数据，导致解析出乱码或崩溃。

运行与测试

我们将所有逻辑串联起来，通过 UDP 发送请求并接收响应。

import socket
import timedef resolve_domain(domain: str) -> list:"""主函数：解析域名"""# 1. 构建查询包query_packet = build_dns_query(domain)# 2. 创建 UDP Socketsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)sock.settimeout(5) # 设置超时，防止无限等待# 3. 发送请求到公共 DNS (8.8.8.8)dns_server = ("8.8.8.8", 53)try:sock.sendto(query_packet, dns_server)# 4. 接收响应start_time = time.time()data, addr = sock.recvfrom(4096)end_time = time.time()# 5. 解析响应result = parse_dns_response(data)# 校验事务 ID 是否匹配，防止响应错乱# 注意：需要传递 txid 进行比对，此处简化print(f"Resolved {domain} in {end_time - start_time:.4f}s")print(f"IP Addresses: {result['answers']}")return result['answers']except socket.timeout:print("Request timed out")return []except Exception as e:print(f"Error: {e}")return []finally:sock.close()if __name__ == "__main__":# 测试解析 www.baidu.netips = resolve_domain("www.baidu.net")if ips:print(f"Success: {ips}")else:print("Failed to resolve")

运行 main.py，你应该能看到类似以下的输出：

Resolved www.baidu.net in 0.0125s
IP Addresses: ['110.242.68.3', '110.242.68.66']
Success: ['110.242.68.3', '110.242.68.66']

如果在某些网络环境下运行失败，请检查你的防火墙是否阻断了 UDP 53 端口的出站流量。这是新手最常遇到的“假性报错”，代码逻辑没问题，但网络层被拦截了。

优化扩展

基础版本跑通后，我们可以从以下几个方向进阶：

支持递归查询与迭代查询：目前的代码假设直接问根服务器或权威服务器。实际中，本地 DNS 会递归查询。我们可以修改 Flags 位，观察不同 DNS 服务器的响应差异。
处理 EDNS0 (Extension Mechanisms for DNS)：RFC 6891 引入了 EDNS0，允许在 OPT 记录中携带额外信息，如更大的 UDP 包大小。随着 HTTPS 证书链变长，标准 512 字节限制经常不够用，支持 EDNS0 是必须的。
并发解析：对于需要解析多个域名的场景，使用 asyncio 或 threading 可以显著提升效率。UDP 是无连接的，天然适合并发。
缓存机制：添加一个简单的内存缓存（TTL 过期自动清除），避免重复查询相同域名，减少网络开销。

在优化时，务必注意 RFC 1035 中关于 TTL 的定义。TTL 是秒数，解析器必须严格遵守，不能随意延长缓存时间，否则会导致 IP 变更后用户仍访问旧服务器，造成故障。