diff options
Diffstat (limited to 'test/test_ipsec_tun_if_esp.py')
| -rw-r--r-- | test/test_ipsec_tun_if_esp.py | 306 |
1 files changed, 302 insertions, 4 deletions
diff --git a/test/test_ipsec_tun_if_esp.py b/test/test_ipsec_tun_if_esp.py index a59baf1dfbf..183012608fe 100644 --- a/test/test_ipsec_tun_if_esp.py +++ b/test/test_ipsec_tun_if_esp.py @@ -15,7 +15,7 @@ from vpp_ipsec_tun_interface import VppIpsecTunInterface from vpp_gre_interface import VppGreInterface from vpp_ipip_tun_interface import VppIpIpTunInterface from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto -from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect +from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect, VppIpsecInterface from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint from vpp_teib import VppTeib @@ -24,13 +24,18 @@ from vpp_papi import VppEnum from vpp_acl import AclRule, VppAcl, VppAclInterface -def config_tun_params(p, encryption_type, tun_if): +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)) crypt_key = mk_scapy_crypt_key(p) - p.tun_dst = tun_if.remote_ip - p.tun_src = tun_if.local_ip + if tun_if: + p.tun_dst = tun_if.remote_ip + p.tun_src = tun_if.local_ip + else: + p.tun_dst = dst + p.tun_src = src + p.scapy_tun_sa = SecurityAssociation( encryption_type, spi=p.vpp_tun_spi, crypt_algo=p.crypt_algo, @@ -2473,5 +2478,298 @@ class TestIpsec6TunProtectTunDrop(TemplateIpsec, self.unconfig_network(p) +class TemplateIpsecItf4(object): + """ IPsec Interface IPv4 """ + + encryption_type = ESP + tun4_encrypt_node_name = "esp4-encrypt-tun" + tun4_decrypt_node_name = "esp4-decrypt-tun" + tun4_input_node = "ipsec4-tun-input" + + 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.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.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.add_vpp_config() + + def config_network(self, p): + p.tun_if = VppIpsecInterface(self) + + 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.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.add_vpp_config() + + def unconfig_network(self, p): + p.route.remove_vpp_config() + p.tun_if.remove_vpp_config() + + def unconfig_protect(self, p): + p.tun_protect.remove_vpp_config() + + def unconfig_sa(self, p): + p.tun_sa_out.remove_vpp_config() + p.tun_sa_in.remove_vpp_config() + + +class TestIpsecItf4(TemplateIpsec, + TemplateIpsecItf4, + IpsecTun4): + """ IPsec Interface IPv4 """ + + def setUp(self): + super(TestIpsecItf4, self).setUp() + + self.tun_if = self.pg0 + + def tearDown(self): + super(TestIpsecItf4, self).tearDown() + + def test_tun_44(self): + """IPSEC interface IPv4""" + + n_pkts = 127 + p = self.ipv4_params + + self.config_network(p) + self.config_sa_tun(p, + self.pg0.local_ip4, + self.pg0.remote_ip4) + self.config_protect(p) + + self.verify_tun_44(p, count=n_pkts) + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], n_pkts) + + p.tun_if.admin_down() + self.verify_tun_dropped_44(p, count=n_pkts) + p.tun_if.admin_up() + self.verify_tun_44(p, count=n_pkts) + + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], 3*n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], 2*n_pkts) + + # it's a v6 packet when its encrypted + self.tun4_encrypt_node_name = "esp6-encrypt-tun" + + self.verify_tun_64(p, count=n_pkts) + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], 4*n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], 3*n_pkts) + + self.tun4_encrypt_node_name = "esp4-encrypt-tun" + + self.vapi.cli("clear interfaces") + + # rekey - create new SAs and update the tunnel protection + np = copy.copy(p) + 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.vpp_tun_sa_id += 1 + 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_protect(np) + self.unconfig_sa(p) + + self.verify_tun_44(np, count=n_pkts) + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], n_pkts) + + # teardown + self.unconfig_protect(np) + self.unconfig_sa(np) + self.unconfig_network(p) + + +class TemplateIpsecItf6(object): + """ IPsec Interface IPv6 """ + + encryption_type = ESP + tun6_encrypt_node_name = "esp6-encrypt-tun" + tun6_decrypt_node_name = "esp6-decrypt-tun" + tun6_input_node = "ipsec6-tun-input" + + 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.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.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.add_vpp_config() + + def config_network(self, p): + p.tun_if = VppIpsecInterface(self) + + p.tun_if.add_vpp_config() + p.tun_if.admin_up() + 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.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.add_vpp_config() + + def unconfig_network(self, p): + p.route.remove_vpp_config() + p.tun_if.remove_vpp_config() + + def unconfig_protect(self, p): + p.tun_protect.remove_vpp_config() + + def unconfig_sa(self, p): + p.tun_sa_out.remove_vpp_config() + p.tun_sa_in.remove_vpp_config() + + +class TestIpsecItf6(TemplateIpsec, + TemplateIpsecItf6, + IpsecTun6): + """ IPsec Interface IPv6 """ + + def setUp(self): + super(TestIpsecItf6, self).setUp() + + self.tun_if = self.pg0 + + def tearDown(self): + super(TestIpsecItf6, self).tearDown() + + def test_tun_44(self): + """IPSEC interface IPv6""" + + n_pkts = 127 + p = self.ipv6_params + + self.config_network(p) + self.config_sa_tun(p, + self.pg0.local_ip6, + self.pg0.remote_ip6) + self.config_protect(p) + + self.verify_tun_66(p, count=n_pkts) + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], n_pkts) + + p.tun_if.admin_down() + self.verify_drop_tun_66(p, count=n_pkts) + p.tun_if.admin_up() + self.verify_tun_66(p, count=n_pkts) + + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], 3*n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], 2*n_pkts) + + # it's a v4 packet when its encrypted + self.tun6_encrypt_node_name = "esp4-encrypt-tun" + + self.verify_tun_46(p, count=n_pkts) + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], 4*n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], 3*n_pkts) + + self.tun6_encrypt_node_name = "esp6-encrypt-tun" + + self.vapi.cli("clear interfaces") + + # rekey - create new SAs and update the tunnel protection + np = copy.copy(p) + 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.vpp_tun_sa_id += 1 + 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_ip6, + self.pg0.remote_ip6) + self.config_protect(np) + self.unconfig_sa(p) + + self.verify_tun_66(np, count=n_pkts) + c = p.tun_if.get_rx_stats() + self.assertEqual(c['packets'], n_pkts) + c = p.tun_if.get_tx_stats() + self.assertEqual(c['packets'], n_pkts) + + # teardown + self.unconfig_protect(np) + self.unconfig_sa(np) + self.unconfig_network(p) + + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner) |