test/test_ipsec_spd_fp_input.py

   1 import socket
   2 import unittest
   3 import ipaddress
   4
   5 from util import ppp
   6 from framework import VppTestRunner
   7 from template_ipsec import IPSecIPv4Fwd
   8 from template_ipsec import IPSecIPv6Fwd
   9 from test_ipsec_esp import TemplateIpsecEsp
  10 import pdb
  11
  12
  13 def debug_signal_handler(signal, frame):
  14     import pdb
  15
  16     pdb.set_trace()
  17
  18
  19 import signal
  20
  21 signal.signal(signal.SIGINT, debug_signal_handler)
  22
  23
  24 class SpdFastPathInbound(IPSecIPv4Fwd):
  25     # In test cases derived from this class, packets in IPv4 FWD path
  26     # are configured to go through IPSec inbound SPD policy lookup.
  27     # Note that order in which the rules are applied is
  28     # PROTECT, BYPASS, DISCARD. Therefore BYPASS rules take
  29     # precedence over DISCARD.
  30     #
  31     # Override setUpConstants to enable inbound fast path in config
  32     @classmethod
  33     def setUpConstants(cls):
  34         super(SpdFastPathInbound, cls).setUpConstants()
  35         cls.vpp_cmdline.extend(["ipsec", "{", "ipv4-inbound-spd-fast-path on", "}"])
  36         cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
  37
  38     @classmethod
  39     def create_enc_stream(self, src_if, dst_if, pkt_count, src_prt=1234, dst_prt=5678):
  40         packets = []
  41         params = self.params[socket.AF_INET]
  42         for i in range(pkt_count):
  43             # create packet info stored in the test case instance
  44             info = self.create_packet_info(src_if, dst_if)
  45             # convert the info into packet payload
  46             payload = self.info_to_payload(info)
  47             # create the packet itself
  48             p = Ether(
  49                 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
  50             ) / params.scapy_tra_sa.encrypt(
  51                 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
  52                 / UDP(sport=src_prt, dport=dst_prt)
  53                 / Raw(payload)
  54             )
  55             # store a copy of the packet in the packet info
  56             info.data = p.copy()
  57             # append the packet to the list
  58             packets.append(p)
  59         # return the created packet list
  60         return packets
  61
  62
  63 class SpdFastPathInboundProtect(TemplateIpsecEsp):
  64     @classmethod
  65     def setUpConstants(cls):
  66         super(SpdFastPathInboundProtect, cls).setUpConstants()
  67         cls.vpp_cmdline.extend(["ipsec", "{", "ipv4-inbound-spd-fast-path on", "}"])
  68         cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
  69
  70     @classmethod
  71     def setUpClass(cls):
  72         super(SpdFastPathInboundProtect, cls).setUpClass()
  73
  74     @classmethod
  75     def tearDownClass(cls):
  76         super(SpdFastPathInboundProtect, cls).tearDownClass()
  77
  78     def setUp(self):
  79         super(SpdFastPathInboundProtect, self).setUp()
  80
  81     def tearDown(self):
  82         self.unconfig_network()
  83         super(SpdFastPathInboundProtect, self).tearDown()
  84
  85
  86 class SpdFastPathIPv6Inbound(IPSecIPv6Fwd):
  87     # In test cases derived from this class, packets in IPvr6 FWD path
  88     # are configured to go through IPSec inbound SPD policy lookup.
  89     # Note that order in which the rules are applied is
  90     # PROTECT, BYPASS, DISCARD. Therefore BYPASS rules take
  91     # precedence over DISCARDi.
  92
  93     # Override setUpConstants to enable inbound fast path in config
  94     @classmethod
  95     def setUpConstants(cls):
  96         super(SpdFastPathIPv6Inbound, cls).setUpConstants()
  97         cls.vpp_cmdline.extend(["ipsec", "{", "ipv6-inbound-spd-fast-path on", "}"])
  98         cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline))
  99
 100
 101 class IPSec4SpdTestCaseBypass(SpdFastPathInbound):
 102     """ IPSec/IPv4 inbound: Policy mode test case with fast path \
 103         (add bypass)"""
 104
 105     def test_ipsec_spd_inbound_bypass(self):
 106         # In this test case, packets in IPv4 FWD path are configured
 107         # to go through IPSec inbound SPD policy lookup.
 108         #
 109         # 2 inbound SPD rules (1 HIGH and 1 LOW) are added.
 110         # - High priority rule action is set to DISCARD.
 111         # - Low priority rule action is set to BYPASS.
 112         #
 113         # Since BYPASS rules take precedence over DISCARD
 114         # (the order being PROTECT, BYPASS, DISCARD) we expect the
 115         # BYPASS rule to match and traffic to be correctly forwarded.
 116         self.create_interfaces(2)
 117         pkt_count = 5
 118
 119         self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
 120
 121         # create input rules
 122         # bypass rule should take precedence over discard rule,
 123         # even though it's lower priority
 124         policy_0 = self.spd_add_rem_policy(  # inbound, priority 10
 125             1,
 126             self.pg1,
 127             self.pg0,
 128             socket.IPPROTO_UDP,
 129             is_out=0,
 130             priority=10,
 131             policy_type="bypass",
 132             ip_range=True,
 133             local_ip_start=self.pg0.remote_ip4,
 134             local_ip_stop=self.pg0.remote_ip4,
 135             remote_ip_start=self.pg1.remote_ip4,
 136             remote_ip_stop=self.pg1.remote_ip4,
 137         )
 138         policy_1 = self.spd_add_rem_policy(  # inbound, priority 15
 139             1,
 140             self.pg1,
 141             self.pg0,
 142             socket.IPPROTO_UDP,
 143             is_out=0,
 144             priority=15,
 145             policy_type="discard",
 146             ip_range=True,
 147             local_ip_start=self.pg0.remote_ip4,
 148             local_ip_stop=self.pg0.remote_ip4,
 149             remote_ip_start=self.pg1.remote_ip4,
 150             remote_ip_stop=self.pg1.remote_ip4,
 151         )
 152
 153         # create output rule so we can capture forwarded packets
 154         policy_2 = self.spd_add_rem_policy(  # outbound, priority 10
 155             1,
 156             self.pg0,
 157             self.pg1,
 158             socket.IPPROTO_UDP,
 159             is_out=1,
 160             priority=10,
 161             policy_type="bypass",
 162         )
 163
 164         # create the packet stream
 165         packets = self.create_stream(self.pg0, self.pg1, pkt_count)
 166         # add the stream to the source interface
 167         self.pg0.add_stream(packets)
 168         self.pg1.enable_capture()
 169         self.pg_start()
 170
 171         # check capture on pg1
 172         capture = self.pg1.get_capture()
 173         for packet in capture:
 174             try:
 175                 self.logger.debug(ppp("SPD Add - Got packet:", packet))
 176             except Exception:
 177                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 178                 raise
 179         self.logger.debug("SPD: Num packets: %s", len(capture.res))
 180
 181         # verify captured packets
 182         self.verify_capture(self.pg0, self.pg1, capture)
 183         # verify all policies matched the expected number of times
 184         self.verify_policy_match(pkt_count, policy_0)
 185         self.verify_policy_match(0, policy_1)
 186         self.verify_policy_match(pkt_count, policy_2)
 187
 188
 189 class IPSec4SpdTestCaseDiscard(SpdFastPathInbound):
 190     """ IPSec/IPv4 inbound: Policy mode test case with fast path \
 191             (add discard)"""
 192
 193     def test_ipsec_spd_inbound_discard(self):
 194         # In this test case, packets in IPv4 FWD path are configured
 195         # to go through IPSec inbound SPD policy lookup.
 196         #
 197         #  Rule action is set to DISCARD.
 198
 199         self.create_interfaces(2)
 200         pkt_count = 5
 201
 202         self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
 203
 204         # create input rules
 205         # bypass rule should take precedence over discard rule,
 206         # even though it's lower priority
 207         policy_0 = self.spd_add_rem_policy(  # inbound, priority 10
 208             1,
 209             self.pg1,
 210             self.pg0,
 211             socket.IPPROTO_UDP,
 212             is_out=0,
 213             priority=10,
 214             policy_type="discard",
 215             ip_range=True,
 216             local_ip_start=self.pg0.remote_ip4,
 217             local_ip_stop=self.pg0.remote_ip4,
 218             remote_ip_start=self.pg1.remote_ip4,
 219             remote_ip_stop=self.pg1.remote_ip4,
 220         )
 221
 222         # create output rule so we can capture forwarded packets
 223         policy_1 = self.spd_add_rem_policy(  # outbound, priority 10
 224             1,
 225             self.pg0,
 226             self.pg1,
 227             socket.IPPROTO_UDP,
 228             is_out=1,
 229             priority=10,
 230             policy_type="bypass",
 231         )
 232
 233         # create the packet stream
 234         packets = self.create_stream(self.pg0, self.pg1, pkt_count)
 235         # add the stream to the source interface
 236         self.pg0.add_stream(packets)
 237         self.pg1.enable_capture()
 238         self.pg_start()
 239
 240         # check capture on pg1
 241         capture = self.pg1.assert_nothing_captured()
 242
 243         # verify all policies matched the expected number of times
 244         self.verify_policy_match(pkt_count, policy_0)
 245         self.verify_policy_match(0, policy_1)
 246
 247
 248 class IPSec4SpdTestCaseProtect(SpdFastPathInboundProtect):
 249     """ IPSec/IPv4 inbound: Policy mode test case with fast path \
 250     (add protect)"""
 251
 252     @classmethod
 253     def setUpClass(cls):
 254         super(IPSec4SpdTestCaseProtect, cls).setUpClass()
 255
 256     @classmethod
 257     def tearDownClass(cls):
 258         super(IPSec4SpdTestCaseProtect, cls).tearDownClass()
 259
 260     def setUp(self):
 261         super(IPSec4SpdTestCaseProtect, self).setUp()
 262
 263     def tearDown(self):
 264         super(IPSec4SpdTestCaseProtect, self).tearDown()
 265
 266     def test_ipsec_spd_inbound_protect(self):
 267         # In this test case, packets in IPv4 FWD path are configured
 268         # to go through IPSec inbound SPD policy lookup.
 269         #
 270         # 2 inbound SPD rules (1 HIGH and 1 LOW) are added.
 271         # - High priority rule action is set to DISCARD.
 272         # - Low priority rule action is set to BYPASS.
 273         #
 274         # Since BYPASS rules take precedence over DISCARD
 275         # (the order being PROTECT, BYPASS, DISCARD) we expect the
 276         # BYPASS rule to match and traffic to be correctly forwarded.
 277
 278         pkt_count = 5
 279         payload_size = 64
 280         p = self.params[socket.AF_INET]
 281         send_pkts = self.gen_encrypt_pkts(
 282             p,
 283             p.scapy_tra_sa,
 284             self.tra_if,
 285             src=self.tra_if.local_ip4,
 286             dst=self.tra_if.remote_ip4,
 287             count=pkt_count,
 288             payload_size=payload_size,
 289         )
 290         recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
 291
 292         self.logger.info(self.vapi.ppcli("show error"))
 293         self.logger.info(self.vapi.ppcli("show ipsec all"))
 294
 295         pkts = p.tra_sa_in.get_stats()["packets"]
 296         self.assertEqual(
 297             pkts,
 298             pkt_count,
 299             "incorrect SA in counts: expected %d != %d" % (pkt_count, pkts),
 300         )
 301         pkts = p.tra_sa_out.get_stats()["packets"]
 302         self.assertEqual(
 303             pkts,
 304             pkt_count,
 305             "incorrect SA out counts: expected %d != %d" % (pkt_count, pkts),
 306         )
 307         self.assertEqual(p.tra_sa_out.get_lost(), 0)
 308         self.assertEqual(p.tra_sa_in.get_lost(), 0)
 309
 310
 311 class IPSec4SpdTestCaseAddIPRange(SpdFastPathInbound):
 312     """ IPSec/IPv4 inbound: Policy mode test case with fast path \
 313         (add  ips  range with any port rule)"""
 314
 315     def test_ipsec_spd_inbound_add(self):
 316         # In this test case, packets in IPv4 FWD path are configured
 317         # to go through IPSec inbound SPD policy lookup.
 318         # 2 SPD bypass rules (1 for inbound and 1 for outbound) are added.
 319         # Traffic sent on pg0 interface should match fast path priority
 320         # rule and should be sent out on pg1 interface.
 321         self.create_interfaces(2)
 322         pkt_count = 5
 323         s_ip_s1 = ipaddress.ip_address(self.pg0.remote_ip4)
 324         s_ip_e1 = ipaddress.ip_address(int(s_ip_s1) + 5)
 325         d_ip_s1 = ipaddress.ip_address(self.pg1.remote_ip4)
 326         d_ip_e1 = ipaddress.ip_address(int(d_ip_s1) + 0)
 327
 328         s_ip_s0 = ipaddress.ip_address(self.pg0.remote_ip4)
 329         s_ip_e0 = ipaddress.ip_address(int(s_ip_s0) + 6)
 330         d_ip_s0 = ipaddress.ip_address(self.pg1.remote_ip4)
 331         d_ip_e0 = ipaddress.ip_address(int(d_ip_s0) + 0)
 332         self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
 333
 334         policy_0 = self.spd_add_rem_policy(  # inbound fast path, priority 10
 335             1,
 336             self.pg0,
 337             self.pg1,
 338             socket.IPPROTO_UDP,
 339             is_out=0,
 340             priority=10,
 341             policy_type="bypass",
 342             ip_range=True,
 343             local_ip_start=s_ip_s0,
 344             local_ip_stop=s_ip_e0,
 345             remote_ip_start=d_ip_s0,
 346             remote_ip_stop=d_ip_e0,
 347         )
 348         policy_1 = self.spd_add_rem_policy(  # outbound, priority 5
 349             1,
 350             self.pg0,
 351             self.pg1,
 352             socket.IPPROTO_UDP,
 353             is_out=1,
 354             priority=5,
 355             policy_type="bypass",
 356             ip_range=True,
 357             local_ip_start=s_ip_s1,
 358             local_ip_stop=s_ip_e1,
 359             remote_ip_start=d_ip_s1,
 360             remote_ip_stop=d_ip_e1,
 361         )
 362
 363         # create the packet stream
 364         packets = self.create_stream(self.pg0, self.pg1, pkt_count)
 365         # add the stream to the source interface + enable capture
 366         self.pg0.add_stream(packets)
 367         self.pg0.enable_capture()
 368         self.pg1.enable_capture()
 369         # start the packet generator
 370         self.pg_start()
 371         # get capture
 372         capture = self.pg1.get_capture()
 373         for packet in capture:
 374             try:
 375                 self.logger.debug(ppp("SPD - Got packet:", packet))
 376             except Exception:
 377                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 378                 raise
 379         self.logger.debug("SPD: Num packets: %s", len(capture.res))
 380
 381         # assert nothing captured on pg0
 382         self.pg0.assert_nothing_captured()
 383         # verify captured packets
 384         self.verify_capture(self.pg0, self.pg1, capture)
 385         # verify all policies matched the expected number of times
 386         self.verify_policy_match(pkt_count, policy_0)
 387         self.verify_policy_match(pkt_count, policy_1)
 388
 389
 390 class IPSec4SpdTestCaseAddAll(SpdFastPathInbound):
 391     """ IPSec/IPv4 inbound: Policy mode test case with fast path \
 392         (add all ips ports rule)"""
 393
 394     def test_ipsec_spd_inbound_add(self):
 395         # In this test case, packets in IPv4 FWD path are configured
 396         # to go through IPSec inbound SPD policy lookup.
 397         # 2 SPD rules (1 HIGH and 1 LOW) are added.
 398         # Low priority rule action is set to BYPASS all ips.
 399         # High priority rule action is set to DISCARD all ips.
 400         # Traffic not sent on pg0 interface when HIGH discard priority rule is added.
 401         # Then LOW priority
 402         # rule is added and send the same traffic to pg0, this time expect
 403         # the traffic is bypassed as bypass takes priority over discard.
 404         self.create_interfaces(2)
 405         pkt_count = 5
 406         self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
 407
 408         policy_0 = self.spd_add_rem_policy(  # inbound, priority 20
 409             1,
 410             self.pg0,
 411             self.pg1,
 412             socket.IPPROTO_UDP,
 413             is_out=0,
 414             priority=20,
 415             policy_type="discard",
 416             all_ips=True,
 417         )
 418
 419         policy_1 = self.spd_add_rem_policy(  # inbound, priority 20
 420             1,
 421             self.pg0,
 422             self.pg1,
 423             socket.IPPROTO_UDP,
 424             is_out=True,
 425             priority=5,
 426             policy_type="bypass",
 427             all_ips=True,
 428         )
 429
 430         # create the packet stream
 431         packets = self.create_stream(self.pg0, self.pg1, pkt_count)
 432         # add the stream to the source interface + enable capture
 433         self.pg0.add_stream(packets)
 434         self.pg0.enable_capture()
 435         self.pg1.enable_capture()
 436         # start the packet generator
 437         self.pg_start()
 438         # assert nothing captured on pg0 and pg1
 439         self.pg0.assert_nothing_captured()
 440         self.pg1.assert_nothing_captured()
 441
 442         policy_2 = self.spd_add_rem_policy(  # inbound, priority 10
 443             1,
 444             self.pg0,
 445             self.pg1,
 446             socket.IPPROTO_UDP,
 447             is_out=0,
 448             priority=10,
 449             policy_type="bypass",
 450             all_ips=True,
 451         )
 452
 453         # create the packet stream
 454         packets = self.create_stream(self.pg0, self.pg1, pkt_count)
 455         # add the stream to the source interface + enable capture
 456         self.pg0.add_stream(packets)
 457         self.pg0.enable_capture()
 458         self.pg1.enable_capture()
 459         # start the packet generator
 460         self.pg_start()
 461         # get capture
 462         capture = self.pg1.get_capture(expected_count=pkt_count)
 463         for packet in capture:
 464             try:
 465                 self.logger.debug(ppp("SPD - Got packet:", packet))
 466             except Exception:
 467                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 468                 raise
 469         self.logger.debug("SPD: Num packets: %s", len(capture.res))
 470
 471         # assert nothing captured on pg0
 472         self.pg0.assert_nothing_captured()
 473         # verify all policies matched the expected number of times
 474         self.verify_policy_match(pkt_count, policy_2)
 475
 476
 477 class IPSec4SpdTestCaseRemove(SpdFastPathInbound):
 478     """ IPSec/IPv4 inbound: Policy mode test case with fast path \
 479         (remove rule)"""
 480
 481     def test_ipsec_spd_inbound_remove(self):
 482         # In this test case, packets in IPv4 FWD path are configured
 483         # to go through IPSec inbound SPD policy lookup.
 484         # 2 SPD rules (1 HIGH and 1 LOW) are added.
 485         # High priority rule action is set to BYPASS.
 486         # Low priority rule action is set to DISCARD.
 487         # High priority rule is then removed.
 488         # Traffic sent on pg0 interface should match low priority
 489         # rule and should be discarded after SPD lookup.
 490         self.create_interfaces(2)
 491         pkt_count = 5
 492         self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
 493         policy_0 = self.spd_add_rem_policy(  # inbound, priority 10
 494             1,
 495             self.pg0,
 496             self.pg1,
 497             socket.IPPROTO_UDP,
 498             is_out=0,
 499             priority=10,
 500             policy_type="bypass",
 501         )
 502         policy_1 = self.spd_add_rem_policy(  # inbound, priority 5
 503             1,
 504             self.pg0,
 505             self.pg1,
 506             socket.IPPROTO_UDP,
 507             is_out=0,
 508             priority=5,
 509             policy_type="discard",
 510         )
 511
 512         policy_out = self.spd_add_rem_policy(  # outbound, priority 10
 513             1,
 514             self.pg0,
 515             self.pg1,
 516             socket.IPPROTO_UDP,
 517             is_out=1,
 518             priority=10,
 519             policy_type="bypass",
 520         )
 521
 522         # create the packet stream
 523         packets = self.create_stream(self.pg0, self.pg1, pkt_count)
 524         # add the stream to the source interface + enable capture
 525         self.pg0.add_stream(packets)
 526         self.pg0.enable_capture()
 527         self.pg1.enable_capture()
 528         # start the packet generator
 529         self.pg_start()
 530         # get capture
 531         capture = self.pg1.get_capture()
 532         for packet in capture:
 533             try:
 534                 self.logger.debug(ppp("SPD - Got packet:", packet))
 535             except Exception:
 536                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 537                 raise
 538
 539         # assert nothing captured on pg0
 540         self.pg0.assert_nothing_captured()
 541         # verify capture on pg1
 542         self.logger.debug("SPD: Num packets: %s", len(capture.res))
 543         self.verify_capture(self.pg0, self.pg1, capture)
 544         # verify all policies matched the expected number of times
 545         self.verify_policy_match(pkt_count, policy_0)
 546         self.verify_policy_match(0, policy_1)
 547         # now remove the bypass rule
 548         self.spd_add_rem_policy(  # outbound, priority 10
 549             1,
 550             self.pg0,
 551             self.pg1,
 552             socket.IPPROTO_UDP,
 553             is_out=0,
 554             priority=10,
 555             policy_type="bypass",
 556             remove=True,
 557         )
 558
 559         # resend the same packets
 560         self.pg0.add_stream(packets)
 561         self.pg0.enable_capture()  # flush the old captures
 562         self.pg1.enable_capture()
 563         self.pg_start()
 564         # assert nothing captured on pg0
 565         self.pg0.assert_nothing_captured()
 566         # all packets will be dropped by SPD rule
 567         self.pg1.assert_nothing_captured()
 568         # verify all policies matched the expected number of times
 569         self.verify_policy_match(pkt_count, policy_0)
 570         self.verify_policy_match(pkt_count, policy_1)
 571
 572
 573 class IPSec4SpdTestCaseReadd(SpdFastPathInbound):
 574     """ IPSec/IPv4 inbound: Policy mode test case with fast path \
 575         (add, remove, re-add)"""
 576
 577     def test_ipsec_spd_inbound_readd(self):
 578         # In this test case, packets in IPv4 FWD path are configured
 579         # to go through IPSec outbound SPD policy lookup.
 580         # 2 SPD rules (1 HIGH and 1 LOW) are added.
 581         # High priority rule action is set to BYPASS.
 582         # Low priority rule action is set to DISCARD.
 583         # Traffic sent on pg0 interface should match high priority
 584         # rule and should be sent out on pg1 interface.
 585         # High priority rule is then removed.
 586         # Traffic sent on pg0 interface should match low priority
 587         # rule and should be discarded after SPD lookup.
 588         # Readd high priority rule.
 589         # Traffic sent on pg0 interface should match high priority
 590         # rule and should be sent out on pg1 interface.
 591         self.create_interfaces(2)
 592         pkt_count = 5
 593         self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
 594         policy_0 = self.spd_add_rem_policy(  # inbound, priority 10
 595             1,
 596             self.pg0,
 597             self.pg1,
 598             socket.IPPROTO_UDP,
 599             is_out=0,
 600             priority=10,
 601             policy_type="bypass",
 602         )
 603         policy_1 = self.spd_add_rem_policy(  # inbound, priority 5
 604             1,
 605             self.pg0,
 606             self.pg1,
 607             socket.IPPROTO_UDP,
 608             is_out=0,
 609             priority=5,
 610             policy_type="discard",
 611         )
 612         policy_2 = self.spd_add_rem_policy(  # outbound, priority 10
 613             1,
 614             self.pg0,
 615             self.pg1,
 616             socket.IPPROTO_UDP,
 617             is_out=1,
 618             priority=10,
 619             policy_type="bypass",
 620         )
 621
 622         # create the packet stream
 623         packets = self.create_stream(self.pg0, self.pg1, pkt_count)
 624         # add the stream to the source interface + enable capture
 625         self.pg0.add_stream(packets)
 626         self.pg0.enable_capture()
 627         self.pg1.enable_capture()
 628         # start the packet generator
 629         self.pg_start()
 630         # get capture
 631         capture = self.pg1.get_capture()
 632         for packet in capture:
 633             try:
 634                 self.logger.debug(ppp("SPD - Got packet:", packet))
 635             except Exception:
 636                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 637                 raise
 638         self.logger.debug("SPD: Num packets: %s", len(capture.res))
 639
 640         # assert nothing captured on pg0
 641         self.pg0.assert_nothing_captured()
 642         # verify capture on pg1
 643         self.verify_capture(self.pg0, self.pg1, capture)
 644         # verify all policies matched the expected number of times
 645         self.verify_policy_match(pkt_count, policy_0)
 646         self.verify_policy_match(0, policy_1)
 647         # remove the bypass rule, leaving only the discard rule
 648         self.spd_add_rem_policy(  # inbound, priority 10
 649             1,
 650             self.pg0,
 651             self.pg1,
 652             socket.IPPROTO_UDP,
 653             is_out=0,
 654             priority=10,
 655             policy_type="bypass",
 656             remove=True,
 657         )
 658
 659         # resend the same packets
 660         self.pg0.add_stream(packets)
 661         self.pg0.enable_capture()  # flush the old captures
 662         self.pg1.enable_capture()
 663         self.pg_start()
 664
 665         # assert nothing captured on pg0
 666         self.pg0.assert_nothing_captured()
 667         # all packets will be dropped by SPD rule
 668         self.pg1.assert_nothing_captured()
 669         # verify all policies matched the expected number of times
 670         self.verify_policy_match(pkt_count, policy_0)
 671         self.verify_policy_match(pkt_count, policy_1)
 672
 673         # now readd the bypass rule
 674         policy_0 = self.spd_add_rem_policy(  # outbound, priority 10
 675             1,
 676             self.pg0,
 677             self.pg1,
 678             socket.IPPROTO_UDP,
 679             is_out=0,
 680             priority=10,
 681             policy_type="bypass",
 682         )
 683
 684         # resend the same packets
 685         self.pg0.add_stream(packets)
 686         self.pg0.enable_capture()  # flush the old captures
 687         self.pg1.enable_capture()
 688         self.pg_start()
 689
 690         # get capture
 691         capture = self.pg1.get_capture(pkt_count)
 692         for packet in capture:
 693             try:
 694                 self.logger.debug(ppp("SPD - Got packet:", packet))
 695             except Exception:
 696                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 697                 raise
 698         self.logger.debug("SPD: Num packets: %s", len(capture.res))
 699
 700         # assert nothing captured on pg0
 701         self.pg0.assert_nothing_captured()
 702         # verify captured packets
 703         self.verify_capture(self.pg0, self.pg1, capture)
 704         # verify all policies matched the expected number of times
 705         self.verify_policy_match(pkt_count, policy_0)
 706         self.verify_policy_match(pkt_count, policy_1)
 707
 708
 709 class IPSec4SpdTestCaseMultiple(SpdFastPathInbound):
 710     """ IPSec/IPv4 inbound: Policy mode test case with fast path \
 711         (multiple interfaces, multiple rules)"""
 712
 713     def test_ipsec_spd_inbound_multiple(self):
 714         # In this test case, packets in IPv4 FWD path are configured to go
 715         # through IPSec outbound SPD policy lookup.
 716         # Multiples rules on multiple interfaces are tested at the same time.
 717         # 3x interfaces are configured, binding the same SPD to each.
 718         # Each interface has 2 SPD rules (1 BYPASS and 1 DISCARD).
 719         # On pg0 & pg1, the BYPASS rule is HIGH priority
 720         # On pg2, the DISCARD rule is HIGH priority
 721         # Traffic should be received on pg0 & pg1 and dropped on pg2.
 722         self.create_interfaces(3)
 723         pkt_count = 5
 724         # bind SPD to all interfaces
 725         self.spd_create_and_intf_add(1, self.pg_interfaces)
 726         # add rules on all interfaces
 727         policy_01 = self.spd_add_rem_policy(  # inbound, priority 10
 728             1,
 729             self.pg0,
 730             self.pg1,
 731             socket.IPPROTO_UDP,
 732             is_out=0,
 733             priority=10,
 734             policy_type="bypass",
 735         )
 736         policy_02 = self.spd_add_rem_policy(  # inbound, priority 5
 737             1,
 738             self.pg0,
 739             self.pg1,
 740             socket.IPPROTO_UDP,
 741             is_out=0,
 742             priority=5,
 743             policy_type="discard",
 744         )
 745
 746         policy_11 = self.spd_add_rem_policy(  # inbound, priority 10
 747             1,
 748             self.pg1,
 749             self.pg2,
 750             socket.IPPROTO_UDP,
 751             is_out=0,
 752             priority=10,
 753             policy_type="bypass",
 754         )
 755         policy_12 = self.spd_add_rem_policy(  # inbound, priority 5
 756             1,
 757             self.pg1,
 758             self.pg2,
 759             socket.IPPROTO_UDP,
 760             is_out=0,
 761             priority=5,
 762             policy_type="discard",
 763         )
 764
 765         policy_21 = self.spd_add_rem_policy(  # inbound, priority 5
 766             1,
 767             self.pg2,
 768             self.pg0,
 769             socket.IPPROTO_UDP,
 770             is_out=0,
 771             priority=5,
 772             policy_type="bypass",
 773         )
 774         policy_22 = self.spd_add_rem_policy(  # inbound, priority 10
 775             1,
 776             self.pg2,
 777             self.pg0,
 778             socket.IPPROTO_UDP,
 779             is_out=0,
 780             priority=10,
 781             policy_type="discard",
 782         )
 783
 784         # interfaces bound to an SPD, will by default drop outbound
 785         # traffic with no matching policies. add catch-all outbound
 786         # bypass rule to SPD:
 787         self.spd_add_rem_policy(  # outbound, all interfaces
 788             1,
 789             None,
 790             None,
 791             socket.IPPROTO_UDP,
 792             is_out=1,
 793             priority=10,
 794             policy_type="bypass",
 795             all_ips=True,
 796         )
 797
 798         # create the packet streams
 799         packets0 = self.create_stream(self.pg0, self.pg1, pkt_count)
 800         packets1 = self.create_stream(self.pg1, self.pg2, pkt_count)
 801         packets2 = self.create_stream(self.pg2, self.pg0, pkt_count)
 802         # add the streams to the source interfaces
 803         self.pg0.add_stream(packets0)
 804         self.pg1.add_stream(packets1)
 805         self.pg2.add_stream(packets2)
 806         # enable capture on all interfaces
 807         for pg in self.pg_interfaces:
 808             pg.enable_capture()
 809         # start the packet generator
 810         self.pg_start()
 811
 812         # get captures
 813         if_caps = []
 814         for pg in [self.pg1, self.pg2]:  # we are expecting captures on pg1/pg2
 815             if_caps.append(pg.get_capture())
 816             for packet in if_caps[-1]:
 817                 try:
 818                     self.logger.debug(ppp("SPD - Got packet:", packet))
 819                 except Exception:
 820                     self.logger.error(ppp("Unexpected or invalid packet:", packet))
 821                     raise
 822         self.logger.debug("SPD: Num packets: %s", len(if_caps[0].res))
 823         self.logger.debug("SPD: Num packets: %s", len(if_caps[1].res))
 824
 825         # verify captures that matched BYPASS rule
 826         self.verify_capture(self.pg0, self.pg1, if_caps[0])
 827         self.verify_capture(self.pg1, self.pg2, if_caps[1])
 828         # verify that traffic to pg0 matched BYPASS rule
 829         # although DISCARD rule had higher prioriy and was not dropped
 830         self.verify_policy_match(pkt_count, policy_21)
 831
 832         # verify all packets that were expected to match rules, matched
 833         # pg0 -> pg1
 834         self.verify_policy_match(pkt_count, policy_01)
 835         self.verify_policy_match(0, policy_02)
 836         # pg1 -> pg2
 837         self.verify_policy_match(pkt_count, policy_11)
 838         self.verify_policy_match(0, policy_12)
 839         # pg2 -> pg0
 840         self.verify_policy_match(0, policy_22)
 841
 842
 843 if __name__ == "__main__":
 844     unittest.main(testRunner=VppTestRunner)