6 from framework import VppTestCase, VppTestRunner
8 from scapy.layers.inet import IP, TCP, UDP, ICMP
9 from scapy.layers.l2 import Ether
13 class TestSNAT(VppTestCase):
14 """ SNAT Test Cases """
18 super(TestSNAT, cls).setUpClass()
21 cls.tcp_port_in = 6303
22 cls.tcp_port_out = 6303
23 cls.udp_port_in = 6304
24 cls.udp_port_out = 6304
26 cls.icmp_id_out = 6305
27 cls.snat_addr = '10.0.0.3'
29 cls.create_pg_interfaces(range(7))
30 cls.interfaces = list(cls.pg_interfaces[0:4])
32 for i in cls.interfaces:
37 cls.pg0.generate_remote_hosts(2)
38 cls.pg0.configure_ipv4_neighbors()
40 cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
42 for i in cls.overlapping_interfaces:
43 i._local_ip4 = "172.16.255.1"
44 i._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
45 i._remote_hosts[0]._ip4 = "172.16.255.2"
46 i.set_table_ip4(i.sw_if_index)
52 super(TestSNAT, cls).tearDownClass()
55 def create_stream_in(self, in_if, out_if):
57 Create packet stream for inside network
59 :param in_if: Inside interface
60 :param out_if: Outside interface
64 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
65 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
66 TCP(sport=self.tcp_port_in))
70 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
71 IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
72 UDP(sport=self.udp_port_in))
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 ICMP(id=self.icmp_id_in, type='echo-request'))
83 def create_stream_out(self, out_if, dst_ip=None):
85 Create packet stream for outside network
87 :param out_if: Outside interface
88 :param dst_ip: Destination IP address (Default use global SNAT address)
91 dst_ip = self.snat_addr
94 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
95 IP(src=out_if.remote_ip4, dst=dst_ip) /
96 TCP(dport=self.tcp_port_out))
100 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
101 IP(src=out_if.remote_ip4, dst=dst_ip) /
102 UDP(dport=self.udp_port_out))
106 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
107 IP(src=out_if.remote_ip4, dst=dst_ip) /
108 ICMP(id=self.icmp_id_out, type='echo-reply'))
113 def verify_capture_out(self, capture, nat_ip=None, same_port=False,
116 Verify captured packets on outside network
118 :param capture: Captured packets
119 :param nat_ip: Translated IP address (Default use global SNAT address)
120 :param same_port: Sorce port number is not translated (Default False)
121 :param packet_num: Expected number of packets (Default 3)
124 nat_ip = self.snat_addr
125 self.assertEqual(packet_num, len(capture))
126 for packet in capture:
128 self.assertEqual(packet[IP].src, nat_ip)
129 if packet.haslayer(TCP):
131 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
133 self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
134 self.tcp_port_out = packet[TCP].sport
135 elif packet.haslayer(UDP):
137 self.assertEqual(packet[UDP].sport, self.udp_port_in)
139 self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
140 self.udp_port_out = packet[UDP].sport
143 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
145 self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
146 self.icmp_id_out = packet[ICMP].id
148 self.logger.error(ppp("Unexpected or invalid packet "
149 "(outside network):", packet))
152 def verify_capture_in(self, capture, in_if, packet_num=3):
154 Verify captured packets on inside network
156 :param capture: Captured packets
157 :param in_if: Inside interface
158 :param packet_num: Expected number of packets (Default 3)
160 self.assertEqual(packet_num, len(capture))
161 for packet in capture:
163 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
164 if packet.haslayer(TCP):
165 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
166 elif packet.haslayer(UDP):
167 self.assertEqual(packet[UDP].dport, self.udp_port_in)
169 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
171 self.logger.error(ppp("Unexpected or invalid packet "
172 "(inside network):", packet))
175 def clear_snat(self):
177 Clear SNAT configuration.
179 interfaces = self.vapi.snat_interface_dump()
180 for intf in interfaces:
181 self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
185 static_mappings = self.vapi.snat_static_mapping_dump()
186 for sm in static_mappings:
187 self.vapi.snat_add_static_mapping(sm.local_ip_address,
188 sm.external_ip_address,
189 local_port=sm.local_port,
190 external_port=sm.external_port,
191 addr_only=sm.addr_only,
195 adresses = self.vapi.snat_address_dump()
196 for addr in adresses:
197 self.vapi.snat_add_address_range(addr.ip_address,
201 def snat_add_static_mapping(self, local_ip, external_ip, local_port=0,
202 external_port=0, vrf_id=0, is_add=1):
204 Add/delete S-NAT static mapping
206 :param local_ip: Local IP address
207 :param external_ip: External IP address
208 :param local_port: Local port number (Optional)
209 :param external_port: External port number (Optional)
210 :param vrf_id: VRF ID (Default 0)
211 :param is_add: 1 if add, 0 if delete (Default add)
214 if local_port and external_port:
216 l_ip = socket.inet_pton(socket.AF_INET, local_ip)
217 e_ip = socket.inet_pton(socket.AF_INET, external_ip)
218 self.vapi.snat_add_static_mapping(l_ip, e_ip, local_port, external_port,
219 addr_only, vrf_id, is_add)
221 def snat_add_address(self, ip, is_add=1):
223 Add/delete S-NAT address
225 :param ip: IP address
226 :param is_add: 1 if add, 0 if delete (Default add)
228 snat_addr = socket.inet_pton(socket.AF_INET, ip)
229 self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add)
231 def test_dynamic(self):
232 """ SNAT dynamic translation test """
234 self.snat_add_address(self.snat_addr)
235 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
236 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
240 pkts = self.create_stream_in(self.pg0, self.pg1)
241 self.pg0.add_stream(pkts)
242 self.pg_enable_capture(self.pg_interfaces)
244 capture = self.pg1.get_capture()
245 self.verify_capture_out(capture)
248 pkts = self.create_stream_out(self.pg1)
249 self.pg1.add_stream(pkts)
250 self.pg_enable_capture(self.pg_interfaces)
252 capture = self.pg0.get_capture()
253 self.verify_capture_in(capture, self.pg0)
255 def test_static_in(self):
256 """ SNAT 1:1 NAT initialized from inside network """
259 self.tcp_port_out = 6303
260 self.udp_port_out = 6304
261 self.icmp_id_out = 6305
263 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
264 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
265 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
269 pkts = self.create_stream_in(self.pg0, self.pg1)
270 self.pg0.add_stream(pkts)
271 self.pg_enable_capture(self.pg_interfaces)
273 capture = self.pg1.get_capture()
274 self.verify_capture_out(capture, nat_ip, True)
277 pkts = self.create_stream_out(self.pg1, nat_ip)
278 self.pg1.add_stream(pkts)
279 self.pg_enable_capture(self.pg_interfaces)
281 capture = self.pg0.get_capture()
282 self.verify_capture_in(capture, self.pg0)
284 def test_static_out(self):
285 """ SNAT 1:1 NAT initialized from outside network """
288 self.tcp_port_out = 6303
289 self.udp_port_out = 6304
290 self.icmp_id_out = 6305
292 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
293 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
294 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
298 pkts = self.create_stream_out(self.pg1, nat_ip)
299 self.pg1.add_stream(pkts)
300 self.pg_enable_capture(self.pg_interfaces)
302 capture = self.pg0.get_capture()
303 self.verify_capture_in(capture, self.pg0)
306 pkts = self.create_stream_in(self.pg0, self.pg1)
307 self.pg0.add_stream(pkts)
308 self.pg_enable_capture(self.pg_interfaces)
310 capture = self.pg1.get_capture()
311 self.verify_capture_out(capture, nat_ip, True)
313 def test_static_with_port_in(self):
314 """ SNAT 1:1 NAT with port initialized from inside network """
316 self.tcp_port_out = 3606
317 self.udp_port_out = 3607
318 self.icmp_id_out = 3608
320 self.snat_add_address(self.snat_addr)
321 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
322 self.tcp_port_in, self.tcp_port_out)
323 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
324 self.udp_port_in, self.udp_port_out)
325 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
326 self.icmp_id_in, self.icmp_id_out)
327 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
328 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
332 pkts = self.create_stream_in(self.pg0, self.pg1)
333 self.pg0.add_stream(pkts)
334 self.pg_enable_capture(self.pg_interfaces)
336 capture = self.pg1.get_capture()
337 self.verify_capture_out(capture)
340 pkts = self.create_stream_out(self.pg1)
341 self.pg1.add_stream(pkts)
342 self.pg_enable_capture(self.pg_interfaces)
344 capture = self.pg0.get_capture()
345 self.verify_capture_in(capture, self.pg0)
347 def test_static_with_port_out(self):
348 """ SNAT 1:1 NAT with port initialized from outside network """
350 self.tcp_port_out = 30606
351 self.udp_port_out = 30607
352 self.icmp_id_out = 30608
354 self.snat_add_address(self.snat_addr)
355 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
356 self.tcp_port_in, self.tcp_port_out)
357 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
358 self.udp_port_in, self.udp_port_out)
359 self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
360 self.icmp_id_in, self.icmp_id_out)
361 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
362 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
366 pkts = self.create_stream_out(self.pg1)
367 self.pg1.add_stream(pkts)
368 self.pg_enable_capture(self.pg_interfaces)
370 capture = self.pg0.get_capture()
371 self.verify_capture_in(capture, self.pg0)
374 pkts = self.create_stream_in(self.pg0, self.pg1)
375 self.pg0.add_stream(pkts)
376 self.pg_enable_capture(self.pg_interfaces)
378 capture = self.pg1.get_capture()
379 self.verify_capture_out(capture)
381 def test_static_vrf_aware(self):
382 """ SNAT 1:1 NAT VRF awareness """
384 nat_ip1 = "10.0.0.30"
385 nat_ip2 = "10.0.0.40"
386 self.tcp_port_out = 6303
387 self.udp_port_out = 6304
388 self.icmp_id_out = 6305
390 self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
391 vrf_id=self.pg4.sw_if_index)
392 self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
393 vrf_id=self.pg4.sw_if_index)
394 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
396 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
397 self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
399 # inside interface VRF match SNAT static mapping VRF
400 pkts = self.create_stream_in(self.pg4, self.pg3)
401 self.pg4.add_stream(pkts)
402 self.pg_enable_capture(self.pg_interfaces)
404 capture = self.pg3.get_capture()
405 self.verify_capture_out(capture, nat_ip1, True)
407 # inside interface VRF don't match SNAT static mapping VRF (packets
409 pkts = self.create_stream_in(self.pg0, self.pg3)
410 self.pg0.add_stream(pkts)
411 self.pg_enable_capture(self.pg_interfaces)
413 self.pg3.assert_nothing_captured()
415 def test_multiple_inside_interfaces(self):
416 """SNAT multiple inside interfaces with non-overlapping address space"""
418 self.snat_add_address(self.snat_addr)
419 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
420 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
421 self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index)
422 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
425 # in2out 1st interface
426 pkts = self.create_stream_in(self.pg0, self.pg3)
427 self.pg0.add_stream(pkts)
428 self.pg_enable_capture(self.pg_interfaces)
430 capture = self.pg3.get_capture()
431 self.verify_capture_out(capture)
433 # out2in 1st interface
434 pkts = self.create_stream_out(self.pg3)
435 self.pg3.add_stream(pkts)
436 self.pg_enable_capture(self.pg_interfaces)
438 capture = self.pg0.get_capture()
439 self.verify_capture_in(capture, self.pg0)
441 # in2out 2nd interface
442 pkts = self.create_stream_in(self.pg1, self.pg3)
443 self.pg1.add_stream(pkts)
444 self.pg_enable_capture(self.pg_interfaces)
446 capture = self.pg3.get_capture()
447 self.verify_capture_out(capture)
449 # out2in 2nd interface
450 pkts = self.create_stream_out(self.pg3)
451 self.pg3.add_stream(pkts)
452 self.pg_enable_capture(self.pg_interfaces)
454 capture = self.pg1.get_capture()
455 self.verify_capture_in(capture, self.pg1)
457 # in2out 3rd interface
458 pkts = self.create_stream_in(self.pg2, self.pg3)
459 self.pg2.add_stream(pkts)
460 self.pg_enable_capture(self.pg_interfaces)
462 capture = self.pg3.get_capture()
463 self.verify_capture_out(capture)
465 # out2in 3rd interface
466 pkts = self.create_stream_out(self.pg3)
467 self.pg3.add_stream(pkts)
468 self.pg_enable_capture(self.pg_interfaces)
470 capture = self.pg2.get_capture()
471 self.verify_capture_in(capture, self.pg2)
473 def test_inside_overlapping_interfaces(self):
474 """ SNAT multiple inside interfaces with overlapping address space """
476 self.snat_add_address(self.snat_addr)
477 self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
479 self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
480 self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
481 self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
483 # in2out 1st interface
484 pkts = self.create_stream_in(self.pg4, self.pg3)
485 self.pg4.add_stream(pkts)
486 self.pg_enable_capture(self.pg_interfaces)
488 capture = self.pg3.get_capture()
489 self.verify_capture_out(capture)
491 # out2in 1st interface
492 pkts = self.create_stream_out(self.pg3)
493 self.pg3.add_stream(pkts)
494 self.pg_enable_capture(self.pg_interfaces)
496 capture = self.pg4.get_capture()
497 self.verify_capture_in(capture, self.pg4)
499 # in2out 2nd interface
500 pkts = self.create_stream_in(self.pg5, self.pg3)
501 self.pg5.add_stream(pkts)
502 self.pg_enable_capture(self.pg_interfaces)
504 capture = self.pg3.get_capture()
505 self.verify_capture_out(capture)
507 # out2in 2nd interface
508 pkts = self.create_stream_out(self.pg3)
509 self.pg3.add_stream(pkts)
510 self.pg_enable_capture(self.pg_interfaces)
512 capture = self.pg5.get_capture()
513 self.verify_capture_in(capture, self.pg5)
515 # in2out 3rd interface
516 pkts = self.create_stream_in(self.pg6, self.pg3)
517 self.pg6.add_stream(pkts)
518 self.pg_enable_capture(self.pg_interfaces)
520 capture = self.pg3.get_capture()
521 self.verify_capture_out(capture)
523 # out2in 3rd interface
524 pkts = self.create_stream_out(self.pg3)
525 self.pg3.add_stream(pkts)
526 self.pg_enable_capture(self.pg_interfaces)
528 capture = self.pg6.get_capture()
529 self.verify_capture_in(capture, self.pg6)
531 def test_hairpinning(self):
532 """ SNAT hairpinning """
534 host = self.pg0.remote_hosts[0]
535 server = self.pg0.remote_hosts[1]
538 server_in_port = 5678
539 server_out_port = 8765
541 self.snat_add_address(self.snat_addr)
542 self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
543 self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
545 # add static mapping for server
546 self.snat_add_static_mapping(server.ip4, self.snat_addr,
547 server_in_port, server_out_port)
549 # send packet from host to server
550 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
551 IP(src=host.ip4, dst=self.snat_addr) /
552 TCP(sport=host_in_port, dport=server_out_port))
553 self.pg0.add_stream(p)
554 self.pg_enable_capture(self.pg_interfaces)
556 capture = self.pg0.get_capture()
557 self.assertEqual(1, len(capture))
562 self.assertEqual(ip.src, self.snat_addr)
563 self.assertEqual(ip.dst, server.ip4)
564 self.assertNotEqual(tcp.sport, host_in_port)
565 self.assertEqual(tcp.dport, server_in_port)
566 host_out_port = tcp.sport
568 self.logger.error(ppp("Unexpected or invalid packet:", p))
571 # send reply from server to host
572 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
573 IP(src=server.ip4, dst=self.snat_addr) /
574 TCP(sport=server_in_port, dport=host_out_port))
575 self.pg0.add_stream(p)
576 self.pg_enable_capture(self.pg_interfaces)
578 capture = self.pg0.get_capture()
579 self.assertEqual(1, len(capture))
584 self.assertEqual(ip.src, self.snat_addr)
585 self.assertEqual(ip.dst, host.ip4)
586 self.assertEqual(tcp.sport, server_out_port)
587 self.assertEqual(tcp.dport, host_in_port)
589 self.logger.error(ppp("Unexpected or invalid packet:"), p)
593 super(TestSNAT, self).tearDown()
594 if not self.vpp_dead:
595 self.logger.info(self.vapi.cli("show snat verbose"))
598 if __name__ == '__main__':
599 unittest.main(testRunner=VppTestRunner)