7 from framework import VppTestCase, VppTestRunner, running_extended_tests
8 from scapy.layers.inet import IP, TCP, UDP, ICMP
9 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
10 from scapy.layers.l2 import Ether, ARP
11 from scapy.data import IP_PROTOS
13 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
14 from time import sleep
17 class MethodHolder(VppTestCase):
18 """ SNAT create capture and verify method holder """
22 super(MethodHolder, cls).setUpClass()
25 super(MethodHolder, self).tearDown()
27 def create_stream_in(self, in_if, out_if, ttl=64):
29 Create packet stream for inside network
31 :param in_if: Inside interface
32 :param out_if: Outside interface
33 :param ttl: TTL of generated packets
37 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
38 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
39 TCP(sport=self.tcp_port_in))
43 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
44 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
45 UDP(sport=self.udp_port_in))
49 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
50 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
51 ICMP(id=self.icmp_id_in, type='echo-request'))
56 def create_stream_out(self, out_if, dst_ip=None, ttl=64):
58 Create packet stream for outside network
60 :param out_if: Outside interface
61 :param dst_ip: Destination IP address (Default use global SNAT address)
62 :param ttl: TTL of generated packets
65 dst_ip = self.snat_addr
68 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
69 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
70 TCP(dport=self.tcp_port_out))
74 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
75 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
76 UDP(dport=self.udp_port_out))
80 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
81 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
82 ICMP(id=self.icmp_id_out, type='echo-reply'))
87 def verify_capture_out(self, capture, nat_ip=None, same_port=False,
90 Verify captured packets on outside network
92 :param capture: Captured packets
93 :param nat_ip: Translated IP address (Default use global SNAT address)
94 :param same_port: Sorce port number is not translated (Default False)
95 :param packet_num: Expected number of packets (Default 3)
98 nat_ip = self.snat_addr
99 self.assertEqual(packet_num, len(capture))
100 for packet in capture:
102 self.assertEqual(packet[IP].src, nat_ip)
103 if packet.haslayer(TCP):
105 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
108 packet[TCP].sport, self.tcp_port_in)
109 self.tcp_port_out = packet[TCP].sport
110 elif packet.haslayer(UDP):
112 self.assertEqual(packet[UDP].sport, self.udp_port_in)
115 packet[UDP].sport, self.udp_port_in)
116 self.udp_port_out = packet[UDP].sport
119 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
121 self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
122 self.icmp_id_out = packet[ICMP].id
124 self.logger.error(ppp("Unexpected or invalid packet "
125 "(outside network):", packet))
128 def verify_capture_in(self, capture, in_if, packet_num=3):
130 Verify captured packets on inside network
132 :param capture: Captured packets
133 :param in_if: Inside interface
134 :param packet_num: Expected number of packets (Default 3)
136 self.assertEqual(packet_num, len(capture))
137 for packet in capture:
139 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
140 if packet.haslayer(TCP):
141 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
142 elif packet.haslayer(UDP):
143 self.assertEqual(packet[UDP].dport, self.udp_port_in)
145 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
147 self.logger.error(ppp("Unexpected or invalid packet "
148 "(inside network):", packet))
151 def verify_capture_no_translation(self, capture, ingress_if, egress_if):
153 Verify captured packet that don't have to be translated
155 :param capture: Captured packets
156 :param ingress_if: Ingress interface
157 :param egress_if: Egress interface
159 for packet in capture:
161 self.assertEqual(packet[IP].src, ingress_if.remote_ip4)
162 self.assertEqual(packet[IP].dst, egress_if.remote_ip4)
163 if packet.haslayer(TCP):
164 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
165 elif packet.haslayer(UDP):
166 self.assertEqual(packet[UDP].sport, self.udp_port_in)
168 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
170 self.logger.error(ppp("Unexpected or invalid packet "
171 "(inside network):", packet))
174 def verify_capture_out_with_icmp_errors(self, capture, src_ip=None,
175 packet_num=3, icmp_type=11):
177 Verify captured packets with ICMP errors on outside network
179 :param capture: Captured packets
180 :param src_ip: Translated IP address or IP address of VPP
181 (Default use global SNAT address)
182 :param packet_num: Expected number of packets (Default 3)
183 :param icmp_type: Type of error ICMP packet
184 we are expecting (Default 11)
187 src_ip = self.snat_addr
188 self.assertEqual(packet_num, len(capture))
189 for packet in capture:
191 self.assertEqual(packet[IP].src, src_ip)
192 self.assertTrue(packet.haslayer(ICMP))
194 self.assertEqual(icmp.type, icmp_type)
195 self.assertTrue(icmp.haslayer(IPerror))
196 inner_ip = icmp[IPerror]
197 if inner_ip.haslayer(TCPerror):
198 self.assertEqual(inner_ip[TCPerror].dport,
200 elif inner_ip.haslayer(UDPerror):
201 self.assertEqual(inner_ip[UDPerror].dport,
204 self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out)
206 self.logger.error(ppp("Unexpected or invalid packet "
207 "(outside network):", packet))
210 def verify_capture_in_with_icmp_errors(self, capture, in_if, packet_num=3,
213 Verify captured packets with ICMP errors on inside network
215 :param capture: Captured packets
216 :param in_if: Inside interface
217 :param packet_num: Expected number of packets (Default 3)
218 :param icmp_type: Type of error ICMP packet
219 we are expecting (Default 11)
221 self.assertEqual(packet_num, len(capture))
222 for packet in capture:
224 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
225 self.assertTrue(packet.haslayer(ICMP))
227 self.assertEqual(icmp.type, icmp_type)
228 self.assertTrue(icmp.haslayer(IPerror))
229 inner_ip = icmp[IPerror]
230 if inner_ip.haslayer(TCPerror):
231 self.assertEqual(inner_ip[TCPerror].sport,
233 elif inner_ip.haslayer(UDPerror):
234 self.assertEqual(inner_ip[UDPerror].sport,
237 self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_in)
239 self.logger.error(ppp("Unexpected or invalid packet "
240 "(inside network):", packet))
243 def verify_ipfix_nat44_ses(self, data):
245 Verify IPFIX NAT44 session create/delete event
247 :param data: Decoded IPFIX data records
249 nat44_ses_create_num = 0
250 nat44_ses_delete_num = 0
251 self.assertEqual(6, len(data))
254 self.assertIn(ord(record[230]), [4, 5])
255 if ord(record[230]) == 4:
256 nat44_ses_create_num += 1
258 nat44_ses_delete_num += 1
260 self.assertEqual(self.pg0.remote_ip4n, record[8])
261 # postNATSourceIPv4Address
262 self.assertEqual(socket.inet_pton(socket.AF_INET, self.snat_addr),
265 self.assertEqual(struct.pack("!I", 0), record[234])
266 # protocolIdentifier/sourceTransportPort/postNAPTSourceTransportPort
267 if IP_PROTOS.icmp == ord(record[4]):
268 self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
269 self.assertEqual(struct.pack("!H", self.icmp_id_out),
271 elif IP_PROTOS.tcp == ord(record[4]):
272 self.assertEqual(struct.pack("!H", self.tcp_port_in),
274 self.assertEqual(struct.pack("!H", self.tcp_port_out),
276 elif IP_PROTOS.udp == ord(record[4]):
277 self.assertEqual(struct.pack("!H", self.udp_port_in),
279 self.assertEqual(struct.pack("!H", self.udp_port_out),
282 self.fail("Invalid protocol")
283 self.assertEqual(3, nat44_ses_create_num)
284 self.assertEqual(3, nat44_ses_delete_num)
286 def verify_ipfix_addr_exhausted(self, data):
288 Verify IPFIX NAT addresses event
290 :param data: Decoded IPFIX data records
292 self.assertEqual(1, len(data))
295 self.assertEqual(ord(record[230]), 3)
297 self.assertEqual(struct.pack("!I", 0), record[283])
300 class TestSNAT(MethodHolder):
301 """ SNAT Test Cases """
305 super(TestSNAT, cls).setUpClass()
308 cls.tcp_port_in = 6303
309 cls.tcp_port_out = 6303
310 cls.udp_port_in = 6304
311 cls.udp_port_out = 6304
312 cls.icmp_id_in = 6305
313 cls.icmp_id_out = 6305
314 cls.snat_addr = '10.0.0.3'
316 cls.create_pg_interfaces(range(8))
317 cls.interfaces = list(cls.pg_interfaces[0:4])
319 for i in cls.interfaces:
324 cls.pg0.generate_remote_hosts(2)
325 cls.pg0.configure_ipv4_neighbors()
327 cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
329 cls.pg4._local_ip4 = "172.16.255.1"
330 cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
331 cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
332 cls.pg4.set_table_ip4(10)
333 cls.pg5._local_ip4 = "172.16.255.3"
334 cls.pg5._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
335 cls.pg5._remote_hosts[0]._ip4 = "172.16.255.4"
336 cls.pg5.set_table_ip4(10)
337 cls.pg6._local_ip4 = "172.16.255.1"
338 cls.pg6._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
339 cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
340 cls.pg6.set_table_ip4(20)
341 for i in cls.overlapping_interfaces:
349 super(TestSNAT, cls).tearDownClass()
352 def clear_snat(self):
354 Clear SNAT configuration.
356 if self.pg7.has_ip4_config:
357 self.pg7.unconfig_ip4()
359 interfaces = self.vapi.snat_interface_addr_dump()
360 for intf in interfaces:
361 self.vapi.snat_add_interface_addr(intf.sw_if_index, is_add=0)
363 self.vapi.snat_ipfix(enable=0)
365 interfaces = self.vapi.snat_interface_dump()
366 for intf in interfaces:
367 self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
371 static_mappings = self.vapi.snat_static_mapping_dump()
372 for sm in static_mappings:
373 self.vapi.snat_add_static_mapping(sm.local_ip_address,
374 sm.external_ip_address,
375 local_port=sm.local_port,
376 external_port=sm.external_port,
377 addr_only=sm.addr_only,
379 protocol=sm.protocol,
382 adresses = self.vapi.snat_address_dump()
383 for addr in adresses:
384 self.vapi.snat_add_address_range(addr.ip_address,
388 def snat_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
389 local_port=0, external_port=0, vrf_id=0,
390 is_add=1, external_sw_if_index=0xFFFFFFFF,
393 Add/delete S-NAT static mapping
395 :param local_ip: Local IP address
396 :param external_ip: External IP address
397 :param local_port: Local port number (Optional)
398 :param external_port: External port number (Optional)
399 :param vrf_id: VRF ID (Default 0)
400 :param is_add: 1 if add, 0 if delete (Default add)
401 :param external_sw_if_index: External interface instead of IP address
402 :param proto: IP protocol (Mandatory if port specified)
405 if local_port and external_port:
407 l_ip = socket.inet_pton(socket.AF_INET, local_ip)
408 e_ip = socket.inet_pton(socket.AF_INET, external_ip)
409 self.vapi.snat_add_static_mapping(
412 external_sw_if_index,
420 def snat_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF):
422 Add/delete S-NAT address
424 :param ip: IP address
425 :param is_add: 1 if add, 0 if delete (Default add)
427 snat_addr = socket.inet_pton(socket.AF_INET, ip)
428 self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add,
431 def test_dynamic(self):
432 """ SNAT dynamic translation test """
434 self.snat_add_address(self.snat_addr)
435 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
436 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
440 pkts = self.create_stream_in(self.pg0, self.pg1)
441 self.pg0.add_stream(pkts)
442 self.pg_enable_capture(self.pg_interfaces)
444 capture = self.pg1.get_capture(len(pkts))
445 self.verify_capture_out(capture)
448 pkts = self.create_stream_out(self.pg1)
449 self.pg1.add_stream(pkts)
450 self.pg_enable_capture(self.pg_interfaces)
452 capture = self.pg0.get_capture(len(pkts))
453 self.verify_capture_in(capture, self.pg0)
455 def test_dynamic_icmp_errors_in2out_ttl_1(self):
456 """ SNAT handling of client packets with TTL=1 """
458 self.snat_add_address(self.snat_addr)
459 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
460 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
463 # Client side - generate traffic
464 pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
465 self.pg0.add_stream(pkts)
466 self.pg_enable_capture(self.pg_interfaces)
469 # Client side - verify ICMP type 11 packets
470 capture = self.pg0.get_capture(len(pkts))
471 self.verify_capture_in_with_icmp_errors(capture, self.pg0)
473 def test_dynamic_icmp_errors_out2in_ttl_1(self):
474 """ SNAT handling of server packets with TTL=1 """
476 self.snat_add_address(self.snat_addr)
477 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
478 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
481 # Client side - create sessions
482 pkts = self.create_stream_in(self.pg0, self.pg1)
483 self.pg0.add_stream(pkts)
484 self.pg_enable_capture(self.pg_interfaces)
487 # Server side - generate traffic
488 capture = self.pg1.get_capture(len(pkts))
489 self.verify_capture_out(capture)
490 pkts = self.create_stream_out(self.pg1, ttl=1)
491 self.pg1.add_stream(pkts)
492 self.pg_enable_capture(self.pg_interfaces)
495 # Server side - verify ICMP type 11 packets
496 capture = self.pg1.get_capture(len(pkts))
497 self.verify_capture_out_with_icmp_errors(capture,
498 src_ip=self.pg1.local_ip4)
500 def test_dynamic_icmp_errors_in2out_ttl_2(self):
501 """ SNAT handling of error responses to client packets with TTL=2 """
503 self.snat_add_address(self.snat_addr)
504 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
505 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
508 # Client side - generate traffic
509 pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
510 self.pg0.add_stream(pkts)
511 self.pg_enable_capture(self.pg_interfaces)
514 # Server side - simulate ICMP type 11 response
515 capture = self.pg1.get_capture(len(pkts))
516 pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
517 IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
518 ICMP(type=11) / packet[IP] for packet in capture]
519 self.pg1.add_stream(pkts)
520 self.pg_enable_capture(self.pg_interfaces)
523 # Client side - verify ICMP type 11 packets
524 capture = self.pg0.get_capture(len(pkts))
525 self.verify_capture_in_with_icmp_errors(capture, self.pg0)
527 def test_dynamic_icmp_errors_out2in_ttl_2(self):
528 """ SNAT handling of error responses to server packets with TTL=2 """
530 self.snat_add_address(self.snat_addr)
531 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
532 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
535 # Client side - create sessions
536 pkts = self.create_stream_in(self.pg0, self.pg1)
537 self.pg0.add_stream(pkts)
538 self.pg_enable_capture(self.pg_interfaces)
541 # Server side - generate traffic
542 capture = self.pg1.get_capture(len(pkts))
543 self.verify_capture_out(capture)
544 pkts = self.create_stream_out(self.pg1, ttl=2)
545 self.pg1.add_stream(pkts)
546 self.pg_enable_capture(self.pg_interfaces)
549 # Client side - simulate ICMP type 11 response
550 capture = self.pg0.get_capture(len(pkts))
551 pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
552 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
553 ICMP(type=11) / packet[IP] for packet in capture]
554 self.pg0.add_stream(pkts)
555 self.pg_enable_capture(self.pg_interfaces)
558 # Server side - verify ICMP type 11 packets
559 capture = self.pg1.get_capture(len(pkts))
560 self.verify_capture_out_with_icmp_errors(capture)
562 def test_ping_out_interface_from_outside(self):
563 """ Ping SNAT out interface from outside network """
565 self.snat_add_address(self.snat_addr)
566 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
567 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
570 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
571 IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) /
572 ICMP(id=self.icmp_id_out, type='echo-request'))
574 self.pg1.add_stream(pkts)
575 self.pg_enable_capture(self.pg_interfaces)
577 capture = self.pg1.get_capture(len(pkts))
578 self.assertEqual(1, len(capture))
581 self.assertEqual(packet[IP].src, self.pg1.local_ip4)
582 self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
583 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
584 self.assertEqual(packet[ICMP].type, 0) # echo reply
586 self.logger.error(ppp("Unexpected or invalid packet "
587 "(outside network):", packet))
590 def test_ping_internal_host_from_outside(self):
591 """ Ping internal host from outside network """
593 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr)
594 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
595 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
599 pkt = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
600 IP(src=self.pg1.remote_ip4, dst=self.snat_addr, ttl=64) /
601 ICMP(id=self.icmp_id_out, type='echo-request'))
602 self.pg1.add_stream(pkt)
603 self.pg_enable_capture(self.pg_interfaces)
605 capture = self.pg0.get_capture(1)
606 self.verify_capture_in(capture, self.pg0, packet_num=1)
607 self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
610 pkt = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
611 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
612 ICMP(id=self.icmp_id_in, type='echo-reply'))
613 self.pg0.add_stream(pkt)
614 self.pg_enable_capture(self.pg_interfaces)
616 capture = self.pg1.get_capture(1)
617 self.verify_capture_out(capture, same_port=True, packet_num=1)
618 self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
620 def test_static_in(self):
621 """ SNAT 1:1 NAT initialized from inside network """
624 self.tcp_port_out = 6303
625 self.udp_port_out = 6304
626 self.icmp_id_out = 6305
628 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
629 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
630 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
634 pkts = self.create_stream_in(self.pg0, self.pg1)
635 self.pg0.add_stream(pkts)
636 self.pg_enable_capture(self.pg_interfaces)
638 capture = self.pg1.get_capture(len(pkts))
639 self.verify_capture_out(capture, nat_ip, True)
642 pkts = self.create_stream_out(self.pg1, nat_ip)
643 self.pg1.add_stream(pkts)
644 self.pg_enable_capture(self.pg_interfaces)
646 capture = self.pg0.get_capture(len(pkts))
647 self.verify_capture_in(capture, self.pg0)
649 def test_static_out(self):
650 """ SNAT 1:1 NAT initialized from outside network """
653 self.tcp_port_out = 6303
654 self.udp_port_out = 6304
655 self.icmp_id_out = 6305
657 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
658 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
659 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
663 pkts = self.create_stream_out(self.pg1, nat_ip)
664 self.pg1.add_stream(pkts)
665 self.pg_enable_capture(self.pg_interfaces)
667 capture = self.pg0.get_capture(len(pkts))
668 self.verify_capture_in(capture, self.pg0)
671 pkts = self.create_stream_in(self.pg0, self.pg1)
672 self.pg0.add_stream(pkts)
673 self.pg_enable_capture(self.pg_interfaces)
675 capture = self.pg1.get_capture(len(pkts))
676 self.verify_capture_out(capture, nat_ip, True)
678 def test_static_with_port_in(self):
679 """ SNAT 1:1 NAT with port initialized from inside network """
681 self.tcp_port_out = 3606
682 self.udp_port_out = 3607
683 self.icmp_id_out = 3608
685 self.snat_add_address(self.snat_addr)
686 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
687 self.tcp_port_in, self.tcp_port_out,
689 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
690 self.udp_port_in, self.udp_port_out,
692 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
693 self.icmp_id_in, self.icmp_id_out,
694 proto=IP_PROTOS.icmp)
695 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
696 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
700 pkts = self.create_stream_in(self.pg0, self.pg1)
701 self.pg0.add_stream(pkts)
702 self.pg_enable_capture(self.pg_interfaces)
704 capture = self.pg1.get_capture(len(pkts))
705 self.verify_capture_out(capture)
708 pkts = self.create_stream_out(self.pg1)
709 self.pg1.add_stream(pkts)
710 self.pg_enable_capture(self.pg_interfaces)
712 capture = self.pg0.get_capture(len(pkts))
713 self.verify_capture_in(capture, self.pg0)
715 def test_static_with_port_out(self):
716 """ SNAT 1:1 NAT with port initialized from outside network """
718 self.tcp_port_out = 30606
719 self.udp_port_out = 30607
720 self.icmp_id_out = 30608
722 self.snat_add_address(self.snat_addr)
723 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
724 self.tcp_port_in, self.tcp_port_out,
726 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
727 self.udp_port_in, self.udp_port_out,
729 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
730 self.icmp_id_in, self.icmp_id_out,
731 proto=IP_PROTOS.icmp)
732 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
733 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
737 pkts = self.create_stream_out(self.pg1)
738 self.pg1.add_stream(pkts)
739 self.pg_enable_capture(self.pg_interfaces)
741 capture = self.pg0.get_capture(len(pkts))
742 self.verify_capture_in(capture, self.pg0)
745 pkts = self.create_stream_in(self.pg0, self.pg1)
746 self.pg0.add_stream(pkts)
747 self.pg_enable_capture(self.pg_interfaces)
749 capture = self.pg1.get_capture(len(pkts))
750 self.verify_capture_out(capture)
752 def test_static_vrf_aware(self):
753 """ SNAT 1:1 NAT VRF awareness """
755 nat_ip1 = "10.0.0.30"
756 nat_ip2 = "10.0.0.40"
757 self.tcp_port_out = 6303
758 self.udp_port_out = 6304
759 self.icmp_id_out = 6305
761 self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
763 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
765 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
767 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
768 self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
770 # inside interface VRF match SNAT static mapping VRF
771 pkts = self.create_stream_in(self.pg4, self.pg3)
772 self.pg4.add_stream(pkts)
773 self.pg_enable_capture(self.pg_interfaces)
775 capture = self.pg3.get_capture(len(pkts))
776 self.verify_capture_out(capture, nat_ip1, True)
778 # inside interface VRF don't match SNAT static mapping VRF (packets
780 pkts = self.create_stream_in(self.pg0, self.pg3)
781 self.pg0.add_stream(pkts)
782 self.pg_enable_capture(self.pg_interfaces)
784 self.pg3.assert_nothing_captured()
786 def test_multiple_inside_interfaces(self):
787 """ SNAT multiple inside interfaces (non-overlapping address space) """
789 self.snat_add_address(self.snat_addr)
790 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
791 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
792 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
795 # between two S-NAT inside interfaces (no translation)
796 pkts = self.create_stream_in(self.pg0, self.pg1)
797 self.pg0.add_stream(pkts)
798 self.pg_enable_capture(self.pg_interfaces)
800 capture = self.pg1.get_capture(len(pkts))
801 self.verify_capture_no_translation(capture, self.pg0, self.pg1)
803 # from S-NAT inside to interface without S-NAT feature (no translation)
804 pkts = self.create_stream_in(self.pg0, self.pg2)
805 self.pg0.add_stream(pkts)
806 self.pg_enable_capture(self.pg_interfaces)
808 capture = self.pg2.get_capture(len(pkts))
809 self.verify_capture_no_translation(capture, self.pg0, self.pg2)
811 # in2out 1st interface
812 pkts = self.create_stream_in(self.pg0, self.pg3)
813 self.pg0.add_stream(pkts)
814 self.pg_enable_capture(self.pg_interfaces)
816 capture = self.pg3.get_capture(len(pkts))
817 self.verify_capture_out(capture)
819 # out2in 1st interface
820 pkts = self.create_stream_out(self.pg3)
821 self.pg3.add_stream(pkts)
822 self.pg_enable_capture(self.pg_interfaces)
824 capture = self.pg0.get_capture(len(pkts))
825 self.verify_capture_in(capture, self.pg0)
827 # in2out 2nd interface
828 pkts = self.create_stream_in(self.pg1, self.pg3)
829 self.pg1.add_stream(pkts)
830 self.pg_enable_capture(self.pg_interfaces)
832 capture = self.pg3.get_capture(len(pkts))
833 self.verify_capture_out(capture)
835 # out2in 2nd interface
836 pkts = self.create_stream_out(self.pg3)
837 self.pg3.add_stream(pkts)
838 self.pg_enable_capture(self.pg_interfaces)
840 capture = self.pg1.get_capture(len(pkts))
841 self.verify_capture_in(capture, self.pg1)
843 def test_inside_overlapping_interfaces(self):
844 """ SNAT multiple inside interfaces with overlapping address space """
846 static_nat_ip = "10.0.0.10"
847 self.snat_add_address(self.snat_addr)
848 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
850 self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
851 self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
852 self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
853 self.snat_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
856 # between S-NAT inside interfaces with same VRF (no translation)
857 pkts = self.create_stream_in(self.pg4, self.pg5)
858 self.pg4.add_stream(pkts)
859 self.pg_enable_capture(self.pg_interfaces)
861 capture = self.pg5.get_capture(len(pkts))
862 self.verify_capture_no_translation(capture, self.pg4, self.pg5)
864 # between S-NAT inside interfaces with different VRF (hairpinning)
865 p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
866 IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
867 TCP(sport=1234, dport=5678))
868 self.pg4.add_stream(p)
869 self.pg_enable_capture(self.pg_interfaces)
871 capture = self.pg6.get_capture(1)
876 self.assertEqual(ip.src, self.snat_addr)
877 self.assertEqual(ip.dst, self.pg6.remote_ip4)
878 self.assertNotEqual(tcp.sport, 1234)
879 self.assertEqual(tcp.dport, 5678)
881 self.logger.error(ppp("Unexpected or invalid packet:", p))
884 # in2out 1st interface
885 pkts = self.create_stream_in(self.pg4, self.pg3)
886 self.pg4.add_stream(pkts)
887 self.pg_enable_capture(self.pg_interfaces)
889 capture = self.pg3.get_capture(len(pkts))
890 self.verify_capture_out(capture)
892 # out2in 1st interface
893 pkts = self.create_stream_out(self.pg3)
894 self.pg3.add_stream(pkts)
895 self.pg_enable_capture(self.pg_interfaces)
897 capture = self.pg4.get_capture(len(pkts))
898 self.verify_capture_in(capture, self.pg4)
900 # in2out 2nd interface
901 pkts = self.create_stream_in(self.pg5, self.pg3)
902 self.pg5.add_stream(pkts)
903 self.pg_enable_capture(self.pg_interfaces)
905 capture = self.pg3.get_capture(len(pkts))
906 self.verify_capture_out(capture)
908 # out2in 2nd interface
909 pkts = self.create_stream_out(self.pg3)
910 self.pg3.add_stream(pkts)
911 self.pg_enable_capture(self.pg_interfaces)
913 capture = self.pg5.get_capture(len(pkts))
914 self.verify_capture_in(capture, self.pg5)
917 addresses = self.vapi.snat_address_dump()
918 self.assertEqual(len(addresses), 1)
919 sessions = self.vapi.snat_user_session_dump(self.pg5.remote_ip4n, 10)
920 self.assertEqual(len(sessions), 3)
921 for session in sessions:
922 self.assertFalse(session.is_static)
923 self.assertEqual(session.inside_ip_address[0:4],
924 self.pg5.remote_ip4n)
925 self.assertEqual(session.outside_ip_address,
926 addresses[0].ip_address)
927 self.assertEqual(sessions[0].protocol, IP_PROTOS.tcp)
928 self.assertEqual(sessions[1].protocol, IP_PROTOS.udp)
929 self.assertEqual(sessions[2].protocol, IP_PROTOS.icmp)
930 self.assertEqual(sessions[0].inside_port, self.tcp_port_in)
931 self.assertEqual(sessions[1].inside_port, self.udp_port_in)
932 self.assertEqual(sessions[2].inside_port, self.icmp_id_in)
933 self.assertEqual(sessions[0].outside_port, self.tcp_port_out)
934 self.assertEqual(sessions[1].outside_port, self.udp_port_out)
935 self.assertEqual(sessions[2].outside_port, self.icmp_id_out)
937 # in2out 3rd interface
938 pkts = self.create_stream_in(self.pg6, self.pg3)
939 self.pg6.add_stream(pkts)
940 self.pg_enable_capture(self.pg_interfaces)
942 capture = self.pg3.get_capture(len(pkts))
943 self.verify_capture_out(capture, static_nat_ip, True)
945 # out2in 3rd interface
946 pkts = self.create_stream_out(self.pg3, static_nat_ip)
947 self.pg3.add_stream(pkts)
948 self.pg_enable_capture(self.pg_interfaces)
950 capture = self.pg6.get_capture(len(pkts))
951 self.verify_capture_in(capture, self.pg6)
953 # general user and session dump verifications
954 users = self.vapi.snat_user_dump()
955 self.assertTrue(len(users) >= 3)
956 addresses = self.vapi.snat_address_dump()
957 self.assertEqual(len(addresses), 1)
959 sessions = self.vapi.snat_user_session_dump(user.ip_address,
961 for session in sessions:
962 self.assertEqual(user.ip_address, session.inside_ip_address)
963 self.assertTrue(session.total_bytes > session.total_pkts > 0)
964 self.assertTrue(session.protocol in
965 [IP_PROTOS.tcp, IP_PROTOS.udp,
969 sessions = self.vapi.snat_user_session_dump(self.pg4.remote_ip4n, 10)
970 self.assertTrue(len(sessions) >= 4)
971 for session in sessions:
972 self.assertFalse(session.is_static)
973 self.assertEqual(session.inside_ip_address[0:4],
974 self.pg4.remote_ip4n)
975 self.assertEqual(session.outside_ip_address,
976 addresses[0].ip_address)
979 sessions = self.vapi.snat_user_session_dump(self.pg6.remote_ip4n, 20)
980 self.assertTrue(len(sessions) >= 3)
981 for session in sessions:
982 self.assertTrue(session.is_static)
983 self.assertEqual(session.inside_ip_address[0:4],
984 self.pg6.remote_ip4n)
985 self.assertEqual(map(ord, session.outside_ip_address[0:4]),
986 map(int, static_nat_ip.split('.')))
987 self.assertTrue(session.inside_port in
988 [self.tcp_port_in, self.udp_port_in,
991 def test_hairpinning(self):
992 """ SNAT hairpinning """
994 host = self.pg0.remote_hosts[0]
995 server = self.pg0.remote_hosts[1]
998 server_in_port = 5678
999 server_out_port = 8765
1001 self.snat_add_address(self.snat_addr)
1002 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1003 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1005 # add static mapping for server
1006 self.snat_add_static_mapping(server.ip4, self.snat_addr,
1007 server_in_port, server_out_port,
1008 proto=IP_PROTOS.tcp)
1010 # send packet from host to server
1011 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
1012 IP(src=host.ip4, dst=self.snat_addr) /
1013 TCP(sport=host_in_port, dport=server_out_port))
1014 self.pg0.add_stream(p)
1015 self.pg_enable_capture(self.pg_interfaces)
1017 capture = self.pg0.get_capture(1)
1022 self.assertEqual(ip.src, self.snat_addr)
1023 self.assertEqual(ip.dst, server.ip4)
1024 self.assertNotEqual(tcp.sport, host_in_port)
1025 self.assertEqual(tcp.dport, server_in_port)
1026 host_out_port = tcp.sport
1028 self.logger.error(ppp("Unexpected or invalid packet:", p))
1031 # send reply from server to host
1032 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
1033 IP(src=server.ip4, dst=self.snat_addr) /
1034 TCP(sport=server_in_port, dport=host_out_port))
1035 self.pg0.add_stream(p)
1036 self.pg_enable_capture(self.pg_interfaces)
1038 capture = self.pg0.get_capture(1)
1043 self.assertEqual(ip.src, self.snat_addr)
1044 self.assertEqual(ip.dst, host.ip4)
1045 self.assertEqual(tcp.sport, server_out_port)
1046 self.assertEqual(tcp.dport, host_in_port)
1048 self.logger.error(ppp("Unexpected or invalid packet:"), p)
1051 def test_max_translations_per_user(self):
1052 """ MAX translations per user - recycle the least recently used """
1054 self.snat_add_address(self.snat_addr)
1055 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1056 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1059 # get maximum number of translations per user
1060 snat_config = self.vapi.snat_show_config()
1062 # send more than maximum number of translations per user packets
1063 pkts_num = snat_config.max_translations_per_user + 5
1065 for port in range(0, pkts_num):
1066 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1067 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1068 TCP(sport=1025 + port))
1070 self.pg0.add_stream(pkts)
1071 self.pg_enable_capture(self.pg_interfaces)
1074 # verify number of translated packet
1075 self.pg1.get_capture(pkts_num)
1077 def test_interface_addr(self):
1078 """ Acquire SNAT addresses from interface """
1079 self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
1081 # no address in NAT pool
1082 adresses = self.vapi.snat_address_dump()
1083 self.assertEqual(0, len(adresses))
1085 # configure interface address and check NAT address pool
1086 self.pg7.config_ip4()
1087 adresses = self.vapi.snat_address_dump()
1088 self.assertEqual(1, len(adresses))
1089 self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n)
1091 # remove interface address and check NAT address pool
1092 self.pg7.unconfig_ip4()
1093 adresses = self.vapi.snat_address_dump()
1094 self.assertEqual(0, len(adresses))
1096 def test_interface_addr_static_mapping(self):
1097 """ Static mapping with addresses from interface """
1098 self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
1099 self.snat_add_static_mapping('1.2.3.4',
1100 external_sw_if_index=self.pg7.sw_if_index)
1102 # static mappings with external interface
1103 static_mappings = self.vapi.snat_static_mapping_dump()
1104 self.assertEqual(1, len(static_mappings))
1105 self.assertEqual(self.pg7.sw_if_index,
1106 static_mappings[0].external_sw_if_index)
1108 # configure interface address and check static mappings
1109 self.pg7.config_ip4()
1110 static_mappings = self.vapi.snat_static_mapping_dump()
1111 self.assertEqual(1, len(static_mappings))
1112 self.assertEqual(static_mappings[0].external_ip_address[0:4],
1113 self.pg7.local_ip4n)
1114 self.assertEqual(0xFFFFFFFF, static_mappings[0].external_sw_if_index)
1116 # remove interface address and check static mappings
1117 self.pg7.unconfig_ip4()
1118 static_mappings = self.vapi.snat_static_mapping_dump()
1119 self.assertEqual(0, len(static_mappings))
1121 def test_ipfix_nat44_sess(self):
1122 """ S-NAT IPFIX logging NAT44 session created/delted """
1123 self.snat_add_address(self.snat_addr)
1124 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1125 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1127 self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
1128 src_address=self.pg3.local_ip4n,
1130 template_interval=10)
1131 self.vapi.snat_ipfix()
1133 pkts = self.create_stream_in(self.pg0, self.pg1)
1134 self.pg0.add_stream(pkts)
1135 self.pg_enable_capture(self.pg_interfaces)
1137 capture = self.pg1.get_capture(len(pkts))
1138 self.verify_capture_out(capture)
1139 self.snat_add_address(self.snat_addr, is_add=0)
1140 self.vapi.cli("ipfix flush") # FIXME this should be an API call
1141 capture = self.pg3.get_capture(3)
1142 ipfix = IPFIXDecoder()
1143 # first load template
1145 self.assertTrue(p.haslayer(IPFIX))
1146 if p.haslayer(Template):
1147 ipfix.add_template(p.getlayer(Template))
1148 # verify events in data set
1150 if p.haslayer(Data):
1151 data = ipfix.decode_data_set(p.getlayer(Set))
1152 self.verify_ipfix_nat44_ses(data)
1154 def test_ipfix_addr_exhausted(self):
1155 """ S-NAT IPFIX logging NAT addresses exhausted """
1156 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1157 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1159 self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
1160 src_address=self.pg3.local_ip4n,
1162 template_interval=10)
1163 self.vapi.snat_ipfix()
1165 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1166 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1168 self.pg0.add_stream(p)
1169 self.pg_enable_capture(self.pg_interfaces)
1171 capture = self.pg1.get_capture(0)
1172 self.vapi.cli("ipfix flush") # FIXME this should be an API call
1173 capture = self.pg3.get_capture(3)
1174 ipfix = IPFIXDecoder()
1175 # first load template
1177 self.assertTrue(p.haslayer(IPFIX))
1178 if p.haslayer(Template):
1179 ipfix.add_template(p.getlayer(Template))
1180 # verify events in data set
1182 if p.haslayer(Data):
1183 data = ipfix.decode_data_set(p.getlayer(Set))
1184 self.verify_ipfix_addr_exhausted(data)
1186 def test_pool_addr_fib(self):
1187 """ S-NAT add pool addresses to FIB """
1188 static_addr = '10.0.0.10'
1189 self.snat_add_address(self.snat_addr)
1190 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1191 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1193 self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr)
1196 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1197 ARP(op=ARP.who_has, pdst=self.snat_addr,
1198 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
1199 self.pg1.add_stream(p)
1200 self.pg_enable_capture(self.pg_interfaces)
1202 capture = self.pg1.get_capture(1)
1203 self.assertTrue(capture[0].haslayer(ARP))
1204 self.assertTrue(capture[0][ARP].op, ARP.is_at)
1207 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1208 ARP(op=ARP.who_has, pdst=static_addr,
1209 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
1210 self.pg1.add_stream(p)
1211 self.pg_enable_capture(self.pg_interfaces)
1213 capture = self.pg1.get_capture(1)
1214 self.assertTrue(capture[0].haslayer(ARP))
1215 self.assertTrue(capture[0][ARP].op, ARP.is_at)
1217 # send ARP to non-SNAT interface
1218 p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1219 ARP(op=ARP.who_has, pdst=self.snat_addr,
1220 psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
1221 self.pg2.add_stream(p)
1222 self.pg_enable_capture(self.pg_interfaces)
1224 capture = self.pg1.get_capture(0)
1226 # remove addresses and verify
1227 self.snat_add_address(self.snat_addr, is_add=0)
1228 self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr,
1231 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1232 ARP(op=ARP.who_has, pdst=self.snat_addr,
1233 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
1234 self.pg1.add_stream(p)
1235 self.pg_enable_capture(self.pg_interfaces)
1237 capture = self.pg1.get_capture(0)
1239 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1240 ARP(op=ARP.who_has, pdst=static_addr,
1241 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
1242 self.pg1.add_stream(p)
1243 self.pg_enable_capture(self.pg_interfaces)
1245 capture = self.pg1.get_capture(0)
1247 def test_vrf_mode(self):
1248 """ S-NAT tenant VRF aware address pool mode """
1252 nat_ip1 = "10.0.0.10"
1253 nat_ip2 = "10.0.0.11"
1255 self.pg0.unconfig_ip4()
1256 self.pg1.unconfig_ip4()
1257 self.pg0.set_table_ip4(vrf_id1)
1258 self.pg1.set_table_ip4(vrf_id2)
1259 self.pg0.config_ip4()
1260 self.pg1.config_ip4()
1262 self.snat_add_address(nat_ip1, vrf_id=vrf_id1)
1263 self.snat_add_address(nat_ip2, vrf_id=vrf_id2)
1264 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1265 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
1266 self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index,
1270 pkts = self.create_stream_in(self.pg0, self.pg2)
1271 self.pg0.add_stream(pkts)
1272 self.pg_enable_capture(self.pg_interfaces)
1274 capture = self.pg2.get_capture(len(pkts))
1275 self.verify_capture_out(capture, nat_ip1)
1278 pkts = self.create_stream_in(self.pg1, self.pg2)
1279 self.pg1.add_stream(pkts)
1280 self.pg_enable_capture(self.pg_interfaces)
1282 capture = self.pg2.get_capture(len(pkts))
1283 self.verify_capture_out(capture, nat_ip2)
1285 def test_vrf_feature_independent(self):
1286 """ S-NAT tenant VRF independent address pool mode """
1288 nat_ip1 = "10.0.0.10"
1289 nat_ip2 = "10.0.0.11"
1291 self.snat_add_address(nat_ip1)
1292 self.snat_add_address(nat_ip2)
1293 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1294 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
1295 self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index,
1299 pkts = self.create_stream_in(self.pg0, self.pg2)
1300 self.pg0.add_stream(pkts)
1301 self.pg_enable_capture(self.pg_interfaces)
1303 capture = self.pg2.get_capture(len(pkts))
1304 self.verify_capture_out(capture, nat_ip1)
1307 pkts = self.create_stream_in(self.pg1, self.pg2)
1308 self.pg1.add_stream(pkts)
1309 self.pg_enable_capture(self.pg_interfaces)
1311 capture = self.pg2.get_capture(len(pkts))
1312 self.verify_capture_out(capture, nat_ip1)
1315 super(TestSNAT, self).tearDown()
1316 if not self.vpp_dead:
1317 self.logger.info(self.vapi.cli("show snat verbose"))
1321 class TestDeterministicNAT(MethodHolder):
1322 """ Deterministic NAT Test Cases """
1325 def setUpConstants(cls):
1326 super(TestDeterministicNAT, cls).setUpConstants()
1327 cls.vpp_cmdline.extend(["snat", "{", "deterministic", "}"])
1330 def setUpClass(cls):
1331 super(TestDeterministicNAT, cls).setUpClass()
1334 cls.tcp_port_in = 6303
1335 cls.tcp_external_port = 6303
1336 cls.udp_port_in = 6304
1337 cls.udp_external_port = 6304
1338 cls.icmp_id_in = 6305
1339 cls.snat_addr = '10.0.0.3'
1341 cls.create_pg_interfaces(range(3))
1342 cls.interfaces = list(cls.pg_interfaces)
1344 for i in cls.interfaces:
1349 cls.pg0.generate_remote_hosts(2)
1350 cls.pg0.configure_ipv4_neighbors()
1353 super(TestDeterministicNAT, cls).tearDownClass()
1356 def create_stream_in(self, in_if, out_if, ttl=64):
1358 Create packet stream for inside network
1360 :param in_if: Inside interface
1361 :param out_if: Outside interface
1362 :param ttl: TTL of generated packets
1366 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
1367 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
1368 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port))
1372 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
1373 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
1374 UDP(sport=self.udp_port_in, dport=self.udp_external_port))
1378 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
1379 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
1380 ICMP(id=self.icmp_id_in, type='echo-request'))
1385 def create_stream_out(self, out_if, dst_ip=None, ttl=64):
1387 Create packet stream for outside network
1389 :param out_if: Outside interface
1390 :param dst_ip: Destination IP address (Default use global SNAT address)
1391 :param ttl: TTL of generated packets
1394 dst_ip = self.snat_addr
1397 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
1398 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
1399 TCP(dport=self.tcp_port_out, sport=self.tcp_external_port))
1403 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
1404 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
1405 UDP(dport=self.udp_port_out, sport=self.udp_external_port))
1409 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
1410 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
1411 ICMP(id=self.icmp_external_id, type='echo-reply'))
1416 def verify_capture_out(self, capture, nat_ip=None, packet_num=3):
1418 Verify captured packets on outside network
1420 :param capture: Captured packets
1421 :param nat_ip: Translated IP address (Default use global SNAT address)
1422 :param same_port: Sorce port number is not translated (Default False)
1423 :param packet_num: Expected number of packets (Default 3)
1426 nat_ip = self.snat_addr
1427 self.assertEqual(packet_num, len(capture))
1428 for packet in capture:
1430 self.assertEqual(packet[IP].src, nat_ip)
1431 if packet.haslayer(TCP):
1432 self.tcp_port_out = packet[TCP].sport
1433 elif packet.haslayer(UDP):
1434 self.udp_port_out = packet[UDP].sport
1436 self.icmp_external_id = packet[ICMP].id
1438 self.logger.error(ppp("Unexpected or invalid packet "
1439 "(outside network):", packet))
1442 def initiate_tcp_session(self, in_if, out_if):
1444 Initiates TCP session
1446 :param in_if: Inside interface
1447 :param out_if: Outside interface
1450 # SYN packet in->out
1451 p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
1452 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
1453 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
1456 self.pg_enable_capture(self.pg_interfaces)
1458 capture = out_if.get_capture(1)
1460 self.tcp_port_out = p[TCP].sport
1462 # SYN + ACK packet out->in
1463 p = (Ether(src=out_if.remote_mac, dst=out_if.local_mac) /
1464 IP(src=out_if.remote_ip4, dst=self.snat_addr) /
1465 TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
1467 out_if.add_stream(p)
1468 self.pg_enable_capture(self.pg_interfaces)
1470 in_if.get_capture(1)
1472 # ACK packet in->out
1473 p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
1474 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
1475 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
1478 self.pg_enable_capture(self.pg_interfaces)
1480 out_if.get_capture(1)
1483 self.logger.error("TCP 3 way handshake failed")
1486 def verify_ipfix_max_entries_per_user(self, data):
1488 Verify IPFIX maximum entries per user exceeded event
1490 :param data: Decoded IPFIX data records
1492 self.assertEqual(1, len(data))
1495 self.assertEqual(ord(record[230]), 13)
1496 # natQuotaExceededEvent
1497 self.assertEqual('\x03\x00\x00\x00', record[466])
1499 self.assertEqual(self.pg0.remote_ip4n, record[8])
1501 def test_deterministic_mode(self):
1502 """ S-NAT run deterministic mode """
1503 in_addr = '172.16.255.0'
1504 out_addr = '172.17.255.50'
1505 in_addr_t = '172.16.255.20'
1506 in_addr_n = socket.inet_aton(in_addr)
1507 out_addr_n = socket.inet_aton(out_addr)
1508 in_addr_t_n = socket.inet_aton(in_addr_t)
1512 snat_config = self.vapi.snat_show_config()
1513 self.assertEqual(1, snat_config.deterministic)
1515 self.vapi.snat_add_det_map(in_addr_n, in_plen, out_addr_n, out_plen)
1517 rep1 = self.vapi.snat_det_forward(in_addr_t_n)
1518 self.assertEqual(rep1.out_addr[:4], out_addr_n)
1519 rep2 = self.vapi.snat_det_reverse(out_addr_n, rep1.out_port_hi)
1520 self.assertEqual(rep2.in_addr[:4], in_addr_t_n)
1522 deterministic_mappings = self.vapi.snat_det_map_dump()
1523 self.assertEqual(len(deterministic_mappings), 1)
1524 dsm = deterministic_mappings[0]
1525 self.assertEqual(in_addr_n, dsm.in_addr[:4])
1526 self.assertEqual(in_plen, dsm.in_plen)
1527 self.assertEqual(out_addr_n, dsm.out_addr[:4])
1528 self.assertEqual(out_plen, dsm.out_plen)
1531 deterministic_mappings = self.vapi.snat_det_map_dump()
1532 self.assertEqual(len(deterministic_mappings), 0)
1534 def test_set_timeouts(self):
1535 """ Set deterministic NAT timeouts """
1536 timeouts_before = self.vapi.snat_det_get_timeouts()
1538 self.vapi.snat_det_set_timeouts(timeouts_before.udp + 10,
1539 timeouts_before.tcp_established + 10,
1540 timeouts_before.tcp_transitory + 10,
1541 timeouts_before.icmp + 10)
1543 timeouts_after = self.vapi.snat_det_get_timeouts()
1545 self.assertNotEqual(timeouts_before.udp, timeouts_after.udp)
1546 self.assertNotEqual(timeouts_before.icmp, timeouts_after.icmp)
1547 self.assertNotEqual(timeouts_before.tcp_established,
1548 timeouts_after.tcp_established)
1549 self.assertNotEqual(timeouts_before.tcp_transitory,
1550 timeouts_after.tcp_transitory)
1552 def test_det_in(self):
1553 """ CGNAT translation test (TCP, UDP, ICMP) """
1555 nat_ip = "10.0.0.10"
1557 self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
1559 socket.inet_aton(nat_ip),
1561 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1562 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1566 pkts = self.create_stream_in(self.pg0, self.pg1)
1567 self.pg0.add_stream(pkts)
1568 self.pg_enable_capture(self.pg_interfaces)
1570 capture = self.pg1.get_capture(len(pkts))
1571 self.verify_capture_out(capture, nat_ip)
1574 pkts = self.create_stream_out(self.pg1, nat_ip)
1575 self.pg1.add_stream(pkts)
1576 self.pg_enable_capture(self.pg_interfaces)
1578 capture = self.pg0.get_capture(len(pkts))
1579 self.verify_capture_in(capture, self.pg0)
1582 sessions = self.vapi.snat_det_session_dump(self.pg0.remote_ip4n)
1583 self.assertEqual(len(sessions), 3)
1587 self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
1588 self.assertEqual(s.in_port, self.tcp_port_in)
1589 self.assertEqual(s.out_port, self.tcp_port_out)
1590 self.assertEqual(s.ext_port, self.tcp_external_port)
1594 self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
1595 self.assertEqual(s.in_port, self.udp_port_in)
1596 self.assertEqual(s.out_port, self.udp_port_out)
1597 self.assertEqual(s.ext_port, self.udp_external_port)
1601 self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
1602 self.assertEqual(s.in_port, self.icmp_id_in)
1603 self.assertEqual(s.out_port, self.icmp_external_id)
1605 def test_multiple_users(self):
1606 """ CGNAT multiple users """
1608 nat_ip = "10.0.0.10"
1610 external_port = 6303
1612 host0 = self.pg0.remote_hosts[0]
1613 host1 = self.pg0.remote_hosts[1]
1615 self.vapi.snat_add_det_map(host0.ip4n,
1617 socket.inet_aton(nat_ip),
1619 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1620 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1624 p = (Ether(src=host0.mac, dst=self.pg0.local_mac) /
1625 IP(src=host0.ip4, dst=self.pg1.remote_ip4) /
1626 TCP(sport=port_in, dport=external_port))
1627 self.pg0.add_stream(p)
1628 self.pg_enable_capture(self.pg_interfaces)
1630 capture = self.pg1.get_capture(1)
1635 self.assertEqual(ip.src, nat_ip)
1636 self.assertEqual(ip.dst, self.pg1.remote_ip4)
1637 self.assertEqual(tcp.dport, external_port)
1638 port_out0 = tcp.sport
1640 self.logger.error(ppp("Unexpected or invalid packet:", p))
1644 p = (Ether(src=host1.mac, dst=self.pg0.local_mac) /
1645 IP(src=host1.ip4, dst=self.pg1.remote_ip4) /
1646 TCP(sport=port_in, dport=external_port))
1647 self.pg0.add_stream(p)
1648 self.pg_enable_capture(self.pg_interfaces)
1650 capture = self.pg1.get_capture(1)
1655 self.assertEqual(ip.src, nat_ip)
1656 self.assertEqual(ip.dst, self.pg1.remote_ip4)
1657 self.assertEqual(tcp.dport, external_port)
1658 port_out1 = tcp.sport
1660 self.logger.error(ppp("Unexpected or invalid packet:", p))
1663 dms = self.vapi.snat_det_map_dump()
1664 self.assertEqual(1, len(dms))
1665 self.assertEqual(2, dms[0].ses_num)
1668 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1669 IP(src=self.pg1.remote_ip4, dst=nat_ip) /
1670 TCP(sport=external_port, dport=port_out0))
1671 self.pg1.add_stream(p)
1672 self.pg_enable_capture(self.pg_interfaces)
1674 capture = self.pg0.get_capture(1)
1679 self.assertEqual(ip.src, self.pg1.remote_ip4)
1680 self.assertEqual(ip.dst, host0.ip4)
1681 self.assertEqual(tcp.dport, port_in)
1682 self.assertEqual(tcp.sport, external_port)
1684 self.logger.error(ppp("Unexpected or invalid packet:", p))
1688 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1689 IP(src=self.pg1.remote_ip4, dst=nat_ip) /
1690 TCP(sport=external_port, dport=port_out1))
1691 self.pg1.add_stream(p)
1692 self.pg_enable_capture(self.pg_interfaces)
1694 capture = self.pg0.get_capture(1)
1699 self.assertEqual(ip.src, self.pg1.remote_ip4)
1700 self.assertEqual(ip.dst, host1.ip4)
1701 self.assertEqual(tcp.dport, port_in)
1702 self.assertEqual(tcp.sport, external_port)
1704 self.logger.error(ppp("Unexpected or invalid packet", p))
1707 # session close api test
1708 self.vapi.snat_det_close_session_out(socket.inet_aton(nat_ip),
1710 self.pg1.remote_ip4n,
1712 dms = self.vapi.snat_det_map_dump()
1713 self.assertEqual(dms[0].ses_num, 1)
1715 self.vapi.snat_det_close_session_in(host0.ip4n,
1717 self.pg1.remote_ip4n,
1719 dms = self.vapi.snat_det_map_dump()
1720 self.assertEqual(dms[0].ses_num, 0)
1722 def test_tcp_session_close_detection_in(self):
1723 """ CGNAT TCP session close initiated from inside network """
1724 self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
1726 socket.inet_aton(self.snat_addr),
1728 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1729 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1732 self.initiate_tcp_session(self.pg0, self.pg1)
1734 # close the session from inside
1736 # FIN packet in -> out
1737 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1738 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1739 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
1741 self.pg0.add_stream(p)
1742 self.pg_enable_capture(self.pg_interfaces)
1744 self.pg1.get_capture(1)
1748 # ACK packet out -> in
1749 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1750 IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
1751 TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
1755 # FIN packet out -> in
1756 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1757 IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
1758 TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
1762 self.pg1.add_stream(pkts)
1763 self.pg_enable_capture(self.pg_interfaces)
1765 self.pg0.get_capture(2)
1767 # ACK packet in -> out
1768 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1769 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1770 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
1772 self.pg0.add_stream(p)
1773 self.pg_enable_capture(self.pg_interfaces)
1775 self.pg1.get_capture(1)
1777 # Check if snat closed the session
1778 dms = self.vapi.snat_det_map_dump()
1779 self.assertEqual(0, dms[0].ses_num)
1781 self.logger.error("TCP session termination failed")
1784 def test_tcp_session_close_detection_out(self):
1785 """ CGNAT TCP session close initiated from outside network """
1786 self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
1788 socket.inet_aton(self.snat_addr),
1790 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1791 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1794 self.initiate_tcp_session(self.pg0, self.pg1)
1796 # close the session from outside
1798 # FIN packet out -> in
1799 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1800 IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
1801 TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
1803 self.pg1.add_stream(p)
1804 self.pg_enable_capture(self.pg_interfaces)
1806 self.pg0.get_capture(1)
1810 # ACK packet in -> out
1811 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1812 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1813 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
1817 # ACK packet in -> out
1818 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1819 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1820 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
1824 self.pg0.add_stream(pkts)
1825 self.pg_enable_capture(self.pg_interfaces)
1827 self.pg1.get_capture(2)
1829 # ACK packet out -> in
1830 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1831 IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
1832 TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
1834 self.pg1.add_stream(p)
1835 self.pg_enable_capture(self.pg_interfaces)
1837 self.pg0.get_capture(1)
1839 # Check if snat closed the session
1840 dms = self.vapi.snat_det_map_dump()
1841 self.assertEqual(0, dms[0].ses_num)
1843 self.logger.error("TCP session termination failed")
1846 @unittest.skipUnless(running_extended_tests(), "part of extended tests")
1847 def test_session_timeout(self):
1848 """ CGNAT session timeouts """
1849 self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
1851 socket.inet_aton(self.snat_addr),
1853 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1854 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1857 self.initiate_tcp_session(self.pg0, self.pg1)
1858 self.vapi.snat_det_set_timeouts(5, 5, 5, 5)
1859 pkts = self.create_stream_in(self.pg0, self.pg1)
1860 self.pg0.add_stream(pkts)
1861 self.pg_enable_capture(self.pg_interfaces)
1863 capture = self.pg1.get_capture(len(pkts))
1866 dms = self.vapi.snat_det_map_dump()
1867 self.assertEqual(0, dms[0].ses_num)
1869 def test_session_limit_per_user(self):
1870 """ CGNAT maximum 1000 sessions per user should be created """
1871 self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
1873 socket.inet_aton(self.snat_addr),
1875 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1876 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1878 self.vapi.set_ipfix_exporter(collector_address=self.pg2.remote_ip4n,
1879 src_address=self.pg2.local_ip4n,
1881 template_interval=10)
1882 self.vapi.snat_ipfix()
1885 for port in range(1025, 2025):
1886 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1887 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1888 UDP(sport=port, dport=port))
1891 self.pg0.add_stream(pkts)
1892 self.pg_enable_capture(self.pg_interfaces)
1894 capture = self.pg1.get_capture(len(pkts))
1896 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1897 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1898 UDP(sport=3001, dport=3002))
1899 self.pg0.add_stream(p)
1900 self.pg_enable_capture(self.pg_interfaces)
1902 capture = self.pg1.assert_nothing_captured()
1904 # verify ICMP error packet
1905 capture = self.pg0.get_capture(1)
1907 self.assertTrue(p.haslayer(ICMP))
1909 self.assertEqual(icmp.type, 3)
1910 self.assertEqual(icmp.code, 1)
1911 self.assertTrue(icmp.haslayer(IPerror))
1912 inner_ip = icmp[IPerror]
1913 self.assertEqual(inner_ip[UDPerror].sport, 3001)
1914 self.assertEqual(inner_ip[UDPerror].dport, 3002)
1916 dms = self.vapi.snat_det_map_dump()
1918 self.assertEqual(1000, dms[0].ses_num)
1920 # verify IPFIX logging
1921 self.vapi.cli("ipfix flush") # FIXME this should be an API call
1922 capture = self.pg2.get_capture(2)
1923 ipfix = IPFIXDecoder()
1924 # first load template
1926 self.assertTrue(p.haslayer(IPFIX))
1927 if p.haslayer(Template):
1928 ipfix.add_template(p.getlayer(Template))
1929 # verify events in data set
1931 if p.haslayer(Data):
1932 data = ipfix.decode_data_set(p.getlayer(Set))
1933 self.verify_ipfix_max_entries_per_user(data)
1935 def clear_snat(self):
1937 Clear SNAT configuration.
1939 self.vapi.snat_ipfix(enable=0)
1940 self.vapi.snat_det_set_timeouts()
1941 deterministic_mappings = self.vapi.snat_det_map_dump()
1942 for dsm in deterministic_mappings:
1943 self.vapi.snat_add_det_map(dsm.in_addr,
1949 interfaces = self.vapi.snat_interface_dump()
1950 for intf in interfaces:
1951 self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
1956 super(TestDeterministicNAT, self).tearDown()
1957 if not self.vpp_dead:
1958 self.logger.info(self.vapi.cli("show snat detail"))
1961 if __name__ == '__main__':
1962 unittest.main(testRunner=VppTestRunner)