X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftemplate_ipsec.py;h=0c1f5a1929839ae48223d88467a1150fbb93355b;hb=c7eaa711f;hp=1caed0dd932a61866fc7977c9b7f33ba4ca123ff;hpb=582eac5c30e78f65ba0127aed2cb7442f4122fd2;p=vpp.git diff --git a/test/template_ipsec.py b/test/template_ipsec.py index 1caed0dd932..0c1f5a19298 100644 --- a/test/template_ipsec.py +++ b/test/template_ipsec.py @@ -15,7 +15,7 @@ from util import ppp, reassemble4, fragment_rfc791, fragment_rfc8200 from vpp_papi import VppEnum -class IPsecIPv4Params(object): +class IPsecIPv4Params: addr_type = socket.AF_INET addr_any = "0.0.0.0" @@ -28,14 +28,19 @@ class IPsecIPv4Params(object): self.remote_tun_if_host6 = '1111::1' self.scapy_tun_sa_id = 100 - self.scapy_tun_spi = 1001 + self.scapy_tun_spi = 1000 self.vpp_tun_sa_id = 200 - self.vpp_tun_spi = 1000 + self.vpp_tun_spi = 2000 self.scapy_tra_sa_id = 300 - self.scapy_tra_spi = 2001 + self.scapy_tra_spi = 3000 self.vpp_tra_sa_id = 400 - self.vpp_tra_spi = 2000 + self.vpp_tra_spi = 4000 + + self.outer_hop_limit = 64 + self.inner_hop_limit = 255 + self.outer_flow_label = 0 + self.inner_flow_label = 0x12345 self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. IPSEC_API_INTEG_ALG_SHA1_96) @@ -49,9 +54,12 @@ class IPsecIPv4Params(object): self.salt = 0 self.flags = 0 self.nat_header = None + self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t. + TUNNEL_API_ENCAP_DECAP_FLAG_NONE) + self.dscp = 0 -class IPsecIPv6Params(object): +class IPsecIPv6Params: addr_type = socket.AF_INET6 addr_any = "0::0" @@ -73,6 +81,11 @@ class IPsecIPv6Params(object): self.vpp_tra_sa_id = 800 self.vpp_tra_spi = 4000 + self.outer_hop_limit = 64 + self.inner_hop_limit = 255 + self.outer_flow_label = 0 + self.inner_flow_label = 0x12345 + self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. IPSEC_API_INTEG_ALG_SHA1_96) self.auth_algo = 'HMAC-SHA1-96' # scapy name @@ -85,10 +98,13 @@ class IPsecIPv6Params(object): self.salt = 0 self.flags = 0 self.nat_header = None + self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t. + TUNNEL_API_ENCAP_DECAP_FLAG_NONE) + self.dscp = 0 def mk_scapy_crypt_key(p): - if p.crypt_algo == "AES-GCM": + if p.crypt_algo in ("AES-GCM", "AES-CTR"): return p.crypt_key + struct.pack("!I", p.salt) else: return p.crypt_key @@ -181,8 +197,10 @@ class TemplateIpsec(VppTestCase): super(TemplateIpsec, cls).tearDownClass() def setup_params(self): - self.ipv4_params = IPsecIPv4Params() - self.ipv6_params = IPsecIPv6Params() + if not hasattr(self, 'ipv4_params'): + self.ipv4_params = IPsecIPv4Params() + if not hasattr(self, 'ipv6_params'): + self.ipv6_params = IPsecIPv6Params() self.params = {self.ipv4_params.addr_type: self.ipv4_params, self.ipv6_params.addr_type: self.ipv6_params} @@ -234,7 +252,9 @@ class TemplateIpsec(VppTestCase): def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=54): return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) / - sa.encrypt(IPv6(src=src, dst=dst) / + sa.encrypt(IPv6(src=src, dst=dst, + hlim=p.inner_hop_limit, + fl=p.inner_flow_label) / ICMPv6EchoRequest(id=0, seq=1, data='X' * payload_size)) for i in range(count)] @@ -244,9 +264,10 @@ class TemplateIpsec(VppTestCase): IP(src=src, dst=dst) / ICMP() / Raw(b'X' * payload_size) for i in range(count)] - def gen_pkts6(self, sw_intf, src, dst, count=1, payload_size=54): + def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=54): return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) / - IPv6(src=src, dst=dst) / + IPv6(src=src, dst=dst, + hlim=p.inner_hop_limit, fl=p.inner_flow_label) / ICMPv6EchoRequest(id=0, seq=1, data='X' * payload_size) for i in range(count)] @@ -790,7 +811,7 @@ class IpsecTun4(object): "incorrect SA in counts: expected %d != %d" % (count, pkts)) pkts = p.tun_sa_out.get_stats(worker)['packets'] - self.assertEqual(pkts, count, + self.assertEqual(pkts, n_frags, "incorrect SA out counts: expected %d != %d" % (count, pkts)) @@ -803,6 +824,15 @@ class IpsecTun4(object): self.assert_equal(rx[IP].dst, self.pg1.remote_ip4) self.assert_packet_checksums_valid(rx) + def verify_esp_padding(self, sa, esp_payload, decrypt_pkt): + align = sa.crypt_algo.block_size + if align < 4: + align = 4 + exp_len = (len(decrypt_pkt) + 2 + (align - 1)) & ~(align - 1) + exp_len += sa.crypt_algo.iv_size + exp_len += sa.crypt_algo.icv_size or sa.auth_algo.icv_size + self.assertEqual(exp_len, len(esp_payload)) + def verify_encrypted(self, p, sa, rxs): decrypt_pkts = [] for rx in rxs: @@ -811,9 +841,12 @@ class IpsecTun4(object): self.assert_packet_checksums_valid(rx) self.assertEqual(len(rx) - len(Ether()), rx[IP].len) try: - decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IP]) + rx_ip = rx[IP] + decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip) if not decrypt_pkt.haslayer(IP): decrypt_pkt = IP(decrypt_pkt[Raw].load) + if rx_ip.proto == socket.IPPROTO_ESP: + self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt) decrypt_pkts.append(decrypt_pkt) self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4) self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host) @@ -914,6 +947,7 @@ class IpsecTun4(object): def verify_tun_64(self, p, count=1): self.vapi.cli("clear errors") + self.vapi.cli("clear ipsec sa") try: send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if, src=p.remote_tun_if_host6, @@ -924,7 +958,7 @@ class IpsecTun4(object): self.assert_equal(recv_pkt[IPv6].src, p.remote_tun_if_host6) self.assert_equal(recv_pkt[IPv6].dst, self.pg1.remote_ip6) self.assert_packet_checksums_valid(recv_pkt) - send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6, + send_pkts = self.gen_pkts6(p, self.pg1, src=self.pg1.remote_ip6, dst=p.remote_tun_if_host6, count=count) recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if) for recv_pkt in recv_pkts: @@ -1013,6 +1047,9 @@ class IpsecTun6(object): self.assert_packet_checksums_valid(rx) self.assertEqual(len(rx) - len(Ether()) - len(IPv6()), rx[IPv6].plen) + self.assert_equal(rx[IPv6].hlim, p.outer_hop_limit) + if p.outer_flow_label: + self.assert_equal(rx[IPv6].fl, p.outer_flow_label) try: decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IPv6]) if not decrypt_pkt.haslayer(IPv6): @@ -1020,6 +1057,8 @@ class IpsecTun6(object): self.assert_packet_checksums_valid(decrypt_pkt) self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6) self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host) + self.assert_equal(decrypt_pkt.hlim, p.inner_hop_limit - 1) + self.assert_equal(decrypt_pkt.fl, p.inner_flow_label) except: self.logger.debug(ppp("Unexpected packet:", rx)) try: @@ -1055,7 +1094,7 @@ class IpsecTun6(object): recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1) self.verify_decrypted6(p_in, recv_pkts) - send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6, + send_pkts = self.gen_pkts6(p_in, self.pg1, src=self.pg1.remote_ip6, dst=p_out.remote_tun_if_host, count=count, payload_size=payload_size) @@ -1087,7 +1126,7 @@ class IpsecTun6(object): self.pg1, n_rx=1) self.verify_decrypted6(p, recv_pkts) - send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6, + send_pkts = self.gen_pkts6(p, self.pg1, src=self.pg1.remote_ip6, dst=p.remote_tun_if_host, count=1, payload_size=64) @@ -1104,6 +1143,7 @@ class IpsecTun6(object): def verify_tun_46(self, p, count=1): """ ipsec 4o6 tunnel basic test """ self.vapi.cli("clear errors") + self.vapi.cli("clear ipsec sa") try: send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if, src=p.remote_tun_if_host4, @@ -1176,7 +1216,7 @@ class IpsecTun6HandoffTests(IpsecTun6): self.pg1, worker=worker) self.verify_decrypted6(p, recv_pkts) - send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6, + send_pkts = self.gen_pkts6(p, self.pg1, src=self.pg1.remote_ip6, dst=p.remote_tun_if_host, count=N_PKTS) recv_pkts = self.send_and_expect(self.pg1, send_pkts,