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_route import VppIpRoute, VppRoutePath, DpoProto, VppIpTable
9 from socket import AF_INET, AF_INET6, inet_pton
12 """ Testipip is a subclass of VPPTestCase classes.
19 def reassemble(listoffragments):
20 buffer = StringIO.StringIO()
21 first = listoffragments[0]
23 for pkt in listoffragments:
24 buffer.seek(pkt[IP].frag*8)
25 buffer.write(pkt[IP].payload)
26 first.len = len(buffer.getvalue()) + 20
29 header = str(first[IP])[:20]
30 return first[IP].__class__(header + buffer.getvalue())
33 class TestIPIP(VppTestCase):
34 """ IPIP Test Case """
38 super(TestIPIP, cls).setUpClass()
39 cls.create_pg_interfaces(range(2))
40 cls.interfaces = list(cls.pg_interfaces)
43 super(TestIPIP, cls).setUp()
44 for i in cls.interfaces:
53 super(TestIPIP, self).tearDown()
55 for i in self.pg_interfaces:
60 def validate(self, rx, expected):
61 self.assertEqual(rx, expected.__class__(str(expected)))
63 def generate_frags(self, payload_length, fragment_size):
64 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
65 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
66 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
67 outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
69 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
70 frags = fragment(outer_ip4, fragment_size)
71 p4_reply = (p_ip4 / p_payload)
73 return frags, p4_reply
76 """ ip{v4,v6} over ip4 test """
77 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
78 p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
79 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
80 p_payload = UDP(sport=1234, dport=1234)
83 rv = self.vapi.ipip_add_tunnel(
84 src_address=self.pg0.local_ip4n,
85 dst_address=self.pg1.remote_ip4n,
86 is_ipv6=0, tc_tos=0xFF)
87 sw_if_index = rv.sw_if_index
89 # Set interface up and enable IP on it
90 self.vapi.sw_interface_set_flags(sw_if_index, 1)
91 self.vapi.sw_interface_set_unnumbered(
92 ip_sw_if_index=self.pg0.sw_if_index,
93 sw_if_index=sw_if_index)
95 # Add IPv4 and IPv6 routes via tunnel interface
96 ip4_via_tunnel = VppIpRoute(
97 self, "130.67.0.0", 16,
98 [VppRoutePath("0.0.0.0",
100 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
101 ip4_via_tunnel.add_vpp_config()
103 ip6_via_tunnel = VppIpRoute(
107 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
108 ip6_via_tunnel.add_vpp_config()
110 # IPv6 in to IPv4 tunnel
111 p6 = (p_ether / p_ip6 / p_payload)
113 p_inner_ip6.hlim -= 1
114 p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
115 proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload)
117 rx = self.send_and_expect(self.pg0, p6*10, self.pg1)
119 self.validate(p[1], p6_reply)
121 # IPv4 in to IPv4 tunnel
122 p4 = (p_ether / p_ip4 / p_payload)
125 p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
127 p_ip4_inner / p_payload)
130 rx = self.send_and_expect(self.pg0, p4*10, self.pg1)
132 self.validate(p[1], p4_reply)
135 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
137 # IPv4 tunnel to IPv4
138 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
139 p4 = (p_ether / IP(src=self.pg1.remote_ip4,
140 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
141 p4_reply = (p_ip4 / p_payload)
143 rx = self.send_and_expect(self.pg1, p4*10, self.pg0)
145 self.validate(p[1], p4_reply)
147 err = '/err/ipip4-input/packets decapsulated'
148 cnt = self.statistics.dump_str(err)
149 self.assertEqual(cnt[err], 10)
151 # IPv4 tunnel to IPv6
152 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
153 p6 = (p_ether / IP(src=self.pg1.remote_ip4,
154 dst=self.pg0.local_ip4) / p_ip6 / p_payload)
155 p6_reply = (p_ip6 / p_payload)
157 rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
159 self.validate(p[1], p6_reply)
161 cnt = self.statistics.dump_str(err)
162 self.assertEqual(cnt[err], 20)
165 # Fragmentation / Reassembly and Re-fragmentation
167 rv = self.vapi.ip_reassembly_enable_disable(
168 sw_if_index=self.pg1.sw_if_index,
171 # Send lots of fragments, verify reassembled packet
172 frags, p4_reply = self.generate_frags(3131, 1400)
174 for i in range(0, 1000):
176 self.pg1.add_stream(f)
177 self.pg_enable_capture()
179 rx = self.pg0.get_capture(1000)
182 self.validate(p[1], p4_reply)
184 cnt = self.statistics.dump_str(err)
185 self.assertEqual(cnt[err], 1020)
189 for i in range(1, 90):
190 frags, p4_reply = self.generate_frags(i * 100, 1000)
193 self.pg_enable_capture()
194 self.pg1.add_stream(f)
196 rx = self.pg0.get_capture(89)
199 self.validate(p[1], r[i])
202 # Now try with re-fragmentation
204 # Send fragments to tunnel head-end, for the tunnel head end
205 # to reassemble and then refragment
207 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
208 frags, p4_reply = self.generate_frags(3123, 1200)
209 self.pg_enable_capture()
210 self.pg1.add_stream(frags)
212 rx = self.pg0.get_capture(6)
213 reass_pkt = reassemble(rx)
216 self.validate(reass_pkt, p4_reply)
218 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
219 frags, p4_reply = self.generate_frags(3123, 1200)
220 self.pg_enable_capture()
221 self.pg1.add_stream(frags)
223 rx = self.pg0.get_capture(2)
224 reass_pkt = reassemble(rx)
227 self.validate(reass_pkt, p4_reply)
229 def test_ipip6(self):
230 """ ip{v4,v6} over ip6 test """
231 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
232 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
233 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
234 p_payload = UDP(sport=1234, dport=1234)
237 rv = self.vapi.ipip_add_tunnel(
238 src_address=self.pg0.local_ip6n,
239 dst_address=self.pg1.remote_ip6n, tc_tos=255)
241 sw_if_index = rv.sw_if_index
243 self.vapi.sw_interface_set_flags(sw_if_index, 1)
244 self.vapi.sw_interface_set_unnumbered(
245 ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
247 # Add IPv4 and IPv6 routes via tunnel interface
248 ip4_via_tunnel = VppIpRoute(
249 self, "130.67.0.0", 16,
250 [VppRoutePath("0.0.0.0",
252 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
253 ip4_via_tunnel.add_vpp_config()
255 ip6_via_tunnel = VppIpRoute(
259 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
260 ip6_via_tunnel.add_vpp_config()
264 # IPv6 in to IPv6 tunnel
265 p6 = (p_ether / p_ip6 / p_payload)
266 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
269 p6_reply[1].hlim -= 1
270 rx = self.send_and_expect(self.pg0, p6*11, self.pg1)
272 self.validate(p[1], p6_reply)
274 # IPv4 in to IPv6 tunnel
275 p4 = (p_ether / p_ip4 / p_payload)
276 p4_reply = (IPv6(src=self.pg0.local_ip6,
277 dst=self.pg1.remote_ip6, hlim=63, tc=42) /
280 rx = self.send_and_expect(self.pg0, p4*11, self.pg1)
282 self.validate(p[1], p4_reply)
286 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
288 # IPv6 tunnel to IPv4
289 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
290 p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
291 dst=self.pg0.local_ip6) / p_ip4 / p_payload)
292 p4_reply = (p_ip4 / p_payload)
294 rx = self.send_and_expect(self.pg1, p4*11, self.pg0)
296 self.validate(p[1], p4_reply)
298 # IPv6 tunnel to IPv6
299 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
300 p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
301 dst=self.pg0.local_ip6) / p_ip6 / p_payload)
302 p6_reply = (p_ip6 / p_payload)
304 rx = self.send_and_expect(self.pg1, p6*11, self.pg0)
306 self.validate(p[1], p6_reply)
308 def test_ipip_create(self):
309 """ ipip create / delete interface test """
310 rv = self.vapi.ipip_add_tunnel(
311 src_address=inet_pton(AF_INET, '1.2.3.4'),
312 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
313 sw_if_index = rv.sw_if_index
314 self.vapi.ipip_del_tunnel(sw_if_index)
316 def test_ipip_vrf_create(self):
317 """ ipip create / delete interface VRF test """
319 t = VppIpTable(self, 20)
321 rv = self.vapi.ipip_add_tunnel(
322 src_address=inet_pton(AF_INET, '1.2.3.4'),
323 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
325 sw_if_index = rv.sw_if_index
326 self.vapi.ipip_del_tunnel(sw_if_index)
328 def payload(self, len):
332 if __name__ == '__main__':
333 unittest.main(testRunner=VppTestRunner)