+class TestIpsecAhTun(TemplateIpsecAh, IpsecTun46Tests):
+ """Ipsec AH - TUN encap tests"""
+
+ def setUp(self):
+ self.ipv4_params = IPsecIPv4Params()
+ self.ipv6_params = IPsecIPv6Params()
+
+ c = (
+ VppEnum.vl_api_tunnel_encap_decap_flags_t.TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_DSCP
+ )
+ c1 = c | (
+ VppEnum.vl_api_tunnel_encap_decap_flags_t.TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_ECN
+ )
+
+ self.ipv4_params.tun_flags = c
+ self.ipv6_params.tun_flags = c1
+
+ super(TestIpsecAhTun, self).setUp()
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy only DSCP
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IP(src=src, dst=dst, tos=5)
+ / UDP(sport=4444, dport=4444)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
+
+ def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy both
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IPv6(src=src, dst=dst, tc=5)
+ / UDP(sport=4444, dport=4444)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
+
+ def verify_encrypted(self, p, sa, rxs):
+ # just check that only the DSCP is copied
+ for rx in rxs:
+ self.assertEqual(rx[IP].tos, 4)
+
+ def verify_encrypted6(self, p, sa, rxs):
+ # just check that the DSCP & ECN are copied
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].tc, 5)
+
+
+class TestIpsecAhTun2(TemplateIpsecAh, IpsecTun46Tests):
+ """Ipsec AH - TUN encap tests"""
+
+ def setUp(self):
+ self.ipv4_params = IPsecIPv4Params()
+ self.ipv6_params = IPsecIPv6Params()
+
+ self.ipv4_params.dscp = 3
+ self.ipv6_params.dscp = 4
+
+ super(TestIpsecAhTun2, self).setUp()
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy only DSCP
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IP(src=src, dst=dst, tos=0)
+ / UDP(sport=4444, dport=4444)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
+
+ def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy both
+ return [
+ Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
+ / IPv6(src=src, dst=dst, tc=0)
+ / UDP(sport=4444, dport=4444)
+ / Raw(b"X" * payload_size)
+ for i in range(count)
+ ]
+
+ def verify_encrypted(self, p, sa, rxs):
+ # just check that only the DSCP is copied
+ for rx in rxs:
+ self.assertEqual(rx[IP].tos, 0xC)
+
+ def verify_encrypted6(self, p, sa, rxs):
+ # just check that the DSCP & ECN are copied
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].tc, 0x10)
+
+
+class TestIpsecAhHandoff(TemplateIpsecAh, IpsecTun6HandoffTests, IpsecTun4HandoffTests):
+ """Ipsec AH Handoff"""
+
+ pass
+
+
+class TestIpsecAhAll(ConfigIpsecAH, IpsecTra4, IpsecTra6, IpsecTun4, IpsecTun6):
+ """Ipsec AH all Algos"""
+
+ def setUp(self):
+ super(TestIpsecAhAll, self).setUp()
+
+ def tearDown(self):
+ super(TestIpsecAhAll, self).tearDown()
+
+ def test_integ_algs(self):
+ """All Engines SHA[1_96, 256, 384, 512] w/ & w/o ESN"""
+ # foreach VPP crypto engine
+ engines = ["ia32", "ipsecmb", "openssl"]
+
+ algos = [
+ {
+ "vpp": VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA1_96,
+ "scapy": "HMAC-SHA1-96",
+ },
+ {
+ "vpp": VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA_256_128,
+ "scapy": "SHA2-256-128",
+ },
+ {
+ "vpp": VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA_384_192,
+ "scapy": "SHA2-384-192",
+ },
+ {
+ "vpp": VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA_512_256,
+ "scapy": "SHA2-512-256",
+ },
+ ]
+
+ flags = [0, (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)]
+
+ #
+ # loop through the VPP engines
+ #
+ for engine in engines:
+ self.vapi.cli("set crypto handler all %s" % engine)
+ #
+ # loop through each of the algorithms
+ #
+ for algo in algos:
+ # with self.subTest(algo=algo['scapy']):
+ for flag in flags:
+ #
+ # setup up the config paramters
+ #
+ self.ipv4_params = IPsecIPv4Params()
+ self.ipv6_params = IPsecIPv6Params()
+
+ self.params = {
+ self.ipv4_params.addr_type: self.ipv4_params,
+ self.ipv6_params.addr_type: self.ipv6_params,
+ }
+
+ for _, p in self.params.items():
+ p.auth_algo_vpp_id = algo["vpp"]
+ p.auth_algo = algo["scapy"]
+ p.flags = p.flags | flag
+
+ #
+ # configure the SPDs. SAs, etc
+ #
+ self.config_network(self.params.values())
+
+ #
+ # run some traffic.
+ # An exhautsive 4o6, 6o4 is not necessary for each algo
+ #
+ self.verify_tra_basic6(count=17)
+ self.verify_tra_basic4(count=17)
+ self.verify_tun_66(self.params[socket.AF_INET6], count=17)
+ self.verify_tun_44(self.params[socket.AF_INET], count=17)
+
+ #
+ # remove the SPDs, SAs, etc
+ #
+ self.unconfig_network()
+
+
+if __name__ == "__main__":