test/test_ipsec_ah.py

   1 import socket
   2 import unittest
   3
   4 from scapy.layers.ipsec import AH
   5
   6 from framework import VppTestRunner
   7 from template_ipsec import TemplateIpsec, IpsecTraTests, IpsecTunTests
   8 from template_ipsec import IpsecTcpTests
   9
  10
  11 class TemplateIpsecAh(TemplateIpsec):
  12     """
  13     Basic test for IPSEC using AH transport and Tunnel mode
  14
  15     TRANSPORT MODE:
  16
  17      ---   encrypt   ---
  18     |pg2| <-------> |VPP|
  19      ---   decrypt   ---
  20
  21     TUNNEL MODE:
  22
  23      ---   encrypt   ---   plain   ---
  24     |pg0| <-------  |VPP| <------ |pg1|
  25      ---             ---           ---
  26
  27      ---   decrypt   ---   plain   ---
  28     |pg0| ------->  |VPP| ------> |pg1|
  29      ---             ---           ---
  30     """
  31
  32     encryption_type = AH
  33
  34     @classmethod
  35     def setUpClass(cls):
  36         super(TemplateIpsecAh, cls).setUpClass()
  37         cls.tun_if = cls.pg0
  38         cls.tra_if = cls.pg2
  39         cls.logger.info(cls.vapi.ppcli("show int addr"))
  40         cls.vapi.ipsec_spd_add_del(cls.tun_spd_id)
  41         cls.vapi.ipsec_interface_add_del_spd(cls.tun_spd_id,
  42                                              cls.tun_if.sw_if_index)
  43         cls.vapi.ipsec_spd_add_del(cls.tra_spd_id)
  44         cls.vapi.ipsec_interface_add_del_spd(cls.tra_spd_id,
  45                                              cls.tra_if.sw_if_index)
  46         for _, p in cls.params.items():
  47             cls.config_ah_tra(p)
  48         cls.logger.info(cls.vapi.ppcli("show ipsec"))
  49         for _, p in cls.params.items():
  50             cls.config_ah_tun(p)
  51         cls.logger.info(cls.vapi.ppcli("show ipsec"))
  52         for _, p in cls.params.items():
  53             src = socket.inet_pton(p.addr_type, p.remote_tun_if_host)
  54             cls.vapi.ip_add_del_route(src, p.addr_len,
  55                                       cls.tun_if.remote_addr_n[p.addr_type],
  56                                       is_ipv6=p.is_ipv6)
  57
  58     @classmethod
  59     def config_ah_tun(cls, params):
  60         addr_type = params.addr_type
  61         is_ipv6 = params.is_ipv6
  62         scapy_tun_sa_id = params.scapy_tun_sa_id
  63         scapy_tun_spi = params.scapy_tun_spi
  64         vpp_tun_sa_id = params.vpp_tun_sa_id
  65         vpp_tun_spi = params.vpp_tun_spi
  66         auth_algo_vpp_id = params.auth_algo_vpp_id
  67         auth_key = params.auth_key
  68         crypt_algo_vpp_id = params.crypt_algo_vpp_id
  69         crypt_key = params.crypt_key
  70         remote_tun_if_host = params.remote_tun_if_host
  71         addr_any = params.addr_any
  72         addr_bcast = params.addr_bcast
  73         cls.vapi.ipsec_sad_add_del_entry(scapy_tun_sa_id, scapy_tun_spi,
  74                                          auth_algo_vpp_id, auth_key,
  75                                          crypt_algo_vpp_id, crypt_key,
  76                                          cls.vpp_ah_protocol,
  77                                          cls.tun_if.local_addr_n[addr_type],
  78                                          cls.tun_if.remote_addr_n[addr_type],
  79                                          is_tunnel=1, is_tunnel_ipv6=is_ipv6)
  80         cls.vapi.ipsec_sad_add_del_entry(vpp_tun_sa_id, vpp_tun_spi,
  81                                          auth_algo_vpp_id, auth_key,
  82                                          crypt_algo_vpp_id, crypt_key,
  83                                          cls.vpp_ah_protocol,
  84                                          cls.tun_if.remote_addr_n[addr_type],
  85                                          cls.tun_if.local_addr_n[addr_type],
  86                                          is_tunnel=1, is_tunnel_ipv6=is_ipv6)
  87         l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any)
  88         l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast)
  89         cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, vpp_tun_sa_id,
  90                                          l_startaddr, l_stopaddr, r_startaddr,
  91                                          r_stopaddr, is_ipv6=is_ipv6,
  92                                          protocol=socket.IPPROTO_AH)
  93         cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, vpp_tun_sa_id,
  94                                          l_startaddr, l_stopaddr, r_startaddr,
  95                                          r_stopaddr, is_outbound=0,
  96                                          is_ipv6=is_ipv6,
  97                                          protocol=socket.IPPROTO_AH)
  98         l_startaddr = l_stopaddr = socket.inet_pton(addr_type,
  99                                                     remote_tun_if_host)
 100         r_startaddr = r_stopaddr = cls.pg1.remote_addr_n[addr_type]
 101         cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, vpp_tun_sa_id,
 102                                          l_startaddr, l_stopaddr, r_startaddr,
 103                                          r_stopaddr, priority=10, policy=3,
 104                                          is_outbound=0, is_ipv6=is_ipv6)
 105         cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, scapy_tun_sa_id,
 106                                          r_startaddr, r_stopaddr, l_startaddr,
 107                                          l_stopaddr, priority=10, policy=3,
 108                                          is_ipv6=is_ipv6)
 109         r_startaddr = r_stopaddr = cls.pg0.local_addr_n[addr_type]
 110         cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, vpp_tun_sa_id,
 111                                          l_startaddr, l_stopaddr, r_startaddr,
 112                                          r_stopaddr, priority=20, policy=3,
 113                                          is_outbound=0, is_ipv6=is_ipv6)
 114         cls.vapi.ipsec_spd_add_del_entry(cls.tun_spd_id, scapy_tun_sa_id,
 115                                          r_startaddr, r_stopaddr, l_startaddr,
 116                                          l_stopaddr, priority=20, policy=3,
 117                                          is_ipv6=is_ipv6)
 118
 119     @classmethod
 120     def config_ah_tra(cls, params):
 121         addr_type = params.addr_type
 122         is_ipv6 = params.is_ipv6
 123         scapy_tra_sa_id = params.scapy_tra_sa_id
 124         scapy_tra_spi = params.scapy_tra_spi
 125         vpp_tra_sa_id = params.vpp_tra_sa_id
 126         vpp_tra_spi = params.vpp_tra_spi
 127         auth_algo_vpp_id = params.auth_algo_vpp_id
 128         auth_key = params.auth_key
 129         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 130         crypt_key = params.crypt_key
 131         addr_any = params.addr_any
 132         addr_bcast = params.addr_bcast
 133         cls.vapi.ipsec_sad_add_del_entry(scapy_tra_sa_id, scapy_tra_spi,
 134                                          auth_algo_vpp_id, auth_key,
 135                                          crypt_algo_vpp_id, crypt_key,
 136                                          cls.vpp_ah_protocol, is_tunnel=0,
 137                                          is_tunnel_ipv6=0)
 138         cls.vapi.ipsec_sad_add_del_entry(vpp_tra_sa_id, vpp_tra_spi,
 139                                          auth_algo_vpp_id, auth_key,
 140                                          crypt_algo_vpp_id, crypt_key,
 141                                          cls.vpp_ah_protocol, is_tunnel=0,
 142                                          is_tunnel_ipv6=0)
 143         l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any)
 144         l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast)
 145         cls.vapi.ipsec_spd_add_del_entry(cls.tra_spd_id, vpp_tra_sa_id,
 146                                          l_startaddr, l_stopaddr, r_startaddr,
 147                                          r_stopaddr, is_ipv6=is_ipv6,
 148                                          protocol=socket.IPPROTO_AH)
 149         cls.vapi.ipsec_spd_add_del_entry(cls.tra_spd_id, scapy_tra_sa_id,
 150                                          l_startaddr, l_stopaddr, r_startaddr,
 151                                          r_stopaddr, is_outbound=0,
 152                                          is_ipv6=is_ipv6,
 153                                          protocol=socket.IPPROTO_AH)
 154         l_startaddr = l_stopaddr = cls.tra_if.local_addr_n[addr_type]
 155         r_startaddr = r_stopaddr = cls.tra_if.remote_addr_n[addr_type]
 156         cls.vapi.ipsec_spd_add_del_entry(cls.tra_spd_id, vpp_tra_sa_id,
 157                                          l_startaddr, l_stopaddr, r_startaddr,
 158                                          r_stopaddr, priority=10, policy=3,
 159                                          is_outbound=0, is_ipv6=is_ipv6)
 160         cls.vapi.ipsec_spd_add_del_entry(cls.tra_spd_id, scapy_tra_sa_id,
 161                                          l_startaddr, l_stopaddr, r_startaddr,
 162                                          r_stopaddr, priority=10,
 163                                          policy=3, is_ipv6=is_ipv6)
 164
 165     def tearDown(self):
 166         super(TemplateIpsecAh, self).tearDown()
 167         if not self.vpp_dead:
 168             self.vapi.cli("show hardware")
 169
 170
 171 class TestIpsecAh1(TemplateIpsecAh, IpsecTraTests, IpsecTunTests):
 172     """ Ipsec AH - TUN & TRA tests """
 173     tra4_encrypt_node_name = "ah4-encrypt"
 174     tra4_decrypt_node_name = "ah4-decrypt"
 175     tra6_encrypt_node_name = "ah6-encrypt"
 176     tra6_decrypt_node_name = "ah6-decrypt"
 177     tun4_encrypt_node_name = "ah4-encrypt"
 178     tun4_decrypt_node_name = "ah4-decrypt"
 179     tun6_encrypt_node_name = "ah6-encrypt"
 180     tun6_decrypt_node_name = "ah6-decrypt"
 181
 182
 183 class TestIpsecAh2(TemplateIpsecAh, IpsecTcpTests):
 184     """ Ipsec AH - TCP tests """
 185     pass
 186
 187
 188 if __name__ == '__main__':
 189     unittest.main(testRunner=VppTestRunner)