Add arp inspection and dhcp snooping tests

framework/python/src/net_orc/network_orchestrator.py

@@ -211,6 +211,7 @@ def _device_discovered(self, mac_addr):
     LOGGER.info(
         f'Device with mac addr {device.mac_addr} has obtained IP address '
         f'{device.ip_addr}')
+    # self._ovs.add_arp_inspection_filter(ip_address=device.ip_addr,mac_address=device.mac_addr)
 
     self._start_device_monitor(device)
 
@@ -682,6 +683,10 @@ def _attach_service_to_network(self, net_module):
                        ' to internet bridge ' + DEVICE_BRIDGE + '. Exiting.')
           sys.exit(1)
 
+  def remove_arp_filters(self):
+    LOGGER.info('Removing ARP inspection filters')
+    self._ovs.delete_arp_inspection_filter()
+
   def restore_net(self):
 
     LOGGER.info('Clearing baseline network')

framework/python/src/net_orc/ovs_control.py

@@ -18,7 +18,9 @@
 DEVICE_BRIDGE = 'tr-d'
 INTERNET_BRIDGE = 'tr-c'
 LOGGER = logger.get_logger('ovs_ctrl')
-
+DEVICER_ARP_COOKIE = '1000'
+UNKNOWN_ARP_COOKIE = '1183'
+CONTAINER_MAC_PREFIX = '9a:02:57:1e:8f'
 
 class OVSControl:
   """OVS Control"""
