Merge branch 'main' into small-reame-fix

JUSTFILE

@@ -64,7 +64,10 @@ load-pytest-kind:
   kind load docker-image cloudbeat-test:latest --name kind-mono
 
 deploy-tests-helm:
-  helm upgrade --wait --timeout={{TIMEOUT}} --install --values tests/deploy/values/ci.yml --namespace kube-system {{TESTS_RELEASE}}  tests/deploy/k8s-cloudbeat-tests/ 
+  helm upgrade --wait --timeout={{TIMEOUT}} --install --values tests/deploy/values/ci.yml --namespace kube-system {{TESTS_RELEASE}}  tests/deploy/k8s-cloudbeat-tests/
+
+deploy-local-tests-helm:
+  helm upgrade --wait --timeout={{TIMEOUT}} --install --values tests/deploy/values/local-host.yml --namespace kube-system {{TESTS_RELEASE}}  tests/deploy/k8s-cloudbeat-tests/
 
 purge-tests:
 	helm del {{TESTS_RELEASE}} -n kube-system

tests/commonlib/docker_wrapper.py

@@ -0,0 +1,30 @@
+"""
+This module provides kubernetes functionality based on original docker SDK python library.
+"""
+
+import docker
+
+
+class DockerWrapper:
+
+    def __init__(self, config=None):
+        if config.base_url != "":
+            self.client = docker.DockerClient(base_url=config.base_url)
+        else:
+            self.client = docker.from_env()
+
+    def exec_command(self, container_name: str, command: str, param_value: str, resource: str):
+        """
+        This function retrieves container by name / id and executes (docker exec) command to container
+        @param container_name: Container id or name
+        @param command: String command to be executed (for docker exec)
+        @param param_value: Command function parameter value to be updated
+        @param resource: Path to resource file
+        @return: Command output, if exists
+        """
+        container = self.client.containers.get(container_id=container_name)
+        command_f = f"{command} {param_value} {resource}"
+        exit_code, output = container.exec_run(cmd=command_f)
+        if exit_code > 0:
+            return ''
+        return output.decode().strip()

tests/commonlib/io_utils.py

@@ -0,0 +1,82 @@
+"""
+This module provides input / output manipulations on streams / files
+"""
+
+import os
+import io
+import json
+import yaml
+import shutil
+from pathlib import Path
+from munch import Munch
+
+
+def get_logs_from_stream(stream: str) -> list[Munch]:
+    """
+    This function converts logs stream to list of Munch objects (dictionaries)
+    @param stream: StringIO stream
+    @return: List of Munch objects
+    """
+
+    logs = io.StringIO(stream)
+    result = []
+    # with open("pod_logs.log", 'a') as pod_log_file:
+    #     pod_log_file.writelines(logs)
+    for log in logs:
+        # current_log = log.split(sep="Z ")[1]
+        if log and "bundles" in log:
+            result.append(Munch(json.loads(log)))
+    return result
+
+
+def get_k8s_yaml_objects(file_path: Path) -> list[str: dict]:
+    """
+    This function loads yaml file, and returns the following list:
+    [ {<k8s_kind> : {<k8s_metadata}}]
+    :param file_path: YAML path
+    :return: [ {<k8s_kind> : {<k8s_metadata}}]
+    """
+    if not file_path:
+        raise Exception(f'{file_path} is required')
+    result_list = []
+    with file_path.open() as yaml_file:
+        yaml_objects = yaml.safe_load_all(yaml_file)
+        for yml_doc in yaml_objects:
+            if yml_doc:
+                doc = Munch(yml_doc)
+                result_list.append({
+                    doc.get('kind'): {key: value for key, value in doc.get('metadata').items()
+                                      if key in ['name', 'namespace']}
+                })
+    return result_list
+
+
+class FsClient:
+
+    @staticmethod
+    def exec_command(container_name: str, command: str, param_value: str, resource: str):
+        """
+        This function executes os command
+        @param container_name: Container node
+        @param command: Linux command to be executed
+        @param param_value: Value to be used in exec command
+        @param resource: File / Resource path
+        @return: None
+        """
+
+        if container_name == '':
+            raise Exception(f"Unknown {container_name} is sent")
+
+        current_resource = Path(resource)
+        if not current_resource.is_file():
+            raise Exception(f"File {resource} does not exist or mount missing.")
+
+        if command == 'chmod':
+            os.chmod(path=resource, mode=int(param_value))
+        elif command == 'chown':
+            uid_gid = param_value.split(':')
+            if len(uid_gid) != 2:
+                raise Exception("User and group parameter shall be separated by ':' ")
+            shutil.chown(path=resource, user=uid_gid[0], group=uid_gid[1])
+        else:
+            raise Exception(f"Command '{command}' still not implemented in test framework")

tests/commonlib/kubernetes.py

@@ -1,20 +1,32 @@
 """
-This module provides kubernetes functionality based on original kuberentes python library.
+This module provides kubernetes functionality based on original kubernetes python library.
 """
 
