test/asf/test_tls.py

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