6 from framework import VppTestRunner
7 from template_ipsec import IPSecIPv4Fwd
8 from template_ipsec import IPSecIPv6Fwd
9 from test_ipsec_esp import TemplateIpsecEsp
13 def debug_signal_handler(signal, frame):
21 signal.signal(signal.SIGINT, debug_signal_handler)
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.
31 # Override setUpConstants to enable inbound fast path in config
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))
39 def create_enc_stream(self, src_if, dst_if, pkt_count, src_prt=1234, dst_prt=5678):
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
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)
55 # store a copy of the packet in the packet info
57 # append the packet to the list
59 # return the created packet list
63 class SpdFastPathInboundProtect(TemplateIpsecEsp):
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))
72 super(SpdFastPathInboundProtect, cls).setUpClass()
75 def tearDownClass(cls):
76 super(SpdFastPathInboundProtect, cls).tearDownClass()
79 super(SpdFastPathInboundProtect, self).setUp()
82 self.unconfig_network()
83 super(SpdFastPathInboundProtect, self).tearDown()
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.
93 # Override setUpConstants to enable inbound fast path in config
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))
101 class IPSec4SpdTestCaseBypass(SpdFastPathInbound):
102 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
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.
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.
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)
119 self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
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
131 policy_type="bypass",
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,
138 policy_1 = self.spd_add_rem_policy( # inbound, priority 15
145 policy_type="discard",
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,
153 # create output rule so we can capture forwarded packets
154 policy_2 = self.spd_add_rem_policy( # outbound, priority 10
161 policy_type="bypass",
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()
171 # check capture on pg1
172 capture = self.pg1.get_capture()
173 for packet in capture:
175 self.logger.debug(ppp("SPD Add - Got packet:", packet))
177 self.logger.error(ppp("Unexpected or invalid packet:", packet))
179 self.logger.debug("SPD: Num packets: %s", len(capture.res))
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)
189 class IPSec4SpdTestCaseDiscard(SpdFastPathInbound):
190 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
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.
197 # Rule action is set to DISCARD.
199 self.create_interfaces(2)
202 self.spd_create_and_intf_add(1, [self.pg1, self.pg0])
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
214 policy_type="discard",
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,
222 # create output rule so we can capture forwarded packets
223 policy_1 = self.spd_add_rem_policy( # outbound, priority 10
230 policy_type="bypass",
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()
240 # check capture on pg1
241 capture = self.pg1.assert_nothing_captured()
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)
248 class IPSec4SpdTestCaseProtect(SpdFastPathInboundProtect):
249 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
254 super(IPSec4SpdTestCaseProtect, cls).setUpClass()
257 def tearDownClass(cls):
258 super(IPSec4SpdTestCaseProtect, cls).tearDownClass()
261 super(IPSec4SpdTestCaseProtect, self).setUp()
264 super(IPSec4SpdTestCaseProtect, self).tearDown()
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.
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.
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.
280 p = self.params[socket.AF_INET]
281 send_pkts = self.gen_encrypt_pkts(
285 src=self.tra_if.local_ip4,
286 dst=self.tra_if.remote_ip4,
288 payload_size=payload_size,
290 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
292 self.logger.info(self.vapi.ppcli("show error"))
293 self.logger.info(self.vapi.ppcli("show ipsec all"))
295 pkts = p.tra_sa_in.get_stats()["packets"]
299 "incorrect SA in counts: expected %d != %d" % (pkt_count, pkts),
301 pkts = p.tra_sa_out.get_stats()["packets"]
305 "incorrect SA out counts: expected %d != %d" % (pkt_count, pkts),
307 self.assertEqual(p.tra_sa_out.get_lost(), 0)
308 self.assertEqual(p.tra_sa_in.get_lost(), 0)
311 class IPSec4SpdTestCaseAddIPRange(SpdFastPathInbound):
312 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
313 (add ips range with any port rule)"""
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)
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)
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])
334 policy_0 = self.spd_add_rem_policy( # inbound fast path, priority 10
341 policy_type="bypass",
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,
348 policy_1 = self.spd_add_rem_policy( # outbound, priority 5
355 policy_type="bypass",
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,
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
372 capture = self.pg1.get_capture()
373 for packet in capture:
375 self.logger.debug(ppp("SPD - Got packet:", packet))
377 self.logger.error(ppp("Unexpected or invalid packet:", packet))
379 self.logger.debug("SPD: Num packets: %s", len(capture.res))
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)
390 class IPSec4SpdTestCaseAddAll(SpdFastPathInbound):
391 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
392 (add all ips ports rule)"""
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.
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)
406 self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
408 policy_0 = self.spd_add_rem_policy( # inbound, priority 20
415 policy_type="discard",
419 policy_1 = self.spd_add_rem_policy( # inbound, priority 20
426 policy_type="bypass",
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
438 # assert nothing captured on pg0 and pg1
439 self.pg0.assert_nothing_captured()
440 self.pg1.assert_nothing_captured()
442 policy_2 = self.spd_add_rem_policy( # inbound, priority 10
449 policy_type="bypass",
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
462 capture = self.pg1.get_capture(expected_count=pkt_count)
463 for packet in capture:
465 self.logger.debug(ppp("SPD - Got packet:", packet))
467 self.logger.error(ppp("Unexpected or invalid packet:", packet))
469 self.logger.debug("SPD: Num packets: %s", len(capture.res))
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)
477 class IPSec4SpdTestCaseRemove(SpdFastPathInbound):
478 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
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)
492 self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
493 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
500 policy_type="bypass",
502 policy_1 = self.spd_add_rem_policy( # inbound, priority 5
509 policy_type="discard",
512 policy_out = self.spd_add_rem_policy( # outbound, priority 10
519 policy_type="bypass",
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
531 capture = self.pg1.get_capture()
532 for packet in capture:
534 self.logger.debug(ppp("SPD - Got packet:", packet))
536 self.logger.error(ppp("Unexpected or invalid packet:", packet))
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
555 policy_type="bypass",
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()
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)
573 class IPSec4SpdTestCaseReadd(SpdFastPathInbound):
574 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
575 (add, remove, re-add)"""
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)
593 self.spd_create_and_intf_add(1, [self.pg0, self.pg1])
594 policy_0 = self.spd_add_rem_policy( # inbound, priority 10
601 policy_type="bypass",
603 policy_1 = self.spd_add_rem_policy( # inbound, priority 5
610 policy_type="discard",
612 policy_2 = self.spd_add_rem_policy( # outbound, priority 10
619 policy_type="bypass",
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
631 capture = self.pg1.get_capture()
632 for packet in capture:
634 self.logger.debug(ppp("SPD - Got packet:", packet))
636 self.logger.error(ppp("Unexpected or invalid packet:", packet))
638 self.logger.debug("SPD: Num packets: %s", len(capture.res))
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
655 policy_type="bypass",
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()
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)
673 # now readd the bypass rule
674 policy_0 = self.spd_add_rem_policy( # outbound, priority 10
681 policy_type="bypass",
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()
691 capture = self.pg1.get_capture(pkt_count)
692 for packet in capture:
694 self.logger.debug(ppp("SPD - Got packet:", packet))
696 self.logger.error(ppp("Unexpected or invalid packet:", packet))
698 self.logger.debug("SPD: Num packets: %s", len(capture.res))
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)
709 class IPSec4SpdTestCaseMultiple(SpdFastPathInbound):
710 """ IPSec/IPv4 inbound: Policy mode test case with fast path \
711 (multiple interfaces, multiple rules)"""
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)
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
734 policy_type="bypass",
736 policy_02 = self.spd_add_rem_policy( # inbound, priority 5
743 policy_type="discard",
746 policy_11 = self.spd_add_rem_policy( # inbound, priority 10
753 policy_type="bypass",
755 policy_12 = self.spd_add_rem_policy( # inbound, priority 5
762 policy_type="discard",
765 policy_21 = self.spd_add_rem_policy( # inbound, priority 5
772 policy_type="bypass",
774 policy_22 = self.spd_add_rem_policy( # inbound, priority 10
781 policy_type="discard",
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
794 policy_type="bypass",
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:
809 # start the packet generator
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]:
818 self.logger.debug(ppp("SPD - Got packet:", packet))
820 self.logger.error(ppp("Unexpected or invalid packet:", packet))
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))
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)
832 # verify all packets that were expected to match rules, matched
834 self.verify_policy_match(pkt_count, policy_01)
835 self.verify_policy_match(0, policy_02)
837 self.verify_policy_match(pkt_count, policy_11)
838 self.verify_policy_match(0, policy_12)
840 self.verify_policy_match(0, policy_22)
843 if __name__ == "__main__":
844 unittest.main(testRunner=VppTestRunner)