[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]

[tor-commits] [ooni-probe/master] Rewrite DNS Tamper test to properly use Twisted DNS client.



commit c904e684ac60f9c25c09d52251cb27966aaab065
Author: Arturo Filastò <arturo@xxxxxxxxxxx>
Date:   Mon Oct 8 02:25:13 2012 +0000

    Rewrite DNS Tamper test to properly use Twisted DNS client.
    * Should be ready for usage
---
 nettests/core/dnstamper.py |  182 ++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 182 insertions(+), 0 deletions(-)

diff --git a/nettests/core/dnstamper.py b/nettests/core/dnstamper.py
new file mode 100644
index 0000000..92948cd
--- /dev/null
+++ b/nettests/core/dnstamper.py
@@ -0,0 +1,182 @@
+# -*- encoding: utf-8 -*-
+#
+#
+#  dnstamper
+#  *********
+#
+#  The test reports censorship if the cardinality of the intersection of
+#  the query result set from the control server and the query result set
+#  from the experimental server is zero, which is to say, if the two sets
+#  have no matching results whatsoever.
+#
+#  NOTE: This test frequently results in false positives due to GeoIP-based
+#  load balancing on major global sites such as google, facebook, and
+#  youtube, etc.
+#
+# :authors: Arturo Filastò, Isis Lovecruft
+# :licence: see LICENSE
+
+from ooni import nettest
+from twisted.internet import defer
+from twisted.names import client
+
+class DNSTamperTest(nettest.TestCase):
+
+    name = "DNS tamper"
+
+    description = "DNS censorship detection test"
+
+    requirements = None
+    inputFile = ['file', 'f', None,
+                 'Input file of list of hostnames to attempt to resolve']
+
+    optParameters = [['controlresolver', 'c', '8.8.8.8',
+                      'Known good DNS server'],
+                     ['testresolvers', 't', None,
+                      'file containing list of DNS resolvers to test against']
+                     ]
+
+    def setUp(self):
+        self.report['test_lookups'] = {}
+        self.report['test_reverse'] = {}
+
+        self.report['control_lookup'] = []
+
+        self.report['a_lookups'] = {}
+
+        self.test_a_lookups = {}
+        self.control_a_lookups = []
+
+        self.test_reverse = {}
+
+        if not self.localOptions['testresolvers']:
+            self.test_resolvers = ['8.8.8.8']
+            return
+        try:
+            fp = open(self.localOptions['testresolvers'])
+        except:
+            raise usage.UsageError("Invalid test resolvers file")
+
+        self.test_resolvers = [x.strip() for x in fp.readlines]
+        fp.close()
+
+    def process_a_answers(self, answers, resolver):
+        all_a = []
+        a_a = []
+        for answer in answers[0]:
+            if answer.type is 1:
+                # A type query
+                r = answer.payload.dottedQuad()
+                self.report['a_lookups'][resolver] = r
+                a_a.append(r)
+            lookup = str(answer.payload)
+            all_a.append(lookup)
+
+        if resolver == 'control':
+            self.report['control_server'] = self.localOptions['controlresolver']
+            self.report['control_lookup'] = all_a
+            self.control_a_lookups = a_a
+        else:
+            self.test_a_lookups[resolver] = a_a
+            self.report['test_lookups'][resolver] = all_a
+
+    def process_ptr_answers(self, answers, resolver):
+        print "Processing PTR ANSWER for %s" % resolver
+        name = None
+        for answer in answers[0]:
+            if answer.type is 12:
+                # PTR type
+                name = str(answer.payload.name)
+
+        if resolver == 'control':
+            self.control_reverse = name
+            self.report['control_reverse'] = name
+        else:
+            self.test_reverse[resolver] = name
+            self.report['test_reverse'][resolver] = name
+
+
+    def ptr_lookup_error(self, failure, resolver):
+        if resolver == 'control':
+            self.report['control_reverse'] = None
+        else:
+            self.report['test_reverse'][resolver] = None
+
+    def a_lookup_error(self, failure, resolver):
+        if resolver == 'control':
+            self.report['control_lookup'] = None
+        else:
+            self.report['test_lookups'][resolver] = None
+
+    def test_lookup(self):
+        list_of_ds = []
+        hostname = self.input
+
+        resolver = [(self.localOptions['controlresolver'], 53)]
+        res = client.createResolver(servers=resolver)
+
+        control_r = res.lookupAllRecords(hostname)
+        control_r.addCallback(self.process_a_answers, 'control')
+        control_r.addErrback(self.a_lookup_error, 'control')
+
+        list_of_ds.append(control_r)
+
+        for test_resolver in self.test_resolvers:
+            resolver = [(test_resolver,53)]
+            res = client.createResolver(servers=resolver)
+
+            d = res.lookupAddress(hostname)
+            d.addCallback(self.process_a_answers, test_resolver)
+            d.addErrback(self.a_lookup_error, test_resolver)
+            list_of_ds.append(d)
+
+        dl = defer.DeferredList(list_of_ds)
+        dl.addCallback(self.do_reverse_lookups)
+        dl.addCallback(self.compare_results)
+        return dl
+
+    def reverse_lookup(self, address, resolver):
+        ptr = '.'.join(address.split('.')[::-1]) + '.in-addr.arpa'
+        r = resolver.lookupPointer(ptr)
+        return r
+
+    def do_reverse_lookups(self, result):
+
+
+        list_of_ds = []
+
+        resolver = [(self.localOptions['controlresolver'], 53)]
+        res = client.createResolver(servers=resolver)
+
+        test_reverse = self.reverse_lookup(self.control_a_lookups[0], res)
+        test_reverse.addCallback(self.process_ptr_answers, 'control')
+        test_reverse.addErrback(self.ptr_lookup_error, 'control')
+
+        list_of_ds.append(test_reverse)
+
+        for test_resolver in self.test_resolvers:
+            d = self.reverse_lookup(self.test_a_lookups[test_resolver][0], res)
+            d.addCallback(self.process_ptr_answers, test_resolver)
+            d.addErrback(self.ptr_lookup_error, test_resolver)
+            list_of_ds.append(d)
+
+        dl = defer.DeferredList(list_of_ds)
+        return dl
+
+    def compare_results(self, *arg, **kw):
+        for test, test_a_lookups in self.test_a_lookups.items():
+            self.report['tampering'] = {}
+            self.report['tampering'][test] = 'unknown'
+
+            if len(set(test_a_lookups) & set(self.control_a_lookups)) > 0:
+                # Address has not tampered with on DNS server
+                self.report['tampering'][test] = False
+
+            elif len(set(self.control_reverse) & set(self.test_reverse[test])) > 0:
+                # Further testing has eliminated false positives
+                self.report['tampering'][test] = 'reverse-match'
+            else:
+                # Reverse DNS on the results returned by returned
+                # which does not match the expected domainname
+                self.report['tampering'][test] = True
+

_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits