X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_nat.py;h=4470a054bed21250c2fc02be171a04d68affa5b8;hb=d042194b4bb0512a81924068013ed4e48f021a2b;hp=6aff932c024c557138b2bdae004237aaf941176d;hpb=9dba7816f6c3b5e4c879f839bb4bc61dcdbde757;p=vpp.git diff --git a/test/test_nat.py b/test/test_nat.py index 6aff932c024..4470a054bed 100644 --- a/test/test_nat.py +++ b/test/test_nat.py @@ -1031,6 +1031,7 @@ class TestNAT44(MethodHolder): twice_nat=sm.twice_nat, out2in_only=sm.out2in_only, tag=sm.tag, + external_sw_if_index=sm.external_sw_if_index, is_add=0) lb_static_mappings = self.vapi.nat44_lb_static_mapping_dump() @@ -1676,6 +1677,37 @@ class TestNAT44(MethodHolder): self.pg_start() self.pg3.assert_nothing_captured() + def test_dynamic_to_static(self): + """ Switch from dynamic translation to 1:1NAT """ + nat_ip = "10.0.0.10" + self.tcp_port_out = 6303 + self.udp_port_out = 6304 + self.icmp_id_out = 6305 + + self.nat44_add_address(self.nat_addr) + self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, + is_inside=0) + + # dynamic + 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_out(capture) + + # 1:1NAT + self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip) + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0) + self.assertEqual(len(sessions), 0) + 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_out(capture, nat_ip, True) + def test_identity_nat(self): """ Identity NAT """ @@ -1775,10 +1807,36 @@ class TestNAT44(MethodHolder): self.logger.error(ppp("Unexpected or invalid packet:", p)) raise - # multiple clients + @unittest.skipUnless(running_extended_tests(), "part of extended tests") + def test_static_lb_multi_clients(self): + """ NAT44 local service load balancing - multiple clients""" + + external_addr_n = socket.inet_pton(socket.AF_INET, self.nat_addr) + external_port = 80 + local_port = 8080 + server1 = self.pg0.remote_hosts[0] + server2 = self.pg0.remote_hosts[1] + + locals = [{'addr': server1.ip4n, + 'port': local_port, + 'probability': 90}, + {'addr': server2.ip4n, + 'port': local_port, + 'probability': 10}] + + self.nat44_add_address(self.nat_addr) + self.vapi.nat44_add_del_lb_static_mapping(external_addr_n, + external_port, + IP_PROTOS.tcp, + local_num=len(locals), + locals=locals) + self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, + is_inside=0) + server1_n = 0 server2_n = 0 - clients = ip4_range(self.pg1.remote_ip4, 10, 20) + clients = ip4_range(self.pg1.remote_ip4, 10, 50) pkts = [] for client in clients: p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / @@ -2406,15 +2464,44 @@ class TestNAT44(MethodHolder): # configure interface address and check static mappings self.pg7.config_ip4() static_mappings = self.vapi.nat44_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) - self.assertEqual((static_mappings[0].tag).split('\0', 1)[0], tag) + self.assertEqual(2, len(static_mappings)) + resolved = False + for sm in static_mappings: + if sm.external_sw_if_index == 0xFFFFFFFF: + self.assertEqual(sm.external_ip_address[0:4], + self.pg7.local_ip4n) + self.assertEqual((sm.tag).split('\0', 1)[0], tag) + resolved = True + self.assertTrue(resolved) # remove interface address and check static mappings self.pg7.unconfig_ip4() static_mappings = self.vapi.nat44_static_mapping_dump() + self.assertEqual(1, len(static_mappings)) + self.assertEqual(self.pg7.sw_if_index, + static_mappings[0].external_sw_if_index) + self.assertEqual((static_mappings[0].tag).split('\0', 1)[0], tag) + + # configure interface address again and check static mappings + self.pg7.config_ip4() + static_mappings = self.vapi.nat44_static_mapping_dump() + self.assertEqual(2, len(static_mappings)) + resolved = False + for sm in static_mappings: + if sm.external_sw_if_index == 0xFFFFFFFF: + self.assertEqual(sm.external_ip_address[0:4], + self.pg7.local_ip4n) + self.assertEqual((sm.tag).split('\0', 1)[0], tag) + resolved = True + self.assertTrue(resolved) + + # remove static mapping + self.nat44_add_static_mapping( + '1.2.3.4', + external_sw_if_index=self.pg7.sw_if_index, + tag=tag, + is_add=0) + static_mappings = self.vapi.nat44_static_mapping_dump() self.assertEqual(0, len(static_mappings)) def test_interface_addr_identity_nat(self): @@ -2437,17 +2524,23 @@ class TestNAT44(MethodHolder): # configure interface address and check identity mappings self.pg7.config_ip4() identity_mappings = self.vapi.nat44_identity_mapping_dump() - self.assertEqual(1, len(identity_mappings)) - self.assertEqual(identity_mappings[0].ip_address, - self.pg7.local_ip4n) - self.assertEqual(0xFFFFFFFF, identity_mappings[0].sw_if_index) - self.assertEqual(port, identity_mappings[0].port) - self.assertEqual(IP_PROTOS.tcp, identity_mappings[0].protocol) + resolved = False + self.assertEqual(2, len(identity_mappings)) + for sm in identity_mappings: + if sm.sw_if_index == 0xFFFFFFFF: + self.assertEqual(identity_mappings[0].ip_address, + self.pg7.local_ip4n) + self.assertEqual(port, identity_mappings[0].port) + self.assertEqual(IP_PROTOS.tcp, identity_mappings[0].protocol) + resolved = True + self.assertTrue(resolved) # remove interface address and check identity mappings self.pg7.unconfig_ip4() identity_mappings = self.vapi.nat44_identity_mapping_dump() - self.assertEqual(0, len(identity_mappings)) + self.assertEqual(1, len(identity_mappings)) + self.assertEqual(self.pg7.sw_if_index, + identity_mappings[0].sw_if_index) def test_ipfix_nat44_sess(self): """ IPFIX logging NAT44 session created/delted """ @@ -3239,6 +3332,187 @@ class TestNAT44(MethodHolder): self.logger.error(ppp("Unexpected or invalid packet:", p)) raise + def test_output_feature_and_service(self): + """ NAT44 interface output feature and services """ + external_addr = '1.2.3.4' + external_port = 80 + local_port = 8080 + + self.vapi.nat44_forwarding_enable_disable(1) + self.nat44_add_address(self.nat_addr) + self.nat44_add_static_mapping(self.pg0.remote_ip4, external_addr, + local_port, external_port, + proto=IP_PROTOS.tcp, out2in_only=1) + self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index, + is_inside=0) + + # from client to service + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=external_addr) / + TCP(sport=12345, dport=external_port)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg0.get_capture(1) + p = capture[0] + server = None + try: + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.dst, self.pg0.remote_ip4) + self.assertEqual(tcp.dport, local_port) + self.check_tcp_checksum(p) + self.check_ip_checksum(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + # from service back to client + p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / + TCP(sport=local_port, dport=12345)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(1) + p = capture[0] + try: + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.src, external_addr) + self.assertEqual(tcp.sport, external_port) + self.check_tcp_checksum(p) + self.check_ip_checksum(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + # from local network host to external network + 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_out(capture) + 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_out(capture) + + # from external network back to local network host + pkts = self.create_stream_out(self.pg1) + 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_in(capture, self.pg0) + + def test_output_feature_and_service2(self): + """ NAT44 interface output feature and service host direct access """ + self.vapi.nat44_forwarding_enable_disable(1) + self.nat44_add_address(self.nat_addr) + self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index, + is_inside=0) + + # session initiaded from service host - translate + 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_out(capture) + + pkts = self.create_stream_out(self.pg1) + 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_in(capture, self.pg0) + + tcp_port_out = self.tcp_port_out + udp_port_out = self.udp_port_out + icmp_id_out = self.icmp_id_out + + # session initiaded from remote host - do not translate + pkts = self.create_stream_out(self.pg1, + self.pg0.remote_ip4, + use_inside_ports=True) + 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_in(capture, self.pg0) + + 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_out(capture, nat_ip=self.pg0.remote_ip4, + same_port=True) + + def test_output_feature_and_service3(self): + """ NAT44 interface output feature and DST NAT """ + external_addr = '1.2.3.4' + external_port = 80 + local_port = 8080 + + self.vapi.nat44_forwarding_enable_disable(1) + self.nat44_add_address(self.nat_addr) + self.nat44_add_static_mapping(self.pg1.remote_ip4, external_addr, + local_port, external_port, + proto=IP_PROTOS.tcp, out2in_only=1) + self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index, + is_inside=0) + self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index, + is_inside=0) + + p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / + IP(src=self.pg0.remote_ip4, dst=external_addr) / + TCP(sport=12345, dport=external_port)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(1) + p = capture[0] + try: + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.src, self.pg0.remote_ip4) + self.assertEqual(tcp.sport, 12345) + self.assertEqual(ip.dst, self.pg1.remote_ip4) + self.assertEqual(tcp.dport, local_port) + self.check_tcp_checksum(p) + self.check_ip_checksum(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) / + TCP(sport=local_port, dport=12345)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg0.get_capture(1) + p = capture[0] + try: + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.src, external_addr) + self.assertEqual(tcp.sport, external_port) + self.assertEqual(ip.dst, self.pg0.remote_ip4) + self.assertEqual(tcp.dport, 12345) + self.check_tcp_checksum(p) + self.check_ip_checksum(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + def test_one_armed_nat44(self): """ One armed NAT44 """ remote_host = self.pg9.remote_hosts[0] @@ -6055,5 +6329,136 @@ class TestDSliteCE(MethodHolder): self.logger.info( self.vapi.cli("show dslite b4-tunnel-endpoint-address")) + +class TestNAT66(MethodHolder): + """ NAT66 Test Cases """ + + @classmethod + def setUpClass(cls): + super(TestNAT66, cls).setUpClass() + + try: + cls.nat_addr = 'fd01:ff::2' + cls.nat_addr_n = socket.inet_pton(socket.AF_INET6, cls.nat_addr) + + cls.create_pg_interfaces(range(2)) + cls.interfaces = list(cls.pg_interfaces) + + for i in cls.interfaces: + i.admin_up() + i.config_ip6() + i.configure_ipv6_neighbors() + + except Exception: + super(TestNAT66, cls).tearDownClass() + raise + + def test_static(self): + """ 1:1 NAT66 test """ + self.vapi.nat66_add_del_interface(self.pg0.sw_if_index) + self.vapi.nat66_add_del_interface(self.pg1.sw_if_index, is_inside=0) + self.vapi.nat66_add_del_static_mapping(self.pg0.remote_ip6n, + self.nat_addr_n) + + # in2out + pkts = [] + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) / + TCP()) + pkts.append(p) + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) / + UDP()) + pkts.append(p) + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) / + ICMPv6EchoRequest()) + pkts.append(p) + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) / + GRE() / IP() / TCP()) + pkts.append(p) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(len(pkts)) + for packet in capture: + try: + self.assertEqual(packet[IPv6].src, self.nat_addr) + self.assertEqual(packet[IPv6].dst, self.pg1.remote_ip6) + if packet.haslayer(TCP): + self.check_tcp_checksum(packet) + elif packet.haslayer(UDP): + self.check_udp_checksum(packet) + elif packet.haslayer(ICMPv6EchoRequest): + self.check_icmpv6_checksum(packet) + except: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + + # out2in + pkts = [] + p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) / + TCP()) + pkts.append(p) + p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) / + UDP()) + pkts.append(p) + p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) / + ICMPv6EchoReply()) + pkts.append(p) + p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) / + GRE() / IP() / TCP()) + pkts.append(p) + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg0.get_capture(len(pkts)) + for packet in capture: + try: + self.assertEqual(packet[IPv6].src, self.pg1.remote_ip6) + self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6) + if packet.haslayer(TCP): + self.check_tcp_checksum(packet) + elif packet.haslayer(UDP): + self.check_udp_checksum(packet) + elif packet.haslayer(ICMPv6EchoReply): + self.check_icmpv6_checksum(packet) + except: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + + sm = self.vapi.nat66_static_mapping_dump() + self.assertEqual(len(sm), 1) + self.assertEqual(sm[0].total_pkts, 8) + + def clear_nat66(self): + """ + Clear NAT66 configuration. + """ + interfaces = self.vapi.nat66_interface_dump() + for intf in interfaces: + self.vapi.nat66_add_del_interface(intf.sw_if_index, + intf.is_inside, + is_add=0) + + static_mappings = self.vapi.nat66_static_mapping_dump() + for sm in static_mappings: + self.vapi.nat66_add_del_static_mapping(sm.local_ip_address, + sm.external_ip_address, + sm.vrf_id, + is_add=0) + + def tearDown(self): + super(TestNAT66, self).tearDown() + if not self.vpp_dead: + self.logger.info(self.vapi.cli("show nat66 interfaces")) + self.logger.info(self.vapi.cli("show nat66 static mappings")) + self.clear_nat66() + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)