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