#!/usr/bin/env python3 import binascii import random import socket import unittest import scapy.compat from scapy.contrib.mpls import MPLS from scapy.contrib.gtp import GTP_U_Header from scapy.layers.inet import IP, UDP, TCP, ICMP, icmptypes, icmpcodes from scapy.layers.inet6 import IPv6 from scapy.layers.l2 import Ether, Dot1Q, ARP from scapy.packet import Raw from six import moves from framework import tag_fixme_vpp_workers from framework import VppTestCase, VppTestRunner from util import ppp from vpp_ip_route import ( VppIpRoute, VppRoutePath, VppIpMRoute, VppMRoutePath, VppMplsIpBind, VppMplsTable, VppIpTable, FibPathType, find_route, VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump, ) from vpp_ip import VppIpPuntPolicer, VppIpPuntRedirect, VppIpPathMtu from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint from vpp_papi import vpp_papi, VppEnum from vpp_neighbor import VppNeighbor from vpp_lo_interface import VppLoInterface from vpp_policer import VppPolicer, PolicerAction NUM_PKTS = 67 class TestIPv4(VppTestCase): """IPv4 Test Case""" @classmethod def setUpClass(cls): super(TestIPv4, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPv4, cls).tearDownClass() def setUp(self): """ Perform test setup before test case. **Config:** - create 3 pg interfaces - untagged pg0 interface - Dot1Q subinterface on pg1 - Dot1AD subinterface on pg2 - setup interfaces: - put it into UP state - set IPv4 addresses - resolve neighbor address using ARP - configure 200 fib entries :ivar list interfaces: pg interfaces and subinterfaces. :ivar dict flows: IPv4 packet flows in test. """ super(TestIPv4, self).setUp() # create 3 pg interfaces self.create_pg_interfaces(range(3)) # create 2 subinterfaces for pg1 and pg2 self.sub_interfaces = [ VppDot1QSubint(self, self.pg1, 100), VppDot1ADSubint(self, self.pg2, 200, 300, 400), ] # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc. self.flows = dict() self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if] self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if] self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if] # packet sizes self.pg_if_packet_sizes = [64, 1500, 9020] self.interfaces = list(self.pg_interfaces) self.interfaces.extend(self.sub_interfaces) # setup all interfaces for i in self.interfaces: i.admin_up() i.config_ip4() i.resolve_arp() # config 2M FIB entries def tearDown(self): """Run standard test teardown and log ``show ip arp``.""" super(TestIPv4, self).tearDown() def show_commands_at_teardown(self): self.logger.info(self.vapi.cli("show ip4 neighbors")) # info(self.vapi.cli("show ip fib")) # many entries def modify_packet(self, src_if, packet_size, pkt): """Add load, set destination IP and extend packet to required packet size for defined interface. :param VppInterface src_if: Interface to create packet for. :param int packet_size: Required packet size. :param Scapy pkt: Packet to be modified. """ dst_if_idx = int(packet_size / 10 % 2) dst_if = self.flows[src_if][dst_if_idx] info = self.create_packet_info(src_if, dst_if) payload = self.info_to_payload(info) p = pkt / Raw(payload) p[IP].dst = dst_if.remote_ip4 info.data = p.copy() if isinstance(src_if, VppSubInterface): p = src_if.add_dot1_layer(p) self.extend_packet(p, packet_size) return p def create_stream(self, src_if): """Create input packet stream for defined interface. :param VppInterface src_if: Interface to create packet stream for. """ hdr_ext = 4 if isinstance(src_if, VppSubInterface) else 0 pkt_tmpl = ( Ether(dst=src_if.local_mac, src=src_if.remote_mac) / IP(src=src_if.remote_ip4) / UDP(sport=1234, dport=1234) ) pkts = [ self.modify_packet(src_if, i, pkt_tmpl) for i in moves.range( self.pg_if_packet_sizes[0], self.pg_if_packet_sizes[1], 10 ) ] pkts_b = [ self.modify_packet(src_if, i, pkt_tmpl) for i in moves.range( self.pg_if_packet_sizes[1] + hdr_ext, self.pg_if_packet_sizes[2] + hdr_ext, 50, ) ] pkts.extend(pkts_b) return pkts def verify_capture(self, dst_if, capture): """Verify captured input packet stream for defined interface. :param VppInterface dst_if: Interface to verify captured packet stream for. :param list capture: Captured packet stream. """ self.logger.info("Verifying capture on interface %s" % dst_if.name) last_info = dict() for i in self.interfaces: last_info[i.sw_if_index] = None is_sub_if = False dst_sw_if_index = dst_if.sw_if_index if hasattr(dst_if, "parent"): is_sub_if = True for packet in capture: if is_sub_if: # Check VLAN tags and Ethernet header packet = dst_if.remove_dot1_layer(packet) self.assertTrue(Dot1Q not in packet) try: ip = packet[IP] udp = packet[UDP] payload_info = self.payload_to_info(packet[Raw]) packet_index = payload_info.index self.assertEqual(payload_info.dst, dst_sw_if_index) self.logger.debug( "Got packet on port %s: src=%u (id=%u)" % (dst_if.name, payload_info.src, packet_index) ) next_info = self.get_next_packet_info_for_interface2( payload_info.src, dst_sw_if_index, last_info[payload_info.src] ) last_info[payload_info.src] = next_info self.assertTrue(next_info is not None) self.assertEqual(packet_index, next_info.index) saved_packet = next_info.data # Check standard fields self.assertEqual(ip.src, saved_packet[IP].src) self.assertEqual(ip.dst, saved_packet[IP].dst) self.assertEqual(udp.sport, saved_packet[UDP].sport) self.assertEqual(udp.dport, saved_packet[UDP].dport) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise for i in self.interfaces: remaining_packet = self.get_next_packet_info_for_interface2( i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index] ) self.assertTrue( remaining_packet is None, "Interface %s: Packet expected from interface %s " "didn't arrive" % (dst_if.name, i.name), ) def test_fib(self): """IPv4 FIB test Test scenario: - Create IPv4 stream for pg0 interface - Create IPv4 tagged streams for pg1's and pg2's sub-interface. - Send and verify received packets on each interface. """ pkts = self.create_stream(self.pg0) self.pg0.add_stream(pkts) for i in self.sub_interfaces: pkts = self.create_stream(i) i.parent.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg0.get_capture() self.verify_capture(self.pg0, pkts) for i in self.sub_interfaces: pkts = i.parent.get_capture() self.verify_capture(i, pkts) class TestIPv4RouteLookup(VppTestCase): """IPv4 Route Lookup Test Case""" routes = [] def route_lookup(self, prefix, exact): return self.vapi.api( self.vapi.papi.ip_route_lookup, { "table_id": 0, "exact": exact, "prefix": prefix, }, ) @classmethod def setUpClass(cls): super(TestIPv4RouteLookup, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPv4RouteLookup, cls).tearDownClass() def setUp(self): super(TestIPv4RouteLookup, self).setUp() drop_nh = VppRoutePath( "127.0.0.1", 0xFFFFFFFF, type=FibPathType.FIB_PATH_TYPE_DROP ) # Add 3 routes r = VppIpRoute(self, "1.1.0.0", 16, [drop_nh]) r.add_vpp_config() self.routes.append(r) r = VppIpRoute(self, "1.1.1.0", 24, [drop_nh]) r.add_vpp_config() self.routes.append(r) r = VppIpRoute(self, "1.1.1.1", 32, [drop_nh]) r.add_vpp_config() self.routes.append(r) def tearDown(self): # Remove the routes we added for r in self.routes: r.remove_vpp_config() super(TestIPv4RouteLookup, self).tearDown() def test_exact_match(self): # Verify we find the host route prefix = "1.1.1.1/32" result = self.route_lookup(prefix, True) assert prefix == str(result.route.prefix) # Verify we find a middle prefix route prefix = "1.1.1.0/24" result = self.route_lookup(prefix, True) assert prefix == str(result.route.prefix) # Verify we do not find an available LPM. with self.vapi.assert_negative_api_retval(): self.route_lookup("1.1.1.2/32", True) def test_longest_prefix_match(self): # verify we find lpm lpm_prefix = "1.1.1.0/24" result = self.route_lookup("1.1.1.2/32", False) assert lpm_prefix == str(result.route.prefix) # Verify we find the exact when not requested result = self.route_lookup(lpm_prefix, False) assert lpm_prefix == str(result.route.prefix) # Can't seem to delete the default route so no negative LPM test. class TestIPv4IfAddrRoute(VppTestCase): """IPv4 Interface Addr Route Test Case""" @classmethod def setUpClass(cls): super(TestIPv4IfAddrRoute, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPv4IfAddrRoute, cls).tearDownClass() def setUp(self): super(TestIPv4IfAddrRoute, self).setUp() # create 1 pg interface self.create_pg_interfaces(range(1)) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() def tearDown(self): super(TestIPv4IfAddrRoute, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip4() i.admin_down() def test_ipv4_ifaddrs_same_prefix(self): """IPv4 Interface Addresses Same Prefix test Test scenario: - Verify no route in FIB for prefix 10.10.10.0/24 - Configure IPv4 address 10.10.10.10/24 on an interface - Verify route in FIB for prefix 10.10.10.0/24 - Configure IPv4 address 10.10.10.20/24 on an interface - Delete 10.10.10.10/24 from interface - Verify route in FIB for prefix 10.10.10.0/24 - Delete 10.10.10.20/24 from interface - Verify no route in FIB for prefix 10.10.10.0/24 """ # create two addresses, verify route not present if_addr1 = VppIpInterfaceAddress(self, self.pg0, "10.10.10.10", 24) if_addr2 = VppIpInterfaceAddress(self, self.pg0, "10.10.10.20", 24) self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.10/24 self.assertFalse(find_route(self, "10.10.10.10", 32)) self.assertFalse(find_route(self, "10.10.10.20", 32)) self.assertFalse(find_route(self, "10.10.10.255", 32)) self.assertFalse(find_route(self, "10.10.10.0", 32)) # configure first address, verify route present if_addr1.add_vpp_config() self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.10/24 self.assertTrue(find_route(self, "10.10.10.10", 32)) self.assertFalse(find_route(self, "10.10.10.20", 32)) self.assertTrue(find_route(self, "10.10.10.255", 32)) self.assertTrue(find_route(self, "10.10.10.0", 32)) # configure second address, delete first, verify route not removed if_addr2.add_vpp_config() if_addr1.remove_vpp_config() self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.10/24 self.assertTrue(if_addr2.query_vpp_config()) # 10.10.10.20/24 self.assertFalse(find_route(self, "10.10.1
/*
 * Copyright (c) 2015 Cisco and/or its affiliates.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
/**
 * @file
 * @brief Host utility functions
 */
