Add tests for TUN sockets & bridge_and_sniff()

.travis/test.sh

@@ -12,6 +12,11 @@ then
   SCAPY_SUDO=""
 fi
 
+if [ "$SCAPY_USE_PCAPDNET" = "yes" ]
+then
+  UT_FLAGS+=" -K not_pcapdnet"
+fi
+
 # AES-CCM, ChaCha20Poly1305 and X25519 were added to Cryptography v2.0
 # but the minimal version mandated by scapy is v1.7
 UT_FLAGS+=" -K crypto_advanced"

test/sendsniff.uts

@@ -4,7 +4,7 @@
 
 ############
 ############
-+ Bridge using tap interfaces
++ Test bridge_and_sniff() using tap sockets
 
 ~ tap linux
 
@@ -13,6 +13,7 @@ tap0, tap1 = [TunTapInterface("tap%d" % i, create=True) for i in range(2)]
 from threading import Thread
 
 = Run a sniff thread on the tap1 **interface**
+* It will terminate when 5 IP packets from 1.2.3.4 have been sniffed
 t_sniff = Thread(
     target=sniff,
     kwargs={"iface": "tap1", "count": 5, "prn": Packet.summary,
@@ -21,12 +22,13 @@ t_sniff = Thread(
 t_sniff.start()
 
 = Run a bridge_and_sniff thread between the taps **sockets**
+* It will terminate when 5 IP packets from 1.2.3.4 have been forwarded
 t_bridge = Thread(target=bridge_and_sniff, args=(tap0, tap1),
                   kwargs={"store": False, "count": 5, 'prn': Packet.summary,
                           "lfilter": lambda p: IP in p and p[IP].src == "1.2.3.4"})
 t_bridge.start()
 
-= Send five packets to the tap0 **interface**
+= Send five IP packets from 1.2.3.4 to the tap0 **interface**
 time.sleep(1)
 sendp([Ether(dst=ETHER_BROADCAST) / IP(src="1.2.3.4") / ICMP()], iface="tap0",
       count=5)
@@ -35,9 +37,8 @@ sendp([Ether(dst=ETHER_BROADCAST) / IP(src="1.2.3.4") / ICMP()], iface="tap0",
 t_bridge.join()
 t_sniff.join()
 
-# Same tests, with "NAT" using xfrm function
-
 = Run a sniff thread on the tap1 **interface**
+* It will terminate when 5 IP packets from 2.3.4.5 have been sniffed
 t_sniff = Thread(
     target=sniff,
     kwargs={"iface": "tap1", "count": 5, "prn": Packet.summary,
@@ -46,6 +47,7 @@ t_sniff = Thread(
 t_sniff.start()
 
 = Run a bridge_and_sniff thread between the taps **sockets**
+* It will "NAT" packets from 1.2.3.4 to 2.3.4.5 and will terminate when 5 IP packets have been forwarded
 def nat_1_2(pkt):
     if IP in pkt and pkt[IP].src == "1.2.3.4":
         pkt[IP].src = "2.3.4.5"
@@ -59,7 +61,7 @@ t_bridge = Thread(target=bridge_and_sniff, args=(tap0, tap1),
                           "lfilter": lambda p: IP in p and p[IP].src == "1.2.3.4"})
 t_bridge.start()
 
-= Send five packets to the tap0 **interface**
+= Send five IP packets from 1.2.3.4 to the tap0 **interface**
 time.sleep(1)
 sendp([Ether(dst=ETHER_BROADCAST) / IP(src="1.2.3.4") / ICMP()], iface="tap0",
       count=5)
@@ -69,7 +71,79 @@ t_bridge.join()
 t_sniff.join()
 
 = Delete the tap interfaces
-tap0.close()
-tap1.close()
-tap0.delete()
-tap1.delete()
+del tap0, tap1
+
+
+############
+############
++ Test bridge_and_sniff() using tun sockets
+
+~ tun linux not_pcapdnet
+
+= Create two tun interfaces
+tun0, tun1 = [TunTapInterface("tun%d" % i, create=True) for i in range(2)]
+from threading import Thread
+
+= Run a sniff thread on the tun1 **interface**
+* It will terminate when 5 IP packets from 1.2.3.4 have been sniffed
+t_sniff = Thread(
+    target=sniff,
+    kwargs={"iface": "tun1", "count": 5, "prn": Packet.summary,
+	    "lfilter": lambda p: IP in p and p[IP].src == "1.2.3.4"}
+)
+t_sniff.start()
+
+= Run a bridge_and_sniff thread between the tuns **sockets**
+* It will terminate when 5 IP packets from 1.2.3.4 have been forwarded.
+t_bridge = Thread(target=bridge_and_sniff, args=(tun0, tun1),
+                  kwargs={"store": False, "count": 5, 'prn': Packet.summary,
+                          "xfrm12": lambda pkt: pkt,
+                          "lfilter": lambda p: IP in p and p[IP].src == "1.2.3.4"})
+t_bridge.start()
+
+= Send five IP packets from 1.2.3.4 to the tun0 **interface**
+time.sleep(1)
+conf.route.add(net="1.2.3.4/32", dev="tun0")
+send(IP(src="1.2.3.4", dst="1.2.3.4") / ICMP(), count=5)
+conf.route.delt(net="1.2.3.4/32", dev="tun0")
+
+= Wait for the threads
+t_bridge.join()
+t_sniff.join()
+
+= Run a sniff thread on the tun1 **interface**
+* It will terminate when 5 IP packets from 2.3.4.5 have been sniffed
+t_sniff = Thread(
+    target=sniff,
+    kwargs={"iface": "tun1", "count": 5, "prn": Packet.summary,
+	    "lfilter": lambda p: IP in p and p[IP].src == "2.3.4.5"}
+)
+t_sniff.start()
+
+= Run a bridge_and_sniff thread between the tuns **sockets**
+* It will "NAT" packets from 1.2.3.4 to 2.3.4.5 and will terminate when 5 IP packets have been forwarded
+def nat_1_2(pkt):
+    if IP in pkt and pkt[IP].src == "1.2.3.4":
+        pkt[IP].src = "2.3.4.5"
+        del pkt[IP].chksum
+        return pkt
+    return False
+
+t_bridge = Thread(target=bridge_and_sniff, args=(tun0, tun1),
+                  kwargs={"store": False, "count": 5, 'prn': Packet.summary,
+                          "xfrm12": nat_1_2,
+                          "lfilter": lambda p: IP in p and p[IP].src == "1.2.3.4"})
+t_bridge.start()
+
+= Send five IP packets from 1.2.3.4 to the tun0 **interface**
+time.sleep(1)
+conf.route.add(net="1.2.3.4/32", dev="tun0")
+send(IP(src="1.2.3.4", dst="1.2.3.4") / ICMP(), count=5)
+conf.route.delt(net="1.2.3.4/32", dev="tun0")
+
+= Wait for the threads
+t_bridge.join()
+t_sniff.join()
+
+= Delete the tun interfaces
+del tun0, tun1