X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_nat.py;h=35e89e39b9cbfc102a6bedcc9a5c56f6b026960a;hb=235a47e371c8ffea352790c001ef39876f2aaef5;hp=c4018cf66d864c9dfc8760bffcc0abe7f381ff95;hpb=ebdf190a9c4a514329de7e5e9b9178c3af055122;p=vpp.git diff --git a/test/test_nat.py b/test/test_nat.py index c4018cf66d8..35e89e39b9c 100644 --- a/test/test_nat.py +++ b/test/test_nat.py @@ -7,7 +7,6 @@ import StringIO import random from framework import VppTestCase, VppTestRunner, running_extended_tests -from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto from scapy.layers.inet import IP, TCP, UDP, ICMP from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply @@ -26,113 +25,173 @@ from util import mactobinary class MethodHolder(VppTestCase): """ NAT create capture and verify method holder """ - @classmethod - def setUpClass(cls): - super(MethodHolder, cls).setUpClass() + def clear_nat44(self): + """ + Clear NAT44 configuration. + """ + if hasattr(self, 'pg7') and hasattr(self, 'pg8'): + # I found no elegant way to do this + self.vapi.ip_add_del_route( + dst_address=self.pg7.remote_ip4n, + dst_address_length=32, + next_hop_address=self.pg7.remote_ip4n, + next_hop_sw_if_index=self.pg7.sw_if_index, + is_add=0) + self.vapi.ip_add_del_route( + dst_address=self.pg8.remote_ip4n, + dst_address_length=32, + next_hop_address=self.pg8.remote_ip4n, + next_hop_sw_if_index=self.pg8.sw_if_index, + is_add=0) - def tearDown(self): - super(MethodHolder, self).tearDown() + for intf in [self.pg7, self.pg8]: + neighbors = self.vapi.ip_neighbor_dump(intf.sw_if_index) + for n in neighbors: + self.vapi.ip_neighbor_add_del(intf.sw_if_index, + n.mac_address, + n.ip_address, + is_add=0) - def check_ip_checksum(self, pkt): - """ - Check IP checksum of the packet + if self.pg7.has_ip4_config: + self.pg7.unconfig_ip4() - :param pkt: Packet to check IP checksum - """ - new = pkt.__class__(str(pkt)) - del new['IP'].chksum - new = new.__class__(str(new)) - self.assertEqual(new['IP'].chksum, pkt['IP'].chksum) + self.vapi.nat44_forwarding_enable_disable(0) - def check_tcp_checksum(self, pkt): - """ - Check TCP checksum in IP packet + interfaces = self.vapi.nat44_interface_addr_dump() + for intf in interfaces: + self.vapi.nat44_add_interface_addr(intf.sw_if_index, + twice_nat=intf.twice_nat, + is_add=0) - :param pkt: Packet to check TCP checksum - """ - new = pkt.__class__(str(pkt)) - del new['TCP'].chksum - new = new.__class__(str(new)) - self.assertEqual(new['TCP'].chksum, pkt['TCP'].chksum) + self.vapi.nat_ipfix(enable=0, src_port=self.ipfix_src_port, + domain_id=self.ipfix_domain_id) + self.ipfix_src_port = 4739 + self.ipfix_domain_id = 1 - def check_udp_checksum(self, pkt): - """ - Check UDP checksum in IP packet + interfaces = self.vapi.nat44_interface_dump() + for intf in interfaces: + if intf.is_inside > 1: + self.vapi.nat44_interface_add_del_feature(intf.sw_if_index, + 0, + is_add=0) + self.vapi.nat44_interface_add_del_feature(intf.sw_if_index, + intf.is_inside, + is_add=0) - :param pkt: Packet to check UDP checksum - """ - new = pkt.__class__(str(pkt)) - del new['UDP'].chksum - new = new.__class__(str(new)) - self.assertEqual(new['UDP'].chksum, pkt['UDP'].chksum) + interfaces = self.vapi.nat44_interface_output_feature_dump() + for intf in interfaces: + self.vapi.nat44_interface_add_del_output_feature(intf.sw_if_index, + intf.is_inside, + is_add=0) - def check_icmp_errror_embedded(self, pkt): - """ - Check ICMP error embeded packet checksum + static_mappings = self.vapi.nat44_static_mapping_dump() + for sm in static_mappings: + self.vapi.nat44_add_del_static_mapping( + sm.local_ip_address, + sm.external_ip_address, + local_port=sm.local_port, + external_port=sm.external_port, + addr_only=sm.addr_only, + vrf_id=sm.vrf_id, + protocol=sm.protocol, + twice_nat=sm.twice_nat, + self_twice_nat=sm.self_twice_nat, + out2in_only=sm.out2in_only, + tag=sm.tag, + external_sw_if_index=sm.external_sw_if_index, + is_add=0) - :param pkt: Packet to check ICMP error embeded packet checksum - """ - if pkt.haslayer(IPerror): - new = pkt.__class__(str(pkt)) - del new['IPerror'].chksum - new = new.__class__(str(new)) - self.assertEqual(new['IPerror'].chksum, pkt['IPerror'].chksum) - - if pkt.haslayer(TCPerror): - new = pkt.__class__(str(pkt)) - del new['TCPerror'].chksum - new = new.__class__(str(new)) - self.assertEqual(new['TCPerror'].chksum, pkt['TCPerror'].chksum) - - if pkt.haslayer(UDPerror): - if pkt['UDPerror'].chksum != 0: - new = pkt.__class__(str(pkt)) - del new['UDPerror'].chksum - new = new.__class__(str(new)) - self.assertEqual(new['UDPerror'].chksum, - pkt['UDPerror'].chksum) - - if pkt.haslayer(ICMPerror): - del new['ICMPerror'].chksum - new = new.__class__(str(new)) - self.assertEqual(new['ICMPerror'].chksum, pkt['ICMPerror'].chksum) - - def check_icmp_checksum(self, pkt): + lb_static_mappings = self.vapi.nat44_lb_static_mapping_dump() + for lb_sm in lb_static_mappings: + self.vapi.nat44_add_del_lb_static_mapping( + lb_sm.external_addr, + lb_sm.external_port, + lb_sm.protocol, + vrf_id=lb_sm.vrf_id, + twice_nat=lb_sm.twice_nat, + self_twice_nat=lb_sm.self_twice_nat, + out2in_only=lb_sm.out2in_only, + tag=lb_sm.tag, + is_add=0, + local_num=0, + locals=[]) + + identity_mappings = self.vapi.nat44_identity_mapping_dump() + for id_m in identity_mappings: + self.vapi.nat44_add_del_identity_mapping( + addr_only=id_m.addr_only, + ip=id_m.ip_address, + port=id_m.port, + sw_if_index=id_m.sw_if_index, + vrf_id=id_m.vrf_id, + protocol=id_m.protocol, + is_add=0) + + adresses = self.vapi.nat44_address_dump() + for addr in adresses: + self.vapi.nat44_add_del_address_range(addr.ip_address, + addr.ip_address, + twice_nat=addr.twice_nat, + is_add=0) + + self.vapi.nat_set_reass() + self.vapi.nat_set_reass(is_ip6=1) + + def nat44_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, twice_nat=0, self_twice_nat=0, + out2in_only=0, tag=""): """ - Check ICMP checksum in IPv4 packet + Add/delete NAT44 static mapping - :param pkt: Packet to check ICMP checksum + :param local_ip: Local IP address + :param external_ip: External IP address + :param local_port: Local port number (Optional) + :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) + :param twice_nat: 1 if translate external host address and port + :param self_twice_nat: 1 if translate external host address and port + whenever external host address equals + local address of internal host + :param out2in_only: if 1 rule is matching only out2in direction + :param tag: Opaque string tag """ - new = pkt.__class__(str(pkt)) - del new['ICMP'].chksum - new = new.__class__(str(new)) - self.assertEqual(new['ICMP'].chksum, pkt['ICMP'].chksum) - if pkt.haslayer(IPerror): - self.check_icmp_errror_embedded(pkt) - - def check_icmpv6_checksum(self, pkt): + addr_only = 1 + if local_port and external_port: + addr_only = 0 + l_ip = socket.inet_pton(socket.AF_INET, local_ip) + e_ip = socket.inet_pton(socket.AF_INET, external_ip) + self.vapi.nat44_add_del_static_mapping( + l_ip, + e_ip, + external_sw_if_index, + local_port, + external_port, + addr_only, + vrf_id, + proto, + twice_nat, + self_twice_nat, + out2in_only, + tag, + is_add) + + def nat44_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF, twice_nat=0): """ - Check ICMPv6 checksum in IPv4 packet + Add/delete NAT44 address - :param pkt: Packet to check ICMPv6 checksum + :param ip: IP address + :param is_add: 1 if add, 0 if delete (Default add) + :param twice_nat: twice NAT address for extenal hosts """ - new = pkt.__class__(str(pkt)) - if pkt.haslayer(ICMPv6DestUnreach): - del new['ICMPv6DestUnreach'].cksum - new = new.__class__(str(new)) - self.assertEqual(new['ICMPv6DestUnreach'].cksum, - pkt['ICMPv6DestUnreach'].cksum) - self.check_icmp_errror_embedded(pkt) - if pkt.haslayer(ICMPv6EchoRequest): - del new['ICMPv6EchoRequest'].cksum - new = new.__class__(str(new)) - self.assertEqual(new['ICMPv6EchoRequest'].cksum, - pkt['ICMPv6EchoRequest'].cksum) - if pkt.haslayer(ICMPv6EchoReply): - del new['ICMPv6EchoReply'].cksum - new = new.__class__(str(new)) - self.assertEqual(new['ICMPv6EchoReply'].cksum, - pkt['ICMPv6EchoReply'].cksum) + nat_addr = socket.inet_pton(socket.AF_INET, ip) + self.vapi.nat44_add_del_address_range(nat_addr, nat_addr, is_add, + vrf_id=vrf_id, + twice_nat=twice_nat) def create_stream_in(self, in_if, out_if, dst_ip=None, ttl=64): """ @@ -383,7 +442,7 @@ class MethodHolder(VppTestCase): for packet in capture: try: if not is_ip6: - self.check_ip_checksum(packet) + self.assert_packet_checksums_valid(packet) self.assertEqual(packet[IP46].src, nat_ip) if dst_ip is not None: self.assertEqual(packet[IP46].dst, dst_ip) @@ -394,7 +453,7 @@ class MethodHolder(VppTestCase): self.assertNotEqual( packet[TCP].sport, self.tcp_port_in) self.tcp_port_out = packet[TCP].sport - self.check_tcp_checksum(packet) + self.assert_packet_checksums_valid(packet) elif packet.haslayer(UDP): if same_port: self.assertEqual(packet[UDP].sport, self.udp_port_in) @@ -408,10 +467,7 @@ class MethodHolder(VppTestCase): else: self.assertNotEqual(packet[ICMP46].id, self.icmp_id_in) self.icmp_id_out = packet[ICMP46].id - if is_ip6: - self.check_icmpv6_checksum(packet) - else: - self.check_icmp_checksum(packet) + self.assert_packet_checksums_valid(packet) except: self.logger.error(ppp("Unexpected or invalid packet " "(outside network):", packet)) @@ -442,16 +498,14 @@ class MethodHolder(VppTestCase): self.assertEqual(packet_num, len(capture)) for packet in capture: try: - self.check_ip_checksum(packet) + self.assert_packet_checksums_valid(packet) self.assertEqual(packet[IP].dst, in_if.remote_ip4) if packet.haslayer(TCP): self.assertEqual(packet[TCP].dport, self.tcp_port_in) - self.check_tcp_checksum(packet) elif packet.haslayer(UDP): self.assertEqual(packet[UDP].dport, self.udp_port_in) else: self.assertEqual(packet[ICMP].id, self.icmp_id_in) - self.check_icmp_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet " "(inside network):", packet)) @@ -471,16 +525,14 @@ class MethodHolder(VppTestCase): try: self.assertEqual(packet[IPv6].src, src_ip) self.assertEqual(packet[IPv6].dst, dst_ip) + self.assert_packet_checksums_valid(packet) if packet.haslayer(TCP): self.assertEqual(packet[TCP].dport, self.tcp_port_in) - self.check_tcp_checksum(packet) elif packet.haslayer(UDP): self.assertEqual(packet[UDP].dport, self.udp_port_in) - self.check_udp_checksum(packet) else: self.assertEqual(packet[ICMPv6EchoReply].id, self.icmp_id_in) - self.check_icmpv6_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet " "(inside network):", packet)) @@ -655,7 +707,7 @@ class MethodHolder(VppTestCase): for p in frags: self.assertEqual(p[IP].src, src) self.assertEqual(p[IP].dst, dst) - self.check_ip_checksum(p) + self.assert_ip_checksum_valid(p) buffer.seek(p[IP].frag * 8) buffer.write(p[IP].payload) ip = frags[0].getlayer(IP) @@ -663,7 +715,7 @@ class MethodHolder(VppTestCase): proto=frags[0][IP].proto) if ip.proto == IP_PROTOS.tcp: p = (ip / TCP(buffer.getvalue())) - self.check_tcp_checksum(p) + self.assert_tcp_checksum_valid(p) elif ip.proto == IP_PROTOS.udp: p = (ip / UDP(buffer.getvalue())) return p @@ -688,9 +740,9 @@ class MethodHolder(VppTestCase): nh=frags[0][IPv6ExtHdrFragment].nh) if ip.nh == IP_PROTOS.tcp: p = (ip / TCP(buffer.getvalue())) - self.check_tcp_checksum(p) elif ip.nh == IP_PROTOS.udp: p = (ip / UDP(buffer.getvalue())) + self.assert_packet_checksums_valid(p) return p def initiate_tcp_session(self, in_if, out_if): @@ -942,6 +994,7 @@ class TestNAT44(MethodHolder): @classmethod def setUpClass(cls): super(TestNAT44, cls).setUpClass() + cls.vapi.cli("set log class nat level debug") try: cls.tcp_port_in = 6303 @@ -1010,215 +1063,50 @@ class TestNAT44(MethodHolder): super(TestNAT44, cls).tearDownClass() raise - def clear_nat44(self): - """ - Clear NAT44 configuration. - """ - # I found no elegant way to do this - self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n, - dst_address_length=32, - next_hop_address=self.pg7.remote_ip4n, - next_hop_sw_if_index=self.pg7.sw_if_index, - is_add=0) - self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n, - dst_address_length=32, - next_hop_address=self.pg8.remote_ip4n, - next_hop_sw_if_index=self.pg8.sw_if_index, - is_add=0) - - for intf in [self.pg7, self.pg8]: - neighbors = self.vapi.ip_neighbor_dump(intf.sw_if_index) - for n in neighbors: - self.vapi.ip_neighbor_add_del(intf.sw_if_index, - n.mac_address, - n.ip_address, - is_add=0) - - if self.pg7.has_ip4_config: - self.pg7.unconfig_ip4() + def test_dynamic(self): + """ NAT44 dynamic translation test """ - self.vapi.nat44_forwarding_enable_disable(0) + 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) - interfaces = self.vapi.nat44_interface_addr_dump() - for intf in interfaces: - self.vapi.nat44_add_interface_addr(intf.sw_if_index, - twice_nat=intf.twice_nat, - is_add=0) + # in2out + 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) - self.vapi.nat_ipfix(enable=0, src_port=self.ipfix_src_port, - domain_id=self.ipfix_domain_id) - self.ipfix_src_port = 4739 - self.ipfix_domain_id = 1 + # out2in + 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) - interfaces = self.vapi.nat44_interface_dump() - for intf in interfaces: - if intf.is_inside > 1: - self.vapi.nat44_interface_add_del_feature(intf.sw_if_index, - 0, - is_add=0) - self.vapi.nat44_interface_add_del_feature(intf.sw_if_index, - intf.is_inside, - is_add=0) + def test_dynamic_icmp_errors_in2out_ttl_1(self): + """ NAT44 handling of client packets with TTL=1 """ - interfaces = self.vapi.nat44_interface_output_feature_dump() - for intf in interfaces: - self.vapi.nat44_interface_add_del_output_feature(intf.sw_if_index, - intf.is_inside, - is_add=0) + 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) - static_mappings = self.vapi.nat44_static_mapping_dump() - for sm in static_mappings: - self.vapi.nat44_add_del_static_mapping( - sm.local_ip_address, - sm.external_ip_address, - local_port=sm.local_port, - external_port=sm.external_port, - addr_only=sm.addr_only, - vrf_id=sm.vrf_id, - protocol=sm.protocol, - twice_nat=sm.twice_nat, - self_twice_nat=sm.self_twice_nat, - out2in_only=sm.out2in_only, - tag=sm.tag, - external_sw_if_index=sm.external_sw_if_index, - is_add=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() - lb_static_mappings = self.vapi.nat44_lb_static_mapping_dump() - for lb_sm in lb_static_mappings: - self.vapi.nat44_add_del_lb_static_mapping( - lb_sm.external_addr, - lb_sm.external_port, - lb_sm.protocol, - vrf_id=lb_sm.vrf_id, - twice_nat=lb_sm.twice_nat, - self_twice_nat=lb_sm.self_twice_nat, - out2in_only=lb_sm.out2in_only, - tag=lb_sm.tag, - is_add=0, - local_num=0, - locals=[]) + # Client side - verify ICMP type 11 packets + capture = self.pg0.get_capture(len(pkts)) + self.verify_capture_in_with_icmp_errors(capture, self.pg0) - identity_mappings = self.vapi.nat44_identity_mapping_dump() - for id_m in identity_mappings: - self.vapi.nat44_add_del_identity_mapping( - addr_only=id_m.addr_only, - ip=id_m.ip_address, - port=id_m.port, - sw_if_index=id_m.sw_if_index, - vrf_id=id_m.vrf_id, - protocol=id_m.protocol, - is_add=0) - - adresses = self.vapi.nat44_address_dump() - for addr in adresses: - self.vapi.nat44_add_del_address_range(addr.ip_address, - addr.ip_address, - twice_nat=addr.twice_nat, - is_add=0) - - self.vapi.nat_set_reass() - self.vapi.nat_set_reass(is_ip6=1) - - def nat44_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, twice_nat=0, self_twice_nat=0, - out2in_only=0, tag=""): - """ - Add/delete NAT44 static mapping - - :param local_ip: Local IP address - :param external_ip: External IP address - :param local_port: Local port number (Optional) - :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) - :param twice_nat: 1 if translate external host address and port - :param self_twice_nat: 1 if translate external host address and port - whenever external host address equals - local address of internal host - :param out2in_only: if 1 rule is matching only out2in direction - :param tag: Opaque string tag - """ - addr_only = 1 - if local_port and external_port: - addr_only = 0 - l_ip = socket.inet_pton(socket.AF_INET, local_ip) - e_ip = socket.inet_pton(socket.AF_INET, external_ip) - self.vapi.nat44_add_del_static_mapping( - l_ip, - e_ip, - external_sw_if_index, - local_port, - external_port, - addr_only, - vrf_id, - proto, - twice_nat, - self_twice_nat, - out2in_only, - tag, - is_add) - - def nat44_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF, twice_nat=0): - """ - Add/delete NAT44 address - - :param ip: IP address - :param is_add: 1 if add, 0 if delete (Default add) - :param twice_nat: twice NAT address for extenal hosts - """ - nat_addr = socket.inet_pton(socket.AF_INET, ip) - self.vapi.nat44_add_del_address_range(nat_addr, nat_addr, is_add, - vrf_id=vrf_id, - twice_nat=twice_nat) - - def test_dynamic(self): - """ NAT44 dynamic translation test """ - - 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) - - # in2out - 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) - - # out2in - 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_dynamic_icmp_errors_in2out_ttl_1(self): - """ NAT44 handling of client packets with TTL=1 """ - - 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) - - # 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): - """ NAT44 handling of server packets with TTL=1 """ + def test_dynamic_icmp_errors_out2in_ttl_1(self): + """ NAT44 handling of server packets with TTL=1 """ self.nat44_add_address(self.nat_addr) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) @@ -1364,7 +1252,7 @@ class TestNAT44(MethodHolder): self.verify_capture_out(capture, same_port=True, packet_num=1) self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp) - def test_forwarding(self): + def _test_forwarding(self): """ NAT44 forwarding test """ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) @@ -1418,6 +1306,19 @@ class TestNAT44(MethodHolder): finally: self.pg0.remote_hosts[0] = host0 + user = self.pg0.remote_hosts[1] + sessions = self.vapi.nat44_user_session_dump(user.ip4n, 0) + self.assertEqual(len(sessions), 3) + self.assertTrue(sessions[0].ext_host_valid) + self.vapi.nat44_del_session( + sessions[0].inside_ip_address, + sessions[0].inside_port, + sessions[0].protocol, + ext_host_address=sessions[0].ext_host_address, + ext_host_port=sessions[0].ext_host_port) + sessions = self.vapi.nat44_user_session_dump(user.ip4n, 0) + self.assertEqual(len(sessions), 2) + finally: self.vapi.nat44_forwarding_enable_disable(0) self.vapi.nat44_add_del_static_mapping(local_ip=real_ip, @@ -1566,138 +1467,6 @@ class TestNAT44(MethodHolder): capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture) - def test_static_with_port_out2(self): - """ 1:1 NAPT symmetrical rule """ - - external_port = 80 - local_port = 8080 - - self.vapi.nat44_forwarding_enable_disable(1) - self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_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.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=self.nat_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 - - # ICMP error - p = (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) / capture[0][IP]) - 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: - self.assertEqual(p[IP].src, self.nat_addr) - inner = p[IPerror] - self.assertEqual(inner.dst, self.nat_addr) - self.assertEqual(inner[TCPerror].dport, external_port) - 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, self.nat_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 - - # ICMP error - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - ICMP(type=11) / capture[0][IP]) - 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: - self.assertEqual(p[IP].dst, self.pg0.remote_ip4) - inner = p[IPerror] - self.assertEqual(inner.src, self.pg0.remote_ip4) - self.assertEqual(inner[TCPerror].sport, local_port) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from client to server (no translation) - 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=12346, dport=local_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 (no translation) - 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=12346)) - 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, local_port) - self.check_tcp_checksum(p) - self.check_ip_checksum(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - def test_static_vrf_aware(self): """ 1:1 NAT VRF awareness """ @@ -1786,243 +1555,13 @@ class TestNAT44(MethodHolder): self.assertEqual(ip.src, self.pg1.remote_ip4) self.assertEqual(tcp.dport, 56789) self.assertEqual(tcp.sport, 12345) - self.check_tcp_checksum(p) - self.check_ip_checksum(p) + self.assert_packet_checksums_valid(p) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise - def test_static_lb(self): - """ NAT44 local service load balancing """ - 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': 70}, - {'addr': server2.ip4n, - 'port': local_port, - 'probability': 30}] - - 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) - - # from client to service - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_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.assertIn(ip.dst, [server1.ip4, server2.ip4]) - if ip.dst == server1.ip4: - server = server1 - else: - server = server2 - 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=server.mac, dst=self.pg0.local_mac) / - IP(src=server.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, self.nat_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 - - @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, 50) - pkts = [] - for client in clients: - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=client, dst=self.nat_addr) / - TCP(sport=12345, dport=external_port)) - 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 p in capture: - if p[IP].dst == server1.ip4: - server1_n += 1 - else: - server2_n += 1 - self.assertTrue(server1_n > server2_n) - - def test_static_lb_2(self): - """ NAT44 local service load balancing (asymmetrical rule) """ - 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': 70}, - {'addr': server2.ip4n, - 'port': local_port, - 'probability': 30}] - - self.vapi.nat44_forwarding_enable_disable(1) - self.vapi.nat44_add_del_lb_static_mapping(external_addr_n, - external_port, - IP_PROTOS.tcp, - out2in_only=1, - 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) - - # from client to service - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_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.assertIn(ip.dst, [server1.ip4, server2.ip4]) - if ip.dst == server1.ip4: - server = server1 - else: - server = server2 - 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=server.mac, dst=self.pg0.local_mac) / - IP(src=server.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, self.nat_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 client to server (no translation) - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=server1.ip4) / - TCP(sport=12346, dport=local_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, server1.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 (no translation) - p = (Ether(src=server1.mac, dst=self.pg0.local_mac) / - IP(src=server1.ip4, dst=self.pg1.remote_ip4) / - TCP(sport=local_port, dport=12346)) - 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, server1.ip4) - self.assertEqual(tcp.sport, local_port) - self.check_tcp_checksum(p) - self.check_ip_checksum(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - def test_multiple_inside_interfaces(self): - """ NAT44 multiple non-overlapping address space inside interfaces """ + def test_multiple_inside_interfaces(self): + """ NAT44 multiple non-overlapping address space inside interfaces """ self.nat44_add_address(self.nat_addr) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) @@ -2202,6 +1741,7 @@ class TestNAT44(MethodHolder): self.assertTrue(session.protocol in [IP_PROTOS.tcp, IP_PROTOS.udp, IP_PROTOS.icmp]) + self.assertFalse(session.ext_host_valid) # pg4 session dump sessions = self.vapi.nat44_user_session_dump(self.pg4.remote_ip4n, 10) @@ -2261,7 +1801,7 @@ class TestNAT44(MethodHolder): self.assertEqual(ip.dst, server.ip4) self.assertNotEqual(tcp.sport, host_in_port) self.assertEqual(tcp.dport, server_in_port) - self.check_tcp_checksum(p) + self.assert_packet_checksums_valid(p) host_out_port = tcp.sport except: self.logger.error(ppp("Unexpected or invalid packet:", p)) @@ -2283,7 +1823,7 @@ class TestNAT44(MethodHolder): self.assertEqual(ip.dst, host.ip4) self.assertEqual(tcp.sport, server_out_port) self.assertEqual(tcp.dport, host_in_port) - self.check_tcp_checksum(p) + self.assert_packet_checksums_valid(p) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise @@ -2334,7 +1874,7 @@ class TestNAT44(MethodHolder): self.assertNotEqual(packet[TCP].sport, self.tcp_port_in) self.assertEqual(packet[TCP].dport, server_tcp_port) self.tcp_port_out = packet[TCP].sport - self.check_tcp_checksum(packet) + self.assert_packet_checksums_valid(packet) elif packet.haslayer(UDP): self.assertNotEqual(packet[UDP].sport, self.udp_port_in) self.assertEqual(packet[UDP].dport, server_udp_port) @@ -2371,7 +1911,7 @@ class TestNAT44(MethodHolder): if packet.haslayer(TCP): self.assertEqual(packet[TCP].dport, self.tcp_port_in) self.assertEqual(packet[TCP].sport, server_tcp_port) - self.check_tcp_checksum(packet) + self.assert_packet_checksums_valid(packet) elif packet.haslayer(UDP): self.assertEqual(packet[UDP].dport, self.udp_port_in) self.assertEqual(packet[UDP].sport, server_udp_port) @@ -2407,7 +1947,7 @@ class TestNAT44(MethodHolder): self.assertEqual(packet[TCP].sport, self.tcp_port_in) self.assertEqual(packet[TCP].dport, server_tcp_port) self.tcp_port_out = packet[TCP].sport - self.check_tcp_checksum(packet) + self.assert_packet_checksums_valid(packet) elif packet.haslayer(UDP): self.assertEqual(packet[UDP].sport, self.udp_port_in) self.assertEqual(packet[UDP].dport, server_udp_port) @@ -2444,7 +1984,7 @@ class TestNAT44(MethodHolder): if packet.haslayer(TCP): self.assertEqual(packet[TCP].dport, self.tcp_port_in) self.assertEqual(packet[TCP].sport, server_tcp_port) - self.check_tcp_checksum(packet) + self.assert_packet_checksums_valid(packet) elif packet.haslayer(UDP): self.assertEqual(packet[UDP].dport, self.udp_port_in) self.assertEqual(packet[UDP].sport, server_udp_port) @@ -2480,6 +2020,31 @@ class TestNAT44(MethodHolder): # verify number of translated packet self.pg1.get_capture(pkts_num) + users = self.vapi.nat44_user_dump() + for user in users: + if user.ip_address == self.pg0.remote_ip4n: + self.assertEqual(user.nsessions, + nat44_config.max_translations_per_user) + self.assertEqual(user.nstaticsessions, 0) + + tcp_port = 22 + self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr, + tcp_port, tcp_port, + proto=IP_PROTOS.tcp) + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / + TCP(sport=tcp_port)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg1.get_capture(1) + users = self.vapi.nat44_user_dump() + for user in users: + if user.ip_address == self.pg0.remote_ip4n: + self.assertEqual(user.nsessions, + nat44_config.max_translations_per_user - 1) + self.assertEqual(user.nstaticsessions, 1) + def test_interface_addr(self): """ Acquire NAT44 addresses from interface """ self.vapi.nat44_add_interface_addr(self.pg7.sw_if_index) @@ -2660,7 +2225,8 @@ class TestNAT44(MethodHolder): self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg1.get_capture(0) + self.pg1.assert_nothing_captured() + sleep(1) self.vapi.cli("ipfix flush") # FIXME this should be an API call capture = self.pg3.get_capture(9) ipfix = IPFIXDecoder() @@ -2717,7 +2283,8 @@ class TestNAT44(MethodHolder): self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - self.pg1.get_capture(0) + self.pg1.assert_nothing_captured() + sleep(1) self.vapi.cli("ipfix flush") # FIXME this should be an API call capture = self.pg3.get_capture(9) ipfix = IPFIXDecoder() @@ -2776,7 +2343,7 @@ class TestNAT44(MethodHolder): self.pg2.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg1.get_capture(0) + self.pg1.assert_nothing_captured() # remove addresses and verify self.nat44_add_address(self.nat_addr, is_add=0) @@ -2789,7 +2356,7 @@ class TestNAT44(MethodHolder): self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg1.get_capture(0) + self.pg1.assert_nothing_captured() p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') / ARP(op=ARP.who_has, pdst=static_addr, @@ -2797,7 +2364,7 @@ class TestNAT44(MethodHolder): self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg1.get_capture(0) + self.pg1.assert_nothing_captured() def test_vrf_mode(self): """ NAT44 tenant VRF aware address pool mode """ @@ -3037,7 +2604,7 @@ class TestNAT44(MethodHolder): self.assertEqual(packet[IP].src, nat_ip) self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) self.assertTrue(packet.haslayer(GRE)) - self.check_ip_checksum(packet) + self.assert_packet_checksums_valid(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise @@ -3057,7 +2624,7 @@ class TestNAT44(MethodHolder): self.assertEqual(packet[IP].src, self.pg1.remote_ip4) self.assertEqual(packet[IP].dst, self.pg0.remote_ip4) self.assertTrue(packet.haslayer(GRE)) - self.check_ip_checksum(packet) + self.assert_packet_checksums_valid(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise @@ -3092,7 +2659,7 @@ class TestNAT44(MethodHolder): self.assertEqual(packet[IP].src, host_nat_ip) self.assertEqual(packet[IP].dst, server.ip4) self.assertTrue(packet.haslayer(GRE)) - self.check_ip_checksum(packet) + self.assert_packet_checksums_valid(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise @@ -3112,155 +2679,34 @@ class TestNAT44(MethodHolder): self.assertEqual(packet[IP].src, server_nat_ip) self.assertEqual(packet[IP].dst, host.ip4) self.assertTrue(packet.haslayer(GRE)) - self.check_ip_checksum(packet) + self.assert_packet_checksums_valid(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise - def test_unknown_proto(self): - """ NAT44 translate packet with unknown protocol """ + def test_output_feature(self): + """ NAT44 interface output feature (in2out postrouting) """ 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) + self.vapi.nat44_interface_add_del_output_feature(self.pg0.sw_if_index) + self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index) + self.vapi.nat44_interface_add_del_output_feature(self.pg3.sw_if_index, + is_inside=0) # in2out - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=self.tcp_port_in, dport=20)) - self.pg0.add_stream(p) + pkts = self.create_stream_in(self.pg0, self.pg3) + self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - p = self.pg1.get_capture(1) + capture = self.pg3.get_capture(len(pkts)) + self.verify_capture_out(capture) - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - GRE() / - IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) / - TCP(sport=1234, dport=1234)) - self.pg0.add_stream(p) + # out2in + pkts = self.create_stream_out(self.pg3) + self.pg3.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - p = self.pg1.get_capture(1) - packet = p[0] - try: - self.assertEqual(packet[IP].src, self.nat_addr) - self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) - self.assertTrue(packet.haslayer(GRE)) - self.check_ip_checksum(packet) - except: - self.logger.error(ppp("Unexpected or invalid packet:", packet)) - raise - - # out2in - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - GRE() / - IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) / - TCP(sport=1234, dport=1234)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - p = self.pg0.get_capture(1) - packet = p[0] - try: - self.assertEqual(packet[IP].src, self.pg1.remote_ip4) - self.assertEqual(packet[IP].dst, self.pg0.remote_ip4) - self.assertTrue(packet.haslayer(GRE)) - self.check_ip_checksum(packet) - except: - self.logger.error(ppp("Unexpected or invalid packet:", packet)) - raise - - def test_hairpinning_unknown_proto(self): - """ NAT44 translate packet with unknown protocol - hairpinning """ - host = self.pg0.remote_hosts[0] - server = self.pg0.remote_hosts[1] - host_in_port = 1234 - host_out_port = 0 - server_in_port = 5678 - server_out_port = 8765 - server_nat_ip = "10.0.0.11" - - 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) - - # add static mapping for server - self.nat44_add_static_mapping(server.ip4, server_nat_ip) - - # host to server - p = (Ether(src=host.mac, dst=self.pg0.local_mac) / - IP(src=host.ip4, dst=server_nat_ip) / - TCP(sport=host_in_port, dport=server_out_port)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(1) - - p = (Ether(dst=self.pg0.local_mac, src=host.mac) / - IP(src=host.ip4, dst=server_nat_ip) / - GRE() / - IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) / - TCP(sport=1234, dport=1234)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - p = self.pg0.get_capture(1) - packet = p[0] - try: - self.assertEqual(packet[IP].src, self.nat_addr) - self.assertEqual(packet[IP].dst, server.ip4) - self.assertTrue(packet.haslayer(GRE)) - self.check_ip_checksum(packet) - except: - self.logger.error(ppp("Unexpected or invalid packet:", packet)) - raise - - # server to host - p = (Ether(dst=self.pg0.local_mac, src=server.mac) / - IP(src=server.ip4, dst=self.nat_addr) / - GRE() / - IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) / - TCP(sport=1234, dport=1234)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - p = self.pg0.get_capture(1) - packet = p[0] - try: - self.assertEqual(packet[IP].src, server_nat_ip) - self.assertEqual(packet[IP].dst, host.ip4) - self.assertTrue(packet.haslayer(GRE)) - self.check_ip_checksum(packet) - except: - self.logger.error(ppp("Unexpected or invalid packet:", packet)) - raise - - def test_output_feature(self): - """ NAT44 interface output feature (in2out postrouting) """ - self.nat44_add_address(self.nat_addr) - self.vapi.nat44_interface_add_del_output_feature(self.pg0.sw_if_index) - self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index) - self.vapi.nat44_interface_add_del_output_feature(self.pg3.sw_if_index, - is_inside=0) - - # in2out - pkts = self.create_stream_in(self.pg0, self.pg3) - self.pg0.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 - pkts = self.create_stream_out(self.pg3) - self.pg3.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) + capture = self.pg0.get_capture(len(pkts)) + self.verify_capture_in(capture, self.pg0) # from non-NAT interface to NAT inside interface pkts = self.create_stream_in(self.pg2, self.pg0) @@ -3360,7 +2806,7 @@ class TestNAT44(MethodHolder): self.assertEqual(ip.dst, server.ip4) self.assertNotEqual(tcp.sport, host_in_port) self.assertEqual(tcp.dport, server_in_port) - self.check_tcp_checksum(p) + self.assert_packet_checksums_valid(p) host_out_port = tcp.sport except: self.logger.error(ppp("Unexpected or invalid packet:", p)) @@ -3382,361 +2828,111 @@ class TestNAT44(MethodHolder): self.assertEqual(ip.dst, host.ip4) self.assertEqual(tcp.sport, server_out_port) self.assertEqual(tcp.dport, host_in_port) - self.check_tcp_checksum(p) + self.assert_packet_checksums_valid(p) except: 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 + def test_one_armed_nat44(self): + """ One armed NAT44 """ + remote_host = self.pg9.remote_hosts[0] + local_host = self.pg9.remote_hosts[1] + external_port = 0 - self.vapi.nat44_forwarding_enable_disable(1) self.nat44_add_address(self.nat_addr) - self.vapi.nat44_add_del_identity_mapping(ip=self.pg1.remote_ip4n) - 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_feature(self.pg0.sw_if_index, + self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index) + self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index, is_inside=0) - 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) + # in2out + p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) / + IP(src=local_host.ip4, dst=remote_host.ip4) / + TCP(sport=12345, dport=80)) + self.pg9.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg0.get_capture(1) + capture = self.pg9.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) + self.assertEqual(ip.src, self.nat_addr) + self.assertEqual(ip.dst, remote_host.ip4) + self.assertNotEqual(tcp.sport, 12345) + external_port = tcp.sport + self.assertEqual(tcp.dport, 80) + self.assert_packet_checksums_valid(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) + # out2in + p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) / + IP(src=remote_host.ip4, dst=self.nat_addr) / + TCP(sport=80, dport=external_port)) + self.pg9.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg1.get_capture(1) + capture = self.pg9.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) + self.assertEqual(ip.src, remote_host.ip4) + self.assertEqual(ip.dst, local_host.ip4) + self.assertEqual(tcp.sport, 80) + self.assertEqual(tcp.dport, 12345) + self.assert_packet_checksums_valid(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) + def test_del_session(self): + """ Delete NAT44 session """ + 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) + 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) + self.pg1.get_capture(len(pkts)) - # 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) + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0) + nsessions = len(sessions) - 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) + self.vapi.nat44_del_session(sessions[0].inside_ip_address, + sessions[0].inside_port, + sessions[0].protocol) + self.vapi.nat44_del_session(sessions[1].outside_ip_address, + sessions[1].outside_port, + sessions[1].protocol, + is_in=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) + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0) + self.assertEqual(nsessions - len(sessions), 2) - 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_set_get_reass(self): + """ NAT44 set/get virtual fragmentation reassembly """ + reas_cfg1 = self.vapi.nat_get_reass() - tcp_port_out = self.tcp_port_out - udp_port_out = self.udp_port_out - icmp_id_out = self.icmp_id_out + self.vapi.nat_set_reass(timeout=reas_cfg1.ip4_timeout + 5, + max_reass=reas_cfg1.ip4_max_reass * 2, + max_frag=reas_cfg1.ip4_max_frag * 2) - # 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) + reas_cfg2 = self.vapi.nat_get_reass() - 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) + self.assertEqual(reas_cfg1.ip4_timeout + 5, reas_cfg2.ip4_timeout) + self.assertEqual(reas_cfg1.ip4_max_reass * 2, reas_cfg2.ip4_max_reass) + self.assertEqual(reas_cfg1.ip4_max_frag * 2, reas_cfg2.ip4_max_frag) - 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.nat_set_reass(drop_frag=1) + self.assertTrue(self.vapi.nat_get_reass().ip4_drop_frag) - 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] - local_host = self.pg9.remote_hosts[1] - external_port = 0 - - self.nat44_add_address(self.nat_addr) - self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index) - self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index, - is_inside=0) - - # in2out - p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) / - IP(src=local_host.ip4, dst=remote_host.ip4) / - TCP(sport=12345, dport=80)) - self.pg9.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg9.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, self.nat_addr) - self.assertEqual(ip.dst, remote_host.ip4) - self.assertNotEqual(tcp.sport, 12345) - external_port = tcp.sport - self.assertEqual(tcp.dport, 80) - self.check_tcp_checksum(p) - self.check_ip_checksum(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # out2in - p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) / - IP(src=remote_host.ip4, dst=self.nat_addr) / - TCP(sport=80, dport=external_port)) - self.pg9.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg9.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, remote_host.ip4) - self.assertEqual(ip.dst, local_host.ip4) - self.assertEqual(tcp.sport, 80) - 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_static(self): - """ One armed NAT44 and 1:1 NAPT symmetrical rule """ - remote_host = self.pg9.remote_hosts[0] - local_host = self.pg9.remote_hosts[1] - external_port = 80 - local_port = 8080 - eh_port_in = 0 - - self.vapi.nat44_forwarding_enable_disable(1) - self.nat44_add_address(self.nat_addr, twice_nat=1) - self.nat44_add_static_mapping(local_host.ip4, self.nat_addr, - local_port, external_port, - proto=IP_PROTOS.tcp, out2in_only=1, - twice_nat=1) - self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index) - self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index, - is_inside=0) - - # from client to service - p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) / - IP(src=remote_host.ip4, dst=self.nat_addr) / - TCP(sport=12345, dport=external_port)) - self.pg9.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg9.get_capture(1) - p = capture[0] - server = None - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.dst, local_host.ip4) - self.assertEqual(ip.src, self.nat_addr) - self.assertEqual(tcp.dport, local_port) - self.assertNotEqual(tcp.sport, 12345) - eh_port_in = tcp.sport - 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.pg9.remote_mac, dst=self.pg9.local_mac) / - IP(src=local_host.ip4, dst=self.nat_addr) / - TCP(sport=local_port, dport=eh_port_in)) - self.pg9.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg9.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, self.nat_addr) - self.assertEqual(ip.dst, remote_host.ip4) - self.assertEqual(tcp.sport, external_port) - 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_del_session(self): - """ Delete NAT44 session """ - 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) - - 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)) - - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0) - nsessions = len(sessions) - - self.vapi.nat44_del_session(sessions[0].inside_ip_address, - sessions[0].inside_port, - sessions[0].protocol) - self.vapi.nat44_del_session(sessions[1].outside_ip_address, - sessions[1].outside_port, - sessions[1].protocol, - is_in=0) - - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0) - self.assertEqual(nsessions - len(sessions), 2) - - def test_set_get_reass(self): - """ NAT44 set/get virtual fragmentation reassembly """ - reas_cfg1 = self.vapi.nat_get_reass() - - self.vapi.nat_set_reass(timeout=reas_cfg1.ip4_timeout + 5, - max_reass=reas_cfg1.ip4_max_reass * 2, - max_frag=reas_cfg1.ip4_max_frag * 2) - - reas_cfg2 = self.vapi.nat_get_reass() - - self.assertEqual(reas_cfg1.ip4_timeout + 5, reas_cfg2.ip4_timeout) - self.assertEqual(reas_cfg1.ip4_max_reass * 2, reas_cfg2.ip4_max_reass) - self.assertEqual(reas_cfg1.ip4_max_frag * 2, reas_cfg2.ip4_max_frag) - - self.vapi.nat_set_reass(drop_frag=1) - self.assertTrue(self.vapi.nat_get_reass().ip4_drop_frag) - - def test_frag_in_order(self): - """ NAT44 translate fragments arriving in order """ + def test_frag_in_order(self): + """ NAT44 translate fragments arriving in order """ 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, @@ -3790,10 +2986,8 @@ class TestNAT44(MethodHolder): def test_reass_hairpinning(self): """ NAT44 fragments hairpinning """ - host = self.pg0.remote_hosts[0] server = self.pg0.remote_hosts[1] host_in_port = random.randint(1025, 65535) - host_out_port = 0 server_in_port = random.randint(1025, 65535) server_out_port = random.randint(1025, 65535) data = "A" * 4 + "B" * 16 + "C" * 3 @@ -3863,26 +3057,751 @@ class TestNAT44(MethodHolder): self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - self.pg1.remote_ip4, - self.pg0.remote_ip4) - self.assertEqual(p[TCP].sport, 20) - self.assertEqual(p[TCP].dport, self.tcp_port_in) - self.assertEqual(data, p[Raw].load) + frags = self.pg0.get_capture(len(pkts)) + p = self.reass_frags_and_verify(frags, + self.pg1.remote_ip4, + self.pg0.remote_ip4) + self.assertEqual(p[TCP].sport, 20) + self.assertEqual(p[TCP].dport, self.tcp_port_in) + self.assertEqual(data, p[Raw].load) + + def test_port_restricted(self): + """ Port restricted NAT44 (MAP-E CE) """ + 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) + self.vapi.cli("nat addr-port-assignment-alg map-e psid 10 " + "psid-offset 6 psid-len 6") + + 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=4567, dport=22)) + 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.dst, self.pg1.remote_ip4) + self.assertEqual(ip.src, self.nat_addr) + self.assertEqual(tcp.dport, 22) + self.assertNotEqual(tcp.sport, 4567) + self.assertEqual((tcp.sport >> 6) & 63, 10) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + def test_ipfix_max_frags(self): + """ IPFIX logging maximum fragments pending reassembly exceeded """ + 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) + self.vapi.nat_set_reass(max_frag=0) + self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n, + src_address=self.pg3.local_ip4n, + path_mtu=512, + template_interval=10) + self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id, + src_port=self.ipfix_src_port) + + data = "A" * 4 + "B" * 16 + "C" * 3 + self.tcp_port_in = random.randint(1025, 65535) + pkts = self.create_stream_frag(self.pg0, + self.pg1.remote_ip4, + self.tcp_port_in, + 20, + data) + self.pg0.add_stream(pkts[-1]) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg1.assert_nothing_captured() + sleep(1) + self.vapi.cli("ipfix flush") # FIXME this should be an API call + capture = self.pg3.get_capture(9) + ipfix = IPFIXDecoder() + # first load template + for p in capture: + self.assertTrue(p.haslayer(IPFIX)) + self.assertEqual(p[IP].src, self.pg3.local_ip4) + self.assertEqual(p[IP].dst, self.pg3.remote_ip4) + self.assertEqual(p[UDP].sport, self.ipfix_src_port) + self.assertEqual(p[UDP].dport, 4739) + self.assertEqual(p[IPFIX].observationDomainID, + self.ipfix_domain_id) + if p.haslayer(Template): + ipfix.add_template(p.getlayer(Template)) + # verify events in data set + for p in capture: + if p.haslayer(Data): + data = ipfix.decode_data_set(p.getlayer(Set)) + self.verify_ipfix_max_fragments_ip4(data, 0, + self.pg0.remote_ip4n) + + def tearDown(self): + super(TestNAT44, self).tearDown() + if not self.vpp_dead: + self.logger.info(self.vapi.cli("show nat44 addresses")) + self.logger.info(self.vapi.cli("show nat44 interfaces")) + self.logger.info(self.vapi.cli("show nat44 static mappings")) + self.logger.info(self.vapi.cli("show nat44 interface address")) + self.logger.info(self.vapi.cli("show nat44 sessions detail")) + self.logger.info(self.vapi.cli("show nat virtual-reassembly")) + self.logger.info(self.vapi.cli("show nat44 hash tables detail")) + self.vapi.cli("nat addr-port-assignment-alg default") + self.clear_nat44() + self.vapi.cli("clear logging") + + +class TestNAT44EndpointDependent(MethodHolder): + """ Endpoint-Dependent mapping and filtering test cases """ + + @classmethod + def setUpConstants(cls): + super(TestNAT44EndpointDependent, cls).setUpConstants() + cls.vpp_cmdline.extend(["nat", "{", "endpoint-dependent", "}"]) + + @classmethod + def setUpClass(cls): + super(TestNAT44EndpointDependent, cls).setUpClass() + cls.vapi.cli("set log class nat level debug") + try: + 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.nat_addr = '10.0.0.3' + cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr) + cls.ipfix_src_port = 4739 + cls.ipfix_domain_id = 1 + cls.tcp_external_port = 80 + + cls.create_pg_interfaces(range(5)) + cls.interfaces = list(cls.pg_interfaces[0:3]) + + for i in cls.interfaces: + i.admin_up() + i.config_ip4() + i.resolve_arp() + + cls.pg0.generate_remote_hosts(3) + cls.pg0.configure_ipv4_neighbors() + + cls.pg3.admin_up() + + cls.pg4.generate_remote_hosts(2) + cls.pg4.config_ip4() + ip_addr_n = socket.inet_pton(socket.AF_INET, "10.0.0.1") + cls.vapi.sw_interface_add_del_address(cls.pg4.sw_if_index, + ip_addr_n, + 24) + cls.pg4.admin_up() + cls.pg4.resolve_arp() + cls.pg4._remote_hosts[1]._ip4 = cls.pg4._remote_hosts[0]._ip4 + cls.pg4.resolve_arp() + + except Exception: + super(TestNAT44EndpointDependent, cls).tearDownClass() + raise + + def test_dynamic(self): + """ NAT44 dynamic translation test """ + + 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) + + # in2out + 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) + + # out2in + 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_forwarding(self): + """ NAT44 forwarding test """ + + 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) + self.vapi.nat44_forwarding_enable_disable(1) + + real_ip = self.pg0.remote_ip4n + alias_ip = self.nat_addr_n + self.vapi.nat44_add_del_static_mapping(local_ip=real_ip, + external_ip=alias_ip) + + try: + # in2out - static mapping match + + 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) + + 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, same_port=True) + + # in2out - no static mapping match + + host0 = self.pg0.remote_hosts[0] + self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1] + try: + pkts = self.create_stream_out(self.pg1, + dst_ip=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) + finally: + self.pg0.remote_hosts[0] = host0 + + user = self.pg0.remote_hosts[1] + sessions = self.vapi.nat44_user_session_dump(user.ip4n, 0) + self.assertEqual(len(sessions), 3) + self.assertTrue(sessions[0].ext_host_valid) + self.vapi.nat44_del_session( + sessions[0].inside_ip_address, + sessions[0].inside_port, + sessions[0].protocol, + ext_host_address=sessions[0].ext_host_address, + ext_host_port=sessions[0].ext_host_port) + sessions = self.vapi.nat44_user_session_dump(user.ip4n, 0) + self.assertEqual(len(sessions), 2) + + finally: + self.vapi.nat44_forwarding_enable_disable(0) + self.vapi.nat44_add_del_static_mapping(local_ip=real_ip, + external_ip=alias_ip, + is_add=0) + + def test_static_lb(self): + """ NAT44 local service load balancing """ + 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': 70}, + {'addr': server2.ip4n, + 'port': local_port, + 'probability': 30}] + + 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) + + # from client to service + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_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.assertIn(ip.dst, [server1.ip4, server2.ip4]) + if ip.dst == server1.ip4: + server = server1 + else: + server = server2 + self.assertEqual(tcp.dport, local_port) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + # from service back to client + p = (Ether(src=server.mac, dst=self.pg0.local_mac) / + IP(src=server.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, self.nat_addr) + self.assertEqual(tcp.sport, external_port) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + sessions = self.vapi.nat44_user_session_dump(server.ip4n, 0) + self.assertEqual(len(sessions), 1) + self.assertTrue(sessions[0].ext_host_valid) + self.vapi.nat44_del_session( + sessions[0].inside_ip_address, + sessions[0].inside_port, + sessions[0].protocol, + ext_host_address=sessions[0].ext_host_address, + ext_host_port=sessions[0].ext_host_port) + sessions = self.vapi.nat44_user_session_dump(server.ip4n, 0) + self.assertEqual(len(sessions), 0) + + @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, 50) + pkts = [] + for client in clients: + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=client, dst=self.nat_addr) / + TCP(sport=12345, dport=external_port)) + 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 p in capture: + if p[IP].dst == server1.ip4: + server1_n += 1 + else: + server2_n += 1 + self.assertTrue(server1_n > server2_n) + + def test_static_lb_2(self): + """ NAT44 local service load balancing (asymmetrical rule) """ + 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': 70}, + {'addr': server2.ip4n, + 'port': local_port, + 'probability': 30}] + + self.vapi.nat44_forwarding_enable_disable(1) + self.vapi.nat44_add_del_lb_static_mapping(external_addr_n, + external_port, + IP_PROTOS.tcp, + out2in_only=1, + 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) + + # from client to service + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_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.assertIn(ip.dst, [server1.ip4, server2.ip4]) + if ip.dst == server1.ip4: + server = server1 + else: + server = server2 + self.assertEqual(tcp.dport, local_port) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + # from service back to client + p = (Ether(src=server.mac, dst=self.pg0.local_mac) / + IP(src=server.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, self.nat_addr) + self.assertEqual(tcp.sport, external_port) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + # from client to server (no translation) + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=server1.ip4) / + TCP(sport=12346, dport=local_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, server1.ip4) + self.assertEqual(tcp.dport, local_port) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + # from service back to client (no translation) + p = (Ether(src=server1.mac, dst=self.pg0.local_mac) / + IP(src=server1.ip4, dst=self.pg1.remote_ip4) / + TCP(sport=local_port, dport=12346)) + 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, server1.ip4) + self.assertEqual(tcp.sport, local_port) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise + + def test_unknown_proto(self): + """ NAT44 translate packet with unknown protocol """ + 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) + + # in2out + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / + TCP(sport=self.tcp_port_in, dport=20)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + p = self.pg1.get_capture(1) + + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / + GRE() / + IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) / + TCP(sport=1234, dport=1234)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + p = self.pg1.get_capture(1) + packet = p[0] + try: + self.assertEqual(packet[IP].src, self.nat_addr) + self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) + self.assertTrue(packet.haslayer(GRE)) + self.assert_packet_checksums_valid(packet) + except: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + + # out2in + p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / + GRE() / + IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) / + TCP(sport=1234, dport=1234)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + p = self.pg0.get_capture(1) + packet = p[0] + try: + self.assertEqual(packet[IP].src, self.pg1.remote_ip4) + self.assertEqual(packet[IP].dst, self.pg0.remote_ip4) + self.assertTrue(packet.haslayer(GRE)) + self.assert_packet_checksums_valid(packet) + except: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + + def test_hairpinning_unknown_proto(self): + """ NAT44 translate packet with unknown protocol - hairpinning """ + host = self.pg0.remote_hosts[0] + server = self.pg0.remote_hosts[1] + host_in_port = 1234 + server_out_port = 8765 + server_nat_ip = "10.0.0.11" + + 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) + + # add static mapping for server + self.nat44_add_static_mapping(server.ip4, server_nat_ip) + + # host to server + p = (Ether(src=host.mac, dst=self.pg0.local_mac) / + IP(src=host.ip4, dst=server_nat_ip) / + TCP(sport=host_in_port, dport=server_out_port)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg0.get_capture(1) + + p = (Ether(dst=self.pg0.local_mac, src=host.mac) / + IP(src=host.ip4, dst=server_nat_ip) / + GRE() / + IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) / + TCP(sport=1234, dport=1234)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + p = self.pg0.get_capture(1) + packet = p[0] + try: + self.assertEqual(packet[IP].src, self.nat_addr) + self.assertEqual(packet[IP].dst, server.ip4) + self.assertTrue(packet.haslayer(GRE)) + self.assert_packet_checksums_valid(packet) + except: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + + # server to host + p = (Ether(dst=self.pg0.local_mac, src=server.mac) / + IP(src=server.ip4, dst=self.nat_addr) / + GRE() / + IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) / + TCP(sport=1234, dport=1234)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + p = self.pg0.get_capture(1) + packet = p[0] + try: + self.assertEqual(packet[IP].src, server_nat_ip) + self.assertEqual(packet[IP].dst, host.ip4) + self.assertTrue(packet.haslayer(GRE)) + self.assert_packet_checksums_valid(packet) + except: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + 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.vapi.nat44_add_del_identity_mapping(ip=self.pg1.remote_ip4n) + 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_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) + + # 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] + try: + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.dst, self.pg0.remote_ip4) + self.assertEqual(tcp.dport, local_port) + self.assert_packet_checksums_valid(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.assert_packet_checksums_valid(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) + + # session initiaded from remote host - do not translate + self.tcp_port_in = 60303 + self.udp_port_in = 60304 + self.icmp_id_in = 60305 + 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_port_restricted(self): - """ Port restricted NAT44 (MAP-E CE) """ + 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.pg1.sw_if_index, + self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index, is_inside=0) - self.vapi.cli("nat addr-port-assignment-alg map-e psid 10 " - "psid-offset 6 psid-len 6") + 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=self.pg1.remote_ip4) / - TCP(sport=4567, dport=22)) + 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() @@ -3891,13 +3810,31 @@ class TestNAT44(MethodHolder): 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(ip.src, self.nat_addr) - self.assertEqual(tcp.dport, 22) - self.assertNotEqual(tcp.sport, 4567) - self.assertEqual((tcp.sport >> 6) & 63, 10) - self.check_tcp_checksum(p) - self.check_ip_checksum(p) + self.assertEqual(tcp.dport, local_port) + self.assert_packet_checksums_valid(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.assert_packet_checksums_valid(p) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise @@ -4006,8 +3943,7 @@ class TestNAT44(MethodHolder): eh_addr_in = ip.src eh_port_in = tcp.sport saved_port_in = tcp.dport - self.check_tcp_checksum(p) - self.check_ip_checksum(p) + self.assert_packet_checksums_valid(p) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise @@ -4027,12 +3963,25 @@ class TestNAT44(MethodHolder): self.assertEqual(ip.src, self.nat_addr) self.assertEqual(tcp.dport, eh_port_out) self.assertEqual(tcp.sport, port_out) - self.check_tcp_checksum(p) - self.check_ip_checksum(p) + self.assert_packet_checksums_valid(p) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise + if eh_translate: + sessions = self.vapi.nat44_user_session_dump(server.ip4n, 0) + self.assertEqual(len(sessions), 1) + self.assertTrue(sessions[0].ext_host_valid) + self.assertTrue(sessions[0].is_twicenat) + self.vapi.nat44_del_session( + sessions[0].inside_ip_address, + sessions[0].inside_port, + sessions[0].protocol, + ext_host_address=sessions[0].ext_host_nat_address, + ext_host_port=sessions[0].ext_host_nat_port) + sessions = self.vapi.nat44_user_session_dump(server.ip4n, 0) + self.assertEqual(len(sessions), 0) + def test_twice_nat(self): """ Twice NAT44 """ self.twice_nat_common() @@ -4061,73 +4010,152 @@ class TestNAT44(MethodHolder): def test_twice_nat_interface_addr(self): """ Acquire twice NAT44 addresses from interface """ - self.vapi.nat44_add_interface_addr(self.pg7.sw_if_index, twice_nat=1) + self.vapi.nat44_add_interface_addr(self.pg3.sw_if_index, twice_nat=1) # no address in NAT pool adresses = self.vapi.nat44_address_dump() self.assertEqual(0, len(adresses)) # configure interface address and check NAT address pool - self.pg7.config_ip4() + self.pg3.config_ip4() adresses = self.vapi.nat44_address_dump() self.assertEqual(1, len(adresses)) - self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n) + self.assertEqual(adresses[0].ip_address[0:4], self.pg3.local_ip4n) self.assertEqual(adresses[0].twice_nat, 1) # remove interface address and check NAT address pool - self.pg7.unconfig_ip4() + self.pg3.unconfig_ip4() adresses = self.vapi.nat44_address_dump() self.assertEqual(0, len(adresses)) - def test_ipfix_max_frags(self): - """ IPFIX logging maximum fragments pending reassembly exceeded """ + def test_tcp_session_close_in(self): + """ Close TCP session from inside network """ + self.tcp_port_out = 10505 self.nat44_add_address(self.nat_addr) + self.nat44_add_static_mapping(self.pg0.remote_ip4, + self.nat_addr, + self.tcp_port_in, + self.tcp_port_out, + proto=IP_PROTOS.tcp, + twice_nat=1) 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) - self.vapi.nat_set_reass(max_frag=0) - self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n, - src_address=self.pg3.local_ip4n, - path_mtu=512, - template_interval=10) - self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id, - src_port=self.ipfix_src_port) - data = "A" * 4 + "B" * 16 + "C" * 3 - self.tcp_port_in = random.randint(1025, 65535) - pkts = self.create_stream_frag(self.pg0, - self.pg1.remote_ip4, - self.tcp_port_in, - 20, - data) - self.pg0.add_stream(pkts[-1]) + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0) + start_sessnum = len(sessions) + + self.initiate_tcp_session(self.pg0, self.pg1) + + # FIN packet in -> out + 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=self.tcp_port_in, dport=self.tcp_external_port, + flags="FA", seq=100, ack=300)) + self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - frags = self.pg1.get_capture(0) - self.vapi.cli("ipfix flush") # FIXME this should be an API call - capture = self.pg3.get_capture(9) - ipfix = IPFIXDecoder() - # first load template - for p in capture: - self.assertTrue(p.haslayer(IPFIX)) - self.assertEqual(p[IP].src, self.pg3.local_ip4) - self.assertEqual(p[IP].dst, self.pg3.remote_ip4) - self.assertEqual(p[UDP].sport, self.ipfix_src_port) - self.assertEqual(p[UDP].dport, 4739) - self.assertEqual(p[IPFIX].observationDomainID, - self.ipfix_domain_id) - if p.haslayer(Template): - ipfix.add_template(p.getlayer(Template)) - # verify events in data set - for p in capture: - if p.haslayer(Data): - data = ipfix.decode_data_set(p.getlayer(Set)) - self.verify_ipfix_max_fragments_ip4(data, 0, - self.pg0.remote_ip4n) + self.pg1.get_capture(1) - def test_tcp_session_close_in(self): + pkts = [] + + # ACK packet out -> in + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / + TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, + flags="A", seq=300, ack=101)) + pkts.append(p) + + # FIN packet out -> in + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / + TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, + flags="FA", seq=300, ack=101)) + pkts.append(p) + + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg0.get_capture(2) + + # ACK packet in -> out + 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=self.tcp_port_in, dport=self.tcp_external_port, + flags="A", seq=101, ack=301)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg1.get_capture(1) + + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, + 0) + self.assertEqual(len(sessions) - start_sessnum, 0) + + def test_tcp_session_close_out(self): + """ Close TCP session from outside network """ + self.tcp_port_out = 10505 + self.nat44_add_address(self.nat_addr) + self.nat44_add_static_mapping(self.pg0.remote_ip4, + self.nat_addr, + self.tcp_port_in, + self.tcp_port_out, + proto=IP_PROTOS.tcp, + twice_nat=1) + 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) + + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0) + start_sessnum = len(sessions) + + self.initiate_tcp_session(self.pg0, self.pg1) + + # FIN packet out -> in + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / + TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, + flags="FA", seq=100, ack=300)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg0.get_capture(1) + + # FIN+ACK packet in -> out + 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=self.tcp_port_in, dport=self.tcp_external_port, + flags="FA", seq=300, ack=101)) + + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg1.get_capture(1) + + # ACK packet out -> in + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / + TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, + flags="A", seq=101, ack=301)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg0.get_capture(1) + + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, + 0) + self.assertEqual(len(sessions) - start_sessnum, 0) + + def test_tcp_session_close_simultaneous(self): """ Close TCP session from inside network """ + self.tcp_port_out = 10505 self.nat44_add_address(self.nat_addr) + self.nat44_add_static_mapping(self.pg0.remote_ip4, + self.nat_addr, + self.tcp_port_in, + self.tcp_port_out, + proto=IP_PROTOS.tcp, + twice_nat=1) 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) @@ -4137,193 +4165,273 @@ class TestNAT44(MethodHolder): self.initiate_tcp_session(self.pg0, self.pg1) - # close the session from inside - try: - # FIN packet in -> out - 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=self.tcp_port_in, dport=self.tcp_external_port, - flags="FA")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) + # FIN packet in -> out + 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=self.tcp_port_in, dport=self.tcp_external_port, + flags="FA", seq=100, ack=300)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg1.get_capture(1) - pkts = [] + # FIN packet out -> in + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / + TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, + flags="FA", seq=300, ack=100)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg0.get_capture(1) - # ACK packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, - flags="A")) - pkts.append(p) + # ACK packet in -> out + 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=self.tcp_port_in, dport=self.tcp_external_port, + flags="A", seq=101, ack=301)) + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg1.get_capture(1) - # FIN packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, - flags="FA")) - pkts.append(p) + # ACK packet out -> in + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / + TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, + flags="A", seq=301, ack=101)) + self.pg1.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.pg0.get_capture(1) - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(2) + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, + 0) + self.assertEqual(len(sessions) - start_sessnum, 0) - # ACK packet in -> out - 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=self.tcp_port_in, dport=self.tcp_external_port, - flags="A")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) + def test_one_armed_nat44_static(self): + """ One armed NAT44 and 1:1 NAPT asymmetrical rule """ + remote_host = self.pg4.remote_hosts[0] + local_host = self.pg4.remote_hosts[1] + external_port = 80 + local_port = 8080 + eh_port_in = 0 + + self.vapi.nat44_forwarding_enable_disable(1) + self.nat44_add_address(self.nat_addr, twice_nat=1) + self.nat44_add_static_mapping(local_host.ip4, self.nat_addr, + local_port, external_port, + proto=IP_PROTOS.tcp, out2in_only=1, + twice_nat=1) + self.vapi.nat44_interface_add_del_feature(self.pg4.sw_if_index) + self.vapi.nat44_interface_add_del_feature(self.pg4.sw_if_index, + is_inside=0) + + # from client to service + p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) / + IP(src=remote_host.ip4, dst=self.nat_addr) / + TCP(sport=12345, dport=external_port)) + self.pg4.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg4.get_capture(1) + p = capture[0] + try: + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.dst, local_host.ip4) + self.assertEqual(ip.src, self.nat_addr) + self.assertEqual(tcp.dport, local_port) + self.assertNotEqual(tcp.sport, 12345) + eh_port_in = tcp.sport + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise - self.initiate_tcp_session(self.pg0, self.pg1) - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, - 0) - self.assertEqual(len(sessions) - start_sessnum, 2) + # from service back to client + p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) / + IP(src=local_host.ip4, dst=self.nat_addr) / + TCP(sport=local_port, dport=eh_port_in)) + self.pg4.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg4.get_capture(1) + p = capture[0] + try: + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.src, self.nat_addr) + self.assertEqual(ip.dst, remote_host.ip4) + self.assertEqual(tcp.sport, external_port) + self.assertEqual(tcp.dport, 12345) + self.assert_packet_checksums_valid(p) except: - self.logger.error("TCP session termination failed") + self.logger.error(ppp("Unexpected or invalid packet:", p)) raise - def test_tcp_session_close_out(self): - """ Close TCP session from outside network """ - self.nat44_add_address(self.nat_addr) + def test_static_with_port_out2(self): + """ 1:1 NAPT asymmetrical rule """ + + external_port = 80 + local_port = 8080 + + self.vapi.nat44_forwarding_enable_disable(1) + self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_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.pg1.sw_if_index, is_inside=0) - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0) - start_sessnum = len(sessions) - - self.initiate_tcp_session(self.pg0, self.pg1) - - # close the session from outside + # from client to service + p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_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] try: - # FIN packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, - flags="FA")) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) - - pkts = [] + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.dst, self.pg0.remote_ip4) + self.assertEqual(tcp.dport, local_port) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise - # ACK packet in -> out - 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=self.tcp_port_in, dport=self.tcp_external_port, - flags="A")) - pkts.append(p) + # ICMP error + p = (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) / capture[0][IP]) + 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: + self.assertEqual(p[IP].src, self.nat_addr) + inner = p[IPerror] + self.assertEqual(inner.dst, self.nat_addr) + self.assertEqual(inner[TCPerror].dport, external_port) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise - # ACK packet in -> out - 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=self.tcp_port_in, dport=self.tcp_external_port, - flags="FA")) - pkts.append(p) + # 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, self.nat_addr) + self.assertEqual(tcp.sport, external_port) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise - self.pg0.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(2) + # ICMP error + p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / + ICMP(type=11) / capture[0][IP]) + 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: + self.assertEqual(p[IP].dst, self.pg0.remote_ip4) + inner = p[IPerror] + self.assertEqual(inner.src, self.pg0.remote_ip4) + self.assertEqual(inner[TCPerror].sport, local_port) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise - # ACK packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, - flags="A")) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) + # from client to server (no translation) + 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=12346, dport=local_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] + try: + ip = p[IP] + tcp = p[TCP] + self.assertEqual(ip.dst, self.pg0.remote_ip4) + self.assertEqual(tcp.dport, local_port) + self.assert_packet_checksums_valid(p) + except: + self.logger.error(ppp("Unexpected or invalid packet:", p)) + raise - self.initiate_tcp_session(self.pg0, self.pg1) - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, - 0) - self.assertEqual(len(sessions) - start_sessnum, 2) + # from service back to client (no translation) + 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=12346)) + 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, local_port) + self.assert_packet_checksums_valid(p) except: - self.logger.error("TCP session termination failed") + self.logger.error(ppp("Unexpected or invalid packet:", p)) raise - def test_tcp_session_close_simultaneous(self): - """ Close TCP session from inside network """ + def test_output_feature(self): + """ NAT44 interface output feature (in2out postrouting) """ + self.vapi.nat44_forwarding_enable_disable(1) 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, + 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) - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0) - start_sessnum = len(sessions) - - self.initiate_tcp_session(self.pg0, self.pg1) - - # close the session from inside - try: - # FIN packet in -> out - 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=self.tcp_port_in, dport=self.tcp_external_port, - flags="FA")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - # FIN packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, - flags="FA")) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) - - # ACK packet in -> out - 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=self.tcp_port_in, dport=self.tcp_external_port, - flags="A")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - # ACK packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, - flags="A")) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) + # in2out + 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) - self.initiate_tcp_session(self.pg0, self.pg1) - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, - 0) - self.assertEqual(len(sessions) - start_sessnum, 2) - except: - self.logger.error("TCP session termination failed") - raise + # out2in + 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 tearDown(self): - super(TestNAT44, self).tearDown() + super(TestNAT44EndpointDependent, self).tearDown() if not self.vpp_dead: self.logger.info(self.vapi.cli("show nat44 addresses")) self.logger.info(self.vapi.cli("show nat44 interfaces")) self.logger.info(self.vapi.cli("show nat44 static mappings")) self.logger.info(self.vapi.cli("show nat44 interface address")) self.logger.info(self.vapi.cli("show nat44 sessions detail")) - self.logger.info(self.vapi.cli("show nat virtual-reassembly")) - self.vapi.cli("nat addr-port-assignment-alg default") + self.logger.info(self.vapi.cli("show nat44 hash tables detail")) self.clear_nat44() + self.vapi.cli("clear logging") class TestNAT44Out2InDPO(MethodHolder): @@ -4337,6 +4445,7 @@ class TestNAT44Out2InDPO(MethodHolder): @classmethod def setUpClass(cls): super(TestNAT44Out2InDPO, cls).setUpClass() + cls.vapi.cli("set log class nat level debug") try: cls.tcp_port_in = 6303 @@ -4454,6 +4563,7 @@ class TestDeterministicNAT(MethodHolder): @classmethod def setUpClass(cls): super(TestDeterministicNAT, cls).setUpClass() + cls.vapi.cli("set log class nat level debug") try: cls.tcp_port_in = 6303 @@ -5370,8 +5480,6 @@ class TestNAT64(MethodHolder): self.udp_port_in = 6304 self.icmp_id_in = 6305 - ses_num_start = self.nat64_get_ses_num() - self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n, self.nat_addr_n) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) @@ -5414,7 +5522,7 @@ class TestNAT64(MethodHolder): inner = packet[IPerror] self.assertEqual(inner.src, self.pg1.remote_ip4) self.assertEqual(inner.dst, self.nat_addr) - self.check_icmp_checksum(packet) + self.assert_packet_checksums_valid(packet) if inner.haslayer(TCPerror): self.assertEqual(inner[TCPerror].dport, self.tcp_port_out) elif inner.haslayer(UDPerror): @@ -5443,7 +5551,7 @@ class TestNAT64(MethodHolder): inner = icmp[IPerror6] self.assertEqual(inner.src, self.pg0.remote_ip6) self.assertEqual(inner.dst, ip.src) - self.check_icmpv6_checksum(packet) + self.assert_icmpv6_checksum_valid(packet) if inner.haslayer(TCPerror): self.assertEqual(inner[TCPerror].sport, self.tcp_port_in) elif inner.haslayer(UDPerror): @@ -5505,15 +5613,14 @@ class TestNAT64(MethodHolder): try: self.assertEqual(packet[IPv6].src, nat_addr_ip6) self.assertEqual(packet[IPv6].dst, server.ip6) + self.assert_packet_checksums_valid(packet) if packet.haslayer(TCP): self.assertNotEqual(packet[TCP].sport, client_tcp_in_port) self.assertEqual(packet[TCP].dport, server_tcp_in_port) - self.check_tcp_checksum(packet) client_tcp_out_port = packet[TCP].sport else: self.assertNotEqual(packet[UDP].sport, client_udp_in_port) self.assertEqual(packet[UDP].dport, server_udp_in_port) - self.check_udp_checksum(packet) client_udp_out_port = packet[UDP].sport except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) @@ -5537,14 +5644,13 @@ class TestNAT64(MethodHolder): try: self.assertEqual(packet[IPv6].src, nat_addr_ip6) self.assertEqual(packet[IPv6].dst, client.ip6) + self.assert_packet_checksums_valid(packet) if packet.haslayer(TCP): self.assertEqual(packet[TCP].sport, server_tcp_out_port) self.assertEqual(packet[TCP].dport, client_tcp_in_port) - self.check_tcp_checksum(packet) else: self.assertEqual(packet[UDP].sport, server_udp_out_port) self.assertEqual(packet[UDP].dport, client_udp_in_port) - self.check_udp_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise @@ -5568,7 +5674,7 @@ class TestNAT64(MethodHolder): inner = icmp[IPerror6] self.assertEqual(inner.src, server.ip6) self.assertEqual(inner.dst, nat_addr_ip6) - self.check_icmpv6_checksum(packet) + self.assert_packet_checksums_valid(packet) if inner.haslayer(TCPerror): self.assertEqual(inner[TCPerror].sport, server_tcp_in_port) self.assertEqual(inner[TCPerror].dport, @@ -5691,7 +5797,7 @@ class TestNAT64(MethodHolder): self.assertEqual(packet[IP].src, self.nat_addr) self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) self.assertTrue(packet.haslayer(GRE)) - self.check_ip_checksum(packet) + self.assert_packet_checksums_valid(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise @@ -5829,8 +5935,7 @@ class TestNAT64(MethodHolder): self.assertNotEqual(tcp.sport, 12345) external_port = tcp.sport self.assertEqual(tcp.dport, 80) - self.check_tcp_checksum(p) - self.check_ip_checksum(p) + self.assert_packet_checksums_valid(p) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise @@ -5851,7 +5956,7 @@ class TestNAT64(MethodHolder): self.assertEqual(ip.dst, self.pg3.remote_ip6) self.assertEqual(tcp.sport, 80) self.assertEqual(tcp.dport, 12345) - self.check_tcp_checksum(p) + self.assert_packet_checksums_valid(p) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise @@ -5909,7 +6014,6 @@ class TestNAT64(MethodHolder): def test_reass_hairpinning(self): """ NAT64 fragments hairpinning """ data = 'a' * 200 - client = self.pg0.remote_hosts[0] server = self.pg0.remote_hosts[1] server_in_port = random.randint(1025, 65535) server_out_port = random.randint(1025, 65535) @@ -6051,7 +6155,8 @@ class TestNAT64(MethodHolder): self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - self.pg1.get_capture(0) + self.pg1.assert_nothing_captured() + sleep(1) self.vapi.cli("ipfix flush") # FIXME this should be an API call capture = self.pg3.get_capture(9) ipfix = IPFIXDecoder() @@ -6078,7 +6183,8 @@ class TestNAT64(MethodHolder): self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - self.pg1.get_capture(0) + self.pg1.assert_nothing_captured() + sleep(1) self.vapi.cli("ipfix flush") # FIXME this should be an API call capture = self.pg3.get_capture(1) # verify events in data set @@ -6114,7 +6220,8 @@ class TestNAT64(MethodHolder): self.pg0.add_stream(pkts[-1]) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - self.pg1.get_capture(0) + self.pg1.assert_nothing_captured() + sleep(1) self.vapi.cli("ipfix flush") # FIXME this should be an API call capture = self.pg3.get_capture(9) ipfix = IPFIXDecoder() @@ -6335,7 +6442,7 @@ class TestDSlite(MethodHolder): self.assertEqual(capture[IP].dst, self.pg0.remote_ip4) self.assertNotEqual(capture[UDP].sport, 20000) self.assertEqual(capture[UDP].dport, 10000) - self.check_ip_checksum(capture) + self.assert_packet_checksums_valid(capture) out_port = capture[UDP].sport p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / @@ -6352,7 +6459,7 @@ class TestDSlite(MethodHolder): self.assertEqual(capture[IP].dst, '192.168.1.1') self.assertEqual(capture[UDP].sport, 10000) self.assertEqual(capture[UDP].dport, 20000) - self.check_ip_checksum(capture) + self.assert_packet_checksums_valid(capture) # TCP p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / @@ -6369,8 +6476,7 @@ class TestDSlite(MethodHolder): self.assertEqual(capture[IP].dst, self.pg0.remote_ip4) self.assertNotEqual(capture[TCP].sport, 20001) self.assertEqual(capture[TCP].dport, 10001) - self.check_ip_checksum(capture) - self.check_tcp_checksum(capture) + self.assert_packet_checksums_valid(capture) out_port = capture[TCP].sport p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / @@ -6387,8 +6493,7 @@ class TestDSlite(MethodHolder): self.assertEqual(capture[IP].dst, '192.168.1.1') self.assertEqual(capture[TCP].sport, 10001) self.assertEqual(capture[TCP].dport, 20001) - self.check_ip_checksum(capture) - self.check_tcp_checksum(capture) + self.assert_packet_checksums_valid(capture) # ICMP p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / @@ -6404,8 +6509,7 @@ class TestDSlite(MethodHolder): self.assertEqual(capture[IP].src, self.nat_addr) self.assertEqual(capture[IP].dst, self.pg0.remote_ip4) self.assertNotEqual(capture[ICMP].id, 4000) - self.check_ip_checksum(capture) - self.check_icmp_checksum(capture) + self.assert_packet_checksums_valid(capture) out_id = capture[ICMP].id p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / @@ -6421,8 +6525,7 @@ class TestDSlite(MethodHolder): self.assertEqual(capture[IP].src, self.pg0.remote_ip4) self.assertEqual(capture[IP].dst, '192.168.1.1') self.assertEqual(capture[ICMP].id, 4000) - self.check_ip_checksum(capture) - self.check_icmp_checksum(capture) + self.assert_packet_checksums_valid(capture) # ping DS-Lite AFTR tunnel endpoint address p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / @@ -6509,7 +6612,7 @@ class TestDSliteCE(MethodHolder): self.assertEqual(capture[IP].dst, self.pg1.remote_ip4) self.assertEqual(capture[UDP].sport, 10000) self.assertEqual(capture[UDP].dport, 20000) - self.check_ip_checksum(capture) + self.assert_packet_checksums_valid(capture) # UDP decapsulation p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / @@ -6526,7 +6629,7 @@ class TestDSliteCE(MethodHolder): self.assertEqual(capture[IP].dst, self.pg0.remote_ip4) self.assertEqual(capture[UDP].sport, 20000) self.assertEqual(capture[UDP].dport, 10000) - self.check_ip_checksum(capture) + self.assert_packet_checksums_valid(capture) # ping DS-Lite B4 tunnel endpoint address p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / @@ -6607,12 +6710,7 @@ class TestNAT66(MethodHolder): 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) + self.assert_packet_checksums_valid(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise @@ -6643,12 +6741,7 @@ class TestNAT66(MethodHolder): 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) + self.assert_packet_checksums_valid(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise @@ -6704,5 +6797,6 @@ class TestNAT66(MethodHolder): self.logger.info(self.vapi.cli("show nat66 static mappings")) self.clear_nat66() + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)