diff options
Diffstat (limited to 'test/test_nat44_ed.py')
| -rw-r--r-- | test/test_nat44_ed.py | 3594 |
1 files changed, 2294 insertions, 1300 deletions
diff --git a/test/test_nat44_ed.py b/test/test_nat44_ed.py index ec8d7c8d159..eed89f1a399 100644 --- a/test/test_nat44_ed.py +++ b/test/test_nat44_ed.py @@ -2,26 +2,31 @@ import unittest from io import BytesIO -from random import randint, shuffle, choice +from random import randint, choice +import re import scapy.compat -from framework import VppTestCase, VppTestRunner +from framework import VppTestCase, VppLoInterface +from asfframework import VppTestRunner, tag_fixme_ubuntu2204, is_distro_ubuntu2204 from scapy.data import IP_PROTOS from scapy.layers.inet import IP, TCP, UDP, ICMP, GRE from scapy.layers.inet import IPerror, TCPerror from scapy.layers.l2 import Ether from scapy.packet import Raw +from statistics import variance from syslog_rfc5424_parser import SyslogMessage, ParseError from syslog_rfc5424_parser.constants import SyslogSeverity -from util import ppp, ip4_range +from util import ppp, pr, ip4_range from vpp_acl import AclRule, VppAcl, VppAclInterface from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_papi import VppEnum +from util import StatsDiff -class NAT44EDTestCase(VppTestCase): +class TestNAT44ED(VppTestCase): + """NAT44ED Test Case""" - nat_addr = '10.0.0.3' + nat_addr = "10.0.10.3" tcp_port_in = 6303 tcp_port_out = 6303 @@ -37,17 +42,17 @@ class NAT44EDTestCase(VppTestCase): max_sessions = 100 def setUp(self): - super(NAT44EDTestCase, self).setUp() + super().setUp() self.plugin_enable() def tearDown(self): - super(NAT44EDTestCase, self).tearDown() + super().tearDown() if not self.vpp_dead: self.plugin_disable() - def plugin_enable(self): - self.vapi.nat44_ed_plugin_enable_disable( - sessions=self.max_sessions, enable=1) + def plugin_enable(self, max_sessions=None): + max_sessions = max_sessions or self.max_sessions + self.vapi.nat44_ed_plugin_enable_disable(sessions=max_sessions, enable=1) def plugin_disable(self): self.vapi.nat44_ed_plugin_enable_disable(enable=0) @@ -70,7 +75,7 @@ class NAT44EDTestCase(VppTestCase): @staticmethod def random_port(): - return randint(1025, 65535) + return randint(1024, 65535) @staticmethod def proto2layer(proto): @@ -85,7 +90,7 @@ class NAT44EDTestCase(VppTestCase): @classmethod def create_and_add_ip4_table(cls, i, table_id=0): - cls.vapi.ip_table_add_del(is_add=1, table={'table_id': table_id}) + cls.vapi.ip_table_add_del(is_add=1, table={"table_id": table_id}) i.set_table_ip4(table_id) @classmethod @@ -103,33 +108,41 @@ class NAT44EDTestCase(VppTestCase): @classmethod def nat_add_interface_address(cls, i): - cls.vapi.nat44_add_del_interface_addr( - sw_if_index=i.sw_if_index, is_add=1) + cls.vapi.nat44_add_del_interface_addr(sw_if_index=i.sw_if_index, is_add=1) def nat_add_inside_interface(self, i): self.vapi.nat44_interface_add_del_feature( - flags=self.config_flags.NAT_IS_INSIDE, - sw_if_index=i.sw_if_index, is_add=1) + flags=self.config_flags.NAT_IS_INSIDE, sw_if_index=i.sw_if_index, is_add=1 + ) def nat_add_outside_interface(self, i): self.vapi.nat44_interface_add_del_feature( - flags=self.config_flags.NAT_IS_OUTSIDE, - sw_if_index=i.sw_if_index, is_add=1) + flags=self.config_flags.NAT_IS_OUTSIDE, sw_if_index=i.sw_if_index, is_add=1 + ) - def nat_add_address(self, address, twice_nat=0, - vrf_id=0xFFFFFFFF, is_add=1): + def nat_add_address(self, address, twice_nat=0, vrf_id=0xFFFFFFFF, is_add=1): flags = self.config_flags.NAT_IS_TWICE_NAT if twice_nat else 0 - self.vapi.nat44_add_del_address_range(first_ip_address=address, - last_ip_address=address, - vrf_id=vrf_id, - is_add=is_add, - flags=flags) - - def nat_add_static_mapping(self, local_ip, external_ip='0.0.0.0', - local_port=0, external_port=0, vrf_id=0, - is_add=1, external_sw_if_index=0xFFFFFFFF, - proto=0, tag="", flags=0): + self.vapi.nat44_add_del_address_range( + first_ip_address=address, + last_ip_address=address, + vrf_id=vrf_id, + is_add=is_add, + flags=flags, + ) + def nat_add_static_mapping( + self, + local_ip, + external_ip="0.0.0.0", + local_port=0, + external_port=0, + vrf_id=0, + is_add=1, + external_sw_if_index=0xFFFFFFFF, + proto=0, + tag="", + flags=0, + ): if not (local_port and external_port): flags |= self.config_flags.NAT_IS_ADDR_ONLY @@ -140,13 +153,17 @@ class NAT44EDTestCase(VppTestCase): external_sw_if_index=external_sw_if_index, local_port=local_port, external_port=external_port, - vrf_id=vrf_id, protocol=proto, + vrf_id=vrf_id, + protocol=proto, flags=flags, - tag=tag) + tag=tag, + ) @classmethod def setUpClass(cls): - super(NAT44EDTestCase, cls).setUpClass() + super().setUpClass() + if is_distro_ubuntu2204 == True and not hasattr(cls, "vpp"): + return cls.create_pg_interfaces(range(12)) cls.interfaces = list(cls.pg_interfaces[:4]) @@ -157,14 +174,14 @@ class NAT44EDTestCase(VppTestCase): cls.configure_ip4_interface(i, hosts=3) # test specific (test-multiple-vrf) - cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 1}) + cls.vapi.ip_table_add_del(is_add=1, table={"table_id": 1}) # test specific (test-one-armed-nat44-static) 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") + 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 @@ -188,36 +205,95 @@ class NAT44EDTestCase(VppTestCase): rl = list() - rl.append(VppIpRoute(cls, "0.0.0.0", 0, - [VppRoutePath("0.0.0.0", 0xffffffff, - nh_table_id=0)], - register=False, table_id=1)) - rl.append(VppIpRoute(cls, "0.0.0.0", 0, - [VppRoutePath(cls.pg1.local_ip4, - cls.pg1.sw_if_index)], - register=False)) - rl.append(VppIpRoute(cls, cls.pg5.remote_ip4, 32, - [VppRoutePath("0.0.0.0", - cls.pg5.sw_if_index)], - register=False, table_id=1)) - rl.append(VppIpRoute(cls, cls.pg6.remote_ip4, 32, - [VppRoutePath("0.0.0.0", - cls.pg6.sw_if_index)], - register=False, table_id=1)) - rl.append(VppIpRoute(cls, cls.pg6.remote_ip4, 16, - [VppRoutePath("0.0.0.0", 0xffffffff, - nh_table_id=1)], - register=False, table_id=0)) + rl.append( + VppIpRoute( + cls, + "0.0.0.0", + 0, + [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=0)], + register=False, + table_id=1, + ) + ) + rl.append( + VppIpRoute( + cls, + "0.0.0.0", + 0, + [VppRoutePath(cls.pg1.local_ip4, cls.pg1.sw_if_index)], + register=False, + ) + ) + rl.append( + VppIpRoute( + cls, + cls.pg5.remote_ip4, + 32, + [VppRoutePath("0.0.0.0", cls.pg5.sw_if_index)], + register=False, + table_id=1, + ) + ) + rl.append( + VppIpRoute( + cls, + cls.pg6.remote_ip4, + 32, + [VppRoutePath("0.0.0.0", cls.pg6.sw_if_index)], + register=False, + table_id=1, + ) + ) + rl.append( + VppIpRoute( + cls, + cls.pg6.remote_ip4, + 16, + [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=1)], + register=False, + table_id=0, + ) + ) for r in rl: r.add_vpp_config() + cls.no_diff = StatsDiff( + { + pg.sw_if_index: { + "/nat44-ed/in2out/fastpath/tcp": 0, + "/nat44-ed/in2out/fastpath/udp": 0, + "/nat44-ed/in2out/fastpath/icmp": 0, + "/nat44-ed/in2out/fastpath/drops": 0, + "/nat44-ed/in2out/slowpath/tcp": 0, + "/nat44-ed/in2out/slowpath/udp": 0, + "/nat44-ed/in2out/slowpath/icmp": 0, + "/nat44-ed/in2out/slowpath/drops": 0, + "/nat44-ed/in2out/fastpath/tcp": 0, + "/nat44-ed/in2out/fastpath/udp": 0, + "/nat44-ed/in2out/fastpath/icmp": 0, + "/nat44-ed/in2out/fastpath/drops": 0, + "/nat44-ed/in2out/slowpath/tcp": 0, + "/nat44-ed/in2out/slowpath/udp": 0, + "/nat44-ed/in2out/slowpath/icmp": 0, + "/nat44-ed/in2out/slowpath/drops": 0, + } + for pg in cls.pg_interfaces + } + ) + def get_err_counter(self, path): return self.statistics.get_err_counter(path) - def reass_hairpinning(self, server_addr, server_in_port, server_out_port, - host_in_port, proto=IP_PROTOS.tcp, - ignore_port=False): + def reass_hairpinning( + self, + server_addr, + server_in_port, + server_out_port, + host_in_port, + proto=IP_PROTOS.tcp, + ignore_port=False, + ): layer = self.proto2layer(proto) if proto == IP_PROTOS.tcp: @@ -226,19 +302,14 @@ class NAT44EDTestCase(VppTestCase): data = b"A" * 16 + b"B" * 16 + b"C" * 3 # send packet from host to server - pkts = self.create_stream_frag(self.pg0, - self.nat_addr, - host_in_port, - server_out_port, - data, - proto) + pkts = self.create_stream_frag( + self.pg0, self.nat_addr, host_in_port, server_out_port, data, proto + ) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - self.nat_addr, - server_addr) + p = self.reass_frags_and_verify(frags, self.nat_addr, server_addr) if proto != IP_PROTOS.icmp: if not ignore_port: self.assertNotEqual(p[layer].sport, host_in_port) @@ -248,8 +319,9 @@ class NAT44EDTestCase(VppTestCase): self.assertNotEqual(p[layer].id, host_in_port) self.assertEqual(data, p[Raw].load) - def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False, - ignore_port=False): + def frag_out_of_order( + self, proto=IP_PROTOS.tcp, dont_translate=False, ignore_port=False + ): layer = self.proto2layer(proto) if proto == IP_PROTOS.tcp: @@ -260,21 +332,22 @@ class NAT44EDTestCase(VppTestCase): 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) self.pg_start() frags = self.pg1.get_capture(len(pkts)) if not dont_translate: - p = self.reass_frags_and_verify(frags, - self.nat_addr, - self.pg1.remote_ip4) + p = self.reass_frags_and_verify( + frags, self.nat_addr, self.pg1.remote_ip4 + ) else: - p = self.reass_frags_and_verify(frags, - self.pg0.remote_ip4, - self.pg1.remote_ip4) + p = self.reass_frags_and_verify( + frags, self.pg0.remote_ip4, self.pg1.remote_ip4 + ) if proto != IP_PROTOS.icmp: if not dont_translate: self.assertEqual(p[layer].dport, 20) @@ -301,17 +374,18 @@ class NAT44EDTestCase(VppTestCase): 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) self.logger.info(self.vapi.cli("show trace")) self.pg_start() frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - self.pg1.remote_ip4, - self.pg0.remote_ip4) + p = self.reass_frags_and_verify( + frags, self.pg1.remote_ip4, self.pg0.remote_ip4 + ) if proto != IP_PROTOS.icmp: self.assertEqual(p[layer].sport, 20) self.assertEqual(p[layer].dport, self.port_in) @@ -327,21 +401,20 @@ class NAT44EDTestCase(VppTestCase): self.assert_ip_checksum_valid(p) buffer.seek(p[IP].frag * 8) buffer.write(bytes(p[IP].payload)) - ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst, - proto=frags[0][IP].proto) + ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst, proto=frags[0][IP].proto) if ip.proto == IP_PROTOS.tcp: - p = (ip / TCP(buffer.getvalue())) + p = ip / TCP(buffer.getvalue()) self.logger.debug(ppp("Reassembled:", p)) self.assert_tcp_checksum_valid(p) elif ip.proto == IP_PROTOS.udp: - p = (ip / UDP(buffer.getvalue()[:8]) / - Raw(buffer.getvalue()[8:])) + p = ip / UDP(buffer.getvalue()[:8]) / Raw(buffer.getvalue()[8:]) elif ip.proto == IP_PROTOS.icmp: - p = (ip / ICMP(buffer.getvalue())) + p = ip / ICMP(buffer.getvalue()) return p - def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False, - ignore_port=False): + def frag_in_order( + self, proto=IP_PROTOS.tcp, dont_translate=False, ignore_port=False + ): layer = self.proto2layer(proto) if proto == IP_PROTOS.tcp: @@ -351,20 +424,19 @@ class NAT44EDTestCase(VppTestCase): self.port_in = self.random_port() # 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() frags = self.pg1.get_capture(len(pkts)) if not dont_translate: - p = self.reass_frags_and_verify(frags, - self.nat_addr, - self.pg1.remote_ip4) + p = self.reass_frags_and_verify(frags, self.nat_addr, self.pg1.remote_ip4) else: - p = self.reass_frags_and_verify(frags, - self.pg0.remote_ip4, - self.pg1.remote_ip4) + p = self.reass_frags_and_verify( + frags, self.pg0.remote_ip4, self.pg1.remote_ip4 + ) if proto != IP_PROTOS.icmp: if not dont_translate: self.assertEqual(p[layer].dport, 20) @@ -391,15 +463,14 @@ class NAT44EDTestCase(VppTestCase): 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() frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - self.pg1.remote_ip4, - self.pg0.remote_ip4) + p = self.reass_frags_and_verify(frags, self.pg1.remote_ip4, self.pg0.remote_ip4) if proto != IP_PROTOS.icmp: self.assertEqual(p[layer].sport, 20) self.assertEqual(p[layer].dport, self.port_in) @@ -407,8 +478,9 @@ class NAT44EDTestCase(VppTestCase): self.assertEqual(p[layer].id, self.port_in) self.assertEqual(data, p[Raw].load) - def verify_capture_out(self, capture, nat_ip=None, same_port=False, - dst_ip=None, ignore_port=False): + def verify_capture_out( + self, capture, nat_ip=None, same_port=False, dst_ip=None, ignore_port=False + ): if nat_ip is None: nat_ip = self.nat_addr for packet in capture: @@ -420,35 +492,30 @@ class NAT44EDTestCase(VppTestCase): if packet.haslayer(TCP): if not ignore_port: if same_port: - self.assertEqual( - packet[TCP].sport, self.tcp_port_in) + self.assertEqual(packet[TCP].sport, self.tcp_port_in) else: - self.assertNotEqual( - packet[TCP].sport, self.tcp_port_in) + self.assertNotEqual(packet[TCP].sport, self.tcp_port_in) self.tcp_port_out = packet[TCP].sport self.assert_packet_checksums_valid(packet) elif packet.haslayer(UDP): if not ignore_port: if same_port: - self.assertEqual( - packet[UDP].sport, self.udp_port_in) + self.assertEqual(packet[UDP].sport, self.udp_port_in) else: - self.assertNotEqual( - packet[UDP].sport, self.udp_port_in) + self.assertNotEqual(packet[UDP].sport, self.udp_port_in) self.udp_port_out = packet[UDP].sport else: if not ignore_port: if same_port: - self.assertEqual( - packet[ICMP].id, self.icmp_id_in) + self.assertEqual(packet[ICMP].id, self.icmp_id_in) else: - self.assertNotEqual( - packet[ICMP].id, self.icmp_id_in) + self.assertNotEqual(packet[ICMP].id, self.icmp_id_in) self.icmp_id_out = packet[ICMP].id self.assert_packet_checksums_valid(packet) except: - self.logger.error(ppp("Unexpected or invalid packet " - "(outside network):", packet)) + self.logger.error( + ppp("Unexpected or invalid packet (outside network):", packet) + ) raise def verify_capture_in(self, capture, in_if): @@ -463,8 +530,9 @@ class NAT44EDTestCase(VppTestCase): else: self.assertEqual(packet[ICMP].id, self.icmp_id_in) except: - self.logger.error(ppp("Unexpected or invalid packet " - "(inside network):", packet)) + self.logger.error( + ppp("Unexpected or invalid packet (inside network):", packet) + ) raise def create_stream_in(self, in_if, out_if, dst_ip=None, ttl=64): @@ -473,27 +541,32 @@ class NAT44EDTestCase(VppTestCase): pkts = [] # TCP - p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) / - TCP(sport=self.tcp_port_in, dport=20)) + p = ( + Ether(dst=in_if.local_mac, src=in_if.remote_mac) + / IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) + / TCP(sport=self.tcp_port_in, dport=20) + ) pkts.extend([p, p]) # UDP - p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) / - UDP(sport=self.udp_port_in, dport=20)) + p = ( + Ether(dst=in_if.local_mac, src=in_if.remote_mac) + / IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) + / UDP(sport=self.udp_port_in, dport=20) + ) pkts.append(p) # ICMP - p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) / - ICMP(id=self.icmp_id_in, type='echo-request')) + p = ( + Ether(dst=in_if.local_mac, src=in_if.remote_mac) + / IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) + / ICMP(id=self.icmp_id_in, type="echo-request") + ) pkts.append(p) return pkts - def create_stream_out(self, out_if, dst_ip=None, ttl=64, - use_inside_ports=False): + def create_stream_out(self, out_if, dst_ip=None, ttl=64, use_inside_ports=False): if dst_ip is None: dst_ip = self.nat_addr if not use_inside_ports: @@ -506,21 +579,27 @@ class NAT44EDTestCase(VppTestCase): icmp_id = self.icmp_id_in pkts = [] # TCP - p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / - IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / - TCP(dport=tcp_port, sport=20)) + p = ( + Ether(dst=out_if.local_mac, src=out_if.remote_mac) + / IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) + / TCP(dport=tcp_port, sport=20) + ) pkts.extend([p, p]) # UDP - p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / - IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / - UDP(dport=udp_port, sport=20)) + p = ( + Ether(dst=out_if.local_mac, src=out_if.remote_mac) + / IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) + / UDP(dport=udp_port, sport=20) + ) pkts.append(p) # ICMP - p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / - IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / - ICMP(id=icmp_id, type='echo-reply')) + p = ( + Ether(dst=out_if.local_mac, src=out_if.remote_mac) + / IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) + / ICMP(id=icmp_id, type="echo-reply") + ) pkts.append(p) return pkts @@ -530,19 +609,34 @@ class NAT44EDTestCase(VppTestCase): port = 6303 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) / - TCP(sport=port + i, dport=20)) + 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) + / TCP(sport=port + i, dport=20) + ) pkts.append(p) return pkts - def create_stream_frag(self, src_if, dst, sport, dport, data, - proto=IP_PROTOS.tcp, echo_reply=False): + def create_udp_stream(self, in_if, out_if, count, base_port=6303): + return [ + ( + Ether(dst=in_if.local_mac, src=in_if.remote_mac) + / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=64) + / UDP(sport=base_port + i, dport=20) + ) + for i in range(count) + ] + + def create_stream_frag( + self, src_if, dst, sport, dport, data, proto=IP_PROTOS.tcp, echo_reply=False + ): if proto == IP_PROTOS.tcp: - p = (IP(src=src_if.remote_ip4, dst=dst) / - TCP(sport=sport, dport=dport) / - Raw(data)) + p = ( + IP(src=src_if.remote_ip4, dst=dst) + / TCP(sport=sport, dport=dport) + / Raw(data) + ) p = p.__class__(scapy.compat.raw(p)) chksum = p[TCP].chksum proto_header = TCP(sport=sport, dport=dport, chksum=chksum) @@ -550,9 +644,9 @@ class NAT44EDTestCase(VppTestCase): proto_header = UDP(sport=sport, dport=dport) elif proto == IP_PROTOS.icmp: if not echo_reply: - proto_header = ICMP(id=sport, type='echo-request') + proto_header = ICMP(id=sport, type="echo-request") else: - proto_header = ICMP(id=sport, type='echo-reply') + proto_header = ICMP(id=sport, type="echo-reply") else: raise Exception("Unsupported protocol") id = self.random_port() @@ -561,34 +655,38 @@ class NAT44EDTestCase(VppTestCase): raw = Raw(data[0:4]) else: raw = Raw(data[0:16]) - p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) / - IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=0, id=id) / - proto_header / - raw) + p = ( + Ether(src=src_if.remote_mac, dst=src_if.local_mac) + / IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=0, id=id) + / proto_header + / raw + ) pkts.append(p) if proto == IP_PROTOS.tcp: raw = Raw(data[4:20]) else: raw = Raw(data[16:32]) - p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) / - IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=3, id=id, - proto=proto) / - raw) + p = ( + Ether(src=src_if.remote_mac, dst=src_if.local_mac) + / IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=3, id=id, proto=proto) + / raw + ) pkts.append(p) if proto == IP_PROTOS.tcp: raw = Raw(data[20:]) else: raw = Raw(data[32:]) - p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) / - IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=proto, - id=id) / - raw) + p = ( + Ether(src=src_if.remote_mac, dst=src_if.local_mac) + / IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=proto, id=id) + / raw + ) pkts.append(p) return pkts - def frag_in_order_in_plus_out(self, in_addr, out_addr, in_port, out_port, - proto=IP_PROTOS.tcp): - + def frag_in_order_in_plus_out( + self, in_addr, out_addr, in_port, out_port, proto=IP_PROTOS.tcp + ): layer = self.proto2layer(proto) if proto == IP_PROTOS.tcp: @@ -599,16 +697,14 @@ class NAT44EDTestCase(VppTestCase): for i in range(2): # out2in - pkts = self.create_stream_frag(self.pg0, out_addr, - port_in, out_port, - data, proto) + pkts = self.create_stream_frag( + self.pg0, out_addr, port_in, out_port, data, proto + ) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() frags = self.pg1.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - self.pg0.remote_ip4, - in_addr) + p = self.reass_frags_and_verify(frags, self.pg0.remote_ip4, in_addr) if proto != IP_PROTOS.icmp: self.assertEqual(p[layer].sport, port_in) self.assertEqual(p[layer].dport, in_port) @@ -618,20 +714,24 @@ class NAT44EDTestCase(VppTestCase): # in2out if proto != IP_PROTOS.icmp: - pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4, - in_port, - p[layer].sport, data, proto) + pkts = self.create_stream_frag( + self.pg1, self.pg0.remote_ip4, in_port, p[layer].sport, data, proto + ) else: - pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4, - p[layer].id, 0, data, proto, - echo_reply=True) + 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) self.pg_start() frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - out_addr, - self.pg0.remote_ip4) + p = self.reass_frags_and_verify(frags, out_addr, self.pg0.remote_ip4) if proto != IP_PROTOS.icmp: self.assertEqual(p[layer].sport, out_port) self.assertEqual(p[layer].dport, port_in) @@ -639,9 +739,9 @@ class NAT44EDTestCase(VppTestCase): self.assertEqual(p[layer].id, port_in) self.assertEqual(data, p[Raw].load) - def frag_out_of_order_in_plus_out(self, in_addr, out_addr, in_port, - out_port, proto=IP_PROTOS.tcp): - + def frag_out_of_order_in_plus_out( + self, in_addr, out_addr, in_port, out_port, proto=IP_PROTOS.tcp + ): layer = self.proto2layer(proto) if proto == IP_PROTOS.tcp: @@ -652,17 +752,15 @@ class NAT44EDTestCase(VppTestCase): for i in range(2): # out2in - pkts = self.create_stream_frag(self.pg0, out_addr, - port_in, out_port, - data, proto) + pkts = self.create_stream_frag( + self.pg0, out_addr, port_in, out_port, data, proto + ) pkts.reverse() self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() frags = self.pg1.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - self.pg0.remote_ip4, - in_addr) + p = self.reass_frags_and_verify(frags, self.pg0.remote_ip4, in_addr) if proto != IP_PROTOS.icmp: self.assertEqual(p[layer].dport, in_port) self.assertEqual(p[layer].sport, port_in) @@ -673,21 +771,25 @@ class NAT44EDTestCase(VppTestCase): # in2out if proto != IP_PROTOS.icmp: - pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4, - in_port, - p[layer].sport, data, proto) + pkts = self.create_stream_frag( + self.pg1, self.pg0.remote_ip4, in_port, p[layer].sport, data, proto + ) else: - pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4, - p[layer].id, 0, data, proto, - echo_reply=True) + 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) self.pg_enable_capture(self.pg_interfaces) self.pg_start() frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - out_addr, - self.pg0.remote_ip4) + p = self.reass_frags_and_verify(frags, out_addr, self.pg0.remote_ip4) if proto != IP_PROTOS.icmp: self.assertEqual(p[layer].sport, out_port) self.assertEqual(p[layer].dport, port_in) @@ -697,9 +799,11 @@ class NAT44EDTestCase(VppTestCase): def init_tcp_session(self, in_if, out_if, in_port, ext_port): # SYN packet in->out - p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) / - IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / - TCP(sport=in_port, dport=ext_port, flags="S")) + p = ( + Ether(src=in_if.remote_mac, dst=in_if.local_mac) + / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="S") + ) in_if.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -708,18 +812,22 @@ class NAT44EDTestCase(VppTestCase): out_port = p[TCP].sport # SYN + ACK packet out->in - p = (Ether(src=out_if.remote_mac, dst=out_if.local_mac) / - IP(src=out_if.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, flags="SA")) + p = ( + Ether(src=out_if.remote_mac, dst=out_if.local_mac) + / IP(src=out_if.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="SA") + ) out_if.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() in_if.get_capture(1) # ACK packet in->out - p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) / - IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / - TCP(sport=in_port, dport=ext_port, flags="A")) + p = ( + Ether(src=in_if.remote_mac, dst=in_if.local_mac) + / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="A") + ) in_if.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -727,9 +835,10 @@ class NAT44EDTestCase(VppTestCase): return out_port - def twice_nat_common(self, self_twice_nat=False, same_pg=False, lb=False, - client_id=None): - twice_nat_addr = '10.0.1.3' + def twice_nat_common( + self, self_twice_nat=False, same_pg=False, lb=False, client_id=None + ): + twice_nat_addr = "10.0.1.3" port_in = 8080 if lb: @@ -756,8 +865,7 @@ class NAT44EDTestCase(VppTestCase): else: pg1 = self.pg1 - eh_translate = ((not self_twice_nat) or (not lb and same_pg) or - client_id == 1) + eh_translate = (not self_twice_nat) or (not lb and same_pg) or client_id == 1 self.nat_add_address(self.nat_addr) self.nat_add_address(twice_nat_addr, twice_nat=1) @@ -769,27 +877,30 @@ class NAT44EDTestCase(VppTestCase): flags |= self.config_flags.NAT_IS_TWICE_NAT if not lb: - self.nat_add_static_mapping(pg0.remote_ip4, self.nat_addr, - port_in, port_out, - proto=IP_PROTOS.tcp, - flags=flags) + self.nat_add_static_mapping( + pg0.remote_ip4, + self.nat_addr, + port_in, + port_out, + proto=IP_PROTOS.tcp, + flags=flags, + ) else: - locals = [{'addr': server1.ip4, - 'port': port_in1, - 'probability': 50, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': port_in2, - 'probability': 50, - 'vrf_id': 0}] + locals = [ + {"addr": server1.ip4, "port": port_in1, "probability": 50, "vrf_id": 0}, + {"addr": server2.ip4, "port": port_in2, "probability": 50, "vrf_id": 0}, + ] out_addr = self.nat_addr - self.vapi.nat44_add_del_lb_static_mapping(is_add=1, flags=flags, - external_addr=out_addr, - external_port=port_out, - protocol=IP_PROTOS.tcp, - local_num=len(locals), - locals=locals) + self.vapi.nat44_add_del_lb_static_mapping( + is_add=1, + flags=flags, + external_addr=out_addr, + external_port=port_out, + protocol=IP_PROTOS.tcp, + local_num=len(locals), + locals=locals, + ) self.nat_add_inside_interface(pg0) self.nat_add_outside_interface(pg1) @@ -804,9 +915,11 @@ class NAT44EDTestCase(VppTestCase): client = self.pg0.remote_hosts[1] else: client = pg1.remote_hosts[0] - p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / - IP(src=client.ip4, dst=self.nat_addr) / - TCP(sport=eh_port_out, dport=port_out)) + p = ( + Ether(src=pg1.remote_mac, dst=pg1.local_mac) + / IP(src=client.ip4, dst=self.nat_addr) + / TCP(sport=eh_port_out, dport=port_out) + ) pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -841,9 +954,11 @@ class NAT44EDTestCase(VppTestCase): self.logger.error(ppp("Unexpected or invalid packet:", p)) raise - p = (Ether(src=server.mac, dst=pg0.local_mac) / - IP(src=server.ip4, dst=eh_addr_in) / - TCP(sport=saved_port_in, dport=eh_port_in)) + p = ( + Ether(src=server.mac, dst=pg0.local_mac) + / IP(src=server.ip4, dst=eh_addr_in) + / TCP(sport=saved_port_in, dport=eh_port_in) + ) pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -864,24 +979,25 @@ class NAT44EDTestCase(VppTestCase): if eh_translate: sessions = self.vapi.nat44_user_session_dump(server.ip4, 0) self.assertEqual(len(sessions), 1) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_EXT_HOST_VALID) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_TWICE_NAT) + self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_EXT_HOST_VALID) + self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_TWICE_NAT) self.logger.info(self.vapi.cli("show nat44 sessions")) self.vapi.nat44_del_session( address=sessions[0].inside_ip_address, port=sessions[0].inside_port, protocol=sessions[0].protocol, - flags=(self.config_flags.NAT_IS_INSIDE | - self.config_flags.NAT_IS_EXT_HOST_VALID), + flags=( + self.config_flags.NAT_IS_INSIDE + | self.config_flags.NAT_IS_EXT_HOST_VALID + ), ext_host_address=sessions[0].ext_host_nat_address, - ext_host_port=sessions[0].ext_host_nat_port) + ext_host_port=sessions[0].ext_host_nat_port, + ) sessions = self.vapi.nat44_user_session_dump(server.ip4, 0) self.assertEqual(len(sessions), 0) - def verify_syslog_sess(self, data, is_add=True, is_ip6=False): - message = data.decode('utf-8') + def verify_syslog_sess(self, data, msgid, is_ip6=False): + message = data.decode("utf-8") try: message = SyslogMessage.parse(message) except ParseError as e: @@ -889,33 +1005,28 @@ class NAT44EDTestCase(VppTestCase): raise else: self.assertEqual(message.severity, SyslogSeverity.info) - self.assertEqual(message.appname, 'NAT') - self.assertEqual(message.msgid, 'SADD' if is_add else 'SDEL') - sd_params = message.sd.get('nsess') + self.assertEqual(message.appname, "NAT") + self.assertEqual(message.msgid, msgid) + sd_params = message.sd.get("nsess") self.assertTrue(sd_params is not None) if is_ip6: - self.assertEqual(sd_params.get('IATYP'), 'IPv6') - self.assertEqual(sd_params.get('ISADDR'), self.pg0.remote_ip6) + self.assertEqual(sd_params.get("IATYP"), "IPv6") + self.assertEqual(sd_params.get("ISADDR"), self.pg0.remote_ip6) else: - self.assertEqual(sd_params.get('IATYP'), 'IPv4') - self.assertEqual(sd_params.get('ISADDR'), self.pg0.remote_ip4) - self.assertTrue(sd_params.get('SSUBIX') is not None) - self.assertEqual(sd_params.get('ISPORT'), "%d" % self.tcp_port_in) - self.assertEqual(sd_params.get('XATYP'), 'IPv4') - self.assertEqual(sd_params.get('XSADDR'), self.nat_addr) - self.assertEqual(sd_params.get('XSPORT'), "%d" % self.tcp_port_out) - self.assertEqual(sd_params.get('PROTO'), "%d" % IP_PROTOS.tcp) - self.assertEqual(sd_params.get('SVLAN'), '0') - self.assertEqual(sd_params.get('XDADDR'), self.pg1.remote_ip4) - self.assertEqual(sd_params.get('XDPORT'), - "%d" % self.tcp_external_port) - - -class TestNAT44ED(NAT44EDTestCase): - """ NAT44ED Test Case """ + self.assertEqual(sd_params.get("IATYP"), "IPv4") + self.assertEqual(sd_params.get("ISADDR"), self.pg0.remote_ip4) + self.assertTrue(sd_params.get("SSUBIX") is not None) + self.assertEqual(sd_params.get("ISPORT"), "%d" % self.tcp_port_in) + self.assertEqual(sd_params.get("XATYP"), "IPv4") + self.assertEqual(sd_params.get("XSADDR"), self.nat_addr) + self.assertEqual(sd_params.get("XSPORT"), "%d" % self.tcp_port_out) + self.assertEqual(sd_params.get("PROTO"), "%d" % IP_PROTOS.tcp) + self.assertEqual(sd_params.get("SVLAN"), "0") + self.assertEqual(sd_params.get("XDADDR"), self.pg1.remote_ip4) + self.assertEqual(sd_params.get("XDPORT"), "%d" % self.tcp_external_port) def test_icmp_error(self): - """ NAT44ED test ICMP error message with inner header""" + """NAT44ED test ICMP error message with inner header""" payload = "H" * 10 @@ -924,42 +1035,54 @@ class TestNAT44ED(NAT44EDTestCase): self.nat_add_outside_interface(self.pg1) # in2out (initiate connection) - p1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - UDP(sport=21, dport=20) / payload) - - self.pg0.add_stream(p1) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg1.get_capture(1)[0] + p1 = [ + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / UDP(sport=21, dport=20) + / payload, + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=21, dport=20, flags="S") + / payload, + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / ICMP(type="echo-request", id=7777) + / payload, + ] + + capture = self.send_and_expect(self.pg0, p1, self.pg1) # out2in (send error message) - p2 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - ICMP(type='dest-unreach', code='port-unreachable') / - capture[IP:]) + p2 = [ + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / ICMP(type="dest-unreach", code="port-unreachable") + / c[IP:] + for c in capture + ] - self.pg1.add_stream(p2) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - - capture = self.pg0.get_capture(1)[0] + capture = self.send_and_expect(self.pg1, p2, self.pg0) - self.logger.info(ppp("p1 packet:", p1)) - self.logger.info(ppp("p2 packet:", p2)) - self.logger.info(ppp("capture packet:", capture)) + for c in capture: + try: + assert c[IP].dst == self.pg0.remote_ip4 + assert c[IPerror].src == self.pg0.remote_ip4 + except AssertionError as a: + raise AssertionError(f"Packet {pr(c)} not translated properly") from a def test_icmp_echo_reply_trailer(self): - """ ICMP echo reply with ethernet trailer""" + """ICMP echo reply with ethernet trailer""" self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) # in2out - p1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - ICMP(type=8, id=0xabcd, seq=0)) + p1 = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / ICMP(type=8, id=0xABCD, seq=0) + ) self.pg0.add_stream(p1) self.pg_enable_capture(self.pg_interfaces) @@ -969,9 +1092,11 @@ class TestNAT44ED(NAT44EDTestCase): self.logger.debug(self.vapi.cli("show trace")) # out2in - p2 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr, id=0xee59) / - ICMP(type=0, id=c[ICMP].id, seq=0)) + p2 = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr, id=0xEE59) + / ICMP(type=0, id=c[ICMP].id, seq=0) + ) # force checksum calculation p2 = p2.__class__(bytes(p2)) @@ -994,7 +1119,7 @@ class TestNAT44ED(NAT44EDTestCase): self.pg0.get_capture(1) def test_users_dump(self): - """ NAT44ED API test - nat44_user_dump """ + """NAT44ED API test - nat44_user_dump""" self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) @@ -1036,9 +1161,9 @@ class TestNAT44ED(NAT44EDTestCase): 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) + 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() @@ -1050,8 +1175,7 @@ class TestNAT44ED(NAT44EDTestCase): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4, - same_port=True) + self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4, same_port=True) finally: self.pg0.remote_hosts[0] = host0 @@ -1082,14 +1206,14 @@ class TestNAT44ED(NAT44EDTestCase): self.assertEqual(non_static_user.nsessions, 3) def test_frag_out_of_order_do_not_translate(self): - """ NAT44ED don't translate fragments arriving out of order """ + """NAT44ED don't translate fragments arriving out of order""" self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) self.vapi.nat44_forwarding_enable_disable(enable=True) self.frag_out_of_order(proto=IP_PROTOS.tcp, dont_translate=True) def test_forwarding(self): - """ NAT44ED forwarding test """ + """NAT44ED forwarding test""" self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) @@ -1098,11 +1222,13 @@ class TestNAT44ED(NAT44EDTestCase): real_ip = self.pg0.remote_ip4 alias_ip = self.nat_addr flags = self.config_flags.NAT_IS_ADDR_ONLY - self.vapi.nat44_add_del_static_mapping(is_add=1, - local_ip_address=real_ip, - external_ip_address=alias_ip, - external_sw_if_index=0xFFFFFFFF, - flags=flags) + self.vapi.nat44_add_del_static_mapping( + is_add=1, + local_ip_address=real_ip, + external_ip_address=alias_ip, + external_sw_if_index=0xFFFFFFFF, + flags=flags, + ) try: # in2out - static mapping match @@ -1126,9 +1252,9 @@ class TestNAT44ED(NAT44EDTestCase): 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) + 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() @@ -1140,24 +1266,27 @@ class TestNAT44ED(NAT44EDTestCase): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4, - same_port=True) + self.verify_capture_out( + capture, nat_ip=self.pg0.remote_ip4, same_port=True + ) finally: self.pg0.remote_hosts[0] = host0 user = self.pg0.remote_hosts[1] sessions = self.vapi.nat44_user_session_dump(user.ip4, 0) self.assertEqual(len(sessions), 3) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_EXT_HOST_VALID) + self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_EXT_HOST_VALID) self.vapi.nat44_del_session( address=sessions[0].inside_ip_address, port=sessions[0].inside_port, protocol=sessions[0].protocol, - flags=(self.config_flags.NAT_IS_INSIDE | - self.config_flags.NAT_IS_EXT_HOST_VALID), + flags=( + self.config_flags.NAT_IS_INSIDE + | self.config_flags.NAT_IS_EXT_HOST_VALID + ), ext_host_address=sessions[0].ext_host_address, - ext_host_port=sessions[0].ext_host_port) + ext_host_port=sessions[0].ext_host_port, + ) sessions = self.vapi.nat44_user_session_dump(user.ip4, 0) self.assertEqual(len(sessions), 2) @@ -1169,15 +1298,17 @@ class TestNAT44ED(NAT44EDTestCase): local_ip_address=real_ip, external_ip_address=alias_ip, external_sw_if_index=0xFFFFFFFF, - flags=flags) + flags=flags, + ) def test_output_feature_and_service2(self): - """ NAT44ED interface output feature and service host direct access """ + """NAT44ED interface output feature and service host direct access""" self.vapi.nat44_forwarding_enable_disable(enable=1) self.nat_add_address(self.nat_addr) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, is_add=1,) + self.vapi.nat44_ed_add_del_output_interface( + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) # session initiated from service host - translate pkts = self.create_stream_in(self.pg0, self.pg1) @@ -1204,9 +1335,9 @@ class TestNAT44ED(NAT44EDTestCase): self.icmp_id_in = 60305 try: - pkts = self.create_stream_out(self.pg1, - self.pg0.remote_ip4, - use_inside_ports=True) + pkts = self.create_stream_out( + self.pg1, self.pg0.remote_ip4, use_inside_ports=True + ) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1218,39 +1349,37 @@ class TestNAT44ED(NAT44EDTestCase): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4, - same_port=True) + self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4, same_port=True) finally: self.tcp_port_in = tcp_port_in self.udp_port_in = udp_port_in self.icmp_id_in = icmp_id_in def test_twice_nat(self): - """ NAT44ED Twice NAT """ + """NAT44ED Twice NAT""" self.twice_nat_common() def test_self_twice_nat_positive(self): - """ NAT44ED Self Twice NAT (positive test) """ + """NAT44ED Self Twice NAT (positive test)""" self.twice_nat_common(self_twice_nat=True, same_pg=True) def test_self_twice_nat_lb_positive(self): - """ NAT44ED Self Twice NAT local service load balancing (positive test) - """ - self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True, - client_id=1) + """NAT44ED Self Twice NAT local service load balancing (positive test)""" + self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True, client_id=1) def test_twice_nat_lb(self): - """ NAT44ED Twice NAT local service load balancing """ + """NAT44ED Twice NAT local service load balancing""" self.twice_nat_common(lb=True) def test_output_feature(self): - """ NAT44ED interface output feature (in2out postrouting) """ + """NAT44ED interface output feature (in2out postrouting)""" self.vapi.nat44_forwarding_enable_disable(enable=1) self.nat_add_address(self.nat_addr) self.nat_add_outside_interface(self.pg0) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, is_add=1) + self.vapi.nat44_ed_add_del_output_interface( + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) # in2out pkts = self.create_stream_in(self.pg0, self.pg1) @@ -1290,33 +1419,37 @@ class TestNAT44ED(NAT44EDTestCase): # in2out 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() - capture = self.pg0.get_capture(len(pkts)) + capture = self.send_and_expect_some(self.pg0, pkts, self.pg0) for p in capture: self.assertIn(ICMP, p) self.assertEqual(p[ICMP].type, 11) # 11 == time-exceeded def test_static_with_port_out2(self): - """ NAT44ED 1:1 NAPT asymmetrical rule """ + """NAT44ED 1:1 NAPT asymmetrical rule""" external_port = 80 local_port = 8080 self.vapi.nat44_forwarding_enable_disable(enable=1) flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr, - local_port, external_port, - proto=IP_PROTOS.tcp, flags=flags) + self.nat_add_static_mapping( + self.pg0.remote_ip4, + self.nat_addr, + local_port, + external_port, + proto=IP_PROTOS.tcp, + flags=flags, + ) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) # 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)) + 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() @@ -1333,9 +1466,12 @@ class TestNAT44ED(NAT44EDTestCase): raise # ICMP error - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - ICMP(type=11) / capture[0][IP]) + p = ( + Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / ICMP(type=11) + / capture[0][IP] + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1351,9 +1487,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from service back to client - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=local_port, dport=12345)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=local_port, dport=12345) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1370,9 +1508,12 @@ class TestNAT44ED(NAT44EDTestCase): raise # ICMP error - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - ICMP(type=11) / capture[0][IP]) + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / ICMP(type=11) + / capture[0][IP] + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1388,9 +1529,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from client to server (no translation) - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) / - TCP(sport=12346, dport=local_port)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) + / TCP(sport=12346, dport=local_port) + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1407,9 +1550,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from service back to client (no translation) - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=local_port, dport=12346)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=local_port, dport=12346) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1426,21 +1571,17 @@ class TestNAT44ED(NAT44EDTestCase): raise def test_static_lb(self): - """ NAT44ED local service load balancing """ + """NAT44ED local service load balancing""" external_addr_n = self.nat_addr external_port = 80 local_port = 8080 server1 = self.pg0.remote_hosts[0] server2 = self.pg0.remote_hosts[1] - locals = [{'addr': server1.ip4, - 'port': local_port, - 'probability': 70, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': local_port, - 'probability': 30, - 'vrf_id': 0}] + locals = [ + {"addr": server1.ip4, "port": local_port, "probability": 70, "vrf_id": 0}, + {"addr": server2.ip4, "port": local_port, "probability": 30, "vrf_id": 0}, + ] self.nat_add_address(self.nat_addr) self.vapi.nat44_add_del_lb_static_mapping( @@ -1449,19 +1590,22 @@ class TestNAT44ED(NAT44EDTestCase): external_port=external_port, protocol=IP_PROTOS.tcp, local_num=len(locals), - locals=locals) + locals=locals, + ) 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) + 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) + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) # 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)) + 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() @@ -1483,9 +1627,11 @@ class TestNAT44ED(NAT44EDTestCase): 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)) + 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() @@ -1503,56 +1649,59 @@ class TestNAT44ED(NAT44EDTestCase): sessions = self.vapi.nat44_user_session_dump(server.ip4, 0) self.assertEqual(len(sessions), 1) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_EXT_HOST_VALID) + self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_EXT_HOST_VALID) self.vapi.nat44_del_session( address=sessions[0].inside_ip_address, port=sessions[0].inside_port, protocol=sessions[0].protocol, - flags=(self.config_flags.NAT_IS_INSIDE | - self.config_flags.NAT_IS_EXT_HOST_VALID), + flags=( + self.config_flags.NAT_IS_INSIDE + | self.config_flags.NAT_IS_EXT_HOST_VALID + ), ext_host_address=sessions[0].ext_host_address, - ext_host_port=sessions[0].ext_host_port) + ext_host_port=sessions[0].ext_host_port, + ) sessions = self.vapi.nat44_user_session_dump(server.ip4, 0) self.assertEqual(len(sessions), 0) def test_static_lb_2(self): - """ NAT44ED local service load balancing (asymmetrical rule) """ + """NAT44ED local service load balancing (asymmetrical rule)""" external_addr = self.nat_addr external_port = 80 local_port = 8080 server1 = self.pg0.remote_hosts[0] server2 = self.pg0.remote_hosts[1] - locals = [{'addr': server1.ip4, - 'port': local_port, - 'probability': 70, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': local_port, - 'probability': 30, - 'vrf_id': 0}] + locals = [ + {"addr": server1.ip4, "port": local_port, "probability": 70, "vrf_id": 0}, + {"addr": server2.ip4, "port": local_port, "probability": 30, "vrf_id": 0}, + ] self.vapi.nat44_forwarding_enable_disable(enable=1) flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.vapi.nat44_add_del_lb_static_mapping(is_add=1, flags=flags, - external_addr=external_addr, - external_port=external_port, - protocol=IP_PROTOS.tcp, - local_num=len(locals), - locals=locals) + self.vapi.nat44_add_del_lb_static_mapping( + is_add=1, + flags=flags, + external_addr=external_addr, + external_port=external_port, + protocol=IP_PROTOS.tcp, + local_num=len(locals), + locals=locals, + ) 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) + 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) + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) # 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)) + 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() @@ -1574,9 +1723,11 @@ class TestNAT44ED(NAT44EDTestCase): 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)) + 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() @@ -1593,9 +1744,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from client to server (no translation) - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=server1.ip4) / - TCP(sport=12346, dport=local_port)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=server1.ip4) + / TCP(sport=12346, dport=local_port) + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1613,9 +1766,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from service back to client (no translation) - p = (Ether(src=server1.mac, dst=self.pg0.local_mac) / - IP(src=server1.ip4, dst=self.pg1.remote_ip4) / - TCP(sport=local_port, dport=12346)) + p = ( + Ether(src=server1.mac, dst=self.pg0.local_mac) + / IP(src=server1.ip4, dst=self.pg1.remote_ip4) + / TCP(sport=local_port, dport=12346) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1632,41 +1787,41 @@ class TestNAT44ED(NAT44EDTestCase): raise def test_lb_affinity(self): - """ NAT44ED local service load balancing affinity """ + """NAT44ED local service load balancing affinity""" external_addr = self.nat_addr external_port = 80 local_port = 8080 server1 = self.pg0.remote_hosts[0] server2 = self.pg0.remote_hosts[1] - locals = [{'addr': server1.ip4, - 'port': local_port, - 'probability': 50, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': local_port, - 'probability': 50, - 'vrf_id': 0}] + locals = [ + {"addr": server1.ip4, "port": local_port, "probability": 50, "vrf_id": 0}, + {"addr": server2.ip4, "port": local_port, "probability": 50, "vrf_id": 0}, + ] self.nat_add_address(self.nat_addr) - self.vapi.nat44_add_del_lb_static_mapping(is_add=1, - external_addr=external_addr, - external_port=external_port, - protocol=IP_PROTOS.tcp, - affinity=10800, - local_num=len(locals), - locals=locals) + self.vapi.nat44_add_del_lb_static_mapping( + is_add=1, + external_addr=external_addr, + external_port=external_port, + protocol=IP_PROTOS.tcp, + affinity=10800, + local_num=len(locals), + locals=locals, + ) 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) + 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) + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=1025, dport=external_port)) + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=1025, dport=external_port) + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1675,22 +1830,26 @@ class TestNAT44ED(NAT44EDTestCase): sessions = self.vapi.nat44_user_session_dump(backend, 0) self.assertEqual(len(sessions), 1) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_EXT_HOST_VALID) + self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_EXT_HOST_VALID) self.vapi.nat44_del_session( address=sessions[0].inside_ip_address, port=sessions[0].inside_port, protocol=sessions[0].protocol, - flags=(self.config_flags.NAT_IS_INSIDE | - self.config_flags.NAT_IS_EXT_HOST_VALID), + flags=( + self.config_flags.NAT_IS_INSIDE + | self.config_flags.NAT_IS_EXT_HOST_VALID + ), ext_host_address=sessions[0].ext_host_address, - ext_host_port=sessions[0].ext_host_port) + ext_host_port=sessions[0].ext_host_port, + ) pkts = [] for port in range(1030, 1100): - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=port, dport=external_port)) + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=port, dport=external_port) + ) pkts.append(p) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) @@ -1700,31 +1859,39 @@ class TestNAT44ED(NAT44EDTestCase): self.assertEqual(p[IP].dst, backend) def test_multiple_vrf_1(self): - """ Multiple VRF - both client & service in VRF1 """ + """Multiple VRF - both client & service in VRF1""" - external_addr = '1.2.3.4' + external_addr = "1.2.3.4" external_port = 80 local_port = 8080 port = 0 flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1) + sw_if_index=self.pg5.sw_if_index, is_add=1 + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1, flags=flags) + sw_if_index=self.pg5.sw_if_index, is_add=1, flags=flags + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg6.sw_if_index, - is_add=1) + sw_if_index=self.pg6.sw_if_index, is_add=1 + ) flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg5.remote_ip4, external_addr, - local_port, external_port, vrf_id=1, - proto=IP_PROTOS.tcp, flags=flags) + self.nat_add_static_mapping( + self.pg5.remote_ip4, + external_addr, + local_port, + external_port, + vrf_id=1, + proto=IP_PROTOS.tcp, + flags=flags, + ) - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=external_addr) / - TCP(sport=12345, dport=external_port)) + p = ( + Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) + / IP(src=self.pg6.remote_ip4, dst=external_addr) + / TCP(sport=12345, dport=external_port) + ) self.pg6.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1740,9 +1907,11 @@ class TestNAT44ED(NAT44EDTestCase): self.logger.error(ppp("Unexpected or invalid packet:", p)) raise - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12345)) + p = ( + Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) + / IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4) + / TCP(sport=local_port, dport=12345) + ) self.pg5.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1759,32 +1928,40 @@ class TestNAT44ED(NAT44EDTestCase): raise def test_multiple_vrf_2(self): - """ Multiple VRF - dynamic NAT from VRF1 to VRF0 (output-feature) """ + """Multiple VRF - dynamic NAT from VRF1 to VRF0 (output-feature)""" - external_addr = '1.2.3.4' + external_addr = "1.2.3.4" external_port = 80 local_port = 8080 port = 0 self.nat_add_address(self.nat_addr) flags = self.config_flags.NAT_IS_INSIDE - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, - is_add=1) + self.vapi.nat44_ed_add_del_output_interface( + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1) + sw_if_index=self.pg5.sw_if_index, is_add=1 + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1, flags=flags) + sw_if_index=self.pg5.sw_if_index, is_add=1, flags=flags + ) flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg5.remote_ip4, external_addr, - local_port, external_port, vrf_id=1, - proto=IP_PROTOS.tcp, flags=flags) + self.nat_add_static_mapping( + self.pg5.remote_ip4, + external_addr, + local_port, + external_port, + vrf_id=1, + proto=IP_PROTOS.tcp, + flags=flags, + ) - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=2345, dport=22)) + p = ( + Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) + / IP(src=self.pg5.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=2345, dport=22) + ) self.pg5.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1800,9 +1977,11 @@ class TestNAT44ED(NAT44EDTestCase): self.logger.error(ppp("Unexpected or invalid packet:", p)) raise - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=22, dport=port)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=22, dport=port) + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1819,23 +1998,23 @@ class TestNAT44ED(NAT44EDTestCase): raise def test_multiple_vrf_3(self): - """ Multiple VRF - client in VRF1, service in VRF0 """ + """Multiple VRF - client in VRF1, service in VRF0""" - external_addr = '1.2.3.4' + external_addr = "1.2.3.4" external_port = 80 local_port = 8080 port = 0 flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1) + sw_if_index=self.pg0.sw_if_index, is_add=1 + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1, flags=flags) + sw_if_index=self.pg0.sw_if_index, is_add=1, flags=flags + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg6.sw_if_index, - is_add=1) + sw_if_index=self.pg6.sw_if_index, is_add=1 + ) flags = self.config_flags.NAT_IS_OUT2IN_ONLY self.nat_add_static_mapping( self.pg0.remote_ip4, @@ -1844,13 +2023,15 @@ class TestNAT44ED(NAT44EDTestCase): vrf_id=0, external_port=external_port, proto=IP_PROTOS.tcp, - flags=flags + flags=flags, ) # from client VRF1 to service VRF0 - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=self.pg0.local_ip4) / - TCP(sport=12346, dport=external_port)) + p = ( + Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) + / IP(src=self.pg6.remote_ip4, dst=self.pg0.local_ip4) + / TCP(sport=12346, dport=external_port) + ) self.pg6.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1867,9 +2048,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from service VRF0 back to client VRF1 - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12346)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) + / TCP(sport=local_port, dport=12346) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1886,35 +2069,43 @@ class TestNAT44ED(NAT44EDTestCase): raise def test_multiple_vrf_4(self): - """ Multiple VRF - client in VRF0, service in VRF1 """ + """Multiple VRF - client in VRF0, service in VRF1""" - external_addr = '1.2.3.4' + external_addr = "1.2.3.4" external_port = 80 local_port = 8080 port = 0 flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1) + sw_if_index=self.pg0.sw_if_index, is_add=1 + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1, flags=flags) + sw_if_index=self.pg0.sw_if_index, is_add=1, flags=flags + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1) + sw_if_index=self.pg5.sw_if_index, is_add=1 + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1, flags=flags) + sw_if_index=self.pg5.sw_if_index, is_add=1, flags=flags + ) flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg5.remote_ip4, external_addr, - local_port, external_port, vrf_id=1, - proto=IP_PROTOS.tcp, flags=flags) + self.nat_add_static_mapping( + self.pg5.remote_ip4, + external_addr, + local_port, + external_port, + vrf_id=1, + proto=IP_PROTOS.tcp, + flags=flags, + ) # from client VRF0 to service VRF1 - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=external_addr) / - TCP(sport=12347, dport=external_port)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=external_addr) + / TCP(sport=12347, dport=external_port) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1931,9 +2122,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from service VRF1 back to client VRF0 - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4) / - TCP(sport=local_port, dport=12347)) + p = ( + Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) + / IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4) + / TCP(sport=local_port, dport=12347) + ) self.pg5.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1950,9 +2143,9 @@ class TestNAT44ED(NAT44EDTestCase): raise def test_multiple_vrf_5(self): - """ Multiple VRF - forwarding - no translation """ + """Multiple VRF - forwarding - no translation""" - external_addr = '1.2.3.4' + external_addr = "1.2.3.4" external_port = 80 local_port = 8080 port = 0 @@ -1960,24 +2153,30 @@ class TestNAT44ED(NAT44EDTestCase): self.vapi.nat44_forwarding_enable_disable(enable=1) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1) + sw_if_index=self.pg0.sw_if_index, is_add=1 + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1, flags=flags) + sw_if_index=self.pg0.sw_if_index, is_add=1, flags=flags + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1) + sw_if_index=self.pg5.sw_if_index, is_add=1 + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1, flags=flags) + sw_if_index=self.pg5.sw_if_index, is_add=1, flags=flags + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg6.sw_if_index, - is_add=1) + sw_if_index=self.pg6.sw_if_index, is_add=1 + ) flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg5.remote_ip4, external_addr, - local_port, external_port, vrf_id=1, - proto=IP_PROTOS.tcp, flags=flags) + self.nat_add_static_mapping( + self.pg5.remote_ip4, + external_addr, + local_port, + external_port, + vrf_id=1, + proto=IP_PROTOS.tcp, + flags=flags, + ) self.nat_add_static_mapping( self.pg0.remote_ip4, external_sw_if_index=self.pg0.sw_if_index, @@ -1985,13 +2184,15 @@ class TestNAT44ED(NAT44EDTestCase): vrf_id=0, external_port=external_port, proto=IP_PROTOS.tcp, - flags=flags + flags=flags, ) # from client to server (both VRF1, no translation) - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=self.pg5.remote_ip4) / - TCP(sport=12348, dport=local_port)) + p = ( + Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) + / IP(src=self.pg6.remote_ip4, dst=self.pg5.remote_ip4) + / TCP(sport=12348, dport=local_port) + ) self.pg6.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2008,9 +2209,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from server back to client (both VRF1, no translation) - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12348)) + p = ( + Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) + / IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4) + / TCP(sport=local_port, dport=12348) + ) self.pg5.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2027,9 +2230,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from client VRF1 to server VRF0 (no translation) - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12349)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) + / TCP(sport=local_port, dport=12349) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2046,9 +2251,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from server VRF0 back to client VRF1 (no translation) - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12349)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) + / TCP(sport=local_port, dport=12349) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2065,9 +2272,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from client VRF0 to server VRF1 (no translation) - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg5.remote_ip4) / - TCP(sport=12344, dport=local_port)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg5.remote_ip4) + / TCP(sport=12344, dport=local_port) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2084,9 +2293,11 @@ class TestNAT44ED(NAT44EDTestCase): raise # from server VRF1 back to client VRF0 (no translation) - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4) / - TCP(sport=local_port, dport=12344)) + p = ( + Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) + / IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4) + / TCP(sport=local_port, dport=12344) + ) self.pg5.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2103,14 +2314,17 @@ class TestNAT44ED(NAT44EDTestCase): raise def test_outside_address_distribution(self): - """ Outside address distribution based on source address """ + """NAT44ED outside address distribution based on source address""" + addresses = 65 x = 100 - nat_addresses = [] - for i in range(1, x): + nat_addresses = [] + nat_distribution = {} + for i in range(1, addresses): a = "10.0.0.%d" % i nat_addresses.append(a) + nat_distribution[a] = set() self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) @@ -2118,7 +2332,10 @@ class TestNAT44ED(NAT44EDTestCase): self.vapi.nat44_add_del_address_range( first_ip_address=nat_addresses[0], last_ip_address=nat_addresses[-1], - vrf_id=0xFFFFFFFF, is_add=1, flags=0) + vrf_id=0xFFFFFFFF, + is_add=1, + flags=0, + ) self.pg0.generate_remote_hosts(x) @@ -2126,11 +2343,12 @@ class TestNAT44ED(NAT44EDTestCase): for i in range(x): info = self.create_packet_info(self.pg0, self.pg1) payload = self.info_to_payload(info) - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_hosts[i].ip4, - dst=self.pg1.remote_ip4) / - UDP(sport=7000+i, dport=8000+i) / - Raw(payload)) + p = ( + Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) + / IP(src=self.pg0.remote_hosts[i].ip4, dst=self.pg1.remote_ip4) + / UDP(sport=7000 + i, dport=8000 + i) + / Raw(payload) + ) info.data = p pkts.append(p) @@ -2145,80 +2363,258 @@ class TestNAT44ED(NAT44EDTestCase): self.assertTrue(info is not None) self.assertEqual(packet_index, info.index) p_sent = info.data - packed = socket.inet_aton(p_sent[IP].src) - numeric = struct.unpack("!L", packed)[0] - numeric = socket.htonl(numeric) - a = nat_addresses[(numeric-1) % len(nat_addresses)] - self.assertEqual( - a, p_recvd[IP].src, - "Invalid packet (src IP %s translated to %s, but expected %s)" - % (p_sent[IP].src, p_recvd[IP].src, a)) + self.assertIn(p_recvd[IP].src, nat_distribution) + nat_distribution[p_recvd[IP].src].add(p_sent[IP].src) + + var = variance(map(len, nat_distribution.values()), x / addresses) + self.assertLess(var, 0.33, msg="Bad outside address distribution") + + def test_dynamic_edge_ports(self): + """NAT44ED dynamic translation test: edge ports""" + + worker_count = self.vpp_worker_count or 1 + port_offset = 1024 + port_per_thread = (65536 - port_offset) // worker_count + port_count = port_per_thread * worker_count + + # worker thread edge ports + thread_edge_ports = {0, port_offset - 1, 65535} + for i in range(0, worker_count): + port_thread_offset = (port_per_thread * i) + port_offset + for port_range_offset in [0, port_per_thread - 1]: + port = port_thread_offset + port_range_offset + thread_edge_ports.add(port) + thread_drop_ports = set( + filter( + lambda x: x not in range(port_offset, port_offset + port_count), + thread_edge_ports, + ) + ) + in_if = self.pg7 + out_if = self.pg8 -class TestNAT44EDMW(TestNAT44ED): - """ NAT44ED MW Test Case """ - vpp_worker_count = 4 - max_sessions = 5000 + self.nat_add_address(self.nat_addr) - @unittest.skip('MW fix required') - def test_users_dump(self): - """ NAT44ED API test - nat44_user_dump """ + try: + self.configure_ip4_interface(in_if, hosts=worker_count) + self.configure_ip4_interface(out_if) - @unittest.skip('MW fix required') - def test_frag_out_of_order_do_not_translate(self): - """ NAT44ED don't translate fragments arriving out of order """ + self.nat_add_inside_interface(in_if) + self.nat_add_outside_interface(out_if) - @unittest.skip('MW fix required') - def test_forwarding(self): - """ NAT44ED forwarding test """ + # in2out + tc1 = self.statistics["/nat44-ed/in2out/slowpath/tcp"] + uc1 = self.statistics["/nat44-ed/in2out/slowpath/udp"] + ic1 = self.statistics["/nat44-ed/in2out/slowpath/icmp"] + dc1 = self.statistics["/nat44-ed/in2out/slowpath/drops"] + + pkt_count = worker_count * len(thread_edge_ports) + + i2o_pkts = [[] for x in range(0, worker_count)] + for i in range(0, worker_count): + remote_host = in_if.remote_hosts[i] + for port in thread_edge_ports: + p = ( + Ether(dst=in_if.local_mac, src=in_if.remote_mac) + / IP(src=remote_host.ip4, dst=out_if.remote_ip4) + / TCP(sport=port, dport=port) + ) + i2o_pkts[i].append(p) + + p = ( + Ether(dst=in_if.local_mac, src=in_if.remote_mac) + / IP(src=remote_host.ip4, dst=out_if.remote_ip4) + / UDP(sport=port, dport=port) + ) + i2o_pkts[i].append(p) + + p = ( + Ether(dst=in_if.local_mac, src=in_if.remote_mac) + / IP(src=remote_host.ip4, dst=out_if.remote_ip4) + / ICMP(id=port, seq=port, type="echo-request") + ) + i2o_pkts[i].append(p) + + for i in range(0, worker_count): + if len(i2o_pkts[i]) > 0: + in_if.add_stream(i2o_pkts[i], worker=i) - @unittest.skip('MW fix required') - def test_twice_nat(self): - """ NAT44ED Twice NAT """ + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = out_if.get_capture(pkt_count * 3) + for packet in capture: + self.assert_packet_checksums_valid(packet) + if packet.haslayer(TCP): + self.assert_in_range( + packet[TCP].sport, + port_offset, + port_offset + port_count, + "src TCP port", + ) + elif packet.haslayer(UDP): + self.assert_in_range( + packet[UDP].sport, + port_offset, + port_offset + port_count, + "src UDP port", + ) + elif packet.haslayer(ICMP): + self.assert_in_range( + packet[ICMP].id, + port_offset, + port_offset + port_count, + "ICMP id", + ) + else: + self.fail( + ppp("Unexpected or invalid packet (outside network):", packet) + ) - @unittest.skip('MW fix required') - def test_twice_nat_lb(self): - """ NAT44ED Twice NAT local service load balancing """ + if_idx = in_if.sw_if_index + tc2 = self.statistics["/nat44-ed/in2out/slowpath/tcp"] + uc2 = self.statistics["/nat44-ed/in2out/slowpath/udp"] + ic2 = self.statistics["/nat44-ed/in2out/slowpath/icmp"] + dc2 = self.statistics["/nat44-ed/in2out/slowpath/drops"] - @unittest.skip('MW fix required') - def test_output_feature(self): - """ NAT44ED interface output feature (in2out postrouting) """ + self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count) + self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count) + self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count) + self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0) - @unittest.skip('MW fix required') - def test_static_with_port_out2(self): - """ NAT44ED 1:1 NAPT asymmetrical rule """ + # out2in + tc1 = self.statistics["/nat44-ed/out2in/fastpath/tcp"] + uc1 = self.statistics["/nat44-ed/out2in/fastpath/udp"] + ic1 = self.statistics["/nat44-ed/out2in/fastpath/icmp"] + dc1 = self.statistics["/nat44-ed/out2in/fastpath/drops"] + dc3 = self.statistics["/nat44-ed/out2in/slowpath/drops"] + + # replies to unchanged thread ports should pass on each worker, + # excluding packets outside dynamic port range + drop_count = worker_count * len(thread_drop_ports) + pass_count = worker_count * len(thread_edge_ports) - drop_count + + o2i_pkts = [[] for x in range(0, worker_count)] + for i in range(0, worker_count): + for port in thread_edge_ports: + p = ( + Ether(dst=out_if.local_mac, src=out_if.remote_mac) + / IP(src=out_if.remote_ip4, dst=self.nat_addr) + / TCP(sport=port, dport=port) + ) + o2i_pkts[i].append(p) + + p = ( + Ether(dst=out_if.local_mac, src=out_if.remote_mac) + / IP(src=out_if.remote_ip4, dst=self.nat_addr) + / UDP(sport=port, dport=port) + ) + o2i_pkts[i].append(p) + + p = ( + Ether(dst=out_if.local_mac, src=out_if.remote_mac) + / IP(src=out_if.remote_ip4, dst=self.nat_addr) + / ICMP(id=port, seq=port, type="echo-reply") + ) + o2i_pkts[i].append(p) + + for i in range(0, worker_count): + if len(o2i_pkts[i]) > 0: + out_if.add_stream(o2i_pkts[i], worker=i) - @unittest.skip('MW fix required') - def test_output_feature_and_service2(self): - """ NAT44ED interface output feature and service host direct access """ + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = in_if.get_capture(pass_count * 3) + for packet in capture: + self.assert_packet_checksums_valid(packet) + if packet.haslayer(TCP): + self.assertIn(packet[TCP].dport, thread_edge_ports, "dst TCP port") + self.assertEqual(packet[TCP].dport, packet[TCP].sport, "TCP ports") + elif packet.haslayer(UDP): + self.assertIn(packet[UDP].dport, thread_edge_ports, "dst UDP port") + self.assertEqual(packet[UDP].dport, packet[UDP].sport, "UDP ports") + elif packet.haslayer(ICMP): + self.assertIn(packet[ICMP].id, thread_edge_ports, "ICMP id") + self.assertEqual(packet[ICMP].id, packet[ICMP].seq, "ICMP id & seq") + else: + self.fail( + ppp("Unexpected or invalid packet (inside network):", packet) + ) + + if_idx = out_if.sw_if_index + tc2 = self.statistics["/nat44-ed/out2in/fastpath/tcp"] + uc2 = self.statistics["/nat44-ed/out2in/fastpath/udp"] + ic2 = self.statistics["/nat44-ed/out2in/fastpath/icmp"] + dc2 = self.statistics["/nat44-ed/out2in/fastpath/drops"] + dc4 = self.statistics["/nat44-ed/out2in/slowpath/drops"] + + self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pass_count) + self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pass_count) + self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pass_count) + self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0) + self.assertEqual( + dc4[:, if_idx].sum() - dc3[:, if_idx].sum(), drop_count * 3 + ) - @unittest.skip('MW fix required') - def test_static_lb(self): - """ NAT44ED local service load balancing """ + finally: + in_if.unconfig() + out_if.unconfig() - @unittest.skip('MW fix required') - def test_static_lb_2(self): - """ NAT44ED local service load balancing (asymmetrical rule) """ + def test_delete_interface(self): + """NAT44ED delete nat interface""" - @unittest.skip('MW fix required') - def test_lb_affinity(self): - """ NAT44ED local service load balancing affinity """ + self.nat_add_address(self.nat_addr) - @unittest.skip('MW fix required') - def test_multiple_vrf(self): - """ NAT44ED Multiple VRF setup """ + interfaces = self.create_loopback_interfaces(4) + self.nat_add_outside_interface(interfaces[0]) + self.nat_add_inside_interface(interfaces[1]) + self.nat_add_outside_interface(interfaces[2]) + self.nat_add_inside_interface(interfaces[2]) + self.vapi.nat44_ed_add_del_output_interface( + sw_if_index=interfaces[3].sw_if_index, is_add=1 + ) - @unittest.skip('MW fix required') - def test_self_twice_nat_positive(self): - """ NAT44ED Self Twice NAT (positive test) """ + nat_sw_if_indices = [ + i.sw_if_index + for i in self.vapi.nat44_interface_dump() + + list(self.vapi.vpp.details_iter(self.vapi.nat44_ed_output_interface_get)) + ] + self.assertEqual(len(nat_sw_if_indices), len(interfaces)) + + loopbacks = [] + for i in interfaces: + # delete nat-enabled interface + self.assertIn(i.sw_if_index, nat_sw_if_indices) + i.remove_vpp_config() + + # create interface with the same index + lo = VppLoInterface(self) + loopbacks.append(lo) + self.assertEqual(lo.sw_if_index, i.sw_if_index) + + # check interface is not nat-enabled + nat_sw_if_indices = [ + i.sw_if_index + for i in self.vapi.nat44_interface_dump() + + list( + self.vapi.vpp.details_iter(self.vapi.nat44_ed_output_interface_get) + ) + ] + self.assertNotIn(lo.sw_if_index, nat_sw_if_indices) + + for i in loopbacks: + i.remove_vpp_config() + + +@tag_fixme_ubuntu2204 +class TestNAT44EDMW(TestNAT44ED): + """NAT44ED MW Test Case""" - @unittest.skip('MW fix required') - def test_self_twice_nat_lb_positive(self): - """ NAT44ED Self Twice NAT local service load balancing (positive test) - """ + vpp_worker_count = 4 + max_sessions = 5000 def test_dynamic(self): - """ NAT44ED dynamic translation test """ + """NAT44ED dynamic translation test""" pkt_count = 1500 tcp_port_offset = 20 udp_port_offset = 20 @@ -2229,27 +2625,33 @@ class TestNAT44EDMW(TestNAT44ED): self.nat_add_outside_interface(self.pg1) # in2out - tc1 = self.statistics['/nat44-ed/in2out/slowpath/tcp'] - uc1 = self.statistics['/nat44-ed/in2out/slowpath/udp'] - ic1 = self.statistics['/nat44-ed/in2out/slowpath/icmp'] - dc1 = self.statistics['/nat44-ed/in2out/slowpath/drops'] + tc1 = self.statistics["/nat44-ed/in2out/slowpath/tcp"] + uc1 = self.statistics["/nat44-ed/in2out/slowpath/udp"] + ic1 = self.statistics["/nat44-ed/in2out/slowpath/icmp"] + dc1 = self.statistics["/nat44-ed/in2out/slowpath/drops"] i2o_pkts = [[] for x in range(0, self.vpp_worker_count)] for i in range(pkt_count): - 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_offset + i, dport=20)) + 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_offset + i, dport=20) + ) i2o_pkts[p[TCP].sport % self.vpp_worker_count].append(p) - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - UDP(sport=udp_port_offset + i, dport=20)) + p = ( + Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / UDP(sport=udp_port_offset + i, dport=20) + ) i2o_pkts[p[UDP].sport % self.vpp_worker_count].append(p) - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - ICMP(id=icmp_id_offset + i, type='echo-request')) + p = ( + Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / ICMP(id=icmp_id_offset + i, type="echo-request") + ) i2o_pkts[p[ICMP].id % self.vpp_worker_count].append(p) for i in range(0, self.vpp_worker_count): @@ -2258,29 +2660,26 @@ class TestNAT44EDMW(TestNAT44ED): self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg1.get_capture(pkt_count * 3) + capture = self.pg1.get_capture(pkt_count * 3, timeout=5) if_idx = self.pg0.sw_if_index - tc2 = self.statistics['/nat44-ed/in2out/slowpath/tcp'] - uc2 = self.statistics['/nat44-ed/in2out/slowpath/udp'] - ic2 = self.statistics['/nat44-ed/in2out/slowpath/icmp'] - dc2 = self.statistics['/nat44-ed/in2out/slowpath/drops'] - - self.assertEqual( - tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count) - self.assertEqual( - uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count) - self.assertEqual( - ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count) + tc2 = self.statistics["/nat44-ed/in2out/slowpath/tcp"] + uc2 = self.statistics["/nat44-ed/in2out/slowpath/udp"] + ic2 = self.statistics["/nat44-ed/in2out/slowpath/icmp"] + dc2 = self.statistics["/nat44-ed/in2out/slowpath/drops"] + + self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count) + self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count) + self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count) self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0) self.logger.info(self.vapi.cli("show trace")) # out2in - tc1 = self.statistics['/nat44-ed/out2in/fastpath/tcp'] - uc1 = self.statistics['/nat44-ed/out2in/fastpath/udp'] - ic1 = self.statistics['/nat44-ed/out2in/fastpath/icmp'] - dc1 = self.statistics['/nat44-ed/out2in/fastpath/drops'] + tc1 = self.statistics["/nat44-ed/out2in/fastpath/tcp"] + uc1 = self.statistics["/nat44-ed/out2in/fastpath/udp"] + ic1 = self.statistics["/nat44-ed/out2in/fastpath/icmp"] + dc1 = self.statistics["/nat44-ed/out2in/fastpath/drops"] recvd_tcp_ports = set() recvd_udp_ports = set() @@ -2300,19 +2699,25 @@ class TestNAT44EDMW(TestNAT44ED): o2i_pkts = [[] for x in range(0, self.vpp_worker_count)] for i in range(pkt_count): - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(dport=choice(recvd_tcp_ports), sport=20)) + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(dport=choice(recvd_tcp_ports), sport=20) + ) o2i_pkts[p[TCP].dport % self.vpp_worker_count].append(p) - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - UDP(dport=choice(recvd_udp_ports), sport=20)) + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / UDP(dport=choice(recvd_udp_ports), sport=20) + ) o2i_pkts[p[UDP].dport % self.vpp_worker_count].append(p) - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - ICMP(id=choice(recvd_icmp_ids), type='echo-reply')) + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / ICMP(id=choice(recvd_icmp_ids), type="echo-reply") + ) o2i_pkts[p[ICMP].id % self.vpp_worker_count].append(p) for i in range(0, self.vpp_worker_count): @@ -2328,41 +2733,50 @@ class TestNAT44EDMW(TestNAT44ED): self.assertEqual(packet[IP].dst, self.pg0.remote_ip4) if packet.haslayer(TCP): self.assert_in_range( - packet[TCP].dport, tcp_port_offset, - tcp_port_offset + pkt_count, "dst TCP port") + packet[TCP].dport, + tcp_port_offset, + tcp_port_offset + pkt_count, + "dst TCP port", + ) elif packet.haslayer(UDP): self.assert_in_range( - packet[UDP].dport, udp_port_offset, - udp_port_offset + pkt_count, "dst UDP port") + packet[UDP].dport, + udp_port_offset, + udp_port_offset + pkt_count, + "dst UDP port", + ) else: self.assert_in_range( - packet[ICMP].id, icmp_id_offset, - icmp_id_offset + pkt_count, "ICMP id") + packet[ICMP].id, + icmp_id_offset, + icmp_id_offset + pkt_count, + "ICMP id", + ) except: - self.logger.error(ppp("Unexpected or invalid packet " - "(inside network):", packet)) + self.logger.error( + ppp("Unexpected or invalid packet (inside network):", packet) + ) raise if_idx = self.pg1.sw_if_index - tc2 = self.statistics['/nat44-ed/out2in/fastpath/tcp'] - uc2 = self.statistics['/nat44-ed/out2in/fastpath/udp'] - ic2 = self.statistics['/nat44-ed/out2in/fastpath/icmp'] - dc2 = self.statistics['/nat44-ed/out2in/fastpath/drops'] - - self.assertEqual( - tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count) - self.assertEqual( - uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count) - self.assertEqual( - ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count) + tc2 = self.statistics["/nat44-ed/out2in/fastpath/tcp"] + uc2 = self.statistics["/nat44-ed/out2in/fastpath/udp"] + ic2 = self.statistics["/nat44-ed/out2in/fastpath/icmp"] + dc2 = self.statistics["/nat44-ed/out2in/fastpath/drops"] + + self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count) + self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count) + self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count) self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0) - sc = self.statistics['/nat44-ed/total-sessions'] - self.assertEqual(sc[:, 0].sum(), len(recvd_tcp_ports) + - len(recvd_udp_ports) + len(recvd_icmp_ids)) + sc = self.statistics["/nat44-ed/total-sessions"] + self.assertEqual( + sc[:, 0].sum(), + len(recvd_tcp_ports) + len(recvd_udp_ports) + len(recvd_icmp_ids), + ) def test_frag_in_order(self): - """ NAT44ED translate fragments arriving in order """ + """NAT44ED translate fragments arriving in order""" self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) @@ -2373,7 +2787,7 @@ class TestNAT44EDMW(TestNAT44ED): self.frag_in_order(proto=IP_PROTOS.icmp, ignore_port=True) def test_frag_in_order_do_not_translate(self): - """ NAT44ED don't translate fragments arriving in order """ + """NAT44ED don't translate fragments arriving in order""" self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) @@ -2383,7 +2797,7 @@ class TestNAT44EDMW(TestNAT44ED): self.frag_in_order(proto=IP_PROTOS.tcp, dont_translate=True) def test_frag_out_of_order(self): - """ NAT44ED translate fragments arriving out of order """ + """NAT44ED translate fragments arriving out of order""" self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) @@ -2394,7 +2808,7 @@ class TestNAT44EDMW(TestNAT44ED): self.frag_out_of_order(proto=IP_PROTOS.icmp, ignore_port=True) def test_frag_in_order_in_plus_out(self): - """ NAT44ED in+out interface fragments in order """ + """NAT44ED in+out interface fragments in order""" in_port = self.random_port() out_port = self.random_port() @@ -2406,39 +2820,29 @@ class TestNAT44EDMW(TestNAT44ED): self.nat_add_outside_interface(self.pg1) # add static mappings for server - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - in_port, - out_port, - proto=IP_PROTOS.tcp) - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - in_port, - out_port, - proto=IP_PROTOS.udp) - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - proto=IP_PROTOS.icmp) + self.nat_add_static_mapping( + self.server_addr, self.nat_addr, in_port, out_port, proto=IP_PROTOS.tcp + ) + self.nat_add_static_mapping( + self.server_addr, self.nat_addr, in_port, out_port, proto=IP_PROTOS.udp + ) + self.nat_add_static_mapping( + self.server_addr, self.nat_addr, proto=IP_PROTOS.icmp + ) # run tests for each protocol - self.frag_in_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.tcp) - self.frag_in_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.udp) - self.frag_in_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.icmp) + self.frag_in_order_in_plus_out( + self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.tcp + ) + self.frag_in_order_in_plus_out( + self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.udp + ) + self.frag_in_order_in_plus_out( + self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.icmp + ) def test_frag_out_of_order_in_plus_out(self): - """ NAT44ED in+out interface fragments out of order """ + """NAT44ED in+out interface fragments out of order""" in_port = self.random_port() out_port = self.random_port() @@ -2450,39 +2854,29 @@ class TestNAT44EDMW(TestNAT44ED): self.nat_add_outside_interface(self.pg1) # add static mappings for server - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - in_port, - out_port, - proto=IP_PROTOS.tcp) - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - in_port, - out_port, - proto=IP_PROTOS.udp) - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - proto=IP_PROTOS.icmp) + self.nat_add_static_mapping( + self.server_addr, self.nat_addr, in_port, out_port, proto=IP_PROTOS.tcp + ) + self.nat_add_static_mapping( + self.server_addr, self.nat_addr, in_port, out_port, proto=IP_PROTOS.udp + ) + self.nat_add_static_mapping( + self.server_addr, self.nat_addr, proto=IP_PROTOS.icmp + ) # run tests for each protocol - self.frag_out_of_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.tcp) - self.frag_out_of_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.udp) - self.frag_out_of_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.icmp) + self.frag_out_of_order_in_plus_out( + self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.tcp + ) + self.frag_out_of_order_in_plus_out( + self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.udp + ) + self.frag_out_of_order_in_plus_out( + self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.icmp + ) def test_reass_hairpinning(self): - """ NAT44ED fragments hairpinning """ + """NAT44ED fragments hairpinning""" server_addr = self.pg0.remote_hosts[1].ip4 @@ -2495,26 +2889,49 @@ class TestNAT44EDMW(TestNAT44ED): self.nat_add_outside_interface(self.pg1) # add static mapping for server - self.nat_add_static_mapping(server_addr, self.nat_addr, - server_in_port, server_out_port, - proto=IP_PROTOS.tcp) - self.nat_add_static_mapping(server_addr, self.nat_addr, - server_in_port, server_out_port, - proto=IP_PROTOS.udp) + self.nat_add_static_mapping( + server_addr, + self.nat_addr, + server_in_port, + server_out_port, + proto=IP_PROTOS.tcp, + ) + self.nat_add_static_mapping( + server_addr, + self.nat_addr, + server_in_port, + server_out_port, + proto=IP_PROTOS.udp, + ) self.nat_add_static_mapping(server_addr, self.nat_addr) - self.reass_hairpinning(server_addr, server_in_port, server_out_port, - host_in_port, proto=IP_PROTOS.tcp, - ignore_port=True) - self.reass_hairpinning(server_addr, server_in_port, server_out_port, - host_in_port, proto=IP_PROTOS.udp, - ignore_port=True) - self.reass_hairpinning(server_addr, server_in_port, server_out_port, - host_in_port, proto=IP_PROTOS.icmp, - ignore_port=True) + self.reass_hairpinning( + server_addr, + server_in_port, + server_out_port, + host_in_port, + proto=IP_PROTOS.tcp, + ignore_port=True, + ) + self.reass_hairpinning( + server_addr, + server_in_port, + server_out_port, + host_in_port, + proto=IP_PROTOS.udp, + ignore_port=True, + ) + self.reass_hairpinning( + server_addr, + server_in_port, + server_out_port, + host_in_port, + proto=IP_PROTOS.icmp, + ignore_port=True, + ) def test_session_limit_per_vrf(self): - """ NAT44ED per vrf session limit """ + """NAT44ED per vrf session limit""" inside = self.pg0 inside_vrf10 = self.pg2 @@ -2522,10 +2939,13 @@ class TestNAT44EDMW(TestNAT44ED): limit = 5 - # 2 interfaces pg0, pg1 (vrf10, limit 1 tcp session) - # non existing vrf_id makes process core dump + # 2 interfaces pg0, pg1 (vrf10, limit 5 tcp sessions) self.vapi.nat44_set_session_limit(session_limit=limit, vrf_id=10) + # expect error when bad is specified + with self.vapi.assert_negative_api_retval(): + self.vapi.nat44_set_session_limit(session_limit=limit, vrf_id=20) + self.nat_add_inside_interface(inside) self.nat_add_inside_interface(inside_vrf10) self.nat_add_outside_interface(outside) @@ -2553,40 +2973,44 @@ class TestNAT44EDMW(TestNAT44ED): capture = outside.get_capture(len(stream)) def test_show_max_translations(self): - """ NAT44ED API test - max translations per thread """ - nat_config = self.vapi.nat_show_config_2() - self.assertEqual(self.max_sessions, - nat_config.max_translations_per_thread) + """NAT44ED API test - max translations per thread""" + config = self.vapi.nat44_show_running_config() + self.assertEqual(self.max_sessions, config.sessions) def test_lru_cleanup(self): - """ NAT44ED LRU cleanup algorithm """ + """NAT44ED LRU cleanup algorithm""" self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) self.vapi.nat_set_timeouts( - udp=1, tcp_established=7440, tcp_transitory=30, icmp=1) + udp=1, tcp_established=7440, tcp_transitory=30, icmp=1 + ) tcp_port_out = self.init_tcp_session(self.pg0, self.pg1, 2000, 80) pkts = [] for i in range(0, self.max_sessions - 1): - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) / - UDP(sport=7000+i, dport=80)) + p = ( + Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) + / UDP(sport=7000 + i, dport=80) + ) pkts.append(p) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg1.get_capture(len(pkts)) - self.sleep(1.5, "wait for timeouts") + self.virtual_sleep(1.5, "wait for timeouts") pkts = [] for i in range(0, self.max_sessions - 1): - p = (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=8000+i, type='echo-request')) + p = ( + 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=8000 + i, type="echo-request") + ) pkts.append(p) self.pg0.add_stream(pkts) @@ -2595,97 +3019,162 @@ class TestNAT44EDMW(TestNAT44ED): self.pg1.get_capture(len(pkts)) def test_session_rst_timeout(self): - """ NAT44ED session RST timeouts """ + """NAT44ED session RST timeouts""" self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) - self.vapi.nat_set_timeouts(udp=300, tcp_established=7440, - tcp_transitory=5, icmp=60) + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=5, icmp=60 + ) + + self.init_tcp_session( + self.pg0, self.pg1, self.tcp_port_in, self.tcp_external_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=self.tcp_port_in, dport=self.tcp_external_port, flags="R") + ) + self.send_and_expect(self.pg0, p, self.pg1) - self.init_tcp_session(self.pg0, self.pg1, self.tcp_port_in, - self.tcp_external_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=self.tcp_port_in, dport=self.tcp_external_port, - flags="R")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) + self.virtual_sleep(6) - self.sleep(6) + # The session is already closed + 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="P") + ) + self.send_and_assert_no_replies(self.pg0, p, self.pg1) - 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 + 1, dport=self.tcp_external_port + 1, - flags="S")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) + # The session can be re-opened + 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="S") + ) + self.send_and_expect(self.pg0, p, self.pg1) - def test_dynamic_out_of_ports(self): - """ NAT44ED dynamic translation test: out of ports """ + def test_session_rst_established_timeout(self): + """NAT44ED session RST timeouts""" + self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) - # in2out and no NAT addresses added - err_old = self.statistics.get_err_counter( - '/err/nat44-ed-in2out-slowpath/out of ports') + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=5, icmp=60 + ) - pkts = self.create_stream_in(self.pg0, self.pg1) - self.pg0.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(0, timeout=1) + self.init_tcp_session( + self.pg0, self.pg1, self.tcp_port_in, self.tcp_external_port + ) - err_new = self.statistics.get_err_counter( - '/err/nat44-ed-in2out-slowpath/out of ports') + # Wait at least the transitory time, the session is in established + # state anyway. RST followed by a data packet should move it to + # transitory state. + self.virtual_sleep(6) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="R") + ) + self.send_and_expect(self.pg0, p, self.pg1) - self.assertEqual(err_new - err_old, len(pkts)) + 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="P") + ) + self.send_and_expect(self.pg0, p, self.pg1) - # in2out after NAT addresses added - self.nat_add_address(self.nat_addr) + # State is transitory, session should be closed after 6 seconds + self.virtual_sleep(6) - err_old = self.statistics.get_err_counter( - '/err/nat44-ed-in2out-slowpath/out of ports') + 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="P") + ) + self.send_and_assert_no_replies(self.pg0, p, self.pg1) + + def test_dynamic_out_of_ports(self): + """NAT44ED dynamic translation test: out of ports""" + + self.nat_add_inside_interface(self.pg0) + self.nat_add_outside_interface(self.pg1) + # in2out and no NAT addresses added 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, ignore_port=True) - err_new = self.statistics.get_err_counter( - '/err/nat44-ed-in2out-slowpath/out of ports') + self.send_and_assert_no_replies( + self.pg0, + pkts, + msg="i2o pkts", + stats_diff=self.no_diff + | { + "err": { + "/err/nat44-ed-in2out-slowpath/out of ports": len(pkts), + }, + self.pg0.sw_if_index: { + "/nat44-ed/in2out/slowpath/drops": len(pkts), + }, + }, + ) + + # in2out after NAT addresses added + self.nat_add_address(self.nat_addr) + + tcpn, udpn, icmpn = ( + sum(x) for x in zip(*((TCP in p, UDP in p, ICMP in p) for p in pkts)) + ) - self.assertEqual(err_new, err_old) + self.send_and_expect( + self.pg0, + pkts, + self.pg1, + msg="i2o pkts", + stats_diff=self.no_diff + | { + "err": { + "/err/nat44-ed-in2out-slowpath/out of ports": 0, + }, + self.pg0.sw_if_index: { + "/nat44-ed/in2out/slowpath/drops": 0, + "/nat44-ed/in2out/slowpath/tcp": tcpn, + "/nat44-ed/in2out/slowpath/udp": udpn, + "/nat44-ed/in2out/slowpath/icmp": icmpn, + }, + }, + ) def test_unknown_proto(self): - """ NAT44ED translate packet with unknown protocol """ + """NAT44ED translate packet with unknown protocol""" self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) # in2out - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=self.tcp_port_in, dport=20)) + p = ( + Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=self.tcp_port_in, dport=20) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg1.get_capture(1) - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - GRE() / - IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) / - TCP(sport=1234, dport=1234)) + p = ( + Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / GRE() + / IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) + / TCP(sport=1234, dport=1234) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2701,11 +3190,13 @@ class TestNAT44EDMW(TestNAT44ED): raise # out2in - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - GRE() / - IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) / - TCP(sport=1234, dport=1234)) + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / GRE() + / IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) + / TCP(sport=1234, dport=1234) + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2721,7 +3212,7 @@ class TestNAT44EDMW(TestNAT44ED): raise def test_hairpinning_unknown_proto(self): - """ NAT44ED translate packet with unknown protocol - hairpinning """ + """NAT44ED translate packet with unknown protocol - hairpinning""" host = self.pg0.remote_hosts[0] server = self.pg0.remote_hosts[1] host_in_port = 1234 @@ -2736,19 +3227,23 @@ class TestNAT44EDMW(TestNAT44ED): self.nat_add_static_mapping(server.ip4, server_nat_ip) # host to server - p = (Ether(src=host.mac, dst=self.pg0.local_mac) / - IP(src=host.ip4, dst=server_nat_ip) / - TCP(sport=host_in_port, dport=server_out_port)) + p = ( + Ether(src=host.mac, dst=self.pg0.local_mac) + / IP(src=host.ip4, dst=server_nat_ip) + / TCP(sport=host_in_port, dport=server_out_port) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg0.get_capture(1) - p = (Ether(dst=self.pg0.local_mac, src=host.mac) / - IP(src=host.ip4, dst=server_nat_ip) / - GRE() / - IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) / - TCP(sport=1234, dport=1234)) + p = ( + Ether(dst=self.pg0.local_mac, src=host.mac) + / IP(src=host.ip4, dst=server_nat_ip) + / GRE() + / IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) + / TCP(sport=1234, dport=1234) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2764,11 +3259,13 @@ class TestNAT44EDMW(TestNAT44ED): raise # server to host - p = (Ether(dst=self.pg0.local_mac, src=server.mac) / - IP(src=server.ip4, dst=self.nat_addr) / - GRE() / - IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) / - TCP(sport=1234, dport=1234)) + p = ( + Ether(dst=self.pg0.local_mac, src=server.mac) + / IP(src=server.ip4, dst=self.nat_addr) + / GRE() + / IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) + / TCP(sport=1234, dport=1234) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2784,8 +3281,8 @@ class TestNAT44EDMW(TestNAT44ED): raise def test_output_feature_and_service(self): - """ NAT44ED interface output feature and services """ - external_addr = '1.2.3.4' + """NAT44ED interface output feature and services""" + external_addr = "1.2.3.4" external_port = 80 local_port = 8080 @@ -2793,22 +3290,33 @@ class TestNAT44EDMW(TestNAT44ED): self.nat_add_address(self.nat_addr) flags = self.config_flags.NAT_IS_ADDR_ONLY self.vapi.nat44_add_del_identity_mapping( - ip_address=self.pg1.remote_ip4, sw_if_index=0xFFFFFFFF, - flags=flags, is_add=1) + ip_address=self.pg1.remote_ip4, + sw_if_index=0xFFFFFFFF, + flags=flags, + is_add=1, + ) flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg0.remote_ip4, external_addr, - local_port, external_port, - proto=IP_PROTOS.tcp, flags=flags) + self.nat_add_static_mapping( + self.pg0.remote_ip4, + external_addr, + local_port, + external_port, + proto=IP_PROTOS.tcp, + flags=flags, + ) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg0) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, is_add=1) + self.vapi.nat44_ed_add_del_output_interface( + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) # from client to service - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=external_addr) / - TCP(sport=12345, dport=external_port)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=external_addr) + / TCP(sport=12345, dport=external_port) + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2825,9 +3333,11 @@ class TestNAT44EDMW(TestNAT44ED): raise # from service back to client - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=local_port, dport=12345)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=local_port, dport=12345) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2866,26 +3376,34 @@ class TestNAT44EDMW(TestNAT44ED): self.verify_capture_in(capture, self.pg0) def test_output_feature_and_service3(self): - """ NAT44ED interface output feature and DST NAT """ - external_addr = '1.2.3.4' + """NAT44ED interface output feature and DST NAT""" + external_addr = "1.2.3.4" external_port = 80 local_port = 8080 self.vapi.nat44_forwarding_enable_disable(enable=1) self.nat_add_address(self.nat_addr) flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg1.remote_ip4, external_addr, - local_port, external_port, - proto=IP_PROTOS.tcp, flags=flags) + self.nat_add_static_mapping( + self.pg1.remote_ip4, + external_addr, + local_port, + external_port, + proto=IP_PROTOS.tcp, + flags=flags, + ) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg0) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, is_add=1) + self.vapi.nat44_ed_add_del_output_interface( + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=external_addr) / - TCP(sport=12345, dport=external_port)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=external_addr) + / TCP(sport=12345, dport=external_port) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2903,9 +3421,11 @@ class TestNAT44EDMW(TestNAT44ED): self.logger.error(ppp("Unexpected or invalid packet:", p)) raise - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) / - TCP(sport=local_port, dport=12345)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) + / TCP(sport=local_port, dport=12345) + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -2924,17 +3444,15 @@ class TestNAT44EDMW(TestNAT44ED): raise def test_self_twice_nat_lb_negative(self): - """ NAT44ED Self Twice NAT local service load balancing (negative test) - """ - self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True, - client_id=2) + """NAT44ED Self Twice NAT local service load balancing (negative test)""" + self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True, client_id=2) def test_self_twice_nat_negative(self): - """ NAT44ED Self Twice NAT (negative test) """ + """NAT44ED Self Twice NAT (negative test)""" self.twice_nat_common(self_twice_nat=True) def test_static_lb_multi_clients(self): - """ NAT44ED local service load balancing - multiple clients""" + """NAT44ED local service load balancing - multiple clients""" external_addr = self.nat_addr external_port = 80 @@ -2943,39 +3461,39 @@ class TestNAT44EDMW(TestNAT44ED): server2 = self.pg0.remote_hosts[1] server3 = self.pg0.remote_hosts[2] - locals = [{'addr': server1.ip4, - 'port': local_port, - 'probability': 90, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': local_port, - 'probability': 10, - 'vrf_id': 0}] + locals = [ + {"addr": server1.ip4, "port": local_port, "probability": 90, "vrf_id": 0}, + {"addr": server2.ip4, "port": local_port, "probability": 10, "vrf_id": 0}, + ] 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) + 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) + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) self.nat_add_address(self.nat_addr) - self.vapi.nat44_add_del_lb_static_mapping(is_add=1, - external_addr=external_addr, - external_port=external_port, - protocol=IP_PROTOS.tcp, - local_num=len(locals), - locals=locals) + self.vapi.nat44_add_del_lb_static_mapping( + is_add=1, + external_addr=external_addr, + external_port=external_port, + protocol=IP_PROTOS.tcp, + local_num=len(locals), + locals=locals, + ) server1_n = 0 server2_n = 0 clients = ip4_range(self.pg1.remote_ip4, 10, 50) pkts = [] for client in clients: - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=client, dst=self.nat_addr) / - TCP(sport=12345, dport=external_port)) + 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) @@ -2989,10 +3507,10 @@ class TestNAT44EDMW(TestNAT44ED): self.assertGreaterEqual(server1_n, server2_n) local = { - 'addr': server3.ip4, - 'port': local_port, - 'probability': 20, - 'vrf_id': 0 + "addr": server3.ip4, + "port": local_port, + "probability": 20, + "vrf_id": 0, } # add new back-end @@ -3001,16 +3519,19 @@ class TestNAT44EDMW(TestNAT44ED): external_addr=external_addr, external_port=external_port, local=local, - protocol=IP_PROTOS.tcp) + protocol=IP_PROTOS.tcp, + ) server1_n = 0 server2_n = 0 server3_n = 0 clients = ip4_range(self.pg1.remote_ip4, 60, 110) 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=12346, dport=external_port)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=client, dst=self.nat_addr) + / TCP(sport=12346, dport=external_port) + ) pkts.append(p) self.assertGreater(len(pkts), 0) self.pg1.add_stream(pkts) @@ -3029,10 +3550,10 @@ class TestNAT44EDMW(TestNAT44ED): self.assertGreater(server3_n, 0) local = { - 'addr': server2.ip4, - 'port': local_port, - 'probability': 10, - 'vrf_id': 0 + "addr": server2.ip4, + "port": local_port, + "probability": 10, + "vrf_id": 0, } # remove one back-end @@ -3041,7 +3562,8 @@ class TestNAT44EDMW(TestNAT44ED): external_addr=external_addr, external_port=external_port, local=local, - protocol=IP_PROTOS.tcp) + protocol=IP_PROTOS.tcp, + ) server1_n = 0 server2_n = 0 server3_n = 0 @@ -3060,39 +3582,106 @@ class TestNAT44EDMW(TestNAT44ED): self.assertEqual(server2_n, 0) self.assertGreater(server3_n, 0) - def test_syslog_sess(self): - """ NAT44ED Test syslog session creation and deletion """ - self.vapi.syslog_set_filter( - self.syslog_severity.SYSLOG_API_SEVERITY_INFO) + # put zzz in front of syslog test name so that it runs as a last test + # setting syslog sender cannot be undone and if it is set, it messes + # with self.send_and_assert_no_replies functionality + def test_zzz_syslog_sess(self): + """NAT44ED Test syslog session creation and deletion""" + self.vapi.syslog_set_filter(self.syslog_severity.SYSLOG_API_SEVERITY_INFO) self.vapi.syslog_set_sender(self.pg3.local_ip4, self.pg3.remote_ip4) self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) - 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=self.tcp_external_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=self.tcp_port_in, dport=self.tcp_external_port) + ) 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_sess(capture[0][Raw].load) + self.verify_syslog_sess(capture[0][Raw].load, "SADD") self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.nat_add_address(self.nat_addr, is_add=0) capture = self.pg3.get_capture(1) - self.verify_syslog_sess(capture[0][Raw].load, False) + self.verify_syslog_sess(capture[0][Raw].load, "SDEL") + + # put zzz in front of syslog test name so that it runs as a last test + # setting syslog sender cannot be undone and if it is set, it messes + # with self.send_and_assert_no_replies functionality + def test_zzz_syslog_sess_reopen(self): + """Syslog events for session reopen""" + self.vapi.syslog_set_filter(self.syslog_severity.SYSLOG_API_SEVERITY_INFO) + self.vapi.syslog_set_sender(self.pg3.local_ip4, self.pg3.remote_ip4) + + self.nat_add_address(self.nat_addr) + self.nat_add_inside_interface(self.pg0) + self.nat_add_outside_interface(self.pg1) + + # SYN in2out + p = ( + Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port) + ) + capture = self.send_and_expect(self.pg0, p, self.pg1)[0] + self.tcp_port_out = capture[0][TCP].sport + capture = self.pg3.get_capture(1) + self.verify_syslog_sess(capture[0][Raw].load, "SADD") + + # SYN out2in + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="SA") + ) + self.send_and_expect(self.pg1, p, self.pg0) + + 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=self.tcp_external_port, flags="A") + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # FIN in2out + p = ( + Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="F") + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # FIN out2in + p = ( + Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="F") + ) + self.send_and_expect(self.pg1, p, self.pg0) + + self.init_tcp_session( + self.pg0, self.pg1, self.tcp_port_in, self.tcp_external_port + ) + + # 2 records should be produced - first one del & add + capture = self.pg3.get_capture(2) + self.verify_syslog_sess(capture[0][Raw].load, "SDEL") + self.verify_syslog_sess(capture[1][Raw].load, "SADD") def test_twice_nat_interface_addr(self): - """ NAT44ED Acquire twice NAT addresses from interface """ + """NAT44ED Acquire twice NAT addresses from interface""" flags = self.config_flags.NAT_IS_TWICE_NAT self.vapi.nat44_add_del_interface_addr( - sw_if_index=self.pg11.sw_if_index, - flags=flags, is_add=1) + sw_if_index=self.pg11.sw_if_index, flags=flags, is_add=1 + ) # no address in NAT pool adresses = self.vapi.nat44_address_dump() @@ -3102,8 +3691,7 @@ class TestNAT44EDMW(TestNAT44ED): self.pg11.config_ip4() adresses = self.vapi.nat44_address_dump() self.assertEqual(1, len(adresses)) - self.assertEqual(str(adresses[0].ip_address), - self.pg11.local_ip4) + self.assertEqual(str(adresses[0].ip_address), self.pg11.local_ip4) self.assertEqual(adresses[0].flags, flags) # remove interface address and check NAT address pool @@ -3112,11 +3700,12 @@ class TestNAT44EDMW(TestNAT44ED): self.assertEqual(0, len(adresses)) def test_output_feature_stateful_acl(self): - """ NAT44ED output feature works with stateful ACL """ + """NAT44ED output feature works with stateful ACL""" self.nat_add_address(self.nat_addr) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, is_add=1) + self.vapi.nat44_ed_add_del_output_interface( + sw_if_index=self.pg1.sw_if_index, is_add=1 + ) # First ensure that the NAT is working sans ACL @@ -3126,14 +3715,16 @@ class TestNAT44EDMW(TestNAT44ED): # send packets into inside intf, ensure received via outside intf pkts_in2out = self.create_stream_in(self.pg0, self.pg1) - capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1, - len(pkts_in2out)) + capture = self.send_and_expect( + self.pg0, pkts_in2out, self.pg1, len(pkts_in2out) + ) self.verify_capture_out(capture, ignore_port=True) # send out2in again, with sessions created it should work now pkts_out2in = self.create_stream_out(self.pg1) - capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0, - len(pkts_out2in)) + capture = self.send_and_expect( + self.pg1, pkts_out2in, self.pg0, len(pkts_out2in) + ) self.verify_capture_in(capture, self.pg0) # Create an ACL blocking everything @@ -3147,8 +3738,9 @@ class TestNAT44EDMW(TestNAT44ED): in2out_acl.add_vpp_config() # apply as input acl on interface and confirm it blocks everything - acl_if = VppAclInterface(self, 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) @@ -3156,8 +3748,9 @@ class TestNAT44EDMW(TestNAT44ED): 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)) + capture = self.send_and_expect( + self.pg0, pkts_in2out, self.pg1, len(pkts_in2out) + ) self.verify_capture_out(capture, ignore_port=True) # send out2in again. ACL state exists so it should work now. @@ -3165,114 +3758,113 @@ class TestNAT44EDMW(TestNAT44ED): for p in pkts_out2in: if p.haslayer(TCP) and p[TCP].flags & 0x02: p[TCP].flags |= 0x10 - capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0, - len(pkts_out2in)) + capture = self.send_and_expect( + self.pg1, pkts_out2in, self.pg0, len(pkts_out2in) + ) self.verify_capture_in(capture, self.pg0) self.logger.info(self.vapi.cli("show trace")) def test_tcp_close(self): - """ NAT44ED Close TCP session from inside network - output feature """ - old_timeouts = self.vapi.nat_get_timeouts() + """NAT44ED Close TCP session from inside network - output feature""" + config = self.vapi.nat44_show_running_config() + old_timeouts = config.timeouts new_transitory = 2 self.vapi.nat_set_timeouts( udp=old_timeouts.udp, tcp_established=old_timeouts.tcp_established, icmp=old_timeouts.icmp, - tcp_transitory=new_transitory) + tcp_transitory=new_transitory, + ) self.vapi.nat44_forwarding_enable_disable(enable=1) self.nat_add_address(self.pg1.local_ip4) - twice_nat_addr = '10.0.1.3' - service_ip = '192.168.16.150' + twice_nat_addr = "10.0.1.3" + service_ip = "192.168.16.150" self.nat_add_address(twice_nat_addr, twice_nat=1) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1) + sw_if_index=self.pg0.sw_if_index, is_add=1 + ) 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_output_feature( - is_add=1, - sw_if_index=self.pg1.sw_if_index) - - flags = (self.config_flags.NAT_IS_OUT2IN_ONLY | - self.config_flags.NAT_IS_TWICE_NAT) - self.nat_add_static_mapping(self.pg0.remote_ip4, - service_ip, 80, 80, - proto=IP_PROTOS.tcp, - flags=flags) + sw_if_index=self.pg0.sw_if_index, flags=flags, is_add=1 + ) + self.vapi.nat44_ed_add_del_output_interface( + is_add=1, sw_if_index=self.pg1.sw_if_index + ) + + flags = ( + self.config_flags.NAT_IS_OUT2IN_ONLY | self.config_flags.NAT_IS_TWICE_NAT + ) + self.nat_add_static_mapping( + self.pg0.remote_ip4, service_ip, 80, 80, proto=IP_PROTOS.tcp, flags=flags + ) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) start_sessnum = len(sessions) # SYN packet out->in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="S")) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(1) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=service_ip) + / TCP(sport=33898, dport=80, flags="S") + ) + capture = self.send_and_expect(self.pg1, p, self.pg0, n_rx=1) p = capture[0] tcp_port = p[TCP].sport # SYN + ACK packet in->out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) / - TCP(sport=80, dport=tcp_port, flags="SA")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) + / TCP(sport=80, dport=tcp_port, flags="SA") + ) + self.send_and_expect(self.pg0, p, self.pg1, n_rx=1) # ACK packet out->in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="A")) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=service_ip) + / TCP(sport=33898, dport=80, flags="A") + ) + self.send_and_expect(self.pg1, p, self.pg0, n_rx=1) # FIN packet in -> out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) / - TCP(sport=80, dport=tcp_port, flags="FA", seq=100, ack=300)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) + / TCP(sport=80, dport=tcp_port, flags="FA", seq=100, ack=300) + ) + self.send_and_expect(self.pg0, p, self.pg1, n_rx=1) # FIN+ACK packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=service_ip) + / TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101) + ) + self.send_and_expect(self.pg1, p, self.pg0, n_rx=1) # ACK packet in -> out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) / - TCP(sport=80, dport=tcp_port, flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - # session now in transitory timeout - # try SYN packet out->in - should be dropped - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="S")) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) + / TCP(sport=80, dport=tcp_port, flags="A", seq=101, ack=301) + ) + self.send_and_expect(self.pg0, p, self.pg1, n_rx=1) + + # session now in transitory timeout, but traffic still flows + # try FIN packet out->in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=service_ip) + / TCP(sport=33898, dport=80, flags="F") + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - self.sleep(new_transitory, "wait for transitory timeout") - self.pg0.assert_nothing_captured(0) + self.virtual_sleep(new_transitory, "wait for transitory timeout") + self.pg0.get_capture(1) # session should still exist sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) @@ -3280,19 +3872,17 @@ class TestNAT44EDMW(TestNAT44ED): # send FIN+ACK packet out -> in - will cause session to be wiped # but won't create a new session - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=service_ip) + / TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101) + ) + self.send_and_assert_no_replies(self.pg1, p) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) self.assertEqual(len(sessions) - start_sessnum, 0) - self.pg0.assert_nothing_captured(0) def test_tcp_session_close_in(self): - """ NAT44ED Close TCP session from inside network """ + """NAT44ED Close TCP session from inside network""" in_port = self.tcp_port_in out_port = 10505 @@ -3301,110 +3891,92 @@ class TestNAT44EDMW(TestNAT44ED): self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) - self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr, - in_port, out_port, proto=IP_PROTOS.tcp, - flags=self.config_flags.NAT_IS_TWICE_NAT) + self.nat_add_static_mapping( + self.pg0.remote_ip4, + self.nat_addr, + in_port, + out_port, + proto=IP_PROTOS.tcp, + flags=self.config_flags.NAT_IS_TWICE_NAT, + ) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) session_n = len(sessions) - self.vapi.nat_set_timeouts(udp=300, tcp_established=7440, - tcp_transitory=2, icmp=5) + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=2, icmp=5 + ) self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port) # FIN packet in -> out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=in_port, dport=ext_port, - flags="FA", seq=100, ack=300)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - + 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=in_port, dport=ext_port, flags="FA", seq=100, ack=300) + ) + self.send_and_expect(self.pg0, p, self.pg1) pkts = [] # ACK packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, - flags="A", seq=300, ack=101)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="A", seq=300, ack=101) + ) pkts.append(p) # FIN packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, - flags="FA", seq=300, ack=101)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=101) + ) pkts.append(p) - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(2) + self.send_and_expect(self.pg1, pkts, self.pg0) # ACK packet in -> out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) + 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=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_expect(self.pg0, p, self.pg1) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) self.assertEqual(len(sessions) - session_n, 1) - out2in_drops = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - in2out_drops = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - - # 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=ext_port, dport=out_port, - 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=in_port, dport=ext_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() + # retransmit FIN packet out -> in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=101) + ) - stats = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - self.assertEqual(stats - out2in_drops, 1) - stats = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - self.assertEqual(stats - in2out_drops, 1) + self.send_and_expect(self.pg1, p, self.pg0) - 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=in_port, dport=ext_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() + # retransmit ACK packet in -> out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + self.virtual_sleep(3) + # retransmit 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=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_assert_no_replies(self.pg0, p) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) self.assertEqual(len(sessions) - session_n, 0) def test_tcp_session_close_out(self): - """ NAT44ED Close TCP session from outside network """ + """NAT44ED Close TCP session from outside network""" in_port = self.tcp_port_in out_port = 10505 @@ -3413,33 +3985,41 @@ class TestNAT44EDMW(TestNAT44ED): self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) - self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr, - in_port, out_port, proto=IP_PROTOS.tcp, - flags=self.config_flags.NAT_IS_TWICE_NAT) + self.nat_add_static_mapping( + self.pg0.remote_ip4, + self.nat_addr, + in_port, + out_port, + proto=IP_PROTOS.tcp, + flags=self.config_flags.NAT_IS_TWICE_NAT, + ) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) session_n = len(sessions) - self.vapi.nat_set_timeouts(udp=300, tcp_established=7440, - tcp_transitory=2, icmp=5) + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=2, icmp=5 + ) _ = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port) # FIN packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, - flags="FA", seq=100, ack=300)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="FA", seq=100, ack=300) + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg0.get_capture(1) # FIN+ACK packet in -> out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=in_port, dport=ext_port, - flags="FA", seq=300, ack=101)) + 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=in_port, dport=ext_port, flags="FA", seq=300, ack=101) + ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) @@ -3447,10 +4027,11 @@ class TestNAT44EDMW(TestNAT44ED): self.pg1.get_capture(1) # ACK packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, - flags="A", seq=101, ack=301)) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="A", seq=101, ack=301) + ) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -3459,54 +4040,35 @@ class TestNAT44EDMW(TestNAT44ED): sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) self.assertEqual(len(sessions) - session_n, 1) - out2in_drops = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - in2out_drops = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - - # 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=ext_port, dport=out_port, - 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=in_port, dport=ext_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.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - self.assertEqual(stats - out2in_drops, 1) - stats = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - self.assertEqual(stats - in2out_drops, 1) + # retransmit FIN packet out -> in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=101) + ) + self.send_and_expect(self.pg1, p, self.pg0) - 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=in_port, dport=ext_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() + # retransmit ACK packet in -> out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + self.virtual_sleep(3) + # retransmit 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=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_assert_no_replies(self.pg0, p) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) self.assertEqual(len(sessions) - session_n, 0) def test_tcp_session_close_simultaneous(self): - """ NAT44ED Close TCP session from inside network """ + """Simultaneous TCP close from both sides""" in_port = self.tcp_port_in ext_port = 10505 @@ -3514,109 +4076,297 @@ class TestNAT44EDMW(TestNAT44ED): self.nat_add_address(self.nat_addr) self.nat_add_inside_interface(self.pg0) self.nat_add_outside_interface(self.pg1) - self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr, - in_port, ext_port, proto=IP_PROTOS.tcp, - flags=self.config_flags.NAT_IS_TWICE_NAT) + self.nat_add_static_mapping( + self.pg0.remote_ip4, + self.nat_addr, + in_port, + ext_port, + proto=IP_PROTOS.tcp, + flags=self.config_flags.NAT_IS_TWICE_NAT, + ) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) session_n = len(sessions) - self.vapi.nat_set_timeouts(udp=300, tcp_established=7440, - tcp_transitory=2, icmp=5) + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=2, icmp=5 + ) out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port) # FIN packet in -> out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=in_port, dport=ext_port, - flags="FA", seq=100, ack=300)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) + 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=in_port, dport=ext_port, flags="FA", seq=100, ack=300) + ) + self.send_and_expect(self.pg0, p, self.pg1) # FIN packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, - flags="FA", seq=300, ack=100)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=100) + ) + self.send_and_expect(self.pg1, p, self.pg0) # ACK packet in -> out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) + 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=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_expect(self.pg0, p, self.pg1) # ACK packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, - flags="A", seq=301, ack=101)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="A", seq=301, ack=101) + ) + self.send_and_expect(self.pg1, p, self.pg0) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) self.assertEqual(len(sessions) - session_n, 1) - out2in_drops = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - in2out_drops = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') + # retransmit FIN packet out -> in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=101) + ) + self.send_and_expect(self.pg1, p, self.pg0) - # 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=ext_port, dport=out_port, - flags="FA", seq=300, ack=101)) + # retransmit ACK packet in -> out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + self.virtual_sleep(3) + # retransmit 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=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.pg_send(self.pg0, p) + self.send_and_assert_no_replies(self.pg0, p) + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) + self.assertEqual(len(sessions) - session_n, 0) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.assert_nothing_captured() + def test_tcp_session_half_reopen_inside(self): + """TCP session in FIN/FIN state not reopened by in2out SYN only""" + in_port = self.tcp_port_in + ext_port = 10505 - # 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=in_port, dport=ext_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() + self.nat_add_address(self.nat_addr) + self.nat_add_inside_interface(self.pg0) + self.nat_add_outside_interface(self.pg1) + self.nat_add_static_mapping( + self.pg0.remote_ip4, + self.nat_addr, + in_port, + ext_port, + proto=IP_PROTOS.tcp, + flags=self.config_flags.NAT_IS_TWICE_NAT, + ) - stats = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - self.assertEqual(stats - out2in_drops, 1) - stats = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - self.assertEqual(stats - in2out_drops, 1) + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) + session_n = len(sessions) + + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=2, icmp=5 + ) + + out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port) + + # FIN packet in -> out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="FA", seq=100, ack=300) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # FIN packet out -> in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=100) + ) + self.send_and_expect(self.pg1, p, self.pg0) - 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=in_port, dport=ext_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) - session_n, 1) + + # send SYN packet in -> out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="S", seq=101, ack=301) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + self.virtual_sleep(3) + # send ACK packet in -> out - session should 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=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_assert_no_replies(self.pg0, p, self.pg1) + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) + self.assertEqual(len(sessions) - session_n, 0) + + def test_tcp_session_half_reopen_outside(self): + """TCP session in FIN/FIN state not reopened by out2in SYN only""" + in_port = self.tcp_port_in + ext_port = 10505 + + self.nat_add_address(self.nat_addr) + self.nat_add_inside_interface(self.pg0) + self.nat_add_outside_interface(self.pg1) + self.nat_add_static_mapping( + self.pg0.remote_ip4, + self.nat_addr, + in_port, + ext_port, + proto=IP_PROTOS.tcp, + flags=self.config_flags.NAT_IS_TWICE_NAT, + ) + + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) + session_n = len(sessions) + + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=2, icmp=5 + ) + + out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port) + + # FIN packet in -> out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="FA", seq=100, ack=300) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # FIN packet out -> in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=100) + ) + self.send_and_expect(self.pg1, p, self.pg0) + + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) + self.assertEqual(len(sessions) - session_n, 1) + + # send SYN packet out -> in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="S", seq=300, ack=101) + ) + self.send_and_expect(self.pg1, p, self.pg0) + + self.virtual_sleep(3) + # send ACK packet in -> out - session should 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=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_assert_no_replies(self.pg0, p, self.pg1) sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) self.assertEqual(len(sessions) - session_n, 0) + def test_tcp_session_reopen(self): + """TCP session in FIN/FIN state reopened by SYN from both sides""" + in_port = self.tcp_port_in + ext_port = 10505 + + self.nat_add_address(self.nat_addr) + self.nat_add_inside_interface(self.pg0) + self.nat_add_outside_interface(self.pg1) + self.nat_add_static_mapping( + self.pg0.remote_ip4, + self.nat_addr, + in_port, + ext_port, + proto=IP_PROTOS.tcp, + flags=self.config_flags.NAT_IS_TWICE_NAT, + ) + + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) + session_n = len(sessions) + + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=2, icmp=5 + ) + + out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port) + + # FIN packet in -> out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="FA", seq=100, ack=300) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # FIN packet out -> in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=100) + ) + self.send_and_expect(self.pg1, p, self.pg0) + + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) + self.assertEqual(len(sessions) - session_n, 1) + + # send SYN packet out -> in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="S", seq=300, ack=101) + ) + self.send_and_expect(self.pg1, p, self.pg0) + + # send SYN packet in -> out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="SA", seq=101, ack=301) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # send ACK packet out -> in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="A", seq=300, ack=101) + ) + self.send_and_expect(self.pg1, p, self.pg0) + + self.virtual_sleep(3) + # send ACK packet in -> out - should be forwarded and session alive + 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=in_port, dport=ext_port, flags="A", seq=101, ack=301) + ) + self.send_and_expect(self.pg0, p, self.pg1) + sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) + self.assertEqual(len(sessions) - session_n, 1) + def test_dynamic_vrf(self): - """ NAT44ED dynamic translation test: different VRF""" + """NAT44ED dynamic translation test: different VRF""" vrf_id_in = 33 vrf_id_out = 34 @@ -3649,30 +4399,28 @@ class TestNAT44EDMW(TestNAT44ED): self.pg7.unconfig() self.pg8.unconfig() - self.vapi.ip_table_add_del(is_add=0, - table={'table_id': vrf_id_in}) - self.vapi.ip_table_add_del(is_add=0, - table={'table_id': vrf_id_out}) + self.vapi.ip_table_add_del(is_add=0, table={"table_id": vrf_id_in}) + self.vapi.ip_table_add_del(is_add=0, table={"table_id": vrf_id_out}) def test_dynamic_output_feature_vrf(self): - """ NAT44ED dynamic translation test: output-feature, VRF""" + """NAT44ED dynamic translation test: output-feature, VRF""" # other then default (0) new_vrf_id = 22 self.nat_add_address(self.nat_addr) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg8.sw_if_index, is_add=1) - + self.vapi.nat44_ed_add_del_output_interface( + sw_if_index=self.pg8.sw_if_index, is_add=1 + ) try: self.configure_ip4_interface(self.pg7, table_id=new_vrf_id) self.configure_ip4_interface(self.pg8, table_id=new_vrf_id) # in2out - tcpn = self.statistics['/nat44-ed/in2out/slowpath/tcp'] - udpn = self.statistics['/nat44-ed/in2out/slowpath/udp'] - icmpn = self.statistics['/nat44-ed/in2out/slowpath/icmp'] - drops = self.statistics['/nat44-ed/in2out/slowpath/drops'] + tcpn = self.statistics["/nat44-ed/in2out/slowpath/tcp"] + udpn = self.statistics["/nat44-ed/in2out/slowpath/udp"] + icmpn = self.statistics["/nat44-ed/in2out/slowpath/icmp"] + drops = self.statistics["/nat44-ed/in2out/slowpath/drops"] pkts = self.create_stream_in(self.pg7, self.pg8) self.pg7.add_stream(pkts) @@ -3682,20 +4430,20 @@ class TestNAT44EDMW(TestNAT44ED): self.verify_capture_out(capture, ignore_port=True) if_idx = self.pg8.sw_if_index - cnt = self.statistics['/nat44-ed/in2out/slowpath/tcp'] + cnt = self.statistics["/nat44-ed/in2out/slowpath/tcp"] self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2) - cnt = self.statistics['/nat44-ed/in2out/slowpath/udp'] + cnt = self.statistics["/nat44-ed/in2out/slowpath/udp"] self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1) - cnt = self.statistics['/nat44-ed/in2out/slowpath/icmp'] + cnt = self.statistics["/nat44-ed/in2out/slowpath/icmp"] self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1) - cnt = self.statistics['/nat44-ed/in2out/slowpath/drops'] + cnt = self.statistics["/nat44-ed/in2out/slowpath/drops"] self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0) # out2in - tcpn = self.statistics['/nat44-ed/out2in/fastpath/tcp'] - udpn = self.statistics['/nat44-ed/out2in/fastpath/udp'] - icmpn = self.statistics['/nat44-ed/out2in/fastpath/icmp'] - drops = self.statistics['/nat44-ed/out2in/fastpath/drops'] + tcpn = self.statistics["/nat44-ed/out2in/fastpath/tcp"] + udpn = self.statistics["/nat44-ed/out2in/fastpath/udp"] + icmpn = self.statistics["/nat44-ed/out2in/fastpath/icmp"] + drops = self.statistics["/nat44-ed/out2in/fastpath/drops"] pkts = self.create_stream_out(self.pg8) self.pg8.add_stream(pkts) @@ -3705,48 +4453,56 @@ class TestNAT44EDMW(TestNAT44ED): self.verify_capture_in(capture, self.pg7) if_idx = self.pg8.sw_if_index - cnt = self.statistics['/nat44-ed/out2in/fastpath/tcp'] + cnt = self.statistics["/nat44-ed/out2in/fastpath/tcp"] self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2) - cnt = self.statistics['/nat44-ed/out2in/fastpath/udp'] + cnt = self.statistics["/nat44-ed/out2in/fastpath/udp"] self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1) - cnt = self.statistics['/nat44-ed/out2in/fastpath/icmp'] + cnt = self.statistics["/nat44-ed/out2in/fastpath/icmp"] self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1) - cnt = self.statistics['/nat44-ed/out2in/fastpath/drops'] + cnt = self.statistics["/nat44-ed/out2in/fastpath/drops"] self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0) - sessions = self.statistics['/nat44-ed/total-sessions'] + sessions = self.statistics["/nat44-ed/total-sessions"] self.assertEqual(sessions[:, 0].sum(), 3) finally: self.pg7.unconfig() self.pg8.unconfig() - self.vapi.ip_table_add_del(is_add=0, - table={'table_id': new_vrf_id}) + self.vapi.ip_table_add_del(is_add=0, table={"table_id": new_vrf_id}) def test_next_src_nat(self): - """ NAT44ED On way back forward packet to nat44-in2out node. """ + """NAT44ED On way back forward packet to nat44-in2out node.""" - twice_nat_addr = '10.0.1.3' + twice_nat_addr = "10.0.1.3" external_port = 80 local_port = 8080 post_twice_nat_port = 0 self.vapi.nat44_forwarding_enable_disable(enable=1) self.nat_add_address(twice_nat_addr, twice_nat=1) - flags = (self.config_flags.NAT_IS_OUT2IN_ONLY | - self.config_flags.NAT_IS_SELF_TWICE_NAT) - self.nat_add_static_mapping(self.pg6.remote_ip4, self.pg1.remote_ip4, - local_port, external_port, - proto=IP_PROTOS.tcp, vrf_id=1, - flags=flags) + flags = ( + self.config_flags.NAT_IS_OUT2IN_ONLY + | self.config_flags.NAT_IS_SELF_TWICE_NAT + ) + self.nat_add_static_mapping( + self.pg6.remote_ip4, + self.pg1.remote_ip4, + local_port, + external_port, + proto=IP_PROTOS.tcp, + vrf_id=1, + flags=flags, + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg6.sw_if_index, - is_add=1) + sw_if_index=self.pg6.sw_if_index, is_add=1 + ) - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=12345, dport=external_port)) + p = ( + Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) + / IP(src=self.pg6.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=12345, dport=external_port) + ) self.pg6.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -3765,9 +4521,11 @@ class TestNAT44EDMW(TestNAT44ED): self.logger.error(ppp("Unexpected or invalid packet:", p)) raise - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=twice_nat_addr) / - TCP(sport=local_port, dport=post_twice_nat_port)) + p = ( + Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) + / IP(src=self.pg6.remote_ip4, dst=twice_nat_addr) + / TCP(sport=local_port, dport=post_twice_nat_port) + ) self.pg6.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -3786,7 +4544,7 @@ class TestNAT44EDMW(TestNAT44ED): raise def test_one_armed_nat44_static(self): - """ NAT44ED One armed NAT and 1:1 NAPT asymmetrical rule """ + """NAT44ED One armed NAT and 1:1 NAPT asymmetrical rule""" remote_host = self.pg4.remote_hosts[0] local_host = self.pg4.remote_hosts[1] @@ -3796,23 +4554,31 @@ class TestNAT44EDMW(TestNAT44ED): self.vapi.nat44_forwarding_enable_disable(enable=1) self.nat_add_address(self.nat_addr, twice_nat=1) - flags = (self.config_flags.NAT_IS_OUT2IN_ONLY | - self.config_flags.NAT_IS_TWICE_NAT) - self.nat_add_static_mapping(local_host.ip4, self.nat_addr, - local_port, external_port, - proto=IP_PROTOS.tcp, flags=flags) + flags = ( + self.config_flags.NAT_IS_OUT2IN_ONLY | self.config_flags.NAT_IS_TWICE_NAT + ) + self.nat_add_static_mapping( + local_host.ip4, + self.nat_addr, + local_port, + external_port, + proto=IP_PROTOS.tcp, + flags=flags, + ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg4.sw_if_index, - is_add=1) + sw_if_index=self.pg4.sw_if_index, is_add=1 + ) self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg4.sw_if_index, - flags=flags, is_add=1) + sw_if_index=self.pg4.sw_if_index, flags=flags, is_add=1 + ) # from client to service - p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) / - IP(src=remote_host.ip4, dst=self.nat_addr) / - TCP(sport=12345, dport=external_port)) + p = ( + Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) + / IP(src=remote_host.ip4, dst=self.nat_addr) + / TCP(sport=12345, dport=external_port) + ) self.pg4.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -3832,9 +4598,11 @@ class TestNAT44EDMW(TestNAT44ED): raise # from service back to client - p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) / - IP(src=local_host.ip4, dst=self.nat_addr) / - TCP(sport=local_port, dport=eh_port_in)) + p = ( + Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) + / IP(src=local_host.ip4, dst=self.nat_addr) + / TCP(sport=local_port, dport=eh_port_in) + ) self.pg4.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -3853,7 +4621,7 @@ class TestNAT44EDMW(TestNAT44ED): raise def test_icmp_error_fwd_outbound(self): - """ NAT44ED ICMP error outbound with forwarding enabled """ + """NAT44ED ICMP error outbound with forwarding enabled""" # Ensure that an outbound ICMP error message is properly associated # with the inbound forward bypass session it is related to. @@ -3865,9 +4633,12 @@ class TestNAT44EDMW(TestNAT44ED): # enable forwarding and initiate connection out2in self.vapi.nat44_forwarding_enable_disable(enable=1) - p1 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) / - UDP(sport=21, dport=20) / payload) + p1 = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) + / UDP(sport=21, dport=20) + / payload + ) self.pg1.add_stream(p1) self.pg_enable_capture(self.pg_interfaces) @@ -3881,10 +4652,12 @@ class TestNAT44EDMW(TestNAT44ED): # session dumps for a user will only look on the worker that the # user is supposed to be mapped to in2out. The forward bypass session # is not necessarily created on that worker. - p2 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - ICMP(type='dest-unreach', code='port-unreachable') / - capture[IP:]) + p2 = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / ICMP(type="dest-unreach", code="port-unreachable") + / capture[IP:] + ) self.pg0.add_stream(p2) self.pg_enable_capture(self.pg_interfaces) @@ -3897,6 +4670,227 @@ class TestNAT44EDMW(TestNAT44ED): self.logger.info(ppp("p2 packet:", p2)) self.logger.info(ppp("capture packet:", capture)) + def test_tcp_session_open_retransmit1(self): + """NAT44ED Open TCP session with SYN,ACK retransmit 1 + + The client does not receive the [SYN,ACK] or the + ACK from the client is lost. Therefore, the [SYN, ACK] + is retransmitted by the server. + """ + + in_port = self.tcp_port_in + ext_port = self.tcp_external_port + payload = "H" * 10 + + self.nat_add_address(self.nat_addr) + self.nat_add_inside_interface(self.pg0) + self.nat_add_outside_interface(self.pg1) + + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=5, icmp=60 + ) + # SYN packet in->out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="S") + ) + p = self.send_and_expect(self.pg0, p, self.pg1)[0] + out_port = p[TCP].sport + + # SYN + ACK packet out->in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="SA") + ) + self.send_and_expect(self.pg1, p, self.pg0) + + # ACK in->out does not arrive + + # resent SYN + ACK packet out->in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="SA") + ) + self.send_and_expect(self.pg1, p, self.pg0) + + # ACK packet in->out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="A") + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # Verify that the data can be transmitted after the transitory time + self.virtual_sleep(6) + + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="PA") + / Raw(payload) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + def test_tcp_session_open_retransmit2(self): + """NAT44ED Open TCP session with SYN,ACK retransmit 2 + + The ACK is lost to the server after the TCP session is opened. + Data is sent by the client, then the [SYN,ACK] is + retransmitted by the server. + """ + + in_port = self.tcp_port_in + ext_port = self.tcp_external_port + payload = "H" * 10 + + self.nat_add_address(self.nat_addr) + self.nat_add_inside_interface(self.pg0) + self.nat_add_outside_interface(self.pg1) + + self.vapi.nat_set_timeouts( + udp=300, tcp_established=7440, tcp_transitory=5, icmp=60 + ) + # SYN packet in->out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="S") + ) + p = self.send_and_expect(self.pg0, p, self.pg1)[0] + out_port = p[TCP].sport + + # SYN + ACK packet out->in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="SA") + ) + self.send_and_expect(self.pg1, p, self.pg0) + + # ACK packet in->out -- not received by the server + 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=in_port, dport=ext_port, flags="A") + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # PUSH + ACK packet in->out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="PA") + / Raw(payload) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # resent SYN + ACK packet out->in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="SA") + ) + self.send_and_expect(self.pg1, p, self.pg0) + + # resent ACK packet in->out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="A") + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # resent PUSH + ACK packet in->out + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="PA") + / Raw(payload) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + # ACK packet out->in + p = ( + Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) + / TCP(sport=ext_port, dport=out_port, flags="A") + ) + self.send_and_expect(self.pg1, p, self.pg0) + + # Verify that the data can be transmitted after the transitory time + self.virtual_sleep(6) + + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / TCP(sport=in_port, dport=ext_port, flags="PA") + / Raw(payload) + ) + self.send_and_expect(self.pg0, p, self.pg1) + + def test_dynamic_ports_exhausted(self): + """NAT44ED dynamic translation test: address ports exhaused""" + + sessions_per_batch = 128 + n_available_ports = 65536 - 1024 + n_sessions = n_available_ports + 2 * sessions_per_batch + + # set high enough session limit for ports to be exhausted + self.plugin_disable() + self.plugin_enable(max_sessions=n_sessions) + + self.nat_add_inside_interface(self.pg0) + self.nat_add_outside_interface(self.pg1) + + # set timeouts to high for sessions to reallistically expire + config = self.vapi.nat44_show_running_config() + old_timeouts = config.timeouts + self.vapi.nat_set_timeouts( + udp=21600, + tcp_established=old_timeouts.tcp_established, + tcp_transitory=old_timeouts.tcp_transitory, + icmp=old_timeouts.icmp, + ) + + # in2out after NAT addresses added + self.nat_add_address(self.nat_addr) + + for i in range(n_sessions // sessions_per_batch): + pkts = self.create_udp_stream( + self.pg0, + self.pg1, + sessions_per_batch, + base_port=i * sessions_per_batch + 100, + ) + + self.pg0.add_stream(pkts) + self.pg_start() + + err = self.statistics.get_err_counter( + "/err/nat44-ed-in2out-slowpath/out of ports" + ) + if err > sessions_per_batch: + break + + # Check for ports to be used no more than once + ports = set() + sessions = self.vapi.cli("show nat44 sessions") + rx = re.compile( + f" *o2i flow: match: saddr {self.pg1.remote_ip4} sport [0-9]+ daddr {self.nat_addr} dport ([0-9]+) proto UDP.*" + ) + for line in sessions.splitlines(): + m = rx.match(line) + if m: + port = int(m.groups()[0]) + self.assertNotIn(port, ports) + ports.add(port) + + self.assertGreaterEqual(err, sessions_per_batch) + -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner) |