Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
Respond to and record dns lookups.
#!/usr/bin/env python
import sys
import random
import string
import datetime
import itertools
import threading
import traceback
import SocketServer
import operator as op
import dnslib
import flask
from flask import request
_response_ip = '' # IP to respond with
_lookups_lock = threading.Lock()
_lookups = []
app = flask.Flask(__name__)
class BaseRequestHandler(SocketServer.BaseRequestHandler):
def get_data(self):
raise NotImplementedError
def send_data(self, data):
raise NotImplementedError
def handle(self):
now ='%Y-%m-%d %H:%M:%S')
(ip_addr, port) = self.client_address
data = self.get_data()
if data is None:
request = dnslib.DNSRecord.parse(data)
domains = [".".join(q.qname.label) for q in request.questions]
domain = ".".join(request.q.qname.label)
print "[+] Got a request for '{}' from: {}".format(domain, ip_addr)
with _lookups_lock:
_lookups.append((now, domain, ip_addr,))
reply = dnslib.DNSRecord(dnslib.DNSHeader(, q=request.q)
rtype=dnslib.QTYPE.A, rclass=1, ttl=60*5, rdata=dnslib.A(_response_ip)))
except Exception:
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:
return None
#raise Exception("Wrong size of TCP packet")
elif sz > len(data) - 2:
return None
#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)
def html_encode(s):
for code in [("'", '&#39;'), ('"', '&quot;'), ('>', '&gt;'), ('<', '&lt;'), ('&', '&amp;')]:
s = s.replace(code[0], code[1])
return s
def status(_id):
key_func = op.itemgetter(1)
hits = []
for k, grp in itertools.groupby(sorted(_lookups, key=key_func), key=key_func):
if k == _id:
hits = ["[{}] {}".format(_[0], html_encode(_[2])) for _ in grp]
return """<html><pre>{}</pre></html>""".format(hits and "\n".join(hits) or 'No Lookups')
def index():
hits = ["[timestamp] - client ip - hostname"]
_sorted = sorted(_lookups, key=op.itemgetter(0))
hits.extend(["[{}] - {} - {}".format(_[0], _[2], html_encode(_[1])) for _ in _sorted])
return """<html><pre>Recent Lookups:\n{}</pre></html>""".format("\n".join(hits))
if __name__ == "__main__":
print "Starting nameservers..."
servers = [
SocketServer.ThreadingUDPServer(('', 53), UDPRequestHandler),
SocketServer.ThreadingTCPServer(('', 53), TCPRequestHandler)]
for s in servers:
thread = threading.Thread(target=s.serve_forever)
thread.daemon = True
print "Starting webserver..."
try:'', port=8000, debug=True, use_reloader=False)
except KeyboardInterrupt:
map(lambda s: s.shutdown(), servers)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment