diff options
Diffstat (limited to 'test/asf/test_ipsec_spd_fp_output.py')
| -rw-r--r-- | test/asf/test_ipsec_spd_fp_output.py | 1418 |
1 files changed, 0 insertions, 1418 deletions
diff --git a/test/asf/test_ipsec_spd_fp_output.py b/test/asf/test_ipsec_spd_fp_output.py deleted file mode 100644 index a92669a4f3f..00000000000 --- a/test/asf/test_ipsec_spd_fp_output.py +++ /dev/null @@ -1,1418 +0,0 @@ -import socket -import unittest -import ipaddress - -from util import ppp -from asfframework import VppTestRunner -from template_ipsec import IPSecIPv4Fwd -from template_ipsec import IPSecIPv6Fwd - - -class SpdFastPathOutbound(IPSecIPv4Fwd): - # Override setUpConstants to enable outbound fast path in config - @classmethod - def setUpConstants(cls): - super(SpdFastPathOutbound, cls).setUpConstants() - cls.vpp_cmdline.extend(["ipsec", "{", "ipv4-outbound-spd-fast-path on", "}"]) - cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline)) - - -class SpdFastPathIPv6Outbound(IPSecIPv6Fwd): - # Override setUpConstants to enable outbound fast path in config - @classmethod - def setUpConstants(cls): - super(SpdFastPathIPv6Outbound, cls).setUpConstants() - cls.vpp_cmdline.extend(["ipsec", "{", "ipv6-outbound-spd-fast-path on", "}"]) - cls.logger.info("VPP modified cmdline is %s" % " ".join(cls.vpp_cmdline)) - - -class IPSec4SpdTestCaseAdd(SpdFastPathOutbound): - """ IPSec/IPv4 outbound: Policy mode test case with fast path \ - (add rule)""" - - def test_ipsec_spd_outbound_add(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. - self.create_interfaces(2) - pkt_count = 5 - s_port_s = 1111 - s_port_e = 1111 - d_port_s = 2222 - d_port_e = 2222 - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - 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="discard", - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - - # create the packet stream - packets = self.create_stream(self.pg0, self.pg1, pkt_count, s_port_s, d_port_s) - # 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(0, policy_1) - - -class IPSec4SpdTestCaseAddPortRange(SpdFastPathOutbound): - """ IPSec/IPv4 outbound: Policy mode test case with fast path \ - (add all ips port range rule)""" - - def test_ipsec_spd_outbound_add(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. - self.create_interfaces(2) - pkt_count = 5 - s_port_s = 1000 - s_port_e = 2023 - d_port_s = 5000 - d_port_e = 6023 - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - all_ips=True, - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - 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="discard", - all_ips=True, - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - - # create the packet stream - packets = self.create_stream(self.pg0, self.pg1, pkt_count, 1333, 5444) - # 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(0, policy_1) - - -class IPSec4SpdTestCaseAddIPRange(SpdFastPathOutbound): - """ IPSec/IPv4 outbound: Policy mode test case with fast path \ - (add ips range with any port rule)""" - - def test_ipsec_spd_outbound_add(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. - self.create_interfaces(2) - pkt_count = 5 - s_ip_s = ipaddress.ip_address(self.pg0.remote_ip4) - s_ip_e = ipaddress.ip_address(int(s_ip_s) + 5) - d_ip_s = ipaddress.ip_address(self.pg1.remote_ip4) - d_ip_e = ipaddress.ip_address(int(d_ip_s) + 0) - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ip_range=True, - local_ip_start=s_ip_s, - local_ip_stop=s_ip_e, - remote_ip_start=d_ip_s, - remote_ip_stop=d_ip_e, - ) - 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="discard", - ip_range=True, - local_ip_start=s_ip_s, - local_ip_stop=s_ip_e, - remote_ip_start=d_ip_s, - remote_ip_stop=d_ip_e, - ) - - # 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(0, policy_1) - - -class IPSec4SpdTestCaseAddIPAndPortRange(SpdFastPathOutbound): - """ IPSec/IPv4 outbound: Policy mode test case with fast path \ - (add all ips range rule)""" - - def test_ipsec_spd_outbound_add(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. - # in this test we define ranges of ports and ip addresses. - self.create_interfaces(2) - pkt_count = 5 - s_port_s = 1000 - s_port_e = 1000 + 1023 - d_port_s = 5000 - d_port_e = 5000 + 1023 - - s_ip_s = ipaddress.ip_address( - int(ipaddress.ip_address(self.pg0.remote_ip4)) - 24 - ) - s_ip_e = ipaddress.ip_address(int(s_ip_s) + 255) - d_ip_s = ipaddress.ip_address(self.pg1.remote_ip4) - d_ip_e = ipaddress.ip_address(int(d_ip_s) + 255) - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ip_range=True, - local_ip_start=s_ip_s, - local_ip_stop=s_ip_e, - remote_ip_start=d_ip_s, - remote_ip_stop=d_ip_e, - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - 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="discard", - ip_range=True, - local_ip_start=s_ip_s, - local_ip_stop=s_ip_e, - remote_ip_start=d_ip_s, - remote_ip_stop=d_ip_e, - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - - # 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(0, policy_1) - - -class IPSec4SpdTestCaseAddAll(SpdFastPathOutbound): - """ IPSec/IPv4 outbound: Policy mode test case with fast path \ - (add all ips ports rule)""" - - def test_ipsec_spd_outbound_add(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. - # Low priority rule action is set to BYPASS all ips. - # High priority rule action is set to DISCARD all ips. - # Traffic sent on pg0 interface when LOW priority rule is added, - # expect the packet is being sent out to pg1. Then HIGH priority - # rule is added and send the same traffic to pg0, this time expect - # the traffic is dropped. - self.create_interfaces(2) - pkt_count = 5 - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - 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() - 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) - - policy_1 = self.spd_add_rem_policy( # outbound, priority 20 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=20, - policy_type="discard", - 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() - - -class IPSec4SpdTestCaseRemove(SpdFastPathOutbound): - """ IPSec/IPv4 outbound: Policy mode test case with fast path \ - (remove rule)""" - - def test_ipsec_spd_outbound_remove(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. - # 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.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ) - 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="discard", - ) - - # 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.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - 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(SpdFastPathOutbound): - """ IPSec/IPv4 outbound: Policy mode test case with fast path \ - (add, remove, re-add)""" - - def test_ipsec_spd_outbound_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.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ) - 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="discard", - ) - - # 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( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - 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.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - 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(SpdFastPathOutbound): - """ IPSec/IPv4 outbound: Policy mode test case with fast path \ - (multiple interfaces, multiple rules)""" - - def test_ipsec_spd_outbound_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( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ) - policy_02 = self.spd_add_rem_policy( # outbound, priority 5 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=5, - policy_type="discard", - ) - - policy_11 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg1, - self.pg2, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ) - policy_12 = self.spd_add_rem_policy( # outbound, priority 5 - 1, - self.pg1, - self.pg2, - socket.IPPROTO_UDP, - is_out=1, - priority=5, - policy_type="discard", - ) - - policy_21 = self.spd_add_rem_policy( # outbound, priority 5 - 1, - self.pg2, - self.pg0, - socket.IPPROTO_UDP, - is_out=1, - priority=5, - policy_type="bypass", - ) - policy_22 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg2, - self.pg0, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="discard", - ) - - # interfaces bound to an SPD, will by default drop inbound - # traffic with no matching policies. add catch-all inbound - # bypass rule to SPD: - self.spd_add_rem_policy( # inbound, all interfaces - 1, - None, - None, - socket.IPPROTO_UDP, - is_out=0, - 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 DISCARD rule and was dropped - self.pg0.assert_nothing_captured() - # 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_21) - self.verify_policy_match(pkt_count, policy_22) - - -class IPSec6SpdTestCaseAdd(SpdFastPathIPv6Outbound): - """ IPSec/IPv6 outbound: Policy mode test case with fast path \ - (add rule)""" - - def test_ipsec_spd_outbound_add(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. - self.create_interfaces(2) - pkt_count = 5 - s_port_s = 1111 - s_port_e = 1111 - d_port_s = 2222 - d_port_e = 2222 - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - 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="discard", - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - - # create the packet stream - packets = self.create_stream(self.pg0, self.pg1, pkt_count, s_port_s, d_port_s) - # 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(0, policy_1) - - -class IPSec6SpdTestCaseAddAll(SpdFastPathIPv6Outbound): - """ IPSec/IPv6 outbound: Policy mode test case with fast path \ - (add all ips ports rule)""" - - def test_ipsec_spd_outbound_add(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. - # Low priority rule action is set to BYPASS all ips. - # High priority rule action is set to DISCARD all ips. - # Traffic sent on pg0 interface when LOW priority rule is added, - # expect the packet is being sent out to pg1. Then HIGH priority - # rule is added and send the same traffic to pg0, this time expect - # the traffic is dropped. - self.create_interfaces(2) - pkt_count = 5 - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - 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() - 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) - - policy_1 = self.spd_add_rem_policy( # outbound, priority 20 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=20, - policy_type="discard", - 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() - - -class IPSec6SpdTestCaseAddPortRange(SpdFastPathIPv6Outbound): - """ IPSec/IPv6 outbound: Policy mode test case with fast path \ - (add all ips port range rule)""" - - def test_ipsec_spd_outbound_add(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. - self.create_interfaces(2) - pkt_count = 5 - s_port_s = 1000 - s_port_e = 2023 - d_port_s = 5000 - d_port_e = 6023 - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - all_ips=True, - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - 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="discard", - all_ips=True, - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - - # create the packet stream - packets = self.create_stream(self.pg0, self.pg1, pkt_count, 1333, 5444) - # 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(0, policy_1) - - -class IPSec6SpdTestCaseAddIPRange(SpdFastPathIPv6Outbound): - """ IPSec/IPv6 outbound: Policy mode test case with fast path \ - (add ips range with any port rule)""" - - def test_ipsec_spd_outbound_add(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. - self.create_interfaces(2) - pkt_count = 5 - s_ip_s = ipaddress.ip_address(self.pg0.remote_ip6) - s_ip_e = ipaddress.ip_address(int(s_ip_s) + 5) - d_ip_s = ipaddress.ip_address(self.pg1.remote_ip6) - d_ip_e = ipaddress.ip_address(int(d_ip_s) + 0) - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ip_range=True, - local_ip_start=s_ip_s, - local_ip_stop=s_ip_e, - remote_ip_start=d_ip_s, - remote_ip_stop=d_ip_e, - ) - 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="discard", - ip_range=True, - local_ip_start=s_ip_s, - local_ip_stop=s_ip_e, - remote_ip_start=d_ip_s, - remote_ip_stop=d_ip_e, - ) - - # 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(0, policy_1) - - -class IPSec6SpdTestCaseAddIPAndPortRange(SpdFastPathIPv6Outbound): - """ IPSec/IPvr6 outbound: Policy mode test case with fast path \ - (add all ips range rule)""" - - def test_ipsec_spd_outbound_add(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. - # in this test we define ranges of ports and ip addresses. - self.create_interfaces(2) - pkt_count = 5 - s_port_s = 1000 - s_port_e = 1000 + 1023 - d_port_s = 5000 - d_port_e = 5000 + 1023 - - s_ip_s = ipaddress.ip_address( - int(ipaddress.ip_address(self.pg0.remote_ip6)) - 24 - ) - s_ip_e = ipaddress.ip_address(int(s_ip_s) + 255) - d_ip_s = ipaddress.ip_address(self.pg1.remote_ip6) - d_ip_e = ipaddress.ip_address(int(d_ip_s) + 255) - self.spd_create_and_intf_add(1, [self.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ip_range=True, - local_ip_start=s_ip_s, - local_ip_stop=s_ip_e, - remote_ip_start=d_ip_s, - remote_ip_stop=d_ip_e, - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - 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="discard", - ip_range=True, - local_ip_start=s_ip_s, - local_ip_stop=s_ip_e, - remote_ip_start=d_ip_s, - remote_ip_stop=d_ip_e, - local_port_start=s_port_s, - local_port_stop=s_port_e, - remote_port_start=d_port_s, - remote_port_stop=d_port_e, - ) - - # 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(0, policy_1) - - -class IPSec6SpdTestCaseReadd(SpdFastPathIPv6Outbound): - """ IPSec/IPv6 outbound: Policy mode test case with fast path \ - (add, remove, re-add)""" - - def test_ipsec_spd_outbound_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.pg1]) - policy_0 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ) - 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="discard", - ) - - # 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( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - 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.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - 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 IPSec6SpdTestCaseMultiple(SpdFastPathIPv6Outbound): - """ IPSec/IPv6 outbound: Policy mode test case with fast path \ - (multiple interfaces, multiple rules)""" - - def test_ipsec_spd_outbound_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( # outbound, priority 10 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ) - policy_02 = self.spd_add_rem_policy( # outbound, priority 5 - 1, - self.pg0, - self.pg1, - socket.IPPROTO_UDP, - is_out=1, - priority=5, - policy_type="discard", - ) - - policy_11 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg1, - self.pg2, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="bypass", - ) - policy_12 = self.spd_add_rem_policy( # outbound, priority 5 - 1, - self.pg1, - self.pg2, - socket.IPPROTO_UDP, - is_out=1, - priority=5, - policy_type="discard", - ) - - policy_21 = self.spd_add_rem_policy( # outbound, priority 5 - 1, - self.pg2, - self.pg0, - socket.IPPROTO_UDP, - is_out=1, - priority=5, - policy_type="bypass", - ) - policy_22 = self.spd_add_rem_policy( # outbound, priority 10 - 1, - self.pg2, - self.pg0, - socket.IPPROTO_UDP, - is_out=1, - priority=10, - policy_type="discard", - ) - - # interfaces bound to an SPD, will by default drop inbound - # traffic with no matching policies. add catch-all inbound - # bypass rule to SPD: - self.spd_add_rem_policy( # inbound, all interfaces - 1, - None, - None, - socket.IPPROTO_UDP, - is_out=0, - 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 DISCARD rule and was dropped - self.pg0.assert_nothing_captured() - # 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_21) - self.verify_policy_match(pkt_count, policy_22) - - -if __name__ == "__main__": - unittest.main(testRunner=VppTestRunner) |