Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save andreif/6069838 to your computer and use it in GitHub Desktop.
Save andreif/6069838 to your computer and use it in GitHub Desktop.
# coding=utf-8
"""
LICENSE http://www.apache.org/licenses/LICENSE-2.0
"""
import datetime
import sys
import time
import threading
import traceback
import SocketServer
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
PORT = 5053
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)],
}
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:\n", reply
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 "\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()
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):
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):
sz = hex(len(data))[2:].zfill(4).decode('hex')
return self.request.sendall(sz + 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..."
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:
time.sleep(1)
sys.stderr.flush()
sys.stdout.flush()
except KeyboardInterrupt:
pass
finally:
for s in servers:
s.shutdown()
@andreif
Copy link
Author

andreif commented Feb 22, 2016

I thought I have shared a version of it with DNSSEC support but maybe it's lost somewhere among my private repos.

@andreif
Copy link
Author

andreif commented Feb 24, 2016

@ialexryan
Copy link

Awesome code snippet! Thank you for sharing. According to this Microsoft page on SOA records, "normally, the retry time is less than the refresh time". Any insight into why you chose the time values you did?

@pati-ni
Copy link

pati-ni commented Jun 4, 2016

This snippet does not work out of the box with the dig command.

$ dig @localhost -p 1866 example.com
; <<>> DiG 9.9.5-3ubuntu0.1-Ubuntu <<>> @localhost -p 1866 example.com
; (1 server found)
;; global options: +cmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 34396
;; flags: qr aa rd ra; QUERY: 1, ANSWER: 0, AUTHORITY: 0, ADDITIONAL: 0

;; QUESTION SECTION:
;example.com.           IN  A

;; Query time: 1 msec
;; SERVER: 127.0.0.1#1866(127.0.0.1)
;; WHEN: Sat Jun 04 12:54:08 EEST 2016
;; MSG SIZE  rcvd: 29

in line 57 qn is "example.com." and D is "example.com"

@serain
Copy link

serain commented Sep 23, 2016

Doesn't work with either dig or nslookup:

~ $ nslookup -port=5053 example.com localhost
Server:     localhost
Address:    127.0.0.1#5053

*** Can't find example.com: No answer
 $ dig @localhost -p 5053 example.com

; <<>> DiG 9.9.5-3ubuntu0.8-Ubuntu <<>> @localhost -p 5053 example.com
; (1 server found)
;; global options: +cmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 39332
;; flags: qr aa rd ra; QUERY: 1, ANSWER: 0, AUTHORITY: 0, ADDITIONAL: 0

;; QUESTION SECTION:
;example.com.           IN  A

;; Query time: 0 msec
;; SERVER: 127.0.0.1#5053(127.0.0.1)
;; WHEN: Fri Sep 23 10:32:33 BST 2016
;; MSG SIZE  rcvd: 29

@aleSilver
Copy link

@pata-ni
error in line 57: if qn == D or qn.endswith('.' + D):

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

Now I got this error:

Traceback (most recent call last):
File "dns-server.py", line 95, in handle
self.send_data(dns_response(data))
File "dns-server.py", line 71, in dns_response
reply.add_ns(RR(rname=D, rtype=QTYPE.NS, rclass=1, ttl=TTL, rdata=rdata))
AttributeError: 'DNSRecord' object has no attribute 'add_ns'

@aleSilver
Copy link

This line did not work
reply.add_ns(RR(rname=D, rtype=QTYPE.SOA, rclass=1, ttl=TTL, rdata=soa_record))

So I changed it with the following
reply.add_answer(RR(rname=D, rtype=QTYPE.NS, rclass=1, ttl=TTL, rdata=rdata))

Now the server worked with dig:

$ dig @127.0.0.1 -p 5053 www.example.com

; <<>> DiG 9.9.5-3ubuntu0.9-Ubuntu <<>> @127.0.0.1 -p 5053 www.example.com
; (1 server found)
;; global options: +cmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 55402
;; flags: qr aa rd ra; QUERY: 1, ANSWER: 3, AUTHORITY: 0, ADDITIONAL: 0

;; QUESTION SECTION:
;www.example.com.       IN  A

;; ANSWER SECTION:
www.example.com.    300 IN  NS  ns1.www.example.com.
www.example.com.    300 IN  NS  ns2.www.example.com.
www.example.com.    300 IN  NS  ns2.www.example.com.

;; Query time: 2 msec
;; SERVER: 127.0.0.1#5053(127.0.0.1)
;; WHEN: Wed Oct 05 17:39:43 CEST 2016
;; MSG SIZE  rcvd: 83

@xrl1
Copy link

xrl1 commented Nov 3, 2016

As @Th4nat0s commented, if the first byte of the transaction id represents a whitespace char (which is allowed), than strip removes him, and the DNS query becomes invalid.
It causes DNSRecord.parse function to raise an exception.

@samuelcolvin
Copy link

I'm afraid there are quite a few errors with this, however it did alert me to the excellent dnslib library, thanks. If only the author had used it's built in tools rather than trying to roll his/her own - for example dnslib has DNSServer which takes away a lot of the pain.

Here's a complete rewrite, where:

  • actually works (tested a fair bit with dig).
  • works with python 3.6 (and probably 3.5, but not tested), should pretty much pass flake8
  • implements some sensible defaults (for ttl, rname etc.) and does a bit of cleverness (eg. returned TLDR SOA record for missing subdomains).
  • is extensible, you should just need to edit ZONES to add whatever other records you wish.

@SeanFelipe
Copy link

SeanFelipe commented Apr 6, 2017

Thanks for the code and comments -- I'm about to use this for a project to help teachers manage what websites students can access, to help reduce distractions in the classroom.

@dugduy
Copy link

dugduy commented Apr 10, 2022

thanks man!!!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment