test/test_gso.py

   1 #!/usr/bin/env python3
   2 """GSO functional tests"""
   3
   4 #
   5 # Add tests for:
   6 # - GSO
   7 # - Verify that sending Jumbo frame without GSO enabled correctly
   8 # - Verify that sending Jumbo frame with GSO enabled correctly
   9 # - Verify that sending Jumbo frame with GSO enabled only on ingress interface
  10 #
  11 import unittest
  12
  13 from scapy.packet import Raw
  14 from scapy.layers.l2 import GRE
  15 from scapy.layers.inet6 import IPv6, Ether, IP, ICMPv6PacketTooBig
  16 from scapy.layers.inet6 import ipv6nh, IPerror6
  17 from scapy.layers.inet import TCP, ICMP
  18 from scapy.layers.vxlan import VXLAN
  19 from scapy.layers.ipsec import ESP
  20
  21 from vpp_papi import VppEnum
  22 from framework import VppTestCase
  23 from asfframework import VppTestRunner
  24 from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto
  25 from vpp_ipip_tun_interface import VppIpIpTunInterface
  26 from vpp_vxlan_tunnel import VppVxlanTunnel
  27 from vpp_gre_interface import VppGreInterface
  28
  29 from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect
  30 from template_ipsec import (
  31     IPsecIPv4Params,
  32     IPsecIPv6Params,
  33     config_tun_params,
  34 )
  35
  36 """ Test_gso is a subclass of VPPTestCase classes.
  37     GSO tests.
  38 """
  39
  40
  41 class TestGSO(VppTestCase):
  42     """GSO Test Case"""
  43
  44     def __init__(self, *args):
  45         VppTestCase.__init__(self, *args)
  46
  47     @classmethod
  48     def setUpClass(self):
  49         super(TestGSO, self).setUpClass()
  50         res = self.create_pg_interfaces(range(2))
  51         res_gso = self.create_pg_interfaces(range(2, 4), 1, 1460)
  52         self.create_pg_interfaces(range(4, 5), 1, 8940)
  53         self.pg_interfaces.append(res[0])
  54         self.pg_interfaces.append(res[1])
  55         self.pg_interfaces.append(res_gso[0])
  56         self.pg_interfaces.append(res_gso[1])
  57
  58     @classmethod
  59     def tearDownClass(self):
  60         super(TestGSO, self).tearDownClass()
  61
  62     def setUp(self):
  63         super(TestGSO, self).setUp()
  64         for i in self.pg_interfaces:
  65             i.admin_up()
  66             i.config_ip4()
  67             i.config_ip6()
  68             i.disable_ipv6_ra()
  69             i.resolve_arp()
  70             i.resolve_ndp()
  71
  72         self.single_tunnel_bd = 10
  73         self.vxlan = VppVxlanTunnel(
  74             self,
  75             src=self.pg0.local_ip4,
  76             dst=self.pg0.remote_ip4,
  77             vni=self.single_tunnel_bd,
  78         )
  79
  80         self.vxlan2 = VppVxlanTunnel(
  81             self,
  82             src=self.pg0.local_ip6,
  83             dst=self.pg0.remote_ip6,
  84             vni=self.single_tunnel_bd,
  85         )
  86
  87         self.ipip4 = VppIpIpTunInterface(
  88             self, self.pg0, self.pg0.local_ip4, self.pg0.remote_ip4
  89         )
  90         self.ipip6 = VppIpIpTunInterface(
  91             self, self.pg0, self.pg0.local_ip6, self.pg0.remote_ip6
  92         )
  93
  94         self.gre4 = VppGreInterface(self, self.pg0.local_ip4, self.pg0.remote_ip4)
  95         self.gre6 = VppGreInterface(self, self.pg0.local_ip6, self.pg0.remote_ip6)
  96
  97     def tearDown(self):
  98         super(TestGSO, self).tearDown()
  99         if not self.vpp_dead:
 100             for i in self.pg_interfaces:
 101                 i.unconfig_ip4()
 102                 i.unconfig_ip6()
 103                 i.admin_down()
 104
 105     def test_gso(self):
 106         """GSO test"""
 107         #
 108         # Send jumbo frame with gso disabled and DF bit is set
 109         #
 110         p4 = (
 111             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 112             / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, flags="DF")
 113             / TCP(sport=1234, dport=1234)
 114             / Raw(b"\xa5" * 65200)
 115         )
 116
 117         rxs = self.send_and_expect(self.pg0, [p4], self.pg0)
 118
 119         for rx in rxs:
 120             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 121             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 122             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
 123             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 124             self.assertEqual(rx[ICMP].type, 3)  # "dest-unreach"
 125             self.assertEqual(rx[ICMP].code, 4)  # "fragmentation-needed"
 126
 127         #
 128         # Send checksum offload frames
 129         #
 130         p40 = (
 131             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 132             / IP(src=self.pg2.remote_ip4, dst=self.pg0.remote_ip4, flags="DF")
 133             / TCP(sport=1234, dport=1234)
 134             / Raw(b"\xa5" * 1460)
 135         )
 136
 137         rxs = self.send_and_expect(self.pg2, 100 * [p40], self.pg0)
 138
 139         for rx in rxs:
 140             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 141             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 142             self.assertEqual(rx[IP].src, self.pg2.remote_ip4)
 143             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 144             payload_len = rx[IP].len - 20 - 20
 145             self.assert_ip_checksum_valid(rx)
 146             self.assert_tcp_checksum_valid(rx)
 147             self.assertEqual(payload_len, len(rx[Raw]))
 148
 149         p60 = (
 150             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 151             / IPv6(src=self.pg2.remote_ip6, dst=self.pg0.remote_ip6)
 152             / TCP(sport=1234, dport=1234)
 153             / Raw(b"\xa5" * 1440)
 154         )
 155
 156         rxs = self.send_and_expect(self.pg2, 100 * [p60], self.pg0)
 157
 158         for rx in rxs:
 159             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 160             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 161             self.assertEqual(rx[IPv6].src, self.pg2.remote_ip6)
 162             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
 163             payload_len = rx[IPv6].plen - 20
 164             self.assert_tcp_checksum_valid(rx)
 165             self.assertEqual(payload_len, len(rx[Raw]))
 166
 167         #
 168         # Send jumbo frame with gso enabled and DF bit is set
 169         # input and output interfaces support GSO
 170         #
 171         self.vapi.feature_gso_enable_disable(
 172             sw_if_index=self.pg3.sw_if_index, enable_disable=1
 173         )
 174         p41 = (
 175             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 176             / IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4, flags="DF")
 177             / TCP(sport=1234, dport=1234)
 178             / Raw(b"\xa5" * 65200)
 179         )
 180
 181         rxs = self.send_and_expect(self.pg2, 100 * [p41], self.pg3, 100)
 182
 183         for rx in rxs:
 184             self.assertEqual(rx[Ether].src, self.pg3.local_mac)
 185             self.assertEqual(rx[Ether].dst, self.pg3.remote_mac)
 186             self.assertEqual(rx[IP].src, self.pg2.remote_ip4)
 187             self.assertEqual(rx[IP].dst, self.pg3.remote_ip4)
 188             self.assertEqual(rx[IP].len, 65240)  # 65200 + 20 (IP) + 20 (TCP)
 189             self.assertEqual(rx[TCP].sport, 1234)
 190             self.assertEqual(rx[TCP].dport, 1234)
 191
 192         #
 193         # ipv6
 194         #
 195         p61 = (
 196             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 197             / IPv6(src=self.pg2.remote_ip6, dst=self.pg3.remote_ip6)
 198             / TCP(sport=1234, dport=1234)
 199             / Raw(b"\xa5" * 65200)
 200         )
 201
 202         rxs = self.send_and_expect(self.pg2, 100 * [p61], self.pg3, 100)
 203
 204         for rx in rxs:
 205             self.assertEqual(rx[Ether].src, self.pg3.local_mac)
 206             self.assertEqual(rx[Ether].dst, self.pg3.remote_mac)
 207             self.assertEqual(rx[IPv6].src, self.pg2.remote_ip6)
 208             self.assertEqual(rx[IPv6].dst, self.pg3.remote_ip6)
 209             self.assertEqual(rx[IPv6].plen, 65220)  # 65200 + 20 (TCP)
 210             self.assertEqual(rx[TCP].sport, 1234)
 211             self.assertEqual(rx[TCP].dport, 1234)
 212
 213         #
 214         # Send jumbo frame with gso enabled only on input interface
 215         # and DF bit is set. GSO packet will be chunked into gso_size
 216         # data payload
 217         #
 218         self.vapi.feature_gso_enable_disable(
 219             sw_if_index=self.pg0.sw_if_index, enable_disable=1
 220         )
 221         p42 = (
 222             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 223             / IP(src=self.pg2.remote_ip4, dst=self.pg0.remote_ip4, flags="DF")
 224             / TCP(sport=1234, dport=1234)
 225             / Raw(b"\xa5" * 65200)
 226         )
 227
 228         rxs = self.send_and_expect(self.pg2, 5 * [p42], self.pg0, 225)
 229         size = 0
 230         for rx in rxs:
 231             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 232             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 233             self.assertEqual(rx[IP].src, self.pg2.remote_ip4)
 234             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 235             payload_len = rx[IP].len - 20 - 20  # len - 20 (IP4) - 20 (TCP)
 236             self.assert_ip_checksum_valid(rx)
 237             self.assert_tcp_checksum_valid(rx)
 238             self.assertEqual(rx[TCP].sport, 1234)
 239             self.assertEqual(rx[TCP].dport, 1234)
 240             self.assertEqual(payload_len, len(rx[Raw]))
 241             size += payload_len
 242         self.assertEqual(size, 65200 * 5)
 243
 244         #
 245         # ipv6
 246         #
 247         p62 = (
 248             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 249             / IPv6(src=self.pg2.remote_ip6, dst=self.pg0.remote_ip6)
 250             / TCP(sport=1234, dport=1234)
 251             / Raw(b"\xa5" * 65200)
 252         )
 253
 254         rxs = self.send_and_expect(self.pg2, 5 * [p62], self.pg0, 225)
 255         size = 0
 256         for rx in rxs:
 257             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 258             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 259             self.assertEqual(rx[IPv6].src, self.pg2.remote_ip6)
 260             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
 261             payload_len = rx[IPv6].plen - 20
 262             self.assert_tcp_checksum_valid(rx)
 263             self.assertEqual(rx[TCP].sport, 1234)
 264             self.assertEqual(rx[TCP].dport, 1234)
 265             self.assertEqual(payload_len, len(rx[Raw]))
 266             size += payload_len
 267         self.assertEqual(size, 65200 * 5)
 268
 269         #
 270         # Send jumbo frame with gso enabled only on input interface
 271         # and DF bit is unset. GSO packet will be fragmented.
 272         #
 273         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [576, 0, 0, 0])
 274         self.vapi.feature_gso_enable_disable(
 275             sw_if_index=self.pg1.sw_if_index, enable_disable=1
 276         )
 277
 278         p43 = (
 279             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 280             / IP(src=self.pg2.remote_ip4, dst=self.pg1.remote_ip4)
 281             / TCP(sport=1234, dport=1234)
 282             / Raw(b"\xa5" * 65200)
 283         )
 284
 285         rxs = self.send_and_expect(self.pg2, 5 * [p43], self.pg1, 5 * 119)
 286         size = 0
 287         for rx in rxs:
 288             self.assertEqual(rx[Ether].src, self.pg1.local_mac)
 289             self.assertEqual(rx[Ether].dst, self.pg1.remote_mac)
 290             self.assertEqual(rx[IP].src, self.pg2.remote_ip4)
 291             self.assertEqual(rx[IP].dst, self.pg1.remote_ip4)
 292             self.assert_ip_checksum_valid(rx)
 293             size += rx[IP].len - 20
 294         size -= 20 * 5  # TCP header
 295         self.assertEqual(size, 65200 * 5)
 296
 297         #
 298         # IPv6
 299         # Send jumbo frame with gso enabled only on input interface.
 300         # ICMPv6 Packet Too Big will be sent back to sender.
 301         #
 302         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
 303         p63 = (
 304             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 305             / IPv6(src=self.pg2.remote_ip6, dst=self.pg1.remote_ip6)
 306             / TCP(sport=1234, dport=1234)
 307             / Raw(b"\xa5" * 65200)
 308         )
 309
 310         rxs = self.send_and_expect_some(self.pg2, 5 * [p63], self.pg2, 5)
 311         for rx in rxs:
 312             self.assertEqual(rx[Ether].src, self.pg2.local_mac)
 313             self.assertEqual(rx[Ether].dst, self.pg2.remote_mac)
 314             self.assertEqual(rx[IPv6].src, self.pg2.local_ip6)
 315             self.assertEqual(rx[IPv6].dst, self.pg2.remote_ip6)
 316             self.assertEqual(rx[IPv6].plen, 1240)  # MTU - IPv6 header
 317             self.assertEqual(ipv6nh[rx[IPv6].nh], "ICMPv6")
 318             self.assertEqual(rx[ICMPv6PacketTooBig].mtu, 1280)
 319             self.assertEqual(rx[IPerror6].src, self.pg2.remote_ip6)
 320             self.assertEqual(rx[IPerror6].dst, self.pg1.remote_ip6)
 321             self.assertEqual(rx[IPerror6].plen - 20, 65200)
 322
 323         #
 324         # Send jumbo frame with gso enabled only on input interface with 9K MTU
 325         # and DF bit is unset. GSO packet will be fragmented. MSS is 8960. GSO
 326         # size will be min(MSS, 2048 - 14 - 20) vlib_buffer_t size
 327         #
 328         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [9000, 0, 0, 0])
 329         self.vapi.sw_interface_set_mtu(self.pg4.sw_if_index, [9000, 0, 0, 0])
 330         p44 = (
 331             Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac)
 332             / IP(src=self.pg4.remote_ip4, dst=self.pg1.remote_ip4)
 333             / TCP(sport=1234, dport=1234)
 334             / Raw(b"\xa5" * 65200)
 335         )
 336
 337         rxs = self.send_and_expect(self.pg4, 5 * [p44], self.pg1, 165)
 338         size = 0
 339         for rx in rxs:
 340             self.assertEqual(rx[Ether].src, self.pg1.local_mac)
 341             self.assertEqual(rx[Ether].dst, self.pg1.remote_mac)
 342             self.assertEqual(rx[IP].src, self.pg4.remote_ip4)
 343             self.assertEqual(rx[IP].dst, self.pg1.remote_ip4)
 344             payload_len = rx[IP].len - 20 - 20  # len - 20 (IP4) - 20 (TCP)
 345             self.assert_ip_checksum_valid(rx)
 346             self.assert_tcp_checksum_valid(rx)
 347             self.assertEqual(payload_len, len(rx[Raw]))
 348             size += payload_len
 349         self.assertEqual(size, 65200 * 5)
 350
 351         #
 352         # IPv6
 353         #
 354         p64 = (
 355             Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac)
 356             / IPv6(src=self.pg4.remote_ip6, dst=self.pg1.remote_ip6)
 357             / TCP(sport=1234, dport=1234)
 358             / Raw(b"\xa5" * 65200)
 359         )
 360
 361         rxs = self.send_and_expect(self.pg4, 5 * [p64], self.pg1, 170)
 362         size = 0
 363         for rx in rxs:
 364             self.assertEqual(rx[Ether].src, self.pg1.local_mac)
 365             self.assertEqual(rx[Ether].dst, self.pg1.remote_mac)
 366             self.assertEqual(rx[IPv6].src, self.pg4.remote_ip6)
 367             self.assertEqual(rx[IPv6].dst, self.pg1.remote_ip6)
 368             payload_len = rx[IPv6].plen - 20
 369             self.assert_tcp_checksum_valid(rx)
 370             self.assertEqual(payload_len, len(rx[Raw]))
 371             size += payload_len
 372         self.assertEqual(size, 65200 * 5)
 373
 374         self.vapi.feature_gso_enable_disable(
 375             sw_if_index=self.pg0.sw_if_index, enable_disable=0
 376         )
 377         self.vapi.feature_gso_enable_disable(
 378             sw_if_index=self.pg1.sw_if_index, enable_disable=0
 379         )
 380
 381     def test_gso_vxlan(self):
 382         """GSO VXLAN test"""
 383         self.logger.info(self.vapi.cli("sh int addr"))
 384         #
 385         # Send jumbo frame with gso enabled only on input interface and
 386         # create VXLAN VTEP on VPP pg0, and put vxlan_tunnel0 and pg2
 387         # into BD.
 388         #
 389
 390         #
 391         # enable ipv4/vxlan
 392         #
 393         self.vxlan.add_vpp_config()
 394         self.vapi.sw_interface_set_l2_bridge(
 395             rx_sw_if_index=self.vxlan.sw_if_index, bd_id=self.single_tunnel_bd
 396         )
 397         self.vapi.sw_interface_set_l2_bridge(
 398             rx_sw_if_index=self.pg2.sw_if_index, bd_id=self.single_tunnel_bd
 399         )
 400         self.vapi.feature_gso_enable_disable(
 401             sw_if_index=self.pg0.sw_if_index, enable_disable=1
 402         )
 403
 404         #
 405         # IPv4/IPv4 - VXLAN
 406         #
 407         p45 = (
 408             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 409             / IP(src=self.pg2.remote_ip4, dst="172.16.3.3", flags="DF")
 410             / TCP(sport=1234, dport=1234)
 411             / Raw(b"\xa5" * 65200)
 412         )
 413
 414         rxs = self.send_and_expect(self.pg2, 5 * [p45], self.pg0, 225)
 415         size = 0
 416         for rx in rxs:
 417             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 418             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 419             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
 420             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 421             self.assert_ip_checksum_valid(rx)
 422             self.assert_udp_checksum_valid(rx, ignore_zero_checksum=False)
 423             self.assertEqual(rx[VXLAN].vni, 10)
 424             inner = rx[VXLAN].payload
 425             self.assertEqual(rx[IP].len - 20 - 8 - 8, len(inner))
 426             self.assertEqual(inner[Ether].src, self.pg2.remote_mac)
 427             self.assertEqual(inner[Ether].dst, "02:fe:60:1e:a2:79")
 428             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
 429             self.assertEqual(inner[IP].dst, "172.16.3.3")
 430             self.assert_ip_checksum_valid(inner)
 431             self.assert_tcp_checksum_valid(inner)
 432             payload_len = inner[IP].len - 20 - 20
 433             self.assertEqual(payload_len, len(inner[Raw]))
 434             size += payload_len
 435         self.assertEqual(size, 65200 * 5)
 436
 437         #
 438         # IPv4/IPv6 - VXLAN
 439         #
 440         p65 = (
 441             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 442             / IPv6(src=self.pg2.remote_ip6, dst="fd01:3::3")
 443             / TCP(sport=1234, dport=1234)
 444             / Raw(b"\xa5" * 65200)
 445         )
 446
 447         rxs = self.send_and_expect(self.pg2, 5 * [p65], self.pg0, 225)
 448         size = 0
 449         for rx in rxs:
 450             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 451             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 452             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
 453             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 454             self.assert_ip_checksum_valid(rx)
 455             self.assert_udp_checksum_valid(rx, ignore_zero_checksum=False)
 456             self.assertEqual(rx[VXLAN].vni, 10)
 457             inner = rx[VXLAN].payload
 458             self.assertEqual(rx[IP].len - 20 - 8 - 8, len(inner))
 459             self.assertEqual(inner[Ether].src, self.pg2.remote_mac)
 460             self.assertEqual(inner[Ether].dst, "02:fe:60:1e:a2:79")
 461             self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
 462             self.assertEqual(inner[IPv6].dst, "fd01:3::3")
 463             self.assert_tcp_checksum_valid(inner)
 464             payload_len = inner[IPv6].plen - 20
 465             self.assertEqual(payload_len, len(inner[Raw]))
 466             size += payload_len
 467         self.assertEqual(size, 65200 * 5)
 468
 469         #
 470         # disable ipv4/vxlan
 471         #
 472         self.vxlan.remove_vpp_config()
 473
 474         #
 475         # enable ipv6/vxlan
 476         #
 477         self.vxlan2.add_vpp_config()
 478         self.vapi.sw_interface_set_l2_bridge(
 479             rx_sw_if_index=self.vxlan2.sw_if_index, bd_id=self.single_tunnel_bd
 480         )
 481
 482         #
 483         # IPv6/IPv4 - VXLAN
 484         #
 485         p46 = (
 486             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 487             / IP(src=self.pg2.remote_ip4, dst="172.16.3.3", flags="DF")
 488             / TCP(sport=1234, dport=1234)
 489             / Raw(b"\xa5" * 65200)
 490         )
 491
 492         rxs = self.send_and_expect(self.pg2, 5 * [p46], self.pg0, 225)
 493         size = 0
 494         for rx in rxs:
 495             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 496             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 497             self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
 498             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
 499             self.assert_udp_checksum_valid(rx, ignore_zero_checksum=False)
 500             self.assertEqual(rx[VXLAN].vni, 10)
 501             inner = rx[VXLAN].payload
 502             self.assertEqual(rx[IPv6].plen - 8 - 8, len(inner))
 503             self.assertEqual(inner[Ether].src, self.pg2.remote_mac)
 504             self.assertEqual(inner[Ether].dst, "02:fe:60:1e:a2:79")
 505             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
 506             self.assertEqual(inner[IP].dst, "172.16.3.3")
 507             self.assert_ip_checksum_valid(inner)
 508             self.assert_tcp_checksum_valid(inner)
 509             payload_len = inner[IP].len - 20 - 20
 510             self.assertEqual(payload_len, len(inner[Raw]))
 511             size += payload_len
 512         self.assertEqual(size, 65200 * 5)
 513
 514         #
 515         # IPv6/IPv6 - VXLAN
 516         #
 517         p66 = (
 518             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 519             / IPv6(src=self.pg2.remote_ip6, dst="fd01:3::3")
 520             / TCP(sport=1234, dport=1234)
 521             / Raw(b"\xa5" * 65200)
 522         )
 523
 524         rxs = self.send_and_expect(self.pg2, 5 * [p66], self.pg0, 225)
 525         size = 0
 526         for rx in rxs:
 527             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 528             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 529             self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
 530             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
 531             self.assert_udp_checksum_valid(rx, ignore_zero_checksum=False)
 532             self.assertEqual(rx[VXLAN].vni, 10)
 533             inner = rx[VXLAN].payload
 534             self.assertEqual(rx[IPv6].plen - 8 - 8, len(inner))
 535             self.assertEqual(inner[Ether].src, self.pg2.remote_mac)
 536             self.assertEqual(inner[Ether].dst, "02:fe:60:1e:a2:79")
 537             self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
 538             self.assertEqual(inner[IPv6].dst, "fd01:3::3")
 539             self.assert_tcp_checksum_valid(inner)
 540             payload_len = inner[IPv6].plen - 20
 541             self.assertEqual(payload_len, len(inner[Raw]))
 542             size += payload_len
 543         self.assertEqual(size, 65200 * 5)
 544
 545         #
 546         # disable ipv4/vxlan
 547         #
 548         self.vxlan2.remove_vpp_config()
 549
 550         self.vapi.feature_gso_enable_disable(
 551             sw_if_index=self.pg0.sw_if_index, enable_disable=0
 552         )
 553
 554     def test_gso_ipip(self):
 555         """GSO IPIP test"""
 556         self.logger.info(self.vapi.cli("sh int addr"))
 557         #
 558         # Send jumbo frame with gso enabled only on input interface and
 559         # create IPIP tunnel on VPP pg0.
 560         #
 561         self.vapi.feature_gso_enable_disable(
 562             sw_if_index=self.pg0.sw_if_index, enable_disable=1
 563         )
 564
 565         #
 566         # enable ipip4
 567         #
 568         self.ipip4.add_vpp_config()
 569
 570         # Set interface up and enable IP on it
 571         self.ipip4.admin_up()
 572         self.ipip4.set_unnumbered(self.pg0.sw_if_index)
 573
 574         # Add IPv4 routes via tunnel interface
 575         self.ip4_via_ip4_tunnel = VppIpRoute(
 576             self,
 577             "172.16.10.0",
 578             24,
 579             [
 580                 VppRoutePath(
 581                     "0.0.0.0",
 582                     self.ipip4.sw_if_index,
 583                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
 584                 )
 585             ],
 586         )
 587         self.ip4_via_ip4_tunnel.add_vpp_config()
 588
 589         #
 590         # IPv4/IPv4 - IPIP
 591         #
 592         p47 = (
 593             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 594             / IP(src=self.pg2.remote_ip4, dst="172.16.10.3", flags="DF")
 595             / TCP(sport=1234, dport=1234)
 596             / Raw(b"\xa5" * 65200)
 597         )
 598
 599         rxs = self.send_and_expect(self.pg2, 5 * [p47], self.pg0, 225)
 600         size = 0
 601         for rx in rxs:
 602             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 603             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 604             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
 605             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 606             self.assert_ip_checksum_valid(rx)
 607             self.assertEqual(rx[IP].proto, 4)  # ipencap
 608             inner = rx[IP].payload
 609             self.assertEqual(rx[IP].len - 20, len(inner))
 610             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
 611             self.assertEqual(inner[IP].dst, "172.16.10.3")
 612             self.assert_ip_checksum_valid(inner)
 613             self.assert_tcp_checksum_valid(inner)
 614             payload_len = inner[IP].len - 20 - 20
 615             self.assertEqual(payload_len, len(inner[Raw]))
 616             size += payload_len
 617         self.assertEqual(size, 65200 * 5)
 618
 619         self.ip6_via_ip4_tunnel = VppIpRoute(
 620             self,
 621             "fd01:10::",
 622             64,
 623             [
 624                 VppRoutePath(
 625                     "::",
 626                     self.ipip4.sw_if_index,
 627                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
 628                 )
 629             ],
 630         )
 631         self.ip6_via_ip4_tunnel.add_vpp_config()
 632         #
 633         # IPv4/IPv6 - IPIP
 634         #
 635         p67 = (
 636             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 637             / IPv6(src=self.pg2.remote_ip6, dst="fd01:10::3")
 638             / TCP(sport=1234, dport=1234)
 639             / Raw(b"\xa5" * 65200)
 640         )
 641
 642         rxs = self.send_and_expect(self.pg2, 5 * [p67], self.pg0, 225)
 643         size = 0
 644         for rx in rxs:
 645             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 646             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 647             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
 648             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 649             self.assert_ip_checksum_valid(rx)
 650             self.assertEqual(rx[IP].proto, 41)  # ipv6
 651             inner = rx[IP].payload
 652             self.assertEqual(rx[IP].len - 20, len(inner))
 653             self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
 654             self.assertEqual(inner[IPv6].dst, "fd01:10::3")
 655             self.assert_tcp_checksum_valid(inner)
 656             payload_len = inner[IPv6].plen - 20
 657             self.assertEqual(payload_len, len(inner[Raw]))
 658             size += payload_len
 659         self.assertEqual(size, 65200 * 5)
 660
 661         #
 662         # Send jumbo frame with gso enabled only on input interface and
 663         # create IPIP tunnel on VPP pg0. Enable gso feature node on ipip
 664         # tunnel - IPSec use case
 665         #
 666         self.vapi.feature_gso_enable_disable(
 667             sw_if_index=self.pg0.sw_if_index, enable_disable=0
 668         )
 669         self.vapi.feature_gso_enable_disable(
 670             sw_if_index=self.ipip4.sw_if_index, enable_disable=1
 671         )
 672
 673         rxs = self.send_and_expect(self.pg2, 5 * [p47], self.pg0, 225)
 674         size = 0
 675         for rx in rxs:
 676             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 677             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 678             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
 679             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 680             self.assert_ip_checksum_valid(rx)
 681             self.assertEqual(rx[IP].proto, 4)  # ipencap
 682             inner = rx[IP].payload
 683             self.assertEqual(rx[IP].len - 20, len(inner))
 684             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
 685             self.assertEqual(inner[IP].dst, "172.16.10.3")
 686             self.assert_ip_checksum_valid(inner)
 687             self.assert_tcp_checksum_valid(inner)
 688             payload_len = inner[IP].len - 20 - 20
 689             self.assertEqual(payload_len, len(inner[Raw]))
 690             size += payload_len
 691         self.assertEqual(size, 65200 * 5)
 692
 693         #
 694         # disable ipip4
 695         #
 696         self.vapi.feature_gso_enable_disable(
 697             sw_if_index=self.ipip4.sw_if_index, enable_disable=0
 698         )
 699         self.ip4_via_ip4_tunnel.remove_vpp_config()
 700         self.ip6_via_ip4_tunnel.remove_vpp_config()
 701         self.ipip4.remove_vpp_config()
 702
 703         #
 704         # enable ipip6
 705         #
 706         self.vapi.feature_gso_enable_disable(
 707             sw_if_index=self.pg0.sw_if_index, enable_disable=1
 708         )
 709         self.ipip6.add_vpp_config()
 710
 711         # Set interface up and enable IP on it
 712         self.ipip6.admin_up()
 713         self.ipip6.set_unnumbered(self.pg0.sw_if_index)
 714
 715         # Add IPv4 routes via tunnel interface
 716         self.ip4_via_ip6_tunnel = VppIpRoute(
 717             self,
 718             "172.16.10.0",
 719             24,
 720             [
 721                 VppRoutePath(
 722                     "0.0.0.0",
 723                     self.ipip6.sw_if_index,
 724                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
 725                 )
 726             ],
 727         )
 728         self.ip4_via_ip6_tunnel.add_vpp_config()
 729
 730         #
 731         # IPv6/IPv4 - IPIP
 732         #
 733         p48 = (
 734             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 735             / IP(src=self.pg2.remote_ip4, dst="172.16.10.3", flags="DF")
 736             / TCP(sport=1234, dport=1234)
 737             / Raw(b"\xa5" * 65200)
 738         )
 739
 740         rxs = self.send_and_expect(self.pg2, 5 * [p48], self.pg0, 225)
 741         size = 0
 742         for rx in rxs:
 743             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 744             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 745             self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
 746             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
 747             self.assertEqual(ipv6nh[rx[IPv6].nh], "IP")
 748             inner = rx[IPv6].payload
 749             self.assertEqual(rx[IPv6].plen, len(inner))
 750             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
 751             self.assertEqual(inner[IP].dst, "172.16.10.3")
 752             self.assert_ip_checksum_valid(inner)
 753             self.assert_tcp_checksum_valid(inner)
 754             payload_len = inner[IP].len - 20 - 20
 755             self.assertEqual(payload_len, len(inner[Raw]))
 756             size += payload_len
 757         self.assertEqual(size, 65200 * 5)
 758
 759         self.ip6_via_ip6_tunnel = VppIpRoute(
 760             self,
 761             "fd01:10::",
 762             64,
 763             [
 764                 VppRoutePath(
 765                     "::",
 766                     self.ipip6.sw_if_index,
 767                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
 768                 )
 769             ],
 770         )
 771         self.ip6_via_ip6_tunnel.add_vpp_config()
 772
 773         #
 774         # IPv6/IPv6 - IPIP
 775         #
 776         p68 = (
 777             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 778             / IPv6(src=self.pg2.remote_ip6, dst="fd01:10::3")
 779             / TCP(sport=1234, dport=1234)
 780             / Raw(b"\xa5" * 65200)
 781         )
 782
 783         rxs = self.send_and_expect(self.pg2, 5 * [p68], self.pg0, 225)
 784         size = 0
 785         for rx in rxs:
 786             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 787             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 788             self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
 789             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
 790             self.assertEqual(ipv6nh[rx[IPv6].nh], "IPv6")
 791             inner = rx[IPv6].payload
 792             self.assertEqual(rx[IPv6].plen, len(inner))
 793             self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
 794             self.assertEqual(inner[IPv6].dst, "fd01:10::3")
 795             self.assert_tcp_checksum_valid(inner)
 796             payload_len = inner[IPv6].plen - 20
 797             self.assertEqual(payload_len, len(inner[Raw]))
 798             size += payload_len
 799         self.assertEqual(size, 65200 * 5)
 800
 801         #
 802         # disable ipip6
 803         #
 804         self.ip4_via_ip6_tunnel.remove_vpp_config()
 805         self.ip6_via_ip6_tunnel.remove_vpp_config()
 806         self.ipip6.remove_vpp_config()
 807
 808         self.vapi.feature_gso_enable_disable(
 809             sw_if_index=self.pg0.sw_if_index, enable_disable=0
 810         )
 811
 812     def test_gso_gre(self):
 813         """GSO GRE test"""
 814         #
 815         # Send jumbo frame with gso enabled only on gre tunnel interface.
 816         # create GRE tunnel on VPP pg0.
 817         #
 818
 819         #
 820         # create gre 4 tunnel
 821         #
 822         self.gre4.add_vpp_config()
 823         self.gre4.admin_up()
 824         self.gre4.config_ip4()
 825
 826         #
 827         # Add a route that resolves the tunnel's destination
 828         #
 829         # Add IPv4 routes via tunnel interface
 830         self.ip4_via_gre4_tunnel = VppIpRoute(
 831             self,
 832             "172.16.10.0",
 833             24,
 834             [
 835                 VppRoutePath(
 836                     "0.0.0.0",
 837                     self.gre4.sw_if_index,
 838                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
 839                 )
 840             ],
 841         )
 842         self.ip4_via_gre4_tunnel.add_vpp_config()
 843
 844         pgre4 = (
 845             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 846             / IP(src=self.pg2.remote_ip4, dst="172.16.10.3", flags="DF")
 847             / TCP(sport=1234, dport=1234)
 848             / Raw(b"\xa5" * 65200)
 849         )
 850
 851         # test when GSO segmentation is disabled, Packets are truncated
 852         rxs = self.send_and_expect(self.pg2, 5 * [pgre4], self.pg0, 5)
 853         for rx in rxs:
 854             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 855             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 856             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
 857             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 858             self.assert_ip_checksum_valid(rx)
 859             self.assertEqual(rx[IP].proto, 0x2F)  # GRE encap
 860             self.assertEqual(rx[GRE].proto, 0x0800)  # IPv4
 861             inner = rx[GRE].payload
 862             self.assertNotEqual(rx[IP].len - 20 - 4, len(inner))
 863             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
 864             self.assertEqual(inner[IP].dst, "172.16.10.3")
 865             self.assert_ip_checksum_valid(inner)
 866             payload_len = inner[IP].len - 20 - 20
 867             self.assertEqual(payload_len, 65200)
 868             # truncated packet to MTU size
 869             self.assertNotEqual(payload_len, len(inner[Raw]))
 870
 871         # enable the GSO segmentation on GRE tunnel
 872         self.vapi.feature_gso_enable_disable(
 873             sw_if_index=self.gre4.sw_if_index, enable_disable=1
 874         )
 875
 876         # test again, this time payload will be chuncked to GSO size (i.e. 1448)
 877         rxs = self.send_and_expect(self.pg2, 5 * [pgre4], self.pg0, 225)
 878         size = 0
 879         for rx in rxs:
 880             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 881             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 882             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
 883             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 884             self.assert_ip_checksum_valid(rx)
 885             self.assertEqual(rx[IP].proto, 0x2F)  # GRE encap
 886             self.assertEqual(rx[GRE].proto, 0x0800)  # IPv4
 887             inner = rx[GRE].payload
 888             self.assertEqual(rx[IP].len - 20 - 4, len(inner))
 889             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
 890             self.assertEqual(inner[IP].dst, "172.16.10.3")
 891             self.assert_ip_checksum_valid(inner)
 892             self.assert_tcp_checksum_valid(inner)
 893             payload_len = inner[IP].len - 20 - 20
 894             self.assertEqual(payload_len, len(inner[Raw]))
 895             size += payload_len
 896         self.assertEqual(size, 65200 * 5)
 897
 898         # Disable the GSO segmentation on GRE tunnel
 899         self.vapi.feature_gso_enable_disable(
 900             sw_if_index=self.gre4.sw_if_index, enable_disable=0
 901         )
 902
 903         # test again when GSO segmentation is disabled, Packets are truncated
 904         rxs = self.send_and_expect(self.pg2, 5 * [pgre4], self.pg0, 5)
 905         for rx in rxs:
 906             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 907             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 908             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
 909             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 910             self.assert_ip_checksum_valid(rx)
 911             self.assertEqual(rx[IP].proto, 0x2F)  # GRE encap
 912             self.assertEqual(rx[GRE].proto, 0x0800)  # IPv4
 913             inner = rx[GRE].payload
 914             self.assertNotEqual(rx[IP].len - 20 - 4, len(inner))
 915             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
 916             self.assertEqual(inner[IP].dst, "172.16.10.3")
 917             self.assert_ip_checksum_valid(inner)
 918             payload_len = inner[IP].len - 20 - 20
 919             self.assertEqual(payload_len, 65200)
 920             # truncated packet to MTU size
 921             self.assertNotEqual(payload_len, len(inner[Raw]))
 922
 923         self.ip4_via_gre4_tunnel.remove_vpp_config()
 924         self.gre4.remove_vpp_config()
 925
 926         self.gre6.add_vpp_config()
 927         self.gre6.admin_up()
 928         self.gre6.config_ip4()
 929
 930         #
 931         # Add a route that resolves the tunnel's destination
 932         # Add IPv6 routes via tunnel interface
 933         #
 934         self.vapi.feature_gso_enable_disable(
 935             sw_if_index=self.gre6.sw_if_index, enable_disable=1
 936         )
 937         self.ip6_via_gre6_tunnel = VppIpRoute(
 938             self,
 939             "fd01:10::",
 940             64,
 941             [
 942                 VppRoutePath(
 943                     "::",
 944                     self.gre6.sw_if_index,
 945                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
 946                 )
 947             ],
 948         )
 949         self.ip6_via_gre6_tunnel.add_vpp_config()
 950
 951         #
 952         # Create IPv6 packet
 953         #
 954         pgre6 = (
 955             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
 956             / IPv6(src=self.pg2.remote_ip6, dst="fd01:10::3")
 957             / TCP(sport=1234, dport=1234)
 958             / Raw(b"\xa5" * 65200)
 959         )
 960
 961         # test when GSO segmentation is enabled, payload will be segmented
 962         # into GSO size (i.e. 1448)
 963         rxs = self.send_and_expect(self.pg2, 5 * [pgre6], self.pg0, 225)
 964         size = 0
 965         for rx in rxs:
 966             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 967             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 968             self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
 969             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
 970             self.assertEqual(ipv6nh[rx[IPv6].nh], "GRE")
 971             self.assertEqual(rx[GRE].proto, 0x86DD)  # IPv6
 972             inner = rx[GRE].payload
 973             self.assertEqual(rx[IPv6].plen - 4, len(inner))
 974             self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
 975             self.assertEqual(inner[IPv6].dst, "fd01:10::3")
 976             self.assert_tcp_checksum_valid(inner)
 977             payload_len = inner[IPv6].plen - 20
 978             self.assertEqual(payload_len, len(inner[Raw]))
 979             size += payload_len
 980         self.assertEqual(size, 65200 * 5)
 981
 982         # disable GSO segmentation
 983         self.vapi.feature_gso_enable_disable(
 984             sw_if_index=self.gre6.sw_if_index, enable_disable=0
 985         )
 986
 987         # test again, this time packets will be truncated
 988         rxs = self.send_and_expect(self.pg2, 5 * [pgre6], self.pg0, 5)
 989         for rx in rxs:
 990             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 991             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 992             self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
 993             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
 994             self.assertEqual(ipv6nh[rx[IPv6].nh], "GRE")
 995             self.assertEqual(rx[GRE].proto, 0x86DD)  # IPv6
 996             inner = rx[GRE].payload
 997             self.assertNotEqual(rx[IPv6].plen - 4, len(inner))
 998             self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
 999             self.assertEqual(inner[IPv6].dst, "fd01:10::3")
