make test: Use VXLAN built in scapy 2.3.3
[vpp.git] / test / test_snat.py
1 #!/usr/bin/env python
2
3 import socket
4 import unittest
5 from logging import *
6
7 from framework import VppTestCase, VppTestRunner
8
9 from scapy.layers.inet import IP, TCP, UDP, ICMP
10 from scapy.layers.l2 import Ether
11
12
13 class TestSNAT(VppTestCase):
14     """ SNAT Test Cases """
15
16     @classmethod
17     def setUpClass(cls):
18         super(TestSNAT, cls).setUpClass()
19
20         try:
21             cls.tcp_port_in = 6303
22             cls.tcp_port_out = 6303
23             cls.udp_port_in = 6304
24             cls.udp_port_out = 6304
25             cls.icmp_id_in = 6305
26             cls.icmp_id_out = 6305
27             cls.snat_addr = '10.0.0.3'
28
29             cls.create_pg_interfaces(range(7))
30             cls.interfaces = list(cls.pg_interfaces[0:4])
31
32             for i in cls.interfaces:
33                 i.admin_up()
34                 i.config_ip4()
35                 i.resolve_arp()
36
37             cls.pg0.generate_remote_hosts(2)
38             cls.pg0.configure_ipv4_neighbors()
39
40             cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
41
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)
47                 i.config_ip4()
48                 i.admin_up()
49                 i.resolve_arp()
50
51         except Exception:
52             super(TestSNAT, cls).tearDownClass()
53             raise
54
55     def create_stream_in(self, in_if, out_if):
56         """
57         Create packet stream for inside network
58
59         :param in_if: Inside interface
60         :param out_if: Outside interface
61         """
62         pkts = []
63         # TCP
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))
67         pkts.append(p)
68
69         # UDP
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))
73         pkts.append(p)
74
75         # ICMP
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'))
79         pkts.append(p)
80
81         return pkts
82
83     def create_stream_out(self, out_if, dst_ip=None):
84         """
85         Create packet stream for outside network
86
87         :param out_if: Outside interface
88         :param dst_ip: Destination IP address (Default use global SNAT address)
89         """
90         if dst_ip is None:
91              dst_ip=self.snat_addr
92         pkts = []
93         # TCP
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))
97         pkts.append(p)
98
99         # UDP
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))
103         pkts.append(p)
104
105         # ICMP
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'))
109         pkts.append(p)
110
111         return pkts
112
113     def verify_capture_out(self, capture, nat_ip=None, same_port=False,
114                            packet_num=3):
115         """
116         Verify captured packets on outside network
117
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)
122         """
123         if nat_ip is None:
124             nat_ip = self.snat_addr
125         self.assertEqual(packet_num, len(capture))
126         for packet in capture:
127             try:
128                 self.assertEqual(packet[IP].src, nat_ip)
129                 if packet.haslayer(TCP):
130                     if same_port:
131                         self.assertEqual(packet[TCP].sport, self.tcp_port_in)
132                     else:
133                         self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
134                     self.tcp_port_out = packet[TCP].sport
135                 elif packet.haslayer(UDP):
136                     if same_port:
137                         self.assertEqual(packet[UDP].sport, self.udp_port_in)
138                     else:
139                         self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
140                     self.udp_port_out = packet[UDP].sport
141                 else:
142                     if same_port:
143                         self.assertEqual(packet[ICMP].id, self.icmp_id_in)
144                     else:
145                         self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
146                     self.icmp_id_out = packet[ICMP].id
147             except:
148                 error("Unexpected or invalid packet (outside network):")
149                 error(packet.show())
150                 raise
151
152     def verify_capture_in(self, capture, in_if, packet_num=3):
153         """
154         Verify captured packets on inside network
155
156         :param capture: Captured packets
157         :param in_if: Inside interface
158         :param packet_num: Expected number of packets (Default 3)
159         """
160         self.assertEqual(packet_num, len(capture))
161         for packet in capture:
162             try:
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)
168                 else:
169                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
170             except:
171                 error("Unexpected or invalid packet (inside network):")
172                 error(packet.show())
173                 raise
174
175     def clear_snat(self):
176         """
177         Clear SNAT configuration.
178         """
179         interfaces = self.vapi.snat_interface_dump()
180         for intf in interfaces:
181             self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
182                                                      intf.is_inside,
183                                                      is_add=0)
184
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,
192                                               vrf_id=sm.vrf_id,
193                                               is_add=0)
194
195         adresses = self.vapi.snat_address_dump()
196         for addr in adresses:
197             self.vapi.snat_add_address_range(addr.ip_address,
198                                              addr.ip_address,
199                                              is_add=0)
200
201     def snat_add_static_mapping(self, local_ip, external_ip, local_port=0,
202                                 external_port=0, vrf_id=0, is_add=1):
203         """
204         Add/delete S-NAT static mapping
205
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)
212         """
213         addr_only = 1
214         if local_port and external_port:
215             addr_only = 0
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)
220
221     def snat_add_address(self, ip, is_add=1):
222         """
223         Add/delete S-NAT address
224
225         :param ip: IP address
226         :param is_add: 1 if add, 0 if delete (Default add)
227         """
228         snat_addr = socket.inet_pton(socket.AF_INET, ip)
229         self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add)
230
231     def test_dynamic(self):
232         """ SNAT dynamic translation test """
233
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,
237                                                  is_inside=0)
238
239         # in2out
240         pkts = self.create_stream_in(self.pg0, self.pg1)
241         self.pg0.add_stream(pkts)
242         self.pg_enable_capture(self.pg_interfaces)
243         self.pg_start()
244         capture = self.pg1.get_capture()
245         self.verify_capture_out(capture)
246
247         # out2in
248         pkts = self.create_stream_out(self.pg1)
249         self.pg1.add_stream(pkts)
250         self.pg_enable_capture(self.pg_interfaces)
251         self.pg_start()
252         capture = self.pg0.get_capture()
253         self.verify_capture_in(capture, self.pg0)
254
255     def test_static_in(self):
256         """ SNAT 1:1 NAT initialized from inside network """
257
258         nat_ip = "10.0.0.10"
259         self.tcp_port_out = 6303
260         self.udp_port_out = 6304
261         self.icmp_id_out = 6305
262
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,
266                                                  is_inside=0)
267
268         # in2out
269         pkts = self.create_stream_in(self.pg0, self.pg1)
270         self.pg0.add_stream(pkts)
271         self.pg_enable_capture(self.pg_interfaces)
272         self.pg_start()
273         capture = self.pg1.get_capture()
274         self.verify_capture_out(capture, nat_ip, True)
275
276         # out2in
277         pkts = self.create_stream_out(self.pg1, nat_ip)
278         self.pg1.add_stream(pkts)
279         self.pg_enable_capture(self.pg_interfaces)
280         self.pg_start()
281         capture = self.pg0.get_capture()
282         self.verify_capture_in(capture, self.pg0)
283
284     def test_static_out(self):
285         """ SNAT 1:1 NAT initialized from outside network """
286
287         nat_ip = "10.0.0.20"
288         self.tcp_port_out = 6303
289         self.udp_port_out = 6304
290         self.icmp_id_out = 6305
291
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,
295                                                  is_inside=0)
296
297         # out2in
298         pkts = self.create_stream_out(self.pg1, nat_ip)
299         self.pg1.add_stream(pkts)
300         self.pg_enable_capture(self.pg_interfaces)
301         self.pg_start()
302         capture = self.pg0.get_capture()
303         self.verify_capture_in(capture, self.pg0)
304
305         # in2out
306         pkts = self.create_stream_in(self.pg0, self.pg1)
307         self.pg0.add_stream(pkts)
308         self.pg_enable_capture(self.pg_interfaces)
309         self.pg_start()
310         capture = self.pg1.get_capture()
311         self.verify_capture_out(capture, nat_ip, True)
312
313     def test_static_with_port_in(self):
314         """ SNAT 1:1 NAT with port initialized from inside network """
315
316         self.tcp_port_out = 3606
317         self.udp_port_out = 3607
318         self.icmp_id_out = 3608
319
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,
329                                                  is_inside=0)
330
331         # in2out
332         pkts = self.create_stream_in(self.pg0, self.pg1)
333         self.pg0.add_stream(pkts)
334         self.pg_enable_capture(self.pg_interfaces)
335         self.pg_start()
336         capture = self.pg1.get_capture()
337         self.verify_capture_out(capture)
338
339         # out2in
340         pkts = self.create_stream_out(self.pg1)
341         self.pg1.add_stream(pkts)
342         self.pg_enable_capture(self.pg_interfaces)
343         self.pg_start()
344         capture = self.pg0.get_capture()
345         self.verify_capture_in(capture, self.pg0)
346
347     def test_static_with_port_out(self):
348         """ SNAT 1:1 NAT with port initialized from outside network """
349
350         self.tcp_port_out = 30606
351         self.udp_port_out = 30607
352         self.icmp_id_out = 30608
353
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,
363                                                  is_inside=0)
364
365         # out2in
366         pkts = self.create_stream_out(self.pg1)
367         self.pg1.add_stream(pkts)
368         self.pg_enable_capture(self.pg_interfaces)
369         self.pg_start()
370         capture = self.pg0.get_capture()
371         self.verify_capture_in(capture, self.pg0)
372
373         # in2out
374         pkts = self.create_stream_in(self.pg0, self.pg1)
375         self.pg0.add_stream(pkts)
376         self.pg_enable_capture(self.pg_interfaces)
377         self.pg_start()
378         capture = self.pg1.get_capture()
379         self.verify_capture_out(capture)
380
381     def test_static_vrf_aware(self):
382         """ SNAT 1:1 NAT VRF awareness """
383
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
389
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,
395                                                  is_inside=0)
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)
398
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)
403         self.pg_start()
404         capture = self.pg3.get_capture()
405         self.verify_capture_out(capture, nat_ip1, True)
406
407         # inside interface VRF don't match SNAT static mapping VRF (packets
408         # are dropped)
409         pkts = self.create_stream_in(self.pg0, self.pg3)
410         self.pg0.add_stream(pkts)
411         self.pg_enable_capture(self.pg_interfaces)
412         self.pg_start()
413         capture = self.pg3.get_capture()
414         self.verify_capture_out(capture, packet_num=0)
415
416     def test_multiple_inside_interfaces(self):
417         """ SNAT multiple inside interfaces with non-overlapping address space """
418
419         self.snat_add_address(self.snat_addr)
420         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
421         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
422         self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index)
423         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
424                                                  is_inside=0)
425
426         # in2out 1st interface
427         pkts = self.create_stream_in(self.pg0, self.pg3)
428         self.pg0.add_stream(pkts)
429         self.pg_enable_capture(self.pg_interfaces)
430         self.pg_start()
431         capture = self.pg3.get_capture()
432         self.verify_capture_out(capture)
433
434         # out2in 1st interface
435         pkts = self.create_stream_out(self.pg3)
436         self.pg3.add_stream(pkts)
437         self.pg_enable_capture(self.pg_interfaces)
438         self.pg_start()
439         capture = self.pg0.get_capture()
440         self.verify_capture_in(capture, self.pg0)
441
442         # in2out 2nd interface
443         pkts = self.create_stream_in(self.pg1, self.pg3)
444         self.pg1.add_stream(pkts)
445         self.pg_enable_capture(self.pg_interfaces)
446         self.pg_start()
447         capture = self.pg3.get_capture()
448         self.verify_capture_out(capture)
449
450         # out2in 2nd interface
451         pkts = self.create_stream_out(self.pg3)
452         self.pg3.add_stream(pkts)
453         self.pg_enable_capture(self.pg_interfaces)
454         self.pg_start()
455         capture = self.pg1.get_capture()
456         self.verify_capture_in(capture, self.pg1)
457
458         # in2out 3rd interface
459         pkts = self.create_stream_in(self.pg2, self.pg3)
460         self.pg2.add_stream(pkts)
461         self.pg_enable_capture(self.pg_interfaces)
462         self.pg_start()
463         capture = self.pg3.get_capture()
464         self.verify_capture_out(capture)
465
466         # out2in 3rd interface
467         pkts = self.create_stream_out(self.pg3)
468         self.pg3.add_stream(pkts)
469         self.pg_enable_capture(self.pg_interfaces)
470         self.pg_start()
471         capture = self.pg2.get_capture()
472         self.verify_capture_in(capture, self.pg2)
473
474     def test_inside_overlapping_interfaces(self):
475         """ SNAT multiple inside interfaces with overlapping address space """
476
477         self.snat_add_address(self.snat_addr)
478         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
479                                                  is_inside=0)
480         self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
481         self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
482         self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
483
484         # in2out 1st interface
485         pkts = self.create_stream_in(self.pg4, self.pg3)
486         self.pg4.add_stream(pkts)
487         self.pg_enable_capture(self.pg_interfaces)
488         self.pg_start()
489         capture = self.pg3.get_capture()
490         self.verify_capture_out(capture)
491
492         # out2in 1st interface
493         pkts = self.create_stream_out(self.pg3)
494         self.pg3.add_stream(pkts)
495         self.pg_enable_capture(self.pg_interfaces)
496         self.pg_start()
497         capture = self.pg4.get_capture()
498         self.verify_capture_in(capture, self.pg4)
499
500         # in2out 2nd interface
501         pkts = self.create_stream_in(self.pg5, self.pg3)
502         self.pg5.add_stream(pkts)
503         self.pg_enable_capture(self.pg_interfaces)
504         self.pg_start()
505         capture = self.pg3.get_capture()
506         self.verify_capture_out(capture)
507
508         # out2in 2nd interface
509         pkts = self.create_stream_out(self.pg3)
510         self.pg3.add_stream(pkts)
511         self.pg_enable_capture(self.pg_interfaces)
512         self.pg_start()
513         capture = self.pg5.get_capture()
514         self.verify_capture_in(capture, self.pg5)
515
516         # in2out 3rd interface
517         pkts = self.create_stream_in(self.pg6, self.pg3)
518         self.pg6.add_stream(pkts)
519         self.pg_enable_capture(self.pg_interfaces)
520         self.pg_start()
521         capture = self.pg3.get_capture()
522         self.verify_capture_out(capture)
523
524         # out2in 3rd interface
525         pkts = self.create_stream_out(self.pg3)
526         self.pg3.add_stream(pkts)
527         self.pg_enable_capture(self.pg_interfaces)
528         self.pg_start()
529         capture = self.pg6.get_capture()
530         self.verify_capture_in(capture, self.pg6)
531
532     def test_hairpinning(self):
533         """ SNAT hairpinning """
534
535         host = self.pg0.remote_hosts[0]
536         server = self.pg0.remote_hosts[1]
537         host_in_port = 1234
538         host_out_port = 0
539         server_in_port = 5678
540         server_out_port = 8765
541
542         self.snat_add_address(self.snat_addr)
543         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
544         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
545                                                  is_inside=0)
546         # add static mapping for server
547         self.snat_add_static_mapping(server.ip4, self.snat_addr,
548                                      server_in_port, server_out_port)
549
550         # send packet from host to server
551         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
552              IP(src=host.ip4, dst=self.snat_addr) /
553              TCP(sport=host_in_port, dport=server_out_port))
554         self.pg0.add_stream(p)
555         self.pg_enable_capture(self.pg_interfaces)
556         self.pg_start()
557         capture = self.pg0.get_capture()
558         self.assertEqual(1, len(capture))
559         p = capture[0]
560         try:
561             ip = p[IP]
562             tcp = p[TCP]
563             self.assertEqual(ip.src, self.snat_addr)
564             self.assertEqual(ip.dst, server.ip4)
565             self.assertNotEqual(tcp.sport, host_in_port)
566             self.assertEqual(tcp.dport, server_in_port)
567             host_out_port = tcp.sport
568         except:
569             error("Unexpected or invalid packet:")
570             error(p.show())
571             raise
572
573         # send reply from server to host
574         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
575              IP(src=server.ip4, dst=self.snat_addr) /
576              TCP(sport=server_in_port, dport=host_out_port))
577         self.pg0.add_stream(p)
578         self.pg_enable_capture(self.pg_interfaces)
579         self.pg_start()
580         capture = self.pg0.get_capture()
581         self.assertEqual(1, len(capture))
582         p = capture[0]
583         try:
584             ip = p[IP]
585             tcp = p[TCP]
586             self.assertEqual(ip.src, self.snat_addr)
587             self.assertEqual(ip.dst, host.ip4)
588             self.assertEqual(tcp.sport, server_out_port)
589             self.assertEqual(tcp.dport, host_in_port)
590         except:
591             error("Unexpected or invalid packet:")
592             error(p.show())
593             raise
594
595
596     def tearDown(self):
597         super(TestSNAT, self).tearDown()
598         if not self.vpp_dead:
599             self.logger.info(self.vapi.cli("show snat verbose"))
600             self.clear_snat()
601
602 if __name__ == '__main__':
603     unittest.main(testRunner=VppTestRunner)