2 """IP{4,6} over IP{v,6} tunnel functional tests"""
5 from scapy.layers.inet6 import IPv6, Ether, IP, UDP
6 from scapy.all import fragment, RandShort
7 from framework import VppTestCase, VppTestRunner
8 from vpp_ip import DpoProto
9 from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
10 from socket import AF_INET, AF_INET6, inet_pton
13 """ Testipip is a subclass of VPPTestCase classes.
20 def reassemble(listoffragments):
21 buffer = StringIO.StringIO()
22 first = listoffragments[0]
24 for pkt in listoffragments:
25 buffer.seek(pkt[IP].frag*8)
26 buffer.write(pkt[IP].payload)
27 first.len = len(buffer.getvalue()) + 20
30 header = str(first[IP])[:20]
31 return first[IP].__class__(header + buffer.getvalue())
34 class TestIPIP(VppTestCase):
35 """ IPIP Test Case """
39 super(TestIPIP, cls).setUpClass()
40 cls.create_pg_interfaces(range(2))
41 cls.interfaces = list(cls.pg_interfaces)
44 super(TestIPIP, cls).setUp()
45 for i in cls.interfaces:
54 super(TestIPIP, self).tearDown()
56 for i in self.pg_interfaces:
61 def validate(self, rx, expected):
62 self.assertEqual(rx, expected.__class__(str(expected)))
64 def generate_frags(self, payload_length, fragment_size):
65 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
66 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
67 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
68 outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
70 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
71 frags = fragment(outer_ip4, fragment_size)
72 p4_reply = (p_ip4 / p_payload)
74 return frags, p4_reply
77 """ ip{v4,v6} over ip4 test """
78 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
79 p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
80 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
81 p_payload = UDP(sport=1234, dport=1234)
84 rv = self.vapi.ipip_add_tunnel(
85 src_address=self.pg0.local_ip4n,
86 dst_address=self.pg1.remote_ip4n,
87 is_ipv6=0, tc_tos=0xFF)
88 sw_if_index = rv.sw_if_index
90 # Set interface up and enable IP on it
91 self.vapi.sw_interface_set_flags(sw_if_index, 1)
92 self.vapi.sw_interface_set_unnumbered(
93 ip_sw_if_index=self.pg0.sw_if_index,
94 sw_if_index=sw_if_index)
96 # Add IPv4 and IPv6 routes via tunnel interface
97 ip4_via_tunnel = VppIpRoute(
98 self, "130.67.0.0", 16,
99 [VppRoutePath("0.0.0.0",
101 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
102 ip4_via_tunnel.add_vpp_config()
104 ip6_via_tunnel = VppIpRoute(
108 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
109 ip6_via_tunnel.add_vpp_config()
111 # IPv6 in to IPv4 tunnel
112 p6 = (p_ether / p_ip6 / p_payload)
114 p_inner_ip6.hlim -= 1
115 p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
116 proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload)
118 rx = self.send_and_expect(self.pg0, p6*10, self.pg1)
120 self.validate(p[1], p6_reply)
122 # IPv4 in to IPv4 tunnel
123 p4 = (p_ether / p_ip4 / p_payload)
126 p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
128 p_ip4_inner / p_payload)
131 rx = self.send_and_expect(self.pg0, p4*10, self.pg1)
133 self.validate(p[1], p4_reply)
136 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
138 # IPv4 tunnel to IPv4
139 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
140 p4 = (p_ether / IP(src=self.pg1.remote_ip4,
141 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
142 p4_reply = (p_ip4 / p_payload)
144 rx = self.send_and_expect(self.pg1, p4*10, self.pg0)
146 self.validate(p[1], p4_reply)
148 err = '/err/ipip4-input/packets decapsulated'
149 cnt = self.statistics.dump_str(err)
150 self.assertEqual(cnt[err], 10)
152 # IPv4 tunnel to IPv6
153 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
154 p6 = (p_ether / IP(src=self.pg1.remote_ip4,
155 dst=self.pg0.local_ip4) / p_ip6 / p_payload)
156 p6_reply = (p_ip6 / p_payload)
158 rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
160 self.validate(p[1], p6_reply)
162 cnt = self.statistics.dump_str(err)
163 self.assertEqual(cnt[err], 20)
166 # Fragmentation / Reassembly and Re-fragmentation
168 rv = self.vapi.ip_reassembly_enable_disable(
169 sw_if_index=self.pg1.sw_if_index,
172 # Send lots of fragments, verify reassembled packet
173 frags, p4_reply = self.generate_frags(3131, 1400)
175 for i in range(0, 1000):
177 self.pg1.add_stream(f)
178 self.pg_enable_capture()
180 rx = self.pg0.get_capture(1000)
183 self.validate(p[1], p4_reply)
185 cnt = self.statistics.dump_str(err)
186 self.assertEqual(cnt[err], 1020)
190 for i in range(1, 90):
191 frags, p4_reply = self.generate_frags(i * 100, 1000)
194 self.pg_enable_capture()
195 self.pg1.add_stream(f)
197 rx = self.pg0.get_capture(89)
200 self.validate(p[1], r[i])
203 # Now try with re-fragmentation
205 # Send fragments to tunnel head-end, for the tunnel head end
206 # to reassemble and then refragment
208 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
209 frags, p4_reply = self.generate_frags(3123, 1200)
210 self.pg_enable_capture()
211 self.pg1.add_stream(frags)
213 rx = self.pg0.get_capture(6)
214 reass_pkt = reassemble(rx)
217 self.validate(reass_pkt, p4_reply)
219 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
220 frags, p4_reply = self.generate_frags(3123, 1200)
221 self.pg_enable_capture()
222 self.pg1.add_stream(frags)
224 rx = self.pg0.get_capture(2)
225 reass_pkt = reassemble(rx)
228 self.validate(reass_pkt, p4_reply)
230 def test_ipip6(self):
231 """ ip{v4,v6} over ip6 test """
232 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
233 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
234 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
235 p_payload = UDP(sport=1234, dport=1234)
238 rv = self.vapi.ipip_add_tunnel(
239 src_address=self.pg0.local_ip6n,
240 dst_address=self.pg1.remote_ip6n, tc_tos=255)
242 sw_if_index = rv.sw_if_index
244 self.vapi.sw_interface_set_flags(sw_if_index, 1)
245 self.vapi.sw_interface_set_unnumbered(
246 ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
248 # Add IPv4 and IPv6 routes via tunnel interface
249 ip4_via_tunnel = VppIpRoute(
250 self, "130.67.0.0", 16,
251 [VppRoutePath("0.0.0.0",
253 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
254 ip4_via_tunnel.add_vpp_config()
256 ip6_via_tunnel = VppIpRoute(
260 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
261 ip6_via_tunnel.add_vpp_config()
265 # IPv6 in to IPv6 tunnel
266 p6 = (p_ether / p_ip6 / p_payload)
267 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
270 p6_reply[1].hlim -= 1
271 rx = self.send_and_expect(self.pg0, p6*11, self.pg1)
273 self.validate(p[1], p6_reply)
275 # IPv4 in to IPv6 tunnel
276 p4 = (p_ether / p_ip4 / p_payload)
277 p4_reply = (IPv6(src=self.pg0.local_ip6,
278 dst=self.pg1.remote_ip6, hlim=63, tc=42) /
281 rx = self.send_and_expect(self.pg0, p4*11, self.pg1)
283 self.validate(p[1], p4_reply)
287 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
289 # IPv6 tunnel to IPv4
290 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
291 p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
292 dst=self.pg0.local_ip6) / p_ip4 / p_payload)
293 p4_reply = (p_ip4 / p_payload)
295 rx = self.send_and_expect(self.pg1, p4*11, self.pg0)
297 self.validate(p[1], p4_reply)
299 # IPv6 tunnel to IPv6
300 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
301 p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
302 dst=self.pg0.local_ip6) / p_ip6 / p_payload)
303 p6_reply = (p_ip6 / p_payload)
305 rx = self.send_and_expect(self.pg1, p6*11, self.pg0)
307 self.validate(p[1], p6_reply)
309 def test_ipip_create(self):
310 """ ipip create / delete interface test """
311 rv = self.vapi.ipip_add_tunnel(
312 src_address=inet_pton(AF_INET, '1.2.3.4'),
313 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
314 sw_if_index = rv.sw_if_index
315 self.vapi.ipip_del_tunnel(sw_if_index)
317 def test_ipip_vrf_create(self):
318 """ ipip create / delete interface VRF test """
320 t = VppIpTable(self, 20)
322 rv = self.vapi.ipip_add_tunnel(
323 src_address=inet_pton(AF_INET, '1.2.3.4'),
324 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
326 sw_if_index = rv.sw_if_index
327 self.vapi.ipip_del_tunnel(sw_if_index)
329 def payload(self, len):
333 if __name__ == '__main__':
334 unittest.main(testRunner=VppTestRunner)