connections: Add "punt-policies"-like mechanisem for the manipulation feature

Author: roykuper13

README.md

@@ -8,6 +8,8 @@ as Python objects.
 Moreover, linux-switch let you manipulate packets before the network switch forwards
 them (see example below). Thus, External binaries/applications that performs
 any logic on packets (for example - VLAN Hopping, NAT, etc) can be tested using linux-switch.
+Also, linux-switch provide a "Punt-Policies"-like mechanisem, which gives you the option
+to filter traffic that the manipulation routine will be recieve.
 
 
 ## Description
@@ -22,6 +24,11 @@ switch device, meaning:
 3. When connecting `Device`s to the `Switch`, the connection type must be specified (access or trunk).
 When using trunk - the switch and the device will send/recieve tagged packet.
 When using access - they'll send untagged packets.
+4. The switch have "Punt-Policies", which means only filtered packets will be forwarded
+to the manipulation routine. The punt-policies feature introduce a `duplicate` mode. When set,
+packets that are filtered (using the punt policies) will be processed by both manipulation routine
+and switch. When not set, packets will be processed by the manipultion routine only, so that routine
+can, for example, drop packets!
 
 
 ## Example
@@ -85,6 +92,7 @@ switch.disconnect_device(dev2)
 switch.term()
 ```
 
-### Manipulations
+### Manipulations(VLAN Hopping) and "Punt-Policies"
 
-For VLAN-Hopping example, see tests/test_manipulation.py (test_vlan_hopping)
+A very good example for VLAN-Hopping and Punt-Policies can be seen
+in tests/test_manipulation (test_vlan_hopping_and_punt_policies).

src/connections.py

@@ -6,8 +6,8 @@
 from contextlib import suppress
 import concurrent.futures
 
-from src.util import PortType, add_vlan_tag, fix_checksums
-from src.manipulation import ManipulateArgs, default_manipulation_cb
+from src.util import PortType, add_vlan_tag, fix_checksums, apply_bpf_filter
+from src.manipulation import ManipulateArgs
 
 
 DeviceEntry = namedtuple('DeviceEntry', [
@@ -21,6 +21,7 @@
 IP_MAX_SIZE = 65535
 
 MAX_WORKERS = 5
+NO_FILTER = None
 
 
 class Connections(object):
@@ -31,7 +32,9 @@ def __init__(self):
         self._devices = list()
         self._message_queue = None
         self._executers = None
-        self.manipulate_cb = default_manipulation_cb
+
+        self._manipulate_cb = None
+        self._manipulate_filter = NO_FILTER
 
     async def _read_message_queue(self):
         while True:
@@ -43,11 +46,15 @@ async def _process_packet(self, packet, src_device_entry):
         # to tag the packet.
         should_inject_raw = False
 
-        if self.manipulate_cb is not None:
-            packet, should_inject_raw = await self._event_loop.run_in_executor(
-                self._executers,
-                self.manipulate_cb,
-                ManipulateArgs(packet, src_device_entry.port_type, src_device_entry.vlan))
+        if self._manipulate_cb is not None:
+            # We'll manipulate if there's no filter or if there's
+            # filter and the packet matches the filter
+            if self._manipulate_filter == NO_FILTER or \
+                    apply_bpf_filter(packet, self._manipulate_filter):
+                packet, should_inject_raw = await self._event_loop.run_in_executor(
+                    self._executers,
+                    self._manipulate_cb,
+                    ManipulateArgs(packet, src_device_entry.port_type, src_device_entry.vlan))
 
         src_vlan = self._get_vlan_by_mac(Ether(packet).src)
         if src_vlan is None:
@@ -139,6 +146,17 @@ def _remove_device_entry(self, dev_to_remove):
 
         self._event_loop.call_soon_threadsafe(_remove_device_entry, self, dev_to_remove)
 
+    def set_manipulation(self, cb, bpf_filter):
+        ''' Assumes cb is in valid form '''
+        def __set_manipulation(cb, bpf_filter):
+            self._manipulate_cb = cb
+            self._manipulate_filter = bpf_filter
+
+        # set_manipulation is called from the main thread.
+        # Since we're affecting the event loop thread, we should call_soon_threadsafe,
+        # so the call will be synchronized.
+        self._event_loop.call_soon_threadsafe(__set_manipulation, cb, bpf_filter)
+
     def start_connections_thread(self):
         self._connections_thread.start()
 

src/switch.py

@@ -1,4 +1,4 @@
-from src.connections import Connections
+from src.connections import Connections, NO_FILTER
 from src.util import shell_run_and_check, PortType
 from src.manipulation import validate_manipulation_cb
 from src.exception import (NamespaceCreationException,
@@ -89,9 +89,9 @@ def disconnect_device(self, dev):
 
         dev.term()
 
-    def set_manipulation(self, cb):
+    def set_manipulation(self, cb, bpf_filter=NO_FILTER):
         if validate_manipulation_cb(cb):
-            self._connections.manipulate_cb = cb
+            self._connections.set_manipulation(cb, bpf_filter)
             return True
 
         return False

src/util.py

@@ -1,7 +1,8 @@
+import os
 import enum
 import subprocess
 from netaddr import IPAddress
-from scapy.all import NoPayload, Ether, Dot1Q, Raw, IP, TCP, UDP
+from scapy.all import NoPayload, Ether, Dot1Q, Raw, IP, TCP, UDP, sniff
 
 
 DOT1Q_ETH_TYPE = 0x8100
@@ -69,3 +70,18 @@ def fix_checksums(packet):
 
     packet = packet.__class__(bytes(packet))
     return Raw(packet).load
+
+
+def apply_bpf_filter(packet, filter):
+    # TODO: this is horrible. Unfortunately scapy's `sniff` (which uses tcpdump)
+    # always outputs annoying stuff to stderr, so we have to redirect it to /dev/null.
+    # For now we'll settle with this, until this issue will be fixed in scapy.
+    stderr_backup = os.dup(2)
+
+    dev_null = os.open('/dev/null', os.O_WRONLY)
+    os.dup2(dev_null, 2)
+
+    filtered = len(sniff(offline=Ether(packet), filter=filter).res) == 1
+
+    os.dup2(stderr_backup, 2)
+    return filtered

tests/test_manipulation.py

@@ -1,5 +1,5 @@
 import threading
-from scapy.all import Ether, IP, Raw
+from scapy.all import Ether, IP, ICMP, Raw
 
 from src.device import Device
 from src.manipulation import ManipulateArgs, ManipulateRet
@@ -17,15 +17,12 @@
 def forward_d1_to_d3(arg: ManipulateArgs) -> ManipulateRet:
     '''
     Replace addresses of d2 with d4, so the packet will be hopped
-    to the second vlan
+    to the second vlan.
     '''
     global GLOBAL_D3_MAC
     global GLOBAL_D4_MAC
     packet = Ether(arg.packet)
 
-    if not (IP in packet and packet[IP].dst == D2_ADDR and packet[IP].src == D1_ADDR):
-        return ManipulateRet(arg.packet, False)
-
     packet[IP].src = D4_ADDR
     packet[IP].dst = D3_ADDR
 
@@ -35,16 +32,19 @@ def forward_d1_to_d3(arg: ManipulateArgs) -> ManipulateRet:
     return ManipulateRet(Raw(packet).load, False)
 
 
-def test_vlan_hopping(switch):
-    '''
+def test_vlan_hopping_and_punt_policies(switch):
+    """
     In this test, d1 is in vlan 20, and it'll try to connect
     to d3, which is part of vlan 10. The manipulation callback works follow:
     For each packet that its source is d1 and destined to d2,
     the callback will change the packet so it'll look like the source is d4,
     and it's destined to d3. The switch will think that the packet came from
     d4 who's part of vlan 10, so it'll forward the packet to d3 (who's also in vlan 10).
     That is an example of a VLAN hopping.
-    '''
+
+    This test also shows the behaviour of the "punt-policies":
+    Only packets from d1 to d2 will be passed to the manipulation callback.
+    """
     global GLOBAL_D3_MAC
     global GLOBAL_D4_MAC
 
@@ -54,27 +54,37 @@ def test_vlan_hopping(switch):
     d3 = Device('c', D3_ADDR, '255.255.255.0')
     d4 = Device('d', D4_ADDR, '255.255.255.0')
 
-    assert switch.connect_device_trunk(d1, 20)
-    assert switch.connect_device_access(d2, 20)
+    switch.connect_device_trunk(d1, 20)
+    switch.connect_device_access(d2, 20)
 
-    assert switch.connect_device_trunk(d3, 10)
-    assert switch.connect_device_access(d4, 10)
+    switch.connect_device_trunk(d3, 10)
+    switch.connect_device_access(d4, 10)
 
     GLOBAL_D3_MAC = d3.get_mac
     GLOBAL_D4_MAC = d4.get_mac
 
-    switch.set_manipulation(forward_d1_to_d3)
+    # Setting the manipulation routine, and punt-policies
+    switch.set_manipulation(forward_d1_to_d3, 'src host {} && dst host {}'.format(
+        D1_ADDR, D2_ADDR))
 
     def _run_listening_nc(dev):
-        out = dev.run_from_namespace('timeout 5s nc -lu 0.0.0.0 4444')
+        out = dev.run_from_namespace('timeout 7s nc -lu 0.0.0.0 4444')
         assert 'hello' == out
 
+    # d3 is going to listen
     t = threading.Thread(target=_run_listening_nc, args=(d3, ))
     t.start()
 
-    out = d1.run_from_namespace('bash -c "echo hello | nc -u 192.168.250.2 4444 -w 3"')
+    # and d1 will connect to it
+    out = d1.run_from_namespace('bash -c "echo hello | nc -u 192.168.250.2 4444 -w 5"')
     assert '' == out
 
+    # Note that the manipulation callback modifies every packet that it recieves,
+    # But it won't recieve this packet (and its reply) since they won't match the
+    # bpf filter that we set before. That's why we expect an echo-reply.
+    out = d3.run_from_namespace('ping -c 1 192.168.1.2')
+    assert '1 packets transmitted, 1 received' in out
+
     switch.disconnect_device(d1)
     switch.disconnect_device(d2)
     switch.disconnect_device(d3)