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

[tor-commits] [ooni-probe/master] Port Isis's captive portal test to the new framework.



commit 3c632b495d15cfeb77577e6f9153f4915453cbfc
Author: Arturo Filastò <art@xxxxxxxxxxxxxx>
Date:   Mon Jul 9 12:41:49 2012 +0200

    Port Isis's captive portal test to the new framework.
---
 ooni/plugins/captiveportal.py |  625 +++++++++++++++++++++++++++++++++++++++++
 1 files changed, 625 insertions(+), 0 deletions(-)

diff --git a/ooni/plugins/captiveportal.py b/ooni/plugins/captiveportal.py
new file mode 100644
index 0000000..9b7bee3
--- /dev/null
+++ b/ooni/plugins/captiveportal.py
@@ -0,0 +1,625 @@
+# -*- coding: utf-8 -*-
+"""
+    captiveportal
+    *************
+
+    This test is a collection of tests to detect the presence of a
+    captive portal. Code is taken, in part, from the old ooni-probe,
+    which was written by Jacob Appelbaum and Arturo Filastò.
+
+    :copyright: (c) 2012 Isis Lovecruft
+    :license: see LICENSE for more details
+"""
+import base64
+import os
+import random
+import re
+import string
+import urllib2
+from urlparse import urlparse
+
+from zope.interface import implements
+from twisted.python import usage
+from twisted.plugin import IPlugin
+
+from ooni.plugoo.assets import Asset
+from ooni.plugoo.tests import ITest, OONITest
+from ooni.protocols import http
+from ooni import log
+
+try:
+    from dns import resolver
+except ImportError:
+    print "The dnspython module was not found. https://crate.io/packages/dnspython/";
+
+__plugoo__ = "captiveportal"
+__desc__ = "Captive portal detection test"
+
+class CaptivePortalArgs(usage.Options):
+    optParameters = [['asset', 'a', None, 'Asset file'],
+                     ['resume', 'r', 0, 'Resume at this index'],
+                     ['experiment-url', 'e', 'http://google.com/', 'Experiment URL'],
+                     ['user-agent', 'u', random.choice(http.useragents),
+                                         'User agent for HTTP requests']
+                     ]
+
+class CaptivePortal(OONITest):
+    """
+    Compares content and status codes of HTTP responses, and attempts
+    to determine if content has been altered.
+    """
+
+    implements(IPlugin, ITest)
+
+    shortName = "captivep"
+    description = "Captive Portal Test"
+    requirements = None
+    options = CaptivePortalArgs
+    # Tells this to be blocking.
+    blocking = True
+
+    def http_fetch(self, url, headers={}):
+        """
+        Parses an HTTP url, fetches it, and returns a urllib2 response
+        object.
+        """
+        url = urlparse(url).geturl()
+        request = urllib2.Request(url, None, headers)
+        response = urllib2.urlopen(request)
+        response_headers = dict(response.headers)
+        return response, response_headers
+
+    def http_content_match_fuzzy_opt(self, experimental_url, control_result,
+                                     headers=None, fuzzy=False):
+        """
+        Makes an HTTP request on port 80 for experimental_url, then
+        compares the response_content of experimental_url with the
+        control_result. Optionally, if the fuzzy parameter is set to
+        True, the response_content is compared with a regex of the
+        control_result. If the response_content from the
+        experimental_url and the control_result match, returns True
+        with the HTTP status code and headers; False, status code, and
+        headers if otherwise.
+        """
+
+        if headers is None:
+            default_ua = self.local_options['user-agent']
+            headers = {'User-Agent': default_ua}
+
+        response, response_headers = self.http_fetch(experimental_url, headers)
+        response_content = response.read()
+        response_code = response.code
+        if response_content is not None:
+            if fuzzy:
+                pattern = re.compile(control_result)
+                match = pattern.search(response_content)
+                if not match:
+                    log.msg("Fuzzy HTTP content comparison for experiment URL")
+                    log.msg("'%s'" % experimental_url)
+                    log.msg("does not match!")
+                    return False, response_code, response_headers
+                else:
+                    log.msg("Fuzzy HTTP content comparison of experiment URL")
+                    log.msg("'%s'" % experimental_url)
+                    log.msg("and the expected control result yielded a match.")
+                    return True, response_code, response_headers
+            else:
+                if str(response_content) != str(control_result):
+                    log.msg("HTTP content comparison of experiment URL")
+                    log.msg("'%s'" % experimental_url)
+                    log.msg("and the expected control result do not match.")
+                    return False, response_code, response_headers
+                else:
+                    return True, response_code, response_headers
+        else:
+            log.warn("HTTP connection appears to have failed.")
+        return False, False, False
+
+    def http_status_code_match(self, experiment_code, control_code):
+        """
+        Compare two HTTP status codes, returns True if they match.
+        """
+        if int(experiment_code) != int(control_code):
+            return False
+        return True
+
+    def http_status_code_no_match(self, experiment_code, control_code):
+        """
+        Compare two HTTP status codes, returns True if they do not match.
+        """
+        if self.http_status_code_match(experiment_code, control_code):
+            return False
+        return True
+
+    def dns_resolve(self, hostname, nameserver=None):
+        """
+        Resolves hostname(s) though nameserver to corresponding
+        address(es). hostname may be either a single hostname string,
+        or a list of strings. If nameserver is not given, use local
+        DNS resolver, and if that fails try using 8.8.8.8.
+        """
+
+        if isinstance(hostname, str):
+            hostname = [hostname]
+
+        if nameserver is not None:
+            res = resolver.Resolver(configure=False)
+            res.nameservers = [nameserver]
+        else:
+            res = resolver.Resolver()
+
+        response = []
+        answer = None
+
+        for hn in hostname:
+            try:
+                answer = res.query(hn)
+            except resolver.NoNameservers:
+                res.nameservers = ['8.8.8.8']
+                try:
+                    answer = res.query(hn)
+                except resolver.NXDOMAIN:
+                    log.msg("DNS resolution for %s returned NXDOMAIN" % hn)
+                    response.append('NXDOMAIN')
+            except resolver.NXDOMAIN:
+                log.msg("DNS resolution for %s returned NXDOMAIN" % hn)
+                response.append('NXDOMAIN')
+            finally:
+                if answer:
+                    for addr in answer:
+                        response.append(addr.address)
+        return response
+
+    def dns_resolve_match(self, experiment_hostname, control_address):
+        """
+        Resolve experiment_hostname, and check to see that it returns
+        an experiment_address which matches the control_address.  If
+        they match, returns True and experiment_address; otherwise
+        returns False and experiment_address.
+        """
+        experiment_address = self.dns_resolve(experiment_hostname)
+        if experiment_address:
+            if len(set(experiment_address) & set([control_address])) > 0:
+                return True, experiment_address
+            else:
+                log.msg("DNS comparison of control '%s' does not" % control_address)
+                log.msg("match experiment response '%s'" % experiment_address)
+                return False, experiment_address
+        else:
+            log.debug("dns_resolve() for %s failed" % experiment_hostname)
+            return None, experiment_address
+
+    def get_auth_nameservers(self, hostname):
+        """
+        Many CPs set a nameserver to be used. Let's query that
+        nameserver for the authoritative nameservers of hostname.
+
+        The equivalent of:
+        $ dig +short NS ooni.nu
+        """
+        res = resolver.Resolver()
+        answer = res.query(hostname, 'NS')
+        auth_nameservers = []
+        for auth in answer:
+            auth_nameservers.append(auth.to_text())
+        return auth_nameservers
+
+    def hostname_to_0x20(self, hostname):
+        """
+        MaKEs yOur HOsTnaME lOoK LiKE THis.
+
+        For more information, see:
+        D. Dagon, et. al. "Increased DNS Forgery Resistance
+        Through 0x20-Bit Encoding". Proc. CSS, 2008.
+        """
+        hostname_0x20 = ''
+        for char in hostname:
+            l33t = random.choice(['caps', 'nocaps'])
+            if l33t == 'caps':
+                hostname_0x20 = hostname_0x20 + char.capitalize()
+            else:
+                hostname_0x20 = hostname_0x20 + char.lower()
+        return hostname_0x20
+
+    def check_0x20_to_auth_ns(self, hostname, sample_size=None):
+        """
+        Resolve a 0x20 DNS request for hostname over hostname's
+        authoritative nameserver(s), and check to make sure that
+        the capitalization in the 0x20 request matches that of the
+        response. Also, check the serial numbers of the SOA (Start
+        of Authority) records on the authoritative nameservers to
+        make sure that they match.
+
+        If sample_size is given, a random sample equal to that number
+        of authoritative nameservers will be queried; default is 5.
+        """
+        log.msg("")
+        log.msg("Testing random capitalization of DNS queries...")
+        log.msg("Testing that Start of Authority serial numbers match...")
+
+        auth_nameservers = self.get_auth_nameservers(hostname)
+
+        if sample_size is None:
+            sample_size = 5
+            resolved_auth_ns = random.sample(self.dns_resolve(auth_nameservers),
+                                             sample_size)
+
+        querynames = []
+        answernames = []
+        serials = []
+
+        # Even when gevent monkey patching is on, the requests here
+        # are sent without being 0x20'd, so we need to 0x20 them.
+        hostname = self.hostname_to_0x20(hostname)
+
+        for auth_ns in resolved_auth_ns:
+            res = resolver.Resolver(configure=False)
+            res.nameservers = [auth_ns]
+            try:
+                answer = res.query(hostname, 'SOA')
+            except resolver.Timeout:
+                continue
+            querynames.append(answer.qname.to_text())
+            answernames.append(answer.rrset.name.to_text())
+            for soa in answer:
+                serials.append(str(soa.serial))
+
+        if len(set(querynames).intersection(answernames)) == 1:
+            log.msg("Capitalization in DNS queries and responses match.")
+            name_match = True
+        else:
+            log.msg("The random capitalization '%s' used in" % hostname)
+            log.msg("DNS queries to that hostname's authoritative")
+            log.msg("nameservers does not match the capitalization in")
+            log.msg("the response.")
+            name_match = False
+
+        if len(set(serials)) == 1:
+            log.msg("Start of Authority serial numbers all match.")
+            serial_match = True
+        else:
+            log.msg("Some SOA serial numbers did not match the rest!")
+            serial_match = False
+
+        ret = name_match, serial_match, querynames, answernames, serials
+
+        if name_match and serial_match:
+            log.msg("Your DNS queries do not appear to be tampered.")
+            return ret
+        elif name_match or serial_match:
+            log.msg("Something is tampering with your DNS queries.")
+            return ret
+        elif not name_match and not serial_match:
+            log.msg("Your DNS queries are definitely being tampered with.")
+            return ret
+
+    def get_random_url_safe_string(self, length):
+        """
+        Returns a random url-safe string of specified length, where
+        0 < length <= 256. The returned string will always start with
+        an alphabetic character.
+        """
+        if (length <= 0):
+            length = 1
+        elif (length > 256):
+            length = 256
+
+        random_ascii = base64.urlsafe_b64encode(os.urandom(int(length)))
+
+        while not random_ascii[:1].isalpha():
+            random_ascii = base64.urlsafe_b64encode(os.urandom(int(length)))
+
+        three_quarters = int((len(random_ascii)) * (3.0/4.0))
+        random_string = random_ascii[:three_quarters]
+        return random_string
+
+    def get_random_hostname(self, length=None):
+        """
+        Returns a random hostname with SLD of specified length. If
+        length is unspecified, length=32 is used.
+
+        These *should* all resolve to NXDOMAIN. If they actually
+        resolve to a box that isn't part of a captive portal that
+        would be rather interesting.
+        """
+        if length is None:
+            length = 32
+
+        random_sld = self.get_random_url_safe_string(length)
+
+        # if it doesn't start with a letter, chuck it.
+        while not random_sld[:1].isalpha():
+            random_sld = self.get_random_url_safe_string(length)
+
+        tld_list = ['.com', '.net', '.org', '.info', '.test', '.invalid']
+        random_tld = urllib2.random.choice(tld_list)
+        random_hostname = random_sld + random_tld
+        return random_hostname
+
+    def compare_random_hostnames(self, hostname_count=None, hostname_length=None):
+        """
+        Get hostname_count number of random hostnames with SLD length
+        of hostname_length, and then attempt DNS resolution. If no
+        arguments are given, default to three hostnames of 32 bytes
+        each. These random hostnames *should* resolve to NXDOMAIN,
+        except in the case where a user is presented with a captive
+        portal and remains unauthenticated, in which case the captive
+        portal may return the address of the authentication page.
+
+        If the cardinality of the intersection of the set of resolved
+        random hostnames and the single element control set
+        (['NXDOMAIN']) are equal to one, then DNS properly resolved.
+
+        Returns true if only NXDOMAINs were returned, otherwise returns
+        False with the relative complement of the control set in the
+        response set.
+        """
+        if hostname_count is None:
+            hostname_count = 3
+
+        log.msg("Generating random hostnames...")
+        log.msg("Resolving DNS for %d random hostnames..." % hostname_count)
+
+        control = ['NXDOMAIN']
+        responses = []
+
+        for x in range(hostname_count):
+            random_hostname = self.get_random_hostname(hostname_length)
+            response_match, response_address = self.dns_resolve_match(random_hostname,
+                                                                      control[0])
+            for address in response_address:
+                if response_match is False:
+                    log.msg("Strangely, DNS resolution of the random hostname")
+                    log.msg("%s actually points to %s"
+                             % (random_hostname, response_address))
+                    responses = responses + [address]
+                else:
+                    responses = responses + [address]
+
+        intersection = set(responses) & set(control)
+        relative_complement = set(responses) - set(control)
+        r = set(responses)
+
+        if len(intersection) == 1:
+            log.msg("All %d random hostnames properly resolved to NXDOMAIN."
+                     % hostname_count)
+            return True, relative_complement
+        elif (len(intersection) == 1) and (len(r) > 1):
+            log.msg("Something odd happened. Some random hostnames correctly")
+            log.msg("resolved to NXDOMAIN, but several others resolved to")
+            log.msg("to the following addresses: %s" % relative_complement)
+            return False, relative_complement
+        elif (len(intersection) == 0) and (len(r) == 1):
+            log.msg("All random hostnames resolved to the IP address ")
+            log.msg("'%s', which is indicative of a captive portal." % r)
+            return False, relative_complement
+        else:
+            log.debug("Apparently, pigs are flying on your network, 'cause a")
+            log.debug("bunch of hostnames made from 32-byte random strings")
+            log.debug("just magically resolved to a bunch of random addresses.")
+            log.debug("That is definitely highly improbable. In fact, my napkin")
+            log.debug("tells me that the probability of just one of those")
+            log.degug("hostnames resolving to an address is 1.68e-59, making")
+            log.debug("it nearly twice as unlikely as an MD5 hash collision.")
+            log.debug("Either someone is seriously messing with your network,")
+            log.debug("or else you are witnessing the impossible. %s" % r)
+            return False, relative_complement
+
+    def google_dns_cp_test(self):
+        """
+        Google Chrome resolves three 10-byte random hostnames.
+        """
+        subtest = "Google Chrome DNS-based"
+
+        log.msg("")
+        log.msg("Running the Google Chrome DNS-based captive portal test...")
+
+        gmatch, google_dns_result = self.compare_random_hostnames(3, 10)
+
+        if gmatch:
+            log.msg("Google Chrome DNS-based captive portal test did not")
+            log.msg("detect a captive portal.")
+            return google_dns_result
+        else:
+            log.msg("Google Chrome DNS-based captive portal test believes")
+            log.msg("you are in a captive portal, or else something very")
+            log.msg("odd is happening with your DNS.")
+            return google_dns_result
+
+    def ms_dns_cp_test(self):
+        """
+        Microsoft "phones home" to a server which will always resolve
+        to the same address.
+        """
+        subtest = "Microsoft NCSI DNS-based"
+
+        log.msg("")
+        log.msg("Running the Microsoft NCSI DNS-based captive portal")
+        log.msg("test...")
+
+        msmatch, ms_dns_result = self.dns_resolve_match("dns.msftncsi.com",
+                                                        "131.107.255.255")
+        if msmatch:
+            log.msg("Microsoft NCSI DNS-based captive portal test did not")
+            log.msg("detect a captive portal.")
+            return ms_dns_result
+        else:
+            log.msg("Microsoft NCSI DNS-based captive portal test ")
+            log.msg("believes you are in a captive portal.")
+            return ms_dns_result
+
+    def run_vendor_dns_tests(self):
+        """
+        Run the vendor DNS tests.
+        """
+        report = {}
+        report['google_dns_cp'] = self.google_dns_cp_test()
+        report['ms_dns_cp'] = self.ms_dns_cp_test()
+
+        return report
+
+    def run_vendor_tests(self, *a, **kw):
+        """
+        These are several vendor tests used to detect the presence of
+        a captive portal. Each test compares HTTP status code and
+        content to the control results and has its own User-Agent
+        string, in order to emulate the test as it would occur on the
+        device it was intended for. Vendor tests are defined in the
+        format:
+        [exp_url, ctrl_result, ctrl_code, ua, test_name]
+        """
+
+        vendor_tests = [['http://www.apple.com/library/test/success.html',
+                         'Success',
+                         '200',
+                         'Mozilla/5.0 (iPhone; U; CPU like Mac OS X; en) AppleWebKit/420+ (KHTML, like Gecko) Version/3.0 Mobile/1A543a Safari/419.3',
+                         'Apple HTTP Captive Portal'],
+                        ['http://tools.ietf.org/html/draft-nottingham-http-portal-02',
+                         '428 Network Authentication Required',
+                         '428',
+                         'Mozilla/5.0 (Windows NT 6.1; rv:5.0) Gecko/20100101 Firefox/5.0',
+                         'W3 Captive Portal'],
+                        ['http://www.msftncsi.com/ncsi.txt',
+                         'Microsoft NCSI',
+                         '200',
+                         'Microsoft NCSI',
+                         'MS HTTP Captive Portal',]]
+
+        cm = self.http_content_match_fuzzy_opt
+        sm = self.http_status_code_match
+        snm = self.http_status_code_no_match
+
+        def compare_content(status_func, fuzzy, experiment_url, control_result,
+                            control_code, headers, test_name):
+            log.msg("")
+            log.msg("Running the %s test..." % test_name)
+
+            content_match, experiment_code, experiment_headers = cm(experiment_url,
+                                                                    control_result,
+                                                                    headers, fuzzy)
+            status_match = status_func(experiment_code, control_code)
+
+            if status_match and content_match:
+                log.msg("The %s test was unable to detect" % test_name)
+                log.msg("a captive portal.")
+                return True
+            else:
+                log.msg("The %s test shows that your network" % test_name)
+                log.msg("is filtered.")
+                return False
+
+        result = []
+        for vt in vendor_tests:
+            report = {}
+            report['vt'] = vt
+
+            experiment_url = vt[0]
+            control_result = vt[1]
+            control_code = vt[2]
+            headers = {'User-Agent': vt[3]}
+            test_name = vt[4]
+
+            args = (experiment_url, control_result, control_code, headers, test_name)
+
+            if test_name == "MS HTTP Captive Portal":
+                report['result'] = compare_content(sm, False, *args)
+
+            elif test_name == "Apple HTTP Captive Portal":
+                report['result'] = compare_content(sm, True, *args)
+
+            elif test_name == "W3 Captive Portal":
+                report['result'] = compare_content(snm, True, *args)
+
+            else:
+                log.warn("Ooni is trying to run an undefined CP vendor test.")
+            result.append(report)
+        return result
+
+    def control(self, experiment_result, args):
+        """
+        Compares the content and status code of the HTTP response for
+        experiment_url with the control_result and control_code
+        respectively. If the status codes match, but the experimental
+        content and control_result do not match, fuzzy matching is enabled
+        to determine if the control_result is at least included somewhere
+        in the experimental content. Returns True if matches are found,
+        and False if otherwise.
+        """
+        experiment_url = self.local_options['experiment-url']
+        control_result = 'XX'
+        control_code = 200
+        ua = self.local_options['user-agent']
+
+        cm = self.http_content_match_fuzzy_opt
+        sm = self.http_status_code_match
+        snm = self.http_status_code_no_match
+
+        log.msg("Running test for '%s'..." % experiment_url)
+        content_match, experiment_code, experiment_headers = cm(experiment_url,
+                                                                control_result)
+        status_match = sm(experiment_code, control_code)
+        if status_match and content_match:
+            log.msg("The test for '%s'" % experiment_url)
+            log.msg("was unable to detect a captive portal.")
+            return experiment_result, True
+
+        elif status_match and not content_match:
+            log.msg("Retrying '%s' with fuzzy match enabled."
+                     % experiment_url)
+            fuzzy_match, experiment_code, experiment_headers = cm(experiment_url,
+                                                                  control_result,
+                                                                  fuzzy=True)
+            if fuzzy_match:
+                return experiment_result, True
+            else:
+                log.msg("Found modified content on '%s'," % experiment_url)
+                log.msg("which could indicate a captive portal.")
+
+                return experiment_result, False
+        else:
+            log.msg("The content comparison test for ")
+            log.msg("'%s'" % experiment_url)
+            log.msg("shows that your HTTP traffic is filtered.")
+            return experiment_result, False
+
+    def experiment(self, args):
+        """
+        Runs the CaptivePortal(Test).
+
+        CONFIG OPTIONS
+        --------------
+
+        If "do_captive_portal_vendor_tests" is set to "true", then vendor
+        specific captive portal HTTP-based tests will be run.
+
+        If "do_captive_portal_dns_tests" is set to "true", then vendor
+        specific captive portal DNS-based tests will be run.
+
+        If "check_dns_requests" is set to "true", then Ooni-probe will
+        attempt to check that your DNS requests are not being tampered with
+        by a captive portal.
+
+        If "captive_portal" = "yourfilename.txt", then user-specified tests
+        will be run.
+
+        Any combination of the above tests can be run.
+        """
+        report = {}
+
+        log.msg("")
+        log.msg("Running vendor tests...")
+        report['vendor_tests'] = self.run_vendor_tests()
+
+        log.msg("")
+        log.msg("Running vendor DNS-based tests...")
+        report['vendor_dns_tests'] = self.run_vendor_dns_tests()
+
+        log.msg("")
+        log.msg("Checking that DNS requests are not being tampered...")
+        report['check0x20'] = self.check_0x20_to_auth_ns('ooni.nu')
+
+        log.msg("")
+        log.msg("Captive portal test finished!")
+        return report
+
+cp = CaptivePortal(None, None, None)



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