5 from random import randint, choice
9 from framework import tag_fixme_ubuntu2204, is_distro_ubuntu2204
10 from framework import VppTestCase, VppTestRunner, VppLoInterface
11 from scapy.data import IP_PROTOS
12 from scapy.layers.inet import IP, TCP, UDP, ICMP, GRE
13 from scapy.layers.inet import IPerror, TCPerror
14 from scapy.layers.l2 import Ether
15 from scapy.packet import Raw
16 from syslog_rfc5424_parser import SyslogMessage, ParseError
17 from syslog_rfc5424_parser.constants import SyslogSeverity
18 from util import ppp, pr, ip4_range
19 from vpp_acl import AclRule, VppAcl, VppAclInterface
20 from vpp_ip_route import VppIpRoute, VppRoutePath
21 from vpp_papi import VppEnum
22 from util import StatsDiff
25 class TestNAT44ED(VppTestCase):
26 """NAT44ED Test Case"""
28 nat_addr = "10.0.10.3"
39 tcp_external_port = 80
52 def plugin_enable(self, max_sessions=None):
53 max_sessions = max_sessions or self.max_sessions
54 self.vapi.nat44_ed_plugin_enable_disable(sessions=max_sessions, enable=1)
56 def plugin_disable(self):
57 self.vapi.nat44_ed_plugin_enable_disable(enable=0)
60 def config_flags(self):
61 return VppEnum.vl_api_nat_config_flags_t
64 def nat44_config_flags(self):
65 return VppEnum.vl_api_nat44_config_flags_t
68 def syslog_severity(self):
69 return VppEnum.vl_api_syslog_severity_t
72 def server_addr(self):
73 return self.pg1.remote_hosts[0].ip4
77 return randint(1024, 65535)
80 def proto2layer(proto):
81 if proto == IP_PROTOS.tcp:
83 elif proto == IP_PROTOS.udp:
85 elif proto == IP_PROTOS.icmp:
88 raise Exception("Unsupported protocol")
91 def create_and_add_ip4_table(cls, i, table_id=0):
92 cls.vapi.ip_table_add_del(is_add=1, table={"table_id": table_id})
93 i.set_table_ip4(table_id)
96 def configure_ip4_interface(cls, i, hosts=0, table_id=None):
98 cls.create_and_add_ip4_table(i, table_id)
105 i.generate_remote_hosts(hosts)
106 i.configure_ipv4_neighbors()
109 def nat_add_interface_address(cls, i):
110 cls.vapi.nat44_add_del_interface_addr(sw_if_index=i.sw_if_index, is_add=1)
112 def nat_add_inside_interface(self, i):
113 self.vapi.nat44_interface_add_del_feature(
114 flags=self.config_flags.NAT_IS_INSIDE, sw_if_index=i.sw_if_index, is_add=1
117 def nat_add_outside_interface(self, i):
118 self.vapi.nat44_interface_add_del_feature(
119 flags=self.config_flags.NAT_IS_OUTSIDE, sw_if_index=i.sw_if_index, is_add=1
122 def nat_add_address(self, address, twice_nat=0, vrf_id=0xFFFFFFFF, is_add=1):
123 flags = self.config_flags.NAT_IS_TWICE_NAT if twice_nat else 0
124 self.vapi.nat44_add_del_address_range(
125 first_ip_address=address,
126 last_ip_address=address,
132 def nat_add_static_mapping(
135 external_ip="0.0.0.0",
140 external_sw_if_index=0xFFFFFFFF,
146 if not (local_port and external_port):
147 flags |= self.config_flags.NAT_IS_ADDR_ONLY
149 self.vapi.nat44_add_del_static_mapping(
151 local_ip_address=local_ip,
152 external_ip_address=external_ip,
153 external_sw_if_index=external_sw_if_index,
154 local_port=local_port,
155 external_port=external_port,
165 if is_distro_ubuntu2204 == True and not hasattr(cls, "vpp"):
168 cls.create_pg_interfaces(range(12))
169 cls.interfaces = list(cls.pg_interfaces[:4])
171 cls.create_and_add_ip4_table(cls.pg2, 10)
173 for i in cls.interfaces:
174 cls.configure_ip4_interface(i, hosts=3)
176 # test specific (test-multiple-vrf)
177 cls.vapi.ip_table_add_del(is_add=1, table={"table_id": 1})
179 # test specific (test-one-armed-nat44-static)
180 cls.pg4.generate_remote_hosts(2)
182 cls.vapi.sw_interface_add_del_address(
183 sw_if_index=cls.pg4.sw_if_index, prefix="10.0.0.1/24"
186 cls.pg4.resolve_arp()
187 cls.pg4._remote_hosts[1]._ip4 = cls.pg4._remote_hosts[0]._ip4
188 cls.pg4.resolve_arp()
190 # test specific interface (pg5)
191 cls.pg5._local_ip4 = "10.1.1.1"
192 cls.pg5._remote_hosts[0]._ip4 = "10.1.1.2"
193 cls.pg5.set_table_ip4(1)
196 cls.pg5.resolve_arp()
198 # test specific interface (pg6)
199 cls.pg6._local_ip4 = "10.1.2.1"
200 cls.pg6._remote_hosts[0]._ip4 = "10.1.2.2"
201 cls.pg6.set_table_ip4(1)
204 cls.pg6.resolve_arp()
213 [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=0)],
223 [VppRoutePath(cls.pg1.local_ip4, cls.pg1.sw_if_index)],
232 [VppRoutePath("0.0.0.0", cls.pg5.sw_if_index)],
242 [VppRoutePath("0.0.0.0", cls.pg6.sw_if_index)],
252 [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=1)],
261 cls.no_diff = StatsDiff(
264 "/nat44-ed/in2out/fastpath/tcp": 0,
265 "/nat44-ed/in2out/fastpath/udp": 0,
266 "/nat44-ed/in2out/fastpath/icmp": 0,
267 "/nat44-ed/in2out/fastpath/drops": 0,
268 "/nat44-ed/in2out/slowpath/tcp": 0,
269 "/nat44-ed/in2out/slowpath/udp": 0,
270 "/nat44-ed/in2out/slowpath/icmp": 0,
271 "/nat44-ed/in2out/slowpath/drops": 0,
272 "/nat44-ed/in2out/fastpath/tcp": 0,
273 "/nat44-ed/in2out/fastpath/udp": 0,
274 "/nat44-ed/in2out/fastpath/icmp": 0,
275 "/nat44-ed/in2out/fastpath/drops": 0,
276 "/nat44-ed/in2out/slowpath/tcp": 0,
277 "/nat44-ed/in2out/slowpath/udp": 0,
278 "/nat44-ed/in2out/slowpath/icmp": 0,
279 "/nat44-ed/in2out/slowpath/drops": 0,
281 for pg in cls.pg_interfaces
285 def get_err_counter(self, path):
286 return self.statistics.get_err_counter(path)
288 def reass_hairpinning(
297 layer = self.proto2layer(proto)
299 if proto == IP_PROTOS.tcp:
300 data = b"A" * 4 + b"B" * 16 + b"C" * 3
302 data = b"A" * 16 + b"B" * 16 + b"C" * 3
304 # send packet from host to server
305 pkts = self.create_stream_frag(
306 self.pg0, self.nat_addr, host_in_port, server_out_port, data, proto
308 self.pg0.add_stream(pkts)
309 self.pg_enable_capture(self.pg_interfaces)
311 frags = self.pg0.get_capture(len(pkts))
312 p = self.reass_frags_and_verify(frags, self.nat_addr, server_addr)
313 if proto != IP_PROTOS.icmp:
315 self.assertNotEqual(p[layer].sport, host_in_port)
316 self.assertEqual(p[layer].dport, server_in_port)
319 self.assertNotEqual(p[layer].id, host_in_port)
320 self.assertEqual(data, p[Raw].load)
322 def frag_out_of_order(
323 self, proto=IP_PROTOS.tcp, dont_translate=False, ignore_port=False
325 layer = self.proto2layer(proto)
327 if proto == IP_PROTOS.tcp:
328 data = b"A" * 4 + b"B" * 16 + b"C" * 3
330 data = b"A" * 16 + b"B" * 16 + b"C" * 3
331 self.port_in = self.random_port()
335 pkts = self.create_stream_frag(
336 self.pg0, self.pg1.remote_ip4, self.port_in, 20, data, proto
339 self.pg0.add_stream(pkts)
340 self.pg_enable_capture(self.pg_interfaces)
342 frags = self.pg1.get_capture(len(pkts))
343 if not dont_translate:
344 p = self.reass_frags_and_verify(
345 frags, self.nat_addr, self.pg1.remote_ip4
348 p = self.reass_frags_and_verify(
349 frags, self.pg0.remote_ip4, self.pg1.remote_ip4
351 if proto != IP_PROTOS.icmp:
352 if not dont_translate:
353 self.assertEqual(p[layer].dport, 20)
355 self.assertNotEqual(p[layer].sport, self.port_in)
357 self.assertEqual(p[layer].sport, self.port_in)
360 if not dont_translate:
361 self.assertNotEqual(p[layer].id, self.port_in)
363 self.assertEqual(p[layer].id, self.port_in)
364 self.assertEqual(data, p[Raw].load)
367 if not dont_translate:
368 dst_addr = self.nat_addr
370 dst_addr = self.pg0.remote_ip4
371 if proto != IP_PROTOS.icmp:
373 dport = p[layer].sport
377 pkts = self.create_stream_frag(
378 self.pg1, dst_addr, sport, dport, data, proto, echo_reply=True
381 self.pg1.add_stream(pkts)
382 self.pg_enable_capture(self.pg_interfaces)
383 self.logger.info(self.vapi.cli("show trace"))
385 frags = self.pg0.get_capture(len(pkts))
386 p = self.reass_frags_and_verify(
387 frags, self.pg1.remote_ip4, self.pg0.remote_ip4
389 if proto != IP_PROTOS.icmp:
390 self.assertEqual(p[layer].sport, 20)
391 self.assertEqual(p[layer].dport, self.port_in)
393 self.assertEqual(p[layer].id, self.port_in)
394 self.assertEqual(data, p[Raw].load)
396 def reass_frags_and_verify(self, frags, src, dst):
399 self.assertEqual(p[IP].src, src)
400 self.assertEqual(p[IP].dst, dst)
401 self.assert_ip_checksum_valid(p)
402 buffer.seek(p[IP].frag * 8)
403 buffer.write(bytes(p[IP].payload))
404 ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst, proto=frags[0][IP].proto)
405 if ip.proto == IP_PROTOS.tcp:
406 p = ip / TCP(buffer.getvalue())
407 self.logger.debug(ppp("Reassembled:", p))
408 self.assert_tcp_checksum_valid(p)
409 elif ip.proto == IP_PROTOS.udp:
410 p = ip / UDP(buffer.getvalue()[:8]) / Raw(buffer.getvalue()[8:])
411 elif ip.proto == IP_PROTOS.icmp:
412 p = ip / ICMP(buffer.getvalue())
416 self, proto=IP_PROTOS.tcp, dont_translate=False, ignore_port=False
418 layer = self.proto2layer(proto)
420 if proto == IP_PROTOS.tcp:
421 data = b"A" * 4 + b"B" * 16 + b"C" * 3
423 data = b"A" * 16 + b"B" * 16 + b"C" * 3
424 self.port_in = self.random_port()
427 pkts = self.create_stream_frag(
428 self.pg0, self.pg1.remote_ip4, self.port_in, 20, data, proto
430 self.pg0.add_stream(pkts)
431 self.pg_enable_capture(self.pg_interfaces)
433 frags = self.pg1.get_capture(len(pkts))
434 if not dont_translate:
435 p = self.reass_frags_and_verify(frags, self.nat_addr, self.pg1.remote_ip4)
437 p = self.reass_frags_and_verify(
438 frags, self.pg0.remote_ip4, self.pg1.remote_ip4
440 if proto != IP_PROTOS.icmp:
441 if not dont_translate:
442 self.assertEqual(p[layer].dport, 20)
444 self.assertNotEqual(p[layer].sport, self.port_in)
446 self.assertEqual(p[layer].sport, self.port_in)
449 if not dont_translate:
450 self.assertNotEqual(p[layer].id, self.port_in)
452 self.assertEqual(p[layer].id, self.port_in)
453 self.assertEqual(data, p[Raw].load)
456 if not dont_translate:
457 dst_addr = self.nat_addr
459 dst_addr = self.pg0.remote_ip4
460 if proto != IP_PROTOS.icmp:
462 dport = p[layer].sport
466 pkts = self.create_stream_frag(
467 self.pg1, dst_addr, sport, dport, data, proto, echo_reply=True
469 self.pg1.add_stream(pkts)
470 self.pg_enable_capture(self.pg_interfaces)
472 frags = self.pg0.get_capture(len(pkts))
473 p = self.reass_frags_and_verify(frags, self.pg1.remote_ip4, self.pg0.remote_ip4)
474 if proto != IP_PROTOS.icmp:
475 self.assertEqual(p[layer].sport, 20)
476 self.assertEqual(p[layer].dport, self.port_in)
478 self.assertEqual(p[layer].id, self.port_in)
479 self.assertEqual(data, p[Raw].load)
481 def verify_capture_out(
482 self, capture, nat_ip=None, same_port=False, dst_ip=None, ignore_port=False
485 nat_ip = self.nat_addr
486 for packet in capture:
488 self.assert_packet_checksums_valid(packet)
489 self.assertEqual(packet[IP].src, nat_ip)
490 if dst_ip is not None:
491 self.assertEqual(packet[IP].dst, dst_ip)
492 if packet.haslayer(TCP):
495 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
497 self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
498 self.tcp_port_out = packet[TCP].sport
499 self.assert_packet_checksums_valid(packet)
500 elif packet.haslayer(UDP):
503 self.assertEqual(packet[UDP].sport, self.udp_port_in)
505 self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
506 self.udp_port_out = packet[UDP].sport
510 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
512 self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
513 self.icmp_id_out = packet[ICMP].id
514 self.assert_packet_checksums_valid(packet)
517 ppp("Unexpected or invalid packet (outside network):", packet)
521 def verify_capture_in(self, capture, in_if):
522 for packet in capture:
524 self.assert_packet_checksums_valid(packet)
525 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
526 if packet.haslayer(TCP):
527 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
528 elif packet.haslayer(UDP):
529 self.assertEqual(packet[UDP].dport, self.udp_port_in)
531 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
534 ppp("Unexpected or invalid packet (inside network):", packet)
538 def create_stream_in(self, in_if, out_if, dst_ip=None, ttl=64):
540 dst_ip = out_if.remote_ip4
545 Ether(dst=in_if.local_mac, src=in_if.remote_mac)
546 / IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl)
547 / TCP(sport=self.tcp_port_in, dport=20)
553 Ether(dst=in_if.local_mac, src=in_if.remote_mac)
554 / IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl)
555 / UDP(sport=self.udp_port_in, dport=20)
561 Ether(dst=in_if.local_mac, src=in_if.remote_mac)
562 / IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl)
563 / ICMP(id=self.icmp_id_in, type="echo-request")
569 def create_stream_out(self, out_if, dst_ip=None, ttl=64, use_inside_ports=False):
571 dst_ip = self.nat_addr
572 if not use_inside_ports:
573 tcp_port = self.tcp_port_out
574 udp_port = self.udp_port_out
575 icmp_id = self.icmp_id_out
577 tcp_port = self.tcp_port_in
578 udp_port = self.udp_port_in
579 icmp_id = self.icmp_id_in
583 Ether(dst=out_if.local_mac, src=out_if.remote_mac)
584 / IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl)
585 / TCP(dport=tcp_port, sport=20)
591 Ether(dst=out_if.local_mac, src=out_if.remote_mac)
592 / IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl)
593 / UDP(dport=udp_port, sport=20)
599 Ether(dst=out_if.local_mac, src=out_if.remote_mac)
600 / IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl)
601 / ICMP(id=icmp_id, type="echo-reply")
607 def create_tcp_stream(self, in_if, out_if, count):
611 for i in range(count):
613 Ether(dst=in_if.local_mac, src=in_if.remote_mac)
614 / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=64)
615 / TCP(sport=port + i, dport=20)
621 def create_udp_stream(self, in_if, out_if, count, base_port=6303):
624 Ether(dst=in_if.local_mac, src=in_if.remote_mac)
625 / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=64)
626 / UDP(sport=base_port + i, dport=20)
628 for i in range(count)
631 def create_stream_frag(
632 self, src_if, dst, sport, dport, data, proto=IP_PROTOS.tcp, echo_reply=False
634 if proto == IP_PROTOS.tcp:
636 IP(src=src_if.remote_ip4, dst=dst)
637 / TCP(sport=sport, dport=dport)
640 p = p.__class__(scapy.compat.raw(p))
641 chksum = p[TCP].chksum
642 proto_header = TCP(sport=sport, dport=dport, chksum=chksum)
643 elif proto == IP_PROTOS.udp:
644 proto_header = UDP(sport=sport, dport=dport)
645 elif proto == IP_PROTOS.icmp:
647 proto_header = ICMP(id=sport, type="echo-request")
649 proto_header = ICMP(id=sport, type="echo-reply")
651 raise Exception("Unsupported protocol")
652 id = self.random_port()
654 if proto == IP_PROTOS.tcp:
657 raw = Raw(data[0:16])
659 Ether(src=src_if.remote_mac, dst=src_if.local_mac)
660 / IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=0, id=id)
665 if proto == IP_PROTOS.tcp:
666 raw = Raw(data[4:20])
668 raw = Raw(data[16:32])
670 Ether(src=src_if.remote_mac, dst=src_if.local_mac)
671 / IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=3, id=id, proto=proto)
675 if proto == IP_PROTOS.tcp:
680 Ether(src=src_if.remote_mac, dst=src_if.local_mac)
681 / IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=proto, id=id)
687 def frag_in_order_in_plus_out(
688 self, in_addr, out_addr, in_port, out_port, proto=IP_PROTOS.tcp
691 layer = self.proto2layer(proto)
693 if proto == IP_PROTOS.tcp:
694 data = b"A" * 4 + b"B" * 16 + b"C" * 3
696 data = b"A" * 16 + b"B" * 16 + b"C" * 3
697 port_in = self.random_port()
701 pkts = self.create_stream_frag(
702 self.pg0, out_addr, port_in, out_port, data, proto
704 self.pg0.add_stream(pkts)
705 self.pg_enable_capture(self.pg_interfaces)
707 frags = self.pg1.get_capture(len(pkts))
708 p = self.reass_frags_and_verify(frags, self.pg0.remote_ip4, in_addr)
709 if proto != IP_PROTOS.icmp:
710 self.assertEqual(p[layer].sport, port_in)
711 self.assertEqual(p[layer].dport, in_port)
713 self.assertEqual(p[layer].id, port_in)
714 self.assertEqual(data, p[Raw].load)
717 if proto != IP_PROTOS.icmp:
718 pkts = self.create_stream_frag(
719 self.pg1, self.pg0.remote_ip4, in_port, p[layer].sport, data, proto
722 pkts = self.create_stream_frag(
731 self.pg1.add_stream(pkts)
732 self.pg_enable_capture(self.pg_interfaces)
734 frags = self.pg0.get_capture(len(pkts))
735 p = self.reass_frags_and_verify(frags, out_addr, self.pg0.remote_ip4)
736 if proto != IP_PROTOS.icmp:
737 self.assertEqual(p[layer].sport, out_port)
738 self.assertEqual(p[layer].dport, port_in)
740 self.assertEqual(p[layer].id, port_in)
741 self.assertEqual(data, p[Raw].load)
743 def frag_out_of_order_in_plus_out(
744 self, in_addr, out_addr, in_port, out_port, proto=IP_PROTOS.tcp
747 layer = self.proto2layer(proto)
749 if proto == IP_PROTOS.tcp:
750 data = b"A" * 4 + b"B" * 16 + b"C" * 3
752 data = b"A" * 16 + b"B" * 16 + b"C" * 3
753 port_in = self.random_port()
757 pkts = self.create_stream_frag(
758 self.pg0, out_addr, port_in, out_port, data, proto
761 self.pg0.add_stream(pkts)
762 self.pg_enable_capture(self.pg_interfaces)
764 frags = self.pg1.get_capture(len(pkts))
765 p = self.reass_frags_and_verify(frags, self.pg0.remote_ip4, in_addr)
766 if proto != IP_PROTOS.icmp:
767 self.assertEqual(p[layer].dport, in_port)
768 self.assertEqual(p[layer].sport, port_in)
769 self.assertEqual(p[layer].dport, in_port)
771 self.assertEqual(p[layer].id, port_in)
772 self.assertEqual(data, p[Raw].load)
775 if proto != IP_PROTOS.icmp:
776 pkts = self.create_stream_frag(
777 self.pg1, self.pg0.remote_ip4, in_port, p[layer].sport, data, proto
780 pkts = self.create_stream_frag(
790 self.pg1.add_stream(pkts)
791 self.pg_enable_capture(self.pg_interfaces)
793 frags = self.pg0.get_capture(len(pkts))
794 p = self.reass_frags_and_verify(frags, out_addr, self.pg0.remote_ip4)
795 if proto != IP_PROTOS.icmp:
796 self.assertEqual(p[layer].sport, out_port)
797 self.assertEqual(p[layer].dport, port_in)
799 self.assertEqual(p[layer].id, port_in)
800 self.assertEqual(data, p[Raw].load)
802 def init_tcp_session(self, in_if, out_if, in_port, ext_port):
805 Ether(src=in_if.remote_mac, dst=in_if.local_mac)
806 / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4)
807 / TCP(sport=in_port, dport=ext_port, flags="S")
810 self.pg_enable_capture(self.pg_interfaces)
812 capture = out_if.get_capture(1)
814 out_port = p[TCP].sport
816 # SYN + ACK packet out->in
818 Ether(src=out_if.remote_mac, dst=out_if.local_mac)
819 / IP(src=out_if.remote_ip4, dst=self.nat_addr)
820 / TCP(sport=ext_port, dport=out_port, flags="SA")
823 self.pg_enable_capture(self.pg_interfaces)
829 Ether(src=in_if.remote_mac, dst=in_if.local_mac)
830 / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4)
831 / TCP(sport=in_port, dport=ext_port, flags="A")
834 self.pg_enable_capture(self.pg_interfaces)
836 out_if.get_capture(1)
840 def twice_nat_common(
841 self, self_twice_nat=False, same_pg=False, lb=False, client_id=None
843 twice_nat_addr = "10.0.1.3"
851 port_in1 = port_in + 1
852 port_in2 = port_in + 2
857 server1 = self.pg0.remote_hosts[0]
858 server2 = self.pg0.remote_hosts[1]
870 eh_translate = (not self_twice_nat) or (not lb and same_pg) or client_id == 1
872 self.nat_add_address(self.nat_addr)
873 self.nat_add_address(twice_nat_addr, twice_nat=1)
877 flags |= self.config_flags.NAT_IS_SELF_TWICE_NAT
879 flags |= self.config_flags.NAT_IS_TWICE_NAT
882 self.nat_add_static_mapping(
892 {"addr": server1.ip4, "port": port_in1, "probability": 50, "vrf_id": 0},
893 {"addr": server2.ip4, "port": port_in2, "probability": 50, "vrf_id": 0},
895 out_addr = self.nat_addr
897 self.vapi.nat44_add_del_lb_static_mapping(
900 external_addr=out_addr,
901 external_port=port_out,
902 protocol=IP_PROTOS.tcp,
903 local_num=len(locals),
906 self.nat_add_inside_interface(pg0)
907 self.nat_add_outside_interface(pg1)
913 assert client_id is not None
915 client = self.pg0.remote_hosts[0]
917 client = self.pg0.remote_hosts[1]
919 client = pg1.remote_hosts[0]
921 Ether(src=pg1.remote_mac, dst=pg1.local_mac)
922 / IP(src=client.ip4, dst=self.nat_addr)
923 / TCP(sport=eh_port_out, dport=port_out)
926 self.pg_enable_capture(self.pg_interfaces)
928 capture = pg0.get_capture(1)
934 if ip.dst == server1.ip4:
940 self.assertEqual(ip.dst, server.ip4)
942 self.assertIn(tcp.dport, [port_in1, port_in2])
944 self.assertEqual(tcp.dport, port_in)
946 self.assertEqual(ip.src, twice_nat_addr)
947 self.assertNotEqual(tcp.sport, eh_port_out)
949 self.assertEqual(ip.src, client.ip4)
950 self.assertEqual(tcp.sport, eh_port_out)
952 eh_port_in = tcp.sport
953 saved_port_in = tcp.dport
954 self.assert_packet_checksums_valid(p)
956 self.logger.error(ppp("Unexpected or invalid packet:", p))
960 Ether(src=server.mac, dst=pg0.local_mac)
961 / IP(src=server.ip4, dst=eh_addr_in)
962 / TCP(sport=saved_port_in, dport=eh_port_in)
965 self.pg_enable_capture(self.pg_interfaces)
967 capture = pg1.get_capture(1)
972 self.assertEqual(ip.dst, client.ip4)
973 self.assertEqual(ip.src, self.nat_addr)
974 self.assertEqual(tcp.dport, eh_port_out)
975 self.assertEqual(tcp.sport, port_out)
976 self.assert_packet_checksums_valid(p)
978 self.logger.error(ppp("Unexpected or invalid packet:", p))
982 sessions = self.vapi.nat44_user_session_dump(server.ip4, 0)
983 self.assertEqual(len(sessions), 1)
984 self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_EXT_HOST_VALID)
985 self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_TWICE_NAT)
986 self.logger.info(self.vapi.cli("show nat44 sessions"))
987 self.vapi.nat44_del_session(
988 address=sessions[0].inside_ip_address,
989 port=sessions[0].inside_port,
990 protocol=sessions[0].protocol,
992 self.config_flags.NAT_IS_INSIDE
993 | self.config_flags.NAT_IS_EXT_HOST_VALID
995 ext_host_address=sessions[0].ext_host_nat_address,
996 ext_host_port=sessions[0].ext_host_nat_port,
998 sessions = self.vapi.nat44_user_session_dump(server.ip4, 0)
999 self.assertEqual(len(sessions), 0)
1001 def verify_syslog_sess(self, data, msgid, is_ip6=False):
1002 message = data.decode("utf-8")
1004 message = SyslogMessage.parse(message)
1005 except ParseError as e:
1006 self.logger.error(e)
1009 self.assertEqual(message.severity, SyslogSeverity.info)
1010 self.assertEqual(message.appname, "NAT")
1011 self.assertEqual(message.msgid, msgid)
1012 sd_params = message.sd.get("nsess")
1013 self.assertTrue(sd_params is not None)
1015 self.assertEqual(sd_params.get("IATYP"), "IPv6")
1016 self.assertEqual(sd_params.get("ISADDR"), self.pg0.remote_ip6)
1018 self.assertEqual(sd_params.get("IATYP"), "IPv4")
1019 self.assertEqual(sd_params.get("ISADDR"), self.pg0.remote_ip4)
1020 self.assertTrue(sd_params.get("SSUBIX") is not None)
1021 self.assertEqual(sd_params.get("ISPORT"), "%d" % self.tcp_port_in)
1022 self.assertEqual(sd_params.get("XATYP"), "IPv4")
1023 self.assertEqual(sd_params.get("XSADDR"), self.nat_addr)
1024 self.assertEqual(sd_params.get("XSPORT"), "%d" % self.tcp_port_out)
1025 self.assertEqual(sd_params.get("PROTO"), "%d" % IP_PROTOS.tcp)
1026 self.assertEqual(sd_params.get("SVLAN"), "0")
1027 self.assertEqual(sd_params.get("XDADDR"), self.pg1.remote_ip4)
1028 self.assertEqual(sd_params.get("XDPORT"), "%d" % self.tcp_external_port)
1030 def test_icmp_error(self):
1031 """NAT44ED test ICMP error message with inner header"""
1035 self.nat_add_address(self.nat_addr)
1036 self.nat_add_inside_interface(self.pg0)
1037 self.nat_add_outside_interface(self.pg1)
1039 # in2out (initiate connection)
1041 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
1042 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
1043 / UDP(sport=21, dport=20)
1045 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
1046 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
1047 / TCP(sport=21, dport=20, flags="S")
1049 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
1050 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
1051 / ICMP(type="echo-request", id=7777)
1055 capture = self.send_and_expect(self.pg0, p1, self.pg1)
1057 # out2in (send error message)
1059 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
1060 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
1061 / ICMP(type="dest-unreach", code="port-unreachable")
1066 capture = self.send_and_expect(self.pg1, p2, self.pg0)
1070 assert c[IP].dst == self.pg0.remote_ip4
1071 assert c[IPerror].src == self.pg0.remote_ip4
1072 except AssertionError as a:
1073 raise AssertionError(f"Packet {pr(c)} not translated properly") from a
1075 def test_icmp_echo_reply_trailer(self):
1076 """ICMP echo reply with ethernet trailer"""
1078 self.nat_add_address(self.nat_addr)
1079 self.nat_add_inside_interface(self.pg0)
1080 self.nat_add_outside_interface(self.pg1)
1084 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
1085 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
1086 / ICMP(type=8, id=0xABCD, seq=0)
1089 self.pg0.add_stream(p1)
1090 self.pg_enable_capture(self.pg_interfaces)
1092 c = self.pg1.get_capture(1)[0]
1094 self.logger.debug(self.vapi.cli("show trace"))
1098 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
1099 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr, id=0xEE59)
1100 / ICMP(type=0, id=c[ICMP].id, seq=0)
1103 # force checksum calculation
1104 p2 = p2.__class__(bytes(p2))
1106 self.logger.debug(ppp("Packet before modification:", p2))
1108 # hex representation of vss monitoring ethernet trailer
1109 # this seems to be just added to end of packet without modifying
1110 # IP or ICMP lengths / checksums
1111 p2 = p2 / Raw("\x00\x00\x52\x54\x00\x46\xab\x04\x84\x18")
1112 # change it so that IP/ICMP is unaffected
1115 self.logger.debug(ppp("Packet with added trailer:", p2))
1117 self.pg1.add_stream(p2)
1118 self.pg_enable_capture(self.pg_interfaces)
1121 self.pg0.get_capture(1)
1123 def test_users_dump(self):
1124 """NAT44ED API test - nat44_user_dump"""
1126 self.nat_add_address(self.nat_addr)
1127 self.nat_add_inside_interface(self.pg0)
1128 self.nat_add_outside_interface(self.pg1)
1130 self.vapi.nat44_forwarding_enable_disable(enable=1)
1132 local_ip = self.pg0.remote_ip4
1133 external_ip = self.nat_addr
1134 self.nat_add_static_mapping(local_ip, external_ip)
1136 users = self.vapi.nat44_user_dump()
1137 self.assertEqual(len(users), 0)
1139 # in2out - static mapping match
1141 pkts = self.create_stream_out(self.pg1)
1142 self.pg1.add_stream(pkts)
1143 self.pg_enable_capture(self.pg_interfaces)
1145 capture = self.pg0.get_capture(len(pkts))
1146 self.verify_capture_in(capture, self.pg0)
1148 pkts = self.create_stream_in(self.pg0, self.pg1)
1149 self.pg0.add_stream(pkts)
1150 self.pg_enable_capture(self.pg_interfaces)
1152 capture = self.pg1.get_capture(len(pkts))
1153 self.verify_capture_out(capture, same_port=True)
1155 users = self.vapi.nat44_user_dump()
1156 self.assertEqual(len(users), 1)
1157 static_user = users[0]
1158 self.assertEqual(static_user.nstaticsessions, 3)
1159 self.assertEqual(static_user.nsessions, 0)
1161 # in2out - no static mapping match (forwarding test)
1163 host0 = self.pg0.remote_hosts[0]
1164 self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1]
1166 pkts = self.create_stream_out(
1167 self.pg1, dst_ip=self.pg0.remote_ip4, use_inside_ports=True
1169 self.pg1.add_stream(pkts)
1170 self.pg_enable_capture(self.pg_interfaces)
1172 capture = self.pg0.get_capture(len(pkts))
1173 self.verify_capture_in(capture, self.pg0)
1175 pkts = self.create_stream_in(self.pg0, self.pg1)
1176 self.pg0.add_stream(pkts)
1177 self.pg_enable_capture(self.pg_interfaces)
1179 capture = self.pg1.get_capture(len(pkts))
1180 self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4, same_port=True)
1182 self.pg0.remote_hosts[0] = host0
1184 users = self.vapi.nat44_user_dump()
1185 self.assertEqual(len(users), 2)
1186 if str(users[0].ip_address) == self.pg0.remote_hosts[0].ip4:
1187 non_static_user = users[1]
1188 static_user = users[0]
1190 non_static_user = users[0]
1191 static_user = users[1]
1192 self.assertEqual(static_user.nstaticsessions, 3)
1193 self.assertEqual(static_user.nsessions, 0)
1194 self.assertEqual(non_static_user.nstaticsessions, 0)
1195 self.assertEqual(non_static_user.nsessions, 3)
1197 users = self.vapi.nat44_user_dump()
1198 self.assertEqual(len(users), 2)
1199 if str(users[0].ip_address) == self.pg0.remote_hosts[0].ip4:
1200 non_static_user = users[1]
1201 static_user = users[0]
1203 non_static_user = users[0]
1204 static_user = users[1]
1205 self.assertEqual(static_user.nstaticsessions, 3)
1206 self.assertEqual(static_user.nsessions, 0)
1207 self.assertEqual(non_static_user.nstaticsessions, 0)
1208 self.assertEqual(non_static_user.nsessions, 3)
1210 def test_frag_out_of_order_do_not_translate(self):
1211 """NAT44ED don't translate fragments arriving out of order"""
1212 self.nat_add_inside_interface(self.pg0)
1213 self.nat_add_outside_interface(self.pg1)
1214 self.vapi.nat44_forwarding_enable_disable(enable=True)
1215 self.frag_out_of_order(proto=IP_PROTOS.tcp, dont_translate=True)
1217 def test_forwarding(self):
1218 """NAT44ED forwarding test"""
1220 self.nat_add_inside_interface(self.pg0)
1221 self.nat_add_outside_interface(self.pg1)
1222 self.vapi.nat44_forwarding_enable_disable(enable=1)
1224 real_ip = self.pg0.remote_ip4
1225 alias_ip = self.nat_addr
1226 flags = self.config_flags.NAT_IS_ADDR_ONLY
1227 self.vapi.nat44_add_del_static_mapping(
1229 local_ip_address=real_ip,
1230 external_ip_address=alias_ip,
1231 external_sw_if_index=0xFFFFFFFF,
1236 # in2out - static mapping match
1238 pkts = self.create_stream_out(self.pg1)
1239 self.pg1.add_stream(pkts)
1240 self.pg_enable_capture(self.pg_interfaces)
1242 capture = self.pg0.get_capture(len(pkts))
1243 self.verify_capture_in(capture, self.pg0)
1245 pkts = self.create_stream_in(self.pg0, self.pg1)
1246 self.pg0.add_stream(pkts)
1247 self.pg_enable_capture(self.pg_interfaces)
1249 capture = self.pg1.get_capture(len(pkts))
1250 self.verify_capture_out(capture, same_port=True)
1252 # in2out - no static mapping match
1254 host0 = self.pg0.remote_hosts[0]
1255 self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1]
1257 pkts = self.create_stream_out(
1258 self.pg1, dst_ip=self.pg0.remote_ip4, use_inside_ports=True
1260 self.pg1.add_stream(pkts)
1261 self.pg_enable_capture(self.pg_interfaces)
1263 capture = self.pg0.get_capture(len(pkts))
1264 self.verify_capture_in(capture, self.pg0)
1266 pkts = self.create_stream_in(self.pg0, self.pg1)
1267 self.pg0.add_stream(pkts)
1268 self.pg_enable_capture(self.pg_interfaces)
1270 capture = self.pg1.get_capture(len(pkts))
1271 self.verify_capture_out(
1272 capture, nat_ip=self.pg0.remote_ip4, same_port=True
1275 self.pg0.remote_hosts[0] = host0
1277 user = self.pg0.remote_hosts[1]
1278 sessions = self.vapi.nat44_user_session_dump(user.ip4, 0)
1279 self.assertEqual(len(sessions), 3)
1280 self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_EXT_HOST_VALID)
1281 self.vapi.nat44_del_session(
1282 address=sessions[0].inside_ip_address,
1283 port=sessions[0].inside_port,
1284 protocol=sessions[0].protocol,
1286 self.config_flags.NAT_IS_INSIDE
1287 | self.config_flags.NAT_IS_EXT_HOST_VALID
1289 ext_host_address=sessions[0].ext_host_address,
1290 ext_host_port=sessions[0].ext_host_port,
1292 sessions = self.vapi.nat44_user_session_dump(user.ip4, 0)
1293 self.assertEqual(len(sessions), 2)
1296 self.vapi.nat44_forwarding_enable_disable(enable=0)
1297 flags = self.config_flags.NAT_IS_ADDR_ONLY
1298 self.vapi.nat44_add_del_static_mapping(
1300 local_ip_address=real_ip,
1301 external_ip_address=alias_ip,
1302 external_sw_if_index=0xFFFFFFFF,
1306 def test_output_feature_and_service2(self):
1307 """NAT44ED interface output feature and service host direct access"""
1308 self.vapi.nat44_forwarding_enable_disable(enable=1)
1309 self.nat_add_address(self.nat_addr)
1311 self.vapi.nat44_ed_add_del_output_interface(
1312 sw_if_index=self.pg1.sw_if_index, is_add=1
1315 # session initiated from service host - translate
1316 pkts = self.create_stream_in(self.pg0, self.pg1)
1317 self.pg0.add_stream(pkts)
1318 self.pg_enable_capture(self.pg_interfaces)
1320 capture = self.pg1.get_capture(len(pkts))
1321 self.verify_capture_out(capture, ignore_port=True)
1323 pkts = self.create_stream_out(self.pg1)
1324 self.pg1.add_stream(pkts)
1325 self.pg_enable_capture(self.pg_interfaces)
1327 capture = self.pg0.get_capture(len(pkts))
1328 self.verify_capture_in(capture, self.pg0)
1330 # session initiated from remote host - do not translate
1331 tcp_port_in = self.tcp_port_in
1332 udp_port_in = self.udp_port_in
1333 icmp_id_in = self.icmp_id_in
1335 self.tcp_port_in = 60303
1336 self.udp_port_in = 60304
1337 self.icmp_id_in = 60305
1340 pkts = self.create_stream_out(
1341 self.pg1, self.pg0.remote_ip4, use_inside_ports=True
1343 self.pg1.add_stream(pkts)
1344 self.pg_enable_capture(self.pg_interfaces)
1346 capture = self.pg0.get_capture(len(pkts))
1347 self.verify_capture_in(capture, self.pg0)
1349 pkts = self.create_stream_in(self.pg0, self.pg1)
1350 self.pg0.add_stream(pkts)
1351 self.pg_enable_capture(self.pg_interfaces)
1353 capture = self.pg1.get_capture(len(pkts))
1354 self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4, same_port=True)
1356 self.tcp_port_in = tcp_port_in
1357 self.udp_port_in = udp_port_in
1358 self.icmp_id_in = icmp_id_in
1360 def test_twice_nat(self):
1361 """NAT44ED Twice NAT"""
1362 self.twice_nat_common()
1364 def test_self_twice_nat_positive(self):
1365 """NAT44ED Self Twice NAT (positive test)"""
1366 self.twice_nat_common(self_twice_nat=True, same_pg=True)
1368 def test_self_twice_nat_lb_positive(self):
1369 """NAT44ED Self Twice NAT local service load balancing (positive test)"""
1370 self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True, client_id=1)
1372 def test_twice_nat_lb(self):
1373 """NAT44ED Twice NAT local service load balancing"""
1374 self.twice_nat_common(lb=True)
1376 def test_output_feature(self):
1377 """NAT44ED interface output feature (in2out postrouting)"""
1378 self.vapi.nat44_forwarding_enable_disable(enable=1)
1379 self.nat_add_address(self.nat_addr)
1381 self.nat_add_outside_interface(self.pg0)
1382 self.vapi.nat44_ed_add_del_output_interface(
1383 sw_if_index=self.pg1.sw_if_index, is_add=1
1387 pkts = self.create_stream_in(self.pg0, self.pg1)
1388 self.pg0.add_stream(pkts)
1389 self.pg_enable_capture(self.pg_interfaces)
1391 capture = self.pg1.get_capture(len(pkts))
1392 self.verify_capture_out(capture, ignore_port=True)
1393 self.logger.debug(self.vapi.cli("show trace"))
1396 pkts = self.create_stream_out(self.pg1)
1397 self.pg1.add_stream(pkts)
1398 self.pg_enable_capture(self.pg_interfaces)
1400 capture = self.pg0.get_capture(len(pkts))
1401 self.verify_capture_in(capture, self.pg0)
1402 self.logger.debug(self.vapi.cli("show trace"))
1405 pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
1406 self.pg0.add_stream(pkts)
1407 self.pg_enable_capture(self.pg_interfaces)
1409 capture = self.pg1.get_capture(len(pkts))
1410 self.verify_capture_out(capture, ignore_port=True)
1411 self.logger.debug(self.vapi.cli("show trace"))
1414 pkts = self.create_stream_out(self.pg1, ttl=2)
1415 self.pg1.add_stream(pkts)
1416 self.pg_enable_capture(self.pg_interfaces)
1418 capture = self.pg0.get_capture(len(pkts))
1419 self.verify_capture_in(capture, self.pg0)
1420 self.logger.debug(self.vapi.cli("show trace"))
1423 pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
1424 capture = self.send_and_expect_some(self.pg0, pkts, self.pg0)
1426 self.assertIn(ICMP, p)
1427 self.assertEqual(p[ICMP].type, 11) # 11 == time-exceeded
1429 def test_static_with_port_out2(self):
1430 """NAT44ED 1:1 NAPT asymmetrical rule"""
1435 self.vapi.nat44_forwarding_enable_disable(enable=1)
1436 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
1437 self.nat_add_static_mapping(
1438 self.pg0.remote_ip4,
1442 proto=IP_PROTOS.tcp,
1446 self.nat_add_inside_interface(self.pg0)
1447 self.nat_add_outside_interface(self.pg1)
1449 # from client to service
1451 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
1452 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
1453 / TCP(sport=12345, dport=external_port)
1455 self.pg1.add_stream(p)
1456 self.pg_enable_capture(self.pg_interfaces)
1458 capture = self.pg0.get_capture(1)
1463 self.assertEqual(ip.dst, self.pg0.remote_ip4)
1464 self.assertEqual(tcp.dport, local_port)
1465 self.assert_packet_checksums_valid(p)
1467 self.logger.error(ppp("Unexpected or invalid packet:", p))
1472 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
1473 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
1477 self.pg0.add_stream(p)
1478 self.pg_enable_capture(self.pg_interfaces)
1480 capture = self.pg1.get_capture(1)
1483 self.assertEqual(p[IP].src, self.nat_addr)
1485 self.assertEqual(inner.dst, self.nat_addr)
1486 self.assertEqual(inner[TCPerror].dport, external_port)
1488 self.logger.error(ppp("Unexpected or invalid packet:", p))
1491 # from service back to client
1493 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
1494 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
1495 / TCP(sport=local_port, dport=12345)
1497 self.pg0.add_stream(p)
1498 self.pg_enable_capture(self.pg_interfaces)
1500 capture = self.pg1.get_capture(1)
1505 self.assertEqual(ip.src, self.nat_addr)
1506 self.assertEqual(tcp.sport, external_port)
1507 self.assert_packet_checksums_valid(p)
1509 self.logger.error(ppp("Unexpected or invalid packet:", p))
1514 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
1515 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
1519 self.pg1.add_stream(p)
1520 self.pg_enable_capture(self.pg_interfaces)
1522 capture = self.pg0.get_capture(1)
1525 self.assertEqual(p[IP].dst, self.pg0.remote_ip4)
1527 self.assertEqual(inner.src, self.pg0.remote_ip4)
1528 self.assertEqual(inner[TCPerror].sport, local_port)
1530 self.logger.error(ppp("Unexpected or invalid packet:", p))
1533 # from client to server (no translation)
1535 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
1536 / IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4)
1537 / TCP(sport=12346, dport=local_port)
1539 self.pg1.add_stream(p)
1540 self.pg_enable_capture(self.pg_interfaces)
1542 capture = self.pg0.get_capture(1)
1547 self.assertEqual(ip.dst, self.pg0.remote_ip4)
1548 self.assertEqual(tcp.dport, local_port)
1549 self.assert_packet_checksums_valid(p)
1551 self.logger.error(ppp("Unexpected or invalid packet:", p))
1554 # from service back to client (no translation)
1556 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
1557 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
1558 / TCP(sport=local_port, dport=12346)
1560 self.pg0.add_stream(p)
1561 self.pg_enable_capture(self.pg_interfaces)
1563 capture = self.pg1.get_capture(1)
1568 self.assertEqual(ip.src, self.pg0.remote_ip4)
1569 self.assertEqual(tcp.sport, local_port)
1570 self.assert_packet_checksums_valid(p)
1572 self.logger.error(ppp("Unexpected or invalid packet:", p))
1575 def test_static_lb(self):
1576 """NAT44ED local service load balancing"""
1577 external_addr_n = self.nat_addr
1580 server1 = self.pg0.remote_hosts[0]
1581 server2 = self.pg0.remote_hosts[1]
1584 {"addr": server1.ip4, "port": local_port, "probability": 70, "vrf_id": 0},
1585 {"addr": server2.ip4, "port": local_port, "probability": 30, "vrf_id": 0},
1588 self.nat_add_address(self.nat_addr)
1589 self.vapi.nat44_add_del_lb_static_mapping(
1591 external_addr=external_addr_n,
1592 external_port=external_port,
1593 protocol=IP_PROTOS.tcp,
1594 local_num=len(locals),
1597 flags = self.config_flags.NAT_IS_INSIDE
1598 self.vapi.nat44_interface_add_del_feature(
1599 sw_if_index=self.pg0.sw_if_index, flags=flags, is_add=1
1601 self.vapi.nat44_interface_add_del_feature(
1602 sw_if_index=self.pg1.sw_if_index, is_add=1
1605 # from client to service
1607 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
1608 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
1609 / TCP(sport=12345, dport=external_port)
1611 self.pg1.add_stream(p)
1612 self.pg_enable_capture(self.pg_interfaces)
1614 capture = self.pg0.get_capture(1)
1620 self.assertIn(ip.dst, [server1.ip4, server2.ip4])
1621 if ip.dst == server1.ip4:
1625 self.assertEqual(tcp.dport, local_port)
1626 self.assert_packet_checksums_valid(p)
1628 self.logger.error(ppp("Unexpected or invalid packet:", p))
1631 # from service back to client
1633 Ether(src=server.mac, dst=self.pg0.local_mac)
1634 / IP(src=server.ip4, dst=self.pg1.remote_ip4)
1635 / TCP(sport=local_port, dport=12345)
1637 self.pg0.add_stream(p)
1638 self.pg_enable_capture(self.pg_interfaces)
1640 capture = self.pg1.get_capture(1)
1645 self.assertEqual(ip.src, self.nat_addr)
1646 self.assertEqual(tcp.sport, external_port)
1647 self.assert_packet_checksums_valid(p)
1649 self.logger.error(ppp("Unexpected or invalid packet:", p))
1652 sessions = self.vapi.nat44_user_session_dump(server.ip4, 0)
1653 self.assertEqual(len(sessions), 1)
1654 self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_EXT_HOST_VALID)
1655 self.vapi.nat44_del_session(
1656 address=sessions[0].inside_ip_address,
1657 port=sessions[0].inside_port,
1658 protocol=sessions[0].protocol,
1660 self.config_flags.NAT_IS_INSIDE
1661 | self.config_flags.NAT_IS_EXT_HOST_VALID
1663 ext_host_address=sessions[0].ext_host_address,
1664 ext_host_port=sessions[0].ext_host_port,
1666 sessions = self.vapi.nat44_user_session_dump(server.ip4, 0)
1667 self.assertEqual(len(sessions), 0)
1669 def test_static_lb_2(self):
1670 """NAT44ED local service load balancing (asymmetrical rule)"""
1671 external_addr = self.nat_addr
1674 server1 = self.pg0.remote_hosts[0]
1675 server2 = self.pg0.remote_hosts[1]
1678 {"addr": server1.ip4, "port": local_port, "probability": 70, "vrf_id": 0},
1679 {"addr": server2.ip4, "port": local_port, "probability": 30, "vrf_id": 0},
1682 self.vapi.nat44_forwarding_enable_disable(enable=1)
1683 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
1684 self.vapi.nat44_add_del_lb_static_mapping(
1687 external_addr=external_addr,
1688 external_port=external_port,
1689 protocol=IP_PROTOS.tcp,
1690 local_num=len(locals),
1693 flags = self.config_flags.NAT_IS_INSIDE
1694 self.vapi.nat44_interface_add_del_feature(
1695 sw_if_index=self.pg0.sw_if_index, flags=flags, is_add=1
1697 self.vapi.nat44_interface_add_del_feature(
1698 sw_if_index=self.pg1.sw_if_index, is_add=1
1701 # from client to service
1703 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
1704 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
1705 / TCP(sport=12345, dport=external_port)
1707 self.pg1.add_stream(p)
1708 self.pg_enable_capture(self.pg_interfaces)
1710 capture = self.pg0.get_capture(1)
1716 self.assertIn(ip.dst, [server1.ip4, server2.ip4])
1717 if ip.dst == server1.ip4:
1721 self.assertEqual(tcp.dport, local_port)
1722 self.assert_packet_checksums_valid(p)
1724 self.logger.error(ppp("Unexpected or invalid packet:", p))
1727 # from service back to client
1729 Ether(src=server.mac, dst=self.pg0.local_mac)
1730 / IP(src=server.ip4, dst=self.pg1.remote_ip4)
1731 / TCP(sport=local_port, dport=12345)
1733 self.pg0.add_stream(p)
1734 self.pg_enable_capture(self.pg_interfaces)
1736 capture = self.pg1.get_capture(1)
1741 self.assertEqual(ip.src, self.nat_addr)
1742 self.assertEqual(tcp.sport, external_port)
1743 self.assert_packet_checksums_valid(p)
1745 self.logger.error(ppp("Unexpected or invalid packet:", p))
1748 # from client to server (no translation)
1750 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
1751 / IP(src=self.pg1.remote_ip4, dst=server1.ip4)
1752 / TCP(sport=12346, dport=local_port)
1754 self.pg1.add_stream(p)
1755 self.pg_enable_capture(self.pg_interfaces)
1757 capture = self.pg0.get_capture(1)
1763 self.assertEqual(ip.dst, server1.ip4)
1764 self.assertEqual(tcp.dport, local_port)
1765 self.assert_packet_checksums_valid(p)
1767 self.logger.error(ppp("Unexpected or invalid packet:", p))
1770 # from service back to client (no translation)
1772 Ether(src=server1.mac, dst=self.pg0.local_mac)
1773 / IP(src=server1.ip4, dst=self.pg1.remote_ip4)
1774 / TCP(sport=local_port, dport=12346)
1776 self.pg0.add_stream(p)
1777 self.pg_enable_capture(self.pg_interfaces)
1779 capture = self.pg1.get_capture(1)
1784 self.assertEqual(ip.src, server1.ip4)
1785 self.assertEqual(tcp.sport, local_port)
1786 self.assert_packet_checksums_valid(p)
1788 self.logger.error(ppp("Unexpected or invalid packet:", p))
1791 def test_lb_affinity(self):
1792 """NAT44ED local service load balancing affinity"""
1793 external_addr = self.nat_addr
1796 server1 = self.pg0.remote_hosts[0]
1797 server2 = self.pg0.remote_hosts[1]
1800 {"addr": server1.ip4, "port": local_port, "probability": 50, "vrf_id": 0},
1801 {"addr": server2.ip4, "port": local_port, "probability": 50, "vrf_id": 0},
1804 self.nat_add_address(self.nat_addr)
1805 self.vapi.nat44_add_del_lb_static_mapping(
1807 external_addr=external_addr,
1808 external_port=external_port,
1809 protocol=IP_PROTOS.tcp,
1811 local_num=len(locals),
1814 flags = self.config_flags.NAT_IS_INSIDE
1815 self.vapi.nat44_interface_add_del_feature(
1816 sw_if_index=self.pg0.sw_if_index, flags=flags, is_add=1
1818 self.vapi.nat44_interface_add_del_feature(
1819 sw_if_index=self.pg1.sw_if_index, is_add=1
1823 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
1824 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
1825 / TCP(sport=1025, dport=external_port)
1827 self.pg1.add_stream(p)
1828 self.pg_enable_capture(self.pg_interfaces)
1830 capture = self.pg0.get_capture(1)
1831 backend = capture[0][IP].dst
1833 sessions = self.vapi.nat44_user_session_dump(backend, 0)
1834 self.assertEqual(len(sessions), 1)
1835 self.assertTrue(sessions[0].flags & self.config_flags.NAT_IS_EXT_HOST_VALID)
1836 self.vapi.nat44_del_session(
1837 address=sessions[0].inside_ip_address,
1838 port=sessions[0].inside_port,
1839 protocol=sessions[0].protocol,
1841 self.config_flags.NAT_IS_INSIDE
1842 | self.config_flags.NAT_IS_EXT_HOST_VALID
1844 ext_host_address=sessions[0].ext_host_address,
1845 ext_host_port=sessions[0].ext_host_port,
1849 for port in range(1030, 1100):
1851 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
1852 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
1853 / TCP(sport=port, dport=external_port)
1856 self.pg1.add_stream(pkts)
1857 self.pg_enable_capture(self.pg_interfaces)
1859 capture = self.pg0.get_capture(len(pkts))
1861 self.assertEqual(p[IP].dst, backend)
1863 def test_multiple_vrf_1(self):
1864 """Multiple VRF - both client & service in VRF1"""
1866 external_addr = "1.2.3.4"
1871 flags = self.config_flags.NAT_IS_INSIDE
1872 self.vapi.nat44_interface_add_del_feature(
1873 sw_if_index=self.pg5.sw_if_index, is_add=1
1875 self.vapi.nat44_interface_add_del_feature(
1876 sw_if_index=self.pg5.sw_if_index, is_add=1, flags=flags
1878 self.vapi.nat44_interface_add_del_feature(
1879 sw_if_index=self.pg6.sw_if_index, is_add=1
1881 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
1882 self.nat_add_static_mapping(
1883 self.pg5.remote_ip4,
1888 proto=IP_PROTOS.tcp,
1893 Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac)
1894 / IP(src=self.pg6.remote_ip4, dst=external_addr)
1895 / TCP(sport=12345, dport=external_port)
1897 self.pg6.add_stream(p)
1898 self.pg_enable_capture(self.pg_interfaces)
1900 capture = self.pg5.get_capture(1)
1905 self.assertEqual(ip.dst, self.pg5.remote_ip4)
1906 self.assertEqual(tcp.dport, local_port)
1907 self.assert_packet_checksums_valid(p)
1909 self.logger.error(ppp("Unexpected or invalid packet:", p))
1913 Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac)
1914 / IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4)
1915 / TCP(sport=local_port, dport=12345)
1917 self.pg5.add_stream(p)
1918 self.pg_enable_capture(self.pg_interfaces)
1920 capture = self.pg6.get_capture(1)
1925 self.assertEqual(ip.src, external_addr)
1926 self.assertEqual(tcp.sport, external_port)
1927 self.assert_packet_checksums_valid(p)
1929 self.logger.error(ppp("Unexpected or invalid packet:", p))
1932 def test_multiple_vrf_2(self):
1933 """Multiple VRF - dynamic NAT from VRF1 to VRF0 (output-feature)"""
1935 external_addr = "1.2.3.4"
1940 self.nat_add_address(self.nat_addr)
1941 flags = self.config_flags.NAT_IS_INSIDE
1942 self.vapi.nat44_ed_add_del_output_interface(
1943 sw_if_index=self.pg1.sw_if_index, is_add=1
1945 self.vapi.nat44_interface_add_del_feature(
1946 sw_if_index=self.pg5.sw_if_index, is_add=1
1948 self.vapi.nat44_interface_add_del_feature(
1949 sw_if_index=self.pg5.sw_if_index, is_add=1, flags=flags
1951 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
1952 self.nat_add_static_mapping(
1953 self.pg5.remote_ip4,
1958 proto=IP_PROTOS.tcp,
1963 Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac)
1964 / IP(src=self.pg5.remote_ip4, dst=self.pg1.remote_ip4)
1965 / TCP(sport=2345, dport=22)
1967 self.pg5.add_stream(p)
1968 self.pg_enable_capture(self.pg_interfaces)
1970 capture = self.pg1.get_capture(1)
1975 self.assertEqual(ip.src, self.nat_addr)
1976 self.assert_packet_checksums_valid(p)
1979 self.logger.error(ppp("Unexpected or invalid packet:", p))
1983 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
1984 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
1985 / TCP(sport=22, dport=port)
1987 self.pg1.add_stream(p)
1988 self.pg_enable_capture(self.pg_interfaces)
1990 capture = self.pg5.get_capture(1)
1995 self.assertEqual(ip.dst, self.pg5.remote_ip4)
1996 self.assertEqual(tcp.dport, 2345)
1997 self.assert_packet_checksums_valid(p)
1999 self.logger.error(ppp("Unexpected or invalid packet:", p))
2002 def test_multiple_vrf_3(self):
2003 """Multiple VRF - client in VRF1, service in VRF0"""
2005 external_addr = "1.2.3.4"
2010 flags = self.config_flags.NAT_IS_INSIDE
2011 self.vapi.nat44_interface_add_del_feature(
2012 sw_if_index=self.pg0.sw_if_index, is_add=1
2014 self.vapi.nat44_interface_add_del_feature(
2015 sw_if_index=self.pg0.sw_if_index, is_add=1, flags=flags
2017 self.vapi.nat44_interface_add_del_feature(
2018 sw_if_index=self.pg6.sw_if_index, is_add=1
2020 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
2021 self.nat_add_static_mapping(
2022 self.pg0.remote_ip4,
2023 external_sw_if_index=self.pg0.sw_if_index,
2024 local_port=local_port,
2026 external_port=external_port,
2027 proto=IP_PROTOS.tcp,
2031 # from client VRF1 to service VRF0
2033 Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac)
2034 / IP(src=self.pg6.remote_ip4, dst=self.pg0.local_ip4)
2035 / TCP(sport=12346, dport=external_port)
2037 self.pg6.add_stream(p)
2038 self.pg_enable_capture(self.pg_interfaces)
2040 capture = self.pg0.get_capture(1)
2045 self.assertEqual(ip.dst, self.pg0.remote_ip4)
2046 self.assertEqual(tcp.dport, local_port)
2047 self.assert_packet_checksums_valid(p)
2049 self.logger.error(ppp("Unexpected or invalid packet:", p))
2052 # from service VRF0 back to client VRF1
2054 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
2055 / IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4)
2056 / TCP(sport=local_port, dport=12346)
2058 self.pg0.add_stream(p)
2059 self.pg_enable_capture(self.pg_interfaces)
2061 capture = self.pg6.get_capture(1)
2066 self.assertEqual(ip.src, self.pg0.local_ip4)
2067 self.assertEqual(tcp.sport, external_port)
2068 self.assert_packet_checksums_valid(p)
2070 self.logger.error(ppp("Unexpected or invalid packet:", p))
2073 def test_multiple_vrf_4(self):
2074 """Multiple VRF - client in VRF0, service in VRF1"""
2076 external_addr = "1.2.3.4"
2081 flags = self.config_flags.NAT_IS_INSIDE
2082 self.vapi.nat44_interface_add_del_feature(
2083 sw_if_index=self.pg0.sw_if_index, is_add=1
2085 self.vapi.nat44_interface_add_del_feature(
2086 sw_if_index=self.pg0.sw_if_index, is_add=1, flags=flags
2088 self.vapi.nat44_interface_add_del_feature(
2089 sw_if_index=self.pg5.sw_if_index, is_add=1
2091 self.vapi.nat44_interface_add_del_feature(
2092 sw_if_index=self.pg5.sw_if_index, is_add=1, flags=flags
2094 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
2095 self.nat_add_static_mapping(
2096 self.pg5.remote_ip4,
2101 proto=IP_PROTOS.tcp,
2105 # from client VRF0 to service VRF1
2107 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
2108 / IP(src=self.pg0.remote_ip4, dst=external_addr)
2109 / TCP(sport=12347, dport=external_port)
2111 self.pg0.add_stream(p)
2112 self.pg_enable_capture(self.pg_interfaces)
2114 capture = self.pg5.get_capture(1)
2119 self.assertEqual(ip.dst, self.pg5.remote_ip4)
2120 self.assertEqual(tcp.dport, local_port)
2121 self.assert_packet_checksums_valid(p)
2123 self.logger.error(ppp("Unexpected or invalid packet:", p))
2126 # from service VRF1 back to client VRF0
2128 Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac)
2129 / IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4)
2130 / TCP(sport=local_port, dport=12347)
2132 self.pg5.add_stream(p)
2133 self.pg_enable_capture(self.pg_interfaces)
2135 capture = self.pg0.get_capture(1)
2140 self.assertEqual(ip.src, external_addr)
2141 self.assertEqual(tcp.sport, external_port)
2142 self.assert_packet_checksums_valid(p)
2144 self.logger.error(ppp("Unexpected or invalid packet:", p))
2147 def test_multiple_vrf_5(self):
2148 """Multiple VRF - forwarding - no translation"""
2150 external_addr = "1.2.3.4"
2155 self.vapi.nat44_forwarding_enable_disable(enable=1)
2156 flags = self.config_flags.NAT_IS_INSIDE
2157 self.vapi.nat44_interface_add_del_feature(
2158 sw_if_index=self.pg0.sw_if_index, is_add=1
2160 self.vapi.nat44_interface_add_del_feature(
2161 sw_if_index=self.pg0.sw_if_index, is_add=1, flags=flags
2163 self.vapi.nat44_interface_add_del_feature(
2164 sw_if_index=self.pg5.sw_if_index, is_add=1
2166 self.vapi.nat44_interface_add_del_feature(
2167 sw_if_index=self.pg5.sw_if_index, is_add=1, flags=flags
2169 self.vapi.nat44_interface_add_del_feature(
2170 sw_if_index=self.pg6.sw_if_index, is_add=1
2172 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
2173 self.nat_add_static_mapping(
2174 self.pg5.remote_ip4,
2179 proto=IP_PROTOS.tcp,
2182 self.nat_add_static_mapping(
2183 self.pg0.remote_ip4,
2184 external_sw_if_index=self.pg0.sw_if_index,
2185 local_port=local_port,
2187 external_port=external_port,
2188 proto=IP_PROTOS.tcp,
2192 # from client to server (both VRF1, no translation)
2194 Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac)
2195 / IP(src=self.pg6.remote_ip4, dst=self.pg5.remote_ip4)
2196 / TCP(sport=12348, dport=local_port)
2198 self.pg6.add_stream(p)
2199 self.pg_enable_capture(self.pg_interfaces)
2201 capture = self.pg5.get_capture(1)
2206 self.assertEqual(ip.dst, self.pg5.remote_ip4)
2207 self.assertEqual(tcp.dport, local_port)
2208 self.assert_packet_checksums_valid(p)
2210 self.logger.error(ppp("Unexpected or invalid packet:", p))
2213 # from server back to client (both VRF1, no translation)
2215 Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac)
2216 / IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4)
2217 / TCP(sport=local_port, dport=12348)
2219 self.pg5.add_stream(p)
2220 self.pg_enable_capture(self.pg_interfaces)
2222 capture = self.pg6.get_capture(1)
2227 self.assertEqual(ip.src, self.pg5.remote_ip4)
2228 self.assertEqual(tcp.sport, local_port)
2229 self.assert_packet_checksums_valid(p)
2231 self.logger.error(ppp("Unexpected or invalid packet:", p))
2234 # from client VRF1 to server VRF0 (no translation)
2236 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
2237 / IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4)
2238 / TCP(sport=local_port, dport=12349)
2240 self.pg0.add_stream(p)
2241 self.pg_enable_capture(self.pg_interfaces)
2243 capture = self.pg6.get_capture(1)
2248 self.assertEqual(ip.src, self.pg0.remote_ip4)
2249 self.assertEqual(tcp.sport, local_port)
2250 self.assert_packet_checksums_valid(p)
2252 self.logger.error(ppp("Unexpected or invalid packet:", p))
2255 # from server VRF0 back to client VRF1 (no translation)
2257 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
2258 / IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4)
2259 / TCP(sport=local_port, dport=12349)
2261 self.pg0.add_stream(p)
2262 self.pg_enable_capture(self.pg_interfaces)
2264 capture = self.pg6.get_capture(1)
2269 self.assertEqual(ip.src, self.pg0.remote_ip4)
2270 self.assertEqual(tcp.sport, local_port)
2271 self.assert_packet_checksums_valid(p)
2273 self.logger.error(ppp("Unexpected or invalid packet:", p))
2276 # from client VRF0 to server VRF1 (no translation)
2278 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
2279 / IP(src=self.pg0.remote_ip4, dst=self.pg5.remote_ip4)
2280 / TCP(sport=12344, dport=local_port)
2282 self.pg0.add_stream(p)
2283 self.pg_enable_capture(self.pg_interfaces)
2285 capture = self.pg5.get_capture(1)
2290 self.assertEqual(ip.dst, self.pg5.remote_ip4)
2291 self.assertEqual(tcp.dport, local_port)
2292 self.assert_packet_checksums_valid(p)
2294 self.logger.error(ppp("Unexpected or invalid packet:", p))
2297 # from server VRF1 back to client VRF0 (no translation)
2299 Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac)
2300 / IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4)
2301 / TCP(sport=local_port, dport=12344)
2303 self.pg5.add_stream(p)
2304 self.pg_enable_capture(self.pg_interfaces)
2306 capture = self.pg0.get_capture(1)
2311 self.assertEqual(ip.src, self.pg5.remote_ip4)
2312 self.assertEqual(tcp.sport, local_port)
2313 self.assert_packet_checksums_valid(p)
2315 self.logger.error(ppp("Unexpected or invalid packet:", p))
2318 def test_outside_address_distribution(self):
2319 """Outside address distribution based on source address"""
2324 for i in range(1, x):
2326 nat_addresses.append(a)
2328 self.nat_add_inside_interface(self.pg0)
2329 self.nat_add_outside_interface(self.pg1)
2331 self.vapi.nat44_add_del_address_range(
2332 first_ip_address=nat_addresses[0],
2333 last_ip_address=nat_addresses[-1],
2339 self.pg0.generate_remote_hosts(x)
2343 info = self.create_packet_info(self.pg0, self.pg1)
2344 payload = self.info_to_payload(info)
2346 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2347 / IP(src=self.pg0.remote_hosts[i].ip4, dst=self.pg1.remote_ip4)
2348 / UDP(sport=7000 + i, dport=8000 + i)
2354 self.pg0.add_stream(pkts)
2355 self.pg_enable_capture(self.pg_interfaces)
2357 recvd = self.pg1.get_capture(len(pkts))
2358 for p_recvd in recvd:
2359 payload_info = self.payload_to_info(p_recvd[Raw])
2360 packet_index = payload_info.index
2361 info = self._packet_infos[packet_index]
2362 self.assertTrue(info is not None)
2363 self.assertEqual(packet_index, info.index)
2365 packed = socket.inet_aton(p_sent[IP].src)
2366 numeric = struct.unpack("!L", packed)[0]
2367 numeric = socket.htonl(numeric)
2368 a = nat_addresses[(numeric - 1) % len(nat_addresses)]
2372 "Invalid packet (src IP %s translated to %s, but expected %s)"
2373 % (p_sent[IP].src, p_recvd[IP].src, a),
2376 def test_dynamic_edge_ports(self):
2377 """NAT44ED dynamic translation test: edge ports"""
2379 worker_count = self.vpp_worker_count or 1
2381 port_per_thread = (65536 - port_offset) // worker_count
2382 port_count = port_per_thread * worker_count
2384 # worker thread edge ports
2385 thread_edge_ports = {0, port_offset - 1, 65535}
2386 for i in range(0, worker_count):
2387 port_thread_offset = (port_per_thread * i) + port_offset
2388 for port_range_offset in [0, port_per_thread - 1]:
2389 port = port_thread_offset + port_range_offset
2390 thread_edge_ports.add(port)
2391 thread_drop_ports = set(
2393 lambda x: x not in range(port_offset, port_offset + port_count),
2401 self.nat_add_address(self.nat_addr)
2404 self.configure_ip4_interface(in_if, hosts=worker_count)
2405 self.configure_ip4_interface(out_if)
2407 self.nat_add_inside_interface(in_if)
2408 self.nat_add_outside_interface(out_if)
2411 tc1 = self.statistics["/nat44-ed/in2out/slowpath/tcp"]
2412 uc1 = self.statistics["/nat44-ed/in2out/slowpath/udp"]
2413 ic1 = self.statistics["/nat44-ed/in2out/slowpath/icmp"]
2414 dc1 = self.statistics["/nat44-ed/in2out/slowpath/drops"]
2416 pkt_count = worker_count * len(thread_edge_ports)
2418 i2o_pkts = [[] for x in range(0, worker_count)]
2419 for i in range(0, worker_count):
2420 remote_host = in_if.remote_hosts[i]
2421 for port in thread_edge_ports:
2423 Ether(dst=in_if.local_mac, src=in_if.remote_mac)
2424 / IP(src=remote_host.ip4, dst=out_if.remote_ip4)
2425 / TCP(sport=port, dport=port)
2427 i2o_pkts[i].append(p)
2430 Ether(dst=in_if.local_mac, src=in_if.remote_mac)
2431 / IP(src=remote_host.ip4, dst=out_if.remote_ip4)
2432 / UDP(sport=port, dport=port)
2434 i2o_pkts[i].append(p)
2437 Ether(dst=in_if.local_mac, src=in_if.remote_mac)
2438 / IP(src=remote_host.ip4, dst=out_if.remote_ip4)
2439 / ICMP(id=port, seq=port, type="echo-request")
2441 i2o_pkts[i].append(p)
2443 for i in range(0, worker_count):
2444 if len(i2o_pkts[i]) > 0:
2445 in_if.add_stream(i2o_pkts[i], worker=i)
2447 self.pg_enable_capture(self.pg_interfaces)
2449 capture = out_if.get_capture(pkt_count * 3)
2450 for packet in capture:
2451 self.assert_packet_checksums_valid(packet)
2452 if packet.haslayer(TCP):
2453 self.assert_in_range(
2456 port_offset + port_count,
2459 elif packet.haslayer(UDP):
2460 self.assert_in_range(
2463 port_offset + port_count,
2466 elif packet.haslayer(ICMP):
2467 self.assert_in_range(
2470 port_offset + port_count,
2475 ppp("Unexpected or invalid packet (outside network):", packet)
2478 if_idx = in_if.sw_if_index
2479 tc2 = self.statistics["/nat44-ed/in2out/slowpath/tcp"]
2480 uc2 = self.statistics["/nat44-ed/in2out/slowpath/udp"]
2481 ic2 = self.statistics["/nat44-ed/in2out/slowpath/icmp"]
2482 dc2 = self.statistics["/nat44-ed/in2out/slowpath/drops"]
2484 self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count)
2485 self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count)
2486 self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count)
2487 self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0)
2490 tc1 = self.statistics["/nat44-ed/out2in/fastpath/tcp"]
2491 uc1 = self.statistics["/nat44-ed/out2in/fastpath/udp"]
2492 ic1 = self.statistics["/nat44-ed/out2in/fastpath/icmp"]
2493 dc1 = self.statistics["/nat44-ed/out2in/fastpath/drops"]
2494 dc3 = self.statistics["/nat44-ed/out2in/slowpath/drops"]
2496 # replies to unchanged thread ports should pass on each worker,
2497 # excluding packets outside dynamic port range
2498 drop_count = worker_count * len(thread_drop_ports)
2499 pass_count = worker_count * len(thread_edge_ports) - drop_count
2501 o2i_pkts = [[] for x in range(0, worker_count)]
2502 for i in range(0, worker_count):
2503 for port in thread_edge_ports:
2505 Ether(dst=out_if.local_mac, src=out_if.remote_mac)
2506 / IP(src=out_if.remote_ip4, dst=self.nat_addr)
2507 / TCP(sport=port, dport=port)
2509 o2i_pkts[i].append(p)
2512 Ether(dst=out_if.local_mac, src=out_if.remote_mac)
2513 / IP(src=out_if.remote_ip4, dst=self.nat_addr)
2514 / UDP(sport=port, dport=port)
2516 o2i_pkts[i].append(p)
2519 Ether(dst=out_if.local_mac, src=out_if.remote_mac)
2520 / IP(src=out_if.remote_ip4, dst=self.nat_addr)
2521 / ICMP(id=port, seq=port, type="echo-reply")
2523 o2i_pkts[i].append(p)
2525 for i in range(0, worker_count):
2526 if len(o2i_pkts[i]) > 0:
2527 out_if.add_stream(o2i_pkts[i], worker=i)
2529 self.pg_enable_capture(self.pg_interfaces)
2531 capture = in_if.get_capture(pass_count * 3)
2532 for packet in capture:
2533 self.assert_packet_checksums_valid(packet)
2534 if packet.haslayer(TCP):
2535 self.assertIn(packet[TCP].dport, thread_edge_ports, "dst TCP port")
2536 self.assertEqual(packet[TCP].dport, packet[TCP].sport, "TCP ports")
2537 elif packet.haslayer(UDP):
2538 self.assertIn(packet[UDP].dport, thread_edge_ports, "dst UDP port")
2539 self.assertEqual(packet[UDP].dport, packet[UDP].sport, "UDP ports")
2540 elif packet.haslayer(ICMP):
2541 self.assertIn(packet[ICMP].id, thread_edge_ports, "ICMP id")
2542 self.assertEqual(packet[ICMP].id, packet[ICMP].seq, "ICMP id & seq")
2545 ppp("Unexpected or invalid packet (inside network):", packet)
2548 if_idx = out_if.sw_if_index
2549 tc2 = self.statistics["/nat44-ed/out2in/fastpath/tcp"]
2550 uc2 = self.statistics["/nat44-ed/out2in/fastpath/udp"]
2551 ic2 = self.statistics["/nat44-ed/out2in/fastpath/icmp"]
2552 dc2 = self.statistics["/nat44-ed/out2in/fastpath/drops"]
2553 dc4 = self.statistics["/nat44-ed/out2in/slowpath/drops"]
2555 self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pass_count)
2556 self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pass_count)
2557 self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pass_count)
2558 self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0)
2560 dc4[:, if_idx].sum() - dc3[:, if_idx].sum(), drop_count * 3
2567 def test_delete_interface(self):
2568 """NAT44ED delete nat interface"""
2570 self.nat_add_address(self.nat_addr)
2572 interfaces = self.create_loopback_interfaces(4)
2573 self.nat_add_outside_interface(interfaces[0])
2574 self.nat_add_inside_interface(interfaces[1])
2575 self.nat_add_outside_interface(interfaces[2])
2576 self.nat_add_inside_interface(interfaces[2])
2577 self.vapi.nat44_ed_add_del_output_interface(
2578 sw_if_index=interfaces[3].sw_if_index, is_add=1
2581 nat_sw_if_indices = [
2583 for i in self.vapi.nat44_interface_dump()
2584 + list(self.vapi.vpp.details_iter(self.vapi.nat44_ed_output_interface_get))
2586 self.assertEqual(len(nat_sw_if_indices), len(interfaces))
2589 for i in interfaces:
2590 # delete nat-enabled interface
2591 self.assertIn(i.sw_if_index, nat_sw_if_indices)
2592 i.remove_vpp_config()
2594 # create interface with the same index
2595 lo = VppLoInterface(self)
2596 loopbacks.append(lo)
2597 self.assertEqual(lo.sw_if_index, i.sw_if_index)
2599 # check interface is not nat-enabled
2600 nat_sw_if_indices = [
2602 for i in self.vapi.nat44_interface_dump()
2604 self.vapi.vpp.details_iter(self.vapi.nat44_ed_output_interface_get)
2607 self.assertNotIn(lo.sw_if_index, nat_sw_if_indices)
2610 i.remove_vpp_config()
2613 @tag_fixme_ubuntu2204
2614 class TestNAT44EDMW(TestNAT44ED):
2615 """NAT44ED MW Test Case"""
2617 vpp_worker_count = 4
2620 def test_dynamic(self):
2621 """NAT44ED dynamic translation test"""
2623 tcp_port_offset = 20
2624 udp_port_offset = 20
2627 self.nat_add_address(self.nat_addr)
2628 self.nat_add_inside_interface(self.pg0)
2629 self.nat_add_outside_interface(self.pg1)
2632 tc1 = self.statistics["/nat44-ed/in2out/slowpath/tcp"]
2633 uc1 = self.statistics["/nat44-ed/in2out/slowpath/udp"]
2634 ic1 = self.statistics["/nat44-ed/in2out/slowpath/icmp"]
2635 dc1 = self.statistics["/nat44-ed/in2out/slowpath/drops"]
2637 i2o_pkts = [[] for x in range(0, self.vpp_worker_count)]
2639 for i in range(pkt_count):
2641 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2642 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
2643 / TCP(sport=tcp_port_offset + i, dport=20)
2645 i2o_pkts[p[TCP].sport % self.vpp_worker_count].append(p)
2648 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2649 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
2650 / UDP(sport=udp_port_offset + i, dport=20)
2652 i2o_pkts[p[UDP].sport % self.vpp_worker_count].append(p)
2655 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2656 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
2657 / ICMP(id=icmp_id_offset + i, type="echo-request")
2659 i2o_pkts[p[ICMP].id % self.vpp_worker_count].append(p)
2661 for i in range(0, self.vpp_worker_count):
2662 if len(i2o_pkts[i]) > 0:
2663 self.pg0.add_stream(i2o_pkts[i], worker=i)
2665 self.pg_enable_capture(self.pg_interfaces)
2667 capture = self.pg1.get_capture(pkt_count * 3, timeout=5)
2669 if_idx = self.pg0.sw_if_index
2670 tc2 = self.statistics["/nat44-ed/in2out/slowpath/tcp"]
2671 uc2 = self.statistics["/nat44-ed/in2out/slowpath/udp"]
2672 ic2 = self.statistics["/nat44-ed/in2out/slowpath/icmp"]
2673 dc2 = self.statistics["/nat44-ed/in2out/slowpath/drops"]
2675 self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count)
2676 self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count)
2677 self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count)
2678 self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0)
2680 self.logger.info(self.vapi.cli("show trace"))
2683 tc1 = self.statistics["/nat44-ed/out2in/fastpath/tcp"]
2684 uc1 = self.statistics["/nat44-ed/out2in/fastpath/udp"]
2685 ic1 = self.statistics["/nat44-ed/out2in/fastpath/icmp"]
2686 dc1 = self.statistics["/nat44-ed/out2in/fastpath/drops"]
2688 recvd_tcp_ports = set()
2689 recvd_udp_ports = set()
2690 recvd_icmp_ids = set()
2694 recvd_tcp_ports.add(p[TCP].sport)
2696 recvd_udp_ports.add(p[UDP].sport)
2698 recvd_icmp_ids.add(p[ICMP].id)
2700 recvd_tcp_ports = list(recvd_tcp_ports)
2701 recvd_udp_ports = list(recvd_udp_ports)
2702 recvd_icmp_ids = list(recvd_icmp_ids)
2704 o2i_pkts = [[] for x in range(0, self.vpp_worker_count)]
2705 for i in range(pkt_count):
2707 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
2708 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
2709 / TCP(dport=choice(recvd_tcp_ports), sport=20)
2711 o2i_pkts[p[TCP].dport % self.vpp_worker_count].append(p)
2714 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
2715 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
2716 / UDP(dport=choice(recvd_udp_ports), sport=20)
2718 o2i_pkts[p[UDP].dport % self.vpp_worker_count].append(p)
2721 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
2722 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
2723 / ICMP(id=choice(recvd_icmp_ids), type="echo-reply")
2725 o2i_pkts[p[ICMP].id % self.vpp_worker_count].append(p)
2727 for i in range(0, self.vpp_worker_count):
2728 if len(o2i_pkts[i]) > 0:
2729 self.pg1.add_stream(o2i_pkts[i], worker=i)
2731 self.pg_enable_capture(self.pg_interfaces)
2733 capture = self.pg0.get_capture(pkt_count * 3)
2734 for packet in capture:
2736 self.assert_packet_checksums_valid(packet)
2737 self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
2738 if packet.haslayer(TCP):
2739 self.assert_in_range(
2742 tcp_port_offset + pkt_count,
2745 elif packet.haslayer(UDP):
2746 self.assert_in_range(
2749 udp_port_offset + pkt_count,
2753 self.assert_in_range(
2756 icmp_id_offset + pkt_count,
2761 ppp("Unexpected or invalid packet (inside network):", packet)
2765 if_idx = self.pg1.sw_if_index
2766 tc2 = self.statistics["/nat44-ed/out2in/fastpath/tcp"]
2767 uc2 = self.statistics["/nat44-ed/out2in/fastpath/udp"]
2768 ic2 = self.statistics["/nat44-ed/out2in/fastpath/icmp"]
2769 dc2 = self.statistics["/nat44-ed/out2in/fastpath/drops"]
2771 self.assertEqual(tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count)
2772 self.assertEqual(uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count)
2773 self.assertEqual(ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count)
2774 self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0)
2776 sc = self.statistics["/nat44-ed/total-sessions"]
2779 len(recvd_tcp_ports) + len(recvd_udp_ports) + len(recvd_icmp_ids),
2782 def test_frag_in_order(self):
2783 """NAT44ED translate fragments arriving in order"""
2785 self.nat_add_address(self.nat_addr)
2786 self.nat_add_inside_interface(self.pg0)
2787 self.nat_add_outside_interface(self.pg1)
2789 self.frag_in_order(proto=IP_PROTOS.tcp, ignore_port=True)
2790 self.frag_in_order(proto=IP_PROTOS.udp, ignore_port=True)
2791 self.frag_in_order(proto=IP_PROTOS.icmp, ignore_port=True)
2793 def test_frag_in_order_do_not_translate(self):
2794 """NAT44ED don't translate fragments arriving in order"""
2796 self.nat_add_address(self.nat_addr)
2797 self.nat_add_inside_interface(self.pg0)
2798 self.nat_add_outside_interface(self.pg1)
2799 self.vapi.nat44_forwarding_enable_disable(enable=True)
2801 self.frag_in_order(proto=IP_PROTOS.tcp, dont_translate=True)
2803 def test_frag_out_of_order(self):
2804 """NAT44ED translate fragments arriving out of order"""
2806 self.nat_add_address(self.nat_addr)
2807 self.nat_add_inside_interface(self.pg0)
2808 self.nat_add_outside_interface(self.pg1)
2810 self.frag_out_of_order(proto=IP_PROTOS.tcp, ignore_port=True)
2811 self.frag_out_of_order(proto=IP_PROTOS.udp, ignore_port=True)
2812 self.frag_out_of_order(proto=IP_PROTOS.icmp, ignore_port=True)
2814 def test_frag_in_order_in_plus_out(self):
2815 """NAT44ED in+out interface fragments in order"""
2817 in_port = self.random_port()
2818 out_port = self.random_port()
2820 self.nat_add_address(self.nat_addr)
2821 self.nat_add_inside_interface(self.pg0)
2822 self.nat_add_outside_interface(self.pg0)
2823 self.nat_add_inside_interface(self.pg1)
2824 self.nat_add_outside_interface(self.pg1)
2826 # add static mappings for server
2827 self.nat_add_static_mapping(
2828 self.server_addr, self.nat_addr, in_port, out_port, proto=IP_PROTOS.tcp
2830 self.nat_add_static_mapping(
2831 self.server_addr, self.nat_addr, in_port, out_port, proto=IP_PROTOS.udp
2833 self.nat_add_static_mapping(
2834 self.server_addr, self.nat_addr, proto=IP_PROTOS.icmp
2837 # run tests for each protocol
2838 self.frag_in_order_in_plus_out(
2839 self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.tcp
2841 self.frag_in_order_in_plus_out(
2842 self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.udp
2844 self.frag_in_order_in_plus_out(
2845 self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.icmp
2848 def test_frag_out_of_order_in_plus_out(self):
2849 """NAT44ED in+out interface fragments out of order"""
2851 in_port = self.random_port()
2852 out_port = self.random_port()
2854 self.nat_add_address(self.nat_addr)
2855 self.nat_add_inside_interface(self.pg0)
2856 self.nat_add_outside_interface(self.pg0)
2857 self.nat_add_inside_interface(self.pg1)
2858 self.nat_add_outside_interface(self.pg1)
2860 # add static mappings for server
2861 self.nat_add_static_mapping(
2862 self.server_addr, self.nat_addr, in_port, out_port, proto=IP_PROTOS.tcp
2864 self.nat_add_static_mapping(
2865 self.server_addr, self.nat_addr, in_port, out_port, proto=IP_PROTOS.udp
2867 self.nat_add_static_mapping(
2868 self.server_addr, self.nat_addr, proto=IP_PROTOS.icmp
2871 # run tests for each protocol
2872 self.frag_out_of_order_in_plus_out(
2873 self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.tcp
2875 self.frag_out_of_order_in_plus_out(
2876 self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.udp
2878 self.frag_out_of_order_in_plus_out(
2879 self.server_addr, self.nat_addr, in_port, out_port, IP_PROTOS.icmp
2882 def test_reass_hairpinning(self):
2883 """NAT44ED fragments hairpinning"""
2885 server_addr = self.pg0.remote_hosts[1].ip4
2887 host_in_port = self.random_port()
2888 server_in_port = self.random_port()
2889 server_out_port = self.random_port()
2891 self.nat_add_address(self.nat_addr)
2892 self.nat_add_inside_interface(self.pg0)
2893 self.nat_add_outside_interface(self.pg1)
2895 # add static mapping for server
2896 self.nat_add_static_mapping(
2901 proto=IP_PROTOS.tcp,
2903 self.nat_add_static_mapping(
2908 proto=IP_PROTOS.udp,
2910 self.nat_add_static_mapping(server_addr, self.nat_addr)
2912 self.reass_hairpinning(
2917 proto=IP_PROTOS.tcp,
2920 self.reass_hairpinning(
2925 proto=IP_PROTOS.udp,
2928 self.reass_hairpinning(
2933 proto=IP_PROTOS.icmp,
2937 def test_session_limit_per_vrf(self):
2938 """NAT44ED per vrf session limit"""
2941 inside_vrf10 = self.pg2
2946 # 2 interfaces pg0, pg1 (vrf10, limit 1 tcp session)
2947 # non existing vrf_id makes process core dump
2948 self.vapi.nat44_set_session_limit(session_limit=limit, vrf_id=10)
2950 self.nat_add_inside_interface(inside)
2951 self.nat_add_inside_interface(inside_vrf10)
2952 self.nat_add_outside_interface(outside)
2955 self.nat_add_interface_address(outside)
2957 # BUG: causing core dump - when bad vrf_id is specified
2958 # self.nat_add_address(outside.local_ip4, vrf_id=20)
2960 stream = self.create_tcp_stream(inside_vrf10, outside, limit * 2)
2961 inside_vrf10.add_stream(stream)
2963 self.pg_enable_capture(self.pg_interfaces)
2966 capture = outside.get_capture(limit)
2968 stream = self.create_tcp_stream(inside, outside, limit * 2)
2969 inside.add_stream(stream)
2971 self.pg_enable_capture(self.pg_interfaces)
2974 capture = outside.get_capture(len(stream))
2976 def test_show_max_translations(self):
2977 """NAT44ED API test - max translations per thread"""
2978 config = self.vapi.nat44_show_running_config()
2979 self.assertEqual(self.max_sessions, config.sessions)
2981 def test_lru_cleanup(self):
2982 """NAT44ED LRU cleanup algorithm"""
2984 self.nat_add_address(self.nat_addr)
2985 self.nat_add_inside_interface(self.pg0)
2986 self.nat_add_outside_interface(self.pg1)
2988 self.vapi.nat_set_timeouts(
2989 udp=1, tcp_established=7440, tcp_transitory=30, icmp=1
2992 tcp_port_out = self.init_tcp_session(self.pg0, self.pg1, 2000, 80)
2994 for i in range(0, self.max_sessions - 1):
2996 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2997 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64)
2998 / UDP(sport=7000 + i, dport=80)
3002 self.pg0.add_stream(pkts)
3003 self.pg_enable_capture(self.pg_interfaces)
3005 self.pg1.get_capture(len(pkts))
3006 self.virtual_sleep(1.5, "wait for timeouts")
3009 for i in range(0, self.max_sessions - 1):
3011 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
3012 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64)
3013 / ICMP(id=8000 + i, type="echo-request")
3017 self.pg0.add_stream(pkts)
3018 self.pg_enable_capture(self.pg_interfaces)
3020 self.pg1.get_capture(len(pkts))
3022 def test_session_rst_timeout(self):
3023 """NAT44ED session RST timeouts"""
3025 self.nat_add_address(self.nat_addr)
3026 self.nat_add_inside_interface(self.pg0)
3027 self.nat_add_outside_interface(self.pg1)
3029 self.vapi.nat_set_timeouts(
3030 udp=300, tcp_established=7440, tcp_transitory=5, icmp=60
3033 self.init_tcp_session(
3034 self.pg0, self.pg1, self.tcp_port_in, self.tcp_external_port
3037 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3038 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3039 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="R")
3041 self.send_and_expect(self.pg0, p, self.pg1)
3043 self.virtual_sleep(6)
3045 # The session is already closed
3047 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3048 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3049 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="P")
3051 self.send_and_assert_no_replies(self.pg0, p, self.pg1)
3053 # The session can be re-opened
3055 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3056 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3057 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="S")
3059 self.send_and_expect(self.pg0, p, self.pg1)
3061 def test_session_rst_established_timeout(self):
3062 """NAT44ED session RST timeouts"""
3064 self.nat_add_address(self.nat_addr)
3065 self.nat_add_inside_interface(self.pg0)
3066 self.nat_add_outside_interface(self.pg1)
3068 self.vapi.nat_set_timeouts(
3069 udp=300, tcp_established=7440, tcp_transitory=5, icmp=60
3072 self.init_tcp_session(
3073 self.pg0, self.pg1, self.tcp_port_in, self.tcp_external_port
3076 # Wait at least the transitory time, the session is in established
3077 # state anyway. RST followed by a data packet should move it to
3079 self.virtual_sleep(6)
3081 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3082 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3083 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="R")
3085 self.send_and_expect(self.pg0, p, self.pg1)
3088 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3089 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3090 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="P")
3092 self.send_and_expect(self.pg0, p, self.pg1)
3094 # State is transitory, session should be closed after 6 seconds
3095 self.virtual_sleep(6)
3098 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3099 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3100 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="P")
3102 self.send_and_assert_no_replies(self.pg0, p, self.pg1)
3104 def test_dynamic_out_of_ports(self):
3105 """NAT44ED dynamic translation test: out of ports"""
3107 self.nat_add_inside_interface(self.pg0)
3108 self.nat_add_outside_interface(self.pg1)
3110 # in2out and no NAT addresses added
3111 pkts = self.create_stream_in(self.pg0, self.pg1)
3113 self.send_and_assert_no_replies(
3117 stats_diff=self.no_diff
3120 "/err/nat44-ed-in2out-slowpath/out of ports": len(pkts),
3122 self.pg0.sw_if_index: {
3123 "/nat44-ed/in2out/slowpath/drops": len(pkts),
3128 # in2out after NAT addresses added
3129 self.nat_add_address(self.nat_addr)
3131 tcpn, udpn, icmpn = (
3132 sum(x) for x in zip(*((TCP in p, UDP in p, ICMP in p) for p in pkts))
3135 self.send_and_expect(
3140 stats_diff=self.no_diff
3143 "/err/nat44-ed-in2out-slowpath/out of ports": 0,
3145 self.pg0.sw_if_index: {
3146 "/nat44-ed/in2out/slowpath/drops": 0,
3147 "/nat44-ed/in2out/slowpath/tcp": tcpn,
3148 "/nat44-ed/in2out/slowpath/udp": udpn,
3149 "/nat44-ed/in2out/slowpath/icmp": icmpn,
3154 def test_unknown_proto(self):
3155 """NAT44ED translate packet with unknown protocol"""
3157 self.nat_add_address(self.nat_addr)
3158 self.nat_add_inside_interface(self.pg0)
3159 self.nat_add_outside_interface(self.pg1)
3163 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
3164 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3165 / TCP(sport=self.tcp_port_in, dport=20)
3167 self.pg0.add_stream(p)
3168 self.pg_enable_capture(self.pg_interfaces)
3170 p = self.pg1.get_capture(1)
3173 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
3174 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3176 / IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4)
3177 / TCP(sport=1234, dport=1234)
3179 self.pg0.add_stream(p)
3180 self.pg_enable_capture(self.pg_interfaces)
3182 p = self.pg1.get_capture(1)
3185 self.assertEqual(packet[IP].src, self.nat_addr)
3186 self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
3187 self.assertEqual(packet.haslayer(GRE), 1)
3188 self.assert_packet_checksums_valid(packet)
3190 self.logger.error(ppp("Unexpected or invalid packet:", packet))
3195 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
3196 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
3198 / IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4)
3199 / TCP(sport=1234, dport=1234)
3201 self.pg1.add_stream(p)
3202 self.pg_enable_capture(self.pg_interfaces)
3204 p = self.pg0.get_capture(1)
3207 self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
3208 self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
3209 self.assertEqual(packet.haslayer(GRE), 1)
3210 self.assert_packet_checksums_valid(packet)
3212 self.logger.error(ppp("Unexpected or invalid packet:", packet))
3215 def test_hairpinning_unknown_proto(self):
3216 """NAT44ED translate packet with unknown protocol - hairpinning"""
3217 host = self.pg0.remote_hosts[0]
3218 server = self.pg0.remote_hosts[1]
3220 server_out_port = 8765
3221 server_nat_ip = "10.0.0.11"
3223 self.nat_add_address(self.nat_addr)
3224 self.nat_add_inside_interface(self.pg0)
3225 self.nat_add_outside_interface(self.pg1)
3227 # add static mapping for server
3228 self.nat_add_static_mapping(server.ip4, server_nat_ip)
3232 Ether(src=host.mac, dst=self.pg0.local_mac)
3233 / IP(src=host.ip4, dst=server_nat_ip)
3234 / TCP(sport=host_in_port, dport=server_out_port)
3236 self.pg0.add_stream(p)
3237 self.pg_enable_capture(self.pg_interfaces)
3239 self.pg0.get_capture(1)
3242 Ether(dst=self.pg0.local_mac, src=host.mac)
3243 / IP(src=host.ip4, dst=server_nat_ip)
3245 / IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4)
3246 / TCP(sport=1234, dport=1234)
3248 self.pg0.add_stream(p)
3249 self.pg_enable_capture(self.pg_interfaces)
3251 p = self.pg0.get_capture(1)
3254 self.assertEqual(packet[IP].src, self.nat_addr)
3255 self.assertEqual(packet[IP].dst, server.ip4)
3256 self.assertEqual(packet.haslayer(GRE), 1)
3257 self.assert_packet_checksums_valid(packet)
3259 self.logger.error(ppp("Unexpected or invalid packet:", packet))
3264 Ether(dst=self.pg0.local_mac, src=server.mac)
3265 / IP(src=server.ip4, dst=self.nat_addr)
3267 / IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4)
3268 / TCP(sport=1234, dport=1234)
3270 self.pg0.add_stream(p)
3271 self.pg_enable_capture(self.pg_interfaces)
3273 p = self.pg0.get_capture(1)
3276 self.assertEqual(packet[IP].src, server_nat_ip)
3277 self.assertEqual(packet[IP].dst, host.ip4)
3278 self.assertEqual(packet.haslayer(GRE), 1)
3279 self.assert_packet_checksums_valid(packet)
3281 self.logger.error(ppp("Unexpected or invalid packet:", packet))
3284 def test_output_feature_and_service(self):
3285 """NAT44ED interface output feature and services"""
3286 external_addr = "1.2.3.4"
3290 self.vapi.nat44_forwarding_enable_disable(enable=1)
3291 self.nat_add_address(self.nat_addr)
3292 flags = self.config_flags.NAT_IS_ADDR_ONLY
3293 self.vapi.nat44_add_del_identity_mapping(
3294 ip_address=self.pg1.remote_ip4,
3295 sw_if_index=0xFFFFFFFF,
3299 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
3300 self.nat_add_static_mapping(
3301 self.pg0.remote_ip4,
3305 proto=IP_PROTOS.tcp,
3309 self.nat_add_inside_interface(self.pg0)
3310 self.nat_add_outside_interface(self.pg0)
3311 self.vapi.nat44_ed_add_del_output_interface(
3312 sw_if_index=self.pg1.sw_if_index, is_add=1
3315 # from client to service
3317 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3318 / IP(src=self.pg1.remote_ip4, dst=external_addr)
3319 / TCP(sport=12345, dport=external_port)
3321 self.pg1.add_stream(p)
3322 self.pg_enable_capture(self.pg_interfaces)
3324 capture = self.pg0.get_capture(1)
3329 self.assertEqual(ip.dst, self.pg0.remote_ip4)
3330 self.assertEqual(tcp.dport, local_port)
3331 self.assert_packet_checksums_valid(p)
3333 self.logger.error(ppp("Unexpected or invalid packet:", p))
3336 # from service back to client
3338 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3339 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3340 / TCP(sport=local_port, dport=12345)
3342 self.pg0.add_stream(p)
3343 self.pg_enable_capture(self.pg_interfaces)
3345 capture = self.pg1.get_capture(1)
3350 self.assertEqual(ip.src, external_addr)
3351 self.assertEqual(tcp.sport, external_port)
3352 self.assert_packet_checksums_valid(p)
3354 self.logger.error(ppp("Unexpected or invalid packet:", p))
3357 # from local network host to external network
3358 pkts = self.create_stream_in(self.pg0, self.pg1)
3359 self.pg0.add_stream(pkts)
3360 self.pg_enable_capture(self.pg_interfaces)
3362 capture = self.pg1.get_capture(len(pkts))
3363 self.verify_capture_out(capture, ignore_port=True)
3364 pkts = self.create_stream_in(self.pg0, self.pg1)
3365 self.pg0.add_stream(pkts)
3366 self.pg_enable_capture(self.pg_interfaces)
3368 capture = self.pg1.get_capture(len(pkts))
3369 self.verify_capture_out(capture, ignore_port=True)
3371 # from external network back to local network host
3372 pkts = self.create_stream_out(self.pg1)
3373 self.pg1.add_stream(pkts)
3374 self.pg_enable_capture(self.pg_interfaces)
3376 capture = self.pg0.get_capture(len(pkts))
3377 self.verify_capture_in(capture, self.pg0)
3379 def test_output_feature_and_service3(self):
3380 """NAT44ED interface output feature and DST NAT"""
3381 external_addr = "1.2.3.4"
3385 self.vapi.nat44_forwarding_enable_disable(enable=1)
3386 self.nat_add_address(self.nat_addr)
3387 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
3388 self.nat_add_static_mapping(
3389 self.pg1.remote_ip4,
3393 proto=IP_PROTOS.tcp,
3397 self.nat_add_inside_interface(self.pg0)
3398 self.nat_add_outside_interface(self.pg0)
3399 self.vapi.nat44_ed_add_del_output_interface(
3400 sw_if_index=self.pg1.sw_if_index, is_add=1
3404 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3405 / IP(src=self.pg0.remote_ip4, dst=external_addr)
3406 / TCP(sport=12345, dport=external_port)
3408 self.pg0.add_stream(p)
3409 self.pg_enable_capture(self.pg_interfaces)
3411 capture = self.pg1.get_capture(1)
3416 self.assertEqual(ip.src, self.pg0.remote_ip4)
3417 self.assertEqual(tcp.sport, 12345)
3418 self.assertEqual(ip.dst, self.pg1.remote_ip4)
3419 self.assertEqual(tcp.dport, local_port)
3420 self.assert_packet_checksums_valid(p)
3422 self.logger.error(ppp("Unexpected or invalid packet:", p))
3426 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3427 / IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4)
3428 / TCP(sport=local_port, dport=12345)
3430 self.pg1.add_stream(p)
3431 self.pg_enable_capture(self.pg_interfaces)
3433 capture = self.pg0.get_capture(1)
3438 self.assertEqual(ip.src, external_addr)
3439 self.assertEqual(tcp.sport, external_port)
3440 self.assertEqual(ip.dst, self.pg0.remote_ip4)
3441 self.assertEqual(tcp.dport, 12345)
3442 self.assert_packet_checksums_valid(p)
3444 self.logger.error(ppp("Unexpected or invalid packet:", p))
3447 def test_self_twice_nat_lb_negative(self):
3448 """NAT44ED Self Twice NAT local service load balancing (negative test)"""
3449 self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True, client_id=2)
3451 def test_self_twice_nat_negative(self):
3452 """NAT44ED Self Twice NAT (negative test)"""
3453 self.twice_nat_common(self_twice_nat=True)
3455 def test_static_lb_multi_clients(self):
3456 """NAT44ED local service load balancing - multiple clients"""
3458 external_addr = self.nat_addr
3461 server1 = self.pg0.remote_hosts[0]
3462 server2 = self.pg0.remote_hosts[1]
3463 server3 = self.pg0.remote_hosts[2]
3466 {"addr": server1.ip4, "port": local_port, "probability": 90, "vrf_id": 0},
3467 {"addr": server2.ip4, "port": local_port, "probability": 10, "vrf_id": 0},
3470 flags = self.config_flags.NAT_IS_INSIDE
3471 self.vapi.nat44_interface_add_del_feature(
3472 sw_if_index=self.pg0.sw_if_index, flags=flags, is_add=1
3474 self.vapi.nat44_interface_add_del_feature(
3475 sw_if_index=self.pg1.sw_if_index, is_add=1
3478 self.nat_add_address(self.nat_addr)
3479 self.vapi.nat44_add_del_lb_static_mapping(
3481 external_addr=external_addr,
3482 external_port=external_port,
3483 protocol=IP_PROTOS.tcp,
3484 local_num=len(locals),
3490 clients = ip4_range(self.pg1.remote_ip4, 10, 50)
3492 for client in clients:
3494 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3495 / IP(src=client, dst=self.nat_addr)
3496 / TCP(sport=12345, dport=external_port)
3499 self.pg1.add_stream(pkts)
3500 self.pg_enable_capture(self.pg_interfaces)
3502 capture = self.pg0.get_capture(len(pkts))
3504 if p[IP].dst == server1.ip4:
3508 self.assertGreaterEqual(server1_n, server2_n)
3511 "addr": server3.ip4,
3518 self.vapi.nat44_lb_static_mapping_add_del_local(
3520 external_addr=external_addr,
3521 external_port=external_port,
3523 protocol=IP_PROTOS.tcp,
3528 clients = ip4_range(self.pg1.remote_ip4, 60, 110)
3530 for client in clients:
3532 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3533 / IP(src=client, dst=self.nat_addr)
3534 / TCP(sport=12346, dport=external_port)
3537 self.assertGreater(len(pkts), 0)
3538 self.pg1.add_stream(pkts)
3539 self.pg_enable_capture(self.pg_interfaces)
3541 capture = self.pg0.get_capture(len(pkts))
3543 if p[IP].dst == server1.ip4:
3545 elif p[IP].dst == server2.ip4:
3549 self.assertGreater(server1_n, 0)
3550 self.assertGreater(server2_n, 0)
3551 self.assertGreater(server3_n, 0)
3554 "addr": server2.ip4,
3560 # remove one back-end
3561 self.vapi.nat44_lb_static_mapping_add_del_local(
3563 external_addr=external_addr,
3564 external_port=external_port,
3566 protocol=IP_PROTOS.tcp,
3571 self.pg1.add_stream(pkts)
3572 self.pg_enable_capture(self.pg_interfaces)
3574 capture = self.pg0.get_capture(len(pkts))
3576 if p[IP].dst == server1.ip4:
3578 elif p[IP].dst == server2.ip4:
3582 self.assertGreater(server1_n, 0)
3583 self.assertEqual(server2_n, 0)
3584 self.assertGreater(server3_n, 0)
3586 # put zzz in front of syslog test name so that it runs as a last test
3587 # setting syslog sender cannot be undone and if it is set, it messes
3588 # with self.send_and_assert_no_replies functionality
3589 def test_zzz_syslog_sess(self):
3590 """NAT44ED Test syslog session creation and deletion"""
3591 self.vapi.syslog_set_filter(self.syslog_severity.SYSLOG_API_SEVERITY_INFO)
3592 self.vapi.syslog_set_sender(self.pg3.local_ip4, self.pg3.remote_ip4)
3594 self.nat_add_address(self.nat_addr)
3595 self.nat_add_inside_interface(self.pg0)
3596 self.nat_add_outside_interface(self.pg1)
3599 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
3600 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3601 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port)
3603 self.pg0.add_stream(p)
3604 self.pg_enable_capture(self.pg_interfaces)
3606 capture = self.pg1.get_capture(1)
3607 self.tcp_port_out = capture[0][TCP].sport
3608 capture = self.pg3.get_capture(1)
3609 self.verify_syslog_sess(capture[0][Raw].load, "SADD")
3611 self.pg_enable_capture(self.pg_interfaces)
3613 self.nat_add_address(self.nat_addr, is_add=0)
3614 capture = self.pg3.get_capture(1)
3615 self.verify_syslog_sess(capture[0][Raw].load, "SDEL")
3617 # put zzz in front of syslog test name so that it runs as a last test
3618 # setting syslog sender cannot be undone and if it is set, it messes
3619 # with self.send_and_assert_no_replies functionality
3620 def test_zzz_syslog_sess_reopen(self):
3621 """Syslog events for session reopen"""
3622 self.vapi.syslog_set_filter(self.syslog_severity.SYSLOG_API_SEVERITY_INFO)
3623 self.vapi.syslog_set_sender(self.pg3.local_ip4, self.pg3.remote_ip4)
3625 self.nat_add_address(self.nat_addr)
3626 self.nat_add_inside_interface(self.pg0)
3627 self.nat_add_outside_interface(self.pg1)
3631 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
3632 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3633 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port)
3635 capture = self.send_and_expect(self.pg0, p, self.pg1)[0]
3636 self.tcp_port_out = capture[0][TCP].sport
3637 capture = self.pg3.get_capture(1)
3638 self.verify_syslog_sess(capture[0][Raw].load, "SADD")
3642 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
3643 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
3644 / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="SA")
3646 self.send_and_expect(self.pg1, p, self.pg0)
3649 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
3650 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3651 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="A")
3653 self.send_and_expect(self.pg0, p, self.pg1)
3657 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
3658 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3659 / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="F")
3661 self.send_and_expect(self.pg0, p, self.pg1)
3665 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
3666 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
3667 / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="F")
3669 self.send_and_expect(self.pg1, p, self.pg0)
3671 self.init_tcp_session(
3672 self.pg0, self.pg1, self.tcp_port_in, self.tcp_external_port
3675 # 2 records should be produced - first one del & add
3676 capture = self.pg3.get_capture(2)
3677 self.verify_syslog_sess(capture[0][Raw].load, "SDEL")
3678 self.verify_syslog_sess(capture[1][Raw].load, "SADD")
3680 def test_twice_nat_interface_addr(self):
3681 """NAT44ED Acquire twice NAT addresses from interface"""
3682 flags = self.config_flags.NAT_IS_TWICE_NAT
3683 self.vapi.nat44_add_del_interface_addr(
3684 sw_if_index=self.pg11.sw_if_index, flags=flags, is_add=1
3687 # no address in NAT pool
3688 adresses = self.vapi.nat44_address_dump()
3689 self.assertEqual(0, len(adresses))
3691 # configure interface address and check NAT address pool
3692 self.pg11.config_ip4()
3693 adresses = self.vapi.nat44_address_dump()
3694 self.assertEqual(1, len(adresses))
3695 self.assertEqual(str(adresses[0].ip_address), self.pg11.local_ip4)
3696 self.assertEqual(adresses[0].flags, flags)
3698 # remove interface address and check NAT address pool
3699 self.pg11.unconfig_ip4()
3700 adresses = self.vapi.nat44_address_dump()
3701 self.assertEqual(0, len(adresses))
3703 def test_output_feature_stateful_acl(self):
3704 """NAT44ED output feature works with stateful ACL"""
3706 self.nat_add_address(self.nat_addr)
3707 self.vapi.nat44_ed_add_del_output_interface(
3708 sw_if_index=self.pg1.sw_if_index, is_add=1
3711 # First ensure that the NAT is working sans ACL
3713 # send packets out2in, no sessions yet so packets should drop
3714 pkts_out2in = self.create_stream_out(self.pg1)
3715 self.send_and_assert_no_replies(self.pg1, pkts_out2in)
3717 # send packets into inside intf, ensure received via outside intf
3718 pkts_in2out = self.create_stream_in(self.pg0, self.pg1)
3719 capture = self.send_and_expect(
3720 self.pg0, pkts_in2out, self.pg1, len(pkts_in2out)
3722 self.verify_capture_out(capture, ignore_port=True)
3724 # send out2in again, with sessions created it should work now
3725 pkts_out2in = self.create_stream_out(self.pg1)
3726 capture = self.send_and_expect(
3727 self.pg1, pkts_out2in, self.pg0, len(pkts_out2in)
3729 self.verify_capture_in(capture, self.pg0)
3731 # Create an ACL blocking everything
3732 out2in_deny_rule = AclRule(is_permit=0)
3733 out2in_acl = VppAcl(self, rules=[out2in_deny_rule])
3734 out2in_acl.add_vpp_config()
3736 # create an ACL to permit/reflect everything
3737 in2out_reflect_rule = AclRule(is_permit=2)
3738 in2out_acl = VppAcl(self, rules=[in2out_reflect_rule])
3739 in2out_acl.add_vpp_config()
3741 # apply as input acl on interface and confirm it blocks everything
3742 acl_if = VppAclInterface(
3743 self, sw_if_index=self.pg1.sw_if_index, n_input=1, acls=[out2in_acl]
3745 acl_if.add_vpp_config()
3746 self.send_and_assert_no_replies(self.pg1, pkts_out2in)
3749 acl_if.acls = [out2in_acl, in2out_acl]
3750 acl_if.add_vpp_config()
3751 # send in2out to generate ACL state (NAT state was created earlier)
3752 capture = self.send_and_expect(
3753 self.pg0, pkts_in2out, self.pg1, len(pkts_in2out)
3755 self.verify_capture_out(capture, ignore_port=True)
3757 # send out2in again. ACL state exists so it should work now.
3758 # TCP packets with the syn flag set also need the ack flag
3759 for p in pkts_out2in:
3760 if p.haslayer(TCP) and p[TCP].flags & 0x02:
3761 p[TCP].flags |= 0x10
3762 capture = self.send_and_expect(
3763 self.pg1, pkts_out2in, self.pg0, len(pkts_out2in)
3765 self.verify_capture_in(capture, self.pg0)
3766 self.logger.info(self.vapi.cli("show trace"))
3768 def test_tcp_close(self):
3769 """NAT44ED Close TCP session from inside network - output feature"""
3770 config = self.vapi.nat44_show_running_config()
3771 old_timeouts = config.timeouts
3773 self.vapi.nat_set_timeouts(
3774 udp=old_timeouts.udp,
3775 tcp_established=old_timeouts.tcp_established,
3776 icmp=old_timeouts.icmp,
3777 tcp_transitory=new_transitory,
3780 self.vapi.nat44_forwarding_enable_disable(enable=1)
3781 self.nat_add_address(self.pg1.local_ip4)
3782 twice_nat_addr = "10.0.1.3"
3783 service_ip = "192.168.16.150"
3784 self.nat_add_address(twice_nat_addr, twice_nat=1)
3786 flags = self.config_flags.NAT_IS_INSIDE
3787 self.vapi.nat44_interface_add_del_feature(
3788 sw_if_index=self.pg0.sw_if_index, is_add=1
3790 self.vapi.nat44_interface_add_del_feature(
3791 sw_if_index=self.pg0.sw_if_index, flags=flags, is_add=1
3793 self.vapi.nat44_ed_add_del_output_interface(
3794 is_add=1, sw_if_index=self.pg1.sw_if_index
3798 self.config_flags.NAT_IS_OUT2IN_ONLY | self.config_flags.NAT_IS_TWICE_NAT
3800 self.nat_add_static_mapping(
3801 self.pg0.remote_ip4, service_ip, 80, 80, proto=IP_PROTOS.tcp, flags=flags
3803 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3804 start_sessnum = len(sessions)
3806 # SYN packet out->in
3808 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3809 / IP(src=self.pg1.remote_ip4, dst=service_ip)
3810 / TCP(sport=33898, dport=80, flags="S")
3812 capture = self.send_and_expect(self.pg1, p, self.pg0, n_rx=1)
3814 tcp_port = p[TCP].sport
3816 # SYN + ACK packet in->out
3818 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3819 / IP(src=self.pg0.remote_ip4, dst=twice_nat_addr)
3820 / TCP(sport=80, dport=tcp_port, flags="SA")
3822 self.send_and_expect(self.pg0, p, self.pg1, n_rx=1)
3824 # ACK packet out->in
3826 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3827 / IP(src=self.pg1.remote_ip4, dst=service_ip)
3828 / TCP(sport=33898, dport=80, flags="A")
3830 self.send_and_expect(self.pg1, p, self.pg0, n_rx=1)
3832 # FIN packet in -> out
3834 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3835 / IP(src=self.pg0.remote_ip4, dst=twice_nat_addr)
3836 / TCP(sport=80, dport=tcp_port, flags="FA", seq=100, ack=300)
3838 self.send_and_expect(self.pg0, p, self.pg1, n_rx=1)
3840 # FIN+ACK packet out -> in
3842 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3843 / IP(src=self.pg1.remote_ip4, dst=service_ip)
3844 / TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101)
3846 self.send_and_expect(self.pg1, p, self.pg0, n_rx=1)
3848 # ACK packet in -> out
3850 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3851 / IP(src=self.pg0.remote_ip4, dst=twice_nat_addr)
3852 / TCP(sport=80, dport=tcp_port, flags="A", seq=101, ack=301)
3854 self.send_and_expect(self.pg0, p, self.pg1, n_rx=1)
3856 # session now in transitory timeout, but traffic still flows
3857 # try FIN packet out->in
3859 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3860 / IP(src=self.pg1.remote_ip4, dst=service_ip)
3861 / TCP(sport=33898, dport=80, flags="F")
3863 self.pg1.add_stream(p)
3864 self.pg_enable_capture(self.pg_interfaces)
3867 self.virtual_sleep(new_transitory, "wait for transitory timeout")
3868 self.pg0.get_capture(1)
3870 # session should still exist
3871 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3872 self.assertEqual(len(sessions) - start_sessnum, 1)
3874 # send FIN+ACK packet out -> in - will cause session to be wiped
3875 # but won't create a new session
3877 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3878 / IP(src=self.pg1.remote_ip4, dst=service_ip)
3879 / TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101)
3881 self.send_and_assert_no_replies(self.pg1, p)
3882 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3883 self.assertEqual(len(sessions) - start_sessnum, 0)
3885 def test_tcp_session_close_in(self):
3886 """NAT44ED Close TCP session from inside network"""
3888 in_port = self.tcp_port_in
3890 ext_port = self.tcp_external_port
3892 self.nat_add_address(self.nat_addr)
3893 self.nat_add_inside_interface(self.pg0)
3894 self.nat_add_outside_interface(self.pg1)
3895 self.nat_add_static_mapping(
3896 self.pg0.remote_ip4,
3900 proto=IP_PROTOS.tcp,
3901 flags=self.config_flags.NAT_IS_TWICE_NAT,
3904 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3905 session_n = len(sessions)
3907 self.vapi.nat_set_timeouts(
3908 udp=300, tcp_established=7440, tcp_transitory=2, icmp=5
3911 self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port)
3913 # FIN packet in -> out
3915 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3916 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3917 / TCP(sport=in_port, dport=ext_port, flags="FA", seq=100, ack=300)
3919 self.send_and_expect(self.pg0, p, self.pg1)
3922 # ACK packet out -> in
3924 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3925 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
3926 / TCP(sport=ext_port, dport=out_port, flags="A", seq=300, ack=101)
3930 # FIN packet out -> in
3932 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3933 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
3934 / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=101)
3938 self.send_and_expect(self.pg1, pkts, self.pg0)
3940 # ACK packet in -> out
3942 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3943 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3944 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
3946 self.send_and_expect(self.pg0, p, self.pg1)
3948 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3949 self.assertEqual(len(sessions) - session_n, 1)
3951 # retransmit FIN packet out -> in
3953 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
3954 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
3955 / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=101)
3958 self.send_and_expect(self.pg1, p, self.pg0)
3960 # retransmit ACK packet in -> out
3962 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3963 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3964 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
3966 self.send_and_expect(self.pg0, p, self.pg1)
3968 self.virtual_sleep(3)
3969 # retransmit ACK packet in -> out - this will cause session to be wiped
3971 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
3972 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
3973 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
3975 self.send_and_assert_no_replies(self.pg0, p)
3976 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3977 self.assertEqual(len(sessions) - session_n, 0)
3979 def test_tcp_session_close_out(self):
3980 """NAT44ED Close TCP session from outside network"""
3982 in_port = self.tcp_port_in
3984 ext_port = self.tcp_external_port
3986 self.nat_add_address(self.nat_addr)
3987 self.nat_add_inside_interface(self.pg0)
3988 self.nat_add_outside_interface(self.pg1)
3989 self.nat_add_static_mapping(
3990 self.pg0.remote_ip4,
3994 proto=IP_PROTOS.tcp,
3995 flags=self.config_flags.NAT_IS_TWICE_NAT,
3998 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3999 session_n = len(sessions)
4001 self.vapi.nat_set_timeouts(
4002 udp=300, tcp_established=7440, tcp_transitory=2, icmp=5
4005 _ = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port)
4007 # FIN packet out -> in
4009 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4010 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4011 / TCP(sport=ext_port, dport=out_port, flags="FA", seq=100, ack=300)
4013 self.pg1.add_stream(p)
4014 self.pg_enable_capture(self.pg_interfaces)
4016 self.pg0.get_capture(1)
4018 # FIN+ACK packet in -> out
4020 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4021 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4022 / TCP(sport=in_port, dport=ext_port, flags="FA", seq=300, ack=101)
4025 self.pg0.add_stream(p)
4026 self.pg_enable_capture(self.pg_interfaces)
4028 self.pg1.get_capture(1)
4030 # ACK packet out -> in
4032 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4033 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4034 / TCP(sport=ext_port, dport=out_port, flags="A", seq=101, ack=301)
4036 self.pg1.add_stream(p)
4037 self.pg_enable_capture(self.pg_interfaces)
4039 self.pg0.get_capture(1)
4041 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4042 self.assertEqual(len(sessions) - session_n, 1)
4044 # retransmit FIN packet out -> in
4046 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4047 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4048 / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=101)
4050 self.send_and_expect(self.pg1, p, self.pg0)
4052 # retransmit ACK packet in -> out
4054 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4055 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4056 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
4058 self.send_and_expect(self.pg0, p, self.pg1)
4060 self.virtual_sleep(3)
4061 # retransmit ACK packet in -> out - this will cause session to be wiped
4063 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4064 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4065 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
4067 self.send_and_assert_no_replies(self.pg0, p)
4068 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4069 self.assertEqual(len(sessions) - session_n, 0)
4071 def test_tcp_session_close_simultaneous(self):
4072 """Simultaneous TCP close from both sides"""
4074 in_port = self.tcp_port_in
4077 self.nat_add_address(self.nat_addr)
4078 self.nat_add_inside_interface(self.pg0)
4079 self.nat_add_outside_interface(self.pg1)
4080 self.nat_add_static_mapping(
4081 self.pg0.remote_ip4,
4085 proto=IP_PROTOS.tcp,
4086 flags=self.config_flags.NAT_IS_TWICE_NAT,
4089 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4090 session_n = len(sessions)
4092 self.vapi.nat_set_timeouts(
4093 udp=300, tcp_established=7440, tcp_transitory=2, icmp=5
4096 out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port)
4098 # FIN packet in -> out
4100 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4101 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4102 / TCP(sport=in_port, dport=ext_port, flags="FA", seq=100, ack=300)
4104 self.send_and_expect(self.pg0, p, self.pg1)
4106 # FIN packet out -> in
4108 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4109 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4110 / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=100)
4112 self.send_and_expect(self.pg1, p, self.pg0)
4114 # ACK packet in -> out
4116 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4117 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4118 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
4120 self.send_and_expect(self.pg0, p, self.pg1)
4122 # ACK packet out -> in
4124 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4125 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4126 / TCP(sport=ext_port, dport=out_port, flags="A", seq=301, ack=101)
4128 self.send_and_expect(self.pg1, p, self.pg0)
4130 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4131 self.assertEqual(len(sessions) - session_n, 1)
4133 # retransmit FIN packet out -> in
4135 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4136 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4137 / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=101)
4139 self.send_and_expect(self.pg1, p, self.pg0)
4141 # retransmit ACK packet in -> out
4143 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4144 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4145 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
4147 self.send_and_expect(self.pg0, p, self.pg1)
4149 self.virtual_sleep(3)
4150 # retransmit ACK packet in -> out - this will cause session to be wiped
4152 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4153 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4154 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
4156 self.pg_send(self.pg0, p)
4157 self.send_and_assert_no_replies(self.pg0, p)
4158 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4159 self.assertEqual(len(sessions) - session_n, 0)
4161 def test_tcp_session_half_reopen_inside(self):
4162 """TCP session in FIN/FIN state not reopened by in2out SYN only"""
4163 in_port = self.tcp_port_in
4166 self.nat_add_address(self.nat_addr)
4167 self.nat_add_inside_interface(self.pg0)
4168 self.nat_add_outside_interface(self.pg1)
4169 self.nat_add_static_mapping(
4170 self.pg0.remote_ip4,
4174 proto=IP_PROTOS.tcp,
4175 flags=self.config_flags.NAT_IS_TWICE_NAT,
4178 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4179 session_n = len(sessions)
4181 self.vapi.nat_set_timeouts(
4182 udp=300, tcp_established=7440, tcp_transitory=2, icmp=5
4185 out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port)
4187 # FIN packet in -> out
4189 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4190 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4191 / TCP(sport=in_port, dport=ext_port, flags="FA", seq=100, ack=300)
4193 self.send_and_expect(self.pg0, p, self.pg1)
4195 # FIN packet out -> in
4197 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4198 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4199 / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=100)
4201 self.send_and_expect(self.pg1, p, self.pg0)
4203 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4204 self.assertEqual(len(sessions) - session_n, 1)
4206 # send SYN packet in -> out
4208 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4209 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4210 / TCP(sport=in_port, dport=ext_port, flags="S", seq=101, ack=301)
4212 self.send_and_expect(self.pg0, p, self.pg1)
4214 self.virtual_sleep(3)
4215 # send ACK packet in -> out - session should be wiped
4217 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4218 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4219 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
4221 self.send_and_assert_no_replies(self.pg0, p, self.pg1)
4222 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4223 self.assertEqual(len(sessions) - session_n, 0)
4225 def test_tcp_session_half_reopen_outside(self):
4226 """TCP session in FIN/FIN state not reopened by out2in SYN only"""
4227 in_port = self.tcp_port_in
4230 self.nat_add_address(self.nat_addr)
4231 self.nat_add_inside_interface(self.pg0)
4232 self.nat_add_outside_interface(self.pg1)
4233 self.nat_add_static_mapping(
4234 self.pg0.remote_ip4,
4238 proto=IP_PROTOS.tcp,
4239 flags=self.config_flags.NAT_IS_TWICE_NAT,
4242 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4243 session_n = len(sessions)
4245 self.vapi.nat_set_timeouts(
4246 udp=300, tcp_established=7440, tcp_transitory=2, icmp=5
4249 out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port)
4251 # FIN packet in -> out
4253 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4254 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4255 / TCP(sport=in_port, dport=ext_port, flags="FA", seq=100, ack=300)
4257 self.send_and_expect(self.pg0, p, self.pg1)
4259 # FIN packet out -> in
4261 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4262 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4263 / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=100)
4265 self.send_and_expect(self.pg1, p, self.pg0)
4267 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4268 self.assertEqual(len(sessions) - session_n, 1)
4270 # send SYN packet out -> in
4272 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4273 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4274 / TCP(sport=ext_port, dport=out_port, flags="S", seq=300, ack=101)
4276 self.send_and_expect(self.pg1, p, self.pg0)
4278 self.virtual_sleep(3)
4279 # send ACK packet in -> out - session should be wiped
4281 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4282 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4283 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
4285 self.send_and_assert_no_replies(self.pg0, p, self.pg1)
4286 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4287 self.assertEqual(len(sessions) - session_n, 0)
4289 def test_tcp_session_reopen(self):
4290 """TCP session in FIN/FIN state reopened by SYN from both sides"""
4291 in_port = self.tcp_port_in
4294 self.nat_add_address(self.nat_addr)
4295 self.nat_add_inside_interface(self.pg0)
4296 self.nat_add_outside_interface(self.pg1)
4297 self.nat_add_static_mapping(
4298 self.pg0.remote_ip4,
4302 proto=IP_PROTOS.tcp,
4303 flags=self.config_flags.NAT_IS_TWICE_NAT,
4306 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4307 session_n = len(sessions)
4309 self.vapi.nat_set_timeouts(
4310 udp=300, tcp_established=7440, tcp_transitory=2, icmp=5
4313 out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port)
4315 # FIN packet in -> out
4317 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4318 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4319 / TCP(sport=in_port, dport=ext_port, flags="FA", seq=100, ack=300)
4321 self.send_and_expect(self.pg0, p, self.pg1)
4323 # FIN packet out -> in
4325 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4326 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4327 / TCP(sport=ext_port, dport=out_port, flags="FA", seq=300, ack=100)
4329 self.send_and_expect(self.pg1, p, self.pg0)
4331 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4332 self.assertEqual(len(sessions) - session_n, 1)
4334 # send SYN packet out -> in
4336 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4337 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4338 / TCP(sport=ext_port, dport=out_port, flags="S", seq=300, ack=101)
4340 self.send_and_expect(self.pg1, p, self.pg0)
4342 # send SYN packet in -> out
4344 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4345 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4346 / TCP(sport=in_port, dport=ext_port, flags="SA", seq=101, ack=301)
4348 self.send_and_expect(self.pg0, p, self.pg1)
4350 # send ACK packet out -> in
4352 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4353 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4354 / TCP(sport=ext_port, dport=out_port, flags="A", seq=300, ack=101)
4356 self.send_and_expect(self.pg1, p, self.pg0)
4358 self.virtual_sleep(3)
4359 # send ACK packet in -> out - should be forwarded and session alive
4361 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4362 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4363 / TCP(sport=in_port, dport=ext_port, flags="A", seq=101, ack=301)
4365 self.send_and_expect(self.pg0, p, self.pg1)
4366 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
4367 self.assertEqual(len(sessions) - session_n, 1)
4369 def test_dynamic_vrf(self):
4370 """NAT44ED dynamic translation test: different VRF"""
4375 self.nat_add_address(self.nat_addr, vrf_id=vrf_id_in)
4378 self.configure_ip4_interface(self.pg7, table_id=vrf_id_in)
4379 self.configure_ip4_interface(self.pg8, table_id=vrf_id_out)
4381 self.nat_add_inside_interface(self.pg7)
4382 self.nat_add_outside_interface(self.pg8)
4384 # just basic stuff nothing special
4385 pkts = self.create_stream_in(self.pg7, self.pg8)
4386 self.pg7.add_stream(pkts)
4387 self.pg_enable_capture(self.pg_interfaces)
4389 capture = self.pg8.get_capture(len(pkts))
4390 self.verify_capture_out(capture, ignore_port=True)
4392 pkts = self.create_stream_out(self.pg8)
4393 self.pg8.add_stream(pkts)
4394 self.pg_enable_capture(self.pg_interfaces)
4396 capture = self.pg7.get_capture(len(pkts))
4397 self.verify_capture_in(capture, self.pg7)
4403 self.vapi.ip_table_add_del(is_add=0, table={"table_id": vrf_id_in})
4404 self.vapi.ip_table_add_del(is_add=0, table={"table_id": vrf_id_out})
4406 def test_dynamic_output_feature_vrf(self):
4407 """NAT44ED dynamic translation test: output-feature, VRF"""
4409 # other then default (0)
4412 self.nat_add_address(self.nat_addr)
4413 self.vapi.nat44_ed_add_del_output_interface(
4414 sw_if_index=self.pg8.sw_if_index, is_add=1
4417 self.configure_ip4_interface(self.pg7, table_id=new_vrf_id)
4418 self.configure_ip4_interface(self.pg8, table_id=new_vrf_id)
4421 tcpn = self.statistics["/nat44-ed/in2out/slowpath/tcp"]
4422 udpn = self.statistics["/nat44-ed/in2out/slowpath/udp"]
4423 icmpn = self.statistics["/nat44-ed/in2out/slowpath/icmp"]
4424 drops = self.statistics["/nat44-ed/in2out/slowpath/drops"]
4426 pkts = self.create_stream_in(self.pg7, self.pg8)
4427 self.pg7.add_stream(pkts)
4428 self.pg_enable_capture(self.pg_interfaces)
4430 capture = self.pg8.get_capture(len(pkts))
4431 self.verify_capture_out(capture, ignore_port=True)
4433 if_idx = self.pg8.sw_if_index
4434 cnt = self.statistics["/nat44-ed/in2out/slowpath/tcp"]
4435 self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2)
4436 cnt = self.statistics["/nat44-ed/in2out/slowpath/udp"]
4437 self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1)
4438 cnt = self.statistics["/nat44-ed/in2out/slowpath/icmp"]
4439 self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1)
4440 cnt = self.statistics["/nat44-ed/in2out/slowpath/drops"]
4441 self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0)
4444 tcpn = self.statistics["/nat44-ed/out2in/fastpath/tcp"]
4445 udpn = self.statistics["/nat44-ed/out2in/fastpath/udp"]
4446 icmpn = self.statistics["/nat44-ed/out2in/fastpath/icmp"]
4447 drops = self.statistics["/nat44-ed/out2in/fastpath/drops"]
4449 pkts = self.create_stream_out(self.pg8)
4450 self.pg8.add_stream(pkts)
4451 self.pg_enable_capture(self.pg_interfaces)
4453 capture = self.pg7.get_capture(len(pkts))
4454 self.verify_capture_in(capture, self.pg7)
4456 if_idx = self.pg8.sw_if_index
4457 cnt = self.statistics["/nat44-ed/out2in/fastpath/tcp"]
4458 self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2)
4459 cnt = self.statistics["/nat44-ed/out2in/fastpath/udp"]
4460 self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1)
4461 cnt = self.statistics["/nat44-ed/out2in/fastpath/icmp"]
4462 self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1)
4463 cnt = self.statistics["/nat44-ed/out2in/fastpath/drops"]
4464 self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0)
4466 sessions = self.statistics["/nat44-ed/total-sessions"]
4467 self.assertEqual(sessions[:, 0].sum(), 3)
4473 self.vapi.ip_table_add_del(is_add=0, table={"table_id": new_vrf_id})
4475 def test_next_src_nat(self):
4476 """NAT44ED On way back forward packet to nat44-in2out node."""
4478 twice_nat_addr = "10.0.1.3"
4481 post_twice_nat_port = 0
4483 self.vapi.nat44_forwarding_enable_disable(enable=1)
4484 self.nat_add_address(twice_nat_addr, twice_nat=1)
4486 self.config_flags.NAT_IS_OUT2IN_ONLY
4487 | self.config_flags.NAT_IS_SELF_TWICE_NAT
4489 self.nat_add_static_mapping(
4490 self.pg6.remote_ip4,
4491 self.pg1.remote_ip4,
4494 proto=IP_PROTOS.tcp,
4498 self.vapi.nat44_interface_add_del_feature(
4499 sw_if_index=self.pg6.sw_if_index, is_add=1
4503 Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac)
4504 / IP(src=self.pg6.remote_ip4, dst=self.pg1.remote_ip4)
4505 / TCP(sport=12345, dport=external_port)
4507 self.pg6.add_stream(p)
4508 self.pg_enable_capture(self.pg_interfaces)
4510 capture = self.pg6.get_capture(1)
4515 self.assertEqual(ip.src, twice_nat_addr)
4516 self.assertNotEqual(tcp.sport, 12345)
4517 post_twice_nat_port = tcp.sport
4518 self.assertEqual(ip.dst, self.pg6.remote_ip4)
4519 self.assertEqual(tcp.dport, local_port)
4520 self.assert_packet_checksums_valid(p)
4522 self.logger.error(ppp("Unexpected or invalid packet:", p))
4526 Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac)
4527 / IP(src=self.pg6.remote_ip4, dst=twice_nat_addr)
4528 / TCP(sport=local_port, dport=post_twice_nat_port)
4530 self.pg6.add_stream(p)
4531 self.pg_enable_capture(self.pg_interfaces)
4533 capture = self.pg6.get_capture(1)
4538 self.assertEqual(ip.src, self.pg1.remote_ip4)
4539 self.assertEqual(tcp.sport, external_port)
4540 self.assertEqual(ip.dst, self.pg6.remote_ip4)
4541 self.assertEqual(tcp.dport, 12345)
4542 self.assert_packet_checksums_valid(p)
4544 self.logger.error(ppp("Unexpected or invalid packet:", p))
4547 def test_one_armed_nat44_static(self):
4548 """NAT44ED One armed NAT and 1:1 NAPT asymmetrical rule"""
4550 remote_host = self.pg4.remote_hosts[0]
4551 local_host = self.pg4.remote_hosts[1]
4556 self.vapi.nat44_forwarding_enable_disable(enable=1)
4557 self.nat_add_address(self.nat_addr, twice_nat=1)
4559 self.config_flags.NAT_IS_OUT2IN_ONLY | self.config_flags.NAT_IS_TWICE_NAT
4561 self.nat_add_static_mapping(
4566 proto=IP_PROTOS.tcp,
4569 flags = self.config_flags.NAT_IS_INSIDE
4570 self.vapi.nat44_interface_add_del_feature(
4571 sw_if_index=self.pg4.sw_if_index, is_add=1
4573 self.vapi.nat44_interface_add_del_feature(
4574 sw_if_index=self.pg4.sw_if_index, flags=flags, is_add=1
4577 # from client to service
4579 Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac)
4580 / IP(src=remote_host.ip4, dst=self.nat_addr)
4581 / TCP(sport=12345, dport=external_port)
4583 self.pg4.add_stream(p)
4584 self.pg_enable_capture(self.pg_interfaces)
4586 capture = self.pg4.get_capture(1)
4591 self.assertEqual(ip.dst, local_host.ip4)
4592 self.assertEqual(ip.src, self.nat_addr)
4593 self.assertEqual(tcp.dport, local_port)
4594 self.assertNotEqual(tcp.sport, 12345)
4595 eh_port_in = tcp.sport
4596 self.assert_packet_checksums_valid(p)
4598 self.logger.error(ppp("Unexpected or invalid packet:", p))
4601 # from service back to client
4603 Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac)
4604 / IP(src=local_host.ip4, dst=self.nat_addr)
4605 / TCP(sport=local_port, dport=eh_port_in)
4607 self.pg4.add_stream(p)
4608 self.pg_enable_capture(self.pg_interfaces)
4610 capture = self.pg4.get_capture(1)
4615 self.assertEqual(ip.src, self.nat_addr)
4616 self.assertEqual(ip.dst, remote_host.ip4)
4617 self.assertEqual(tcp.sport, external_port)
4618 self.assertEqual(tcp.dport, 12345)
4619 self.assert_packet_checksums_valid(p)
4621 self.logger.error(ppp("Unexpected or invalid packet:", p))
4624 def test_icmp_error_fwd_outbound(self):
4625 """NAT44ED ICMP error outbound with forwarding enabled"""
4627 # Ensure that an outbound ICMP error message is properly associated
4628 # with the inbound forward bypass session it is related to.
4631 self.nat_add_address(self.nat_addr)
4632 self.nat_add_inside_interface(self.pg0)
4633 self.nat_add_outside_interface(self.pg1)
4635 # enable forwarding and initiate connection out2in
4636 self.vapi.nat44_forwarding_enable_disable(enable=1)
4638 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4639 / IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4)
4640 / UDP(sport=21, dport=20)
4644 self.pg1.add_stream(p1)
4645 self.pg_enable_capture(self.pg_interfaces)
4647 capture = self.pg0.get_capture(1)[0]
4649 self.logger.info(self.vapi.cli("show nat44 sessions"))
4651 # reply with ICMP error message in2out
4652 # We cannot reliably retrieve forward bypass sessions via the API.
4653 # session dumps for a user will only look on the worker that the
4654 # user is supposed to be mapped to in2out. The forward bypass session
4655 # is not necessarily created on that worker.
4657 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4658 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4659 / ICMP(type="dest-unreach", code="port-unreachable")
4663 self.pg0.add_stream(p2)
4664 self.pg_enable_capture(self.pg_interfaces)
4666 capture = self.pg1.get_capture(1)[0]
4668 self.logger.info(self.vapi.cli("show nat44 sessions"))
4670 self.logger.info(ppp("p1 packet:", p1))
4671 self.logger.info(ppp("p2 packet:", p2))
4672 self.logger.info(ppp("capture packet:", capture))
4674 def test_tcp_session_open_retransmit1(self):
4675 """NAT44ED Open TCP session with SYN,ACK retransmit 1
4677 The client does not receive the [SYN,ACK] or the
4678 ACK from the client is lost. Therefore, the [SYN, ACK]
4679 is retransmitted by the server.
4682 in_port = self.tcp_port_in
4683 ext_port = self.tcp_external_port
4686 self.nat_add_address(self.nat_addr)
4687 self.nat_add_inside_interface(self.pg0)
4688 self.nat_add_outside_interface(self.pg1)
4690 self.vapi.nat_set_timeouts(
4691 udp=300, tcp_established=7440, tcp_transitory=5, icmp=60
4693 # SYN packet in->out
4695 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4696 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4697 / TCP(sport=in_port, dport=ext_port, flags="S")
4699 p = self.send_and_expect(self.pg0, p, self.pg1)[0]
4700 out_port = p[TCP].sport
4702 # SYN + ACK packet out->in
4704 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4705 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4706 / TCP(sport=ext_port, dport=out_port, flags="SA")
4708 self.send_and_expect(self.pg1, p, self.pg0)
4710 # ACK in->out does not arrive
4712 # resent SYN + ACK packet out->in
4714 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4715 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4716 / TCP(sport=ext_port, dport=out_port, flags="SA")
4718 self.send_and_expect(self.pg1, p, self.pg0)
4720 # ACK packet in->out
4722 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4723 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4724 / TCP(sport=in_port, dport=ext_port, flags="A")
4726 self.send_and_expect(self.pg0, p, self.pg1)
4728 # Verify that the data can be transmitted after the transitory time
4729 self.virtual_sleep(6)
4732 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4733 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4734 / TCP(sport=in_port, dport=ext_port, flags="PA")
4737 self.send_and_expect(self.pg0, p, self.pg1)
4739 def test_tcp_session_open_retransmit2(self):
4740 """NAT44ED Open TCP session with SYN,ACK retransmit 2
4742 The ACK is lost to the server after the TCP session is opened.
4743 Data is sent by the client, then the [SYN,ACK] is
4744 retransmitted by the server.
4747 in_port = self.tcp_port_in
4748 ext_port = self.tcp_external_port
4751 self.nat_add_address(self.nat_addr)
4752 self.nat_add_inside_interface(self.pg0)
4753 self.nat_add_outside_interface(self.pg1)
4755 self.vapi.nat_set_timeouts(
4756 udp=300, tcp_established=7440, tcp_transitory=5, icmp=60
4758 # SYN packet in->out
4760 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4761 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4762 / TCP(sport=in_port, dport=ext_port, flags="S")
4764 p = self.send_and_expect(self.pg0, p, self.pg1)[0]
4765 out_port = p[TCP].sport
4767 # SYN + ACK packet out->in
4769 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4770 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4771 / TCP(sport=ext_port, dport=out_port, flags="SA")
4773 self.send_and_expect(self.pg1, p, self.pg0)
4775 # ACK packet in->out -- not received by the server
4777 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4778 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4779 / TCP(sport=in_port, dport=ext_port, flags="A")
4781 self.send_and_expect(self.pg0, p, self.pg1)
4783 # PUSH + ACK packet in->out
4785 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4786 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4787 / TCP(sport=in_port, dport=ext_port, flags="PA")
4790 self.send_and_expect(self.pg0, p, self.pg1)
4792 # resent SYN + ACK packet out->in
4794 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4795 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4796 / TCP(sport=ext_port, dport=out_port, flags="SA")
4798 self.send_and_expect(self.pg1, p, self.pg0)
4800 # resent ACK packet in->out
4802 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4803 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4804 / TCP(sport=in_port, dport=ext_port, flags="A")
4806 self.send_and_expect(self.pg0, p, self.pg1)
4808 # resent PUSH + ACK packet in->out
4810 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4811 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4812 / TCP(sport=in_port, dport=ext_port, flags="PA")
4815 self.send_and_expect(self.pg0, p, self.pg1)
4817 # ACK packet out->in
4819 Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
4820 / IP(src=self.pg1.remote_ip4, dst=self.nat_addr)
4821 / TCP(sport=ext_port, dport=out_port, flags="A")
4823 self.send_and_expect(self.pg1, p, self.pg0)
4825 # Verify that the data can be transmitted after the transitory time
4826 self.virtual_sleep(6)
4829 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
4830 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
4831 / TCP(sport=in_port, dport=ext_port, flags="PA")
4834 self.send_and_expect(self.pg0, p, self.pg1)
4836 def test_dynamic_ports_exhausted(self):
4837 """NAT44ED dynamic translation test: address ports exhaused"""
4839 sessions_per_batch = 128
4840 n_available_ports = 65536 - 1024
4841 n_sessions = n_available_ports + 2 * sessions_per_batch
4843 # set high enough session limit for ports to be exhausted
4844 self.plugin_disable()
4845 self.plugin_enable(max_sessions=n_sessions)
4847 self.nat_add_inside_interface(self.pg0)
4848 self.nat_add_outside_interface(self.pg1)
4850 # set timeouts to high for sessions to reallistically expire
4851 config = self.vapi.nat44_show_running_config()
4852 old_timeouts = config.timeouts
4853 self.vapi.nat_set_timeouts(
4855 tcp_established=old_timeouts.tcp_established,
4856 tcp_transitory=old_timeouts.tcp_transitory,
4857 icmp=old_timeouts.icmp,
4860 # in2out after NAT addresses added
4861 self.nat_add_address(self.nat_addr)
4863 for i in range(n_sessions // sessions_per_batch):
4864 pkts = self.create_udp_stream(
4868 base_port=i * sessions_per_batch + 100,
4871 self.pg0.add_stream(pkts)
4874 err = self.statistics.get_err_counter(
4875 "/err/nat44-ed-in2out-slowpath/out of ports"
4877 if err > sessions_per_batch:
4880 # Check for ports to be used no more than once
4882 sessions = self.vapi.cli("show nat44 sessions")
4884 f" *o2i flow: match: saddr {self.pg1.remote_ip4} sport [0-9]+ daddr {self.nat_addr} dport ([0-9]+) proto UDP.*"
4886 for line in sessions.splitlines():
4889 port = int(m.groups()[0])
4890 self.assertNotIn(port, ports)
4893 self.assertGreaterEqual(err, sessions_per_batch)
4896 if __name__ == "__main__":
4897 unittest.main(testRunner=VppTestRunner)