| 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) |