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, \
- IpsecTun6HandoffTests, IpsecTun4HandoffTests, config_tun_params
+from template_ipsec import (
+ TemplateIpsec,
+ IpsecTun4Tests,
+ IpsecTun6Tests,
+ IpsecTun4,
+ IpsecTun6,
+ IpsecTcpTests,
+ mk_scapy_crypt_key,
+ IpsecTun6HandoffTests,
+ IpsecTun4HandoffTests,
+ config_tun_params,
+)
from vpp_gre_interface import VppGreInterface
from vpp_ipip_tun_interface import VppIpIpTunInterface
-from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto, VppMplsLabel, \
- VppMplsTable, VppMplsRoute, FibPathProto
+from vpp_ip_route import (
+ VppIpRoute,
+ VppRoutePath,
+ DpoProto,
+ VppMplsLabel,
+ VppMplsTable,
+ VppMplsRoute,
+ FibPathProto,
+)
from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect, VppIpsecInterface
from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint
def config_tun_params(p, encryption_type, tun_if, src=None, dst=None):
ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
- esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
- IPSEC_API_SAD_FLAG_USE_ESN))
+ esn_en = bool(
+ p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
+ )
crypt_key = mk_scapy_crypt_key(p)
if tun_if:
p.tun_dst = tun_if.remote_ip
p.tun_src = src
if p.nat_header:
- is_default_port = (p.nat_header.dport == 4500)
+ 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)
+ 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,
+ 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,
- tunnel_header=ip_class_by_addr_type[p.addr_type](
- src=p.tun_dst,
- dst=p.tun_src),
+ auth_algo=p.auth_algo,
+ auth_key=p.auth_key,
+ tunnel_header=ip_class_by_addr_type[p.addr_type](src=p.tun_dst, dst=p.tun_src),
nat_t_header=outbound_nat_header,
- esn_en=esn_en)
+ esn_en=esn_en,
+ )
p.vpp_tun_sa = SecurityAssociation(
- encryption_type, spi=p.scapy_tun_spi,
+ encryption_type,
+ spi=p.scapy_tun_spi,
crypt_algo=p.crypt_algo,
crypt_key=crypt_key,
- auth_algo=p.auth_algo, auth_key=p.auth_key,
- tunnel_header=ip_class_by_addr_type[p.addr_type](
- dst=p.tun_dst,
- src=p.tun_src),
+ auth_algo=p.auth_algo,
+ auth_key=p.auth_key,
+ tunnel_header=ip_class_by_addr_type[p.addr_type](dst=p.tun_dst, src=p.tun_src),
nat_t_header=p.nat_header,
- esn_en=esn_en)
+ esn_en=esn_en,
+ )
def config_tra_params(p, encryption_type, tun_if):
ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
- esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
- IPSEC_API_SAD_FLAG_USE_ESN))
+ esn_en = bool(
+ p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
+ )
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)
+ 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)
+ 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,
+ 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,
+ auth_algo=p.auth_algo,
+ auth_key=p.auth_key,
esn_en=esn_en,
- nat_t_header=outbound_nat_header)
+ nat_t_header=outbound_nat_header,
+ )
p.vpp_tun_sa = SecurityAssociation(
- encryption_type, spi=p.scapy_tun_spi,
+ encryption_type,
+ spi=p.scapy_tun_spi,
crypt_algo=p.crypt_algo,
crypt_key=crypt_key,
- auth_algo=p.auth_algo, auth_key=p.auth_key,
+ auth_algo=p.auth_algo,
+ auth_key=p.auth_key,
esn_en=esn_en,
- nat_t_header=p.nat_header)
+ nat_t_header=p.nat_header,
+ )
class TemplateIpsec4TunProtect(object):
- """ IPsec IPv4 Tunnel protect """
+ """IPsec IPv4 Tunnel protect"""
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
def config_sa_tra(self, p):
config_tun_params(p, self.encryption_type, p.tun_if)
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- flags=p.flags)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ flags=p.flags,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- flags=p.flags)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ flags=p.flags,
+ )
p.tun_sa_in.add_vpp_config()
def config_sa_tun(self, p):
config_tun_params(p, self.encryption_type, p.tun_if)
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.tun_if.local_addr[p.addr_type],
- self.tun_if.remote_addr[p.addr_type],
- flags=p.flags)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.tun_if.local_addr[p.addr_type],
+ self.tun_if.remote_addr[p.addr_type],
+ flags=p.flags,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.tun_if.remote_addr[p.addr_type],
- self.tun_if.local_addr[p.addr_type],
- flags=p.flags)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.tun_if.remote_addr[p.addr_type],
+ self.tun_if.local_addr[p.addr_type],
+ flags=p.flags,
+ )
p.tun_sa_in.add_vpp_config()
def config_protect(self, p):
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
def config_network(self, p):
- if hasattr(p, 'tun_dst'):
+ if hasattr(p, "tun_dst"):
tun_dst = p.tun_dst
else:
tun_dst = self.pg0.remote_ip4
- p.tun_if = VppIpIpTunInterface(self, self.pg0,
- self.pg0.local_ip4,
- tun_dst)
+ p.tun_if = VppIpIpTunInterface(self, self.pg0, self.pg0.local_ip4, tun_dst)
p.tun_if.add_vpp_config()
p.tun_if.admin_up()
p.tun_if.config_ip4()
p.tun_if.config_ip6()
- p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
- [VppRoutePath(p.tun_if.remote_ip4,
- 0xffffffff)])
+ p.route = VppIpRoute(
+ self,
+ p.remote_tun_if_host,
+ 32,
+ [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)],
+ )
p.route.add_vpp_config()
- r = VppIpRoute(self, p.remote_tun_if_host6, 128,
- [VppRoutePath(p.tun_if.remote_ip6,
- 0xffffffff,
- proto=DpoProto.DPO_PROTO_IP6)])
+ r = VppIpRoute(
+ self,
+ p.remote_tun_if_host6,
+ 128,
+ [
+ VppRoutePath(
+ p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
+ )
+ ],
+ )
r.add_vpp_config()
def unconfig_network(self, p):
p.tun_sa_in.remove_vpp_config()
-class TemplateIpsec4TunIfEsp(TemplateIpsec4TunProtect,
- TemplateIpsec):
- """ IPsec tunnel interface tests """
+class TemplateIpsec4TunIfEsp(TemplateIpsec4TunProtect, TemplateIpsec):
+ """IPsec tunnel interface tests"""
encryption_type = ESP
super(TemplateIpsec4TunIfEsp, self).tearDown()
-class TemplateIpsec4TunIfEspUdp(TemplateIpsec4TunProtect,
- TemplateIpsec):
- """ IPsec UDP tunnel interface tests """
+class TemplateIpsec4TunIfEspUdp(TemplateIpsec4TunProtect, TemplateIpsec):
+ """IPsec UDP tunnel interface tests"""
tun4_encrypt_node_name = "esp4-encrypt-tun"
tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
def config_sa_tra(self, p):
config_tun_params(p, self.encryption_type, p.tun_if)
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- flags=p.flags,
- udp_src=p.nat_header.sport,
- udp_dst=p.nat_header.dport)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ flags=p.flags,
+ udp_src=p.nat_header.sport,
+ udp_dst=p.nat_header.dport,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- flags=p.flags,
- udp_src=p.nat_header.sport,
- udp_dst=p.nat_header.dport)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ flags=p.flags,
+ udp_src=p.nat_header.sport,
+ udp_dst=p.nat_header.dport,
+ )
p.tun_sa_in.add_vpp_config()
def setUp(self):
super(TemplateIpsec4TunIfEspUdp, self).setUp()
p = self.ipv4_params
- p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
- IPSEC_API_SAD_FLAG_UDP_ENCAP)
+ p.flags = VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_UDP_ENCAP
p.nat_header = UDP(sport=5454, dport=4500)
self.tun_if = self.pg0
class TestIpsec4TunIfEsp1(TemplateIpsec4TunIfEsp, IpsecTun4Tests):
- """ Ipsec ESP - TUN tests """
+ """Ipsec ESP - TUN tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-tun"
tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
def test_tun_basic64(self):
- """ ipsec 6o4 tunnel basic test """
+ """ipsec 6o4 tunnel basic test"""
self.tun4_encrypt_node_name = "esp4-encrypt-tun"
self.verify_tun_64(self.params[socket.AF_INET], count=1)
def test_tun_burst64(self):
- """ ipsec 6o4 tunnel basic test """
+ """ipsec 6o4 tunnel basic test"""
self.tun4_encrypt_node_name = "esp4-encrypt-tun"
self.verify_tun_64(self.params[socket.AF_INET], count=257)
def test_tun_basic_frag44(self):
- """ ipsec 4o4 tunnel frag basic test """
+ """ipsec 4o4 tunnel frag basic test"""
self.tun4_encrypt_node_name = "esp4-encrypt-tun"
p = self.ipv4_params
- self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
- [1500, 0, 0, 0])
- self.verify_tun_44(self.params[socket.AF_INET],
- count=1, payload_size=1800, n_rx=2)
- self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
- [9000, 0, 0, 0])
+ self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index, [1500, 0, 0, 0])
+ self.verify_tun_44(
+ self.params[socket.AF_INET], count=1, payload_size=1800, n_rx=2
+ )
+ self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index, [9000, 0, 0, 0])
class TestIpsec4TunIfEspUdp(TemplateIpsec4TunIfEspUdp, IpsecTun4Tests):
- """ Ipsec ESP UDP tests """
+ """Ipsec ESP UDP tests"""
tun4_input_node = "ipsec4-tun-input"
super(TestIpsec4TunIfEspUdp, self).setUp()
def test_keepalive(self):
- """ IPSEC NAT Keepalive """
+ """IPSEC NAT Keepalive"""
self.verify_keepalive(self.ipv4_params)
class TestIpsec4TunIfEspUdpGCM(TemplateIpsec4TunIfEspUdp, IpsecTun4Tests):
- """ Ipsec ESP UDP GCM tests """
+ """Ipsec ESP UDP GCM tests"""
tun4_input_node = "ipsec4-tun-input"
def setUp(self):
super(TestIpsec4TunIfEspUdpGCM, self).setUp()
p = self.ipv4_params
- p.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_NONE)
- p.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_AES_GCM_256)
+ p.auth_algo_vpp_id = VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
+ p.crypt_algo_vpp_id = (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_GCM_256
+ )
p.crypt_algo = "AES-GCM"
p.auth_algo = "NULL"
p.crypt_key = b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"
class TestIpsec4TunIfEsp2(TemplateIpsec4TunIfEsp, IpsecTcpTests):
- """ Ipsec ESP - TCP tests """
+ """Ipsec ESP - TCP tests"""
+
pass
class TemplateIpsec6TunProtect(object):
- """ IPsec IPv6 Tunnel protect """
+ """IPsec IPv6 Tunnel protect"""
def config_sa_tra(self, p):
config_tun_params(p, self.encryption_type, p.tun_if)
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_in.add_vpp_config()
def config_sa_tun(self, p):
config_tun_params(p, self.encryption_type, p.tun_if)
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.tun_if.local_addr[p.addr_type],
- self.tun_if.remote_addr[p.addr_type])
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.tun_if.local_addr[p.addr_type],
+ self.tun_if.remote_addr[p.addr_type],
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.tun_if.remote_addr[p.addr_type],
- self.tun_if.local_addr[p.addr_type])
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.tun_if.remote_addr[p.addr_type],
+ self.tun_if.local_addr[p.addr_type],
+ )
p.tun_sa_in.add_vpp_config()
def config_protect(self, p):
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
def config_network(self, p):
- if hasattr(p, 'tun_dst'):
+ if hasattr(p, "tun_dst"):
tun_dst = p.tun_dst
else:
tun_dst = self.pg0.remote_ip6
- p.tun_if = VppIpIpTunInterface(self, self.pg0,
- self.pg0.local_ip6,
- tun_dst)
+ p.tun_if = VppIpIpTunInterface(self, self.pg0, self.pg0.local_ip6, tun_dst)
p.tun_if.add_vpp_config()
p.tun_if.admin_up()
p.tun_if.config_ip6()
p.tun_if.config_ip4()
- p.route = VppIpRoute(self, p.remote_tun_if_host, 128,
- [VppRoutePath(p.tun_if.remote_ip6,
- 0xffffffff,
- proto=DpoProto.DPO_PROTO_IP6)])
+ p.route = VppIpRoute(
+ self,
+ p.remote_tun_if_host,
+ 128,
+ [
+ VppRoutePath(
+ p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
+ )
+ ],
+ )
p.route.add_vpp_config()
- r = VppIpRoute(self, p.remote_tun_if_host4, 32,
- [VppRoutePath(p.tun_if.remote_ip4,
- 0xffffffff)])
+ r = VppIpRoute(
+ self,
+ p.remote_tun_if_host4,
+ 32,
+ [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)],
+ )
r.add_vpp_config()
def unconfig_network(self, p):
p.tun_sa_in.remove_vpp_config()
-class TemplateIpsec6TunIfEsp(TemplateIpsec6TunProtect,
- TemplateIpsec):
- """ IPsec tunnel interface tests """
+class TemplateIpsec6TunIfEsp(TemplateIpsec6TunProtect, TemplateIpsec):
+ """IPsec tunnel interface tests"""
encryption_type = ESP
super(TemplateIpsec6TunIfEsp, self).tearDown()
-class TestIpsec6TunIfEsp1(TemplateIpsec6TunIfEsp,
- IpsecTun6Tests):
- """ Ipsec ESP - TUN tests """
+class TestIpsec6TunIfEsp1(TemplateIpsec6TunIfEsp, IpsecTun6Tests):
+ """Ipsec ESP - TUN tests"""
+
tun6_encrypt_node_name = "esp6-encrypt-tun"
tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
def test_tun_basic46(self):
- """ ipsec 4o6 tunnel basic test """
+ """ipsec 4o6 tunnel basic test"""
self.tun6_encrypt_node_name = "esp6-encrypt-tun"
self.verify_tun_46(self.params[socket.AF_INET6], count=1)
def test_tun_burst46(self):
- """ ipsec 4o6 tunnel burst test """
+ """ipsec 4o6 tunnel burst test"""
self.tun6_encrypt_node_name = "esp6-encrypt-tun"
self.verify_tun_46(self.params[socket.AF_INET6], count=257)
-class TestIpsec6TunIfEspHandoff(TemplateIpsec6TunIfEsp,
- IpsecTun6HandoffTests):
- """ Ipsec ESP 6 Handoff tests """
+class TestIpsec6TunIfEspHandoff(TemplateIpsec6TunIfEsp, IpsecTun6HandoffTests):
+ """Ipsec ESP 6 Handoff tests"""
+
tun6_encrypt_node_name = "esp6-encrypt-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 """
+ """ESP 6o6 tunnel with policer worker hand-off test"""
self.vapi.cli("clear errors")
self.vapi.cli("clear ipsec sa")
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)
+ 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
# 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)
+ 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"))
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)
+ 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(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.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'])
+ 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 """
+class TestIpsec4TunIfEspHandoff(TemplateIpsec4TunIfEsp, IpsecTun4HandoffTests):
+ """Ipsec ESP 4 Handoff tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-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 """
+ """ESP 4o4 tunnel with policer worker hand-off test"""
self.vapi.cli("clear errors")
self.vapi.cli("clear ipsec sa")
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)
+ 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
# 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)
+ 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"))
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)
+ 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(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.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'])
+ 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
-class TestIpsec4MultiTunIfEsp(TemplateIpsec4TunProtect,
- TemplateIpsec,
- IpsecTun4):
- """ IPsec IPv4 Multi Tunnel interface """
+class TestIpsec4MultiTunIfEsp(TemplateIpsec4TunProtect, TemplateIpsec, IpsecTun4):
+ """IPsec IPv4 Multi Tunnel interface"""
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
super(TestIpsec4MultiTunIfEsp, self).tearDown()
def test_tun_44(self):
- """Multiple IPSEC tunnel interfaces """
+ """Multiple IPSEC tunnel interfaces"""
for p in self.multi_params:
self.verify_tun_44(p, count=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 """
+ """Round-robin packets acrros multiple interface"""
tx = []
for p in self.multi_params:
- tx = tx + self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
- src=p.remote_tun_if_host,
- dst=self.pg1.remote_ip4)
+ tx = tx + self.gen_encrypt_pkts(
+ p,
+ p.scapy_tun_sa,
+ self.tun_if,
+ src=p.remote_tun_if_host,
+ dst=self.pg1.remote_ip4,
+ )
rxs = self.send_and_expect(self.tun_if, tx, self.pg1)
for rx, p in zip(rxs, self.multi_params):
tx = []
for p in self.multi_params:
- tx = tx + self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
- dst=p.remote_tun_if_host)
+ tx = tx + self.gen_pkts(
+ self.pg1, src=self.pg1.remote_ip4, dst=p.remote_tun_if_host
+ )
rxs = self.send_and_expect(self.pg1, tx, self.tun_if)
for rx, p in zip(rxs, self.multi_params):
self.verify_encrypted(p, p.vpp_tun_sa, [rx])
-class TestIpsec4TunIfEspAll(TemplateIpsec4TunProtect,
- TemplateIpsec,
- IpsecTun4):
- """ IPsec IPv4 Tunnel interface all Algos """
+class TestIpsec4TunIfEspAll(TemplateIpsec4TunProtect, TemplateIpsec, IpsecTun4):
+ """IPsec IPv4 Tunnel interface all Algos"""
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
# change the key and the SPI
#
np = copy.copy(p)
- p.crypt_key = b'X' + p.crypt_key[1:]
+ p.crypt_key = b"X" + p.crypt_key[1:]
p.scapy_tun_spi += 1
p.scapy_tun_sa_id += 1
p.vpp_tun_spi += 1
config_tun_params(p, self.encryption_type, p.tun_if)
- p.tun_sa_out = VppIpsecSA(self,
- p.scapy_tun_sa_id,
- p.scapy_tun_spi,
- p.auth_algo_vpp_id,
- p.auth_key,
- p.crypt_algo_vpp_id,
- p.crypt_key,
- self.vpp_esp_protocol,
- flags=p.flags,
- salt=p.salt)
- p.tun_sa_in = VppIpsecSA(self,
- p.vpp_tun_sa_id,
- p.vpp_tun_spi,
- p.auth_algo_vpp_id,
- p.auth_key,
- p.crypt_algo_vpp_id,
- p.crypt_key,
- self.vpp_esp_protocol,
- flags=p.flags,
- salt=p.salt)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ flags=p.flags,
+ salt=p.salt,
+ )
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ flags=p.flags,
+ salt=p.salt,
+ )
p.tun_sa_in.add_vpp_config()
p.tun_sa_out.add_vpp_config()
self.logger.info(self.vapi.cli("sh ipsec sa"))
def test_tun_44(self):
- """IPSEC tunnel all algos """
+ """IPSEC tunnel all algos"""
# foreach VPP crypto engine
engines = ["ia32", "ipsecmb", "openssl"]
# foreach crypto algorithm
- algos = [{'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_AES_GCM_128),
- 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_NONE),
- 'scapy-crypto': "AES-GCM",
- 'scapy-integ': "NULL",
- 'key': b"JPjyOWBeVEQiMe7h",
- 'salt': 3333},
- {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_AES_GCM_192),
- 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_NONE),
- 'scapy-crypto': "AES-GCM",
- 'scapy-integ': "NULL",
- 'key': b"JPjyOWBeVEQiMe7hJPjyOWBe",
- 'salt': 0},
- {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_AES_GCM_256),
- 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_NONE),
- 'scapy-crypto': "AES-GCM",
- 'scapy-integ': "NULL",
- 'key': b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
- 'salt': 9999},
- {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_AES_CBC_128),
- 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_SHA1_96),
- 'scapy-crypto': "AES-CBC",
- 'scapy-integ': "HMAC-SHA1-96",
- 'salt': 0,
- 'key': b"JPjyOWBeVEQiMe7h"},
- {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_AES_CBC_192),
- 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_SHA_512_256),
- 'scapy-crypto': "AES-CBC",
- 'scapy-integ': "SHA2-512-256",
- 'salt': 0,
- 'key': b"JPjyOWBeVEQiMe7hJPjyOWBe"},
- {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_AES_CBC_256),
- 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_SHA_256_128),
- 'scapy-crypto': "AES-CBC",
- 'scapy-integ': "SHA2-256-128",
- 'salt': 0,
- 'key': b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"},
- {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_NONE),
- 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_SHA1_96),
- 'scapy-crypto': "NULL",
- 'scapy-integ': "HMAC-SHA1-96",
- 'salt': 0,
- 'key': b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}]
+ algos = [
+ {
+ "vpp-crypto": (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_GCM_128
+ ),
+ "vpp-integ": (
+ VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
+ ),
+ "scapy-crypto": "AES-GCM",
+ "scapy-integ": "NULL",
+ "key": b"JPjyOWBeVEQiMe7h",
+ "salt": 3333,
+ },
+ {
+ "vpp-crypto": (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_GCM_192
+ ),
+ "vpp-integ": (
+ VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
+ ),
+ "scapy-crypto": "AES-GCM",
+ "scapy-integ": "NULL",
+ "key": b"JPjyOWBeVEQiMe7hJPjyOWBe",
+ "salt": 0,
+ },
+ {
+ "vpp-crypto": (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_GCM_256
+ ),
+ "vpp-integ": (
+ VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
+ ),
+ "scapy-crypto": "AES-GCM",
+ "scapy-integ": "NULL",
+ "key": b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
+ "salt": 9999,
+ },
+ {
+ "vpp-crypto": (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_128
+ ),
+ "vpp-integ": (
+ VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA1_96
+ ),
+ "scapy-crypto": "AES-CBC",
+ "scapy-integ": "HMAC-SHA1-96",
+ "salt": 0,
+ "key": b"JPjyOWBeVEQiMe7h",
+ },
+ {
+ "vpp-crypto": (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_192
+ ),
+ "vpp-integ": (
+ VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA_512_256
+ ),
+ "scapy-crypto": "AES-CBC",
+ "scapy-integ": "SHA2-512-256",
+ "salt": 0,
+ "key": b"JPjyOWBeVEQiMe7hJPjyOWBe",
+ },
+ {
+ "vpp-crypto": (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_256
+ ),
+ "vpp-integ": (
+ VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA_256_128
+ ),
+ "scapy-crypto": "AES-CBC",
+ "scapy-integ": "SHA2-256-128",
+ "salt": 0,
+ "key": b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
+ },
+ {
+ "vpp-crypto": (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_NONE
+ ),
+ "vpp-integ": (
+ VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA1_96
+ ),
+ "scapy-crypto": "NULL",
+ "scapy-integ": "HMAC-SHA1-96",
+ "salt": 0,
+ "key": b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
+ },
+ ]
for engine in engines:
self.vapi.cli("set crypto handler all %s" % engine)
# with self.subTest(algo=algo['scapy']):
p = self.ipv4_params
- p.auth_algo_vpp_id = algo['vpp-integ']
- p.crypt_algo_vpp_id = algo['vpp-crypto']
- p.crypt_algo = algo['scapy-crypto']
- p.auth_algo = algo['scapy-integ']
- p.crypt_key = algo['key']
- p.salt = algo['salt']
+ p.auth_algo_vpp_id = algo["vpp-integ"]
+ p.crypt_algo_vpp_id = algo["vpp-crypto"]
+ p.crypt_algo = algo["scapy-crypto"]
+ p.auth_algo = algo["scapy-integ"]
+ p.crypt_key = algo["key"]
+ p.salt = algo["salt"]
#
# rekey the tunnel
self.verify_tun_44(p, count=127)
-class TestIpsec4TunIfEspNoAlgo(TemplateIpsec4TunProtect,
- TemplateIpsec,
- IpsecTun4):
- """ IPsec IPv4 Tunnel interface no Algos """
+class TestIpsec4TunIfEspNoAlgo(TemplateIpsec4TunProtect, TemplateIpsec, IpsecTun4):
+ """IPsec IPv4 Tunnel interface no Algos"""
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
self.tun_if = self.pg0
p = self.ipv4_params
- p.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_NONE)
- p.auth_algo = 'NULL'
+ p.auth_algo_vpp_id = VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
+ p.auth_algo = "NULL"
p.auth_key = []
- p.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_NONE)
- p.crypt_algo = 'NULL'
+ p.crypt_algo_vpp_id = (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_NONE
+ )
+ p.crypt_algo = "NULL"
p.crypt_key = []
def tearDown(self):
super(TestIpsec4TunIfEspNoAlgo, self).tearDown()
def test_tun_44(self):
- """ IPSec SA with NULL algos """
+ """IPSec SA with NULL algos"""
p = self.ipv4_params
self.config_network(p)
self.config_sa_tra(p)
self.config_protect(p)
- tx = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
- dst=p.remote_tun_if_host)
+ tx = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4, dst=p.remote_tun_if_host)
self.send_and_assert_no_replies(self.pg1, tx)
self.unconfig_protect(p)
@tag_fixme_vpp_workers
-class TestIpsec6MultiTunIfEsp(TemplateIpsec6TunProtect,
- TemplateIpsec,
- IpsecTun6):
- """ IPsec IPv6 Multi Tunnel interface """
+class TestIpsec6MultiTunIfEsp(TemplateIpsec6TunProtect, TemplateIpsec, IpsecTun6):
+ """IPsec IPv6 Multi Tunnel interface"""
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
super(TestIpsec6MultiTunIfEsp, self).tearDown()
def test_tun_66(self):
- """Multiple IPSEC tunnel interfaces """
+ """Multiple IPSEC tunnel interfaces"""
for p in self.multi_params:
self.verify_tun_66(p, count=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 """
+class TestIpsecGreTebIfEsp(TemplateIpsec, IpsecTun4Tests):
+ """Ipsec GRE TEB ESP - TUN tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-tun"
tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
omac = "00:11:22:33:44:55"
- def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=self.pg0.remote_ip4,
- dst=self.pg0.local_ip4) /
- GRE() /
- Ether(dst=self.omac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts(self, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(dst=self.omac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / GRE()
+ / Ether(dst=self.omac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(dst=self.omac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted(self, p, rxs):
for rx in rxs:
bd1 = VppBridgeDomain(self, 1)
bd1.add_vpp_config()
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.pg0.remote_ip4,
- self.pg0.local_ip4)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.pg0.remote_ip4,
+ self.pg0.local_ip4,
+ )
p.tun_sa_in.add_vpp_config()
- p.tun_if = VppGreInterface(self,
- self.pg0.local_ip4,
- self.pg0.remote_ip4,
- type=(VppEnum.vl_api_gre_tunnel_type_t.
- GRE_API_TUNNEL_TYPE_TEB))
+ p.tun_if = VppGreInterface(
+ self,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4,
+ type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB),
+ )
p.tun_if.add_vpp_config()
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
super(TestIpsecGreTebIfEsp, self).tearDown()
-class TestIpsecGreTebVlanIfEsp(TemplateIpsec,
- IpsecTun4Tests):
- """ Ipsec GRE TEB ESP - TUN tests """
+class TestIpsecGreTebVlanIfEsp(TemplateIpsec, IpsecTun4Tests):
+ """Ipsec GRE TEB ESP - TUN tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-tun"
tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
omac = "00:11:22:33:44:55"
- def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=self.pg0.remote_ip4,
- dst=self.pg0.local_ip4) /
- GRE() /
- Ether(dst=self.omac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts(self, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(dst=self.omac) /
- Dot1Q(vlan=11) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / GRE()
+ / Ether(dst=self.omac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(dst=self.omac)
+ / Dot1Q(vlan=11)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted(self, p, rxs):
for rx in rxs:
self.pg1_11 = VppDot1QSubint(self, self.pg1, 11)
self.vapi.l2_interface_vlan_tag_rewrite(
- sw_if_index=self.pg1_11.sw_if_index, vtr_op=L2_VTR_OP.L2_POP_1,
- push_dot1q=11)
+ sw_if_index=self.pg1_11.sw_if_index,
+ vtr_op=L2_VTR_OP.L2_POP_1,
+ push_dot1q=11,
+ )
self.pg1_11.admin_up()
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.pg0.remote_ip4,
- self.pg0.local_ip4)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.pg0.remote_ip4,
+ self.pg0.local_ip4,
+ )
p.tun_sa_in.add_vpp_config()
- p.tun_if = VppGreInterface(self,
- self.pg0.local_ip4,
- self.pg0.remote_ip4,
- type=(VppEnum.vl_api_gre_tunnel_type_t.
- GRE_API_TUNNEL_TYPE_TEB))
+ p.tun_if = VppGreInterface(
+ self,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4,
+ type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB),
+ )
p.tun_if.add_vpp_config()
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
self.pg1_11.remove_vpp_config()
-class TestIpsecGreTebIfEspTra(TemplateIpsec,
- IpsecTun4Tests):
- """ Ipsec GRE TEB ESP - Tra tests """
+class TestIpsecGreTebIfEspTra(TemplateIpsec, IpsecTun4Tests):
+ """Ipsec GRE TEB ESP - Tra tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-tun"
tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
omac = "00:11:22:33:44:55"
- def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=self.pg0.remote_ip4,
- dst=self.pg0.local_ip4) /
- GRE() /
- Ether(dst=self.omac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts(self, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(dst=self.omac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / GRE()
+ / Ether(dst=self.omac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(dst=self.omac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted(self, p, rxs):
for rx in rxs:
bd1 = VppBridgeDomain(self, 1)
bd1.add_vpp_config()
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_in.add_vpp_config()
- p.tun_if = VppGreInterface(self,
- self.pg0.local_ip4,
- self.pg0.remote_ip4,
- type=(VppEnum.vl_api_gre_tunnel_type_t.
- GRE_API_TUNNEL_TYPE_TEB))
+ p.tun_if = VppGreInterface(
+ self,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4,
+ type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB),
+ )
p.tun_if.add_vpp_config()
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
super(TestIpsecGreTebIfEspTra, self).tearDown()
-class TestIpsecGreTebUdpIfEspTra(TemplateIpsec,
- IpsecTun4Tests):
- """ Ipsec GRE TEB UDP ESP - Tra tests """
+class TestIpsecGreTebUdpIfEspTra(TemplateIpsec, IpsecTun4Tests):
+ """Ipsec GRE TEB UDP ESP - Tra tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-tun"
tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
encryption_type = ESP
omac = "00:11:22:33:44:55"
- def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=self.pg0.remote_ip4,
- dst=self.pg0.local_ip4) /
- GRE() /
- Ether(dst=self.omac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts(self, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(dst=self.omac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / GRE()
+ / Ether(dst=self.omac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(dst=self.omac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted(self, p, rxs):
for rx in rxs:
p = self.ipv4_params
p = self.ipv4_params
- p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
- IPSEC_API_SAD_FLAG_UDP_ENCAP)
+ p.flags = VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_UDP_ENCAP
p.nat_header = UDP(sport=5454, dport=4545)
bd1 = VppBridgeDomain(self, 1)
bd1.add_vpp_config()
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- flags=p.flags,
- udp_src=5454,
- udp_dst=4545)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ flags=p.flags,
+ udp_src=5454,
+ udp_dst=4545,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- flags=(p.flags |
- VppEnum.vl_api_ipsec_sad_flags_t.
- IPSEC_API_SAD_FLAG_IS_INBOUND),
- udp_src=4545,
- udp_dst=5454)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ flags=(
+ p.flags | VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_IS_INBOUND
+ ),
+ udp_src=4545,
+ udp_dst=5454,
+ )
p.tun_sa_in.add_vpp_config()
- p.tun_if = VppGreInterface(self,
- self.pg0.local_ip4,
- self.pg0.remote_ip4,
- type=(VppEnum.vl_api_gre_tunnel_type_t.
- GRE_API_TUNNEL_TYPE_TEB))
+ p.tun_if = VppGreInterface(
+ self,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4,
+ type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB),
+ )
p.tun_if.add_vpp_config()
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
super(TestIpsecGreTebUdpIfEspTra, self).tearDown()
-class TestIpsecGreIfEsp(TemplateIpsec,
- IpsecTun4Tests):
- """ Ipsec GRE ESP - TUN tests """
+class TestIpsecGreIfEsp(TemplateIpsec, IpsecTun4Tests):
+ """Ipsec GRE ESP - TUN tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-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,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=self.pg0.remote_ip4,
- dst=self.pg0.local_ip4) /
- GRE() /
- IP(src=self.pg1.local_ip4,
- dst=self.pg1.remote_ip4) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts(self, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / GRE()
+ / IP(src=self.pg1.local_ip4, dst=self.pg1.remote_ip4)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted(self, p, rxs):
for rx in rxs:
bd1 = VppBridgeDomain(self, 1)
bd1.add_vpp_config()
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- self.pg0.remote_ip4,
- self.pg0.local_ip4)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ self.pg0.remote_ip4,
+ self.pg0.local_ip4,
+ )
p.tun_sa_in.add_vpp_config()
- p.tun_if = VppGreInterface(self,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ p.tun_if = VppGreInterface(self, self.pg0.local_ip4, self.pg0.remote_ip4)
p.tun_if.add_vpp_config()
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
p.tun_if.admin_up()
p.tun_if.config_ip4()
config_tun_params(p, self.encryption_type, p.tun_if)
- VppIpRoute(self, "1.1.1.2", 32,
- [VppRoutePath(p.tun_if.remote_ip4,
- 0xffffffff)]).add_vpp_config()
+ VppIpRoute(
+ self, "1.1.1.2", 32, [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)]
+ ).add_vpp_config()
def tearDown(self):
p = self.ipv4_params
super(TestIpsecGreIfEsp, self).tearDown()
-class TestIpsecGreIfEspTra(TemplateIpsec,
- IpsecTun4Tests):
- """ Ipsec GRE ESP - TRA tests """
+class TestIpsecGreIfEspTra(TemplateIpsec, IpsecTun4Tests):
+ """Ipsec GRE ESP - TRA tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-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,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=self.pg0.remote_ip4,
- dst=self.pg0.local_ip4) /
- GRE() /
- IP(src=self.pg1.local_ip4,
- dst=self.pg1.remote_ip4) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_encrypt_non_ip_pkts(self, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=self.pg0.remote_ip4,
- dst=self.pg0.local_ip4) /
- GRE() /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts(self, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / GRE()
+ / IP(src=self.pg1.local_ip4, dst=self.pg1.remote_ip4)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_encrypt_non_ip_pkts(self, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / GRE()
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted(self, p, rxs):
for rx in rxs:
p = self.ipv4_params
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_in.add_vpp_config()
- p.tun_if = VppGreInterface(self,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ p.tun_if = VppGreInterface(self, self.pg0.local_ip4, self.pg0.remote_ip4)
p.tun_if.add_vpp_config()
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
p.tun_if.admin_up()
p.tun_if.config_ip4()
config_tra_params(p, self.encryption_type, p.tun_if)
- VppIpRoute(self, "1.1.1.2", 32,
- [VppRoutePath(p.tun_if.remote_ip4,
- 0xffffffff)]).add_vpp_config()
+ VppIpRoute(
+ self, "1.1.1.2", 32, [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)]
+ ).add_vpp_config()
def tearDown(self):
p = self.ipv4_params
def test_gre_non_ip(self):
p = self.ipv4_params
- tx = self.gen_encrypt_non_ip_pkts(p.scapy_tun_sa, self.tun_if,
- src=p.remote_tun_if_host,
- dst=self.pg1.remote_ip6)
+ tx = self.gen_encrypt_non_ip_pkts(
+ p.scapy_tun_sa,
+ self.tun_if,
+ src=p.remote_tun_if_host,
+ 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[0])
+ node_name = "/err/%s/unsupported payload" % self.tun4_decrypt_node_name[0]
self.assertEqual(1, self.statistics.get_err_counter(node_name))
-class TestIpsecGre6IfEspTra(TemplateIpsec,
- IpsecTun6Tests):
- """ Ipsec GRE ESP - TRA tests """
+class TestIpsecGre6IfEspTra(TemplateIpsec, IpsecTun6Tests):
+ """Ipsec GRE ESP - TRA tests"""
+
tun6_encrypt_node_name = "esp6-encrypt-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,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IPv6(src=self.pg0.remote_ip6,
- dst=self.pg0.local_ip6) /
- GRE() /
- IPv6(src=self.pg1.local_ip6,
- dst=self.pg1.remote_ip6) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts6(self, p, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- IPv6(src="1::1", dst="1::2") /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6)
+ / GRE()
+ / IPv6(src=self.pg1.local_ip6, dst=self.pg1.remote_ip6)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IPv6(src="1::1", dst="1::2")
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted6(self, p, rxs):
for rx in rxs:
bd1 = VppBridgeDomain(self, 1)
bd1.add_vpp_config()
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_in.add_vpp_config()
- p.tun_if = VppGreInterface(self,
- self.pg0.local_ip6,
- self.pg0.remote_ip6)
+ p.tun_if = VppGreInterface(self, self.pg0.local_ip6, self.pg0.remote_ip6)
p.tun_if.add_vpp_config()
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
p.tun_if.admin_up()
p.tun_if.config_ip6()
config_tra_params(p, self.encryption_type, p.tun_if)
- r = VppIpRoute(self, "1::2", 128,
- [VppRoutePath(p.tun_if.remote_ip6,
- 0xffffffff,
- proto=DpoProto.DPO_PROTO_IP6)])
+ r = VppIpRoute(
+ self,
+ "1::2",
+ 128,
+ [
+ VppRoutePath(
+ p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
+ )
+ ],
+ )
r.add_vpp_config()
def tearDown(self):
class TestIpsecMGreIfEspTra4(TemplateIpsec, IpsecTun4):
- """ Ipsec mGRE ESP v4 TRA tests """
+ """Ipsec mGRE ESP v4 TRA tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-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,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=p.tun_dst,
- dst=self.pg0.local_ip4) /
- GRE() /
- IP(src=self.pg1.local_ip4,
- dst=self.pg1.remote_ip4) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts(self, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- IP(src="1.1.1.1", dst=dst) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=p.tun_dst, dst=self.pg0.local_ip4)
+ / GRE()
+ / IP(src=self.pg1.local_ip4, dst=self.pg1.remote_ip4)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IP(src="1.1.1.1", dst=dst)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted(self, p, rxs):
for rx in rxs:
N_NHS = 16
self.tun_if = self.pg0
p = self.ipv4_params
- p.tun_if = VppGreInterface(self,
- self.pg0.local_ip4,
- "0.0.0.0",
- mode=(VppEnum.vl_api_tunnel_mode_t.
- TUNNEL_API_MODE_MP))
+ p.tun_if = VppGreInterface(
+ self,
+ self.pg0.local_ip4,
+ "0.0.0.0",
+ mode=(VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_MP),
+ )
p.tun_if.add_vpp_config()
p.tun_if.admin_up()
p.tun_if.config_ip4()
p.scapy_tra_spi = p.scapy_tra_spi + ii
p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
p.vpp_tra_spi = p.vpp_tra_spi + ii
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_in.add_vpp_config()
p.tun_protect = VppIpsecTunProtect(
p.tun_if,
p.tun_sa_out,
[p.tun_sa_in],
- nh=p.tun_if.remote_hosts[ii].ip4)
+ nh=p.tun_if.remote_hosts[ii].ip4,
+ )
p.tun_protect.add_vpp_config()
config_tra_params(p, self.encryption_type, p.tun_if)
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()
+ 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()
# in this v4 variant add the teibs after the protect
- p.teib = VppTeib(self, p.tun_if,
- p.tun_if.remote_hosts[ii].ip4,
- self.pg0.remote_hosts[ii].ip4).add_vpp_config()
+ p.teib = VppTeib(
+ self,
+ p.tun_if,
+ p.tun_if.remote_hosts[ii].ip4,
+ self.pg0.remote_hosts[ii].ip4,
+ ).add_vpp_config()
p.tun_dst = self.pg0.remote_hosts[ii].ip4
self.logger.info(self.vapi.cli("sh ipsec protect-hash"))
class TestIpsecMGreIfEspTra6(TemplateIpsec, IpsecTun6):
- """ Ipsec mGRE ESP v6 TRA tests """
+ """Ipsec mGRE ESP v6 TRA tests"""
+
tun6_encrypt_node_name = "esp6-encrypt-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,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IPv6(src=p.tun_dst,
- dst=self.pg0.local_ip6) /
- GRE() /
- IPv6(src=self.pg1.local_ip6,
- dst=self.pg1.remote_ip6) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts6(self, p, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- IPv6(src="1::1", dst=dst) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IPv6(src=p.tun_dst, dst=self.pg0.local_ip6)
+ / GRE()
+ / IPv6(src=self.pg1.local_ip6, dst=self.pg1.remote_ip6)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IPv6(src="1::1", dst=dst)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted6(self, p, rxs):
for rx in rxs:
N_NHS = 16
self.tun_if = self.pg0
p = self.ipv6_params
- p.tun_if = VppGreInterface(self,
- self.pg0.local_ip6,
- "::",
- mode=(VppEnum.vl_api_tunnel_mode_t.
- TUNNEL_API_MODE_MP))
+ p.tun_if = VppGreInterface(
+ self,
+ self.pg0.local_ip6,
+ "::",
+ mode=(VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_MP),
+ )
p.tun_if.add_vpp_config()
p.tun_if.admin_up()
p.tun_if.config_ip6()
p.scapy_tra_spi = p.scapy_tra_spi + ii
p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
p.vpp_tra_spi = p.vpp_tra_spi + ii
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ )
p.tun_sa_in.add_vpp_config()
# in this v6 variant add the teibs first then the protection
p.tun_dst = self.pg0.remote_hosts[ii].ip6
- VppTeib(self, p.tun_if,
- p.tun_if.remote_hosts[ii].ip6,
- p.tun_dst).add_vpp_config()
+ VppTeib(
+ self, p.tun_if, p.tun_if.remote_hosts[ii].ip6, p.tun_dst
+ ).add_vpp_config()
p.tun_protect = VppIpsecTunProtect(
self,
p.tun_if,
p.tun_sa_out,
[p.tun_sa_in],
- nh=p.tun_if.remote_hosts[ii].ip6)
+ nh=p.tun_if.remote_hosts[ii].ip6,
+ )
p.tun_protect.add_vpp_config()
config_tra_params(p, self.encryption_type, p.tun_if)
self.multi_params.append(p)
- VppIpRoute(self, p.remote_tun_if_host, 128,
- [VppRoutePath(p.tun_if.remote_hosts[ii].ip6,
- p.tun_if.sw_if_index)]).add_vpp_config()
+ VppIpRoute(
+ self,
+ p.remote_tun_if_host,
+ 128,
+ [VppRoutePath(p.tun_if.remote_hosts[ii].ip6, p.tun_if.sw_if_index)],
+ ).add_vpp_config()
p.tun_dst = self.pg0.remote_hosts[ii].ip6
self.logger.info(self.vapi.cli("sh log"))
@tag_fixme_vpp_workers
-class TestIpsec4TunProtect(TemplateIpsec,
- TemplateIpsec4TunProtect,
- IpsecTun4):
- """ IPsec IPv4 Tunnel protect - transport mode"""
+class TestIpsec4TunProtect(TemplateIpsec, TemplateIpsec4TunProtect, IpsecTun4):
+ """IPsec IPv4 Tunnel protect - transport mode"""
def setUp(self):
super(TestIpsec4TunProtect, self).setUp()
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
- np.crypt_key = b'X' + p.crypt_key[1:]
+ np.crypt_key = b"X" + p.crypt_key[1:]
np.scapy_tun_spi += 100
np.scapy_tun_sa_id += 1
np.vpp_tun_spi += 100
@tag_fixme_vpp_workers
-class TestIpsec4TunProtectUdp(TemplateIpsec,
- TemplateIpsec4TunProtect,
- IpsecTun4):
- """ IPsec IPv4 Tunnel protect - transport mode"""
+class TestIpsec4TunProtectUdp(TemplateIpsec, TemplateIpsec4TunProtect, IpsecTun4):
+ """IPsec IPv4 Tunnel protect - transport mode"""
def setUp(self):
super(TestIpsec4TunProtectUdp, self).setUp()
self.tun_if = self.pg0
p = self.ipv4_params
- p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
- IPSEC_API_SAD_FLAG_UDP_ENCAP)
+ p.flags = VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_UDP_ENCAP
p.nat_header = UDP(sport=4500, dport=4500)
self.config_network(p)
self.config_sa_tra(p)
self.assertEqual(p.tun_if.get_tx_stats(), 127)
def test_keepalive(self):
- """ IPSEC NAT Keepalive """
+ """IPSEC NAT Keepalive"""
self.verify_keepalive(self.ipv4_params)
@tag_fixme_vpp_workers
-class TestIpsec4TunProtectTun(TemplateIpsec,
- TemplateIpsec4TunProtect,
- IpsecTun4):
- """ IPsec IPv4 Tunnel protect - tunnel mode"""
+class TestIpsec4TunProtectTun(TemplateIpsec, TemplateIpsec4TunProtect, IpsecTun4):
+ """IPsec IPv4 Tunnel protect - tunnel mode"""
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
def tearDown(self):
super(TestIpsec4TunProtectTun, self).tearDown()
- def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=sw_intf.remote_ip4,
- dst=sw_intf.local_ip4) /
- IP(src=src, dst=dst) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts(self, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- IP(src=src, dst=dst) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=sw_intf.remote_ip4, dst=sw_intf.local_ip4)
+ / IP(src=src, dst=dst)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IP(src=src, dst=dst)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted(self, p, rxs):
for rx in rxs:
raise
def test_tun_44(self):
- """IPSEC tunnel protect """
+ """IPSEC tunnel protect"""
p = self.ipv4_params
# also add an output features on the tunnel and physical interface
# so we test they still work
- r_all = AclRule(True,
- src_prefix="0.0.0.0/0",
- dst_prefix="0.0.0.0/0",
- proto=0)
+ 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()
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
- np.crypt_key = b'X' + p.crypt_key[1:]
+ np.crypt_key = b"X" + p.crypt_key[1:]
np.scapy_tun_spi += 100
np.scapy_tun_sa_id += 1
np.vpp_tun_spi += 100
self.unconfig_network(p)
-class TestIpsec4TunProtectTunDrop(TemplateIpsec,
- TemplateIpsec4TunProtect,
- IpsecTun4):
- """ IPsec IPv4 Tunnel protect - tunnel mode - drop"""
+class TestIpsec4TunProtectTunDrop(TemplateIpsec, TemplateIpsec4TunProtect, IpsecTun4):
+ """IPsec IPv4 Tunnel protect - tunnel mode - drop"""
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
def tearDown(self):
super(TestIpsec4TunProtectTunDrop, self).tearDown()
- def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=sw_intf.remote_ip4,
- dst="5.5.5.5") /
- IP(src=src, dst=dst) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=sw_intf.remote_ip4, dst="5.5.5.5")
+ / IP(src=src, dst=dst)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
def test_tun_drop_44(self):
- """IPSEC tunnel protect bogus tunnel header """
+ """IPSEC tunnel protect bogus tunnel header"""
p = self.ipv4_params
self.config_sa_tun(p)
self.config_protect(p)
- tx = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
- src=p.remote_tun_if_host,
- dst=self.pg1.remote_ip4,
- count=63)
+ tx = self.gen_encrypt_pkts(
+ p,
+ p.scapy_tun_sa,
+ self.tun_if,
+ src=p.remote_tun_if_host,
+ dst=self.pg1.remote_ip4,
+ count=63,
+ )
self.send_and_assert_no_replies(self.tun_if, tx)
# teardown
@tag_fixme_vpp_workers
-class TestIpsec6TunProtect(TemplateIpsec,
- TemplateIpsec6TunProtect,
- IpsecTun6):
- """ IPsec IPv6 Tunnel protect - transport mode"""
+class TestIpsec6TunProtect(TemplateIpsec, TemplateIpsec6TunProtect, IpsecTun6):
+ """IPsec IPv6 Tunnel protect - transport mode"""
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
- np.crypt_key = b'X' + p.crypt_key[1:]
+ np.crypt_key = b"X" + p.crypt_key[1:]
np.scapy_tun_spi += 100
np.scapy_tun_sa_id += 1
np.vpp_tun_spi += 100
# bounce the interface state
p.tun_if.admin_down()
self.verify_drop_tun_66(np, count=127)
- node = ('/err/ipsec6-tun-input/%s' %
- 'ipsec packets received on disabled interface')
+ node = (
+ "/err/ipsec6-tun-input/%s" % "ipsec packets received on disabled interface"
+ )
self.assertEqual(127, self.statistics.get_err_counter(node))
p.tun_if.admin_up()
self.verify_tun_66(np, count=127)
# 2) swap output SA to [new]
# 3) use only [new] input SA
np3 = copy.copy(np)
- np3.crypt_key = b'Z' + p.crypt_key[1:]
+ np3.crypt_key = b"Z" + p.crypt_key[1:]
np3.scapy_tun_spi += 100
np3.scapy_tun_sa_id += 1
np3.vpp_tun_spi += 100
self.config_sa_tra(np3)
# step 1;
- p.tun_protect.update_vpp_config(np.tun_sa_out,
- [np.tun_sa_in, np3.tun_sa_in])
+ p.tun_protect.update_vpp_config(np.tun_sa_out, [np.tun_sa_in, np3.tun_sa_in])
self.verify_tun_66(np, np, count=127)
self.verify_tun_66(np3, np, count=127)
# step 2;
- p.tun_protect.update_vpp_config(np3.tun_sa_out,
- [np.tun_sa_in, np3.tun_sa_in])
+ p.tun_protect.update_vpp_config(np3.tun_sa_out, [np.tun_sa_in, np3.tun_sa_in])
self.verify_tun_66(np, np3, count=127)
self.verify_tun_66(np3, np3, count=127)
# step 1;
- p.tun_protect.update_vpp_config(np3.tun_sa_out,
- [np3.tun_sa_in])
+ 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_rx_66(np, count=127)
- self.assertEqual(p.tun_if.get_rx_stats(), 127*9)
- self.assertEqual(p.tun_if.get_tx_stats(), 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
@tag_fixme_vpp_workers
-class TestIpsec6TunProtectTun(TemplateIpsec,
- TemplateIpsec6TunProtect,
- IpsecTun6):
- """ IPsec IPv6 Tunnel protect - tunnel mode"""
+class TestIpsec6TunProtectTun(TemplateIpsec, TemplateIpsec6TunProtect, IpsecTun6):
+ """IPsec IPv6 Tunnel protect - tunnel mode"""
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
def tearDown(self):
super(TestIpsec6TunProtectTun, self).tearDown()
- def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IPv6(src=sw_intf.remote_ip6,
- dst=sw_intf.local_ip6) /
- IPv6(src=src, dst=dst) /
- UDP(sport=1166, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts6(self, p, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- IPv6(src=src, dst=dst) /
- UDP(sport=1166, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IPv6(src=sw_intf.remote_ip6, dst=sw_intf.local_ip6)
+ / IPv6(src=src, dst=dst)
+ / UDP(sport=1166, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IPv6(src=src, dst=dst)
+ / UDP(sport=1166, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted6(self, p, rxs):
for rx in rxs:
raise
def test_tun_66(self):
- """IPSEC tunnel protect """
+ """IPSEC tunnel protect"""
p = self.ipv6_params
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
- np.crypt_key = b'X' + p.crypt_key[1:]
+ np.crypt_key = b"X" + p.crypt_key[1:]
np.scapy_tun_spi += 100
np.scapy_tun_sa_id += 1
np.vpp_tun_spi += 100
self.unconfig_network(p)
-class TestIpsec6TunProtectTunDrop(TemplateIpsec,
- TemplateIpsec6TunProtect,
- IpsecTun6):
- """ IPsec IPv6 Tunnel protect - tunnel mode - drop"""
+class TestIpsec6TunProtectTunDrop(TemplateIpsec, TemplateIpsec6TunProtect, IpsecTun6):
+ """IPsec IPv6 Tunnel protect - tunnel mode - drop"""
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
def tearDown(self):
super(TestIpsec6TunProtectTunDrop, self).tearDown()
- def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
+ def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
# the IP destination of the revelaed packet does not match
# that assigned to the tunnel
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IPv6(src=sw_intf.remote_ip6,
- dst="5::5") /
- IPv6(src=src, dst=dst) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IPv6(src=sw_intf.remote_ip6, dst="5::5")
+ / IPv6(src=src, dst=dst)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
def test_tun_drop_66(self):
- """IPSEC 6 tunnel protect bogus tunnel header """
+ """IPSEC 6 tunnel protect bogus tunnel header"""
p = self.ipv6_params
self.config_sa_tun(p)
self.config_protect(p)
- tx = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
- src=p.remote_tun_if_host,
- dst=self.pg1.remote_ip6,
- count=63)
+ tx = self.gen_encrypt_pkts6(
+ p,
+ p.scapy_tun_sa,
+ self.tun_if,
+ src=p.remote_tun_if_host,
+ dst=self.pg1.remote_ip6,
+ count=63,
+ )
self.send_and_assert_no_replies(self.tun_if, tx)
self.unconfig_protect(p)
class TemplateIpsecItf4(object):
- """ IPsec Interface IPv4 """
+ """IPsec Interface IPv4"""
encryption_type = ESP
tun4_encrypt_node_name = "esp4-encrypt-tun"
def config_sa_tun(self, p, src, dst):
config_tun_params(p, self.encryption_type, None, src, dst)
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- src, dst,
- flags=p.flags)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ src,
+ dst,
+ flags=p.flags,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- dst, src,
- flags=p.flags)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ dst,
+ src,
+ flags=p.flags,
+ )
p.tun_sa_in.add_vpp_config()
def config_protect(self, p):
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
- def config_network(self, p, instance=0xffffffff):
+ def config_network(self, p, instance=0xFFFFFFFF):
p.tun_if = VppIpsecInterface(self, instance=instance)
p.tun_if.add_vpp_config()
p.tun_if.config_ip4()
p.tun_if.config_ip6()
- p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
- [VppRoutePath(p.tun_if.remote_ip4,
- 0xffffffff)])
+ p.route = VppIpRoute(
+ self,
+ p.remote_tun_if_host,
+ 32,
+ [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)],
+ )
p.route.add_vpp_config()
- r = VppIpRoute(self, p.remote_tun_if_host6, 128,
- [VppRoutePath(p.tun_if.remote_ip6,
- 0xffffffff,
- proto=DpoProto.DPO_PROTO_IP6)])
+ r = VppIpRoute(
+ self,
+ p.remote_tun_if_host6,
+ 128,
+ [
+ VppRoutePath(
+ p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
+ )
+ ],
+ )
r.add_vpp_config()
def unconfig_network(self, p):
@tag_fixme_vpp_workers
-class TestIpsecItf4(TemplateIpsec,
- TemplateIpsecItf4,
- IpsecTun4):
- """ IPsec Interface IPv4 """
+class TestIpsecItf4(TemplateIpsec, TemplateIpsecItf4, IpsecTun4):
+ """IPsec Interface IPv4"""
def setUp(self):
super(TestIpsecItf4, self).setUp()
p = self.ipv4_params
self.config_network(p)
- config_tun_params(p, self.encryption_type, None,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ 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_sa_tun(p, self.pg0.local_ip4, self.pg0.remote_ip4)
self.config_protect(p)
self.verify_tun_44(p, count=n_pkts)
p.tun_if.admin_up()
self.verify_tun_44(p, count=n_pkts)
- self.assertEqual(p.tun_if.get_rx_stats(), 3*n_pkts)
- self.assertEqual(p.tun_if.get_tx_stats(), 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)
- self.assertEqual(p.tun_if.get_rx_stats(), 4*n_pkts)
- self.assertEqual(p.tun_if.get_tx_stats(), 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"
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
- np.crypt_key = b'X' + p.crypt_key[1:]
+ np.crypt_key = b"X" + p.crypt_key[1:]
np.scapy_tun_spi += 100
np.scapy_tun_sa_id += 1
np.vpp_tun_spi += 100
np.tun_if.local_spi = p.vpp_tun_spi
np.tun_if.remote_spi = p.scapy_tun_spi
- self.config_sa_tun(np,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ self.config_sa_tun(np, self.pg0.local_ip4, self.pg0.remote_ip4)
self.config_protect(np)
self.unconfig_sa(p)
n_pkts = 127
p = copy.copy(self.ipv4_params)
- p.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
- IPSEC_API_INTEG_ALG_NONE)
- p.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
- IPSEC_API_CRYPTO_ALG_NONE)
+ p.auth_algo_vpp_id = VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
+ p.crypt_algo_vpp_id = (
+ VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_NONE
+ )
p.crypt_algo = "NULL"
p.auth_algo = "NULL"
self.config_network(p)
- self.config_sa_tun(p,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ self.config_sa_tun(p, self.pg0.local_ip4, self.pg0.remote_ip4)
self.config_protect(p)
self.logger.info(self.vapi.cli("sh ipsec sa"))
p = self.ipv4_params
self.config_network(p)
- self.config_sa_tun(p,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ 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)
+ 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
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)
+ 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.unconfig_network(p)
-class TestIpsecItf4MPLS(TemplateIpsec,
- TemplateIpsecItf4,
- IpsecTun4):
- """ IPsec Interface MPLSoIPv4 """
+class TestIpsecItf4MPLS(TemplateIpsec, TemplateIpsecItf4, IpsecTun4):
+ """IPsec Interface MPLSoIPv4"""
tun4_encrypt_node_name = "esp-mpls-encrypt-tun"
def tearDown(self):
super(TestIpsecItf4MPLS, self).tearDown()
- def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(MPLS(label=44, ttl=3) /
- IP(src=src, dst=dst) /
- UDP(sport=1166, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ MPLS(label=44, ttl=3)
+ / IP(src=src, dst=dst)
+ / UDP(sport=1166, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
def verify_encrypted(self, p, sa, rxs):
for rx in rxs:
self.config_network(p)
# deag MPLS routes from the tunnel
- r4 = VppMplsRoute(self, 44, 1,
- [VppRoutePath(
- self.pg1.remote_ip4,
- self.pg1.sw_if_index)]).add_vpp_config()
- p.route.modify([VppRoutePath(p.tun_if.remote_ip4,
- p.tun_if.sw_if_index,
- labels=[VppMplsLabel(44)])])
+ r4 = VppMplsRoute(
+ self, 44, 1, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)]
+ ).add_vpp_config()
+ p.route.modify(
+ [
+ VppRoutePath(
+ p.tun_if.remote_ip4, p.tun_if.sw_if_index, labels=[VppMplsLabel(44)]
+ )
+ ]
+ )
p.tun_if.enable_mpls()
- self.config_sa_tun(p,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ 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)
class TemplateIpsecItf6(object):
- """ IPsec Interface IPv6 """
+ """IPsec Interface IPv6"""
encryption_type = ESP
tun6_encrypt_node_name = "esp6-encrypt-tun"
def config_sa_tun(self, p, src, dst):
config_tun_params(p, self.encryption_type, None, src, dst)
- if not hasattr(p, 'tun_flags'):
+ if not hasattr(p, "tun_flags"):
p.tun_flags = None
- if not hasattr(p, 'hop_limit'):
+ if not hasattr(p, "hop_limit"):
p.hop_limit = 255
- p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- src, dst,
- flags=p.flags,
- tun_flags=p.tun_flags,
- hop_limit=p.hop_limit)
+ p.tun_sa_out = VppIpsecSA(
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ src,
+ dst,
+ flags=p.flags,
+ tun_flags=p.tun_flags,
+ hop_limit=p.hop_limit,
+ )
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
- self.vpp_esp_protocol,
- dst, src,
- flags=p.flags)
+ p.tun_sa_in = VppIpsecSA(
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
+ self.vpp_esp_protocol,
+ dst,
+ src,
+ flags=p.flags,
+ )
p.tun_sa_in.add_vpp_config()
def config_protect(self, p):
- p.tun_protect = VppIpsecTunProtect(self,
- p.tun_if,
- p.tun_sa_out,
- [p.tun_sa_in])
+ p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
p.tun_protect.add_vpp_config()
def config_network(self, p):
p.tun_if.config_ip4()
p.tun_if.config_ip6()
- r = VppIpRoute(self, p.remote_tun_if_host4, 32,
- [VppRoutePath(p.tun_if.remote_ip4,
- 0xffffffff)])
+ r = VppIpRoute(
+ self,
+ p.remote_tun_if_host4,
+ 32,
+ [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)],
+ )
r.add_vpp_config()
- p.route = VppIpRoute(self, p.remote_tun_if_host, 128,
- [VppRoutePath(p.tun_if.remote_ip6,
- 0xffffffff,
- proto=DpoProto.DPO_PROTO_IP6)])
+ p.route = VppIpRoute(
+ self,
+ p.remote_tun_if_host,
+ 128,
+ [
+ VppRoutePath(
+ p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
+ )
+ ],
+ )
p.route.add_vpp_config()
def unconfig_network(self, p):
@tag_fixme_vpp_workers
-class TestIpsecItf6(TemplateIpsec,
- TemplateIpsecItf6,
- IpsecTun6):
- """ IPsec Interface IPv6 """
+class TestIpsecItf6(TemplateIpsec, TemplateIpsecItf6, IpsecTun6):
+ """IPsec Interface IPv6"""
def setUp(self):
super(TestIpsecItf6, self).setUp()
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)
+ 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_sa_tun(p, self.pg0.local_ip6, self.pg0.remote_ip6)
self.config_protect(p)
self.verify_tun_66(p, count=n_pkts)
p.tun_if.admin_up()
self.verify_tun_66(p, count=n_pkts)
- self.assertEqual(p.tun_if.get_rx_stats(), 3*n_pkts)
- self.assertEqual(p.tun_if.get_tx_stats(), 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)
- self.assertEqual(p.tun_if.get_rx_stats(), 4*n_pkts)
- self.assertEqual(p.tun_if.get_tx_stats(), 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"
# rekey - create new SAs and update the tunnel protection
np = copy.copy(p)
- np.crypt_key = b'X' + p.crypt_key[1:]
+ np.crypt_key = b"X" + p.crypt_key[1:]
np.scapy_tun_spi += 100
np.scapy_tun_sa_id += 1
np.vpp_tun_spi += 100
np.tun_if.remote_spi = p.scapy_tun_spi
np.inner_hop_limit = 24
np.outer_hop_limit = 128
- np.inner_flow_label = 0xabcde
- np.outer_flow_label = 0xabcde
+ np.inner_flow_label = 0xABCDE
+ np.outer_flow_label = 0xABCDE
np.hop_limit = 128
np.tun_flags = tf.TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_FLOW_LABEL
- self.config_sa_tun(np,
- self.pg0.local_ip6,
- self.pg0.remote_ip6)
+ self.config_sa_tun(np, self.pg0.local_ip6, self.pg0.remote_ip6)
self.config_protect(np)
self.unconfig_sa(p)
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_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)
+ 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
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)
+ 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)
class TestIpsecMIfEsp4(TemplateIpsec, IpsecTun4):
- """ Ipsec P2MP ESP v4 tests """
+ """Ipsec P2MP ESP v4 tests"""
+
tun4_encrypt_node_name = "esp4-encrypt-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,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(IP(src=self.pg1.local_ip4,
- dst=self.pg1.remote_ip4) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
-
- def gen_pkts(self, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- IP(src="1.1.1.1", dst=dst) /
- UDP(sport=1144, dport=2233) /
- Raw(b'X' * payload_size)
- for i in range(count)]
+ def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ IP(src=self.pg1.local_ip4, dst=self.pg1.remote_ip4)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IP(src="1.1.1.1", dst=dst)
+ / UDP(sport=1144, dport=2233)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
def verify_decrypted(self, p, rxs):
for rx in rxs:
def verify_encrypted(self, p, sa, rxs):
for rx in rxs:
try:
- self.assertEqual(rx[IP].tos,
- VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF << 2)
+ self.assertEqual(
+ rx[IP].tos, VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF << 2
+ )
self.assertEqual(rx[IP].ttl, p.hop_limit)
pkt = sa.decrypt(rx[IP])
if not pkt.haslayer(IP):
N_NHS = 16
self.tun_if = self.pg0
p = self.ipv4_params
- p.tun_if = VppIpsecInterface(self,
- mode=(VppEnum.vl_api_tunnel_mode_t.
- TUNNEL_API_MODE_MP))
+ p.tun_if = VppIpsecInterface(
+ self, mode=(VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_MP)
+ )
p.tun_if.add_vpp_config()
p.tun_if.admin_up()
p.tun_if.config_ip4()
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)
+ 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()
p.scapy_tra_spi = p.scapy_tra_spi + ii
p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
p.vpp_tra_spi = p.vpp_tra_spi + ii
- p.hop_limit = ii+10
+ p.hop_limit = ii + 10
p.tun_sa_out = VppIpsecSA(
- self, p.scapy_tun_sa_id, p.scapy_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
+ self,
+ p.scapy_tun_sa_id,
+ p.scapy_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
self.vpp_esp_protocol,
self.pg0.local_ip4,
self.pg0.remote_hosts[ii].ip4,
dscp=VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF,
- hop_limit=p.hop_limit)
+ hop_limit=p.hop_limit,
+ )
p.tun_sa_out.add_vpp_config()
p.tun_sa_in = VppIpsecSA(
- self, p.vpp_tun_sa_id, p.vpp_tun_spi,
- p.auth_algo_vpp_id, p.auth_key,
- p.crypt_algo_vpp_id, p.crypt_key,
+ self,
+ p.vpp_tun_sa_id,
+ p.vpp_tun_spi,
+ p.auth_algo_vpp_id,
+ p.auth_key,
+ p.crypt_algo_vpp_id,
+ p.crypt_key,
self.vpp_esp_protocol,
self.pg0.remote_hosts[ii].ip4,
self.pg0.local_ip4,
dscp=VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF,
- hop_limit=p.hop_limit)
+ hop_limit=p.hop_limit,
+ )
p.tun_sa_in.add_vpp_config()
p.tun_protect = VppIpsecTunProtect(
p.tun_if,
p.tun_sa_out,
[p.tun_sa_in],
- nh=p.tun_if.remote_hosts[ii].ip4)
+ nh=p.tun_if.remote_hosts[ii].ip4,
+ )
p.tun_protect.add_vpp_config()
- config_tun_params(p, self.encryption_type, None,
- self.pg0.local_ip4,
- self.pg0.remote_hosts[ii].ip4)
+ config_tun_params(
+ p,
+ self.encryption_type,
+ None,
+ self.pg0.local_ip4,
+ self.pg0.remote_hosts[ii].ip4,
+ )
self.multi_params.append(p)
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()
+ 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
self.verify_tun_44(p, count=N_PKTS)
-class TestIpsecItf6MPLS(TemplateIpsec,
- TemplateIpsecItf6,
- IpsecTun6):
- """ IPsec Interface MPLSoIPv6 """
+class TestIpsecItf6MPLS(TemplateIpsec, TemplateIpsecItf6, IpsecTun6):
+ """IPsec Interface MPLSoIPv6"""
tun6_encrypt_node_name = "esp-mpls-encrypt-tun"
def tearDown(self):
super(TestIpsecItf6MPLS, self).tearDown()
- def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1,
- payload_size=100):
- return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
- sa.encrypt(MPLS(label=66, ttl=3) /
- IPv6(src=src, dst=dst) /
- UDP(sport=1166, dport=2233) /
- Raw(b'X' * payload_size))
- for i in range(count)]
+ def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / sa.encrypt(
+ MPLS(label=66, ttl=3)
+ / IPv6(src=src, dst=dst)
+ / UDP(sport=1166, dport=2233)
+ / Raw(b"X" * payload_size)
+ )
+ for i in range(count)
+ ]
def verify_encrypted6(self, p, sa, rxs):
for rx in rxs:
self.config_network(p)
# deag MPLS routes from the tunnel
- r6 = VppMplsRoute(self, 66, 1,
- [VppRoutePath(
- self.pg1.remote_ip6,
- self.pg1.sw_if_index)],
- eos_proto=f.FIB_PATH_NH_PROTO_IP6).add_vpp_config()
- p.route.modify([VppRoutePath(p.tun_if.remote_ip6,
- p.tun_if.sw_if_index,
- labels=[VppMplsLabel(66)])])
+ r6 = VppMplsRoute(
+ self,
+ 66,
+ 1,
+ [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
+ eos_proto=f.FIB_PATH_NH_PROTO_IP6,
+ ).add_vpp_config()
+ p.route.modify(
+ [
+ VppRoutePath(
+ p.tun_if.remote_ip6, p.tun_if.sw_if_index, labels=[VppMplsLabel(66)]
+ )
+ ]
+ )
p.tun_if.enable_mpls()
- self.config_sa_tun(p,
- self.pg0.local_ip6,
- self.pg0.remote_ip6)
+ 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)
self.unconfig_network(p)
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff