add registry scanning system test

configurations/registry/check_aws.json

@@ -0,0 +1,4 @@
+{
+  "registryURI": "015253967648.dkr.ecr.eu-central-1.amazonaws.com",
+  "accessKeyID": "AKIAQHDJVCMQAN4N4I55"
+}

configurations/registry/check_azure.json

@@ -0,0 +1,4 @@
+{
+  "loginServer": "refaelregistrytest.azurecr.io",
+  "username": "systemtests"
+}

configurations/registry/check_google.json

@@ -0,0 +1,3 @@
+{
+  "registryURI": "europe-docker.pkg.dev"
+}

configurations/registry/check_quay.json

@@ -0,0 +1,4 @@
+{
+  "containerRegistryName": "quay.io",
+  "robotAccountName": "systemtests+systemtests"
+}

configurations/registry/create_quay.json

@@ -0,0 +1,6 @@
+{
+  "containerRegistryName": "quay.io",
+  "robotAccountName": "systemtests+systemtests",
+  "scanFrequency": "*/10 5 1 * *",
+  "repositories": ["systemtests/webgoat"]
+}

configurations/system/tests.py

@@ -6,6 +6,7 @@
 from .tests_cases import KubescapeTests, KSMicroserviceTests, RuntimeTests
 from .tests_cases.ks_vuln_scan_tests import KsVulnerabilityScanningTests
 from .tests_cases.notifications_tests import NotificationSTests
+from .tests_cases.registry_tests import RegistryTests
 from .tests_cases.payments_tests import PaymentTests
 from .tests_cases.relevant_vuln_scanning_tests import RelevantVulnerabilityScanningTests
 from .tests_cases.seccomp_tests import SeccompProfileTests
@@ -32,6 +33,7 @@ def all_tests_names():
     tests.extend(TestUtil.get_class_methods(IntegrationsTests))
     tests.extend(TestUtil.get_class_methods(SeccompProfileTests))
     tests.extend(TestUtil.get_class_methods(WorkflowsTests))
+    tests.extend(TestUtil.get_class_methods(RegistryTests))
     return tests
 
 
@@ -66,6 +68,8 @@ def get_test(test_name):
         return SeccompProfileTests().__getattribute__(test_name)()
     if test_name in TestUtil.get_class_methods(WorkflowsTests):
         return WorkflowsTests().__getattribute__(test_name)()
+    if test_name in TestUtil.get_class_methods(RegistryTests):
+        return RegistryTests().__getattribute__(test_name)()
 
 
 ALL_TESTS = all_tests_names()

configurations/system/tests_cases/registry_tests.py

@@ -0,0 +1,20 @@
+import inspect
+from os.path import join
+
+from systest_utils.statics import DEFAULT_NOTIFICATIONS_PATHS, DEFAULT_NOTIFICATIONS_DEPLOYMENT_PATH, \
+    DEFAULT_REGISTRY_PATHS
+from tests_scripts.users_notifications.alert_notifications import get_messages_from_teams_channel, \
+    enrich_teams_alert_channel, get_messages_from_slack_channel, enrich_slack_alert_channel
+from .structures import TestConfiguration
+
+class RegistryTests(object):
+
+    @staticmethod
+    def test_registry_scanning():
+        from tests_scripts.registry.registry_connectors import RegistryChecker
+        return TestConfiguration(
+            name=inspect.currentframe().f_code.co_name,
+            test_obj=RegistryChecker,
+            check_payload_file=join(DEFAULT_REGISTRY_PATHS, "check_{}.json"),
+            create_payload_file=join(DEFAULT_REGISTRY_PATHS, "create_quay.json") # we do sanity only for one provider
+        )

infrastructure/backend_api.py

@@ -98,6 +98,8 @@ class NotExistingCustomer(Exception):
 API_NOTIFICATIONS_UNSUBSCRIBE = "/api/v1/notifications/unsubscribe"
 API_NOTIFICATIONS_ALERTCHANNEL = "/api/v1/notifications/alertChannel"
 
+API_REGISTRY_MANAGEMENT = "/api/v1/registry/management"
+
 API_ATTACK_CHAINS = "/api/v1/attackchains"
 
 API_NETWORK_POLICIES = "/api/v1/networkpolicies"
@@ -2179,6 +2181,54 @@ def remove_alert_channel(self, guid) -> requests.Response:
                     self.customer, res.status_code, res.text))
         return res
 
