+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+
+ data = "A" * 4 + "B" * 16 + "C" * 3
+ random.randint(1025, 65535)
+
+ # in2out
+ pkts = self.create_stream_frag(self.pg0,
+ self.pg1.remote_ip4,
+ self.tcp_port_in,
+ 20,
+ data)
+ pkts.reverse()
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ frags = self.pg1.get_capture(len(pkts))
+ p = self.reass_frags_and_verify(frags,
+ self.nat_addr,
+ self.pg1.remote_ip4)
+ self.assertEqual(p[TCP].dport, 20)
+ self.assertNotEqual(p[TCP].sport, self.tcp_port_in)
+ self.tcp_port_out = p[TCP].sport
+ self.assertEqual(data, p[Raw].load)
+
+ # out2in
+ pkts = self.create_stream_frag(self.pg1,
+ self.nat_addr,
+ 20,
+ self.tcp_port_out,
+ data)
+ pkts.reverse()
+ 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)
+
+ 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 twice_nat_common(self, self_twice_nat=False, same_pg=False, lb=False,
+ client_id=None):
+ twice_nat_addr = '10.0.1.3'
+
+ port_in = 8080
+ if lb:
+ if not same_pg:
+ port_in1 = port_in
+ port_in2 = port_in
+ else:
+ port_in1 = port_in+1
+ port_in2 = port_in+2
+
+ port_out = 80
+ eh_port_out = 4567
+
+ server1 = self.pg0.remote_hosts[0]
+ server2 = self.pg0.remote_hosts[1]
+ if lb and same_pg:
+ server2 = server1
+ if not lb:
+ server = server1
+
+ pg0 = self.pg0
+ if same_pg:
+ pg1 = self.pg0
+ else:
+ pg1 = self.pg1
+
+ eh_translate = ((not self_twice_nat) or (not lb and same_pg) or
+ client_id == 1)
+
+ self.nat44_add_address(self.nat_addr)
+ self.nat44_add_address(twice_nat_addr, twice_nat=1)
+ if not lb:
+ self.nat44_add_static_mapping(pg0.remote_ip4, self.nat_addr,
+ port_in, port_out,
+ proto=IP_PROTOS.tcp,
+ twice_nat=int(not self_twice_nat),
+ self_twice_nat=int(self_twice_nat))
+ else:
+ locals = [{'addr': server1.ip4n,
+ 'port': port_in1,
+ 'probability': 50},
+ {'addr': server2.ip4n,
+ 'port': port_in2,
+ 'probability': 50}]
+ out_addr_n = socket.inet_pton(socket.AF_INET, self.nat_addr)
+ self.vapi.nat44_add_del_lb_static_mapping(out_addr_n,
+ port_out,
+ IP_PROTOS.tcp,
+ twice_nat=int(
+ not self_twice_nat),
+ self_twice_nat=int(
+ self_twice_nat),
+ local_num=len(locals),
+ locals=locals)
+ self.vapi.nat44_interface_add_del_feature(pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(pg1.sw_if_index,
+ is_inside=0)
+
+ if same_pg:
+ if not lb:
+ client = server
+ else:
+ assert client_id is not None
+ if client_id == 1:
+ client = self.pg0.remote_hosts[0]
+ elif client_id == 2:
+ client = self.pg0.remote_hosts[1]
+ else:
+ client = pg1.remote_hosts[0]
+ p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
+ IP(src=client.ip4, dst=self.nat_addr) /
+ TCP(sport=eh_port_out, dport=port_out))
+ pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = pg0.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ if lb:
+ if ip.dst == server1.ip4:
+ server = server1
+ port_in = port_in1
+ else:
+ server = server2
+ port_in = port_in2
+ self.assertEqual(ip.dst, server.ip4)
+ if lb and same_pg:
+ self.assertIn(tcp.dport, [port_in1, port_in2])
+ else:
+ self.assertEqual(tcp.dport, port_in)
+ if eh_translate:
+ self.assertEqual(ip.src, twice_nat_addr)
+ self.assertNotEqual(tcp.sport, eh_port_out)
+ else:
+ self.assertEqual(ip.src, client.ip4)
+ self.assertEqual(tcp.sport, eh_port_out)
+ eh_addr_in = ip.src
+ eh_port_in = tcp.sport
+ saved_port_in = tcp.dport
+ self.assert_packet_checksums_valid(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ p = (Ether(src=server.mac, dst=pg0.local_mac) /
+ IP(src=server.ip4, dst=eh_addr_in) /
+ TCP(sport=saved_port_in, dport=eh_port_in))
+ pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = pg1.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ self.assertEqual(ip.dst, client.ip4)
+ self.assertEqual(ip.src, self.nat_addr)
+ self.assertEqual(tcp.dport, eh_port_out)
+ self.assertEqual(tcp.sport, port_out)
+ 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()
+
+ def test_self_twice_nat_positive(self):
+ """ Self Twice NAT44 (positive test) """
+ self.twice_nat_common(self_twice_nat=True, same_pg=True)
+
+ def test_self_twice_nat_negative(self):
+ """ Self Twice NAT44 (negative test) """
+ self.twice_nat_common(self_twice_nat=True)
+
+ def test_twice_nat_lb(self):
+ """ Twice NAT44 local service load balancing """
+ self.twice_nat_common(lb=True)
+
+ def test_self_twice_nat_lb_positive(self):
+ """ Self Twice NAT44 local service load balancing (positive test) """
+ self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True,
+ client_id=1)
+
+ def test_self_twice_nat_lb_negative(self):
+ """ Self Twice NAT44 local service load balancing (negative test) """
+ self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True,
+ client_id=2)
+
+ 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)
+
+ # 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()
+ 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].twice_nat, 1)
+
+ # remove interface address and check NAT address pool
+ self.pg7.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 """
+ 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()
+ 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)
+
+ def test_tcp_session_close_in(self):
+ """ Close TCP session from inside network """
+ 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)
+
+ 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", 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 = []
+
+ # 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)
+
+ 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, 1)
+ except:
+ self.logger.error("TCP session termination failed")
+ raise
+
+ def test_tcp_session_close_out(self):
+ """ Close TCP session from outside network """
+ 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)
+
+ 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
+ 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", 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)
+
+ 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, 1)
+ except:
+ self.logger.error("TCP session termination failed")
+ raise
+
+ def test_tcp_session_close_simultaneous(self):
+ """ Close TCP session from inside network """
+ 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)
+
+ 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", seq=100, ack=300))
+ 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", 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 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)
+
+ # 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.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, 1)
+ except:
+ self.logger.error("TCP session termination failed")
+ raise
+
+ 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.vapi.cli("nat addr-port-assignment-alg default")
+ self.clear_nat44()
+
+
+class TestNAT44Out2InDPO(MethodHolder):
+ """ NAT44 Test Cases using out2in DPO """
+
+ @classmethod
+ def setUpConstants(cls):
+ super(TestNAT44Out2InDPO, cls).setUpConstants()
+ cls.vpp_cmdline.extend(["nat", "{", "out2in dpo", "}"])
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestNAT44Out2InDPO, cls).setUpClass()
+
+ 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.dst_ip4 = '192.168.70.1'
+
+ cls.create_pg_interfaces(range(2))
+
+ cls.pg0.admin_up()
+ cls.pg0.config_ip4()
+ cls.pg0.resolve_arp()
+
+ cls.pg1.admin_up()
+ cls.pg1.config_ip6()
+ cls.pg1.resolve_ndp()
+
+ cls.vapi.ip_add_del_route(is_ipv6=True, dst_address='\x00'*16,
+ dst_address_length=0,
+ next_hop_address=cls.pg1.remote_ip6n,
+ next_hop_sw_if_index=cls.pg1.sw_if_index)
+
+ except Exception:
+ super(TestNAT44Out2InDPO, cls).tearDownClass()
+ raise
+
+ def configure_xlat(self):
+ self.dst_ip6_pfx = '1:2:3::'
+ self.dst_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
+ self.dst_ip6_pfx)
+ self.dst_ip6_pfx_len = 96
+ self.src_ip6_pfx = '4:5:6::'
+ self.src_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
+ self.src_ip6_pfx)
+ self.src_ip6_pfx_len = 96
+ self.vapi.map_add_domain(self.dst_ip6_pfx_n, self.dst_ip6_pfx_len,
+ self.src_ip6_pfx_n, self.src_ip6_pfx_len,
+ '\x00\x00\x00\x00', 0, is_translation=1,
+ is_rfc6052=1)
+
+ def test_464xlat_ce(self):
+ """ Test 464XLAT CE with NAT44 """
+
+ self.configure_xlat()
+
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_add_del_address_range(self.nat_addr_n, self.nat_addr_n)
Code Review / vpp.git / blobdiff