pip uninstall -y docker
pip install -U docker==2.7.0
+#reinstall pyopenssl library
+echo "Reinstall pyopenssl library."
+pip uninstall pyopenssl -y
+pip install pyopenssl==17.5.0
+
#Disable proxy - for local run
unset http_proxy https_proxy
[Documentation] Run with correct env and expected exit code 0
Run Cert Service Client And Validate JKS File Creation And Client Exit Code ${VALID_ENV_FILE} 0
+Cert Service Client successfully creates keystore and truststore with expected data
+ [Tags] AAF-CERT-SERVICE
+ [Documentation] Run with correct env and JKS files created with correct data
+ Run Cert Service Client And Validate JKS Files Contain Expected Data ${VALID_ENV_FILE} 0
+
Run Cert Service Client Container And Validate Exit Code And API Response
[Tags] AAF-CERT-SERVICE
[Documentation] Run with invalid CaName env and expected exit code 5
import os
import shutil
import re
-from OpenSSL import crypto
+from EnvsReader import EnvsReader
from docker.types import Mount
ARCHIVES_PATH = os.getenv("WORKSPACE") + "/archives/"
-MOUNT_PATH = os.getenv("WORKSPACE") + "/tests/aaf/certservice/tmp"
ERROR_API_REGEX = 'Error on API response.*[0-9]{3}'
RESPONSE_CODE_REGEX = '[0-9]{3}'
-
class CertClientManager:
+ def __init__(self, mount_path):
+ self.mount_path = mount_path
+
def run_client_container(self, client_image, container_name, path_to_env, request_url, network):
self.create_mount_dir()
client = docker.from_env()
- environment = self.read_list_env_from_file(path_to_env)
+ environment = EnvsReader().read_env_list_from_file(path_to_env)
environment.append("REQUEST_URL=" + request_url)
container = client.containers.run(
image=client_image,
environment=environment,
network=network,
user='root', #Run container as root to avoid permission issues with volume mount access
- mounts=[Mount(target='/var/certs', source=MOUNT_PATH, type='bind')],
+ mounts=[Mount(target='/var/certs', source=self.mount_path, type='bind')],
detach=True
)
exitcode = container.wait()
return exitcode
- def read_list_env_from_file(self, path):
- f = open(path, "r")
- r_list = []
- for line in f:
- line = line.strip()
- if line[0] != "#":
- r_list.append(line)
- return r_list
-
def remove_client_container_and_save_logs(self, container_name, log_file_name):
client = docker.from_env()
container = client.containers.get(container_name)
- text_file = open(ARCHIVES_PATH + "container_" + log_file_name + ".log", "w")
+ text_file = open(ARCHIVES_PATH + "client_container_" + log_file_name + ".log", "w")
text_file.write(container.logs())
text_file.close()
container.remove()
self.remove_mount_dir()
- def can_open_keystore_and_truststore_with_pass(self):
- keystore_pass_path = MOUNT_PATH + '/keystore.pass'
- keystore_jks_path = MOUNT_PATH + '/keystore.jks'
- can_open_keystore = self.can_open_jks_file_by_pass_file(keystore_pass_path, keystore_jks_path)
-
- truststore_pass_path = MOUNT_PATH + '/truststore.pass'
- truststore_jks_path = MOUNT_PATH + '/truststore.jks'
- can_open_truststore = self.can_open_jks_file_by_pass_file(truststore_pass_path, truststore_jks_path)
-
- return can_open_keystore & can_open_truststore
-
- def can_open_jks_file_by_pass_file(self, pass_file_path, jks_file_path):
- try:
- password = open(pass_file_path, 'rb').read()
- crypto.load_pkcs12(open(jks_file_path, 'rb').read(), password)
- return True
- except:
- return False
-
def create_mount_dir(self):
- if not os.path.exists(MOUNT_PATH):
- os.makedirs(MOUNT_PATH)
+ if not os.path.exists(self.mount_path):
+ os.makedirs(self.mount_path)
def remove_mount_dir(self):
- shutil.rmtree(MOUNT_PATH)
+ shutil.rmtree(self.mount_path)
def can_find_api_response_in_logs(self, container_name):
logs = self.get_container_logs(container_name)
--- /dev/null
+
+class EnvsReader:
+
+ def read_env_list_from_file(self, path):
+ f = open(path, "r")
+ r_list = []
+ for line in f:
+ line = line.strip()
+ if line[0] != "#":
+ r_list.append(line)
+ return r_list
--- /dev/null
+from OpenSSL import crypto
+from cryptography.x509.oid import ExtensionOID
+from cryptography import x509
+from EnvsReader import EnvsReader
+
+class JksFilesValidator:
+
+ def __init__(self, mount_path):
+ self.keystorePassPath = mount_path + '/keystore.pass'
+ self.keystoreJksPath = mount_path + '/keystore.jks'
+ self.truststorePassPath = mount_path + '/truststore.pass'
+ self.truststoreJksPath = mount_path + '/truststore.jks'
+
+ def get_and_compare_data(self, path_to_env):
+ data = self.get_data(path_to_env)
+ return data, self.contains_expected_data(data)
+
+ def can_open_keystore_and_truststore_with_pass(self):
+ can_open_keystore = self.can_open_jks_file_with_pass_file(self.keystorePassPath, self.keystoreJksPath)
+ can_open_truststore = self.can_open_jks_file_with_pass_file(self.truststorePassPath, self.truststoreJksPath)
+
+ return can_open_keystore & can_open_truststore
+
+ def can_open_jks_file_with_pass_file(self, pass_file_path, jks_file_path):
+ try:
+ self.get_certificate(pass_file_path, jks_file_path)
+ return True
+ except:
+ return False
+
+ def get_data(self, path_to_env):
+ envs = self.get_envs_as_dict(EnvsReader().read_env_list_from_file(path_to_env))
+ certificate = self.get_certificate(self.keystorePassPath, self.keystoreJksPath)
+ data = self.get_owner_data_from_certificate(certificate)
+ data['SANS'] = self.get_sans(certificate)
+ return type('', (object,), {"expectedData": envs, "actualData": data})
+
+ def contains_expected_data(self, data):
+ expectedData = data.expectedData
+ actualData = data.actualData
+ return cmp(expectedData, actualData) == 0
+
+ def get_owner_data_from_certificate(self, certificate):
+ list = certificate.get_subject().get_components()
+ return dict((k, v) for k, v in list)
+
+ def get_certificate(self, pass_file_path, jks_file_path):
+ password = open(pass_file_path, 'rb').read()
+ crypto.load_pkcs12(open(jks_file_path, 'rb').read(), password)
+ return crypto.load_pkcs12(open(jks_file_path, 'rb').read(), password).get_certificate()
+
+ def get_sans(self, cert):
+ extension = cert.to_cryptography().extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME)
+ dnsList = extension.value.get_values_for_type(x509.DNSName)
+ return ':'.join(map(lambda dns: dns.encode('ascii','ignore'), dnsList))
+
+ def get_envs_as_dict(self, list):
+ envs = self.get_list_of_pairs_by_mappings(list)
+ return self.remove_nones_from_dict(envs)
+
+ def remove_nones_from_dict(self, dictionary):
+ return dict((k, v) for k, v in dictionary.iteritems() if k is not None)
+
+ def get_list_of_pairs_by_mappings(self, list):
+ mappings = self.get_mappings()
+ listOfEnvs = map(lambda k: k.split('='), list)
+ return dict((mappings.get(a[0]), a[1]) for a in listOfEnvs)
+
+ def get_mappings(self):
+ return {'COMMON_NAME':'CN', 'ORGANIZATION':'O', 'ORGANIZATION_UNIT':'OU', 'LOCATION':'L', 'STATE':'ST', 'COUNTRY':'C', 'SANS':'SANS'}
*** Settings ***
+Resource ../../../common.robot
+Resource ./cert-service-properties.robot
Library RequestsLibrary
Library HttpLibrary.HTTP
Library Collections
-Library ../libraries/CertClientManager.py
-Resource ../../../common.robot
-Resource ./cert-service-properties.robot
+Library ../libraries/CertClientManager.py ${MOUNT_PATH}
+Library ../libraries/JksFilesValidator.py ${MOUNT_PATH}
*** Keywords ***
Should Be Equal As Strings ${exit_code} ${expected_exit_code} Client return: ${exitcode} exit code, but expected: ${expected_exit_code}
Should Be True ${can_open} Cannot Open Keystore/TrustStore by passpshase
+Run Cert Service Client And Validate JKS Files Contain Expected Data
+ [Documentation] Run Cert Service Client Container And Validate JKS Files Contain Expected Data
+ [Arguments] ${env_file} ${expected_exit_code}
+ ${exit_code}= Run Client Container ${DOCKER_CLIENT_IMAGE} ${CLIENT_CONTAINER_NAME} ${env_file} ${CERT_SERVICE_ADDRESS}${CERT_SERVICE_ENDPOINT} ${CERT_SERVICE_NETWORK}
+ ${data} ${isEqual}= Get And Compare Data ${env_file}
+ Remove Client Container And Save Logs ${CLIENT_CONTAINER_NAME} positive_path_with_data
+ Should Be Equal As Strings ${exit_code} ${expected_exit_code} Client return: ${exitcode} exit code, but expected: ${expected_exit_code}
+ Should Be True ${isEqual} Keystore doesn't contain ${data.expectedData}. Actual data is: ${data.actualData}
+
Run Cert Service Client And Validate Http Response Code And Client Exit Code
[Documentation] Run Cert Service Client Container And Validate Exit Code
[Arguments] ${env_file} ${expected_api_response_code} ${expected_exit_code}
${DOCKER_CLIENT_IMAGE} nexus3.onap.org:10001/onap/org.onap.aaf.certservice.aaf-certservice-client:latest
${CLIENT_CONTAINER_NAME} %{ClientContainerName}
${CERT_SERVICE_NETWORK} certservice_certservice
+${MOUNT_PATH} %{WORKSPACE}/tests/aaf/certservice/tmp
Code Review / integration / csit.git / commitdiff