@@ -0,0 +1,2409 @@
+from twisted.internet import defer%0Afrom twisted.python import usage%0A%0Afrom scapy.all import IP, UDP, DNS, DNSQR%0A%0Afrom ooni.templates import scapyt%0Afrom ooni.utils import log%0A%0Aclass UsageOptions(usage.Options):%0A    optParameters = %5B%5B'resolver', 'r', None,%0A                    'Specify the resolver that should be used for DNS queries (ip:port)'%5D,%0A                    %5B'hostname', 'h', None,%0A                        'Specify the hostname of a censored site'%5D,%0A                    %5B'backend', 'b', '8.8.8.8:53',%0A                        'Specify the IP address of a good DNS resolver (ip:port)'%5D%0A                    %5D%0A%0A%0Aclass DNSSpoof(scapyt.ScapyTest):%0A    name = %22DNS Spoof%22%0A    timeout = 2%0A%0A    usageOptions = UsageOptions%0A%0A    requiredOptions = %5B'hostname', 'resolver'%5D%0A%0A    def setUp(self):%0A        self.resolverAddr, self.resolverPort = self.localOptions%5B'resolver'%5D.split(':')%0A        self.resolverPort = int(self.resolverPort)%0A%0A        self.controlResolverAddr, self.controlResolverPort = self.localOptions%5B'backend'%5D.split(':')%0A        self.controlResolverPort = int(self.controlResolverPort)%0A%0A        self.hostname = self.localOptions%5B'hostname'%5D%0A%0A    def postProcessor(self, report):%0A        %22%22%22%0A        This is not tested, but the concept is that if the two responses%0A        match up then spoofing is occuring.%0A        %22%22%22%0A        test_answer = report%5B'test_a_lookup'%5D%5B'answered_packets'%5D%5B0%5D%5B1%5D%0A        control_answer = report%5B'test_control_a_lookup'%5D%5B'answered_packets'%5D%5B0%5D%5B1%5D%0A        if test_answer%5BUDP%5D == control_answer%5BUDP%5D:%0A                self.report%5B'spoofing'%5D = True%0A        else:%0A            self.report%5B'spoofing'%5D = False%0A        return%0A%0A    @defer.inlineCallbacks%0A    def test_a_lookup(self):%0A        question = IP(dst=self.resolverAddr)/UDP()/DNS(rd=1,%0A                qd=DNSQR(qtype=%22A%22, qclass=%22IN%22, qname=self.hostname))%0A        log.msg(%22Performing query to %25s with %25s:%25s%22 %25 (self.hostname, self.resolverAddr, self.resolverPort))%0A        answered, unanswered = yield self.sr1(question)%0A%0A    @defer.inlineCallbacks%0A    def test_control_a_lookup(self):%0A        question = IP(dst=self.controlResolverAddr)/UDP()/DNS(rd=1,%0A                qd=DNSQR(qtype=%22A%22, qclass=%22IN%22, qname=self.hostname))%0A        log.msg(%22Performing query to %25s with %25s:%25s%22 %25 (self.hostname,%0A            self.controlResolverAddr, self.controlResolverPort))%0A        answered, unanswered = yield self.sr1(question)%0A%0A%0A

Implement test that detects DNS spoofing * To be run with a known good resolver and the default resolver of the country being tested, from inside the country with a hostname of a site that is known to be censored.