5 from random import randint
8 from framework import VppTestCase, VppTestRunner
9 from scapy.data import IP_PROTOS
10 from scapy.layers.inet import IP, TCP, UDP, ICMP, GRE
11 from scapy.layers.inet import IPerror, TCPerror
12 from scapy.layers.l2 import Ether
13 from scapy.packet import Raw
14 from syslog_rfc5424_parser import SyslogMessage, ParseError
15 from syslog_rfc5424_parser.constants import SyslogSeverity
16 from util import ppp, ip4_range
17 from vpp_acl import AclRule, VppAcl, VppAclInterface
18 from vpp_ip_route import VppIpRoute, VppRoutePath
19 from vpp_papi import VppEnum
22 class NAT44EDTestCase(VppTestCase):
35 tcp_external_port = 80
40 super(NAT44EDTestCase, self).setUp()
44 super(NAT44EDTestCase, self).tearDown()
48 def plugin_enable(self):
49 self.vapi.nat44_ed_plugin_enable_disable(
50 sessions=self.max_sessions, enable=1)
52 def plugin_disable(self):
53 self.vapi.nat44_ed_plugin_enable_disable(enable=0)
56 def config_flags(self):
57 return VppEnum.vl_api_nat_config_flags_t
60 def nat44_config_flags(self):
61 return VppEnum.vl_api_nat44_config_flags_t
64 def syslog_severity(self):
65 return VppEnum.vl_api_syslog_severity_t
68 def server_addr(self):
69 return self.pg1.remote_hosts[0].ip4
73 return randint(1025, 65535)
76 def proto2layer(proto):
77 if proto == IP_PROTOS.tcp:
79 elif proto == IP_PROTOS.udp:
81 elif proto == IP_PROTOS.icmp:
84 raise Exception("Unsupported protocol")
87 def create_and_add_ip4_table(cls, i, table_id=0):
88 cls.vapi.ip_table_add_del(is_add=1, table={'table_id': table_id})
89 i.set_table_ip4(table_id)
92 def configure_ip4_interface(cls, i, hosts=0, table_id=None):
94 cls.create_and_add_ip4_table(i, table_id)
102 i.generate_remote_hosts(hosts)
103 i.configure_ipv4_neighbors()
106 def nat_add_interface_address(cls, i):
107 cls.vapi.nat44_add_del_interface_addr(
108 sw_if_index=i.sw_if_index, is_add=1)
110 def nat_add_inside_interface(self, i):
111 self.vapi.nat44_interface_add_del_feature(
112 flags=self.config_flags.NAT_IS_INSIDE,
113 sw_if_index=i.sw_if_index, is_add=1)
115 def nat_add_outside_interface(self, i):
116 self.vapi.nat44_interface_add_del_feature(
117 flags=self.config_flags.NAT_IS_OUTSIDE,
118 sw_if_index=i.sw_if_index, is_add=1)
120 def nat_add_address(self, address, twice_nat=0,
121 vrf_id=0xFFFFFFFF, is_add=1):
122 flags = self.config_flags.NAT_IS_TWICE_NAT if twice_nat else 0
123 self.vapi.nat44_add_del_address_range(first_ip_address=address,
124 last_ip_address=address,
129 def nat_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
130 local_port=0, external_port=0, vrf_id=0,
131 is_add=1, external_sw_if_index=0xFFFFFFFF,
132 proto=0, tag="", flags=0):
134 if not (local_port and external_port):
135 flags |= self.config_flags.NAT_IS_ADDR_ONLY
137 self.vapi.nat44_add_del_static_mapping(
139 local_ip_address=local_ip,
140 external_ip_address=external_ip,
141 external_sw_if_index=external_sw_if_index,
142 local_port=local_port,
143 external_port=external_port,
144 vrf_id=vrf_id, protocol=proto,
150 super(NAT44EDTestCase, cls).setUpClass()
152 cls.create_pg_interfaces(range(12))
153 cls.interfaces = list(cls.pg_interfaces[:4])
155 cls.create_and_add_ip4_table(cls.pg2, 10)
157 for i in cls.interfaces:
158 cls.configure_ip4_interface(i, hosts=3)
160 # test specific (test-multiple-vrf)
161 cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 1})
163 # test specific (test-one-armed-nat44-static)
164 cls.pg4.generate_remote_hosts(2)
166 cls.vapi.sw_interface_add_del_address(
167 sw_if_index=cls.pg4.sw_if_index,
168 prefix="10.0.0.1/24")
170 cls.pg4.resolve_arp()
171 cls.pg4._remote_hosts[1]._ip4 = cls.pg4._remote_hosts[0]._ip4
172 cls.pg4.resolve_arp()
174 # test specific interface (pg5)
175 cls.pg5._local_ip4 = "10.1.1.1"
176 cls.pg5._remote_hosts[0]._ip4 = "10.1.1.2"
177 cls.pg5.set_table_ip4(1)
180 cls.pg5.resolve_arp()
182 # test specific interface (pg6)
183 cls.pg6._local_ip4 = "10.1.2.1"
184 cls.pg6._remote_hosts[0]._ip4 = "10.1.2.2"
185 cls.pg6.set_table_ip4(1)
188 cls.pg6.resolve_arp()
192 rl.append(VppIpRoute(cls, "0.0.0.0", 0,
193 [VppRoutePath("0.0.0.0", 0xffffffff,
195 register=False, table_id=1))
196 rl.append(VppIpRoute(cls, "0.0.0.0", 0,
197 [VppRoutePath(cls.pg1.local_ip4,
198 cls.pg1.sw_if_index)],
200 rl.append(VppIpRoute(cls, cls.pg5.remote_ip4, 32,
201 [VppRoutePath("0.0.0.0",
202 cls.pg5.sw_if_index)],
203 register=False, table_id=1))
204 rl.append(VppIpRoute(cls, cls.pg6.remote_ip4, 32,
205 [VppRoutePath("0.0.0.0",
206 cls.pg6.sw_if_index)],
207 register=False, table_id=1))
208 rl.append(VppIpRoute(cls, cls.pg6.remote_ip4, 16,
209 [VppRoutePath("0.0.0.0", 0xffffffff,
211 register=False, table_id=0))
216 def get_err_counter(self, path):
217 return self.statistics.get_err_counter(path)
219 def get_stats_counter(self, path, worker=0):
220 return self.statistics.get_counter(path)[worker]
222 def reass_hairpinning(self, server_addr, server_in_port, server_out_port,
223 host_in_port, proto=IP_PROTOS.tcp,
225 layer = self.proto2layer(proto)
227 if proto == IP_PROTOS.tcp:
228 data = b"A" * 4 + b"B" * 16 + b"C" * 3
230 data = b"A" * 16 + b"B" * 16 + b"C" * 3
232 # send packet from host to server
233 pkts = self.create_stream_frag(self.pg0,
239 self.pg0.add_stream(pkts)
240 self.pg_enable_capture(self.pg_interfaces)
242 frags = self.pg0.get_capture(len(pkts))
243 p = self.reass_frags_and_verify(frags,
246 if proto != IP_PROTOS.icmp:
248 self.assertNotEqual(p[layer].sport, host_in_port)
249 self.assertEqual(p[layer].dport, server_in_port)
252 self.assertNotEqual(p[layer].id, host_in_port)
253 self.assertEqual(data, p[Raw].load)
255 def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False,
257 layer = self.proto2layer(proto)
259 if proto == IP_PROTOS.tcp:
260 data = b"A" * 4 + b"B" * 16 + b"C" * 3
262 data = b"A" * 16 + b"B" * 16 + b"C" * 3
263 self.port_in = self.random_port()
267 pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4,
268 self.port_in, 20, data, proto)
270 self.pg0.add_stream(pkts)
271 self.pg_enable_capture(self.pg_interfaces)
273 frags = self.pg1.get_capture(len(pkts))
274 if not dont_translate:
275 p = self.reass_frags_and_verify(frags,
279 p = self.reass_frags_and_verify(frags,
282 if proto != IP_PROTOS.icmp:
283 if not dont_translate:
284 self.assertEqual(p[layer].dport, 20)
286 self.assertNotEqual(p[layer].sport, self.port_in)
288 self.assertEqual(p[layer].sport, self.port_in)
291 if not dont_translate:
292 self.assertNotEqual(p[layer].id, self.port_in)
294 self.assertEqual(p[layer].id, self.port_in)
295 self.assertEqual(data, p[Raw].load)
298 if not dont_translate:
299 dst_addr = self.nat_addr
301 dst_addr = self.pg0.remote_ip4
302 if proto != IP_PROTOS.icmp:
304 dport = p[layer].sport
308 pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport,
309 data, proto, echo_reply=True)
311 self.pg1.add_stream(pkts)
312 self.pg_enable_capture(self.pg_interfaces)
313 self.logger.info(self.vapi.cli("show trace"))
315 frags = self.pg0.get_capture(len(pkts))
316 p = self.reass_frags_and_verify(frags,
319 if proto != IP_PROTOS.icmp:
320 self.assertEqual(p[layer].sport, 20)
321 self.assertEqual(p[layer].dport, self.port_in)
323 self.assertEqual(p[layer].id, self.port_in)
324 self.assertEqual(data, p[Raw].load)
326 def reass_frags_and_verify(self, frags, src, dst):
329 self.assertEqual(p[IP].src, src)
330 self.assertEqual(p[IP].dst, dst)
331 self.assert_ip_checksum_valid(p)
332 buffer.seek(p[IP].frag * 8)
333 buffer.write(bytes(p[IP].payload))
334 ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst,
335 proto=frags[0][IP].proto)
336 if ip.proto == IP_PROTOS.tcp:
337 p = (ip / TCP(buffer.getvalue()))
338 self.logger.debug(ppp("Reassembled:", p))
339 self.assert_tcp_checksum_valid(p)
340 elif ip.proto == IP_PROTOS.udp:
341 p = (ip / UDP(buffer.getvalue()[:8]) /
342 Raw(buffer.getvalue()[8:]))
343 elif ip.proto == IP_PROTOS.icmp:
344 p = (ip / ICMP(buffer.getvalue()))
347 def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False,
349 layer = self.proto2layer(proto)
351 if proto == IP_PROTOS.tcp:
352 data = b"A" * 4 + b"B" * 16 + b"C" * 3
354 data = b"A" * 16 + b"B" * 16 + b"C" * 3
355 self.port_in = self.random_port()
358 pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4,
359 self.port_in, 20, data, proto)
360 self.pg0.add_stream(pkts)
361 self.pg_enable_capture(self.pg_interfaces)
363 frags = self.pg1.get_capture(len(pkts))
364 if not dont_translate:
365 p = self.reass_frags_and_verify(frags,
369 p = self.reass_frags_and_verify(frags,
372 if proto != IP_PROTOS.icmp:
373 if not dont_translate:
374 self.assertEqual(p[layer].dport, 20)
376 self.assertNotEqual(p[layer].sport, self.port_in)
378 self.assertEqual(p[layer].sport, self.port_in)
381 if not dont_translate:
382 self.assertNotEqual(p[layer].id, self.port_in)
384 self.assertEqual(p[layer].id, self.port_in)
385 self.assertEqual(data, p[Raw].load)
388 if not dont_translate:
389 dst_addr = self.nat_addr
391 dst_addr = self.pg0.remote_ip4
392 if proto != IP_PROTOS.icmp:
394 dport = p[layer].sport
398 pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport, data,
399 proto, echo_reply=True)
400 self.pg1.add_stream(pkts)
401 self.pg_enable_capture(self.pg_interfaces)
403 frags = self.pg0.get_capture(len(pkts))
404 p = self.reass_frags_and_verify(frags,
407 if proto != IP_PROTOS.icmp:
408 self.assertEqual(p[layer].sport, 20)
409 self.assertEqual(p[layer].dport, self.port_in)
411 self.assertEqual(p[layer].id, self.port_in)
412 self.assertEqual(data, p[Raw].load)
414 def verify_capture_out(self, capture, nat_ip=None, same_port=False,
415 dst_ip=None, ignore_port=False):
417 nat_ip = self.nat_addr
418 for packet in capture:
420 self.assert_packet_checksums_valid(packet)
421 self.assertEqual(packet[IP].src, nat_ip)
422 if dst_ip is not None:
423 self.assertEqual(packet[IP].dst, dst_ip)
424 if packet.haslayer(TCP):
428 packet[TCP].sport, self.tcp_port_in)
431 packet[TCP].sport, self.tcp_port_in)
432 self.tcp_port_out = packet[TCP].sport
433 self.assert_packet_checksums_valid(packet)
434 elif packet.haslayer(UDP):
438 packet[UDP].sport, self.udp_port_in)
441 packet[UDP].sport, self.udp_port_in)
442 self.udp_port_out = packet[UDP].sport
447 packet[ICMP].id, self.icmp_id_in)
450 packet[ICMP].id, self.icmp_id_in)
451 self.icmp_id_out = packet[ICMP].id
452 self.assert_packet_checksums_valid(packet)
454 self.logger.error(ppp("Unexpected or invalid packet "
455 "(outside network):", packet))
458 def verify_capture_in(self, capture, in_if):
459 for packet in capture:
461 self.assert_packet_checksums_valid(packet)
462 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
463 if packet.haslayer(TCP):
464 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
465 elif packet.haslayer(UDP):
466 self.assertEqual(packet[UDP].dport, self.udp_port_in)
468 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
470 self.logger.error(ppp("Unexpected or invalid packet "
471 "(inside network):", packet))
474 def create_stream_in(self, in_if, out_if, dst_ip=None, ttl=64):
476 dst_ip = out_if.remote_ip4
480 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
481 IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
482 TCP(sport=self.tcp_port_in, dport=20))
486 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
487 IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
488 UDP(sport=self.udp_port_in, dport=20))
492 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
493 IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
494 ICMP(id=self.icmp_id_in, type='echo-request'))
499 def create_stream_out(self, out_if, dst_ip=None, ttl=64,
500 use_inside_ports=False):
502 dst_ip = self.nat_addr
503 if not use_inside_ports:
504 tcp_port = self.tcp_port_out
505 udp_port = self.udp_port_out
506 icmp_id = self.icmp_id_out
508 tcp_port = self.tcp_port_in
509 udp_port = self.udp_port_in
510 icmp_id = self.icmp_id_in
513 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
514 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
515 TCP(dport=tcp_port, sport=20))
519 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
520 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
521 UDP(dport=udp_port, sport=20))
525 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
526 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
527 ICMP(id=icmp_id, type='echo-reply'))
532 def create_tcp_stream(self, in_if, out_if, count):
536 for i in range(count):
537 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
538 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=64) /
539 TCP(sport=port + i, dport=20))
544 def create_stream_frag(self, src_if, dst, sport, dport, data,
545 proto=IP_PROTOS.tcp, echo_reply=False):
546 if proto == IP_PROTOS.tcp:
547 p = (IP(src=src_if.remote_ip4, dst=dst) /
548 TCP(sport=sport, dport=dport) /
550 p = p.__class__(scapy.compat.raw(p))
551 chksum = p[TCP].chksum
552 proto_header = TCP(sport=sport, dport=dport, chksum=chksum)
553 elif proto == IP_PROTOS.udp:
554 proto_header = UDP(sport=sport, dport=dport)
555 elif proto == IP_PROTOS.icmp:
557 proto_header = ICMP(id=sport, type='echo-request')
559 proto_header = ICMP(id=sport, type='echo-reply')
561 raise Exception("Unsupported protocol")
562 id = self.random_port()
564 if proto == IP_PROTOS.tcp:
567 raw = Raw(data[0:16])
568 p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
569 IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=0, id=id) /
573 if proto == IP_PROTOS.tcp:
574 raw = Raw(data[4:20])
576 raw = Raw(data[16:32])
577 p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
578 IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=3, id=id,
582 if proto == IP_PROTOS.tcp:
586 p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
587 IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=proto,
593 def frag_in_order_in_plus_out(self, in_addr, out_addr, in_port, out_port,
594 proto=IP_PROTOS.tcp):
596 layer = self.proto2layer(proto)
598 if proto == IP_PROTOS.tcp:
599 data = b"A" * 4 + b"B" * 16 + b"C" * 3
601 data = b"A" * 16 + b"B" * 16 + b"C" * 3
602 port_in = self.random_port()
606 pkts = self.create_stream_frag(self.pg0, out_addr,
609 self.pg0.add_stream(pkts)
610 self.pg_enable_capture(self.pg_interfaces)
612 frags = self.pg1.get_capture(len(pkts))
613 p = self.reass_frags_and_verify(frags,
616 if proto != IP_PROTOS.icmp:
617 self.assertEqual(p[layer].sport, port_in)
618 self.assertEqual(p[layer].dport, in_port)
620 self.assertEqual(p[layer].id, port_in)
621 self.assertEqual(data, p[Raw].load)
624 if proto != IP_PROTOS.icmp:
625 pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4,
627 p[layer].sport, data, proto)
629 pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4,
630 p[layer].id, 0, data, proto,
632 self.pg1.add_stream(pkts)
633 self.pg_enable_capture(self.pg_interfaces)
635 frags = self.pg0.get_capture(len(pkts))
636 p = self.reass_frags_and_verify(frags,
639 if proto != IP_PROTOS.icmp:
640 self.assertEqual(p[layer].sport, out_port)
641 self.assertEqual(p[layer].dport, port_in)
643 self.assertEqual(p[layer].id, port_in)
644 self.assertEqual(data, p[Raw].load)
646 def frag_out_of_order_in_plus_out(self, in_addr, out_addr, in_port,
647 out_port, proto=IP_PROTOS.tcp):
649 layer = self.proto2layer(proto)
651 if proto == IP_PROTOS.tcp:
652 data = b"A" * 4 + b"B" * 16 + b"C" * 3
654 data = b"A" * 16 + b"B" * 16 + b"C" * 3
655 port_in = self.random_port()
659 pkts = self.create_stream_frag(self.pg0, out_addr,
663 self.pg0.add_stream(pkts)
664 self.pg_enable_capture(self.pg_interfaces)
666 frags = self.pg1.get_capture(len(pkts))
667 p = self.reass_frags_and_verify(frags,
670 if proto != IP_PROTOS.icmp:
671 self.assertEqual(p[layer].dport, in_port)
672 self.assertEqual(p[layer].sport, port_in)
673 self.assertEqual(p[layer].dport, in_port)
675 self.assertEqual(p[layer].id, port_in)
676 self.assertEqual(data, p[Raw].load)
679 if proto != IP_PROTOS.icmp:
680 pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4,
682 p[layer].sport, data, proto)
684 pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4,
685 p[layer].id, 0, data, proto,
688 self.pg1.add_stream(pkts)
689 self.pg_enable_capture(self.pg_interfaces)
691 frags = self.pg0.get_capture(len(pkts))
692 p = self.reass_frags_and_verify(frags,
695 if proto != IP_PROTOS.icmp:
696 self.assertEqual(p[layer].sport, out_port)
697 self.assertEqual(p[layer].dport, port_in)
699 self.assertEqual(p[layer].id, port_in)
700 self.assertEqual(data, p[Raw].load)
702 def init_tcp_session(self, in_if, out_if, in_port, ext_port):
704 p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
705 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
706 TCP(sport=in_port, dport=ext_port, flags="S"))
708 self.pg_enable_capture(self.pg_interfaces)
710 capture = out_if.get_capture(1)
712 out_port = p[TCP].sport
714 # SYN + ACK packet out->in
715 p = (Ether(src=out_if.remote_mac, dst=out_if.local_mac) /
716 IP(src=out_if.remote_ip4, dst=self.nat_addr) /
717 TCP(sport=ext_port, dport=out_port, flags="SA"))
719 self.pg_enable_capture(self.pg_interfaces)
724 p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
725 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
726 TCP(sport=in_port, dport=ext_port, flags="A"))
728 self.pg_enable_capture(self.pg_interfaces)
730 out_if.get_capture(1)
734 def twice_nat_common(self, self_twice_nat=False, same_pg=False, lb=False,
736 twice_nat_addr = '10.0.1.3'
744 port_in1 = port_in + 1
745 port_in2 = port_in + 2
750 server1 = self.pg0.remote_hosts[0]
751 server2 = self.pg0.remote_hosts[1]
763 eh_translate = ((not self_twice_nat) or (not lb and same_pg) or
766 self.nat_add_address(self.nat_addr)
767 self.nat_add_address(twice_nat_addr, twice_nat=1)
771 flags |= self.config_flags.NAT_IS_SELF_TWICE_NAT
773 flags |= self.config_flags.NAT_IS_TWICE_NAT
776 self.nat_add_static_mapping(pg0.remote_ip4, self.nat_addr,
781 locals = [{'addr': server1.ip4,
785 {'addr': server2.ip4,
789 out_addr = self.nat_addr
791 self.vapi.nat44_add_del_lb_static_mapping(is_add=1, flags=flags,
792 external_addr=out_addr,
793 external_port=port_out,
794 protocol=IP_PROTOS.tcp,
795 local_num=len(locals),
797 self.nat_add_inside_interface(pg0)
798 self.nat_add_outside_interface(pg1)
804 assert client_id is not None
806 client = self.pg0.remote_hosts[0]
808 client = self.pg0.remote_hosts[1]
810 client = pg1.remote_hosts[0]
811 p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
812 IP(src=client.ip4, dst=self.nat_addr) /
813 TCP(sport=eh_port_out, dport=port_out))
815 self.pg_enable_capture(self.pg_interfaces)
817 capture = pg0.get_capture(1)
823 if ip.dst == server1.ip4:
829 self.assertEqual(ip.dst, server.ip4)
831 self.assertIn(tcp.dport, [port_in1, port_in2])
833 self.assertEqual(tcp.dport, port_in)
835 self.assertEqual(ip.src, twice_nat_addr)
836 self.assertNotEqual(tcp.sport, eh_port_out)
838 self.assertEqual(ip.src, client.ip4)
839 self.assertEqual(tcp.sport, eh_port_out)
841 eh_port_in = tcp.sport
842 saved_port_in = tcp.dport
843 self.assert_packet_checksums_valid(p)
845 self.logger.error(ppp("Unexpected or invalid packet:", p))
848 p = (Ether(src=server.mac, dst=pg0.local_mac) /
849 IP(src=server.ip4, dst=eh_addr_in) /
850 TCP(sport=saved_port_in, dport=eh_port_in))
852 self.pg_enable_capture(self.pg_interfaces)
854 capture = pg1.get_capture(1)
859 self.assertEqual(ip.dst, client.ip4)
860 self.assertEqual(ip.src, self.nat_addr)
861 self.assertEqual(tcp.dport, eh_port_out)
862 self.assertEqual(tcp.sport, port_out)
863 self.assert_packet_checksums_valid(p)
865 self.logger.error(ppp("Unexpected or invalid packet:", p))
869 sessions = self.vapi.nat44_user_session_dump(server.ip4, 0)
870 self.assertEqual(len(sessions), 1)
871 self.assertTrue(sessions[0].flags &
872 self.config_flags.NAT_IS_EXT_HOST_VALID)
873 self.assertTrue(sessions[0].flags &
874 self.config_flags.NAT_IS_TWICE_NAT)
875 self.logger.info(self.vapi.cli("show nat44 sessions"))
876 self.vapi.nat44_del_session(
877 address=sessions[0].inside_ip_address,
878 port=sessions[0].inside_port,
879 protocol=sessions[0].protocol,
880 flags=(self.config_flags.NAT_IS_INSIDE |
881 self.config_flags.NAT_IS_EXT_HOST_VALID),
882 ext_host_address=sessions[0].ext_host_nat_address,
883 ext_host_port=sessions[0].ext_host_nat_port)
884 sessions = self.vapi.nat44_user_session_dump(server.ip4, 0)
885 self.assertEqual(len(sessions), 0)
887 def verify_syslog_sess(self, data, is_add=True, is_ip6=False):
888 message = data.decode('utf-8')
890 message = SyslogMessage.parse(message)
891 except ParseError as e:
895 self.assertEqual(message.severity, SyslogSeverity.info)
896 self.assertEqual(message.appname, 'NAT')
897 self.assertEqual(message.msgid, 'SADD' if is_add else 'SDEL')
898 sd_params = message.sd.get('nsess')
899 self.assertTrue(sd_params is not None)
901 self.assertEqual(sd_params.get('IATYP'), 'IPv6')
902 self.assertEqual(sd_params.get('ISADDR'), self.pg0.remote_ip6)
904 self.assertEqual(sd_params.get('IATYP'), 'IPv4')
905 self.assertEqual(sd_params.get('ISADDR'), self.pg0.remote_ip4)
906 self.assertTrue(sd_params.get('SSUBIX') is not None)
907 self.assertEqual(sd_params.get('ISPORT'), "%d" % self.tcp_port_in)
908 self.assertEqual(sd_params.get('XATYP'), 'IPv4')
909 self.assertEqual(sd_params.get('XSADDR'), self.nat_addr)
910 self.assertEqual(sd_params.get('XSPORT'), "%d" % self.tcp_port_out)
911 self.assertEqual(sd_params.get('PROTO'), "%d" % IP_PROTOS.tcp)
912 self.assertEqual(sd_params.get('SVLAN'), '0')
913 self.assertEqual(sd_params.get('XDADDR'), self.pg1.remote_ip4)
914 self.assertEqual(sd_params.get('XDPORT'),
915 "%d" % self.tcp_external_port)
918 class TestNAT44ED(NAT44EDTestCase):
919 """ NAT44ED Test Case """
921 def test_users_dump(self):
922 """ NAT44ED API test - nat44_user_dump """
924 self.nat_add_address(self.nat_addr)
925 self.nat_add_inside_interface(self.pg0)
926 self.nat_add_outside_interface(self.pg1)
928 self.vapi.nat44_forwarding_enable_disable(enable=1)
930 local_ip = self.pg0.remote_ip4
931 external_ip = self.nat_addr
932 self.nat_add_static_mapping(local_ip, external_ip)
934 users = self.vapi.nat44_user_dump()
935 self.assertEqual(len(users), 0)
937 # in2out - static mapping match
939 pkts = self.create_stream_out(self.pg1)
940 self.pg1.add_stream(pkts)
941 self.pg_enable_capture(self.pg_interfaces)
943 capture = self.pg0.get_capture(len(pkts))
944 self.verify_capture_in(capture, self.pg0)
946 pkts = self.create_stream_in(self.pg0, self.pg1)
947 self.pg0.add_stream(pkts)
948 self.pg_enable_capture(self.pg_interfaces)
950 capture = self.pg1.get_capture(len(pkts))
951 self.verify_capture_out(capture, same_port=True)
953 users = self.vapi.nat44_user_dump()
954 self.assertEqual(len(users), 1)
955 static_user = users[0]
956 self.assertEqual(static_user.nstaticsessions, 3)
957 self.assertEqual(static_user.nsessions, 0)
959 # in2out - no static mapping match (forwarding test)
961 host0 = self.pg0.remote_hosts[0]
962 self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1]
964 pkts = self.create_stream_out(self.pg1,
965 dst_ip=self.pg0.remote_ip4,
966 use_inside_ports=True)
967 self.pg1.add_stream(pkts)
968 self.pg_enable_capture(self.pg_interfaces)
970 capture = self.pg0.get_capture(len(pkts))
971 self.verify_capture_in(capture, self.pg0)
973 pkts = self.create_stream_in(self.pg0, self.pg1)
974 self.pg0.add_stream(pkts)
975 self.pg_enable_capture(self.pg_interfaces)
977 capture = self.pg1.get_capture(len(pkts))
978 self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4,
981 self.pg0.remote_hosts[0] = host0
983 users = self.vapi.nat44_user_dump()
984 self.assertEqual(len(users), 2)
985 if str(users[0].ip_address) == self.pg0.remote_hosts[0].ip4:
986 non_static_user = users[1]
987 static_user = users[0]
989 non_static_user = users[0]
990 static_user = users[1]
991 self.assertEqual(static_user.nstaticsessions, 3)
992 self.assertEqual(static_user.nsessions, 0)
993 self.assertEqual(non_static_user.nstaticsessions, 0)
994 self.assertEqual(non_static_user.nsessions, 3)
996 users = self.vapi.nat44_user_dump()
997 self.assertEqual(len(users), 2)
998 if str(users[0].ip_address) == self.pg0.remote_hosts[0].ip4:
999 non_static_user = users[1]
1000 static_user = users[0]
1002 non_static_user = users[0]
1003 static_user = users[1]
1004 self.assertEqual(static_user.nstaticsessions, 3)
1005 self.assertEqual(static_user.nsessions, 0)
1006 self.assertEqual(non_static_user.nstaticsessions, 0)
1007 self.assertEqual(non_static_user.nsessions, 3)
1009 def test_frag_out_of_order_do_not_translate(self):
1010 """ NAT44ED don't translate fragments arriving out of order """
1011 self.nat_add_inside_interface(self.pg0)
1012 self.nat_add_outside_interface(self.pg1)
1013 self.vapi.nat44_forwarding_enable_disable(enable=True)
1014 self.frag_out_of_order(proto=IP_PROTOS.tcp, dont_translate=True)
1016 def test_forwarding(self):
1017 """ NAT44ED forwarding test """
1019 self.nat_add_inside_interface(self.pg0)
1020 self.nat_add_outside_interface(self.pg1)
1021 self.vapi.nat44_forwarding_enable_disable(enable=1)
1023 real_ip = self.pg0.remote_ip4
1024 alias_ip = self.nat_addr
1025 flags = self.config_flags.NAT_IS_ADDR_ONLY
1026 self.vapi.nat44_add_del_static_mapping(is_add=1,
1027 local_ip_address=real_ip,
1028 external_ip_address=alias_ip,
1029 external_sw_if_index=0xFFFFFFFF,
1033 # in2out - static mapping match
1035 pkts = self.create_stream_out(self.pg1)
1036 self.pg1.add_stream(pkts)
1037 self.pg_enable_capture(self.pg_interfaces)
1039 capture = self.pg0.get_capture(len(pkts))
1040 self.verify_capture_in(capture, self.pg0)
1042 pkts = self.create_stream_in(self.pg0, self.pg1)
1043 self.pg0.add_stream(pkts)
1044 self.pg_enable_capture(self.pg_interfaces)
1046 capture = self.pg1.get_capture(len(pkts))
1047 self.verify_capture_out(capture, same_port=True)
1049 # in2out - no static mapping match
1051 host0 = self.pg0.remote_hosts[0]
1052 self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1]
1054 pkts = self.create_stream_out(self.pg1,
1055 dst_ip=self.pg0.remote_ip4,
1056 use_inside_ports=True)
1057 self.pg1.add_stream(pkts)
1058 self.pg_enable_capture(self.pg_interfaces)
1060 capture = self.pg0.get_capture(len(pkts))
1061 self.verify_capture_in(capture, self.pg0)
1063 pkts = self.create_stream_in(self.pg0, self.pg1)
1064 self.pg0.add_stream(pkts)
1065 self.pg_enable_capture(self.pg_interfaces)
1067 capture = self.pg1.get_capture(len(pkts))
1068 self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4,
1071 self.pg0.remote_hosts[0] = host0
1073 user = self.pg0.remote_hosts[1]
1074 sessions = self.vapi.nat44_user_session_dump(user.ip4, 0)
1075 self.assertEqual(len(sessions), 3)
1076 self.assertTrue(sessions[0].flags &
1077 self.config_flags.NAT_IS_EXT_HOST_VALID)
1078 self.vapi.nat44_del_session(
1079 address=sessions[0].inside_ip_address,
1080 port=sessions[0].inside_port,
1081 protocol=sessions[0].protocol,
1082 flags=(self.config_flags.NAT_IS_INSIDE |
1083 self.config_flags.NAT_IS_EXT_HOST_VALID),
1084 ext_host_address=sessions[0].ext_host_address,
1085 ext_host_port=sessions[0].ext_host_port)
1086 sessions = self.vapi.nat44_user_session_dump(user.ip4, 0)
1087 self.assertEqual(len(sessions), 2)
1090 self.vapi.nat44_forwarding_enable_disable(enable=0)
1091 flags = self.config_flags.NAT_IS_ADDR_ONLY
1092 self.vapi.nat44_add_del_static_mapping(
1094 local_ip_address=real_ip,
1095 external_ip_address=alias_ip,
1096 external_sw_if_index=0xFFFFFFFF,
1099 def test_output_feature_and_service2(self):
1100 """ NAT44ED interface output feature and service host direct access """
1101 self.vapi.nat44_forwarding_enable_disable(enable=1)
1102 self.nat_add_address(self.nat_addr)
1104 self.vapi.nat44_interface_add_del_output_feature(
1105 sw_if_index=self.pg1.sw_if_index, is_add=1,)
1107 # session initiated from service host - translate
1108 pkts = self.create_stream_in(self.pg0, self.pg1)
1109 self.pg0.add_stream(pkts)
1110 self.pg_enable_capture(self.pg_interfaces)
1112 capture = self.pg1.get_capture(len(pkts))
1113 self.verify_capture_out(capture, ignore_port=True)
1115 pkts = self.create_stream_out(self.pg1)
1116 self.pg1.add_stream(pkts)
1117 self.pg_enable_capture(self.pg_interfaces)
1119 capture = self.pg0.get_capture(len(pkts))
1120 self.verify_capture_in(capture, self.pg0)
1122 # session initiated from remote host - do not translate
1123 tcp_port_in = self.tcp_port_in
1124 udp_port_in = self.udp_port_in
1125 icmp_id_in = self.icmp_id_in
1127 self.tcp_port_in = 60303
1128 self.udp_port_in = 60304
1129 self.icmp_id_in = 60305
1132 pkts = self.create_stream_out(self.pg1,
1133 self.pg0.remote_ip4,
1134 use_inside_ports=True)
1135 self.pg1.add_stream(pkts)
1136 self.pg_enable_capture(self.pg_interfaces)
1138 capture = self.pg0.get_capture(len(pkts))
1139 self.verify_capture_in(capture, self.pg0)
1141 pkts = self.create_stream_in(self.pg0, self.pg1)
1142 self.pg0.add_stream(pkts)
1143 self.pg_enable_capture(self.pg_interfaces)
1145 capture = self.pg1.get_capture(len(pkts))
1146 self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4,
1149 self.tcp_port_in = tcp_port_in
1150 self.udp_port_in = udp_port_in
1151 self.icmp_id_in = icmp_id_in
1153 def test_twice_nat(self):
1154 """ NAT44ED Twice NAT """
1155 self.twice_nat_common()
1157 def test_self_twice_nat_positive(self):
1158 """ NAT44ED Self Twice NAT (positive test) """
1159 self.twice_nat_common(self_twice_nat=True, same_pg=True)
1161 def test_self_twice_nat_lb_positive(self):
1162 """ NAT44ED Self Twice NAT local service load balancing (positive test)
1164 self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True,
1167 def test_twice_nat_lb(self):
1168 """ NAT44ED Twice NAT local service load balancing """
1169 self.twice_nat_common(lb=True)
1171 def test_output_feature(self):
1172 """ NAT44ED interface output feature (in2out postrouting) """
1173 self.vapi.nat44_forwarding_enable_disable(enable=1)
1174 self.nat_add_address(self.nat_addr)
1176 self.nat_add_outside_interface(self.pg0)
1177 self.vapi.nat44_interface_add_del_output_feature(
1178 sw_if_index=self.pg1.sw_if_index, is_add=1)
1181 pkts = self.create_stream_in(self.pg0, self.pg1)
1182 self.pg0.add_stream(pkts)
1183 self.pg_enable_capture(self.pg_interfaces)
1185 capture = self.pg1.get_capture(len(pkts))
1186 self.verify_capture_out(capture, ignore_port=True)
1189 pkts = self.create_stream_out(self.pg1)
1190 self.pg1.add_stream(pkts)
1191 self.pg_enable_capture(self.pg_interfaces)
1193 capture = self.pg0.get_capture(len(pkts))
1194 self.verify_capture_in(capture, self.pg0)
1196 def test_static_with_port_out2(self):
1197 """ NAT44ED 1:1 NAPT asymmetrical rule """
1202 self.vapi.nat44_forwarding_enable_disable(enable=1)
1203 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
1204 self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1205 local_port, external_port,
1206 proto=IP_PROTOS.tcp, flags=flags)
1208 self.nat_add_inside_interface(self.pg0)
1209 self.nat_add_outside_interface(self.pg1)
1211 # from client to service
1212 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1213 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1214 TCP(sport=12345, dport=external_port))
1215 self.pg1.add_stream(p)
1216 self.pg_enable_capture(self.pg_interfaces)
1218 capture = self.pg0.get_capture(1)
1223 self.assertEqual(ip.dst, self.pg0.remote_ip4)
1224 self.assertEqual(tcp.dport, local_port)
1225 self.assert_packet_checksums_valid(p)
1227 self.logger.error(ppp("Unexpected or invalid packet:", p))
1231 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1232 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1233 ICMP(type=11) / capture[0][IP])
1234 self.pg0.add_stream(p)
1235 self.pg_enable_capture(self.pg_interfaces)
1237 capture = self.pg1.get_capture(1)
1240 self.assertEqual(p[IP].src, self.nat_addr)
1242 self.assertEqual(inner.dst, self.nat_addr)
1243 self.assertEqual(inner[TCPerror].dport, external_port)
1245 self.logger.error(ppp("Unexpected or invalid packet:", p))
1248 # from service back to client
1249 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1250 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1251 TCP(sport=local_port, dport=12345))
1252 self.pg0.add_stream(p)
1253 self.pg_enable_capture(self.pg_interfaces)
1255 capture = self.pg1.get_capture(1)
1260 self.assertEqual(ip.src, self.nat_addr)
1261 self.assertEqual(tcp.sport, external_port)
1262 self.assert_packet_checksums_valid(p)
1264 self.logger.error(ppp("Unexpected or invalid packet:", p))
1268 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1269 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1270 ICMP(type=11) / capture[0][IP])
1271 self.pg1.add_stream(p)
1272 self.pg_enable_capture(self.pg_interfaces)
1274 capture = self.pg0.get_capture(1)
1277 self.assertEqual(p[IP].dst, self.pg0.remote_ip4)
1279 self.assertEqual(inner.src, self.pg0.remote_ip4)
1280 self.assertEqual(inner[TCPerror].sport, local_port)
1282 self.logger.error(ppp("Unexpected or invalid packet:", p))
1285 # from client to server (no translation)
1286 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1287 IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) /
1288 TCP(sport=12346, dport=local_port))
1289 self.pg1.add_stream(p)
1290 self.pg_enable_capture(self.pg_interfaces)
1292 capture = self.pg0.get_capture(1)
1297 self.assertEqual(ip.dst, self.pg0.remote_ip4)
1298 self.assertEqual(tcp.dport, local_port)
1299 self.assert_packet_checksums_valid(p)
1301 self.logger.error(ppp("Unexpected or invalid packet:", p))
1304 # from service back to client (no translation)
1305 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1306 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1307 TCP(sport=local_port, dport=12346))
1308 self.pg0.add_stream(p)
1309 self.pg_enable_capture(self.pg_interfaces)
1311 capture = self.pg1.get_capture(1)
1316 self.assertEqual(ip.src, self.pg0.remote_ip4)
1317 self.assertEqual(tcp.sport, local_port)
1318 self.assert_packet_checksums_valid(p)
1320 self.logger.error(ppp("Unexpected or invalid packet:", p))
1323 def test_static_lb(self):
1324 """ NAT44ED local service load balancing """
1325 external_addr_n = self.nat_addr
1328 server1 = self.pg0.remote_hosts[0]
1329 server2 = self.pg0.remote_hosts[1]
1331 locals = [{'addr': server1.ip4,
1335 {'addr': server2.ip4,
1340 self.nat_add_address(self.nat_addr)
1341 self.vapi.nat44_add_del_lb_static_mapping(
1343 external_addr=external_addr_n,
1344 external_port=external_port,
1345 protocol=IP_PROTOS.tcp,
1346 local_num=len(locals),
1348 flags = self.config_flags.NAT_IS_INSIDE
1349 self.vapi.nat44_interface_add_del_feature(
1350 sw_if_index=self.pg0.sw_if_index,
1351 flags=flags, is_add=1)
1352 self.vapi.nat44_interface_add_del_feature(
1353 sw_if_index=self.pg1.sw_if_index,
1356 # from client to service
1357 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1358 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1359 TCP(sport=12345, dport=external_port))
1360 self.pg1.add_stream(p)
1361 self.pg_enable_capture(self.pg_interfaces)
1363 capture = self.pg0.get_capture(1)
1369 self.assertIn(ip.dst, [server1.ip4, server2.ip4])
1370 if ip.dst == server1.ip4:
1374 self.assertEqual(tcp.dport, local_port)
1375 self.assert_packet_checksums_valid(p)
1377 self.logger.error(ppp("Unexpected or invalid packet:", p))
1380 # from service back to client
1381 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
1382 IP(src=server.ip4, dst=self.pg1.remote_ip4) /
1383 TCP(sport=local_port, dport=12345))
1384 self.pg0.add_stream(p)
1385 self.pg_enable_capture(self.pg_interfaces)
1387 capture = self.pg1.get_capture(1)
1392 self.assertEqual(ip.src, self.nat_addr)
1393 self.assertEqual(tcp.sport, external_port)
1394 self.assert_packet_checksums_valid(p)
1396 self.logger.error(ppp("Unexpected or invalid packet:", p))
1399 sessions = self.vapi.nat44_user_session_dump(server.ip4, 0)
1400 self.assertEqual(len(sessions), 1)
1401 self.assertTrue(sessions[0].flags &
1402 self.config_flags.NAT_IS_EXT_HOST_VALID)
1403 self.vapi.nat44_del_session(
1404 address=sessions[0].inside_ip_address,
1405 port=sessions[0].inside_port,
1406 protocol=sessions[0].protocol,
1407 flags=(self.config_flags.NAT_IS_INSIDE |
1408 self.config_flags.NAT_IS_EXT_HOST_VALID),
1409 ext_host_address=sessions[0].ext_host_address,
1410 ext_host_port=sessions[0].ext_host_port)
1411 sessions = self.vapi.nat44_user_session_dump(server.ip4, 0)
1412 self.assertEqual(len(sessions), 0)
1414 def test_static_lb_2(self):
1415 """ NAT44ED local service load balancing (asymmetrical rule) """
1416 external_addr = self.nat_addr
1419 server1 = self.pg0.remote_hosts[0]
1420 server2 = self.pg0.remote_hosts[1]
1422 locals = [{'addr': server1.ip4,
1426 {'addr': server2.ip4,
1431 self.vapi.nat44_forwarding_enable_disable(enable=1)
1432 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
1433 self.vapi.nat44_add_del_lb_static_mapping(is_add=1, flags=flags,
1434 external_addr=external_addr,
1435 external_port=external_port,
1436 protocol=IP_PROTOS.tcp,
1437 local_num=len(locals),
1439 flags = self.config_flags.NAT_IS_INSIDE
1440 self.vapi.nat44_interface_add_del_feature(
1441 sw_if_index=self.pg0.sw_if_index,
1442 flags=flags, is_add=1)
1443 self.vapi.nat44_interface_add_del_feature(
1444 sw_if_index=self.pg1.sw_if_index,
1447 # from client to service
1448 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1449 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1450 TCP(sport=12345, dport=external_port))
1451 self.pg1.add_stream(p)
1452 self.pg_enable_capture(self.pg_interfaces)
1454 capture = self.pg0.get_capture(1)
1460 self.assertIn(ip.dst, [server1.ip4, server2.ip4])
1461 if ip.dst == server1.ip4:
1465 self.assertEqual(tcp.dport, local_port)
1466 self.assert_packet_checksums_valid(p)
1468 self.logger.error(ppp("Unexpected or invalid packet:", p))
1471 # from service back to client
1472 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
1473 IP(src=server.ip4, dst=self.pg1.remote_ip4) /
1474 TCP(sport=local_port, dport=12345))
1475 self.pg0.add_stream(p)
1476 self.pg_enable_capture(self.pg_interfaces)
1478 capture = self.pg1.get_capture(1)
1483 self.assertEqual(ip.src, self.nat_addr)
1484 self.assertEqual(tcp.sport, external_port)
1485 self.assert_packet_checksums_valid(p)
1487 self.logger.error(ppp("Unexpected or invalid packet:", p))
1490 # from client to server (no translation)
1491 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1492 IP(src=self.pg1.remote_ip4, dst=server1.ip4) /
1493 TCP(sport=12346, dport=local_port))
1494 self.pg1.add_stream(p)
1495 self.pg_enable_capture(self.pg_interfaces)
1497 capture = self.pg0.get_capture(1)
1503 self.assertEqual(ip.dst, server1.ip4)
1504 self.assertEqual(tcp.dport, local_port)
1505 self.assert_packet_checksums_valid(p)
1507 self.logger.error(ppp("Unexpected or invalid packet:", p))
1510 # from service back to client (no translation)
1511 p = (Ether(src=server1.mac, dst=self.pg0.local_mac) /
1512 IP(src=server1.ip4, dst=self.pg1.remote_ip4) /
1513 TCP(sport=local_port, dport=12346))
1514 self.pg0.add_stream(p)
1515 self.pg_enable_capture(self.pg_interfaces)
1517 capture = self.pg1.get_capture(1)
1522 self.assertEqual(ip.src, server1.ip4)
1523 self.assertEqual(tcp.sport, local_port)
1524 self.assert_packet_checksums_valid(p)
1526 self.logger.error(ppp("Unexpected or invalid packet:", p))
1529 def test_lb_affinity(self):
1530 """ NAT44ED local service load balancing affinity """
1531 external_addr = self.nat_addr
1534 server1 = self.pg0.remote_hosts[0]
1535 server2 = self.pg0.remote_hosts[1]
1537 locals = [{'addr': server1.ip4,
1541 {'addr': server2.ip4,
1546 self.nat_add_address(self.nat_addr)
1547 self.vapi.nat44_add_del_lb_static_mapping(is_add=1,
1548 external_addr=external_addr,
1549 external_port=external_port,
1550 protocol=IP_PROTOS.tcp,
1552 local_num=len(locals),
1554 flags = self.config_flags.NAT_IS_INSIDE
1555 self.vapi.nat44_interface_add_del_feature(
1556 sw_if_index=self.pg0.sw_if_index,
1557 flags=flags, is_add=1)
1558 self.vapi.nat44_interface_add_del_feature(
1559 sw_if_index=self.pg1.sw_if_index,
1562 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1563 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1564 TCP(sport=1025, dport=external_port))
1565 self.pg1.add_stream(p)
1566 self.pg_enable_capture(self.pg_interfaces)
1568 capture = self.pg0.get_capture(1)
1569 backend = capture[0][IP].dst
1571 sessions = self.vapi.nat44_user_session_dump(backend, 0)
1572 self.assertEqual(len(sessions), 1)
1573 self.assertTrue(sessions[0].flags &
1574 self.config_flags.NAT_IS_EXT_HOST_VALID)
1575 self.vapi.nat44_del_session(
1576 address=sessions[0].inside_ip_address,
1577 port=sessions[0].inside_port,
1578 protocol=sessions[0].protocol,
1579 flags=(self.config_flags.NAT_IS_INSIDE |
1580 self.config_flags.NAT_IS_EXT_HOST_VALID),
1581 ext_host_address=sessions[0].ext_host_address,
1582 ext_host_port=sessions[0].ext_host_port)
1585 for port in range(1030, 1100):
1586 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1587 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1588 TCP(sport=port, dport=external_port))
1590 self.pg1.add_stream(pkts)
1591 self.pg_enable_capture(self.pg_interfaces)
1593 capture = self.pg0.get_capture(len(pkts))
1595 self.assertEqual(p[IP].dst, backend)
1597 def test_multiple_vrf(self):
1598 """ NAT44ED Multiple VRF setup """
1600 external_addr = '1.2.3.4'
1605 self.vapi.nat44_forwarding_enable_disable(enable=1)
1606 self.nat_add_address(self.nat_addr)
1607 flags = self.config_flags.NAT_IS_INSIDE
1608 self.vapi.nat44_interface_add_del_feature(
1609 sw_if_index=self.pg0.sw_if_index,
1611 self.vapi.nat44_interface_add_del_feature(
1612 sw_if_index=self.pg0.sw_if_index,
1613 is_add=1, flags=flags)
1614 self.vapi.nat44_interface_add_del_output_feature(
1615 sw_if_index=self.pg1.sw_if_index,
1617 self.vapi.nat44_interface_add_del_feature(
1618 sw_if_index=self.pg5.sw_if_index,
1620 self.vapi.nat44_interface_add_del_feature(
1621 sw_if_index=self.pg5.sw_if_index,
1622 is_add=1, flags=flags)
1623 self.vapi.nat44_interface_add_del_feature(
1624 sw_if_index=self.pg6.sw_if_index,
1626 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
1627 self.nat_add_static_mapping(self.pg5.remote_ip4, external_addr,
1628 local_port, external_port, vrf_id=1,
1629 proto=IP_PROTOS.tcp, flags=flags)
1630 self.nat_add_static_mapping(
1631 self.pg0.remote_ip4,
1632 external_sw_if_index=self.pg0.sw_if_index,
1633 local_port=local_port,
1635 external_port=external_port,
1636 proto=IP_PROTOS.tcp,
1640 # from client to service (both VRF1)
1641 p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) /
1642 IP(src=self.pg6.remote_ip4, dst=external_addr) /
1643 TCP(sport=12345, dport=external_port))
1644 self.pg6.add_stream(p)
1645 self.pg_enable_capture(self.pg_interfaces)
1647 capture = self.pg5.get_capture(1)
1652 self.assertEqual(ip.dst, self.pg5.remote_ip4)
1653 self.assertEqual(tcp.dport, local_port)
1654 self.assert_packet_checksums_valid(p)
1656 self.logger.error(ppp("Unexpected or invalid packet:", p))
1659 # from service back to client (both VRF1)
1660 p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) /
1661 IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4) /
1662 TCP(sport=local_port, dport=12345))
1663 self.pg5.add_stream(p)
1664 self.pg_enable_capture(self.pg_interfaces)
1666 capture = self.pg6.get_capture(1)
1671 self.assertEqual(ip.src, external_addr)
1672 self.assertEqual(tcp.sport, external_port)
1673 self.assert_packet_checksums_valid(p)
1675 self.logger.error(ppp("Unexpected or invalid packet:", p))
1678 # dynamic NAT from VRF1 to VRF0 (output-feature)
1679 p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) /
1680 IP(src=self.pg5.remote_ip4, dst=self.pg1.remote_ip4) /
1681 TCP(sport=2345, dport=22))
1682 self.pg5.add_stream(p)
1683 self.pg_enable_capture(self.pg_interfaces)
1685 capture = self.pg1.get_capture(1)
1690 self.assertEqual(ip.src, self.nat_addr)
1691 self.assert_packet_checksums_valid(p)
1694 self.logger.error(ppp("Unexpected or invalid packet:", p))
1697 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1698 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1699 TCP(sport=22, dport=port))
1700 self.pg1.add_stream(p)
1701 self.pg_enable_capture(self.pg_interfaces)
1703 capture = self.pg5.get_capture(1)
1708 self.assertEqual(ip.dst, self.pg5.remote_ip4)
1709 self.assertEqual(tcp.dport, 2345)
1710 self.assert_packet_checksums_valid(p)
1712 self.logger.error(ppp("Unexpected or invalid packet:", p))
1715 # from client VRF1 to service VRF0
1716 p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) /
1717 IP(src=self.pg6.remote_ip4, dst=self.pg0.local_ip4) /
1718 TCP(sport=12346, dport=external_port))
1719 self.pg6.add_stream(p)
1720 self.pg_enable_capture(self.pg_interfaces)
1722 capture = self.pg0.get_capture(1)
1727 self.assertEqual(ip.dst, self.pg0.remote_ip4)
1728 self.assertEqual(tcp.dport, local_port)
1729 self.assert_packet_checksums_valid(p)
1731 self.logger.error(ppp("Unexpected or invalid packet:", p))
1734 # from service VRF0 back to client VRF1
1735 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1736 IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) /
1737 TCP(sport=local_port, dport=12346))
1738 self.pg0.add_stream(p)
1739 self.pg_enable_capture(self.pg_interfaces)
1741 capture = self.pg6.get_capture(1)
1746 self.assertEqual(ip.src, self.pg0.local_ip4)
1747 self.assertEqual(tcp.sport, external_port)
1748 self.assert_packet_checksums_valid(p)
1750 self.logger.error(ppp("Unexpected or invalid packet:", p))
1753 # from client VRF0 to service VRF1
1754 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1755 IP(src=self.pg0.remote_ip4, dst=external_addr) /
1756 TCP(sport=12347, dport=external_port))
1757 self.pg0.add_stream(p)
1758 self.pg_enable_capture(self.pg_interfaces)
1760 capture = self.pg5.get_capture(1)
1765 self.assertEqual(ip.dst, self.pg5.remote_ip4)
1766 self.assertEqual(tcp.dport, local_port)
1767 self.assert_packet_checksums_valid(p)
1769 self.logger.error(ppp("Unexpected or invalid packet:", p))
1772 # from service VRF1 back to client VRF0
1773 p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) /
1774 IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4) /
1775 TCP(sport=local_port, dport=12347))
1776 self.pg5.add_stream(p)
1777 self.pg_enable_capture(self.pg_interfaces)
1779 capture = self.pg0.get_capture(1)
1784 self.assertEqual(ip.src, external_addr)
1785 self.assertEqual(tcp.sport, external_port)
1786 self.assert_packet_checksums_valid(p)
1788 self.logger.error(ppp("Unexpected or invalid packet:", p))
1791 # from client to server (both VRF1, no translation)
1792 p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) /
1793 IP(src=self.pg6.remote_ip4, dst=self.pg5.remote_ip4) /
1794 TCP(sport=12348, dport=local_port))
1795 self.pg6.add_stream(p)
1796 self.pg_enable_capture(self.pg_interfaces)
1798 capture = self.pg5.get_capture(1)
1803 self.assertEqual(ip.dst, self.pg5.remote_ip4)
1804 self.assertEqual(tcp.dport, local_port)
1805 self.assert_packet_checksums_valid(p)
1807 self.logger.error(ppp("Unexpected or invalid packet:", p))
1810 # from server back to client (both VRF1, no translation)
1811 p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) /
1812 IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4) /
1813 TCP(sport=local_port, dport=12348))
1814 self.pg5.add_stream(p)
1815 self.pg_enable_capture(self.pg_interfaces)
1817 capture = self.pg6.get_capture(1)
1822 self.assertEqual(ip.src, self.pg5.remote_ip4)
1823 self.assertEqual(tcp.sport, local_port)
1824 self.assert_packet_checksums_valid(p)
1826 self.logger.error(ppp("Unexpected or invalid packet:", p))
1829 # from client VRF1 to server VRF0 (no translation)
1830 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1831 IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) /
1832 TCP(sport=local_port, dport=12349))
1833 self.pg0.add_stream(p)
1834 self.pg_enable_capture(self.pg_interfaces)
1836 capture = self.pg6.get_capture(1)
1841 self.assertEqual(ip.src, self.pg0.remote_ip4)
1842 self.assertEqual(tcp.sport, local_port)
1843 self.assert_packet_checksums_valid(p)
1845 self.logger.error(ppp("Unexpected or invalid packet:", p))
1848 # from server VRF0 back to client VRF1 (no translation)
1849 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1850 IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) /
1851 TCP(sport=local_port, dport=12349))
1852 self.pg0.add_stream(p)
1853 self.pg_enable_capture(self.pg_interfaces)
1855 capture = self.pg6.get_capture(1)
1860 self.assertEqual(ip.src, self.pg0.remote_ip4)
1861 self.assertEqual(tcp.sport, local_port)
1862 self.assert_packet_checksums_valid(p)
1864 self.logger.error(ppp("Unexpected or invalid packet:", p))
1867 # from client VRF0 to server VRF1 (no translation)
1868 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1869 IP(src=self.pg0.remote_ip4, dst=self.pg5.remote_ip4) /
1870 TCP(sport=12344, dport=local_port))
1871 self.pg0.add_stream(p)
1872 self.pg_enable_capture(self.pg_interfaces)
1874 capture = self.pg5.get_capture(1)
1879 self.assertEqual(ip.dst, self.pg5.remote_ip4)
1880 self.assertEqual(tcp.dport, local_port)
1881 self.assert_packet_checksums_valid(p)
1883 self.logger.error(ppp("Unexpected or invalid packet:", p))
1886 # from server VRF1 back to client VRF0 (no translation)
1887 p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) /
1888 IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4) /
1889 TCP(sport=local_port, dport=12344))
1890 self.pg5.add_stream(p)
1891 self.pg_enable_capture(self.pg_interfaces)
1893 capture = self.pg0.get_capture(1)
1898 self.assertEqual(ip.src, self.pg5.remote_ip4)
1899 self.assertEqual(tcp.sport, local_port)
1900 self.assert_packet_checksums_valid(p)
1902 self.logger.error(ppp("Unexpected or invalid packet:", p))
1905 def test_outside_address_distribution(self):
1906 """ Outside address distribution based on source address """
1911 for i in range(1, x):
1913 nat_addresses.append(a)
1915 self.nat_add_inside_interface(self.pg0)
1916 self.nat_add_outside_interface(self.pg1)
1918 self.vapi.nat44_add_del_address_range(
1919 first_ip_address=nat_addresses[0],
1920 last_ip_address=nat_addresses[-1],
1921 vrf_id=0xFFFFFFFF, is_add=1, flags=0)
1923 self.pg0.generate_remote_hosts(x)
1927 info = self.create_packet_info(self.pg0, self.pg1)
1928 payload = self.info_to_payload(info)
1929 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1930 IP(src=self.pg0.remote_hosts[i].ip4,
1931 dst=self.pg1.remote_ip4) /
1932 UDP(sport=7000+i, dport=8000+i) /
1937 self.pg0.add_stream(pkts)
1938 self.pg_enable_capture(self.pg_interfaces)
1940 recvd = self.pg1.get_capture(len(pkts))
1941 for p_recvd in recvd:
1942 payload_info = self.payload_to_info(p_recvd[Raw])
1943 packet_index = payload_info.index
1944 info = self._packet_infos[packet_index]
1945 self.assertTrue(info is not None)
1946 self.assertEqual(packet_index, info.index)
1948 packed = socket.inet_aton(p_sent[IP].src)
1949 numeric = struct.unpack("!L", packed)[0]
1950 numeric = socket.htonl(numeric)
1951 a = nat_addresses[(numeric-1) % len(nat_addresses)]
1954 "Invalid packet (src IP %s translated to %s, but expected %s)"
1955 % (p_sent[IP].src, p_recvd[IP].src, a))
1958 class TestNAT44EDMW(TestNAT44ED):
1959 """ NAT44ED MW Test Case """
1960 vpp_worker_count = 1
1962 def get_stats_counter(self, path, worker=1):
1963 return super(TestNAT44EDMW, self).get_stats_counter(path, worker)
1965 @unittest.skip('MW fix required')
1966 def test_users_dump(self):
1967 """ NAT44ED API test - nat44_user_dump """
1969 @unittest.skip('MW fix required')
1970 def test_frag_out_of_order_do_not_translate(self):
1971 """ NAT44ED don't translate fragments arriving out of order """
1973 @unittest.skip('MW fix required')
1974 def test_forwarding(self):
1975 """ NAT44ED forwarding test """
1977 @unittest.skip('MW fix required')
1978 def test_twice_nat(self):
1979 """ NAT44ED Twice NAT """
1981 @unittest.skip('MW fix required')
1982 def test_twice_nat_lb(self):
1983 """ NAT44ED Twice NAT local service load balancing """
1985 @unittest.skip('MW fix required')
1986 def test_output_feature(self):
1987 """ NAT44ED interface output feature (in2out postrouting) """
1989 @unittest.skip('MW fix required')
1990 def test_static_with_port_out2(self):
1991 """ NAT44ED 1:1 NAPT asymmetrical rule """
1993 @unittest.skip('MW fix required')
1994 def test_output_feature_and_service2(self):
1995 """ NAT44ED interface output feature and service host direct access """
1997 @unittest.skip('MW fix required')
1998 def test_static_lb(self):
1999 """ NAT44ED local service load balancing """
2001 @unittest.skip('MW fix required')
2002 def test_static_lb_2(self):
2003 """ NAT44ED local service load balancing (asymmetrical rule) """
2005 @unittest.skip('MW fix required')
2006 def test_lb_affinity(self):
2007 """ NAT44ED local service load balancing affinity """
2009 @unittest.skip('MW fix required')
2010 def test_multiple_vrf(self):
2011 """ NAT44ED Multiple VRF setup """
2013 @unittest.skip('MW fix required')
2014 def test_self_twice_nat_positive(self):
2015 """ NAT44ED Self Twice NAT (positive test) """
2017 @unittest.skip('MW fix required')
2018 def test_self_twice_nat_lb_positive(self):
2019 """ NAT44ED Self Twice NAT local service load balancing (positive test)
2022 def test_dynamic(self):
2023 """ NAT44ED dynamic translation test """
2025 self.nat_add_address(self.nat_addr)
2026 self.nat_add_inside_interface(self.pg0)
2027 self.nat_add_outside_interface(self.pg1)
2030 tc1 = self.get_stats_counter('/nat44-ed/in2out/slowpath/tcp')
2031 uc1 = self.get_stats_counter('/nat44-ed/in2out/slowpath/udp')
2032 ic1 = self.get_stats_counter('/nat44-ed/in2out/slowpath/icmp')
2033 dc1 = self.get_stats_counter('/nat44-ed/in2out/slowpath/drops')
2035 pkts = self.create_stream_in(self.pg0, self.pg1)
2036 # TODO: specify worker=idx, also stats have to
2037 # know from which worker to take capture
2038 self.pg0.add_stream(pkts)
2039 self.pg_enable_capture(self.pg_interfaces)
2041 capture = self.pg1.get_capture(len(pkts))
2042 self.verify_capture_out(capture, ignore_port=True)
2044 if_idx = self.pg0.sw_if_index
2045 tc2 = self.get_stats_counter('/nat44-ed/in2out/slowpath/tcp')
2046 uc2 = self.get_stats_counter('/nat44-ed/in2out/slowpath/udp')
2047 ic2 = self.get_stats_counter('/nat44-ed/in2out/slowpath/icmp')
2048 dc2 = self.get_stats_counter('/nat44-ed/in2out/slowpath/drops')
2050 self.assertEqual(tc2[if_idx] - tc1[if_idx], 2)
2051 self.assertEqual(uc2[if_idx] - uc1[if_idx], 1)
2052 self.assertEqual(ic2[if_idx] - ic1[if_idx], 1)
2053 self.assertEqual(dc2[if_idx] - dc1[if_idx], 0)
2056 tc1 = self.get_stats_counter('/nat44-ed/out2in/fastpath/tcp')
2057 uc1 = self.get_stats_counter('/nat44-ed/out2in/fastpath/udp')
2058 ic1 = self.get_stats_counter('/nat44-ed/out2in/fastpath/icmp')
2059 dc1 = self.get_stats_counter('/nat44-ed/out2in/fastpath/drops')
2061 pkts = self.create_stream_out(self.pg1)
2062 self.pg1.add_stream(pkts)
2063 self.pg_enable_capture(self.pg_interfaces)
2065 capture = self.pg0.get_capture(len(pkts))
2066 self.verify_capture_in(capture, self.pg0)
2068 if_idx = self.pg1.sw_if_index
2069 tc2 = self.get_stats_counter('/nat44-ed/out2in/fastpath/tcp')
2070 uc2 = self.get_stats_counter('/nat44-ed/out2in/fastpath/udp')
2071 ic2 = self.get_stats_counter('/nat44-ed/out2in/fastpath/icmp')
2072 dc2 = self.get_stats_counter('/nat44-ed/out2in/fastpath/drops')
2074 self.assertEqual(tc2[if_idx] - tc1[if_idx], 2)
2075 self.assertEqual(uc2[if_idx] - uc1[if_idx], 1)
2076 self.assertEqual(ic2[if_idx] - ic1[if_idx], 1)
2077 self.assertEqual(dc2[if_idx] - dc1[if_idx], 0)
2079 sc = self.get_stats_counter('/nat44-ed/total-sessions')
2080 self.assertEqual(sc[0], 3)
2082 def test_frag_in_order(self):
2083 """ NAT44ED translate fragments arriving in order """
2085 self.nat_add_address(self.nat_addr)
2086 self.nat_add_inside_interface(self.pg0)
2087 self.nat_add_outside_interface(self.pg1)
2089 self.frag_in_order(proto=IP_PROTOS.tcp, ignore_port=True)
2090 self.frag_in_order(proto=IP_PROTOS.udp, ignore_port=True)
2091 self.frag_in_order(proto=IP_PROTOS.icmp, ignore_port=True)
2093 def test_frag_in_order_do_not_translate(self):
2094 """ NAT44ED don't translate fragments arriving in order """
2096 self.nat_add_address(self.nat_addr)
2097 self.nat_add_inside_interface(self.pg0)
2098 self.nat_add_outside_interface(self.pg1)
2099 self.vapi.nat44_forwarding_enable_disable(enable=True)
2101 self.frag_in_order(proto=IP_PROTOS.tcp, dont_translate=True)
2103 def test_frag_out_of_order(self):
2104 """ NAT44ED translate fragments arriving out of order """
2106 self.nat_add_address(self.nat_addr)
2107 self.nat_add_inside_interface(self.pg0)
2108 self.nat_add_outside_interface(self.pg1)
2110 self.frag_out_of_order(proto=IP_PROTOS.tcp, ignore_port=True)
2111 self.frag_out_of_order(proto=IP_PROTOS.udp, ignore_port=True)
2112 self.frag_out_of_order(proto=IP_PROTOS.icmp, ignore_port=True)
2114 def test_frag_in_order_in_plus_out(self):
2115 """ NAT44ED in+out interface fragments in order """
2117 in_port = self.random_port()
2118 out_port = self.random_port()
2120 self.nat_add_address(self.nat_addr)
2121 self.nat_add_inside_interface(self.pg0)
2122 self.nat_add_outside_interface(self.pg0)
2123 self.nat_add_inside_interface(self.pg1)
2124 self.nat_add_outside_interface(self.pg1)
2126 # add static mappings for server
2127 self.nat_add_static_mapping(self.server_addr,
2131 proto=IP_PROTOS.tcp)
2132 self.nat_add_static_mapping(self.server_addr,
2136 proto=IP_PROTOS.udp)
2137 self.nat_add_static_mapping(self.server_addr,
2139 proto=IP_PROTOS.icmp)
2141 # run tests for each protocol
2142 self.frag_in_order_in_plus_out(self.server_addr,
2147 self.frag_in_order_in_plus_out(self.server_addr,
2152 self.frag_in_order_in_plus_out(self.server_addr,
2158 def test_frag_out_of_order_in_plus_out(self):
2159 """ NAT44ED in+out interface fragments out of order """
2161 in_port = self.random_port()
2162 out_port = self.random_port()
2164 self.nat_add_address(self.nat_addr)
2165 self.nat_add_inside_interface(self.pg0)
2166 self.nat_add_outside_interface(self.pg0)
2167 self.nat_add_inside_interface(self.pg1)
2168 self.nat_add_outside_interface(self.pg1)
2170 # add static mappings for server
2171 self.nat_add_static_mapping(self.server_addr,
2175 proto=IP_PROTOS.tcp)
2176 self.nat_add_static_mapping(self.server_addr,
2180 proto=IP_PROTOS.udp)
2181 self.nat_add_static_mapping(self.server_addr,
2183 proto=IP_PROTOS.icmp)
2185 # run tests for each protocol
2186 self.frag_out_of_order_in_plus_out(self.server_addr,
2191 self.frag_out_of_order_in_plus_out(self.server_addr,
2196 self.frag_out_of_order_in_plus_out(self.server_addr,
2202 def test_reass_hairpinning(self):
2203 """ NAT44ED fragments hairpinning """
2205 server_addr = self.pg0.remote_hosts[1].ip4
2207 host_in_port = self.random_port()
2208 server_in_port = self.random_port()
2209 server_out_port = self.random_port()
2211 self.nat_add_address(self.nat_addr)
2212 self.nat_add_inside_interface(self.pg0)
2213 self.nat_add_outside_interface(self.pg1)
2215 # add static mapping for server
2216 self.nat_add_static_mapping(server_addr, self.nat_addr,
2217 server_in_port, server_out_port,
2218 proto=IP_PROTOS.tcp)
2219 self.nat_add_static_mapping(server_addr, self.nat_addr,
2220 server_in_port, server_out_port,
2221 proto=IP_PROTOS.udp)
2222 self.nat_add_static_mapping(server_addr, self.nat_addr)
2224 self.reass_hairpinning(server_addr, server_in_port, server_out_port,
2225 host_in_port, proto=IP_PROTOS.tcp,
2227 self.reass_hairpinning(server_addr, server_in_port, server_out_port,
2228 host_in_port, proto=IP_PROTOS.udp,
2230 self.reass_hairpinning(server_addr, server_in_port, server_out_port,
2231 host_in_port, proto=IP_PROTOS.icmp,
2234 def test_session_limit_per_vrf(self):
2235 """ NAT44ED per vrf session limit """
2238 inside_vrf10 = self.pg2
2243 # 2 interfaces pg0, pg1 (vrf10, limit 1 tcp session)
2244 # non existing vrf_id makes process core dump
2245 self.vapi.nat44_set_session_limit(session_limit=limit, vrf_id=10)
2247 self.nat_add_inside_interface(inside)
2248 self.nat_add_inside_interface(inside_vrf10)
2249 self.nat_add_outside_interface(outside)
2252 self.nat_add_interface_address(outside)
2254 # BUG: causing core dump - when bad vrf_id is specified
2255 # self.nat_add_address(outside.local_ip4, vrf_id=20)
2257 stream = self.create_tcp_stream(inside_vrf10, outside, limit * 2)
2258 inside_vrf10.add_stream(stream)
2260 self.pg_enable_capture(self.pg_interfaces)
2263 capture = outside.get_capture(limit)
2265 stream = self.create_tcp_stream(inside, outside, limit * 2)
2266 inside.add_stream(stream)
2268 self.pg_enable_capture(self.pg_interfaces)
2271 capture = outside.get_capture(len(stream))
2273 def test_show_max_translations(self):
2274 """ NAT44ED API test - max translations per thread """
2275 nat_config = self.vapi.nat_show_config_2()
2276 self.assertEqual(self.max_sessions,
2277 nat_config.max_translations_per_thread)
2279 def test_lru_cleanup(self):
2280 """ NAT44ED LRU cleanup algorithm """
2282 self.nat_add_address(self.nat_addr)
2283 self.nat_add_inside_interface(self.pg0)
2284 self.nat_add_outside_interface(self.pg1)
2286 self.vapi.nat_set_timeouts(
2287 udp=1, tcp_established=7440, tcp_transitory=30, icmp=1)
2289 tcp_port_out = self.init_tcp_session(self.pg0, self.pg1, 2000, 80)
2291 for i in range(0, self.max_sessions - 1):
2292 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2293 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
2294 UDP(sport=7000+i, dport=80))
2297 self.pg0.add_stream(pkts)
2298 self.pg_enable_capture(self.pg_interfaces)
2300 self.pg1.get_capture(len(pkts))
2301 self.sleep(1.5, "wait for timeouts")
2304 for i in range(0, self.max_sessions - 1):
2305 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2306 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
2307 ICMP(id=8000+i, type='echo-request'))
2310 self.pg0.add_stream(pkts)
2311 self.pg_enable_capture(self.pg_interfaces)
2313 self.pg1.get_capture(len(pkts))
2315 def test_session_rst_timeout(self):
2316 """ NAT44ED session RST timeouts """
2318 self.nat_add_address(self.nat_addr)
2319 self.nat_add_inside_interface(self.pg0)
2320 self.nat_add_outside_interface(self.pg1)
2322 self.vapi.nat_set_timeouts(udp=300, tcp_established=7440,
2323 tcp_transitory=5, icmp=60)
2325 self.init_tcp_session(self.pg0, self.pg1, self.tcp_port_in,
2326 self.tcp_external_port)
2327 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2328 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2329 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
2331 self.pg0.add_stream(p)
2332 self.pg_enable_capture(self.pg_interfaces)
2334 self.pg1.get_capture(1)
2338 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2339 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2340 TCP(sport=self.tcp_port_in + 1, dport=self.tcp_external_port + 1,
2342 self.pg0.add_stream(p)
2343 self.pg_enable_capture(self.pg_interfaces)
2345 self.pg1.get_capture(1)
2347 def test_dynamic_out_of_ports(self):
2348 """ NAT44ED dynamic translation test: out of ports """
2350 self.nat_add_inside_interface(self.pg0)
2351 self.nat_add_outside_interface(self.pg1)
2353 # in2out and no NAT addresses added
2354 err_old = self.statistics.get_err_counter(
2355 '/err/nat44-ed-in2out-slowpath/out of ports')
2357 pkts = self.create_stream_in(self.pg0, self.pg1)
2358 self.pg0.add_stream(pkts)
2359 self.pg_enable_capture(self.pg_interfaces)
2361 self.pg1.get_capture(0, timeout=1)
2363 err_new = self.statistics.get_err_counter(
2364 '/err/nat44-ed-in2out-slowpath/out of ports')
2366 self.assertEqual(err_new - err_old, len(pkts))
2368 # in2out after NAT addresses added
2369 self.nat_add_address(self.nat_addr)
2371 err_old = self.statistics.get_err_counter(
2372 '/err/nat44-ed-in2out-slowpath/out of ports')
2374 pkts = self.create_stream_in(self.pg0, self.pg1)
2375 self.pg0.add_stream(pkts)
2376 self.pg_enable_capture(self.pg_interfaces)
2378 capture = self.pg1.get_capture(len(pkts))
2379 self.verify_capture_out(capture, ignore_port=True)
2381 err_new = self.statistics.get_err_counter(
2382 '/err/nat44-ed-in2out-slowpath/out of ports')
2384 self.assertEqual(err_new, err_old)
2386 def test_unknown_proto(self):
2387 """ NAT44ED translate packet with unknown protocol """
2389 self.nat_add_address(self.nat_addr)
2390 self.nat_add_inside_interface(self.pg0)
2391 self.nat_add_outside_interface(self.pg1)
2394 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2395 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2396 TCP(sport=self.tcp_port_in, dport=20))
2397 self.pg0.add_stream(p)
2398 self.pg_enable_capture(self.pg_interfaces)
2400 p = self.pg1.get_capture(1)
2402 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2403 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2405 IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) /
2406 TCP(sport=1234, dport=1234))
2407 self.pg0.add_stream(p)
2408 self.pg_enable_capture(self.pg_interfaces)
2410 p = self.pg1.get_capture(1)
2413 self.assertEqual(packet[IP].src, self.nat_addr)
2414 self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
2415 self.assertEqual(packet.haslayer(GRE), 1)
2416 self.assert_packet_checksums_valid(packet)
2418 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2422 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
2423 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
2425 IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) /
2426 TCP(sport=1234, dport=1234))
2427 self.pg1.add_stream(p)
2428 self.pg_enable_capture(self.pg_interfaces)
2430 p = self.pg0.get_capture(1)
2433 self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
2434 self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
2435 self.assertEqual(packet.haslayer(GRE), 1)
2436 self.assert_packet_checksums_valid(packet)
2438 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2441 def test_hairpinning_unknown_proto(self):
2442 """ NAT44ED translate packet with unknown protocol - hairpinning """
2443 host = self.pg0.remote_hosts[0]
2444 server = self.pg0.remote_hosts[1]
2446 server_out_port = 8765
2447 server_nat_ip = "10.0.0.11"
2449 self.nat_add_address(self.nat_addr)
2450 self.nat_add_inside_interface(self.pg0)
2451 self.nat_add_outside_interface(self.pg1)
2453 # add static mapping for server
2454 self.nat_add_static_mapping(server.ip4, server_nat_ip)
2457 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
2458 IP(src=host.ip4, dst=server_nat_ip) /
2459 TCP(sport=host_in_port, dport=server_out_port))
2460 self.pg0.add_stream(p)
2461 self.pg_enable_capture(self.pg_interfaces)
2463 self.pg0.get_capture(1)
2465 p = (Ether(dst=self.pg0.local_mac, src=host.mac) /
2466 IP(src=host.ip4, dst=server_nat_ip) /
2468 IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) /
2469 TCP(sport=1234, dport=1234))
2470 self.pg0.add_stream(p)
2471 self.pg_enable_capture(self.pg_interfaces)
2473 p = self.pg0.get_capture(1)
2476 self.assertEqual(packet[IP].src, self.nat_addr)
2477 self.assertEqual(packet[IP].dst, server.ip4)
2478 self.assertEqual(packet.haslayer(GRE), 1)
2479 self.assert_packet_checksums_valid(packet)
2481 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2485 p = (Ether(dst=self.pg0.local_mac, src=server.mac) /
2486 IP(src=server.ip4, dst=self.nat_addr) /
2488 IP(src=self.pg2.remote_ip4, dst=self.pg2.remote_ip4) /
2489 TCP(sport=1234, dport=1234))
2490 self.pg0.add_stream(p)
2491 self.pg_enable_capture(self.pg_interfaces)
2493 p = self.pg0.get_capture(1)
2496 self.assertEqual(packet[IP].src, server_nat_ip)
2497 self.assertEqual(packet[IP].dst, host.ip4)
2498 self.assertEqual(packet.haslayer(GRE), 1)
2499 self.assert_packet_checksums_valid(packet)
2501 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2504 def test_output_feature_and_service(self):
2505 """ NAT44ED interface output feature and services """
2506 external_addr = '1.2.3.4'
2510 self.vapi.nat44_forwarding_enable_disable(enable=1)
2511 self.nat_add_address(self.nat_addr)
2512 flags = self.config_flags.NAT_IS_ADDR_ONLY
2513 self.vapi.nat44_add_del_identity_mapping(
2514 ip_address=self.pg1.remote_ip4, sw_if_index=0xFFFFFFFF,
2515 flags=flags, is_add=1)
2516 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
2517 self.nat_add_static_mapping(self.pg0.remote_ip4, external_addr,
2518 local_port, external_port,
2519 proto=IP_PROTOS.tcp, flags=flags)
2521 self.nat_add_inside_interface(self.pg0)
2522 self.nat_add_outside_interface(self.pg0)
2523 self.vapi.nat44_interface_add_del_output_feature(
2524 sw_if_index=self.pg1.sw_if_index, is_add=1)
2526 # from client to service
2527 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
2528 IP(src=self.pg1.remote_ip4, dst=external_addr) /
2529 TCP(sport=12345, dport=external_port))
2530 self.pg1.add_stream(p)
2531 self.pg_enable_capture(self.pg_interfaces)
2533 capture = self.pg0.get_capture(1)
2538 self.assertEqual(ip.dst, self.pg0.remote_ip4)
2539 self.assertEqual(tcp.dport, local_port)
2540 self.assert_packet_checksums_valid(p)
2542 self.logger.error(ppp("Unexpected or invalid packet:", p))
2545 # from service back to client
2546 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2547 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2548 TCP(sport=local_port, dport=12345))
2549 self.pg0.add_stream(p)
2550 self.pg_enable_capture(self.pg_interfaces)
2552 capture = self.pg1.get_capture(1)
2557 self.assertEqual(ip.src, external_addr)
2558 self.assertEqual(tcp.sport, external_port)
2559 self.assert_packet_checksums_valid(p)
2561 self.logger.error(ppp("Unexpected or invalid packet:", p))
2564 # from local network host to external network
2565 pkts = self.create_stream_in(self.pg0, self.pg1)
2566 self.pg0.add_stream(pkts)
2567 self.pg_enable_capture(self.pg_interfaces)
2569 capture = self.pg1.get_capture(len(pkts))
2570 self.verify_capture_out(capture, ignore_port=True)
2571 pkts = self.create_stream_in(self.pg0, self.pg1)
2572 self.pg0.add_stream(pkts)
2573 self.pg_enable_capture(self.pg_interfaces)
2575 capture = self.pg1.get_capture(len(pkts))
2576 self.verify_capture_out(capture, ignore_port=True)
2578 # from external network back to local network host
2579 pkts = self.create_stream_out(self.pg1)
2580 self.pg1.add_stream(pkts)
2581 self.pg_enable_capture(self.pg_interfaces)
2583 capture = self.pg0.get_capture(len(pkts))
2584 self.verify_capture_in(capture, self.pg0)
2586 def test_output_feature_and_service3(self):
2587 """ NAT44ED interface output feature and DST NAT """
2588 external_addr = '1.2.3.4'
2592 self.vapi.nat44_forwarding_enable_disable(enable=1)
2593 self.nat_add_address(self.nat_addr)
2594 flags = self.config_flags.NAT_IS_OUT2IN_ONLY
2595 self.nat_add_static_mapping(self.pg1.remote_ip4, external_addr,
2596 local_port, external_port,
2597 proto=IP_PROTOS.tcp, flags=flags)
2599 self.nat_add_inside_interface(self.pg0)
2600 self.nat_add_outside_interface(self.pg0)
2601 self.vapi.nat44_interface_add_del_output_feature(
2602 sw_if_index=self.pg1.sw_if_index, is_add=1)
2604 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2605 IP(src=self.pg0.remote_ip4, dst=external_addr) /
2606 TCP(sport=12345, dport=external_port))
2607 self.pg0.add_stream(p)
2608 self.pg_enable_capture(self.pg_interfaces)
2610 capture = self.pg1.get_capture(1)
2615 self.assertEqual(ip.src, self.pg0.remote_ip4)
2616 self.assertEqual(tcp.sport, 12345)
2617 self.assertEqual(ip.dst, self.pg1.remote_ip4)
2618 self.assertEqual(tcp.dport, local_port)
2619 self.assert_packet_checksums_valid(p)
2621 self.logger.error(ppp("Unexpected or invalid packet:", p))
2624 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
2625 IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) /
2626 TCP(sport=local_port, dport=12345))
2627 self.pg1.add_stream(p)
2628 self.pg_enable_capture(self.pg_interfaces)
2630 capture = self.pg0.get_capture(1)
2635 self.assertEqual(ip.src, external_addr)
2636 self.assertEqual(tcp.sport, external_port)
2637 self.assertEqual(ip.dst, self.pg0.remote_ip4)
2638 self.assertEqual(tcp.dport, 12345)
2639 self.assert_packet_checksums_valid(p)
2641 self.logger.error(ppp("Unexpected or invalid packet:", p))
2644 def test_self_twice_nat_lb_negative(self):
2645 """ NAT44ED Self Twice NAT local service load balancing (negative test)
2647 self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True,
2650 def test_self_twice_nat_negative(self):
2651 """ NAT44ED Self Twice NAT (negative test) """
2652 self.twice_nat_common(self_twice_nat=True)
2654 def test_static_lb_multi_clients(self):
2655 """ NAT44ED local service load balancing - multiple clients"""
2657 external_addr = self.nat_addr
2660 server1 = self.pg0.remote_hosts[0]
2661 server2 = self.pg0.remote_hosts[1]
2662 server3 = self.pg0.remote_hosts[2]
2664 locals = [{'addr': server1.ip4,
2668 {'addr': server2.ip4,
2673 flags = self.config_flags.NAT_IS_INSIDE
2674 self.vapi.nat44_interface_add_del_feature(
2675 sw_if_index=self.pg0.sw_if_index,
2676 flags=flags, is_add=1)
2677 self.vapi.nat44_interface_add_del_feature(
2678 sw_if_index=self.pg1.sw_if_index,
2681 self.nat_add_address(self.nat_addr)
2682 self.vapi.nat44_add_del_lb_static_mapping(is_add=1,
2683 external_addr=external_addr,
2684 external_port=external_port,
2685 protocol=IP_PROTOS.tcp,
2686 local_num=len(locals),
2691 clients = ip4_range(self.pg1.remote_ip4, 10, 50)
2693 for client in clients:
2694 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
2695 IP(src=client, dst=self.nat_addr) /
2696 TCP(sport=12345, dport=external_port))
2698 self.pg1.add_stream(pkts)
2699 self.pg_enable_capture(self.pg_interfaces)
2701 capture = self.pg0.get_capture(len(pkts))
2703 if p[IP].dst == server1.ip4:
2707 self.assertGreater(server1_n, server2_n)
2710 'addr': server3.ip4,
2717 self.vapi.nat44_lb_static_mapping_add_del_local(
2719 external_addr=external_addr,
2720 external_port=external_port,
2722 protocol=IP_PROTOS.tcp)
2726 clients = ip4_range(self.pg1.remote_ip4, 60, 110)
2728 for client in clients:
2729 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
2730 IP(src=client, dst=self.nat_addr) /
2731 TCP(sport=12346, dport=external_port))
2733 self.assertGreater(len(pkts), 0)
2734 self.pg1.add_stream(pkts)
2735 self.pg_enable_capture(self.pg_interfaces)
2737 capture = self.pg0.get_capture(len(pkts))
2739 if p[IP].dst == server1.ip4:
2741 elif p[IP].dst == server2.ip4:
2745 self.assertGreater(server1_n, 0)
2746 self.assertGreater(server2_n, 0)
2747 self.assertGreater(server3_n, 0)
2750 'addr': server2.ip4,
2756 # remove one back-end
2757 self.vapi.nat44_lb_static_mapping_add_del_local(
2759 external_addr=external_addr,
2760 external_port=external_port,
2762 protocol=IP_PROTOS.tcp)
2766 self.pg1.add_stream(pkts)
2767 self.pg_enable_capture(self.pg_interfaces)
2769 capture = self.pg0.get_capture(len(pkts))
2771 if p[IP].dst == server1.ip4:
2773 elif p[IP].dst == server2.ip4:
2777 self.assertGreater(server1_n, 0)
2778 self.assertEqual(server2_n, 0)
2779 self.assertGreater(server3_n, 0)
2781 def test_syslog_sess(self):
2782 """ NAT44ED Test syslog session creation and deletion """
2783 self.vapi.syslog_set_filter(
2784 self.syslog_severity.SYSLOG_API_SEVERITY_INFO)
2785 self.vapi.syslog_set_sender(self.pg3.local_ip4, self.pg3.remote_ip4)
2787 self.nat_add_address(self.nat_addr)
2788 self.nat_add_inside_interface(self.pg0)
2789 self.nat_add_outside_interface(self.pg1)
2791 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2792 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2793 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port))
2794 self.pg0.add_stream(p)
2795 self.pg_enable_capture(self.pg_interfaces)
2797 capture = self.pg1.get_capture(1)
2798 self.tcp_port_out = capture[0][TCP].sport
2799 capture = self.pg3.get_capture(1)
2800 self.verify_syslog_sess(capture[0][Raw].load)
2802 self.pg_enable_capture(self.pg_interfaces)
2804 self.nat_add_address(self.nat_addr, is_add=0)
2805 capture = self.pg3.get_capture(1)
2806 self.verify_syslog_sess(capture[0][Raw].load, False)
2808 def test_twice_nat_interface_addr(self):
2809 """ NAT44ED Acquire twice NAT addresses from interface """
2810 flags = self.config_flags.NAT_IS_TWICE_NAT
2811 self.vapi.nat44_add_del_interface_addr(
2812 sw_if_index=self.pg11.sw_if_index,
2813 flags=flags, is_add=1)
2815 # no address in NAT pool
2816 adresses = self.vapi.nat44_address_dump()
2817 self.assertEqual(0, len(adresses))
2819 # configure interface address and check NAT address pool
2820 self.pg11.config_ip4()
2821 adresses = self.vapi.nat44_address_dump()
2822 self.assertEqual(1, len(adresses))
2823 self.assertEqual(str(adresses[0].ip_address),
2824 self.pg11.local_ip4)
2825 self.assertEqual(adresses[0].flags, flags)
2827 # remove interface address and check NAT address pool
2828 self.pg11.unconfig_ip4()
2829 adresses = self.vapi.nat44_address_dump()
2830 self.assertEqual(0, len(adresses))
2832 def test_output_feature_stateful_acl(self):
2833 """ NAT44ED output feature works with stateful ACL """
2835 self.nat_add_address(self.nat_addr)
2836 self.vapi.nat44_interface_add_del_output_feature(
2837 sw_if_index=self.pg0.sw_if_index,
2838 flags=self.config_flags.NAT_IS_INSIDE, is_add=1)
2839 self.vapi.nat44_interface_add_del_output_feature(
2840 sw_if_index=self.pg1.sw_if_index,
2841 flags=self.config_flags.NAT_IS_OUTSIDE, is_add=1)
2843 # First ensure that the NAT is working sans ACL
2845 # send packets out2in, no sessions yet so packets should drop
2846 pkts_out2in = self.create_stream_out(self.pg1)
2847 self.send_and_assert_no_replies(self.pg1, pkts_out2in)
2849 # send packets into inside intf, ensure received via outside intf
2850 pkts_in2out = self.create_stream_in(self.pg0, self.pg1)
2851 capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1,
2853 self.verify_capture_out(capture, ignore_port=True)
2855 # send out2in again, with sessions created it should work now
2856 pkts_out2in = self.create_stream_out(self.pg1)
2857 capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0,
2859 self.verify_capture_in(capture, self.pg0)
2861 # Create an ACL blocking everything
2862 out2in_deny_rule = AclRule(is_permit=0)
2863 out2in_acl = VppAcl(self, rules=[out2in_deny_rule])
2864 out2in_acl.add_vpp_config()
2866 # create an ACL to permit/reflect everything
2867 in2out_reflect_rule = AclRule(is_permit=2)
2868 in2out_acl = VppAcl(self, rules=[in2out_reflect_rule])
2869 in2out_acl.add_vpp_config()
2871 # apply as input acl on interface and confirm it blocks everything
2872 acl_if = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
2873 n_input=1, acls=[out2in_acl])
2874 acl_if.add_vpp_config()
2875 self.send_and_assert_no_replies(self.pg1, pkts_out2in)
2878 acl_if.acls = [out2in_acl, in2out_acl]
2879 acl_if.add_vpp_config()
2880 # send in2out to generate ACL state (NAT state was created earlier)
2881 capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1,
2883 self.verify_capture_out(capture, ignore_port=True)
2885 # send out2in again. ACL state exists so it should work now.
2886 # TCP packets with the syn flag set also need the ack flag
2887 for p in pkts_out2in:
2888 if p.haslayer(TCP) and p[TCP].flags & 0x02:
2889 p[TCP].flags |= 0x10
2890 capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0,
2892 self.verify_capture_in(capture, self.pg0)
2893 self.logger.info(self.vapi.cli("show trace"))
2895 def test_tcp_close(self):
2896 """ NAT44ED Close TCP session from inside network - output feature """
2897 old_timeouts = self.vapi.nat_get_timeouts()
2899 self.vapi.nat_set_timeouts(
2900 udp=old_timeouts.udp,
2901 tcp_established=old_timeouts.tcp_established,
2902 icmp=old_timeouts.icmp,
2903 tcp_transitory=new_transitory)
2905 self.vapi.nat44_forwarding_enable_disable(enable=1)
2906 self.nat_add_address(self.pg1.local_ip4)
2907 twice_nat_addr = '10.0.1.3'
2908 service_ip = '192.168.16.150'
2909 self.nat_add_address(twice_nat_addr, twice_nat=1)
2911 flags = self.config_flags.NAT_IS_INSIDE
2912 self.vapi.nat44_interface_add_del_feature(
2913 sw_if_index=self.pg0.sw_if_index,
2915 self.vapi.nat44_interface_add_del_feature(
2916 sw_if_index=self.pg0.sw_if_index,
2917 flags=flags, is_add=1)
2918 self.vapi.nat44_interface_add_del_output_feature(
2920 sw_if_index=self.pg1.sw_if_index)
2922 flags = (self.config_flags.NAT_IS_OUT2IN_ONLY |
2923 self.config_flags.NAT_IS_TWICE_NAT)
2924 self.nat_add_static_mapping(self.pg0.remote_ip4,
2926 proto=IP_PROTOS.tcp,
2928 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
2929 start_sessnum = len(sessions)
2931 # SYN packet out->in
2932 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
2933 IP(src=self.pg1.remote_ip4, dst=service_ip) /
2934 TCP(sport=33898, dport=80, flags="S"))
2935 self.pg1.add_stream(p)
2936 self.pg_enable_capture(self.pg_interfaces)
2938 capture = self.pg0.get_capture(1)
2940 tcp_port = p[TCP].sport
2942 # SYN + ACK packet in->out
2943 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2944 IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) /
2945 TCP(sport=80, dport=tcp_port, flags="SA"))
2946 self.pg0.add_stream(p)
2947 self.pg_enable_capture(self.pg_interfaces)
2949 self.pg1.get_capture(1)
2951 # ACK packet out->in
2952 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
2953 IP(src=self.pg1.remote_ip4, dst=service_ip) /
2954 TCP(sport=33898, dport=80, flags="A"))
2955 self.pg1.add_stream(p)
2956 self.pg_enable_capture(self.pg_interfaces)
2958 self.pg0.get_capture(1)
2960 # FIN packet in -> out
2961 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2962 IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) /
2963 TCP(sport=80, dport=tcp_port, flags="FA", seq=100, ack=300))
2964 self.pg0.add_stream(p)
2965 self.pg_enable_capture(self.pg_interfaces)
2967 self.pg1.get_capture(1)
2969 # FIN+ACK packet out -> in
2970 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
2971 IP(src=self.pg1.remote_ip4, dst=service_ip) /
2972 TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101))
2973 self.pg1.add_stream(p)
2974 self.pg_enable_capture(self.pg_interfaces)
2976 self.pg0.get_capture(1)
2978 # ACK packet in -> out
2979 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2980 IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) /
2981 TCP(sport=80, dport=tcp_port, flags="A", seq=101, ack=301))
2982 self.pg0.add_stream(p)
2983 self.pg_enable_capture(self.pg_interfaces)
2985 self.pg1.get_capture(1)
2987 # session now in transitory timeout
2988 # try SYN packet out->in - should be dropped
2989 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
2990 IP(src=self.pg1.remote_ip4, dst=service_ip) /
2991 TCP(sport=33898, dport=80, flags="S"))
2992 self.pg1.add_stream(p)
2993 self.pg_enable_capture(self.pg_interfaces)
2996 self.sleep(new_transitory, "wait for transitory timeout")
2997 self.pg0.assert_nothing_captured(0)
2999 # session should still exist
3000 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3001 self.assertEqual(len(sessions) - start_sessnum, 1)
3003 # send FIN+ACK packet out -> in - will cause session to be wiped
3004 # but won't create a new session
3005 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3006 IP(src=self.pg1.remote_ip4, dst=service_ip) /
3007 TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101))
3008 self.pg1.add_stream(p)
3009 self.pg_enable_capture(self.pg_interfaces)
3012 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3013 self.assertEqual(len(sessions) - start_sessnum, 0)
3014 self.pg0.assert_nothing_captured(0)
3016 def test_tcp_session_close_in(self):
3017 """ NAT44ED Close TCP session from inside network """
3019 in_port = self.tcp_port_in
3021 ext_port = self.tcp_external_port
3023 self.nat_add_address(self.nat_addr)
3024 self.nat_add_inside_interface(self.pg0)
3025 self.nat_add_outside_interface(self.pg1)
3026 self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
3027 in_port, out_port, proto=IP_PROTOS.tcp,
3028 flags=self.config_flags.NAT_IS_TWICE_NAT)
3030 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3031 session_n = len(sessions)
3033 self.vapi.nat_set_timeouts(udp=300, tcp_established=7440,
3034 tcp_transitory=2, icmp=5)
3036 self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port)
3038 # FIN packet in -> out
3039 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3040 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3041 TCP(sport=in_port, dport=ext_port,
3042 flags="FA", seq=100, ack=300))
3043 self.pg0.add_stream(p)
3044 self.pg_enable_capture(self.pg_interfaces)
3046 self.pg1.get_capture(1)
3050 # ACK packet out -> in
3051 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3052 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3053 TCP(sport=ext_port, dport=out_port,
3054 flags="A", seq=300, ack=101))
3057 # FIN packet out -> in
3058 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3059 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3060 TCP(sport=ext_port, dport=out_port,
3061 flags="FA", seq=300, ack=101))
3064 self.pg1.add_stream(pkts)
3065 self.pg_enable_capture(self.pg_interfaces)
3067 self.pg0.get_capture(2)
3069 # ACK packet in -> out
3070 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3071 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3072 TCP(sport=in_port, dport=ext_port,
3073 flags="A", seq=101, ack=301))
3074 self.pg0.add_stream(p)
3075 self.pg_enable_capture(self.pg_interfaces)
3077 self.pg1.get_capture(1)
3079 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3080 self.assertEqual(len(sessions) - session_n, 1)
3082 out2in_drops = self.get_err_counter(
3083 '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
3084 in2out_drops = self.get_err_counter(
3085 '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
3087 # extra FIN packet out -> in - this should be dropped
3088 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3089 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3090 TCP(sport=ext_port, dport=out_port,
3091 flags="FA", seq=300, ack=101))
3093 self.pg1.add_stream(p)
3094 self.pg_enable_capture(self.pg_interfaces)
3096 self.pg0.assert_nothing_captured()
3098 # extra ACK packet in -> out - this should be dropped
3099 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3100 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3101 TCP(sport=in_port, dport=ext_port,
3102 flags="A", seq=101, ack=301))
3103 self.pg0.add_stream(p)
3104 self.pg_enable_capture(self.pg_interfaces)
3106 self.pg1.assert_nothing_captured()
3108 stats = self.get_err_counter(
3109 '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
3110 self.assertEqual(stats - out2in_drops, 1)
3111 stats = self.get_err_counter(
3112 '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
3113 self.assertEqual(stats - in2out_drops, 1)
3116 # extra ACK packet in -> out - this will cause session to be wiped
3117 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3118 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3119 TCP(sport=in_port, dport=ext_port,
3120 flags="A", seq=101, ack=301))
3121 self.pg0.add_stream(p)
3122 self.pg_enable_capture(self.pg_interfaces)
3124 self.pg1.assert_nothing_captured()
3125 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3126 self.assertEqual(len(sessions) - session_n, 0)
3128 def test_tcp_session_close_out(self):
3129 """ NAT44ED Close TCP session from outside network """
3131 in_port = self.tcp_port_in
3133 ext_port = self.tcp_external_port
3135 self.nat_add_address(self.nat_addr)
3136 self.nat_add_inside_interface(self.pg0)
3137 self.nat_add_outside_interface(self.pg1)
3138 self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
3139 in_port, out_port, proto=IP_PROTOS.tcp,
3140 flags=self.config_flags.NAT_IS_TWICE_NAT)
3142 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3143 session_n = len(sessions)
3145 self.vapi.nat_set_timeouts(udp=300, tcp_established=7440,
3146 tcp_transitory=2, icmp=5)
3148 _ = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port)
3150 # FIN packet out -> in
3151 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3152 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3153 TCP(sport=ext_port, dport=out_port,
3154 flags="FA", seq=100, ack=300))
3155 self.pg1.add_stream(p)
3156 self.pg_enable_capture(self.pg_interfaces)
3158 self.pg0.get_capture(1)
3160 # FIN+ACK packet in -> out
3161 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3162 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3163 TCP(sport=in_port, dport=ext_port,
3164 flags="FA", seq=300, ack=101))
3166 self.pg0.add_stream(p)
3167 self.pg_enable_capture(self.pg_interfaces)
3169 self.pg1.get_capture(1)
3171 # ACK packet out -> in
3172 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3173 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3174 TCP(sport=ext_port, dport=out_port,
3175 flags="A", seq=101, ack=301))
3176 self.pg1.add_stream(p)
3177 self.pg_enable_capture(self.pg_interfaces)
3179 self.pg0.get_capture(1)
3181 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3182 self.assertEqual(len(sessions) - session_n, 1)
3184 out2in_drops = self.get_err_counter(
3185 '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
3186 in2out_drops = self.get_err_counter(
3187 '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
3189 # extra FIN packet out -> in - this should be dropped
3190 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3191 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3192 TCP(sport=ext_port, dport=out_port,
3193 flags="FA", seq=300, ack=101))
3195 self.pg1.add_stream(p)
3196 self.pg_enable_capture(self.pg_interfaces)
3198 self.pg0.assert_nothing_captured()
3200 # extra ACK packet in -> out - this should be dropped
3201 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3202 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3203 TCP(sport=in_port, dport=ext_port,
3204 flags="A", seq=101, ack=301))
3205 self.pg0.add_stream(p)
3206 self.pg_enable_capture(self.pg_interfaces)
3208 self.pg1.assert_nothing_captured()
3210 stats = self.get_err_counter(
3211 '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
3212 self.assertEqual(stats - out2in_drops, 1)
3213 stats = self.get_err_counter(
3214 '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
3215 self.assertEqual(stats - in2out_drops, 1)
3218 # extra ACK packet in -> out - this will cause session to be wiped
3219 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3220 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3221 TCP(sport=in_port, dport=ext_port,
3222 flags="A", seq=101, ack=301))
3223 self.pg0.add_stream(p)
3224 self.pg_enable_capture(self.pg_interfaces)
3226 self.pg1.assert_nothing_captured()
3227 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3228 self.assertEqual(len(sessions) - session_n, 0)
3230 def test_tcp_session_close_simultaneous(self):
3231 """ NAT44ED Close TCP session from inside network """
3233 in_port = self.tcp_port_in
3236 self.nat_add_address(self.nat_addr)
3237 self.nat_add_inside_interface(self.pg0)
3238 self.nat_add_outside_interface(self.pg1)
3239 self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
3240 in_port, ext_port, proto=IP_PROTOS.tcp,
3241 flags=self.config_flags.NAT_IS_TWICE_NAT)
3243 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3244 session_n = len(sessions)
3246 self.vapi.nat_set_timeouts(udp=300, tcp_established=7440,
3247 tcp_transitory=2, icmp=5)
3249 out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port)
3251 # FIN packet in -> out
3252 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3253 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3254 TCP(sport=in_port, dport=ext_port,
3255 flags="FA", seq=100, ack=300))
3256 self.pg0.add_stream(p)
3257 self.pg_enable_capture(self.pg_interfaces)
3259 self.pg1.get_capture(1)
3261 # FIN packet out -> in
3262 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3263 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3264 TCP(sport=ext_port, dport=out_port,
3265 flags="FA", seq=300, ack=100))
3266 self.pg1.add_stream(p)
3267 self.pg_enable_capture(self.pg_interfaces)
3269 self.pg0.get_capture(1)
3271 # ACK packet in -> out
3272 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3273 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3274 TCP(sport=in_port, dport=ext_port,
3275 flags="A", seq=101, ack=301))
3276 self.pg0.add_stream(p)
3277 self.pg_enable_capture(self.pg_interfaces)
3279 self.pg1.get_capture(1)
3281 # ACK packet out -> in
3282 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3283 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3284 TCP(sport=ext_port, dport=out_port,
3285 flags="A", seq=301, ack=101))
3286 self.pg1.add_stream(p)
3287 self.pg_enable_capture(self.pg_interfaces)
3289 self.pg0.get_capture(1)
3291 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3292 self.assertEqual(len(sessions) - session_n, 1)
3294 out2in_drops = self.get_err_counter(
3295 '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
3296 in2out_drops = self.get_err_counter(
3297 '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
3299 # extra FIN packet out -> in - this should be dropped
3300 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3301 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3302 TCP(sport=ext_port, dport=out_port,
3303 flags="FA", seq=300, ack=101))
3305 self.pg1.add_stream(p)
3306 self.pg_enable_capture(self.pg_interfaces)
3308 self.pg0.assert_nothing_captured()
3310 # extra ACK packet in -> out - this should be dropped
3311 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3312 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3313 TCP(sport=in_port, dport=ext_port,
3314 flags="A", seq=101, ack=301))
3315 self.pg0.add_stream(p)
3316 self.pg_enable_capture(self.pg_interfaces)
3318 self.pg1.assert_nothing_captured()
3320 stats = self.get_err_counter(
3321 '/err/nat44-ed-out2in/drops due to TCP in transitory timeout')
3322 self.assertEqual(stats - out2in_drops, 1)
3323 stats = self.get_err_counter(
3324 '/err/nat44-ed-in2out/drops due to TCP in transitory timeout')
3325 self.assertEqual(stats - in2out_drops, 1)
3328 # extra ACK packet in -> out - this will cause session to be wiped
3329 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3330 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3331 TCP(sport=in_port, dport=ext_port,
3332 flags="A", seq=101, ack=301))
3333 self.pg0.add_stream(p)
3334 self.pg_enable_capture(self.pg_interfaces)
3336 self.pg1.assert_nothing_captured()
3337 sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
3338 self.assertEqual(len(sessions) - session_n, 0)
3340 def test_dynamic_output_feature_vrf(self):
3341 """ NAT44ED dynamic translation test: output-feature, VRF"""
3343 # other then default (0)
3346 self.nat_add_address(self.nat_addr)
3347 flags = self.config_flags.NAT_IS_INSIDE
3348 self.vapi.nat44_interface_add_del_output_feature(
3349 sw_if_index=self.pg7.sw_if_index,
3350 flags=flags, is_add=1)
3351 self.vapi.nat44_interface_add_del_output_feature(
3352 sw_if_index=self.pg8.sw_if_index,
3356 self.configure_ip4_interface(self.pg7, table_id=new_vrf_id)
3357 self.configure_ip4_interface(self.pg8, table_id=new_vrf_id)
3360 tcpn = self.get_stats_counter(
3361 '/nat44-ed/in2out/slowpath/tcp')
3362 udpn = self.get_stats_counter(
3363 '/nat44-ed/in2out/slowpath/udp')
3364 icmpn = self.get_stats_counter(
3365 '/nat44-ed/in2out/slowpath/icmp')
3366 drops = self.get_stats_counter(
3367 '/nat44-ed/in2out/slowpath/drops')
3369 pkts = self.create_stream_in(self.pg7, self.pg8)
3370 self.pg7.add_stream(pkts)
3371 self.pg_enable_capture(self.pg_interfaces)
3373 capture = self.pg8.get_capture(len(pkts))
3374 self.verify_capture_out(capture, ignore_port=True)
3376 if_idx = self.pg7.sw_if_index
3377 cnt = self.get_stats_counter(
3378 '/nat44-ed/in2out/slowpath/tcp')
3379 self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
3380 cnt = self.get_stats_counter(
3381 '/nat44-ed/in2out/slowpath/udp')
3382 self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
3383 cnt = self.get_stats_counter(
3384 '/nat44-ed/in2out/slowpath/icmp')
3385 self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
3386 cnt = self.get_stats_counter(
3387 '/nat44-ed/in2out/slowpath/drops')
3388 self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
3391 tcpn = self.get_stats_counter(
3392 '/nat44-ed/out2in/fastpath/tcp')
3393 udpn = self.get_stats_counter(
3394 '/nat44-ed/out2in/fastpath/udp')
3395 icmpn = self.get_stats_counter(
3396 '/nat44-ed/out2in/fastpath/icmp')
3397 drops = self.get_stats_counter(
3398 '/nat44-ed/out2in/fastpath/drops')
3400 pkts = self.create_stream_out(self.pg8)
3401 self.pg8.add_stream(pkts)
3402 self.pg_enable_capture(self.pg_interfaces)
3404 capture = self.pg7.get_capture(len(pkts))
3405 self.verify_capture_in(capture, self.pg7)
3407 if_idx = self.pg8.sw_if_index
3408 cnt = self.get_stats_counter(
3409 '/nat44-ed/out2in/fastpath/tcp')
3410 self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
3411 cnt = self.get_stats_counter(
3412 '/nat44-ed/out2in/fastpath/udp')
3413 self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
3414 cnt = self.get_stats_counter(
3415 '/nat44-ed/out2in/fastpath/icmp')
3416 self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
3417 cnt = self.get_stats_counter(
3418 '/nat44-ed/out2in/fastpath/drops')
3419 self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
3421 sessions = self.get_stats_counter('/nat44-ed/total-sessions')
3422 self.assertEqual(sessions[0], 3)
3425 self.configure_ip4_interface(self.pg7, table_id=0)
3426 self.configure_ip4_interface(self.pg8, table_id=0)
3428 self.vapi.ip_table_add_del(is_add=0,
3429 table={'table_id': new_vrf_id})
3431 def test_next_src_nat(self):
3432 """ NAT44ED On way back forward packet to nat44-in2out node. """
3434 twice_nat_addr = '10.0.1.3'
3437 post_twice_nat_port = 0
3439 self.vapi.nat44_forwarding_enable_disable(enable=1)
3440 self.nat_add_address(twice_nat_addr, twice_nat=1)
3441 flags = (self.config_flags.NAT_IS_OUT2IN_ONLY |
3442 self.config_flags.NAT_IS_SELF_TWICE_NAT)
3443 self.nat_add_static_mapping(self.pg6.remote_ip4, self.pg1.remote_ip4,
3444 local_port, external_port,
3445 proto=IP_PROTOS.tcp, vrf_id=1,
3447 self.vapi.nat44_interface_add_del_feature(
3448 sw_if_index=self.pg6.sw_if_index,
3451 p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) /
3452 IP(src=self.pg6.remote_ip4, dst=self.pg1.remote_ip4) /
3453 TCP(sport=12345, dport=external_port))
3454 self.pg6.add_stream(p)
3455 self.pg_enable_capture(self.pg_interfaces)
3457 capture = self.pg6.get_capture(1)
3462 self.assertEqual(ip.src, twice_nat_addr)
3463 self.assertNotEqual(tcp.sport, 12345)
3464 post_twice_nat_port = tcp.sport
3465 self.assertEqual(ip.dst, self.pg6.remote_ip4)
3466 self.assertEqual(tcp.dport, local_port)
3467 self.assert_packet_checksums_valid(p)
3469 self.logger.error(ppp("Unexpected or invalid packet:", p))
3472 p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) /
3473 IP(src=self.pg6.remote_ip4, dst=twice_nat_addr) /
3474 TCP(sport=local_port, dport=post_twice_nat_port))
3475 self.pg6.add_stream(p)
3476 self.pg_enable_capture(self.pg_interfaces)
3478 capture = self.pg6.get_capture(1)
3483 self.assertEqual(ip.src, self.pg1.remote_ip4)
3484 self.assertEqual(tcp.sport, external_port)
3485 self.assertEqual(ip.dst, self.pg6.remote_ip4)
3486 self.assertEqual(tcp.dport, 12345)
3487 self.assert_packet_checksums_valid(p)
3489 self.logger.error(ppp("Unexpected or invalid packet:", p))
3492 def test_one_armed_nat44_static(self):
3493 """ NAT44ED One armed NAT and 1:1 NAPT asymmetrical rule """
3495 remote_host = self.pg4.remote_hosts[0]
3496 local_host = self.pg4.remote_hosts[1]
3501 self.vapi.nat44_forwarding_enable_disable(enable=1)
3502 self.nat_add_address(self.nat_addr, twice_nat=1)
3503 flags = (self.config_flags.NAT_IS_OUT2IN_ONLY |
3504 self.config_flags.NAT_IS_TWICE_NAT)
3505 self.nat_add_static_mapping(local_host.ip4, self.nat_addr,
3506 local_port, external_port,
3507 proto=IP_PROTOS.tcp, flags=flags)
3508 flags = self.config_flags.NAT_IS_INSIDE
3509 self.vapi.nat44_interface_add_del_feature(
3510 sw_if_index=self.pg4.sw_if_index,
3512 self.vapi.nat44_interface_add_del_feature(
3513 sw_if_index=self.pg4.sw_if_index,
3514 flags=flags, is_add=1)
3516 # from client to service
3517 p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
3518 IP(src=remote_host.ip4, dst=self.nat_addr) /
3519 TCP(sport=12345, dport=external_port))
3520 self.pg4.add_stream(p)
3521 self.pg_enable_capture(self.pg_interfaces)
3523 capture = self.pg4.get_capture(1)
3528 self.assertEqual(ip.dst, local_host.ip4)
3529 self.assertEqual(ip.src, self.nat_addr)
3530 self.assertEqual(tcp.dport, local_port)
3531 self.assertNotEqual(tcp.sport, 12345)
3532 eh_port_in = tcp.sport
3533 self.assert_packet_checksums_valid(p)
3535 self.logger.error(ppp("Unexpected or invalid packet:", p))
3538 # from service back to client
3539 p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
3540 IP(src=local_host.ip4, dst=self.nat_addr) /
3541 TCP(sport=local_port, dport=eh_port_in))
3542 self.pg4.add_stream(p)
3543 self.pg_enable_capture(self.pg_interfaces)
3545 capture = self.pg4.get_capture(1)
3550 self.assertEqual(ip.src, self.nat_addr)
3551 self.assertEqual(ip.dst, remote_host.ip4)
3552 self.assertEqual(tcp.sport, external_port)
3553 self.assertEqual(tcp.dport, 12345)
3554 self.assert_packet_checksums_valid(p)
3556 self.logger.error(ppp("Unexpected or invalid packet:", p))
3560 if __name__ == '__main__':
3561 unittest.main(testRunner=VppTestRunner)