如何自己动手实现一个简易DNS服务器？

域名系统（DNS）是互联网最基础的服务之一，它把人类易记的域名转换成IP地址。虽然大多数开发者每天都在间接使用DNS，但很少有人真正动手实现过DNS服务器。本文的目标不是做一个生产级服务器，而是通过最小可用版本，理解DNS协议报文格式、UDP通信和解析转发流程。我们会使用Python标准库中的socket和struct，不需要安装其他依赖。实现后的程序将监听UDP 53端口，能够回答指定域名的A记录查询，并支持将无法回答的查询转发到上游公共DNS。

一、DNS报文结构与编码方式

DNS查询和响应在UDP上传输时使用同一个报文结构。每个DNS报文由五部分组成：头部、问题区、应答区、授权区和附加区。头部固定12字节，后四个区长度可变。头部包含一个16位标识符，用于匹配请求和响应；一个标志字段，用来区分查询或响应、是否递归等；后面是四个计数字段，分别表示问题、应答、授权和附加区的记录数量。对简易服务器来说，最关键的是标识符、标志位和问题计数。

问题区用来描述要查询的域名和类型。域名不是直接以点分字符串存储，而是按标签编码：每个标签先写长度，再写标签内容，最后以0字节结束。例如ipipp.com会被编码成05 69 70 69 70 70 03 63 6f 6d 00。查询类型通常为A记录，对应整数值1，表示希望获得IPv4地址。查询类固定为IN，对应整数值1。理解了这些字段后，就可以用Python的struct模块来解析和构造报文。

下面代码展示了如何把点分域名转换为DNS协议中的标签编码，以及如何从标签编码还原域名。编码时需要处理根域名和结尾0字节，解码时要跳过压缩指针，但我们在最小实现中暂不处理报文中的压缩指针，只解析问题区不涉及压缩的情况。

import struct
import socket

def encode_domain(domain):
    result = b''
    for label in domain.rstrip('.').split('.'):
        result += bytes([len(label)]) + label.encode()
    result += b'\x00'
    return result

def decode_domain(data, offset):
    labels = []
    while True:
        length = data[offset]
        offset += 1
        if length == 0:
            break
        labels.append(data[offset:offset+length].decode())
        offset += length
    return '.'.join(labels), offset

二、用UDP Socket搭建查询响应服务

DNS服务器通常使用UDP 53端口监听请求。UDP无连接，每个请求都在独立数据报中处理，很适合简单的请求响应模型。Python的socket模块可以快速创建UDP套接字，绑定到0.0.0.0:53后进入循环，每次recvfrom收到数据和客户端地址，解析问题区，再构造响应包发送回去。由于53端口是特权端口，在Linux和macOS下通常需要root权限，测试时也可以先绑定到5353端口避免权限问题。

构造响应包时，先复制请求中的标识符到头部，把标志位设置成0x8180，这是一个标准响应：QR为1表示响应，RD和RA为1表示支持递归。问题计数保持不变，应答计数设置为1。然后原样复制请求中的问题区，再在应答区添加一条A记录。A记录包含域名指针、类型、类、TTL、数据长度和IPv4地址。域名指针通常在响应中使用0xC00C这样的压缩指针，指向问题区中的域名位置，但本文为简化，可以直接把问题区的域名再次编码一遍放进应答区，虽然增大了一点包体积，但不影响功能。

def build_response(request_data, ip):
    transaction_id = request_data[:2]
    flags = b'\x81\x80'
    qdcount = b'\x00\x01'
    ancount = b'\x00\x01'
    nscount = b'\x00\x00'
    arcount = b'\x00\x00'
    header = transaction_id + flags + qdcount + ancount + nscount + arcount

    # 问题区原样复制
    question_end = request_data.find(b'\x00', 12) + 5
    question = request_data[12:question_end]

    # 应答区：域名指针、类型A、类IN、TTL、数据长度、IP
    answer = b'\xc0\x0c'
    answer += struct.pack('!HHIH', 1, 1, 300, 4)
    answer += socket.inet_aton(ip)

    return header + question + answer

def start_server(ip='127.0.0.1', port=53):
    server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server.bind(('0.0.0.0', port))
    print(f'DNS server listening on port {port}')
    while True:
        request_data, client_addr = server.recvfrom(512)
        domain, _ = decode_domain(request_data, 12)
        print(f'query: {domain} from {client_addr}')
        response = build_response(request_data, ip)
        server.sendto(response, client_addr)

