from ipaddress import IPv6Network
from util import ppc, ppp
from socket import inet_pton, AF_INET
+from vpp_acl import AclRule, VppAcl, VppAclInterface
# NAT HA protocol event data
self.port_in = random.randint(1025, 65535)
# in2out
- pkts = self.create_stream_frag(self.pg0,
- self.pg1.remote_ip4,
- self.port_in,
- 20,
- data,
- proto)
+ pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4,
+ self.port_in, 20, data, proto)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
else:
sport = p[layer].id
dport = 0
- pkts = self.create_stream_frag(self.pg1,
- dst_addr,
- sport,
- dport,
- data,
- proto,
- echo_reply=True)
+ pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport, data,
+ proto, echo_reply=True)
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
for i in range(2):
# out2in
- pkts = self.create_stream_frag(self.pg0,
- self.server_out_addr,
- self.port_in,
- self.server_out_port,
- data,
- proto)
+ pkts = self.create_stream_frag(self.pg0, self.server_out_addr,
+ self.port_in, self.server_out_port,
+ data, proto)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
# in2out
if proto != IP_PROTOS.icmp:
- pkts = self.create_stream_frag(self.pg1,
- self.pg0.remote_ip4,
+ pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4,
self.server_in_port,
- p[layer].sport,
- data,
- proto)
+ p[layer].sport, data, proto)
else:
- pkts = self.create_stream_frag(self.pg1,
- self.pg0.remote_ip4,
- p[layer].id,
- 0,
- data,
- proto,
+ pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4,
+ p[layer].id, 0, data, proto,
echo_reply=True)
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
for i in range(2):
# in2out
- pkts = self.create_stream_frag(self.pg0,
- self.pg1.remote_ip4,
- self.port_in,
- 20,
- data,
- proto)
+ pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4,
+ self.port_in, 20, data, proto)
pkts.reverse()
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
else:
sport = p[layer].id
dport = 0
- pkts = self.create_stream_frag(self.pg1,
- dst_addr,
- sport,
- dport,
- data,
- proto,
- echo_reply=True)
+ pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport,
+ data, proto, echo_reply=True)
pkts.reverse()
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
for i in range(2):
# out2in
- pkts = self.create_stream_frag(self.pg0,
- self.server_out_addr,
- self.port_in,
- self.server_out_port,
- data,
- proto)
+ pkts = self.create_stream_frag(self.pg0, self.server_out_addr,
+ self.port_in, self.server_out_port,
+ data, proto)
pkts.reverse()
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
# in2out
if proto != IP_PROTOS.icmp:
- pkts = self.create_stream_frag(self.pg1,
- self.pg0.remote_ip4,
+ pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4,
self.server_in_port,
- p[layer].sport,
- data,
- proto)
+ p[layer].sport, data, proto)
else:
- pkts = self.create_stream_frag(self.pg1,
- self.pg0.remote_ip4,
- p[layer].id,
- 0,
- data,
- proto,
+ pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4,
+ p[layer].id, 0, data, proto,
echo_reply=True)
pkts.reverse()
self.pg1.add_stream(pkts)
super(TestNAT44, 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.ipfix_src_port = 4739
- cls.ipfix_domain_id = 1
- cls.tcp_external_port = 80
- cls.udp_external_port = 69
-
- cls.create_pg_interfaces(range(10))
- cls.interfaces = list(cls.pg_interfaces[0:4])
-
- 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.pg1.generate_remote_hosts(1)
- cls.pg1.configure_ipv4_neighbors()
-
- cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
- cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
- cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
-
- cls.pg4._local_ip4 = "172.16.255.1"
- cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
- cls.pg4.set_table_ip4(10)
- cls.pg5._local_ip4 = "172.17.255.3"
- cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
- cls.pg5.set_table_ip4(10)
- cls.pg6._local_ip4 = "172.16.255.1"
- cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
- cls.pg6.set_table_ip4(20)
- for i in cls.overlapping_interfaces:
- i.config_ip4()
- i.admin_up()
- i.resolve_arp()
-
- cls.pg7.admin_up()
- cls.pg8.admin_up()
-
- cls.pg9.generate_remote_hosts(2)
- cls.pg9.config_ip4()
- cls.vapi.sw_interface_add_del_address(
- sw_if_index=cls.pg9.sw_if_index,
- prefix="10.0.0.1/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
+ 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.ipfix_src_port = 4739
+ cls.ipfix_domain_id = 1
+ cls.tcp_external_port = 80
+ cls.udp_external_port = 69
+
+ cls.create_pg_interfaces(range(10))
+ cls.interfaces = list(cls.pg_interfaces[0:4])
+
+ 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.pg1.generate_remote_hosts(1)
+ cls.pg1.configure_ipv4_neighbors()
+
+ cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
+ cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
+ cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
+
+ cls.pg4._local_ip4 = "172.16.255.1"
+ cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
+ cls.pg4.set_table_ip4(10)
+ cls.pg5._local_ip4 = "172.17.255.3"
+ cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
+ cls.pg5.set_table_ip4(10)
+ cls.pg6._local_ip4 = "172.16.255.1"
+ cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
+ cls.pg6.set_table_ip4(20)
+ for i in cls.overlapping_interfaces:
+ i.config_ip4()
+ i.admin_up()
+ i.resolve_arp()
+
+ cls.pg7.admin_up()
+ cls.pg8.admin_up()
+
+ cls.pg9.generate_remote_hosts(2)
+ cls.pg9.config_ip4()
+ cls.vapi.sw_interface_add_del_address(
+ sw_if_index=cls.pg9.sw_if_index,
+ prefix="10.0.0.1/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()
@classmethod
def tearDownClass(cls):
self.logger.info(self.vapi.cli("show nat ha"))
-class TestNAT44EndpointDependent2(MethodHolder):
- """ Endpoint-Dependent session test cases """
-
- icmp_timeout = 2
-
- @classmethod
- def setUpConstants(cls):
- super(TestNAT44EndpointDependent2, cls).setUpConstants()
- cls.vpp_cmdline.extend(["nat", "{", "endpoint-dependent",
- "translation", "hash", "buckets", "1",
- "icmp", "timeout", str(cls.icmp_timeout), "}"])
-
- @classmethod
- def setUpClass(cls):
- super(TestNAT44EndpointDependent2, cls).setUpClass()
- try:
- translation_buckets = 1
- cls.max_translations = 10 * translation_buckets
-
- cls.create_pg_interfaces(range(2))
- cls.interfaces = list(cls.pg_interfaces[0:2])
-
- for i in cls.interfaces:
- i.admin_up()
- i.config_ip4()
- i.resolve_arp()
-
- cls.pg0.generate_remote_hosts(1)
- cls.pg0.configure_ipv4_neighbors()
-
- cls.pg1.generate_remote_hosts(1)
- cls.pg1.configure_ipv4_neighbors()
-
- except Exception:
- super(TestNAT44EndpointDependent2, cls).tearDownClass()
- raise
-
- def create_icmp_stream(self, in_if, out_if, count):
- """
- Create ICMP packet stream for inside network
-
- :param in_if: Inside interface
- :param out_if: Outside interface
- :param count: Number of packets
- """
-
- self.assertTrue(count > 0)
- icmp_id = random.randint(0, 65535 - (count - 1))
-
- pkts = list()
- for i in range(count):
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=64) /
- ICMP(id=icmp_id + i, type='echo-request'))
- pkts.append(p)
- return pkts
-
- def send_pkts(self, pkts, expected=None):
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- return self.pg1.get_capture(
- len(pkts) if expected is None else expected)
-
- def test_session_cleanup(self):
- """ NAT44 session cleanup test """
-
- self.nat44_add_address(self.pg1.local_ip4)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
- sw_if_index=self.pg0.sw_if_index,
- flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
-
- nat_config = self.vapi.nat_show_config()
- self.assertEqual(1, nat_config.endpoint_dependent)
-
- pkts = self.create_icmp_stream(self.pg0, self.pg1,
- self.max_translations + 2)
- sz = len(pkts)
-
- # positive test
- self.send_pkts(pkts[0:self.max_translations])
-
- # false positive test
- self.send_pkts(pkts[self.max_translations:sz - 1], 0)
-
- sleep(self.icmp_timeout)
-
- # positive test
- self.send_pkts(pkts[self.max_translations + 1:sz])
-
-
class TestNAT44EndpointDependent(MethodHolder):
""" Endpoint-Dependent mapping and filtering test cases """
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.ipfix_src_port = 4739
- cls.ipfix_domain_id = 1
- cls.tcp_external_port = 80
-
- cls.create_pg_interfaces(range(9))
- 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()
- cls.vapi.sw_interface_add_del_address(
- sw_if_index=cls.pg4.sw_if_index,
- prefix="10.0.0.1/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()
-
- zero_ip4 = socket.inet_pton(socket.AF_INET, "0.0.0.0")
- cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 1})
-
- cls.pg5._local_ip4 = "10.1.1.1"
- cls.pg5._remote_hosts[0]._ip4 = "10.1.1.2"
- cls.pg5.set_table_ip4(1)
- cls.pg5.config_ip4()
- cls.pg5.admin_up()
- r1 = VppIpRoute(cls, cls.pg5.remote_ip4, 32,
- [VppRoutePath("0.0.0.0",
- cls.pg5.sw_if_index)],
- table_id=1,
- register=False)
- r1.add_vpp_config()
-
- cls.pg6._local_ip4 = "10.1.2.1"
- cls.pg6._remote_hosts[0]._ip4 = "10.1.2.2"
- cls.pg6.set_table_ip4(1)
- cls.pg6.config_ip4()
- cls.pg6.admin_up()
-
- r2 = VppIpRoute(cls, cls.pg6.remote_ip4, 32,
- [VppRoutePath("0.0.0.0",
- cls.pg6.sw_if_index)],
- table_id=1,
- register=False)
- r3 = VppIpRoute(cls, cls.pg6.remote_ip4, 16,
- [VppRoutePath("0.0.0.0",
- 0xffffffff,
- nh_table_id=1)],
- table_id=0,
- register=False)
- r4 = VppIpRoute(cls, "0.0.0.0", 0,
- [VppRoutePath("0.0.0.0", 0xffffffff,
- nh_table_id=0)],
- table_id=1,
- register=False)
- r5 = VppIpRoute(cls, "0.0.0.0", 0,
- [VppRoutePath(cls.pg1.local_ip4,
- cls.pg1.sw_if_index)],
- register=False)
- r2.add_vpp_config()
- r3.add_vpp_config()
- r4.add_vpp_config()
- r5.add_vpp_config()
-
- cls.pg5.resolve_arp()
- cls.pg6.resolve_arp()
-
- cls.pg7.admin_up()
- cls.pg7.config_ip4()
- cls.pg7.resolve_arp()
- cls.pg7.generate_remote_hosts(3)
- cls.pg7.configure_ipv4_neighbors()
-
- cls.pg8.admin_up()
- cls.pg8.config_ip4()
- cls.pg8.resolve_arp()
-
- except Exception:
- super(TestNAT44EndpointDependent, cls).tearDownClass()
- raise
+
+ 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.ipfix_src_port = 4739
+ cls.ipfix_domain_id = 1
+ cls.tcp_external_port = 80
+
+ cls.create_pg_interfaces(range(9))
+ 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()
+ cls.vapi.sw_interface_add_del_address(
+ sw_if_index=cls.pg4.sw_if_index,
+ prefix="10.0.0.1/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()
+
+ zero_ip4 = socket.inet_pton(socket.AF_INET, "0.0.0.0")
+ cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 1})
+
+ cls.pg5._local_ip4 = "10.1.1.1"
+ cls.pg5._remote_hosts[0]._ip4 = "10.1.1.2"
+ cls.pg5.set_table_ip4(1)
+ cls.pg5.config_ip4()
+ cls.pg5.admin_up()
+ r1 = VppIpRoute(cls, cls.pg5.remote_ip4, 32,
+ [VppRoutePath("0.0.0.0",
+ cls.pg5.sw_if_index)],
+ table_id=1,
+ register=False)
+ r1.add_vpp_config()
+
+ cls.pg6._local_ip4 = "10.1.2.1"
+ cls.pg6._remote_hosts[0]._ip4 = "10.1.2.2"
+ cls.pg6.set_table_ip4(1)
+ cls.pg6.config_ip4()
+ cls.pg6.admin_up()
+
+ r2 = VppIpRoute(cls, cls.pg6.remote_ip4, 32,
+ [VppRoutePath("0.0.0.0",
+ cls.pg6.sw_if_index)],
+ table_id=1,
+ register=False)
+ r3 = VppIpRoute(cls, cls.pg6.remote_ip4, 16,
+ [VppRoutePath("0.0.0.0",
+ 0xffffffff,
+ nh_table_id=1)],
+ table_id=0,
+ register=False)
+ r4 = VppIpRoute(cls, "0.0.0.0", 0,
+ [VppRoutePath("0.0.0.0", 0xffffffff,
+ nh_table_id=0)],
+ table_id=1,
+ register=False)
+ r5 = VppIpRoute(cls, "0.0.0.0", 0,
+ [VppRoutePath(cls.pg1.local_ip4,
+ cls.pg1.sw_if_index)],
+ register=False)
+ r2.add_vpp_config()
+ r3.add_vpp_config()
+ r4.add_vpp_config()
+ r5.add_vpp_config()
+
+ cls.pg5.resolve_arp()
+ cls.pg6.resolve_arp()
+
+ cls.pg7.admin_up()
+ cls.pg7.config_ip4()
+ cls.pg7.resolve_arp()
+ cls.pg7.generate_remote_hosts(3)
+ cls.pg7.configure_ipv4_neighbors()
+
+ cls.pg8.admin_up()
+ cls.pg8.config_ip4()
+ cls.pg8.resolve_arp()
+
+ def setUp(self):
+ super(TestNAT44EndpointDependent, self).setUp()
+ self.vapi.nat_set_timeouts(
+ udp=300, tcp_established=7440, tcp_transitory=240, icmp=60)
@classmethod
def tearDownClass(cls):
sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
start_sessnum = len(sessions)
+ self.vapi.nat_set_timeouts(udp=300, tcp_established=7440,
+ tcp_transitory=2, icmp=5)
+
self.initiate_tcp_session(self.pg0, self.pg1)
# FIN packet in -> out
self.pg_start()
self.pg1.get_capture(1)
- sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4,
- 0)
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
+ self.assertEqual(len(sessions) - start_sessnum, 1)
+
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
+ out2in_drops = stats[0]
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
+ in2out_drops = stats[0]
+
+ # extra FIN packet out -> in - this should be dropped
+ 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))
+
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg0.assert_nothing_captured()
+
+ # extra ACK packet in -> out - this should be dropped
+ 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.assert_nothing_captured()
+
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
+ self.assertEqual(stats[0] - out2in_drops, 1)
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
+ self.assertEqual(stats[0] - in2out_drops, 1)
+
+ self.sleep(3)
+ # extra ACK packet in -> out - this will cause session to be wiped
+ 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.assert_nothing_captured()
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
self.assertEqual(len(sessions) - start_sessnum, 0)
def test_tcp_session_close_out(self):
sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
start_sessnum = len(sessions)
+ self.vapi.nat_set_timeouts(udp=300, tcp_established=7440,
+ tcp_transitory=2, icmp=5)
+
self.initiate_tcp_session(self.pg0, self.pg1)
# FIN packet out -> in
self.pg_start()
self.pg0.get_capture(1)
- sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4,
- 0)
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
+ self.assertEqual(len(sessions) - start_sessnum, 1)
+
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
+ out2in_drops = stats[0]
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
+ in2out_drops = stats[0]
+
+ # extra FIN packet out -> in - this should be dropped
+ 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))
+
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg0.assert_nothing_captured()
+
+ # extra ACK packet in -> out - this should be dropped
+ 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.assert_nothing_captured()
+
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
+ self.assertEqual(stats[0] - out2in_drops, 1)
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
+ self.assertEqual(stats[0] - in2out_drops, 1)
+
+ self.sleep(3)
+ # extra ACK packet in -> out - this will cause session to be wiped
+ 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.assert_nothing_captured()
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
self.assertEqual(len(sessions) - start_sessnum, 0)
def test_tcp_session_close_simultaneous(self):
sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
start_sessnum = len(sessions)
+ self.vapi.nat_set_timeouts(udp=300, tcp_established=7440,
+ tcp_transitory=2, icmp=5)
+
self.initiate_tcp_session(self.pg0, self.pg1)
# FIN packet in -> out
self.pg_start()
self.pg0.get_capture(1)
- sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4,
- 0)
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
+ self.assertEqual(len(sessions) - start_sessnum, 1)
+
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
+ out2in_drops = stats[0]
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
+ in2out_drops = stats[0]
+
+ # extra FIN packet out -> in - this should be dropped
+ 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))
+
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg0.assert_nothing_captured()
+
+ # extra ACK packet in -> out - this should be dropped
+ 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.assert_nothing_captured()
+
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
+ self.assertEqual(stats[0] - out2in_drops, 1)
+ stats = self.statistics.get_counter(
+ '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
+ self.assertEqual(stats[0] - in2out_drops, 1)
+
+ self.sleep(3)
+ # extra ACK packet in -> out - this will cause session to be wiped
+ 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.assert_nothing_captured()
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
self.assertEqual(len(sessions) - start_sessnum, 0)
def test_one_armed_nat44_static(self):
self.verify_capture_in(capture, self.pg0)
# Create an ACL blocking everything
- out2in_deny_rule = {
- 'is_permit': 0,
- 'is_ipv6': 0,
- 'src_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
- 'src_ip_prefix_len': 0,
- 'dst_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
- 'dst_ip_prefix_len': 0,
- 'srcport_or_icmptype_first': 0,
- 'srcport_or_icmptype_last': 65535,
- 'dstport_or_icmpcode_first': 0,
- 'dstport_or_icmpcode_last': 65535,
- 'proto': 0,
- }
- out2in_rules = [out2in_deny_rule]
- res = self.vapi.acl_add_replace(0xffffffff, out2in_rules)
- self.assertEqual(res.retval, 0, "error adding out2in ACL")
- out2in_acl = res.acl_index
+ out2in_deny_rule = AclRule(is_permit=0)
+ out2in_acl = VppAcl(self, rules=[out2in_deny_rule])
+ out2in_acl.add_vpp_config()
+
+ # create an ACL to permit/reflect everything
+ in2out_reflect_rule = AclRule(is_permit=2)
+ in2out_acl = VppAcl(self, rules=[in2out_reflect_rule])
+ in2out_acl.add_vpp_config()
# apply as input acl on interface and confirm it blocks everything
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- n_input=1,
- acls=[out2in_acl])
+ acl_if = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
+ n_input=1, acls=[out2in_acl])
+ acl_if.add_vpp_config()
self.send_and_assert_no_replies(self.pg1, pkts_out2in)
- # create an ACL to permit/reflect everything
- in2out_reflect_rule = {
- 'is_permit': 2,
- 'is_ipv6': 0,
- 'src_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
- 'src_ip_prefix_len': 0,
- 'dst_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
- 'dst_ip_prefix_len': 0,
- 'srcport_or_icmptype_first': 0,
- 'srcport_or_icmptype_last': 65535,
- 'dstport_or_icmpcode_first': 0,
- 'dstport_or_icmpcode_last': 65535,
- 'proto': 0,
- }
- in2out_rules = [in2out_reflect_rule]
- res = self.vapi.acl_add_replace(0xffffffff, in2out_rules)
- self.assertEqual(res.retval, 0, "error adding in2out ACL")
- in2out_acl = res.acl_index
-
# apply output acl
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- n_input=1,
- acls=[out2in_acl, in2out_acl])
+ acl_if.acls = [out2in_acl, in2out_acl]
+ acl_if.add_vpp_config()
# send in2out to generate ACL state (NAT state was created earlier)
capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1,
len(pkts_in2out))
self.verify_capture_in(capture, self.pg0)
self.logger.info(self.vapi.cli("show trace"))
- # Clean up
- # Remove ACLs from interface
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- n_input=0,
- acls=[])
- # delete ACLs
- self.vapi.acl_del(acl_index=out2in_acl, expected_retval=0)
- self.vapi.acl_del(acl_index=in2out_acl, expected_retval=0)
-
def test_multiple_vrf(self):
""" Multiple VRF setup """
external_addr = '1.2.3.4'
super(TestNAT44Out2InDPO, 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.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()
-
- r1 = VppIpRoute(cls, "::", 0,
- [VppRoutePath(cls.pg1.remote_ip6,
- cls.pg1.sw_if_index)],
- register=False)
- r1.add_vpp_config()
-
- except Exception:
- super(TestNAT44Out2InDPO, cls).tearDownClass()
- raise
+ 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.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()
+
+ r1 = VppIpRoute(cls, "::", 0,
+ [VppRoutePath(cls.pg1.remote_ip6,
+ cls.pg1.sw_if_index)],
+ register=False)
+ r1.add_vpp_config()
@classmethod
def tearDownClass(cls):
super(TestDeterministicNAT, cls).setUpClass()
cls.vapi.cli("set log class nat level debug")
- try:
- cls.tcp_port_in = 6303
- cls.tcp_external_port = 6303
- cls.udp_port_in = 6304
- cls.udp_external_port = 6304
- cls.icmp_id_in = 6305
- cls.nat_addr = '10.0.0.3'
-
- cls.create_pg_interfaces(range(3))
- cls.interfaces = list(cls.pg_interfaces)
-
- for i in cls.interfaces:
- i.admin_up()
- i.config_ip4()
- i.resolve_arp()
-
- cls.pg0.generate_remote_hosts(2)
- cls.pg0.configure_ipv4_neighbors()
-
- except Exception:
- super(TestDeterministicNAT, cls).tearDownClass()
- raise
+ cls.tcp_port_in = 6303
+ cls.tcp_external_port = 6303
+ cls.udp_port_in = 6304
+ cls.udp_external_port = 6304
+ cls.icmp_id_in = 6305
+ cls.nat_addr = '10.0.0.3'
+
+ cls.create_pg_interfaces(range(3))
+ cls.interfaces = list(cls.pg_interfaces)
+
+ for i in cls.interfaces:
+ i.admin_up()
+ i.config_ip4()
+ i.resolve_arp()
+
+ cls.pg0.generate_remote_hosts(2)
+ cls.pg0.configure_ipv4_neighbors()
@classmethod
def tearDownClass(cls):
def setUpClass(cls):
super(TestNAT64, 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.tcp_external_port = 80
- cls.nat_addr = '10.0.0.3'
- cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
- cls.vrf1_id = 10
- cls.vrf1_nat_addr = '10.0.10.3'
- cls.ipfix_src_port = 4739
- cls.ipfix_domain_id = 1
-
- cls.create_pg_interfaces(range(6))
- 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(is_add=1,
- table={'table_id': cls.vrf1_id,
- 'is_ip6': 1})
-
- cls.pg_interfaces[2].set_table_ip6(cls.vrf1_id)
-
- cls.pg0.generate_remote_hosts(2)
-
- for i in cls.ip6_interfaces:
- i.admin_up()
- i.config_ip6()
- i.configure_ipv6_neighbors()
-
- for i in cls.ip4_interfaces:
- i.admin_up()
- 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()
-
- cls.pg5.admin_up()
- cls.pg5.config_ip6()
-
- except Exception:
- super(TestNAT64, cls).tearDownClass()
- raise
+ 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.tcp_external_port = 80
+ cls.nat_addr = '10.0.0.3'
+ cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
+ cls.vrf1_id = 10
+ cls.vrf1_nat_addr = '10.0.10.3'
+ cls.ipfix_src_port = 4739
+ cls.ipfix_domain_id = 1
+
+ cls.create_pg_interfaces(range(6))
+ 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(is_add=1,
+ table={'table_id': cls.vrf1_id,
+ 'is_ip6': 1})
+
+ cls.pg_interfaces[2].set_table_ip6(cls.vrf1_id)
+
+ cls.pg0.generate_remote_hosts(2)
+
+ for i in cls.ip6_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.configure_ipv6_neighbors()
+
+ for i in cls.ip4_interfaces:
+ i.admin_up()
+ 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()
+
+ cls.pg5.admin_up()
+ cls.pg5.config_ip6()
@classmethod
def tearDownClass(cls):
def setUpClass(cls):
super(TestNAT66, cls).setUpClass()
- try:
- cls.nat_addr = 'fd01:ff::2'
-
- cls.create_pg_interfaces(range(2))
- cls.interfaces = list(cls.pg_interfaces)
+ cls.nat_addr = 'fd01:ff::2'
- for i in cls.interfaces:
- i.admin_up()
- i.config_ip6()
- i.configure_ipv6_neighbors()
+ cls.create_pg_interfaces(range(2))
+ cls.interfaces = list(cls.pg_interfaces)
- except Exception:
- super(TestNAT66, cls).tearDownClass()
- raise
+ for i in cls.interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.configure_ipv6_neighbors()
@classmethod
def tearDownClass(cls):
Code Review / vpp.git / blobdiff