tests/aaf/certservice/libraries/CertClientManager.py

   1 import docker
   2 import os
   3 import shutil
   4 import re
   5 from OpenSSL import crypto
   6 from docker.types import Mount
   7
   8 ARCHIVES_PATH = os.getenv("WORKSPACE") + "/archives/"
   9 MOUNT_PATH = os.getenv("WORKSPACE") + "/tests/aaf/certservice/tmp"
  10
  11 ERROR_API_REGEX = 'Error on API response.*[0-9]{3}'
  12 RESPONSE_CODE_REGEX = '[0-9]{3}'
  13
  14
  15 class CertClientManager:
  16
  17     def run_client_container(self, client_image, container_name, path_to_env, request_url, network):
  18         self.create_mount_dir()
  19         client = docker.from_env()
  20         environment = self.read_list_env_from_file(path_to_env)
  21         environment.append("REQUEST_URL=" + request_url)
  22         container = client.containers.run(
  23             image=client_image,
  24             name=container_name,
  25             environment=environment,
  26             network=network,
  27             user='root', #Run container as root to avoid permission issues with volume mount access
  28             mounts=[Mount(target='/var/certs', source=MOUNT_PATH, type='bind')],
  29             detach=True
  30         )
  31         exitcode = container.wait()
  32         return exitcode
  33
  34     def read_list_env_from_file(self, path):
  35         f = open(path, "r")
  36         r_list = []
  37         for line in f:
  38             line = line.strip()
  39             if line[0] != "#":
  40                 r_list.append(line)
  41         return r_list
  42
  43     def remove_client_container_and_save_logs(self, container_name, log_file_name):
  44         client = docker.from_env()
  45         container = client.containers.get(container_name)
  46         text_file = open(ARCHIVES_PATH + "container_" + log_file_name + ".log", "w")
  47         text_file.write(container.logs())
  48         text_file.close()
  49         container.remove()
  50         self.remove_mount_dir()
  51
  52     def can_open_keystore_and_truststore_with_pass(self):
  53         keystore_pass_path = MOUNT_PATH + '/keystore.pass'
  54         keystore_jks_path = MOUNT_PATH + '/keystore.jks'
  55         can_open_keystore = self.can_open_jks_file_by_pass_file(keystore_pass_path, keystore_jks_path)
  56
  57         truststore_pass_path = MOUNT_PATH + '/truststore.pass'
  58         truststore_jks_path = MOUNT_PATH + '/truststore.jks'
  59         can_open_truststore = self.can_open_jks_file_by_pass_file(truststore_pass_path, truststore_jks_path)
  60
  61         return can_open_keystore & can_open_truststore
  62
  63     def can_open_jks_file_by_pass_file(self, pass_file_path, jks_file_path):
  64         try:
  65             password = open(pass_file_path, 'rb').read()
  66             crypto.load_pkcs12(open(jks_file_path, 'rb').read(), password)
  67             return True
  68         except:
  69             return False
  70
  71     def create_mount_dir(self):
  72         if not os.path.exists(MOUNT_PATH):
  73             os.makedirs(MOUNT_PATH)
  74
  75     def remove_mount_dir(self):
  76         shutil.rmtree(MOUNT_PATH)
  77
  78     def can_find_api_response_in_logs(self, container_name):
  79         logs = self.get_container_logs(container_name)
  80         api_logs = re.findall(ERROR_API_REGEX, logs)
  81         if api_logs:
  82             return True
  83         else:
  84             return False
  85
  86     def get_api_response_from_logs(self, container_name):
  87         logs = self.get_container_logs(container_name)
  88         error_api_message = re.findall(ERROR_API_REGEX, logs)
  89         code = re.findall(RESPONSE_CODE_REGEX, error_api_message[0])
  90         return code[0]
  91
  92     def get_container_logs(self, container_name):
  93         client = docker.from_env()
  94         container = client.containers.get(container_name)
  95         logs = container.logs()
  96         return logs