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