Accounts - cloud and posture

configurations/system/tests.py

@@ -13,6 +13,7 @@
 from .tests_cases.security_risks_tests import SecurityRisksTests
 from .tests_cases.vuln_scan_tests import VulnerabilityScanningTests
 from .tests_cases.workflows_tests import WorkflowsTests
+from .tests_cases.accounts_tests import AccountsTests
 
 
 def all_tests_names():
@@ -34,6 +35,7 @@ def all_tests_names():
     tests.extend(TestUtil.get_class_methods(SeccompProfileTests))
     tests.extend(TestUtil.get_class_methods(WorkflowsTests))
     tests.extend(TestUtil.get_class_methods(RegistryTests))
+    tests.extend(TestUtil.get_class_methods(AccountsTests))
     return tests
 
 
@@ -70,6 +72,8 @@ def get_test(test_name):
         return WorkflowsTests().__getattribute__(test_name)()
     if test_name in TestUtil.get_class_methods(RegistryTests):
         return RegistryTests().__getattribute__(test_name)()
+    if test_name in TestUtil.get_class_methods(AccountsTests):
+        return AccountsTests().__getattribute__(test_name)()
 
 
 ALL_TESTS = all_tests_names()

configurations/system/tests_cases/accounts_tests.py

@@ -0,0 +1,14 @@
+import inspect
+
+from .structures import KubescapeConfiguration
+
+
+class AccountsTests(object):
+
+    @staticmethod
+    def accounts():
+        from tests_scripts.accounts.accounts import Accounts
+        return KubescapeConfiguration(
+            name=inspect.currentframe().f_code.co_name,
+            test_obj=Accounts,
+        )

infrastructure/backend_api.py

@@ -142,6 +142,13 @@ class NotExistingCustomer(Exception):
 API_WEBHOOKS = "/api/v1/notifications/teams"
 API_TEAMS_TEST_MESSAGE = "/api/v1/notifications/teams/testMessage"
 
+API_ACCOUNTS = "/api/v1/accounts"
+API_ACCOUNTS_CLOUD_LIST = "/api/v1/accounts/cloud/list"
+API_ACCOUNTS_KUBERNETES_LIST = "/api/v1/accounts/kubernetes/list"
+API_ACCOUNTS_AWS_REGIONS = "/api/v1/accounts/aws/regions"
+API_UNIQUEVALUES_ACCOUNTS_CLOUD= "/api/v1/uniqueValues/accounts/cloud"
+API_UNIQUEVALUES_ACCOUNTS_KUBERNETES = "/api/v1/uniqueValues/accounts/kubernetes"
+
 
 
 def deco_cookie(func):
@@ -3085,7 +3092,113 @@ def test_webhook_message(self, body):
         return r.json()
 
 
+    def get_cloud_accounts(self,  body=None, **kwargs):
+        url = API_ACCOUNTS_CLOUD_LIST
+        if body is None:
+            body = {"pageSize": 150, "pageNum": 1}
+
+        params = {"customerGUID": self.selected_tenant_id}
+        if kwargs:
+            params.update(**kwargs)
+        r = self.post(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error accessing cloud accounts. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+    def get_kubernetes_accounts(self,  body=None, **kwargs):
+        url = API_ACCOUNTS_KUBERNETES_LIST
+        if body is None:
+            body = {"pageSize": 150, "pageNum": 1}
+
+        params = {"customerGUID": self.selected_tenant_id}
+        if kwargs:
+            params.update(**kwargs)
+        r = self.post(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error accessing cloud accounts. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+
+    def get_cloud_accounts_uniquevalues(self, body):
+        params = {"customerGUID": self.selected_tenant_id}
+
+        Logger.logger.info("get_cloud_accounts_uniquevalues body: %s" % body)
 
+        r = self.post(API_UNIQUEVALUES_ACCOUNTS_CLOUD, params=params, json=body)
+
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error accessing dashboard. Request: get_cloud_accounts_uniquevalues "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+    def get_kubernetes_accounts_uniquevalues(self, body):
+        params = {"customerGUID": self.selected_tenant_id}
+
+        Logger.logger.info("get_kubernetes_accounts_uniquevalues body: %s" % body)
+
+        r = self.post(API_UNIQUEVALUES_ACCOUNTS_KUBERNETES, params=params, json=body)
+
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error accessing dashboard. Request: get_kubernetes_accounts_uniquevalues "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+    def create_cloud_account(self, body, provider):
+        url = API_ACCOUNTS
+        params = {"customerGUID": self.selected_tenant_id,
+                  "provider": provider}
+        r = self.post(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error creating cloud account. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+    def delete_cloud_account(self, guid):
+        url = API_ACCOUNTS
+        params = {"customerGUID": self.selected_tenant_id}
+        body = {
+            "innerFilters": [
+                {
+                    "guid": guid
+                }
+            ]
+        }
+        r = self.delete(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error deleting cloud account. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+
+
+    def update_cloud_account(self, body, provider):
+        url = API_ACCOUNTS
+        params = {"customerGUID": self.selected_tenant_id,
+                  "provider": provider}
+        r = self.put(url, params=params, json=body)
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error updating cloud account. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
+
+
+    def get_aws_regions(self):
+        url = API_ACCOUNTS_AWS_REGIONS
+        r = self.get(url, params={"customerGUID": self.selected_tenant_id})
+        if not 200 <= r.status_code < 300:
+            raise Exception(
+                'Error accessing AWS regions. Customer: "%s" (code: %d, message: %s)' % (
+                    self.customer, r.status_code, r.text))
+        return r.json()
 
 
 

system_test_mapping.json

@@ -1689,5 +1689,18 @@
     "description": "Checks workflows configurations",
     "skip_on_environment": "",
     "owner": "[email protected]"
+  },
+  "accounts": {
+    "target": [
+      "Backend"
+    ],
+    "target_repositories": [
+      "config-service-dummy",
+      "cadashboardbe-dummy",
+      "event-ingester-service-dummy"
+    ],
+    "description": "Checks accounts",
+    "skip_on_environment": "",
+    "owner": "[email protected]"
   }
 }