dnsfuzz.py

# http://notmysock.org/blog/hacks/a-twisted-dns-story.html
# https://gist.github.com/johnboxall/1147973
# twistd -y dnsfuzz.py

import sys
import socket

from twisted.python import log
from twisted.internet.protocol import Factory, Protocol
from twisted.internet import reactor
from twisted.names import dns, client, server


class FuzzDNSDatagramProtocol(dns.DNSDatagramProtocol):

    _original_writeMessage = dns.DNSDatagramProtocol.writeMessage

    def writeMessage(self, message, address):
        """Original implementation:
        
        self.transport.write(message.toStr(), address)

        """
        # TODO: fuzz message
        log.msg('Fuzzed message: %s' % repr(message.toStr()))
        dns.DNSDatagramProtocol.writeMessage(self, message, address)

    def _query(self, queries, timeout, id, writeMessage):
        # The original implementation of query also uses writeMessage() to send a
        # query. For the sake of a fuzzing DNS server that responds with a fuzzed
        # DNS response, we don't actually want to fuzz the queries forwarded to
        # the real DNS servers.
        writeMessage = self._original_writeMessage
        dns.DNSMixin._query(self, queries, timeout, id, writeMessage)


class FuzzDNSProtocol(dns.DNSProtocol):

    _original_writeMessage = dns.DNSProtocol.writeMessage

    def writeMessage(self, message): 
        """Original implementation:

        s = message.toStr()
        self.transport.write(struct.pack('!H', len(s)) + s)  

        """
        # TODO: fuzz message
        log.msg('Fuzzed message: %s' % repr(message.toStr()))
        dns.DNSProtocol.writeMessage(self, message)

    def _query(self, queries, timeout, id, writeMessage):
        # The original implementation of query also uses writeMessage() to send a
        # query. For the sake of a fuzzing DNS server that responds with a fuzzed
        # DNS response, we don't actually want to fuzz the queries forwarded to
        # the real DNS servers.
        writeMessage = self._original_writeMessage
        dns.DNSMixin._query(self, queries, timeout, id, writeMessage)


verbosity = 2

resolver = client.Resolver(servers=[('127.0.1.1', 53)])      # set the server we forward original queries to
f = server.DNSServerFactory(clients=[resolver], verbose=verbosity)
f.protocol = FuzzDNSProtocol
p = FuzzDNSDatagramProtocol(f)
f.noisy = p.noisy = verbosity

log.startLogging(sys.stdout)

reactor.listenUDP(5553, p)
reactor.listenTCP(5553, f)
reactor.run()           # test from a client, e.g. with `dig @<this_host> -p 5553 www.example.com`