X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_snat.py;h=01bbe10aa3bcc2673c54de9dfa761a5861a322e1;hb=694396dc589b4fe75b1fad02fde1d3c3cdaeef04;hp=1abb3daaa40bd4bb49880e5e69b7dbfba0f7b3ae;hpb=eea28d78a3173341727aafee4c414bcb01001339;p=vpp.git diff --git a/test/test_snat.py b/test/test_snat.py index 1abb3daaa40..01bbe10aa3b 100644 --- a/test/test_snat.py +++ b/test/test_snat.py @@ -6,7 +6,8 @@ import struct from framework import VppTestCase, VppTestRunner from scapy.layers.inet import IP, TCP, UDP, ICMP -from scapy.layers.l2 import Ether +from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror +from scapy.layers.l2 import Ether, ARP from scapy.data import IP_PROTOS from util import ppp from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder @@ -41,11 +42,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() @@ -56,59 +65,61 @@ class TestSNAT(VppTestCase): super(TestSNAT, cls).tearDownClass() raise - def create_stream_in(self, in_if, out_if): + def create_stream_in(self, in_if, out_if, ttl=64): """ Create packet stream for inside network :param in_if: Inside interface :param out_if: Outside interface + :param ttl: TTL of generated packets """ pkts = [] # TCP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / + IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / TCP(sport=self.tcp_port_in)) pkts.append(p) # UDP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / + IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / UDP(sport=self.udp_port_in)) pkts.append(p) # ICMP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / + IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / ICMP(id=self.icmp_id_in, type='echo-request')) pkts.append(p) return pkts - def create_stream_out(self, out_if, dst_ip=None): + def create_stream_out(self, out_if, dst_ip=None, ttl=64): """ Create packet stream for outside network :param out_if: Outside interface :param dst_ip: Destination IP address (Default use global SNAT address) + :param ttl: TTL of generated packets """ if dst_ip is None: dst_ip = self.snat_addr pkts = [] # TCP p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / - IP(src=out_if.remote_ip4, dst=dst_ip) / + IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / TCP(dport=self.tcp_port_out)) pkts.append(p) # UDP p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / - IP(src=out_if.remote_ip4, dst=dst_ip) / + IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / UDP(dport=self.udp_port_out)) pkts.append(p) # ICMP p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / - IP(src=out_if.remote_ip4, dst=dst_ip) / + IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / ICMP(id=self.icmp_id_out, type='echo-reply')) pkts.append(p) @@ -178,6 +189,98 @@ 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_capture_out_with_icmp_errors(self, capture, src_ip=None, + packet_num=3, icmp_type=11): + """ + Verify captured packets with ICMP errors on outside network + + :param capture: Captured packets + :param src_ip: Translated IP address or IP address of VPP + (Default use global SNAT address) + :param packet_num: Expected number of packets (Default 3) + :param icmp_type: Type of error ICMP packet + we are expecting (Default 11) + """ + if src_ip is None: + src_ip = self.snat_addr + self.assertEqual(packet_num, len(capture)) + for packet in capture: + try: + self.assertEqual(packet[IP].src, src_ip) + self.assertTrue(packet.haslayer(ICMP)) + icmp = packet[ICMP] + self.assertEqual(icmp.type, icmp_type) + self.assertTrue(icmp.haslayer(IPerror)) + inner_ip = icmp[IPerror] + if inner_ip.haslayer(TCPerror): + self.assertEqual(inner_ip[TCPerror].dport, + self.tcp_port_out) + elif inner_ip.haslayer(UDPerror): + self.assertEqual(inner_ip[UDPerror].dport, + self.udp_port_out) + else: + self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out) + except: + self.logger.error(ppp("Unexpected or invalid packet " + "(outside network):", packet)) + raise + + def verify_capture_in_with_icmp_errors(self, capture, in_if, packet_num=3, + icmp_type=11): + """ + Verify captured packets with ICMP errors on inside network + + :param capture: Captured packets + :param in_if: Inside interface + :param packet_num: Expected number of packets (Default 3) + :param icmp_type: Type of error ICMP packet + we are expecting (Default 11) + """ + self.assertEqual(packet_num, len(capture)) + for packet in capture: + try: + self.assertEqual(packet[IP].dst, in_if.remote_ip4) + self.assertTrue(packet.haslayer(ICMP)) + icmp = packet[ICMP] + self.assertEqual(icmp.type, icmp_type) + self.assertTrue(icmp.haslayer(IPerror)) + inner_ip = icmp[IPerror] + if inner_ip.haslayer(TCPerror): + self.assertEqual(inner_ip[TCPerror].sport, + self.tcp_port_in) + elif inner_ip.haslayer(UDPerror): + self.assertEqual(inner_ip[UDPerror].sport, + self.udp_port_in) + else: + self.assertEqual(inner_ip[ICMPerror].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 @@ -238,6 +341,9 @@ class TestSNAT(VppTestCase): """ Clear SNAT configuration. """ + if self.pg7.has_ip4_config: + self.pg7.unconfig_ip4() + interfaces = self.vapi.snat_interface_addr_dump() for intf in interfaces: self.vapi.snat_add_interface_addr(intf.sw_if_index, is_add=0) @@ -258,6 +364,7 @@ class TestSNAT(VppTestCase): external_port=sm.external_port, addr_only=sm.addr_only, vrf_id=sm.vrf_id, + protocol=sm.protocol, is_add=0) adresses = self.vapi.snat_address_dump() @@ -266,8 +373,10 @@ class TestSNAT(VppTestCase): addr.ip_address, is_add=0) - def snat_add_static_mapping(self, local_ip, external_ip, local_port=0, - external_port=0, vrf_id=0, is_add=1): + def snat_add_static_mapping(self, local_ip, external_ip='0.0.0.0', + local_port=0, external_port=0, vrf_id=0, + is_add=1, external_sw_if_index=0xFFFFFFFF, + proto=0): """ Add/delete S-NAT static mapping @@ -277,6 +386,8 @@ class TestSNAT(VppTestCase): :param external_port: External port number (Optional) :param vrf_id: VRF ID (Default 0) :param is_add: 1 if add, 0 if delete (Default add) + :param external_sw_if_index: External interface instead of IP address + :param proto: IP protocol (Mandatory if port specified) """ addr_only = 1 if local_port and external_port: @@ -286,10 +397,12 @@ class TestSNAT(VppTestCase): self.vapi.snat_add_static_mapping( l_ip, e_ip, + external_sw_if_index, local_port, external_port, addr_only, vrf_id, + proto, is_add) def snat_add_address(self, ip, is_add=1): @@ -326,6 +439,141 @@ class TestSNAT(VppTestCase): capture = self.pg0.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg0) + def test_dynamic_icmp_errors_in2out_ttl_1(self): + """ SNAT handling of client packets with TTL=1 """ + + 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, + is_inside=0) + + # Client side - generate traffic + pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Client side - verify ICMP type 11 packets + capture = self.pg0.get_capture(len(pkts)) + self.verify_capture_in_with_icmp_errors(capture, self.pg0) + + def test_dynamic_icmp_errors_out2in_ttl_1(self): + """ SNAT handling of server packets with TTL=1 """ + + 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, + is_inside=0) + + # Client side - create sessions + pkts = self.create_stream_in(self.pg0, self.pg1) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - generate traffic + capture = self.pg1.get_capture(len(pkts)) + self.verify_capture_out(capture) + pkts = self.create_stream_out(self.pg1, ttl=1) + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - verify ICMP type 11 packets + capture = self.pg1.get_capture(len(pkts)) + self.verify_capture_out_with_icmp_errors(capture, + src_ip=self.pg1.local_ip4) + + def test_dynamic_icmp_errors_in2out_ttl_2(self): + """ SNAT handling of error responses to client packets with TTL=2 """ + + 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, + is_inside=0) + + # Client side - generate traffic + pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - simulate ICMP type 11 response + capture = self.pg1.get_capture(len(pkts)) + pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IP(src=self.pg1.remote_ip4, dst=self.snat_addr) / + ICMP(type=11) / packet[IP] for packet in capture] + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Client side - verify ICMP type 11 packets + capture = self.pg0.get_capture(len(pkts)) + self.verify_capture_in_with_icmp_errors(capture, self.pg0) + + def test_dynamic_icmp_errors_out2in_ttl_2(self): + """ SNAT handling of error responses to server packets with TTL=2 """ + + 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, + is_inside=0) + + # Client side - create sessions + pkts = self.create_stream_in(self.pg0, self.pg1) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - generate traffic + capture = self.pg1.get_capture(len(pkts)) + self.verify_capture_out(capture) + pkts = self.create_stream_out(self.pg1, ttl=2) + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Client side - simulate ICMP type 11 response + capture = self.pg0.get_capture(len(pkts)) + pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / + ICMP(type=11) / packet[IP] for packet in capture] + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - verify ICMP type 11 packets + capture = self.pg1.get_capture(len(pkts)) + self.verify_capture_out_with_icmp_errors(capture) + + def test_ping_out_interface_from_outside(self): + """ Ping SNAT out interface from outside """ + + 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, + is_inside=0) + + p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) / + ICMP(id=self.icmp_id_out, type='echo-request')) + pkts = [p] + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(len(pkts)) + self.assertEqual(1, len(capture)) + packet = capture[0] + try: + self.assertEqual(packet[IP].src, self.pg1.local_ip4) + self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) + self.assertEqual(packet[ICMP].id, self.icmp_id_in) + self.assertEqual(packet[ICMP].type, 0) # echo reply + except: + self.logger.error(ppp("Unexpected or invalid packet " + "(outside network):", packet)) + raise + def test_static_in(self): """ SNAT 1:1 NAT initialized from inside network """ @@ -393,11 +641,14 @@ class TestSNAT(VppTestCase): self.snat_add_address(self.snat_addr) self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr, - self.tcp_port_in, self.tcp_port_out) + self.tcp_port_in, self.tcp_port_out, + proto=IP_PROTOS.tcp) self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr, - self.udp_port_in, self.udp_port_out) + self.udp_port_in, self.udp_port_out, + proto=IP_PROTOS.udp) self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr, - self.icmp_id_in, self.icmp_id_out) + self.icmp_id_in, self.icmp_id_out, + proto=IP_PROTOS.icmp) self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) @@ -427,11 +678,14 @@ class TestSNAT(VppTestCase): self.snat_add_address(self.snat_addr) self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr, - self.tcp_port_in, self.tcp_port_out) + self.tcp_port_in, self.tcp_port_out, + proto=IP_PROTOS.tcp) self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr, - self.udp_port_in, self.udp_port_out) + self.udp_port_in, self.udp_port_out, + proto=IP_PROTOS.udp) self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr, - self.icmp_id_in, self.icmp_id_out) + self.icmp_id_in, self.icmp_id_out, + proto=IP_PROTOS.icmp) self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) @@ -462,9 +716,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) @@ -487,17 +741,30 @@ class TestSNAT(VppTestCase): self.pg3.assert_nothing_captured() def test_multiple_inside_interfaces(self): - """ - SNAT multiple inside interfaces with non-overlapping address space - """ + """ SNAT multiple inside interfaces (non-overlapping address space) """ 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 +797,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 +876,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() @@ -620,7 +902,8 @@ class TestSNAT(VppTestCase): is_inside=0) # add static mapping for server self.snat_add_static_mapping(server.ip4, self.snat_addr, - server_in_port, server_out_port) + server_in_port, server_out_port, + proto=IP_PROTOS.tcp) # send packet from host to server p = (Ether(src=host.mac, dst=self.pg0.local_mac) / @@ -701,12 +984,38 @@ class TestSNAT(VppTestCase): self.pg7.config_ip4() adresses = self.vapi.snat_address_dump() self.assertEqual(1, len(adresses)) + self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n) # remove interface address and check NAT address pool self.pg7.unconfig_ip4() adresses = self.vapi.snat_address_dump() self.assertEqual(0, len(adresses)) + def test_interface_addr_static_mapping(self): + """ Static mapping with addresses from interface """ + self.vapi.snat_add_interface_addr(self.pg7.sw_if_index) + self.snat_add_static_mapping('1.2.3.4', + external_sw_if_index=self.pg7.sw_if_index) + + # static mappings with external interface + static_mappings = self.vapi.snat_static_mapping_dump() + self.assertEqual(1, len(static_mappings)) + self.assertEqual(self.pg7.sw_if_index, + static_mappings[0].external_sw_if_index) + + # configure interface address and check static mappings + self.pg7.config_ip4() + static_mappings = self.vapi.snat_static_mapping_dump() + self.assertEqual(1, len(static_mappings)) + self.assertEqual(static_mappings[0].external_ip_address[0:4], + self.pg7.local_ip4n) + self.assertEqual(0xFFFFFFFF, static_mappings[0].external_sw_if_index) + + # remove interface address and check static mappings + self.pg7.unconfig_ip4() + static_mappings = self.vapi.snat_static_mapping_dump() + self.assertEqual(0, len(static_mappings)) + def test_ipfix_nat44_sess(self): """ S-NAT IPFIX logging NAT44 session created/delted """ self.snat_add_address(self.snat_addr) @@ -772,6 +1081,67 @@ class TestSNAT(VppTestCase): data = ipfix.decode_data_set(p.getlayer(Set)) self.verify_ipfix_addr_exhausted(data) + def test_pool_addr_fib(self): + """ S-NAT add pool addresses to FIB """ + static_addr = '10.0.0.10' + 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, + is_inside=0) + self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr) + + # SNAT address + p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') / + ARP(op=ARP.who_has, pdst=self.snat_addr, + psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(1) + self.assertTrue(capture[0].haslayer(ARP)) + self.assertTrue(capture[0][ARP].op, ARP.is_at) + + # 1:1 NAT address + p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') / + ARP(op=ARP.who_has, pdst=static_addr, + psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(1) + self.assertTrue(capture[0].haslayer(ARP)) + self.assertTrue(capture[0][ARP].op, ARP.is_at) + + # send ARP to non-SNAT interface + p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') / + ARP(op=ARP.who_has, pdst=self.snat_addr, + psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac)) + self.pg2.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(0) + + # remove addresses and verify + self.snat_add_address(self.snat_addr, is_add=0) + self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr, + is_add=0) + + p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') / + ARP(op=ARP.who_has, pdst=self.snat_addr, + psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(0) + + p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') / + ARP(op=ARP.who_has, pdst=static_addr, + psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(0) + def tearDown(self): super(TestSNAT, self).tearDown() if not self.vpp_dead: