Test Templates

ooni.templates.httpt Module

class ooni.templates.httpt.HTTPTest[source]

Bases: ooni.nettest.NetTestCase

A utility class for dealing with HTTP based testing. It provides methods to be overriden for dealing with HTTP based testing. The main functions to look at are processResponseBody and processResponseHeader that are invoked once the headers have been received and once the request body has been received.

To perform requests over Tor you will have to use the special URL schema “shttp”. For example to request / on example.com you will have to do specify as URL “shttp://example.com/”.

XXX all of this requires some refactoring.

addToReport(request, response=None, response_body=None, failure_string=None)[source]

Adds to the report the specified request and response.

Args:

request (dict): A dict describing the request that was made

response (instance): An instance of
:class:twisted.web.client.Response. Note: headers is our modified True Headers version.

failure (instance): An instance of :class:twisted.internet.failure.Failure

baseParameters = [['socksproxy', 's', None, 'Specify a socks proxy to use for requests (ip:port)']]
doRequest(url, method='GET', headers={}, body=None, headers_processor=None, body_processor=None, use_tor=False)[source]

Perform an HTTP request with the specified method and headers.

Args:

url (str): the full URL of the request. The scheme may be either
http, https, or httpo for http over Tor Hidden Service.

Kwargs:

method (str): the HTTP method name to use for the request

headers (dict): the request headers to send

body (str): the request body

headers_processor : a function to be used for processing the HTTP
header responses (defaults to self.processResponseHeaders). This function takes as argument the HTTP headers as a dict.
body_processory: a function to be used for processing the HTTP
response body (defaults to self.processResponseBody). This function takes the response body as an argument.
use_tor (bool): specify if the HTTP request should be done over Tor
or not.
followRedirects = False
name = 'HTTP Test'
processInputs()[source]
processRedirect(location)[source]

Handle a redirection via a 3XX HTTP status code.

Here you may place logic that evaluates the destination that you are being redirected to. Matches against known censor redirects, etc.

Note: if self.followRedirects is set to True, then this method will
never be called. XXX perhaps we may want to hook _handleResponse in RedirectAgent to call processRedirect every time we get redirected.

Args:

location (str): the url that we are being redirected to.
processResponseBody(body)[source]

Overwrite this method if you wish to interact with the response body of every request that is made.

Args:

body (str): The body of the HTTP response
processResponseHeaders(headers)[source]

This should take care of dealing with the returned HTTP headers.

Args:

headers (dict): The returned header fields.
randomizeUA = False
randomize_useragent(request)[source]
version = '0.1.1'
exception ooni.templates.httpt.InvalidSocksProxyOption[source]

Bases: exceptions.Exception

class ooni.templates.httpt.StreamListener(request)[source]

Bases: txtorcon.interface.StreamListenerMixin

stream_succeeded(stream)[source]

ooni.templates.scapyt Module

class ooni.templates.scapyt.BaseScapyTest[source]

Bases: ooni.nettest.NetTestCase

The report of a test run with scapy looks like this:

report:
sent_packets: [
{ ‘raw_packet’: BASE64Encoding of packet, ‘summary’: ‘IP / TCP 192.168.2.66:ftp_data > 8.8.8.8:http S’ }

]

answered_packets: []

baseFlags = [['ipsrc', 's', 'Does *not* check if IP src and ICMP IP citation matches when processing answers'], ['seqack', 'k', 'Check if TCP sequence number and ACK match in the ICMP citation when processing answers'], ['ipid', 'i', 'Check if the IPID matches when processing answers']]
finishedSendReceive(packets)[source]

This gets called when all packets have been sent and received.

name = 'Base Scapy Test'
requiresRoot = True
send(packets, *arg, **kw)[source]

Wrapper around scapy.sendrecv.send for sending of packets at layer 3

sr(packets, timeout=None, *arg, **kw)[source]

Wrapper around scapy.sendrecv.sr for sending and receiving of packets at layer 3.

sr1(packets, *arg, **kw)[source]
version = 0.1
ooni.templates.scapyt.ScapyTest

alias of BaseScapyTest

ooni.templates.dnst Module

class ooni.templates.dnst.DNSTest[source]

Bases: ooni.nettest.NetTestCase

addToReport(query, resolver=None, query_type=None, answers=None, name=None, addrs=None, failure=None)[source]
dnsLookup(hostname, dns_type, dns_server=None)[source]

Performs a DNS lookup and returns the response.

Hostname:is the hostname to perform the DNS lookup on
Dns_type:type of lookup ‘NS’/’A’/’SOA’
Dns_server:is the dns_server that should be used for the lookup as a tuple of ip port (ex. (“127.0.0.1”, 53))
name = 'Base DNS Test'
performALookup(hostname, dns_server=None)[source]

Performs an A lookup and returns an array containg all the dotted quad IP addresses in the response.

Hostname:

is the hostname to perform the A lookup on

Dns_server:

is the dns_server that should be used for the lookup as a tuple of ip port (ex. (“127.0.0.1”, 53))

if None, system dns settings will be used

performNSLookup(hostname, dns_server=None)[source]

Performs a NS lookup and returns an array containg all nameservers in the response.

Hostname:

is the hostname to perform the NS lookup on

Dns_server:

is the dns_server that should be used for the lookup as a tuple of ip port (ex. (“127.0.0.1”, 53))

if None, system dns settings will be used

performPTRLookup(address, dns_server=None)[source]

Does a reverse DNS lookup on the input ip address

Address:

the IP Address as a dotted quad to do a reverse lookup on.

Dns_server:

is the dns_server that should be used for the lookup as a tuple of ip port (ex. (“127.0.0.1”, 53))

if None, system dns settings will be used

performSOALookup(hostname, dns_server=None)[source]

Performs a SOA lookup and returns the response (name,serial).

Hostname:

is the hostname to perform the SOA lookup on

Dns_server:

is the dns_server that should be used for the lookup as a tuple of ip port (ex. (“127.0.0.1”, 53))

if None, system dns settings will be used

queryTimeout = [1]
requiresRoot = False
version = 0.1
ooni.templates.dnst.connectionLost(self, reason=None)[source]

Taken from Twisted 13.1.0 to suppress log.msg printing.

ooni.templates.dnst.representAnswer(answer)[source]

ooni.templates.tcpt Module

class ooni.templates.tcpt.TCPSender[source]

Bases: twisted.internet.protocol.Protocol

dataReceived(data)[source]

We receive data until the total amount of data received reaches that which we have sent. At that point we append the received data to the report and we fire the callback of the test template sendPayload function.

This is used in pair with a TCP Echo server.

The reason why we put the data received inside of an array is that in future we may want to expand this to support state and do something similar to what daphne does, but without the mutation.

XXX Actually daphne will probably be refactored to be a subclass of the TCP Test Template.

sendPayload(payload)[source]

Write the payload to the wire and set the expected size of the payload we are to receive.

Args:

payload: the data to be sent on the wire.
class ooni.templates.tcpt.TCPSenderFactory[source]

Bases: twisted.internet.protocol.Factory

buildProtocol(addr)[source]
class ooni.templates.tcpt.TCPTest[source]

Bases: ooni.nettest.NetTestCase

address = None
name = 'Base TCP Test'
port = None
requiresRoot = False
sendPayload(payload)[source]
timeout = 5
version = '0.1'