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, IpsecTra46Tests, IpsecTun46Tests, \
   8     config_tun_params, config_tra_params
   9 from template_ipsec import IpsecTcpTests
  10 from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\
  11         VppIpsecSpdItfBinding
  12 from vpp_ip_route import VppIpRoute, VppRoutePath
  13 from vpp_ip import DpoProto
  14 from vpp_papi import VppEnum
  15
  16
  17 class TemplateIpsecAh(TemplateIpsec):
  18     """
  19     Basic test for IPSEC using AH transport and Tunnel mode
  20
  21     TRANSPORT MODE:
  22
  23      ---   encrypt   ---
  24     |pg2| <-------> |VPP|
  25      ---   decrypt   ---
  26
  27     TUNNEL MODE:
  28
  29      ---   encrypt   ---   plain   ---
  30     |pg0| <-------  |VPP| <------ |pg1|
  31      ---             ---           ---
  32
  33      ---   decrypt   ---   plain   ---
  34     |pg0| ------->  |VPP| ------> |pg1|
  35      ---             ---           ---
  36     """
  37
  38     def setUp(self):
  39         super(TemplateIpsecAh, self).setUp()
  40
  41         self.encryption_type = AH
  42         self.tun_if = self.pg0
  43         self.tra_if = self.pg2
  44         self.logger.info(self.vapi.ppcli("show int addr"))
  45
  46         self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
  47         self.tra_spd.add_vpp_config()
  48         VppIpsecSpdItfBinding(self, self.tra_spd,
  49                               self.tra_if).add_vpp_config()
  50         self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
  51         self.tun_spd.add_vpp_config()
  52         VppIpsecSpdItfBinding(self, self.tun_spd,
  53                               self.tun_if).add_vpp_config()
  54
  55         for _, p in self.params.items():
  56             self.config_ah_tra(p)
  57             config_tra_params(p, self.encryption_type)
  58             self.logger.info(self.vapi.ppcli("show ipsec"))
  59         for _, p in self.params.items():
  60             self.config_ah_tun(p)
  61             self.logger.info(self.vapi.ppcli("show ipsec"))
  62         for _, p in self.params.items():
  63             d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
  64             VppIpRoute(self,  p.remote_tun_if_host, p.addr_len,
  65                        [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
  66                                      0xffffffff,
  67                                      proto=d)],
  68                        is_ip6=p.is_ipv6).add_vpp_config()
  69
  70     def tearDown(self):
  71         super(TemplateIpsecAh, self).tearDown()
  72         if not self.vpp_dead:
  73             self.vapi.cli("show hardware")
  74
  75     def config_ah_tun(self, params):
  76         addr_type = params.addr_type
  77         scapy_tun_sa_id = params.scapy_tun_sa_id
  78         scapy_tun_spi = params.scapy_tun_spi
  79         vpp_tun_sa_id = params.vpp_tun_sa_id
  80         vpp_tun_spi = params.vpp_tun_spi
  81         auth_algo_vpp_id = params.auth_algo_vpp_id
  82         auth_key = params.auth_key
  83         crypt_algo_vpp_id = params.crypt_algo_vpp_id
  84         crypt_key = params.crypt_key
  85         remote_tun_if_host = params.remote_tun_if_host
  86         addr_any = params.addr_any
  87         addr_bcast = params.addr_bcast
  88         e = VppEnum.vl_api_ipsec_spd_action_t
  89
  90         params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
  91                                       auth_algo_vpp_id, auth_key,
  92                                       crypt_algo_vpp_id, crypt_key,
  93                                       self.vpp_ah_protocol,
  94                                       self.tun_if.local_addr[addr_type],
  95                                       self.tun_if.remote_addr[addr_type])
  96         params.tun_sa_in.add_vpp_config()
  97         params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
  98                                        auth_algo_vpp_id, auth_key,
  99                                        crypt_algo_vpp_id, crypt_key,
 100                                        self.vpp_ah_protocol,
 101                                        self.tun_if.remote_addr[addr_type],
 102                                        self.tun_if.local_addr[addr_type])
 103         params.tun_sa_out.add_vpp_config()
 104
 105         params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
 106                                                     vpp_tun_sa_id,
 107                                                     addr_any, addr_bcast,
 108                                                     addr_any, addr_bcast,
 109                                                     socket.IPPROTO_AH)
 110         params.spd_policy_in_any.add_vpp_config()
 111         params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
 112                                                      vpp_tun_sa_id,
 113                                                      addr_any, addr_bcast,
 114                                                      addr_any, addr_bcast,
 115                                                      socket.IPPROTO_AH,
 116                                                      is_outbound=0)
 117         params.spd_policy_out_any.add_vpp_config()
 118
 119         VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 120                          remote_tun_if_host,
 121                          remote_tun_if_host,
 122                          self.pg1.remote_addr[addr_type],
 123                          self.pg1.remote_addr[addr_type],
 124                          0, priority=10,
 125                          policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 126                          is_outbound=0).add_vpp_config()
 127         VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 128                          self.pg1.remote_addr[addr_type],
 129                          self.pg1.remote_addr[addr_type],
 130                          remote_tun_if_host,
 131                          remote_tun_if_host,
 132                          0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 133                          priority=10).add_vpp_config()
 134
 135         VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 136                          remote_tun_if_host,
 137                          remote_tun_if_host,
 138                          self.pg0.local_addr[addr_type],
 139                          self.pg0.local_addr[addr_type],
 140                          0, priority=20,
 141                          policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 142                          is_outbound=0).add_vpp_config()
 143         VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 144                          self.pg0.local_addr[addr_type],
 145                          self.pg0.local_addr[addr_type],
 146                          remote_tun_if_host,
 147                          remote_tun_if_host,
 148                          0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 149                          priority=20).add_vpp_config()
 150
 151     def config_ah_tra(self, params):
 152         addr_type = params.addr_type
 153         scapy_tra_sa_id = params.scapy_tra_sa_id
 154         scapy_tra_spi = params.scapy_tra_spi
 155         vpp_tra_sa_id = params.vpp_tra_sa_id
 156         vpp_tra_spi = params.vpp_tra_spi
 157         auth_algo_vpp_id = params.auth_algo_vpp_id
 158         auth_key = params.auth_key
 159         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 160         crypt_key = params.crypt_key
 161         addr_any = params.addr_any
 162         addr_bcast = params.addr_bcast
 163         flags = (VppEnum.vl_api_ipsec_sad_flags_t.
 164                  IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
 165         e = VppEnum.vl_api_ipsec_spd_action_t
 166
 167         params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
 168                                       auth_algo_vpp_id, auth_key,
 169                                       crypt_algo_vpp_id, crypt_key,
 170                                       self.vpp_ah_protocol,
 171                                       flags=flags)
 172         params.tra_sa_in.add_vpp_config()
 173         params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
 174                                        auth_algo_vpp_id, auth_key,
 175                                        crypt_algo_vpp_id, crypt_key,
 176                                        self.vpp_ah_protocol,
 177                                        flags=flags)
 178         params.tra_sa_out.add_vpp_config()
 179
 180         VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 181                          addr_any, addr_bcast,
 182                          addr_any, addr_bcast,
 183                          socket.IPPROTO_AH).add_vpp_config()
 184         VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
 185                          addr_any, addr_bcast,
 186                          addr_any, addr_bcast,
 187                          socket.IPPROTO_AH,
 188                          is_outbound=0).add_vpp_config()
 189
 190         VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 191                          self.tra_if.local_addr[addr_type],
 192                          self.tra_if.local_addr[addr_type],
 193                          self.tra_if.remote_addr[addr_type],
 194                          self.tra_if.remote_addr[addr_type],
 195                          0, priority=10,
 196                          policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 197                          is_outbound=0).add_vpp_config()
 198         VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
 199                          self.tra_if.local_addr[addr_type],
 200                          self.tra_if.local_addr[addr_type],
 201                          self.tra_if.remote_addr[addr_type],
 202                          self.tra_if.remote_addr[addr_type],
 203                          0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 204                          priority=10).add_vpp_config()
 205
 206
 207 class TestIpsecAh1(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
 208     """ Ipsec AH - TUN & TRA tests """
 209     tra4_encrypt_node_name = "ah4-encrypt"
 210     tra4_decrypt_node_name = "ah4-decrypt"
 211     tra6_encrypt_node_name = "ah6-encrypt"
 212     tra6_decrypt_node_name = "ah6-decrypt"
 213     tun4_encrypt_node_name = "ah4-encrypt"
 214     tun4_decrypt_node_name = "ah4-decrypt"
 215     tun6_encrypt_node_name = "ah6-encrypt"
 216     tun6_decrypt_node_name = "ah6-decrypt"
 217
 218
 219 class TestIpsecAh2(TemplateIpsecAh, IpsecTcpTests):
 220     """ Ipsec AH - TCP tests """
 221     pass
 222
 223
 224 if __name__ == '__main__':
 225     unittest.main(testRunner=VppTestRunner)