import threading import socket import socketserver class Handler(socketserver.BaseRequestHandler): def handle(self): request_data = self.request[0] # 将请求转发到 114 DNS redirect_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) redirect_socket.sendto(request_data, ('114.114.114.114', 53)) response_data, address = redirect_socket.recvfrom(1024) # 将114响应响应给客户 client_socket = self.request[1] client_socket.sendto(response_data, self.client_address) class Server(socketserver.ThreadingMixIn, socketserver.UDPServer): : # 一下ip需换成自己电脑的ip server = Server(('172.16.42.254', 53), Handler) with server: server_thread = threading.Thread(target=server.serve_forever) server_thread.daemon = True server_thread.start() print('The DNS server is running at 172.16.42.254...') server_thread.join()

现在我们的DNS服务器就可以进行转发工作了。运行以上程序（需root权限），然后用nsloop命令，向我们的服务器发送DNS请求，一切OK：

$ nslookup baidu.com 172.16.42.254 Server: 172.16.42.254 Address: 172.16.42.254#53 Non-authoritative answer: Name: baidu.com Address: 123.125.114.144 Name: baidu.com Address: 180.149.132.47 Name: baidu.com Address: 111.13.101.208 Name: baidu.com Address: 220.181.57.217 4.2 缓存功能

如果仅仅做一下代理转发，那也太无聊了。现在我们再加上缓存功能，即它能够将其他DNS服务器的响应结果缓存起来。当收到请求时，若请求主机号在缓存中，且没有过期，则直接响应缓存结果；否则进行上一功能中的操作。这一功能的关键在于对DNS消息的解析，代码如下：

class Message: (, authority): self.header = header self.question = question self.answer = answer self.authority = authority self.additional = additional @classmethod def from_bytes(cls, data): scanner = Scanner(data) # 读取header header = dict() header['ID'] = scanner.next_bytes(2) header['QR'] = scanner.next_bits(1) header['OPCODE'] = scanner.next_bits(4) header['AA'] = scanner.next_bits(1) header['TC'] = scanner.next_bits(1) header['RD'] = scanner.next_bits(1) header['RA'] = scanner.next_bits(1) header['Z'] = scanner.next_bits(3) header['RCODE'] = scanner.next_bits(4) header['QDCOUNT'] = scanner.next_bytes(2) header['ANCOUNT'] = scanner.next_bytes(2) header['NSCOUNT'] = scanner.next_bytes(2) header['ARCOUNT'] = scanner.next_bytes(2) print('header:', header) # 读取question questions = list() for _ in range(header['QDCOUNT']): question = dict() question['QNAME'] = scanner.next_bytes_until(lambda current, _: current == 0) scanner.next_bytes(1) # 跳过0 question['QTYPE'] = scanner.next_bytes(2) question['QCLASS'] = scanner.next_bytes(2) questions.append(question) print('questions:', questions) message = Message(header) # 读取answer、authority、additional rrs = list() for i in range(header['ANCOUNT'] + header['NSCOUNT'] + header['ARCOUNT']): rr = dict() rr['NAME'] = cls.handle_compression(scanner) rr['TYPE'] = scanner.next_bytes(2) rr['CLASS'] = scanner.next_bytes(2) rr['TTL'] = scanner.next_bytes(4) rr['RDLENGTH'] = scanner.next_bytes(2) : # A记录 r_data = scanner.next_bytes(rr['RDLENGTH'], False) rr[) y, map(lambda num: str(num), r_data)) rr['TYPE'] == 5: # NS与CNAME记录 rr['RDATA'] = cls.handle_compression(scanner, rr['RDLENGTH']) rrs.append(rr) answer, authority, additional = list(), list(), list() for i, rr in enumerate(rrs): if i < header['ANCOUNT']: answer.append(rr) elif i < header['ANCOUNT'] + header['NSCOUNT']: authority.append(rr) else: additional.append(rr) print('answer:', answer) print('authority:', authority) print('additional:', additional) return message ("inf")): byte = scanner.next_bytes() : # a pointer pointer ) + scanner.next_bytes() return cls.handle_compression(Scanner(scanner.data, pointer)) data current offset > length) : # a sequence of labels ending in a zero octet scanner.next_bytes() return data # a sequence of labels ending with a pointer result cls.handle_compression(Scanner(scanner.data, *scanner.position())) scanner.next_bytes(2) # 跳过2个字节的指针 return result

其中用到了一个自定义的Scanner类，用来帮助我们从bytes中按字节或位读取数据，其定义如下：