快捷導(dǎo)航

Python3簡(jiǎn)易DNS服務(wù)器實(shí)現(xiàn)方式

更新時(shí)間：2025年07月05日 14:32:56 作者：言之。

本文介紹使用Python3和dnslib庫(kù)開(kāi)發(fā)簡(jiǎn)單DNS服務(wù)器的實(shí)現(xiàn)方案,支持配置資源記錄及dig查詢,涵蓋JSON配置、日志記錄、端口權(quán)限與安全注意事項(xiàng),并建議擴(kuò)展記錄類型以提升功能

實(shí)現(xiàn)方案

我們將使用socketserver和dnslib庫(kù)來(lái)構(gòu)建這個(gè)DNS服務(wù)器。dnslib庫(kù)能幫助我們處理DNS協(xié)議的復(fù)雜細(xì)節(jié)。

1. 安裝依賴

首先確保安裝了dnslib庫(kù)：

pip install dnslib

2. DNS服務(wù)器實(shí)現(xiàn)代碼

#!/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查詢請(qǐng)求
        """
        reply = request.reply()
        qname = request.q.qname
        qtype = request.q.qtype
      
        # 記錄查詢?nèi)罩?
        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():
    """
    啟動(dòng)DNS服務(wù)器
    """
    resolver = SimpleResolver()
  
    # 創(chuàng)建DNS服務(wù)器，監(jiān)聽(tīng)UDP 53端口
    server = DNSServer(
        resolver,
        port=53,
        address="0.0.0.0",  # 監(jiān)聽(tīng)所有接口
        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. 運(yùn)行服務(wù)器

由于DNS服務(wù)需要使用53端口，在Linux/macOS上需要以root權(quán)限運(yùn)行：

sudo python3 dns_server.py

4. 使用dig測(cè)試

打開(kāi)另一個(gè)終端窗口，使用dig命令測(cè)試：

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

進(jìn)階功能

1. 從配置文件加載資源記錄

我們可以改進(jìn)代碼，從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. 添加日志記錄

我們可以添加更詳細(xì)的日志記錄：

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