5 from scapy.layers.inet import IP, UDP
6 from scapy.layers.inet6 import IPv6, Raw
7 from scapy.layers.l2 import Ether, ARP
9 from util import reassemble4
10 from vpp_object import VppObject
11 from framework import VppTestCase
12 from asfframework import VppTestRunner
13 from vpp_ipip_tun_interface import VppIpIpTunInterface
14 from template_ipsec import (
18 from template_ipsec import (
22 from test_ipsec_tun_if_esp import TemplateIpsecItf4
25 class VppLcpPair(VppObject):
26 def __init__(self, test, phy, host):
31 def add_vpp_config(self):
32 self._test.vapi.cli("test lcp add phy %s host %s" % (self.phy, self.host))
33 self._test.registry.register(self, self._test.logger)
36 def remove_vpp_config(self):
37 self._test.vapi.cli("test lcp del phy %s host %s" % (self.phy, self.host))
40 return "lcp:%d:%d" % (self.phy.sw_if_index, self.host.sw_if_index)
42 def query_vpp_config(self):
43 pairs = list(self._test.vapi.vpp.details_iter(self._test.vapi.lcp_itf_pair_get))
47 p.phy_sw_if_index == self.phy.sw_if_index
48 and p.host_sw_if_index == self.host.sw_if_index
54 class TestLinuxCP(VppTestCase):
55 """Linux Control Plane"""
57 extra_vpp_plugin_config = [
64 "linux_cp_unittest_plugin.so",
72 super(TestLinuxCP, cls).setUpClass()
75 def tearDownClass(cls):
76 super(TestLinuxCP, cls).tearDownClass()
79 super(TestLinuxCP, self).setUp()
81 # create 4 pg interfaces so we can create two pairs
82 self.create_pg_interfaces(range(4))
84 # create on ip4 and one ip6 pg tun
85 self.pg_interfaces += self.create_pg_ip4_interfaces(range(4, 5))
86 self.pg_interfaces += self.create_pg_ip6_interfaces(range(5, 6))
88 for i in self.pg_interfaces:
92 for i in self.pg_interfaces:
94 super(TestLinuxCP, self).tearDown()
96 def test_linux_cp_tap(self):
103 arp_opts = {"who-has": 1, "is-at": 2}
105 # create two pairs, wihch a bunch of hots on the phys
106 hosts = [self.pg0, self.pg1]
107 phys = [self.pg2, self.pg3]
112 phy.generate_remote_hosts(4)
113 phy.configure_ipv4_neighbors()
115 pair1 = VppLcpPair(self, phys[0], hosts[0]).add_vpp_config()
116 pair2 = VppLcpPair(self, phys[1], hosts[1]).add_vpp_config()
118 self.logger.info(self.vapi.cli("sh lcp adj verbose"))
119 self.logger.info(self.vapi.cli("sh lcp"))
126 for phy, host in zip(phys, hosts):
127 for j in range(N_HOSTS):
129 Ether(src=phy.local_mac, dst=phy.remote_hosts[j].mac)
130 / IP(src=phy.local_ip4, dst=phy.remote_hosts[j].ip4)
131 / UDP(sport=1234, dport=1234)
135 rxs = self.send_and_expect(host, [p], phy)
137 # verify packet is unchanged
139 self.assertEqual(p.show2(True), rx.show2(True))
141 # ARPs x-connect to phy
142 p = Ether(dst="ff:ff:ff:ff:ff:ff", src=phy.remote_hosts[j].mac) / ARP(
144 hwdst=phy.remote_hosts[j].mac,
147 pdst=phy.remote_hosts[j].ip4,
150 rxs = self.send_and_expect(host, [p], phy)
152 # verify packet is unchanged
154 self.assertEqual(p.show2(True), rx.show2(True))
157 for phy, host in zip(phys, hosts):
158 for j in range(N_HOSTS):
160 Ether(dst=phy.local_mac, src=phy.remote_hosts[j].mac)
161 / IP(dst=phy.local_ip4, src=phy.remote_hosts[j].ip4)
162 / UDP(sport=1234, dport=1234)
166 rxs = self.send_and_expect(phy, [p], host)
168 # verify packet is unchanged
170 self.assertEqual(p.show2(True), rx.show2(True))
172 # ARPs rx'd on the phy are sent to the host
173 p = Ether(dst="ff:ff:ff:ff:ff:ff", src=phy.remote_hosts[j].mac) / ARP(
175 hwsrc=phy.remote_hosts[j].mac,
178 psrc=phy.remote_hosts[j].ip4,
181 rxs = self.send_and_expect(phy, [p], host)
183 # verify packet is unchanged
185 self.assertEqual(p.show2(True), rx.show2(True))
191 def test_linux_cp_tun(self):
199 # create two pairs, wihch a bunch of hots on the phys
200 hosts = [self.pg4, self.pg5]
208 tun4 = VppIpIpTunInterface(
209 self, phy, phy.local_ip4, phy.remote_ip4
211 tun6 = VppIpIpTunInterface(
212 self, phy, phy.local_ip6, phy.remote_ip6
221 pair1 = VppLcpPair(self, tuns[0], hosts[0]).add_vpp_config()
222 pair2 = VppLcpPair(self, tuns[1], hosts[1]).add_vpp_config()
224 self.logger.info(self.vapi.cli("sh lcp adj verbose"))
225 self.logger.info(self.vapi.cli("sh lcp"))
226 self.logger.info(self.vapi.cli("sh ip punt redirect"))
233 p = IP(src=tun4.local_ip4, dst="2.2.2.2") / UDP(sport=1234, dport=1234) / Raw()
235 rxs = self.send_and_expect(self.pg4, p * N_PKTS, phy)
237 # verify inner packet is unchanged and has the tunnel encap
239 self.assertEqual(rx[Ether].dst, phy.remote_mac)
240 self.assertEqual(rx[IP].dst, phy.remote_ip4)
241 self.assertEqual(rx[IP].src, phy.local_ip4)
242 inner = IP(rx[IP].payload)
243 self.assertEqual(inner.src, tun4.local_ip4)
244 self.assertEqual(inner.dst, "2.2.2.2")
247 p = IPv6(src=tun6.local_ip6, dst="2::2") / UDP(sport=1234, dport=1234) / Raw()
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")
261 Ether(dst=phy.local_mac, src=phy.remote_mac)
262 / IP(dst=phy.local_ip4, src=phy.remote_ip4)
263 / IP(dst=tun4.local_ip4, src=tun4.remote_ip4)
264 / UDP(sport=1234, dport=1234)
268 rxs = self.send_and_expect(phy, p * N_PKTS, self.pg4)
271 self.assertEqual(rx[IP].dst, tun4.local_ip4)
272 self.assertEqual(rx[IP].src, tun4.remote_ip4)
276 Ether(dst=phy.local_mac, src=phy.remote_mac)
277 / IPv6(dst=phy.local_ip6, src=phy.remote_ip6)
278 / IPv6(dst=tun6.local_ip6, src=tun6.remote_ip6)
279 / UDP(sport=1234, dport=1234)
283 rxs = self.send_and_expect(phy, p * N_PKTS, self.pg5)
286 self.assertEqual(rx[IPv6].dst, tun6.local_ip6)
287 self.assertEqual(rx[IPv6].src, tun6.remote_ip6)
297 class TestLinuxCPIpsec(TemplateIpsec, TemplateIpsecItf4, IpsecTun4):
298 """IPsec Interface IPv4"""
300 extra_vpp_plugin_config = [
302 "linux_cp_plugin.so",
307 "linux_cp_unittest_plugin.so",
314 super(TestLinuxCPIpsec, self).setUp()
316 self.tun_if = self.pg0
317 self.pg_interfaces += self.create_pg_ip4_interfaces(range(3, 4))
318 self.pg_interfaces += self.create_pg_ip6_interfaces(range(4, 5))
321 super(TestLinuxCPIpsec, self).tearDown()
323 def verify_encrypted(self, p, sa, rxs):
327 self.assertEqual(rx[UDP].dport, 4500)
328 self.assert_packet_checksums_valid(rx)
329 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
332 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
333 if not decrypt_pkt.haslayer(IP):
334 decrypt_pkt = IP(decrypt_pkt[Raw].load)
335 if rx_ip.proto == socket.IPPROTO_ESP:
336 self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
337 decrypt_pkts.append(decrypt_pkt)
338 self.assert_equal(decrypt_pkt.src, p.tun_if.local_ip4)
339 self.assert_equal(decrypt_pkt.dst, p.tun_if.remote_ip4)
341 self.logger.debug(ppp("Unexpected packet:", rx))
343 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
347 pkts = reassemble4(decrypt_pkts)
349 self.assert_packet_checksums_valid(pkt)
351 def verify_decrypted(self, p, rxs):
354 self.assert_equal(rx[IP].src, p.tun_if.remote_ip4)
355 self.assert_equal(rx[IP].dst, p.tun_if.local_ip4)
356 self.assert_packet_checksums_valid(rx)
358 def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=54):
360 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
363 / UDP(sport=1111, dport=2222)
364 / Raw(b"X" * payload_size)
366 for i in range(count)
369 def test_linux_cp_ipsec4_tun(self):
370 """Linux CP Ipsec TUN"""
377 # the pg that paris with the tunnel
380 # tunnel and protection setup
383 self.config_network(p)
384 self.config_sa_tun(p, self.pg0.local_ip4, self.pg0.remote_ip4)
385 self.config_protect(p)
387 pair = VppLcpPair(self, p.tun_if, self.host).add_vpp_config()
389 self.logger.info(self.vapi.cli("sh int addr"))
390 self.logger.info(self.vapi.cli("sh lcp"))
391 self.logger.info(self.vapi.cli("sh ip punt redirect"))
399 IP(src=p.tun_if.local_ip4, dst=p.tun_if.remote_ip4)
400 / UDP(sport=1234, dport=1234)
404 rxs = self.send_and_expect(self.host, pkt * N_PKTS, self.tun_if)
405 self.verify_encrypted(p, p.vpp_tun_sa, rxs)
408 pkts = self.gen_encrypt_pkts(
412 src=p.tun_if.remote_ip4,
413 dst=p.tun_if.local_ip4,
416 rxs = self.send_and_expect(self.tun_if, pkts, self.host)
417 self.verify_decrypted(p, rxs)
420 pair.remove_vpp_config()
421 self.unconfig_protect(p)
423 self.unconfig_network(p)
426 if __name__ == "__main__":
427 unittest.main(testRunner=VppTestRunner)