#include <vppinfra/format.h>
#include <vlib/vlib.h>

#include <vlib/threads.h>
#include <vnet/vnet.h>
#include <vppinfra/format.h>

/**
 * @brief GDB callable function: vl - Return vector length of vector
 *
 * @param *p - void - address of vector
 *
 * @return length - u32
 *
 */
u32
vl (void *p)
{
  return vec_len (p);
}

/**
 * @brief GDB callable function: pe - call pool_elts - number of elements in a pool
 *
 * @param *v - void - address of pool
 *
 * @return number - uword
 *
 */
uword
pe (void *v)
{
  return (pool_elts (v));
}

/**
 * @brief GDB callable function: pifi - call pool_is_free_index - is passed index free?
 *
 * @param *p - void - address of pool
 * @param *index - u32
 *
 * @return 0|1 - int
 *
 */
int
pifi (void *p, u32 index)
{
  return pool_is_free_index (p, index);
}

/**
 * @brief GDB callable function: debug_hex_bytes - return formatted hex string
 *
 * @param *s - u8
 * @param n - u32 - number of bytes to format
 *
 */
void
debug_hex_bytes (u8 * s, u32 n)
{
  fformat (stderr, "%U\n", format_hex_bytes, s, n);
}

/**
 * @brief GDB callable function: vlib_dump_frame_ownership
 *
 */
void
vlib_dump_frame_ownership (void)
{
  vlib_main_t *vm = vlib_get_main ();
  vlib_node_main_t *nm = &vm->node_main;
  vlib_node_runtime_t *this_node_runtime;
  vlib_next_frame_t *nf;
  u32 first_nf_index;
  u32 index;

  vec_foreach (this_node_runtime, nm->nodes_by_type[VLIB_NODE_TYPE_INTERNAL])
  {
    first_nf_index = this_node_runtime->next_frame_index;

    for (index = first_nf_index; index < first_nf_index +
	 this_node_runtime->n_next_nodes; index++)
      {
	vlib_node_runtime_t *owned_runtime;
	nf = vec_elt_at_index (vm->node_main.next_frames, index);
	if (nf->flags & VLIB_FRAME_OWNER)
	  {
	    owned_runtime = vec_elt_at_index (nm->nodes_by_type[0],
					      nf->node_runtime_index);
	    fformat (stderr,
		     "%s next index %d owns enqueue rights to %s\n",
		     nm->nodes[this_node_runtime->node_index]->name,
		     index - first_nf_index,
		     nm->nodes[owned_runtime->node_index]->name);
	    fformat (stderr, "  nf index %d nf->frame_index %d\n",
		     nf - vm->node_main.next_frames, nf->frame_index);
	  }
      }
  }
}

/**
 * @brief GDB callable function: vlib_runtime_index_to_node_name
 *
 * Takes node index and will return the node name.
 *
 * @param index - u32
 */
void
vlib_runtime_index_to_node_name (u32 index)
{
  vlib_main_t *vm = vlib_get_main ();
  vlib_node_main_t *nm = &vm->node_main;

  if (index >= vec_len (nm->nodes))
    {
      fformat (stderr, "%d out of range, max %d\n", vec_len (nm->nodes));
      return;
    }

  fformat (stderr, "node runtime index %d name %s\n", index,
	   nm->nodes[index]->name);
}

void
gdb_show_errors (int verbose)
{
  extern vlib_cli_command_t vlib_cli_show_errors;
  unformat_input_t input;
  vlib_main_t *vm = vlib_get_main ();

  if (verbose == 0)
    unformat_init_string (&input, "verbose 0", 9);
  else if (verbose == 1)
    unformat_init_string (&input, "verbose 1", 9);
  else
    {
      fformat (stderr, "verbose not 0 or 1\n");
      return;
    }

  vlib_cli_show_errors.function (vm, &input, 0 /* cmd */ );
  unformat_free (&input);
}

void
gdb_show_session (int verbose)
{
  extern vlib_cli_command_t vlib_cli_show_session_command;
  unformat_input_t input;
  vlib_main_t *vm = vlib_get_main ();

  if (verbose == 0)
    unformat_init_string (&input, "verbose 0", 9);
  else if (verbose == 1)
    unformat_init_string (&input, "verbose 1", 9);
  else if (verbose == 2)
    unformat_init_string (&input, "verbose 2", 9);
  else
    {
      fformat (stderr, "verbose not 0 - 2\n");
      return;
    }

  vlib_cli_show_session_command.function (vm, &input, 0 /* cmd */ );
  unformat_free (&input);
}

/**
 * @brief GDB callable function: show_gdb_command_fn - show gdb
 *
 * Shows list of functions for VPP available in GDB
 *
 * @return error - clib_error_t
 */
static clib_error_t *
show_gdb_command_fn (vlib_main_t * vm,
		     unformat_input_t * input, vlib_cli_command_t * cmd)
{
  vlib_cli_output (vm, "vl(p) returns vec_len(p)");
  vlib_cli_output (vm, "vb(b) returns vnet_buffer(b) [opaque]");
  vlib_cli_output (vm, "vb2(b) returns vnet_buffer2(b) [opaque2]");
  vlib_cli_output (vm, "pe(p) returns pool_elts(p)");
  vlib_cli_output (vm, "pifi(p, i) returns pool_is_free_index(p, i)");
  vlib_cli_output (vm, "gdb_show_errors(0|1) dumps error counters");
  vlib_cli_output (vm, "gdb_show_session dumps session counters");
  vlib_cli_output (vm, "debug_hex_bytes (ptr, n_bytes) dumps n_bytes in hex");
  vlib_cli_output (vm, "vlib_dump_frame_ownership() does what it says");
  vlib_cli_output (vm, "vlib_runtime_index_to_node_name (index) prints NN");

  return 0;
}

