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 # IPv4 tunnel to IPv6
148 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
149 p6 = (p_ether / IP(src=self.pg1.remote_ip4,
150 dst=self.pg0.local_ip4) / p_ip6 / p_payload)
151 p6_reply = (p_ip6 / p_payload)
153 rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
155 self.validate(p[1], p6_reply)
158 # Fragmentation / Reassembly and Re-fragmentation
160 rv = self.vapi.ip_reassembly_enable_disable(
161 sw_if_index=self.pg1.sw_if_index,
164 # Send lots of fragments, verify reassembled packet
165 frags, p4_reply = self.generate_frags(3131, 1400)
167 for i in range(0, 1000):
169 self.pg1.add_stream(f)
170 self.pg_enable_capture()
172 rx = self.pg0.get_capture(1000)
175 self.validate(p[1], p4_reply)
179 for i in range(1, 90):
180 frags, p4_reply = self.generate_frags(i * 100, 1000)
183 self.pg_enable_capture()
184 self.pg1.add_stream(f)
186 rx = self.pg0.get_capture(89)
189 self.validate(p[1], r[i])
192 # Now try with re-fragmentation
194 # Send fragments to tunnel head-end, for the tunnel head end
195 # to reassemble and then refragment
197 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
198 frags, p4_reply = self.generate_frags(3123, 1200)
199 self.pg_enable_capture()
200 self.pg1.add_stream(frags)
202 rx = self.pg0.get_capture(6)
203 reass_pkt = reassemble(rx)
206 self.validate(reass_pkt, p4_reply)
208 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 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(2)
214 reass_pkt = reassemble(rx)
217 self.validate(reass_pkt, p4_reply)
219 def test_ipip6(self):
220 """ ip{v4,v6} over ip6 test """
221 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
222 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
223 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
224 p_payload = UDP(sport=1234, dport=1234)
227 rv = self.vapi.ipip_add_tunnel(
228 src_address=self.pg0.local_ip6n,
229 dst_address=self.pg1.remote_ip6n, tc_tos=255)
231 sw_if_index = rv.sw_if_index
233 self.vapi.sw_interface_set_flags(sw_if_index, 1)
234 self.vapi.sw_interface_set_unnumbered(
235 ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
237 # Add IPv4 and IPv6 routes via tunnel interface
238 ip4_via_tunnel = VppIpRoute(
239 self, "130.67.0.0", 16,
240 [VppRoutePath("0.0.0.0",
242 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
243 ip4_via_tunnel.add_vpp_config()
245 ip6_via_tunnel = VppIpRoute(
249 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
250 ip6_via_tunnel.add_vpp_config()
254 # IPv6 in to IPv6 tunnel
255 p6 = (p_ether / p_ip6 / p_payload)
256 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
259 p6_reply[1].hlim -= 1
260 rx = self.send_and_expect(self.pg0, p6*11, self.pg1)
262 self.validate(p[1], p6_reply)
264 # IPv4 in to IPv6 tunnel
265 p4 = (p_ether / p_ip4 / p_payload)
266 p4_reply = (IPv6(src=self.pg0.local_ip6,
267 dst=self.pg1.remote_ip6, hlim=63, tc=42) /
270 rx = self.send_and_expect(self.pg0, p4*11, self.pg1)
272 self.validate(p[1], p4_reply)
276 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
278 # IPv6 tunnel to IPv4
279 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
280 p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
281 dst=self.pg0.local_ip6) / p_ip4 / p_payload)
282 p4_reply = (p_ip4 / p_payload)
284 rx = self.send_and_expect(self.pg1, p4*11, self.pg0)
286 self.validate(p[1], p4_reply)
288 # IPv6 tunnel to IPv6
289 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
290 p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
291 dst=self.pg0.local_ip6) / p_ip6 / p_payload)
292 p6_reply = (p_ip6 / p_payload)
294 rx = self.send_and_expect(self.pg1, p6*11, self.pg0)
296 self.validate(p[1], p6_reply)
298 def test_ipip_create(self):
299 """ ipip create / delete interface test """
300 rv = self.vapi.ipip_add_tunnel(
301 src_address=inet_pton(AF_INET, '1.2.3.4'),
302 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
303 sw_if_index = rv.sw_if_index
304 self.vapi.ipip_del_tunnel(sw_if_index)
306 def test_ipip_vrf_create(self):
307 """ ipip create / delete interface VRF test """
309 t = VppIpTable(self, 20)
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,
315 sw_if_index = rv.sw_if_index
316 self.vapi.ipip_del_tunnel(sw_if_index)
318 def payload(self, len):
322 if __name__ == '__main__':
323 unittest.main(testRunner=VppTestRunner)