test/test_tls.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4 import os
   5 import re
   6 import subprocess
   7
   8 from framework import VppTestCase, VppTestRunner
   9 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
  10
  11
  12 def checkQat():
  13     r = os.path.exists("/dev/qat_dev_processes")
  14     if r:
  15         return True
  16     else:
  17         # print("NO QAT! EXIT!")
  18         return False
  19
  20
  21 def checkOpenSSLVersion():
  22     ret = False
  23     r = "OPENSSL_ROOT_DIR" in os.environ
  24     if r:
  25         ssl = os.environ["OPENSSL_ROOT_DIR"] + "/bin/openssl version"
  26         p = subprocess.Popen(ssl, stdin=subprocess.PIPE,
  27                              stdout=subprocess.PIPE, shell=True)
  28         p.wait()
  29         output = p.stdout.read()
  30         status = p.returncode
  31
  32         if status:
  33             pass
  34             # print("openssl version error!")
  35         else:
  36             ssl_ver_src = re.findall(r"(\d+)\.+\d+.+\d+", output)
  37             ssl_ver = int(ssl_ver_src[0])
  38             if ssl_ver < 3:
  39                 ret = False
  40             else:
  41                 ret = True
  42     else:
  43         # print("NO OPENSSL_ROOT_DIR!")
  44         pass
  45
  46     return ret
  47
  48
  49 def checkAll():
  50     ret = checkQat() & checkOpenSSLVersion()
  51     return ret
  52
  53
  54 class TestTLS(VppTestCase):
  55     """ TLS Qat Test Case. """
  56
  57     @classmethod
  58     def setUpClass(cls):
  59         super(TestTLS, cls).setUpClass()
  60
  61     @classmethod
  62     def tearDownClass(cls):
  63         super(TestTLS, cls).tearDownClass()
  64
  65     def setUp(self):
  66         super(TestTLS, self).setUp()
  67
  68         self.vapi.session_enable_disable(is_enabled=1)
  69         self.create_loopback_interfaces(2)
  70
  71         table_id = 0
  72
  73         for i in self.lo_interfaces:
  74             i.admin_up()
  75
  76             if table_id != 0:
  77                 tbl = VppIpTable(self, table_id)
  78                 tbl.add_vpp_config()
  79
  80             i.set_table_ip4(table_id)
  81             i.config_ip4()
  82             table_id += 1
  83
  84         # Configure namespaces
  85         self.vapi.app_namespace_add_del(namespace_id="0",
  86                                         sw_if_index=self.loop0.sw_if_index)
  87         self.vapi.app_namespace_add_del(namespace_id="1",
  88                                         sw_if_index=self.loop1.sw_if_index)
  89
  90     def tearDown(self):
  91         for i in self.lo_interfaces:
  92             i.unconfig_ip4()
  93             i.set_table_ip4(0)
  94             i.admin_down()
  95         self.vapi.session_enable_disable(is_enabled=0)
  96         super(TestTLS, self).tearDown()
  97
  98     @unittest.skipUnless(checkAll(),
  99                          "QAT or OpenSSL not satisfied,skip.")
 100     def test_tls_transfer(self):
 101         """ TLS qat echo client/server transfer """
 102
 103         # Add inter-table routes
 104         ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
 105                             [VppRoutePath("0.0.0.0",
 106                                           0xffffffff,
 107                                           nh_table_id=1)])
 108
 109         ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
 110                             [VppRoutePath("0.0.0.0",
 111                                           0xffffffff,
 112                                           nh_table_id=0)], table_id=1)
 113         ip_t01.add_vpp_config()
 114         ip_t10.add_vpp_config()
 115
 116         # Enable QAT engine and TLS async
 117         r = self.vapi.tls_openssl_set_engine(
 118                 async_enable=1,
 119                 engine="qat",
 120                 algorithm="RSA,PKEY_CRYPTO",
 121                 ciphers="RSA")
 122         self.assertIsNotNone(r,
 123                              'No response msg ')
 124
 125         # Start builtin server and client
 126         uri = "tls://" + self.loop0.local_ip4 + "/1234"
 127         error = self.vapi.cli("test echo server appns 0 fifo-size 4 "
 128                               "tls-engine 1 uri " +
 129                               uri)
 130         if error:
 131             self.logger.critical(error)
 132             self.assertNotIn("failed", error)
 133
 134         error = self.vapi.cli("test echo client mbytes 10 appns 1 "
 135                               "fifo-size 4 no-output test-bytes "
 136                               "tls-engine 1 "
 137                               "syn-timeout 2 uri " + uri)
 138         if error:
 139             self.logger.critical(error)
 140             self.assertNotIn("failed", error)
 141
 142         # Delete inter-table routes
 143         ip_t01.remove_vpp_config()
 144         ip_t10.remove_vpp_config()
 145
 146
 147 if __name__ == '__main__':
 148     unittest.main(testRunner=VppTestRunner)