| import socket |
| import unittest |
| from scapy.layers.ipsec import ESP |
| |
| from framework import VppTestRunner |
| from template_ipsec import IpsecTraTests, IpsecTunTests |
| from template_ipsec import TemplateIpsec, IpsecTcpTests |
| |
| |
| class TemplateIpsecEsp(TemplateIpsec): |
| """ |
| Basic test for ipsec esp sanity - tunnel and transport modes. |
| |
| Below 4 cases are covered as part of this test |
| 1) ipsec esp v4 transport basic test - IPv4 Transport mode |
| scenario using HMAC-SHA1-96 intergrity algo |
| 2) ipsec esp v4 transport burst test |
| Above test for 257 pkts |
| 3) ipsec esp 4o4 tunnel basic test - IPv4 Tunnel mode |
| scenario using HMAC-SHA1-96 intergrity algo |
| 4) ipsec esp 4o4 tunnel burst test |
| Above test for 257 pkts |
| |
| TRANSPORT MODE: |
| |
| --- encrypt --- |
| |pg2| <-------> |VPP| |
| --- decrypt --- |
| |
| TUNNEL MODE: |
| |
| --- encrypt --- plain --- |
| |pg0| <------- |VPP| <------ |pg1| |
| --- --- --- |
| |
| --- decrypt --- plain --- |
| |pg0| -------> |VPP| ------> |pg1| |
| --- --- --- |
| """ |
| |
| def setUp(self): |
| super(TemplateIpsecEsp, self).setUp() |
| self.encryption_type = ESP |
| self.tun_if = self.pg0 |
| self.tra_if = self.pg2 |
| self.logger.info(self.vapi.ppcli("show int addr")) |
| self.vapi.ipsec_spd_add_del(self.tra_spd_id) |
| self.vapi.ipsec_interface_add_del_spd(self.tra_spd_id, |
| self.tra_if.sw_if_index) |
| for _, p in self.params.items(): |
| self.config_esp_tra(p) |
| self.configure_sa_tra(p) |
| self.logger.info(self.vapi.ppcli("show ipsec")) |
| self.vapi.ipsec_spd_add_del(self.tun_spd_id) |
| self.vapi.ipsec_interface_add_del_spd(self.tun_spd_id, |
| self.tun_if.sw_if_index) |
| for _, p in self.params.items(): |
| self.config_esp_tun(p) |
| self.logger.info(self.vapi.ppcli("show ipsec")) |
| for _, p in self.params.items(): |
| src = socket.inet_pton(p.addr_type, p.remote_tun_if_host) |
| self.vapi.ip_add_del_route( |
| src, p.addr_len, self.tun_if.remote_addr_n[p.addr_type], |
| is_ipv6=p.is_ipv6) |
| |
| def tearDown(self): |
| for _, p in self.params.items(): |
| self.unconfig_esp_tun(p) |
| for _, p in self.params.items(): |
| self.unconfig_esp_tra(p) |
| |
| self.vapi.ipsec_interface_add_del_spd(self.tun_spd_id, |
| self.tun_if.sw_if_index, |
| is_add=0) |
| self.vapi.ipsec_spd_add_del(self.tun_spd_id, is_add=0) |
| self.vapi.ipsec_interface_add_del_spd(self.tra_spd_id, |
| self.tra_if.sw_if_index, |
| is_add=0) |
| self.vapi.ipsec_spd_add_del(self.tra_spd_id, |
| is_add=0) |
| for _, p in self.params.items(): |
| src = socket.inet_pton(p.addr_type, p.remote_tun_if_host) |
| self.vapi.ip_add_del_route( |
| src, p.addr_len, self.tun_if.remote_addr_n[p.addr_type], |
| is_ipv6=p.is_ipv6, is_add=0) |
| |
| super(TemplateIpsecEsp, self).tearDown() |
| if not self.vpp_dead: |
| self.vapi.cli("show hardware") |
| |
| def config_esp_tun(self, params): |
| addr_type = params.addr_type |
| is_ipv6 = params.is_ipv6 |
| scapy_tun_sa_id = params.scapy_tun_sa_id |
| scapy_tun_spi = params.scapy_tun_spi |
| vpp_tun_sa_id = params.vpp_tun_sa_id |
| vpp_tun_spi = params.vpp_tun_spi |
| auth_algo_vpp_id = params.auth_algo_vpp_id |
| auth_key = params.auth_key |
| crypt_algo_vpp_id = params.crypt_algo_vpp_id |
| crypt_key = params.crypt_key |
| remote_tun_if_host = params.remote_tun_if_host |
| addr_any = params.addr_any |
| addr_bcast = params.addr_bcast |
| self.vapi.ipsec_sad_add_del_entry(scapy_tun_sa_id, scapy_tun_spi, |
| auth_algo_vpp_id, auth_key, |
| crypt_algo_vpp_id, crypt_key, |
| self.vpp_esp_protocol, |
| self.tun_if.local_addr_n[addr_type], |
| self.tun_if.remote_addr_n[addr_type], |
| is_tunnel=1, is_tunnel_ipv6=is_ipv6) |
| self.vapi.ipsec_sad_add_del_entry(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_n[addr_type], |
| self.tun_if.local_addr_n[addr_type], |
| is_tunnel=1, is_tunnel_ipv6=is_ipv6) |
| l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any) |
| l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast) |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, is_ipv6=is_ipv6, |
| protocol=socket.IPPROTO_ESP) |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, is_outbound=0, |
| protocol=socket.IPPROTO_ESP, |
| is_ipv6=is_ipv6) |
| l_startaddr = l_stopaddr = socket.inet_pton(addr_type, |
| remote_tun_if_host) |
| r_startaddr = r_stopaddr = self.pg1.remote_addr_n[addr_type] |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, vpp_tun_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, priority=10, policy=3, |
| is_ipv6=is_ipv6, is_outbound=0) |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id, |
| r_startaddr, r_stopaddr, l_startaddr, |
| l_stopaddr, priority=10, policy=3, |
| is_ipv6=is_ipv6) |
| l_startaddr = l_stopaddr = socket.inet_pton(addr_type, |
| remote_tun_if_host) |
| r_startaddr = r_stopaddr = self.pg0.local_addr_n[addr_type] |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, vpp_tun_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, priority=20, policy=3, |
| is_outbound=0, is_ipv6=is_ipv6) |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id, |
| r_startaddr, r_stopaddr, l_startaddr, |
| l_stopaddr, priority=20, policy=3, |
| is_ipv6=is_ipv6) |
| |
| def unconfig_esp_tun(self, params): |
| addr_type = params.addr_type |
| is_ipv6 = params.is_ipv6 |
| scapy_tun_sa_id = params.scapy_tun_sa_id |
| scapy_tun_spi = params.scapy_tun_spi |
| vpp_tun_sa_id = params.vpp_tun_sa_id |
| vpp_tun_spi = params.vpp_tun_spi |
| auth_algo_vpp_id = params.auth_algo_vpp_id |
| auth_key = params.auth_key |
| crypt_algo_vpp_id = params.crypt_algo_vpp_id |
| crypt_key = params.crypt_key |
| remote_tun_if_host = params.remote_tun_if_host |
| addr_any = params.addr_any |
| addr_bcast = params.addr_bcast |
| l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any) |
| l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast) |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, is_ipv6=is_ipv6, |
| protocol=socket.IPPROTO_ESP, |
| is_add=0) |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, is_outbound=0, |
| protocol=socket.IPPROTO_ESP, |
| is_ipv6=is_ipv6, |
| is_add=0) |
| l_startaddr = l_stopaddr = socket.inet_pton(addr_type, |
| remote_tun_if_host) |
| r_startaddr = r_stopaddr = self.pg1.remote_addr_n[addr_type] |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, vpp_tun_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, priority=10, policy=3, |
| is_ipv6=is_ipv6, is_outbound=0, |
| is_add=0) |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id, |
| r_startaddr, r_stopaddr, l_startaddr, |
| l_stopaddr, priority=10, policy=3, |
| is_ipv6=is_ipv6, is_add=0) |
| l_startaddr = l_stopaddr = socket.inet_pton(addr_type, |
| remote_tun_if_host) |
| r_startaddr = r_stopaddr = self.pg0.local_addr_n[addr_type] |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, vpp_tun_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, priority=20, policy=3, |
| is_outbound=0, is_ipv6=is_ipv6, |
| is_add=0) |
| self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id, |
| r_startaddr, r_stopaddr, l_startaddr, |
| l_stopaddr, priority=20, policy=3, |
| is_ipv6=is_ipv6, |
| is_add=0) |
| self.vapi.ipsec_sad_add_del_entry(scapy_tun_sa_id, scapy_tun_spi, |
| auth_algo_vpp_id, auth_key, |
| crypt_algo_vpp_id, crypt_key, |
| self.vpp_esp_protocol, |
| self.tun_if.local_addr_n[addr_type], |
| self.tun_if.remote_addr_n[addr_type], |
| is_tunnel=1, is_tunnel_ipv6=is_ipv6, |
| is_add=0) |
| self.vapi.ipsec_sad_add_del_entry(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_n[addr_type], |
| self.tun_if.local_addr_n[addr_type], |
| is_tunnel=1, is_tunnel_ipv6=is_ipv6, |
| is_add=0) |
| |
| def config_esp_tra(self, params): |
| addr_type = params.addr_type |
| is_ipv6 = params.is_ipv6 |
| scapy_tra_sa_id = params.scapy_tra_sa_id |
| scapy_tra_spi = params.scapy_tra_spi |
| vpp_tra_sa_id = params.vpp_tra_sa_id |
| vpp_tra_spi = params.vpp_tra_spi |
| auth_algo_vpp_id = params.auth_algo_vpp_id |
| auth_key = params.auth_key |
| crypt_algo_vpp_id = params.crypt_algo_vpp_id |
| crypt_key = params.crypt_key |
| addr_any = params.addr_any |
| addr_bcast = params.addr_bcast |
| self.vapi.ipsec_sad_add_del_entry(scapy_tra_sa_id, scapy_tra_spi, |
| auth_algo_vpp_id, auth_key, |
| crypt_algo_vpp_id, crypt_key, |
| self.vpp_esp_protocol, is_tunnel=0, |
| use_anti_replay=1) |
| self.vapi.ipsec_sad_add_del_entry(vpp_tra_sa_id, vpp_tra_spi, |
| auth_algo_vpp_id, auth_key, |
| crypt_algo_vpp_id, crypt_key, |
| self.vpp_esp_protocol, is_tunnel=0, |
| use_anti_replay=1) |
| l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any) |
| l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast) |
| self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, is_ipv6=is_ipv6, |
| protocol=socket.IPPROTO_ESP) |
| self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, is_outbound=0, |
| is_ipv6=is_ipv6, |
| protocol=socket.IPPROTO_ESP) |
| l_startaddr = l_stopaddr = self.tra_if.local_addr_n[addr_type] |
| r_startaddr = r_stopaddr = self.tra_if.remote_addr_n[addr_type] |
| self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, priority=10, policy=3, |
| is_outbound=0, is_ipv6=is_ipv6) |
| self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, scapy_tra_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, priority=10, policy=3, |
| is_ipv6=is_ipv6) |
| |
| def unconfig_esp_tra(self, params): |
| addr_type = params.addr_type |
| is_ipv6 = params.is_ipv6 |
| scapy_tra_sa_id = params.scapy_tra_sa_id |
| scapy_tra_spi = params.scapy_tra_spi |
| vpp_tra_sa_id = params.vpp_tra_sa_id |
| vpp_tra_spi = params.vpp_tra_spi |
| auth_algo_vpp_id = params.auth_algo_vpp_id |
| auth_key = params.auth_key |
| crypt_algo_vpp_id = params.crypt_algo_vpp_id |
| crypt_key = params.crypt_key |
| addr_any = params.addr_any |
| addr_bcast = params.addr_bcast |
| l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any) |
| l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast) |
| self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, is_ipv6=is_ipv6, |
| protocol=socket.IPPROTO_ESP, |
| is_add=0) |
| self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, is_outbound=0, |
| is_ipv6=is_ipv6, |
| protocol=socket.IPPROTO_ESP, |
| is_add=0) |
| l_startaddr = l_stopaddr = self.tra_if.local_addr_n[addr_type] |
| r_startaddr = r_stopaddr = self.tra_if.remote_addr_n[addr_type] |
| self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, priority=10, policy=3, |
| is_outbound=0, is_ipv6=is_ipv6, |
| is_add=0) |
| self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, scapy_tra_sa_id, |
| l_startaddr, l_stopaddr, r_startaddr, |
| r_stopaddr, priority=10, policy=3, |
| is_ipv6=is_ipv6, |
| is_add=0) |
| self.vapi.ipsec_sad_add_del_entry(scapy_tra_sa_id, scapy_tra_spi, |
| auth_algo_vpp_id, auth_key, |
| crypt_algo_vpp_id, crypt_key, |
| self.vpp_esp_protocol, is_tunnel=0, |
| use_anti_replay=1, |
| is_add=0) |
| self.vapi.ipsec_sad_add_del_entry(vpp_tra_sa_id, vpp_tra_spi, |
| auth_algo_vpp_id, auth_key, |
| crypt_algo_vpp_id, crypt_key, |
| self.vpp_esp_protocol, is_tunnel=0, |
| use_anti_replay=1, |
| is_add=0) |
| |
| |
| class TestIpsecEsp1(TemplateIpsecEsp, IpsecTraTests, IpsecTunTests): |
| """ Ipsec ESP - TUN & TRA tests """ |
| tra4_encrypt_node_name = "esp4-encrypt" |
| tra4_decrypt_node_name = "esp4-decrypt" |
| tra6_encrypt_node_name = "esp6-encrypt" |
| tra6_decrypt_node_name = "esp6-decrypt" |
| tun4_encrypt_node_name = "esp4-encrypt" |
| tun4_decrypt_node_name = "esp4-decrypt" |
| tun6_encrypt_node_name = "esp6-encrypt" |
| tun6_decrypt_node_name = "esp6-decrypt" |
| |
| |
| class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests): |
| """ Ipsec ESP - TCP tests """ |
| pass |
| |
| |
| if __name__ == '__main__': |
| unittest.main(testRunner=VppTestRunner) |