1000             payload_len = inner[IPv6].plen - 20
1001             self.assertEqual(payload_len, 65200)
1002             # packets are truncated to MTU size
1003             self.assertNotEqual(payload_len, len(inner[Raw]))
1004
1005         self.ip6_via_gre6_tunnel.remove_vpp_config()
1006         self.gre6.remove_vpp_config()
1007
1008     def test_gso_ipsec(self):
1009         """GSO IPSEC test"""
1010         #
1011         # Send jumbo frame with gso enabled only on input interface and
1012         # create IPIP tunnel on VPP pg0.
1013         #
1014
1015         #
1016         # enable ipip4
1017         #
1018         self.ipip4.add_vpp_config()
1019         self.vapi.feature_gso_enable_disable(
1020             sw_if_index=self.ipip4.sw_if_index, enable_disable=1
1021         )
1022
1023         # Add IPv4 routes via tunnel interface
1024         self.ip4_via_ip4_tunnel = VppIpRoute(
1025             self,
1026             "172.16.10.0",
1027             24,
1028             [
1029                 VppRoutePath(
1030                     "0.0.0.0",
1031                     self.ipip4.sw_if_index,
1032                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
1033                 )
1034             ],
1035         )
1036         self.ip4_via_ip4_tunnel.add_vpp_config()
1037
1038         # IPSec config
1039         self.ipv4_params = IPsecIPv4Params()
1040         self.encryption_type = ESP
1041         config_tun_params(self.ipv4_params, self.encryption_type, self.ipip4)
1042
1043         self.tun_sa_in_v4 = VppIpsecSA(
1044             self,
1045             self.ipv4_params.scapy_tun_sa_id,
1046             self.ipv4_params.scapy_tun_spi,
1047             self.ipv4_params.auth_algo_vpp_id,
1048             self.ipv4_params.auth_key,
1049             self.ipv4_params.crypt_algo_vpp_id,
1050             self.ipv4_params.crypt_key,
1051             VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_ESP,
1052         )
1053         self.tun_sa_in_v4.add_vpp_config()
1054
1055         self.tun_sa_out_v4 = VppIpsecSA(
1056             self,
1057             self.ipv4_params.vpp_tun_sa_id,
1058             self.ipv4_params.vpp_tun_spi,
1059             self.ipv4_params.auth_algo_vpp_id,
1060             self.ipv4_params.auth_key,
1061             self.ipv4_params.crypt_algo_vpp_id,
1062             self.ipv4_params.crypt_key,
1063             VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_ESP,
1064         )
1065         self.tun_sa_out_v4.add_vpp_config()
1066
1067         self.tun_protect_v4 = VppIpsecTunProtect(
1068             self, self.ipip4, self.tun_sa_out_v4, [self.tun_sa_in_v4]
1069         )
1070
1071         self.tun_protect_v4.add_vpp_config()
1072
1073         # Set interface up and enable IP on it
1074         self.ipip4.admin_up()
1075         self.ipip4.set_unnumbered(self.pg0.sw_if_index)
1076
1077         #
1078         # IPv4/IPv4 - IPSEC
1079         #
1080         ipsec44 = (
1081             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
1082             / IP(src=self.pg2.remote_ip4, dst="172.16.10.3", flags="DF")
1083             / TCP(sport=1234, dport=1234)
1084             / Raw(b"\xa5" * 65200)
1085         )
1086
1087         rxs = self.send_and_expect(self.pg2, [ipsec44], self.pg0, 45)
1088         size = 0
1089         for rx in rxs:
1090             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
1091             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
1092             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
1093             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1094             self.assertEqual(rx[IP].proto, 50)  # ESP
1095             self.assertEqual(rx[ESP].spi, self.ipv4_params.vpp_tun_spi)
1096             inner = self.ipv4_params.vpp_tun_sa.decrypt(rx[IP])
1097             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
1098             self.assertEqual(inner[IP].dst, "172.16.10.3")
1099             size += inner[IP].len - 20 - 20
1100         self.assertEqual(size, 65200)
1101
1102         self.ip6_via_ip4_tunnel = VppIpRoute(
1103             self,
1104             "fd01:10::",
1105             64,
1106             [
1107                 VppRoutePath(
1108                     "::",
1109                     self.ipip4.sw_if_index,
1110                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
1111                 )
1112             ],
1113         )
1114         self.ip6_via_ip4_tunnel.add_vpp_config()
1115         #
1116         # IPv4/IPv6 - IPSEC
1117         #
1118         ipsec46 = (
1119             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
1120             / IPv6(src=self.pg2.remote_ip6, dst="fd01:10::3")
1121             / TCP(sport=1234, dport=1234)
1122             / Raw(b"\xa5" * 65200)
1123         )
1124
1125         rxs = self.send_and_expect(self.pg2, [ipsec46], self.pg0, 45)
1126         size = 0
1127         for rx in rxs:
1128             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
1129             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
1130             self.assertEqual(rx[IP].src, self.pg0.local_ip4)
1131             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
1132             self.assertEqual(rx[IP].proto, 50)  # ESP
1133             self.assertEqual(rx[ESP].spi, self.ipv4_params.vpp_tun_spi)
1134             inner = self.ipv4_params.vpp_tun_sa.decrypt(rx[IP])
1135             self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
1136             self.assertEqual(inner[IPv6].dst, "fd01:10::3")
1137             size += inner[IPv6].plen - 20
1138         self.assertEqual(size, 65200)
1139
1140         # disable IPSec
1141         self.tun_protect_v4.remove_vpp_config()
1142         self.tun_sa_in_v4.remove_vpp_config()
1143         self.tun_sa_out_v4.remove_vpp_config()
1144
1145         #
1146         # disable ipip4
1147         #
1148         self.vapi.feature_gso_enable_disable(self.ipip4.sw_if_index, enable_disable=0)
1149         self.ip4_via_ip4_tunnel.remove_vpp_config()
1150         self.ip6_via_ip4_tunnel.remove_vpp_config()
1151         self.ipip4.remove_vpp_config()
1152
1153         #
1154         # enable ipip6
1155         #
1156         self.ipip6.add_vpp_config()
1157         self.vapi.feature_gso_enable_disable(self.ipip6.sw_if_index, enable_disable=1)
1158
1159         # Set interface up and enable IP on it
1160         self.ipip6.admin_up()
1161         self.ipip6.set_unnumbered(self.pg0.sw_if_index)
1162
1163         # Add IPv4 routes via tunnel interface
1164         self.ip4_via_ip6_tunnel = VppIpRoute(
1165             self,
1166             "172.16.10.0",
1167             24,
1168             [
1169                 VppRoutePath(
1170                     "0.0.0.0",
1171                     self.ipip6.sw_if_index,
1172                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
1173                 )
1174             ],
1175         )
1176         self.ip4_via_ip6_tunnel.add_vpp_config()
1177
1178         # IPSec config
1179         self.ipv6_params = IPsecIPv6Params()
1180         self.encryption_type = ESP
1181         config_tun_params(self.ipv6_params, self.encryption_type, self.ipip6)
1182         self.tun_sa_in_v6 = VppIpsecSA(
1183             self,
1184             self.ipv6_params.scapy_tun_sa_id,
1185             self.ipv6_params.scapy_tun_spi,
1186             self.ipv6_params.auth_algo_vpp_id,
1187             self.ipv6_params.auth_key,
1188             self.ipv6_params.crypt_algo_vpp_id,
1189             self.ipv6_params.crypt_key,
1190             VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_ESP,
1191         )
1192         self.tun_sa_in_v6.add_vpp_config()
1193
1194         self.tun_sa_out_v6 = VppIpsecSA(
1195             self,
1196             self.ipv6_params.vpp_tun_sa_id,
1197             self.ipv6_params.vpp_tun_spi,
1198             self.ipv6_params.auth_algo_vpp_id,
1199             self.ipv6_params.auth_key,
1200             self.ipv6_params.crypt_algo_vpp_id,
1201             self.ipv6_params.crypt_key,
1202             VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_ESP,
1203         )
1204         self.tun_sa_out_v6.add_vpp_config()
1205
1206         self.tun_protect_v6 = VppIpsecTunProtect(
1207             self, self.ipip6, self.tun_sa_out_v6, [self.tun_sa_in_v6]
1208         )
1209
1210         self.tun_protect_v6.add_vpp_config()
1211
1212         #
1213         # IPv6/IPv4 - IPSEC
1214         #
1215         ipsec64 = (
1216             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
1217             / IP(src=self.pg2.remote_ip4, dst="172.16.10.3", flags="DF")
1218             / TCP(sport=1234, dport=1234)
1219             / Raw(b"\xa5" * 65200)
1220         )
1221
1222         rxs = self.send_and_expect(self.pg2, [ipsec64], self.pg0, 45)
1223         size = 0
1224         for rx in rxs:
1225             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
1226             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
1227             self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
1228             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1229             self.assertEqual(ipv6nh[rx[IPv6].nh], "ESP Header")
1230             self.assertEqual(rx[ESP].spi, self.ipv6_params.vpp_tun_spi)
1231             inner = self.ipv6_params.vpp_tun_sa.decrypt(rx[IPv6])
1232             self.assertEqual(inner[IP].src, self.pg2.remote_ip4)
1233             self.assertEqual(inner[IP].dst, "172.16.10.3")
1234             size += inner[IP].len - 20 - 20
1235         self.assertEqual(size, 65200)
1236
1237         self.ip6_via_ip6_tunnel = VppIpRoute(
1238             self,
1239             "fd01:10::",
1240             64,
1241             [
1242                 VppRoutePath(
1243                     "::",
1244                     self.ipip6.sw_if_index,
1245                     proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
1246                 )
1247             ],
1248         )
1249         self.ip6_via_ip6_tunnel.add_vpp_config()
1250
1251         #
1252         # IPv6/IPv6 - IPSEC
1253         #
1254         ipsec66 = (
1255             Ether(src=self.pg2.remote_mac, dst="02:fe:60:1e:a2:79")
1256             / IPv6(src=self.pg2.remote_ip6, dst="fd01:10::3")
1257             / TCP(sport=1234, dport=1234)
1258             / Raw(b"\xa5" * 65200)
1259         )
1260
1261         rxs = self.send_and_expect(self.pg2, [ipsec66], self.pg0, 45)
1262         size = 0
1263         for rx in rxs:
1264             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
1265             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
1266             self.assertEqual(rx[IPv6].src, self.pg0.local_ip6)
1267             self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6)
1268             self.assertEqual(ipv6nh[rx[IPv6].nh], "ESP Header")
1269             self.assertEqual(rx[ESP].spi, self.ipv6_params.vpp_tun_spi)
1270             inner = self.ipv6_params.vpp_tun_sa.decrypt(rx[IPv6])
1271             self.assertEqual(inner[IPv6].src, self.pg2.remote_ip6)
1272             self.assertEqual(inner[IPv6].dst, "fd01:10::3")
1273             size += inner[IPv6].plen - 20
1274         self.assertEqual(size, 65200)
1275
1276         # disable IPSec
1277         self.tun_protect_v6.remove_vpp_config()
1278         self.tun_sa_in_v6.remove_vpp_config()
1279         self.tun_sa_out_v6.remove_vpp_config()
1280
1281         #
1282         # disable ipip6
1283         #
1284         self.ip4_via_ip6_tunnel.remove_vpp_config()
1285         self.ip6_via_ip6_tunnel.remove_vpp_config()
1286         self.ipip6.remove_vpp_config()
1287
1288         self.vapi.feature_gso_enable_disable(self.pg0.sw_if_index, enable_disable=0)
1289
1290
1291 if __name__ == "__main__":
1292     unittest.main(testRunner=VppTestRunner)