test/test_punt.py

   1 #!/usr/bin/env python3
   2 import binascii
   3 import random
   4 import socket
   5 import os
   6 import threading
   7 import struct
   8 import copy
   9 import fcntl
  10 import time
  11
  12 from struct import unpack, unpack_from
  13
  14 try:
  15     import unittest2 as unittest
  16 except ImportError:
  17     import unittest
  18
  19 from util import ppp, ppc
  20 from re import compile
  21 import scapy.compat
  22 from scapy.packet import Raw
  23 from scapy.layers.l2 import Ether
  24 from scapy.layers.inet import IP, UDP, ICMP
  25 from scapy.layers.ipsec import ESP
  26 import scapy.layers.inet6 as inet6
  27 from scapy.layers.inet6 import IPv6, ICMPv6DestUnreach
  28 from scapy.contrib.ospf import OSPF_Hdr, OSPFv3_Hello
  29 from framework import tag_fixme_vpp_workers
  30 from framework import VppTestCase, VppTestRunner
  31
  32 from vpp_ip import DpoProto
  33 from vpp_ip_route import VppIpRoute, VppRoutePath
  34 from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect, VppIpsecInterface
  35 from vpp_papi import VppEnum
  36
  37 NUM_PKTS = 67
  38
  39
  40 class serverSocketThread(threading.Thread):
  41     """ Socket server thread"""
  42
  43     def __init__(self, threadID, sockName):
  44         threading.Thread.__init__(self)
  45         self.threadID = threadID
  46         self.sockName = sockName
  47         self.sock = None
  48         self.rx_pkts = []
  49         self.keep_running = True
  50
  51     def rx_packets(self):
  52         # Wait for some packets on socket
  53         while self.keep_running:
  54             try:
  55                 data = self.sock.recv(65536)
  56
  57                 # punt socket metadata
  58                 # packet_desc = data[0:8]
  59
  60                 # Ethernet
  61                 self.rx_pkts.append(Ether(data[8:]))
  62             except IOError as e:
  63                 if e.errno == 11:
  64                     # nothing to receive, sleep a little
  65                     time.sleep(0.1)
  66                     pass
  67                 else:
  68                     raise
  69
  70     def run(self):
  71         self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
  72         try:
  73             os.unlink(self.sockName)
  74         except:
  75             pass
  76         self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)
  77         self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)
  78         fcntl.fcntl(self.sock, fcntl.F_SETFL, os.O_NONBLOCK)
  79         self.sock.bind(self.sockName)
  80
  81         self.rx_packets()
  82
  83     def close(self):
  84         self.sock.close()
  85         self.keep_running = False
  86         return self.rx_pkts
  87
  88
  89 class TestPuntSocket(VppTestCase):
  90     """ Punt Socket """
  91
  92     ports = [1111, 2222, 3333, 4444]
  93     sock_servers = list()
  94     # FIXME: nr_packets > 3 results in failure
  95     # nr_packets = 3 makes the test unstable
  96     nr_packets = 2
  97
  98     @classmethod
  99     def setUpClass(cls):
 100         super(TestPuntSocket, cls).setUpClass()
 101
 102     @classmethod
 103     def tearDownClass(cls):
 104         super(TestPuntSocket, cls).tearDownClass()
 105
 106     @classmethod
 107     def setUpConstants(cls):
 108         cls.extra_vpp_punt_config = [
 109             "punt", "{", "socket", cls.tempdir+"/socket_punt", "}"]
 110         super(TestPuntSocket, cls).setUpConstants()
 111
 112     def setUp(self):
 113         super(TestPuntSocket, self).setUp()
 114         random.seed()
 115
 116         self.create_pg_interfaces(range(2))
 117         for i in self.pg_interfaces:
 118             i.admin_up()
 119
 120     def tearDown(self):
 121         del self.sock_servers[:]
 122         super(TestPuntSocket, self).tearDown()
 123
 124     def socket_client_create(self, sock_name, id=None):
 125         thread = serverSocketThread(id, sock_name)
 126         self.sock_servers.append(thread)
 127         thread.start()
 128         return thread
 129
 130     def socket_client_close(self):
 131         rx_pkts = []
 132         for thread in self.sock_servers:
 133             rx_pkts += thread.close()
 134             thread.join()
 135         return rx_pkts
 136
 137     def verify_port(self, pr, vpr):
 138         self.assertEqual(vpr.punt.type, pr['type'])
 139         self.assertEqual(vpr.punt.punt.l4.port,
 140                          pr['punt']['l4']['port'])
 141         self.assertEqual(vpr.punt.punt.l4.protocol,
 142                          pr['punt']['l4']['protocol'])
 143         self.assertEqual(vpr.punt.punt.l4.af,
 144                          pr['punt']['l4']['af'])
 145
 146     def verify_exception(self, pr, vpr):
 147         self.assertEqual(vpr.punt.type, pr['type'])
 148         self.assertEqual(vpr.punt.punt.exception.id,
 149                          pr['punt']['exception']['id'])
 150
 151     def verify_ip_proto(self, pr, vpr):
 152         self.assertEqual(vpr.punt.type, pr['type'])
 153         self.assertEqual(vpr.punt.punt.ip_proto.af,
 154                          pr['punt']['ip_proto']['af'])
 155         self.assertEqual(vpr.punt.punt.ip_proto.protocol,
 156                          pr['punt']['ip_proto']['protocol'])
 157
 158     def verify_udp_pkts(self, rxs, n_rx, port):
 159         n_match = 0
 160         for rx in rxs:
 161             self.assertTrue(rx.haslayer(UDP))
 162             if rx[UDP].dport == port:
 163                 n_match += 1
 164         self.assertEqual(n_match, n_rx)
 165
 166
 167 def set_port(pr, port):
 168     pr['punt']['l4']['port'] = port
 169     return pr
 170
 171
 172 def set_reason(pr, reason):
 173     pr['punt']['exception']['id'] = reason
 174     return pr
 175
 176
 177 def mk_vpp_cfg4():
 178     pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
 179     af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4
 180     udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
 181     punt_l4 = {
 182         'type': pt_l4,
 183         'punt': {
 184             'l4': {
 185                 'af': af_ip4,
 186                 'protocol': udp_proto
 187             }
 188         }
 189     }
 190     return punt_l4
 191
 192
 193 def mk_vpp_cfg6():
 194     pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
 195     af_ip6 = VppEnum.vl_api_address_family_t.ADDRESS_IP6
 196     udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
 197     punt_l4 = {
 198         'type': pt_l4,
 199         'punt': {
 200             'l4': {
 201                 'af': af_ip6,
 202                 'protocol': udp_proto
 203             }
 204         }
 205     }
 206     return punt_l4
 207
 208
 209 class TestIP4PuntSocket(TestPuntSocket):
 210     """ Punt Socket for IPv4 UDP """
 211
 212     @classmethod
 213     def setUpClass(cls):
 214         super(TestIP4PuntSocket, cls).setUpClass()
 215
 216     @classmethod
 217     def tearDownClass(cls):
 218         super(TestIP4PuntSocket, cls).tearDownClass()
 219
 220     def setUp(self):
 221         super(TestIP4PuntSocket, self).setUp()
 222
 223         for i in self.pg_interfaces:
 224             i.config_ip4()
 225             i.resolve_arp()
 226
 227     def tearDown(self):
 228         super(TestIP4PuntSocket, self).tearDown()
 229         for i in self.pg_interfaces:
 230             i.unconfig_ip4()
 231             i.admin_down()
 232
 233     def test_punt_socket_dump(self):
 234         """ Punt socket registration/deregistration"""
 235
 236         pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
 237         af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4
 238         udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
 239
 240         punts = self.vapi.punt_socket_dump(type=pt_l4)
 241         self.assertEqual(len(punts), 0)
 242
 243         #
 244         # configure a punt socket
 245         #
 246         punt_l4 = mk_vpp_cfg4()
 247
 248         self.vapi.punt_socket_register(set_port(punt_l4, 1111),
 249                                        "%s/socket_punt_1111" % self.tempdir)
 250         self.vapi.punt_socket_register(set_port(punt_l4, 2222),
 251                                        "%s/socket_punt_2222" % self.tempdir)
 252         punts = self.vapi.punt_socket_dump(type=pt_l4)
 253         self.assertEqual(len(punts), 2)
 254         self.verify_port(set_port(punt_l4, 1111), punts[0])
 255         self.verify_port(set_port(punt_l4, 2222), punts[1])
 256
 257         #
 258         # deregister a punt socket
 259         #
 260         self.vapi.punt_socket_deregister(set_port(punt_l4, 1111))
 261         punts = self.vapi.punt_socket_dump(type=pt_l4)
 262         self.assertEqual(len(punts), 1)
 263
 264         #
 265         # configure a punt socket again
 266         #
 267         self.vapi.punt_socket_register(set_port(punt_l4, 1111),
 268                                        "%s/socket_punt_1111" % self.tempdir)
 269         self.vapi.punt_socket_register(set_port(punt_l4, 3333),
 270                                        "%s/socket_punt_3333" % self.tempdir)
 271         punts = self.vapi.punt_socket_dump(type=pt_l4)
 272         self.assertEqual(len(punts), 3)
 273
 274         self.logger.info(self.vapi.cli("sh punt sock reg"))
 275
 276         #
 277         # deregister all punt socket
 278         #
 279         self.vapi.punt_socket_deregister(set_port(punt_l4, 1111))
 280         self.vapi.punt_socket_deregister(set_port(punt_l4, 2222))
 281         self.vapi.punt_socket_deregister(set_port(punt_l4, 3333))
 282         punts = self.vapi.punt_socket_dump(type=pt_l4)
 283         self.assertEqual(len(punts), 0)
 284
 285     def test_punt_socket_traffic_single_port_single_socket(self):
 286         """ Punt socket traffic single port single socket"""
 287
 288         port = self.ports[0]
 289         pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
 290         punt_l4 = set_port(mk_vpp_cfg4(), port)
 291
 292         p = (Ether(src=self.pg0.remote_mac,
 293                    dst=self.pg0.local_mac) /
 294              IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
 295              UDP(sport=9876, dport=port) /
 296              Raw(b'\xa5' * 100))
 297
 298         pkts = p * self.nr_packets
 299
 300         punts = self.vapi.punt_socket_dump(type=pt_l4)
 301         self.assertEqual(len(punts), 0)
 302
 303         #
 304         # expect ICMP - port unreachable for all packets
 305         #
 306         rx = self.send_and_expect(self.pg0, pkts, self.pg0)
 307
 308         for p in rx:
 309             self.assertEqual(int(p[IP].proto), 1)   # ICMP
 310             self.assertEqual(int(p[ICMP].code), 3)  # unreachable
 311
 312         #
 313         # configure a punt socket
 314         #
 315         self.socket_client_create("%s/socket_%d" % (self.tempdir, port))
 316         self.vapi.punt_socket_register(punt_l4, "%s/socket_%d" %
 317                                        (self.tempdir, port))
 318         punts = self.vapi.punt_socket_dump(type=pt_l4)
 319         self.assertEqual(len(punts), 1)
 320
 321         #
 322         # expect punt socket and no packets on pg0
 323         #
 324         self.send_and_assert_no_replies(self.pg0, pkts)
 325         rx = self.socket_client_close()
 326         self.verify_udp_pkts(rx, len(pkts), port)
 327
 328         #
 329         # remove punt socket. expect ICMP - port unreachable for all packets
 330         #
 331         self.vapi.punt_socket_deregister(punt_l4)
 332         punts = self.vapi.punt_socket_dump(type=pt_l4)
 333         self.assertEqual(len(punts), 0)
 334
 335         rx = self.send_and_expect(self.pg0, pkts, self.pg0)
 336         for p in rx:
 337             self.assertEqual(int(p[IP].proto), 1)   # ICMP
 338             self.assertEqual(int(p[ICMP].code), 3)  # unreachable
 339
 340     def test_punt_socket_traffic_multi_ports_multi_sockets(self):
 341         """ Punt socket traffic multi ports and multi sockets"""
 342
 343         punt_l4 = mk_vpp_cfg4()
 344
 345         # configuration for each UDP port
 346         cfgs = dict()
 347
 348         #
 349         # create stream of packets for each port
 350         #
 351         for port in self.ports:
 352             # choose port from port list
 353             cfgs[port] = {}
 354
 355             pkt = (Ether(src=self.pg0.remote_mac,
 356                          dst=self.pg0.local_mac) /
 357                    IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
 358                    UDP(sport=9876, dport=port) /
 359                    Raw(b'\xa5' * 100))
 360             cfgs[port]['pkts'] = pkt * self.nr_packets
 361             cfgs[port]['port'] = port
 362             cfgs[port]['vpp'] = copy.deepcopy(set_port(punt_l4, port))
 363
 364             # configure punt sockets
 365             cfgs[port]['sock'] = self.socket_client_create(
 366                 "%s/socket_%d" % (self.tempdir, port))
 367             self.vapi.punt_socket_register(
 368                 cfgs[port]['vpp'],
 369                 "%s/socket_%d" % (self.tempdir, port))
 370
 371         #
 372         # send the packets that get punted
 373         #
 374         for cfg in cfgs.values():
 375             self.send_and_assert_no_replies(self.pg0, cfg['pkts'])
 376
 377         #
 378         # test that we got the excepted packets on the expected socket
 379         #
 380         for cfg in cfgs.values():
 381             rx = cfg['sock'].close()
 382             self.verify_udp_pkts(rx, len(cfg['pkts']), cfg['port'])
 383             self.vapi.punt_socket_deregister(cfg['vpp'])
 384
 385     def test_punt_socket_traffic_multi_ports_single_socket(self):
 386         """ Punt socket traffic multi ports and single socket"""
 387
 388         pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
 389         punt_l4 = mk_vpp_cfg4()
 390
 391         #
 392         # create stream of packets with each port
 393         #
 394         pkts = []
 395         for port in self.ports:
 396             # choose port from port list
 397             pkt = (Ether(src=self.pg0.remote_mac,
 398                          dst=self.pg0.local_mac) /
 399                    IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
 400                    UDP(sport=9876, dport=port) /
 401                    Raw(b'\xa5' * 100))
 402             pkts += pkt * self.nr_packets
 403
 404         #
 405         # configure a punt socket
 406         #
 407         self.socket_client_create("%s/socket_multi" % self.tempdir)
 408         for p in self.ports:
 409             self.vapi.punt_socket_register(set_port(punt_l4, p),
 410                                            "%s/socket_multi" % self.tempdir)
 411         punts = self.vapi.punt_socket_dump(type=pt_l4)
 412         self.assertEqual(len(punts), len(self.ports))
 413
 414         #
 415         # expect punt socket and no packets on pg0
 416         #
 417         self.send_and_assert_no_replies(self.pg0, pkts)
 418         self.logger.info(self.vapi.cli("show trace"))
 419         rx = self.socket_client_close()
 420
 421         for p in self.ports:
 422             self.verify_udp_pkts(rx, self.nr_packets, p)
 423             self.vapi.punt_socket_deregister(set_port(punt_l4, p))
 424         punts = self.vapi.punt_socket_dump(type=pt_l4)
 425         self.assertEqual(len(punts), 0)
 426
 427
 428 class TestIP6PuntSocket(TestPuntSocket):
 429     """ Punt Socket for IPv6 UDP """
 430
 431     @classmethod
 432     def setUpClass(cls):
 433         super(TestIP6PuntSocket, cls).setUpClass()
 434
 435     @classmethod
 436     def tearDownClass(cls):
 437         super(TestIP6PuntSocket, cls).tearDownClass()
 438
 439     def setUp(self):
 440         super(TestIP6PuntSocket, self).setUp()
 441
 442         for i in self.pg_interfaces:
 443             i.config_ip6()
 444             i.resolve_ndp()
 445
 446     def tearDown(self):
 447         super(TestIP6PuntSocket, self).tearDown()
 448         for i in self.pg_interfaces:
 449             i.unconfig_ip6()
 450             i.admin_down()
 451
 452     def test_punt_socket_dump(self):
 453         """ Punt socket registration """
 454
 455         pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
 456         af_ip6 = VppEnum.vl_api_address_family_t.ADDRESS_IP6
 457         udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
 458         #
 459         # configure a punt socket
 460         #
 461         punt_l4 = {
 462             'type': pt_l4,
 463             'punt': {
 464                 'l4': {
 465                     'af': af_ip6,
 466                     'protocol': udp_proto
 467                 }
 468             }
 469         }
 470
 471         punts = self.vapi.punt_socket_dump(type=pt_l4)
 472         self.assertEqual(len(punts), 0)
 473
 474         #
 475         # configure a punt socket
 476         #
 477         self.vapi.punt_socket_register(set_port(punt_l4, 1111),
 478                                        "%s/socket_1111" % self.tempdir)
 479         self.vapi.punt_socket_register(set_port(punt_l4, 2222),
 480                                        "%s/socket_2222" % self.tempdir)
 481         punts = self.vapi.punt_socket_dump(type=pt_l4)
 482         self.assertEqual(len(punts), 2)
 483         self.verify_port(set_port(punt_l4, 1111), punts[0])
 484         self.verify_port(set_port(punt_l4, 2222), punts[1])
 485
 486         #
 487         # deregister a punt socket
 488         #
 489         self.vapi.punt_socket_deregister(set_port(punt_l4, 1111))
 490         punts = self.vapi.punt_socket_dump(type=pt_l4)
 491         self.assertEqual(len(punts), 1)
 492
 493         #
 494         # configure a punt socket again
 495         #
 496         self.vapi.punt_socket_register(set_port(punt_l4, 1111),
 497                                        "%s/socket_1111" % self.tempdir)
 498         punts = self.vapi.punt_socket_dump(type=pt_l4)
 499         self.assertEqual(len(punts), 2)
 500
 501         #
 502         # deregister all punt socket
 503         #
 504         self.vapi.punt_socket_deregister(set_port(punt_l4, 1111))
 505         self.vapi.punt_socket_deregister(set_port(punt_l4, 2222))
 506         self.vapi.punt_socket_deregister(set_port(punt_l4, 3333))
 507         punts = self.vapi.punt_socket_dump(type=pt_l4)
 508         self.assertEqual(len(punts), 0)
 509
 510     def test_punt_socket_traffic_single_port_single_socket(self):
 511         """ Punt socket traffic single port single socket"""
 512
 513         port = self.ports[0]
 514         pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
 515         af_ip6 = VppEnum.vl_api_address_family_t.ADDRESS_IP6
 516         udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
 517         punt_l4 = {
 518             'type': pt_l4,
 519             'punt': {
 520                 'l4': {
 521                     'af': af_ip6,
 522                     'protocol': udp_proto,
 523                     'port': port,
 524                 }
 525             }
 526         }
 527
 528         p = (Ether(src=self.pg0.remote_mac,
 529                    dst=self.pg0.local_mac) /
 530              IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
 531              inet6.UDP(sport=9876, dport=port) /
 532              Raw(b'\xa5' * 100))
 533
 534         pkts = p * self.nr_packets
 535
 536         punts = self.vapi.punt_socket_dump(type=pt_l4)
 537         self.assertEqual(len(punts), 0)
 538
 539         #
 540         # expect ICMPv6 - destination unreachable for all packets
 541         #
 542         self.vapi.cli("clear trace")
 543         self.pg0.add_stream(pkts)
 544         self.pg_enable_capture(self.pg_interfaces)
 545         self.pg_start()
 546         # FIXME - when punt socket deregister is implemented
 547         # rx = self.pg0.get_capture(self.nr_packets)
 548         # for p in rx:
 549         #     self.assertEqual(int(p[IPv6].nh), 58)                # ICMPv6
 550         #     self.assertEqual(int(p[ICMPv6DestUnreach].code),4)  # unreachable
 551
 552         #
 553         # configure a punt socket
 554         #
 555         self.socket_client_create("%s/socket_%d" % (self.tempdir, port))
 556         self.vapi.punt_socket_register(punt_l4, "%s/socket_%d" %
 557                                        (self.tempdir, port))
 558         punts = self.vapi.punt_socket_dump(type=pt_l4)
 559         self.assertEqual(len(punts), 1)
 560
 561         #
 562         # expect punt socket and no packets on pg0
 563         #
 564         self.vapi.cli("clear errors")
 565         self.vapi.cli("clear trace")
 566         self.pg0.add_stream(pkts)
 567         self.pg_enable_capture(self.pg_interfaces)
 568         self.pg_start()
 569         self.pg0.get_capture(0)
 570         self.logger.info(self.vapi.cli("show trace"))
 571         rx = self.socket_client_close()
 572         self.verify_udp_pkts(rx, len(pkts), port)
 573
 574         #
 575         # remove punt socket. expect ICMP - dest. unreachable for all packets
 576         #
 577         self.vapi.punt_socket_deregister(punt_l4)
 578         punts = self.vapi.punt_socket_dump(type=pt_l4)
 579         self.assertEqual(len(punts), 0)
 580         self.pg0.add_stream(pkts)
 581         self.pg_enable_capture(self.pg_interfaces)
 582         self.pg_start()
 583         # FIXME - when punt socket deregister is implemented
 584         # self.pg0.get_capture(nr_packets)
 585
 586     def test_punt_socket_traffic_multi_ports_multi_sockets(self):
 587         """ Punt socket traffic multi ports and multi sockets"""
 588
 589         punt_l4 = mk_vpp_cfg6()
 590
 591         # configuration for each UDP port
 592         cfgs = dict()
 593
 594         #
 595         # create stream of packets for each port
 596         #
 597         for port in self.ports:
 598             # choose port from port list
 599             cfgs[port] = {}
 600
 601             pkt = (Ether(src=self.pg0.remote_mac,
 602                          dst=self.pg0.local_mac) /
 603                    IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
 604                    UDP(sport=9876, dport=port) /
 605                    Raw(b'\xa5' * 100))
 606             cfgs[port]['pkts'] = pkt * self.nr_packets
 607             cfgs[port]['port'] = port
 608             cfgs[port]['vpp'] = copy.deepcopy(set_port(punt_l4, port))
 609
 610             # configure punt sockets
 611             cfgs[port]['sock'] = self.socket_client_create(
 612                 "%s/socket_%d" % (self.tempdir, port))
 613             self.vapi.punt_socket_register(
 614                 cfgs[port]['vpp'],
 615                 "%s/socket_%d" % (self.tempdir, port))
 616
 617         #
 618         # send the packets that get punted
 619         #
 620         for cfg in cfgs.values():
 621             self.send_and_assert_no_replies(self.pg0, cfg['pkts'])
 622
 623         #
 624         # test that we got the excepted packets on the expected socket
 625         #
 626         for cfg in cfgs.values():
 627             rx = cfg['sock'].close()
 628             self.verify_udp_pkts(rx, len(cfg['pkts']), cfg['port'])
 629             self.vapi.punt_socket_deregister(cfg['vpp'])
 630
 631     def test_punt_socket_traffic_multi_ports_single_socket(self):
 632         """ Punt socket traffic multi ports and single socket"""
 633
 634         pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
 635         af_ip6 = VppEnum.vl_api_address_family_t.ADDRESS_IP6
 636         udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
 637         punt_l4 = {
 638             'type': pt_l4,
 639             'punt': {
 640                 'l4': {
 641                     'af': af_ip6,
 642                     'protocol': udp_proto,
 643                 }
 644             }
 645         }
 646
 647         #
 648         # create stream of packets with each port
 649         #
 650         pkts = []
 651         for port in self.ports:
 652             # choose port from port list
 653             pkt = (Ether(src=self.pg0.remote_mac,
 654                          dst=self.pg0.local_mac) /
 655                    IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
 656                    UDP(sport=9876, dport=port) /
 657                    Raw(b'\xa5' * 100))
 658             pkts += pkt * self.nr_packets
 659
 660         #
 661         # no punt socket
 662         #
 663         punts = self.vapi.punt_socket_dump(type=pt_l4)
 664         self.assertEqual(len(punts), 0)
 665
 666         #
 667         # configure a punt socket
 668         #
 669         self.socket_client_create("%s/socket_multi" % self.tempdir)
 670         for p in self.ports:
 671             self.vapi.punt_socket_register(set_port(punt_l4, p),
 672                                            "%s/socket_multi" % self.tempdir)
 673         punts = self.vapi.punt_socket_dump(type=pt_l4)
 674         self.assertEqual(len(punts), len(self.ports))
 675
 676         #
 677         # expect punt socket and no packets on pg0
 678         #
 679         self.vapi.cli("clear errors")
 680         self.vapi.cli("clear trace")
 681         self.pg0.add_stream(pkts)
 682         self.pg_enable_capture(self.pg_interfaces)
 683         self.pg_start()
 684         # give a chance to punt socket to collect all packets
 685         self.sleep(1)
 686         self.pg0.get_capture(0)
 687         rx = self.socket_client_close()
 688
 689         for p in self.ports:
 690             self.verify_udp_pkts(rx, self.nr_packets, p)
 691             self.vapi.punt_socket_deregister(set_port(punt_l4, p))
 692         punts = self.vapi.punt_socket_dump(type=pt_l4)
 693         self.assertEqual(len(punts), 0)
 694
 695
 696 class TestExceptionPuntSocket(TestPuntSocket):
 697     """ Punt Socket for Exceptions """
 698
 699     @classmethod
 700     def setUpClass(cls):
 701         super(TestExceptionPuntSocket, cls).setUpClass()
 702
 703     @classmethod
 704     def tearDownClass(cls):
 705         super(TestExceptionPuntSocket, cls).tearDownClass()
 706
 707     def setUp(self):
 708         super(TestExceptionPuntSocket, self).setUp()
 709
 710         self.create_pg_interfaces(range(2))
 711         for i in self.pg_interfaces:
 712             i.config_ip4()
 713             i.resolve_arp()
 714
 715     def tearDown(self):
 716         super(TestExceptionPuntSocket, self).tearDown()
 717         for i in self.pg_interfaces:
 718             i.unconfig_ip4()
 719             i.admin_down()
 720
 721     def test_registration(self):
 722         """ Punt socket registration/deregistration"""
 723
 724         pt_ex = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_EXCEPTION
 725
 726         punts = self.vapi.punt_socket_dump(type=pt_ex)
 727         self.assertEqual(len(punts), 0)
 728
 729         #
 730         # configure a punt socket
 731         #
 732         punt_ex = {
 733             'type': pt_ex,
 734             'punt': {
 735                 'exception': {}
 736             }
 737         }
 738
 739         self.vapi.punt_socket_register(set_reason(punt_ex, 1),
 740                                        "%s/socket_punt_1" % self.tempdir)
 741         self.vapi.punt_socket_register(set_reason(punt_ex, 2),
 742                                        "%s/socket_punt_2" % self.tempdir)
 743         punts = self.vapi.punt_socket_dump(type=pt_ex)
 744         self.assertEqual(len(punts), 2)
 745         self.verify_exception(set_reason(punt_ex, 1), punts[0])
 746         self.verify_exception(set_reason(punt_ex, 2), punts[1])
 747
 748         #
 749         # deregister a punt socket
 750         #
 751         self.vapi.punt_socket_deregister(set_reason(punt_ex, 1))
 752         punts = self.vapi.punt_socket_dump(type=pt_ex)
 753         self.assertEqual(len(punts), 1)
 754
 755         #
 756         # configure a punt socket again
 757         #
 758         self.vapi.punt_socket_register(set_reason(punt_ex, 1),
 759                                        "%s/socket_punt_1" % self.tempdir)
 760         self.vapi.punt_socket_register(set_reason(punt_ex, 3),
 761                                        "%s/socket_punt_3" % self.tempdir)
 762         punts = self.vapi.punt_socket_dump(type=pt_ex)
 763         self.assertEqual(len(punts), 3)
 764
 765         self.logger.info(self.vapi.cli("sh punt sock reg exception"))
 766
 767         #
 768         # deregister all punt socket
 769         #
 770         self.vapi.punt_socket_deregister(set_reason(punt_ex, 1))
 771         self.vapi.punt_socket_deregister(set_reason(punt_ex, 2))
 772         self.vapi.punt_socket_deregister(set_reason(punt_ex, 3))
 773         punts = self.vapi.punt_socket_dump(type=pt_ex)
 774         self.assertEqual(len(punts), 0)
 775
 776     def verify_esp_pkts(self, rxs, n_sent, spi, has_udp):
 777         self.assertEqual(len(rxs), n_sent)
 778         for rx in rxs:
 779             self.assertTrue(rx.haslayer(IP))
 780             self.assertTrue(rx.haslayer(ESP))
 781             self.assertEqual(rx[ESP].spi, spi)
 782             if has_udp:
 783                 self.assertTrue(rx.haslayer(UDP))
 784
 785     def test_traffic(self):
 786         """ Punt socket traffic """
 787
 788         port = self.ports[0]
 789         pt_ex = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_EXCEPTION
 790         punt_ex = {
 791             'type': pt_ex,
 792             'punt': {
 793                 'exception': {}
 794             }
 795         }
 796
 797         #
 798         # we're dealing with IPSec tunnels punting for no-such-tunnel
 799         # (SPI=0 goes to ikev2)
 800         #
 801         cfgs = dict()
 802         cfgs['ipsec4-no-such-tunnel'] = {'spi': 99,
 803                                          'udp': False,
 804                                          'itf': self.pg0}
 805
 806         #
 807         # find the VPP ID for these punt exception reasin
 808         #
 809         rs = self.vapi.punt_reason_dump()
 810         for key in cfgs:
 811             for r in rs:
 812                 print(r.reason.name)
 813                 print(key)
 814                 if r.reason.name == key:
 815                     cfgs[key]['id'] = r.reason.id
 816                     cfgs[key]['vpp'] = copy.deepcopy(
 817                         set_reason(punt_ex,
 818                                    cfgs[key]['id']))
 819                     break
 820
 821         #
 822         # configure punt sockets
 823         #
 824         for cfg in cfgs.values():
 825             cfg['sock'] = self.socket_client_create("%s/socket_%d" %
 826                                                     (self.tempdir, cfg['id']))
 827             self.vapi.punt_socket_register(
 828                 cfg['vpp'], "%s/socket_%d" % (self.tempdir, cfg['id']))
 829
 830         #
 831         # create packet streams for 'no-such-tunnel' exception
 832         #
 833         for cfg in cfgs.values():
 834             pkt = (Ether(src=cfg['itf'].remote_mac,
 835                          dst=cfg['itf'].local_mac) /
 836                    IP(src=cfg['itf'].remote_ip4,
 837                       dst=cfg['itf'].local_ip4))
 838             if (cfg['udp']):
 839                 pkt = pkt / UDP(sport=666, dport=4500)
 840             pkt = (pkt / ESP(spi=cfg['spi'], seq=3) /
 841                    Raw(b'\xa5' * 100))
 842             cfg['pkts'] = [pkt]
 843
 844         #
 845         # send packets for each SPI we expect to be punted
 846         #
 847         for cfg in cfgs.values():
 848             self.send_and_assert_no_replies(cfg['itf'], cfg['pkts'])
 849
 850         #
 851         # verify the punted packets arrived on the associated socket
 852         #
 853         for cfg in cfgs.values():
 854             rx = cfg['sock'].close()
 855             self.verify_esp_pkts(rx, len(cfg['pkts']),
 856                                  cfg['spi'], cfg['udp'])
 857
 858         #
 859         # add some tunnels, make sure it still punts
 860         #
 861         tun = VppIpsecInterface(self).add_vpp_config()
 862         sa_in = VppIpsecSA(self, 11, 11,
 863                            (VppEnum.vl_api_ipsec_integ_alg_t.
 864                             IPSEC_API_INTEG_ALG_SHA1_96),
 865                            b"0123456701234567",
 866                            (VppEnum.vl_api_ipsec_crypto_alg_t.
 867                             IPSEC_API_CRYPTO_ALG_AES_CBC_128),
 868                            b"0123456701234567",
 869                            50,
 870                            self.pg0.local_ip4,
 871                            self.pg0.remote_ip4).add_vpp_config()
 872         sa_out = VppIpsecSA(self, 22, 22,
 873                             (VppEnum.vl_api_ipsec_integ_alg_t.
 874                              IPSEC_API_INTEG_ALG_SHA1_96),
 875                             b"0123456701234567",
 876                             (VppEnum.vl_api_ipsec_crypto_alg_t.
 877                              IPSEC_API_CRYPTO_ALG_AES_CBC_128),
 878                             b"0123456701234567",
 879                             50,
 880                             self.pg0.local_ip4,
 881                             self.pg0.remote_ip4).add_vpp_config()
 882         protect = VppIpsecTunProtect(self, tun,
 883                                      sa_out,
 884                                      [sa_in]).add_vpp_config()
 885
 886         #
 887         # send packets for each SPI we expect to be punted
 888         #
 889         for cfg in cfgs.values():
 890             self.send_and_assert_no_replies(cfg['itf'], cfg['pkts'])
 891
 892         #
 893         # verify the punted packets arrived on the associated socket
 894         #
 895         for cfg in cfgs.values():
 896             rx = cfg['sock'].close()
 897             self.verify_esp_pkts(rx, len(cfg['pkts']),
 898                                  cfg['spi'], cfg['udp'])
 899         #
 900         # socket deregister
 901         #
 902         for cfg in cfgs.values():
 903             self.vapi.punt_socket_deregister(cfg['vpp'])
 904
 905
 906 class TestIpProtoPuntSocket(TestPuntSocket):
 907     """ Punt Socket for IP packets """
 908
 909     @classmethod
 910     def setUpClass(cls):
 911         super(TestIpProtoPuntSocket, cls).setUpClass()
 912
 913     @classmethod
 914     def tearDownClass(cls):
 915         super(TestIpProtoPuntSocket, cls).tearDownClass()
 916
 917     def setUp(self):
 918         super(TestIpProtoPuntSocket, self).setUp()
 919
 920         for i in self.pg_interfaces:
 921             i.config_ip4()
 922             i.resolve_arp()
 923
 924     def tearDown(self):
 925         super(TestIpProtoPuntSocket, self).tearDown()
 926         for i in self.pg_interfaces:
 927             i.unconfig_ip4()
 928             i.admin_down()
 929
 930     def test_registration(self):
 931         """ Punt socket registration/deregistration"""
 932
 933         af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4
 934         pt_ip = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_IP_PROTO
 935         proto_ospf = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_OSPF
 936         proto_eigrp = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_EIGRP
 937
 938         punts = self.vapi.punt_socket_dump(type=pt_ip)
 939         self.assertEqual(len(punts), 0)
 940
 941         #
 942         # configure a punt socket
 943         #
 944         punt_ospf = {
 945             'type': pt_ip,
 946             'punt': {
 947                 'ip_proto': {
 948                     'af': af_ip4,
 949                     'protocol': proto_ospf
 950                 }
 951             }
 952         }
 953         punt_eigrp = {
 954             'type': pt_ip,
 955             'punt': {
 956                 'ip_proto': {
 957                     'af': af_ip4,
 958                     'protocol': proto_eigrp
 959                 }
 960             }
 961         }
 962
 963         self.vapi.punt_socket_register(punt_ospf,
 964                                        "%s/socket_punt_1" % self.tempdir)
 965         self.vapi.punt_socket_register(punt_eigrp,
 966                                        "%s/socket_punt_2" % self.tempdir)
 967         self.logger.info(self.vapi.cli("sh punt sock reg ip"))
 968         punts = self.vapi.punt_socket_dump(type=pt_ip)
 969         self.assertEqual(len(punts), 2)
 970         self.verify_ip_proto(punt_ospf, punts[0])
 971         self.verify_ip_proto(punt_eigrp, punts[1])
 972
 973         #
 974         # deregister a punt socket
 975         #
 976         self.vapi.punt_socket_deregister(punt_ospf)
 977         punts = self.vapi.punt_socket_dump(type=pt_ip)
 978         self.assertEqual(len(punts), 1)
 979
 980         #
 981         # configure a punt socket again
 982         #
 983         self.vapi.punt_socket_register(punt_ospf,
 984                                        "%s/socket_punt_3" % self.tempdir)
 985         punts = self.vapi.punt_socket_dump(type=pt_ip)
 986         self.assertEqual(len(punts), 2)
 987
 988         self.logger.info(self.vapi.cli("sh punt sock reg exception"))
 989
 990         #
 991         # deregister all punt socket
 992         #
 993         self.vapi.punt_socket_deregister(punt_eigrp)
 994         self.vapi.punt_socket_deregister(punt_ospf)
 995         punts = self.vapi.punt_socket_dump(type=pt_ip)
 996         self.assertEqual(len(punts), 0)
 997
 998     def verify_ospf_pkts(self, rxs, n_sent):
 999         self.assertEqual(len(rxs), n_sent)
