4 from util import ip4_range, reassemble4
6 from framework import VppTestCase
7 from asfframework import VppTestRunner
8 from template_bd import BridgeDomain
10 from scapy.layers.l2 import Ether
11 from scapy.layers.l2 import ARP
12 from scapy.packet import Raw, bind_layers
13 from scapy.layers.inet import IP, UDP
14 from scapy.layers.vxlan import VXLAN
17 from vpp_ip_route import VppIpRoute, VppRoutePath
18 from vpp_vxlan_tunnel import VppVxlanTunnel
19 from vpp_ip import INVALID_INDEX
20 from vpp_neighbor import VppNeighbor
23 class TestVxlan(BridgeDomain, VppTestCase):
26 def __init__(self, *args):
27 BridgeDomain.__init__(self)
28 VppTestCase.__init__(self, *args)
30 def encapsulate(self, pkt, vni):
32 Encapsulate the original payload frame by adding VXLAN header with its
33 UDP, IP and Ethernet fields
36 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
37 / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
38 / UDP(sport=self.dport, dport=self.dport, chksum=0)
39 / VXLAN(vni=vni, flags=self.flags)
43 def ip_range(self, start, end):
44 """range of remote ip's"""
45 return ip4_range(self.pg0.remote_ip4, start, end)
47 def encap_mcast(self, pkt, src_ip, src_mac, vni):
49 Encapsulate the original payload frame by adding VXLAN header with its
50 UDP, IP and Ethernet fields
53 Ether(src=src_mac, dst=self.mcast_mac)
54 / IP(src=src_ip, dst=self.mcast_ip4)
55 / UDP(sport=self.dport, dport=self.dport, chksum=0)
56 / VXLAN(vni=vni, flags=self.flags)
60 def decapsulate(self, pkt):
62 Decapsulate the original payload frame by removing VXLAN header
64 # check if is set I flag
65 self.assertEqual(pkt[VXLAN].flags, int("0x8", 16))
66 return pkt[VXLAN].payload
68 # Method for checking VXLAN encapsulation.
70 def check_encapsulation(self, pkt, vni, local_only=False, mcast_pkt=False):
71 # TODO: add error messages
72 # Verify source MAC is VPP_MAC and destination MAC is MY_MAC resolved
74 self.assertEqual(pkt[Ether].src, self.pg0.local_mac)
77 self.assertEqual(pkt[Ether].dst, self.pg0.remote_mac)
79 self.assertEqual(pkt[Ether].dst, type(self).mcast_mac)
80 # Verify VXLAN tunnel source IP is VPP_IP and destination IP is MY_IP.
81 self.assertEqual(pkt[IP].src, self.pg0.local_ip4)
84 self.assertEqual(pkt[IP].dst, self.pg0.remote_ip4)
86 self.assertEqual(pkt[IP].dst, type(self).mcast_ip4)
87 # Verify UDP destination port is VXLAN 4789, source UDP port could be
89 self.assertEqual(pkt[UDP].dport, self.dport)
91 self.assert_udp_checksum_valid(pkt)
93 self.assertEqual(pkt[VXLAN].vni, vni)
96 def create_vxlan_flood_test_bd(cls, vni, n_ucast_tunnels, port):
97 # Create 10 ucast vxlan tunnels under bd
99 ip_range_end = ip_range_start + n_ucast_tunnels
100 next_hop_address = cls.pg0.remote_ip4
101 for dest_ip4 in ip4_range(next_hop_address, ip_range_start, ip_range_end):
102 # add host route so dest_ip4 will not be resolved
107 [VppRoutePath(next_hop_address, INVALID_INDEX)],
114 src=cls.pg0.local_ip4,
121 cls.vapi.sw_interface_set_l2_bridge(r.sw_if_index, bd_id=vni)
124 def add_del_shared_mcast_dst_load(cls, port, is_add):
126 add or del tunnels sharing the same mcast dst
127 to test vxlan ref_count mechanism
129 n_shared_dst_tunnels = 20
131 vni_end = vni_start + n_shared_dst_tunnels
132 for vni in range(vni_start, vni_end):
135 src=cls.pg0.local_ip4,
144 if r.sw_if_index == 0xFFFFFFFF:
145 raise ValueError("bad sw_if_index: ~0")
147 r.remove_vpp_config()
150 def add_shared_mcast_dst_load(cls, port):
151 cls.add_del_shared_mcast_dst_load(port=port, is_add=1)
154 def del_shared_mcast_dst_load(cls, port):
155 cls.add_del_shared_mcast_dst_load(port=port, is_add=0)
158 def add_del_mcast_tunnels_load(cls, port, is_add):
160 add or del tunnels to test vxlan stability
162 n_distinct_dst_tunnels = 200
164 ip_range_end = ip_range_start + n_distinct_dst_tunnels
165 for dest_ip4 in ip4_range(cls.mcast_ip4, ip_range_start, ip_range_end):
166 vni = bytearray(socket.inet_pton(socket.AF_INET, dest_ip4))[3]
169 src=cls.pg0.local_ip4,
179 r.remove_vpp_config()
182 def add_mcast_tunnels_load(cls, port):
183 cls.add_del_mcast_tunnels_load(port=port, is_add=1)
186 def del_mcast_tunnels_load(cls, port):
187 cls.add_del_mcast_tunnels_load(port=port, is_add=0)
189 # Class method to start the VXLAN test case.
190 # Overrides setUpClass method in VppTestCase class.
191 # Python try..except statement is used to ensure that the tear down of
192 # the class will be executed even if exception is raised.
193 # @param cls The class pointer.
196 super(TestVxlan, cls).setUpClass()
201 # Create 2 pg interfaces.
202 cls.create_pg_interfaces(range(4))
203 for pg in cls.pg_interfaces:
206 # Configure IPv4 addresses on VPP pg0.
209 # Resolve MAC address for VPP's IP address on pg0.
210 cls.pg0.resolve_arp()
212 # Our Multicast address
213 cls.mcast_ip4 = "239.1.1.1"
214 cls.mcast_mac = util.mcast_ip_to_mac(cls.mcast_ip4)
220 def tearDownClass(cls):
221 super(TestVxlan, cls).tearDownClass()
224 super(TestVxlan, self).setUp()
226 def createVxLANInterfaces(self, port=4789):
227 # Create VXLAN VTEP on VPP pg0, and put vxlan_tunnel0 and pg1
231 self.single_tunnel_vni = 0x12345
232 self.single_tunnel_bd = 1
235 src=self.pg0.local_ip4,
236 dst=self.pg0.remote_ip4,
239 vni=self.single_tunnel_vni,
242 self.vapi.sw_interface_set_l2_bridge(
243 rx_sw_if_index=r.sw_if_index, bd_id=self.single_tunnel_bd
245 self.vapi.sw_interface_set_l2_bridge(
246 rx_sw_if_index=self.pg1.sw_if_index, bd_id=self.single_tunnel_bd
249 # Setup vni 2 to test multicast flooding
250 self.n_ucast_tunnels = 10
251 self.mcast_flood_bd = 2
252 self.create_vxlan_flood_test_bd(
253 self.mcast_flood_bd, self.n_ucast_tunnels, self.dport
257 src=self.pg0.local_ip4,
262 vni=self.mcast_flood_bd,
265 self.vapi.sw_interface_set_l2_bridge(
266 rx_sw_if_index=r.sw_if_index, bd_id=self.mcast_flood_bd
268 self.vapi.sw_interface_set_l2_bridge(
269 rx_sw_if_index=self.pg2.sw_if_index, bd_id=self.mcast_flood_bd
272 # Add and delete mcast tunnels to check stability
273 self.add_shared_mcast_dst_load(self.dport)
274 self.add_mcast_tunnels_load(self.dport)
275 self.del_shared_mcast_dst_load(self.dport)
276 self.del_mcast_tunnels_load(self.dport)
278 # Setup vni 3 to test unicast flooding
279 self.ucast_flood_bd = 3
280 self.create_vxlan_flood_test_bd(
281 self.ucast_flood_bd, self.n_ucast_tunnels, self.dport
283 self.vapi.sw_interface_set_l2_bridge(
284 rx_sw_if_index=self.pg3.sw_if_index, bd_id=self.ucast_flood_bd
287 # Set scapy listen custom port for VxLAN
288 bind_layers(UDP, VXLAN, dport=self.dport)
290 def encap_big_packet(self):
291 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1500, 0, 0, 0])
294 Ether(src="00:00:00:00:00:02", dst="00:00:00:00:00:01")
295 / IP(src="4.3.2.1", dst="1.2.3.4")
296 / UDP(sport=20000, dport=10000)
297 / Raw(b"\xa5" * 1450)
300 self.pg1.add_stream([frame])
302 self.pg0.enable_capture()
306 # Pick first received frame and check if it's correctly encapsulated.
307 out = self.pg0.get_capture(2)
309 pkt = reassemble4(out)
311 self.check_encapsulation(pkt, self.single_tunnel_vni)
313 payload = self.decapsulate(pkt)
315 # self.assert_eq_pkts(payload, frame)
318 Tests with default port (4789)
321 def test_decap(self):
322 """Decapsulation test
325 self.createVxLANInterfaces()
326 super(TestVxlan, self).test_decap()
328 def test_encap(self):
329 """Encapsulation test
332 self.createVxLANInterfaces()
333 super(TestVxlan, self).test_encap()
335 def test_encap_big_packet(self):
336 """Encapsulation test send big frame from pg1
337 Verify receipt of encapsulated frames on pg0
339 self.createVxLANInterfaces()
340 self.encap_big_packet()
342 def test_ucast_flood(self):
343 """Unicast flood test
346 self.createVxLANInterfaces()
347 super(TestVxlan, self).test_ucast_flood()
349 def test_mcast_flood(self):
350 """Multicast flood test
353 self.createVxLANInterfaces()
354 super(TestVxlan, self).test_mcast_flood()
356 def test_mcast_rcv(self):
357 """Multicast receive test
360 self.createVxLANInterfaces()
361 super(TestVxlan, self).test_mcast_rcv()
364 Tests with custom port
367 def test_decap_custom_port(self):
368 """Decapsulation test custom port
371 self.createVxLANInterfaces(1111)
372 super(TestVxlan, self).test_decap()
374 def test_encap_custom_port(self):
375 """Encapsulation test custom port
378 self.createVxLANInterfaces(1111)
379 super(TestVxlan, self).test_encap()
381 def test_ucast_flood_custom_port(self):
382 """Unicast flood test custom port
385 self.createVxLANInterfaces(1111)
386 super(TestVxlan, self).test_ucast_flood()
388 def test_mcast_flood_custom_port(self):
389 """Multicast flood test custom port
392 self.createVxLANInterfaces(1111)
393 super(TestVxlan, self).test_mcast_flood()
395 def test_mcast_rcv_custom_port(self):
396 """Multicast receive test custom port
399 self.createVxLANInterfaces(1111)
400 super(TestVxlan, self).test_mcast_rcv()
402 # Method to define VPP actions before tear down of the test case.
403 # Overrides tearDown method in VppTestCase class.
404 # @param self The object pointer.
407 super(TestVxlan, self).tearDown()
409 def show_commands_at_teardown(self):
410 self.logger.info(self.vapi.cli("show bridge-domain 1 detail"))
411 self.logger.info(self.vapi.cli("show bridge-domain 2 detail"))
412 self.logger.info(self.vapi.cli("show bridge-domain 3 detail"))
413 self.logger.info(self.vapi.cli("show vxlan tunnel"))
416 class TestVxlan2(VppTestCase):
417 """VXLAN Test Case"""
420 super(TestVxlan2, self).setUp()
422 # Create 2 pg interfaces.
423 self.create_pg_interfaces(range(4))
424 for pg in self.pg_interfaces:
427 # Configure IPv4 addresses on VPP pg0.
428 self.pg0.config_ip4()
429 self.pg0.resolve_arp()
432 super(TestVxlan2, self).tearDown()
434 def test_xconnect(self):
435 """VXLAN source address not local"""
438 # test the broken configuration of a VXLAN tunnel whose
439 # source address is not local ot the box. packets sent
440 # through the tunnel should be dropped
442 t = VppVxlanTunnel(self, src="10.0.0.5", dst=self.pg0.local_ip4, vni=1000)
446 self.vapi.sw_interface_set_l2_xconnect(
447 t.sw_if_index, self.pg1.sw_if_index, enable=1
449 self.vapi.sw_interface_set_l2_xconnect(
450 self.pg1.sw_if_index, t.sw_if_index, enable=1
454 Ether(src="00:11:22:33:44:55", dst="00:00:00:11:22:33")
455 / IP(src="4.3.2.1", dst="1.2.3.4")
456 / UDP(sport=20000, dport=10000)
457 / Raw(b"\xa5" * 1450)
460 rx = self.send_and_assert_no_replies(self.pg1, [p])
463 class TestVxlanL2Mode(VppTestCase):
464 """VXLAN Test Case"""
467 super(TestVxlanL2Mode, self).setUp()
469 # Create 2 pg interfaces.
470 self.create_pg_interfaces(range(2))
471 for pg in self.pg_interfaces:
474 # Configure IPv4 addresses on VPP pg0.
475 self.pg0.config_ip4()
476 self.pg0.resolve_arp()
478 # Configure IPv4 addresses on VPP pg1.
479 self.pg1.config_ip4()
482 super(TestVxlanL2Mode, self).tearDown()
484 def test_l2_mode(self):
487 self, src=self.pg0.local_ip4, dst=self.pg0.remote_ip4, vni=1000, is_l3=False
493 dstIP = t.local_ip4[:-1] + "2"
495 # Create a packet to send
497 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
498 / IP(src=self.pg1.local_ip4, dst=dstIP)
499 / UDP(sport=555, dport=556)
504 rx = self.send_and_expect(self.pg1, [p], self.pg0)
506 self.assertEqual(p[Ether].dst, self.pg0.remote_mac)
507 self.assertEqual(p[Ether].src, self.pg0.local_mac)
508 self.assertEqual(p[ARP].op, 1)
509 self.assertEqual(p[ARP].pdst, dstIP)
512 VppNeighbor(self, t.sw_if_index, self.pg1.remote_mac, dstIP).add_vpp_config()
516 rx = self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)
517 self.assertEqual(NUM_PKTS, len(rx))
520 if __name__ == "__main__":
521 unittest.main(testRunner=VppTestRunner)