tests: replace pycodestyle with black
Drop pycodestyle for code style checking in favor of black. Black is
much faster, stable PEP8 compliant code style checker offering also
automatic formatting. It aims to be very stable and produce smallest
diffs. It's used by many small and big projects.
Running checkstyle with black takes a few seconds with a terse output.
Thus, test-checkstyle-diff is no longer necessary.
Expand scope of checkstyle to all python files in the repo, replacing
test-checkstyle with checkstyle-python.
Also, fixstyle-python is now available for automatic style formatting.
Note: python virtualenv has been consolidated in test/Makefile,
test/requirements*.txt which will eventually be moved to a central
location. This is required to simply the automated generation of
docker executor images in the CI.
Type: improvement
Change-Id: I022a326603485f58585e879ac0f697fceefbc9c8
Signed-off-by: Klement Sekera <klement.sekera@gmail.com>
Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
diff --git a/test/test_nat44_ed.py b/test/test_nat44_ed.py
index 21eebb2..7745987 100644
--- a/test/test_nat44_ed.py
+++ b/test/test_nat44_ed.py
@@ -21,9 +21,9 @@
class TestNAT44ED(VppTestCase):
- """ NAT44ED Test Case """
+ """NAT44ED Test Case"""
- nat_addr = '10.0.10.3'
+ nat_addr = "10.0.10.3"
tcp_port_in = 6303
tcp_port_out = 6303
@@ -48,8 +48,7 @@
self.plugin_disable()
def plugin_enable(self):
- self.vapi.nat44_ed_plugin_enable_disable(
- sessions=self.max_sessions, enable=1)
+ self.vapi.nat44_ed_plugin_enable_disable(sessions=self.max_sessions, enable=1)
def plugin_disable(self):
self.vapi.nat44_ed_plugin_enable_disable(enable=0)
@@ -87,7 +86,7 @@
@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
@@ -105,32 +104,41 @@
@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)
+ 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):
+ 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
@@ -142,9 +150,11 @@
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):
@@ -159,14 +169,14 @@
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
@@ -190,58 +200,95 @@
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,
+ 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
}
- 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:
@@ -250,19 +297,14 @@
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)
@@ -272,8 +314,9 @@
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:
@@ -284,21 +327,22 @@
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)
@@ -325,17 +369,18 @@
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)
@@ -351,21 +396,20 @@
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:
@@ -375,20 +419,19 @@
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)
@@ -415,15 +458,14 @@
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)
@@ -431,8 +473,9 @@
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:
@@ -444,35 +487,30 @@
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):
@@ -487,8 +525,9 @@
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):
@@ -497,27 +536,32 @@
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:
@@ -530,21 +574,27 @@
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
@@ -554,19 +604,24 @@
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_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)
@@ -574,9 +629,9 @@
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()
@@ -585,33 +640,38 @@
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)
@@ -623,16 +683,14 @@
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)
@@ -642,20 +700,24 @@
# 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)
@@ -663,8 +725,9 @@
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)
@@ -676,17 +739,15 @@
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)
@@ -697,21 +758,25 @@
# 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)
@@ -721,9 +786,11 @@
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()
@@ -732,18 +799,22 @@
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()
@@ -751,9 +822,10 @@
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:
@@ -780,8 +852,7 @@
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)
@@ -793,27 +864,30 @@
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)
@@ -828,9 +902,11 @@
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()
@@ -865,9 +941,11 @@
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()
@@ -888,24 +966,25 @@
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, msgid, is_ip6=False):
- message = data.decode('utf-8')
+ message = data.decode("utf-8")
try:
message = SyslogMessage.parse(message)
except ParseError as e:
@@ -913,29 +992,28 @@
raise
else:
self.assertEqual(message.severity, SyslogSeverity.info)
- self.assertEqual(message.appname, 'NAT')
+ self.assertEqual(message.appname, "NAT")
self.assertEqual(message.msgid, msgid)
- sd_params = message.sd.get('nsess')
+ 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)
+ 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
@@ -944,25 +1022,31 @@
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,
- 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,
- ]
+ 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') /
- c[IP:]
- for c in capture]
+ 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
+ ]
capture = self.send_and_expect(self.pg1, p2, self.pg0)
@@ -971,20 +1055,21 @@
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
+ 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)
@@ -994,9 +1079,11 @@
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))
@@ -1019,7 +1106,7 @@
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)
@@ -1061,9 +1148,9 @@
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()
@@ -1075,8 +1162,7 @@
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
@@ -1107,14 +1193,14 @@
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)
@@ -1123,11 +1209,13 @@
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
@@ -1151,9 +1239,9 @@
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()
@@ -1165,24 +1253,27 @@
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)
@@ -1194,16 +1285,17 @@
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_ed_add_del_output_interface(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
+ 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)
@@ -1230,9 +1322,9 @@
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()
@@ -1244,40 +1336,37 @@
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_ed_add_del_output_interface(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg1.sw_if_index, is_add=1
+ )
# in2out
pkts = self.create_stream_in(self.pg0, self.pg1)
@@ -1323,24 +1412,31 @@
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()
@@ -1357,9 +1453,12 @@
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()
@@ -1375,9 +1474,11 @@
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()
@@ -1394,9 +1495,12 @@
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()
@@ -1412,9 +1516,11 @@
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()
@@ -1431,9 +1537,11 @@
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()
@@ -1450,21 +1558,17 @@
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(
@@ -1473,19 +1577,22 @@
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()
@@ -1507,9 +1614,11 @@
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()
@@ -1527,56 +1636,59 @@
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()
@@ -1598,9 +1710,11 @@
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()
@@ -1617,9 +1731,11 @@
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()
@@ -1637,9 +1753,11 @@
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()
@@ -1656,41 +1774,41 @@
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()
@@ -1699,22 +1817,26 @@
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)
@@ -1724,31 +1846,39 @@
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()
@@ -1764,9 +1894,11 @@
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()
@@ -1783,9 +1915,9 @@
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
@@ -1793,22 +1925,30 @@
self.nat_add_address(self.nat_addr)
flags = self.config_flags.NAT_IS_INSIDE
self.vapi.nat44_ed_add_del_output_interface(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
+ 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()
@@ -1824,9 +1964,11 @@
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()
@@ -1843,23 +1985,23 @@
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,
@@ -1868,13 +2010,15 @@
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()
@@ -1891,9 +2035,11 @@
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()
@@ -1910,35 +2056,43 @@
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()
@@ -1955,9 +2109,11 @@
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()
@@ -1974,9 +2130,9 @@
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
@@ -1984,24 +2140,30 @@
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,
@@ -2009,13 +2171,15 @@
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()
@@ -2032,9 +2196,11 @@
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()
@@ -2051,9 +2217,11 @@
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()
@@ -2070,9 +2238,11 @@
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()
@@ -2089,9 +2259,11 @@
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()
@@ -2108,9 +2280,11 @@
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()
@@ -2127,7 +2301,7 @@
raise
def test_outside_address_distribution(self):
- """ Outside address distribution based on source address """
+ """Outside address distribution based on source address"""
x = 100
nat_addresses = []
@@ -2142,7 +2316,10 @@
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)
@@ -2150,11 +2327,12 @@
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)
@@ -2172,20 +2350,23 @@
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)]
+ a = nat_addresses[(numeric - 1) % len(nat_addresses)]
self.assertEqual(
- a, p_recvd[IP].src,
+ 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))
+ % (p_sent[IP].src, p_recvd[IP].src, a),
+ )
class TestNAT44EDMW(TestNAT44ED):
- """ NAT44ED MW Test Case """
+ """NAT44ED MW Test Case"""
+
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
@@ -2196,27 +2377,33 @@
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):
@@ -2228,26 +2415,23 @@
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']
+ 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(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()
@@ -2267,19 +2451,25 @@
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):
@@ -2295,41 +2485,50 @@
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']
+ 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(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)
@@ -2340,7 +2539,7 @@
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)
@@ -2350,7 +2549,7 @@
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)
@@ -2361,7 +2560,7 @@
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()
@@ -2373,39 +2572,29 @@
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()
@@ -2417,39 +2606,29 @@
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
@@ -2462,26 +2641,49 @@
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
@@ -2520,27 +2722,29 @@
capture = outside.get_capture(len(stream))
def test_show_max_translations(self):
- """ NAT44ED API test - 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)
+ 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)
@@ -2551,9 +2755,11 @@
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)
@@ -2562,79 +2768,89 @@
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.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.virtual_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"))
+ 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)
# 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"))
+ 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_session_rst_established_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)
+ self.init_tcp_session(
+ self.pg0, self.pg1, self.tcp_port_in, self.tcp_external_port
+ )
# 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"))
+ 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)
- 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"))
+ 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)
# State is transitory, session should be closed after 6 seconds
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="P"))
+ 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 """
+ """NAT44ED dynamic translation test: out of ports"""
self.nat_add_inside_interface(self.pg0)
self.nat_add_outside_interface(self.pg1)
@@ -2643,60 +2859,71 @@
pkts = self.create_stream_in(self.pg0, self.pg1)
self.send_and_assert_no_replies(
- self.pg0, pkts, msg="i2o pkts",
- stats_diff=self.no_diff | {
+ self.pg0,
+ pkts,
+ msg="i2o pkts",
+ stats_diff=self.no_diff
+ | {
"err": {
- '/err/nat44-ed-in2out-slowpath/out of ports': len(pkts),
+ "/err/nat44-ed-in2out-slowpath/out of ports": len(pkts),
},
self.pg0.sw_if_index: {
- '/nat44-ed/in2out/slowpath/drops': len(pkts),
+ "/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)))
+ tcpn, udpn, icmpn = (
+ sum(x) for x in zip(*((TCP in p, UDP in p, ICMP in p) for p in pkts))
+ )
self.send_and_expect(
- self.pg0, pkts, self.pg1, msg="i2o pkts",
- stats_diff=self.no_diff | {
+ self.pg0,
+ pkts,
+ self.pg1,
+ msg="i2o pkts",
+ stats_diff=self.no_diff
+ | {
"err": {
- '/err/nat44-ed-in2out-slowpath/out of ports': 0,
+ "/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,
+ "/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()
@@ -2712,11 +2939,13 @@
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()
@@ -2732,7 +2961,7 @@
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
@@ -2747,19 +2976,23 @@
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()
@@ -2775,11 +3008,13 @@
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()
@@ -2795,8 +3030,8 @@
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
@@ -2804,22 +3039,33 @@
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_ed_add_del_output_interface(
- 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=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()
@@ -2836,9 +3082,11 @@
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()
@@ -2877,26 +3125,34 @@
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_ed_add_del_output_interface(
- sw_if_index=self.pg1.sw_if_index, is_add=1)
+ 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()
@@ -2914,9 +3170,11 @@
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()
@@ -2935,17 +3193,15 @@
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
@@ -2954,39 +3210,39 @@
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)
@@ -3000,10 +3256,10 @@
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
@@ -3012,16 +3268,19 @@
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)
@@ -3040,10 +3299,10 @@
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
@@ -3052,7 +3311,8 @@
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
@@ -3075,39 +3335,39 @@
# 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)
+ """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, 'SADD')
+ 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, 'SDEL')
+ 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)
+ """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)
@@ -3115,55 +3375,62 @@
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))
+ 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')
+ 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'))
+ 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"))
+ 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"))
+ 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"))
+ 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)
+ 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')
+ 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()
@@ -3173,8 +3440,7 @@
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
@@ -3183,11 +3449,12 @@
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_ed_add_del_output_interface(
- sw_if_index=self.pg1.sw_if_index, is_add=1)
+ sw_if_index=self.pg1.sw_if_index, is_add=1
+ )
# First ensure that the NAT is working sans ACL
@@ -3197,14 +3464,16 @@
# 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
@@ -3218,8 +3487,9 @@
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)
@@ -3227,8 +3497,9 @@
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.
@@ -3236,13 +3507,14 @@
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 """
+ """NAT44ED Close TCP session from inside network - output feature"""
config = self.vapi.nat44_show_running_config()
old_timeouts = config.timeouts
new_transitory = 2
@@ -3250,77 +3522,92 @@
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)
+ 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)
+ 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)
+ 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"))
+ 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"))
+ 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"))
+ 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))
+ 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))
+ 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))
+ 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"))
+ 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()
@@ -3334,15 +3621,17 @@
# 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))
+ 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)
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
@@ -3351,79 +3640,92 @@
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))
+ 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.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))
+ 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)
# 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))
+ 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)
# 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))
+ 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))
+ 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
@@ -3432,33 +3734,41 @@
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)
@@ -3466,10 +3776,11 @@
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()
@@ -3479,31 +3790,34 @@
self.assertEqual(len(sessions) - session_n, 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))
+ 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)
# 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))
+ 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))
+ 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):
- """ Simultaneous TCP close from both sides """
+ """Simultaneous TCP close from both sides"""
in_port = self.tcp_port_in
ext_port = 10505
@@ -3511,252 +3825,297 @@
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))
+ 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))
+ 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))
+ 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))
+ 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)
# 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))
+ 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)
# 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))
+ 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))
+ 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)
def test_tcp_session_half_reopen_inside(self):
- """ TCP session in FIN/FIN state not reopened by in2out SYN only """
+ """TCP session in FIN/FIN state not reopened by in2out 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)
+ 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))
+ 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))
+ 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 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))
+ 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))
+ 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 """
+ """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)
+ 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))
+ 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))
+ 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))
+ 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))
+ 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 """
+ """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)
+ 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))
+ 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))
+ 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))
+ 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))
+ 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))
+ 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))
+ 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
@@ -3789,30 +4148,28 @@
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_ed_add_del_output_interface(
- sw_if_index=self.pg8.sw_if_index,
- is_add=1)
+ 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)
@@ -3822,20 +4179,20 @@
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)
@@ -3845,48 +4202,56 @@
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()
@@ -3905,9 +4270,11 @@
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()
@@ -3926,7 +4293,7 @@
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]
@@ -3936,23 +4303,31 @@
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()
@@ -3972,9 +4347,11 @@
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()
@@ -3993,7 +4370,7 @@
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.
@@ -4005,9 +4382,12 @@
# 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)
@@ -4021,10 +4401,12 @@
# 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)
@@ -4038,11 +4420,11 @@
self.logger.info(ppp("capture packet:", capture))
def test_tcp_session_open_retransmit1(self):
- """ NAT44ED Open TCP session with SYN,ACK retransmit 1
+ """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.
+ 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
@@ -4053,50 +4435,61 @@
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
+ )
# 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 = (
+ 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"))
+ 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"))
+ 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"))
+ 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))
+ 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
+ """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.
+ 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
@@ -4107,68 +4500,87 @@
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
+ )
# 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 = (
+ 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"))
+ 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"))
+ 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))
+ 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"))
+ 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"))
+ 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))
+ 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"))
+ 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))
+ 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)
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)