Writing OONI tests

The OONI testing API is heavily influenced and partially based on the python unittest module and twisted.trial.

Test Cases

The atom of OONI Testing is called a Test Case. A test case class may contain multiple Test Methods.

If the test you plan to write is not listed on the Tor OONI trac page, you should add it to the list and then add a description about it following the test template.

Tests are driven by inputs. For every input a new test instance is created, internally the _setUp method is called that is defined inside of test templates, then the setUp method that is overwritable by users.

Gotchas: never call reactor.start of reactor.stop inside of your test method and all will be good.

Inputs

Inputs are what is given as input to every iteration of the Test Case. If you have 100 inputs, then every test case will be run 100 times.

To configure a static set of inputs you should define the ooni.nettest.NetTestCase attribute inputs. The test will be run len(inputs) times. Any iterable object is a valid inputs attribute.

If you would like to have inputs be determined from a user specified input file, then you must set the inputFile attribute. This is an array that specifies what command line option may be used to control this value.

By default the inputProcessor is set to read the file line by line and strip newline characters. To change this behavior you must set the inputProcessor attribute to a function that takes as argument a file descriptor and yield the next item. The default inputProcessor looks like this:

def lineByLine(filename):
    fp = open(filename)
    for x in fp.xreadlines():
        yield x.strip()
    fp.close()

For example, if you wanted to modify inputProcessor to read entries from a CSV file, you could use:

def inputProcessor(self, filename):
    with open(filename) as csvFile:
        reader = DictReader(csvFile)
        for entry in reader:
            yield entry

The inputs iterator is unique per NetTestCase and shared by all its measurement tests. This provides opportunities for each measurement test in a NetTestCase to communicate with the iterator to do things such as adding additional values to be passed to measurement tests as a later input.

Note

Deleting/removing the current item from the inputs iterator will not stop other measurement tests from operating on that input. When a NetTestCase is run a single input is taken from the inputs iterator by the OONI test loader and run against all of the individual measurement tests within that NetTestCase. Removing an input from the inputs iterator during a measurement will not stop that input from being called on all other measurement tests within the NetTestCase.

Here is one example of how you can take advantage of shared inputs iterator when writing your own OONI tests. If you have a list of urls and you want to make sure that you always test the HTTP equivalent of any HTTPS urls provided you can send values back to a custom generator in your postProcessor.

To do this the first thing you would need to create is a URL generator that can accept values sent to it.

::
class UrlGeneratorWithSend(object):
def __init__(self):