+    def check_registry(self, payload, provider) -> requests.Response:
+        res = self.post(API_REGISTRY_MANAGEMENT + "/" + provider + "/repositories", cookies=self.selected_tenant_cookie, data=json.dumps(payload))
+        if not 200 <= res.status_code < 300:
+            raise Exception(
+                'Error accessing dashboard. Request: check registry "%s" (code: %d, message: %s)' % (
+                    self.customer, res.status_code, res.text))
+        return res
+
+    def create_registry(self, payload, provider) -> requests.Response:
+        res = self.post(API_REGISTRY_MANAGEMENT + "/" + provider, cookies=self.selected_tenant_cookie, data=json.dumps(payload))
+        if not 200 <= res.status_code < 300:
+            raise Exception(
+                'Error accessing dashboard. Request: create registry "%s" (code: %d, message: %s)' % (
+                    self.customer, res.status_code, res.text))
+        return res
+
+    def get_registry(self, provider, guid) -> requests.Response:
+        res = self.get(API_REGISTRY_MANAGEMENT + "/" + provider + "/" + guid, cookies=self.selected_tenant_cookie)
+        if not 200 <= res.status_code < 300:
+            raise Exception(
+                'Error accessing dashboard. Request: get registry "%s" (code: %d, message: %s)' % (
+                    self.customer, res.status_code, res.text))
+        return res
+
+    def get_all_registries(self, provider) -> requests.Response:
+        res = self.get(API_REGISTRY_MANAGEMENT + "/" + provider, cookies=self.selected_tenant_cookie)
+        if not 200 <= res.status_code < 300:
+            raise Exception(
+                'Error accessing dashboard. Request: get registry "%s" (code: %d, message: %s)' % (
+                    self.customer, res.status_code, res.text))
+        return res
+
+    def update_registry(self, payload, provider, guid) -> requests.Response:
+        res = self.put(API_REGISTRY_MANAGEMENT + "/" + provider + "/" + guid, cookies=self.selected_tenant_cookie, data=json.dumps(payload))
+        if not 200 <= res.status_code < 300:
+            raise Exception(
+                'Error accessing dashboard. Request: update registry "%s" (code: %d, message: %s)' % (
+                    self.customer, res.status_code, res.text))
+        return res
+
+    def delete_registry(self, provider, guid) -> requests.Response:
+        res = self.delete(API_REGISTRY_MANAGEMENT + "/" + provider + "/" + guid, cookies=self.selected_tenant_cookie)
+        if not 200 <= res.status_code < 300:
+            raise Exception(
+                'Error accessing dashboard. Request: delete registry "%s" (code: %d, message: %s)' % (
+                    self.customer, res.status_code, res.text))
+        return res
+
     def get_attack_chains(self, cluster_name=None):
         params = {"customerGUID": self.selected_tenant_id}
 

system_test_mapping.json

@@ -1230,6 +1230,19 @@
     "skip_on_environment": "production,production-us",
     "owner": ""
   },
