from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
from scapy.contrib.mpls import MPLS
+from framework import tag_fixme_vpp_workers
from framework import VppTestRunner
from template_ipsec import TemplateIpsec, IpsecTun4Tests, IpsecTun6Tests, \
IpsecTun4, IpsecTun6, IpsecTcpTests, mk_scapy_crypt_key, \
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):
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
tun4_input_node = "ipsec4-tun-input"
def config_sa_tra(self, p):
""" IPsec UDP tunnel interface tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
@classmethod
class TestIpsec4TunIfEsp1(TemplateIpsec4TunIfEsp, IpsecTun4Tests):
""" Ipsec ESP - TUN tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
def test_tun_basic64(self):
""" ipsec 6o4 tunnel basic test """
IpsecTun6Tests):
""" Ipsec ESP - TUN tests """
tun6_encrypt_node_name = "esp6-encrypt-tun"
- tun6_decrypt_node_name = "esp6-decrypt-tun"
+ tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
def test_tun_basic46(self):
""" ipsec 4o6 tunnel basic test """
IpsecTun6HandoffTests):
""" Ipsec ESP 6 Handoff tests """
tun6_encrypt_node_name = "esp6-encrypt-tun"
- tun6_decrypt_node_name = "esp6-decrypt-tun"
+ tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
+
+ 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 == 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):
""" Ipsec ESP 4 Handoff tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
+ 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 == 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,
TemplateIpsec,
IpsecTun4):
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
def setUp(self):
super(TestIpsec4MultiTunIfEsp, self).setUp()
"""Multiple IPSEC tunnel interfaces """
for p in self.multi_params:
self.verify_tun_44(p, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 127)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 127)
+ self.assertEqual(p.tun_if.get_rx_stats(), 127)
+ self.assertEqual(p.tun_if.get_tx_stats(), 127)
def test_tun_rr_44(self):
""" Round-robin packets acrros multiple interface """
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
def setUp(self):
super(TestIpsec4TunIfEspAll, self).setUp()
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
def setUp(self):
super(TestIpsec4TunIfEspNoAlgo, self).setUp()
self.unconfig_network(p)
+@tag_fixme_vpp_workers
class TestIpsec6MultiTunIfEsp(TemplateIpsec6TunProtect,
TemplateIpsec,
IpsecTun6):
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
- tun6_decrypt_node_name = "esp6-decrypt-tun"
+ tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
def setUp(self):
super(TestIpsec6MultiTunIfEsp, self).setUp()
"""Multiple IPSEC tunnel interfaces """
for p in self.multi_params:
self.verify_tun_66(p, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 127)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 127)
+ self.assertEqual(p.tun_if.get_rx_stats(), 127)
+ self.assertEqual(p.tun_if.get_tx_stats(), 127)
class TestIpsecGreTebIfEsp(TemplateIpsec,
IpsecTun4Tests):
""" Ipsec GRE TEB ESP - TUN tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
omac = "00:11:22:33:44:55"
IpsecTun4Tests):
""" Ipsec GRE TEB ESP - TUN tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
omac = "00:11:22:33:44:55"
IpsecTun4Tests):
""" Ipsec GRE TEB ESP - Tra tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
omac = "00:11:22:33:44:55"
IpsecTun4Tests):
""" Ipsec GRE TEB UDP ESP - Tra tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
omac = "00:11:22:33:44:55"
IpsecTun4Tests):
""" Ipsec GRE ESP - TUN tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
IpsecTun4Tests):
""" Ipsec GRE ESP - TRA tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
dst=self.pg1.remote_ip6)
self.send_and_assert_no_replies(self.tun_if, tx)
node_name = ('/err/%s/unsupported payload' %
- self.tun4_decrypt_node_name)
+ self.tun4_decrypt_node_name[0])
self.assertEqual(1, self.statistics.get_err_counter(node_name))
IpsecTun6Tests):
""" Ipsec GRE ESP - TRA tests """
tun6_encrypt_node_name = "esp6-encrypt-tun"
- tun6_decrypt_node_name = "esp6-decrypt-tun"
+ tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
encryption_type = ESP
def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1,
class TestIpsecMGreIfEspTra4(TemplateIpsec, IpsecTun4):
""" Ipsec mGRE ESP v4 TRA tests """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
class TestIpsecMGreIfEspTra6(TemplateIpsec, IpsecTun6):
""" Ipsec mGRE ESP v6 TRA tests """
tun6_encrypt_node_name = "esp6-encrypt-tun"
- tun6_decrypt_node_name = "esp6-decrypt-tun"
+ tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
encryption_type = ESP
def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1,
self.verify_tun_66(p, count=63)
+@tag_fixme_vpp_workers
class TestIpsec4TunProtect(TemplateIpsec,
TemplateIpsec4TunProtect,
IpsecTun4):
self.config_protect(p)
self.verify_tun_44(p, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 127)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 127)
+ self.assertEqual(p.tun_if.get_rx_stats(), 127)
+ self.assertEqual(p.tun_if.get_tx_stats(), 127)
self.vapi.cli("clear ipsec sa")
self.verify_tun_64(p, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 254)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 254)
+ self.assertEqual(p.tun_if.get_rx_stats(), 254)
+ self.assertEqual(p.tun_if.get_tx_stats(), 254)
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
self.unconfig_sa(p)
self.verify_tun_44(np, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 381)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 381)
+ self.assertEqual(p.tun_if.get_rx_stats(), 381)
+ self.assertEqual(p.tun_if.get_tx_stats(), 381)
# teardown
self.unconfig_protect(np)
self.unconfig_network(p)
+@tag_fixme_vpp_workers
class TestIpsec4TunProtectUdp(TemplateIpsec,
TemplateIpsec4TunProtect,
IpsecTun4):
p = self.ipv4_params
self.verify_tun_44(p, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 127)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 127)
+ self.assertEqual(p.tun_if.get_rx_stats(), 127)
+ self.assertEqual(p.tun_if.get_tx_stats(), 127)
def test_keepalive(self):
""" IPSEC NAT Keepalive """
self.verify_keepalive(self.ipv4_params)
+@tag_fixme_vpp_workers
class TestIpsec4TunProtectTun(TemplateIpsec,
TemplateIpsec4TunProtect,
IpsecTun4):
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
def setUp(self):
super(TestIpsec4TunProtectTun, self).setUp()
self.verify_tun_44(p, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 127)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 127)
+ self.assertEqual(p.tun_if.get_rx_stats(), 127)
+ self.assertEqual(p.tun_if.get_tx_stats(), 127)
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
self.unconfig_sa(p)
self.verify_tun_44(np, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 254)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 254)
+ self.assertEqual(p.tun_if.get_rx_stats(), 254)
+ self.assertEqual(p.tun_if.get_tx_stats(), 254)
# teardown
self.unconfig_protect(np)
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
def setUp(self):
super(TestIpsec4TunProtectTunDrop, self).setUp()
self.unconfig_network(p)
+@tag_fixme_vpp_workers
class TestIpsec6TunProtect(TemplateIpsec,
TemplateIpsec6TunProtect,
IpsecTun6):
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
- tun6_decrypt_node_name = "esp6-decrypt-tun"
+ tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
def setUp(self):
super(TestIpsec6TunProtect, self).setUp()
self.config_protect(p)
self.verify_tun_66(p, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 127)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 127)
+ self.assertEqual(p.tun_if.get_rx_stats(), 127)
+ self.assertEqual(p.tun_if.get_tx_stats(), 127)
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
self.unconfig_sa(p)
self.verify_tun_66(np, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 254)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 254)
+ self.assertEqual(p.tun_if.get_rx_stats(), 254)
+ self.assertEqual(p.tun_if.get_tx_stats(), 254)
# bounce the interface state
p.tun_if.admin_down()
self.verify_tun_66(np3, np3, count=127)
self.verify_drop_tun_66(np, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 127*9)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 127*8)
+ self.assertEqual(p.tun_if.get_rx_stats(), 127*9)
+ self.assertEqual(p.tun_if.get_tx_stats(), 127*8)
self.unconfig_sa(np)
# teardown
self.config_protect(p)
self.verify_tun_46(p, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 127)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 127)
+ self.assertEqual(p.tun_if.get_rx_stats(), 127)
+ self.assertEqual(p.tun_if.get_tx_stats(), 127)
# teardown
self.unconfig_protect(p)
self.unconfig_network(p)
+@tag_fixme_vpp_workers
class TestIpsec6TunProtectTun(TemplateIpsec,
TemplateIpsec6TunProtect,
IpsecTun6):
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
- tun6_decrypt_node_name = "esp6-decrypt-tun"
+ tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
def setUp(self):
super(TestIpsec6TunProtectTun, self).setUp()
self.verify_tun_66(p, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 127)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 127)
+ self.assertEqual(p.tun_if.get_rx_stats(), 127)
+ self.assertEqual(p.tun_if.get_tx_stats(), 127)
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
self.unconfig_sa(p)
self.verify_tun_66(np, count=127)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 254)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 254)
+ self.assertEqual(p.tun_if.get_rx_stats(), 254)
+ self.assertEqual(p.tun_if.get_tx_stats(), 254)
# teardown
self.unconfig_protect(np)
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
- tun6_decrypt_node_name = "esp6-decrypt-tun"
+ tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
def setUp(self):
super(TestIpsec6TunProtectTunDrop, self).setUp()
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
tun4_input_node = "ipsec4-tun-input"
def config_sa_tun(self, p, src, dst):
p.tun_sa_in.remove_vpp_config()
+@tag_fixme_vpp_workers
class TestIpsecItf4(TemplateIpsec,
TemplateIpsecItf4,
IpsecTun4):
self.config_protect(p)
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)
+ self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)
p.tun_if.admin_down()
self.verify_tun_dropped_44(p, count=n_pkts)
p.tun_if.admin_up()
self.verify_tun_44(p, count=n_pkts)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 3*n_pkts)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 2*n_pkts)
+ self.assertEqual(p.tun_if.get_rx_stats(), 3*n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), 2*n_pkts)
# it's a v6 packet when its encrypted
self.tun4_encrypt_node_name = "esp6-encrypt-tun"
self.verify_tun_64(p, count=n_pkts)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 4*n_pkts)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 3*n_pkts)
+ self.assertEqual(p.tun_if.get_rx_stats(), 4*n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), 3*n_pkts)
self.tun4_encrypt_node_name = "esp4-encrypt-tun"
self.unconfig_sa(p)
self.verify_tun_44(np, 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)
+ self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)
# teardown
self.unconfig_protect(np)
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)
+ self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), 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,
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
- tun6_decrypt_node_name = "esp6-decrypt-tun"
+ tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
tun6_input_node = "ipsec6-tun-input"
def config_sa_tun(self, p, src, dst):
p.tun_sa_in.remove_vpp_config()
+@tag_fixme_vpp_workers
class TestIpsecItf6(TemplateIpsec,
TemplateIpsecItf6,
IpsecTun6):
self.config_protect(p)
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)
+ self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)
p.tun_if.admin_down()
self.verify_drop_tun_66(p, count=n_pkts)
p.tun_if.admin_up()
self.verify_tun_66(p, count=n_pkts)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 3*n_pkts)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 2*n_pkts)
+ self.assertEqual(p.tun_if.get_rx_stats(), 3*n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), 2*n_pkts)
# it's a v4 packet when its encrypted
self.tun6_encrypt_node_name = "esp4-encrypt-tun"
self.verify_tun_46(p, count=n_pkts)
- c = p.tun_if.get_rx_stats()
- self.assertEqual(c['packets'], 4*n_pkts)
- c = p.tun_if.get_tx_stats()
- self.assertEqual(c['packets'], 3*n_pkts)
+ self.assertEqual(p.tun_if.get_rx_stats(), 4*n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), 3*n_pkts)
self.tun6_encrypt_node_name = "esp6-encrypt-tun"
self.unconfig_sa(p)
self.verify_tun_66(np, 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)
+ self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)
# teardown
self.unconfig_protect(np)
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)
+ self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
+ self.assertEqual(p.tun_if.get_tx_stats(), 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 """
tun4_encrypt_node_name = "esp4-encrypt-tun"
- tun4_decrypt_node_name = "esp4-decrypt-tun"
+ tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,