Bugfixes, high level configuration classes. (#2)

* bug fixes

* remove unused import

* add missing undocumented api

* implement high level tenant config

* more high level config impls

Author: nleach999

cxone_api/__init__.py

@@ -130,7 +130,7 @@ def __init__(self):
 
 class ApiDEU(CxOneApiEndpoint):
     def __init__(self):
-        super().__init__("deu.checkmarx.net")
+        super().__init__("deu.ast.checkmarx.net")
 
 class ApiANZ(CxOneApiEndpoint):
     def __init__(self):

cxone_api/exceptions.py

@@ -1,4 +1,5 @@
 import re
+from typing import Type, Any, List
 
 class EndpointException(BaseException):
     pass
@@ -35,3 +36,17 @@ class ResponseException(BaseException):
 
 class ScanException(BaseException):
     pass
+
+class ConfigurationException(BaseException):
+
+    @staticmethod
+    def wrong_type(value : Any, type : Type):
+        return ConfigurationException(f"Value of [{value}] is not of type [{type}].")
+
+    @staticmethod
+    def not_in_enum(value : Any, valid_values : List[str]):
+        return ConfigurationException(f"Value of [{value}] is not one of {valid_values}.")
+
+    @staticmethod
+    def read_only():
+        return ConfigurationException("Configuration values are read-only.")

cxone_api/high/scan_configuration/__init__.py

@@ -0,0 +1,77 @@
+from typing import List, Dict
+from .base_config import BaseScanConfiguration
+from ...client import CxOneClient
+from ..util import json_on_ok
+from ...exceptions import ResponseException, ConfigurationException
+from ...low.scan_configuration import \
+  retrieve_tenant_configuration, update_tenant_configuration, delete_tenant_configuration, \
+  retrieve_project_configuration, update_project_configuration, delete_project_configuration, \
+  retrieve_scan_configuration
+
+
+class TenantScanConfiguration(BaseScanConfiguration):
+
+  def __init__(self, client : CxOneClient):
+    super().__init__(client)
+
+  def _origin_level(self) -> str:
+    return "Tenant"
+
+  async def _load_config(self) -> List[Dict]:
+    return json_on_ok(await retrieve_tenant_configuration(self.client))
+
+  async def _write_config(self, items : List[Dict]) -> None:
+    resp = await update_tenant_configuration(self.client, items)
+    if not resp.ok:
+      raise ResponseException(resp)
+
+  async def _write_deletes(self, keys : List[str]) -> None:
+    resp = await delete_tenant_configuration(self.client, config_keys=','.join(keys))
+    if not resp.ok:
+      raise ResponseException(resp)
+
+class ProjectScanConfiguration(BaseScanConfiguration):
+  def __init__(self, client : CxOneClient, project_id : str):
+    super().__init__(client)
+    self.__project_id = project_id
+
+  @property
+  def project_id(self) -> str:
+    return self.__project_id
+
+  def _origin_level(self) -> str:
+    return "Project"
+
+  async def _load_config(self) -> List[Dict]:
+    return json_on_ok(await retrieve_project_configuration(self.client, project_id=self.project_id))
+
+  async def _write_config(self, items : List[Dict]) -> None:
+    resp = await update_project_configuration(self.client, items, project_id=self.project_id)
+    if not resp.ok:
+      raise ResponseException(resp)
+
+  async def _write_deletes(self, keys : List[str]) -> None:
+    resp = await delete_project_configuration(self.client, config_keys=','.join(keys), project_id=self.project_id)
+    if not resp.ok:
+      raise ResponseException(resp)
+
+class ScanConfiguration(ProjectScanConfiguration):
+  def __init__(self, client : CxOneClient, project_id : str, scan_id : str):
+    super().__init__(client, project_id)
+    self.__scan_id = scan_id
+
+  @property
+  def scan_id(self) -> str:
+    return self.__scan_id
+
+  def _origin_level(self) -> str:
+    return "Scan"
+
+  async def _load_config(self) -> List[Dict]:
+    return json_on_ok(await retrieve_scan_configuration(self.client, project_id=self.project_id, scan_id=self.scan_id))
+
+  async def _write_config(self, items : List[Dict]) -> None:
+    raise ConfigurationException.read_only()
+
+  async def _write_deletes(self, keys : List[str]) -> None:
+    raise ConfigurationException.read_only()

cxone_api/high/scan_configuration/base_config.py

@@ -0,0 +1,31 @@
+from .categories import SastScanConfig, ScaScanConfig, ApiSecurityScanConfig, ContainerScanConfig
+from .element_handler import CxOneConfigElementHandler
+from ...client import CxOneClient
+
+
+class BaseScanConfiguration(CxOneConfigElementHandler):
+  def __init__(self, client : CxOneClient):
+    super().__init__(client)
+    self.__sast = SastScanConfig(self)
+    self.__sca = ScaScanConfig(self)
+    self.__api = ApiSecurityScanConfig(self)
+    self.__containers = ContainerScanConfig(self)
+  
+  @property
+  def SAST(self) -> SastScanConfig:
+    return self.__sast
+
+  @property
+  def SCA(self) -> ScaScanConfig:
+    return self.__sca
+
+  @property
+  def APISec(self) -> ApiSecurityScanConfig:
+    return self.__api
+
+  @property
+  def Containers(self) -> ContainerScanConfig:
+    return self.__containers
+
+  def _origin_level(self) -> str:
+    raise NotImplemented("_origin_level")

cxone_api/high/scan_configuration/categories.py

@@ -0,0 +1,191 @@
+from .element_handler import CxOneConfigElementHandler
+from ...exceptions import ConfigurationException, CommunicationException
+from ...low.misc import retrieve_preset_list
+from ..util import json_on_ok
+from typing import Any, Type, List, Awaitable, Union
+from .value_types import ScanLanguageMode, RestListEnum
+import asyncio
+
+class BaseConfigurationCategory:
+  def __init__(self, container : CxOneConfigElementHandler):
+    self.__config = container
+  
+  @property
+  def _config(self) -> CxOneConfigElementHandler:
+    return self.__config
+
+
+class ConfigurationPropertyHandler:
+
+  def __init__(self, handler : CxOneConfigElementHandler, prop_key : str, value_type : Union[Type, RestListEnum]):
+    self.__handler = handler
+    self.__key = prop_key
+    self.__value_type = value_type
+
+  def __convert_from(self, value : Any) -> Any:
+    if isinstance(self.__value_type, RestListEnum):
+      return str(value)
+
+    if self.__value_type is bool:
+      if isinstance(value, str):
+        return (not value == '') or bool(value)
+      else:
+        return bool(value)
+    else:
+      return value
+
+  def __convert_to(self, value : Any) -> Any:
+    if self.__value_type is bool:
+      return str(bool(value)).lower()
+    
+    if self.__value_type is str and not isinstance(value, str):
+      return str(value)
+    
+    return value
+  
+
+  async def setValue(self, value : Any) -> None:
+    if isinstance(self.__value_type, RestListEnum):
+      if not await self.__value_type.is_valid(value):
+        raise ConfigurationException.not_in_enum(value, await self.__value_type.get_enum())
+    elif not isinstance(value, self.__value_type) and \
+      not isinstance(self.__value_type(value), self.__value_type):
+      raise ConfigurationException.wrong_type(value, self.__value_type)
+
+    await self.__handler._set_value(self.__key, self.__convert_to(value))
+
+  async def getValue(self) -> Any:
+    return self.__convert_from(await self.__handler._get_value(self.__key))
+
+  async def setOverride(self, value : bool) -> None:
+    await self.__handler._set_override(self.__key, value)
+
+  async def getOverride(self) -> bool:
+    return await self.__handler._get_override(self.__key)
+
+  async def reset(self) -> None:
+    await self.__handler._delete_config(self.__key)
+
+
+class SastPresetType(RestListEnum):
+
+  def __init__(self, retrieve_lambda : Awaitable[List[str]]):
+    super().__init__()
+    self.__get_lambda = retrieve_lambda
+    self.__lock = asyncio.Lock()
+    self.__enum = None
+
+  async def get_enum(self) -> List[str]:
+    async with self.__lock:
+      if self.__enum == None:
+        self.__enum = await self.__get_lambda()
+
+    return self.__enum
+
+
+class SastScanConfig(BaseConfigurationCategory):
+  def __init__(self, container : CxOneConfigElementHandler):
+    super().__init__(container)
+    self.__fastscan = ConfigurationPropertyHandler(self._config, "scan.config.sast.fastScanMode", bool)
+    self.__exclusions = ConfigurationPropertyHandler(self._config, "scan.config.sast.filter", str)
+    self.__incremental = ConfigurationPropertyHandler(self._config, "scan.config.sast.incremental", bool)
+    self.__rec = ConfigurationPropertyHandler(self._config, "scan.config.sast.recommendedExclusions", bool)
+    self.__verbose = ConfigurationPropertyHandler(self._config, "scan.config.sast.engineVerbose", bool)
+    self.__mode = ConfigurationPropertyHandler(self._config, "scan.config.sast.languageMode", ScanLanguageMode)
+
+    async def get_presets() -> List[str]:
+      pl_resp = await retrieve_preset_list(self._config.client)
+      if not pl_resp.ok:
+        raise CommunicationException(pl_resp)
+      
+      pl = json_on_ok(pl_resp)
+
+      return [x['name'] for x in pl]
+
+    self.__preset = ConfigurationPropertyHandler(self._config, "scan.config.sast.presetName", 
+                                                 SastPresetType(get_presets))
+
+  @property
+  def FastScan(self) -> ConfigurationPropertyHandler:
+    return self.__fastscan
+  
+  @property
+  def Exclusions(self) -> ConfigurationPropertyHandler:
+    return self.__exclusions
+
+  @property
+  def IncrementalScan(self) -> ConfigurationPropertyHandler:
+    return self.__incremental
+
+  @property
+  def Preset(self) -> ConfigurationPropertyHandler:
+    return self.__preset
+
+  @property
+  def RecommendedExclusions(self) -> ConfigurationPropertyHandler:
+    return self.__rec
+
+  @property
+  def EngineVerbose(self) -> ConfigurationPropertyHandler:
+    return self.__verbose
+
+  @property
+  def LanguageMode(self) -> ConfigurationPropertyHandler:
+    return self.__mode
+
+class ScaScanConfig(BaseConfigurationCategory):
+  def __init__(self, container : CxOneConfigElementHandler):
+    super().__init__(container)
+    self.__exclusions = ConfigurationPropertyHandler(self._config, "scan.config.sca.filter", str)
+    self.__obs_pattern = ConfigurationPropertyHandler(self._config, "scan.config.sca.obfuscatePackagesPattern", str)
+    self.__exp_path = ConfigurationPropertyHandler(self._config, "scan.config.sca.ExploitablePath", bool)
+    self.__exp_path_days = ConfigurationPropertyHandler(self._config, "scan.config.sca.LastSastScanTime", str)
+
+  @property
+  def Exclusions(self) -> ConfigurationPropertyHandler:
+    return self.__exclusions
+
+  @property
+  def ObfuscatePackagesPattern(self) -> ConfigurationPropertyHandler:
+    return self.__obs_pattern
+
+  @property
+  def ExploitablePath(self) -> ConfigurationPropertyHandler:
+    return self.__exp_path
+
+  @property
+  def ExploitablePathLastScanDays(self) -> ConfigurationPropertyHandler:
+    return self.__exp_path_days
+
+class ApiSecurityScanConfig(BaseConfigurationCategory):
+  def __init__(self, container : CxOneConfigElementHandler):
+    super().__init__(container)
+    self.__exclusions = ConfigurationPropertyHandler(self._config, "scan.config.apisec.swaggerFilter", str)
+
+  @property
+  def SwaggerFilter(self) -> ConfigurationPropertyHandler:
+    return self.__exclusions
+
+class ContainerScanConfig(BaseConfigurationCategory):
+  def __init__(self, container : CxOneConfigElementHandler):
+    super().__init__(container)
+    self.__exclusions = ConfigurationPropertyHandler(self._config, "scan.config.containers.filesFilter", str)
+    self.__package_filter = ConfigurationPropertyHandler(self._config, "scan.config.containers.packagesFilter", str)
+    self.__img_tag_filter = ConfigurationPropertyHandler(self._config, "scan.config.containers.imagesFilter", str)
+    self.__exclude_non_final = ConfigurationPropertyHandler(self._config, "scan.config.containers.nonFinalStagesFilter", bool)
+
+  @property
+  def Exclusions(self) -> ConfigurationPropertyHandler:
+    return self.__exclusions
+
+  @property
+  def ImageTagFilter(self) -> ConfigurationPropertyHandler:
+    return self.__img_tag_filter
+
+  @property
+  def PrivatePackageFilter(self) -> ConfigurationPropertyHandler:
+    return self.__package_filter
+
+  @property
+  def ExclusionNonFinalStagesFilter(self) -> ConfigurationPropertyHandler:
+    return self.__exclude_non_final