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 (
20 from template_ipsec import (
27 from test_ipsec_tun_if_esp import TemplateIpsecItf4
28 from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect, VppIpsecInterface
31 class VppLcpPair(VppObject):
32 def __init__(self, test, phy, host):
37 def add_vpp_config(self):
38 self._test.vapi.cli("test lcp add phy %s host %s" % (self.phy, self.host))
39 self._test.registry.register(self, self._test.logger)
42 def remove_vpp_config(self):
43 self._test.vapi.cli("test lcp del phy %s host %s" % (self.phy, self.host))
46 return "lcp:%d:%d" % (self.phy.sw_if_index, self.host.sw_if_index)
48 def query_vpp_config(self):
49 pairs = list(self._test.vapi.vpp.details_iter(self._test.vapi.lcp_itf_pair_get))
53 p.phy_sw_if_index == self.phy.sw_if_index
54 and p.host_sw_if_index == self.host.sw_if_index
60 class TestLinuxCP(VppTestCase):
61 """Linux Control Plane"""
63 extra_vpp_plugin_config = [
70 "linux_cp_unittest_plugin.so",
78 super(TestLinuxCP, cls).setUpClass()
81 def tearDownClass(cls):
82 super(TestLinuxCP, cls).tearDownClass()
85 super(TestLinuxCP, self).setUp()
87 # create 4 pg interfaces so we can create two pairs
88 self.create_pg_interfaces(range(4))
90 # create on ip4 and one ip6 pg tun
91 self.pg_interfaces += self.create_pg_ip4_interfaces(range(4, 5))
92 self.pg_interfaces += self.create_pg_ip6_interfaces(range(5, 6))
94 for i in self.pg_interfaces:
98 for i in self.pg_interfaces:
100 super(TestLinuxCP, self).tearDown()
102 def test_linux_cp_tap(self):
109 arp_opts = {"who-has": 1, "is-at": 2}
111 # create two pairs, wihch a bunch of hots on the phys
112 hosts = [self.pg0, self.pg1]
113 phys = [self.pg2, self.pg3]
118 phy.generate_remote_hosts(4)
119 phy.configure_ipv4_neighbors()
121 pair1 = VppLcpPair(self, phys[0], hosts[0]).add_vpp_config()
122 pair2 = VppLcpPair(self, phys[1], hosts[1]).add_vpp_config()
124 self.logger.info(self.vapi.cli("sh lcp adj verbose"))
125 self.logger.info(self.vapi.cli("sh lcp"))
132 for phy, host in zip(phys, hosts):
133 for j in range(N_HOSTS):
135 Ether(src=phy.local_mac, dst=phy.remote_hosts[j].mac)
136 / IP(src=phy.local_ip4, dst=phy.remote_hosts[j].ip4)
137 / UDP(sport=1234, dport=1234)
141 rxs = self.send_and_expect(host, [p], phy)
143 # verify packet is unchanged
145 self.assertEqual(p.show2(True), rx.show2(True))
147 # ARPs x-connect to phy
148 p = Ether(dst="ff:ff:ff:ff:ff:ff", src=phy.remote_hosts[j].mac) / ARP(
150 hwdst=phy.remote_hosts[j].mac,
153 pdst=phy.remote_hosts[j].ip4,
156 rxs = self.send_and_expect(host, [p], phy)
158 # verify packet is unchanged
160 self.assertEqual(p.show2(True), rx.show2(True))
163 for phy, host in zip(phys, hosts):
164 for j in range(N_HOSTS):
166 Ether(dst=phy.local_mac, src=phy.remote_hosts[j].mac)
167 / IP(dst=phy.local_ip4, src=phy.remote_hosts[j].ip4)
168 / UDP(sport=1234, dport=1234)
172 rxs = self.send_and_expect(phy, [p], host)
174 # verify packet is unchanged
176 self.assertEqual(p.show2(True), rx.show2(True))
178 # ARPs rx'd on the phy are sent to the host
179 p = Ether(dst="ff:ff:ff:ff:ff:ff", src=phy.remote_hosts[j].mac) / ARP(
181 hwsrc=phy.remote_hosts[j].mac,
184 psrc=phy.remote_hosts[j].ip4,
187 rxs = self.send_and_expect(phy, [p], host)
189 # verify packet is unchanged
191 self.assertEqual(p.show2(True), rx.show2(True))
197 def test_linux_cp_tun(self):
205 # create two pairs, wihch a bunch of hots on the phys
206 hosts = [self.pg4, self.pg5]
214 tun4 = VppIpIpTunInterface(
215 self, phy, phy.local_ip4, phy.remote_ip4
217 tun6 = VppIpIpTunInterface(
218 self, phy, phy.local_ip6, phy.remote_ip6
227 pair1 = VppLcpPair(self, tuns[0], hosts[0]).add_vpp_config()
228 pair2 = VppLcpPair(self, tuns[1], hosts[1]).add_vpp_config()
230 self.logger.info(self.vapi.cli("sh lcp adj verbose"))
231 self.logger.info(self.vapi.cli("sh lcp"))
232 self.logger.info(self.vapi.cli("sh ip punt redirect"))
239 p = IP(src=tun4.local_ip4, dst="2.2.2.2") / UDP(sport=1234, dport=1234) / Raw()
241 rxs = self.send_and_expect(self.pg4, p * N_PKTS, phy)
243 # verify inner packet is unchanged and has the tunnel encap
245 self.assertEqual(rx[Ether].dst, phy.remote_mac)
246 self.assertEqual(rx[IP].dst, phy.remote_ip4)
247 self.assertEqual(rx[IP].src, phy.local_ip4)
248 inner = IP(rx[IP].payload)
249 self.assertEqual(inner.src, tun4.local_ip4)
250 self.assertEqual(inner.dst, "2.2.2.2")
253 p = IPv6(src=tun6.local_ip6, dst="2::2") / UDP(sport=1234, dport=1234) / Raw()
255 rxs = self.send_and_expect(self.pg5, p * N_PKTS, phy)
257 # verify inner packet is unchanged and has the tunnel encap
259 self.assertEqual(rx[IPv6].dst, phy.remote_ip6)
260 self.assertEqual(rx[IPv6].src, phy.local_ip6)
261 inner = IPv6(rx[IPv6].payload)
262 self.assertEqual(inner.src, tun6.local_ip6)
263 self.assertEqual(inner.dst, "2::2")
267 Ether(dst=phy.local_mac, src=phy.remote_mac)
268 / IP(dst=phy.local_ip4, src=phy.remote_ip4)
269 / IP(dst=tun4.local_ip4, src=tun4.remote_ip4)
270 / UDP(sport=1234, dport=1234)
274 rxs = self.send_and_expect(phy, p * N_PKTS, self.pg4)
277 self.assertEqual(rx[IP].dst, tun4.local_ip4)
278 self.assertEqual(rx[IP].src, tun4.remote_ip4)
282 Ether(dst=phy.local_mac, src=phy.remote_mac)
283 / IPv6(dst=phy.local_ip6, src=phy.remote_ip6)
284 / IPv6(dst=tun6.local_ip6, src=tun6.remote_ip6)
285 / UDP(sport=1234, dport=1234)
289 rxs = self.send_and_expect(phy, p * N_PKTS, self.pg5)
292 self.assertEqual(rx[IPv6].dst, tun6.local_ip6)
293 self.assertEqual(rx[IPv6].src, tun6.remote_ip6)
303 class TestLinuxCPIpsec(TemplateIpsec, TemplateIpsecItf4, IpsecTun4):
304 """IPsec Interface IPv4"""
306 extra_vpp_plugin_config = [
308 "linux_cp_plugin.so",
313 "linux_cp_unittest_plugin.so",
320 super(TestLinuxCPIpsec, self).setUp()
322 self.tun_if = self.pg0
323 self.pg_interfaces += self.create_pg_ip4_interfaces(range(3, 4))
324 self.pg_interfaces += self.create_pg_ip6_interfaces(range(4, 5))
327 super(TestLinuxCPIpsec, self).tearDown()
329 def verify_encrypted(self, p, sa, rxs):
333 self.assertEqual(rx[UDP].dport, 4500)
334 self.assert_packet_checksums_valid(rx)
335 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
338 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
339 if not decrypt_pkt.haslayer(IP):
340 decrypt_pkt = IP(decrypt_pkt[Raw].load)
341 if rx_ip.proto == socket.IPPROTO_ESP:
342 self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
343 decrypt_pkts.append(decrypt_pkt)
344 self.assert_equal(decrypt_pkt.src, p.tun_if.local_ip4)
345 self.assert_equal(decrypt_pkt.dst, p.tun_if.remote_ip4)
347 self.logger.debug(ppp("Unexpected packet:", rx))
349 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
353 pkts = reassemble4(decrypt_pkts)
355 self.assert_packet_checksums_valid(pkt)
357 def verify_decrypted(self, p, rxs):
360 self.assert_equal(rx[IP].src, p.tun_if.remote_ip4)
361 self.assert_equal(rx[IP].dst, p.tun_if.local_ip4)
362 self.assert_packet_checksums_valid(rx)
364 def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=54):
366 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
369 / UDP(sport=1111, dport=2222)
370 / Raw(b"X" * payload_size)
372 for i in range(count)
375 def test_linux_cp_ipsec4_tun(self):
376 """Linux CP Ipsec TUN"""
383 # the pg that paris with the tunnel
386 # tunnel and protection setup
389 self.config_network(p)
390 self.config_sa_tun(p, self.pg0.local_ip4, self.pg0.remote_ip4)
391 self.config_protect(p)
393 pair = VppLcpPair(self, p.tun_if, self.host).add_vpp_config()
395 self.logger.info(self.vapi.cli("sh int addr"))
396 self.logger.info(self.vapi.cli("sh lcp"))
397 self.logger.info(self.vapi.cli("sh ip punt redirect"))
405 IP(src=p.tun_if.local_ip4, dst=p.tun_if.remote_ip4)
406 / UDP(sport=1234, dport=1234)
410 rxs = self.send_and_expect(self.host, pkt * N_PKTS, self.tun_if)
411 self.verify_encrypted(p, p.vpp_tun_sa, rxs)
414 pkts = self.gen_encrypt_pkts(
418 src=p.tun_if.remote_ip4,
419 dst=p.tun_if.local_ip4,
422 rxs = self.send_and_expect(self.tun_if, pkts, self.host)
423 self.verify_decrypted(p, rxs)
426 pair.remove_vpp_config()
427 self.unconfig_protect(p)
429 self.unconfig_network(p)
432 if __name__ == "__main__":
433 unittest.main(testRunner=VppTestRunner)