diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/test_snat.py | 111 |
1 files changed, 86 insertions, 25 deletions
diff --git a/test/test_snat.py b/test/test_snat.py index 1abb3daaa40..09fdb108c0b 100644 --- a/test/test_snat.py +++ b/test/test_snat.py @@ -41,11 +41,19 @@ class TestSNAT(VppTestCase): cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7])) + cls.pg4._local_ip4 = "172.16.255.1" + cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4) + cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2" + cls.pg4.set_table_ip4(10) + cls.pg5._local_ip4 = "172.16.255.3" + cls.pg5._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4) + cls.pg5._remote_hosts[0]._ip4 = "172.16.255.4" + cls.pg5.set_table_ip4(10) + cls.pg6._local_ip4 = "172.16.255.1" + cls.pg6._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4) + cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2" + cls.pg6.set_table_ip4(20) for i in cls.overlapping_interfaces: - i._local_ip4 = "172.16.255.1" - i._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4) - i._remote_hosts[0]._ip4 = "172.16.255.2" - i.set_table_ip4(i.sw_if_index) i.config_ip4() i.admin_up() i.resolve_arp() @@ -178,6 +186,29 @@ class TestSNAT(VppTestCase): "(inside network):", packet)) raise + def verify_capture_no_translation(self, capture, ingress_if, egress_if): + """ + Verify captured packet that don't have to be translated + + :param capture: Captured packets + :param ingress_if: Ingress interface + :param egress_if: Egress interface + """ + for packet in capture: + try: + self.assertEqual(packet[IP].src, ingress_if.remote_ip4) + self.assertEqual(packet[IP].dst, egress_if.remote_ip4) + if packet.haslayer(TCP): + self.assertEqual(packet[TCP].sport, self.tcp_port_in) + elif packet.haslayer(UDP): + self.assertEqual(packet[UDP].sport, self.udp_port_in) + else: + self.assertEqual(packet[ICMP].id, self.icmp_id_in) + except: + self.logger.error(ppp("Unexpected or invalid packet " + "(inside network):", packet)) + raise + def verify_ipfix_nat44_ses(self, data): """ Verify IPFIX NAT44 session create/delete event @@ -462,9 +493,9 @@ class TestSNAT(VppTestCase): self.icmp_id_out = 6305 self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1, - vrf_id=self.pg4.sw_if_index) + vrf_id=10) self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2, - vrf_id=self.pg4.sw_if_index) + vrf_id=10) self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index, is_inside=0) self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) @@ -494,10 +525,25 @@ class TestSNAT(VppTestCase): self.snat_add_address(self.snat_addr) self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index) - self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index) self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index, is_inside=0) + # between two S-NAT inside interfaces (no translation) + pkts = self.create_stream_in(self.pg0, self.pg1) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(len(pkts)) + self.verify_capture_no_translation(capture, self.pg0, self.pg1) + + # from S-NAT inside to interface without S-NAT feature (no translation) + pkts = self.create_stream_in(self.pg0, self.pg2) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg2.get_capture(len(pkts)) + self.verify_capture_no_translation(capture, self.pg0, self.pg2) + # in2out 1st interface pkts = self.create_stream_in(self.pg0, self.pg3) self.pg0.add_stream(pkts) @@ -530,31 +576,46 @@ class TestSNAT(VppTestCase): capture = self.pg1.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg1) - # in2out 3rd interface - pkts = self.create_stream_in(self.pg2, self.pg3) - self.pg2.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg3.get_capture(len(pkts)) - self.verify_capture_out(capture) - - # out2in 3rd interface - pkts = self.create_stream_out(self.pg3) - self.pg3.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg2.get_capture(len(pkts)) - self.verify_capture_in(capture, self.pg2) - def test_inside_overlapping_interfaces(self): """ SNAT multiple inside interfaces with overlapping address space """ + static_nat_ip = "10.0.0.10" self.snat_add_address(self.snat_addr) self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index, is_inside=0) self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index) self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index) self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index) + self.snat_add_static_mapping(self.pg6.remote_ip4, static_nat_ip, + vrf_id=20) + + # between S-NAT inside interfaces with same VRF (no translation) + pkts = self.create_stream_in(self.pg4, self.pg5) + self.pg4.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg5.get_capture(len(pkts)) + self.verify_capture_no_translation(capture, self.pg4, self.pg5) + + # between S-NAT inside interfaces with different VRF (hairpinning) + p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) / + IP(src=self.pg4.remote_ip4, dst=static_nat_ip) / + TCP(sport=1234, dport=5678)) + self.pg4.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg6.get_capture(1) + p = capture[0] + try: + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.src, self.snat_addr) + self.assertEqual(ip.dst, self.pg6.remote_ip4) + self.assertNotEqual(tcp.sport, 1234) + self.assertEqual(tcp.dport, 5678) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise # in2out 1st interface pkts = self.create_stream_in(self.pg4, self.pg3) @@ -594,10 +655,10 @@ class TestSNAT(VppTestCase): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg3.get_capture(len(pkts)) - self.verify_capture_out(capture) + self.verify_capture_out(capture, static_nat_ip, True) # out2in 3rd interface - pkts = self.create_stream_out(self.pg3) + pkts = self.create_stream_out(self.pg3, static_nat_ip) self.pg3.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() |