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
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)
@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
@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
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):
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
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:
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)
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:
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)
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)
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:
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)
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)
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:
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):
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):
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:
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
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)
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()
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)
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)
# 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)
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)
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)
# 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)
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()
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()
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:
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)
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)
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()
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()
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:
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
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)
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)
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))
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)
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()
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
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)
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
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()
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)
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)
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()
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)
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()
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()
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()
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()
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()
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()
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(
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()
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()
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()
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()
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()
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()
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()
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)
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()
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()
raise
def test_multiple_vrf_2(self):
- """ Multiple VRF - dynamic NAT from VRF1 to VRF0 (output-feature) """
+ """Multiple VRF - dynamic NAT from VRF1 to VRF0 (output-feature)"""
- external_addr = '1.2.3.4'
+ external_addr = "1.2.3.4"
external_port = 80
local_port = 8080
port = 0
self.nat_add_address(self.nat_addr)
flags = self.config_flags.NAT_IS_INSIDE
self.vapi.nat44_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()
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()
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,
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()
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()
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()
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()
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
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,
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()
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()
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()
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()
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()
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()
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 = []
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)
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)
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
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):
capture = self.pg1.get_capture(pkt_count * 3, timeout=5)
if_idx = self.pg0.sw_if_index
- tc2 = self.statistics['/nat44-ed/in2out/slowpath/tcp']
- uc2 = self.statistics['/nat44-ed/in2out/slowpath/udp']
- ic2 = self.statistics['/nat44-ed/in2out/slowpath/icmp']
- dc2 = self.statistics['/nat44-ed/in2out/slowpath/drops']
-
- self.assertEqual(
- tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count)
- self.assertEqual(
- uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count)
- self.assertEqual(
- ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count)
+ tc2 = self.statistics["/nat44-ed/in2out/slowpath/tcp"]
+ uc2 = self.statistics["/nat44-ed/in2out/slowpath/udp"]
+ ic2 = self.statistics["/nat44-ed/in2out/slowpath/icmp"]
+ dc2 = self.statistics["/nat44-ed/in2out/slowpath/drops"]
+
+ self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count)
+ self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count)
+ self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count)
self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0)
self.logger.info(self.vapi.cli("show trace"))
# out2in
- tc1 = self.statistics['/nat44-ed/out2in/fastpath/tcp']
- uc1 = self.statistics['/nat44-ed/out2in/fastpath/udp']
- ic1 = self.statistics['/nat44-ed/out2in/fastpath/icmp']
- dc1 = self.statistics['/nat44-ed/out2in/fastpath/drops']
+ tc1 = self.statistics["/nat44-ed/out2in/fastpath/tcp"]
+ uc1 = self.statistics["/nat44-ed/out2in/fastpath/udp"]
+ ic1 = self.statistics["/nat44-ed/out2in/fastpath/icmp"]
+ dc1 = self.statistics["/nat44-ed/out2in/fastpath/drops"]
recvd_tcp_ports = set()
recvd_udp_ports = set()
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):
self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
if packet.haslayer(TCP):
self.assert_in_range(
- packet[TCP].dport, tcp_port_offset,
- tcp_port_offset + pkt_count, "dst TCP port")
+ packet[TCP].dport,
+ tcp_port_offset,
+ tcp_port_offset + pkt_count,
+ "dst TCP port",
+ )
elif packet.haslayer(UDP):
self.assert_in_range(
- packet[UDP].dport, udp_port_offset,
- udp_port_offset + pkt_count, "dst UDP port")
+ packet[UDP].dport,
+ udp_port_offset,
+ udp_port_offset + pkt_count,
+ "dst UDP port",
+ )
else:
self.assert_in_range(
- packet[ICMP].id, icmp_id_offset,
- icmp_id_offset + pkt_count, "ICMP id")
+ packet[ICMP].id,
+ icmp_id_offset,
+ icmp_id_offset + pkt_count,
+ "ICMP id",
+ )
except:
- self.logger.error(ppp("Unexpected or invalid packet "
- "(inside network):", packet))
+ self.logger.error(
+ ppp("Unexpected or invalid packet (inside network):", packet)
+ )
raise
if_idx = self.pg1.sw_if_index
- tc2 = self.statistics['/nat44-ed/out2in/fastpath/tcp']
- uc2 = self.statistics['/nat44-ed/out2in/fastpath/udp']
- ic2 = self.statistics['/nat44-ed/out2in/fastpath/icmp']
- dc2 = self.statistics['/nat44-ed/out2in/fastpath/drops']
-
- self.assertEqual(
- tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count)
- self.assertEqual(
- uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count)
- self.assertEqual(
- ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count)
+ tc2 = self.statistics["/nat44-ed/out2in/fastpath/tcp"]
+ uc2 = self.statistics["/nat44-ed/out2in/fastpath/udp"]
+ ic2 = self.statistics["/nat44-ed/out2in/fastpath/icmp"]
+ dc2 = self.statistics["/nat44-ed/out2in/fastpath/drops"]
+
+ self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count)
+ self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count)
+ self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count)
self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0)
- sc = self.statistics['/nat44-ed/total-sessions']
- self.assertEqual(sc[:, 0].sum(), len(recvd_tcp_ports) +
- len(recvd_udp_ports) + len(recvd_icmp_ids))
+ sc = self.statistics["/nat44-ed/total-sessions"]
+ self.assertEqual(
+ sc[:, 0].sum(),
+ len(recvd_tcp_ports) + len(recvd_udp_ports) + len(recvd_icmp_ids),
+ )
def test_frag_in_order(self):
- """ NAT44ED translate fragments arriving in order """
+ """NAT44ED translate fragments arriving in order"""
self.nat_add_address(self.nat_addr)
self.nat_add_inside_interface(self.pg0)
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)
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)
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()
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()
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
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
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)
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)
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)
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()
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()
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
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()
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()
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
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()
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()
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()
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()
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
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)
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
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)
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
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
# 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)
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()
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
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
# 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
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)
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.
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
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)
-
- 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)
+ is_add=1, sw_if_index=self.pg1.sw_if_index
+ )
+
+ flags = (
+ self.config_flags.NAT_IS_OUT2IN_ONLY | self.config_flags.NAT_IS_TWICE_NAT
+ )
+ self.nat_add_static_mapping(
+ self.pg0.remote_ip4, service_ip, 80, 80, proto=IP_PROTOS.tcp, flags=flags
+ )
sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
start_sessnum = len(sessions)
# SYN packet out->in
- p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=service_ip) /
- TCP(sport=33898, dport=80, flags="S"))
+ 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()
# 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
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
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)
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()
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
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
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)
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)
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()
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()
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]
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()
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()
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.
# 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)
# 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)
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
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
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)
Code Review / vpp.git / blobdiff