self.vapi.nat_set_reass()
self.vapi.nat_set_reass(is_ip6=1)
self.verify_no_nat44_user()
+ self.vapi.nat_set_timeouts()
+ self.vapi.nat_set_addr_and_port_alloc_alg()
+ self.vapi.nat_set_mss_clamping()
def nat44_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
local_port=0, external_port=0, vrf_id=0,
return pkts
def verify_capture_out(self, capture, nat_ip=None, same_port=False,
- packet_num=3, dst_ip=None, is_ip6=False):
+ dst_ip=None, is_ip6=False):
"""
Verify captured packets on outside network
:param capture: Captured packets
:param nat_ip: Translated IP address (Default use global NAT address)
:param same_port: Sorce port number is not translated (Default False)
- :param packet_num: Expected number of packets (Default 3)
:param dst_ip: Destination IP address (Default do not verify)
:param is_ip6: If L3 protocol is IPv6 (Default False)
"""
ICMP46 = ICMP
if nat_ip is None:
nat_ip = self.nat_addr
- self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
if not is_ip6:
raise
def verify_capture_out_ip6(self, capture, nat_ip, same_port=False,
- packet_num=3, dst_ip=None):
+ dst_ip=None):
"""
Verify captured packets on outside network
:param capture: Captured packets
:param nat_ip: Translated IP address
:param same_port: Sorce port number is not translated (Default False)
- :param packet_num: Expected number of packets (Default 3)
:param dst_ip: Destination IP address (Default do not verify)
"""
- return self.verify_capture_out(capture, nat_ip, same_port, packet_num,
- dst_ip, True)
+ return self.verify_capture_out(capture, nat_ip, same_port, dst_ip,
+ True)
- def verify_capture_in(self, capture, in_if, packet_num=3):
+ def verify_capture_in(self, capture, in_if):
"""
Verify captured packets on inside network
:param capture: Captured packets
:param in_if: Inside interface
- :param packet_num: Expected number of packets (Default 3)
"""
- self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
self.assert_packet_checksums_valid(packet)
"(inside network):", packet))
raise
- def verify_capture_in_ip6(self, capture, src_ip, dst_ip, packet_num=3):
+ def verify_capture_in_ip6(self, capture, src_ip, dst_ip):
"""
Verify captured IPv6 packets on inside network
:param capture: Captured packets
:param src_ip: Source IP
:param dst_ip: Destination IP address
- :param packet_num: Expected number of packets (Default 3)
"""
- self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
self.assertEqual(packet[IPv6].src, src_ip)
raise
def verify_capture_out_with_icmp_errors(self, capture, src_ip=None,
- packet_num=3, icmp_type=11):
+ icmp_type=11):
"""
Verify captured packets with ICMP errors on outside network
:param capture: Captured packets
:param src_ip: Translated IP address or IP address of VPP
(Default use global NAT address)
- :param packet_num: Expected number of packets (Default 3)
:param icmp_type: Type of error ICMP packet
we are expecting (Default 11)
"""
if src_ip is None:
src_ip = self.nat_addr
- self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
self.assertEqual(packet[IP].src, src_ip)
"(outside network):", packet))
raise
- def verify_capture_in_with_icmp_errors(self, capture, in_if, packet_num=3,
- icmp_type=11):
+ def verify_capture_in_with_icmp_errors(self, capture, in_if, icmp_type=11):
"""
Verify captured packets with ICMP errors on inside network
:param capture: Captured packets
:param in_if: Inside interface
- :param packet_num: Expected number of packets (Default 3)
:param icmp_type: Type of error ICMP packet
we are expecting (Default 11)
"""
- self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
self.assertEqual(packet[IP].dst, in_if.remote_ip4)
"(inside network):", packet))
raise
- def create_stream_frag(self, src_if, dst, sport, dport, data):
+ def create_stream_frag(self, src_if, dst, sport, dport, data,
+ proto=IP_PROTOS.tcp, echo_reply=False):
"""
Create fragmented packet stream
:param src_if: Source interface
:param dst: Destination IPv4 address
- :param sport: Source TCP port
- :param dport: Destination TCP port
+ :param sport: Source port
+ :param dport: Destination port
:param data: Payload data
+ :param proto: protocol (TCP, UDP, ICMP)
+ :param echo_reply: use echo_reply if protocol is ICMP
:returns: Fragmets
"""
+ if proto == IP_PROTOS.tcp:
+ p = (IP(src=src_if.remote_ip4, dst=dst) /
+ TCP(sport=sport, dport=dport) /
+ Raw(data))
+ p = p.__class__(str(p))
+ chksum = p['TCP'].chksum
+ proto_header = TCP(sport=sport, dport=dport, chksum=chksum)
+ elif proto == IP_PROTOS.udp:
+ proto_header = UDP(sport=sport, dport=dport)
+ elif proto == IP_PROTOS.icmp:
+ if not echo_reply:
+ proto_header = ICMP(id=sport, type='echo-request')
+ else:
+ proto_header = ICMP(id=sport, type='echo-reply')
+ else:
+ raise Exception("Unsupported protocol")
id = random.randint(0, 65535)
- p = (IP(src=src_if.remote_ip4, dst=dst) /
- TCP(sport=sport, dport=dport) /
- Raw(data))
- p = p.__class__(str(p))
- chksum = p['TCP'].chksum
pkts = []
+ if proto == IP_PROTOS.tcp:
+ 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) /
- TCP(sport=sport, dport=dport, chksum=chksum) /
- Raw(data[0:4]))
+ 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=IP_PROTOS.tcp) /
- Raw(data[4:20]))
+ 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=IP_PROTOS.tcp,
+ IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=proto,
id=id) /
- Raw(data[20:]))
+ raw)
pkts.append(p)
return pkts
self.assert_ip_checksum_valid(p)
buffer.seek(p[IP].frag * 8)
buffer.write(p[IP].payload)
- ip = frags[0].getlayer(IP)
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()))
self.assert_tcp_checksum_valid(p)
elif ip.proto == IP_PROTOS.udp:
- p = (ip / UDP(buffer.getvalue()))
+ p = (ip / UDP(buffer.getvalue()[:8]) /
+ Raw(buffer.getvalue()[8:]))
+ elif ip.proto == IP_PROTOS.icmp:
+ p = (ip / ICMP(buffer.getvalue()))
return p
def reass_frags_and_verify_ip6(self, frags, src, dst):
users = self.vapi.nat44_user_dump()
self.assertEqual(len(users), 0)
+ def verify_ipfix_max_entries_per_user(self, data, limit, src_addr):
+ """
+ Verify IPFIX maximum entries per user exceeded event
+
+ :param data: Decoded IPFIX data records
+ :param limit: Number of maximum entries per user
+ :param src_addr: IPv4 source address
+ """
+ self.assertEqual(1, len(data))
+ record = data[0]
+ # natEvent
+ self.assertEqual(ord(record[230]), 13)
+ # natQuotaExceededEvent
+ self.assertEqual(struct.pack("I", 3), record[466])
+ # maxEntriesPerUser
+ self.assertEqual(struct.pack("I", limit), record[473])
+ # sourceIPv4Address
+ self.assertEqual(src_addr, record[8])
+
+ def verify_mss_value(self, pkt, mss):
+ """
+ Verify TCP MSS value
+
+ :param pkt:
+ :param 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':
+ self.assertEqual(option[1], mss)
+ self.assert_tcp_checksum_valid(pkt)
+
+ @staticmethod
+ def proto2layer(proto):
+ if proto == IP_PROTOS.tcp:
+ return TCP
+ elif proto == IP_PROTOS.udp:
+ return UDP
+ elif proto == IP_PROTOS.icmp:
+ return ICMP
+ else:
+ raise Exception("Unsupported protocol")
+
+ def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False):
+ layer = self.proto2layer(proto)
+
+ if proto == IP_PROTOS.tcp:
+ data = "A" * 4 + "B" * 16 + "C" * 3
+ else:
+ data = "A" * 16 + "B" * 16 + "C" * 3
+ self.port_in = random.randint(1025, 65535)
+
+ reass = self.vapi.nat_reass_dump()
+ reass_n_start = len(reass)
+
+ # in2out
+ 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)
+ else:
+ 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)
+ self.assertNotEqual(p[layer].sport, self.port_in)
+ else:
+ self.assertEqual(p[layer].sport, self.port_in)
+ else:
+ if not dont_translate:
+ self.assertNotEqual(p[layer].id, self.port_in)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
+ self.assertEqual(data, p[Raw].load)
+
+ # out2in
+ if not dont_translate:
+ dst_addr = self.nat_addr
+ else:
+ dst_addr = self.pg0.remote_ip4
+ if proto != IP_PROTOS.icmp:
+ sport = 20
+ dport = p[layer].sport
+ else:
+ sport = p[layer].id
+ dport = 0
+ 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)
+ if proto != IP_PROTOS.icmp:
+ self.assertEqual(p[layer].sport, 20)
+ self.assertEqual(p[layer].dport, self.port_in)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
+ self.assertEqual(data, p[Raw].load)
+
+ reass = self.vapi.nat_reass_dump()
+ reass_n_end = len(reass)
+
+ self.assertEqual(reass_n_end - reass_n_start, 2)
+
+ def frag_in_order_in_plus_out(self, proto=IP_PROTOS.tcp):
+ layer = self.proto2layer(proto)
+
+ if proto == IP_PROTOS.tcp:
+ data = "A" * 4 + "B" * 16 + "C" * 3
+ else:
+ data = "A" * 16 + "B" * 16 + "C" * 3
+ self.port_in = random.randint(1025, 65535)
+
+ for i in range(2):
+ reass = self.vapi.nat_reass_dump()
+ reass_n_start = len(reass)
+
+ # out2in
+ pkts = self.create_stream_frag(self.pg0,
+ self.server_out_addr,
+ self.port_in,
+ self.server_out_port,
+ data,
+ proto)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ frags = self.pg1.get_capture(len(pkts))
+ p = self.reass_frags_and_verify(frags,
+ self.pg0.remote_ip4,
+ self.server_in_addr)
+ if proto != IP_PROTOS.icmp:
+ self.assertEqual(p[layer].sport, self.port_in)
+ self.assertEqual(p[layer].dport, self.server_in_port)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
+ self.assertEqual(data, p[Raw].load)
+
+ # in2out
+ if proto != IP_PROTOS.icmp:
+ pkts = self.create_stream_frag(self.pg1,
+ self.pg0.remote_ip4,
+ self.server_in_port,
+ p[layer].sport,
+ data,
+ proto)
+ else:
+ pkts = self.create_stream_frag(self.pg1,
+ self.pg0.remote_ip4,
+ p[layer].id,
+ 0,
+ data,
+ proto,
+ echo_reply=True)
+ 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.server_out_addr,
+ self.pg0.remote_ip4)
+ if proto != IP_PROTOS.icmp:
+ self.assertEqual(p[layer].sport, self.server_out_port)
+ self.assertEqual(p[layer].dport, self.port_in)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
+ self.assertEqual(data, p[Raw].load)
+
+ reass = self.vapi.nat_reass_dump()
+ reass_n_end = len(reass)
+
+ self.assertEqual(reass_n_end - reass_n_start, 2)
+
+ def reass_hairpinning(self, proto=IP_PROTOS.tcp):
+ layer = self.proto2layer(proto)
+
+ if proto == IP_PROTOS.tcp:
+ data = "A" * 4 + "B" * 16 + "C" * 3
+ else:
+ data = "A" * 16 + "B" * 16 + "C" * 3
+
+ # send packet from host to server
+ pkts = self.create_stream_frag(self.pg0,
+ self.nat_addr,
+ self.host_in_port,
+ self.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,
+ self.server.ip4)
+ if proto != IP_PROTOS.icmp:
+ self.assertNotEqual(p[layer].sport, self.host_in_port)
+ self.assertEqual(p[layer].dport, self.server_in_port)
+ else:
+ self.assertNotEqual(p[layer].id, self.host_in_port)
+ self.assertEqual(data, p[Raw].load)
+
+ def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False):
+ layer = self.proto2layer(proto)
+
+ if proto == IP_PROTOS.tcp:
+ data = "A" * 4 + "B" * 16 + "C" * 3
+ else:
+ data = "A" * 16 + "B" * 16 + "C" * 3
+ self.port_in = random.randint(1025, 65535)
+
+ for i in range(2):
+ # in2out
+ 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)
+ else:
+ 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)
+ self.assertNotEqual(p[layer].sport, self.port_in)
+ else:
+ self.assertEqual(p[layer].sport, self.port_in)
+ else:
+ if not dont_translate:
+ self.assertNotEqual(p[layer].id, self.port_in)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
+ self.assertEqual(data, p[Raw].load)
+
+ # out2in
+ if not dont_translate:
+ dst_addr = self.nat_addr
+ else:
+ dst_addr = self.pg0.remote_ip4
+ if proto != IP_PROTOS.icmp:
+ sport = 20
+ dport = p[layer].sport
+ else:
+ sport = p[layer].id
+ dport = 0
+ 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)
+ if proto != IP_PROTOS.icmp:
+ self.assertEqual(p[layer].sport, 20)
+ self.assertEqual(p[layer].dport, self.port_in)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
+ self.assertEqual(data, p[Raw].load)
+
+ def frag_out_of_order_in_plus_out(self, proto=IP_PROTOS.tcp):
+ layer = self.proto2layer(proto)
+
+ if proto == IP_PROTOS.tcp:
+ data = "A" * 4 + "B" * 16 + "C" * 3
+ else:
+ data = "A" * 16 + "B" * 16 + "C" * 3
+ self.port_in = random.randint(1025, 65535)
+
+ for i in range(2):
+ # out2in
+ pkts = self.create_stream_frag(self.pg0,
+ self.server_out_addr,
+ self.port_in,
+ self.server_out_port,
+ data,
+ proto)
+ pkts.reverse()
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ frags = self.pg1.get_capture(len(pkts))
+ p = self.reass_frags_and_verify(frags,
+ self.pg0.remote_ip4,
+ self.server_in_addr)
+ if proto != IP_PROTOS.icmp:
+ self.assertEqual(p[layer].dport, self.server_in_port)
+ self.assertEqual(p[layer].sport, self.port_in)
+ self.assertEqual(p[layer].dport, self.server_in_port)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
+ self.assertEqual(data, p[Raw].load)
+
+ # in2out
+ if proto != IP_PROTOS.icmp:
+ pkts = self.create_stream_frag(self.pg1,
+ self.pg0.remote_ip4,
+ self.server_in_port,
+ p[layer].sport,
+ data,
+ proto)
+ else:
+ pkts = self.create_stream_frag(self.pg1,
+ self.pg0.remote_ip4,
+ p[layer].id,
+ 0,
+ data,
+ proto,
+ echo_reply=True)
+ pkts.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.server_out_addr,
+ self.pg0.remote_ip4)
+ if proto != IP_PROTOS.icmp:
+ self.assertEqual(p[layer].sport, self.server_out_port)
+ self.assertEqual(p[layer].dport, self.port_in)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
+ self.assertEqual(data, p[Raw].load)
+
class TestNAT44(MethodHolder):
""" NAT44 Test Cases """
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(len(pkts))
- self.assertEqual(1, len(capture))
packet = capture[0]
try:
self.assertEqual(packet[IP].src, self.pg1.local_ip4)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg0.get_capture(1)
- self.verify_capture_in(capture, self.pg0, packet_num=1)
+ self.verify_capture_in(capture, self.pg0)
self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
# in2out
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(1)
- self.verify_capture_out(capture, same_port=True, packet_num=1)
+ self.verify_capture_out(capture, same_port=True)
self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
def test_forwarding(self):
def test_frag_in_order(self):
""" NAT44 translate fragments arriving in order """
+
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
def test_reass_hairpinning(self):
""" NAT44 fragments hairpinning """
+
server = self.pg0.remote_hosts[1]
host_in_port = random.randint(1025, 65535)
server_in_port = random.randint(1025, 65535)
def test_frag_out_of_order(self):
""" NAT44 translate fragments arriving out of order """
+
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
is_inside=0)
- self.vapi.cli("nat addr-port-assignment-alg map-e psid 10 "
- "psid-offset 6 psid-len 6")
+ self.vapi.nat_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) /
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
+ def test_port_range(self):
+ """ External address port range """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ self.vapi.nat_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))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(3)
+ for p in capture:
+ tcp = p[TCP]
+ self.assertGreaterEqual(tcp.sport, 1025)
+ self.assertLessEqual(tcp.sport, 1027)
+
def test_ipfix_max_frags(self):
""" IPFIX logging maximum fragments pending reassembly exceeded """
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
is_inside=0)
- self.vapi.nat_set_reass(max_frag=0)
+ self.vapi.nat_set_reass(max_frag=1)
self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
src_address=self.pg3.local_ip4n,
path_mtu=512,
self.tcp_port_in,
20,
data)
- self.pg0.add_stream(pkts[-1])
+ pkts.reverse()
+ self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.pg1.assert_nothing_captured()
for p in capture:
if p.haslayer(Data):
data = ipfix.decode_data_set(p.getlayer(Set))
- self.verify_ipfix_max_fragments_ip4(data, 0,
+ self.verify_ipfix_max_fragments_ip4(data, 1,
self.pg0.remote_ip4n)
def test_multiple_outside_vrf(self):
self.pg1.resolve_arp()
self.pg2.resolve_arp()
+ @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ def test_session_timeout(self):
+ """ NAT44 session timeouts """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ self.vapi.nat_set_timeouts(udp=5)
+
+ max_sessions = 1000
+ 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) /
+ UDP(sport=1025, dport=53))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.get_capture(max_sessions)
+
+ sleep(6)
+
+ 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) /
+ UDP(sport=1026, dport=53))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.get_capture(max_sessions)
+
+ nsessions = 0
+ users = self.vapi.nat44_user_dump()
+ for user in users:
+ nsessions = nsessions + user.nsessions
+ self.assertLess(nsessions, 2 * max_sessions)
+
+ def test_mss_clamping(self):
+ """ TCP MSS clamping """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+
+ 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.nat_set_mss_clamping(enable=1, mss_value=1000)
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ # Negotiated MSS value greater than configured - changed
+ self.verify_mss_value(capture[0], 1000)
+
+ self.vapi.nat_set_mss_clamping(enable=0)
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ # MSS clamping disabled - negotiated MSS unchanged
+ self.verify_mss_value(capture[0], 1400)
+
+ self.vapi.nat_set_mss_clamping(enable=1, mss_value=1500)
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ # Negotiated MSS value smaller than configured - unchanged
+ self.verify_mss_value(capture[0], 1400)
+
def tearDown(self):
super(TestNAT44, self).tearDown()
if not self.vpp_dead:
self.logger.info(self.vapi.cli("show nat44 sessions detail"))
self.logger.info(self.vapi.cli("show nat virtual-reassembly"))
self.logger.info(self.vapi.cli("show nat44 hash tables detail"))
- self.vapi.cli("nat addr-port-assignment-alg default")
+ self.logger.info(self.vapi.cli("show nat timeouts"))
+ self.logger.info(
+ self.vapi.cli("show nat addr-port-assignment-alg"))
self.clear_nat44()
self.vapi.cli("clear logging")
super(TestNAT44EndpointDependent, cls).tearDownClass()
raise
+ def test_frag_in_order(self):
+ """ NAT44 translate fragments arriving in order """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ 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_in_order_dont_translate(self):
+ """ NAT44 don't translate fragments arriving in order """
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ self.vapi.nat44_forwarding_enable_disable(enable=True)
+ self.frag_in_order(proto=IP_PROTOS.tcp, dont_translate=True)
+
+ def test_frag_out_of_order(self):
+ """ NAT44 translate fragments arriving out of order """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ 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_frag_out_of_order_dont_translate(self):
+ """ NAT44 don't translate fragments arriving out of order """
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ self.vapi.nat44_forwarding_enable_disable(enable=True)
+ self.frag_out_of_order(proto=IP_PROTOS.tcp, dont_translate=True)
+
+ def test_frag_in_order_in_plus_out(self):
+ """ in+out interface fragments in order """
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index,
+ is_inside=0)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+
+ self.server = self.pg1.remote_hosts[0]
+
+ self.server_in_addr = self.server.ip4
+ self.server_out_addr = '11.11.11.11'
+ self.server_in_port = random.randint(1025, 65535)
+ self.server_out_port = random.randint(1025, 65535)
+
+ self.nat44_add_address(self.server_out_addr)
+
+ # add static mappings for server
+ self.nat44_add_static_mapping(self.server_in_addr,
+ self.server_out_addr,
+ self.server_in_port,
+ self.server_out_port,
+ proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(self.server_in_addr,
+ self.server_out_addr,
+ self.server_in_port,
+ self.server_out_port,
+ proto=IP_PROTOS.udp)
+ self.nat44_add_static_mapping(self.server_in_addr,
+ self.server_out_addr,
+ proto=IP_PROTOS.icmp)
+
+ self.vapi.nat_set_reass(timeout=10)
+
+ self.frag_in_order_in_plus_out(proto=IP_PROTOS.tcp)
+ self.frag_in_order_in_plus_out(proto=IP_PROTOS.udp)
+ self.frag_in_order_in_plus_out(proto=IP_PROTOS.icmp)
+
+ def test_frag_out_of_order_in_plus_out(self):
+ """ in+out interface fragments out of order """
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index,
+ is_inside=0)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+
+ self.server = self.pg1.remote_hosts[0]
+
+ self.server_in_addr = self.server.ip4
+ self.server_out_addr = '11.11.11.11'
+ self.server_in_port = random.randint(1025, 65535)
+ self.server_out_port = random.randint(1025, 65535)
+
+ self.nat44_add_address(self.server_out_addr)
+
+ # add static mappings for server
+ self.nat44_add_static_mapping(self.server_in_addr,
+ self.server_out_addr,
+ self.server_in_port,
+ self.server_out_port,
+ proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(self.server_in_addr,
+ self.server_out_addr,
+ self.server_in_port,
+ self.server_out_port,
+ proto=IP_PROTOS.udp)
+ self.nat44_add_static_mapping(self.server_in_addr,
+ self.server_out_addr,
+ proto=IP_PROTOS.icmp)
+
+ self.vapi.nat_set_reass(timeout=10)
+
+ self.frag_out_of_order_in_plus_out(proto=IP_PROTOS.tcp)
+ self.frag_out_of_order_in_plus_out(proto=IP_PROTOS.udp)
+ self.frag_out_of_order_in_plus_out(proto=IP_PROTOS.icmp)
+
+ def test_reass_hairpinning(self):
+ """ NAT44 fragments hairpinning """
+ self.server = self.pg0.remote_hosts[1]
+ self.host_in_port = random.randint(1025, 65535)
+ self.server_in_port = random.randint(1025, 65535)
+ self.server_out_port = random.randint(1025, 65535)
+
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ # add static mapping for server
+ self.nat44_add_static_mapping(self.server.ip4, self.nat_addr,
+ self.server_in_port,
+ self.server_out_port,
+ proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(self.server.ip4, self.nat_addr,
+ self.server_in_port,
+ self.server_out_port,
+ proto=IP_PROTOS.udp)
+ self.nat44_add_static_mapping(self.server.ip4, self.nat_addr,
+ proto=IP_PROTOS.icmp)
+
+ self.reass_hairpinning(proto=IP_PROTOS.tcp)
+ self.reass_hairpinning(proto=IP_PROTOS.udp)
+ self.reass_hairpinning(proto=IP_PROTOS.icmp)
+
def test_dynamic(self):
""" NAT44 dynamic translation test """
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
+ def test_lb_affinity(self):
+ """ NAT44 local service load balancing affinity """
+ external_addr_n = socket.inet_pton(socket.AF_INET, self.nat_addr)
+ external_port = 80
+ local_port = 8080
+ server1 = self.pg0.remote_hosts[0]
+ server2 = self.pg0.remote_hosts[1]
+
+ locals = [{'addr': server1.ip4n,
+ 'port': local_port,
+ 'probability': 50,
+ 'vrf_id': 0},
+ {'addr': server2.ip4n,
+ 'port': local_port,
+ 'probability': 50,
+ 'vrf_id': 0}]
+
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_add_del_lb_static_mapping(external_addr_n,
+ external_port,
+ IP_PROTOS.tcp,
+ affinity=10800,
+ local_num=len(locals),
+ locals=locals)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
+ TCP(sport=1025, dport=external_port))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ backend = capture[0][IP].dst
+
+ sessions = self.vapi.nat44_user_session_dump(
+ socket.inet_pton(socket.AF_INET, backend), 0)
+ self.assertEqual(len(sessions), 1)
+ self.assertTrue(sessions[0].ext_host_valid)
+ self.vapi.nat44_del_session(
+ sessions[0].inside_ip_address,
+ sessions[0].inside_port,
+ sessions[0].protocol,
+ ext_host_address=sessions[0].ext_host_address,
+ ext_host_port=sessions[0].ext_host_port)
+
+ pkts = []
+ for port in range(1030, 1100):
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
+ TCP(sport=port, dport=external_port))
+ pkts.append(p)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for p in capture:
+ self.assertEqual(p[IP].dst, backend)
+
def test_unknown_proto(self):
""" NAT44 translate packet with unknown protocol """
self.nat44_add_address(self.nat_addr)
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
+ @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ def test_session_timeout(self):
+ """ NAT44 session timeouts """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ self.vapi.nat_set_timeouts(icmp=5)
+
+ max_sessions = 1000
+ 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) /
+ ICMP(id=1025, type='echo-request'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.get_capture(max_sessions)
+
+ sleep(10)
+
+ 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) /
+ ICMP(id=1026, type='echo-request'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.get_capture(max_sessions)
+
+ nsessions = 0
+ users = self.vapi.nat44_user_dump()
+ for user in users:
+ nsessions = nsessions + user.nsessions
+ self.assertLess(nsessions, 2 * max_sessions)
+
+ @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ def test_session_limit_per_user(self):
+ """ Maximum sessions per user limit """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ self.vapi.set_ipfix_exporter(collector_address=self.pg2.remote_ip4n,
+ src_address=self.pg2.local_ip4n,
+ path_mtu=512,
+ template_interval=10)
+ self.vapi.nat_set_timeouts(udp=5)
+
+ # get maximum number of translations per user
+ nat44_config = self.vapi.nat_show_config()
+
+ pkts = []
+ for port in range(0, nat44_config.max_translations_per_user):
+ p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ UDP(sport=1025 + port, dport=1025 + port))
+ pkts.append(p)
+
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+
+ self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port)
+
+ p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ UDP(sport=3001, dport=3002))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.assert_nothing_captured()
+
+ # verify IPFIX logging
+ self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ sleep(1)
+ capture = self.pg2.get_capture(10)
+ ipfix = IPFIXDecoder()
+ # first load template
+ for p in capture:
+ self.assertTrue(p.haslayer(IPFIX))
+ if p.haslayer(Template):
+ ipfix.add_template(p.getlayer(Template))
+ # verify events in data set
+ for p in capture:
+ if p.haslayer(Data):
+ data = ipfix.decode_data_set(p.getlayer(Set))
+ self.verify_ipfix_max_entries_per_user(
+ data,
+ nat44_config.max_translations_per_user,
+ self.pg0.remote_ip4n)
+
+ sleep(6)
+ p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ UDP(sport=3001, dport=3002))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.get_capture(1)
+
def tearDown(self):
super(TestNAT44EndpointDependent, self).tearDown()
if not self.vpp_dead:
self.logger.info(self.vapi.cli("show nat44 interface address"))
self.logger.info(self.vapi.cli("show nat44 sessions detail"))
self.logger.info(self.vapi.cli("show nat44 hash tables detail"))
+ self.logger.info(self.vapi.cli("show nat timeouts"))
self.clear_nat44()
self.vapi.cli("clear logging")
return pkts
- def verify_capture_out(self, capture, nat_ip=None, packet_num=3):
+ def verify_capture_out(self, capture, nat_ip=None):
"""
Verify captured packets on outside network
:param capture: Captured packets
:param nat_ip: Translated IP address (Default use global NAT address)
:param same_port: Sorce port number is not translated (Default False)
- :param packet_num: Expected number of packets (Default 3)
"""
if nat_ip is None:
nat_ip = self.nat_addr
- self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
self.assertEqual(packet[IP].src, nat_ip)
"(outside network):", packet))
raise
- def verify_ipfix_max_entries_per_user(self, data):
- """
- Verify IPFIX maximum entries per user exceeded event
-
- :param data: Decoded IPFIX data records
- """
- self.assertEqual(1, len(data))
- record = data[0]
- # natEvent
- self.assertEqual(ord(record[230]), 13)
- # natQuotaExceededEvent
- self.assertEqual('\x03\x00\x00\x00', record[466])
- # maxEntriesPerUser
- self.assertEqual('\xe8\x03\x00\x00', record[473])
- # sourceIPv4Address
- self.assertEqual(self.pg0.remote_ip4n, record[8])
-
def test_deterministic_mode(self):
""" NAT plugin run deterministic mode """
in_addr = '172.16.255.0'
def test_set_timeouts(self):
""" Set deterministic NAT timeouts """
- timeouts_before = self.vapi.nat_det_get_timeouts()
+ timeouts_before = self.vapi.nat_get_timeouts()
- self.vapi.nat_det_set_timeouts(timeouts_before.udp + 10,
- timeouts_before.tcp_established + 10,
- timeouts_before.tcp_transitory + 10,
- timeouts_before.icmp + 10)
+ self.vapi.nat_set_timeouts(timeouts_before.udp + 10,
+ timeouts_before.tcp_established + 10,
+ timeouts_before.tcp_transitory + 10,
+ timeouts_before.icmp + 10)
- timeouts_after = self.vapi.nat_det_get_timeouts()
+ timeouts_after = self.vapi.nat_get_timeouts()
self.assertNotEqual(timeouts_before.udp, timeouts_after.udp)
self.assertNotEqual(timeouts_before.icmp, timeouts_after.icmp)
is_inside=0)
self.initiate_tcp_session(self.pg0, self.pg1)
- self.vapi.nat_det_set_timeouts(5, 5, 5, 5)
+ self.vapi.nat_set_timeouts(5, 5, 5, 5)
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
for p in capture:
if p.haslayer(Data):
data = ipfix.decode_data_set(p.getlayer(Set))
- self.verify_ipfix_max_entries_per_user(data)
+ self.verify_ipfix_max_entries_per_user(data,
+ 1000,
+ self.pg0.remote_ip4n)
def clear_nat_det(self):
"""
Clear deterministic NAT configuration.
"""
self.vapi.nat_ipfix(enable=0)
- self.vapi.nat_det_set_timeouts()
+ self.vapi.nat_set_timeouts()
deterministic_mappings = self.vapi.nat_det_map_dump()
for dsm in deterministic_mappings:
self.vapi.nat_det_add_del_map(dsm.in_addr,
super(TestDeterministicNAT, self).tearDown()
if not self.vpp_dead:
self.logger.info(self.vapi.cli("show nat44 interfaces"))
+ self.logger.info(self.vapi.cli("show nat timeouts"))
self.logger.info(
self.vapi.cli("show nat44 deterministic mappings"))
- self.logger.info(
- self.vapi.cli("show nat44 deterministic timeouts"))
self.logger.info(
self.vapi.cli("show nat44 deterministic sessions"))
self.clear_nat_det()
# Wait for Neighbor Solicitation
capture = self.pg5.get_capture(len(pkts))
- self.assertEqual(1, len(capture))
packet = capture[0]
try:
self.assertEqual(packet[IPv6].src, self.pg5.local_ip6)
# Wait for ping reply
capture = self.pg5.get_capture(len(pkts))
- self.assertEqual(1, len(capture))
packet = capture[0]
try:
self.assertEqual(packet[IPv6].src, self.pg5.local_ip6)
def test_set_timeouts(self):
""" Set NAT64 timeouts """
# verify default values
- timeouts = self.vapi.nat64_get_timeouts()
+ timeouts = self.vapi.nat_get_timeouts()
self.assertEqual(timeouts.udp, 300)
self.assertEqual(timeouts.icmp, 60)
- self.assertEqual(timeouts.tcp_trans, 240)
- self.assertEqual(timeouts.tcp_est, 7440)
- self.assertEqual(timeouts.tcp_incoming_syn, 6)
+ self.assertEqual(timeouts.tcp_transitory, 240)
+ self.assertEqual(timeouts.tcp_established, 7440)
# set and verify custom values
- self.vapi.nat64_set_timeouts(udp=200, icmp=30, tcp_trans=250,
- tcp_est=7450, tcp_incoming_syn=10)
- timeouts = self.vapi.nat64_get_timeouts()
+ self.vapi.nat_set_timeouts(udp=200, icmp=30, tcp_transitory=250,
+ tcp_established=7450)
+ timeouts = self.vapi.nat_get_timeouts()
self.assertEqual(timeouts.udp, 200)
self.assertEqual(timeouts.icmp, 30)
- self.assertEqual(timeouts.tcp_trans, 250)
- self.assertEqual(timeouts.tcp_est, 7450)
- self.assertEqual(timeouts.tcp_incoming_syn, 10)
+ self.assertEqual(timeouts.tcp_transitory, 250)
+ self.assertEqual(timeouts.tcp_established, 7450)
def test_dynamic(self):
""" NAT64 dynamic translation test """
self.nat_addr_n)
self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
- self.vapi.nat64_set_timeouts(icmp=5, tcp_trans=5, tcp_est=5)
+ self.vapi.nat_set_timeouts(icmp=5, tcp_transitory=5, tcp_established=5)
pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.nat_addr_n)
self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
- self.vapi.nat_set_reass(max_frag=0, is_ip6=1)
+ self.vapi.nat_set_reass(max_frag=1, is_ip6=1)
self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
src_address=self.pg3.local_ip4n,
path_mtu=512,
data = 'a' * 200
pkts = self.create_stream_frag_ip6(self.pg0, self.pg1.remote_ip4,
self.tcp_port_in, 20, data)
- self.pg0.add_stream(pkts[-1])
+ pkts.reverse()
+ self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.pg1.assert_nothing_captured()
for p in capture:
if p.haslayer(Data):
data = ipfix.decode_data_set(p.getlayer(Set))
- self.verify_ipfix_max_fragments_ip6(data, 0,
+ self.verify_ipfix_max_fragments_ip6(data, 1,
self.pg0.remote_ip6n)
def test_ipfix_bib_ses(self):
self.ipfix_src_port = 4739
self.ipfix_domain_id = 1
- self.vapi.nat64_set_timeouts()
+ self.vapi.nat_set_timeouts()
interfaces = self.vapi.nat64_interface_dump()
for intf in interfaces:
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(1)
- self.assertEqual(1, len(capture))
capture = capture[0]
self.assertEqual(capture[IPv6].src, aftr_ip6)
self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(1)
- self.assertEqual(1, len(capture))
capture = capture[0]
self.assertEqual(capture[IPv6].src, b4_ip6)
self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)