diff options
| -rw-r--r-- | test/test_acl_plugin.py | 2 | ||||
| -rw-r--r-- | test/test_acl_plugin_macip.py | 8 | ||||
| -rw-r--r-- | test/test_bfd.py | 4 | ||||
| -rw-r--r-- | test/test_ipsec_nat.py | 148 | ||||
| -rw-r--r-- | test/vpp_sub_interface.py | 8 |
5 files changed, 84 insertions, 86 deletions
diff --git a/test/test_acl_plugin.py b/test/test_acl_plugin.py index b2d97565d56..f3ab3812716 100644 --- a/test/test_acl_plugin.py +++ b/test/test_acl_plugin.py @@ -1423,7 +1423,7 @@ class TestACLplugin(VppTestCase): # create an interface intf = [] - intf.append(VppLoInterface(self, 0)) + intf.append(VppLoInterface(self)) # Apply rules self.apply_rules_to(rules, "permit ipv4 tcp", intf[0].sw_if_index) diff --git a/test/test_acl_plugin_macip.py b/test/test_acl_plugin_macip.py index 8bcef25c49a..f35db55fb7e 100644 --- a/test/test_acl_plugin_macip.py +++ b/test/test_acl_plugin_macip.py @@ -1080,8 +1080,8 @@ class TestMACIP(MethodHolder): self.apply_macip_rules(self.create_rules(acl_count=3, rules_count=[3, 5, 4])) - intf.append(VppLoInterface(self, 0)) - intf.append(VppLoInterface(self, 1)) + intf.append(VppLoInterface(self)) + intf.append(VppLoInterface(self)) sw_if_index0 = intf[0].sw_if_index self.vapi.macip_acl_interface_add_del(sw_if_index0, 1) @@ -1103,8 +1103,8 @@ class TestMACIP(MethodHolder): self.assertEqual(reply.acls[sw_if_index0], 4294967295) self.assertEqual(reply.acls[sw_if_index1], 0) - intf.append(VppLoInterface(self, 2)) - intf.append(VppLoInterface(self, 3)) + intf.append(VppLoInterface(self)) + intf.append(VppLoInterface(self)) sw_if_index2 = intf[2].sw_if_index sw_if_index3 = intf[3].sw_if_index self.vapi.macip_acl_interface_add_del(sw_if_index2, 1) diff --git a/test/test_bfd.py b/test/test_bfd.py index 3afe9428025..b3ac6356d2f 100644 --- a/test/test_bfd.py +++ b/test/test_bfd.py @@ -1407,7 +1407,7 @@ class BFD4TestCase(VppTestCase): def test_intf_deleted(self): """ interface with bfd session deleted """ - intf = VppLoInterface(self, 0) + intf = VppLoInterface(self) intf.config_ip4() intf.admin_up() sw_if_index = intf.sw_if_index @@ -1616,7 +1616,7 @@ class BFD6TestCase(VppTestCase): def test_intf_deleted(self): """ interface with bfd session deleted """ - intf = VppLoInterface(self, 0) + intf = VppLoInterface(self) intf.config_ip6() intf.admin_up() sw_if_index = intf.sw_if_index diff --git a/test/test_ipsec_nat.py b/test/test_ipsec_nat.py index 7a5aca6bb7b..6d9dc109add 100644 --- a/test/test_ipsec_nat.py +++ b/test/test_ipsec_nat.py @@ -6,10 +6,10 @@ 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 framework import VppTestCase +from template_ipsec import TemplateIpsec -class IPSecNATTestCase(VppTestCase): +class IPSecNATTestCase(TemplateIpsec): """ IPSec/NAT TRANSPORT MODE: @@ -31,26 +31,20 @@ class IPSecNATTestCase(VppTestCase): --- --- --- """ - remote_pg0_client_addr = '1.1.1.1' - @classmethod def setUpClass(cls): super(IPSecNATTestCase, cls).setUpClass() - cls.create_pg_interfaces(range(2)) - for i in cls.pg_interfaces: - i.configure_ipv4_neighbors() - i.admin_up() - i.config_ip4() - i.resolve_arp() - cls.tcp_port_in = 6303 cls.tcp_port_out = 6303 cls.udp_port_in = 6304 cls.udp_port_out = 6304 cls.icmp_id_in = 6305 cls.icmp_id_out = 6305 + cls.tun_if = cls.pg0 cls.config_esp_tun() cls.logger.info(cls.vapi.ppcli("show ipsec")) + client = socket.inet_pton(socket.AF_INET, cls.remote_tun_if_host) + cls.vapi.ip_add_del_route(client, 32, cls.pg0.remote_ip4n) def create_stream_plain(self, src_mac, dst_mac, src_ip, dst_ip): return [ @@ -88,7 +82,7 @@ class IPSecNATTestCase(VppTestCase): for packet in capture: try: self.assert_packet_checksums_valid(packet) - self.assert_equal(packet[IP].src, self.pg0.remote_ip4, + self.assert_equal(packet[IP].src, self.tun_if.remote_ip4, "decrypted packet source address") self.assert_equal(packet[IP].dst, self.pg1.remote_ip4, "decrypted packet destination address") @@ -123,7 +117,7 @@ class IPSecNATTestCase(VppTestCase): decrypt_pkt = sa.decrypt(packet[IP]) self.assert_equal(decrypt_pkt[IP].src, self.pg1.remote_ip4, "encrypted packet source address") - self.assert_equal(decrypt_pkt[IP].dst, self.pg0.remote_ip4, + self.assert_equal(decrypt_pkt[IP].dst, self.tun_if.remote_ip4, "encrypted packet destination address") # if decrypt_pkt.haslayer(TCP): # self.tcp_port_out = decrypt_pkt[TCP].sport @@ -138,101 +132,99 @@ class IPSecNATTestCase(VppTestCase): @classmethod def config_esp_tun(cls): - spd_id = 1 - remote_sa_id = 10 - local_sa_id = 20 - scapy_tun_spi = 1001 - vpp_tun_spi = 1000 - client = socket.inet_pton(socket.AF_INET, cls.remote_pg0_client_addr) - cls.vapi.ip_add_del_route(client, 32, cls.pg0.remote_ip4n) - cls.vapi.ipsec_sad_add_del_entry(remote_sa_id, scapy_tun_spi, + cls.vapi.ipsec_sad_add_del_entry(cls.scapy_tun_sa_id, + cls.scapy_tun_spi, + cls.auth_algo_vpp_id, cls.auth_key, + cls.crypt_algo_vpp_id, + cls.crypt_key, cls.vpp_esp_protocol, cls.pg1.remote_ip4n, - cls.pg0.remote_ip4n, - integrity_key_length=20, - crypto_key_length=16, - protocol=1, udp_encap=1) - cls.vapi.ipsec_sad_add_del_entry(local_sa_id, vpp_tun_spi, - cls.pg0.remote_ip4n, - cls.pg1.remote_ip4n, - integrity_key_length=20, - crypto_key_length=16, - protocol=1, udp_encap=1) - cls.vapi.ipsec_spd_add_del(spd_id) - cls.vapi.ipsec_interface_add_del_spd(spd_id, cls.pg0.sw_if_index) + cls.tun_if.remote_ip4n) + cls.vapi.ipsec_sad_add_del_entry(cls.vpp_tun_sa_id, + cls.vpp_tun_spi, + cls.auth_algo_vpp_id, cls.auth_key, + cls.crypt_algo_vpp_id, + cls.crypt_key, cls.vpp_esp_protocol, + cls.tun_if.remote_ip4n, + cls.pg1.remote_ip4n) + cls.vapi.ipsec_spd_add_del(cls.tun_spd_id) + cls.vapi.ipsec_interface_add_del_spd(cls.tun_spd_id, + cls.tun_if.sw_if_index) l_startaddr = r_startaddr = socket.inet_pton(socket.AF_INET, "0.0.0.0") l_stopaddr = r_stopaddr = socket.inet_pton(socket.AF_INET, "255.255.255.255") - cls.vapi.ipsec_spd_add_del_entry(spd_id, l_startaddr, l_stopaddr, - r_startaddr, r_stopaddr, + cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, cls.vpp_tun_sa_id, + l_startaddr, l_stopaddr, r_startaddr, + r_stopaddr, protocol=socket.IPPROTO_ESP) - cls.vapi.ipsec_spd_add_del_entry(spd_id, l_startaddr, l_stopaddr, - r_startaddr, r_stopaddr, - protocol=socket.IPPROTO_ESP, - is_outbound=0) - cls.vapi.ipsec_spd_add_del_entry(spd_id, l_startaddr, l_stopaddr, - r_startaddr, r_stopaddr, - remote_port_start=4500, + cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, cls.scapy_tun_sa_id, + l_startaddr, l_stopaddr, r_startaddr, + r_stopaddr, is_outbound=0, + protocol=socket.IPPROTO_ESP) + cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, cls.vpp_tun_sa_id, + l_startaddr, l_stopaddr, r_startaddr, + r_stopaddr, remote_port_start=4500, remote_port_stop=4500, protocol=socket.IPPROTO_UDP) - cls.vapi.ipsec_spd_add_del_entry(spd_id, l_startaddr, l_stopaddr, - r_startaddr, r_stopaddr, - remote_port_start=4500, + cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, cls.scapy_tun_sa_id, + l_startaddr, l_stopaddr, r_startaddr, + r_stopaddr, remote_port_start=4500, remote_port_stop=4500, protocol=socket.IPPROTO_UDP, is_outbound=0) - l_startaddr = l_stopaddr = cls.pg0.remote_ip4n + l_startaddr = l_stopaddr = cls.tun_if.remote_ip4n r_startaddr = r_stopaddr = cls.pg1.remote_ip4n - cls.vapi.ipsec_spd_add_del_entry(spd_id, l_startaddr, l_stopaddr, - r_startaddr, r_stopaddr, - priority=10, policy=3, - is_outbound=0, sa_id=local_sa_id) - cls.vapi.ipsec_spd_add_del_entry(spd_id, r_startaddr, r_stopaddr, - l_startaddr, l_stopaddr, - priority=10, policy=3, - sa_id=remote_sa_id) + cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, cls.vpp_tun_sa_id, + l_startaddr, l_stopaddr, r_startaddr, + r_stopaddr, priority=10, policy=3, + is_outbound=0) + cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, cls.scapy_tun_sa_id, + r_startaddr, r_stopaddr, l_startaddr, + l_stopaddr, priority=10, policy=3) def test_ipsec_nat_tun(self): """ IPSec/NAT tunnel test case """ - local_tun_sa = SecurityAssociation(ESP, spi=0x000003e9, - crypt_algo='AES-CBC', - crypt_key='JPjyOWBeVEQiMe7h', - auth_algo='HMAC-SHA1-96', - auth_key='C91KUR9GYMm5GfkEvNjX', + scapy_tun_sa = SecurityAssociation(ESP, + spi=self.scapy_tun_spi, + crypt_algo=self.crypt_algo, + crypt_key=self.crypt_key, + auth_algo=self.auth_algo, + auth_key=self.auth_key, tunnel_header=IP( src=self.pg1.remote_ip4, - dst=self.pg0.remote_ip4), + dst=self.tun_if.remote_ip4), nat_t_header=UDP( sport=4500, dport=4500)) # in2out - from private network to public pkts = self.create_stream_plain( self.pg1.remote_mac, self.pg1.local_mac, - self.pg1.remote_ip4, self.pg0.remote_ip4) + self.pg1.remote_ip4, self.tun_if.remote_ip4) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg0.get_capture(len(pkts)) - self.verify_capture_encrypted(capture, local_tun_sa) - - remote_tun_sa = SecurityAssociation(ESP, spi=0x000003e8, - crypt_algo='AES-CBC', - crypt_key='JPjyOWBeVEQiMe7h', - auth_algo='HMAC-SHA1-96', - auth_key='C91KUR9GYMm5GfkEvNjX', - tunnel_header=IP( - src=self.pg0.remote_ip4, - dst=self.pg1.remote_ip4), - nat_t_header=UDP( - sport=4500, - dport=4500)) + capture = self.tun_if.get_capture(len(pkts)) + self.verify_capture_encrypted(capture, scapy_tun_sa) + + vpp_tun_sa = SecurityAssociation(ESP, + spi=self.vpp_tun_spi, + crypt_algo=self.crypt_algo, + crypt_key=self.crypt_key, + auth_algo=self.auth_algo, + auth_key=self.auth_key, + tunnel_header=IP( + src=self.tun_if.remote_ip4, + dst=self.pg1.remote_ip4), + nat_t_header=UDP( + sport=4500, + dport=4500)) # out2in - from public network to private pkts = self.create_stream_encrypted( - self.pg0.remote_mac, self.pg0.local_mac, - self.pg0.remote_ip4, self.pg1.remote_ip4, remote_tun_sa) + self.tun_if.remote_mac, self.tun_if.local_mac, + self.tun_if.remote_ip4, self.pg1.remote_ip4, vpp_tun_sa) self.logger.info(ppc("Sending packets:", pkts)) - self.pg0.add_stream(pkts) + self.tun_if.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) diff --git a/test/vpp_sub_interface.py b/test/vpp_sub_interface.py index 3c726b29d3f..255cfffd827 100644 --- a/test/vpp_sub_interface.py +++ b/test/vpp_sub_interface.py @@ -1,8 +1,8 @@ from scapy.layers.l2 import Dot1Q from abc import abstractmethod, ABCMeta -from vpp_interface import VppInterface from vpp_pg_interface import VppPGInterface from vpp_papi_provider import L2_VTR_OP +from vpp_interface import VppInterface class VppSubInterface(VppPGInterface): @@ -50,6 +50,12 @@ class VppSubInterface(VppPGInterface): def create_ndp_req(self): pass + def resolve_arp(self): + super(VppSubInterface, self).resolve_arp(self.parent) + + def resolve_ndp(self): + super(VppSubInterface, self).resolve_ndp(self.parent) + @abstractmethod def add_dot1_layer(self, pkt): pass |