ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

IPSEC-GRE; tests

Browse Source 
failing test disabled on ARM

Change-Id: I6b7535cd8f51fdaf9786ba77f9f61a7d8d049bbd
Signed-off-by: Neale Ranns <nranns@cisco.com>
This commit is contained in:
Neale Ranns
2019-03-29 20:23:58 +00:00
committed by Damjan Marion
parent 2b5ba9501c
commit f05e732e4a
4 changed files with 203 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
test/template_ipsec.py
View File

@@ -472,6 +472,33 @@ class IpsecTun4(object):
self.assert_packet_counter_equal(self.tun4_encrypt_node_name, count)
self.assert_packet_counter_equal(self.tun4_decrypt_node_name, count)
def verify_decrypted(self, p, rxs):
for rx in rxs:
self.assert_equal(rx[IP].src, p.remote_tun_if_host)
self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
self.assert_packet_checksums_valid(rx)
def verify_encrypted(self, p, sa, rxs):
decrypt_pkts = []
for rx in rxs:
try:
decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IP])
if not decrypt_pkt.haslayer(IP):
decrypt_pkt = IP(decrypt_pkt[Raw].load)
decrypt_pkts.append(decrypt_pkt)
self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
except:
self.logger.debug(ppp("Unexpected packet:", rx))
try:
self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
except:
pass
raise
pkts = reassemble4(decrypt_pkts)
for pkt in pkts:
self.assert_packet_checksums_valid(pkt)
def verify_tun_44(self, p, count=1, payload_size=64, n_rx=None):
self.vapi.cli("clear errors")
if not n_rx:
@@ -483,35 +510,15 @@ class IpsecTun4(object):
dst=self.pg1.remote_ip4,
count=count)
recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
for recv_pkt in recv_pkts:
self.assert_equal(recv_pkt[IP].src, p.remote_tun_if_host)
self.assert_equal(recv_pkt[IP].dst, self.pg1.remote_ip4)
self.assert_packet_checksums_valid(recv_pkt)
self.verify_decrypted(p, recv_pkts)
send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
dst=p.remote_tun_if_host, count=count,
payload_size=payload_size)
recv_pkts = self.send_and_expect(self.pg1, send_pkts,
self.tun_if, n_rx)
decrypt_pkts = []
for recv_pkt in recv_pkts:
try:
decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IP])
if not decrypt_pkt.haslayer(IP):
decrypt_pkt = IP(decrypt_pkt[Raw].load)
decrypt_pkts.append(decrypt_pkt)
self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
except:
self.logger.debug(ppp("Unexpected packet:", recv_pkt))
try:
self.logger.debug(
ppp("Decrypted packet:", decrypt_pkt))
except:
pass
raise
pkts = reassemble4(decrypt_pkts)
for pkt in pkts:
self.assert_packet_checksums_valid(pkt)
self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
finally:
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show ipsec"))

116
test/test_ipsec_tun_if_esp.py
View File

