diff options
Diffstat (limited to 'test/test_ipsec_spd_fp_input.py')
| -rw-r--r-- | test/test_ipsec_spd_fp_input.py | 885 |
1 files changed, 885 insertions, 0 deletions
diff --git a/test/test_ipsec_spd_fp_input.py b/test/test_ipsec_spd_fp_input.py new file mode 100644 index 00000000000..ec4a7c71d7b --- /dev/null +++ b/test/test_ipsec_spd_fp_input.py @@ -0,0 +1,885 @@ +import socket +import unittest +import ipaddress + +from util import ppp +from asfframework import VppTestRunner +from template_ipsec import IPSecIPv4Fwd +from template_ipsec import IPSecIPv6Fwd +from test_ipsec_esp import TemplateIpsecEsp +from template_ipsec import SpdFastPathTemplate + + +def debug_signal_handler(signal, frame): + import pdb + + pdb.set_trace() + + +import signal + +signal.signal(signal.SIGINT, debug_signal_handler) + + +class SpdFastPathInbound(SpdFastPathTemplate): + # In test cases derived from this class, packets in IPv4 FWD path + # are configured to go through IPSec inbound SPD policy lookup. + # Note that order in which the rules are applied is + # PROTECT, BYPASS, DISCARD. Therefore BYPASS rules take + # precedence over DISCARD. + # + # Override setUpConstants to enable inbound fast path in config + @classmethod + def setUpConstants(cls): + super(SpdFastPathInbound, cls).setUpConstants() + cls.vpp_cmdline.extend(["ipsec", "{", "ipv4-inbound-spd-fast-path on", "}"]) + cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline)) + + +class SpdFastPathInboundProtect(TemplateIpsecEsp): + @classmethod + def setUpConstants(cls): + super(SpdFastPathInboundProtect, cls).setUpConstants() + cls.vpp_cmdline.extend(["ipsec", "{", "ipv4-inbound-spd-fast-path on", "}"]) + cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline)) + + @classmethod + def setUpClass(cls): + super(SpdFastPathInboundProtect, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(SpdFastPathInboundProtect, cls).tearDownClass() + + def setUp(self): + super(SpdFastPathInboundProtect, self).setUp() + + def tearDown(self): + self.unconfig_network() + super(SpdFastPathInboundProtect, self).tearDown() + + +class SpdFastPathIPv6Inbound(IPSecIPv6Fwd): + # In test cases derived from this class, packets in IPvr6 FWD path + # are configured to go through IPSec inbound SPD policy lookup. + # Note that order in which the rules are applied is + # PROTECT, BYPASS, DISCARD. Therefore BYPASS rules take + # precedence over DISCARDi. + + # Override setUpConstants to enable inbound fast path in config + @classmethod + def setUpConstants(cls): + super(SpdFastPathIPv6Inbound, cls).setUpConstants() + cls.vpp_cmdline.extend(["ipsec", "{", "ipv6-inbound-spd-fast-path on", "}"]) + cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline)) + + +class SpdFastPathIPv6InboundProtect(TemplateIpsecEsp): + @classmethod + def setUpConstants(cls): + super(SpdFastPathIPv6InboundProtect, cls).setUpConstants() + cls.vpp_cmdline.extend(["ipsec", "{", "ipv6-inbound-spd-fast-path on", "}"]) + cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline)) + + @classmethod + def setUpClass(cls): + super(SpdFastPathIPv6InboundProtect, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(SpdFastPathIPv6InboundProtect, cls).tearDownClass() + + def setUp(self): + super(SpdFastPathIPv6InboundProtect, self).setUp() + + def tearDown(self): + self.unconfig_network() + super(SpdFastPathIPv6InboundProtect, self).tearDown() + + +class IPSec4SpdTestCaseBypass(SpdFastPathInbound): + """ IPSec/IPv4 inbound: Policy mode test case with fast path \ + (add bypass)""" + + def test_ipsec_spd_inbound_bypass(self): + # In this test case, packets in IPv4 FWD path are configured + # to go through IPSec inbound SPD policy lookup. + # + # 2 inbound SPD rules (1 HIGH and 1 LOW) are added. + # - High priority rule action is set to DISCARD. + # - Low priority rule action is set to BYPASS. + # + # Since BYPASS rules take precedence over DISCARD + # (the order being PROTECT, BYPASS, DISCARD) we expect the + # BYPASS rule to match and traffic to be correctly forwarded. + self.create_interfaces(2) + pkt_count = 5 + + self.spd_create_and_intf_add(1, [self.pg1, self.pg0]) + + # create input rules + # bypass rule should take precedence over discard rule, + # even though it's lower priority, because for input policies + # matching PROTECT policies precedes matching BYPASS policies + # which preceeds matching for DISCARD policies. + # Any hit stops the process. + policy_0 = self.spd_add_rem_policy( # inbound, priority 10 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + ip_range=True, + local_ip_start=self.pg1.remote_ip4, + local_ip_stop=self.pg1.remote_ip4, + remote_ip_start=self.pg0.remote_ip4, + remote_ip_stop=self.pg0.remote_ip4, + ) + policy_1 = self.spd_add_rem_policy( # inbound, priority 15 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=15, + policy_type="discard", + ip_range=True, + local_ip_start=self.pg1.remote_ip4, + local_ip_stop=self.pg1.remote_ip4, + remote_ip_start=self.pg0.remote_ip4, + remote_ip_stop=self.pg0.remote_ip4, + ) + + # create output rule so we can capture forwarded packets + policy_2 = self.spd_add_rem_policy( # outbound, priority 10 + 1, + self.pg0, + self.pg1, + socket.IPPROTO_UDP, + is_out=1, + priority=10, + policy_type="bypass", + ) + + # create the packet stream + packets = self.create_stream(self.pg0, self.pg1, pkt_count) + # add the stream to the source interface + self.pg0.add_stream(packets) + self.pg1.enable_capture() + self.pg_start() + + # check capture on pg1 + capture = self.pg1.get_capture() + for packet in capture: + try: + self.logger.debug(ppp("SPD Add - Got packet:", packet)) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + self.logger.debug("SPD: Num packets: %s", len(capture.res)) + + # verify captured packets + self.verify_capture(self.pg0, self.pg1, capture) + # verify all policies matched the expected number of times + self.verify_policy_match(pkt_count, policy_0) + self.verify_policy_match(0, policy_1) + self.verify_policy_match(pkt_count, policy_2) + + +class IPSec4SpdTestCaseDiscard(SpdFastPathInbound): + """ IPSec/IPv4 inbound: Policy mode test case with fast path \ + (add discard)""" + + def test_ipsec_spd_inbound_discard(self): + # In this test case, packets in IPv4 FWD path are configured + # to go through IPSec inbound SPD policy lookup. + # + # Rule action is set to DISCARD. + + self.create_interfaces(2) + pkt_count = 5 + + self.spd_create_and_intf_add(1, [self.pg1, self.pg0]) + + # create input rules + # bypass rule should take precedence over discard rule, + # even though it's lower priority + policy_0 = self.spd_add_rem_policy( # inbound, priority 10 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="discard", + ) + + # create output rule so we can capture forwarded packets + policy_1 = self.spd_add_rem_policy( # outbound, priority 10 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=1, + priority=10, + policy_type="bypass", + ) + + # create the packet stream + packets = self.create_stream(self.pg0, self.pg1, pkt_count) + # add the stream to the source interface + self.pg0.add_stream(packets) + self.pg1.enable_capture() + self.pg_start() + + # check capture on pg1 + capture = self.pg1.assert_nothing_captured() + + # verify all policies matched the expected number of times + self.verify_policy_match(pkt_count, policy_0) + self.verify_policy_match(0, policy_1) + + +class IPSec4SpdTestCaseProtect(SpdFastPathInboundProtect): + """ IPSec/IPv4 inbound: Policy mode test case with fast path \ + (add protect)""" + + @classmethod + def setUpClass(cls): + super(IPSec4SpdTestCaseProtect, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(IPSec4SpdTestCaseProtect, cls).tearDownClass() + + def setUp(self): + super(IPSec4SpdTestCaseProtect, self).setUp() + + def tearDown(self): + super(IPSec4SpdTestCaseProtect, self).tearDown() + + def test_ipsec_spd_inbound_protect(self): + # In this test case, encrypted packets in IPv4 + # PROTECT path are configured + # to go through IPSec inbound SPD policy lookup. + + pkt_count = 5 + payload_size = 64 + p = self.params[socket.AF_INET] + send_pkts = self.gen_encrypt_pkts( + p, + p.scapy_tra_sa, + self.tra_if, + src=self.tra_if.remote_ip4, + dst=self.tra_if.local_ip4, + count=pkt_count, + payload_size=payload_size, + ) + recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if) + + self.logger.info(self.vapi.ppcli("show error")) + self.logger.info(self.vapi.ppcli("show ipsec all")) + + pkts = p.tra_sa_in.get_stats()["packets"] + self.assertEqual( + pkts, + pkt_count, + "incorrect SA in counts: expected %d != %d" % (pkt_count, pkts), + ) + pkts = p.tra_sa_out.get_stats()["packets"] + self.assertEqual( + pkts, + pkt_count, + "incorrect SA out counts: expected %d != %d" % (pkt_count, pkts), + ) + self.assertEqual(p.tra_sa_out.get_err("lost"), 0) + self.assertEqual(p.tra_sa_in.get_err("lost"), 0) + + +class IPSec4SpdTestCaseAddIPRange(SpdFastPathInbound): + """ IPSec/IPv4 inbound: Policy mode test case with fast path \ + (add ips range with any port rule)""" + + def test_ipsec_spd_inbound_add(self): + # In this test case, packets in IPv4 FWD path are configured + # to go through IPSec inbound SPD policy lookup. + # 2 SPD bypass rules (1 for inbound and 1 for outbound) are added. + # Traffic sent on pg0 interface should match fast path priority + # rule and should be sent out on pg1 interface. + self.create_interfaces(2) + pkt_count = 5 + s_ip_s1 = ipaddress.ip_address(self.pg0.remote_ip4) + s_ip_e1 = ipaddress.ip_address(int(s_ip_s1) + 5) + d_ip_s1 = ipaddress.ip_address(self.pg1.remote_ip4) + d_ip_e1 = ipaddress.ip_address(int(d_ip_s1) + 0) + + s_ip_s0 = ipaddress.ip_address(self.pg0.remote_ip4) + s_ip_e0 = ipaddress.ip_address(int(s_ip_s0) + 6) + d_ip_s0 = ipaddress.ip_address(self.pg1.remote_ip4) + d_ip_e0 = ipaddress.ip_address(int(d_ip_s0) + 0) + self.spd_create_and_intf_add(1, [self.pg1, self.pg0]) + + policy_0 = self.spd_add_rem_policy( # inbound fast path, priority 10 + 1, + self.pg0, + self.pg1, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + ip_range=True, + local_ip_start=d_ip_s0, + local_ip_stop=d_ip_e0, + remote_ip_start=s_ip_s0, + remote_ip_stop=s_ip_e0, + ) + policy_1 = self.spd_add_rem_policy( # outbound, priority 5 + 1, + self.pg0, + self.pg1, + socket.IPPROTO_UDP, + is_out=1, + priority=5, + policy_type="bypass", + ip_range=True, + local_ip_start=s_ip_s1, + local_ip_stop=s_ip_e1, + remote_ip_start=d_ip_s1, + remote_ip_stop=d_ip_e1, + ) + + # create the packet stream + packets = self.create_stream(self.pg0, self.pg1, pkt_count) + # add the stream to the source interface + enable capture + self.pg0.add_stream(packets) + self.pg0.enable_capture() + self.pg1.enable_capture() + # start the packet generator + self.pg_start() + # get capture + capture = self.pg1.get_capture() + for packet in capture: + try: + self.logger.debug(ppp("SPD - Got packet:", packet)) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + self.logger.debug("SPD: Num packets: %s", len(capture.res)) + + # assert nothing captured on pg0 + self.pg0.assert_nothing_captured() + # verify captured packets + self.verify_capture(self.pg0, self.pg1, capture) + # verify all policies matched the expected number of times + self.verify_policy_match(pkt_count, policy_0) + self.verify_policy_match(pkt_count, policy_1) + + +class IPSec4SpdTestCaseAddAll(SpdFastPathInbound): + """ IPSec/IPv4 inbound: Policy mode test case with fast path \ + (add all ips ports rule)""" + + def test_ipsec_spd_inbound_add(self): + # In this test case, packets in IPv4 FWD path are configured + # to go through IPSec inbound SPD policy lookup. + # 2 SPD rules (1 HIGH and 1 LOW) are added. + # Low priority rule action is set to BYPASS all ips. + # High priority rule action is set to DISCARD all ips. + # Traffic not sent on pg0 interface when HIGH discard priority rule is added. + # Then LOW priority + # rule is added and send the same traffic to pg0, this time expect + # the traffic is bypassed as bypass takes priority over discard. + self.create_interfaces(2) + pkt_count = 5 + self.spd_create_and_intf_add(1, [self.pg0, self.pg1]) + + policy_0 = self.spd_add_rem_policy( # inbound, priority 20 + 1, + self.pg0, + self.pg1, + socket.IPPROTO_UDP, + is_out=0, + priority=20, + policy_type="discard", + all_ips=True, + ) + + policy_1 = self.spd_add_rem_policy( # inbound, priority 20 + 1, + self.pg0, + self.pg1, + socket.IPPROTO_UDP, + is_out=True, + priority=5, + policy_type="bypass", + all_ips=True, + ) + + # create the packet stream + packets = self.create_stream(self.pg0, self.pg1, pkt_count) + # add the stream to the source interface + enable capture + self.pg0.add_stream(packets) + self.pg0.enable_capture() + self.pg1.enable_capture() + # start the packet generator + self.pg_start() + # assert nothing captured on pg0 and pg1 + self.pg0.assert_nothing_captured() + self.pg1.assert_nothing_captured() + + policy_2 = self.spd_add_rem_policy( # inbound, priority 10 + 1, + self.pg0, + self.pg1, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + all_ips=True, + ) + + # create the packet stream + packets = self.create_stream(self.pg0, self.pg1, pkt_count) + # add the stream to the source interface + enable capture + self.pg0.add_stream(packets) + self.pg0.enable_capture() + self.pg1.enable_capture() + # start the packet generator + self.pg_start() + # get capture + capture = self.pg1.get_capture(expected_count=pkt_count) + for packet in capture: + try: + self.logger.debug(ppp("SPD - Got packet:", packet)) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + self.logger.debug("SPD: Num packets: %s", len(capture.res)) + + # assert nothing captured on pg0 + self.pg0.assert_nothing_captured() + # verify all policies matched the expected number of times + self.verify_policy_match(pkt_count, policy_2) + + +class IPSec4SpdTestCaseRemove(SpdFastPathInbound): + """ IPSec/IPv4 inbound: Policy mode test case with fast path \ + (remove rule)""" + + def test_ipsec_spd_inbound_remove(self): + # In this test case, packets in IPv4 FWD path are configured + # to go through IPSec inbound SPD policy lookup. + # 2 SPD rules (1 HIGH and 1 LOW) are added. + # High priority rule action is set to BYPASS. + # Low priority rule action is set to DISCARD. + # High priority rule is then removed. + # Traffic sent on pg0 interface should match low priority + # rule and should be discarded after SPD lookup. + self.create_interfaces(2) + pkt_count = 5 + self.spd_create_and_intf_add(1, [self.pg0, self.pg1]) + policy_0 = self.spd_add_rem_policy( # inbound, priority 10 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + ) + policy_1 = self.spd_add_rem_policy( # inbound, priority 5 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=5, + policy_type="discard", + ) + + policy_out = self.spd_add_rem_policy( # outbound, priority 10 + 1, + self.pg0, + self.pg1, + socket.IPPROTO_UDP, + is_out=1, + priority=10, + policy_type="bypass", + ) + + # create the packet stream + packets = self.create_stream(self.pg0, self.pg1, pkt_count) + # add the stream to the source interface + enable capture + self.pg0.add_stream(packets) + self.pg0.enable_capture() + self.pg1.enable_capture() + # start the packet generator + self.pg_start() + # get capture + capture = self.pg1.get_capture() + for packet in capture: + try: + self.logger.debug(ppp("SPD - Got packet:", packet)) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + + # assert nothing captured on pg0 + self.pg0.assert_nothing_captured() + # verify capture on pg1 + self.logger.debug("SPD: Num packets: %s", len(capture.res)) + self.verify_capture(self.pg0, self.pg1, capture) + # verify all policies matched the expected number of times + self.verify_policy_match(pkt_count, policy_0) + self.verify_policy_match(0, policy_1) + # now remove the bypass rule + self.spd_add_rem_policy( # outbound, priority 10 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + remove=True, + ) + + # resend the same packets + self.pg0.add_stream(packets) + self.pg0.enable_capture() # flush the old captures + self.pg1.enable_capture() + self.pg_start() + # assert nothing captured on pg0 + self.pg0.assert_nothing_captured() + # all packets will be dropped by SPD rule + self.pg1.assert_nothing_captured() + # verify all policies matched the expected number of times + self.verify_policy_match(pkt_count, policy_0) + self.verify_policy_match(pkt_count, policy_1) + + +class IPSec4SpdTestCaseReadd(SpdFastPathInbound): + """ IPSec/IPv4 inbound: Policy mode test case with fast path \ + (add, remove, re-add)""" + + def test_ipsec_spd_inbound_readd(self): + # In this test case, packets in IPv4 FWD path are configured + # to go through IPSec outbound SPD policy lookup. + # 2 SPD rules (1 HIGH and 1 LOW) are added. + # High priority rule action is set to BYPASS. + # Low priority rule action is set to DISCARD. + # Traffic sent on pg0 interface should match high priority + # rule and should be sent out on pg1 interface. + # High priority rule is then removed. + # Traffic sent on pg0 interface should match low priority + # rule and should be discarded after SPD lookup. + # Readd high priority rule. + # Traffic sent on pg0 interface should match high priority + # rule and should be sent out on pg1 interface. + self.create_interfaces(2) + pkt_count = 5 + self.spd_create_and_intf_add(1, [self.pg0, self.pg1]) + policy_0 = self.spd_add_rem_policy( # inbound, priority 10 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + ) + policy_1 = self.spd_add_rem_policy( # inbound, priority 5 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=5, + policy_type="discard", + ) + policy_2 = self.spd_add_rem_policy( # outbound, priority 10 + 1, + self.pg0, + self.pg1, + socket.IPPROTO_UDP, + is_out=1, + priority=10, + policy_type="bypass", + ) + + # create the packet stream + packets = self.create_stream(self.pg0, self.pg1, pkt_count) + # add the stream to the source interface + enable capture + self.pg0.add_stream(packets) + self.pg0.enable_capture() + self.pg1.enable_capture() + # start the packet generator + self.pg_start() + # get capture + capture = self.pg1.get_capture() + for packet in capture: + try: + self.logger.debug(ppp("SPD - Got packet:", packet)) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + self.logger.debug("SPD: Num packets: %s", len(capture.res)) + + # assert nothing captured on pg0 + self.pg0.assert_nothing_captured() + # verify capture on pg1 + self.verify_capture(self.pg0, self.pg1, capture) + # verify all policies matched the expected number of times + self.verify_policy_match(pkt_count, policy_0) + self.verify_policy_match(0, policy_1) + # remove the bypass rule, leaving only the discard rule + self.spd_add_rem_policy( # inbound, priority 10 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + remove=True, + ) + + # resend the same packets + self.pg0.add_stream(packets) + self.pg0.enable_capture() # flush the old captures + self.pg1.enable_capture() + self.pg_start() + + # assert nothing captured on pg0 + self.pg0.assert_nothing_captured() + # all packets will be dropped by SPD rule + self.pg1.assert_nothing_captured() + # verify all policies matched the expected number of times + self.verify_policy_match(pkt_count, policy_0) + self.verify_policy_match(pkt_count, policy_1) + + # now readd the bypass rule + policy_0 = self.spd_add_rem_policy( # outbound, priority 10 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + ) + + # resend the same packets + self.pg0.add_stream(packets) + self.pg0.enable_capture() # flush the old captures + self.pg1.enable_capture() + self.pg_start() + + # get capture + capture = self.pg1.get_capture(pkt_count) + for packet in capture: + try: + self.logger.debug(ppp("SPD - Got packet:", packet)) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + self.logger.debug("SPD: Num packets: %s", len(capture.res)) + + # assert nothing captured on pg0 + self.pg0.assert_nothing_captured() + # verify captured packets + self.verify_capture(self.pg0, self.pg1, capture) + # verify all policies matched the expected number of times + self.verify_policy_match(pkt_count, policy_0) + self.verify_policy_match(pkt_count, policy_1) + + +class IPSec4SpdTestCaseMultiple(SpdFastPathInbound): + """ IPSec/IPv4 inbound: Policy mode test case with fast path \ + (multiple interfaces, multiple rules)""" + + def test_ipsec_spd_inbound_multiple(self): + # In this test case, packets in IPv4 FWD path are configured to go + # through IPSec outbound SPD policy lookup. + # Multiples rules on multiple interfaces are tested at the same time. + # 3x interfaces are configured, binding the same SPD to each. + # Each interface has 2 SPD rules (1 BYPASS and 1 DISCARD). + # On pg0 & pg1, the BYPASS rule is HIGH priority + # On pg2, the DISCARD rule is HIGH priority + # Traffic should be received on pg0 & pg1 and dropped on pg2. + self.create_interfaces(3) + pkt_count = 5 + # bind SPD to all interfaces + self.spd_create_and_intf_add(1, self.pg_interfaces) + # add rules on all interfaces + policy_01 = self.spd_add_rem_policy( # inbound, priority 10 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + ) + policy_02 = self.spd_add_rem_policy( # inbound, priority 5 + 1, + self.pg1, + self.pg0, + socket.IPPROTO_UDP, + is_out=0, + priority=5, + policy_type="discard", + ) + + policy_11 = self.spd_add_rem_policy( # inbound, priority 10 + 1, + self.pg2, + self.pg1, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="bypass", + ) + policy_12 = self.spd_add_rem_policy( # inbound, priority 5 + 1, + self.pg2, + self.pg1, + socket.IPPROTO_UDP, + is_out=0, + priority=5, + policy_type="discard", + ) + + policy_21 = self.spd_add_rem_policy( # inbound, priority 5 + 1, + self.pg0, + self.pg2, + socket.IPPROTO_UDP, + is_out=0, + priority=5, + policy_type="bypass", + ) + policy_22 = self.spd_add_rem_policy( # inbound, priority 10 + 1, + self.pg0, + self.pg2, + socket.IPPROTO_UDP, + is_out=0, + priority=10, + policy_type="discard", + ) + + # interfaces bound to an SPD, will by default drop outbound + # traffic with no matching policies. add catch-all outbound + # bypass rule to SPD: + self.spd_add_rem_policy( # outbound, all interfaces + 1, + None, + None, + socket.IPPROTO_UDP, + is_out=1, + priority=10, + policy_type="bypass", + all_ips=True, + ) + + # create the packet streams + packets0 = self.create_stream(self.pg0, self.pg1, pkt_count) + packets1 = self.create_stream(self.pg1, self.pg2, pkt_count) + packets2 = self.create_stream(self.pg2, self.pg0, pkt_count) + # add the streams to the source interfaces + self.pg0.add_stream(packets0) + self.pg1.add_stream(packets1) + self.pg2.add_stream(packets2) + # enable capture on all interfaces + for pg in self.pg_interfaces: + pg.enable_capture() + # start the packet generator + self.pg_start() + + # get captures + if_caps = [] + for pg in [self.pg1, self.pg2]: # we are expecting captures on pg1/pg2 + if_caps.append(pg.get_capture()) + for packet in if_caps[-1]: + try: + self.logger.debug(ppp("SPD - Got packet:", packet)) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + self.logger.debug("SPD: Num packets: %s", len(if_caps[0].res)) + self.logger.debug("SPD: Num packets: %s", len(if_caps[1].res)) + + # verify captures that matched BYPASS rule + self.verify_capture(self.pg0, self.pg1, if_caps[0]) + self.verify_capture(self.pg1, self.pg2, if_caps[1]) + # verify that traffic to pg0 matched BYPASS rule + # although DISCARD rule had higher prioriy and was not dropped + self.verify_policy_match(pkt_count, policy_21) + + # verify all packets that were expected to match rules, matched + # pg0 -> pg1 + self.verify_policy_match(pkt_count, policy_01) + self.verify_policy_match(0, policy_02) + # pg1 -> pg2 + self.verify_policy_match(pkt_count, policy_11) + self.verify_policy_match(0, policy_12) + # pg2 -> pg0 + self.verify_policy_match(0, policy_22) + + +class IPSec6SpdTestCaseProtect(SpdFastPathIPv6InboundProtect): + """ IPSec/IPv6 inbound: Policy mode test case with fast path \ + (add protect)""" + + @classmethod + def setUpClass(cls): + super(IPSec6SpdTestCaseProtect, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(IPSec6SpdTestCaseProtect, cls).tearDownClass() + + def setUp(self): + super(IPSec6SpdTestCaseProtect, self).setUp() + + def tearDown(self): + super(IPSec6SpdTestCaseProtect, self).tearDown() + + def test_ipsec6_spd_inbound_protect(self): + pkt_count = 5 + payload_size = 64 + p = self.params[socket.AF_INET6] + send_pkts = self.gen_encrypt_pkts6( + p, + p.scapy_tra_sa, + self.tra_if, + src=self.tra_if.remote_ip6, + dst=self.tra_if.local_ip6, + count=pkt_count, + payload_size=payload_size, + ) + recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if) + + self.logger.info(self.vapi.ppcli("show error")) + self.logger.info(self.vapi.ppcli("show ipsec all")) + pkts = p.tra_sa_in.get_stats()["packets"] + self.assertEqual( + pkts, + pkt_count, + "incorrect SA in counts: expected %d != %d" % (pkt_count, pkts), + ) + pkts = p.tra_sa_out.get_stats()["packets"] + self.assertEqual( + pkts, + pkt_count, + "incorrect SA out counts: expected %d != %d" % (pkt_count, pkts), + ) + self.assertEqual(p.tra_sa_out.get_err("lost"), 0) + self.assertEqual(p.tra_sa_in.get_err("lost"), 0) + + +if __name__ == "__main__": + unittest.main(testRunner=VppTestRunner) |