3. DNS查询报文构造与解析

任务目标

构造一个DNS标准查询报文（A记录），通过UDP发送至指定DNS服务器，并解析返回的响应报文，从中提取目标域名的IPv4地址。本案例聚焦于DNS报文格式的二进制封装与解析，不涉及DNS缓存、EDNS扩展或TCP传输。

环境依赖

Python 3.8+，仅使用标准库 socket、struct、random。无需第三方库。

核心实现

import socket
import struct
import random

def build_query(domain):
    """构造DNS查询报文（A记录，递归请求）"""
    txid = random.randint(0, 0xFFFF)
    header = struct.pack('>HHHHHH', txid, 0x0100, 1, 0, 0, 0)
    
    # QNAME：将域名转为长度前缀格式
    qname = b''
    for label in domain.split('.'):
        qname += bytes([len(label)]) + label.encode('ascii')
    qname += b'\x00'
    
    # QTYPE=A(1), QCLASS=IN(1)
    question = qname + struct.pack('>HH', 1, 1)
    return header + question, txid

def parse_response(data, txid):
    """解析DNS响应，返回IP地址列表"""
    if len(data) < 12:
        raise ValueError('报文过短')
    
    resp_txid, flags, qdcount, ancount, _, _ = struct.unpack('>HHHHHH', data[:12])
    if resp_txid != txid:
        raise ValueError('事务ID不匹配')
    if flags & 0x8000 == 0:
        raise ValueError('非响应报文')
    if flags & 0x000F != 0:
        raise ValueError(f'DNS错误码: {flags & 0x000F}')
    
    offset = 12
    # 跳过问题部分
    for _ in range(qdcount):
        while data[offset] != 0:
            offset += data[offset] + 1
        offset += 5  # 结束符(1) + QTYPE(2) + QCLASS(2)
    
    # 解析回答部分
    ips = []
    for _ in range(ancount):
        # 跳过NAME（可能是指针）
        if data[offset] & 0xC0 == 0xC0:
            offset += 2
        else:
            while data[offset] != 0:
                offset += data[offset] + 1
            offset += 1
        
        rtype, rclass, ttl, rdlength = struct.unpack('>HHIH', data[offset:offset+10])
        offset += 10
        
        if rtype == 1 and rclass == 1 and rdlength == 4:  # A记录
            ip = socket.inet_ntoa(data[offset:offset+4])
            ips.append(ip)
        offset += rdlength
    
    return ips

def resolve(domain, dns_server='8.8.8.8', timeout=3):
    """发送DNS查询并返回IP列表"""
    query, txid = build_query(domain)
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(timeout)
    try:
        sock.sendto(query, (dns_server, 53))
        response, _ = sock.recvfrom(512)
    finally:
        sock.close()
    
    return parse_response(response, txid)

if __name__ == '__main__':
    domain = 'www.example.com'
    ips = resolve(domain)
    print(f'{domain} -> {ips}')

运行与验证

1. 将上述代码保存为 dns_query.py。
2. 在终端执行：python3 dns_query.py
3. 预期输出（IP可能因DNS服务器不同而变化）：www.example.com -> ['93.184.216.34']

验证判定标准：程序能打印出目标域名对应的合法IPv4地址，且无异常抛出。可通过修改代码末尾的 domain 变量为其他公网域名（如 www.baidu.com）再次运行，均能返回非空IP列表，即视为通过。

常见问题

• 超时无响应：检查本机网络连通性，或更换 dns_server 为 114.114.114.114 等公共DNS。
• 解析CNAME：本实现仅提取A记录，若响应中目标域名为CNAME别名，需递归解析回答部分中的A记录（当前代码已支持跳过CNAME并继续读取后续A记录）。
• 报文截断：UDP响应超过512字节时可能被截断，本案例面向常规小型查询，如需完整处理需实现TCP回退，不在当前目标范围内。