import unittest
from scapy.packet import Raw
-from scapy.layers.inet6 import IPv6, Ether, IP, UDP, ICMPv6PacketTooBig
+from scapy.layers.l2 import GRE
+from scapy.layers.inet6 import IPv6, Ether, IP, ICMPv6PacketTooBig
from scapy.layers.inet6 import ipv6nh, IPerror6
from scapy.layers.inet import TCP, ICMP
from scapy.layers.vxlan import VXLAN
-from scapy.data import ETH_P_IP, ETH_P_IPV6, ETH_P_ARP
-from scapy.layers.ipsec import SecurityAssociation, ESP
+from scapy.layers.ipsec import ESP
from vpp_papi import VppEnum
-from framework import VppTestCase, VppTestRunner
-from vpp_object import VppObject
-from vpp_interface import VppInterface
-from vpp_ip import DpoProto
+from framework import VppTestCase
+from asfframework import VppTestRunner
from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto
from vpp_ipip_tun_interface import VppIpIpTunInterface
from vpp_vxlan_tunnel import VppVxlanTunnel
-from socket import AF_INET, AF_INET6, inet_pton
-from util import reassemble4
+from vpp_gre_interface import VppGreInterface
from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect
from template_ipsec import (
IPsecIPv4Params,
IPsecIPv6Params,
- mk_scapy_crypt_key,
config_tun_params,
)
self, self.pg0, self.pg0.local_ip6, self.pg0.remote_ip6
)
+ self.gre4 = VppGreInterface(self, self.pg0.local_ip4, self.pg0.remote_ip4)
+ self.gre6 = VppGreInterface(self, self.pg0.local_ip6, self.pg0.remote_ip6)
+
def tearDown(self):
super(TestGSO, self).tearDown()
if not self.vpp_dead:
sw_if_index=self.pg0.sw_if_index, enable_disable=0
)
+ def test_gso_gre(self):
+ """GSO GRE test"""
+ #
+ # Send jumbo frame with gso enabled only on gre tunnel interface.
+ # create GRE tunnel on VPP pg0.
+ #
+
+ #
+ # create gre 4 tunnel
+ #
+ self.gre4.add_vpp_config()
+ self.gre4.admin_up()
+ self.gre4.config_ip4()
+
+ #
+ # Add a route that resolves the tunnel's destination
+ #
+ # Add IPv4 routes via tunnel interface
+ self.ip4_via_gre4_tunnel = VppIpRoute(
+ self,
+ "172.16.10.0",
+ 24,
+ [
+ VppRoutePath(
+ "0.0.0.0",
+ self.gre4.sw_if_index,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
+ )
+ ],
+ )
+ self.ip4_via_gre4_tunnel.add_vpp_config()
+
+ pgre4 = (
+ Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
+ / IP(src=self.pg2.remote_ip4, dst="172.16.10.3", flags="DF")
+ / TCP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 65200)
+ )
+
+ # test when GSO segmentation is disabled, Packets are truncated
+ rxs = self.send_and_expect(self.pg2, 5 * [pgre4], self.pg0, 5)
+ for rx in rxs:
+ self.assertEqual(rx[Ether].src, self.pg0.local_mac)
+ self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
+ self.assertEqual(rx[IP].src, self.pg0.local_ip4)
+ self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
+ self.assert_ip_checksum_valid(rx)
+ self.assertEqual(rx[IP].proto, 0x2F) # GRE encap
+ self.assertEqual(rx[GRE].proto, 0x0800) # IPv4
+ inner = rx[GRE].payload
+ self.assertNotEqual(rx[IP].len - 20 - 4, len(inner))
+ self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
+ self.assertEqual(inner[IP].dst, "172.16.10.3")
+ self.assert_ip_checksum_valid(inner)
+ payload_len = inner[IP].len - 20 - 20
+ self.assertEqual(payload_len, 65200)
+ # truncated packet to MTU size
+ self.assertNotEqual(payload_len, len(inner[Raw]))
+
+ # enable the GSO segmentation on GRE tunnel
+ self.vapi.feature_gso_enable_disable(
+ sw_if_index=self.gre4.sw_if_index, enable_disable=1
+ )
+
+ # test again, this time payload will be chuncked to GSO size (i.e. 1448)
+ rxs = self.send_and_expect(self.pg2, 5 * [pgre4], self.pg0, 225)
+ size = 0
+ for rx in rxs:
+ self.assertEqual(rx[Ether].src, self.pg0.local_mac)
+ self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
+ self.assertEqual(rx[IP].src, self.pg0.local_ip4)
+ self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
+ self.assert_ip_checksum_valid(rx)
+ self.assertEqual(rx[IP].proto, 0x2F) # GRE encap
+ self.assertEqual(rx[GRE].proto, 0x0800) # IPv4
+ inner = rx[GRE].payload
+ self.assertEqual(rx[IP].len - 20 - 4, len(inner))
+ self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
+ self.assertEqual(inner[IP].dst, "172.16.10.3")
+ self.assert_ip_checksum_valid(inner)
+ self.assert_tcp_checksum_valid(inner)
+ payload_len = inner[IP].len - 20 - 20
+ self.assertEqual(payload_len, len(inner[Raw]))
+ size += payload_len
+ self.assertEqual(size, 65200 * 5)
+
+ # Disable the GSO segmentation on GRE tunnel
+ self.vapi.feature_gso_enable_disable(
+ sw_if_index=self.gre4.sw_if_index, enable_disable=0
+ )
+
+ # test again when GSO segmentation is disabled, Packets are truncated
+ rxs = self.send_and_expect(self.pg2, 5 * [pgre4], self.pg0, 5)
+ for rx in rxs:
+ self.assertEqual(rx[Ether].src, self.pg0.local_mac)
+ self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
+ self.assertEqual(rx[IP].src, self.pg0.local_ip4)
+ self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
+ self.assert_ip_checksum_valid(rx)
+ self.assertEqual(rx[IP].proto, 0x2F) # GRE encap
+ self.assertEqual(rx[GRE].proto, 0x0800) # IPv4
+ inner = rx[GRE].payload
+ self.assertNotEqual(rx[IP].len - 20 - 4, len(inner))
+ self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
+ self.assertEqual(inner[IP].dst, "172.16.10.3")
+ self.assert_ip_checksum_valid(inner)
+ payload_len = inner[IP].len - 20 - 20
+ self.assertEqual(payload_len, 65200)
+ # truncated packet to MTU size
+ self.assertNotEqual(payload_len, len(inner[Raw]))
+
+ self.ip4_via_gre4_tunnel.remove_vpp_config()
+ self.gre4.remove_vpp_config()
+
+ self.gre6.add_vpp_config()
+ self.gre6.admin_up()
+ self.gre6.config_ip4()
+
+ #
+ # Add a route that resolves the tunnel's destination
+ # Add IPv6 routes via tunnel interface
+ #
+ self.vapi.feature_gso_enable_disable(
+ sw_if_index=self.gre6.sw_if_index, enable_disable=1
+ )
+ self.ip6_via_gre6_tunnel = VppIpRoute(
+ self,
+ "fd01:10::",
+ 64,
+ [
+ VppRoutePath(
+ "::",
+ self.gre6.sw_if_index,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ )
+ ],
+ )
+ self.ip6_via_gre6_tunnel.add_vpp_config()
+
+ #
+ # Create IPv6 packet
+ #
+ pgre6 = (
+ Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
+ / IPv6(src=self.pg2.remote_ip6, dst="fd01:10::3")
+ / TCP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 65200)
+ )
+
+ # test when GSO segmentation is enabled, payload will be segmented
+ # into GSO size (i.e. 1448)
+ rxs = self.send_and_expect(self.pg2, 5 * [pgre6], self.pg0, 225)
+ size = 0
+ for rx in rxs:
+ self.assertEqual(rx[Ether].src, self.pg0.local_mac)
+ self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
+ self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
+ self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
+ self.assertEqual(ipv6nh[rx[IPv6].nh], "GRE")
+ self.assertEqual(rx[GRE].proto, 0x86DD) # IPv6
+ inner = rx[GRE].payload
+ self.assertEqual(rx[IPv6].plen - 4, len(inner))
+ self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
+ self.assertEqual(inner[IPv6].dst, "fd01:10::3")
+ self.assert_tcp_checksum_valid(inner)
+ payload_len = inner[IPv6].plen - 20
+ self.assertEqual(payload_len, len(inner[Raw]))
+ size += payload_len
+ self.assertEqual(size, 65200 * 5)
+
+ # disable GSO segmentation
+ self.vapi.feature_gso_enable_disable(
+ sw_if_index=self.gre6.sw_if_index, enable_disable=0
+ )
+
+ # test again, this time packets will be truncated
+ rxs = self.send_and_expect(self.pg2, 5 * [pgre6], self.pg0, 5)
+ for rx in rxs:
+ self.assertEqual(rx[Ether].src, self.pg0.local_mac)
+ self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
+ self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
+ self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
+ self.assertEqual(ipv6nh[rx[IPv6].nh], "GRE")
+ self.assertEqual(rx[GRE].proto, 0x86DD) # IPv6
+ inner = rx[GRE].payload
+ self.assertNotEqual(rx[IPv6].plen - 4, len(inner))
+ self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
+ self.assertEqual(inner[IPv6].dst, "fd01:10::3")
+ payload_len = inner[IPv6].plen - 20
+ self.assertEqual(payload_len, 65200)
+ # packets are truncated to MTU size
+ self.assertNotEqual(payload_len, len(inner[Raw]))
+
+ self.ip6_via_gre6_tunnel.remove_vpp_config()
+ self.gre6.remove_vpp_config()
+
def test_gso_ipsec(self):
"""GSO IPSEC test"""
#
self.tun_sa_in_v4 = VppIpsecSA(
self,
- self.ipv4_params.vpp_tun_sa_id,
- self.ipv4_params.vpp_tun_spi,
+ self.ipv4_params.scapy_tun_sa_id,
+ self.ipv4_params.scapy_tun_spi,
self.ipv4_params.auth_algo_vpp_id,
self.ipv4_params.auth_key,
self.ipv4_params.crypt_algo_vpp_id,
self.tun_sa_out_v4 = VppIpsecSA(
self,
- self.ipv4_params.scapy_tun_sa_id,
- self.ipv4_params.scapy_tun_spi,
+ self.ipv4_params.vpp_tun_sa_id,
+ self.ipv4_params.vpp_tun_spi,
self.ipv4_params.auth_algo_vpp_id,
self.ipv4_params.auth_key,
self.ipv4_params.crypt_algo_vpp_id,
self.assertEqual(rx[IP].src, self.pg0.local_ip4)
self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
self.assertEqual(rx[IP].proto, 50) # ESP
- self.assertEqual(rx[ESP].spi, self.ipv4_params.scapy_tun_spi)
+ self.assertEqual(rx[ESP].spi, self.ipv4_params.vpp_tun_spi)
inner = self.ipv4_params.vpp_tun_sa.decrypt(rx[IP])
self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
self.assertEqual(inner[IP].dst, "172.16.10.3")
self.assertEqual(rx[IP].src, self.pg0.local_ip4)
self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
self.assertEqual(rx[IP].proto, 50) # ESP
- self.assertEqual(rx[ESP].spi, self.ipv4_params.scapy_tun_spi)
+ self.assertEqual(rx[ESP].spi, self.ipv4_params.vpp_tun_spi)
inner = self.ipv4_params.vpp_tun_sa.decrypt(rx[IP])
self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
self.assertEqual(inner[IPv6].dst, "fd01:10::3")
config_tun_params(self.ipv6_params, self.encryption_type, self.ipip6)
self.tun_sa_in_v6 = VppIpsecSA(
self,
- self.ipv6_params.vpp_tun_sa_id,
- self.ipv6_params.vpp_tun_spi,
+ self.ipv6_params.scapy_tun_sa_id,
+ self.ipv6_params.scapy_tun_spi,
self.ipv6_params.auth_algo_vpp_id,
self.ipv6_params.auth_key,
self.ipv6_params.crypt_algo_vpp_id,
self.tun_sa_out_v6 = VppIpsecSA(
self,
- self.ipv6_params.scapy_tun_sa_id,
- self.ipv6_params.scapy_tun_spi,
+ self.ipv6_params.vpp_tun_sa_id,
+ self.ipv6_params.vpp_tun_spi,
self.ipv6_params.auth_algo_vpp_id,
self.ipv6_params.auth_key,
self.ipv6_params.crypt_algo_vpp_id,
self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
self.assertEqual(ipv6nh[rx[IPv6].nh], "ESP Header")
- self.assertEqual(rx[ESP].spi, self.ipv6_params.scapy_tun_spi)
+ self.assertEqual(rx[ESP].spi, self.ipv6_params.vpp_tun_spi)
inner = self.ipv6_params.vpp_tun_sa.decrypt(rx[IPv6])
self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
self.assertEqual(inner[IP].dst, "172.16.10.3")
self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
self.assertEqual(ipv6nh[rx[IPv6].nh], "ESP Header")
- self.assertEqual(rx[ESP].spi, self.ipv6_params.scapy_tun_spi)
+ self.assertEqual(rx[ESP].spi, self.ipv6_params.vpp_tun_spi)
inner = self.ipv6_params.vpp_tun_sa.decrypt(rx[IPv6])
self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
self.assertEqual(inner[IPv6].dst, "fd01:10::3")
Code Review / vpp.git / blobdiff