diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/template_ipsec.py | 2 | ||||
| -rw-r--r-- | test/test_ipsec_tun_if_esp.py | 45 | ||||
| -rw-r--r-- | test/vpp_ipsec.py | 20 |
3 files changed, 61 insertions, 6 deletions
diff --git a/test/template_ipsec.py b/test/template_ipsec.py index 9d9ea3a86d3..d00216c7308 100644 --- a/test/template_ipsec.py +++ b/test/template_ipsec.py @@ -1291,7 +1291,7 @@ class IpsecTun4(object): decrypt_pkts = [] for rx in rxs: if p.nat_header: - self.assertEqual(rx[UDP].dport, 4500) + self.assertEqual(rx[UDP].dport, p.nat_header.dport) self.assert_packet_checksums_valid(rx) self.assertEqual(len(rx) - len(Ether()), rx[IP].len) try: diff --git a/test/test_ipsec_tun_if_esp.py b/test/test_ipsec_tun_if_esp.py index 61a66d40a4e..fe05f98e6e6 100644 --- a/test/test_ipsec_tun_if_esp.py +++ b/test/test_ipsec_tun_if_esp.py @@ -300,7 +300,7 @@ class TemplateIpsec4TunIfEspUdp(TemplateIpsec4TunProtect, TemplateIpsec): # which strips them self.assertTrue(rx.haslayer(UDP)) self.assert_equal(rx[UDP].sport, p.nat_header.sport) - self.assert_equal(rx[UDP].dport, 4500) + self.assert_equal(rx[UDP].dport, p.nat_header.dport) pkt = sa.decrypt(rx[IP]) if not pkt.haslayer(IP): @@ -344,7 +344,8 @@ class TemplateIpsec4TunIfEspUdp(TemplateIpsec4TunProtect, TemplateIpsec): p.crypt_algo_vpp_id, p.crypt_key, self.vpp_esp_protocol, - flags=p.flags, + flags=p.flags + | VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_IS_INBOUND, udp_src=p.nat_header.sport, udp_dst=p.nat_header.dport, ) @@ -429,6 +430,24 @@ class TestIpsec4TunIfEspUdpGCM(TemplateIpsec4TunIfEspUdp, IpsecTun4Tests): p.salt = 0 +class TestIpsec4TunIfEspUdpUpdate(TemplateIpsec4TunIfEspUdp, IpsecTun4Tests): + """Ipsec ESP UDP update tests""" + + tun4_input_node = "ipsec4-tun-input" + + def setUp(self): + super(TestIpsec4TunIfEspUdpUpdate, self).setUp() + p = self.ipv4_params + p.nat_header = UDP(sport=6565, dport=7676) + config_tun_params(p, self.encryption_type, p.tun_if) + p.tun_sa_in.update_vpp_config( + udp_src=p.nat_header.dport, udp_dst=p.nat_header.sport + ) + p.tun_sa_out.update_vpp_config( + udp_src=p.nat_header.sport, udp_dst=p.nat_header.dport + ) + + class TestIpsec4TunIfEsp2(TemplateIpsec4TunIfEsp, IpsecTcpTests): """Ipsec ESP - TCP tests""" @@ -583,7 +602,7 @@ class TemplateIpsec6TunIfEspUdp(TemplateIpsec6TunProtect, TemplateIpsec): # which strips them self.assertTrue(rx.haslayer(UDP)) self.assert_equal(rx[UDP].sport, p.nat_header.sport) - self.assert_equal(rx[UDP].dport, 4500) + self.assert_equal(rx[UDP].dport, p.nat_header.dport) pkt = sa.decrypt(rx[IP]) if not pkt.haslayer(IP): @@ -629,7 +648,8 @@ class TemplateIpsec6TunIfEspUdp(TemplateIpsec6TunProtect, TemplateIpsec): p.crypt_algo_vpp_id, p.crypt_key, self.vpp_esp_protocol, - flags=p.flags, + flags=p.flags + | VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_IS_INBOUND, udp_src=p.nat_header.sport, udp_dst=p.nat_header.dport, ) @@ -2957,7 +2977,8 @@ class TemplateIpsecItf4(object): self.vpp_esp_protocol, dst, src, - flags=p.flags, + flags=p.flags + | VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_IS_INBOUND, ) p.tun_sa_in.add_vpp_config() @@ -3063,6 +3084,20 @@ class TestIpsecItf4(TemplateIpsec, TemplateIpsecItf4, IpsecTun4): self.tun4_encrypt_node_name = "esp4-encrypt-tun" + # update the SA tunnel + config_tun_params( + p, self.encryption_type, None, self.pg2.local_ip4, self.pg2.remote_ip4 + ) + p.tun_sa_in.update_vpp_config( + is_tun=True, tun_src=self.pg2.remote_ip4, tun_dst=self.pg2.local_ip4 + ) + p.tun_sa_out.update_vpp_config( + is_tun=True, tun_src=self.pg2.local_ip4, tun_dst=self.pg2.remote_ip4 + ) + self.verify_tun_44(p, count=n_pkts) + self.assertEqual(p.tun_if.get_rx_stats(), 5 * n_pkts) + self.assertEqual(p.tun_if.get_tx_stats(), 4 * n_pkts) + self.vapi.cli("clear interfaces") # rekey - create new SAs and update the tunnel protection diff --git a/test/vpp_ipsec.py b/test/vpp_ipsec.py index eb0209fc57a..f50d491c396 100644 --- a/test/vpp_ipsec.py +++ b/test/vpp_ipsec.py @@ -295,6 +295,26 @@ class VppIpsecSA(VppObject): self.test.registry.register(self, self.test.logger) return self + def update_vpp_config( + self, udp_src=None, udp_dst=None, is_tun=False, tun_src=None, tun_dst=None + ): + if is_tun: + if tun_src: + self.tun_src = ip_address(text_type(tun_src)) + if tun_dst: + self.tun_dst = ip_address(text_type(tun_dst)) + if udp_src: + self.udp_src = udp_src + if udp_dst: + self.udp_dst = udp_dst + self.test.vapi.ipsec_sad_entry_update( + sad_id=self.id, + is_tun=is_tun, + tunnel=self.tunnel_encode(), + udp_src_port=udp_src, + udp_dst_port=udp_dst, + ) + def remove_vpp_config(self): self.test.vapi.ipsec_sad_entry_del(id=self.id) |