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 = self.statistics.get_counter(
149 '/err/ipip4-input/packets decapsulated')
150 self.assertEqual(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 err = self.statistics.get_counter(
163 '/err/ipip4-input/packets decapsulated')
164 self.assertEqual(err, 20)
167 # Fragmentation / Reassembly and Re-fragmentation
169 rv = self.vapi.ip_reassembly_enable_disable(
170 sw_if_index=self.pg1.sw_if_index,
173 # Send lots of fragments, verify reassembled packet
174 frags, p4_reply = self.generate_frags(3131, 1400)
176 for i in range(0, 1000):
178 self.pg1.add_stream(f)
179 self.pg_enable_capture()
181 rx = self.pg0.get_capture(1000)
184 self.validate(p[1], p4_reply)
186 err = self.statistics.get_counter(
187 '/err/ipip4-input/packets decapsulated')
188 self.assertEqual(err, 1020)
192 for i in range(1, 90):
193 frags, p4_reply = self.generate_frags(i * 100, 1000)
196 self.pg_enable_capture()
197 self.pg1.add_stream(f)
199 rx = self.pg0.get_capture(89)
202 self.validate(p[1], r[i])
205 # Now try with re-fragmentation
207 # Send fragments to tunnel head-end, for the tunnel head end
208 # to reassemble and then refragment
210 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
211 frags, p4_reply = self.generate_frags(3123, 1200)
212 self.pg_enable_capture()
213 self.pg1.add_stream(frags)
215 rx = self.pg0.get_capture(6)
216 reass_pkt = reassemble(rx)
219 self.validate(reass_pkt, p4_reply)
221 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
222 frags, p4_reply = self.generate_frags(3123, 1200)
223 self.pg_enable_capture()
224 self.pg1.add_stream(frags)
226 rx = self.pg0.get_capture(2)
227 reass_pkt = reassemble(rx)
230 self.validate(reass_pkt, p4_reply)
232 def test_ipip6(self):
233 """ ip{v4,v6} over ip6 test """
234 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
235 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
236 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
237 p_payload = UDP(sport=1234, dport=1234)
240 rv = self.vapi.ipip_add_tunnel(
241 src_address=self.pg0.local_ip6n,
242 dst_address=self.pg1.remote_ip6n, tc_tos=255)
244 sw_if_index = rv.sw_if_index
246 self.vapi.sw_interface_set_flags(sw_if_index, 1)
247 self.vapi.sw_interface_set_unnumbered(
248 ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
250 # Add IPv4 and IPv6 routes via tunnel interface
251 ip4_via_tunnel = VppIpRoute(
252 self, "130.67.0.0", 16,
253 [VppRoutePath("0.0.0.0",
255 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
256 ip4_via_tunnel.add_vpp_config()
258 ip6_via_tunnel = VppIpRoute(
262 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
263 ip6_via_tunnel.add_vpp_config()
267 # IPv6 in to IPv6 tunnel
268 p6 = (p_ether / p_ip6 / p_payload)
269 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
272 p6_reply[1].hlim -= 1
273 rx = self.send_and_expect(self.pg0, p6*11, self.pg1)
275 self.validate(p[1], p6_reply)
277 # IPv4 in to IPv6 tunnel
278 p4 = (p_ether / p_ip4 / p_payload)
279 p4_reply = (IPv6(src=self.pg0.local_ip6,
280 dst=self.pg1.remote_ip6, hlim=63, tc=42) /
283 rx = self.send_and_expect(self.pg0, p4*11, self.pg1)
285 self.validate(p[1], p4_reply)
289 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
291 # IPv6 tunnel to IPv4
292 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
293 p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
294 dst=self.pg0.local_ip6) / p_ip4 / p_payload)
295 p4_reply = (p_ip4 / p_payload)
297 rx = self.send_and_expect(self.pg1, p4*11, self.pg0)
299 self.validate(p[1], p4_reply)
301 # IPv6 tunnel to IPv6
302 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
303 p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
304 dst=self.pg0.local_ip6) / p_ip6 / p_payload)
305 p6_reply = (p_ip6 / p_payload)
307 rx = self.send_and_expect(self.pg1, p6*11, self.pg0)
309 self.validate(p[1], p6_reply)
311 def test_ipip_create(self):
312 """ ipip create / delete interface test """
313 rv = self.vapi.ipip_add_tunnel(
314 src_address=inet_pton(AF_INET, '1.2.3.4'),
315 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
316 sw_if_index = rv.sw_if_index
317 self.vapi.ipip_del_tunnel(sw_if_index)
319 def test_ipip_vrf_create(self):
320 """ ipip create / delete interface VRF test """
322 t = VppIpTable(self, 20)
324 rv = self.vapi.ipip_add_tunnel(
325 src_address=inet_pton(AF_INET, '1.2.3.4'),
326 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
328 sw_if_index = rv.sw_if_index
329 self.vapi.ipip_del_tunnel(sw_if_index)
331 def payload(self, len):
335 if __name__ == '__main__':
336 unittest.main(testRunner=VppTestRunner)