[tor-commits] [ooni-probe/master] Implement basic keyword filtering detection test.

[art at torproject.org]

commit 9e7cf8c35eb29ed56e8bca2da1bc3055a50df1aa
Author: Arturo Filastò <art at fuffa.org>
Date:   Sun Nov 11 15:48:22 2012 +0100

    Implement basic keyword filtering detection test.
    * XXX Does not use the bisection method
    * XXX Does not establish a TCP session, but just sends
---
 nettests/core/keyword_filtering.py |   52 ++++++++++++++++++++++++++++++++++++
 1 files changed, 52 insertions(+), 0 deletions(-)

diff --git a/nettests/core/keyword_filtering.py b/nettests/core/keyword_filtering.py
new file mode 100644
index 0000000..9eec4ff
--- /dev/null
+++ b/nettests/core/keyword_filtering.py
@@ -0,0 +1,52 @@
+# -*- encoding: utf-8 -*-
+#
+# :authors: Arturo Filastò
+# :licence: see LICENSE
+
+from twisted.python import usage
+from twisted.internet import defer
+
+from ooni.utils import log
+from ooni.templates import scapyt
+
+from scapy.all import *
+
+class UsageOptions(usage.Options):
+    optParameters = [
+                    ['backend', 'b', '127.0.0.1:57002', 'Test backend running TCP echo'],
+                    ['timeout', 't', 5, 'Timeout after which to give up waiting for RST packets']
+                    ]
+
+class KeywordFiltering(scapyt.BaseScapyTest):
+    name = "Keyword Filtering detection based on RST packets"
+    author = "Arturo Filastò"
+    version = "0.1"
+
+    usageOptions = UsageOptions
+
+    inputFile = ['file', 'f', None, 
+            'List of keywords to use for censorship testing']
+
+    def test_tcp_keyword_filtering(self):
+        """
+        Places the keyword to be tested in the payload of a TCP packet.
+        XXX need to implement bisection method for enumerating keywords.
+            though this should not be an issue since we are testing all 
+            the keywords in parallel.
+        """
+        def finished(packets):
+            log.debug("Finished running TCP traceroute test on port %s" % port)
+            answered, unanswered = packets
+            self.report['rst_packets'] = []
+            for snd, rcv in answered:
+                # The received packet has the RST flag
+                if rcv[TCP].flags == 4:
+                    self.report['rst_packets'].append(rcv)
+
+        backend_ip, backend_port = self.localOptions['backend']
+        keyword_to_test = str(self.input)
+        packets = IP(dst=backend_ip,id=RandShort())/TCP(dport=backend_port)/keyword_to_test
+        d = self.sr(packets, timeout=timeout)
+        d.addCallback(finished)
+        return d
+