5 from scapy.layers.inet import IP, UDP
6 from scapy.layers.inet6 import IPv6, Raw
7 from scapy.layers.l2 import Ether, ARP, Dot1Q
9 from util import reassemble4
10 from vpp_object import VppObject
11 from framework import VppTestCase, VppTestRunner
12 from vpp_ipip_tun_interface import VppIpIpTunInterface
13 from template_ipsec import TemplateIpsec, IpsecTun4Tests, \
14 IpsecTun4, mk_scapy_crypt_key, config_tun_params
15 from template_ipsec import TemplateIpsec, IpsecTun4Tests, \
16 IpsecTun4, mk_scapy_crypt_key, config_tun_params
17 from test_ipsec_tun_if_esp import TemplateIpsecItf4
18 from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect, VppIpsecInterface
21 class VppLcpPair(VppObject):
22 def __init__(self, test, phy, host):
27 def add_vpp_config(self):
28 self._test.vapi.cli("test lcp add phy %s host %s" %
29 (self.phy, self.host))
30 self._test.registry.register(self, self._test.logger)
33 def remove_vpp_config(self):
34 self._test.vapi.cli("test lcp del phy %s host %s" %
35 (self.phy, self.host))
38 return "lcp:%d:%d" % (self.phy.sw_if_index,
39 self.host.sw_if_index)
41 def query_vpp_config(self):
42 pairs = list(self._test.vapi.vpp.details_iter(
43 self._test.vapi.lcp_itf_pair_get))
46 if p.phy_sw_if_index == self.phy.sw_if_index and \
47 p.host_sw_if_index == self.host.sw_if_index:
52 class TestLinuxCP(VppTestCase):
53 """ Linux Control Plane """
55 extra_vpp_plugin_config = ["plugin",
59 "linux_cp_unittest_plugin.so",
64 super(TestLinuxCP, cls).setUpClass()
67 def tearDownClass(cls):
68 super(TestLinuxCP, cls).tearDownClass()
71 super(TestLinuxCP, self).setUp()
73 # create 4 pg interfaces so we can create two pairs
74 self.create_pg_interfaces(range(4))
76 # create on ip4 and one ip6 pg tun
77 self.pg_interfaces += self.create_pg_ip4_interfaces(range(4, 5))
78 self.pg_interfaces += self.create_pg_ip6_interfaces(range(5, 6))
80 for i in self.pg_interfaces:
84 for i in self.pg_interfaces:
86 super(TestLinuxCP, self).tearDown()
88 def test_linux_cp_tap(self):
95 arp_opts = {"who-has": 1, "is-at": 2}
97 # create two pairs, wihch a bunch of hots on the phys
98 hosts = [self.pg0, self.pg1]
99 phys = [self.pg2, self.pg3]
104 phy.generate_remote_hosts(4)
105 phy.configure_ipv4_neighbors()
107 pair1 = VppLcpPair(self, phys[0], hosts[0]).add_vpp_config()
108 pair2 = VppLcpPair(self, phys[1], hosts[1]).add_vpp_config()
110 self.logger.info(self.vapi.cli("sh lcp adj verbose"))
111 self.logger.info(self.vapi.cli("sh lcp"))
118 for phy, host in zip(phys, hosts):
119 for j in range(N_HOSTS):
120 p = (Ether(src=phy.local_mac,
121 dst=phy.remote_hosts[j].mac) /
122 IP(src=phy.local_ip4,
123 dst=phy.remote_hosts[j].ip4) /
124 UDP(sport=1234, dport=1234) /
127 rxs = self.send_and_expect(host, [p], phy)
129 # verify packet is unchanged
131 self.assertEqual(p.show2(True), rx.show2(True))
133 # ARPs x-connect to phy
134 p = (Ether(dst="ff:ff:ff:ff:ff:ff",
135 src=phy.remote_hosts[j].mac) /
137 hwdst=phy.remote_hosts[j].mac,
140 pdst=phy.remote_hosts[j].ip4))
142 rxs = self.send_and_expect(host, [p], phy)
144 # verify packet is unchanged
146 self.assertEqual(p.show2(True), rx.show2(True))
149 for phy, host in zip(phys, hosts):
150 for j in range(N_HOSTS):
151 p = (Ether(dst=phy.local_mac,
152 src=phy.remote_hosts[j].mac) /
153 IP(dst=phy.local_ip4,
154 src=phy.remote_hosts[j].ip4) /
155 UDP(sport=1234, dport=1234) /
158 rxs = self.send_and_expect(phy, [p], host)
160 # verify packet is unchanged
162 self.assertEqual(p.show2(True), rx.show2(True))
164 # ARPs rx'd on the phy are sent to the host
165 p = (Ether(dst="ff:ff:ff:ff:ff:ff",
166 src=phy.remote_hosts[j].mac) /
168 hwsrc=phy.remote_hosts[j].mac,
171 psrc=phy.remote_hosts[j].ip4))
173 rxs = self.send_and_expect(phy, [p], host)
175 # verify packet is unchanged
177 self.assertEqual(p.show2(True), rx.show2(True))
183 def test_linux_cp_tun(self):
191 # create two pairs, wihch a bunch of hots on the phys
192 hosts = [self.pg4, self.pg5]
200 tun4 = VppIpIpTunInterface(
204 phy.remote_ip4).add_vpp_config()
205 tun6 = VppIpIpTunInterface(
209 phy.remote_ip6).add_vpp_config()
217 pair1 = VppLcpPair(self, tuns[0], hosts[0]).add_vpp_config()
218 pair2 = VppLcpPair(self, tuns[1], hosts[1]).add_vpp_config()
220 self.logger.info(self.vapi.cli("sh lcp adj verbose"))
221 self.logger.info(self.vapi.cli("sh lcp"))
222 self.logger.info(self.vapi.cli("sh ip punt redirect"))
229 p = (IP(src=tun4.local_ip4, dst="2.2.2.2") /
230 UDP(sport=1234, dport=1234) /
233 rxs = self.send_and_expect(self.pg4, p * N_PKTS, phy)
235 # verify inner packet is unchanged and has the tunnel encap
237 self.assertEqual(rx[Ether].dst, phy.remote_mac)
238 self.assertEqual(rx[IP].dst, phy.remote_ip4)
239 self.assertEqual(rx[IP].src, phy.local_ip4)
240 inner = IP(rx[IP].payload)
241 self.assertEqual(inner.src, tun4.local_ip4)
242 self.assertEqual(inner.dst, "2.2.2.2")
245 p = (IPv6(src=tun6.local_ip6, dst="2::2") /
246 UDP(sport=1234, dport=1234) /
249 rxs = self.send_and_expect(self.pg5, p * N_PKTS, phy)
251 # verify inner packet is unchanged and has the tunnel encap
253 self.assertEqual(rx[IPv6].dst, phy.remote_ip6)
254 self.assertEqual(rx[IPv6].src, phy.local_ip6)
255 inner = IPv6(rx[IPv6].payload)
256 self.assertEqual(inner.src, tun6.local_ip6)
257 self.assertEqual(inner.dst, "2::2")
260 p = (Ether(dst=phy.local_mac, src=phy.remote_mac) /
261 IP(dst=phy.local_ip4, src=phy.remote_ip4) /
262 IP(dst=tun4.local_ip4, src=tun4.remote_ip4) /
263 UDP(sport=1234, dport=1234) /
266 rxs = self.send_and_expect(phy, p * N_PKTS, self.pg4)
269 self.assertEqual(rx[IP].dst, tun4.local_ip4)
270 self.assertEqual(rx[IP].src, tun4.remote_ip4)
273 p = (Ether(dst=phy.local_mac, src=phy.remote_mac) /
274 IPv6(dst=phy.local_ip6, src=phy.remote_ip6) /
275 IPv6(dst=tun6.local_ip6, src=tun6.remote_ip6) /
276 UDP(sport=1234, dport=1234) /
279 rxs = self.send_and_expect(phy, p * N_PKTS, self.pg5)
282 self.assertEqual(rx[IPv6].dst, tun6.local_ip6)
283 self.assertEqual(rx[IPv6].src, tun6.remote_ip6)
293 class TestLinuxCPIpsec(TemplateIpsec,
296 """ IPsec Interface IPv4 """
298 extra_vpp_plugin_config = ["plugin",
299 "linux_cp_plugin.so",
302 "linux_cp_unittest_plugin.so",
306 super(TestLinuxCPIpsec, self).setUp()
308 self.tun_if = self.pg0
309 self.pg_interfaces += self.create_pg_ip4_interfaces(range(3, 4))
310 self.pg_interfaces += self.create_pg_ip6_interfaces(range(4, 5))
313 super(TestLinuxCPIpsec, self).tearDown()
315 def verify_encrypted(self, p, sa, rxs):
319 self.assertEqual(rx[UDP].dport, 4500)
320 self.assert_packet_checksums_valid(rx)
321 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
324 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
325 if not decrypt_pkt.haslayer(IP):
326 decrypt_pkt = IP(decrypt_pkt[Raw].load)
327 if rx_ip.proto == socket.IPPROTO_ESP:
328 self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
329 decrypt_pkts.append(decrypt_pkt)
330 self.assert_equal(decrypt_pkt.src, p.tun_if.local_ip4)
331 self.assert_equal(decrypt_pkt.dst, p.tun_if.remote_ip4)
333 self.logger.debug(ppp("Unexpected packet:", rx))
335 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
339 pkts = reassemble4(decrypt_pkts)
341 self.assert_packet_checksums_valid(pkt)
343 def verify_decrypted(self, p, rxs):
346 self.assert_equal(rx[IP].src, p.tun_if.remote_ip4)
347 self.assert_equal(rx[IP].dst, p.tun_if.local_ip4)
348 self.assert_packet_checksums_valid(rx)
350 def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
352 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
353 sa.encrypt(IP(src=src, dst=dst) /
354 UDP(sport=1111, dport=2222) /
355 Raw(b'X' * payload_size))
356 for i in range(count)]
358 def test_linux_cp_ipsec4_tun(self):
359 """ Linux CP Ipsec TUN """
366 # the pg that paris with the tunnel
369 # tunnel and protection setup
372 self.config_network(p)
373 self.config_sa_tun(p,
376 self.config_protect(p)
378 pair = VppLcpPair(self, p.tun_if, self.host).add_vpp_config()
380 self.logger.info(self.vapi.cli("sh int addr"))
381 self.logger.info(self.vapi.cli("sh lcp"))
382 self.logger.info(self.vapi.cli("sh ip punt redirect"))
389 pkt = (IP(src=p.tun_if.local_ip4,
390 dst=p.tun_if.remote_ip4) /
391 UDP(sport=1234, dport=1234) /
394 rxs = self.send_and_expect(self.host, pkt * N_PKTS, self.tun_if)
395 self.verify_encrypted(p, p.vpp_tun_sa, rxs)
398 pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
399 src=p.tun_if.remote_ip4,
400 dst=p.tun_if.local_ip4,
402 rxs = self.send_and_expect(self.tun_if, pkts, self.host)
403 self.verify_decrypted(p, rxs)
406 pair.remove_vpp_config()
407 self.unconfig_protect(p)
409 self.unconfig_network(p)
412 if __name__ == '__main__':
413 unittest.main(testRunner=VppTestRunner)