上述代码中，问题区结束位置通过从偏移12开始查找第一个0字节来确定。找到域名结束的0字节后，再往后跳过4字节的类型和类字段，就得到了问题区的完整结束位置。这个简单假设成立的前提是请求中没有其他压缩指针，对于普通A记录查询足够使用。实际DNS请求还可能包含EDNS0扩展，处理起来会更复杂，但本文先聚焦核心链路。

三、增加上游转发与本地缓存

一个只会返回固定IP的DNS服务器实际用途有限。更合理的行为是：如果本地配置了某些域名的映射则直接返回；否则把请求原封不动转发给上游公共DNS服务器，如8.8.8.8或114.114.114.114，再把上游响应回传给客户端。转发时需要注意，客户端请求是递归查询，上游返回的响应中可能包含压缩指针，回传时不需要修改，直接把UDP数据包透传即可。我们只需创建一个新的UDP套接字发送给上游，等待响应，然后发送给原客户端。

转发会带来延迟和重复查询。可以在内存中维护一个简单字典作为缓存，键为域名和类型，值为解析结果和过期时间。当收到查询时先查缓存，命中则构造响应；未命中则转发并缓存。生产环境还会考虑TTL、缓存淘汰、并发安全等问题，但本文的简易实现只保存IPv4地址和过期时间戳，能让重复查询明显加快。

import time

UPSTREAM = ('8.8.8.8', 53)
cache = {}

def lookup_upstream(request_data):
    upstream_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    upstream_socket.settimeout(3)
    upstream_socket.sendto(request_data, UPSTREAM)
    response, _ = upstream_socket.recvfrom(512)
    upstream_socket.close()
    return response

def extract_ip_from_response(response):
    # 简化处理：跳过头部和问题区，从应答区读取RDATA最后的IPv4地址
    question_end = response.find(b'\x00', 12) + 5
    answer_start = question_end
    # 应答区第一字节通常是指针0xC0，接着是偏移，再跳过类型、类、TTL
    answer_start += 2 + 2 + 2 + 4 + 2
    return socket.inet_ntoa(response[answer_start:answer_start+4])

def handle_query(request_data, client_addr, server_socket):
    domain, _ = decode_domain(request_data, 12)
    key = domain.lower()
    if key in cache:
        expire, ip = cache[key]
        if time.time() < expire:
            print(f'cache hit: {key}')
            server_socket.sendto(build_response(request_data, ip), client_addr)
            return
    response = lookup_upstream(request_data)
    ip = extract_ip_from_response(response)
    cache[key] = (time.time() + 300, ip)
    server_socket.sendto(response, client_addr)

extract_ip_from_response函数做了很多简化假设：它只处理应答区第一条记录，并且假设域名部分占用2字节压缩指针，类型、类、TTL、数据长度字段按标准长度跳过。如果上游响应包含CNAME链或多条A记录，这个函数可能取到错误位置。但对于原理演示已经足够。若要解析更复杂的响应，可以引入压缩指针检测和循环读取资源记录的逻辑。

四、测试、调试与常见问题

完成代码后，可以使用dig命令在Linux或macOS上直接测试。假设程序绑定在5353端口，执行dig @127.0.0.1 -p 5353 ipipp.com A，应该能看到返回的IP地址。若绑定在53端口，则dig @127.0.0.1 ipipp.com A即可。Windows用户可安装nslookup或使用WSL，也可以写一个简单的Python客户端发送构造好的查询。

测试中最常见的问题是端口被占用或权限不足。53端口通常被systemd-resolved、dnsmasq等系统服务占用，可以先停止这些服务或改用5353。如果程序没有响应，请检查防火墙是否放行UDP端口。抓包工具如Wireshark可以观察请求与响应的字节级内容，帮助确认头部字段是否正确。在Linux下也可以使用tcpdump -i lo udp port 5353 -X查看十六进制和ASCII数据。

另一个容易出错的地方是字节序。DNS报文头部的多字节字段全部使用网络字节序，即大端。Python的struct.pack和unpack中首字符使用感叹号可以自动处理平台差异，不要直接使用默认的本机字节序。域名编码时长度值必须是无符号单字节，否则超过127的标签会解析错误，虽然常规域名标签不会超过63字节，但这是协议规范。

最终，这个简易DNS服务器虽然离生产级还差很多，比如未实现TCP模式、DNSSEC、EDNS0和完整压缩指针处理，但已经覆盖了DNS服务器的核心链路：监听UDP、解析问题、查询本地或上游、返回应答。理解这些基础后，再阅读dnspython、CoreDNS或BIND源码会轻松很多。

DNS服务器域名解析Python网络编程修改时间：2026-08-27 09:24:00