/* *INDENT-OFF* */
VLIB_CLI_COMMAND (show_gdb_funcs_command, static) = {
  .path = "show gdb",
  .short_help = "Describe functions which can be called from gdb",
  .function = show_gdb_command_fn,
};
/* *INDENT-ON* */

vnet_buffer_opaque_t *
vb (void *vb_arg)
{
  vlib_buffer_t *b = (vlib_buffer_t *) vb_arg;
  vnet_buffer_opaque_t *rv;

  rv = vnet_buffer (b);

  return rv;
}

vnet_buffer_opaque2_t *
vb2 (void *vb_arg)
{
  vlib_buffer_t *b = (vlib_buffer_t *) vb_arg;
  vnet_buffer_opaque2_t *rv;

  rv = vnet_buffer2 (b);

  return rv;
}


/* Cafeteria plan, maybe you don't want these functions */
clib_error_t *
gdb_func_init (vlib_main_t * vm)
{
  return 0;
}

VLIB_INIT_FUNCTION (gdb_func_init);

/*
 * fd.io coding-style-patch-verification: ON
 *
 * Local Variables:
 * eval: (c-set-style "gnu")
 * End:
 */
nd_expect_only(self.pg0, port_ip_pkts, self.pg2) # # this case gtp v1 teid key LB # self.vapi.set_ip_flow_hash_v3( af=af.ADDRESS_IP4, table_id=0, flow_hash_config=( fhcv2.IP_API_V2_FLOW_HASH_SRC_IP | fhcv2.IP_API_V2_FLOW_HASH_PROTO | fhcv2.IP_API_V2_FLOW_HASH_GTPV1_TEID ), ) self.logger.info(self.vapi.cli("show ip fib")) self.send_and_expect_load_balancing( self.pg0, src_gtp_pkts, [self.pg1, self.pg2] ) self.send_and_expect_only(self.pg0, port_gtp_pkts, self.pg2) # # change the flow hash config back to defaults # self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, proto=1, sport=1, dport=1) # # Recursive prefixes # - testing that 2 stages of load-balancing occurs and there is no # polarisation (i.e. only 2 of 4 paths are used) # port_pkts = [] src_pkts = [] for ii in range(257): port_pkts.append( ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(dst="1.1.1.1", src="20.0.0.1") / UDP(sport=1234, dport=1234 + ii) / Raw(b"\xa5" * 100) ) ) src_pkts.append( ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(dst="1.1.1.1", src="20.0.0.%d" % ii) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) ) route_10_0_0_2 = VppIpRoute( self, "10.0.0.2", 32, [ VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index), VppRoutePath(self.pg4.remote_ip4, self.pg4.sw_if_index), ], ) route_10_0_0_2.add_vpp_config() route_1_1_1_1 = VppIpRoute( self, "1.1.1.1", 32, [ VppRoutePath("10.0.0.2", 0xFFFFFFFF), VppRoutePath("10.0.0.1", 0xFFFFFFFF), ], ) route_1_1_1_1.add_vpp_config() # # inject the packet on pg0 - expect load-balancing across all 4 paths # self.vapi.cli("clear trace") self.send_and_expect_load_balancing( self.pg0, port_pkts, [self.pg1, self.pg2, self.pg3, self.pg4] ) self.send_and_expect_load_balancing( self.pg0, src_pkts, [self.pg1, self.pg2, self.pg3, self.pg4] ) # # bring down pg1 expect LB to adjust to use only those that are up # self.pg1.link_down() rx = self.send_and_expect_load_balancing( self.pg0, src_pkts, [self.pg2, self.pg3, self.pg4] ) self.assertEqual(len(src_pkts), self.total_len(rx)) # # bring down pg2 expect LB to adjust to use only those that are up # self.pg2.link_down() rx = self.send_and_expect_load_balancing( self.pg0, src_pkts, [self.pg3, self.pg4] ) self.assertEqual(len(src_pkts), self.total_len(rx)) # # bring the links back up - expect LB over all again # self.pg1.link_up() self.pg2.link_up() rx = self.send_and_expect_load_balancing( self.pg0, src_pkts, [self.pg1, self.pg2, self.pg3, self.pg4] ) self.assertEqual(len(src_pkts), self.total_len(rx)) # # The same link-up/down but this time admin state # self.pg1.admin_down() self.pg2.admin_down() rx = self.send_and_expect_load_balancing( self.pg0, src_pkts, [self.pg3, self.pg4] ) self.assertEqual(len(src_pkts), self.total_len(rx)) self.pg1.admin_up() self.pg2.admin_up() self.pg1.resolve_arp() self.pg2.resolve_arp() rx = self.send_and_expect_load_balancing( self.pg0, src_pkts, [self.pg1, self.pg2, self.pg3, self.pg4] ) self.assertEqual(len(src_pkts), self.total_len(rx)) # # Recursive prefixes # - testing that 2 stages of load-balancing, no choices # port_pkts = [] for ii in range(257): port_pkts.append( ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(dst="1.1.1.2", src="20.0.0.2") / UDP(sport=1234, dport=1234 + ii) / Raw(b"\xa5" * 100) ) ) route_10_0_0_3 = VppIpRoute( self, "10.0.0.3", 32, [VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index)], ) route_10_0_0_3.add_vpp_config() route_1_1_1_2 = VppIpRoute( self, "1.1.1.2", 32, [VppRoutePath("10.0.0.3", 0xFFFFFFFF)] ) route_1_1_1_2.add_vpp_config() # # inject the packet on pg0 - rx only on via routes output interface # self.vapi.cli("clear trace") self.send_and_expect_only(self.pg0, port_pkts, self.pg3) # # Add a LB route in the presence of a down link - expect no # packets over the down link # self.pg3.link_down() route_10_0_0_3 = VppIpRoute( self, "10.0.0.3", 32, [ VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index), VppRoutePath(self.pg4.remote_ip4, self.pg4.sw_if_index), ], ) route_10_0_0_3.add_vpp_config() port_pkts = [] for ii in range(257): port_pkts.append( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(dst="10.0.0.3", src="20.0.0.2") / UDP(sport=1234, dport=1234 + ii) / Raw(b"\xa5" * 100) ) self.send_and_expect_only(self.pg0, port_pkts, self.pg4) # bring the link back up self.pg3.link_up() rx = self.send_and_expect_load_balancing( self.pg0, port_pkts, [self.pg3, self.pg4] ) self.assertEqual(len(src_pkts), self.total_len(rx)) class TestIPVlan0(VppTestCase): """IPv4 VLAN-0""" @classmethod def setUpClass(cls): super(TestIPVlan0, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPVlan0, cls).tearDownClass() def setUp(self): super(TestIPVlan0, self).setUp() self.create_pg_interfaces(range(2)) mpls_tbl = VppMplsTable(self, 0) mpls_tbl.add_vpp_config() for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() i.enable_mpls() def tearDown(self): for i in self.pg_interfaces: i.disable_mpls() i.unconfig_ip4() i.admin_down() super(TestIPVlan0, self).tearDown() def test_ip_vlan_0(self): """IP VLAN-0""" pkts = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / Dot1Q(vlan=0) / IP(dst=self.pg1.remote_ip4, src=self.pg0.remote_ip4) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) * NUM_PKTS # # Expect that packets sent on VLAN-0 are forwarded on the # main interface. # self.send_and_expect(self.pg0, pkts, self.pg1) class IPPuntSetup(object): """Setup for IPv4 Punt Police/Redirect""" def punt_setup(self): self.create_pg_interfaces(range(4)) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() # use UDP packet that have a port we need to explicitly # register to get punted. pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4 af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4 udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP punt_udp = { "type": pt_l4, "punt": { "l4": { "af": af_ip4, "protocol": udp_proto, "port": 1234, } }, } self.vapi.set_punt(is_add=1, punt=punt_udp) af_ip6 = VppEnum.vl_api_address_family_t.ADDRESS_IP6 punt_udp = { "type": pt_l4, "punt": { "l4": { "af": af_ip6, "protocol": udp_proto, "port": 1236, } }, } self.vapi.set_punt(is_add=1, punt=punt_udp) self.pkt = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) def punt_teardown(self): for i in self.pg_interfaces: i.unconfig_ip4() i.admin_down() class TestIPPunt(IPPuntSetup, VppTestCase): """IPv4 Punt Police/Redirect""" def setUp(self): super().setUp() super().punt_setup() def tearDown(self): super().punt_teardown() super().tearDown() def test_ip_punt_api_validation(self): """IP punt API parameter validation""" nh_addr = self.pg1.remote_ip4 punt = { "rx_sw_if_index": self.pg0.sw_if_index, "af": VppEnum.vl_api_address_family_t.ADDRESS_IP4, "n_paths": 1000000, "paths": [], } with self.assertRaises(vpp_papi.VPPIOError): self.vapi.add_del_ip_punt_redirect_v2(punt=punt, is_add=True) punt = { "rx_sw_if_index": self.pg0.sw_if_index, "af": VppEnum.vl_api_address_family_t.ADDRESS_IP4, "n_paths": 0, "paths": [], } self.vapi.add_del_ip_punt_redirect_v2(punt=punt, is_add=True) def test_ip_punt(self): """IP punt police and redirect""" pkts = self.pkt * 1025 # # Configure a punt redirect via pg1. # nh_addr = self.pg1.remote_ip4 ip_punt_redirect = VppIpPuntRedirect( self, self.pg0.sw_if_index, self.pg1.sw_if_index, nh_addr ) ip_punt_redirect.add_vpp_config() self.send_and_expect(self.pg0, pkts, self.pg1) # # add a policer # policer = VppPolicer(self, "ip4-punt", 400, 0, 10, 0, rate_type=1) policer.add_vpp_config() ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index) ip_punt_policer.add_vpp_config() self.vapi.cli("clear trace") self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() # # the number of packet received should be greater than 0, # but not equal to the number sent, since some were policed # rx = self.pg1._get_capture(1) stats = policer.get_stats() # Single rate policer - expect conform, violate but no exceed self.assertGreater(stats["conform_packets"], 0) self.assertEqual(stats["exceed_packets"], 0) self.assertGreater(stats["violate_packets"], 0) self.assertGreater(len(rx), 0) self.assertLess(len(rx), len(pkts)) # # remove the policer. back to full rx # ip_punt_policer.remove_vpp_config() policer.remove_vpp_config() self.send_and_expect(self.pg0, pkts, self.pg1) # # remove the redirect. expect full drop. # ip_punt_redirect.remove_vpp_config() self.send_and_assert_no_replies(self.pg0, pkts, "IP no punt config") # # Add a redirect that is not input port selective # ip_punt_redirect = VppIpPuntRedirect( self, 0xFFFFFFFF, self.pg1.sw_if_index, nh_addr ) ip_punt_redirect.add_vpp_config() self.send_and_expect(self.pg0, pkts, self.pg1) ip_punt_redirect.remove_vpp_config() def test_ip_punt_vrf(self): """IP punt/local with VRFs""" # use a punt redirect to test if for-us packets are accepted pkts = self.pkt * 1025 vlans_pg0 = [VppDot1QSubint(self, self.pg0, v) for v in range(100, 104)] vlans_pg1 = [VppDot1QSubint(self, self.pg1, v) for v in range(100, 104)] tbl4 = [VppIpTable(self, v).add_vpp_config() for v in range(100, 104)] tbl6 = [VppIpTable(self, v, True).add_vpp_config() for v in range(100, 104)] for v in vlans_pg0 + vlans_pg1: v.admin_up() v.set_table_ip4(v.vlan) v.set_table_ip6(v.vlan) v.config_ip4() v.config_ip6() v.resolve_arp() v.resolve_ndp() [ VppIpPuntRedirect( self, vlans_pg0[i].sw_if_index, vlans_pg1[i].sw_if_index, vlans_pg1[i].remote_ip4, ).add_vpp_config() for i in range(4) ] [ VppIpPuntRedirect( self, vlans_pg0[i].sw_if_index, vlans_pg1[i].sw_if_index, vlans_pg1[i].remote_ip6, ).add_vpp_config() for i in range(4) ] pkts = [ ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / Dot1Q(vlan=i.vlan) / IP(src=i.remote_ip4, dst=i.local_ip4) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) for i in vlans_pg0 ] self.send_and_expect(self.pg0, pkts, self.pg1) # # IPv4 # # we reject packets for source addresses in the wrong vlan/VRF pkts = [ ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / Dot1Q(vlan=i.vlan) / IP(src="1.1.1.1", dst=i.local_ip4) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) for i in vlans_pg0 ] # single and dual loop self.send_and_assert_no_replies(self.pg0, [pkts[0]]) self.send_and_assert_no_replies(self.pg0, pkts) self.assert_error_counter_equal("/err/ip4-local/src_lookup_miss", len(pkts) + 1) # using the same source in different tables, should reject # for the table that the source is not present in # the first packet in the stream is drop pkts = [ ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / Dot1Q(vlan=i.vlan) / IP(src=vlans_pg0[0].remote_ip4, dst=i.local_ip4) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) for i in vlans_pg0 ] # single loop accept and drop # followed by both in the same frame/loop self.send_and_expect(self.pg0, [pkts[0]], self.pg1) self.send_and_assert_no_replies(self.pg0, [pkts[1]]) self.send_and_expect(self.pg0, pkts * 4, self.pg1, n_rx=4) # using the same source in different tables, should reject # for the table that the source is not present in # the first packet in the stream is accept pkts = [ ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / Dot1Q(vlan=i.vlan) / IP(src=vlans_pg0[3].remote_ip4, dst=i.local_ip4) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) for i in vlans_pg0 ] # single loop accept and drop # followed by both in the same frame/loop self.send_and_expect(self.pg0, [pkts[3]], self.pg1) self.send_and_assert_no_replies(self.pg0, [pkts[1]]) self.send_and_expect(self.pg0, pkts * 4, self.pg1, n_rx=4) # # IPv6 # # we reject packets for source addresses in the wrong vlan/VRF pkts = [ ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / Dot1Q(vlan=i.vlan) / IPv6(src="1::1", dst=i.local_ip6) / UDP(sport=1236, dport=1236) / Raw(b"\xa5" * 100) ) for i in vlans_pg0 ] # single and dual loop self.send_and_assert_no_replies(self.pg0, [pkts[0]]) self.send_and_assert_no_replies(self.pg0, pkts) self.assert_error_counter_equal("/err/ip6-input/src_lookup_miss", len(pkts) + 1) # using the same source in different tables, should reject # for the table that the source is not present in # the first packet in the stream is drop pkts = [ ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / Dot1Q(vlan=i.vlan) / IPv6(src=vlans_pg0[0].remote_ip6, dst=i.local_ip6) / UDP(sport=1236, dport=1236) / Raw(b"\xa5" * 100) ) for i in vlans_pg0 ] # single loop accept and drop # followed by both in the same frame/loop self.send_and_expect(self.pg0, [pkts[0]], self.pg1) self.send_and_assert_no_replies(self.pg0, [pkts[1]]) self.send_and_expect(self.pg0, pkts * 4, self.pg1, n_rx=4) # using the same source in different tables, should reject # for the table that the source is not present in # the first packet in the stream is accept pkts = [ ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / Dot1Q(vlan=i.vlan) / IPv6(src=vlans_pg0[3].remote_ip6, dst=i.local_ip6) / UDP(sport=1236, dport=1236) / Raw(b"\xa5" * 100) ) for i in vlans_pg0 ] # single loop accept and drop # followed by both in the same frame/loop self.send_and_expect(self.pg0, [pkts[3]], self.pg1) self.send_and_assert_no_replies(self.pg0, [pkts[1]]) self.send_and_expect(self.pg0, pkts * 4, self.pg1, n_rx=4) for v in vlans_pg0 + vlans_pg1: v.unconfig_ip4() v.unconfig_ip6() v.set_table_ip4(0) v.set_table_ip6(0) def test_ip_punt_dump(self): """IP4 punt redirect dump""" # # Configure a punt redirects # nh_address = self.pg3.remote_ip4 ipr_03 = VppIpPuntRedirect( self, self.pg0.sw_if_index, self.pg3.sw_if_index, nh_address ) ipr_13 = VppIpPuntRedirect( self, self.pg1.sw_if_index, self.pg3.sw_if_index, nh_address ) ipr_23 = VppIpPuntRedirect( self, self.pg2.sw_if_index, self.pg3.sw_if_index, "0.0.0.0" ) ipr_03.add_vpp_config() ipr_13.add_vpp_config() ipr_23.add_vpp_config() # # Dump pg0 punt redirects # self.assertTrue(ipr_03.query_vpp_config()) self.assertTrue(ipr_13.query_vpp_config()) self.assertTrue(ipr_23.query_vpp_config()) # # Dump punt redirects for all interfaces # punts = self.vapi.ip_punt_redirect_dump(0xFFFFFFFF) self.assertEqual(len(punts), 3) for p in punts: self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index) self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip4) self.assertEqual(str(punts[2].punt.nh), "0.0.0.0") class TestIPPuntHandoff(IPPuntSetup, VppTestCase): """IPv4 Punt Policer thread handoff""" vpp_worker_count = 2 def setUp(self): super(TestIPPuntHandoff, self).setUp() super(TestIPPuntHandoff, self).punt_setup() def tearDown(self): super(TestIPPuntHandoff, self).punt_teardown() super(TestIPPuntHandoff, self).tearDown() def test_ip_punt_policer_handoff(self): """IP4 punt policer thread handoff""" pkts = self.pkt * NUM_PKTS # # Configure a punt redirect via pg1. # nh_addr = self.pg1.remote_ip4 ip_punt_redirect = VppIpPuntRedirect( self, self.pg0.sw_if_index, self.pg1.sw_if_index, nh_addr ) ip_punt_redirect.add_vpp_config() action_tx = PolicerAction( VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0 ) # # This policer drops no packets, we are just # testing that they get to the right thread. # policer = VppPolicer( self, "ip4-punt", 400, 0, 10, 0, 1, 0, 0, False, action_tx, action_tx, action_tx, ) policer.add_vpp_config() ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index) ip_punt_policer.add_vpp_config() for worker in [0, 1]: self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker) self.logger.debug(self.vapi.cli("show trace max 100")) # Combined stats, all threads stats = policer.get_stats() # Single rate policer - expect conform, violate but no exceed self.assertGreater(stats["conform_packets"], 0) self.assertEqual(stats["exceed_packets"], 0) self.assertGreater(stats["violate_packets"], 0) # Worker 0, should have done all the policing stats0 = policer.get_stats(worker=0) self.assertEqual(stats, stats0) # Worker 1, should have handed everything off stats1 = policer.get_stats(worker=1) self.assertEqual(stats1["conform_packets"], 0) self.assertEqual(stats1["exceed_packets"], 0) self.assertEqual(stats1["violate_packets"], 0) # Bind the policer to worker 1 and repeat policer.bind_vpp_config(1, True) for worker in [0, 1]: self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker) self.logger.debug(self.vapi.cli("show trace max 100")) # The 2 workers should now have policed the same amount stats = policer.get_stats() stats0 = policer.get_stats(worker=0) stats1 = policer.get_stats(worker=1) self.assertGreater(stats0["conform_packets"], 0) self.assertEqual(stats0["exceed_packets"], 0) self.assertGreater(stats0["violate_packets"], 0) self.assertGreater(stats1["conform_packets"], 0) self.assertEqual(stats1["exceed_packets"], 0) self.assertGreater(stats1["violate_packets"], 0) self.assertEqual( stats0["conform_packets"] + stats1["conform_packets"], stats["conform_packets"], ) self.assertEqual( stats0["violate_packets"] + stats1["violate_packets"], stats["violate_packets"], ) # Unbind the policer and repeat policer.bind_vpp_config(1, False) for worker in [0, 1]: self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker) self.logger.debug(self.vapi.cli("show trace max 100")) # The policer should auto-bind to worker 0 when packets arrive stats = policer.get_stats() stats0new = policer.get_stats(worker=0) stats1new = policer.get_stats(worker=1) self.assertGreater(stats0new["conform_packets"], stats0["conform_packets"]) self.assertEqual(stats0new["exceed_packets"], 0) self.assertGreater(stats0new["violate_packets"], stats0["violate_packets"]) self.assertEqual(stats1, stats1new) # # Clean up # ip_punt_policer.remove_vpp_config() policer.remove_vpp_config() ip_punt_redirect.remove_vpp_config() class TestIPDeag(VppTestCase): """IPv4 Deaggregate Routes""" @classmethod def setUpClass(cls): super(TestIPDeag, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPDeag, cls).tearDownClass() def setUp(self): super(TestIPDeag, self).setUp() self.create_pg_interfaces(range(3)) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() def tearDown(self): super(TestIPDeag, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip4() i.admin_down() def test_ip_deag(self): """IP Deag Routes""" # # Create a table to be used for: # 1 - another destination address lookup # 2 - a source address lookup # table_dst = VppIpTable(self, 1) table_src = VppIpTable(self, 2) table_dst.add_vpp_config() table_src.add_vpp_config() # # Add a route in the default table to point to a deag/ # second lookup in each of these tables # route_to_dst = VppIpRoute( self, "1.1.1.1", 32, [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=1)] ) route_to_src = VppIpRoute( self, "1.1.1.2", 32, [ VppRoutePath( "0.0.0.0", 0xFFFFFFFF, nh_table_id=2, type=FibPathType.FIB_PATH_TYPE_SOURCE_LOOKUP, ) ], ) route_to_dst.add_vpp_config() route_to_src.add_vpp_config() # # packets to these destination are dropped, since they'll # hit the respective default routes in the second table # p_dst = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src="5.5.5.5", dst="1.1.1.1") / TCP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) p_src = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src="2.2.2.2", dst="1.1.1.2") / TCP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) pkts_dst = p_dst * 257 pkts_src = p_src * 257 self.send_and_assert_no_replies(self.pg0, pkts_dst, "IP in dst table") self.send_and_assert_no_replies(self.pg0, pkts_src, "IP in src table") # # add a route in the dst table to forward via pg1 # route_in_dst = VppIpRoute( self, "1.1.1.1", 32, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)], table_id=1, ) route_in_dst.add_vpp_config() self.send_and_expect(self.pg0, pkts_dst, self.pg1) # # add a route in the src table to forward via pg2 # route_in_src = VppIpRoute( self, "2.2.2.2", 32, [VppRoutePath(self.pg2.remote_ip4, self.pg2.sw_if_index)], table_id=2, ) route_in_src.add_vpp_config() self.send_and_expect(self.pg0, pkts_src, self.pg2) # # loop in the lookup DP # route_loop = VppIpRoute( self, "2.2.2.3", 32, [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=0)] ) route_loop.add_vpp_config() p_l = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src="2.2.2.4", dst="2.2.2.3") / TCP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) self.send_and_assert_no_replies(self.pg0, p_l * 257, "IP lookup loop") class TestIPInput(VppTestCase): """IPv4 Input Exceptions""" @classmethod def setUpClass(cls): super(TestIPInput, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPInput, cls).tearDownClass() def setUp(self): super(TestIPInput, self).setUp() self.create_pg_interfaces(range(2)) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() def tearDown(self): super(TestIPInput, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip4() i.admin_down() def test_ip_input(self): """IP Input Exceptions""" # i can't find a way in scapy to construct an IP packet # with a length less than the IP header length # # Packet too short - this is forwarded # p_short = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, len=40) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) rx = self.send_and_expect(self.pg0, p_short * NUM_PKTS, self.pg1) # # Packet too long - this is dropped # p_long = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, len=400) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) rx = self.send_and_assert_no_replies(self.pg0, p_long * NUM_PKTS, "too long") # # bad chksum - this is dropped # p_chksum = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, chksum=400) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) rx = self.send_and_assert_no_replies( self.pg0, p_chksum * NUM_PKTS, "bad checksum" ) # # bad version - this is dropped # p_ver = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, version=3) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) rx = self.send_and_assert_no_replies( self.pg0, p_ver * NUM_PKTS, "funky version" ) # # fragment offset 1 - this is dropped # p_frag = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, frag=1) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) rx = self.send_and_assert_no_replies(self.pg0, p_frag * NUM_PKTS, "frag offset") # # TTL expired packet # p_ttl = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=1) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 100) ) rxs = self.send_and_expect_some(self.pg0, p_ttl * NUM_PKTS, self.pg0) for rx in rxs: icmp = rx[ICMP] self.assertEqual(icmptypes[icmp.type], "time-exceeded") self.assertEqual(icmpcodes[icmp.type][icmp.code], "ttl-zero-during-transit") self.assertEqual(icmp.src, self.pg0.remote_ip4) self.assertEqual(icmp.dst, self.pg1.remote_ip4) # # MTU exceeded # p_mtu = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=10, flags="DF") / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 2000) ) self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1500, 0, 0, 0]) rxs = self.send_and_expect_some(self.pg0, p_mtu * NUM_PKTS, self.pg0) for rx in rxs: icmp = rx[ICMP] self.assertEqual(icmptypes[icmp.type], "dest-unreach") self.assertEqual(icmpcodes[icmp.type][icmp.code], "fragmentation-needed") self.assertEqual(icmp.nexthopmtu, 1500) self.assertEqual(icmp.src, self.pg0.remote_ip4) self.assertEqual(icmp.dst, self.pg1.remote_ip4) self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2500, 0, 0, 0]) rx = self.send_and_expect(self.pg0, p_mtu * NUM_PKTS, self.pg1) # Reset MTU for subsequent tests self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [9000, 0, 0, 0]) # # source address 0.0.0.0 and 25.255.255.255 and for-us # p_s0 = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src="0.0.0.0", dst=self.pg0.local_ip4) / ICMP(id=4, seq=4) / Raw(load=b"\x0a" * 18) ) rx = self.send_and_assert_no_replies(self.pg0, p_s0 * 17) p_s0 = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src="255.255.255.255", dst=self.pg0.local_ip4) / ICMP(id=4, seq=4) / Raw(load=b"\x0a" * 18) ) rx = self.send_and_assert_no_replies(self.pg0, p_s0 * 17) class TestIPDirectedBroadcast(VppTestCase): """IPv4 Directed Broadcast""" @classmethod def setUpClass(cls): super(TestIPDirectedBroadcast, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPDirectedBroadcast, cls).tearDownClass() def setUp(self): super(TestIPDirectedBroadcast, self).setUp() self.create_pg_interfaces(range(2)) for i in self.pg_interfaces: i.admin_up() def tearDown(self): super(TestIPDirectedBroadcast, self).tearDown() for i in self.pg_interfaces: i.admin_down() def test_ip_input(self): """IP Directed Broadcast""" # # set the directed broadcast on pg0 first, then config IP4 addresses # for pg1 directed broadcast is always disabled self.vapi.sw_interface_set_ip_directed_broadcast(self.pg0.sw_if_index, 1) p0 = ( Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / IP(src="1.1.1.1", dst=self.pg0._local_ip4_bcast) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 2000) ) p1 = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src="1.1.1.1", dst=self.pg1._local_ip4_bcast) / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 2000) ) self.pg0.config_ip4() self.pg0.resolve_arp() self.pg1.config_ip4() self.pg1.resolve_arp() # # test packet is L2 broadcast # rx = self.send_and_expect(self.pg1, p0 * NUM_PKTS, self.pg0) self.assertTrue(rx[0][Ether].dst, "ff:ff:ff:ff:ff:ff") self.send_and_assert_no_replies( self.pg0, p1 * NUM_PKTS, "directed broadcast disabled" ) # # toggle directed broadcast on pg0 # self.vapi.sw_interface_set_ip_directed_broadcast(self.pg0.sw_if_index, 0) self.send_and_assert_no_replies( self.pg1, p0 * NUM_PKTS, "directed broadcast disabled" ) self.vapi.sw_interface_set_ip_directed_broadcast(self.pg0.sw_if_index, 1) rx = self.send_and_expect(self.pg1, p0 * NUM_PKTS, self.pg0) self.pg0.unconfig_ip4() self.pg1.unconfig_ip4() class TestIPLPM(VppTestCase): """IPv4 longest Prefix Match""" @classmethod def setUpClass(cls): super(TestIPLPM, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPLPM, cls).tearDownClass() def setUp(self): super(TestIPLPM, self).setUp() self.create_pg_interfaces(range(4)) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() def tearDown(self): super(TestIPLPM, self).tearDown() for i in self.pg_interfaces: i.admin_down() i.unconfig_ip4() def test_ip_lpm(self): """IP longest Prefix Match""" s_24 = VppIpRoute( self, "10.1.2.0", 24, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)], ) s_24.add_vpp_config() s_8 = VppIpRoute( self, "10.0.0.0", 8, [VppRoutePath(self.pg2.remote_ip4, self.pg2.sw_if_index)], ) s_8.add_vpp_config() p_8 = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src="1.1.1.1", dst="10.1.1.1") / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 2000) ) p_24 = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src="1.1.1.1", dst="10.1.2.1") / UDP(sport=1234, dport=1234) / Raw(b"\xa5" * 2000) ) self.logger.info(self.vapi.cli("sh ip fib mtrie")) rx = self.send_and_expect(self.pg0, p_8 * NUM_PKTS, self.pg2) rx = self.send_and_expect(self.pg0, p_24 * NUM_PKTS, self.pg1) @tag_fixme_vpp_workers class TestIPv4Frag(VppTestCase): """IPv4 fragmentation""" @classmethod def setUpClass(cls): super(TestIPv4Frag, cls).setUpClass() cls.create_pg_interfaces([0, 1]) cls.src_if = cls.pg0 cls.dst_if = cls.pg1 # setup all interfaces for i in cls.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() @classmethod def tearDownClass(cls): super(TestIPv4Frag, cls).tearDownClass() def test_frag_large_packets(self): """Fragmentation of large packets""" self.vapi.cli("adjacency counters enable") p = ( Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) / IP(src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4) / UDP(sport=1234, dport=5678) / Raw() ) self.extend_packet(p, 6000, "abcde") saved_payload = p[Raw].load nbr = VppNeighbor( self, self.dst_if.sw_if_index, self.dst_if.remote_mac, self.dst_if.remote_ip4, ).add_vpp_config() # Force fragmentation by setting MTU of output interface # lower than packet size self.vapi.sw_interface_set_mtu(self.dst_if.sw_if_index, [5000, 0, 0, 0]) self.pg_enable_capture() self.src_if.add_stream(p) self.pg_start() # Expecting 3 fragments because size of created fragments currently # cannot be larger then VPP buffer size (which is 2048) packets = self.dst_if.get_capture(3) # we should show 3 packets thru the neighbor self.assertEqual(3, nbr.get_stats()["packets"]) # Assume VPP sends the fragments in order payload = b"" for p in packets: payload_offset = p.frag * 8 if payload_offset > 0: payload_offset -= 8 # UDP header is not in payload self.assert_equal(payload_offset, len(payload)) payload += p[Raw].load self.assert_equal(payload, saved_payload, "payload") class TestIPReplace(VppTestCase): """IPv4 Table Replace""" @classmethod def setUpClass(cls): super(TestIPReplace, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPReplace, cls).tearDownClass() def setUp(self): super(TestIPReplace, self).setUp() self.create_pg_interfaces(range(4)) table_id = 1 self.tables = [] for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() i.generate_remote_hosts(2) self.tables.append(VppIpTable(self, table_id).add_vpp_config()) table_id += 1 def tearDown(self): super(TestIPReplace, self).tearDown() for i in self.pg_interfaces: i.admin_down() i.unconfig_ip4() def test_replace(self): """IP Table Replace""" MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t N_ROUTES = 20 links = [self.pg0, self.pg1, self.pg2, self.pg3] routes = [[], [], [], []] # load up the tables with some routes for ii, t in enumerate(self.tables): for jj in range(N_ROUTES): uni = VppIpRoute( self, "10.0.0.%d" % jj, 32, [ VppRoutePath( links[ii].remote_hosts[0].ip4, links[ii].sw_if_index ), VppRoutePath( links[ii].remote_hosts[1].ip4, links[ii].sw_if_index ), ], table_id=t.table_id, ).add_vpp_config() multi = VppIpMRoute( self, "0.0.0.0", "239.0.0.%d" % jj, 32, MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE, [ VppMRoutePath( self.pg0.sw_if_index, MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT, ), VppMRoutePath( self.pg1.sw_if_index, MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD, ), VppMRoutePath( self.pg2.sw_if_index, MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD, ), VppMRoutePath( self.pg3.sw_if_index, MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD, ), ], table_id=t.table_id, ).add_vpp_config() routes[ii].append({"uni": uni, "multi": multi}) # # replace the tables a few times # for kk in range(3): # replace_begin each table for t in self.tables: t.replace_begin() # all the routes are still there for ii, t in enumerate(self.tables): dump = t.dump() mdump = t.mdump() for r in routes[ii]: self.assertTrue(find_route_in_dump(dump, r["uni"], t)) self.assertTrue(find_mroute_in_dump(mdump, r["multi"], t)) # redownload the even numbered routes for ii, t in enumerate(self.tables): for jj in range(0, N_ROUTES, 2): routes[ii][jj]["uni"].add_vpp_config() routes[ii][jj]["multi"].add_vpp_config() # signal each table replace_end for t in self.tables: t.replace_end() # we should find the even routes, but not the odd for ii, t in enumerate(self.tables): dump = t.dump() mdump = t.mdump() for jj in range(0, N_ROUTES, 2): self.assertTrue(find_route_in_dump(dump, routes[ii][jj]["uni"], t)) self.assertTrue( find_mroute_in_dump(mdump, routes[ii][jj]["multi"], t) ) for jj in range(1, N_ROUTES - 1, 2): self.assertFalse(find_route_in_dump(dump, routes[ii][jj]["uni"], t)) self.assertFalse( find_mroute_in_dump(mdump, routes[ii][jj]["multi"], t) ) # reload all the routes for ii, t in enumerate(self.tables): for r in routes[ii]: r["uni"].add_vpp_config() r["multi"].add_vpp_config() # all the routes are still there for ii, t in enumerate(self.tables): dump = t.dump() mdump = t.mdump() for r in routes[ii]: self.assertTrue(find_route_in_dump(dump, r["uni"], t)) self.assertTrue(find_mroute_in_dump(mdump, r["multi"], t)) # # finally flush the tables for good measure # for t in self.tables: t.flush() self.assertEqual(len(t.dump()), 5) self.assertEqual(len(t.mdump()), 3) class TestIPCover(VppTestCase): """IPv4 Table Cover""" @classmethod def setUpClass(cls): super(TestIPCover, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIPCover, cls).tearDownClass() def setUp(self): super(TestIPCover, self).setUp() self.create_pg_interfaces(range(4)) table_id = 1 self.tables = [] for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() i.generate_remote_hosts(2) self.tables.append(VppIpTable(self, table_id).add_vpp_config()) table_id += 1 def tearDown(self): super(TestIPCover, self).tearDown() for i in self.pg_interfaces: i.admin_down() i.unconfig_ip4() def test_cover(self): """IP Table Cover""" # add a loop back with a /32 prefix lo = VppLoInterface(self) lo.admin_up() a = VppIpInterfaceAddress(self, lo, "127.0.0.1", 32).add_vpp_config() # add a neighbour that matches the loopback's /32 nbr = VppNeighbor( self, lo.sw_if_index, lo.remote_mac, "127.0.0.1" ).add_vpp_config() # add the default route which will be the cover for /32 r = VppIpRoute( self, "0.0.0.0", 0, [VppRoutePath("127.0.0.1", lo.sw_if_index)], register=False, ).add_vpp_config() # add/remove/add a longer mask cover r8 = VppIpRoute( self, "127.0.0.0", 8, [VppRoutePath("127.0.0.1", lo.sw_if_index)] ).add_vpp_config() r8.remove_vpp_config() r8.add_vpp_config() r8.remove_vpp_config() # remove the default route r.remove_vpp_config() # remove the interface prefix a.remove_vpp_config() class TestIP4Replace(VppTestCase): """IPv4 Interface Address Replace""" @classmethod def setUpClass(cls): super(TestIP4Replace, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestIP4Replace, cls).tearDownClass() def setUp(self): super(TestIP4Replace, self).setUp() self.create_pg_interfaces(range(4)) for i in self.pg_interfaces: i.admin_up() def tearDown(self): super(TestIP4Replace, self).tearDown() for i in self.pg_interfaces: i.admin_down() def get_n_pfxs(self, intf): return len(self.vapi.ip_address_dump(intf.sw_if_index)) def test_replace(self): """IP interface address replace""" intf_pfxs = [[], [], [], []] # add prefixes to each of the interfaces for i in range(len(self.pg_interfaces)): intf = self.pg_interfaces[i] # 172.16.x.1/24 addr = "172.16.%d.1" % intf.sw_if_index a = VppIpInterfaceAddress(self, intf, addr, 24).add_vpp_config() intf_pfxs[i].append(a) # 172.16.x.2/24 - a different address in the same subnet as above addr = "172.16.%d.2" % intf.sw_if_index a = VppIpInterfaceAddress(self, intf, addr, 24).add_vpp_config() intf_pfxs[i].append(a) # 172.15.x.2/24 - a different address and subnet addr = "172.15.%d.2" % intf.sw_if_index a = VppIpInterfaceAddress(self, intf, addr, 24).add_vpp_config() intf_pfxs[i].append(a) # a dump should n_address in it for intf in self.pg_interfaces: self.assertEqual(self.get_n_pfxs(intf), 3) # # remove all the address thru a replace # self.vapi.sw_interface_address_replace_begin() self.vapi.sw_interface_address_replace_end() for intf in self.pg_interfaces: self.assertEqual(self.get_n_pfxs(intf), 0) # # add all the interface addresses back # for p in intf_pfxs: for v in p: v.add_vpp_config() for intf in self.pg_interfaces: self.assertEqual(self.get_n_pfxs(intf), 3) # # replace again, but this time update/re-add the address on the first # two interfaces # self.vapi.sw_interface_address_replace_begin() for p in intf_pfxs[:2]: for v in p: v.add_vpp_config() self.vapi.sw_interface_address_replace_end() # on the first two the address still exist, # on the other two they do not for intf in self.pg_interfaces[:2]: self.assertEqual(self.get_n_pfxs(intf), 3) for p in intf_pfxs[:2]: for v in p: self.assertTrue(v.query_vpp_config()) for intf in self.pg_interfaces[2:]: self.assertEqual(self.get_n_pfxs(intf), 0) # # add all the interface addresses back on the last two # for p in intf_pfxs[2:]: for v in p: v.add_vpp_config() for intf in self.pg_interfaces: self.assertEqual(self.get_n_pfxs(intf), 3) # # replace again, this time add different prefixes on all the interfaces # self.vapi.sw_interface_address_replace_begin() pfxs = [] for intf in self.pg_interfaces: # 172.18.x.1/24 addr = "172.18.%d.1" % intf.sw_if_index pfxs.append(VppIpInterfaceAddress(self, intf, addr, 24).add_vpp_config()) self.vapi.sw_interface_address_replace_end() # only .18 should exist on each interface for intf in self.pg_interfaces: self.assertEqual(self.get_n_pfxs(intf), 1) for pfx in pfxs: self.assertTrue(pfx.query_vpp_config()) # # remove everything # self.vapi.sw_interface_address_replace_begin() self.vapi.sw_interface_address_replace_end() for intf in self.pg_interfaces: self.assertEqual(self.get_n_pfxs(intf), 0) # # add prefixes to each interface. post-begin add the prefix from # interface X onto interface Y. this would normally be an error # since it would generate a 'duplicate address' warning. but in # this case, since what is newly downloaded is sane, it's ok # for intf in self.pg_interfaces: # 172.18.x.1/24 addr = "172.18.%d.1" % intf.sw_if_index VppIpInterfaceAddress(self, intf, addr, 24).add_vpp_config() self.vapi.sw_interface_address_replace_begin() pfxs = [] for intf in self.pg_interfaces: # 172.18.x.1/24 addr = "172.18.%d.1" % (intf.sw_if_index + 1) pfxs.append(VppIpInterfaceAddress(self, intf, addr, 24).add_vpp_config()) self.vapi.sw_interface_address_replace_end() self.logger.info(self.vapi.cli("sh int addr")) for intf in self.pg_interfaces: self.assertEqual(self.get_n_pfxs(intf), 1) for pfx in pfxs: self.assertTrue(pfx.query_vpp_config()) class TestIPv4PathMTU(VppTestCase): """IPv4 Path MTU""" @classmethod def setUpClass(cls): super(TestIPv4PathMTU, cls).setUpClass() cls.create_pg_interfaces(range(2)) # setup all interfaces for i in cls.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() @classmethod def tearDownClass(cls): super(TestIPv4PathMTU, cls).tearDownClass() def test_path_mtu(self): """Path MTU""" # # The goal here is not to test that fragmentation works correctly, # that's done elsewhere, the intent is to ensure that the Path MTU # settings are honoured. # self.vapi.cli("adjacency counters enable") # set the interface MTU to a reasonable value self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1800, 0, 0, 0]) self.pg1.generate_remote_hosts(4) p_2k = ( Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / UDP(sport=1234, dport=5678) / Raw(b"0xa" * 640) ) p_1k = ( Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / UDP(sport=1234, dport=5678) / Raw(b"0xa" * 320) ) nbr = VppNeighbor( self, self.pg1.sw_if_index, self.pg1.remote_mac, self.pg1.remote_ip4 ).add_vpp_config() # this is now the interface MTU frags self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2) self.send_and_expect(self.pg0, [p_1k], self.pg1) # drop the path MTU for this neighbour to below the interface MTU # expect more frags pmtu = VppIpPathMtu(self, self.pg1.remote_ip4, 900).add_vpp_config() self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3) self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2) # print/format the adj delegate self.logger.info(self.vapi.cli("sh adj 5")) # increase the path MTU to more than the interface # expect to use the interface MTU pmtu.modify(8192) self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2) self.send_and_expect(self.pg0, [p_1k], self.pg1) # go back to an MTU from the path # wrap the call around mark-n-sweep to enusre updates clear stale self.vapi.ip_path_mtu_replace_begin() pmtu.modify(900) self.vapi.ip_path_mtu_replace_end() self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3) self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2) # raise the interface's MTU # should still use that of the path self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2000, 0, 0, 0]) self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3) self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2) # set path high and interface low pmtu.modify(2000) self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [900, 0, 0, 0]) self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3) self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2) # remove the path MTU using the mark-n-sweep semantics self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1800, 0, 0, 0]) self.vapi.ip_path_mtu_replace_begin() self.vapi.ip_path_mtu_replace_end() self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2) self.send_and_expect(self.pg0, [p_1k], self.pg1) # # set path MTU for a neighbour that doesn't exist, yet # pmtu2 = VppIpPathMtu(self, self.pg1.remote_hosts[2].ip4, 900).add_vpp_config() p_2k = ( Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_hosts[2].ip4) / UDP(sport=1234, dport=5678) / Raw(b"0xa" * 640) ) p_1k = ( Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_hosts[2].ip4) / UDP(sport=1234, dport=5678) / Raw(b"0xa" * 320) ) nbr2 = VppNeighbor( self, self.pg1.sw_if_index, self.pg1.remote_hosts[2].mac, self.pg1.remote_hosts[2].ip4, ).add_vpp_config() # should frag to the path MTU self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3) self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2) # remove and re-add the neighbour nbr2.remove_vpp_config() nbr2.add_vpp_config() # should frag to the path MTU self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3) self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2) # # set PMTUs for many peers # N_HOSTS = 16 self.pg1.generate_remote_hosts(16) self.pg1.configure_ipv4_neighbors() for h in range(N_HOSTS): pmtu = VppIpPathMtu(self, self.pg1.remote_hosts[h].ip4, 900) pmtu.add_vpp_config() self.assertTrue(pmtu.query_vpp_config()) self.logger.info(self.vapi.cli("sh ip pmtu")) dump = list(self.vapi.vpp.details_iter(self.vapi.ip_path_mtu_get)) self.assertEqual(N_HOSTS, len(dump)) for h in range(N_HOSTS): p_2k[IP].dst = self.pg1.remote_hosts[h].ip4 p_1k[IP].dst = self.pg1.remote_hosts[h].ip4 # should frag to the path MTU self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3) self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2) class TestIPv4ItfRebind(VppTestCase): """IPv4 Interface Bind w/ attached routes""" def setUp(self): super(TestIPv4ItfRebind, self).setUp() self.create_pg_interfaces(range(3)) def tearDown(self): super(TestIPv4ItfRebind, self).tearDown() def test_rebind(self): """Import to no import""" TABLE_ID = 1 tbl = VppIpTable(self, TABLE_ID).add_vpp_config() self.pg1.set_table_ip4(TABLE_ID) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() # add an attached route via an pg0 # in a different table. this prefix should import rt = VppIpRoute( self, self.pg0.local_ip4, 24, [VppRoutePath("0.0.0.0", self.pg0.sw_if_index)], table_id=TABLE_ID, ).add_vpp_config() p = ( Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) / UDP(sport=1234, dport=5678) / Raw(b"0xa" * 640) ) rx = self.send_and_expect(self.pg1, [p], self.pg0) self.assertFalse(rx[0].haslayer(ARP)) # then bind pg0 to a new table # so the prefix no longer imports self.pg0.unconfig_ip4() self.pg0.set_table_ip4(TABLE_ID) self.pg0.config_ip4() self.pg0.resolve_arp() rx = self.send_and_expect(self.pg1, [p], self.pg0) self.assertFalse(rx[0].haslayer(ARP)) # revert back to imported self.pg0.unconfig_ip4() self.pg0.set_table_ip4(0) self.pg0.config_ip4() self.pg0.resolve_arp() rx = self.send_and_expect(self.pg1, [p], self.pg0) self.assertFalse(rx[0].haslayer(ARP)) # cleanup for i in self.pg_interfaces: i.unconfig_ip4() i.set_table_ip4(0) i.admin_down() rt.remove_vpp_config() tbl.remove_vpp_config() def test_delete(self): """Swap import tables""" TABLE_ID1 = 1 tbl1_4 = VppIpTable(self, TABLE_ID1).add_vpp_config() tbl1_6 = VppIpTable(self, TABLE_ID1, True).add_vpp_config() TABLE_ID2 = 2 tbl2_4 = VppIpTable(self, TABLE_ID2).add_vpp_config() tbl2_6 = VppIpTable(self, TABLE_ID2, True).add_vpp_config() # table mappings self.pg1.set_table_ip4(TABLE_ID1) self.pg1.set_table_ip6(TABLE_ID1) self.pg2.set_table_ip4(TABLE_ID2) self.pg2.set_table_ip6(TABLE_ID2) for i in self.pg_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() # add an attached route in the default table via pg0 # this should import to table 1 rt4 = VppIpRoute( self, self.pg1.local_ip4, 24, [VppRoutePath("0.0.0.0", self.pg1.sw_if_index)], ).add_vpp_config() rt6 = VppIpRoute( self, self.pg1.local_ip6, 64, [VppRoutePath("0.0.0.0", self.pg1.sw_if_index)], ).add_vpp_config() p1 = ( Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg1.remote_ip4, dst=self.pg1.remote_ip4) / UDP(sport=1234, dport=5678) / Raw(b"0xa" * 640) ) # inject into table 0 rx = self.send_and_expect(self.pg0, [p1], self.pg1) self.assertFalse(rx[0].haslayer(ARP)) # swap the attached interface to table 2 self.pg1.unconfig_ip4() self.pg1.unconfig_ip6() self.pg1.set_table_ip4(TABLE_ID2) self.pg1.set_table_ip6(TABLE_ID2) self.pg1.config_ip4() self.pg1.config_ip6() self.pg1.resolve_arp() # delete table 1 tbl1_4.flush() tbl1_6.flush() tbl1_4.remove_vpp_config() tbl1_6.remove_vpp_config() rx = self.send_and_expect(self.pg0, [p1], self.pg1) self.assertFalse(rx[0].haslayer(ARP)) for i in self.pg_interfaces: i.unconfig_ip4() i.unconfig_ip6() i.set_table_ip4(0) i.set_table_ip6(0) i.admin_down() if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)