diff options
Diffstat (limited to 'test/test_ipsec_tun_if_esp.py')
| -rw-r--r-- | test/test_ipsec_tun_if_esp.py | 236 |
1 files changed, 236 insertions, 0 deletions
diff --git a/test/test_ipsec_tun_if_esp.py b/test/test_ipsec_tun_if_esp.py index 49c4d63161d..6f7752915d0 100644 --- a/test/test_ipsec_tun_if_esp.py +++ b/test/test_ipsec_tun_if_esp.py @@ -25,6 +25,7 @@ from util import ppp from vpp_papi import VppEnum from vpp_papi_provider import CliFailedCommandError from vpp_acl import AclRule, VppAcl, VppAclInterface +from vpp_policer import PolicerAction, VppPolicer def config_tun_params(p, encryption_type, tun_if, src=None, dst=None): @@ -469,6 +470,71 @@ class TestIpsec6TunIfEspHandoff(TemplateIpsec6TunIfEsp, tun6_encrypt_node_name = "esp6-encrypt-tun" tun6_decrypt_node_name = "esp6-decrypt-tun" + def test_tun_handoff_66_police(self): + """ ESP 6o6 tunnel with policer worker hand-off test """ + self.vapi.cli("clear errors") + self.vapi.cli("clear ipsec sa") + + N_PKTS = 15 + p = self.params[socket.AF_INET6] + + action_tx = PolicerAction( + VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, + 0) + policer = VppPolicer(self, "pol1", 80, 0, 1000, 0, + conform_action=action_tx, + exceed_action=action_tx, + violate_action=action_tx) + policer.add_vpp_config() + + # Start policing on tun + policer.apply_vpp_config(p.tun_if.sw_if_index, True) + + for pol_bind in [1, 0]: + policer.bind_vpp_config(pol_bind, True) + + # inject alternately on worker 0 and 1. + for worker in [0, 1, 0, 1]: + send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, + self.tun_if, + src=p.remote_tun_if_host, + dst=self.pg1.remote_ip6, + count=N_PKTS) + recv_pkts = self.send_and_expect(self.tun_if, send_pkts, + self.pg1, worker=worker) + self.verify_decrypted6(p, recv_pkts) + self.logger.debug(self.vapi.cli("show trace max 100")) + + stats = policer.get_stats() + stats0 = policer.get_stats(worker=0) + stats1 = policer.get_stats(worker=1) + + if pol_bind is 1: + # First pass: Worker 1, should have done all the policing + self.assertEqual(stats, stats1) + + # Worker 0, should have handed everything off + self.assertEqual(stats0['conform_packets'], 0) + self.assertEqual(stats0['exceed_packets'], 0) + self.assertEqual(stats0['violate_packets'], 0) + else: + # Second pass: both workers should have policed equal amounts + self.assertGreater(stats1['conform_packets'], 0) + self.assertEqual(stats1['exceed_packets'], 0) + self.assertGreater(stats1['violate_packets'], 0) + + self.assertGreater(stats0['conform_packets'], 0) + self.assertEqual(stats0['exceed_packets'], 0) + self.assertGreater(stats0['violate_packets'], 0) + + self.assertEqual(stats0['conform_packets'] + + stats0['violate_packets'], + stats1['conform_packets'] + + stats1['violate_packets']) + + policer.apply_vpp_config(p.tun_if.sw_if_index, False) + policer.remove_vpp_config() + class TestIpsec4TunIfEspHandoff(TemplateIpsec4TunIfEsp, IpsecTun4HandoffTests): @@ -476,6 +542,71 @@ class TestIpsec4TunIfEspHandoff(TemplateIpsec4TunIfEsp, tun4_encrypt_node_name = "esp4-encrypt-tun" tun4_decrypt_node_name = "esp4-decrypt-tun" + def test_tun_handoff_44_police(self): + """ ESP 4o4 tunnel with policer worker hand-off test """ + self.vapi.cli("clear errors") + self.vapi.cli("clear ipsec sa") + + N_PKTS = 15 + p = self.params[socket.AF_INET] + + action_tx = PolicerAction( + VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, + 0) + policer = VppPolicer(self, "pol1", 80, 0, 1000, 0, + conform_action=action_tx, + exceed_action=action_tx, + violate_action=action_tx) + policer.add_vpp_config() + + # Start policing on tun + policer.apply_vpp_config(p.tun_if.sw_if_index, True) + + for pol_bind in [1, 0]: + policer.bind_vpp_config(pol_bind, True) + + # inject alternately on worker 0 and 1. + for worker in [0, 1, 0, 1]: + send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, + self.tun_if, + src=p.remote_tun_if_host, + dst=self.pg1.remote_ip4, + count=N_PKTS) + recv_pkts = self.send_and_expect(self.tun_if, send_pkts, + self.pg1, worker=worker) + self.verify_decrypted(p, recv_pkts) + self.logger.debug(self.vapi.cli("show trace max 100")) + + stats = policer.get_stats() + stats0 = policer.get_stats(worker=0) + stats1 = policer.get_stats(worker=1) + + if pol_bind is 1: + # First pass: Worker 1, should have done all the policing + self.assertEqual(stats, stats1) + + # Worker 0, should have handed everything off + self.assertEqual(stats0['conform_packets'], 0) + self.assertEqual(stats0['exceed_packets'], 0) + self.assertEqual(stats0['violate_packets'], 0) + else: + # Second pass: both workers should have policed equal amounts + self.assertGreater(stats1['conform_packets'], 0) + self.assertEqual(stats1['exceed_packets'], 0) + self.assertGreater(stats1['violate_packets'], 0) + + self.assertGreater(stats0['conform_packets'], 0) + self.assertEqual(stats0['exceed_packets'], 0) + self.assertGreater(stats0['violate_packets'], 0) + + self.assertEqual(stats0['conform_packets'] + + stats0['violate_packets'], + stats1['conform_packets'] + + stats1['violate_packets']) + + policer.apply_vpp_config(p.tun_if.sw_if_index, False) + policer.remove_vpp_config() + @tag_fixme_vpp_workers class TestIpsec4MultiTunIfEsp(TemplateIpsec4TunProtect, @@ -2580,6 +2711,56 @@ class TestIpsecItf4(TemplateIpsec, self.unconfig_sa(p) self.unconfig_network(p) + def test_tun_44_police(self): + """IPSEC interface IPv4 with input policer""" + n_pkts = 127 + p = self.ipv4_params + + self.config_network(p) + self.config_sa_tun(p, + self.pg0.local_ip4, + self.pg0.remote_ip4) + self.config_protect(p) + + action_tx = PolicerAction( + VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, + 0) + policer = VppPolicer(self, "pol1", 80, 0, 1000, 0, + conform_action=action_tx, + exceed_action=action_tx, + violate_action=action_tx) + policer.add_vpp_config() + + # Start policing on tun + policer.apply_vpp_config(p.tun_if.sw_if_index, True) + + self.verify_tun_44(p, count=n_pkts) + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], n_pkts) + + stats = policer.get_stats() + + # Single rate, 2 colour policer - expect conform, violate but no exceed + self.assertGreater(stats['conform_packets'], 0) + self.assertEqual(stats['exceed_packets'], 0) + self.assertGreater(stats['violate_packets'], 0) + + # Stop policing on tun + policer.apply_vpp_config(p.tun_if.sw_if_index, False) + self.verify_tun_44(p, count=n_pkts) + + # No new policer stats + statsnew = policer.get_stats() + self.assertEqual(stats, statsnew) + + # teardown + policer.remove_vpp_config() + self.unconfig_protect(p) + self.unconfig_sa(p) + self.unconfig_network(p) + class TestIpsecItf4MPLS(TemplateIpsec, TemplateIpsecItf4, @@ -2822,6 +3003,61 @@ class TestIpsecItf6(TemplateIpsec, self.unconfig_sa(np) self.unconfig_network(p) + def test_tun_66_police(self): + """IPSEC interface IPv6 with input policer""" + tf = VppEnum.vl_api_tunnel_encap_decap_flags_t + n_pkts = 127 + p = self.ipv6_params + p.inner_hop_limit = 24 + p.outer_hop_limit = 23 + p.outer_flow_label = 243224 + p.tun_flags = tf.TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_HOP_LIMIT + + self.config_network(p) + self.config_sa_tun(p, + self.pg0.local_ip6, + self.pg0.remote_ip6) + self.config_protect(p) + + action_tx = PolicerAction( + VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, + 0) + policer = VppPolicer(self, "pol1", 80, 0, 1000, 0, + conform_action=action_tx, + exceed_action=action_tx, + violate_action=action_tx) + policer.add_vpp_config() + + # Start policing on tun + policer.apply_vpp_config(p.tun_if.sw_if_index, True) + + self.verify_tun_66(p, count=n_pkts) + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], n_pkts) + + stats = policer.get_stats() + + # Single rate, 2 colour policer - expect conform, violate but no exceed + self.assertGreater(stats['conform_packets'], 0) + self.assertEqual(stats['exceed_packets'], 0) + self.assertGreater(stats['violate_packets'], 0) + + # Stop policing on tun + policer.apply_vpp_config(p.tun_if.sw_if_index, False) + self.verify_tun_66(p, count=n_pkts) + + # No new policer stats + statsnew = policer.get_stats() + self.assertEqual(stats, statsnew) + + # teardown + policer.remove_vpp_config() + self.unconfig_protect(p) + self.unconfig_sa(p) + self.unconfig_network(p) + class TestIpsecMIfEsp4(TemplateIpsec, IpsecTun4): """ Ipsec P2MP ESP v4 tests """ |