@@ -2,11 +2,17 @@ import unittest
import socket
import copy
from scapy.layers.ipsec import ESP
from framework import VppTestRunner
from scapy.layers.l2 import Ether, Raw, GRE
from scapy.layers.inet import IP, UDP
from framework import VppTestRunner, is_skip_aarch64_set, is_platform_aarch64
from template_ipsec import TemplateIpsec, IpsecTun4Tests, IpsecTun6Tests, \
IpsecTun4, IpsecTun6, IpsecTcpTests, config_tun_params
from vpp_ipsec_tun_interface import VppIpsecTunInterface
from vpp_ipsec_tun_interface import VppIpsecTunInterface, \
VppIpsecGRETunInterface
from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
from vpp_ipsec import VppIpsecSA
from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
from util import ppp
class TemplateIpsec4TunIfEsp(TemplateIpsec):
@@ -248,5 +254,111 @@ class TestIpsec6MultiTunIfEsp(TemplateIpsec, IpsecTun6):
self.assertEqual(c['packets'], 127)
class TemplateIpsecGRETunIfEsp(TemplateIpsec):
""" IPsec GRE tunnel interface tests """
encryption_type = ESP
omac = "00:11:22:33:44:55"
def gen_encrypt_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() /
Ether(dst=self.omac) /
IP(src="1.1.1.1", dst="1.1.1.2") /
UDP(sport=1144, dport=2233) /
Raw('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('X' * payload_size)
for i in range(count)]
def verify_decrypted(self, p, rxs):
for rx in rxs:
self.assert_equal(rx[Ether].dst, self.omac)
self.assert_equal(rx[IP].dst, "1.1.1.2")
def verify_encrypted(self, p, sa, rxs):
for rx in rxs:
try:
pkt = sa.decrypt(rx[IP])
if not pkt.haslayer(IP):
pkt = IP(pkt[Raw].load)
self.assert_packet_checksums_valid(pkt)
self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
self.assertTrue(pkt.haslayer(GRE))
e = pkt[Ether]
self.assertEqual(e[Ether].dst, self.omac)
self.assertEqual(e[IP].dst, "1.1.1.2")
except (IndexError, AssertionError):
self.logger.debug(ppp("Unexpected packet:", rx))
try:
self.logger.debug(ppp("Decrypted packet:", pkt))
except:
pass
raise
def setUp(self):
super(TemplateIpsecGRETunIfEsp, self).setUp()
self.tun_if = self.pg0
p = self.ipv4_params
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.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.add_vpp_config()
self.tun = VppIpsecGRETunInterface(self, self.pg0,
p.tun_sa_out.id,
p.tun_sa_in.id)
self.tun.add_vpp_config()
self.tun.admin_up()
self.tun.config_ip4()
VppIpRoute(self, p.remote_tun_if_host, 32,
[VppRoutePath(self.tun.remote_ip4,
0xffffffff)]).add_vpp_config()
VppBridgeDomainPort(self, bd1, self.tun).add_vpp_config()
VppBridgeDomainPort(self, bd1, self.pg1).add_vpp_config()
def tearDown(self):
if not self.vpp_dead:
self.vapi.cli("show hardware")
self.tun.unconfig_ip4()
super(TemplateIpsecGRETunIfEsp, self).tearDown()
@unittest.skipIf(is_skip_aarch64_set and is_platform_aarch64,
"test doesn't work on aarch64")
class TestIpsecGRETunIfEsp1(TemplateIpsecGRETunIfEsp, IpsecTun4Tests):
""" Ipsec GRE ESP - TUN tests """
tun4_encrypt_node_name = "esp4-encrypt"
tun4_decrypt_node_name = "esp4-decrypt"
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)

45
test/vpp_ipsec_tun_interface.py
View File

@@ -45,3 +45,48 @@ class VppIpsecTunInterface(VppTunnelInterface):
def object_id(self):
return "ipsec-tun-if-%d" % self._sw_if_index
class VppIpsecGRETunInterface(VppTunnelInterface):
"""
VPP IPsec GRE Tunnel interface
this creates headers
IP / ESP / IP / GRE / payload
i.e. it's GRE over IPSEC, rather than IPSEC over GRE.
"""
def __init__(self, test, parent_if, sa_out, sa_in):
super(VppIpsecGRETunInterface, self).__init__(test, parent_if)
self.sa_in = sa_in
self.sa_out = sa_out
def add_vpp_config(self):
r = self.test.vapi.ipsec_gre_tunnel_add_del(
self.parent_if.local_ip4n,
self.parent_if.remote_ip4n,
self.sa_out,
self.sa_in)
self.set_sw_if_index(r.sw_if_index)
self.generate_remote_hosts()
self.test.registry.register(self, self.test.logger)
def remove_vpp_config(self):
self.test.vapi.ipsec_gre_tunnel_add_del(
self.parent_if.local_ip4n,
self.parent_if.remote_ip4n,
self.sa_out,
self.sa_in,
is_add=0)
def query_vpp_config(self):
ts = self.test.vapi.ipsec_gre_tunnel_dump(sw_if_index=0xffffffff)
for t in ts:
if t.tunnel.sw_if_index == self._sw_if_index:
return True
return False
def __str__(self):
return self.object_id()
def object_id(self):
return "ipsec-gre-tun-if-%d" % self._sw_if_index

13
test/vpp_papi_provider.py
View File

@@ -2498,6 +2498,19 @@ class VppPapiProvider(object):
'show_instance': show_instance
})
def ipsec_gre_tunnel_add_del(self, local_ip, remote_ip,
sa_out, sa_in, is_add=1):
return self.api(self.papi.ipsec_gre_tunnel_add_del,
{
'is_add': is_add,
'tunnel': {
'src': local_ip,
'dst': remote_ip,
'local_sa_id': sa_out,
'remote_sa_id': sa_in
}
})
def ipsec_select_backend(self, protocol, index):
return self.api(self.papi.ipsec_select_backend,
{'protocol': protocol, 'index': index})