7 from framework import VppTestCase, VppTestRunner
8 from scapy.layers.inet import IP, TCP, UDP, ICMP
9 from scapy.layers.l2 import Ether, ARP
10 from scapy.data import IP_PROTOS
12 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
15 class TestSNAT(VppTestCase):
16 """ SNAT Test Cases """
20 super(TestSNAT, cls).setUpClass()
23 cls.tcp_port_in = 6303
24 cls.tcp_port_out = 6303
25 cls.udp_port_in = 6304
26 cls.udp_port_out = 6304
28 cls.icmp_id_out = 6305
29 cls.snat_addr = '10.0.0.3'
31 cls.create_pg_interfaces(range(8))
32 cls.interfaces = list(cls.pg_interfaces[0:4])
34 for i in cls.interfaces:
39 cls.pg0.generate_remote_hosts(2)
40 cls.pg0.configure_ipv4_neighbors()
42 cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
44 cls.pg4._local_ip4 = "172.16.255.1"
45 cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
46 cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
47 cls.pg4.set_table_ip4(10)
48 cls.pg5._local_ip4 = "172.16.255.3"
49 cls.pg5._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
50 cls.pg5._remote_hosts[0]._ip4 = "172.16.255.4"
51 cls.pg5.set_table_ip4(10)
52 cls.pg6._local_ip4 = "172.16.255.1"
53 cls.pg6._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
54 cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
55 cls.pg6.set_table_ip4(20)
56 for i in cls.overlapping_interfaces:
64 super(TestSNAT, cls).tearDownClass()
67 def create_stream_in(self, in_if, out_if):
69 Create packet stream for inside network
71 :param in_if: Inside interface
72 :param out_if: Outside interface
76 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
77 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
78 TCP(sport=self.tcp_port_in))
82 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
83 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
84 UDP(sport=self.udp_port_in))
88 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
89 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
90 ICMP(id=self.icmp_id_in, type='echo-request'))
95 def create_stream_out(self, out_if, dst_ip=None):
97 Create packet stream for outside network
99 :param out_if: Outside interface
100 :param dst_ip: Destination IP address (Default use global SNAT address)
103 dst_ip = self.snat_addr
106 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
107 IP(src=out_if.remote_ip4, dst=dst_ip) /
108 TCP(dport=self.tcp_port_out))
112 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
113 IP(src=out_if.remote_ip4, dst=dst_ip) /
114 UDP(dport=self.udp_port_out))
118 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
119 IP(src=out_if.remote_ip4, dst=dst_ip) /
120 ICMP(id=self.icmp_id_out, type='echo-reply'))
125 def verify_capture_out(self, capture, nat_ip=None, same_port=False,
128 Verify captured packets on outside network
130 :param capture: Captured packets
131 :param nat_ip: Translated IP address (Default use global SNAT address)
132 :param same_port: Sorce port number is not translated (Default False)
133 :param packet_num: Expected number of packets (Default 3)
136 nat_ip = self.snat_addr
137 self.assertEqual(packet_num, len(capture))
138 for packet in capture:
140 self.assertEqual(packet[IP].src, nat_ip)
141 if packet.haslayer(TCP):
143 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
146 packet[TCP].sport, self.tcp_port_in)
147 self.tcp_port_out = packet[TCP].sport
148 elif packet.haslayer(UDP):
150 self.assertEqual(packet[UDP].sport, self.udp_port_in)
153 packet[UDP].sport, self.udp_port_in)
154 self.udp_port_out = packet[UDP].sport
157 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
159 self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
160 self.icmp_id_out = packet[ICMP].id
162 self.logger.error(ppp("Unexpected or invalid packet "
163 "(outside network):", packet))
166 def verify_capture_in(self, capture, in_if, packet_num=3):
168 Verify captured packets on inside network
170 :param capture: Captured packets
171 :param in_if: Inside interface
172 :param packet_num: Expected number of packets (Default 3)
174 self.assertEqual(packet_num, len(capture))
175 for packet in capture:
177 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
178 if packet.haslayer(TCP):
179 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
180 elif packet.haslayer(UDP):
181 self.assertEqual(packet[UDP].dport, self.udp_port_in)
183 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
185 self.logger.error(ppp("Unexpected or invalid packet "
186 "(inside network):", packet))
189 def verify_capture_no_translation(self, capture, ingress_if, egress_if):
191 Verify captured packet that don't have to be translated
193 :param capture: Captured packets
194 :param ingress_if: Ingress interface
195 :param egress_if: Egress interface
197 for packet in capture:
199 self.assertEqual(packet[IP].src, ingress_if.remote_ip4)
200 self.assertEqual(packet[IP].dst, egress_if.remote_ip4)
201 if packet.haslayer(TCP):
202 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
203 elif packet.haslayer(UDP):
204 self.assertEqual(packet[UDP].sport, self.udp_port_in)
206 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
208 self.logger.error(ppp("Unexpected or invalid packet "
209 "(inside network):", packet))
212 def verify_ipfix_nat44_ses(self, data):
214 Verify IPFIX NAT44 session create/delete event
216 :param data: Decoded IPFIX data records
218 nat44_ses_create_num = 0
219 nat44_ses_delete_num = 0
220 self.assertEqual(6, len(data))
223 self.assertIn(ord(record[230]), [4, 5])
224 if ord(record[230]) == 4:
225 nat44_ses_create_num += 1
227 nat44_ses_delete_num += 1
229 self.assertEqual(self.pg0.remote_ip4n, record[8])
230 # postNATSourceIPv4Address
231 self.assertEqual(socket.inet_pton(socket.AF_INET, self.snat_addr),
234 self.assertEqual(struct.pack("!I", 0), record[234])
235 # protocolIdentifier/sourceTransportPort/postNAPTSourceTransportPort
236 if IP_PROTOS.icmp == ord(record[4]):
237 self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
238 self.assertEqual(struct.pack("!H", self.icmp_id_out),
240 elif IP_PROTOS.tcp == ord(record[4]):
241 self.assertEqual(struct.pack("!H", self.tcp_port_in),
243 self.assertEqual(struct.pack("!H", self.tcp_port_out),
245 elif IP_PROTOS.udp == ord(record[4]):
246 self.assertEqual(struct.pack("!H", self.udp_port_in),
248 self.assertEqual(struct.pack("!H", self.udp_port_out),
251 self.fail("Invalid protocol")
252 self.assertEqual(3, nat44_ses_create_num)
253 self.assertEqual(3, nat44_ses_delete_num)
255 def verify_ipfix_addr_exhausted(self, data):
257 Verify IPFIX NAT addresses event
259 :param data: Decoded IPFIX data records
261 self.assertEqual(1, len(data))
264 self.assertEqual(ord(record[230]), 3)
266 self.assertEqual(struct.pack("!I", 0), record[283])
268 def clear_snat(self):
270 Clear SNAT configuration.
272 if self.pg7.has_ip4_config:
273 self.pg7.unconfig_ip4()
275 interfaces = self.vapi.snat_interface_addr_dump()
276 for intf in interfaces:
277 self.vapi.snat_add_interface_addr(intf.sw_if_index, is_add=0)
279 self.vapi.snat_ipfix(enable=0)
281 interfaces = self.vapi.snat_interface_dump()
282 for intf in interfaces:
283 self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
287 static_mappings = self.vapi.snat_static_mapping_dump()
288 for sm in static_mappings:
289 self.vapi.snat_add_static_mapping(sm.local_ip_address,
290 sm.external_ip_address,
291 local_port=sm.local_port,
292 external_port=sm.external_port,
293 addr_only=sm.addr_only,
295 protocol=sm.protocol,
298 adresses = self.vapi.snat_address_dump()
299 for addr in adresses:
300 self.vapi.snat_add_address_range(addr.ip_address,
304 def snat_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
305 local_port=0, external_port=0, vrf_id=0,
306 is_add=1, external_sw_if_index=0xFFFFFFFF,
309 Add/delete S-NAT static mapping
311 :param local_ip: Local IP address
312 :param external_ip: External IP address
313 :param local_port: Local port number (Optional)
314 :param external_port: External port number (Optional)
315 :param vrf_id: VRF ID (Default 0)
316 :param is_add: 1 if add, 0 if delete (Default add)
317 :param external_sw_if_index: External interface instead of IP address
318 :param proto: IP protocol (Mandatory if port specified)
321 if local_port and external_port:
323 l_ip = socket.inet_pton(socket.AF_INET, local_ip)
324 e_ip = socket.inet_pton(socket.AF_INET, external_ip)
325 self.vapi.snat_add_static_mapping(
328 external_sw_if_index,
336 def snat_add_address(self, ip, is_add=1):
338 Add/delete S-NAT address
340 :param ip: IP address
341 :param is_add: 1 if add, 0 if delete (Default add)
343 snat_addr = socket.inet_pton(socket.AF_INET, ip)
344 self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add)
346 def test_dynamic(self):
347 """ SNAT dynamic translation test """
349 self.snat_add_address(self.snat_addr)
350 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
351 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
355 pkts = self.create_stream_in(self.pg0, self.pg1)
356 self.pg0.add_stream(pkts)
357 self.pg_enable_capture(self.pg_interfaces)
359 capture = self.pg1.get_capture(len(pkts))
360 self.verify_capture_out(capture)
363 pkts = self.create_stream_out(self.pg1)
364 self.pg1.add_stream(pkts)
365 self.pg_enable_capture(self.pg_interfaces)
367 capture = self.pg0.get_capture(len(pkts))
368 self.verify_capture_in(capture, self.pg0)
370 def test_static_in(self):
371 """ SNAT 1:1 NAT initialized from inside network """
374 self.tcp_port_out = 6303
375 self.udp_port_out = 6304
376 self.icmp_id_out = 6305
378 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
379 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
380 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
384 pkts = self.create_stream_in(self.pg0, self.pg1)
385 self.pg0.add_stream(pkts)
386 self.pg_enable_capture(self.pg_interfaces)
388 capture = self.pg1.get_capture(len(pkts))
389 self.verify_capture_out(capture, nat_ip, True)
392 pkts = self.create_stream_out(self.pg1, nat_ip)
393 self.pg1.add_stream(pkts)
394 self.pg_enable_capture(self.pg_interfaces)
396 capture = self.pg0.get_capture(len(pkts))
397 self.verify_capture_in(capture, self.pg0)
399 def test_static_out(self):
400 """ SNAT 1:1 NAT initialized from outside network """
403 self.tcp_port_out = 6303
404 self.udp_port_out = 6304
405 self.icmp_id_out = 6305
407 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
408 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
409 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
413 pkts = self.create_stream_out(self.pg1, nat_ip)
414 self.pg1.add_stream(pkts)
415 self.pg_enable_capture(self.pg_interfaces)
417 capture = self.pg0.get_capture(len(pkts))
418 self.verify_capture_in(capture, self.pg0)
421 pkts = self.create_stream_in(self.pg0, self.pg1)
422 self.pg0.add_stream(pkts)
423 self.pg_enable_capture(self.pg_interfaces)
425 capture = self.pg1.get_capture(len(pkts))
426 self.verify_capture_out(capture, nat_ip, True)
428 def test_static_with_port_in(self):
429 """ SNAT 1:1 NAT with port initialized from inside network """
431 self.tcp_port_out = 3606
432 self.udp_port_out = 3607
433 self.icmp_id_out = 3608
435 self.snat_add_address(self.snat_addr)
436 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
437 self.tcp_port_in, self.tcp_port_out,
439 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
440 self.udp_port_in, self.udp_port_out,
442 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
443 self.icmp_id_in, self.icmp_id_out,
444 proto=IP_PROTOS.icmp)
445 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
446 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
450 pkts = self.create_stream_in(self.pg0, self.pg1)
451 self.pg0.add_stream(pkts)
452 self.pg_enable_capture(self.pg_interfaces)
454 capture = self.pg1.get_capture(len(pkts))
455 self.verify_capture_out(capture)
458 pkts = self.create_stream_out(self.pg1)
459 self.pg1.add_stream(pkts)
460 self.pg_enable_capture(self.pg_interfaces)
462 capture = self.pg0.get_capture(len(pkts))
463 self.verify_capture_in(capture, self.pg0)
465 def test_static_with_port_out(self):
466 """ SNAT 1:1 NAT with port initialized from outside network """
468 self.tcp_port_out = 30606
469 self.udp_port_out = 30607
470 self.icmp_id_out = 30608
472 self.snat_add_address(self.snat_addr)
473 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
474 self.tcp_port_in, self.tcp_port_out,
476 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
477 self.udp_port_in, self.udp_port_out,
479 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
480 self.icmp_id_in, self.icmp_id_out,
481 proto=IP_PROTOS.icmp)
482 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
483 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
487 pkts = self.create_stream_out(self.pg1)
488 self.pg1.add_stream(pkts)
489 self.pg_enable_capture(self.pg_interfaces)
491 capture = self.pg0.get_capture(len(pkts))
492 self.verify_capture_in(capture, self.pg0)
495 pkts = self.create_stream_in(self.pg0, self.pg1)
496 self.pg0.add_stream(pkts)
497 self.pg_enable_capture(self.pg_interfaces)
499 capture = self.pg1.get_capture(len(pkts))
500 self.verify_capture_out(capture)
502 def test_static_vrf_aware(self):
503 """ SNAT 1:1 NAT VRF awareness """
505 nat_ip1 = "10.0.0.30"
506 nat_ip2 = "10.0.0.40"
507 self.tcp_port_out = 6303
508 self.udp_port_out = 6304
509 self.icmp_id_out = 6305
511 self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
513 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
515 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
517 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
518 self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
520 # inside interface VRF match SNAT static mapping VRF
521 pkts = self.create_stream_in(self.pg4, self.pg3)
522 self.pg4.add_stream(pkts)
523 self.pg_enable_capture(self.pg_interfaces)
525 capture = self.pg3.get_capture(len(pkts))
526 self.verify_capture_out(capture, nat_ip1, True)
528 # inside interface VRF don't match SNAT static mapping VRF (packets
530 pkts = self.create_stream_in(self.pg0, self.pg3)
531 self.pg0.add_stream(pkts)
532 self.pg_enable_capture(self.pg_interfaces)
534 self.pg3.assert_nothing_captured()
536 def test_multiple_inside_interfaces(self):
537 """ SNAT multiple inside interfaces (non-overlapping address space) """
539 self.snat_add_address(self.snat_addr)
540 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
541 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
542 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
545 # between two S-NAT inside interfaces (no translation)
546 pkts = self.create_stream_in(self.pg0, self.pg1)
547 self.pg0.add_stream(pkts)
548 self.pg_enable_capture(self.pg_interfaces)
550 capture = self.pg1.get_capture(len(pkts))
551 self.verify_capture_no_translation(capture, self.pg0, self.pg1)
553 # from S-NAT inside to interface without S-NAT feature (no translation)
554 pkts = self.create_stream_in(self.pg0, self.pg2)
555 self.pg0.add_stream(pkts)
556 self.pg_enable_capture(self.pg_interfaces)
558 capture = self.pg2.get_capture(len(pkts))
559 self.verify_capture_no_translation(capture, self.pg0, self.pg2)
561 # in2out 1st interface
562 pkts = self.create_stream_in(self.pg0, self.pg3)
563 self.pg0.add_stream(pkts)
564 self.pg_enable_capture(self.pg_interfaces)
566 capture = self.pg3.get_capture(len(pkts))
567 self.verify_capture_out(capture)
569 # out2in 1st interface
570 pkts = self.create_stream_out(self.pg3)
571 self.pg3.add_stream(pkts)
572 self.pg_enable_capture(self.pg_interfaces)
574 capture = self.pg0.get_capture(len(pkts))
575 self.verify_capture_in(capture, self.pg0)
577 # in2out 2nd interface
578 pkts = self.create_stream_in(self.pg1, self.pg3)
579 self.pg1.add_stream(pkts)
580 self.pg_enable_capture(self.pg_interfaces)
582 capture = self.pg3.get_capture(len(pkts))
583 self.verify_capture_out(capture)
585 # out2in 2nd interface
586 pkts = self.create_stream_out(self.pg3)
587 self.pg3.add_stream(pkts)
588 self.pg_enable_capture(self.pg_interfaces)
590 capture = self.pg1.get_capture(len(pkts))
591 self.verify_capture_in(capture, self.pg1)
593 def test_inside_overlapping_interfaces(self):
594 """ SNAT multiple inside interfaces with overlapping address space """
596 static_nat_ip = "10.0.0.10"
597 self.snat_add_address(self.snat_addr)
598 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
600 self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
601 self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
602 self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
603 self.snat_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
606 # between S-NAT inside interfaces with same VRF (no translation)
607 pkts = self.create_stream_in(self.pg4, self.pg5)
608 self.pg4.add_stream(pkts)
609 self.pg_enable_capture(self.pg_interfaces)
611 capture = self.pg5.get_capture(len(pkts))
612 self.verify_capture_no_translation(capture, self.pg4, self.pg5)
614 # between S-NAT inside interfaces with different VRF (hairpinning)
615 p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
616 IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
617 TCP(sport=1234, dport=5678))
618 self.pg4.add_stream(p)
619 self.pg_enable_capture(self.pg_interfaces)
621 capture = self.pg6.get_capture(1)
626 self.assertEqual(ip.src, self.snat_addr)
627 self.assertEqual(ip.dst, self.pg6.remote_ip4)
628 self.assertNotEqual(tcp.sport, 1234)
629 self.assertEqual(tcp.dport, 5678)
631 self.logger.error(ppp("Unexpected or invalid packet:", p))
634 # in2out 1st interface
635 pkts = self.create_stream_in(self.pg4, self.pg3)
636 self.pg4.add_stream(pkts)
637 self.pg_enable_capture(self.pg_interfaces)
639 capture = self.pg3.get_capture(len(pkts))
640 self.verify_capture_out(capture)
642 # out2in 1st interface
643 pkts = self.create_stream_out(self.pg3)
644 self.pg3.add_stream(pkts)
645 self.pg_enable_capture(self.pg_interfaces)
647 capture = self.pg4.get_capture(len(pkts))
648 self.verify_capture_in(capture, self.pg4)
650 # in2out 2nd interface
651 pkts = self.create_stream_in(self.pg5, self.pg3)
652 self.pg5.add_stream(pkts)
653 self.pg_enable_capture(self.pg_interfaces)
655 capture = self.pg3.get_capture(len(pkts))
656 self.verify_capture_out(capture)
658 # out2in 2nd interface
659 pkts = self.create_stream_out(self.pg3)
660 self.pg3.add_stream(pkts)
661 self.pg_enable_capture(self.pg_interfaces)
663 capture = self.pg5.get_capture(len(pkts))
664 self.verify_capture_in(capture, self.pg5)
666 # in2out 3rd interface
667 pkts = self.create_stream_in(self.pg6, self.pg3)
668 self.pg6.add_stream(pkts)
669 self.pg_enable_capture(self.pg_interfaces)
671 capture = self.pg3.get_capture(len(pkts))
672 self.verify_capture_out(capture, static_nat_ip, True)
674 # out2in 3rd interface
675 pkts = self.create_stream_out(self.pg3, static_nat_ip)
676 self.pg3.add_stream(pkts)
677 self.pg_enable_capture(self.pg_interfaces)
679 capture = self.pg6.get_capture(len(pkts))
680 self.verify_capture_in(capture, self.pg6)
682 def test_hairpinning(self):
683 """ SNAT hairpinning """
685 host = self.pg0.remote_hosts[0]
686 server = self.pg0.remote_hosts[1]
689 server_in_port = 5678
690 server_out_port = 8765
692 self.snat_add_address(self.snat_addr)
693 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
694 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
696 # add static mapping for server
697 self.snat_add_static_mapping(server.ip4, self.snat_addr,
698 server_in_port, server_out_port,
701 # send packet from host to server
702 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
703 IP(src=host.ip4, dst=self.snat_addr) /
704 TCP(sport=host_in_port, dport=server_out_port))
705 self.pg0.add_stream(p)
706 self.pg_enable_capture(self.pg_interfaces)
708 capture = self.pg0.get_capture(1)
713 self.assertEqual(ip.src, self.snat_addr)
714 self.assertEqual(ip.dst, server.ip4)
715 self.assertNotEqual(tcp.sport, host_in_port)
716 self.assertEqual(tcp.dport, server_in_port)
717 host_out_port = tcp.sport
719 self.logger.error(ppp("Unexpected or invalid packet:", p))
722 # send reply from server to host
723 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
724 IP(src=server.ip4, dst=self.snat_addr) /
725 TCP(sport=server_in_port, dport=host_out_port))
726 self.pg0.add_stream(p)
727 self.pg_enable_capture(self.pg_interfaces)
729 capture = self.pg0.get_capture(1)
734 self.assertEqual(ip.src, self.snat_addr)
735 self.assertEqual(ip.dst, host.ip4)
736 self.assertEqual(tcp.sport, server_out_port)
737 self.assertEqual(tcp.dport, host_in_port)
739 self.logger.error(ppp("Unexpected or invalid packet:"), p)
742 def test_max_translations_per_user(self):
743 """ MAX translations per user - recycle the least recently used """
745 self.snat_add_address(self.snat_addr)
746 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
747 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
750 # get maximum number of translations per user
751 snat_config = self.vapi.snat_show_config()
753 # send more than maximum number of translations per user packets
754 pkts_num = snat_config.max_translations_per_user + 5
756 for port in range(0, pkts_num):
757 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
758 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
759 TCP(sport=1025 + port))
761 self.pg0.add_stream(pkts)
762 self.pg_enable_capture(self.pg_interfaces)
765 # verify number of translated packet
766 self.pg1.get_capture(pkts_num)
768 def test_interface_addr(self):
769 """ Acquire SNAT addresses from interface """
770 self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
772 # no address in NAT pool
773 adresses = self.vapi.snat_address_dump()
774 self.assertEqual(0, len(adresses))
776 # configure interface address and check NAT address pool
777 self.pg7.config_ip4()
778 adresses = self.vapi.snat_address_dump()
779 self.assertEqual(1, len(adresses))
780 self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n)
782 # remove interface address and check NAT address pool
783 self.pg7.unconfig_ip4()
784 adresses = self.vapi.snat_address_dump()
785 self.assertEqual(0, len(adresses))
787 def test_interface_addr_static_mapping(self):
788 """ Static mapping with addresses from interface """
789 self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
790 self.snat_add_static_mapping('1.2.3.4',
791 external_sw_if_index=self.pg7.sw_if_index)
794 static_mappings = self.vapi.snat_static_mapping_dump()
795 self.assertEqual(0, len(static_mappings))
797 # configure interface address and check static mappings
798 self.pg7.config_ip4()
799 static_mappings = self.vapi.snat_static_mapping_dump()
800 self.assertEqual(1, len(static_mappings))
801 self.assertEqual(static_mappings[0].external_ip_address[0:4],
804 # remove interface address and check static mappings
805 self.pg7.unconfig_ip4()
806 static_mappings = self.vapi.snat_static_mapping_dump()
807 self.assertEqual(0, len(static_mappings))
809 def test_ipfix_nat44_sess(self):
810 """ S-NAT IPFIX logging NAT44 session created/delted """
811 self.snat_add_address(self.snat_addr)
812 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
813 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
815 self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
816 src_address=self.pg3.local_ip4n,
818 template_interval=10)
819 self.vapi.snat_ipfix()
821 pkts = self.create_stream_in(self.pg0, self.pg1)
822 self.pg0.add_stream(pkts)
823 self.pg_enable_capture(self.pg_interfaces)
825 capture = self.pg1.get_capture(len(pkts))
826 self.verify_capture_out(capture)
827 self.snat_add_address(self.snat_addr, is_add=0)
828 self.vapi.cli("ipfix flush") # FIXME this should be an API call
829 capture = self.pg3.get_capture(3)
830 ipfix = IPFIXDecoder()
831 # first load template
833 self.assertTrue(p.haslayer(IPFIX))
834 if p.haslayer(Template):
835 ipfix.add_template(p.getlayer(Template))
836 # verify events in data set
839 data = ipfix.decode_data_set(p.getlayer(Set))
840 self.verify_ipfix_nat44_ses(data)
842 def test_ipfix_addr_exhausted(self):
843 """ S-NAT IPFIX logging NAT addresses exhausted """
844 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
845 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
847 self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
848 src_address=self.pg3.local_ip4n,
850 template_interval=10)
851 self.vapi.snat_ipfix()
853 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
854 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
856 self.pg0.add_stream(p)
857 self.pg_enable_capture(self.pg_interfaces)
859 capture = self.pg1.get_capture(0)
860 self.vapi.cli("ipfix flush") # FIXME this should be an API call
861 capture = self.pg3.get_capture(3)
862 ipfix = IPFIXDecoder()
863 # first load template
865 self.assertTrue(p.haslayer(IPFIX))
866 if p.haslayer(Template):
867 ipfix.add_template(p.getlayer(Template))
868 # verify events in data set
871 data = ipfix.decode_data_set(p.getlayer(Set))
872 self.verify_ipfix_addr_exhausted(data)
874 def test_pool_addr_fib(self):
875 """ S-NAT add pool addresses to FIB """
876 static_addr = '10.0.0.10'
877 self.snat_add_address(self.snat_addr)
878 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
879 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
881 self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr)
884 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
885 ARP(op=ARP.who_has, pdst=self.snat_addr,
886 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
887 self.pg1.add_stream(p)
888 self.pg_enable_capture(self.pg_interfaces)
890 capture = self.pg1.get_capture(1)
891 self.assertTrue(capture[0].haslayer(ARP))
892 self.assertTrue(capture[0][ARP].op, ARP.is_at)
895 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
896 ARP(op=ARP.who_has, pdst=static_addr,
897 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
898 self.pg1.add_stream(p)
899 self.pg_enable_capture(self.pg_interfaces)
901 capture = self.pg1.get_capture(1)
902 self.assertTrue(capture[0].haslayer(ARP))
903 self.assertTrue(capture[0][ARP].op, ARP.is_at)
905 # send ARP to non-SNAT interface
906 p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
907 ARP(op=ARP.who_has, pdst=self.snat_addr,
908 psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
909 self.pg2.add_stream(p)
910 self.pg_enable_capture(self.pg_interfaces)
912 capture = self.pg1.get_capture(0)
914 # remove addresses and verify
915 self.snat_add_address(self.snat_addr, is_add=0)
916 self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr,
919 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
920 ARP(op=ARP.who_has, pdst=self.snat_addr,
921 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
922 self.pg1.add_stream(p)
923 self.pg_enable_capture(self.pg_interfaces)
925 capture = self.pg1.get_capture(0)
927 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
928 ARP(op=ARP.who_has, pdst=static_addr,
929 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
930 self.pg1.add_stream(p)
931 self.pg_enable_capture(self.pg_interfaces)
933 capture = self.pg1.get_capture(0)
936 super(TestSNAT, self).tearDown()
937 if not self.vpp_dead:
938 self.logger.info(self.vapi.cli("show snat verbose"))
942 if __name__ == '__main__':
943 unittest.main(testRunner=VppTestRunner)