Python3 简易DNS服务器实现

news2025/5/19 1:25:23

使用Python3开发一个简单的DNS服务器，支持配置资源记录(RR)，并能通过dig命令进行查询。

让自己理解DNS原理

实现方案

我们将使用socketserver和dnslib库来构建这个DNS服务器。dnslib库能帮助我们处理DNS协议的复杂细节。

1. 安装依赖

首先确保安装了dnslib库：

pip install dnslib

2. DNS服务器实现代码

#!/usr/bin/env python3
import socketserver
from dnslib import DNSRecord, DNSHeader, QTYPE, RR, A, AAAA, CNAME, TXT
from dnslib.server import DNSServer, BaseResolver

class SimpleResolver(BaseResolver):
    """
    自定义DNS解析器，包含我们配置的资源记录
    """
    def __init__(self):
        # 初始化资源记录字典
        self.records = {
            # A记录 (域名 -> IPv4)
            'example.com.': [
                RR('example.com.', QTYPE.A, rdata=A('93.184.216.34'), ttl=60),
                RR('example.com.', QTYPE.A, rdata=A('93.184.216.35'), ttl=60)
            ],
            # AAAA记录 (域名 -> IPv6)
            'ipv6.example.com.': [
                RR('ipv6.example.com.', QTYPE.AAAA, rdata=AAAA('2606:2800:220:1:248:1893:25c8:1946'), ttl=60)
            ],
            # CNAME记录 (别名)
            'www.example.com.': [
                RR('www.example.com.', QTYPE.CNAME, rdata=CNAME('example.com.'), ttl=60)
            ],
            # TXT记录
            'txt.example.com.': [
                RR('txt.example.com.', QTYPE.TXT, rdata=TXT('This is a test TXT record'), ttl=60)
            ]
        }
  
    def resolve(self, request, handler):
        """
        处理DNS查询请求
        """
        reply = request.reply()
        qname = request.q.qname
        qtype = request.q.qtype
      
        # 记录查询日志
        print(f"Received query: {qname} (Type: {QTYPE[qtype]})")
      
        # 检查是否有匹配的记录
        if str(qname) in self.records:
            for record in self.records[str(qname)]:
                if record.rtype == qtype or qtype == QTYPE.ANY:
                    reply.add_answer(record)
      
        return reply

def main():
    """
    启动DNS服务器
    """
    resolver = SimpleResolver()
  
    # 创建DNS服务器，监听UDP 53端口
    server = DNSServer(
        resolver,
        port=53,
        address="0.0.0.0",  # 监听所有接口
        tcp=False           # 仅使用UDP
    )
  
    print("Starting DNS server on port 53...")
    try:
        server.start()
    except KeyboardInterrupt:
        server.stop()
        print("\nDNS server stopped")

if __name__ == '__main__':
    main()

3. 运行服务器

由于DNS服务需要使用53端口，在Linux/macOS上需要以root权限运行：

sudo python3 dns_server.py

4. 使用dig测试

打开另一个终端窗口，使用dig命令测试：

# 查询A记录
dig @127.0.0.1 example.com

# 输出中有
# example.com.            60      IN      A       93.184.216.34
# example.com.            60      IN      A       93.184.216.35


# 查询AAAA记录
dig @127.0.0.1 -t AAAA ipv6.example.com

# 查询CNAME记录
dig @127.0.0.1 -t CNAME www.example.com

# 查询TXT记录
dig @127.0.0.1 -t TXT txt.example.com

进阶功能

1. 从配置文件加载资源记录

我们可以改进代码，从JSON文件加载资源记录：

import json

class ConfigurableResolver(BaseResolver):
    def __init__(self, config_file='dns_config.json'):
        self.records = {}
        self.load_config(config_file)
  
    def load_config(self, config_file):
        with open(config_file) as f:
            config = json.load(f)
      
        for domain, records in config.items():
            self.records[domain] = []
            for record in records:
                rtype = record['type'].upper()
                if rtype == 'A':
                    self.records[domain].append(
                        RR(domain, QTYPE.A, rdata=A(record['value']), ttl=record.get('ttl', 60))
                    )
                elif rtype == 'AAAA':
                    self.records[domain].append(
                        RR(domain, QTYPE.AAAA, rdata=AAAA(record['value']), ttl=record.get('ttl', 60))
                    )
                elif rtype == 'CNAME':
                    self.records[domain].append(
                        RR(domain, QTYPE.CNAME, rdata=CNAME(record['value']), ttl=record.get('ttl', 60))
                    )
                elif rtype == 'TXT':
                    self.records[domain].append(
                        RR(domain, QTYPE.TXT, rdata=TXT(record['value']), ttl=record.get('ttl', 60))
                    )

示例配置文件dns_config.json:

{
    "example.com.": [
        {"type": "A", "value": "93.184.216.34", "ttl": 300},
        {"type": "A", "value": "93.184.216.35", "ttl": 300}
    ],
    "ipv6.example.com.": [
        {"type": "AAAA", "value": "2606:2800:220:1:248:1893:25c8:1946"}
    ],
    "www.example.com.": [
        {"type": "CNAME", "value": "example.com."}
    ],
    "txt.example.com.": [
        {"type": "TXT", "value": "This is a test TXT record"}
    ]
}

2. 添加日志记录

我们可以添加更详细的日志记录：

import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('dns_server.log'),
        logging.StreamHandler()
    ]
)

class LoggingResolver(BaseResolver):
    def resolve(self, request, handler):
        client_ip = handler.client_address[0]
        qname = request.q.qname
        qtype = QTYPE[request.q.qtype]
      
        logging.info(f"Query from {client_ip}: {qname} (Type: {qtype})")
      
        reply = super().resolve(request, handler)
      
        if reply.rr:
            for answer in reply.rr:
                logging.info(f"Responded with: {answer}")
        else:
            logging.warning(f"No records found for {qname} (Type: {qtype})")
      
        return reply

注意事项

端口权限：DNS服务器需要使用53端口，在Unix-like系统上需要root权限。
防火墙设置：确保防火墙允许UDP 53端口的传入连接。
系统DNS缓存：测试时可能需要清除本地DNS缓存：
- macOS: sudo killall -HUP mDNSResponder
- Linux: 取决于发行版，可能是systemd-resolve --flush-caches
性能考虑：这个实现是单线程的，对于高负载环境，可以考虑使用多线程或异步IO。
安全性：生产环境应考虑添加DNS查询限制、防止DNS放大攻击等安全措施。