VPP-1277: IPIP - Copy TOS/TC from inner packet to outer.
[vpp.git] / test / test_ipip.py
1 #!/usr/bin/env python
2 """IP{4,6} over IP{v,6} tunnel functional tests"""
3
4 import unittest
5 from scapy.layers.inet6 import IPv6, Ether, IP, UDP
6 from framework import VppTestCase, VppTestRunner
7 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
8 from socket import AF_INET, AF_INET6, inet_pton
9
10 """ Testipip is a subclass of  VPPTestCase classes.
11
12 IPIP tests.
13
14 """
15
16
17 class TestIPIP(VppTestCase):
18     """ IPIP Test Case """
19
20     @classmethod
21     def setUpClass(cls):
22         super(TestIPIP, cls).setUpClass()
23         cls.create_pg_interfaces(range(2))
24         cls.interfaces = list(cls.pg_interfaces)
25
26     def setUp(cls):
27         super(TestIPIP, cls).setUp()
28         for i in cls.interfaces:
29             i.admin_up()
30             i.config_ip4()
31             i.config_ip6()
32             i.disable_ipv6_ra()
33             i.resolve_arp()
34             i.resolve_ndp()
35
36     def tearDown(self):
37         super(TestIPIP, self).tearDown()
38         if not self.vpp_dead:
39             for i in self.pg_interfaces:
40                 i.unconfig_ip4()
41                 i.unconfig_ip6()
42                 i.admin_down()
43
44     def validate(self, rx, expected):
45         self.assertEqual(rx, expected.__class__(str(expected)))
46
47     def test_ipip4(self):
48         """ ip{v4,v6} over ip4 test """
49         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
50         p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
51         p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
52         p_payload = UDP(sport=1234, dport=1234)
53
54         # IPv4 transport
55         rv = self.vapi.ipip_add_tunnel(
56             src_address=self.pg0.local_ip4n,
57             dst_address=self.pg1.remote_ip4n,
58             is_ipv6=0, tc_tos=0xFF)
59         sw_if_index = rv.sw_if_index
60
61         # Set interface up and enable IP on it
62         self.vapi.sw_interface_set_flags(sw_if_index, 1)
63         self.vapi.sw_interface_set_unnumbered(
64             ip_sw_if_index=self.pg0.sw_if_index,
65             sw_if_index=sw_if_index)
66
67         # Add IPv4 and IPv6 routes via tunnel interface
68         ip4_via_tunnel = VppIpRoute(
69             self, "130.67.0.0", 16,
70             [VppRoutePath("0.0.0.0",
71                           sw_if_index,
72                           proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
73         ip4_via_tunnel.add_vpp_config()
74
75         ip6_via_tunnel = VppIpRoute(
76             self, "dead::", 16,
77             [VppRoutePath("::",
78                           sw_if_index,
79                           proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
80         ip6_via_tunnel.add_vpp_config()
81
82         # IPv6 in to IPv4 tunnel
83         p6 = (p_ether / p_ip6 / p_payload)
84         p_inner_ip6 = p_ip6
85         p_inner_ip6.hlim -= 1
86         p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
87                        proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload)
88         p6_reply.ttl -= 1
89         rx = self.send_and_expect(self.pg0, p6*10, self.pg1)
90         for p in rx:
91             self.validate(p[1], p6_reply)
92
93         # IPv4 in to IPv4 tunnel
94         p4 = (p_ether / p_ip4 / p_payload)
95         p_ip4_inner = p_ip4
96         p_ip4_inner.ttl -= 1
97         p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
98                        tos=42) /
99                     p_ip4_inner / p_payload)
100         p4_reply.ttl -= 1
101         p4_reply.id = 0
102         rx = self.send_and_expect(self.pg0, p4*10, self.pg1)
103         for p in rx:
104             self.validate(p[1], p4_reply)
105
106         # Decapsulation
107         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
108
109         # IPv4 tunnel to IPv4
110         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
111         p4 = (p_ether / IP(src=self.pg1.remote_ip4,
112                            dst=self.pg0.local_ip4) / p_ip4 / p_payload)
113         p4_reply = (p_ip4 / p_payload)
114         p4_reply.ttl -= 1
115         rx = self.send_and_expect(self.pg1, p4*10, self.pg0)
116         for p in rx:
117             self.validate(p[1], p4_reply)
118
119         # IPv4 tunnel to IPv6
120         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
121         p6 = (p_ether / IP(src=self.pg1.remote_ip4,
122                            dst=self.pg0.local_ip4) / p_ip6 / p_payload)
123         p6_reply = (p_ip6 / p_payload)
124         p6_reply.hlim = 63
125         rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
126         for p in rx:
127             self.validate(p[1], p6_reply)
128
129     def test_ipip6(self):
130         """ ip{v4,v6} over ip6 test """
131         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
132         p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
133         p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
134         p_payload = UDP(sport=1234, dport=1234)
135
136         # IPv6 transport
137         rv = self.vapi.ipip_add_tunnel(
138             src_address=self.pg0.local_ip6n,
139             dst_address=self.pg1.remote_ip6n, tc_tos=255)
140
141         sw_if_index = rv.sw_if_index
142
143         self.vapi.sw_interface_set_flags(sw_if_index, 1)
144         self.vapi.sw_interface_set_unnumbered(
145             ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
146
147         # Add IPv4 and IPv6 routes via tunnel interface
148         ip4_via_tunnel = VppIpRoute(
149             self, "130.67.0.0", 16,
150             [VppRoutePath("0.0.0.0",
151                           sw_if_index,
152                           proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
153         ip4_via_tunnel.add_vpp_config()
154
155         ip6_via_tunnel = VppIpRoute(
156             self, "dead::", 16,
157             [VppRoutePath("::",
158                           sw_if_index,
159                           proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
160         ip6_via_tunnel.add_vpp_config()
161
162         # Encapsulation
163
164         # IPv6 in to IPv6 tunnel
165         p6 = (p_ether / p_ip6 / p_payload)
166         p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
167                          hlim=63, tc=42) /
168                     p_ip6 / p_payload)
169         p6_reply[1].hlim -= 1
170         rx = self.send_and_expect(self.pg0, p6*11, self.pg1)
171         for p in rx:
172             self.validate(p[1], p6_reply)
173
174         # IPv4 in to IPv6 tunnel
175         p4 = (p_ether / p_ip4 / p_payload)
176         p4_reply = (IPv6(src=self.pg0.local_ip6,
177                          dst=self.pg1.remote_ip6, hlim=63, tc=42) /
178                     p_ip4 / p_payload)
179         p4_reply[1].ttl -= 1
180         rx = self.send_and_expect(self.pg0, p4*11, self.pg1)
181         for p in rx:
182             self.validate(p[1], p4_reply)
183
184         # Decapsulation
185
186         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
187
188         # IPv6 tunnel to IPv4
189         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
190         p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
191                              dst=self.pg0.local_ip6) / p_ip4 / p_payload)
192         p4_reply = (p_ip4 / p_payload)
193         p4_reply.ttl -= 1
194         rx = self.send_and_expect(self.pg1, p4*11, self.pg0)
195         for p in rx:
196             self.validate(p[1], p4_reply)
197
198         # IPv6 tunnel to IPv6
199         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
200         p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
201                              dst=self.pg0.local_ip6) / p_ip6 / p_payload)
202         p6_reply = (p_ip6 / p_payload)
203         p6_reply.hlim = 63
204         rx = self.send_and_expect(self.pg1, p6*11, self.pg0)
205         for p in rx:
206             self.validate(p[1], p6_reply)
207
208     def test_ipip_create(self):
209         """ ipip create / delete interface test """
210         rv = self.vapi.ipip_add_tunnel(
211             src_address=inet_pton(AF_INET, '1.2.3.4'),
212             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
213         sw_if_index = rv.sw_if_index
214         self.vapi.ipip_del_tunnel(sw_if_index)
215
216
217 if __name__ == '__main__':
218     unittest.main(testRunner=VppTestRunner)