“”“Create initial list and set generator state.”“” self.urls = [“http://www.torproject.org”,

self.current = 0

def __iter__(self):
return self
def __next__(self):
try:
cur = self.urls[self.current] self.current += 1 return cur
except IndexError:
raise StopIteration

# Python 2 & 3 generator compatibility next = __next__

def send(self, returned):

“”“Appends a value to self.urls when activated”“” if returned is not None:

print(“Value {0} sent to generator”.format(returned)) self.urls.append(returned)

With this generator created you can now assign it as the inputs to a NetTestCase and send values back to it.

::

class TestUrlList(nettest.NetTestCase):

# Adding custom generator here inputs = UrlGeneratorWithSend()

def postProcessor(self, measurements):

“”“If any HTTPS url’s are passed send back an HTTP url.”“” if re.match(“^https”, self.input):

http_version = re.sub(“https”, “http”, self.input, 1) self.inputs.send(http_version)

return self.report

def test_url(self):
self.report[‘tested’] = [self.input] return defer.succeed(1)

Setup and command line passing

Tests may define the setUp method that will be called every time the Test Case object is instantiated, in here you may place some common logic to all your Test Methods that should be run before any testing occurs.

Command line arguments can be parsed thanks to the twisted twisted.python.usage.UsageOptions class.

You will have to subclass this and define the NetTestCase attribute usageOptions to point to a subclass of this.

class UsageOptions(usage.Options):
  optParameters = [['backend', 'b', 'http://127.0.0.1:57001',
                      'URL of the test backend to use']
                  ]

class MyTestCase(nettest.NetTestCase):
  usageOptions = UsageOptions

  inputFile = ['file', 'f', None, "Some foo file"]
  requiredOptions = ['backend']

  def test_my_test(self):
    self.localOptions['backend']

You will then be able to access the parsed command line arguments via the class attribute localOptions.

The requiredOptions attributes specifies an array of parameters that are required for the test to run properly.

inputFile is a special class attribute that will be used for processing of the inputFile. The filename that is read here will be given to the ooni.nettest.NetTestCase.inputProcessor method that will yield, by default, one line of the file at a time.

Test Methods

These shall be defined inside of your ooni.nettest.NetTestCase subclass. These will be class methods.

All class methods that are prefixed with test_ shall be run. Functions that are relevant to your test should be all lowercase separated by underscore.

To add data to the test report you may write directly to the report object like so:

def test_my_function():
    result = do_something()
    self.report['something'] = result

OONI will then handle the writing of the data to the final test report.

To access the current input you can use the input attribute, for example:

def test_with_input():
    do_something_with_input(self.input)

This will at each iteration over the list of inputs do something with the input.

Test Templates

Test templates assist you in writing tests. They already contain all the common functionality that is useful to running a test of that type. They also take care of writing the data they collect that is relevant to the test run to the report file.

Currently implemented test templates are ooni.templates.scapyt for tests based on Scapy, ooni.templates.tcpt for tests based on TCP, ooni.templates.httpt for tests based on HTTP, and ooni.templates.dnst for tests based on DNS.

Scapy based tests

Scapy based tests will be a subclass of ooni.templates.scapyt.BaseScapyTest.

It provides a wrapper around the scapy send and receive function that will write the sent and received packets to the report with sanitization of the src and destination IP addresses.

It has the same syntax as the Scapy sr function, except that it will return a deferred.

To implement a simple ICMP ping based on this function you can do like so (Taken from nettest.examples.example_scapyt.ExampleICMPPingScapy)

from twisted.python import usage

from scapy.all import IP, ICMP

from ooni.templates import scapyt

class UsageOptions(usage.Options):
    optParameters = [['target', 't', '127.0.0.1', "Specify the target to ping"]]

class ExampleICMPPingScapy(scapyt.BaseScapyTest):
    name = "Example ICMP Ping Test"

    usageOptions = UsageOptions

    def test_icmp_ping(self):
        def finished(packets):
            print packets
            answered, unanswered = packets
            for snd, rcv in answered:
                rcv.show()

        packets = IP(dst=self.localOptions['target'])/ICMP()
        d = self.sr(packets)
        d.addCallback(finished)
        return d

The arguments taken by self.sr() are exactly the same as the scapy send and receive function, the only difference is that instead of using the regular scapy super socket it uses our twisted driven wrapper around it.

Alternatively this test can also be written using the twisted.defer.inlineCallbacks() decorator, that makes it look more similar to regular sequential code.

from twisted.python import usage
from twisted.internet import defer

from scapy.all import IP, ICMP

from ooni.templates import scapyt

class UsageOptions(usage.Options):
    optParameters = [['target', 't', '127.0.0.1', "Specify the target to ping"]]

class ExampleICMPPingScapyYield(scapyt.BaseScapyTest):
    name = "Example ICMP Ping Test"

    usageOptions = UsageOptions

    @defer.inlineCallbacks
    def test_icmp_ping(self):
        packets = IP(dst=self.localOptions['target'])/ICMP()
        answered, unanswered = yield self.sr(packets)
        for snd, rcv in answered:
            rcv.show()

Report Format

###########################################
# OONI Probe Report for Example ICMP Ping Test test
# Thu Nov 22 18:20:43 2012
###########################################
---
{probe_asn: null, probe_cc: null, probe_ip: 127.0.0.1, software_name: ooniprobe, software_version: 0.0.7.1-alpha,
  start_time: 1353601243.0, test_name: Example ICMP Ping Test, test_version: 0.1}
...
---
input: null
report:
  answer_flags: [ipsrc]
  answered_packets:
  - - raw_packet: !!binary |
        RQAAHAEdAAAuAbjKCAgICH8AAAEAAAAAAAAAAA==
      summary: IP / ICMP 8.8.8.8 > 127.0.0.1 echo-reply 0
  sent_packets:
  - - raw_packet: !!binary |
        RQAAHAABAABAAevPfwAAAQgICAgIAPf/AAAAAA==
      summary: IP / ICMP 127.0.0.1 > 8.8.8.8 echo-request 0
test_name: test_icmp_ping
test_started: 1353604843.553605
...

TCP based tests

TCP based tests will subclass ooni.templates.tcpt.TCPTest.

This test template facilitates the sending of TCP payloads to the wire and recording the response.

from twisted.internet.error import ConnectionRefusedError
from ooni.utils import log
from ooni.templates import tcpt

class ExampleTCPT(tcpt.TCPTest):
    def test_hello_world(self):
        def got_response(response):
            print "Got this data %s" % response

        def connection_failed(failure):
            failure.trap(ConnectionRefusedError)
            print "Connection Refused"

        self.address = "127.0.0.1"
        self.port = 57002
        payload = "Hello World!\n\r"
        d = self.sendPayload(payload)
        d.addErrback(connection_failed)
        d.addCallback(got_response)
        return d

The possible failures for a TCP connection are:

twisted.internet.error.NoRouteError that corresponds to errno.ENETUNREACH

twisted.internet.error.ConnectionRefusedError that corresponds to errno.ECONNREFUSED

twisted.internet.error.TCPTimedOutError that corresponds to errno.ETIMEDOUT

Report format

The basic report of a TCP test looks like the following (this is an report generated by running the above example against a TCP echo server).

###########################################
# OONI Probe Report for Base TCP Test test
# Thu Nov 22 18:18:28 2012
###########################################
---
{probe_asn: null, probe_cc: null, probe_ip: 127.0.0.1, software_name: ooniprobe, software_version: 0.0.7.1-alpha,
  start_time: 1353601108.0, test_name: Base TCP Test, test_version: '0.1'}
...
---
input: null
report:
  errors: []
  received: ["Hello World!\n\r"]
  sent: ["Hello World!\n\r"]
test_name: test_hello_world
test_started: 1353604708.705081
...

TODO finish this with more details

HTTP based tests

HTTP based tests will be a subclass of ooni.templates.httpt.HTTPTest.

It provides methods ooni.templates.httpt.HTTPTest.processResponseBody() and ooni.templates.httpt.HTTPTest.processResponseHeaders() for interacting with the response body and headers respectively.

For example, to implement a HTTP test that returns the sha256 hash of the response body (based on nettests.examples.example_httpt):

from ooni.utils import log
from ooni.templates import httpt
from hashlib import sha256

class SHA256HTTPBodyTest(httpt.HTTPTest):
    name = "ChecksumHTTPBodyTest"
    author = "Aaron Gibson"
    version = 0.1

    inputFile = ['url file', 'f', None,
            'List of URLS to perform GET requests to']
    requiredOptions = ['url file']

    def test_http(self):
        if self.input:
            url = self.input
            return self.doRequest(url)
        else:
            raise Exception("No input specified")

    def processResponseBody(self, body):
        body_sha256sum = sha256(body).hexdigest()
        self.report['checksum'] = body_sha256sum

Report format

###########################################
# OONI Probe Report for ChecksumHTTPBodyTest test
# Thu Dec  6 17:31:57 2012
###########################################
---
options:
  collector: null
  help: 0
  logfile: null
  pcapfile: null
  reportfile: null
  resume: 0
  subargs: [-f, hosts]
  test: nettests/examples/example_http_checksum.py
probe_asn: null
probe_cc: null
probe_ip: 127.0.0.1
software_name: ooniprobe
software_version: 0.0.7.1-alpha
start_time: 1354786317.0
test_name: ChecksumHTTPBodyTest
test_version: 0.1
...
---
input: http://www.google.com
report:
  agent: agent
  checksum: d630fa2efd547d3656e349e96ff7af5496889dad959e8e29212af1ff843e7aa1
  requests:
  - request:
      body: null
      headers:
      - - User-Agent
        - - [Opera/9.00 (Windows NT 5.1; U; en), 'Opera 9.0, Windows XP']
      method: GET
      url: http://www.google.com
    response:
      body: '<!doctype html><html ... snip ...  </html>'
      code: 200
      headers:
      - - X-XSS-Protection
        - [1; mode=block]
      - - Set-Cookie
        - ['PREF=ID=fada4216eb3684f9:FF=0:TM=1354800717:LM=1354800717:S=IT-2GCkNAocyXlVa;
            expires=Sat, 06-Dec-2014 13:31:57 GMT; path=/; domain=.google.com', 'NID=66=KWaLbNQumuGuYf0HrWlGm54u9l-DKJwhFCMQXfhQPZM-qniRhmF6QRGXUKXb_8CIUuCOHnyoC5oAX5jWNrsfk-LLJLW530UiMp6hemTtDMh_e6GSiEB4GR3yOP_E0TCN;
            expires=Fri, 07-Jun-2013 13:31:57 GMT; path=/; domain=.google.com; HttpOnly']
      - - Expires
        - ['-1']
      - - Server
        - [gws]
      - - Connection
        - [close]
      - - Cache-Control
        - ['private, max-age=0']
      - - Date
        - ['Thu, 06 Dec 2012 13:31:57 GMT']
      - - P3P
        - ['CP="This is not a P3P policy! See http://www.google.com/support/accounts/bin/answer.py?hl=en&answer=151657
            for more info."']
      - - Content-Type
        - [text/html; charset=UTF-8]
      - - X-Frame-Options
        - [SAMEORIGIN]
  socksproxy: null
test_name: test_http
test_runtime: 0.08298492431640625
test_started: 1354800717.478403
...

DNS based tests

DNS based tests will be a subclass of ooni.templates.dnst.DNSTest.

It provides methods ooni.templates.dnst.DNSTest.performPTRLookup() and ooni.templates.dnst.DNSTest.performALookup()

For example (taken from nettests.examples.example_dnst):

from ooni.templates.dnst import DNSTest

class ExampleDNSTest(DNSTest):
    def test_a_lookup(self):
        def gotResult(result):
            # Result is an array containing all the A record lookup results
            print result

        d = self.performALookup('torproject.org', ('8.8.8.8', 53))
        d.addCallback(gotResult)
        return d

Report format

###########################################
# OONI Probe Report for Base DNS Test test
# Thu Dec  6 17:42:51 2012
###########################################
---
options:
  collector: null
  help: 0
  logfile: null
  pcapfile: null
  reportfile: null
  resume: 0
  subargs: []
  test: nettests/examples/example_dnst.py
probe_asn: null
probe_cc: null
probe_ip: 127.0.0.1
software_name: ooniprobe
software_version: 0.0.7.1-alpha
start_time: 1354786971.0
test_name: Base DNS Test
test_version: 0.1
...
---
input: null
report:
  queries:
  - addrs: [82.195.75.101, 86.59.30.40, 38.229.72.14, 38.229.72.16]
    answers:
    - [<RR name=torproject.org type=A class=IN ttl=782s auth=False>, <A address=82.195.75.101
        ttl=782>]
    - [<RR name=torproject.org type=A class=IN ttl=782s auth=False>, <A address=86.59.30.40
        ttl=782>]
    - [<RR name=torproject.org type=A class=IN ttl=782s auth=False>, <A address=38.229.72.14
        ttl=782>]
    - [<RR name=torproject.org type=A class=IN ttl=782s auth=False>, <A address=38.229.72.16
        ttl=782>]
    query: '[Query(''torproject.org'', 1, 1)]'
    query_type: A
    resolver: [8.8.8.8, 53]
test_name: test_a_lookup
test_runtime: 0.028924942016601562
test_started: 1354801371.980114
...

For a more complex example, see: nettests.blocking.dnsconsistency