@@ -0,0 +1,1962 @@
+# -*- encoding: utf-8 -*-%0Afrom twisted.python import usage%0Afrom twisted.internet import defer%0A%0Afrom ooni.templates import dnst%0Afrom ooni import nettest%0Afrom ooni.utils import log%0A%0Aclass UsageOptions(usage.Options):%0A    optParameters = %5B%0A            %5B'resolver', 'r', '8.8.8.1', 'an invalid DNS resolver'%5D,%0A            %5B'timeout', 't', 3, 'timeout after which we should consider the query failed'%5D%0A    %5D%0A%0Aclass DNSInjectionTest(dnst.DNSTest):%0A    %22%22%22%0A    This test detects DNS spoofed DNS responses by performing UDP based DNS%0A    queries towards an invalid DNS resolver.%0A%0A    For it to work we must be traversing the network segment of a machine that%0A    is actively injecting DNS query answers.%0A    %22%22%22%0A    name = %22DNS Injection%22%0A    description = %22Checks for injection of spoofed DNS answers%22%0A    version = %220.1%22%0A    authors = %22Arturo Filast%C3%B2%22%0A%0A    inputFile = %5B'file', 'f', None,%0A                 'Input file of list of hostnames to attempt to resolve'%5D%0A%0A    usageOptions = UsageOptions%0A    requiredOptions = %5B'resolver', 'file'%5D%0A%0A    def setUp(self):%0A        self.resolver = (self.localOptions%5B'resolver'%5D, 53)%0A        self.queryTimeout = %5Bself.localOptions%5B'timeout'%5D%5D%0A%0A    def inputProcessor(self, filename):%0A        fp = open(filename)%0A        for line in fp:%0A            if line.startswith('http://'):%0A                yield line.replace('http://', '').replace('/', '').strip()%0A            else:%0A                yield x.strip()%0A        fp.close()%0A%0A    def test_injection(self):%0A        self.report%5B'injected'%5D = None%0A%0A        d = self.performALookup(self.input, self.resolver)%0A        @d.addCallback%0A        def cb(res):%0A            log.msg(%22The DNS query for %25s is injected%22 %25 self.input)%0A            self.report%5B'injected'%5D = True%0A%0A        @d.addErrback%0A        def err(err):%0A            err.trap(defer.TimeoutError)%0A            log.msg(%22The DNS query for %25s is not injected%22 %25 self.input)%0A            self.report%5B'injected'%5D = False%0A%0A        return d%0A%0A

Add DNS injection test for detecting censorship when DNS inject happens