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
7 from framework import VppTestCase, VppTestRunner
8 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
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)))
64 """ ip{v4,v6} over ip4 test """
65 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
66 p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
67 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
68 p_payload = UDP(sport=1234, dport=1234)
71 rv = self.vapi.ipip_add_tunnel(
72 src_address=self.pg0.local_ip4n,
73 dst_address=self.pg1.remote_ip4n,
74 is_ipv6=0, tc_tos=0xFF)
75 sw_if_index = rv.sw_if_index
77 # Set interface up and enable IP on it
78 self.vapi.sw_interface_set_flags(sw_if_index, 1)
79 self.vapi.sw_interface_set_unnumbered(
80 ip_sw_if_index=self.pg0.sw_if_index,
81 sw_if_index=sw_if_index)
83 # Add IPv4 and IPv6 routes via tunnel interface
84 ip4_via_tunnel = VppIpRoute(
85 self, "130.67.0.0", 16,
86 [VppRoutePath("0.0.0.0",
88 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
89 ip4_via_tunnel.add_vpp_config()
91 ip6_via_tunnel = VppIpRoute(
95 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
96 ip6_via_tunnel.add_vpp_config()
98 # IPv6 in to IPv4 tunnel
99 p6 = (p_ether / p_ip6 / p_payload)
101 p_inner_ip6.hlim -= 1
102 p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
103 proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload)
105 rx = self.send_and_expect(self.pg0, p6*10, self.pg1)
107 self.validate(p[1], p6_reply)
109 # IPv4 in to IPv4 tunnel
110 p4 = (p_ether / p_ip4 / p_payload)
113 p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
115 p_ip4_inner / p_payload)
118 rx = self.send_and_expect(self.pg0, p4*10, self.pg1)
120 self.validate(p[1], p4_reply)
123 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
125 # IPv4 tunnel to IPv4
126 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
127 p4 = (p_ether / IP(src=self.pg1.remote_ip4,
128 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
129 p4_reply = (p_ip4 / p_payload)
131 rx = self.send_and_expect(self.pg1, p4*10, self.pg0)
133 self.validate(p[1], p4_reply)
135 # IPv4 tunnel to IPv6
136 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
137 p6 = (p_ether / IP(src=self.pg1.remote_ip4,
138 dst=self.pg0.local_ip4) / p_ip6 / p_payload)
139 p6_reply = (p_ip6 / p_payload)
141 rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
143 self.validate(p[1], p6_reply)
145 # Fragmentation / Reassembly and Re-fragmentation
146 rv = self.vapi.ip_reassembly_enable_disable(
147 sw_if_index=self.pg1.sw_if_index,
150 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
151 p_payload = UDP(sport=1234, dport=1234) / self.payload(3123)
152 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
153 outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
154 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
155 frags = fragment(outer_ip4, 1400)
156 p4_reply = (p_ip4 / p_payload)
159 self.pg_enable_capture()
160 self.pg1.add_stream(frags)
162 rx = self.pg0.get_capture(1)
164 self.validate(p[1], p4_reply)
166 # Now try with re-fragmentation
167 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
168 self.pg_enable_capture()
169 self.pg1.add_stream(frags)
171 rx = self.pg0.get_capture(6)
172 reass_pkt = reassemble(rx)
175 self.validate(reass_pkt, p4_reply)
177 def test_ipip6(self):
178 """ ip{v4,v6} over ip6 test """
179 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
180 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
181 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
182 p_payload = UDP(sport=1234, dport=1234)
185 rv = self.vapi.ipip_add_tunnel(
186 src_address=self.pg0.local_ip6n,
187 dst_address=self.pg1.remote_ip6n, tc_tos=255)
189 sw_if_index = rv.sw_if_index
191 self.vapi.sw_interface_set_flags(sw_if_index, 1)
192 self.vapi.sw_interface_set_unnumbered(
193 ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
195 # Add IPv4 and IPv6 routes via tunnel interface
196 ip4_via_tunnel = VppIpRoute(
197 self, "130.67.0.0", 16,
198 [VppRoutePath("0.0.0.0",
200 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
201 ip4_via_tunnel.add_vpp_config()
203 ip6_via_tunnel = VppIpRoute(
207 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
208 ip6_via_tunnel.add_vpp_config()
212 # IPv6 in to IPv6 tunnel
213 p6 = (p_ether / p_ip6 / p_payload)
214 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
217 p6_reply[1].hlim -= 1
218 rx = self.send_and_expect(self.pg0, p6*11, self.pg1)
220 self.validate(p[1], p6_reply)
222 # IPv4 in to IPv6 tunnel
223 p4 = (p_ether / p_ip4 / p_payload)
224 p4_reply = (IPv6(src=self.pg0.local_ip6,
225 dst=self.pg1.remote_ip6, hlim=63, tc=42) /
228 rx = self.send_and_expect(self.pg0, p4*11, self.pg1)
230 self.validate(p[1], p4_reply)
234 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
236 # IPv6 tunnel to IPv4
237 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
238 p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
239 dst=self.pg0.local_ip6) / p_ip4 / p_payload)
240 p4_reply = (p_ip4 / p_payload)
242 rx = self.send_and_expect(self.pg1, p4*11, self.pg0)
244 self.validate(p[1], p4_reply)
246 # IPv6 tunnel to IPv6
247 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
248 p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
249 dst=self.pg0.local_ip6) / p_ip6 / p_payload)
250 p6_reply = (p_ip6 / p_payload)
252 rx = self.send_and_expect(self.pg1, p6*11, self.pg0)
254 self.validate(p[1], p6_reply)
256 def test_ipip_create(self):
257 """ ipip create / delete interface test """
258 rv = self.vapi.ipip_add_tunnel(
259 src_address=inet_pton(AF_INET, '1.2.3.4'),
260 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
261 sw_if_index = rv.sw_if_index
262 self.vapi.ipip_del_tunnel(sw_if_index)
264 def payload(self, len):
268 if __name__ == '__main__':
269 unittest.main(testRunner=VppTestRunner)