+ self.vapi.punt_socket_deregister(punt_ospf)
+ punts = self.vapi.punt_socket_dump(type=pt_ip)
+ self.assertEqual(len(punts), 1)
+
+ #
+ # configure a punt socket again
+ #
+ self.vapi.punt_socket_register(punt_ospf, "%s/socket_punt_3" % self.tempdir)
+ punts = self.vapi.punt_socket_dump(type=pt_ip)
+ self.assertEqual(len(punts), 2)
+
+ self.logger.info(self.vapi.cli("sh punt sock reg exception"))
+
+ #
+ # deregister all punt socket
+ #
+ self.vapi.punt_socket_deregister(punt_eigrp)
+ self.vapi.punt_socket_deregister(punt_ospf)
+ punts = self.vapi.punt_socket_dump(type=pt_ip)
+ self.assertEqual(len(punts), 0)
+
+ def verify_ospf_pkts(self, rxs, n_sent):
+ self.assertEqual(len(rxs), n_sent)
+ for rx in rxs:
+ self.assertTrue(rx.haslayer(OSPF_Hdr))
+
+ def test_traffic(self):
+ """Punt socket traffic"""
+
+ af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4
+ pt_ip = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_IP_PROTO
+ proto_ospf = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_OSPF
+
+ #
+ # configure a punt socket to capture OSPF packets
+ #
+ punt_ospf = {
+ "type": pt_ip,
+ "punt": {"ip_proto": {"af": af_ip4, "protocol": proto_ospf}},
+ }
+
+ #
+ # create packet streams and configure a punt sockets
+ #
+ pkt = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / OSPF_Hdr()
+ / OSPFv3_Hello()
+ )
+ pkts = pkt * 7
+
+ sock = self.socket_client_create("%s/socket_1" % self.tempdir)
+ self.vapi.punt_socket_register(punt_ospf, "%s/socket_1" % self.tempdir)
+
+ #
+ # send packets for each SPI we expect to be punted
+ #
+ self.send_and_assert_no_replies(self.pg0, pkts)
+
+ #
+ # verify the punted packets arrived on the associated socket
+ #
+ rx = sock.close()
+ self.verify_ospf_pkts(rx, len(pkts))
+ self.vapi.punt_socket_deregister(punt_ospf)
+
+
+class TestDot1QPuntSocket(TestPuntSocket):
+ """Punt Socket for 802.1Q (dot1q)"""
+
+ def setUp(self):
+ super(TestDot1QPuntSocket, self).setUp()
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip4()
+ i.resolve_arp()
+
+ def tearDown(self):
+ super(TestDot1QPuntSocket, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip4()
+ i.admin_down()
+
+ def test_dot1q_header_punt(self):
+ """Punt socket traffic with Dot1q header"""
+
+ port = self.ports[0]
+ pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
+ punt_l4 = set_port(mk_vpp_cfg4(), port)
+
+ # VLAN ID
+ vlan_id = 100
+
+ # Create a subinterface with the VLAN ID
+ subif = VppDot1QSubint(self, self.pg0, vlan_id)
+ subif.admin_up()
+ subif.config_ip4()
+
+ # Configure an IP address on the subinterface
+ subif_ip4 = subif.local_ip4
+
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / Dot1Q(vlan=vlan_id)
+ / IP(src=self.pg0.remote_ip4, dst=subif_ip4)
+ / UDP(sport=9876, dport=port)
+ / Raw(b"\xa5" * 100)
+ )
+
+ pkts = p * self.nr_packets
+
+ # Expect ICMP - port unreachable for all packets
+ rx = self.send_and_expect_some(self.pg0, pkts, self.pg0)
+
+ for p in rx:
+ self.assertEqual(int(p[IP].proto), 1) # ICMP
+ self.assertEqual(int(p[ICMP].code), 3) # unreachable
+
+ # Configure a punt socket
+ self.socket_client_create("%s/socket_%d" % (self.tempdir, port))
+ self.vapi.punt_socket_register(punt_l4, "%s/socket_%d" % (self.tempdir, port))
+ punts = self.vapi.punt_socket_dump(type=pt_l4)
+ self.assertEqual(len(punts), 1)
+
+ # Expect punt socket and no packets on pg0
+ self.send_and_assert_no_replies(self.pg0, pkts)
+ rx = self.socket_client_close()
+ self.logger.info("RXPKT")
+ self.logger.info(rx)
+ self.verify_udp_pkts(rx, len(pkts), port)
+ for pkt in rx:
+ self.assertEqual(pkt[Ether].src, self.pg0.remote_mac)
+ self.assertEqual(pkt[Ether].dst, self.pg0.local_mac)
+ self.assertEqual(pkt[Dot1Q].vlan, 100)
+
+ # Remove punt socket. Expect ICMP - port unreachable for all packets
+ self.vapi.punt_socket_deregister(punt_l4)
+ punts = self.vapi.punt_socket_dump(type=pt_l4)
+ self.assertEqual(len(punts), 0)
+
+ rx = self.send_and_expect_some(self.pg0, pkts, self.pg0)
+ for p in rx:
+ self.assertEqual(int(p[IP].proto), 1) # ICMP
+ self.assertEqual(int(p[ICMP].code), 3) # unreachable
+
+
+@tag_fixme_vpp_workers
+class TestPunt(VppTestCase):
+ """Exception Punt Test Case"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestPunt, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestPunt, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestPunt, self).setUp()
+
+ self.create_pg_interfaces(range(4))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip4()
+ i.resolve_arp()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ for i in self.pg_interfaces:
+ i.unconfig_ip4()
+ i.unconfig_ip6()
+ i.admin_down()
+ super(TestPunt, self).tearDown()
+
+ def test_punt(self):
+ """Exception Path testing"""
+
+ #
+ # dump the punt registered reasons
+ # search for a few we know should be there
+ #
+ rs = self.vapi.punt_reason_dump()
+
+ reasons = [
+ "ipsec6-no-such-tunnel",
+ "ipsec4-no-such-tunnel",
+ "ipsec4-spi-o-udp-0",
+ ]
+
+ for reason in reasons:
+ found = False
+ for r in rs:
+ if r.reason.name == reason:
+ found = True
+ break
+ self.assertTrue(found)
+
+ #
+ # Using the test CLI we will hook in a exception path to
+ # send ACL deny packets out of pg0 and pg1.
+ # the ACL is src,dst = 1.1.1.1,1.1.1.2
+ #
+ ip_1_1_1_2 = VppIpRoute(
+ self,
+ "1.1.1.2",
+ 32,
+ [VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index)],
+ )
+ ip_1_1_1_2.add_vpp_config()
+ ip_1_2 = VppIpRoute(
+ self,
+ "1::2",
+ 128,
+ [
+ VppRoutePath(
+ self.pg3.remote_ip6,
+ self.pg3.sw_if_index,
+ proto=DpoProto.DPO_PROTO_IP6,
+ )
+ ],
+ )
+ ip_1_2.add_vpp_config()
+
+ p4 = (
+ Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ p6 = (
+ Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
+ / IPv6(src="1::1", dst="1::2")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ self.send_and_expect(self.pg2, p4 * 1, self.pg3)
+ self.send_and_expect(self.pg2, p6 * 1, self.pg3)
+
+ #
+ # apply the punting features
+ #
+ self.vapi.cli("test punt pg2")
+
+ #
+ # dump the punt reasons to learn the IDs assigned
+ #
+ rs = self.vapi.punt_reason_dump(reason={"name": "reason-v4"})
+ r4 = rs[0].reason.id
+ rs = self.vapi.punt_reason_dump(reason={"name": "reason-v6"})
+ r6 = rs[0].reason.id
+
+ #
+ # pkts now dropped
+ #
+ self.send_and_assert_no_replies(self.pg2, p4 * NUM_PKTS)
+ self.send_and_assert_no_replies(self.pg2, p6 * NUM_PKTS)
+
+ #
+ # Check state:
+ # 1 - node error counters
+ # 2 - per-reason counters
+ # 2, 3 are the index of the assigned punt reason
+ #
+ stats = self.statistics.get_err_counter("/err/punt-dispatch/No registrations")
+ self.assertEqual(stats, 2 * NUM_PKTS)
+
+ stats = self.statistics.get_counter("/net/punt")
+ self.assertEqual(stats[0][r4]["packets"], NUM_PKTS)
+ self.assertEqual(stats[0][r6]["packets"], NUM_PKTS)
+
+ #
+ # use the test CLI to test a client that punts exception
+ # packets out of pg0
+ #
+ self.vapi.cli("test punt pg0 %s" % self.pg0.remote_ip4)
+ self.vapi.cli("test punt pg0 %s" % self.pg0.remote_ip6)
+
+ rx4s = self.send_and_expect(self.pg2, p4 * NUM_PKTS, self.pg0)
+ rx6s = self.send_and_expect(self.pg2, p6 * NUM_PKTS, self.pg0)
+
+ #
+ # check the packets come out IP unmodified but destined to pg0 host
+ #
+ for rx in rx4s:
+ self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
+ self.assertEqual(rx[Ether].src, self.pg0.local_mac)
+ self.assertEqual(p4[IP].dst, rx[IP].dst)
+ self.assertEqual(p4[IP].ttl, rx[IP].ttl)
+ for rx in rx6s:
+ self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
+ self.assertEqual(rx[Ether].src, self.pg0.local_mac)
+ self.assertEqual(p6[IPv6].dst, rx[IPv6].dst)
+ self.assertEqual(p6[IPv6].hlim, rx[IPv6].hlim)
+
+ stats = self.statistics.get_counter("/net/punt")
+ self.assertEqual(stats[0][r4]["packets"], 2 * NUM_PKTS)
+ self.assertEqual(stats[0][r6]["packets"], 2 * NUM_PKTS)
+
+ #
+ # add another registration for the same reason to send packets
+ # out of pg1
+ #
+ self.vapi.cli("test punt pg1 %s" % self.pg1.remote_ip4)
+ self.vapi.cli("test punt pg1 %s" % self.pg1.remote_ip6)
+