Python3 简易DNS服务器实现
使用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, BaseResolverclass 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.qnameqtype = 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 replydef 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 jsonclass 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.qnameqtype = 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
- macOS:
-
性能考虑:这个实现是单线程的,对于高负载环境,可以考虑使用多线程或异步IO。
-
安全性:生产环境应考虑添加DNS查询限制、防止DNS放大攻击等安全措施。
这个实现提供了基本的DNS服务器功能,你可以根据需要扩展更多记录类型(MX, NS, SOA等)或添加更复杂的逻辑。