+ except Exception:
+ super(TestNAT44, 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
+ tcpn = self.statistics.get_counter(
+ '/err/nat44-in2out-slowpath/TCP packets')
+ udpn = self.statistics.get_counter(
+ '/err/nat44-in2out-slowpath/UDP packets')
+ icmpn = self.statistics.get_counter(
+ '/err/nat44-in2out-slowpath/ICMP packets')
+ totaln = self.statistics.get_counter(
+ '/err/nat44-in2out-slowpath/good in2out packets processed')
+
+ 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)
+
+ err = self.statistics.get_counter(
+ '/err/nat44-in2out-slowpath/TCP packets')
+ self.assertEqual(err - tcpn, 1)
+ err = self.statistics.get_counter(
+ '/err/nat44-in2out-slowpath/UDP packets')
+ self.assertEqual(err - udpn, 1)
+ err = self.statistics.get_counter(
+ '/err/nat44-in2out-slowpath/ICMP packets')
+ self.assertEqual(err - icmpn, 1)
+ err = self.statistics.get_counter(
+ '/err/nat44-in2out-slowpath/good in2out packets processed')
+ self.assertEqual(err - totaln, 3)
+
+ # out2in
+ tcpn = self.statistics.get_counter('/err/nat44-out2in/TCP packets')
+ udpn = self.statistics.get_counter('/err/nat44-out2in/UDP packets')
+ icmpn = self.statistics.get_counter('/err/nat44-out2in/ICMP packets')
+ totaln = self.statistics.get_counter(
+ '/err/nat44-out2in/good out2in packets processed')
+
+ 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)
+
+ err = self.statistics.get_counter('/err/nat44-out2in/TCP packets')
+ self.assertEqual(err - tcpn, 1)
+ err = self.statistics.get_counter('/err/nat44-out2in/UDP packets')
+ self.assertEqual(err - udpn, 1)
+ err = self.statistics.get_counter('/err/nat44-out2in/ICMP packets')
+ self.assertEqual(err - icmpn, 1)
+ err = self.statistics.get_counter(
+ '/err/nat44-out2in/good out2in packets processed')
+ self.assertEqual(err - totaln, 3)
+
+ users = self.statistics.get_counter('/nat44/total-users')
+ self.assertEqual(users[0][0], 1)
+ sessions = self.statistics.get_counter('/nat44/total-sessions')
+ self.assertEqual(sessions[0][0], 3)
+
+ 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 """
+
+ 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 - create sessions
+ pkts = self.create_stream_in(self.pg0, self.pg1)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Server side - generate traffic
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture)
+ pkts = self.create_stream_out(self.pg1, ttl=1)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Server side - verify ICMP type 11 packets
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out_with_icmp_errors(capture,
+ src_ip=self.pg1.local_ip4)
+
+ def test_dynamic_icmp_errors_in2out_ttl_2(self):
+ """ NAT44 handling of error responses to client packets with TTL=2 """
+
+ 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=2)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Server side - simulate ICMP type 11 response
+ capture = self.pg1.get_capture(len(pkts))
+ pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
+ ICMP(type=11) / packet[IP] for packet in capture]
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Client side - verify ICMP type 11 packets
+ capture = self.pg0.get_capture(len(pkts))
+ self.verify_capture_in_with_icmp_errors(capture, self.pg0)
+
+ def test_dynamic_icmp_errors_out2in_ttl_2(self):
+ """ NAT44 handling of error responses to server packets with TTL=2 """
+
+ 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 - create sessions
+ pkts = self.create_stream_in(self.pg0, self.pg1)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Server side - generate traffic
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture)
+ pkts = self.create_stream_out(self.pg1, ttl=2)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Client side - simulate ICMP type 11 response
+ capture = self.pg0.get_capture(len(pkts))
+ pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ ICMP(type=11) / packet[IP] for packet in capture]
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Server side - verify ICMP type 11 packets
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out_with_icmp_errors(capture)
+
+ def test_ping_out_interface_from_outside(self):
+ """ Ping NAT44 out interface 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)
+
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) /
+ ICMP(id=self.icmp_id_out, type='echo-request'))
+ pkts = [p]
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ packet = capture[0]
+ try:
+ self.assertEqual(packet[IP].src, self.pg1.local_ip4)
+ self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
+ self.assertEqual(packet[ICMP].id, self.icmp_id_in)
+ self.assertEqual(packet[ICMP].type, 0) # echo reply
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet "
+ "(outside network):", packet))
+ raise
+
+ def test_ping_internal_host_from_outside(self):
+ """ Ping internal host from outside network """
+
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, 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)
+
+ # out2in
+ pkt = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr, ttl=64) /
+ ICMP(id=self.icmp_id_out, type='echo-request'))
+ self.pg1.add_stream(pkt)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ self.verify_capture_in(capture, self.pg0)
+ self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
+
+ # in2out
+ pkt = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
+ ICMP(id=self.icmp_id_in, type='echo-reply'))
+ self.pg0.add_stream(pkt)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ self.verify_capture_out(capture, same_port=True)
+ self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
+
+ 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:
+ # 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)
+
+ # 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
+
+ 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_in(self):
+ """ 1:1 NAT initialized from inside network """
+
+ nat_ip = "10.0.0.10"
+ self.tcp_port_out = 6303
+ self.udp_port_out = 6304
+ self.icmp_id_out = 6305
+
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
+ 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)
+ sm = self.vapi.nat44_static_mapping_dump()
+ self.assertEqual(len(sm), 1)
+ self.assertEqual((sm[0].tag).split('\0', 1)[0], '')
+ self.assertEqual(sm[0].protocol, 0)
+ self.assertEqual(sm[0].local_port, 0)
+ self.assertEqual(sm[0].external_port, 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, nat_ip, True)
+
+ # out2in
+ pkts = self.create_stream_out(self.pg1, nat_ip)
+ 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_static_out(self):
+ """ 1:1 NAT initialized from outside network """
+
+ nat_ip = "10.0.0.20"
+ self.tcp_port_out = 6303
+ self.udp_port_out = 6304
+ self.icmp_id_out = 6305
+ tag = "testTAG"
+
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip, tag=tag)
+ 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)
+ sm = self.vapi.nat44_static_mapping_dump()
+ self.assertEqual(len(sm), 1)
+ self.assertEqual((sm[0].tag).split('\0', 1)[0], tag)
+
+ # out2in
+ pkts = self.create_stream_out(self.pg1, nat_ip)
+ 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)
+
+ # 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, nat_ip, True)
+
+ def test_static_with_port_in(self):
+ """ 1:1 NAPT initialized from inside network """
+
+ self.tcp_port_out = 3606
+ self.udp_port_out = 3607
+ self.icmp_id_out = 3608
+
+ 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)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.udp_port_in, self.udp_port_out,
+ proto=IP_PROTOS.udp)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.icmp_id_in, self.icmp_id_out,
+ proto=IP_PROTOS.icmp)
+ 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_static_with_port_out(self):
+ """ 1:1 NAPT initialized from outside network """
+
+ self.tcp_port_out = 30606
+ self.udp_port_out = 30607
+ self.icmp_id_out = 30608
+
+ 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)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.udp_port_in, self.udp_port_out,
+ proto=IP_PROTOS.udp)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.icmp_id_in, self.icmp_id_out,
+ proto=IP_PROTOS.icmp)
+ 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)
+
+ # 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)
+
+ # 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)
+
+ def test_static_vrf_aware(self):
+ """ 1:1 NAT VRF awareness """
+
+ nat_ip1 = "10.0.0.30"
+ nat_ip2 = "10.0.0.40"
+ self.tcp_port_out = 6303
+ self.udp_port_out = 6304
+ self.icmp_id_out = 6305
+
+ self.nat44_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
+ vrf_id=10)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
+ vrf_id=10)
+ self.vapi.nat44_interface_add_del_feature(self.pg3.sw_if_index,
+ is_inside=0)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg4.sw_if_index)
+
+ # inside interface VRF match NAT44 static mapping VRF
+ pkts = self.create_stream_in(self.pg4, self.pg3)
+ self.pg4.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, nat_ip1, True)
+
+ # inside interface VRF don't match NAT44 static mapping VRF (packets
+ # are dropped)
+ pkts = self.create_stream_in(self.pg0, self.pg3)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg3.assert_nothing_captured()
+
+ def test_dynamic_to_static(self):
+ """ Switch from dynamic translation to 1:1NAT """
+ nat_ip = "10.0.0.10"
+ self.tcp_port_out = 6303
+ self.udp_port_out = 6304
+ self.icmp_id_out = 6305
+
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+
+ # dynamic
+ pkts = self.create_stream_in(self.pg0, self.pg1)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture)
+
+ # 1:1NAT
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
+ self.assertEqual(len(sessions), 0)
+ pkts = self.create_stream_in(self.pg0, self.pg1)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip, True)
+
+ def test_identity_nat(self):
+ """ Identity NAT """
+
+ self.vapi.nat44_add_del_identity_mapping(ip=self.pg0.remote_ip4n)
+ 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)
+
+ 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=12345, dport=56789))
+ 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(ip.src, self.pg1.remote_ip4)
+ self.assertEqual(tcp.dport, 56789)
+ self.assertEqual(tcp.sport, 12345)
+ self.assert_packet_checksums_valid(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
+ self.assertEqual(len(sessions), 0)
+ self.vapi.nat44_add_del_identity_mapping(ip=self.pg0.remote_ip4n,
+ vrf_id=1)
+ identity_mappings = self.vapi.nat44_identity_mapping_dump()
+ self.assertEqual(len(identity_mappings), 2)
+
+ 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)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg3.sw_if_index,
+ is_inside=0)
+
+ # between two NAT44 inside interfaces (no translation)
+ pkts = self.create_stream_in(self.pg0, self.pg1)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_no_translation(capture, self.pg0, self.pg1)
+
+ # from NAT44 inside to interface without NAT44 feature (no translation)
+ pkts = self.create_stream_in(self.pg0, self.pg2)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg2.get_capture(len(pkts))
+ self.verify_capture_no_translation(capture, self.pg0, self.pg2)
+
+ # in2out 1st interface
+ pkts = self.create_stream_in(self.pg0, self.pg3)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg3.get_capture(len(pkts))
+ self.verify_capture_out(capture)
+
+ # out2in 1st interface
+ pkts = self.create_stream_out(self.pg3)
+ self.pg3.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg0)
+
+ # in2out 2nd interface
+ pkts = self.create_stream_in(self.pg1, self.pg3)
+ self.pg1.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 2nd interface
+ pkts = self.create_stream_out(self.pg3)
+ self.pg3.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg1)
+
+ def test_inside_overlapping_interfaces(self):
+ """ NAT44 multiple inside interfaces with overlapping address space """
+
+ static_nat_ip = "10.0.0.10"
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg3.sw_if_index,
+ is_inside=0)
+ self.vapi.nat44_interface_add_del_feature(self.pg4.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg5.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg6.sw_if_index)
+ self.nat44_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
+ vrf_id=20)
+
+ # between NAT44 inside interfaces with same VRF (no translation)
+ pkts = self.create_stream_in(self.pg4, self.pg5)
+ self.pg4.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg5.get_capture(len(pkts))
+ self.verify_capture_no_translation(capture, self.pg4, self.pg5)
+
+ # between NAT44 inside interfaces with different VRF (hairpinning)
+ p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
+ IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
+ TCP(sport=1234, dport=5678))
+ self.pg4.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg6.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ self.assertEqual(ip.src, self.nat_addr)
+ self.assertEqual(ip.dst, self.pg6.remote_ip4)
+ self.assertNotEqual(tcp.sport, 1234)
+ self.assertEqual(tcp.dport, 5678)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ # in2out 1st interface
+ pkts = self.create_stream_in(self.pg4, self.pg3)
+ self.pg4.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 1st interface
+ pkts = self.create_stream_out(self.pg3)
+ self.pg3.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg4.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg4)
+
+ # in2out 2nd interface
+ pkts = self.create_stream_in(self.pg5, self.pg3)
+ self.pg5.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 2nd interface
+ pkts = self.create_stream_out(self.pg3)
+ self.pg3.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg5.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg5)
+
+ # pg5 session dump
+ addresses = self.vapi.nat44_address_dump()
+ self.assertEqual(len(addresses), 1)
+ sessions = self.vapi.nat44_user_session_dump(self.pg5.remote_ip4n, 10)
+ self.assertEqual(len(sessions), 3)
+ for session in sessions:
+ self.assertFalse(session.is_static)
+ self.assertEqual(session.inside_ip_address[0:4],
+ self.pg5.remote_ip4n)
+ self.assertEqual(session.outside_ip_address,
+ addresses[0].ip_address)
+ self.assertEqual(sessions[0].protocol, IP_PROTOS.tcp)
+ self.assertEqual(sessions[1].protocol, IP_PROTOS.udp)
+ self.assertEqual(sessions[2].protocol, IP_PROTOS.icmp)
+ self.assertEqual(sessions[0].inside_port, self.tcp_port_in)
+ self.assertEqual(sessions[1].inside_port, self.udp_port_in)
+ self.assertEqual(sessions[2].inside_port, self.icmp_id_in)
+ self.assertEqual(sessions[0].outside_port, self.tcp_port_out)
+ self.assertEqual(sessions[1].outside_port, self.udp_port_out)
+ self.assertEqual(sessions[2].outside_port, self.icmp_id_out)
+
+ # in2out 3rd interface
+ pkts = self.create_stream_in(self.pg6, self.pg3)
+ self.pg6.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, static_nat_ip, True)
+
+ # out2in 3rd interface
+ pkts = self.create_stream_out(self.pg3, static_nat_ip)
+ self.pg3.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg6.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg6)
+
+ # general user and session dump verifications
+ users = self.vapi.nat44_user_dump()
+ self.assertGreaterEqual(len(users), 3)
+ addresses = self.vapi.nat44_address_dump()
+ self.assertEqual(len(addresses), 1)
+ for user in users:
+ sessions = self.vapi.nat44_user_session_dump(user.ip_address,
+ user.vrf_id)
+ for session in sessions:
+ self.assertEqual(user.ip_address, session.inside_ip_address)
+ self.assertTrue(session.total_bytes > session.total_pkts > 0)
+ 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)
+ self.assertGreaterEqual(len(sessions), 4)
+ for session in sessions:
+ self.assertFalse(session.is_static)
+ self.assertEqual(session.inside_ip_address[0:4],
+ self.pg4.remote_ip4n)
+ self.assertEqual(session.outside_ip_address,
+ addresses[0].ip_address)
+
+ # pg6 session dump
+ sessions = self.vapi.nat44_user_session_dump(self.pg6.remote_ip4n, 20)
+ self.assertGreaterEqual(len(sessions), 3)
+ for session in sessions:
+ self.assertTrue(session.is_static)
+ self.assertEqual(session.inside_ip_address[0:4],
+ self.pg6.remote_ip4n)
+ self.assertEqual(session.outside_ip_address,
+ socket.inet_pton(socket.AF_INET, static_nat_ip))
+ self.assertTrue(session.inside_port in
+ [self.tcp_port_in, self.udp_port_in,
+ self.icmp_id_in])
+
+ def test_hairpinning(self):
+ """ NAT44 hairpinning - 1:1 NAPT """
+
+ 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
+
+ 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, self.nat_addr,
+ server_in_port, server_out_port,
+ proto=IP_PROTOS.tcp)
+
+ # send packet from host to server
+ p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
+ IP(src=host.ip4, dst=self.nat_addr) /
+ 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 = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ self.assertEqual(ip.src, self.nat_addr)
+ self.assertEqual(ip.dst, server.ip4)
+ self.assertNotEqual(tcp.sport, host_in_port)
+ self.assertEqual(tcp.dport, server_in_port)
+ self.assert_packet_checksums_valid(p)
+ host_out_port = tcp.sport
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ # send reply from server to host
+ p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
+ IP(src=server.ip4, dst=self.nat_addr) /
+ TCP(sport=server_in_port, dport=host_out_port))
+ self.pg0.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, self.nat_addr)
+ self.assertEqual(ip.dst, host.ip4)
+ self.assertEqual(tcp.sport, server_out_port)
+ self.assertEqual(tcp.dport, host_in_port)
+ self.assert_packet_checksums_valid(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ def test_hairpinning2(self):
+ """ NAT44 hairpinning - 1:1 NAT"""
+
+ server1_nat_ip = "10.0.0.10"
+ server2_nat_ip = "10.0.0.11"
+ host = self.pg0.remote_hosts[0]
+ server1 = self.pg0.remote_hosts[1]
+ server2 = self.pg0.remote_hosts[2]
+ server_tcp_port = 22
+ server_udp_port = 20
+
+ 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 servers
+ self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
+ self.nat44_add_static_mapping(server2.ip4, server2_nat_ip)
+
+ # host to server1
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=host.ip4, dst=server1_nat_ip) /
+ TCP(sport=self.tcp_port_in, dport=server_tcp_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=host.ip4, dst=server1_nat_ip) /
+ UDP(sport=self.udp_port_in, dport=server_udp_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=host.ip4, dst=server1_nat_ip) /
+ ICMP(id=self.icmp_id_in, type='echo-request'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IP].src, self.nat_addr)
+ self.assertEqual(packet[IP].dst, server1.ip4)
+ if packet.haslayer(TCP):
+ 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.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)
+ self.udp_port_out = packet[UDP].sport
+ else:
+ self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
+ self.icmp_id_out = packet[ICMP].id
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # server1 to host
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server1.ip4, dst=self.nat_addr) /
+ TCP(sport=server_tcp_port, dport=self.tcp_port_out))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server1.ip4, dst=self.nat_addr) /
+ UDP(sport=server_udp_port, dport=self.udp_port_out))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server1.ip4, dst=self.nat_addr) /
+ ICMP(id=self.icmp_id_out, type='echo-reply'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IP].src, server1_nat_ip)
+ self.assertEqual(packet[IP].dst, host.ip4)
+ if packet.haslayer(TCP):
+ self.assertEqual(packet[TCP].dport, self.tcp_port_in)
+ self.assertEqual(packet[TCP].sport, server_tcp_port)
+ 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)
+ else:
+ self.assertEqual(packet[ICMP].id, self.icmp_id_in)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # server2 to server1
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server2.ip4, dst=server1_nat_ip) /
+ TCP(sport=self.tcp_port_in, dport=server_tcp_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server2.ip4, dst=server1_nat_ip) /
+ UDP(sport=self.udp_port_in, dport=server_udp_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server2.ip4, dst=server1_nat_ip) /
+ ICMP(id=self.icmp_id_in, type='echo-request'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IP].src, server2_nat_ip)
+ self.assertEqual(packet[IP].dst, server1.ip4)
+ if packet.haslayer(TCP):
+ 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.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)
+ self.udp_port_out = packet[UDP].sport
+ else:
+ self.assertEqual(packet[ICMP].id, self.icmp_id_in)
+ self.icmp_id_out = packet[ICMP].id
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # server1 to server2
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server1.ip4, dst=server2_nat_ip) /
+ TCP(sport=server_tcp_port, dport=self.tcp_port_out))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server1.ip4, dst=server2_nat_ip) /
+ UDP(sport=server_udp_port, dport=self.udp_port_out))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server1.ip4, dst=server2_nat_ip) /
+ ICMP(id=self.icmp_id_out, type='echo-reply'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IP].src, server1_nat_ip)
+ self.assertEqual(packet[IP].dst, server2.ip4)
+ if packet.haslayer(TCP):
+ self.assertEqual(packet[TCP].dport, self.tcp_port_in)
+ self.assertEqual(packet[TCP].sport, server_tcp_port)
+ 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)
+ else:
+ self.assertEqual(packet[ICMP].id, self.icmp_id_in)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ def test_max_translations_per_user(self):
+ """ MAX translations per user - recycle the least recently used """
+
+ 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)
+
+ # get maximum number of translations per user
+ nat44_config = self.vapi.nat_show_config()
+
+ # send more than maximum number of translations per user packets
+ pkts_num = nat44_config.max_translations_per_user + 5
+ pkts = []
+ for port in range(0, pkts_num):
+ 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=1025 + port))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # 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_del_interface_addr(self.pg7.sw_if_index)
+
+ # 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)
+
+ # 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_interface_addr_static_mapping(self):
+ """ Static mapping with addresses from interface """
+ tag = "testTAG"
+
+ self.vapi.nat44_add_del_interface_addr(self.pg7.sw_if_index)
+ self.nat44_add_static_mapping(
+ '1.2.3.4',
+ external_sw_if_index=self.pg7.sw_if_index,
+ tag=tag)
+
+ # static mappings with external interface
+ static_mappings = self.vapi.nat44_static_mapping_dump()
+ self.assertEqual(1, len(static_mappings))
+ self.assertEqual(self.pg7.sw_if_index,
+ static_mappings[0].external_sw_if_index)
+ self.assertEqual((static_mappings[0].tag).split('\0', 1)[0], tag)
+
+ # configure interface address and check static mappings
+ self.pg7.config_ip4()
+ static_mappings = self.vapi.nat44_static_mapping_dump()
+ self.assertEqual(2, len(static_mappings))
+ resolved = False
+ for sm in static_mappings:
+ if sm.external_sw_if_index == 0xFFFFFFFF:
+ self.assertEqual(sm.external_ip_address[0:4],
+ self.pg7.local_ip4n)
+ self.assertEqual((sm.tag).split('\0', 1)[0], tag)
+ resolved = True
+ self.assertTrue(resolved)
+
+ # remove interface address and check static mappings
+ self.pg7.unconfig_ip4()
+ static_mappings = self.vapi.nat44_static_mapping_dump()
+ self.assertEqual(1, len(static_mappings))
+ self.assertEqual(self.pg7.sw_if_index,
+ static_mappings[0].external_sw_if_index)
+ self.assertEqual((static_mappings[0].tag).split('\0', 1)[0], tag)
+
+ # configure interface address again and check static mappings
+ self.pg7.config_ip4()
+ static_mappings = self.vapi.nat44_static_mapping_dump()
+ self.assertEqual(2, len(static_mappings))
+ resolved = False
+ for sm in static_mappings:
+ if sm.external_sw_if_index == 0xFFFFFFFF:
+ self.assertEqual(sm.external_ip_address[0:4],
+ self.pg7.local_ip4n)
+ self.assertEqual((sm.tag).split('\0', 1)[0], tag)
+ resolved = True
+ self.assertTrue(resolved)
+
+ # remove static mapping
+ self.nat44_add_static_mapping(
+ '1.2.3.4',
+ external_sw_if_index=self.pg7.sw_if_index,
+ tag=tag,
+ is_add=0)
+ static_mappings = self.vapi.nat44_static_mapping_dump()
+ self.assertEqual(0, len(static_mappings))
+
+ def test_interface_addr_identity_nat(self):
+ """ Identity NAT with addresses from interface """
+
+ port = 53053
+ self.vapi.nat44_add_del_interface_addr(self.pg7.sw_if_index)
+ self.vapi.nat44_add_del_identity_mapping(
+ sw_if_index=self.pg7.sw_if_index,
+ port=port,
+ protocol=IP_PROTOS.tcp,
+ addr_only=0)
+
+ # identity mappings with external interface
+ identity_mappings = self.vapi.nat44_identity_mapping_dump()
+ self.assertEqual(1, len(identity_mappings))
+ self.assertEqual(self.pg7.sw_if_index,
+ identity_mappings[0].sw_if_index)
+
+ # configure interface address and check identity mappings
+ self.pg7.config_ip4()
+ identity_mappings = self.vapi.nat44_identity_mapping_dump()
+ resolved = False
+ self.assertEqual(2, len(identity_mappings))
+ for sm in identity_mappings:
+ if sm.sw_if_index == 0xFFFFFFFF:
+ self.assertEqual(identity_mappings[0].ip_address,
+ self.pg7.local_ip4n)
+ self.assertEqual(port, identity_mappings[0].port)
+ self.assertEqual(IP_PROTOS.tcp, identity_mappings[0].protocol)
+ resolved = True
+ self.assertTrue(resolved)
+
+ # remove interface address and check identity mappings
+ self.pg7.unconfig_ip4()
+ identity_mappings = self.vapi.nat44_identity_mapping_dump()
+ self.assertEqual(1, len(identity_mappings))
+ self.assertEqual(self.pg7.sw_if_index,
+ identity_mappings[0].sw_if_index)
+
+ def test_ipfix_nat44_sess(self):
+ """ IPFIX logging NAT44 session created/delted """
+ self.ipfix_domain_id = 10
+ self.ipfix_src_port = 20202
+ colector_port = 30303
+ bind_layers(UDP, IPFIX, dport=30303)
+ 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.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
+ src_address=self.pg3.local_ip4n,
+ path_mtu=512,
+ template_interval=10,
+ collector_port=colector_port)
+ self.vapi.nat_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port)
+
+ 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.nat44_add_address(self.nat_addr, is_add=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, colector_port)
+ 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_nat44_ses(data)
+
+ def test_ipfix_addr_exhausted(self):
+ """ IPFIX logging NAT addresses exhausted """
+ 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.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_enable_disable(domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port)
+
+ 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=3025))
+ self.pg0.add_stream(p)
+ 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_addr_exhausted(data)
+
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ def test_ipfix_max_sessions(self):
+ """ IPFIX logging maximum session entries 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)
+
+ nat44_config = self.vapi.nat_show_config()
+ max_sessions = 10 * nat44_config.translation_buckets
+
+ pkts = []
+ for i in range(0, max_sessions):
+ src = "10.10.%u.%u" % ((i & 0xFF00) >> 8, i & 0xFF)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=src, dst=self.pg1.remote_ip4) /
+ TCP(sport=1025))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ self.pg1.get_capture(max_sessions)
+ 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_enable_disable(domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port)
+
+ 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=1025))
+ self.pg0.add_stream(p)
+ 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_sessions(data, max_sessions)
+
+ def test_syslog_apmap(self):
+ """ Test syslog address and port mapping creation and deletion """
+ self.vapi.syslog_set_filter(SYSLOG_SEVERITY.INFO)
+ self.vapi.syslog_set_sender(self.pg3.local_ip4n, self.pg3.remote_ip4n)
+ 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)
+
+ 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()
+ capture = self.pg1.get_capture(1)
+ self.tcp_port_out = capture[0][TCP].sport
+ capture = self.pg3.get_capture(1)
+ self.verify_syslog_apmap(capture[0][Raw].load)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.nat44_add_address(self.nat_addr, is_add=0)
+ capture = self.pg3.get_capture(1)
+ self.verify_syslog_apmap(capture[0][Raw].load, False)
+
+ def test_pool_addr_fib(self):
+ """ NAT44 add pool addresses to FIB """
+ static_addr = '10.0.0.10'
+ 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.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr)
+
+ # NAT44 address
+ p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
+ ARP(op=ARP.who_has, pdst=self.nat_addr,
+ psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ self.assertTrue(capture[0].haslayer(ARP))
+ self.assertTrue(capture[0][ARP].op, ARP.is_at)
+
+ # 1:1 NAT address
+ p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
+ ARP(op=ARP.who_has, pdst=static_addr,
+ psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ self.assertTrue(capture[0].haslayer(ARP))
+ self.assertTrue(capture[0][ARP].op, ARP.is_at)
+
+ # send ARP to non-NAT44 interface
+ p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
+ ARP(op=ARP.who_has, pdst=self.nat_addr,
+ psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
+ self.pg2.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.assert_nothing_captured()
+
+ # remove addresses and verify
+ self.nat44_add_address(self.nat_addr, is_add=0)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr,
+ is_add=0)
+
+ p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
+ ARP(op=ARP.who_has, pdst=self.nat_addr,
+ psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ 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,
+ psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.assert_nothing_captured()
+
+ def test_vrf_mode(self):
+ """ NAT44 tenant VRF aware address pool mode """
+
+ vrf_id1 = 1
+ vrf_id2 = 2
+ nat_ip1 = "10.0.0.10"
+ nat_ip2 = "10.0.0.11"
+
+ self.pg0.unconfig_ip4()
+ self.pg1.unconfig_ip4()
+ self.vapi.ip_table_add_del(is_add=1, table_id=vrf_id1)
+ self.vapi.ip_table_add_del(is_add=1, table_id=vrf_id2)
+ self.pg0.set_table_ip4(vrf_id1)
+ self.pg1.set_table_ip4(vrf_id2)
+ self.pg0.config_ip4()
+ self.pg1.config_ip4()
+ self.pg0.resolve_arp()
+ self.pg1.resolve_arp()
+
+ self.nat44_add_address(nat_ip1, vrf_id=vrf_id1)
+ self.nat44_add_address(nat_ip2, vrf_id=vrf_id2)
+ 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.pg2.sw_if_index,
+ is_inside=0)
+
+ try:
+ # first VRF
+ pkts = self.create_stream_in(self.pg0, self.pg2)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg2.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip1)
+
+ # second VRF
+ pkts = self.create_stream_in(self.pg1, self.pg2)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg2.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip2)
+
+ finally:
+ self.pg0.unconfig_ip4()
+ self.pg1.unconfig_ip4()
+ self.pg0.set_table_ip4(0)
+ self.pg1.set_table_ip4(0)
+ self.pg0.config_ip4()
+ self.pg1.config_ip4()
+ self.pg0.resolve_arp()
+ self.pg1.resolve_arp()
+ self.vapi.ip_table_add_del(is_add=0, table_id=vrf_id1)
+ self.vapi.ip_table_add_del(is_add=0, table_id=vrf_id2)
+
+ def test_vrf_feature_independent(self):
+ """ NAT44 tenant VRF independent address pool mode """
+
+ nat_ip1 = "10.0.0.10"
+ nat_ip2 = "10.0.0.11"
+
+ self.nat44_add_address(nat_ip1)
+ self.nat44_add_address(nat_ip2, vrf_id=99)
+ 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.pg2.sw_if_index,
+ is_inside=0)
+
+ # first VRF
+ pkts = self.create_stream_in(self.pg0, self.pg2)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg2.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip1)
+
+ # second VRF
+ pkts = self.create_stream_in(self.pg1, self.pg2)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg2.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip1)
+
+ def test_dynamic_ipless_interfaces(self):
+ """ NAT44 interfaces without configured IP address """
+
+ self.vapi.ip_neighbor_add_del(
+ self.pg7.sw_if_index,
+ self.pg7.remote_mac,
+ self.pg7.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))
+ self.vapi.ip_neighbor_add_del(
+ self.pg8.sw_if_index,
+ self.pg8.remote_mac,
+ self.pg8.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))
+
+ 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)
+ 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)
+
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index,
+ is_inside=0)
+
+ # in2out
+ pkts = self.create_stream_in(self.pg7, self.pg8)
+ self.pg7.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg8.get_capture(len(pkts))
+ self.verify_capture_out(capture)
+
+ # out2in
+ pkts = self.create_stream_out(self.pg8, self.nat_addr)
+ self.pg8.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg7.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg7)
+
+ def test_static_ipless_interfaces(self):
+ """ NAT44 interfaces without configured IP address - 1:1 NAT """
+
+ self.vapi.ip_neighbor_add_del(
+ self.pg7.sw_if_index,
+ self.pg7.remote_mac,
+ self.pg7.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))
+ self.vapi.ip_neighbor_add_del(
+ self.pg8.sw_if_index,
+ self.pg8.remote_mac,
+ self.pg8.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))