import scapy.compat
from framework import VppTestCase, VppTestRunner
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
-from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
- IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
- PacketListField
+from scapy.all import (
+ bind_layers,
+ Packet,
+ ByteEnumField,
+ ShortField,
+ IPField,
+ IntField,
+ LongField,
+ XByteField,
+ FlagsField,
+ FieldLenField,
+ PacketListField,
+)
from scapy.data import IP_PROTOS
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
# NAT HA protocol event data
class Event(Packet):
name = "Event"
- fields_desc = [ByteEnumField("event_type", None,
- {1: "add", 2: "del", 3: "refresh"}),
- ByteEnumField("protocol", None,
- {0: "other", 1: "udp", 2: "tcp", 3: "icmp"}),
- ShortField("flags", 0),
- IPField("in_addr", None),
- IPField("out_addr", None),
- ShortField("in_port", None),
- ShortField("out_port", None),
- IPField("eh_addr", None),
- IPField("ehn_addr", None),
- ShortField("eh_port", None),
- ShortField("ehn_port", None),
- IntField("fib_index", None),
- IntField("total_pkts", 0),
- LongField("total_bytes", 0)]
+ fields_desc = [
+ ByteEnumField("event_type", None, {1: "add", 2: "del", 3: "refresh"}),
+ ByteEnumField("protocol", None, {0: "other", 1: "udp", 2: "tcp", 3: "icmp"}),
+ ShortField("flags", 0),
+ IPField("in_addr", None),
+ IPField("out_addr", None),
+ ShortField("in_port", None),
+ ShortField("out_port", None),
+ IPField("eh_addr", None),
+ IPField("ehn_addr", None),
+ ShortField("eh_port", None),
+ ShortField("ehn_port", None),
+ IntField("fib_index", None),
+ IntField("total_pkts", 0),
+ LongField("total_bytes", 0),
+ ]
def extract_padding(self, s):
return "", s
# NAT HA protocol header
class HANATStateSync(Packet):
name = "HA NAT state sync"
- fields_desc = [XByteField("version", 1),
- FlagsField("flags", 0, 8, ['ACK']),
- FieldLenField("count", None, count_of="events"),
- IntField("sequence_number", 1),
- IntField("thread_index", 0),
- PacketListField("events", [], Event,
- count_from=lambda pkt: pkt.count)]
+ fields_desc = [
+ XByteField("version", 1),
+ FlagsField("flags", 0, 8, ["ACK"]),
+ FieldLenField("count", None, count_of="events"),
+ IntField("sequence_number", 1),
+ IntField("thread_index", 0),
+ PacketListField("events", [], Event, count_from=lambda pkt: pkt.count),
+ ]
class MethodHolder(VppTestCase):
- """ NAT create capture and verify method holder """
+ """NAT create capture and verify method holder"""
@property
def config_flags(self):
def SYSLOG_SEVERITY(self):
return VppEnum.vl_api_syslog_severity_t
- def nat44_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 nat44_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,
+ ):
"""
Add/delete NAT44EI static mapping
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,
+ )
def nat44_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF):
"""
:param ip: IP address
:param is_add: 1 if add, 0 if delete (Default add)
"""
- self.vapi.nat44_ei_add_del_address_range(first_ip_address=ip,
- last_ip_address=ip,
- vrf_id=vrf_id,
- is_add=is_add)
+ self.vapi.nat44_ei_add_del_address_range(
+ first_ip_address=ip, last_ip_address=ip, vrf_id=vrf_id, is_add=is_add
+ )
def create_routes_and_neigbors(self):
- r1 = VppIpRoute(self, self.pg7.remote_ip4, 32,
- [VppRoutePath(self.pg7.remote_ip4,
- self.pg7.sw_if_index)])
- r2 = VppIpRoute(self, self.pg8.remote_ip4, 32,
- [VppRoutePath(self.pg8.remote_ip4,
- self.pg8.sw_if_index)])
+ r1 = VppIpRoute(
+ self,
+ self.pg7.remote_ip4,
+ 32,
+ [VppRoutePath(self.pg7.remote_ip4, self.pg7.sw_if_index)],
+ )
+ r2 = VppIpRoute(
+ self,
+ self.pg8.remote_ip4,
+ 32,
+ [VppRoutePath(self.pg8.remote_ip4, self.pg8.sw_if_index)],
+ )
r1.add_vpp_config()
r2.add_vpp_config()
- n1 = VppNeighbor(self,
- self.pg7.sw_if_index,
- self.pg7.remote_mac,
- self.pg7.remote_ip4,
- is_static=1)
- n2 = VppNeighbor(self,
- self.pg8.sw_if_index,
- self.pg8.remote_mac,
- self.pg8.remote_ip4,
- is_static=1)
+ n1 = VppNeighbor(
+ self,
+ self.pg7.sw_if_index,
+ self.pg7.remote_mac,
+ self.pg7.remote_ip4,
+ is_static=1,
+ )
+ n2 = VppNeighbor(
+ self,
+ self.pg8.sw_if_index,
+ self.pg8.remote_mac,
+ self.pg8.remote_ip4,
+ is_static=1,
+ )
n1.add_vpp_config()
n2.add_vpp_config()
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
pref_n[13] = ip4_n[1]
pref_n[14] = ip4_n[2]
pref_n[15] = ip4_n[3]
- packed_pref_n = b''.join([scapy.compat.chb(x) for x in pref_n])
+ packed_pref_n = b"".join([scapy.compat.chb(x) for x in pref_n])
return socket.inet_ntop(socket.AF_INET6, packed_pref_n)
- 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):
"""
Create packet stream for outside network
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
"""
pkts = []
# TCP
- p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
- IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
- TCP(dport=self.tcp_port_out, sport=20))
+ p = (
+ Ether(dst=out_if.local_mac, src=out_if.remote_mac)
+ / IPv6(src=src_ip, dst=dst_ip, hlim=hl)
+ / TCP(dport=self.tcp_port_out, sport=20)
+ )
pkts.append(p)
# UDP
- p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
- IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
- UDP(dport=self.udp_port_out, sport=20))
+ p = (
+ Ether(dst=out_if.local_mac, src=out_if.remote_mac)
+ / IPv6(src=src_ip, dst=dst_ip, hlim=hl)
+ / UDP(dport=self.udp_port_out, sport=20)
+ )
pkts.append(p)
# ICMP
- p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
- IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
- ICMPv6EchoReply(id=self.icmp_id_out))
+ p = (
+ Ether(dst=out_if.local_mac, src=out_if.remote_mac)
+ / IPv6(src=src_ip, dst=dst_ip, hlim=hl)
+ / ICMPv6EchoReply(id=self.icmp_id_out)
+ )
pkts.append(p)
return pkts
- def verify_capture_out(self, capture, nat_ip=None, same_port=False,
- dst_ip=None, is_ip6=False, ignore_port=False):
+ def verify_capture_out(
+ self,
+ capture,
+ nat_ip=None,
+ same_port=False,
+ dst_ip=None,
+ is_ip6=False,
+ ignore_port=False,
+ ):
"""
Verify captured packets on outside network
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[ICMP46].id, self.icmp_id_in)
+ self.assertEqual(packet[ICMP46].id, self.icmp_id_in)
else:
- self.assertNotEqual(
- packet[ICMP46].id, self.icmp_id_in)
+ self.assertNotEqual(packet[ICMP46].id, self.icmp_id_in)
self.icmp_id_out = packet[ICMP46].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_out_ip6(self, capture, nat_ip, same_port=False,
- dst_ip=None):
+ def verify_capture_out_ip6(self, capture, nat_ip, same_port=False, dst_ip=None):
"""
Verify captured packets on outside network
:param same_port: Source port number is not translated (Default False)
:param dst_ip: Destination IP address (Default do not verify)
"""
- return self.verify_capture_out(capture, nat_ip, same_port, dst_ip,
- True)
+ return self.verify_capture_out(capture, nat_ip, same_port, dst_ip, True)
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 verify_capture_no_translation(self, capture, ingress_if, egress_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 verify_capture_out_with_icmp_errors(self, capture, src_ip=None,
- icmp_type=11):
+ def verify_capture_out_with_icmp_errors(self, capture, src_ip=None, icmp_type=11):
"""
Verify captured packets with ICMP errors on outside network
self.assertTrue(icmp.haslayer(IPerror))
inner_ip = icmp[IPerror]
if inner_ip.haslayer(TCPerror):
- self.assertEqual(inner_ip[TCPerror].dport,
- self.tcp_port_out)
+ self.assertEqual(inner_ip[TCPerror].dport, self.tcp_port_out)
elif inner_ip.haslayer(UDPerror):
- self.assertEqual(inner_ip[UDPerror].dport,
- self.udp_port_out)
+ self.assertEqual(inner_ip[UDPerror].dport, self.udp_port_out)
else:
self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out)
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_with_icmp_errors(self, capture, in_if, icmp_type=11):
self.assertTrue(icmp.haslayer(IPerror))
inner_ip = icmp[IPerror]
if inner_ip.haslayer(TCPerror):
- self.assertEqual(inner_ip[TCPerror].sport,
- self.tcp_port_in)
+ self.assertEqual(inner_ip[TCPerror].sport, self.tcp_port_in)
elif inner_ip.haslayer(UDPerror):
- self.assertEqual(inner_ip[UDPerror].sport,
- self.udp_port_in)
+ self.assertEqual(inner_ip[UDPerror].sport, self.udp_port_in)
else:
self.assertEqual(inner_ip[ICMPerror].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_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
+ ):
"""
Create fragmented packet stream
:returns: Fragments
"""
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 = random.randint(0, 65535)
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
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 verify_ipfix_nat44_ses(self, data):
else:
nat44_ses_delete_num += 1
# sourceIPv4Address
- self.assertEqual(self.pg0.remote_ip4,
- str(ipaddress.IPv4Address(record[8])))
+ self.assertEqual(self.pg0.remote_ip4, str(ipaddress.IPv4Address(record[8])))
# postNATSourceIPv4Address
- self.assertEqual(socket.inet_pton(socket.AF_INET, self.nat_addr),
- record[225])
+ self.assertEqual(
+ socket.inet_pton(socket.AF_INET, self.nat_addr), record[225]
+ )
# ingressVRFID
self.assertEqual(struct.pack("!I", 0), record[234])
# protocolIdentifier/sourceTransportPort
# /postNAPTSourceTransportPort
if IP_PROTOS.icmp == scapy.compat.orb(record[4]):
self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
- self.assertEqual(struct.pack("!H", self.icmp_id_out),
- record[227])
+ self.assertEqual(struct.pack("!H", self.icmp_id_out), record[227])
elif IP_PROTOS.tcp == scapy.compat.orb(record[4]):
- self.assertEqual(struct.pack("!H", self.tcp_port_in),
- record[7])
- self.assertEqual(struct.pack("!H", self.tcp_port_out),
- record[227])
+ self.assertEqual(struct.pack("!H", self.tcp_port_in), record[7])
+ self.assertEqual(struct.pack("!H", self.tcp_port_out), record[227])
elif IP_PROTOS.udp == scapy.compat.orb(record[4]):
- self.assertEqual(struct.pack("!H", self.udp_port_in),
- record[7])
- self.assertEqual(struct.pack("!H", self.udp_port_out),
- record[227])
+ self.assertEqual(struct.pack("!H", self.udp_port_in), record[7])
+ self.assertEqual(struct.pack("!H", self.udp_port_out), record[227])
else:
self.fail(f"Invalid protocol {scapy.compat.orb(record[4])}")
self.assertEqual(3, nat44_ses_create_num)
self.assertEqual(struct.pack("!I", limit), record[471])
def verify_no_nat44_user(self):
- """ Verify that there is no NAT44EI user """
+ """Verify that there is no NAT44EI user"""
users = self.vapi.nat44_ei_user_dump()
self.assertEqual(len(users), 0)
- users = self.statistics['/nat44-ei/total-users']
+ users = self.statistics["/nat44-ei/total-users"]
self.assertEqual(users[0][0], 0)
- sessions = self.statistics['/nat44-ei/total-sessions']
+ sessions = self.statistics["/nat44-ei/total-sessions"]
self.assertEqual(sessions[0][0], 0)
def verify_syslog_apmap(self, data, is_add=True):
- 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.msgid, 'APMADD' if is_add else 'APMDEL')
- sd_params = message.sd.get('napmap')
+ self.assertEqual(message.appname, "NAT")
+ self.assertEqual(message.msgid, "APMADD" if is_add else "APMDEL")
+ sd_params = message.sd.get("napmap")
self.assertTrue(sd_params is not None)
- self.assertEqual(sd_params.get('IATYP'), 'IPv4')
- self.assertEqual(sd_params.get('ISADDR'), self.pg0.remote_ip4)
- 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.assertTrue(sd_params.get('SSUBIX') is not None)
- self.assertEqual(sd_params.get('SVLAN'), '0')
+ self.assertEqual(sd_params.get("IATYP"), "IPv4")
+ self.assertEqual(sd_params.get("ISADDR"), self.pg0.remote_ip4)
+ 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.assertTrue(sd_params.get("SSUBIX") is not None)
+ self.assertEqual(sd_params.get("SVLAN"), "0")
def verify_mss_value(self, pkt, mss):
if not pkt.haslayer(IP) or not pkt.haslayer(TCP):
raise TypeError("Not a TCP/IP packet")
for option in pkt[TCP].options:
- if option[0] == 'MSS':
+ if option[0] == "MSS":
self.assertEqual(option[1], mss)
self.assert_tcp_checksum_valid(pkt)
else:
raise Exception("Unsupported protocol")
- 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 = random.randint(1025, 65535)
# in2out
- pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4,
- self.port_in, 20, data, proto)
+ pkts = self.create_stream_frag(
+ self.pg0, self.pg1.remote_ip4, self.port_in, 20, data, proto
+ )
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
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 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)
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.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)
class TestNAT44EI(MethodHolder):
- """ NAT44EI Test Cases """
+ """NAT44EI Test Cases"""
max_translations = 10240
max_users = 10240
cls.udp_port_out = 6304
cls.icmp_id_in = 6305
cls.icmp_id_out = 6305
- cls.nat_addr = '10.0.0.3'
+ cls.nat_addr = "10.0.0.3"
cls.ipfix_src_port = 4739
cls.ipfix_domain_id = 1
cls.tcp_external_port = 80
cls.pg1.configure_ipv4_neighbors()
cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
- cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
- cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
+ cls.vapi.ip_table_add_del(is_add=1, table={"table_id": 10})
+ cls.vapi.ip_table_add_del(is_add=1, table={"table_id": 20})
cls.pg4._local_ip4 = "172.16.255.1"
cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
cls.pg9.generate_remote_hosts(2)
cls.pg9.config_ip4()
cls.vapi.sw_interface_add_del_address(
- sw_if_index=cls.pg9.sw_if_index,
- prefix="10.0.0.1/24")
+ sw_if_index=cls.pg9.sw_if_index, prefix="10.0.0.1/24"
+ )
cls.pg9.admin_up()
cls.pg9.resolve_arp()
def plugin_enable(self):
self.vapi.nat44_ei_plugin_enable_disable(
- sessions=self.max_translations,
- users=self.max_users, enable=1)
+ sessions=self.max_translations, users=self.max_users, enable=1
+ )
def setUp(self):
super(TestNAT44EI, self).setUp()
super(TestNAT44EI, self).tearDown()
if not self.vpp_dead:
self.vapi.nat44_ei_ipfix_enable_disable(
- domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port,
- enable=0)
+ domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port, enable=0
+ )
self.ipfix_src_port = 4739
self.ipfix_domain_id = 1
self.vapi.cli("clear logging")
def test_clear_sessions(self):
- """ NAT44EI session clearing test """
+ """NAT44EI session clearing test"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture)
- sessions = self.statistics['/nat44-ei/total-sessions']
+ sessions = self.statistics["/nat44-ei/total-sessions"]
self.assertGreater(sessions[:, 0].sum(), 0, "Session count invalid")
self.logger.info("sessions before clearing: %s" % sessions[0][0])
self.vapi.cli("clear nat44 ei sessions")
- sessions = self.statistics['/nat44-ei/total-sessions']
+ sessions = self.statistics["/nat44-ei/total-sessions"]
self.assertEqual(sessions[:, 0].sum(), 0, "Session count invalid")
self.logger.info("sessions after clearing: %s" % sessions[0][0])
def test_dynamic(self):
- """ NAT44EI dynamic translation test """
+ """NAT44EI dynamic translation test"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# in2out
- tcpn = self.statistics['/nat44-ei/in2out/slowpath/tcp']
- udpn = self.statistics['/nat44-ei/in2out/slowpath/udp']
- icmpn = self.statistics['/nat44-ei/in2out/slowpath/icmp']
- drops = self.statistics['/nat44-ei/in2out/slowpath/drops']
+ tcpn = self.statistics["/nat44-ei/in2out/slowpath/tcp"]
+ udpn = self.statistics["/nat44-ei/in2out/slowpath/udp"]
+ icmpn = self.statistics["/nat44-ei/in2out/slowpath/icmp"]
+ drops = self.statistics["/nat44-ei/in2out/slowpath/drops"]
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.verify_capture_out(capture)
if_idx = self.pg0.sw_if_index
- cnt = self.statistics['/nat44-ei/in2out/slowpath/tcp']
+ cnt = self.statistics["/nat44-ei/in2out/slowpath/tcp"]
self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2)
- cnt = self.statistics['/nat44-ei/in2out/slowpath/udp']
+ cnt = self.statistics["/nat44-ei/in2out/slowpath/udp"]
self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1)
- cnt = self.statistics['/nat44-ei/in2out/slowpath/icmp']
+ cnt = self.statistics["/nat44-ei/in2out/slowpath/icmp"]
self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1)
- cnt = self.statistics['/nat44-ei/in2out/slowpath/drops']
+ cnt = self.statistics["/nat44-ei/in2out/slowpath/drops"]
self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0)
# out2in
- tcpn = self.statistics['/nat44-ei/out2in/slowpath/tcp']
- udpn = self.statistics['/nat44-ei/out2in/slowpath/udp']
- icmpn = self.statistics['/nat44-ei/out2in/slowpath/icmp']
- drops = self.statistics['/nat44-ei/out2in/slowpath/drops']
+ tcpn = self.statistics["/nat44-ei/out2in/slowpath/tcp"]
+ udpn = self.statistics["/nat44-ei/out2in/slowpath/udp"]
+ icmpn = self.statistics["/nat44-ei/out2in/slowpath/icmp"]
+ drops = self.statistics["/nat44-ei/out2in/slowpath/drops"]
pkts = self.create_stream_out(self.pg1)
self.pg1.add_stream(pkts)
self.verify_capture_in(capture, self.pg0)
if_idx = self.pg1.sw_if_index
- cnt = self.statistics['/nat44-ei/out2in/slowpath/tcp']
+ cnt = self.statistics["/nat44-ei/out2in/slowpath/tcp"]
self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2)
- cnt = self.statistics['/nat44-ei/out2in/slowpath/udp']
+ cnt = self.statistics["/nat44-ei/out2in/slowpath/udp"]
self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1)
- cnt = self.statistics['/nat44-ei/out2in/slowpath/icmp']
+ cnt = self.statistics["/nat44-ei/out2in/slowpath/icmp"]
self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1)
- cnt = self.statistics['/nat44-ei/out2in/slowpath/drops']
+ cnt = self.statistics["/nat44-ei/out2in/slowpath/drops"]
self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0)
- users = self.statistics['/nat44-ei/total-users']
+ users = self.statistics["/nat44-ei/total-users"]
self.assertEqual(users[:, 0].sum(), 1)
- sessions = self.statistics['/nat44-ei/total-sessions']
+ sessions = self.statistics["/nat44-ei/total-sessions"]
self.assertEqual(sessions[:, 0].sum(), 3)
def test_dynamic_icmp_errors_in2out_ttl_1(self):
- """ NAT44EI handling of client packets with TTL=1 """
+ """NAT44EI handling of client packets with TTL=1"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# Client side - generate traffic
pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
self.verify_capture_in_with_icmp_errors(capture, self.pg0)
def test_dynamic_icmp_errors_out2in_ttl_1(self):
- """ NAT44EI handling of server packets with TTL=1 """
+ """NAT44EI handling of server packets with TTL=1"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# Client side - create sessions
pkts = self.create_stream_in(self.pg0, self.pg1)
capture = self.send_and_expect_some(self.pg1, pkts, self.pg1)
# Server side - verify ICMP type 11 packets
- self.verify_capture_out_with_icmp_errors(capture,
- src_ip=self.pg1.local_ip4)
+ self.verify_capture_out_with_icmp_errors(capture, src_ip=self.pg1.local_ip4)
def test_dynamic_icmp_errors_in2out_ttl_2(self):
- """ NAT44EI handling of error responses to client packets with TTL=2
- """
+ """NAT44EI handling of error responses to client packets with TTL=2"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# Client side - generate traffic
pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
# Server side - simulate ICMP type 11 response
capture = self.pg1.get_capture(len(pkts))
- pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
- ICMP(type=11) / packet[IP] for packet in capture]
+ pkts = [
+ Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
+ / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
+ / ICMP(type=11)
+ / packet[IP]
+ for packet in capture
+ ]
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.verify_capture_in_with_icmp_errors(capture, self.pg0)
def test_dynamic_icmp_errors_out2in_ttl_2(self):
- """ NAT44EI handling of error responses to server packets with TTL=2
- """
+ """NAT44EI handling of error responses to server packets with TTL=2"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# Client side - create sessions
pkts = self.create_stream_in(self.pg0, self.pg1)
# Client side - simulate ICMP type 11 response
capture = self.pg0.get_capture(len(pkts))
- pkts = [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) / packet[IP] for packet in capture]
+ pkts = [
+ 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)
+ / packet[IP]
+ for packet in capture
+ ]
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.verify_capture_out_with_icmp_errors(capture)
def test_ping_out_interface_from_outside(self):
- """ NAT44EI ping out interface from outside network """
+ """NAT44EI ping out interface from outside network"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- 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.pg1.local_ip4) /
- ICMP(id=self.icmp_id_out, type='echo-request'))
+ 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.pg1.local_ip4)
+ / ICMP(id=self.icmp_id_out, type="echo-request")
+ )
pkts = [p]
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.assertEqual(packet[ICMP].id, self.icmp_id_in)
self.assertEqual(packet[ICMP].type, 0) # echo reply
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 test_ping_internal_host_from_outside(self):
- """ NAT44EI ping internal host from outside network """
+ """NAT44EI ping internal host from outside network"""
self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# out2in
- pkt = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.nat_addr, ttl=64) /
- ICMP(id=self.icmp_id_out, type='echo-request'))
+ pkt = (
+ Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
+ / IP(src=self.pg1.remote_ip4, dst=self.nat_addr, ttl=64)
+ / ICMP(id=self.icmp_id_out, type="echo-request")
+ )
self.pg1.add_stream(pkt)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
# in2out
- pkt = (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=self.icmp_id_in, type='echo-reply'))
+ pkt = (
+ 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=self.icmp_id_in, type="echo-reply")
+ )
self.pg0.add_stream(pkt)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
def test_forwarding(self):
- """ NAT44EI forwarding test """
+ """NAT44EI forwarding test"""
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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.vapi.nat44_ei_forwarding_enable_disable(enable=1)
real_ip = self.pg0.remote_ip4
alias_ip = self.nat_addr
flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
self.vapi.nat44_ei_add_del_static_mapping(
- is_add=1, local_ip_address=real_ip,
+ is_add=1,
+ local_ip_address=real_ip,
external_ip_address=alias_ip,
external_sw_if_index=0xFFFFFFFF,
- flags=flags)
+ flags=flags,
+ )
try:
# 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
local_ip_address=real_ip,
external_ip_address=alias_ip,
external_sw_if_index=0xFFFFFFFF,
- flags=flags)
+ flags=flags,
+ )
def test_static_in(self):
- """ NAT44EI 1:1 NAT initialized from inside network """
+ """NAT44EI 1:1 NAT initialized from inside network"""
nat_ip = "10.0.0.10"
self.tcp_port_out = 6303
self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
sm = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(len(sm), 1)
- self.assertEqual(sm[0].tag, '')
+ self.assertEqual(sm[0].tag, "")
self.assertEqual(sm[0].protocol, 0)
self.assertEqual(sm[0].local_port, 0)
self.assertEqual(sm[0].external_port, 0)
self.verify_capture_in(capture, self.pg0)
def test_static_out(self):
- """ NAT44EI 1:1 NAT initialized from outside network """
+ """NAT44EI 1:1 NAT initialized from outside network"""
nat_ip = "10.0.0.20"
self.tcp_port_out = 6303
self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip, tag=tag)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
sm = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(len(sm), 1)
self.assertEqual(sm[0].tag, tag)
self.verify_capture_out(capture, nat_ip, True)
def test_static_with_port_in(self):
- """ NAT44EI 1:1 NAPT initialized from inside network """
+ """NAT44EI 1:1 NAPT initialized from inside network"""
self.tcp_port_out = 3606
self.udp_port_out = 3607
self.icmp_id_out = 3608
self.nat44_add_address(self.nat_addr)
- self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
- self.tcp_port_in, self.tcp_port_out,
- proto=IP_PROTOS.tcp)
- self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
- self.udp_port_in, self.udp_port_out,
- proto=IP_PROTOS.udp)
- self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
- self.icmp_id_in, self.icmp_id_out,
- proto=IP_PROTOS.icmp)
+ self.nat44_add_static_mapping(
+ self.pg0.remote_ip4,
+ self.nat_addr,
+ self.tcp_port_in,
+ self.tcp_port_out,
+ proto=IP_PROTOS.tcp,
+ )
+ self.nat44_add_static_mapping(
+ self.pg0.remote_ip4,
+ self.nat_addr,
+ self.udp_port_in,
+ self.udp_port_out,
+ proto=IP_PROTOS.udp,
+ )
+ self.nat44_add_static_mapping(
+ self.pg0.remote_ip4,
+ self.nat_addr,
+ self.icmp_id_in,
+ self.icmp_id_out,
+ proto=IP_PROTOS.icmp,
+ )
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# in2out
pkts = self.create_stream_in(self.pg0, self.pg1)
self.verify_capture_in(capture, self.pg0)
def test_static_with_port_out(self):
- """ NAT44EI 1:1 NAPT initialized from outside network """
+ """NAT44EI 1:1 NAPT initialized from outside network"""
self.tcp_port_out = 30606
self.udp_port_out = 30607
self.icmp_id_out = 30608
self.nat44_add_address(self.nat_addr)
- self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
- self.tcp_port_in, self.tcp_port_out,
- proto=IP_PROTOS.tcp)
- self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
- self.udp_port_in, self.udp_port_out,
- proto=IP_PROTOS.udp)
- self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
- self.icmp_id_in, self.icmp_id_out,
- proto=IP_PROTOS.icmp)
+ self.nat44_add_static_mapping(
+ self.pg0.remote_ip4,
+ self.nat_addr,
+ self.tcp_port_in,
+ self.tcp_port_out,
+ proto=IP_PROTOS.tcp,
+ )
+ self.nat44_add_static_mapping(
+ self.pg0.remote_ip4,
+ self.nat_addr,
+ self.udp_port_in,
+ self.udp_port_out,
+ proto=IP_PROTOS.udp,
+ )
+ self.nat44_add_static_mapping(
+ self.pg0.remote_ip4,
+ self.nat_addr,
+ self.icmp_id_in,
+ self.icmp_id_out,
+ proto=IP_PROTOS.icmp,
+ )
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# out2in
pkts = self.create_stream_out(self.pg1)
self.verify_capture_out(capture)
def test_static_vrf_aware(self):
- """ NAT44EI 1:1 NAT VRF awareness """
+ """NAT44EI 1:1 NAT VRF awareness"""
nat_ip1 = "10.0.0.30"
nat_ip2 = "10.0.0.40"
self.udp_port_out = 6304
self.icmp_id_out = 6305
- self.nat44_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
- vrf_id=10)
- self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
- vrf_id=10)
+ self.nat44_add_static_mapping(self.pg4.remote_ip4, nat_ip1, vrf_id=10)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip2, vrf_id=10)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg3.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg3.sw_if_index, is_add=1
+ )
self.vapi.nat44_ei_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_ei_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
+ )
# inside interface VRF match NAT44EI static mapping VRF
pkts = self.create_stream_in(self.pg4, self.pg3)
self.pg3.assert_nothing_captured()
def test_dynamic_to_static(self):
- """ NAT44EI Switch from dynamic translation to 1:1NAT """
+ """NAT44EI Switch from dynamic translation to 1:1NAT"""
nat_ip = "10.0.0.10"
self.tcp_port_out = 6303
self.udp_port_out = 6304
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# dynamic
pkts = self.create_stream_in(self.pg0, self.pg1)
self.verify_capture_out(capture, nat_ip, True)
def test_identity_nat(self):
- """ NAT44EI Identity NAT """
+ """NAT44EI Identity NAT"""
flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
self.vapi.nat44_ei_add_del_identity_mapping(
- ip_address=self.pg0.remote_ip4, sw_if_index=0xFFFFFFFF,
- flags=flags, is_add=1)
+ ip_address=self.pg0.remote_ip4,
+ sw_if_index=0xFFFFFFFF,
+ flags=flags,
+ is_add=1,
+ )
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
-
- 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=12345, dport=56789))
+ sw_if_index=self.pg1.sw_if_index, is_add=1
+ )
+
+ 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=12345, dport=56789)
+ )
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertEqual(len(sessions), 0)
flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
self.vapi.nat44_ei_add_del_identity_mapping(
- ip_address=self.pg0.remote_ip4, sw_if_index=0xFFFFFFFF,
- flags=flags, vrf_id=1, is_add=1)
+ ip_address=self.pg0.remote_ip4,
+ sw_if_index=0xFFFFFFFF,
+ flags=flags,
+ vrf_id=1,
+ is_add=1,
+ )
identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
self.assertEqual(len(identity_mappings), 2)
def test_multiple_inside_interfaces(self):
- """ NAT44EI multiple non-overlapping address space inside interfaces
- """
+ """NAT44EI multiple non-overlapping address space inside interfaces"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- flags=flags, is_add=1)
+ sw_if_index=self.pg1.sw_if_index, flags=flags, is_add=1
+ )
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg3.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg3.sw_if_index, is_add=1
+ )
# between two NAT44EI inside interfaces (no translation)
pkts = self.create_stream_in(self.pg0, self.pg1)
self.verify_capture_in(capture, self.pg1)
def test_inside_overlapping_interfaces(self):
- """ NAT44EI multiple inside interfaces with overlapping address space
- """
+ """NAT44EI multiple inside interfaces with overlapping address space"""
static_nat_ip = "10.0.0.10"
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg3.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg3.sw_if_index, is_add=1
+ )
self.vapi.nat44_ei_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
+ )
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg5.sw_if_index,
- flags=flags, is_add=1)
+ sw_if_index=self.pg5.sw_if_index, flags=flags, is_add=1
+ )
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg6.sw_if_index,
- flags=flags, is_add=1)
- self.nat44_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
- vrf_id=20)
+ sw_if_index=self.pg6.sw_if_index, flags=flags, is_add=1
+ )
+ self.nat44_add_static_mapping(self.pg6.remote_ip4, static_nat_ip, vrf_id=20)
# between NAT44EI inside interfaces with same VRF (no translation)
pkts = self.create_stream_in(self.pg4, self.pg5)
self.verify_capture_no_translation(capture, self.pg4, self.pg5)
# between NAT44EI inside interfaces with different VRF (hairpinning)
- p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
- IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
- TCP(sport=1234, dport=5678))
+ p = (
+ Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac)
+ / IP(src=self.pg4.remote_ip4, dst=static_nat_ip)
+ / TCP(sport=1234, dport=5678)
+ )
self.pg4.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
# pg5 session dump
addresses = self.vapi.nat44_ei_address_dump()
self.assertEqual(len(addresses), 1)
- sessions = self.vapi.nat44_ei_user_session_dump(
- self.pg5.remote_ip4, 10)
+ sessions = self.vapi.nat44_ei_user_session_dump(self.pg5.remote_ip4, 10)
self.assertEqual(len(sessions), 3)
for session in sessions:
- self.assertFalse(session.flags &
- self.config_flags.NAT44_EI_STATIC_MAPPING)
- self.assertEqual(str(session.inside_ip_address),
- self.pg5.remote_ip4)
- self.assertEqual(session.outside_ip_address,
- addresses[0].ip_address)
+ self.assertFalse(session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
+ self.assertEqual(str(session.inside_ip_address), self.pg5.remote_ip4)
+ self.assertEqual(session.outside_ip_address, addresses[0].ip_address)
self.assertEqual(sessions[0].protocol, IP_PROTOS.tcp)
self.assertEqual(sessions[1].protocol, IP_PROTOS.udp)
self.assertEqual(sessions[2].protocol, IP_PROTOS.icmp)
addresses = self.vapi.nat44_ei_address_dump()
self.assertEqual(len(addresses), 1)
for user in users:
- sessions = self.vapi.nat44_ei_user_session_dump(user.ip_address,
- user.vrf_id)
+ sessions = self.vapi.nat44_ei_user_session_dump(
+ user.ip_address, user.vrf_id
+ )
for session in sessions:
self.assertEqual(user.ip_address, session.inside_ip_address)
self.assertTrue(session.total_bytes > session.total_pkts > 0)
- self.assertTrue(session.protocol in
- [IP_PROTOS.tcp, IP_PROTOS.udp,
- IP_PROTOS.icmp])
+ self.assertTrue(
+ session.protocol in [IP_PROTOS.tcp, IP_PROTOS.udp, IP_PROTOS.icmp]
+ )
# pg4 session dump
- sessions = self.vapi.nat44_ei_user_session_dump(
- self.pg4.remote_ip4, 10)
+ sessions = self.vapi.nat44_ei_user_session_dump(self.pg4.remote_ip4, 10)
self.assertGreaterEqual(len(sessions), 4)
for session in sessions:
- self.assertFalse(
- session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
- self.assertEqual(str(session.inside_ip_address),
- self.pg4.remote_ip4)
- self.assertEqual(session.outside_ip_address,
- addresses[0].ip_address)
+ self.assertFalse(session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
+ self.assertEqual(str(session.inside_ip_address), self.pg4.remote_ip4)
+ self.assertEqual(session.outside_ip_address, addresses[0].ip_address)
# pg6 session dump
- sessions = self.vapi.nat44_ei_user_session_dump(
- self.pg6.remote_ip4, 20)
+ sessions = self.vapi.nat44_ei_user_session_dump(self.pg6.remote_ip4, 20)
self.assertGreaterEqual(len(sessions), 3)
for session in sessions:
+ self.assertTrue(session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
+ self.assertEqual(str(session.inside_ip_address), self.pg6.remote_ip4)
+ self.assertEqual(str(session.outside_ip_address), static_nat_ip)
self.assertTrue(
- session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
- self.assertEqual(str(session.inside_ip_address),
- self.pg6.remote_ip4)
- self.assertEqual(str(session.outside_ip_address),
- static_nat_ip)
- self.assertTrue(session.inside_port in
- [self.tcp_port_in, self.udp_port_in,
- self.icmp_id_in])
+ session.inside_port
+ in [self.tcp_port_in, self.udp_port_in, self.icmp_id_in]
+ )
def test_hairpinning(self):
- """ NAT44EI hairpinning - 1:1 NAPT """
+ """NAT44EI hairpinning - 1:1 NAPT"""
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# add static mapping for server
- self.nat44_add_static_mapping(server.ip4, self.nat_addr,
- server_in_port, server_out_port,
- proto=IP_PROTOS.tcp)
-
- cnt = self.statistics['/nat44-ei/hairpinning']
+ self.nat44_add_static_mapping(
+ server.ip4,
+ self.nat_addr,
+ server_in_port,
+ server_out_port,
+ proto=IP_PROTOS.tcp,
+ )
+
+ cnt = self.statistics["/nat44-ei/hairpinning"]
# send packet from host to server
- p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=self.nat_addr) /
- TCP(sport=host_in_port, dport=server_out_port))
+ p = (
+ Ether(src=host.mac, dst=self.pg0.local_mac)
+ / IP(src=host.ip4, dst=self.nat_addr)
+ / 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.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- after = self.statistics['/nat44-ei/hairpinning']
+ after = self.statistics["/nat44-ei/hairpinning"]
if_idx = self.pg0.sw_if_index
self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(), 1)
# send reply from server to host
- p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
- IP(src=server.ip4, dst=self.nat_addr) /
- TCP(sport=server_in_port, dport=host_out_port))
+ p = (
+ Ether(src=server.mac, dst=self.pg0.local_mac)
+ / IP(src=server.ip4, dst=self.nat_addr)
+ / TCP(sport=server_in_port, dport=host_out_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
- after = self.statistics['/nat44-ei/hairpinning']
+ after = self.statistics["/nat44-ei/hairpinning"]
if_idx = self.pg0.sw_if_index
- self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(),
- 2+(1 if self.vpp_worker_count > 0 else 0))
+ self.assertEqual(
+ after[:, if_idx].sum() - cnt[:, if_idx].sum(),
+ 2 + (1 if self.vpp_worker_count > 0 else 0),
+ )
def test_hairpinning2(self):
- """ NAT44EI hairpinning - 1:1 NAT"""
+ """NAT44EI hairpinning - 1:1 NAT"""
server1_nat_ip = "10.0.0.10"
server2_nat_ip = "10.0.0.11"
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# add static mapping for servers
self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
# host to server1
pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=host.ip4, dst=server1_nat_ip) /
- TCP(sport=self.tcp_port_in, dport=server_tcp_port))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=host.ip4, dst=server1_nat_ip)
+ / TCP(sport=self.tcp_port_in, dport=server_tcp_port)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=host.ip4, dst=server1_nat_ip) /
- UDP(sport=self.udp_port_in, dport=server_udp_port))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=host.ip4, dst=server1_nat_ip)
+ / UDP(sport=self.udp_port_in, dport=server_udp_port)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=host.ip4, dst=server1_nat_ip) /
- ICMP(id=self.icmp_id_in, type='echo-request'))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=host.ip4, dst=server1_nat_ip)
+ / ICMP(id=self.icmp_id_in, type="echo-request")
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
# server1 to host
pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.nat_addr) /
- TCP(sport=server_tcp_port, dport=self.tcp_port_out))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=self.nat_addr)
+ / TCP(sport=server_tcp_port, dport=self.tcp_port_out)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.nat_addr) /
- UDP(sport=server_udp_port, dport=self.udp_port_out))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=self.nat_addr)
+ / UDP(sport=server_udp_port, dport=self.udp_port_out)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.nat_addr) /
- ICMP(id=self.icmp_id_out, type='echo-reply'))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=self.nat_addr)
+ / ICMP(id=self.icmp_id_out, type="echo-reply")
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
# server2 to server1
pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server2.ip4, dst=server1_nat_ip) /
- TCP(sport=self.tcp_port_in, dport=server_tcp_port))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server2.ip4, dst=server1_nat_ip)
+ / TCP(sport=self.tcp_port_in, dport=server_tcp_port)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server2.ip4, dst=server1_nat_ip) /
- UDP(sport=self.udp_port_in, dport=server_udp_port))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server2.ip4, dst=server1_nat_ip)
+ / UDP(sport=self.udp_port_in, dport=server_udp_port)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server2.ip4, dst=server1_nat_ip) /
- ICMP(id=self.icmp_id_in, type='echo-request'))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server2.ip4, dst=server1_nat_ip)
+ / ICMP(id=self.icmp_id_in, type="echo-request")
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
# server1 to server2
pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=server2_nat_ip) /
- TCP(sport=server_tcp_port, dport=self.tcp_port_out))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=server2_nat_ip)
+ / TCP(sport=server_tcp_port, dport=self.tcp_port_out)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=server2_nat_ip) /
- UDP(sport=server_udp_port, dport=self.udp_port_out))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=server2_nat_ip)
+ / UDP(sport=server_udp_port, dport=self.udp_port_out)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=server2_nat_ip) /
- ICMP(id=self.icmp_id_out, type='echo-reply'))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=server2_nat_ip)
+ / ICMP(id=self.icmp_id_out, type="echo-reply")
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
raise
def test_hairpinning_avoid_inf_loop(self):
- """ NAT44EI hairpinning - 1:1 NAPT avoid infinite loop """
+ """NAT44EI hairpinning - 1:1 NAPT avoid infinite loop"""
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# add static mapping for server
- self.nat44_add_static_mapping(server.ip4, self.nat_addr,
- server_in_port, server_out_port,
- proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(
+ server.ip4,
+ self.nat_addr,
+ server_in_port,
+ server_out_port,
+ proto=IP_PROTOS.tcp,
+ )
# add another static mapping that maps pg0.local_ip4 address to itself
self.nat44_add_static_mapping(self.pg0.local_ip4, self.pg0.local_ip4)
# send packet from host to VPP (the packet should get dropped)
- p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=self.pg0.local_ip4) /
- TCP(sport=host_in_port, dport=server_out_port))
+ p = (
+ Ether(src=host.mac, dst=self.pg0.local_mac)
+ / IP(src=host.ip4, dst=self.pg0.local_ip4)
+ / TCP(sport=host_in_port, dport=server_out_port)
+ )
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
# Here VPP used to crash due to an infinite loop
- cnt = self.statistics['/nat44-ei/hairpinning']
+ cnt = self.statistics["/nat44-ei/hairpinning"]
# send packet from host to server
- p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=self.nat_addr) /
- TCP(sport=host_in_port, dport=server_out_port))
+ p = (
+ Ether(src=host.mac, dst=self.pg0.local_mac)
+ / IP(src=host.ip4, dst=self.nat_addr)
+ / 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.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- after = self.statistics['/nat44-ei/hairpinning']
+ after = self.statistics["/nat44-ei/hairpinning"]
if_idx = self.pg0.sw_if_index
self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(), 1)
# send reply from server to host
- p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
- IP(src=server.ip4, dst=self.nat_addr) /
- TCP(sport=server_in_port, dport=host_out_port))
+ p = (
+ Ether(src=server.mac, dst=self.pg0.local_mac)
+ / IP(src=server.ip4, dst=self.nat_addr)
+ / TCP(sport=server_in_port, dport=host_out_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
- after = self.statistics['/nat44-ei/hairpinning']
+ after = self.statistics["/nat44-ei/hairpinning"]
if_idx = self.pg0.sw_if_index
- self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(),
- 2+(1 if self.vpp_worker_count > 0 else 0))
+ self.assertEqual(
+ after[:, if_idx].sum() - cnt[:, if_idx].sum(),
+ 2 + (1 if self.vpp_worker_count > 0 else 0),
+ )
def test_interface_addr(self):
- """ NAT44EI acquire addresses from interface """
+ """NAT44EI acquire addresses from interface"""
self.vapi.nat44_ei_add_del_interface_addr(
- is_add=1,
- sw_if_index=self.pg7.sw_if_index)
+ is_add=1, sw_if_index=self.pg7.sw_if_index
+ )
# no address in NAT pool
addresses = self.vapi.nat44_ei_address_dump()
self.assertEqual(0, len(addresses))
def test_interface_addr_static_mapping(self):
- """ NAT44EI Static mapping with addresses from interface """
+ """NAT44EI Static mapping with addresses from interface"""
tag = "testTAG"
self.vapi.nat44_ei_add_del_interface_addr(
- is_add=1,
- sw_if_index=self.pg7.sw_if_index)
+ is_add=1, sw_if_index=self.pg7.sw_if_index
+ )
self.nat44_add_static_mapping(
- '1.2.3.4',
- external_sw_if_index=self.pg7.sw_if_index,
- tag=tag)
+ "1.2.3.4", external_sw_if_index=self.pg7.sw_if_index, tag=tag
+ )
# static mappings with external interface
static_mappings = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(1, len(static_mappings))
- self.assertEqual(self.pg7.sw_if_index,
- static_mappings[0].external_sw_if_index)
+ self.assertEqual(self.pg7.sw_if_index, static_mappings[0].external_sw_if_index)
self.assertEqual(static_mappings[0].tag, tag)
# configure interface address and check static mappings
resolved = False
for sm in static_mappings:
if sm.external_sw_if_index == 0xFFFFFFFF:
- self.assertEqual(str(sm.external_ip_address),
- self.pg7.local_ip4)
+ self.assertEqual(str(sm.external_ip_address), self.pg7.local_ip4)
self.assertEqual(sm.tag, tag)
resolved = True
self.assertTrue(resolved)
self.pg7.unconfig_ip4()
static_mappings = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(1, len(static_mappings))
- self.assertEqual(self.pg7.sw_if_index,
- static_mappings[0].external_sw_if_index)
+ self.assertEqual(self.pg7.sw_if_index, static_mappings[0].external_sw_if_index)
self.assertEqual(static_mappings[0].tag, tag)
# configure interface address again and check static mappings
resolved = False
for sm in static_mappings:
if sm.external_sw_if_index == 0xFFFFFFFF:
- self.assertEqual(str(sm.external_ip_address),
- self.pg7.local_ip4)
+ self.assertEqual(str(sm.external_ip_address), self.pg7.local_ip4)
self.assertEqual(sm.tag, tag)
resolved = True
self.assertTrue(resolved)
# remove static mapping
self.nat44_add_static_mapping(
- '1.2.3.4',
- external_sw_if_index=self.pg7.sw_if_index,
- tag=tag,
- is_add=0)
+ "1.2.3.4", external_sw_if_index=self.pg7.sw_if_index, tag=tag, is_add=0
+ )
static_mappings = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(0, len(static_mappings))
def test_interface_addr_identity_nat(self):
- """ NAT44EI Identity NAT with addresses from interface """
+ """NAT44EI Identity NAT with addresses from interface"""
port = 53053
self.vapi.nat44_ei_add_del_interface_addr(
- is_add=1,
- sw_if_index=self.pg7.sw_if_index)
+ is_add=1, sw_if_index=self.pg7.sw_if_index
+ )
self.vapi.nat44_ei_add_del_identity_mapping(
- ip_address=b'0',
+ ip_address=b"0",
sw_if_index=self.pg7.sw_if_index,
port=port,
protocol=IP_PROTOS.tcp,
- is_add=1)
+ is_add=1,
+ )
# identity mappings with external interface
identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
self.assertEqual(1, len(identity_mappings))
- self.assertEqual(self.pg7.sw_if_index,
- identity_mappings[0].sw_if_index)
+ self.assertEqual(self.pg7.sw_if_index, identity_mappings[0].sw_if_index)
# configure interface address and check identity mappings
self.pg7.config_ip4()
self.assertEqual(2, len(identity_mappings))
for sm in identity_mappings:
if sm.sw_if_index == 0xFFFFFFFF:
- self.assertEqual(str(identity_mappings[0].ip_address),
- self.pg7.local_ip4)
+ self.assertEqual(
+ str(identity_mappings[0].ip_address), self.pg7.local_ip4
+ )
self.assertEqual(port, identity_mappings[0].port)
self.assertEqual(IP_PROTOS.tcp, identity_mappings[0].protocol)
resolved = True
self.pg7.unconfig_ip4()
identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
self.assertEqual(1, len(identity_mappings))
- self.assertEqual(self.pg7.sw_if_index,
- identity_mappings[0].sw_if_index)
+ self.assertEqual(self.pg7.sw_if_index, identity_mappings[0].sw_if_index)
def test_ipfix_nat44_sess(self):
- """ NAT44EI IPFIX logging NAT44EI session created/deleted """
+ """NAT44EI IPFIX logging NAT44EI session created/deleted"""
self.ipfix_domain_id = 10
self.ipfix_src_port = 20202
collector_port = 30303
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
- self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
- src_address=self.pg3.local_ip4,
- path_mtu=512,
- template_interval=10,
- collector_port=collector_port)
- self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port,
- enable=1)
+ sw_if_index=self.pg1.sw_if_index, is_add=1
+ )
+ self.vapi.set_ipfix_exporter(
+ collector_address=self.pg3.remote_ip4,
+ src_address=self.pg3.local_ip4,
+ path_mtu=512,
+ template_interval=10,
+ collector_port=collector_port,
+ )
+ self.vapi.nat44_ei_ipfix_enable_disable(
+ domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port, enable=1
+ )
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
self.assertEqual(p[UDP].sport, self.ipfix_src_port)
self.assertEqual(p[UDP].dport, collector_port)
- self.assertEqual(p[IPFIX].observationDomainID,
- self.ipfix_domain_id)
+ self.assertEqual(p[IPFIX].observationDomainID, self.ipfix_domain_id)
if p.haslayer(Template):
ipfix.add_template(p.getlayer(Template))
# verify events in data set
self.verify_ipfix_nat44_ses(data)
def test_ipfix_addr_exhausted(self):
- """ NAT44EI IPFIX logging NAT addresses exhausted """
+ """NAT44EI IPFIX logging NAT addresses exhausted"""
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
- self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
- src_address=self.pg3.local_ip4,
- path_mtu=512,
- template_interval=10)
- self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port,
- enable=1)
-
- p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=3025))
+ sw_if_index=self.pg1.sw_if_index, is_add=1
+ )
+ self.vapi.set_ipfix_exporter(
+ collector_address=self.pg3.remote_ip4,
+ src_address=self.pg3.local_ip4,
+ path_mtu=512,
+ template_interval=10,
+ )
+ self.vapi.nat44_ei_ipfix_enable_disable(
+ domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port, enable=1
+ )
+
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
+ / TCP(sport=3025)
+ )
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
self.assertEqual(p[UDP].sport, self.ipfix_src_port)
self.assertEqual(p[UDP].dport, 4739)
- self.assertEqual(p[IPFIX].observationDomainID,
- self.ipfix_domain_id)
+ self.assertEqual(p[IPFIX].observationDomainID, self.ipfix_domain_id)
if p.haslayer(Template):
ipfix.add_template(p.getlayer(Template))
# verify events in data set
self.verify_ipfix_addr_exhausted(data)
def test_ipfix_max_sessions(self):
- """ NAT44EI IPFIX logging maximum session entries exceeded """
+ """NAT44EI IPFIX logging maximum session entries exceeded"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
max_sessions_per_thread = self.max_translations
max_sessions = max(1, self.vpp_worker_count) * max_sessions_per_thread
pkts = []
for i in range(0, max_sessions):
src = "10.10.%u.%u" % ((i & 0xFF00) >> 8, i & 0xFF)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=src, dst=self.pg1.remote_ip4) /
- TCP(sport=1025))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=src, dst=self.pg1.remote_ip4)
+ / TCP(sport=1025)
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.pg1.get_capture(max_sessions)
- self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
- src_address=self.pg3.local_ip4,
- path_mtu=512,
- template_interval=10)
- self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port,
- enable=1)
-
- 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=1025))
+ self.vapi.set_ipfix_exporter(
+ collector_address=self.pg3.remote_ip4,
+ src_address=self.pg3.local_ip4,
+ path_mtu=512,
+ template_interval=10,
+ )
+ self.vapi.nat44_ei_ipfix_enable_disable(
+ domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port, enable=1
+ )
+
+ 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=1025)
+ )
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
self.assertEqual(p[UDP].sport, self.ipfix_src_port)
self.assertEqual(p[UDP].dport, 4739)
- self.assertEqual(p[IPFIX].observationDomainID,
- self.ipfix_domain_id)
+ self.assertEqual(p[IPFIX].observationDomainID, self.ipfix_domain_id)
if p.haslayer(Template):
ipfix.add_template(p.getlayer(Template))
# verify events in data set
self.verify_ipfix_max_sessions(data, max_sessions_per_thread)
def test_syslog_apmap(self):
- """ NAT44EI syslog address and port mapping creation and deletion """
- self.vapi.syslog_set_filter(
- self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_INFO)
+ """NAT44EI syslog address and port mapping 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.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
-
- 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))
+ sw_if_index=self.pg1.sw_if_index, is_add=1
+ )
+
+ 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()
self.verify_syslog_apmap(capture[0][Raw].load, False)
def test_pool_addr_fib(self):
- """ NAT44EI add pool addresses to FIB """
- static_addr = '10.0.0.10'
+ """NAT44EI add pool addresses to FIB"""
+ static_addr = "10.0.0.10"
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr)
# NAT44EI address
- p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=self.nat_addr,
- psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
+ p = Ether(src=self.pg1.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
+ op=ARP.who_has,
+ pdst=self.nat_addr,
+ psrc=self.pg1.remote_ip4,
+ hwsrc=self.pg1.remote_mac,
+ )
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertTrue(capture[0][ARP].op, ARP.is_at)
# 1:1 NAT address
- p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=static_addr,
- psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
+ p = Ether(src=self.pg1.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
+ op=ARP.who_has,
+ pdst=static_addr,
+ psrc=self.pg1.remote_ip4,
+ hwsrc=self.pg1.remote_mac,
+ )
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertTrue(capture[0][ARP].op, ARP.is_at)
# send ARP to non-NAT44EI interface
- p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=self.nat_addr,
- psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
+ p = Ether(src=self.pg2.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
+ op=ARP.who_has,
+ pdst=self.nat_addr,
+ psrc=self.pg2.remote_ip4,
+ hwsrc=self.pg2.remote_mac,
+ )
self.pg2.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
# remove addresses and verify
self.nat44_add_address(self.nat_addr, is_add=0)
- self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr,
- is_add=0)
-
- p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=self.nat_addr,
- psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr, is_add=0)
+
+ p = Ether(src=self.pg1.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
+ op=ARP.who_has,
+ pdst=self.nat_addr,
+ psrc=self.pg1.remote_ip4,
+ hwsrc=self.pg1.remote_mac,
+ )
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.pg1.assert_nothing_captured()
- p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=static_addr,
- psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
+ p = Ether(src=self.pg1.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
+ op=ARP.who_has,
+ pdst=static_addr,
+ psrc=self.pg1.remote_ip4,
+ hwsrc=self.pg1.remote_mac,
+ )
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.pg1.assert_nothing_captured()
def test_vrf_mode(self):
- """ NAT44EI tenant VRF aware address pool mode """
+ """NAT44EI tenant VRF aware address pool mode"""
vrf_id1 = 1
vrf_id2 = 2
self.pg0.unconfig_ip4()
self.pg1.unconfig_ip4()
- self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id1})
- self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id2})
+ self.vapi.ip_table_add_del(is_add=1, table={"table_id": vrf_id1})
+ self.vapi.ip_table_add_del(is_add=1, table={"table_id": vrf_id2})
self.pg0.set_table_ip4(vrf_id1)
self.pg1.set_table_ip4(vrf_id2)
self.pg0.config_ip4()
self.nat44_add_address(nat_ip2, vrf_id=vrf_id2)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- flags=flags, is_add=1)
+ sw_if_index=self.pg1.sw_if_index, flags=flags, is_add=1
+ )
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg2.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg2.sw_if_index, is_add=1
+ )
try:
# first VRF
self.pg1.config_ip4()
self.pg0.resolve_arp()
self.pg1.resolve_arp()
- self.vapi.ip_table_add_del(is_add=0, table={'table_id': vrf_id1})
- self.vapi.ip_table_add_del(is_add=0, table={'table_id': vrf_id2})
+ self.vapi.ip_table_add_del(is_add=0, table={"table_id": vrf_id1})
+ self.vapi.ip_table_add_del(is_add=0, table={"table_id": vrf_id2})
def test_vrf_feature_independent(self):
- """ NAT44EI tenant VRF independent address pool mode """
+ """NAT44EI tenant VRF independent address pool mode"""
nat_ip1 = "10.0.0.10"
nat_ip2 = "10.0.0.11"
self.nat44_add_address(nat_ip2, vrf_id=99)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- flags=flags, is_add=1)
+ sw_if_index=self.pg1.sw_if_index, flags=flags, is_add=1
+ )
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg2.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg2.sw_if_index, is_add=1
+ )
# first VRF
pkts = self.create_stream_in(self.pg0, self.pg2)
self.verify_capture_out(capture, nat_ip1)
def test_dynamic_ipless_interfaces(self):
- """ NAT44EI interfaces without configured IP address """
+ """NAT44EI interfaces without configured IP address"""
self.create_routes_and_neigbors()
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg7.sw_if_index,
- flags=flags, is_add=1)
+ sw_if_index=self.pg7.sw_if_index, flags=flags, is_add=1
+ )
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg8.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg8.sw_if_index, is_add=1
+ )
# in2out
pkts = self.create_stream_in(self.pg7, self.pg8)
self.verify_capture_in(capture, self.pg7)
def test_static_ipless_interfaces(self):
- """ NAT44EI interfaces without configured IP address - 1:1 NAT """
+ """NAT44EI interfaces without configured IP address - 1:1 NAT"""
self.create_routes_and_neigbors()
self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg7.sw_if_index,
- flags=flags, is_add=1)
+ sw_if_index=self.pg7.sw_if_index, flags=flags, is_add=1
+ )
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg8.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg8.sw_if_index, is_add=1
+ )
# out2in
pkts = self.create_stream_out(self.pg8)
self.verify_capture_out(capture, self.nat_addr, True)
def test_static_with_port_ipless_interfaces(self):
- """ NAT44EI interfaces without configured IP address - 1:1 NAPT """
+ """NAT44EI interfaces without configured IP address - 1:1 NAPT"""
self.tcp_port_out = 30606
self.udp_port_out = 30607
self.create_routes_and_neigbors()
self.nat44_add_address(self.nat_addr)
- self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
- self.tcp_port_in, self.tcp_port_out,
- proto=IP_PROTOS.tcp)
- self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
- self.udp_port_in, self.udp_port_out,
- proto=IP_PROTOS.udp)
- self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
- self.icmp_id_in, self.icmp_id_out,
- proto=IP_PROTOS.icmp)
+ self.nat44_add_static_mapping(
+ self.pg7.remote_ip4,
+ self.nat_addr,
+ self.tcp_port_in,
+ self.tcp_port_out,
+ proto=IP_PROTOS.tcp,
+ )
+ self.nat44_add_static_mapping(
+ self.pg7.remote_ip4,
+ self.nat_addr,
+ self.udp_port_in,
+ self.udp_port_out,
+ proto=IP_PROTOS.udp,
+ )
+ self.nat44_add_static_mapping(
+ self.pg7.remote_ip4,
+ self.nat_addr,
+ self.icmp_id_in,
+ self.icmp_id_out,
+ proto=IP_PROTOS.icmp,
+ )
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg7.sw_if_index,
- flags=flags, is_add=1)
+ sw_if_index=self.pg7.sw_if_index, flags=flags, is_add=1
+ )
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg8.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg8.sw_if_index, is_add=1
+ )
# out2in
pkts = self.create_stream_out(self.pg8)
self.verify_capture_out(capture)
def test_static_unknown_proto(self):
- """ NAT44EI 1:1 translate packet with unknown protocol """
+ """NAT44EI 1:1 translate packet with unknown protocol"""
nat_ip = "10.0.0.10"
self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# in2out
- 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.pg3.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.pg3.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=nat_ip) /
- GRE() /
- IP(src=self.pg3.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=nat_ip)
+ / GRE()
+ / IP(src=self.pg3.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_static_unknown_proto(self):
- """ NAT44EI 1:1 translate packet with unknown protocol - hairpinning
- """
+ """NAT44EI 1:1 translate packet with unknown protocol - hairpinning"""
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
self.nat44_add_static_mapping(server.ip4, server_nat_ip)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# host to server
- 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.pg3.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.pg3.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=host_nat_ip) /
- GRE() /
- IP(src=self.pg3.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=host_nat_ip)
+ / GRE()
+ / IP(src=self.pg3.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(self):
- """ NAT44EI output feature (in2out postrouting) """
+ """NAT44EI output feature (in2out postrouting)"""
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_ei_add_del_output_interface(
- sw_if_index=self.pg3.sw_if_index, is_add=1)
+ sw_if_index=self.pg3.sw_if_index, is_add=1
+ )
# in2out
pkts = self.create_stream_in(self.pg0, self.pg3)
self.verify_capture_no_translation(capture, self.pg2, self.pg0)
def test_output_feature_vrf_aware(self):
- """ NAT44EI output feature VRF aware (in2out postrouting) """
+ """NAT44EI output feature VRF aware (in2out postrouting)"""
nat_ip_vrf10 = "10.0.0.10"
nat_ip_vrf20 = "10.0.0.20"
- r1 = VppIpRoute(self, self.pg3.remote_ip4, 32,
- [VppRoutePath(self.pg3.remote_ip4,
- self.pg3.sw_if_index)],
- table_id=10)
- r2 = VppIpRoute(self, self.pg3.remote_ip4, 32,
- [VppRoutePath(self.pg3.remote_ip4,
- self.pg3.sw_if_index)],
- table_id=20)
+ r1 = VppIpRoute(
+ self,
+ self.pg3.remote_ip4,
+ 32,
+ [VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index)],
+ table_id=10,
+ )
+ r2 = VppIpRoute(
+ self,
+ self.pg3.remote_ip4,
+ 32,
+ [VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index)],
+ table_id=20,
+ )
r1.add_vpp_config()
r2.add_vpp_config()
self.nat44_add_address(nat_ip_vrf10, vrf_id=10)
self.nat44_add_address(nat_ip_vrf20, vrf_id=20)
self.vapi.nat44_ei_add_del_output_interface(
- sw_if_index=self.pg3.sw_if_index, is_add=1)
+ sw_if_index=self.pg3.sw_if_index, is_add=1
+ )
# in2out VRF 10
pkts = self.create_stream_in(self.pg4, self.pg3)
self.verify_capture_in(capture, self.pg6)
def test_output_feature_hairpinning(self):
- """ NAT44EI output feature hairpinning (in2out postrouting) """
+ """NAT44EI output feature hairpinning (in2out postrouting)"""
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
host_in_port = 1234
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_ei_add_del_output_interface(
- 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_ei_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
+ )
# add static mapping for server
- self.nat44_add_static_mapping(server.ip4, self.nat_addr,
- server_in_port, server_out_port,
- proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(
+ server.ip4,
+ self.nat_addr,
+ server_in_port,
+ server_out_port,
+ proto=IP_PROTOS.tcp,
+ )
# send packet from host to server
- p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=self.nat_addr) /
- TCP(sport=host_in_port, dport=server_out_port))
+ p = (
+ Ether(src=host.mac, dst=self.pg0.local_mac)
+ / IP(src=host.ip4, dst=self.nat_addr)
+ / TCP(sport=host_in_port, dport=server_out_port)
+ )
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
raise
# send reply from server to host
- p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
- IP(src=server.ip4, dst=self.nat_addr) /
- TCP(sport=server_in_port, dport=host_out_port))
+ p = (
+ Ether(src=server.mac, dst=self.pg0.local_mac)
+ / IP(src=server.ip4, dst=self.nat_addr)
+ / TCP(sport=server_in_port, dport=host_out_port)
+ )
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
raise
def test_one_armed_nat44(self):
- """ NAT44EI One armed NAT """
+ """NAT44EI One armed NAT"""
remote_host = self.pg9.remote_hosts[0]
local_host = self.pg9.remote_hosts[1]
external_port = 0
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg9.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg9.sw_if_index, is_add=1
+ )
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg9.sw_if_index,
- flags=flags, is_add=1)
+ sw_if_index=self.pg9.sw_if_index, flags=flags, is_add=1
+ )
# in2out
- p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
- IP(src=local_host.ip4, dst=remote_host.ip4) /
- TCP(sport=12345, dport=80))
+ p = (
+ Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac)
+ / IP(src=local_host.ip4, dst=remote_host.ip4)
+ / TCP(sport=12345, dport=80)
+ )
self.pg9.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
raise
# out2in
- p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
- IP(src=remote_host.ip4, dst=self.nat_addr) /
- TCP(sport=80, dport=external_port))
+ p = (
+ Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac)
+ / IP(src=remote_host.ip4, dst=self.nat_addr)
+ / TCP(sport=80, dport=external_port)
+ )
self.pg9.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
else:
node = "nat44-ei-classify"
- err = self.statistics.get_err_counter('/err/%s/next in2out' % node)
+ err = self.statistics.get_err_counter("/err/%s/next in2out" % node)
self.assertEqual(err, 1)
- err = self.statistics.get_err_counter('/err/%s/next out2in' % node)
+ err = self.statistics.get_err_counter("/err/%s/next out2in" % node)
self.assertEqual(err, 1)
def test_del_session(self):
- """ NAT44EI delete session """
+ """NAT44EI delete session"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
address=sessions[0].inside_ip_address,
port=sessions[0].inside_port,
protocol=sessions[0].protocol,
- flags=self.config_flags.NAT44_EI_IF_INSIDE)
+ flags=self.config_flags.NAT44_EI_IF_INSIDE,
+ )
self.vapi.nat44_ei_del_session(
address=sessions[1].outside_ip_address,
port=sessions[1].outside_port,
- protocol=sessions[1].protocol)
+ protocol=sessions[1].protocol,
+ )
sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
self.assertEqual(nsessions - len(sessions), 2)
address=sessions[0].inside_ip_address,
port=sessions[0].inside_port,
protocol=sessions[0].protocol,
- flags=self.config_flags.NAT44_EI_IF_INSIDE)
+ flags=self.config_flags.NAT44_EI_IF_INSIDE,
+ )
self.verify_no_nat44_user()
def test_frag_in_order(self):
- """ NAT44EI translate fragments arriving in order """
+ """NAT44EI translate fragments arriving in order"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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.frag_in_order(proto=IP_PROTOS.tcp)
self.frag_in_order(proto=IP_PROTOS.udp)
self.frag_in_order(proto=IP_PROTOS.icmp)
def test_frag_forwarding(self):
- """ NAT44EI forwarding fragment test """
+ """NAT44EI forwarding fragment test"""
self.vapi.nat44_ei_add_del_interface_addr(
- is_add=1,
- sw_if_index=self.pg1.sw_if_index)
+ is_add=1, sw_if_index=self.pg1.sw_if_index
+ )
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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.vapi.nat44_ei_forwarding_enable_disable(enable=1)
data = b"A" * 16 + b"B" * 16 + b"C" * 3
- pkts = self.create_stream_frag(self.pg1,
- self.pg0.remote_ip4,
- 4789,
- 4789,
- data,
- proto=IP_PROTOS.udp)
+ pkts = self.create_stream_frag(
+ self.pg1, self.pg0.remote_ip4, 4789, 4789, data, proto=IP_PROTOS.udp
+ )
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)
self.assertEqual(p[UDP].sport, 4789)
self.assertEqual(p[UDP].dport, 4789)
self.assertEqual(data, p[Raw].load)
def test_reass_hairpinning(self):
- """ NAT44EI fragments hairpinning """
+ """NAT44EI fragments hairpinning"""
server_addr = self.pg0.remote_hosts[1].ip4
host_in_port = random.randint(1025, 65535)
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# add static mapping for server
- self.nat44_add_static_mapping(server_addr, self.nat_addr,
- server_in_port,
- server_out_port,
- proto=IP_PROTOS.tcp)
- self.nat44_add_static_mapping(server_addr, self.nat_addr,
- server_in_port,
- server_out_port,
- proto=IP_PROTOS.udp)
+ self.nat44_add_static_mapping(
+ server_addr,
+ self.nat_addr,
+ server_in_port,
+ server_out_port,
+ proto=IP_PROTOS.tcp,
+ )
+ self.nat44_add_static_mapping(
+ server_addr,
+ self.nat_addr,
+ server_in_port,
+ server_out_port,
+ proto=IP_PROTOS.udp,
+ )
self.nat44_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)
- self.reass_hairpinning(server_addr, server_in_port, server_out_port,
- host_in_port, proto=IP_PROTOS.udp)
- self.reass_hairpinning(server_addr, server_in_port, server_out_port,
- host_in_port, proto=IP_PROTOS.icmp)
+ self.reass_hairpinning(
+ server_addr,
+ server_in_port,
+ server_out_port,
+ host_in_port,
+ proto=IP_PROTOS.tcp,
+ )
+ self.reass_hairpinning(
+ server_addr,
+ server_in_port,
+ server_out_port,
+ host_in_port,
+ proto=IP_PROTOS.udp,
+ )
+ self.reass_hairpinning(
+ server_addr,
+ server_in_port,
+ server_out_port,
+ host_in_port,
+ proto=IP_PROTOS.icmp,
+ )
def test_frag_out_of_order(self):
- """ NAT44EI translate fragments arriving out of order """
+ """NAT44EI translate fragments arriving out of order"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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.frag_out_of_order(proto=IP_PROTOS.tcp)
self.frag_out_of_order(proto=IP_PROTOS.udp)
self.frag_out_of_order(proto=IP_PROTOS.icmp)
def test_port_restricted(self):
- """ NAT44EI Port restricted NAT44EI (MAP-E CE) """
+ """NAT44EI Port restricted NAT44EI (MAP-E CE)"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
- self.vapi.nat44_ei_set_addr_and_port_alloc_alg(alg=1,
- psid_offset=6,
- psid_length=6,
- psid=10)
-
- 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=4567, dport=22))
+ sw_if_index=self.pg1.sw_if_index, is_add=1
+ )
+ self.vapi.nat44_ei_set_addr_and_port_alloc_alg(
+ alg=1, psid_offset=6, psid_length=6, psid=10
+ )
+
+ 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=4567, dport=22)
+ )
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
raise
def test_port_range(self):
- """ NAT44EI External address port range """
+ """NAT44EI External address port range"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
- self.vapi.nat44_ei_set_addr_and_port_alloc_alg(alg=2,
- start_port=1025,
- end_port=1027)
+ sw_if_index=self.pg1.sw_if_index, is_add=1
+ )
+ self.vapi.nat44_ei_set_addr_and_port_alloc_alg(
+ alg=2, start_port=1025, end_port=1027
+ )
pkts = []
for port in range(0, 5):
- 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=1125 + 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=1125 + port)
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.assertLessEqual(tcp.sport, 1027)
def test_multiple_outside_vrf(self):
- """ NAT44EI Multiple outside VRF """
+ """NAT44EI Multiple outside VRF"""
vrf_id1 = 1
vrf_id2 = 2
self.pg1.unconfig_ip4()
self.pg2.unconfig_ip4()
- self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id1})
- self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id2})
+ self.vapi.ip_table_add_del(is_add=1, table={"table_id": vrf_id1})
+ self.vapi.ip_table_add_del(is_add=1, table={"table_id": vrf_id2})
self.pg1.set_table_ip4(vrf_id1)
self.pg2.set_table_ip4(vrf_id2)
self.pg1.config_ip4()
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg2.sw_if_index,
- is_add=1)
+ sw_if_index=self.pg2.sw_if_index, is_add=1
+ )
try:
# first VRF
self.pg2.resolve_arp()
def test_mss_clamping(self):
- """ NAT44EI TCP MSS clamping """
+ """NAT44EI TCP MSS clamping"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- 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=self.pg1.remote_ip4) /
- TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
- flags="S", options=[('MSS', 1400)]))
+ 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=self.pg1.remote_ip4)
+ / TCP(
+ sport=self.tcp_port_in,
+ dport=self.tcp_external_port,
+ flags="S",
+ options=[("MSS", 1400)],
+ )
+ )
self.vapi.nat44_ei_set_mss_clamping(enable=1, mss_value=1000)
self.pg0.add_stream(p)
self.verify_mss_value(capture[0], 1400)
def test_ha_send(self):
- """ NAT44EI Send HA session synchronization events (active) """
+ """NAT44EI Send HA session synchronization events (active)"""
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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.nat44_add_address(self.nat_addr)
self.vapi.nat44_ei_ha_set_listener(
- ip_address=self.pg3.local_ip4, port=12345, path_mtu=512)
+ ip_address=self.pg3.local_ip4, port=12345, path_mtu=512
+ )
self.vapi.nat44_ei_ha_set_failover(
- ip_address=self.pg3.remote_ip4, port=12346,
- session_refresh_interval=10)
+ ip_address=self.pg3.remote_ip4, port=12346, session_refresh_interval=10
+ )
bind_layers(UDP, HANATStateSync, sport=12345)
# create sessions
self.verify_capture_out(capture)
# active send HA events
self.vapi.nat44_ei_ha_flush()
- stats = self.statistics['/nat44-ei/ha/add-event-send']
+ stats = self.statistics["/nat44-ei/ha/add-event-send"]
self.assertEqual(stats[:, 0].sum(), 3)
capture = self.pg3.get_capture(1)
p = capture[0]
self.assertEqual(event.fib_index, 0)
# ACK received events
- ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
- IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
- UDP(sport=12346, dport=12345) /
- HANATStateSync(sequence_number=seq, flags='ACK',
- thread_index=hanat.thread_index))
+ ack = (
+ Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac)
+ / IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4)
+ / UDP(sport=12346, dport=12345)
+ / HANATStateSync(
+ sequence_number=seq, flags="ACK", thread_index=hanat.thread_index
+ )
+ )
self.pg3.add_stream(ack)
self.pg_start()
- stats = self.statistics['/nat44-ei/ha/ack-recv']
+ stats = self.statistics["/nat44-ei/ha/ack-recv"]
self.assertEqual(stats[:, 0].sum(), 1)
# delete one session
self.pg_enable_capture(self.pg_interfaces)
self.vapi.nat44_ei_del_session(
- address=self.pg0.remote_ip4, port=self.tcp_port_in,
- protocol=IP_PROTOS.tcp, flags=self.config_flags.NAT44_EI_IF_INSIDE)
+ address=self.pg0.remote_ip4,
+ port=self.tcp_port_in,
+ protocol=IP_PROTOS.tcp,
+ flags=self.config_flags.NAT44_EI_IF_INSIDE,
+ )
self.vapi.nat44_ei_ha_flush()
- stats = self.statistics['/nat44-ei/ha/del-event-send']
+ stats = self.statistics["/nat44-ei/ha/del-event-send"]
self.assertEqual(stats[:, 0].sum(), 1)
capture = self.pg3.get_capture(1)
p = capture[0]
# do not send ACK, active retry send HA event again
self.pg_enable_capture(self.pg_interfaces)
self.virtual_sleep(12)
- stats = self.statistics['/nat44-ei/ha/retry-count']
+ stats = self.statistics["/nat44-ei/ha/retry-count"]
self.assertEqual(stats[:, 0].sum(), 3)
- stats = self.statistics['/nat44-ei/ha/missed-count']
+ stats = self.statistics["/nat44-ei/ha/missed-count"]
self.assertEqual(stats[:, 0].sum(), 1)
capture = self.pg3.get_capture(3)
for packet in capture:
self.pg_start()
self.pg0.get_capture(2)
self.vapi.nat44_ei_ha_flush()
- stats = self.statistics['/nat44-ei/ha/refresh-event-send']
+ stats = self.statistics["/nat44-ei/ha/refresh-event-send"]
self.assertEqual(stats[:, 0].sum(), 2)
capture = self.pg3.get_capture(1)
p = capture[0]
self.assertEqual(event.total_pkts, 2)
self.assertGreater(event.total_bytes, 0)
- stats = self.statistics['/nat44-ei/ha/ack-recv']
- ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
- IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
- UDP(sport=12346, dport=12345) /
- HANATStateSync(sequence_number=seq, flags='ACK',
- thread_index=hanat.thread_index))
+ stats = self.statistics["/nat44-ei/ha/ack-recv"]
+ ack = (
+ Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac)
+ / IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4)
+ / UDP(sport=12346, dport=12345)
+ / HANATStateSync(
+ sequence_number=seq, flags="ACK", thread_index=hanat.thread_index
+ )
+ )
self.pg3.add_stream(ack)
self.pg_start()
- stats = self.statistics['/nat44-ei/ha/ack-recv']
+ stats = self.statistics["/nat44-ei/ha/ack-recv"]
self.assertEqual(stats[:, 0].sum(), 2)
def test_ha_recv(self):
- """ NAT44EI Receive HA session synchronization events (passive) """
+ """NAT44EI Receive HA session synchronization events (passive)"""
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
- self.vapi.nat44_ei_ha_set_listener(ip_address=self.pg3.local_ip4,
- port=12345, path_mtu=512)
+ sw_if_index=self.pg1.sw_if_index, is_add=1
+ )
+ self.vapi.nat44_ei_ha_set_listener(
+ ip_address=self.pg3.local_ip4, port=12345, path_mtu=512
+ )
bind_layers(UDP, HANATStateSync, sport=12345)
# this is a bit tricky - HA dictates thread index due to how it's
# IP address) and out2in (based on outside port)
# first choose a thread index which is correct for IP
- thread_index = get_nat44_ei_in2out_worker_index(self.pg0.remote_ip4,
- self.vpp_worker_count)
+ thread_index = get_nat44_ei_in2out_worker_index(
+ self.pg0.remote_ip4, self.vpp_worker_count
+ )
# now pick a port which is correct for given thread
- port_per_thread = int((0xffff-1024) / max(1, self.vpp_worker_count))
+ port_per_thread = int((0xFFFF - 1024) / max(1, self.vpp_worker_count))
self.tcp_port_out = 1024 + random.randint(1, port_per_thread)
self.udp_port_out = 1024 + random.randint(1, port_per_thread)
if self.vpp_worker_count > 0:
self.udp_port_out += port_per_thread * (thread_index - 1)
# send HA session add events to failover/passive
- p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
- IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
- UDP(sport=12346, dport=12345) /
- HANATStateSync(sequence_number=1, events=[
- Event(event_type='add', protocol='tcp',
- in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
- in_port=self.tcp_port_in, out_port=self.tcp_port_out,
- eh_addr=self.pg1.remote_ip4,
- ehn_addr=self.pg1.remote_ip4,
- eh_port=self.tcp_external_port,
- ehn_port=self.tcp_external_port, fib_index=0),
- Event(event_type='add', protocol='udp',
- in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
- in_port=self.udp_port_in, out_port=self.udp_port_out,
- eh_addr=self.pg1.remote_ip4,
- ehn_addr=self.pg1.remote_ip4,
- eh_port=self.udp_external_port,
- ehn_port=self.udp_external_port, fib_index=0)],
- thread_index=thread_index))
+ p = (
+ Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac)
+ / IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4)
+ / UDP(sport=12346, dport=12345)
+ / HANATStateSync(
+ sequence_number=1,
+ events=[
+ Event(
+ event_type="add",
+ protocol="tcp",
+ in_addr=self.pg0.remote_ip4,
+ out_addr=self.nat_addr,
+ in_port=self.tcp_port_in,
+ out_port=self.tcp_port_out,
+ eh_addr=self.pg1.remote_ip4,
+ ehn_addr=self.pg1.remote_ip4,
+ eh_port=self.tcp_external_port,
+ ehn_port=self.tcp_external_port,
+ fib_index=0,
+ ),
+ Event(
+ event_type="add",
+ protocol="udp",
+ in_addr=self.pg0.remote_ip4,
+ out_addr=self.nat_addr,
+ in_port=self.udp_port_in,
+ out_port=self.udp_port_out,
+ eh_addr=self.pg1.remote_ip4,
+ ehn_addr=self.pg1.remote_ip4,
+ eh_port=self.udp_external_port,
+ ehn_port=self.udp_external_port,
+ fib_index=0,
+ ),
+ ],
+ thread_index=thread_index,
+ )
+ )
self.pg3.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
raise
else:
self.assertEqual(hanat.sequence_number, 1)
- self.assertEqual(hanat.flags, 'ACK')
+ self.assertEqual(hanat.flags, "ACK")
self.assertEqual(hanat.version, 1)
self.assertEqual(hanat.thread_index, thread_index)
- stats = self.statistics['/nat44-ei/ha/ack-send']
+ stats = self.statistics["/nat44-ei/ha/ack-send"]
self.assertEqual(stats[:, 0].sum(), 1)
- stats = self.statistics['/nat44-ei/ha/add-event-recv']
+ stats = self.statistics["/nat44-ei/ha/add-event-recv"]
self.assertEqual(stats[:, 0].sum(), 2)
- users = self.statistics['/nat44-ei/total-users']
+ users = self.statistics["/nat44-ei/total-users"]
self.assertEqual(users[:, 0].sum(), 1)
- sessions = self.statistics['/nat44-ei/total-sessions']
+ sessions = self.statistics["/nat44-ei/total-sessions"]
self.assertEqual(sessions[:, 0].sum(), 2)
users = self.vapi.nat44_ei_user_dump()
self.assertEqual(len(users), 1)
- self.assertEqual(str(users[0].ip_address),
- self.pg0.remote_ip4)
+ self.assertEqual(str(users[0].ip_address), self.pg0.remote_ip4)
# there should be 2 sessions created by HA
sessions = self.vapi.nat44_ei_user_session_dump(
- users[0].ip_address, users[0].vrf_id)
+ users[0].ip_address, users[0].vrf_id
+ )
self.assertEqual(len(sessions), 2)
for session in sessions:
- self.assertEqual(str(session.inside_ip_address),
- self.pg0.remote_ip4)
- self.assertEqual(str(session.outside_ip_address),
- self.nat_addr)
- self.assertIn(session.inside_port,
- [self.tcp_port_in, self.udp_port_in])
- self.assertIn(session.outside_port,
- [self.tcp_port_out, self.udp_port_out])
+ self.assertEqual(str(session.inside_ip_address), self.pg0.remote_ip4)
+ self.assertEqual(str(session.outside_ip_address), self.nat_addr)
+ self.assertIn(session.inside_port, [self.tcp_port_in, self.udp_port_in])
+ self.assertIn(session.outside_port, [self.tcp_port_out, self.udp_port_out])
self.assertIn(session.protocol, [IP_PROTOS.tcp, IP_PROTOS.udp])
# send HA session delete event to failover/passive
- p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
- IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
- UDP(sport=12346, dport=12345) /
- HANATStateSync(sequence_number=2, events=[
- Event(event_type='del', protocol='udp',
- in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
- in_port=self.udp_port_in, out_port=self.udp_port_out,
- eh_addr=self.pg1.remote_ip4,
- ehn_addr=self.pg1.remote_ip4,
- eh_port=self.udp_external_port,
- ehn_port=self.udp_external_port, fib_index=0)],
- thread_index=thread_index))
+ p = (
+ Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac)
+ / IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4)
+ / UDP(sport=12346, dport=12345)
+ / HANATStateSync(
+ sequence_number=2,
+ events=[
+ Event(
+ event_type="del",
+ protocol="udp",
+ in_addr=self.pg0.remote_ip4,
+ out_addr=self.nat_addr,
+ in_port=self.udp_port_in,
+ out_port=self.udp_port_out,
+ eh_addr=self.pg1.remote_ip4,
+ ehn_addr=self.pg1.remote_ip4,
+ eh_port=self.udp_external_port,
+ ehn_port=self.udp_external_port,
+ fib_index=0,
+ )
+ ],
+ thread_index=thread_index,
+ )
+ )
self.pg3.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
raise
else:
self.assertEqual(hanat.sequence_number, 2)
- self.assertEqual(hanat.flags, 'ACK')
+ self.assertEqual(hanat.flags, "ACK")
self.assertEqual(hanat.version, 1)
users = self.vapi.nat44_ei_user_dump()
self.assertEqual(len(users), 1)
- self.assertEqual(str(users[0].ip_address),
- self.pg0.remote_ip4)
+ self.assertEqual(str(users[0].ip_address), self.pg0.remote_ip4)
# now we should have only 1 session, 1 deleted by HA
- sessions = self.vapi.nat44_ei_user_session_dump(users[0].ip_address,
- users[0].vrf_id)
+ sessions = self.vapi.nat44_ei_user_session_dump(
+ users[0].ip_address, users[0].vrf_id
+ )
self.assertEqual(len(sessions), 1)
- stats = self.statistics['/nat44-ei/ha/del-event-recv']
+ stats = self.statistics["/nat44-ei/ha/del-event-recv"]
self.assertEqual(stats[:, 0].sum(), 1)
- stats = self.statistics.get_err_counter(
- '/err/nat44-ei-ha/pkts-processed')
+ stats = self.statistics.get_err_counter("/err/nat44-ei-ha/pkts-processed")
self.assertEqual(stats, 2)
# send HA session refresh event to failover/passive
- p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
- IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
- UDP(sport=12346, dport=12345) /
- HANATStateSync(sequence_number=3, events=[
- Event(event_type='refresh', protocol='tcp',
- in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
- in_port=self.tcp_port_in, out_port=self.tcp_port_out,
- eh_addr=self.pg1.remote_ip4,
- ehn_addr=self.pg1.remote_ip4,
- eh_port=self.tcp_external_port,
- ehn_port=self.tcp_external_port, fib_index=0,
- total_bytes=1024, total_pkts=2)],
- thread_index=thread_index))
+ p = (
+ Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac)
+ / IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4)
+ / UDP(sport=12346, dport=12345)
+ / HANATStateSync(
+ sequence_number=3,
+ events=[
+ Event(
+ event_type="refresh",
+ protocol="tcp",
+ in_addr=self.pg0.remote_ip4,
+ out_addr=self.nat_addr,
+ in_port=self.tcp_port_in,
+ out_port=self.tcp_port_out,
+ eh_addr=self.pg1.remote_ip4,
+ ehn_addr=self.pg1.remote_ip4,
+ eh_port=self.tcp_external_port,
+ ehn_port=self.tcp_external_port,
+ fib_index=0,
+ total_bytes=1024,
+ total_pkts=2,
+ )
+ ],
+ thread_index=thread_index,
+ )
+ )
self.pg3.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
raise
else:
self.assertEqual(hanat.sequence_number, 3)
- self.assertEqual(hanat.flags, 'ACK')
+ self.assertEqual(hanat.flags, "ACK")
self.assertEqual(hanat.version, 1)
users = self.vapi.nat44_ei_user_dump()
self.assertEqual(len(users), 1)
- self.assertEqual(str(users[0].ip_address),
- self.pg0.remote_ip4)
+ self.assertEqual(str(users[0].ip_address), self.pg0.remote_ip4)
sessions = self.vapi.nat44_ei_user_session_dump(
- users[0].ip_address, users[0].vrf_id)
+ users[0].ip_address, users[0].vrf_id
+ )
self.assertEqual(len(sessions), 1)
session = sessions[0]
self.assertEqual(session.total_bytes, 1024)
self.assertEqual(session.total_pkts, 2)
- stats = self.statistics['/nat44-ei/ha/refresh-event-recv']
+ stats = self.statistics["/nat44-ei/ha/refresh-event-recv"]
self.assertEqual(stats[:, 0].sum(), 1)
- stats = self.statistics.get_err_counter(
- '/err/nat44-ei-ha/pkts-processed')
+ stats = self.statistics.get_err_counter("/err/nat44-ei-ha/pkts-processed")
self.assertEqual(stats, 3)
# send packet to test session created by HA
- 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))
+ 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)
+ )
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
return self.vapi.nat44_ei_show_fq_options().frame_queue_nelts
def test_set_frame_queue_nelts(self):
- """ NAT44EI API test - worker handoff frame queue elements """
+ """NAT44EI API test - worker handoff frame queue elements"""
self.assertEqual(self.reconfigure_frame_queue_nelts(512), 512)
def show_commands_at_teardown(self):
self.logger.info(self.vapi.cli("show nat44 ei sessions detail"))
self.logger.info(self.vapi.cli("show nat44 ei hash tables detail"))
self.logger.info(self.vapi.cli("show nat44 ei ha"))
- self.logger.info(
- self.vapi.cli("show nat44 ei addr-port-assignment-alg"))
+ self.logger.info(self.vapi.cli("show nat44 ei addr-port-assignment-alg"))
def test_outside_address_distribution(self):
- """ Outside address distribution based on source address """
+ """Outside address distribution based on source address"""
x = 100
nat_addresses = []
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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.vapi.nat44_ei_add_del_address_range(
first_ip_address=nat_addresses[0],
last_ip_address=nat_addresses[-1],
- vrf_id=0xFFFFFFFF, is_add=1)
+ vrf_id=0xFFFFFFFF,
+ is_add=1,
+ )
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),
+ )
def test_default_user_sessions(self):
- """ NAT44EI default per-user session limit is used and reported """
+ """NAT44EI default per-user session limit is used and reported"""
nat44_ei_config = self.vapi.nat44_ei_show_running_config()
# a nonzero default should be reported for user_sessions
self.assertNotEqual(nat44_ei_config.user_sessions, 0)
class TestNAT44Out2InDPO(MethodHolder):
- """ NAT44EI Test Cases using out2in DPO """
+ """NAT44EI Test Cases using out2in DPO"""
@classmethod
def setUpClass(cls):
cls.udp_port_out = 6304
cls.icmp_id_in = 6305
cls.icmp_id_out = 6305
- cls.nat_addr = '10.0.0.3'
- cls.dst_ip4 = '192.168.70.1'
+ cls.nat_addr = "10.0.0.3"
+ cls.dst_ip4 = "192.168.70.1"
cls.create_pg_interfaces(range(2))
cls.pg1.config_ip6()
cls.pg1.resolve_ndp()
- r1 = VppIpRoute(cls, "::", 0,
- [VppRoutePath(cls.pg1.remote_ip6,
- cls.pg1.sw_if_index)],
- register=False)
+ r1 = VppIpRoute(
+ cls,
+ "::",
+ 0,
+ [VppRoutePath(cls.pg1.remote_ip6, cls.pg1.sw_if_index)],
+ register=False,
+ )
r1.add_vpp_config()
def setUp(self):
self.vapi.cli("clear logging")
def configure_xlat(self):
- self.dst_ip6_pfx = '1:2:3::'
- self.dst_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
- self.dst_ip6_pfx)
+ self.dst_ip6_pfx = "1:2:3::"
+ self.dst_ip6_pfx_n = socket.inet_pton(socket.AF_INET6, self.dst_ip6_pfx)
self.dst_ip6_pfx_len = 96
- self.src_ip6_pfx = '4:5:6::'
- self.src_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
- self.src_ip6_pfx)
+ self.src_ip6_pfx = "4:5:6::"
+ self.src_ip6_pfx_n = socket.inet_pton(socket.AF_INET6, self.src_ip6_pfx)
self.src_ip6_pfx_len = 96
- self.vapi.map_add_domain(self.dst_ip6_pfx_n, self.dst_ip6_pfx_len,
- self.src_ip6_pfx_n, self.src_ip6_pfx_len,
- '\x00\x00\x00\x00', 0)
-
- @unittest.skip('Temporary disabled')
+ self.vapi.map_add_domain(
+ self.dst_ip6_pfx_n,
+ self.dst_ip6_pfx_len,
+ self.src_ip6_pfx_n,
+ self.src_ip6_pfx_len,
+ "\x00\x00\x00\x00",
+ 0,
+ )
+
+ @unittest.skip("Temporary disabled")
def test_464xlat_ce(self):
- """ Test 464XLAT CE with NAT44EI """
+ """Test 464XLAT CE with NAT44EI"""
self.configure_xlat()
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_add_del_address_range(
first_ip_address=self.nat_addr_n,
last_ip_address=self.nat_addr_n,
- vrf_id=0xFFFFFFFF, is_add=1)
+ vrf_id=0xFFFFFFFF,
+ is_add=1,
+ )
- out_src_ip6 = self.compose_ip6(self.dst_ip4, self.dst_ip6_pfx,
- self.dst_ip6_pfx_len)
- out_dst_ip6 = self.compose_ip6(self.nat_addr, self.src_ip6_pfx,
- self.src_ip6_pfx_len)
+ out_src_ip6 = self.compose_ip6(
+ self.dst_ip4, self.dst_ip6_pfx, self.dst_ip6_pfx_len
+ )
+ out_dst_ip6 = self.compose_ip6(
+ self.nat_addr, self.src_ip6_pfx, self.src_ip6_pfx_len
+ )
try:
pkts = self.create_stream_in(self.pg0, self.pg1, self.dst_ip4)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out_ip6(capture, nat_ip=out_dst_ip6,
- dst_ip=out_src_ip6)
+ self.verify_capture_out_ip6(capture, nat_ip=out_dst_ip6, dst_ip=out_src_ip6)
- pkts = self.create_stream_out_ip6(self.pg1, out_src_ip6,
- out_dst_ip6)
+ pkts = self.create_stream_out_ip6(self.pg1, out_src_ip6, out_dst_ip6)
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.verify_capture_in(capture, self.pg0)
finally:
self.vapi.nat44_ei_interface_add_del_feature(
- sw_if_index=self.pg0.sw_if_index,
- flags=flags)
+ sw_if_index=self.pg0.sw_if_index, flags=flags
+ )
self.vapi.nat44_ei_add_del_address_range(
first_ip_address=self.nat_addr_n,
last_ip_address=self.nat_addr_n,
- vrf_id=0xFFFFFFFF)
+ vrf_id=0xFFFFFFFF,
+ )
- @unittest.skip('Temporary disabled')
+ @unittest.skip("Temporary disabled")
def test_464xlat_ce_no_nat(self):
- """ Test 464XLAT CE without NAT44EI """
+ """Test 464XLAT CE without NAT44EI"""
self.configure_xlat()
- out_src_ip6 = self.compose_ip6(self.dst_ip4, self.dst_ip6_pfx,
- self.dst_ip6_pfx_len)
- out_dst_ip6 = self.compose_ip6(self.pg0.remote_ip4, self.src_ip6_pfx,
- self.src_ip6_pfx_len)
+ out_src_ip6 = self.compose_ip6(
+ self.dst_ip4, self.dst_ip6_pfx, self.dst_ip6_pfx_len
+ )
+ out_dst_ip6 = self.compose_ip6(
+ self.pg0.remote_ip4, self.src_ip6_pfx, self.src_ip6_pfx_len
+ )
pkts = self.create_stream_in(self.pg0, self.pg1, self.dst_ip4)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out_ip6(capture, dst_ip=out_src_ip6,
- nat_ip=out_dst_ip6, same_port=True)
+ self.verify_capture_out_ip6(
+ capture, dst_ip=out_src_ip6, nat_ip=out_dst_ip6, same_port=True
+ )
pkts = self.create_stream_out_ip6(self.pg1, out_src_ip6, out_dst_ip6)
self.pg1.add_stream(pkts)
class TestNAT44EIMW(MethodHolder):
- """ NAT44EI Test Cases (multiple workers) """
+ """NAT44EI Test Cases (multiple workers)"""
+
vpp_worker_count = 2
max_translations = 10240
max_users = 10240
cls.udp_port_out = 6304
cls.icmp_id_in = 6305
cls.icmp_id_out = 6305
- cls.nat_addr = '10.0.0.3'
+ cls.nat_addr = "10.0.0.3"
cls.ipfix_src_port = 4739
cls.ipfix_domain_id = 1
cls.tcp_external_port = 80
cls.pg1.configure_ipv4_neighbors()
cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
- cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
- cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
+ cls.vapi.ip_table_add_del(is_add=1, table={"table_id": 10})
+ cls.vapi.ip_table_add_del(is_add=1, table={"table_id": 20})
cls.pg4._local_ip4 = "172.16.255.1"
cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
cls.pg9.generate_remote_hosts(2)
cls.pg9.config_ip4()
cls.vapi.sw_interface_add_del_address(
- sw_if_index=cls.pg9.sw_if_index,
- prefix="10.0.0.1/24")
+ sw_if_index=cls.pg9.sw_if_index, prefix="10.0.0.1/24"
+ )
cls.pg9.admin_up()
cls.pg9.resolve_arp()
def setUp(self):
super(TestNAT44EIMW, self).setUp()
self.vapi.nat44_ei_plugin_enable_disable(
- sessions=self.max_translations,
- users=self.max_users, enable=1)
+ sessions=self.max_translations, users=self.max_users, enable=1
+ )
def tearDown(self):
super(TestNAT44EIMW, self).tearDown()
if not self.vpp_dead:
self.vapi.nat44_ei_ipfix_enable_disable(
- domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port,
- enable=0)
+ domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port, enable=0
+ )
self.ipfix_src_port = 4739
self.ipfix_domain_id = 1
self.vapi.cli("clear logging")
def test_hairpinning(self):
- """ NAT44EI hairpinning - 1:1 NAPT """
+ """NAT44EI hairpinning - 1:1 NAPT"""
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# add static mapping for server
- self.nat44_add_static_mapping(server.ip4, self.nat_addr,
- server_in_port, server_out_port,
- proto=IP_PROTOS.tcp)
-
- cnt = self.statistics['/nat44-ei/hairpinning']
+ self.nat44_add_static_mapping(
+ server.ip4,
+ self.nat_addr,
+ server_in_port,
+ server_out_port,
+ proto=IP_PROTOS.tcp,
+ )
+
+ cnt = self.statistics["/nat44-ei/hairpinning"]
# send packet from host to server
- p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=self.nat_addr) /
- TCP(sport=host_in_port, dport=server_out_port))
+ p = (
+ Ether(src=host.mac, dst=self.pg0.local_mac)
+ / IP(src=host.ip4, dst=self.nat_addr)
+ / 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.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- after = self.statistics['/nat44-ei/hairpinning']
+ after = self.statistics["/nat44-ei/hairpinning"]
if_idx = self.pg0.sw_if_index
self.assertEqual(after[worker_2][if_idx] - cnt[worker_1][if_idx], 1)
# send reply from server to host
- p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
- IP(src=server.ip4, dst=self.nat_addr) /
- TCP(sport=server_in_port, dport=host_out_port))
+ p = (
+ Ether(src=server.mac, dst=self.pg0.local_mac)
+ / IP(src=server.ip4, dst=self.nat_addr)
+ / TCP(sport=server_in_port, dport=host_out_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
- after = self.statistics['/nat44-ei/hairpinning']
+ after = self.statistics["/nat44-ei/hairpinning"]
if_idx = self.pg0.sw_if_index
self.assertEqual(after[worker_1][if_idx] - cnt[worker_1][if_idx], 1)
self.assertEqual(after[worker_2][if_idx] - cnt[worker_2][if_idx], 2)
def test_hairpinning2(self):
- """ NAT44EI hairpinning - 1:1 NAT"""
+ """NAT44EI hairpinning - 1:1 NAT"""
server1_nat_ip = "10.0.0.10"
server2_nat_ip = "10.0.0.11"
self.nat44_add_address(self.nat_addr)
flags = self.config_flags.NAT44_EI_IF_INSIDE
self.vapi.nat44_ei_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_ei_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
+ )
# add static mapping for servers
self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
# host to server1
pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=host.ip4, dst=server1_nat_ip) /
- TCP(sport=self.tcp_port_in, dport=server_tcp_port))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=host.ip4, dst=server1_nat_ip)
+ / TCP(sport=self.tcp_port_in, dport=server_tcp_port)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=host.ip4, dst=server1_nat_ip) /
- UDP(sport=self.udp_port_in, dport=server_udp_port))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=host.ip4, dst=server1_nat_ip)
+ / UDP(sport=self.udp_port_in, dport=server_udp_port)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=host.ip4, dst=server1_nat_ip) /
- ICMP(id=self.icmp_id_in, type='echo-request'))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=host.ip4, dst=server1_nat_ip)
+ / ICMP(id=self.icmp_id_in, type="echo-request")
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
# server1 to host
pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.nat_addr) /
- TCP(sport=server_tcp_port, dport=self.tcp_port_out))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=self.nat_addr)
+ / TCP(sport=server_tcp_port, dport=self.tcp_port_out)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.nat_addr) /
- UDP(sport=server_udp_port, dport=self.udp_port_out))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=self.nat_addr)
+ / UDP(sport=server_udp_port, dport=self.udp_port_out)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.nat_addr) /
- ICMP(id=self.icmp_id_out, type='echo-reply'))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=self.nat_addr)
+ / ICMP(id=self.icmp_id_out, type="echo-reply")
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
# server2 to server1
pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server2.ip4, dst=server1_nat_ip) /
- TCP(sport=self.tcp_port_in, dport=server_tcp_port))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server2.ip4, dst=server1_nat_ip)
+ / TCP(sport=self.tcp_port_in, dport=server_tcp_port)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server2.ip4, dst=server1_nat_ip) /
- UDP(sport=self.udp_port_in, dport=server_udp_port))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server2.ip4, dst=server1_nat_ip)
+ / UDP(sport=self.udp_port_in, dport=server_udp_port)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server2.ip4, dst=server1_nat_ip) /
- ICMP(id=self.icmp_id_in, type='echo-request'))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server2.ip4, dst=server1_nat_ip)
+ / ICMP(id=self.icmp_id_in, type="echo-request")
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
# server1 to server2
pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=server2_nat_ip) /
- TCP(sport=server_tcp_port, dport=self.tcp_port_out))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=server2_nat_ip)
+ / TCP(sport=server_tcp_port, dport=self.tcp_port_out)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=server2_nat_ip) /
- UDP(sport=server_udp_port, dport=self.udp_port_out))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=server2_nat_ip)
+ / UDP(sport=server_udp_port, dport=self.udp_port_out)
+ )
pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=server2_nat_ip) /
- ICMP(id=self.icmp_id_out, type='echo-reply'))
+ p = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=server1.ip4, dst=server2_nat_ip)
+ / ICMP(id=self.icmp_id_out, type="echo-reply")
+ )
pkts.append(p)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
raise
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff