[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]
[tor-commits] [ooni-probe/master] Move captive portal tests to proper subdirectories
commit c85c627a7d12ccfeda7010fed6d88a58241e5978
Author: Arturo Filastò <art@xxxxxxxxxxxxxx>
Date: Mon Jul 9 16:35:13 2012 +0200
Move captive portal tests to proper subdirectories
---
assets/captive_portal_tests.txt | 4 -
ooni/assets/captive_portal_tests.txt | 4 +
ooni/plugins/captiveportal.py | 651 ++++++++++++++++++++++++++++++++++
tests/captiveportal.py | 651 ----------------------------------
4 files changed, 655 insertions(+), 655 deletions(-)
diff --git a/assets/captive_portal_tests.txt b/assets/captive_portal_tests.txt
deleted file mode 100644
index 7007411..0000000
--- a/assets/captive_portal_tests.txt
+++ /dev/null
@@ -1,4 +0,0 @@
-
-http://ooni.nu, Open Observatory of Network Interference, 200
-https://w2.eff.org/Censorship/Internet_censorship_bills/barlow_0296.declaration, let us now take our leave of them, 200
-https://www.torproject.org, Defend yourself against network surveillance and traffic analysis, 200
diff --git a/ooni/assets/captive_portal_tests.txt b/ooni/assets/captive_portal_tests.txt
new file mode 100644
index 0000000..7007411
--- /dev/null
+++ b/ooni/assets/captive_portal_tests.txt
@@ -0,0 +1,4 @@
+
+http://ooni.nu, Open Observatory of Network Interference, 200
+https://w2.eff.org/Censorship/Internet_censorship_bills/barlow_0296.declaration, let us now take our leave of them, 200
+https://www.torproject.org, Defend yourself against network surveillance and traffic analysis, 200
diff --git a/ooni/plugins/captiveportal.py b/ooni/plugins/captiveportal.py
new file mode 100644
index 0000000..ffa86c4
--- /dev/null
+++ b/ooni/plugins/captiveportal.py
@@ -0,0 +1,651 @@
+# -*- 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 plugoo.assets import Asset
+from plugoo.tests import Test
+
+try:
+ from dns import resolver
+except ImportError:
+ print "The dnspython module was not found. https://crate.io/packages/dnspython/";
+
+try:
+ from gevent import monkey
+ monkey.patch_all(socket=True, dns=False, time=True, select=False, thread=True,
+ os=True, ssl=True, httplib=False, aggressive=True)
+except ImportError:
+ print "The gevent module was not found. https://crate.io/packages/gevent/";
+
+__plugoo__ = "captiveportal"
+__desc__ = "Captive portal detection test"
+
+
+class CaptivePortalAsset(Asset):
+ """
+ Parses captive_portal_tests.txt into an Asset.
+ """
+ def __init__(self, file=None):
+ self = Asset.__init__(self, file)
+
+ def parse_line(self, line):
+ self = Asset.parse_line(self, line)
+ return line.replace('\n', '').split(', ')
+
+class CaptivePortal(Test):
+ """
+ Compares content and status codes of HTTP responses, and attempts
+ to determine if content has been altered.
+ """
+ def __init__(self, ooni, name=__plugoo__):
+ Test.__init__(self, ooni, name)
+ self.default_ua = ooni.config.tests.default_ua
+
+ 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.
+ """
+ log = self.logger
+
+ if headers is None:
+ default_ua = self.default_ua
+ 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.info("Fuzzy HTTP content comparison for experiment URL")
+ log.info("'%s'" % experimental_url)
+ log.info("does not match!")
+ return False, response_code, response_headers
+ else:
+ log.info("Fuzzy HTTP content comparison of experiment URL")
+ log.info("'%s'" % experimental_url)
+ log.info("and the expected control result yielded a match.")
+ return True, response_code, response_headers
+ else:
+ if str(response_content) != str(control_result):
+ log.info("HTTP content comparison of experiment URL")
+ log.info("'%s'" % experimental_url)
+ log.info("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.
+ """
+ log = self.logger
+
+ 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.info("DNS resolution for %s returned NXDOMAIN" % hn)
+ response.append('NXDOMAIN')
+ except resolver.NXDOMAIN:
+ log.info("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.
+ """
+ log = self.logger
+
+ 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.info("DNS comparison of control '%s' does not" % control_address)
+ log.info("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 = self.logger
+ log.info("")
+ log.info("Testing random capitalization of DNS queries...")
+ log.info("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.info("Capitalization in DNS queries and responses match.")
+ name_match = True
+ else:
+ log.info("The random capitalization '%s' used in" % hostname)
+ log.info("DNS queries to that hostname's authoritative")
+ log.info("nameservers does not match the capitalization in")
+ log.info("the response.")
+ name_match = False
+
+ if len(set(serials)) == 1:
+ log.info("Start of Authority serial numbers all match.")
+ serial_match = True
+ else:
+ log.info("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.info("Your DNS queries do not appear to be tampered.")
+ return ret
+ elif name_match or serial_match:
+ log.info("Something is tampering with your DNS queries.")
+ return ret
+ elif not name_match and not serial_match:
+ log.info("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.
+ """
+ log = self.logger
+
+ 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.
+ """
+ log = self.logger
+
+ if hostname_count is None:
+ hostname_count = 3
+
+ log.info("Generating random hostnames...")
+ log.info("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.info("Strangely, DNS resolution of the random hostname")
+ log.info("%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.info("All %d random hostnames properly resolved to NXDOMAIN."
+ % hostname_count)
+ return True, relative_complement
+ elif (len(intersection) == 1) and (len(r) > 1):
+ log.info("Something odd happened. Some random hostnames correctly")
+ log.info("resolved to NXDOMAIN, but several others resolved to")
+ log.info("to the following addresses: %s" % relative_complement)
+ return False, relative_complement
+ elif (len(intersection) == 0) and (len(r) == 1):
+ log.info("All random hostnames resolved to the IP address ")
+ log.info("'%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.
+ """
+ log = self.logger
+ subtest = "Google Chrome DNS-based"
+
+ log.info("")
+ log.info("Running the Google Chrome DNS-based captive portal test...")
+
+ gmatch, google_dns_result = self.compare_random_hostnames(3, 10)
+
+ if gmatch:
+ log.info("Google Chrome DNS-based captive portal test did not")
+ log.info("detect a captive portal.")
+ return google_dns_result
+ else:
+ log.info("Google Chrome DNS-based captive portal test believes")
+ log.info("you are in a captive portal, or else something very")
+ log.info("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.
+ """
+ log = self.logger
+ subtest = "Microsoft NCSI DNS-based"
+
+ log.info("")
+ log.info("Running the Microsoft NCSI DNS-based captive portal")
+ log.info("test...")
+
+ msmatch, ms_dns_result = self.dns_resolve_match("dns.msftncsi.com",
+ "131.107.255.255")
+ if msmatch:
+ log.info("Microsoft NCSI DNS-based captive portal test did not")
+ log.info("detect a captive portal.")
+ return ms_dns_result
+ else:
+ log.info("Microsoft NCSI DNS-based captive portal test ")
+ log.info("believes you are in a captive portal.")
+ return ms_dns_result
+
+ def run_vendor_dns_tests(self):
+ """
+ Run the vendor DNS tests.
+ """
+ self.google_dns_cp_test()
+ self.ms_dns_cp_test()
+
+ return
+
+ 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
+ log = self.logger
+
+ def compare_content(status_func, fuzzy, experiment_url, control_result,
+ control_code, headers, test_name):
+ log.info("")
+ log.info("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.info("The %s test was unable to detect" % test_name)
+ log.info("a captive portal.")
+ else:
+ log.info("The %s test shows that your network" % test_name)
+ log.info("is filtered.")
+
+ for vt in vendor_tests:
+ 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":
+ compare_content(sm, False, *args)
+
+ elif test_name == "Apple HTTP Captive Portal":
+ compare_content(sm, True, *args)
+
+ elif test_name == "W3 Captive Portal":
+ compare_content(snm, True, *args)
+
+ else:
+ log.warn("Ooni is trying to run an undefined CP vendor test.")
+
+ def experiment(self, *a, **kw):
+ """
+ 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.
+ """
+ if (os.path.isfile(os.path.join(self.config.main.assetdir,
+ self.config.tests.captive_portal))):
+ kw['data'].append(None)
+ kw['data'].append('user-defined')
+
+ experiment_url = kw['data'][0]
+ control_result = kw['data'][1]
+ control_code = kw['data'][2]
+ ua = kw['data'][3]
+ test_name = kw['data'][4]
+
+ cm = self.http_content_match_fuzzy_opt
+ sm = self.http_status_code_match
+ snm = self.http_status_code_no_match
+
+ log = self.logger
+
+ if test_name == "user-defined":
+ log.info("Running %s test for '%s'..." % (test_name, 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.info("The %s test for '%s'" % (test_name, experiment_url))
+ log.info("was unable to detect a captive portal.")
+ return True, test_name
+ elif status_match and not content_match:
+ log.info("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 True, test_name
+ else:
+ log.info("Found modified content on '%s'," % experiment_url)
+ log.info("which could indicate a captive portal.")
+
+ return False, test_name
+ else:
+ log.info("The content comparison test for ")
+ log.info("'%s'" % experiment_url)
+ log.info("shows that your HTTP traffic is filtered.")
+ return False, test_name
+
+ else:
+ log.warn("Ooni is trying to run an undefined captive portal test.")
+ return False, test_name
+
+
+def run(ooni):
+ """
+ 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.
+ """
+ config = ooni.config
+ log = ooni.logger
+
+ assets = []
+ if (os.path.isfile(os.path.join(config.main.assetdir,
+ config.tests.captive_portal))):
+ assets.append(CaptivePortalAsset(os.path.join(config.main.assetdir,
+ config.tests.captive_portal)))
+
+ captiveportal = CaptivePortal(ooni)
+ log.info("Starting captive portal test...")
+ captiveportal.run(assets, {'index': 1})
+
+ if config.tests.do_captive_portal_vendor_tests:
+ log.info("")
+ log.info("Running vendor tests...")
+ captiveportal.run_vendor_tests()
+
+ if config.tests.do_captive_portal_vendor_dns_tests:
+ log.info("")
+ log.info("Running vendor DNS-based tests...")
+ captiveportal.run_vendor_dns_tests()
+
+ if config.tests.check_dns_requests:
+ log.info("")
+ log.info("Checking that DNS requests are not being tampered...")
+ captiveportal.check_0x20_to_auth_ns('ooni.nu')
+
+ log.info("")
+ log.info("Captive portal test finished!")
diff --git a/tests/captiveportal.py b/tests/captiveportal.py
deleted file mode 100644
index ffa86c4..0000000
--- a/tests/captiveportal.py
+++ /dev/null
@@ -1,651 +0,0 @@
-# -*- 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 plugoo.assets import Asset
-from plugoo.tests import Test
-
-try:
- from dns import resolver
-except ImportError:
- print "The dnspython module was not found. https://crate.io/packages/dnspython/";
-
-try:
- from gevent import monkey
- monkey.patch_all(socket=True, dns=False, time=True, select=False, thread=True,
- os=True, ssl=True, httplib=False, aggressive=True)
-except ImportError:
- print "The gevent module was not found. https://crate.io/packages/gevent/";
-
-__plugoo__ = "captiveportal"
-__desc__ = "Captive portal detection test"
-
-
-class CaptivePortalAsset(Asset):
- """
- Parses captive_portal_tests.txt into an Asset.
- """
- def __init__(self, file=None):
- self = Asset.__init__(self, file)
-
- def parse_line(self, line):
- self = Asset.parse_line(self, line)
- return line.replace('\n', '').split(', ')
-
-class CaptivePortal(Test):
- """
- Compares content and status codes of HTTP responses, and attempts
- to determine if content has been altered.
- """
- def __init__(self, ooni, name=__plugoo__):
- Test.__init__(self, ooni, name)
- self.default_ua = ooni.config.tests.default_ua
-
- 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.
- """
- log = self.logger
-
- if headers is None:
- default_ua = self.default_ua
- 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.info("Fuzzy HTTP content comparison for experiment URL")
- log.info("'%s'" % experimental_url)
- log.info("does not match!")
- return False, response_code, response_headers
- else:
- log.info("Fuzzy HTTP content comparison of experiment URL")
- log.info("'%s'" % experimental_url)
- log.info("and the expected control result yielded a match.")
- return True, response_code, response_headers
- else:
- if str(response_content) != str(control_result):
- log.info("HTTP content comparison of experiment URL")
- log.info("'%s'" % experimental_url)
- log.info("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.
- """
- log = self.logger
-
- 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.info("DNS resolution for %s returned NXDOMAIN" % hn)
- response.append('NXDOMAIN')
- except resolver.NXDOMAIN:
- log.info("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.
- """
- log = self.logger
-
- 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.info("DNS comparison of control '%s' does not" % control_address)
- log.info("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 = self.logger
- log.info("")
- log.info("Testing random capitalization of DNS queries...")
- log.info("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.info("Capitalization in DNS queries and responses match.")
- name_match = True
- else:
- log.info("The random capitalization '%s' used in" % hostname)
- log.info("DNS queries to that hostname's authoritative")
- log.info("nameservers does not match the capitalization in")
- log.info("the response.")
- name_match = False
-
- if len(set(serials)) == 1:
- log.info("Start of Authority serial numbers all match.")
- serial_match = True
- else:
- log.info("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.info("Your DNS queries do not appear to be tampered.")
- return ret
- elif name_match or serial_match:
- log.info("Something is tampering with your DNS queries.")
- return ret
- elif not name_match and not serial_match:
- log.info("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.
- """
- log = self.logger
-
- 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.
- """
- log = self.logger
-
- if hostname_count is None:
- hostname_count = 3
-
- log.info("Generating random hostnames...")
- log.info("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.info("Strangely, DNS resolution of the random hostname")
- log.info("%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.info("All %d random hostnames properly resolved to NXDOMAIN."
- % hostname_count)
- return True, relative_complement
- elif (len(intersection) == 1) and (len(r) > 1):
- log.info("Something odd happened. Some random hostnames correctly")
- log.info("resolved to NXDOMAIN, but several others resolved to")
- log.info("to the following addresses: %s" % relative_complement)
- return False, relative_complement
- elif (len(intersection) == 0) and (len(r) == 1):
- log.info("All random hostnames resolved to the IP address ")
- log.info("'%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.
- """
- log = self.logger
- subtest = "Google Chrome DNS-based"
-
- log.info("")
- log.info("Running the Google Chrome DNS-based captive portal test...")
-
- gmatch, google_dns_result = self.compare_random_hostnames(3, 10)
-
- if gmatch:
- log.info("Google Chrome DNS-based captive portal test did not")
- log.info("detect a captive portal.")
- return google_dns_result
- else:
- log.info("Google Chrome DNS-based captive portal test believes")
- log.info("you are in a captive portal, or else something very")
- log.info("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.
- """
- log = self.logger
- subtest = "Microsoft NCSI DNS-based"
-
- log.info("")
- log.info("Running the Microsoft NCSI DNS-based captive portal")
- log.info("test...")
-
- msmatch, ms_dns_result = self.dns_resolve_match("dns.msftncsi.com",
- "131.107.255.255")
- if msmatch:
- log.info("Microsoft NCSI DNS-based captive portal test did not")
- log.info("detect a captive portal.")
- return ms_dns_result
- else:
- log.info("Microsoft NCSI DNS-based captive portal test ")
- log.info("believes you are in a captive portal.")
- return ms_dns_result
-
- def run_vendor_dns_tests(self):
- """
- Run the vendor DNS tests.
- """
- self.google_dns_cp_test()
- self.ms_dns_cp_test()
-
- return
-
- 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
- log = self.logger
-
- def compare_content(status_func, fuzzy, experiment_url, control_result,
- control_code, headers, test_name):
- log.info("")
- log.info("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.info("The %s test was unable to detect" % test_name)
- log.info("a captive portal.")
- else:
- log.info("The %s test shows that your network" % test_name)
- log.info("is filtered.")
-
- for vt in vendor_tests:
- 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":
- compare_content(sm, False, *args)
-
- elif test_name == "Apple HTTP Captive Portal":
- compare_content(sm, True, *args)
-
- elif test_name == "W3 Captive Portal":
- compare_content(snm, True, *args)
-
- else:
- log.warn("Ooni is trying to run an undefined CP vendor test.")
-
- def experiment(self, *a, **kw):
- """
- 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.
- """
- if (os.path.isfile(os.path.join(self.config.main.assetdir,
- self.config.tests.captive_portal))):
- kw['data'].append(None)
- kw['data'].append('user-defined')
-
- experiment_url = kw['data'][0]
- control_result = kw['data'][1]
- control_code = kw['data'][2]
- ua = kw['data'][3]
- test_name = kw['data'][4]
-
- cm = self.http_content_match_fuzzy_opt
- sm = self.http_status_code_match
- snm = self.http_status_code_no_match
-
- log = self.logger
-
- if test_name == "user-defined":
- log.info("Running %s test for '%s'..." % (test_name, 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.info("The %s test for '%s'" % (test_name, experiment_url))
- log.info("was unable to detect a captive portal.")
- return True, test_name
- elif status_match and not content_match:
- log.info("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 True, test_name
- else:
- log.info("Found modified content on '%s'," % experiment_url)
- log.info("which could indicate a captive portal.")
-
- return False, test_name
- else:
- log.info("The content comparison test for ")
- log.info("'%s'" % experiment_url)
- log.info("shows that your HTTP traffic is filtered.")
- return False, test_name
-
- else:
- log.warn("Ooni is trying to run an undefined captive portal test.")
- return False, test_name
-
-
-def run(ooni):
- """
- 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.
- """
- config = ooni.config
- log = ooni.logger
-
- assets = []
- if (os.path.isfile(os.path.join(config.main.assetdir,
- config.tests.captive_portal))):
- assets.append(CaptivePortalAsset(os.path.join(config.main.assetdir,
- config.tests.captive_portal)))
-
- captiveportal = CaptivePortal(ooni)
- log.info("Starting captive portal test...")
- captiveportal.run(assets, {'index': 1})
-
- if config.tests.do_captive_portal_vendor_tests:
- log.info("")
- log.info("Running vendor tests...")
- captiveportal.run_vendor_tests()
-
- if config.tests.do_captive_portal_vendor_dns_tests:
- log.info("")
- log.info("Running vendor DNS-based tests...")
- captiveportal.run_vendor_dns_tests()
-
- if config.tests.check_dns_requests:
- log.info("")
- log.info("Checking that DNS requests are not being tampered...")
- captiveportal.check_0x20_to_auth_ns('ooni.nu')
-
- log.info("")
- log.info("Captive portal test finished!")
_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits