Futher refactoring TCP observer functions

pathspider/observer/tcp.py

@@ -1,10 +1,43 @@
 
+TCP_CWR = 0x80
+TCP_ECE = 0x40
+TCP_ACK = 0x10
+TCP_SYN = 0x02
+
+TCP_SEC = ( TCP_SYN | TCP_ECE | TCP_CWR )
+TCP_SAEW = (TCP_SYN | TCP_ACK | TCP_ECE | TCP_CWR)
+TCP_SAE = (TCP_SYN | TCP_ACK | TCP_ECE)
+
 def tcp_setup(rec, ip):
+    rec['fwd_syn_flags'] = None
+    rec['rec_syn_flags'] = None
+
     rec['fwd_fin'] = False
-    rec['fwd_rst'] = False
     rec['rev_fin'] = False
+    rec['fwd_rst'] = False
     rec['rev_rst'] = False
 
+    rec['tcp_connected'] = False
+
+    return True
+
+def tcp_handshake(rec, tcp, rev):
+    flags = tcp.flags
+
+    if flags & TCP_SYN:
+        if rev == 0:
+            rec['fwd_syn_flags'] = flags
+        if rev == 1:
+            rec['rev_syn_flags'] = flags
+
+    # TODO: This test could perhaps be improved upon.
+    # This test is intended to catch the completion of the 3WHS.
+    if (not tcp.connected and rev == 0 and
+       rec['fwd_syn_flags'] is not None and
+       rec['rev_syn_flags'] is not None and
+       flags & TCP_ACK):
+        rec['tcp_connected'] = True
+
     return True
 
 def tcp_complete(rec, tcp, rev): # pylint: disable=W0612,W0613

pathspider/plugins/dscp.py

@@ -16,6 +16,7 @@
 from pathspider.observer import basic_count
 
 from pathspider.observer.tcp import tcp_setup
+from pathspider.observer.tcp import tcp_handshake
 from pathspider.observer.tcp import tcp_complete
 
 Connection = collections.namedtuple("Connection", ["client", "port", "state"])
@@ -30,8 +31,13 @@
 ## Chain functions
 
 def dscp_setup(rec, ip):
-    rec['fwd_dscp'] = None
-    rec['rev_dscp'] = None
+    if ip.tcp:
+        # we'll only care about these if it's TCP
+        rec['fwd_syn_dscp'] = None
+        rec['rev_syn_dscp'] = None
+
+    rec['fwd_data_dscp'] = None
+    rec['rev_data_dscp'] = None
     return True
 
 def dscp_extract(rec, ip, rev):
@@ -136,7 +142,7 @@ def create_observer(self):
                             new_flow_chain=[basic_flow, tcp_setup, dscp_setup],
                             ip4_chain=[basic_count, dscp_extract],
                             ip6_chain=[basic_count, dscp_extract],
-                            tcp_chain=[tcp_complete])
+                            tcp_chain=[tcp_handshake, tcp_complete])
         except:
             logger.error("Observer not cooperating, abandon ship")
             traceback.print_exc()

pathspider/plugins/ecn.py

@@ -16,7 +16,10 @@
 from pathspider.observer import basic_flow
 from pathspider.observer import basic_count
 from pathspider.observer.tcp import tcp_setup
+from pathspider.observer.tcp import tcp_handshake
 from pathspider.observer.tcp import tcp_complete
+from pathspider.observer.tcp import TCP_SAE
+from pathspider.observer.tcp import TCP_SAEW
 
 Connection = collections.namedtuple("Connection", ["client", "port", "state", "tstart"])
 SpiderRecord = collections.namedtuple("SpiderRecord", ["ip", "rport", "port",
@@ -29,35 +32,15 @@
 
 USER_AGENT = "pathspider"
 
-TCP_CWR = 0x80
-TCP_ECE = 0x40
-TCP_ACK = 0x10
-TCP_SYN = 0x02
-
-TCP_SEC = ( TCP_SYN | TCP_ECE | TCP_CWR )
-TCP_SAEW = (TCP_SYN | TCP_ACK | TCP_ECE | TCP_CWR)
-TCP_SAE = (TCP_SYN | TCP_ACK | TCP_ECE)
-
 ## Chain functions
 
-def ecnsetup(rec, ip):
+def ecn_setup(rec, ip):
     fields = ['fwd_ez', 'fwd_eo', 'fwd_ce', 'rev_ez', 'rev_eo', 'rev_ce']
     for field in fields:
         rec[field] = False
     return True
 
-def ecnflags(rec, tcp, rev):
-    flags = tcp.flags
-
-    if flags & TCP_SYN:
-        if rev == 0:
-            rec['fwd_syn_flags'] = flags
-        if rev == 1:
-            rec['rev_syn_flags'] = flags
-
-    return True
-
-def ecncode(rec, ip, rev):
+def ecn_code(rec, ip, rev):
     EZ = 0x01
     EO = 0x02
     CE = 0x03
@@ -171,10 +154,10 @@ def create_observer(self):
         logger.info("Creating observer")
         try:
             return Observer(self.libtrace_uri,
-                            new_flow_chain=[basic_flow, tcp_setup, ecnsetup],
-                            ip4_chain=[basic_count, ecncode],
-                            ip6_chain=[basic_count, ecncode],
-                            tcp_chain=[ecnflags, tcp_complete])
+                            new_flow_chain=[basic_flow, tcp_setup, ecn_setup],
+                            ip4_chain=[basic_count, ecn_code],
+                            ip6_chain=[basic_count, ecn_code],
+                            tcp_chain=[tcp_handshake, tcp_complete])
         except:
             logger.error("Observer not cooperating, abandon ship")
             traceback.print_exc()
@@ -207,10 +190,10 @@ def combine_flows(self, flow):
                 cond_conn = 'ecn.connectivity.offline'
 
             if flows[1]['rev_syn_flags'] & TCP_SAEW == TCP_SAE:
-                negotiated = False
+                negotiated = True
                 conditions.append('ecn.negotiated')
             else:
-                negotiated = True
+                negotiated = False
                 conditions.append('ecn.not_negotiated')
 
             if flows[1]['rev_ez']: