CM-31107 - Fix new CLI endpoints that should be applied to SCA scans only

cycode/cli/commands/scan/code_scanner.py

@@ -439,7 +439,7 @@ def perform_scan_async(
     scan_async_result = cycode_client.zipped_file_scan_async(zipped_documents, scan_type, scan_parameters)
     logger.debug('scan request has been triggered successfully, scan id: %s', scan_async_result.scan_id)
 
-    return poll_scan_results(cycode_client, scan_async_result.scan_id)
+    return poll_scan_results(cycode_client, scan_async_result.scan_id, scan_type)
 
 
 def perform_commit_range_scan_async(
@@ -455,11 +455,14 @@ def perform_commit_range_scan_async(
     )
 
     logger.debug('scan request has been triggered successfully, scan id: %s', scan_async_result.scan_id)
-    return poll_scan_results(cycode_client, scan_async_result.scan_id, timeout)
+    return poll_scan_results(cycode_client, scan_async_result.scan_id, scan_type, timeout)
 
 
 def poll_scan_results(
-    cycode_client: 'ScanClient', scan_id: str, polling_timeout: Optional[int] = None
+    cycode_client: 'ScanClient',
+    scan_id: str,
+    scan_type: str,
+    polling_timeout: Optional[int] = None,
 ) -> ZippedFileScanResult:
     if polling_timeout is None:
         polling_timeout = configuration_manager.get_scan_polling_timeout_in_seconds()
@@ -468,14 +471,14 @@ def poll_scan_results(
     end_polling_time = time.time() + polling_timeout
 
     while time.time() < end_polling_time:
-        scan_details = cycode_client.get_scan_details(scan_id)
+        scan_details = cycode_client.get_scan_details(scan_type, scan_id)
 
         if scan_details.scan_update_at is not None and scan_details.scan_update_at != last_scan_update_at:
             last_scan_update_at = scan_details.scan_update_at
             print_debug_scan_details(scan_details)
 
         if scan_details.scan_status == consts.SCAN_STATUS_COMPLETED:
-            return _get_scan_result(cycode_client, scan_id, scan_details)
+            return _get_scan_result(cycode_client, scan_type, scan_id, scan_details)
 
         if scan_details.scan_status == consts.SCAN_STATUS_ERROR:
             raise custom_exceptions.ScanAsyncError(
@@ -759,14 +762,14 @@ def _does_severity_match_severity_threshold(severity: str, severity_threshold: s
 
 
 def _get_scan_result(
-    cycode_client: 'ScanClient', scan_id: str, scan_details: 'ScanDetailsResponse'
+    cycode_client: 'ScanClient', scan_type: str, scan_id: str, scan_details: 'ScanDetailsResponse'
 ) -> ZippedFileScanResult:
     if not scan_details.detections_count:
         return init_default_scan_result(scan_id, scan_details.metadata)
 
-    wait_for_detections_creation(cycode_client, scan_id, scan_details.detections_count)
+    wait_for_detections_creation(cycode_client, scan_type, scan_id, scan_details.detections_count)
 
-    scan_detections = cycode_client.get_scan_detections(scan_id)
+    scan_detections = cycode_client.get_scan_detections(scan_type, scan_id)
     return ZippedFileScanResult(
         did_detect=True,
         detections_per_file=_map_detections_per_file(scan_detections),
@@ -792,15 +795,17 @@ def _try_get_report_url(metadata_json: Optional[str]) -> Optional[str]:
         return None
 
 
-def wait_for_detections_creation(cycode_client: 'ScanClient', scan_id: str, expected_detections_count: int) -> None:
+def wait_for_detections_creation(
+    cycode_client: 'ScanClient', scan_type: str, scan_id: str, expected_detections_count: int
+) -> None:
     logger.debug('Waiting for detections to be created')
 
     scan_persisted_detections_count = 0
     polling_timeout = consts.DETECTIONS_COUNT_VERIFICATION_TIMEOUT_IN_SECONDS
     end_polling_time = time.time() + polling_timeout
 
     while time.time() < end_polling_time:
-        scan_persisted_detections_count = cycode_client.get_scan_detections_count(scan_id)
+        scan_persisted_detections_count = cycode_client.get_scan_detections_count(scan_type, scan_id)
         logger.debug(
             f'Excepted {expected_detections_count} detections, got {scan_persisted_detections_count} detections '
             f'({expected_detections_count - scan_persisted_detections_count} more; '

cycode/cyclient/scan_client.py

@@ -20,22 +20,43 @@ def __init__(
         self.scan_cycode_client = scan_cycode_client
         self.scan_config = scan_config
 
-        self.SCAN_CONTROLLER_PATH = 'api/v1/cli-scan'
-        self.DETECTIONS_SERVICE_CONTROLLER_PATH = 'api/v1/detections/cli'
+        self._SCAN_CONTROLLER_PATH = 'api/v1/scan'
+        self._SCAN_CONTROLLER_PATH_SCA = 'api/v1/cli-scan'
+
+        self._DETECTIONS_SERVICE_CONTROLLER_PATH = 'api/v1/detections'
+        self._DETECTIONS_SERVICE_CONTROLLER_PATH_SCA = 'api/v1/detections/cli'
+
         self.POLICIES_SERVICE_CONTROLLER_PATH_V3 = 'api/v3/policies'
 
         self._hide_response_log = hide_response_log
 
+    def get_scan_controller_path(self, scan_type: str) -> str:
+        if scan_type == consts.SCA_SCAN_TYPE:
+            return self._SCAN_CONTROLLER_PATH_SCA
+
+        return self._SCAN_CONTROLLER_PATH
+
+    def get_detections_service_controller_path(self, scan_type: str) -> str:
+        if scan_type == consts.SCA_SCAN_TYPE:
+            return self._DETECTIONS_SERVICE_CONTROLLER_PATH_SCA
+
+        return self._DETECTIONS_SERVICE_CONTROLLER_PATH
+
+    def get_scan_service_url_path(self, scan_type: str) -> str:
+        service_path = self.scan_config.get_service_name(scan_type)
+        controller_path = self.get_scan_controller_path(scan_type)
+        return f'{service_path}/{controller_path}'
+
     def content_scan(self, scan_type: str, file_name: str, content: str, is_git_diff: bool = True) -> models.ScanResult:
-        path = f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/content'
+        path = f'{self.get_scan_service_url_path(scan_type)}/content'
         body = {'name': file_name, 'content': content, 'is_git_diff': is_git_diff}
         response = self.scan_cycode_client.post(
             url_path=path, body=body, hide_response_content_log=self._hide_response_log
         )
         return self.parse_scan_response(response)
 
     def get_zipped_file_scan_url_path(self, scan_type: str) -> str:
-        return f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/zipped-file'
+        return f'{self.get_scan_service_url_path(scan_type)}/zipped-file'
 
     def zipped_file_scan(
         self, scan_type: str, zip_file: InMemoryZip, scan_id: str, scan_parameters: dict, is_git_diff: bool = False
@@ -54,9 +75,7 @@ def zipped_file_scan(
     def get_zipped_file_scan_async_url_path(self, scan_type: str) -> str:
         async_scan_type = self.scan_config.get_async_scan_type(scan_type)
         async_entity_type = self.scan_config.get_async_entity_type(scan_type)
-
-        url_prefix = self.scan_config.get_scans_prefix()
-        return f'{url_prefix}/{self.SCAN_CONTROLLER_PATH}/{async_scan_type}/{async_entity_type}'
+        return f'{self.get_scan_service_url_path(scan_type)}/{async_scan_type}/{async_entity_type}'
 
     def zipped_file_scan_async(
         self, zip_file: InMemoryZip, scan_type: str, scan_parameters: dict, is_git_diff: bool = False
@@ -77,9 +96,7 @@ def multiple_zipped_file_scan_async(
         scan_parameters: dict,
         is_git_diff: bool = False,
     ) -> models.ScanInitializationResponse:
-        url_path = (
-            f'{self.scan_config.get_scans_prefix()}/{self.SCAN_CONTROLLER_PATH}/{scan_type}/repository/commit-range'
-        )
+        url_path = f'{self.get_scan_service_url_path(scan_type)}/{scan_type}/repository/commit-range'
         files = {
             'file_from_commit': ('multiple_files_scan.zip', from_commit_zip_file.read()),
             'file_to_commit': ('multiple_files_scan.zip', to_commit_zip_file.read()),
@@ -91,11 +108,12 @@ def multiple_zipped_file_scan_async(
         )
         return models.ScanInitializationResponseSchema().load(response.json())
 
-    def get_scan_details_path(self, scan_id: str) -> str:
-        return f'{self.scan_config.get_scans_prefix()}/{self.SCAN_CONTROLLER_PATH}/{scan_id}'
+    def get_scan_details_path(self, scan_type: str, scan_id: str) -> str:
+        return f'{self.get_scan_service_url_path(scan_type)}/{scan_id}'
 
-    def get_scan_details(self, scan_id: str) -> models.ScanDetailsResponse:
-        response = self.scan_cycode_client.get(url_path=self.get_scan_details_path(scan_id))
+    def get_scan_details(self, scan_type: str, scan_id: str) -> models.ScanDetailsResponse:
+        path = self.get_scan_details_path(scan_type, scan_id)
+        response = self.scan_cycode_client.get(url_path=path)
         return models.ScanDetailsResponseSchema().load(response.json())
 
     def get_detection_rules_path(self) -> str:
@@ -150,10 +168,10 @@ def get_detection_rules(
         # we are filtering rules by ids in-place for smooth migration when backend will be ready
         return self._filter_detection_rules_by_ids(self.parse_detection_rules_response(response), detection_rules_ids)
 
-    def get_scan_detections_path(self) -> str:
-        return f'{self.scan_config.get_detections_prefix()}/{self.DETECTIONS_SERVICE_CONTROLLER_PATH}/detections'
+    def get_scan_detections_path(self, scan_type: str) -> str:
+        return f'{self.scan_config.get_detections_prefix()}/{self.get_detections_service_controller_path(scan_type)}'
 
-    def get_scan_detections(self, scan_id: str) -> List[dict]:
+    def get_scan_detections(self, scan_type: str, scan_id: str) -> List[dict]:
         params = {'scan_id': scan_id}
 
         page_size = 200
@@ -166,8 +184,9 @@ def get_scan_detections(self, scan_id: str) -> List[dict]:
             params['page_size'] = page_size
             params['page_number'] = page_number
 
+            path = f'{self.get_scan_detections_path(scan_type)}/detections'
             response = self.scan_cycode_client.get(
-                url_path=self.get_scan_detections_path(),
+                url_path=path,
                 params=params,
                 hide_response_content_log=self._hide_response_log,
             ).json()
@@ -178,29 +197,27 @@ def get_scan_detections(self, scan_id: str) -> List[dict]:
 
         return detections
 
-    def get_get_scan_detections_count_path(self) -> str:
-        return f'{self.scan_config.get_detections_prefix()}/{self.DETECTIONS_SERVICE_CONTROLLER_PATH}/count'
+    def get_get_scan_detections_count_path(self, scan_type: str) -> str:
+        return f'{self.get_scan_detections_path(scan_type)}/count'
 
-    def get_scan_detections_count(self, scan_id: str) -> int:
+    def get_scan_detections_count(self, scan_type: str, scan_id: str) -> int:
         response = self.scan_cycode_client.get(
-            url_path=self.get_get_scan_detections_count_path(), params={'scan_id': scan_id}
+            url_path=self.get_get_scan_detections_count_path(scan_type), params={'scan_id': scan_id}
         )
         return response.json().get('count', 0)
 
     def commit_range_zipped_file_scan(
         self, scan_type: str, zip_file: InMemoryZip, scan_id: str
     ) -> models.ZippedFileScanResult:
-        url_path = (
-            f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/commit-range-zipped-file'
-        )
+        url_path = f'{self.get_scan_service_url_path(scan_type)}/commit-range-zipped-file'
         files = {'file': ('multiple_files_scan.zip', zip_file.read())}
         response = self.scan_cycode_client.post(
             url_path=url_path, data={'scan_id': scan_id}, files=files, hide_response_content_log=self._hide_response_log
         )
         return self.parse_zipped_file_scan_response(response)
 
     def get_report_scan_status_path(self, scan_type: str, scan_id: str) -> str:
-        return f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/{scan_id}/status'
+        return f'{self.get_scan_service_url_path(scan_type)}/{scan_id}/status'
 
     def report_scan_status(self, scan_type: str, scan_id: str, scan_status: dict) -> None:
         self.scan_cycode_client.post(url_path=self.get_report_scan_status_path(scan_type, scan_id), body=scan_status)

cycode/cyclient/scan_config_base.py

@@ -20,10 +20,6 @@ def get_async_entity_type(_: str) -> str:
         # we are migrating to "zippedfile" entity type. will be used later
         return 'repository'
 
-    @abstractmethod
-    def get_scans_prefix(self) -> str:
-        ...
-
     @abstractmethod
     def get_detections_prefix(self) -> str:
         ...
@@ -39,9 +35,6 @@ def get_service_name(self, scan_type: str) -> str:
         # sca and sast
         return '5004'
 
-    def get_scans_prefix(self) -> str:
-        return '5004'
-
     def get_detections_prefix(self) -> str:
         return '5016'
 
@@ -56,8 +49,5 @@ def get_service_name(self, scan_type: str) -> str:
         # sca and sast
         return 'scans'
 
-    def get_scans_prefix(self) -> str:
-        return 'scans'
-
     def get_detections_prefix(self) -> str:
         return 'detections'

process_executable_file.py

@@ -51,7 +51,12 @@ def get_hashes_of_every_file_in_the_directory(dir_path: Path) -> DirHashes:
     hashes = []
 
     for root, _, files in os.walk(dir_path):
-        hashes.extend(get_hashes_of_many_files(root, files,))
+        hashes.extend(
+            get_hashes_of_many_files(
+                root,
+                files,
+            )
+        )
 
     return hashes
 
@@ -60,7 +65,7 @@ def normalize_hashes_db(hashes: DirHashes, dir_path: Path) -> DirHashes:
     normalized_hashes = []
 
     for file_hash, file_path in hashes:
-        relative_file_path = file_path[file_path.find(dir_path.name):]
+        relative_file_path = file_path[file_path.find(dir_path.name) :]
         normalized_hashes.append((file_hash, relative_file_path))
 
     # sort by file path

tests/cyclient/mocked_responses/scan_client.py

@@ -119,9 +119,9 @@ def get_scan_detections_count_response(url: str) -> responses.Response:
     return responses.Response(method=responses.GET, url=url, json=json_response, status=200)
 
 
-def get_scan_detections_url(scan_client: ScanClient) -> str:
+def get_scan_detections_url(scan_client: ScanClient, scan_type: str) -> str:
     api_url = scan_client.scan_cycode_client.api_url
-    service_url = scan_client.get_scan_detections_path()
+    service_url = scan_client.get_scan_detections_path(scan_type)
     return f'{api_url}/{service_url}'
 
 
@@ -168,7 +168,9 @@ def mock_scan_async_responses(
     responses_module.add(get_scan_details_response(get_scan_details_url(scan_id, scan_client), scan_id))
     responses_module.add(get_detection_rules_response(get_detection_rules_url(scan_client)))
     responses_module.add(get_scan_detections_count_response(get_scan_detections_count_url(scan_client)))
-    responses_module.add(get_scan_detections_response(get_scan_detections_url(scan_client), scan_id, zip_content_path))
+    responses_module.add(
+        get_scan_detections_response(get_scan_detections_url(scan_client, scan_type), scan_id, zip_content_path)
+    )
     responses_module.add(get_report_scan_status_response(get_report_scan_status_url(scan_type, scan_id, scan_client)))
 
 

tests/cyclient/scan_config/test_default_scan_config.py

@@ -10,12 +10,6 @@ def test_get_service_name() -> None:
     assert default_scan_config.get_service_name('sast') == 'scans'
 
 
-def test_get_scans_prefix() -> None:
-    default_scan_config = DefaultScanConfig()
-
-    assert default_scan_config.get_scans_prefix() == 'scans'
-
-
 def test_get_detections_prefix() -> None:
     default_scan_config = DefaultScanConfig()
 

tests/cyclient/scan_config/test_dev_scan_config.py

@@ -10,12 +10,6 @@ def test_get_service_name() -> None:
     assert dev_scan_config.get_service_name('sast') == '5004'
 
 
-def test_get_scans_prefix() -> None:
-    dev_scan_config = DevScanConfig()
-
-    assert dev_scan_config.get_scans_prefix() == '5004'
-
-
 def test_get_detections_prefix() -> None:
     dev_scan_config = DevScanConfig()