Skip to content

Instantly share code, notes, and snippets.

@andreif

andreif/Simple DNS server (UDP and TCP) in Python using dnslib.py

Last active April 9, 2024 07:19
Download ZIP

Revisions

  1. andreif revised this gist Feb 24, 2016. 1 changed file with 3 additions and 0 deletions.
    3 changes: 3 additions & 0 deletions Simple DNS server (UDP and TCP) in Python using dnslib.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,4 +1,7 @@
    # coding=utf-8
    """
    LICENSE http://www.apache.org/licenses/LICENSE-2.0
    """
    import datetime
    import sys
    import time
  2. andreif renamed this gist Jul 24, 2013. 1 changed file with 0 additions and 0 deletions.
    0 Python simple DNS server with dnslib.py → ...r (UDP and TCP) in Python using dnslib.py
    View file
    File renamed without changes.
  3. andreif revised this gist Jul 24, 2013. 1 changed file with 2 additions and 3 deletions.
    5 changes: 2 additions & 3 deletions Python simple DNS server with dnslib.py
    View file
    Original file line number Diff line number Diff line change
    @@ -65,8 +65,7 @@ def dns_response(data):

    reply.add_ns(RR(rname=D, rtype=QTYPE.SOA, rclass=1, ttl=TTL, rdata=soa_record))

    print "---- Reply"
    print reply
    print "---- Reply:\n", reply

    return reply.pack()

    @@ -81,7 +80,7 @@ def send_data(self, data):

    def handle(self):
    now = datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
    print "---- %s request %s (%s %s):" % (self.__class__.__name__[:3], now, self.client_address[0],
    print "\n\n%s request %s (%s %s):" % (self.__class__.__name__[:3], now, self.client_address[0],
    self.client_address[1])
    try:
    data = self.get_data()
  4. andreif revised this gist Jul 24, 2013. 1 changed file with 12 additions and 2 deletions.
    14 changes: 12 additions & 2 deletions Python simple DNS server with dnslib.py
    View file
    Original file line number Diff line number Diff line change
    @@ -94,10 +94,17 @@ def handle(self):
    class TCPRequestHandler(BaseRequestHandler):

    def get_data(self):
    return self.request.recv(8192).strip()[2:]
    data = self.request.recv(8192).strip()
    sz = int(data[:2].encode('hex'), 16)
    if sz < len(data) - 2:
    raise Exception("Wrong size of TCP packet")
    elif sz > len(data) - 2:
    raise Exception("Too big TCP packet")
    return data[2:]

    def send_data(self, data):
    return self.request.sendall(data)
    sz = hex(len(data))[2:].zfill(4).decode('hex')
    return self.request.sendall(sz + data)


    class UDPRequestHandler(BaseRequestHandler):
    @@ -125,6 +132,9 @@ def send_data(self, data):
    try:
    while 1:
    time.sleep(1)
    sys.stderr.flush()
    sys.stdout.flush()

    except KeyboardInterrupt:
    pass
    finally:
  5. andreif revised this gist Jul 24, 2013. 1 changed file with 2 additions and 1 deletion.
    3 changes: 2 additions & 1 deletion Python simple DNS server with dnslib.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,6 +1,7 @@
    # coding=utf-8
    import datetime
    import sys
    import time
    import threading
    import traceback
    import SocketServer
    @@ -123,7 +124,7 @@ def send_data(self, data):

    try:
    while 1:
    pass
    time.sleep(1)
    except KeyboardInterrupt:
    pass
    finally:
  6. andreif revised this gist Jul 24, 2013. 1 changed file with 58 additions and 17 deletions.
    75 changes: 58 additions & 17 deletions Python simple DNS server with dnslib.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,5 +1,9 @@
    # coding=utf-8
    import socket
    import datetime
    import sys
    import threading
    import traceback
    import SocketServer
    from dnslib import *


    @@ -11,6 +15,7 @@ def __getattr__(self, item):
    D = DomainName('example.com')
    IP = '127.0.0.1'
    TTL = 60 * 5
    PORT = 5053

    soa_record = SOA(
    mname=D.ns1, # primary name server
    @@ -31,7 +36,6 @@ def __getattr__(self, item):
    D.mail: [A(IP)],
    D.andrei: [CNAME(D)],
    }
    # TODO: DNSKEY, RRSIG


    def dns_response(data):
    @@ -66,25 +70,62 @@ def dns_response(data):
    return reply.pack()


    class BaseRequestHandler(SocketServer.BaseRequestHandler):

    def get_data(self):
    raise NotImplementedError

    def send_data(self, data):
    raise NotImplementedError

    def handle(self):
    now = datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
    print "---- %s request %s (%s %s):" % (self.__class__.__name__[:3], now, self.client_address[0],
    self.client_address[1])
    try:
    data = self.get_data()
    print len(data), data.encode('hex') # repr(data).replace('\\x', '')[1:-1]
    self.send_data(dns_response(data))
    except Exception:
    traceback.print_exc(file=sys.stderr)


    class TCPRequestHandler(BaseRequestHandler):

    def get_data(self):
    return self.request.recv(8192).strip()[2:]

    def send_data(self, data):
    return self.request.sendall(data)


    class UDPRequestHandler(BaseRequestHandler):

    def get_data(self):
    return self.request[0].strip()

    def send_data(self, data):
    return self.request[1].sendto(data, self.client_address)


    if __name__ == '__main__':
    print "Starting nameserver..."

    udps = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    udps.bind(('', 5053))
    servers = [
    SocketServer.ThreadingUDPServer(('', PORT), UDPRequestHandler),
    SocketServer.ThreadingTCPServer(('', PORT), TCPRequestHandler),
    ]
    for s in servers:
    thread = threading.Thread(target=s.serve_forever) # that thread will start one more thread for each request
    thread.daemon = True # exit the server thread when the main thread terminates
    thread.start()
    print "%s server loop running in thread: %s" % (s.RequestHandlerClass.__name__[:3], thread.name)

    try:
    while 1:
    print "\nWaiting for query...\n"
    data, addr = udps.recvfrom(8192)

    try:
    print "---- Request (%s %s):" % tuple(addr)
    udps.sendto(dns_response(data), addr)

    except:
    import sys
    import traceback
    traceback.print_exc(file=sys.stderr)

    pass
    except KeyboardInterrupt:
    udps.close()
    pass
    finally:
    for s in servers:
    s.shutdown()
  7. andreif revised this gist Jul 24, 2013. 1 changed file with 2 additions and 1 deletion.
    3 changes: 2 additions & 1 deletion Python simple DNS server with dnslib.py
    View file
    Original file line number Diff line number Diff line change
    @@ -82,8 +82,9 @@ def dns_response(data):
    udps.sendto(dns_response(data), addr)

    except:
    import sys
    import traceback
    traceback.print_exc()
    traceback.print_exc(file=sys.stderr)

    except KeyboardInterrupt:
    udps.close()
  8. andreif created this gist Jul 24, 2013.
    89 changes: 89 additions & 0 deletions Python simple DNS server with dnslib.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,89 @@
    # coding=utf-8
    import socket
    from dnslib import *


    class DomainName(str):
    def __getattr__(self, item):
    return DomainName(item + '.' + self)


    D = DomainName('example.com')
    IP = '127.0.0.1'
    TTL = 60 * 5

    soa_record = SOA(
    mname=D.ns1, # primary name server
    rname=D.andrei, # email of the domain administrator
    times=(
    201307231, # serial number
    60 * 60 * 1, # refresh
    60 * 60 * 3, # retry
    60 * 60 * 24, # expire
    60 * 60 * 1, # minimum
    )
    )
    ns_records = [NS(D.ns1), NS(D.ns2)]
    records = {
    D: [A(IP), AAAA((0,) * 16), MX(D.mail), soa_record] + ns_records,
    D.ns1: [A(IP)], # MX and NS records must never point to a CNAME alias (RFC 2181 section 10.3)
    D.ns2: [A(IP)],
    D.mail: [A(IP)],
    D.andrei: [CNAME(D)],
    }
    # TODO: DNSKEY, RRSIG


    def dns_response(data):
    request = DNSRecord.parse(data)

    print request

    reply = DNSRecord(DNSHeader(id=request.header.id, qr=1, aa=1, ra=1), q=request.q)

    qname = request.q.qname
    qn = str(qname)
    qtype = request.q.qtype
    qt = QTYPE[qtype]

    if qn == D or qn.endswith('.' + D):

    for name, rrs in records.iteritems():
    if name == qn:
    for rdata in rrs:
    rqt = rdata.__class__.__name__
    if qt in ['*', rqt]:
    reply.add_answer(RR(rname=qname, rtype=QTYPE[rqt], rclass=1, ttl=TTL, rdata=rdata))

    for rdata in ns_records:
    reply.add_ns(RR(rname=D, rtype=QTYPE.NS, rclass=1, ttl=TTL, rdata=rdata))

    reply.add_ns(RR(rname=D, rtype=QTYPE.SOA, rclass=1, ttl=TTL, rdata=soa_record))

    print "---- Reply"
    print reply

    return reply.pack()


    if __name__ == '__main__':
    print "Starting nameserver..."

    udps = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    udps.bind(('', 5053))

    try:
    while 1:
    print "\nWaiting for query...\n"
    data, addr = udps.recvfrom(8192)

    try:
    print "---- Request (%s %s):" % tuple(addr)
    udps.sendto(dns_response(data), addr)

    except:
    import traceback
    traceback.print_exc()

    except KeyboardInterrupt:
    udps.close()