| import socket |
| import unittest |
| import ipaddress |
| |
| from util import ppp |
| from framework import VppTestRunner |
| from template_ipsec import IPSecIPv4Fwd |
| from template_ipsec import IPSecIPv6Fwd |
| from test_ipsec_esp import TemplateIpsecEsp |
| |
| |
| def debug_signal_handler(signal, frame): |
| import pdb |
| |
| pdb.set_trace() |
| |
| |
| import signal |
| |
| signal.signal(signal.SIGINT, debug_signal_handler) |
| |
| |
| class SpdFastPathInbound(IPSecIPv4Fwd): |
| # 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) |