Test script for ArchiveTeam/wpull#365

dnscrashtest.md

dnscrashtest.py tests whether dnspython handles invalid messages correctly. With a broken dnspython (all released versions as of 2018-10-14), you get a ValueError. With a fixed dnspython (either a more recent commit from the master branch, or a patched dns.inet.is_multicast, which can be enabled by uncommenting line 20), you get an NXDOMAIN response.

dnscrashtest.py

import dns.query
import dns.rcode
import dns.resolver
import socket


def nowait(*args):
	pass

dns.query._wait_for_readable = nowait
dns.query._wait_for_writable = nowait


dns__inet__is_multicast = dns.inet.is_multicast
def is_multicast(text):
	try:
		return dns__inet__is_multicast(text)
	except Exception:
		return False
#dns.inet.is_multicast = is_multicast


class TriggerSocket(socket.socket):
	'''A socket that triggers dnspython's ValueError crash'''
	def __init__(self, *args, **kwargs):
		super().__init__(*args, **kwargs)
		self.__targetAddress = None
		self.__query = None
		self.__recvCount = 0

	def sendto(self, data, destination):
		self.__query = dns.message.from_wire(data)
		self.__targetAddress = destination

	def recvfrom(self, bufsize):
		self.__recvCount += 1
		if self.__recvCount == 1:
			# First call, return bogus multicast
			return b'fuckyou', ('255.255.255.255', 12345)
		else:
			# Second or later call, return NXDOMAIN
			r = dns.message.make_response(self.__query)
			r.set_rcode(dns.rcode.NXDOMAIN)
			return r.to_wire(), self.__targetAddress

dns.query.socket_factory = TriggerSocket


print(dns.query.udp(dns.message.make_query('example.org', 'A'), '1.1.1.1', ignore_unexpected = True))