Portieris app sanity tests

- portieris app functional test with clusterimagepolicy
- portieris app functional test with imagepolicy

Change-Id: Ida25ec148ff4f28a15a28818a15d811df84b7cb5
Signed-off-by: Thomas Sunil <sunil.thomas@windriver.com>
This commit is contained in:
Thomas Sunil
2025-09-23 13:47:48 -04:00
parent 7a3501b980
commit eb4d2d461f
13 changed files with 503 additions and 1 deletions

View File

@@ -6,5 +6,19 @@
"stepca_server_url": "external_acme_server_url",
// ACME server issuer
"stepca_server_issuer": "external_acme_server_issuer"
"stepca_server_issuer": "external_acme_server_issuer",
// Portieris configuration
"portieris": {
"registry_server_hostname": "registry_server_hostname",
"registry_server_port": "registry_server_port",
"trust_server": "https://registry_server_hostname:trust_port",
"signed_image_name": "registry_server_hostname:registry_server_port/signed_repo/image:tag",
"unsigned_image_name": "registry_server_hostname:registry_server_port/unsigned_repo/image:tag",
"registry_credentials": {
"username": "registry_username",
"password": "registry_password"
},
"registry_ca_cert": "registry_ca_cert"
}
}

View File

@@ -16,6 +16,16 @@ class SecurityConfig:
self.stepca_server_url = security_dict["stepca_server_url"]
self.stepca_server_issuer = security_dict["stepca_server_issuer"]
# Portieris configuration
portieris_config = security_dict.get("portieris", {})
self.portieris_registry_hostname = portieris_config.get("registry_server_hostname", "")
self.portieris_registry_port = portieris_config.get("registry_server_port", "")
self.portieris_trust_server = portieris_config.get("trust_server", "")
self.portieris_signed_image_name = portieris_config.get("signed_image_name", "")
self.portieris_unsigned_image_name = portieris_config.get("unsigned_image_name", "")
self.portieris_registry_credentials = portieris_config.get("registry_credentials", {})
self.portieris_registry_ca_cert = portieris_config.get("registry_ca_cert", "")
def get_domain_name(self) -> str:
"""Getter for the domain name.
@@ -39,3 +49,75 @@ class SecurityConfig:
str: StepCA server issuer.
"""
return self.stepca_server_issuer
def get_portieris_registry_hostname(self) -> str:
"""Getter for Portieris registry hostname.
Returns:
str: Registry hostname.
"""
return self.portieris_registry_hostname
def get_portieris_registry_port(self) -> str:
"""Getter for Portieris registry port.
Returns:
str: Registry port.
"""
return self.portieris_registry_port
def get_portieris_registry_server(self) -> str:
"""Getter for Portieris registry server (hostname:port).
Returns:
str: Registry server in hostname:port format.
"""
return f"{self.portieris_registry_hostname}:{self.portieris_registry_port}"
def get_portieris_trust_server(self) -> str:
"""Getter for Portieris trust server.
Returns:
str: Trust server URL.
"""
return self.portieris_trust_server
def get_portieris_signed_image_name(self) -> str:
"""Getter for Portieris signed image name.
Returns:
str: Signed image name.
"""
return self.portieris_signed_image_name
def get_portieris_registry_username(self) -> str:
"""Getter for Portieris registry username.
Returns:
str: Registry username.
"""
return self.portieris_registry_credentials.get("username", "registry_username")
def get_portieris_registry_password(self) -> str:
"""Getter for Portieris registry password.
Returns:
str: Registry password.
"""
return self.portieris_registry_credentials.get("password", "registry_password")
def get_portieris_unsigned_image_name(self) -> str:
"""Getter for Portieris unsigned image name.
Returns:
str: Unsigned image name.
"""
return self.portieris_unsigned_image_name
def get_portieris_registry_ca_cert(self) -> str:
"""Getter for Portieris registry CA certificate.
Returns:
str: Registry CA certificate content.
"""
return self.portieris_registry_ca_cert