-from kubernetes import client, config
+from kubernetes import client, config, utils
 
 
 class KubernetesHelper:
 
     def __init__(self, is_in_cluster_config: bool = False):
 
         if is_in_cluster_config:
-            config.load_incluster_config()
+            self.config = config.load_incluster_config()
         else:
-            config.load_kube_config()
+            self.config = config.load_kube_config()
 
         self.core_v1_client = client.CoreV1Api()
+        self.app_api = client.AppsV1Api()
+        self.rbac_api = client.RbacAuthorizationV1Api()
+        self.api_client = client.api_client.ApiClient(configuration=self.config)
+        self.dispatch_delete = {
+            'ConfigMap': self.core_v1_client.delete_namespaced_config_map,
+            'ServiceAccount': self.core_v1_client.delete_namespaced_service_account,
+            'DaemonSet': self.app_api.delete_namespaced_daemon_set,
+            'Role': self.rbac_api.delete_namespaced_role,
+            'RoleBinding': self.rbac_api.delete_namespaced_role_binding,
+            'ClusterRoleBinding': self.rbac_api.delete_cluster_role_binding,
+            'ClusterRole': self.rbac_api.delete_cluster_role
+        }
 
     def get_agent_pod_instances(self, agent_name: str, namespace: str):
         """
@@ -36,3 +48,48 @@ def get_cluster_nodes(self):
         nodes = self.core_v1_client.list_node()
         return nodes.items
 
+    def get_pod_logs(self, pod_name: str, namespace: str, **kwargs):
+        """
+        This function returns pod logs
+        @param pod_name: Name of pod
+        @param namespace: Pod namespace
+        @param kwargs:
+        @return: Pod logs stream
+        """
+
+        return self.core_v1_client.read_namespaced_pod_log(name=pod_name, namespace=namespace, **kwargs)
+
+    def start_agent(self, yaml_file: str, namespace: str):
+        """
+        This function deploys cloudbeat agent from yaml file
+        :return:
+        """
+
+        return utils.create_from_yaml(k8s_client=self.api_client,
+                                      yaml_file=yaml_file,
+                                      namespace=namespace,
+                                      verbose=True)
+
+    def stop_agent(self, yaml_objects_list: list):
+        """
+        This function will delete all cloudbeat kubernetes resources.
+        Currently, there is no ability to remove through utils due to the following:
+        https://github.com/kubernetes-client/python/pull/1392
+        So below is cloud-security-posture own implementation.
+        :return: V1Object - result
+        """
+        result_list = []
+        for yaml_object in yaml_objects_list:
+            for dict_key in yaml_object:
+                result_list.append(self._delete_resources(resource_type=dict_key, **yaml_object[dict_key]))
+        return result_list
+
+    def _delete_resources(self, resource_type: str, **kwargs):
+        """
+        This is internal method for executing delete method depends on resource type.
+        Binding is done using dispatch_delete dictionary.
+        @param resource_type: Kubernetes resource to be deleted.
+        @param kwargs: Depends on resource type, it may be a name / name and namespace.
+        @return:
+        """
+        return self.dispatch_delete[resource_type](**kwargs)

tests/configuration.py

@@ -3,17 +3,18 @@
 
 """
 import os
-
+from distutils.util import strtobool
 from munch import Munch
 
 # --- Cloudbeat agent environment definition ----------------
 agent = Munch()
 agent.name = os.getenv('AGENT_NAME', 'cloudbeat')
 agent.namespace = os.getenv('AGENT_NAMESPACE', 'kube-system')
+agent.findings_timeout = 90
 
 # --- Kubernetes environment definition --------------------
 kubernetes = Munch()
-kubernetes.is_in_cluster_config = os.getenv('KUBERNETES_IN_CLUSTER', False)
+kubernetes.is_in_cluster_config = bool(strtobool(os.getenv('KUBERNETES_IN_CLUSTER', 'False')))
 
 # --- Elasticsearch environment definition --------------------------------
 elasticsearch = Munch()
@@ -25,3 +26,12 @@
 elasticsearch.protocol = os.getenv('ES_PROTOCOL', 'http')
 elasticsearch.url = f"{elasticsearch.protocol}://{elasticsearch.hosts}:{elasticsearch.port}"
 elasticsearch.cis_index = os.getenv('CIS_INDEX', "*cis_kubernetes_benchmark.findings*")
+
+# --- Docker environment definition
+docker = Munch()
+docker.base_url = os.getenv('DOCKER_URL', "")
+docker.use_docker = bool(strtobool(os.getenv('USE_DOCKER', 'True')))
+
+# Printing all environment keys
+for key, value in sorted(os.environ.items()):
+    print('{}: {}'.format(key, value))

tests/conftest.py

@@ -2,20 +2,53 @@
 import configuration
 from commonlib.kubernetes import KubernetesHelper
 from commonlib.elastic_wrapper import ElasticWrapper
