ipsec: set l2_len for GRE-TEB tunnel decap

[vpp.git] / test / test_ipsec_tun_if_esp.py

diff --git a/test/test_ipsec_tun_if_esp.py b/test/test_ipsec_tun_if_esp.py
index 1a1ce80..57c90f9 100644 (file)
--- a/test/test_ipsec_tun_if_esp.py
+++ b/test/test_ipsec_tun_if_esp.py
@@ -3,7 +3,7 @@ import socket
 import copy
 
 from scapy.layers.ipsec import SecurityAssociation, ESP
 import copy
 
 from scapy.layers.ipsec import SecurityAssociation, ESP

-from scapy.layers.l2 import Ether, GRE
+from scapy.layers.l2 import Ether, GRE, Dot1Q

 from scapy.packet import Raw
 from scapy.layers.inet import IP, UDP
 from scapy.layers.inet6 import IPv6
 from scapy.packet import Raw
 from scapy.layers.inet import IP, UDP
 from scapy.layers.inet6 import IPv6


@@ -17,6 +17,7 @@ from vpp_ipip_tun_interface import VppIpIpTunInterface
 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
 from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect
 from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
 from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect
 from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort

+from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint

 from util import ppp
 from vpp_papi import VppEnum
 
 from util import ppp
 from vpp_papi import VppEnum
 


@@ -781,6 +782,126 @@ class TestIpsecGreTebIfEsp(TemplateIpsec,
         super(TestIpsecGreTebIfEsp, self).tearDown()
 
 
         super(TestIpsecGreTebIfEsp, self).tearDown()
 
 

+class TestIpsecGreTebVlanIfEsp(TemplateIpsec,
+                               IpsecTun4Tests):
+    """ Ipsec GRE TEB ESP - TUN tests """
+    tun4_encrypt_node_name = "esp4-encrypt-tun"
+    tun4_decrypt_node_name = "esp4-decrypt-tun"
+    encryption_type = ESP
+    omac = "00:11:22:33:44:55"
+
+    def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
+                         payload_size=100):
+        return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+                sa.encrypt(IP(src=self.pg0.remote_ip4,
+                              dst=self.pg0.local_ip4) /
+                           GRE() /
+                           Ether(dst=self.omac) /
+                           IP(src="1.1.1.1", dst="1.1.1.2") /
+                           UDP(sport=1144, dport=2233) /
+                           Raw(b'X' * payload_size))
+                for i in range(count)]
+
+    def gen_pkts(self, sw_intf, src, dst, count=1,
+                 payload_size=100):
+        return [Ether(dst=self.omac) /
+                Dot1Q(vlan=11) /
+                IP(src="1.1.1.1", dst="1.1.1.2") /
+                UDP(sport=1144, dport=2233) /
+                Raw(b'X' * payload_size)
+                for i in range(count)]
+
+    def verify_decrypted(self, p, rxs):
+        for rx in rxs:
+            self.assert_equal(rx[Ether].dst, self.omac)
+            self.assert_equal(rx[Dot1Q].vlan, 11)
+            self.assert_equal(rx[IP].dst, "1.1.1.2")
+
+    def verify_encrypted(self, p, sa, rxs):
+        for rx in rxs:
+            try:
+                pkt = sa.decrypt(rx[IP])
+                if not pkt.haslayer(IP):
+                    pkt = IP(pkt[Raw].load)
+                self.assert_packet_checksums_valid(pkt)
+                self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
+                self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
+                self.assertTrue(pkt.haslayer(GRE))
+                e = pkt[Ether]
+                self.assertEqual(e[Ether].dst, self.omac)
+                self.assertFalse(e.haslayer(Dot1Q))
+                self.assertEqual(e[IP].dst, "1.1.1.2")
+            except (IndexError, AssertionError):
+                self.logger.debug(ppp("Unexpected packet:", rx))
+                try:
+                    self.logger.debug(ppp("Decrypted packet:", pkt))
+                except:
+                    pass
+                raise
+
+    def setUp(self):
+        super(TestIpsecGreTebVlanIfEsp, self).setUp()
+
+        self.tun_if = self.pg0
+
+        p = self.ipv4_params
+
+        bd1 = VppBridgeDomain(self, 1)
+        bd1.add_vpp_config()
+
+        self.pg1_11 = VppDot1QSubint(self, self.pg1, 11)
+        self.vapi.l2_interface_vlan_tag_rewrite(
+            sw_if_index=self.pg1_11.sw_if_index, vtr_op=L2_VTR_OP.L2_POP_1,
+            push_dot1q=11)
+        self.pg1_11.admin_up()
+
+        p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
+                                  p.auth_algo_vpp_id, p.auth_key,
+                                  p.crypt_algo_vpp_id, p.crypt_key,
+                                  self.vpp_esp_protocol,
+                                  self.pg0.local_ip4,
+                                  self.pg0.remote_ip4)
+        p.tun_sa_out.add_vpp_config()
+
+        p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
+                                 p.auth_algo_vpp_id, p.auth_key,
+                                 p.crypt_algo_vpp_id, p.crypt_key,
+                                 self.vpp_esp_protocol,
+                                 self.pg0.remote_ip4,
+                                 self.pg0.local_ip4)
+        p.tun_sa_in.add_vpp_config()
+
+        p.tun_if = VppGreInterface(self,
+                                   self.pg0.local_ip4,
+                                   self.pg0.remote_ip4,
+                                   type=(VppEnum.vl_api_gre_tunnel_type_t.
+                                         GRE_API_TUNNEL_TYPE_TEB))
+        p.tun_if.add_vpp_config()
+
+        p.tun_protect = VppIpsecTunProtect(self,
+                                           p.tun_if,
+                                           p.tun_sa_out,
+                                           [p.tun_sa_in])
+
+        p.tun_protect.add_vpp_config()
+
+        p.tun_if.admin_up()
+        p.tun_if.config_ip4()
+        config_tun_params(p, self.encryption_type, p.tun_if)
+
+        VppBridgeDomainPort(self, bd1, p.tun_if).add_vpp_config()
+        VppBridgeDomainPort(self, bd1, self.pg1_11).add_vpp_config()
+
+        self.vapi.cli("clear ipsec sa")
+
+    def tearDown(self):
+        p = self.ipv4_params
+        p.tun_if.unconfig_ip4()
+        super(TestIpsecGreTebVlanIfEsp, self).tearDown()
+        self.pg1_11.admin_down()
+        self.pg1_11.remove_vpp_config()
+
+

 class TestIpsecGreTebIfEspTra(TemplateIpsec,
                               IpsecTun4Tests):
     """ Ipsec GRE TEB ESP - Tra tests """
 class TestIpsecGreTebIfEspTra(TemplateIpsec,
                               IpsecTun4Tests):
     """ Ipsec GRE TEB ESP - Tra tests """