View File

@@ -0,0 +1,29 @@
from framework.ssh.ssh_connection import SSHConnection
from keywords.base_keyword import BaseKeyword
class DockerTrustKeywords(BaseKeyword):
"""Keywords for Docker trust operations."""
def __init__(self, ssh_connection: SSHConnection):
"""Initialize Docker trust keywords.
Args:
ssh_connection (SSHConnection): SSH connection to the active controller.
"""
self.ssh_connection = ssh_connection
def inspect_docker_trust(self, image_name: str, trust_server: str) -> str:
"""Inspect Docker trust signatures for an image.
Args:
image_name (str): Name of the Docker image to inspect.
trust_server (str): Docker trust server URL.
Returns:
str: Docker trust inspection output.
"""
cmd = f"DOCKER_CONTENT_TRUST=1 DOCKER_CONTENT_TRUST_SERVER={trust_server} docker trust inspect {image_name}"
output = self.ssh_connection.send_as_sudo(cmd)
self.validate_success_return_code(self.ssh_connection)
return "\n".join(output) if isinstance(output, list) else str(output)

View File

@@ -26,3 +26,16 @@ class KubectlFileApplyKeywords(BaseKeyword):
"""
self.ssh_connection.send(export_k8s_config(f"kubectl apply -f {yaml_file}"))
self.validate_success_return_code(self.ssh_connection)
def kubectl_apply_with_error(self, yaml_file: str) -> str:
"""
Apply Kubernetes resource and return output message.
Args:
yaml_file (str): Path to the YAML file containing the resource definition.
Returns:
str: Output message from kubectl apply command.
"""
output = self.ssh_connection.send(export_k8s_config(f"kubectl apply -f {yaml_file}"))
return "\n".join(output) if isinstance(output, list) else str(output)

View File

@@ -0,0 +1,25 @@
from framework.ssh.ssh_connection import SSHConnection
from keywords.base_keyword import BaseKeyword
from keywords.k8s.k8s_command_wrapper import export_k8s_config
class KubectlDeleteImagePolicyKeywords(BaseKeyword):
"""Keywords for deleting Kubernetes image policies."""
def __init__(self, ssh_connection: SSHConnection):
"""Initialize kubectl delete image policy keywords.
Args:
ssh_connection (SSHConnection): SSH connection to the active controller.
"""
self.ssh_connection = ssh_connection
def delete_all_imagepolicies(self) -> None:
"""Delete all image policies in the cluster."""
self.ssh_connection.send(export_k8s_config("kubectl delete imagepolicy --all --ignore-not-found=true"))
self.validate_success_return_code(self.ssh_connection)
def delete_all_clusterimagepolicies(self) -> None:
"""Delete all cluster image policies in the cluster."""
self.ssh_connection.send(export_k8s_config("kubectl delete clusterimagepolicy --all --ignore-not-found=true"))
self.validate_success_return_code(self.ssh_connection)

View File

@@ -117,3 +117,15 @@ class KubectlGetPodsOutput:
if len(pods) == 0:
raise ValueError(f"No pods found starting with '{starts_with}'.")
return pods[0].get_name()
def get_pods_with_status(self, status: str) -> [KubectlPodObject]:
"""
Returns list of pods with the specified status.
Args:
status (str): The status to filter by (e.g., "Running", "Pending").
Returns:
[KubectlPodObject]: List of pods with the specified status.
"""
return [pod for pod in self.kubectl_pod if pod.get_status() == status]

View File

@@ -0,0 +1,30 @@
from framework.ssh.ssh_connection import SSHConnection
from keywords.base_keyword import BaseKeyword
class LsKeywords(BaseKeyword):
"""Keywords for ls command operations."""
def __init__(self, ssh_connection: SSHConnection):
"""Initialize ls keywords.
Args:
ssh_connection (SSHConnection): SSH connection to the active controller.
"""
self.ssh_connection = ssh_connection
def get_first_matching_file(self, pattern: str) -> str:
"""Get the first file matching the given pattern.
Args:
pattern (str): File pattern to match.
Returns:
str: First matching file path.
"""
output = self.ssh_connection.send(f"ls {pattern}")
self.validate_success_return_code(self.ssh_connection)
if isinstance(output, list):
return output[0].strip()
return output.strip()

View File

@@ -0,0 +1 @@
caCert: {{ registry_ca_cert }}

View File

@@ -0,0 +1,11 @@
apiVersion: portieris.cloud.ibm.com/v1
kind: ClusterImagePolicy
metadata:
name: clusterpolicy
spec:
repositories:
- name: "*"
policy:
trust:
enabled: true
trustServer: "{{ trust_server }}"

View File

@@ -0,0 +1,12 @@
apiVersion: portieris.cloud.ibm.com/v1
kind: ImagePolicy
metadata:
name: allow-custom
namespace: {{ namespace }}
spec:
repositories:
- name: "{{ registry_server }}:{{ registry_port }}/{{ signed_repo }}/*"
policy:
trust:
enabled: true
trustServer: "{{ trust_server }}"

View File

@@ -0,0 +1,15 @@
apiVersion: v1
kind: Pod
metadata:
name: {{ test_pod_name }}
namespace: {{ namespace }}
spec:
containers:
- command:
- sleep
- '3600'
image: {{ signed_image_name }}
imagePullPolicy: Always
name: {{ test_pod_name }}
imagePullSecrets:
- name: {{ pull_secret_name }}

View File

@@ -0,0 +1,15 @@
apiVersion: v1
kind: Pod
metadata:
name: {{ test_pod_name }}
namespace: {{ namespace }}
spec:
containers:
- command:
- sleep
- '3600'
image: {{ unsigned_image_name }}
imagePullPolicy: Always
name: {{ test_pod_name }}
imagePullSecrets:
- name: {{ pull_secret_name }}

View File

@@ -0,0 +1,243 @@
from pytest import mark
from config.configuration_manager import ConfigurationManager
from config.docker.objects.registry import Registry
from config.security.objects.security_config import SecurityConfig
from framework.logging.automation_logger import get_logger
from framework.resources.resource_finder import get_stx_resource_path
from framework.ssh.ssh_connection import SSHConnection
from framework.validation.validation import validate_equals, validate_not_equals, validate_str_contains
from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from keywords.cloud_platform.system.application.system_application_apply_keywords import SystemApplicationApplyKeywords
from keywords.cloud_platform.system.application.system_application_delete_keywords import SystemApplicationDeleteInput, SystemApplicationDeleteKeywords
from keywords.cloud_platform.system.application.system_application_list_keywords import SystemApplicationListKeywords
from keywords.cloud_platform.system.application.system_application_remove_keywords import SystemApplicationRemoveInput, SystemApplicationRemoveKeywords
from keywords.cloud_platform.system.application.system_application_upload_keywords import SystemApplicationUploadInput, SystemApplicationUploadKeywords
from keywords.cloud_platform.system.helm.system_helm_keywords import SystemHelmKeywords
from keywords.docker.trust.docker_trust_keywords import DockerTrustKeywords
from keywords.files.yaml_keywords import YamlKeywords
from keywords.k8s.files.kubectl_file_apply_keywords import KubectlFileApplyKeywords
from keywords.k8s.imagepolicy.kubectl_delete_imagepolicy_keywords import KubectlDeleteImagePolicyKeywords
from keywords.k8s.namespace.kubectl_create_namespace_keywords import KubectlCreateNamespacesKeywords
from keywords.k8s.namespace.kubectl_delete_namespace_keywords import KubectlDeleteNamespaceKeywords
from keywords.k8s.pods.kubectl_get_pods_keywords import KubectlGetPodsKeywords
from keywords.k8s.secret.kubectl_create_secret_keywords import KubectlCreateSecretsKeywords
from keywords.linux.ls.ls_keywords import LsKeywords
APP_NAME = "portieris"
CHART_PATH = "/usr/local/share/applications/helm/portieris-[0-9]*"
NAMESPACE = "pvtest"
def setup_portieris_environment(ssh_connection: SSHConnection, security_config: SecurityConfig) -> None:
"""Setup Portieris application and test environment.
Args:
ssh_connection (SSHConnection): SSH connection to active controller.
security_config (SecurityConfig): Security configuration object.
"""
# Setup Portieris if not present
system_app_list = SystemApplicationListKeywords(ssh_connection)
if not system_app_list.is_app_present(APP_NAME):
get_logger().log_info(f"Uploading {APP_NAME} application")
ls_keywords = LsKeywords(ssh_connection)
actual_chart = ls_keywords.get_first_matching_file(CHART_PATH)
upload_input = SystemApplicationUploadInput()
upload_input.set_tar_file_path(actual_chart)
upload_input.set_app_name(APP_NAME)
system_app_upload = SystemApplicationUploadKeywords(ssh_connection)
system_app_upload.system_application_upload(upload_input)
# Setup helm overrides and apply application
system_app_apply = SystemApplicationApplyKeywords(ssh_connection)
if not system_app_apply.is_already_applied(APP_NAME):
get_logger().log_info(f"Setting up {APP_NAME} helm overrides for caCert")
helm_keywords = SystemHelmKeywords(ssh_connection)
yaml_keywords = YamlKeywords(ssh_connection)
template_file = get_stx_resource_path("resources/cloud_platform/security/portieris/caCert.yaml")
replacement_dict = {"registry_ca_cert": security_config.get_portieris_registry_ca_cert()}
portieris_overrides = yaml_keywords.generate_yaml_file_from_template(template_file, replacement_dict, "caCert.yaml", "/tmp")
helm_keywords.helm_override_update(APP_NAME, "portieris", "portieris", portieris_overrides)
get_logger().log_info(f"Applying {APP_NAME} application")
system_app_apply.system_application_apply(APP_NAME)
# Wait for Portieris pods
kubectl_pods = KubectlGetPodsKeywords(ssh_connection)
pods_output = kubectl_pods.get_pods("portieris")
running_pods = pods_output.get_pods_with_status("Running")
validate_equals(len(running_pods) > 0, True, "At least one Portieris pod should be running")
# Create namespace and registry secret
kubectl_create_ns = KubectlCreateNamespacesKeywords(ssh_connection)
kubectl_create_ns.create_namespaces(NAMESPACE)
registry_hostname = security_config.get_portieris_registry_hostname()
username = security_config.get_portieris_registry_username()
password = security_config.get_portieris_registry_password()
registry = Registry("registry", registry_hostname, username, password)
kubectl_create_secret = KubectlCreateSecretsKeywords(ssh_connection)
kubectl_create_secret.create_secret_for_registry(registry, "registry-secret", NAMESPACE)
def cleanup_portieris_environment(ssh_connection: SSHConnection) -> None:
"""Clean up Portieris test resources.
Args:
ssh_connection (SSHConnection): SSH connection to active controller.
"""
get_logger().log_info("Cleaning up Portieris test resources")
kubectl_delete_ns = KubectlDeleteNamespaceKeywords(ssh_connection)
kubectl_delete_ns.cleanup_namespace(NAMESPACE)
kubectl_delete_policies = KubectlDeleteImagePolicyKeywords(ssh_connection)
kubectl_delete_policies.delete_all_clusterimagepolicies()
kubectl_delete_policies.delete_all_imagepolicies()
system_app_list = SystemApplicationListKeywords(ssh_connection)
if system_app_list.is_app_present(APP_NAME):
get_logger().log_info(f"Removing {APP_NAME} application")
system_app_apply = SystemApplicationApplyKeywords(ssh_connection)
if system_app_apply.is_already_applied(APP_NAME):
remove_input = SystemApplicationRemoveInput()
remove_input.set_app_name(APP_NAME)
system_app_remove = SystemApplicationRemoveKeywords(ssh_connection)
system_app_remove.system_application_remove(remove_input)
delete_input = SystemApplicationDeleteInput()
delete_input.set_app_name(APP_NAME)
delete_input.set_force_deletion(True)
system_app_delete = SystemApplicationDeleteKeywords(ssh_connection)
system_app_delete.get_system_application_delete(delete_input)
@mark.p1
def test_portieris_image_security_policy(request):
"""Test Portieris image security with image policy.
Steps:
- Setup Portieris application and test environment
- Apply image policy configuration
- Test unsigned image deployment (should be rejected)
- Validate signed image signatures
- Test signed image deployment (should be accepted)
"""
get_logger().log_info("Starting Portieris image security test")
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
security_config = ConfigurationManager.get_security_config()
def cleanup():
cleanup_portieris_environment(ssh_connection)
request.addfinalizer(cleanup)
setup_portieris_environment(ssh_connection, security_config)
# Initialize keyword classes directly in test
yaml_keywords = YamlKeywords(ssh_connection)
kubectl_file_apply = KubectlFileApplyKeywords(ssh_connection)
kubectl_pods = KubectlGetPodsKeywords(ssh_connection)
docker_trust = DockerTrustKeywords(ssh_connection)
# Apply image policy
policy_template = get_stx_resource_path("resources/cloud_platform/security/portieris/image-policy.yaml")
registry_hostname = security_config.get_portieris_registry_hostname()
registry_port = security_config.get_portieris_registry_port()
replacement_dict = {"registry_server": registry_hostname, "registry_port": registry_port, "signed_repo": "wrcp-test-signed", "trust_server": security_config.get_portieris_trust_server(), "namespace": NAMESPACE}
policy_file = yaml_keywords.generate_yaml_file_from_template(policy_template, replacement_dict, "image-policy.yaml", "/tmp")
kubectl_file_apply.apply_resource_from_yaml(policy_file)
# Test unsigned image (should be rejected)
get_logger().log_info("Testing unsigned image deployment (should be rejected)")
unsigned_template = get_stx_resource_path("resources/cloud_platform/security/portieris/unsigned-image.yaml")
replacement_dict = {"namespace": NAMESPACE, "test_pod_name": "test-pod", "unsigned_image_name": security_config.get_portieris_unsigned_image_name()}
pod_file = yaml_keywords.generate_yaml_file_from_template(unsigned_template, replacement_dict, "unsigned-image-policy.yaml", "/tmp")
# Use kubectl_apply_with_error and validate in test case
error_output = kubectl_file_apply.kubectl_apply_with_error(pod_file)
return_code = ssh_connection.get_return_code()
validate_not_equals(return_code, 0, "kubectl apply should fail for policy rejection")
validate_str_contains(error_output, "trust.hooks.securityenforcement.admission.cloud.ibm.com", "Output should contain Portieris admission webhook")
# Validate signatures using docker trust keywords
get_logger().log_info("Verifying signed image has valid signatures")
signed_image = security_config.get_portieris_signed_image_name()
trust_server = security_config.get_portieris_trust_server()
trust_output = docker_trust.inspect_docker_trust(signed_image, trust_server)
validate_str_contains(trust_output, "Signers", "Signed image should have valid signatures")
# Test signed image (should be accepted)
get_logger().log_info("Testing signed image deployment (should be accepted)")
signed_template = get_stx_resource_path("resources/cloud_platform/security/portieris/signed-image.yaml")
replacement_dict = {"namespace": NAMESPACE, "test_pod_name": "test-pod", "signed_image_name": security_config.get_portieris_signed_image_name(), "pull_secret_name": "registry-secret"}
pod_file = yaml_keywords.generate_yaml_file_from_template(signed_template, replacement_dict, "signed-image-policy.yaml", "/tmp")
kubectl_file_apply.apply_resource_from_yaml(pod_file)
pod_running = kubectl_pods.wait_for_pod_status("test-pod", "Running", NAMESPACE, 300)
validate_equals(pod_running, True, "Signed image pod should be running")
@mark.p1
def test_portieris_cluster_image_policy(request):
"""Test Portieris image security with cluster image policy.
Steps:
- Setup Portieris application and test environment
- Apply cluster image policy configuration
- Test unsigned image deployment (should be rejected)
- Validate signed image signatures
- Test signed image deployment (should be accepted)
"""
get_logger().log_info("Starting Portieris cluster image policy test")
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
security_config = ConfigurationManager.get_security_config()
def cleanup():
cleanup_portieris_environment(ssh_connection)
request.addfinalizer(cleanup)
setup_portieris_environment(ssh_connection, security_config)
# Initialize keyword classes directly in test
yaml_keywords = YamlKeywords(ssh_connection)
kubectl_file_apply = KubectlFileApplyKeywords(ssh_connection)
kubectl_pods = KubectlGetPodsKeywords(ssh_connection)
docker_trust = DockerTrustKeywords(ssh_connection)
# Apply cluster image policy
policy_template = get_stx_resource_path("resources/cloud_platform/security/portieris/cluster-image-policy.yaml")
registry_hostname = security_config.get_portieris_registry_hostname()
registry_port = security_config.get_portieris_registry_port()
replacement_dict = {"registry_server": registry_hostname, "registry_port": registry_port, "signed_repo": "wrcp-test-signed", "trust_server": security_config.get_portieris_trust_server(), "namespace": NAMESPACE}
policy_file = yaml_keywords.generate_yaml_file_from_template(policy_template, replacement_dict, "cluster-image-policy.yaml", "/tmp")
kubectl_file_apply.apply_resource_from_yaml(policy_file)
# Test unsigned image (should be rejected)
get_logger().log_info("Testing unsigned image deployment (should be rejected)")
unsigned_template = get_stx_resource_path("resources/cloud_platform/security/portieris/unsigned-image.yaml")
replacement_dict = {"namespace": NAMESPACE, "test_pod_name": "test-pod", "unsigned_image_name": security_config.get_portieris_unsigned_image_name()}
pod_file = yaml_keywords.generate_yaml_file_from_template(unsigned_template, replacement_dict, "unsigned-cluster-image-policy.yaml", "/tmp")
# Use kubectl_apply_with_error and validate in test case
error_output = kubectl_file_apply.kubectl_apply_with_error(pod_file)
return_code = ssh_connection.get_return_code()
validate_not_equals(return_code, 0, "kubectl apply should fail for policy rejection")
validate_str_contains(error_output, "trust.hooks.securityenforcement.admission.cloud.ibm.com", "Output should contain Portieris admission webhook")
# Validate signatures using docker trust keywords
get_logger().log_info("Verifying signed image has valid signatures")
signed_image = security_config.get_portieris_signed_image_name()
trust_server = security_config.get_portieris_trust_server()
trust_output = docker_trust.inspect_docker_trust(signed_image, trust_server)
validate_str_contains(trust_output, "Signers", "Signed image should have valid signatures")
# Test signed image (should be accepted)
get_logger().log_info("Testing signed image deployment (should be accepted)")
signed_template = get_stx_resource_path("resources/cloud_platform/security/portieris/signed-image.yaml")
replacement_dict = {"namespace": NAMESPACE, "test_pod_name": "test-pod", "signed_image_name": security_config.get_portieris_signed_image_name(), "pull_secret_name": "registry-secret"}
pod_file = yaml_keywords.generate_yaml_file_from_template(signed_template, replacement_dict, "signed-cluster-image-policy.yaml", "/tmp")
kubectl_file_apply.apply_resource_from_yaml(pod_file)
pod_running = kubectl_pods.wait_for_pod_status("test-pod", "Running", NAMESPACE, 300)
validate_equals(pod_running, True, "Signed image pod should be running")