Python で書かれた簡単な DNS サーバーの例

@author: RobinTang
'''

import socketserver
import struct

# DNS Query
class SinDNSQuery:
    def __init__(self, data):
        i = 1
        self.name = ''
        while True:
            d = data[i]
            if d == 0:
                break;
            if d < 32:
self.name = self.name + '.'
else:
self.name = self.name + chr(d)
i = i + 1
self.querybytes = data[0:i + 1]
(self.type, self.classify) = struct.unpack('>HH', data[i + 1:i + 5])
        self.len = i + 5
    def getbytes(self):
        return self.querybytes + struct.pack('>HH', self.type, self.classify)

# DNS Answer RRS
# this class is also can be use as Authority RRS or Additional RRS
class SinDNSAnswer:
    def __init__(self, ip):
        self.name = 49164
        self.type = 1
        self.classify = 1
        self.timetolive = 190
        self.datalength = 4
        self.ip = ip
    def getbytes(self):
        res = struct.pack('>HHHLH', self.name, self.type, self.classify, self.timetolive, self.datalength)
        s = self.ip.split('.')
        res = res + struct.pack('BBBB', int(s[0]), int(s[1]), int(s[2]), int(s[3]))
        return res

# DNS フレーム
# DNS クエリ フレームによって初期化する必要があります
class SinDNSFrame:
def __init__(self, data):
(self.id, self.flags, self.quests, self.answers, self.author, self.addition) = struct.unpack('>HHHHHH', data[0:12])
self.query = SinDNSQuery(data[12:])
def getname (self):
return self.query.name
def setip(self, ip):
self.answer = SinDNSAswer(ip)
self.answers = 1
self.flags = 33152
def getbytes(self):
res = struct.pack('>HHHHHH', self.id, self.flags, self.quests, self.answers, self.author, self.addition)
res = res + self.query.getbytes()
if self.answers != 0:
res = res + self.answer.getbytes()
return res
# UDPHandler を返すDNS クエリを処理します
class SinDNSUDPHandler(socketserver.BaseRequestHandler):
def handle(self):
data = self.request[0].strip()
dns = SinDNSFrame(data)
socket = self.request[1]
namemap = SinDNSServer.namemap
if(dns.query.type==1):
# これが A レコードのクエリである場合、応答します

name = dns.getname();
if namemap.__contains__(name):
# レコードがある場合は応答します
dns.setip(namemap[name])
ソケット.sendto( dns.getbytes(), self.client_address)
elif namemap.__contains__('*'):
# 応答のデフォルトアドレス
dns.setip(namemap['*'])
ソケット.sendto (dns.getbytes(), self.client_address)
else:
# 無視します
ソケット.sendto(data, self.client_address)
else:
# これがクエリでない場合は、レコード、無視します
ソケット.sendto(data, self.client_address)

# DNS サーバー
# レコード クエリのみをサポートします
# ユーザーは、単純な DNS サーバーを作成できます
class SinDNSServer:
def __init__(self, port=53):
SinDNSServer.namemap = {}
self.port = port
def addname(self, name, ip):
SinDNSServer.namemap[name] = ip
def start(self):
HOST, PORT = "0.0.0.0", self.port
server =ソケットサーバー.UDPServer((HOST, PORT), SinDNSUDPHandler)
server.serve_forever()

# さて、テストします
if __name__ == "__main__":
sev = SinDNSServer()
sev.addname('www.aa.com', '192.168.0.1') # add a A レコード
sev.addname('www.bb.com', '192.168.0.2') # A レコードを追加
sev.addname('*', '0.0.0.0') # デフォルトのアドレス
sev.start() # DNS サーバーを開始します

# これで、「nslookup」コマンドを使用してテストできます
# 「nslookup www.aa.com」など