from scapy.layers.ipsec import SecurityAssociation, ESP
from scapy.layers.l2 import Ether, GRE, Dot1Q
-from scapy.packet import Raw
+from scapy.packet import Raw, bind_layers
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
from scapy.contrib.mpls import MPLS
from vpp_papi import VppEnum
from vpp_papi_provider import CliFailedCommandError
from vpp_acl import AclRule, VppAcl, VppAclInterface
+from vpp_policer import PolicerAction, VppPolicer, Dir
def config_tun_params(p, encryption_type, tun_if, src=None, dst=None):
p.tun_dst = dst
p.tun_src = src
+ if p.nat_header:
+ is_default_port = (p.nat_header.dport == 4500)
+ else:
+ is_default_port = True
+
+ if is_default_port:
+ outbound_nat_header = p.nat_header
+ else:
+ outbound_nat_header = UDP(sport=p.nat_header.dport,
+ dport=p.nat_header.sport)
+ bind_layers(UDP, ESP, dport=p.nat_header.dport)
+
p.scapy_tun_sa = SecurityAssociation(
encryption_type, spi=p.vpp_tun_spi,
crypt_algo=p.crypt_algo,
tunnel_header=ip_class_by_addr_type[p.addr_type](
src=p.tun_dst,
dst=p.tun_src),
- nat_t_header=p.nat_header,
+ nat_t_header=outbound_nat_header,
esn_en=esn_en)
p.vpp_tun_sa = SecurityAssociation(
encryption_type, spi=p.scapy_tun_spi,
crypt_key = mk_scapy_crypt_key(p)
p.tun_dst = tun_if.remote_ip
p.tun_src = tun_if.local_ip
+
+ if p.nat_header:
+ is_default_port = (p.nat_header.dport == 4500)
+ else:
+ is_default_port = True
+
+ if is_default_port:
+ outbound_nat_header = p.nat_header
+ else:
+ outbound_nat_header = UDP(sport=p.nat_header.dport,
+ dport=p.nat_header.sport)
+ bind_layers(UDP, ESP, dport=p.nat_header.dport)
+
p.scapy_tun_sa = SecurityAssociation(
encryption_type, spi=p.vpp_tun_spi,
crypt_algo=p.crypt_algo,
crypt_key=crypt_key,
auth_algo=p.auth_algo, auth_key=p.auth_key,
esn_en=esn_en,
- nat_t_header=p.nat_header)
+ nat_t_header=outbound_nat_header)
p.vpp_tun_sa = SecurityAssociation(
encryption_type, spi=p.scapy_tun_spi,
crypt_algo=p.crypt_algo,
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, Dir.RX, 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, Dir.RX, 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, Dir.RX, 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, Dir.RX, False)
+ policer.remove_vpp_config()
@tag_fixme_vpp_workers
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()
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"
flags=(p.flags |
VppEnum.vl_api_ipsec_sad_flags_t.
IPSEC_API_SAD_FLAG_IS_INBOUND),
- udp_src=5454,
- udp_dst=4545)
+ udp_src=4545,
+ udp_dst=5454)
p.tun_sa_in.add_vpp_config()
p.tun_if = VppGreInterface(self,
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.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)
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 """
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()
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()
p.tun_protect.update_vpp_config(np3.tun_sa_out,
[np3.tun_sa_in])
self.verify_tun_66(np3, np3, count=127)
- self.verify_drop_tun_66(np, count=127)
+ self.verify_drop_tun_rx_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)
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 = self.ipv4_params
self.config_network(p)
+ config_tun_params(p, self.encryption_type, None,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4)
+ self.verify_tun_dropped_44(p, count=n_pkts)
self.config_sa_tun(p,
self.pg0.local_ip4,
self.pg0.remote_ip4)
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.pg0.remote_ip4)
self.config_protect(p)
+ self.logger.info(self.vapi.cli("sh ipsec sa"))
self.verify_tun_44(p, count=n_pkts)
# teardown
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, Dir.RX, 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, Dir.RX, 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):
def tearDown(self):
super(TestIpsecItf6, self).tearDown()
- def test_tun_44(self):
+ def test_tun_66(self):
"""IPSEC interface IPv6"""
tf = VppEnum.vl_api_tunnel_encap_decap_flags_t
p.tun_flags = tf.TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_HOP_LIMIT
self.config_network(p)
+ config_tun_params(p, self.encryption_type, None,
+ self.pg0.local_ip6,
+ self.pg0.remote_ip6)
+ self.verify_drop_tun_66(p, count=n_pkts)
self.config_sa_tun(p,
self.pg0.local_ip6,
self.pg0.remote_ip6)
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, Dir.RX, 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, Dir.RX, 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,
self.pg0.generate_remote_hosts(N_NHS)
self.pg0.configure_ipv4_neighbors()
+ r_all = AclRule(True,
+ src_prefix="0.0.0.0/0",
+ dst_prefix="0.0.0.0/0",
+ proto=0)
+ a = VppAcl(self, [r_all]).add_vpp_config()
+
+ VppAclInterface(self, self.pg0.sw_if_index, [a]).add_vpp_config()
+ VppAclInterface(self, p.tun_if.sw_if_index, [a]).add_vpp_config()
+
# setup some SAs for several next-hops on the interface
self.multi_params = []
self.pg0.remote_hosts[ii].ip4)
self.multi_params.append(p)
- VppIpRoute(self, p.remote_tun_if_host, 32,
- [VppRoutePath(p.tun_if.remote_hosts[ii].ip4,
- p.tun_if.sw_if_index)]).add_vpp_config()
+ p.via_tun_route = VppIpRoute(
+ self, p.remote_tun_if_host, 32,
+ [VppRoutePath(p.tun_if.remote_hosts[ii].ip4,
+ p.tun_if.sw_if_index)]).add_vpp_config()
p.tun_dst = self.pg0.remote_hosts[ii].ip4
for p in self.multi_params:
self.verify_tun_44(p, count=N_PKTS)
+ # remove one tunnel protect, the rest should still work
+ self.multi_params[0].tun_protect.remove_vpp_config()
+ self.verify_tun_dropped_44(self.multi_params[0], count=N_PKTS)
+ self.multi_params[0].via_tun_route.remove_vpp_config()
+ self.verify_tun_dropped_44(self.multi_params[0], count=N_PKTS)
+
+ for p in self.multi_params[1:]:
+ self.verify_tun_44(p, count=N_PKTS)
+
+ self.multi_params[0].tun_protect.add_vpp_config()
+ self.multi_params[0].via_tun_route.add_vpp_config()
+
+ for p in self.multi_params:
+ self.verify_tun_44(p, count=N_PKTS)
+
class TestIpsecItf6MPLS(TemplateIpsec,
TemplateIpsecItf6,
Code Review / vpp.git / blobdiff