7 from framework import VppTestCase, VppTestRunner
8 from scapy.layers.inet import IP, TCP, UDP, ICMP
9 from scapy.layers.l2 import Ether
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 for i in cls.overlapping_interfaces:
45 i._local_ip4 = "172.16.255.1"
46 i._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
47 i._remote_hosts[0]._ip4 = "172.16.255.2"
48 i.set_table_ip4(i.sw_if_index)
56 super(TestSNAT, cls).tearDownClass()
59 def create_stream_in(self, in_if, out_if):
61 Create packet stream for inside network
63 :param in_if: Inside interface
64 :param out_if: Outside interface
68 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
69 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
70 TCP(sport=self.tcp_port_in))
74 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
75 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
76 UDP(sport=self.udp_port_in))
80 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
81 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
82 ICMP(id=self.icmp_id_in, type='echo-request'))
87 def create_stream_out(self, out_if, dst_ip=None):
89 Create packet stream for outside network
91 :param out_if: Outside interface
92 :param dst_ip: Destination IP address (Default use global SNAT address)
95 dst_ip = self.snat_addr
98 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
99 IP(src=out_if.remote_ip4, dst=dst_ip) /
100 TCP(dport=self.tcp_port_out))
104 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
105 IP(src=out_if.remote_ip4, dst=dst_ip) /
106 UDP(dport=self.udp_port_out))
110 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
111 IP(src=out_if.remote_ip4, dst=dst_ip) /
112 ICMP(id=self.icmp_id_out, type='echo-reply'))
117 def verify_capture_out(self, capture, nat_ip=None, same_port=False,
120 Verify captured packets on outside network
122 :param capture: Captured packets
123 :param nat_ip: Translated IP address (Default use global SNAT address)
124 :param same_port: Sorce port number is not translated (Default False)
125 :param packet_num: Expected number of packets (Default 3)
128 nat_ip = self.snat_addr
129 self.assertEqual(packet_num, len(capture))
130 for packet in capture:
132 self.assertEqual(packet[IP].src, nat_ip)
133 if packet.haslayer(TCP):
135 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
138 packet[TCP].sport, self.tcp_port_in)
139 self.tcp_port_out = packet[TCP].sport
140 elif packet.haslayer(UDP):
142 self.assertEqual(packet[UDP].sport, self.udp_port_in)
145 packet[UDP].sport, self.udp_port_in)
146 self.udp_port_out = packet[UDP].sport
149 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
151 self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
152 self.icmp_id_out = packet[ICMP].id
154 self.logger.error(ppp("Unexpected or invalid packet "
155 "(outside network):", packet))
158 def verify_capture_in(self, capture, in_if, packet_num=3):
160 Verify captured packets on inside network
162 :param capture: Captured packets
163 :param in_if: Inside interface
164 :param packet_num: Expected number of packets (Default 3)
166 self.assertEqual(packet_num, len(capture))
167 for packet in capture:
169 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
170 if packet.haslayer(TCP):
171 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
172 elif packet.haslayer(UDP):
173 self.assertEqual(packet[UDP].dport, self.udp_port_in)
175 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
177 self.logger.error(ppp("Unexpected or invalid packet "
178 "(inside network):", packet))
181 def verify_ipfix_nat44_ses(self, data):
183 Verify IPFIX NAT44 session create/delete event
185 :param data: Decoded IPFIX data records
187 nat44_ses_create_num = 0
188 nat44_ses_delete_num = 0
189 self.assertEqual(6, len(data))
192 self.assertIn(ord(record[230]), [4, 5])
193 if ord(record[230]) == 4:
194 nat44_ses_create_num += 1
196 nat44_ses_delete_num += 1
198 self.assertEqual(self.pg0.remote_ip4n, record[8])
199 # postNATSourceIPv4Address
200 self.assertEqual(socket.inet_pton(socket.AF_INET, self.snat_addr),
203 self.assertEqual(struct.pack("!I", 0), record[234])
204 # protocolIdentifier/sourceTransportPort/postNAPTSourceTransportPort
205 if IP_PROTOS.icmp == ord(record[4]):
206 self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
207 self.assertEqual(struct.pack("!H", self.icmp_id_out),
209 elif IP_PROTOS.tcp == ord(record[4]):
210 self.assertEqual(struct.pack("!H", self.tcp_port_in),
212 self.assertEqual(struct.pack("!H", self.tcp_port_out),
214 elif IP_PROTOS.udp == ord(record[4]):
215 self.assertEqual(struct.pack("!H", self.udp_port_in),
217 self.assertEqual(struct.pack("!H", self.udp_port_out),
220 self.fail("Invalid protocol")
221 self.assertEqual(3, nat44_ses_create_num)
222 self.assertEqual(3, nat44_ses_delete_num)
224 def verify_ipfix_addr_exhausted(self, data):
226 Verify IPFIX NAT addresses event
228 :param data: Decoded IPFIX data records
230 self.assertEqual(1, len(data))
233 self.assertEqual(ord(record[230]), 3)
235 self.assertEqual(struct.pack("!I", 0), record[283])
237 def clear_snat(self):
239 Clear SNAT configuration.
241 interfaces = self.vapi.snat_interface_addr_dump()
242 for intf in interfaces:
243 self.vapi.snat_add_interface_addr(intf.sw_if_index, is_add=0)
245 self.vapi.snat_ipfix(enable=0)
247 interfaces = self.vapi.snat_interface_dump()
248 for intf in interfaces:
249 self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
253 static_mappings = self.vapi.snat_static_mapping_dump()
254 for sm in static_mappings:
255 self.vapi.snat_add_static_mapping(sm.local_ip_address,
256 sm.external_ip_address,
257 local_port=sm.local_port,
258 external_port=sm.external_port,
259 addr_only=sm.addr_only,
263 adresses = self.vapi.snat_address_dump()
264 for addr in adresses:
265 self.vapi.snat_add_address_range(addr.ip_address,
269 def snat_add_static_mapping(self, local_ip, external_ip, local_port=0,
270 external_port=0, vrf_id=0, is_add=1):
272 Add/delete S-NAT static mapping
274 :param local_ip: Local IP address
275 :param external_ip: External IP address
276 :param local_port: Local port number (Optional)
277 :param external_port: External port number (Optional)
278 :param vrf_id: VRF ID (Default 0)
279 :param is_add: 1 if add, 0 if delete (Default add)
282 if local_port and external_port:
284 l_ip = socket.inet_pton(socket.AF_INET, local_ip)
285 e_ip = socket.inet_pton(socket.AF_INET, external_ip)
286 self.vapi.snat_add_static_mapping(
295 def snat_add_address(self, ip, is_add=1):
297 Add/delete S-NAT address
299 :param ip: IP address
300 :param is_add: 1 if add, 0 if delete (Default add)
302 snat_addr = socket.inet_pton(socket.AF_INET, ip)
303 self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add)
305 def test_dynamic(self):
306 """ SNAT dynamic translation test """
308 self.snat_add_address(self.snat_addr)
309 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
310 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
314 pkts = self.create_stream_in(self.pg0, self.pg1)
315 self.pg0.add_stream(pkts)
316 self.pg_enable_capture(self.pg_interfaces)
318 capture = self.pg1.get_capture(len(pkts))
319 self.verify_capture_out(capture)
322 pkts = self.create_stream_out(self.pg1)
323 self.pg1.add_stream(pkts)
324 self.pg_enable_capture(self.pg_interfaces)
326 capture = self.pg0.get_capture(len(pkts))
327 self.verify_capture_in(capture, self.pg0)
329 def test_static_in(self):
330 """ SNAT 1:1 NAT initialized from inside network """
333 self.tcp_port_out = 6303
334 self.udp_port_out = 6304
335 self.icmp_id_out = 6305
337 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
338 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
339 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
343 pkts = self.create_stream_in(self.pg0, self.pg1)
344 self.pg0.add_stream(pkts)
345 self.pg_enable_capture(self.pg_interfaces)
347 capture = self.pg1.get_capture(len(pkts))
348 self.verify_capture_out(capture, nat_ip, True)
351 pkts = self.create_stream_out(self.pg1, nat_ip)
352 self.pg1.add_stream(pkts)
353 self.pg_enable_capture(self.pg_interfaces)
355 capture = self.pg0.get_capture(len(pkts))
356 self.verify_capture_in(capture, self.pg0)
358 def test_static_out(self):
359 """ SNAT 1:1 NAT initialized from outside network """
362 self.tcp_port_out = 6303
363 self.udp_port_out = 6304
364 self.icmp_id_out = 6305
366 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
367 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
368 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
372 pkts = self.create_stream_out(self.pg1, nat_ip)
373 self.pg1.add_stream(pkts)
374 self.pg_enable_capture(self.pg_interfaces)
376 capture = self.pg0.get_capture(len(pkts))
377 self.verify_capture_in(capture, self.pg0)
380 pkts = self.create_stream_in(self.pg0, self.pg1)
381 self.pg0.add_stream(pkts)
382 self.pg_enable_capture(self.pg_interfaces)
384 capture = self.pg1.get_capture(len(pkts))
385 self.verify_capture_out(capture, nat_ip, True)
387 def test_static_with_port_in(self):
388 """ SNAT 1:1 NAT with port initialized from inside network """
390 self.tcp_port_out = 3606
391 self.udp_port_out = 3607
392 self.icmp_id_out = 3608
394 self.snat_add_address(self.snat_addr)
395 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
396 self.tcp_port_in, self.tcp_port_out)
397 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
398 self.udp_port_in, self.udp_port_out)
399 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
400 self.icmp_id_in, self.icmp_id_out)
401 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
402 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
406 pkts = self.create_stream_in(self.pg0, self.pg1)
407 self.pg0.add_stream(pkts)
408 self.pg_enable_capture(self.pg_interfaces)
410 capture = self.pg1.get_capture(len(pkts))
411 self.verify_capture_out(capture)
414 pkts = self.create_stream_out(self.pg1)
415 self.pg1.add_stream(pkts)
416 self.pg_enable_capture(self.pg_interfaces)
418 capture = self.pg0.get_capture(len(pkts))
419 self.verify_capture_in(capture, self.pg0)
421 def test_static_with_port_out(self):
422 """ SNAT 1:1 NAT with port initialized from outside network """
424 self.tcp_port_out = 30606
425 self.udp_port_out = 30607
426 self.icmp_id_out = 30608
428 self.snat_add_address(self.snat_addr)
429 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
430 self.tcp_port_in, self.tcp_port_out)
431 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
432 self.udp_port_in, self.udp_port_out)
433 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
434 self.icmp_id_in, self.icmp_id_out)
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_out(self.pg1)
441 self.pg1.add_stream(pkts)
442 self.pg_enable_capture(self.pg_interfaces)
444 capture = self.pg0.get_capture(len(pkts))
445 self.verify_capture_in(capture, self.pg0)
448 pkts = self.create_stream_in(self.pg0, self.pg1)
449 self.pg0.add_stream(pkts)
450 self.pg_enable_capture(self.pg_interfaces)
452 capture = self.pg1.get_capture(len(pkts))
453 self.verify_capture_out(capture)
455 def test_static_vrf_aware(self):
456 """ SNAT 1:1 NAT VRF awareness """
458 nat_ip1 = "10.0.0.30"
459 nat_ip2 = "10.0.0.40"
460 self.tcp_port_out = 6303
461 self.udp_port_out = 6304
462 self.icmp_id_out = 6305
464 self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
465 vrf_id=self.pg4.sw_if_index)
466 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
467 vrf_id=self.pg4.sw_if_index)
468 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
470 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
471 self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
473 # inside interface VRF match SNAT static mapping VRF
474 pkts = self.create_stream_in(self.pg4, self.pg3)
475 self.pg4.add_stream(pkts)
476 self.pg_enable_capture(self.pg_interfaces)
478 capture = self.pg3.get_capture(len(pkts))
479 self.verify_capture_out(capture, nat_ip1, True)
481 # inside interface VRF don't match SNAT static mapping VRF (packets
483 pkts = self.create_stream_in(self.pg0, self.pg3)
484 self.pg0.add_stream(pkts)
485 self.pg_enable_capture(self.pg_interfaces)
487 self.pg3.assert_nothing_captured()
489 def test_multiple_inside_interfaces(self):
491 SNAT multiple inside interfaces with non-overlapping address space
494 self.snat_add_address(self.snat_addr)
495 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
496 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
497 self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index)
498 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
501 # in2out 1st interface
502 pkts = self.create_stream_in(self.pg0, self.pg3)
503 self.pg0.add_stream(pkts)
504 self.pg_enable_capture(self.pg_interfaces)
506 capture = self.pg3.get_capture(len(pkts))
507 self.verify_capture_out(capture)
509 # out2in 1st interface
510 pkts = self.create_stream_out(self.pg3)
511 self.pg3.add_stream(pkts)
512 self.pg_enable_capture(self.pg_interfaces)
514 capture = self.pg0.get_capture(len(pkts))
515 self.verify_capture_in(capture, self.pg0)
517 # in2out 2nd interface
518 pkts = self.create_stream_in(self.pg1, self.pg3)
519 self.pg1.add_stream(pkts)
520 self.pg_enable_capture(self.pg_interfaces)
522 capture = self.pg3.get_capture(len(pkts))
523 self.verify_capture_out(capture)
525 # out2in 2nd interface
526 pkts = self.create_stream_out(self.pg3)
527 self.pg3.add_stream(pkts)
528 self.pg_enable_capture(self.pg_interfaces)
530 capture = self.pg1.get_capture(len(pkts))
531 self.verify_capture_in(capture, self.pg1)
533 # in2out 3rd interface
534 pkts = self.create_stream_in(self.pg2, self.pg3)
535 self.pg2.add_stream(pkts)
536 self.pg_enable_capture(self.pg_interfaces)
538 capture = self.pg3.get_capture(len(pkts))
539 self.verify_capture_out(capture)
541 # out2in 3rd interface
542 pkts = self.create_stream_out(self.pg3)
543 self.pg3.add_stream(pkts)
544 self.pg_enable_capture(self.pg_interfaces)
546 capture = self.pg2.get_capture(len(pkts))
547 self.verify_capture_in(capture, self.pg2)
549 def test_inside_overlapping_interfaces(self):
550 """ SNAT multiple inside interfaces with overlapping address space """
552 self.snat_add_address(self.snat_addr)
553 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
555 self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
556 self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
557 self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
559 # in2out 1st interface
560 pkts = self.create_stream_in(self.pg4, self.pg3)
561 self.pg4.add_stream(pkts)
562 self.pg_enable_capture(self.pg_interfaces)
564 capture = self.pg3.get_capture(len(pkts))
565 self.verify_capture_out(capture)
567 # out2in 1st interface
568 pkts = self.create_stream_out(self.pg3)
569 self.pg3.add_stream(pkts)
570 self.pg_enable_capture(self.pg_interfaces)
572 capture = self.pg4.get_capture(len(pkts))
573 self.verify_capture_in(capture, self.pg4)
575 # in2out 2nd interface
576 pkts = self.create_stream_in(self.pg5, self.pg3)
577 self.pg5.add_stream(pkts)
578 self.pg_enable_capture(self.pg_interfaces)
580 capture = self.pg3.get_capture(len(pkts))
581 self.verify_capture_out(capture)
583 # out2in 2nd interface
584 pkts = self.create_stream_out(self.pg3)
585 self.pg3.add_stream(pkts)
586 self.pg_enable_capture(self.pg_interfaces)
588 capture = self.pg5.get_capture(len(pkts))
589 self.verify_capture_in(capture, self.pg5)
591 # in2out 3rd interface
592 pkts = self.create_stream_in(self.pg6, self.pg3)
593 self.pg6.add_stream(pkts)
594 self.pg_enable_capture(self.pg_interfaces)
596 capture = self.pg3.get_capture(len(pkts))
597 self.verify_capture_out(capture)
599 # out2in 3rd interface
600 pkts = self.create_stream_out(self.pg3)
601 self.pg3.add_stream(pkts)
602 self.pg_enable_capture(self.pg_interfaces)
604 capture = self.pg6.get_capture(len(pkts))
605 self.verify_capture_in(capture, self.pg6)
607 def test_hairpinning(self):
608 """ SNAT hairpinning """
610 host = self.pg0.remote_hosts[0]
611 server = self.pg0.remote_hosts[1]
614 server_in_port = 5678
615 server_out_port = 8765
617 self.snat_add_address(self.snat_addr)
618 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
619 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
621 # add static mapping for server
622 self.snat_add_static_mapping(server.ip4, self.snat_addr,
623 server_in_port, server_out_port)
625 # send packet from host to server
626 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
627 IP(src=host.ip4, dst=self.snat_addr) /
628 TCP(sport=host_in_port, dport=server_out_port))
629 self.pg0.add_stream(p)
630 self.pg_enable_capture(self.pg_interfaces)
632 capture = self.pg0.get_capture(1)
637 self.assertEqual(ip.src, self.snat_addr)
638 self.assertEqual(ip.dst, server.ip4)
639 self.assertNotEqual(tcp.sport, host_in_port)
640 self.assertEqual(tcp.dport, server_in_port)
641 host_out_port = tcp.sport
643 self.logger.error(ppp("Unexpected or invalid packet:", p))
646 # send reply from server to host
647 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
648 IP(src=server.ip4, dst=self.snat_addr) /
649 TCP(sport=server_in_port, dport=host_out_port))
650 self.pg0.add_stream(p)
651 self.pg_enable_capture(self.pg_interfaces)
653 capture = self.pg0.get_capture(1)
658 self.assertEqual(ip.src, self.snat_addr)
659 self.assertEqual(ip.dst, host.ip4)
660 self.assertEqual(tcp.sport, server_out_port)
661 self.assertEqual(tcp.dport, host_in_port)
663 self.logger.error(ppp("Unexpected or invalid packet:"), p)
666 def test_max_translations_per_user(self):
667 """ MAX translations per user - recycle the least recently used """
669 self.snat_add_address(self.snat_addr)
670 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
671 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
674 # get maximum number of translations per user
675 snat_config = self.vapi.snat_show_config()
677 # send more than maximum number of translations per user packets
678 pkts_num = snat_config.max_translations_per_user + 5
680 for port in range(0, pkts_num):
681 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
682 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
683 TCP(sport=1025 + port))
685 self.pg0.add_stream(pkts)
686 self.pg_enable_capture(self.pg_interfaces)
689 # verify number of translated packet
690 self.pg1.get_capture(pkts_num)
692 def test_interface_addr(self):
693 """ Acquire SNAT addresses from interface """
694 self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
696 # no address in NAT pool
697 adresses = self.vapi.snat_address_dump()
698 self.assertEqual(0, len(adresses))
700 # configure interface address and check NAT address pool
701 self.pg7.config_ip4()
702 adresses = self.vapi.snat_address_dump()
703 self.assertEqual(1, len(adresses))
705 # remove interface address and check NAT address pool
706 self.pg7.unconfig_ip4()
707 adresses = self.vapi.snat_address_dump()
708 self.assertEqual(0, len(adresses))
710 def test_ipfix_nat44_sess(self):
711 """ S-NAT IPFIX logging NAT44 session created/delted """
712 self.snat_add_address(self.snat_addr)
713 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
714 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
716 self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
717 src_address=self.pg3.local_ip4n,
719 template_interval=10)
720 self.vapi.snat_ipfix()
722 pkts = self.create_stream_in(self.pg0, self.pg1)
723 self.pg0.add_stream(pkts)
724 self.pg_enable_capture(self.pg_interfaces)
726 capture = self.pg1.get_capture(len(pkts))
727 self.verify_capture_out(capture)
728 self.snat_add_address(self.snat_addr, is_add=0)
729 self.vapi.cli("ipfix flush") # FIXME this should be an API call
730 capture = self.pg3.get_capture(3)
731 ipfix = IPFIXDecoder()
732 # first load template
734 self.assertTrue(p.haslayer(IPFIX))
735 if p.haslayer(Template):
736 ipfix.add_template(p.getlayer(Template))
737 # verify events in data set
740 data = ipfix.decode_data_set(p.getlayer(Set))
741 self.verify_ipfix_nat44_ses(data)
743 def test_ipfix_addr_exhausted(self):
744 """ S-NAT IPFIX logging NAT addresses exhausted """
745 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
746 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
748 self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
749 src_address=self.pg3.local_ip4n,
751 template_interval=10)
752 self.vapi.snat_ipfix()
754 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
755 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
757 self.pg0.add_stream(p)
758 self.pg_enable_capture(self.pg_interfaces)
760 capture = self.pg1.get_capture(0)
761 self.vapi.cli("ipfix flush") # FIXME this should be an API call
762 capture = self.pg3.get_capture(3)
763 ipfix = IPFIXDecoder()
764 # first load template
766 self.assertTrue(p.haslayer(IPFIX))
767 if p.haslayer(Template):
768 ipfix.add_template(p.getlayer(Template))
769 # verify events in data set
772 data = ipfix.decode_data_set(p.getlayer(Set))
773 self.verify_ipfix_addr_exhausted(data)
776 super(TestSNAT, self).tearDown()
777 if not self.vpp_dead:
778 self.logger.info(self.vapi.cli("show snat verbose"))
782 if __name__ == '__main__':
783 unittest.main(testRunner=VppTestRunner)