+from commonlib.docker_wrapper import DockerWrapper
+from commonlib.io_utils import FsClient
 
 
 @pytest.fixture(scope="session", autouse=True)
 def k8s():
+    """
+    This function (fixture) instantiates KubernetesHelper depends on configuration.
+    When executing tests code local, kubeconfig file is used for connecting to K8s cluster.
+    When code executed as container (pod / job) in K8s cluster in cluster configuration is used.
+    @return: Kubernetes Helper instance.
+    """
     return KubernetesHelper(is_in_cluster_config=configuration.kubernetes.is_in_cluster_config)
 
 
 @pytest.fixture(scope="session", autouse=True)
 def cloudbeat_agent():
+    """
+    This function (fixture) retrieves agent configuration, defined in configuration.py file.
+    @return: Agent config
+    """
     return configuration.agent
 
 
 @pytest.fixture(scope="session", autouse=True)
 def elastic_client():
+    """
+    This function (fixture) instantiate ElasticWrapper.
+    @return: ElasticWrapper client
+    """
     elastic_config = configuration.elasticsearch
     es_client = ElasticWrapper(elastic_params=elastic_config)
     return es_client
+
+
+@pytest.fixture(scope="session", autouse=True)
+def api_client():
+    """
+    This function (fixture) instantiates client depends on configuration.
+    For local development mode, the docker api may be used.
+    For production mode (deployment to k8s cluster), FsClient shall be used.
+    @return: Client (docker / FsClient).
+    """
+    docker_config = configuration.docker
+    print(f"Config use_docker value: {docker_config.use_docker}")
+    if docker_config.use_docker:
+        client = DockerWrapper(config=docker_config)
+    else:
+        client = FsClient
+    return client

tests/deploy/k8s-cloudbeat-tests/templates/cloudbeat-test-configmap.yaml

@@ -8,4 +8,6 @@ data:
   KUBERNETES_IN_CLUSTER: "true"
   ES_USER: "elastic"
   ES_PASSWORD: "changeme"
-  ES_HOST: "elasticsearch-master.kube-system"
+  ES_HOST: "elasticsearch-master.kube-system"
+  USE_DOCKER: "false"
+  TEST_MARKER: {{ .Values.testData.marker }}

tests/deploy/k8s-cloudbeat-tests/templates/cloudbeat-test-cr.yaml

@@ -12,7 +12,7 @@ rules:
       - events
       - pods
       - services
-    verbs: ["get", "list", "watch"]
+    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
   # Enable this rule only if planing to use Kubernetes keystore
   #- apiGroups: [""]
   #  resources:

tests/deploy/k8s-cloudbeat-tests/templates/tests/cloudbeat-test-pod.yaml

@@ -16,15 +16,21 @@ spec:
     - /bin/sh
     - -c
     - |
-      pytest -rA --disable-warnings --alluredir=/usr/src/app/tests/report
+      pytest -rA --disable-warnings -m ${TEST_MARKER} --alluredir=/usr/src/app/tests/report
     envFrom:
       - configMapRef:
           name: {{ .Values.serviceAccount.name }}-configmap
     volumeMounts:
-      - mountPath: /usr/src/app/tests/report
-        name: cloudbeat-test-pv-storage
+      - name: cloudbeat-test-pv-storage
+        mountPath: /usr/src/app/tests/report
+      - name: etc-kubernetes
+        mountPath: /etc/kubernetes/manifests
+        readOnly: false
   restartPolicy: Never
   volumes:
     - name: cloudbeat-test-pv-storage
       hostPath:
-        path: /tmp/data
+        path: /tmp/data
+    - name: etc-kubernetes
+      hostPath:
+        path: /etc/kubernetes/manifests

tests/deploy/k8s-cloudbeat-tests/values.yaml

@@ -11,6 +11,9 @@ image:
   # Overrides the image tag whose default is the chart appVersion.
   tag: "latest"
 
+testData:
+  marker: rules
+
 imagePullSecrets: []
 nameOverride: ""
 fullnameOverride: ""
@@ -116,7 +119,7 @@ elasticsearch:
 
   # Request smaller persistent volumes.
   volumeClaimTemplate:
-    accessModes: [ "ReadWriteOnce" ]
+    accessModes: ["ReadWriteOnce"]
     storageClassName: "standard"
     resources:
       requests:
@@ -125,7 +128,7 @@ elasticsearch:
   # Storage settings
   persistence:
     enabled: false
-  
+
   #disable ES tests
   tests:
-    enabled: false
+    enabled: false

tests/deploy/values/ci.yml

@@ -1,2 +1,5 @@
 cloudbeat:
-  deploy: true
+  deploy: true
+
+testData:
+  marker: ci_cloudbeat

tests/deploy/values/local-host.yml

@@ -1,2 +1,5 @@
 cloudbeat:
-  deploy: false
+  deploy: false
+
+testData:
+  marker: rules