+  "test_registry_scanning": {
+    "target": [
+      "In cluster",
+      "Backend"
+    ],
+    "target_repositories": [
+      "cadashboardbe",
+      "event-ingester-service"
+    ],
+    "description": "",
+    "skip_on_environment": "",
+    "owner": ""
+  },
   "relevant_data_is_appended": {
     "target": [
       "In cluster",

systest_utils/statics.py

@@ -75,6 +75,9 @@
 DEFAULT_WORKFLOWS_PATHS = os.path.abspath(os.path.join('configurations', 'workflows_notifications'))
 DEFAULT_WORKFLOWS_DEPLOYMENT_PATH = os.path.join(DEFAULT_WORKFLOWS_PATHS, 'deployments')
 
+# registry
+DEFAULT_REGISTRY_PATHS = os.path.abspath(os.path.join('configurations', 'registry'))
+
 # kdr
 DEFAULT_KDR_DEPLOYMENT_PATH = os.path.join(DEFAULT_K8S_PATHS, 'deployments')
 

tests_scripts/registry/base_registry.py

@@ -0,0 +1,14 @@
+from tests_scripts import base_test
+from systest_utils import statics
+
+
+class BaseRegistry(base_test.BaseTest):
+
+
+    def __init__(self, test_obj=None, backend=None, test_driver=None):
+        super().__init__(test_driver=test_driver, test_obj=test_obj, backend=backend)
+
+    def cleanup(self, **kwargs):
+        super().cleanup(**kwargs)        
+        return statics.SUCCESS, ""
+

tests_scripts/registry/registry_connectors.py

@@ -0,0 +1,138 @@
+import base64
+import json
+import os
+from typing import List, Dict
+
+from systest_utils import Logger, TestUtil, statics
+from tests_scripts.helm.base_helm import BaseHelm
+
+REGISTRY_PROVIDERS: List[Dict[str, any]] = [
+    {
+        "provider": "quay",
+        "secret_field_name": "robotAccountToken",
+        "secret_value_env_var": "QUAY_REGISTRY_ACCESS_TOKEN",
+    },
+    {
+        "provider": "aws",
+        "secret_field_name": "secretAccessKey",
+        "secret_value_env_var": "AWS_REGISTRY_SECRET_KEY",
+    },
+    {
+        "provider": "azure",
+        "secret_field_name": "accessToken",
+        "secret_value_env_var": "AZURE_REGISTRY_ACCESS_TOKEN",
+    },
+    {
+        "provider": "google",
+        "secret_field_name": "key",
+        "secret_value_env_var": "GOOGLE_REGISTRY_KEY",
+    },
+]
+
+class RegistryChecker(BaseHelm):
+
+    def __init__(self, test_obj=None, backend=None, kubernetes_obj=None, test_driver=None):
+        super(RegistryChecker, self).__init__(test_driver=test_driver, test_obj=test_obj, backend=backend,
+                                              kubernetes_obj=kubernetes_obj)
+
+    def start(self):
+        Logger.logger.info('Stage 1: Install kubescape with helm-chart')
+        cluster, _ = self.setup(apply_services=False)
+        self.install_kubescape()
+        Logger.logger.info('Stage 2: Check registries connection')
+        self.check_registries_connection(cluster)
+        Logger.logger.info('Stage 3: Check quay.io registry CRUD operations')
+        self.check_registry_crud(cluster)
+        return self.cleanup(cluster=cluster)
+
+    def check_registry_crud(self, cluster):
+        quay_config = REGISTRY_PROVIDERS[0]
+        provider = quay_config["provider"]
+
+        Logger.logger.info("Loading creation payload file")
+        file_path = self.test_obj["create_payload_file"]
+        with open(file_path, 'r') as file:
+            data = json.load(file)
+        data["clusterName"] = cluster
+        Logger.logger.info("Loading creation secret from env var")
+        secret = os.getenv(quay_config["secret_value_env_var"])
+        data[quay_config["secret_field_name"]] = secret
+
+        Logger.logger.info("Calling creation API")
+        created = self.backend.create_registry(data, provider)
+        assert created, "Expected created registry"
+        guid = created.json()["guid"]
+
+        Logger.logger.info("Checking if registry scan is completed")
+        self.assert_registry_scan_completed(provider, guid)
+
+        registry = self.backend.get_registry(provider, guid)
+        assert registry, "Expected to get registry"
+        update_payload = registry.json()
+        update_payload["scanFrequency"] = "30 * * 5 *"
+        Logger.logger.info("Updating registry with new scan scanFrequency")
+        updated = self.backend.update_registry(update_payload, provider, guid)
+        assert updated, "Expected updated registry"
+        assert updated.json()["scanFrequency"] == "30 * * 5 *", "Expected scanFrequency to be updated"
+
+        Logger.logger.info("Deleting registry")
+        self.backend.delete_registry(provider, guid)
+        registries = self.get_all_quay_registries_for_cluster(cluster)
+        assert len(registries) == 0, "Expected to have no registries"
+        self.cleanup(cluster=cluster)
+
+    def check_registries_connection(self, cluster):
+        for provider_config in REGISTRY_PROVIDERS:
+            provider = provider_config["provider"]
+            Logger.logger.info(f'{provider}: Loading payload file')
+            file_path = self.test_obj["check_payload_file"].format(provider)
+            with open(file_path, 'r') as file:
+                data = json.load(file)
+            data["clusterName"] = cluster
+            Logger.logger.info(f'{provider}: Loading secrets from env var')
+            secret = os.getenv(provider_config["secret_value_env_var"])
+            if provider == "google":
+                secret = json.loads(base64.b64decode(secret).decode('utf-8'))
+            data[provider_config["secret_field_name"]] = secret
+            Logger.logger.info(f'{provider}: Calling repositories API')
+            repositories_response = self.backend.check_registry(data, provider)
+            assert repositories_response, "Expected repositories"
+            assert any("systemtests/webgoat" in item for item in repositories_response.json()), \
+                f"'systemtests/webgoat' not found in any item of {repositories_response.json()}"
+
+    def cleanup(self, **kwargs):
+        self.delete_all_quay_registries_for_cluster(kwargs['cluster'])
+        return super().cleanup()
+
+    def install_kubescape(self, helm_kwargs: dict = None):
+        self.add_and_upgrade_armo_to_repo()
+        self.install_armo_helm_chart(helm_kwargs=helm_kwargs)
+        self.verify_running_pods(namespace=statics.CA_NAMESPACE_FROM_HELM_NAME)
+
+    def assert_registry_scan_completed(self, provider, guid):
+        TestUtil.sleep(30, "waiting for registry scan to complete")
+        for i in range(60):
+            try:
+                registry = self.backend.get_registry(provider, guid)
+                assert registry, "Expected registry"
+                assert registry.json()["scanStatus"] == "Completed"
+                break
+            except AssertionError:
+                if i == 59:
+                    raise
+                TestUtil.sleep(2, "waiting for registry scan to complete")
+
+    def delete_all_quay_registries_for_cluster(self, cluster):
+        leftovers = self.get_all_quay_registries_for_cluster(cluster)
+        for r in leftovers:
+            self.backend.delete_registry(r["provider"], r["guid"])
+
+    def get_all_quay_registries_for_cluster(self, cluster):
+        ret = []
+        resp = self.backend.get_all_registries("quay").json()
+        if resp['total']['value'] == 0:
+            return ret
+        for r in resp['response']:
+            if r["clusterName"] == cluster:
+                ret.append(r)
+        return ret