1000         for rx in rxs:
1001             self.assertTrue(rx.haslayer(OSPF_Hdr))
1002
1003     def test_traffic(self):
1004         """ Punt socket traffic """
1005
1006         af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4
1007         pt_ip = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_IP_PROTO
1008         proto_ospf = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_OSPF
1009
1010         #
1011         # configure a punt socket to capture OSPF packets
1012         #
1013         punt_ospf = {
1014             'type': pt_ip,
1015             'punt': {
1016                 'ip_proto': {
1017                     'af': af_ip4,
1018                     'protocol': proto_ospf
1019                 }
1020             }
1021         }
1022
1023         #
1024         # create packet streams and configure a punt sockets
1025         #
1026         pkt = (Ether(src=self.pg0.remote_mac,
1027                      dst=self.pg0.local_mac) /
1028                IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
1029                OSPF_Hdr() /
1030                OSPFv3_Hello())
1031         pkts = pkt * 7
1032
1033         sock = self.socket_client_create("%s/socket_1" % self.tempdir)
1034         self.vapi.punt_socket_register(
1035             punt_ospf, "%s/socket_1" % self.tempdir)
1036
1037         #
1038         # send packets for each SPI we expect to be punted
1039         #
1040         self.send_and_assert_no_replies(self.pg0, pkts)
1041
1042         #
1043         # verify the punted packets arrived on the associated socket
1044         #
1045         rx = sock.close()
1046         self.verify_ospf_pkts(rx, len(pkts))
1047         self.vapi.punt_socket_deregister(punt_ospf)
1048
1049
1050 @tag_fixme_vpp_workers
1051 class TestPunt(VppTestCase):
1052     """ Exception Punt Test Case """
1053
1054     @classmethod
1055     def setUpClass(cls):
1056         super(TestPunt, cls).setUpClass()
1057
1058     @classmethod
1059     def tearDownClass(cls):
1060         super(TestPunt, cls).tearDownClass()
1061
1062     def setUp(self):
1063         super(TestPunt, self).setUp()
1064
1065         self.create_pg_interfaces(range(4))
1066
1067         for i in self.pg_interfaces:
1068             i.admin_up()
1069             i.config_ip4()
1070             i.resolve_arp()
1071             i.config_ip6()
1072             i.resolve_ndp()
1073
1074     def tearDown(self):
1075         for i in self.pg_interfaces:
1076             i.unconfig_ip4()
1077             i.unconfig_ip6()
1078             i.admin_down()
1079         super(TestPunt, self).tearDown()
1080
1081     def test_punt(self):
1082         """ Exception Path testing """
1083
1084         #
1085         # dump the punt registered reasons
1086         #  search for a few we know should be there
1087         #
1088         rs = self.vapi.punt_reason_dump()
1089
1090         reasons = ["ipsec6-no-such-tunnel",
1091                    "ipsec4-no-such-tunnel",
1092                    "ipsec4-spi-o-udp-0"]
1093
1094         for reason in reasons:
1095             found = False
1096             for r in rs:
1097                 if r.reason.name == reason:
1098                     found = True
1099                     break
1100             self.assertTrue(found)
1101
1102         #
1103         # Using the test CLI we will hook in a exception path to
1104         # send ACL deny packets out of pg0 and pg1.
1105         # the ACL is src,dst = 1.1.1.1,1.1.1.2
1106         #
1107         ip_1_1_1_2 = VppIpRoute(self, "1.1.1.2", 32,
1108                                 [VppRoutePath(self.pg3.remote_ip4,
1109                                               self.pg3.sw_if_index)])
1110         ip_1_1_1_2.add_vpp_config()
1111         ip_1_2 = VppIpRoute(self, "1::2", 128,
1112                             [VppRoutePath(self.pg3.remote_ip6,
1113                                           self.pg3.sw_if_index,
1114                                           proto=DpoProto.DPO_PROTO_IP6)])
1115         ip_1_2.add_vpp_config()
1116
1117         p4 = (Ether(src=self.pg2.remote_mac,
1118                     dst=self.pg2.local_mac) /
1119               IP(src="1.1.1.1", dst="1.1.1.2") /
1120               UDP(sport=1234, dport=1234) /
1121               Raw(b'\xa5' * 100))
1122         p6 = (Ether(src=self.pg2.remote_mac,
1123                     dst=self.pg2.local_mac) /
1124               IPv6(src="1::1", dst="1::2") /
1125               UDP(sport=1234, dport=1234) /
1126               Raw(b'\xa5' * 100))
1127         self.send_and_expect(self.pg2, p4*1, self.pg3)
1128         self.send_and_expect(self.pg2, p6*1, self.pg3)
1129
1130         #
1131         # apply the punting features
1132         #
1133         self.vapi.cli("test punt pg2")
1134
1135         #
1136         # dump the punt reasons to learn the IDs assigned
1137         #
1138         rs = self.vapi.punt_reason_dump(reason={'name': "reason-v4"})
1139         r4 = rs[0].reason.id
1140         rs = self.vapi.punt_reason_dump(reason={'name': "reason-v6"})
1141         r6 = rs[0].reason.id
1142
1143         #
1144         # pkts now dropped
1145         #
1146         self.send_and_assert_no_replies(self.pg2, p4*NUM_PKTS)
1147         self.send_and_assert_no_replies(self.pg2, p6*NUM_PKTS)
1148
1149         #
1150         # Check state:
1151         #  1 - node error counters
1152         #  2 - per-reason counters
1153         #    2, 3 are the index of the assigned punt reason
1154         #
1155         stats = self.statistics.get_err_counter(
1156             "/err/punt-dispatch/No registrations")
1157         self.assertEqual(stats, 2*NUM_PKTS)
1158
1159         stats = self.statistics.get_counter("/net/punt")
1160         self.assertEqual(stats[0][r4]['packets'], NUM_PKTS)
1161         self.assertEqual(stats[0][r6]['packets'], NUM_PKTS)
1162
1163         #
1164         # use the test CLI to test a client that punts exception
1165         # packets out of pg0
1166         #
1167         self.vapi.cli("test punt pg0 %s" % self.pg0.remote_ip4)
1168         self.vapi.cli("test punt pg0 %s" % self.pg0.remote_ip6)
1169
1170         rx4s = self.send_and_expect(self.pg2, p4*NUM_PKTS, self.pg0)
1171         rx6s = self.send_and_expect(self.pg2, p6*NUM_PKTS, self.pg0)
1172
1173         #
1174         # check the packets come out IP unmodified but destined to pg0 host
1175         #
1176         for rx in rx4s:
1177             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
1178             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
1179             self.assertEqual(p4[IP].dst, rx[IP].dst)
1180             self.assertEqual(p4[IP].ttl, rx[IP].ttl)
1181         for rx in rx6s:
1182             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
1183             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
1184             self.assertEqual(p6[IPv6].dst, rx[IPv6].dst)
1185             self.assertEqual(p6[IPv6].hlim, rx[IPv6].hlim)
1186
1187         stats = self.statistics.get_counter("/net/punt")
1188         self.assertEqual(stats[0][r4]['packets'], 2*NUM_PKTS)
1189         self.assertEqual(stats[0][r6]['packets'], 2*NUM_PKTS)
1190
1191         #
1192         # add another registration for the same reason to send packets
1193         # out of pg1
1194         #
1195         self.vapi.cli("test punt pg1 %s" % self.pg1.remote_ip4)
1196         self.vapi.cli("test punt pg1 %s" % self.pg1.remote_ip6)
1197
1198         self.vapi.cli("clear trace")
1199         self.pg2.add_stream(p4 * NUM_PKTS)
1200         self.pg_enable_capture(self.pg_interfaces)
1201         self.pg_start()
1202
1203         rxd = self.pg0.get_capture(NUM_PKTS)
1204         for rx in rxd:
1205             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
1206             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
1207             self.assertEqual(p4[IP].dst, rx[IP].dst)
1208             self.assertEqual(p4[IP].ttl, rx[IP].ttl)
1209         rxd = self.pg1.get_capture(NUM_PKTS)
1210         for rx in rxd:
1211             self.assertEqual(rx[Ether].dst, self.pg1.remote_mac)
1212             self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1213             self.assertEqual(p4[IP].dst, rx[IP].dst)
1214             self.assertEqual(p4[IP].ttl, rx[IP].ttl)
1215
1216         self.vapi.cli("clear trace")
1217         self.pg2.add_stream(p6 * NUM_PKTS)
1218         self.pg_enable_capture(self.pg_interfaces)
1219         self.pg_start()
1220
1221         rxd = self.pg0.get_capture(NUM_PKTS)
1222         for rx in rxd:
1223             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
1224             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
1225             self.assertEqual(p6[IPv6].dst, rx[IPv6].dst)
1226             self.assertEqual(p6[IPv6].hlim, rx[IPv6].hlim)
1227         rxd = self.pg1.get_capture(NUM_PKTS)
1228         for rx in rxd:
1229             self.assertEqual(rx[Ether].dst, self.pg1.remote_mac)
1230             self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1231             self.assertEqual(p6[IPv6].dst, rx[IPv6].dst)
1232             self.assertEqual(p6[IPv6].hlim, rx[IPv6].hlim)
1233
1234         stats = self.statistics.get_counter("/net/punt")
1235         self.assertEqual(stats[0][r4]['packets'], 3*NUM_PKTS)
1236         self.assertEqual(stats[0][r6]['packets'], 3*NUM_PKTS)
1237
1238         self.logger.info(self.vapi.cli("show vlib graph punt-dispatch"))
1239         self.logger.info(self.vapi.cli("show punt client"))
1240         self.logger.info(self.vapi.cli("show punt reason"))
1241         self.logger.info(self.vapi.cli("show punt stats"))
1242         self.logger.info(self.vapi.cli("show punt db"))
1243
1244
1245 if __name__ == '__main__':
1246     unittest.main(testRunner=VppTestRunner)