Merge pull request #509 from guedou/codecov_8

[coverage] Better unit tests for inet.py

scapy/layers/inet.py

@@ -387,14 +387,6 @@ class IP(Packet, IPTools):
         l = self.len - (self.ihl << 2)
         return s[:l],s[l:]
 
-    def send(self, s, slp=0):
-        for p in self:
-            try:
-                s.sendto(str(p), (p.dst,0))
-            except socket.error, msg:
-                log_runtime.error(msg)
-            if slp:
-                time.sleep(slp)
     def route(self):
         dst = self.dst
         if isinstance(dst,Gen):
@@ -829,7 +821,15 @@ def fragment(pkt, fragsize=1480):
             lst.append(q)
     return lst
 
+@conf.commands.register
 def overlap_frag(p, overlap, fragsize=8, overlap_fragsize=None):
+    """Build overlapping fragments to bypass NIPS
+
+p:                the original packet
+overlap:          the overlapping data
+fragsize:         the fragment size of the packet
+overlap_fragsize: the fragment size of the overlapping packet"""
+
     if overlap_fragsize is None:
         overlap_fragsize = fragsize
     q = p.copy()
@@ -985,11 +985,12 @@ def _packetlist_timeskew_graph(self, ip, **kargs):
 
     # Build a list of tuples (creation_time, replied_timestamp)
     c = []
+    tsf = ICMPTimeStampField("", None)
     for p in b:
         opts = p.getlayer(TCP).options
         for o in opts:
             if o[0] == "Timestamp":
-                c.append((p.time,o[1][0]))
+                c.append((p.time, tsf.any2i("", o[1][0])))
 
     # Stop if the list is empty
     if not c:
@@ -1586,6 +1587,8 @@ class TCP_client(Automaton):
 ## Reporting stuff ##
 #####################
 
+
+@conf.commands.register
 def report_ports(target, ports):
     """portscan a target and output a LaTeX table
 report_ports(target, ports) -> string"""
