X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ipsec_nat.py;h=07670d71b03682936306972607379c25344dedb5;hb=097fa66b986f06281f603767d321ab13ab6c88c3;hp=aa6d87cb96e8cbbb067bd402c0e3a74e73860ea5;hpb=311124e21b9b8ca11073d82408ffb81d98790969;p=vpp.git diff --git a/test/test_ipsec_nat.py b/test/test_ipsec_nat.py index aa6d87cb96e..07670d71b03 100644 --- a/test/test_ipsec_nat.py +++ b/test/test_ipsec_nat.py @@ -2,14 +2,18 @@ import socket +import scapy.compat from scapy.layers.l2 import Ether from scapy.layers.inet import ICMP, IP, TCP, UDP from scapy.layers.ipsec import SecurityAssociation, ESP + from util import ppp, ppc from template_ipsec import TemplateIpsec -from vpp_ipsec import * +from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\ + VppIpsecSpdItfBinding from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_ip import DpoProto +from vpp_papi import VppEnum class IPSecNATTestCase(TemplateIpsec): @@ -34,6 +38,14 @@ class IPSecNATTestCase(TemplateIpsec): icmp_id_in = 6305 icmp_id_out = 6305 + @classmethod + def setUpClass(cls): + super(IPSecNATTestCase, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(IPSecNATTestCase, cls).tearDownClass() + def setUp(self): super(IPSecNATTestCase, self).setUp() self.tun_if = self.pg0 @@ -45,14 +57,13 @@ class IPSecNATTestCase(TemplateIpsec): p = self.ipv4_params self.config_esp_tun(p) - self.logger.info(self.vapi.ppcli("show ipsec")) + self.logger.info(self.vapi.ppcli("show ipsec all")) d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4 VppIpRoute(self, p.remote_tun_if_host, p.addr_len, [VppRoutePath(self.tun_if.remote_addr[p.addr_type], 0xffffffff, - proto=d)], - is_ip6=p.is_ipv6).add_vpp_config() + proto=d)]).add_vpp_config() def tearDown(self): super(IPSecNATTestCase, self).tearDown() @@ -124,9 +135,9 @@ class IPSecNATTestCase(TemplateIpsec): def verify_capture_encrypted(self, capture, sa): for packet in capture: try: - copy = packet.__class__(str(packet)) + copy = packet.__class__(scapy.compat.raw(packet)) del copy[UDP].len - copy = packet.__class__(str(copy)) + copy = packet.__class__(scapy.compat.raw(copy)) self.assert_equal(packet[UDP].len, copy[UDP].len, "UDP header length") self.assert_packet_checksums_valid(packet) @@ -154,6 +165,9 @@ class IPSecNATTestCase(TemplateIpsec): crypt_key = params.crypt_key addr_any = params.addr_any addr_bcast = params.addr_bcast + flags = (VppEnum.vl_api_ipsec_sad_flags_t. + IPSEC_API_SAD_FLAG_UDP_ENCAP) + e = VppEnum.vl_api_ipsec_spd_action_t VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi, auth_algo_vpp_id, auth_key, @@ -161,14 +175,14 @@ class IPSecNATTestCase(TemplateIpsec): self.vpp_esp_protocol, self.pg1.remote_addr[addr_type], self.tun_if.remote_addr[addr_type], - udp_encap=1).add_vpp_config() + flags=flags).add_vpp_config() VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi, auth_algo_vpp_id, auth_key, crypt_algo_vpp_id, crypt_key, self.vpp_esp_protocol, self.tun_if.remote_addr[addr_type], self.pg1.remote_addr[addr_type], - udp_encap=1).add_vpp_config() + flags=flags).add_vpp_config() VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id, addr_any, addr_bcast, @@ -197,14 +211,16 @@ class IPSecNATTestCase(TemplateIpsec): self.tun_if.remote_addr[addr_type], self.pg1.remote_addr[addr_type], self.pg1.remote_addr[addr_type], - 0, priority=10, policy=3, + 0, priority=10, + policy=e.IPSEC_API_SPD_ACTION_PROTECT, is_outbound=0).add_vpp_config() VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id, self.pg1.remote_addr[addr_type], self.pg1.remote_addr[addr_type], self.tun_if.remote_addr[addr_type], self.tun_if.remote_addr[addr_type], - 0, priority=10, policy=3).add_vpp_config() + 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT, + priority=10).add_vpp_config() def test_ipsec_nat_tun(self): """ IPSec/NAT tunnel test case """