Python3简易DNS服务器实现方式

pip install dnslib

#!/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()

sudo python3 dns_server.py

# 查询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

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))
                    )

{
    "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"}
    ]
}

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