@@ -1608,8 +1611,13 @@ report_ports(target, ports) -> string"""
     return rep
 
 
-
+@conf.commands.register
 def IPID_count(lst, funcID=lambda x:x[1].id, funcpres=lambda x:x[1].summary()):
+    """Identify IP id values classes in a list of packets
+
+lst:      a list of packets
+funcID:   a function that returns IP id values
+funcpres: a function used to summarize packets"""
     idlst = map(funcID, lst)
     idlst.sort()
     classes = [idlst[0]]+map(lambda x:x[1],filter(lambda (x,y): abs(x-y)>50, map(lambda x,y: (x,y),idlst[:-1], idlst[1:])))
@@ -1620,6 +1628,7 @@ def IPID_count(lst, funcID=lambda x:x[1].id, funcpres=lambda x:x[1].summary()):
         print "%5i" % id, pr
     
     
+@conf.commands.register
 def fragleak(target,sport=123, dport=123, timeout=0.2, onlyasc=0):
     load = "XXXXYYYYYYYYYY"
 #    getmacbyip(target)
@@ -1670,6 +1679,8 @@ def fragleak(target,sport=123, dport=123, timeout=0.2, onlyasc=0):
     except KeyboardInterrupt:
         pass
 
+
+@conf.commands.register
 def fragleak2(target, timeout=0.4, onlyasc=0):
     found={}
     try:

test/regression.uts

@@ -202,12 +202,13 @@ atol("www.secdev.org") == 3642339845
 * Those test are here mainly to check nothing has been broken
 * and to catch Exceptions
 
-= Building some packets packet
+= Building some packets
 ~ basic IP TCP UDP NTP LLC SNAP Dot11
 IP()/TCP()
 Ether()/IP()/UDP()/NTP()
 Dot11()/LLC()/SNAP()/IP()/TCP()/"XXX"
 IP(ttl=25)/TCP(sport=12, dport=42)
+IP().summary()
 
 = Manipulating some packets
 ~ basic IP TCP
@@ -7386,3 +7387,119 @@ r1 == '@\x04\x08\x08\x08\x08'
 
 r2 = b.dec(r1)[0]
 r2.val == '8.8.8.8'
+
+
++ inet.py
+
+= IPv4 - ICMPTimeStampField
+test = ICMPTimeStampField("test", None)
+value = test.any2i("", "07:28:28.07")
+value == 26908070
+test.i2repr("", value) == '7:28:28.70'
+
+= IPv4 - UDP null checksum
+IP(str(IP()/UDP()/Raw("\xff\xff\x01\x6a")))[UDP].chksum == 0xFFFF
+
+= IPv4 - (IP|UDP|TCP|ICMP)Error
+query = IP(dst="192.168.0.1", src="192.168.0.254", ttl=1)/UDP()/DNS()
+answer = IP(dst="192.168.0.254", src="192.168.0.2", ttl=1)/ICMP()/IPerror(dst="192.168.0.1", src="192.168.0.254", ttl=0)/UDPerror()/DNS()
+
+query = IP(dst="192.168.0.1", src="192.168.0.254", ttl=1)/UDP()/DNS()
+answer = IP(dst="192.168.0.254", src="192.168.0.2")/ICMP(type=11)/IPerror(dst="192.168.0.1", src="192.168.0.254", ttl=0)/UDPerror()/DNS()
+answer.answers(query) == True
+
+query = IP(dst="192.168.0.1", src="192.168.0.254", ttl=1)/TCP()
+answer = IP(dst="192.168.0.254", src="192.168.0.2")/ICMP(type=11)/IPerror(dst="192.168.0.1", src="192.168.0.254", ttl=0)/TCPerror()
+
+answer.answers(query) == True
+
+query = IP(dst="192.168.0.1", src="192.168.0.254", ttl=1)/ICMP()/"scapy"
+answer = IP(dst="192.168.0.254", src="192.168.0.2")/ICMP(type=11)/IPerror(dst="192.168.0.1", src="192.168.0.254", ttl=0)/ICMPerror()/"scapy"
+
+answer.answers(query) == True
+
+= IPv4 - utilities
+l = overlap_frag(IP(dst="1.2.3.4")/ICMP()/("AB"*8), ICMP()/("CD"*8))
+len(l) == 6
+[len(str(IP.payload)) for p in l] == [38, 38, 38, 38, 38, 38]
+[(p.frag, p.flags.MF) for p in [IP(str(p)) for p in l]] == [(0, True), (1, True), (2, True), (0, True), (1, True), (2, False)]
+
+= IPv4 - traceroute utilities
+ip_ttl = [("192.168.0.%d" % i, i) for i in xrange(1, 10)]
+
+tr_packets = [ (IP(dst="192.168.0.1", src="192.168.0.254", ttl=ttl)/TCP(options=[("Timestamp", "00:00:%.2d.00" % ttl)])/"scapy",
+                IP(dst="192.168.0.254", src=ip)/ICMP(type=11)/IPerror(dst="192.168.0.1", src="192.168.0.254", ttl=0)/TCPerror()/"scapy")
+               for (ip, ttl) in ip_ttl ]
+
+tr = TracerouteResult(tr_packets)
+tr.get_trace() == {'192.168.0.1': {1: ('192.168.0.1', False), 2: ('192.168.0.2', False), 3: ('192.168.0.3', False), 4: ('192.168.0.4', False), 5: ('192.168.0.5', False), 6: ('192.168.0.6', False), 7: ('192.168.0.7', False), 8: ('192.168.0.8', False), 9: ('192.168.0.9', False)}}
+
+
+result_show = ""
+def test_show():
+    def write(s):
+        global result_show
+        result_show += s
+    mock_stdout = mock.Mock()
+    mock_stdout.write = write
+    sys.stdout = mock_stdout
+    tr = TracerouteResult(tr_packets)
+    tr.show()
+    sys.stdout = sys.__stdout__
+    expected = "  192.168.0.1:tcp80  \n"
+    expected += "1 192.168.0.1     11 \n"
+    expected += "2 192.168.0.2     11 \n"
+    expected += "3 192.168.0.3     11 \n"
+    expected += "4 192.168.0.4     11 \n"
+    expected += "5 192.168.0.5     11 \n"
+    expected += "6 192.168.0.6     11 \n"
+    expected += "7 192.168.0.7     11 \n"
+    expected += "8 192.168.0.8     11 \n"
+    expected += "9 192.168.0.9     11 \n"
+    index_result = result_show.index("1")
+    index_expected = expected.index("1")
+    assert(result_show[index_result:] == expected[index_expected:])
+
+test_show()
+
+
+@mock.patch("scapy.layers.inet.plt")
+def test_timeskew_graph(mock_plt):
+    def fake_plot(data, **kwargs):
+        return data
+    mock_plt.plot = fake_plot
+    srl = SndRcvList([(a, a) for a in [IP(str(p[0])) for p in tr_packets]])
+    ret = srl.timeskew_graph("192.168.0.254")
+    len(ret) == 9
+    ret[0][1] == 0.0
+
+test_timeskew_graph()
+
+
+tr = TracerouteResult(tr_packets)
+saved_AS_resolver = conf.AS_resolver
+conf.AS_resolver = None
+tr.make_graph()
+print len(tr.graphdef) == 491
+print tr.graphdef.startswith("digraph trace {") == True
+print '"192.168.0.9" ->' in tr.graphdef == True
+conf.AS_resolver = conf.AS_resolver
+
+= IPv4 - reporting
+
+result_IPID_count = ""
+def test_IPID_count():
+    def write(s):
+        global result_IPID_count
+        result_IPID_count += s
+    mock_stdout = mock.Mock()
+    mock_stdout.write = write
+    sys.stdout = mock_stdout
+    random.seed(0x2807)
+    IPID_count([(IP()/UDP(), IP(id=random.randint(0, 65535))/UDP()) for i in range(3)])
+    sys.stdout = sys.__stdout__
+    lines = result_IPID_count.split("\n")
+    print len(lines) == 5
+    print lines[0] == "Probably 3 classes: [4613, 53881, 58437]"
+
+test_IPID_count()