@@ -51,6 +53,12 @@ def add_port(self, port, bridge_name):
                              add-port {bridge_name} {port}""")
     return success
 
+  def delete_flow(self, bridge_name, flow):
+    # Delete a flow from the bridge using ovs-ofctl commands
+    LOGGER.debug(f'Deleting flow {flow} from bridge: {bridge_name}')
+    success = util.run_command(f'ovs-ofctl del-flows {bridge_name} \'{flow}\'')
+    return success
+
   def get_bridge_ports(self, bridge_name):
     # Get a list of all the ports on a bridge
     response = util.run_command(f'ovs-vsctl list-ports {bridge_name}',
@@ -125,6 +133,13 @@ def create_baseline_net(self, verify=True):
     self.add_flow(bridge_name=DEVICE_BRIDGE,
                   flow='table=0, dl_dst=01:80:c2:00:00:03, actions=flood')
 
+    # Add a DHCP snooping equivalent to the device bridge
+    # ToDo Define these IP's dynamically
+    dhcp_server_primary_ip = '10.10.10.2'
+    dhcp_server_secondary_ip = '10.10.10.3'
+    self.add_dhcp_filters(dhcp_server_primary_ip=dhcp_server_primary_ip,
+      dhcp_server_secondary_ip=dhcp_server_secondary_ip)
+
     # Set ports up
     self.set_bridge_up(DEVICE_BRIDGE)
     self.set_bridge_up(INTERNET_BRIDGE)
@@ -136,6 +151,49 @@ def create_baseline_net(self, verify=True):
     else:
       return None
 
+  def add_dhcp_filters(self,dhcp_server_primary_ip,dhcp_server_secondary_ip):
+
+    # Allow DHCP traffic from primary server
+    allow_primary_dhcp_server = (
+      f'table=0, dl_type=0x800, priority=65535, tp_src=67, tp_dst=68, nw_src={dhcp_server_primary_ip}, actions=normal')
+    self.add_flow(bridge_name=DEVICE_BRIDGE,flow=allow_primary_dhcp_server)
+
+    # Allow DHCP traffic from secondary server
+    allow_secondary_dhcp_server = (
+      f'table=0, dl_type=0x800, priority=65535, tp_src=67, tp_dst=68, nw_src={dhcp_server_secondary_ip}, actions=normal')
+    self.add_flow(bridge_name=DEVICE_BRIDGE,flow=allow_secondary_dhcp_server)
+
+    # Drop DHCP packets not associated with known servers
+    drop_dhcp_flow = 'table=0, dl_type=0x800, priority=0, tp_src=67, tp_dst=68, actions=drop'
+    self.add_flow(bridge_name=DEVICE_BRIDGE,flow=drop_dhcp_flow)
+
+  def add_arp_inspection_filter(self,ip_address,mac_address):
+    # Allow ARP packets with known MAC-to-IP mappings
+    allow_known_arps= f'table=0, cookie={DEVICER_ARP_COOKIE}, priority=65535, arp, arp_tpa={ip_address}, arp_tha={mac_address}, action=normal'
+    self.add_flow(bridge_name=DEVICE_BRIDGE,flow=allow_known_arps)
+
+    DHCP1_MAC = f'{CONTAINER_MAC_PREFIX}:02'
+    DHCP2_MAC = f'{CONTAINER_MAC_PREFIX}:03'
+    DHCP1_IP = '10.10.10.2'
+    DHCP2_IP = '10.10.10.3'
+
+    dhcp_1_arps= f'table=0, priority=65535, arp, arp_tpa={DHCP1_IP}, arp_tha={DHCP1_MAC}, action=normal'
+    dhcp_2_arps= f'table=0, priority=65535, arp, arp_tpa={DHCP2_IP}, arp_tha={DHCP2_MAC}, action=normal'
+    self.add_flow(bridge_name=DEVICE_BRIDGE,flow=dhcp_1_arps)
+    self.add_flow(bridge_name=DEVICE_BRIDGE,flow=dhcp_2_arps)
+
+    # Drop ARP packets with unknown MAC-to-IP mappings
+    drop_unknown_arps = (
+        f'table=0, cookie={UNKNOWN_ARP_COOKIE} priority=100, arp, '
+        f'action=drop'
+    )
+    self.add_flow(bridge_name=DEVICE_BRIDGE,flow=drop_unknown_arps)
+
+  def delete_arp_inspection_filter(self):
+    self.delete_flow(bridge_name=DEVICE_BRIDGE,flow=f'cookie={DEVICER_ARP_COOKIE}/-1')
+    self.delete_flow(bridge_name=DEVICE_BRIDGE,flow=f'cookie={UNKNOWN_ARP_COOKIE}/-1')
+
+
   def delete_bridge(self, bridge_name):
     LOGGER.debug('Deleting OVS Bridge: ' + bridge_name)
     # Delete the bridge using ovs-vsctl commands

framework/python/src/test_orc/test_orchestrator.py

@@ -339,6 +339,9 @@ def _run_test_module(self, module):
 
       client = docker.from_env()
 
+      # if module.name == 'connection':
+      #   self._net_orc.remove_arp_filters()
+
       module.container = client.containers.run(
           module.image_name,
           auto_remove=True,

modules/test/conn/conf/module_config.json

@@ -13,6 +13,18 @@
       "timeout": 1800
     },
     "tests": [
+      {
+        "name": "connection.switch.arp_inspection",
+        "test_description": "The device implements ARP correctly as per RFC826",
+        "expected_behavior": "Device continues to operate correctly when ARP inspection is enabled on the switch.  No functionality is lost with ARP inspection enabled.",
+        "required_result": "Required"
+      },
+      {
+        "name": "connection.switch.dhcp_snooping",
+        "test_description": "The device operates as a DHCP client and operates correctly when DHCP snooping is enabled on a switch.",
+        "expected_behavior": "Device continues to operate correctly when DHCP snooping is enabled on the switch.",
+        "required_result": "Required"
+      },
       {
         "name": "connection.dhcp_address",
         "test_description": "The device under test has received an IP address from the DHCP server and responds to an ICMP echo (ping) request",
@@ -34,7 +46,7 @@
       },
       {
         "name": "connection.mac_oui",
-        "test_description": "The device under test hs a MAC address prefix that is registered against a known manufacturer.",
+        "test_description": "The device under test has a MAC address prefix that is registered against a known manufacturer.",
         "expected_behavior": "The MAC address prefix is registered in the IEEE Organizationally Unique Identifier database.",
         "required_result": "Required",
         "recommendations": [

modules/test/conn/python/src/connection_module.py

@@ -15,7 +15,7 @@
 import util
 import time
 import traceback
-from scapy.all import rdpcap, DHCP, Ether, IPv6, ICMPv6ND_NS
+from scapy.all import rdpcap, DHCP, ARP, Ether, IPv6, ICMPv6ND_NS
 from test_module import TestModule
 from dhcp1.client import Client as DHCPClient1
 from dhcp2.client import Client as DHCPClient2
@@ -74,6 +74,62 @@ def __init__(self, module):
     # response = self.dhcp1_client.set_dhcp_range('10.10.10.20','10.10.10.30')
     # print("Set Range: " + str(response))
 
+  def _connection_switch_arp_inspection(self):
+    LOGGER.info('Running connection.switch.arp_inspection')
+
+    no_arp = True
+
+    # Read all the pcap files
+    packets = rdpcap(STARTUP_CAPTURE_FILE) + rdpcap(MONITOR_CAPTURE_FILE)
+    for packet in packets:
+
+      # We are not interested in packets unless they are ARP packets
+      if not ARP in packet:
+        continue
+
+      # We are only interested in packets from the device
+      if packet.src != self._device_mac:
+        continue
+
+      # Get the ARP packet
+      arp_packet = packet[ARP]
+      no_arp = False
+
+      # Check MAC address matches IP address
+      if (arp_packet.hwsrc == self._device_mac and
+          arp_packet.psrc != self._device_ipv4_addr):
+        LOGGER.info(f'Bad ARP packet detected for MAC: {self._device_mac}')
+        LOGGER.info(f'ARP packet IP {arp_packet.psrc} does not match {self._device_ipv4_addr}')
+        return False, 'Device is sending false ARP response'
+
+    if no_arp:
+      return None, 'No ARP packets from the device found'
+
+    return True, 'Device uses ARP'
+
+  def _connection_switch_dhcp_snooping(self):
+    LOGGER.info('Running connection.switch.dhcp_snooping')
+
+    disallowed_dhcp_types = [2, 4, 5, 6, 9, 10, 12, 13, 15, 17]
+
+    # Read all the pcap files
+    packets = rdpcap(STARTUP_CAPTURE_FILE) + rdpcap(MONITOR_CAPTURE_FILE)
+    for packet in packets:
+
+      # We are not interested in packets unless they are DHCP packets
+      if not DHCP in packet:
+        continue
+
+      # We are only interested in packets from the device
+      if packet.src != self._device_mac:
+        continue
+
+      dhcp_type = self._get_dhcp_type(packet)
+      if dhcp_type in disallowed_dhcp_types:
+        return False, 'Device has sent disallowed DHCP message'
+
+    return True, 'Device does not act as a DHCP server'
+
   def _connection_private_address(self, config):
     LOGGER.info('Running connection.private_address')
     return self._run_subnet_test(config)
@@ -141,7 +197,7 @@ def _connection_single_ip(self):
       if DHCP in packet:
         for option in packet[DHCP].options:
           # message-type, option 3 = DHCPREQUEST
-          if 'message-type' in option and option[1] == 3:
+          if self._get_dhcp_type(packet) == 3:
             mac_address = packet[Ether].src
             LOGGER.info('DHCPREQUEST detected MAC address: ' + mac_address)
             if not mac_address.startswith(TR_CONTAINER_MAC_PREFIX):
@@ -160,6 +216,11 @@ def _connection_single_ip(self):
     else:
       return result, 'Device is using multiple IP addresses'
 
+  def _get_dhcp_type(self, packet):
+    for option in packet[DHCP].options:
+      if 'message-type' in option:
+        return option[1]
+
   def _connection_target_ping(self):
     LOGGER.info('Running connection.target_ping')