test/test_ip4_irb.py

   1 #!/usr/bin/env python3
   2 """IRB Test Case HLD:
   3
   4 **config**
   5     - L2 MAC learning enabled in l2bd
   6     - 2 routed interfaces untagged, bvi (Bridge Virtual Interface)
   7     - 2 bridged interfaces in l2bd with bvi
   8
   9 **test**
  10     - sending ip4 eth pkts between routed interfaces
  11         - 2 routed interfaces
  12         - 2 bridged interfaces
  13
  14     - 64B, 512B, 1518B, 9200B (ether_size)
  15
  16     - burst of pkts per interface
  17         - 257pkts per burst
  18         - routed pkts hitting different FIB entries
  19         - bridged pkts hitting different MAC entries
  20
  21 **verify**
  22     - all packets received correctly
  23
  24 """
  25
  26 import unittest
  27 from random import choice
  28
  29 from scapy.packet import Raw
  30 from scapy.layers.l2 import Ether
  31 from scapy.layers.inet import IP, UDP
  32
  33 from framework import VppTestCase, VppTestRunner
  34 from vpp_papi import MACAddress
  35 from vpp_l2 import L2_PORT_TYPE
  36
  37
  38 class TestIpIrb(VppTestCase):
  39     """IRB Test Case"""
  40
  41     @classmethod
  42     def setUpClass(cls):
  43         """
  44         #. Create BD with MAC learning enabled and put interfaces to this BD.
  45         #. Configure IPv4 addresses on BVI interface and routed interface.
  46         #. Configure MAC address binding to IPv4 neighbors on bvi0.
  47         #. Configure MAC address on pg2.
  48         #. BVI interface has remote hosts, one half of hosts are
  49            behind pg0 second behind pg1.
  50         """
  51         super(TestIpIrb, cls).setUpClass()
  52
  53         cls.pg_if_packet_sizes = [64, 512, 1518, 9018]  # packet sizes
  54         cls.bd_id = 10
  55         cls.remote_hosts_count = 250
  56
  57         # create 3 pg interfaces, 1 BVI interface
  58         cls.create_pg_interfaces(range(3))
  59         cls.create_bvi_interfaces(1)
  60
  61         cls.interfaces = list(cls.pg_interfaces)
  62         cls.interfaces.extend(cls.bvi_interfaces)
  63
  64         for i in cls.interfaces:
  65             i.admin_up()
  66
  67         # Create BD with MAC learning enabled and put interfaces to this BD
  68         cls.vapi.sw_interface_set_l2_bridge(
  69             rx_sw_if_index=cls.bvi0.sw_if_index,
  70             bd_id=cls.bd_id,
  71             port_type=L2_PORT_TYPE.BVI,
  72         )
  73         cls.vapi.sw_interface_set_l2_bridge(
  74             rx_sw_if_index=cls.pg0.sw_if_index, bd_id=cls.bd_id
  75         )
  76         cls.vapi.sw_interface_set_l2_bridge(
  77             rx_sw_if_index=cls.pg1.sw_if_index, bd_id=cls.bd_id
  78         )
  79
  80         # Configure IPv4 addresses on BVI interface and routed interface
  81         cls.bvi0.config_ip4()
  82         cls.pg2.config_ip4()
  83
  84         # Configure MAC address binding to IPv4 neighbors on bvi0
  85         cls.bvi0.generate_remote_hosts(cls.remote_hosts_count)
  86         cls.bvi0.configure_ipv4_neighbors()
  87         # configure MAC address on pg2
  88         cls.pg2.resolve_arp()
  89
  90         # BVI interface has remote hosts, one half of hosts are behind
  91         # pg0 second behind pg1
  92         half = cls.remote_hosts_count // 2
  93         cls.pg0.remote_hosts = cls.bvi0.remote_hosts[:half]
  94         cls.pg1.remote_hosts = cls.bvi0.remote_hosts[half:]
  95
  96     @classmethod
  97     def tearDownClass(cls):
  98         super(TestIpIrb, cls).tearDownClass()
  99
 100     def tearDown(self):
 101         """Run standard test teardown and log ``show l2patch``,
 102         ``show l2fib verbose``,``show bridge-domain <bd_id> detail``,
 103         ``show ip neighbors``.
 104         """
 105         super(TestIpIrb, self).tearDown()
 106
 107     def show_commands_at_teardown(self):
 108         self.logger.info(self.vapi.cli("show l2patch"))
 109         self.logger.info(self.vapi.cli("show l2fib verbose"))
 110         self.logger.info(self.vapi.cli("show bridge-domain %s detail" % self.bd_id))
 111         self.logger.info(self.vapi.cli("show ip neighbors"))
 112
 113     def create_stream(self, src_ip_if, dst_ip_if, packet_sizes):
 114         pkts = []
 115         for i in range(0, 257):
 116             remote_dst_host = choice(dst_ip_if.remote_hosts)
 117             info = self.create_packet_info(src_ip_if, dst_ip_if)
 118             payload = self.info_to_payload(info)
 119             p = (
 120                 Ether(dst=src_ip_if.local_mac, src=src_ip_if.remote_mac)
 121                 / IP(src=src_ip_if.remote_ip4, dst=remote_dst_host.ip4)
 122                 / UDP(sport=1234, dport=1234)
 123                 / Raw(payload)
 124             )
 125             info.data = p.copy()
 126             size = packet_sizes[(i // 2) % len(packet_sizes)]
 127             self.extend_packet(p, size)
 128             pkts.append(p)
 129         return pkts
 130
 131     def create_stream_l2_to_ip(self, src_l2_if, src_ip_if, dst_ip_if, packet_sizes):
 132         pkts = []
 133         for i in range(0, 257):
 134             info = self.create_packet_info(src_ip_if, dst_ip_if)
 135             payload = self.info_to_payload(info)
 136
 137             host = choice(src_l2_if.remote_hosts)
 138
 139             p = (
 140                 Ether(src=host.mac, dst=src_ip_if.local_mac)
 141                 / IP(src=host.ip4, dst=dst_ip_if.remote_ip4)
 142                 / UDP(sport=1234, dport=1234)
 143                 / Raw(payload)
 144             )
 145
 146             info.data = p.copy()
 147             size = packet_sizes[(i // 2) % len(packet_sizes)]
 148             self.extend_packet(p, size)
 149
 150             pkts.append(p)
 151         return pkts
 152
 153     def verify_capture_l2_to_ip(self, dst_ip_if, src_ip_if, capture):
 154         last_info = dict()
 155         for i in self.interfaces:
 156             last_info[i.sw_if_index] = None
 157
 158         dst_ip_sw_if_index = dst_ip_if.sw_if_index
 159
 160         for packet in capture:
 161             ip = packet[IP]
 162             udp = packet[IP][UDP]
 163             payload_info = self.payload_to_info(packet[IP][UDP][Raw])
 164
 165             self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
 166
 167             next_info = self.get_next_packet_info_for_interface2(
 168                 payload_info.src, dst_ip_sw_if_index, last_info[payload_info.src]
 169             )
 170             last_info[payload_info.src] = next_info
 171             self.assertTrue(next_info is not None)
 172             saved_packet = next_info.data
 173             self.assertTrue(next_info is not None)
 174
 175             # MAC: src, dst
 176             self.assertEqual(packet.src, dst_ip_if.local_mac)
 177             self.assertEqual(packet.dst, dst_ip_if.remote_mac)
 178
 179             # IP: src, dst
 180             host = src_ip_if.host_by_ip4(ip.src)
 181             self.assertIsNotNone(host)
 182             self.assertEqual(ip.dst, saved_packet[IP].dst)
 183             self.assertEqual(ip.dst, dst_ip_if.remote_ip4)
 184
 185             # UDP:
 186             self.assertEqual(udp.sport, saved_packet[UDP].sport)
 187             self.assertEqual(udp.dport, saved_packet[UDP].dport)
 188
 189     def verify_capture(self, dst_ip_if, src_ip_if, capture):
 190         last_info = dict()
 191         for i in self.interfaces:
 192             last_info[i.sw_if_index] = None
 193
 194         dst_ip_sw_if_index = dst_ip_if.sw_if_index
 195
 196         for packet in capture:
 197             ip = packet[IP]
 198             udp = packet[IP][UDP]
 199             payload_info = self.payload_to_info(packet[IP][UDP][Raw])
 200             packet_index = payload_info.index
 201
 202             self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
 203
 204             next_info = self.get_next_packet_info_for_interface2(
 205                 payload_info.src, dst_ip_sw_if_index, last_info[payload_info.src]
 206             )
 207             last_info[payload_info.src] = next_info
 208             self.assertTrue(next_info is not None)
 209             self.assertEqual(packet_index, next_info.index)
 210             saved_packet = next_info.data
 211             self.assertTrue(next_info is not None)
 212
 213             # MAC: src, dst
 214             self.assertEqual(packet.src, dst_ip_if.local_mac)
 215             host = dst_ip_if.host_by_mac(packet.dst)
 216
 217             # IP: src, dst
 218             self.assertEqual(ip.src, src_ip_if.remote_ip4)
 219             self.assertEqual(ip.dst, saved_packet[IP].dst)
 220             self.assertEqual(ip.dst, host.ip4)
 221
 222             # UDP:
 223             self.assertEqual(udp.sport, saved_packet[UDP].sport)
 224             self.assertEqual(udp.dport, saved_packet[UDP].dport)
 225
 226     def test_ip4_irb_1(self):
 227         """IPv4 IRB test 1
 228
 229         Test scenario:
 230             - ip traffic from pg2 interface must ends in both pg0 and pg1
 231             - arp entry present in bvi0 interface for destination IP
 232             - no l2 entry configured, pg0 and pg1 are same
 233         """
 234
 235         stream = self.create_stream(self.pg2, self.bvi0, self.pg_if_packet_sizes)
 236         self.pg2.add_stream(stream)
 237
 238         self.pg_enable_capture(self.pg_interfaces)
 239         self.pg_start()
 240
 241         packet_count = self.get_packet_count_for_if_idx(self.bvi0.sw_if_index)
 242
 243         rcvd1 = self.pg0.get_capture(packet_count)
 244         rcvd2 = self.pg1.get_capture(packet_count)
 245
 246         self.verify_capture(self.bvi0, self.pg2, rcvd1)
 247         self.verify_capture(self.bvi0, self.pg2, rcvd2)
 248
 249         self.assertListEqual(rcvd1.res, rcvd2.res)
 250
 251     def send_and_verify_l2_to_ip(self):
 252         stream1 = self.create_stream_l2_to_ip(
 253             self.pg0, self.bvi0, self.pg2, self.pg_if_packet_sizes
 254         )
 255         stream2 = self.create_stream_l2_to_ip(
 256             self.pg1, self.bvi0, self.pg2, self.pg_if_packet_sizes
 257         )
 258         self.vapi.cli("clear trace")
 259         self.pg0.add_stream(stream1)
 260         self.pg1.add_stream(stream2)
 261
 262         self.pg_enable_capture(self.pg_interfaces)
 263         self.pg_start()
 264
 265         rcvd = self.pg2.get_capture(514)
 266         self.verify_capture_l2_to_ip(self.pg2, self.bvi0, rcvd)
 267
 268     def test_ip4_irb_2(self):
 269         """IPv4 IRB test 2
 270
 271         Test scenario:
 272             - ip traffic from pg0 and pg1 ends on pg2
 273         """
 274         self.send_and_verify_l2_to_ip()
 275
 276         # change the BVI's mac and resed traffic
 277         self.bvi0.set_mac(MACAddress("00:00:00:11:11:33"))
 278
 279         self.send_and_verify_l2_to_ip()
 280         # check it wasn't flooded
 281         self.pg1.assert_nothing_captured(remark="UU Flood")
 282
 283
 284 if __name__ == "__main__":
 285     unittest.main(testRunner=VppTestRunner)