from util import ppp
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from time import sleep
+from util import ip4_range
class MethodHolder(VppTestCase):
cls.ipfix_src_port = 4739
cls.ipfix_domain_id = 1
- cls.create_pg_interfaces(range(9))
+ cls.create_pg_interfaces(range(10))
cls.interfaces = list(cls.pg_interfaces[0:4])
for i in cls.interfaces:
cls.pg0.configure_ipv4_neighbors()
cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
+ cls.vapi.ip_table_add_del(10, is_add=1)
+ cls.vapi.ip_table_add_del(20, is_add=1)
cls.pg4._local_ip4 = "172.16.255.1"
cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
cls.pg7.admin_up()
cls.pg8.admin_up()
+ cls.pg9.generate_remote_hosts(2)
+ cls.pg9.config_ip4()
+ ip_addr_n = socket.inet_pton(socket.AF_INET, "10.0.0.1")
+ cls.vapi.sw_interface_add_del_address(cls.pg9.sw_if_index,
+ ip_addr_n,
+ 24)
+ cls.pg9.admin_up()
+ cls.pg9.resolve_arp()
+ cls.pg9._remote_hosts[1]._ip4 = cls.pg9._remote_hosts[0]._ip4
+ cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
+ cls.pg9.resolve_arp()
+
except Exception:
super(TestNAT44, cls).tearDownClass()
raise
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)
protocol=sm.protocol,
is_add=0)
+ 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,
+ lb_sm.vrf_id,
+ is_add=0)
+
adresses = self.vapi.nat44_address_dump()
for addr in adresses:
self.vapi.nat44_add_del_address_range(addr.ip_address,
self.pg_start()
self.pg3.assert_nothing_captured()
+ 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
+
+ # multiple clients
+ server1_n = 0
+ server2_n = 0
+ clients = ip4_range(self.pg1.remote_ip4, 10, 20)
+ 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_multiple_inside_interfaces(self):
""" NAT44 multiple non-overlapping address space inside interfaces """
self.pg0.unconfig_ip4()
self.pg1.unconfig_ip4()
+ self.vapi.ip_table_add_del(vrf_id1, is_add=1)
+ self.vapi.ip_table_add_del(vrf_id2, is_add=1)
self.pg0.set_table_ip4(vrf_id1)
self.pg1.set_table_ip4(vrf_id2)
self.pg0.config_ip4()
capture = self.pg2.get_capture(len(pkts))
self.verify_capture_out(capture, nat_ip2)
+ self.pg0.unconfig_ip4()
+ self.pg1.unconfig_ip4()
+ self.pg0.set_table_ip4(0)
+ self.pg1.set_table_ip4(0)
+ self.vapi.ip_table_add_del(vrf_id1, is_add=0)
+ self.vapi.ip_table_add_del(vrf_id2, is_add=0)
+
def test_vrf_feature_independent(self):
""" NAT44 tenant VRF independent address pool mode """
""" 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.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.pg1)
+ 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.pg1.get_capture(len(pkts))
+ capture = self.pg3.get_capture(len(pkts))
self.verify_capture_out(capture)
# out2in
- pkts = self.create_stream_out(self.pg1)
- self.pg1.add_stream(pkts)
+ 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)
+ # from non-NAT interface to NAT inside interface
+ pkts = self.create_stream_in(self.pg2, self.pg0)
+ self.pg2.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ self.verify_capture_no_translation(capture, self.pg2, self.pg0)
+
def test_output_feature_vrf_aware(self):
""" NAT44 interface output feature VRF aware (in2out postrouting) """
nat_ip_vrf10 = "10.0.0.10"
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 tearDown(self):
super(TestNAT44, self).tearDown()
if not self.vpp_dead:
cls.vrf1_nat_addr_n = socket.inet_pton(socket.AF_INET,
cls.vrf1_nat_addr)
- cls.create_pg_interfaces(range(3))
+ cls.create_pg_interfaces(range(4))
cls.ip6_interfaces = list(cls.pg_interfaces[0:1])
cls.ip6_interfaces.append(cls.pg_interfaces[2])
cls.ip4_interfaces = list(cls.pg_interfaces[1:2])
+ cls.vapi.ip_table_add_del(cls.vrf1_id, is_add=1, is_ipv6=1)
+
cls.pg_interfaces[2].set_table_ip6(cls.vrf1_id)
cls.pg0.generate_remote_hosts(2)
i.config_ip4()
i.resolve_arp()
+ cls.pg3.admin_up()
+ cls.pg3.config_ip4()
+ cls.pg3.resolve_arp()
+ cls.pg3.config_ip6()
+ cls.pg3.configure_ipv6_neighbors()
+
except Exception:
super(TestNAT64, cls).tearDownClass()
raise
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
+ def test_one_armed_nat64(self):
+ """ One armed NAT64 """
+ external_port = 0
+ remote_host_ip6 = self.compose_ip6(self.pg3.remote_ip4,
+ '64:ff9b::',
+ 96)
+
+ self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
+ self.nat_addr_n)
+ self.vapi.nat64_add_del_interface(self.pg3.sw_if_index)
+ self.vapi.nat64_add_del_interface(self.pg3.sw_if_index, is_inside=0)
+
+ # in2out
+ p = (Ether(src=self.pg3.remote_mac, dst=self.pg3.local_mac) /
+ IPv6(src=self.pg3.remote_ip6, dst=remote_host_ip6) /
+ TCP(sport=12345, dport=80))
+ self.pg3.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg3.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.pg3.remote_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.pg3.remote_mac, dst=self.pg3.local_mac) /
+ IP(src=self.pg3.remote_ip4, dst=self.nat_addr) /
+ TCP(sport=80, dport=external_port))
+ self.pg3.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg3.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IPv6]
+ tcp = p[TCP]
+ self.assertEqual(ip.src, remote_host_ip6)
+ self.assertEqual(ip.dst, self.pg3.remote_ip6)
+ self.assertEqual(tcp.sport, 80)
+ self.assertEqual(tcp.dport, 12345)
+ self.check_tcp_checksum(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
def nat64_get_ses_num(self):
"""
Return number of active NAT64 sessions.
interfaces = self.vapi.nat64_interface_dump()
for intf in interfaces:
+ if intf.is_inside > 1:
+ self.vapi.nat64_add_del_interface(intf.sw_if_index,
+ 0,
+ is_add=0)
self.vapi.nat64_add_del_interface(intf.sw_if_index,
intf.is_inside,
is_add=0)
Code Review / vpp.git / blobdiff