CM-48357 - Fix SCA restore error handling (#309)

Author: MarshalX

cycode/cli/apps/scan/code_scanner.py

@@ -683,8 +683,8 @@ def try_get_git_remote_url(path: str) -> Optional[str]:
         remote_url = git_proxy.get_repo(path).remotes[0].config_reader.get('url')
         logger.debug('Found Git remote URL, %s', {'remote_url': remote_url, 'path': path})
         return remote_url
-    except Exception as e:
-        logger.debug('Failed to get Git remote URL', exc_info=e)
+    except Exception:
+        logger.debug('Failed to get Git remote URL. Probably not a Git repository')
         return None
 
 
@@ -706,7 +706,9 @@ def _get_plastic_repository_name(path: str) -> Optional[str]:
             f'--fieldseparator={consts.PLASTIC_VCS_DATA_SEPARATOR}',
         ]
 
-        status = shell(command=command, timeout=consts.PLASTIC_VSC_CLI_TIMEOUT, working_directory=path)
+        status = shell(
+            command=command, timeout=consts.PLASTIC_VSC_CLI_TIMEOUT, working_directory=path, silent_exc_info=True
+        )
         if not status:
             logger.debug('Failed to get Plastic repository name (command failed)')
             return None
@@ -717,8 +719,8 @@ def _get_plastic_repository_name(path: str) -> Optional[str]:
             return None
 
         return status_parts[2].strip()
-    except Exception as e:
-        logger.debug('Failed to get Plastic repository name', exc_info=e)
+    except Exception:
+        logger.debug('Failed to get Plastic repository name. Probably not a Plastic repository')
         return None
 
 
@@ -738,7 +740,9 @@ def _get_plastic_repository_list(working_dir: Optional[str] = None) -> dict[str,
     try:
         command = ['cm', 'repo', 'ls', f'--format={{repname}}{consts.PLASTIC_VCS_DATA_SEPARATOR}{{repguid}}']
 
-        status = shell(command=command, timeout=consts.PLASTIC_VSC_CLI_TIMEOUT, working_directory=working_dir)
+        status = shell(
+            command=command, timeout=consts.PLASTIC_VSC_CLI_TIMEOUT, working_directory=working_dir, silent_exc_info=True
+        )
         if not status:
             logger.debug('Failed to get Plastic repository list (command failed)')
             return repo_name_to_guid

cycode/cli/files_collector/sca/base_restore_dependencies.py

@@ -1,9 +1,9 @@
+import os
 from abc import ABC, abstractmethod
 from typing import Optional
 
 import typer
 
-from cycode.cli.logger import logger
 from cycode.cli.models import Document
 from cycode.cli.utils.path_utils import get_file_content, get_file_dir, get_path_from_context, join_paths
 from cycode.cli.utils.shell_executor import shell
@@ -15,30 +15,27 @@ def build_dep_tree_path(path: str, generated_file_name: str) -> str:
 
 def execute_commands(
     commands: list[list[str]],
-    file_name: str,
-    command_timeout: int,
-    dependencies_file_name: Optional[str] = None,
+    timeout: int,
+    output_file_path: Optional[str] = None,
     working_directory: Optional[str] = None,
 ) -> Optional[str]:
     try:
-        all_dependencies = []
+        outputs = []
 
-        # Run all commands and collect outputs
         for command in commands:
-            dependencies = shell(command=command, timeout=command_timeout, working_directory=working_directory)
-            all_dependencies.append(dependencies)  # Collect each command's output
+            command_output = shell(command=command, timeout=timeout, working_directory=working_directory)
+            if command_output:
+                outputs.append(command_output)
 
-        dependencies = '\n'.join(all_dependencies)
+        joined_output = '\n'.join(outputs)
 
-        # Write all collected outputs to the file if dependencies_file_name is provided
-        if dependencies_file_name:
-            with open(dependencies_file_name, 'w') as output_file:  # Open once in 'w' mode to start fresh
-                output_file.writelines(dependencies)
-    except Exception as e:
-        logger.debug('Failed to restore dependencies via shell command, %s', {'filename': file_name}, exc_info=e)
+        if output_file_path:
+            with open(output_file_path, 'w', encoding='UTF-8') as output_file:
+                output_file.writelines(joined_output)
+    except Exception:
         return None
 
-    return dependencies
+    return joined_output
 
 
 class BaseRestoreDependencies(ABC):
@@ -64,27 +61,25 @@ def try_restore_dependencies(self, document: Document) -> Optional[Document]:
         relative_restore_file_path = build_dep_tree_path(document.path, self.get_lock_file_name())
         working_directory_path = self.get_working_directory(document)
 
-        if self.verify_restore_file_already_exist(restore_file_path):
-            restore_file_content = get_file_content(restore_file_path)
-        else:
-            output_file_path = restore_file_path if self.create_output_file_manually else None
-            execute_commands(
+        if not self.verify_restore_file_already_exist(restore_file_path):
+            output = execute_commands(
                 self.get_commands(manifest_file_path),
-                manifest_file_path,
                 self.command_timeout,
-                output_file_path,
-                working_directory_path,
+                output_file_path=restore_file_path if self.create_output_file_manually else None,
+                working_directory=working_directory_path,
             )
-            restore_file_content = get_file_content(restore_file_path)
+            if output is None:  # one of the commands failed
+                return None
 
+        restore_file_content = get_file_content(restore_file_path)
         return Document(relative_restore_file_path, restore_file_content, self.is_git_diff)
 
     def get_working_directory(self, document: Document) -> Optional[str]:
         return None
 
-    @abstractmethod
-    def verify_restore_file_already_exist(self, restore_file_path: str) -> bool:
-        pass
+    @staticmethod
+    def verify_restore_file_already_exist(restore_file_path: str) -> bool:
+        return os.path.isfile(restore_file_path)
 
     @abstractmethod
     def is_project(self, document: Document) -> bool:

cycode/cli/files_collector/sca/go/restore_go_dependencies.py

@@ -44,8 +44,5 @@ def get_commands(self, manifest_file_path: str) -> list[list[str]]:
     def get_lock_file_name(self) -> str:
         return GO_RESTORE_FILE_NAME
 
-    def verify_restore_file_already_exist(self, restore_file_path: str) -> bool:
-        return os.path.isfile(restore_file_path)
-
     def get_working_directory(self, document: Document) -> Optional[str]:
         return os.path.dirname(document.absolute_path)

cycode/cli/files_collector/sca/maven/restore_gradle_dependencies.py

@@ -42,22 +42,19 @@ def get_commands(self, manifest_file_path: str) -> list[list[str]]:
     def get_lock_file_name(self) -> str:
         return BUILD_GRADLE_DEP_TREE_FILE_NAME
 
-    def verify_restore_file_already_exist(self, restore_file_path: str) -> bool:
-        return os.path.isfile(restore_file_path)
-
     def get_working_directory(self, document: Document) -> Optional[str]:
         return get_path_from_context(self.ctx) if self.is_gradle_sub_projects() else None
 
     def get_all_projects(self) -> set[str]:
-        projects_output = shell(
+        output = shell(
             command=BUILD_GRADLE_ALL_PROJECTS_COMMAND,
             timeout=BUILD_GRADLE_ALL_PROJECTS_TIMEOUT,
             working_directory=get_path_from_context(self.ctx),
         )
+        if not output:
+            return set()
 
-        projects = re.findall(ALL_PROJECTS_REGEX, projects_output)
-
-        return set(projects)
+        return set(re.findall(ALL_PROJECTS_REGEX, output))
 
     def get_commands_for_sub_projects(self, manifest_file_path: str) -> list[list[str]]:
         project_name = os.path.basename(os.path.dirname(manifest_file_path))

cycode/cli/files_collector/sca/maven/restore_maven_dependencies.py

@@ -1,4 +1,3 @@
-import os
 from os import path
 from typing import Optional
 
@@ -30,9 +29,6 @@ def get_commands(self, manifest_file_path: str) -> list[list[str]]:
     def get_lock_file_name(self) -> str:
         return join_paths('target', MAVEN_CYCLONE_DEP_TREE_FILE_NAME)
 
-    def verify_restore_file_already_exist(self, restore_file_path: str) -> bool:
-        return os.path.isfile(restore_file_path)
-
     def try_restore_dependencies(self, document: Document) -> Optional[Document]:
         restore_dependencies_document = super().try_restore_dependencies(document)
         manifest_file_path = self.get_manifest_file_path(document)
@@ -51,8 +47,8 @@ def restore_from_secondary_command(
         self, document: Document, manifest_file_path: str, restore_dependencies_document: Optional[Document]
     ) -> Optional[Document]:
         # TODO(MarshalX): does it even work? Ignored restore_dependencies_document arg
-        secondary_restore_command = create_secondary_restore_command(manifest_file_path)
-        backup_restore_content = execute_commands(secondary_restore_command, manifest_file_path, self.command_timeout)
+        secondary_restore_command = create_secondary_restore_commands(manifest_file_path)
+        backup_restore_content = execute_commands(secondary_restore_command, self.command_timeout)
         restore_dependencies_document = Document(
             build_dep_tree_path(document.path, MAVEN_DEP_TREE_FILE_NAME), backup_restore_content, self.is_git_diff
         )
@@ -64,13 +60,15 @@ def restore_from_secondary_command(
         return restore_dependencies
 
 
-def create_secondary_restore_command(manifest_file_path: str) -> list[str]:
+def create_secondary_restore_commands(manifest_file_path: str) -> list[list[str]]:
     return [
-        'mvn',
-        'dependency:tree',
-        '-B',
-        '-DoutputType=text',
-        '-f',
-        manifest_file_path,
-        f'-DoutputFile={MAVEN_DEP_TREE_FILE_NAME}',
+        [
+            'mvn',
+            'dependency:tree',
+            '-B',
+            '-DoutputType=text',
+            '-f',
+            manifest_file_path,
+            f'-DoutputFile={MAVEN_DEP_TREE_FILE_NAME}',
+        ]
     ]

cycode/cli/files_collector/sca/npm/restore_npm_dependencies.py

@@ -33,9 +33,6 @@ def get_commands(self, manifest_file_path: str) -> list[list[str]]:
     def get_lock_file_name(self) -> str:
         return NPM_LOCK_FILE_NAME
 
-    def verify_restore_file_already_exist(self, restore_file_path: str) -> bool:
-        return os.path.isfile(restore_file_path)
-
     @staticmethod
     def prepare_manifest_file_path_for_command(manifest_file_path: str) -> str:
         return manifest_file_path.replace(os.sep + NPM_MANIFEST_FILE_NAME, '')

cycode/cli/files_collector/sca/nuget/restore_nuget_dependencies.py

@@ -1,5 +1,3 @@
-import os
-
 import typer
 
 from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies
@@ -21,6 +19,3 @@ def get_commands(self, manifest_file_path: str) -> list[list[str]]:
 
     def get_lock_file_name(self) -> str:
         return NUGET_LOCK_FILE_NAME
-
-    def verify_restore_file_already_exist(self, restore_file_path: str) -> bool:
-        return os.path.isfile(restore_file_path)

cycode/cli/files_collector/sca/ruby/restore_ruby_dependencies.py

@@ -18,8 +18,5 @@ def get_commands(self, manifest_file_path: str) -> list[list[str]]:
     def get_lock_file_name(self) -> str:
         return RUBY_LOCK_FILE_NAME
 
-    def verify_restore_file_already_exist(self, restore_file_path: str) -> bool:
-        return os.path.isfile(restore_file_path)
-
     def get_working_directory(self, document: Document) -> Optional[str]:
         return os.path.dirname(document.absolute_path)

cycode/cli/files_collector/sca/sbt/restore_sbt_dependencies.py

@@ -18,8 +18,5 @@ def get_commands(self, manifest_file_path: str) -> list[list[str]]:
     def get_lock_file_name(self) -> str:
         return SBT_LOCK_FILE_NAME
 
-    def verify_restore_file_already_exist(self, restore_file_path: str) -> bool:
-        return os.path.isfile(restore_file_path)
-
     def get_working_directory(self, document: Document) -> Optional[str]:
         return os.path.dirname(document.absolute_path)

cycode/cli/utils/shell_executor.py

@@ -16,6 +16,7 @@ def shell(
     command: Union[str, list[str]],
     timeout: int = _SUBPROCESS_DEFAULT_TIMEOUT_SEC,
     working_directory: Optional[str] = None,
+    silent_exc_info: bool = False,
 ) -> Optional[str]:
     logger.debug('Executing shell command: %s', command)
 
@@ -27,12 +28,15 @@ def shell(
 
         return result.stdout.decode('UTF-8').strip()
     except subprocess.CalledProcessError as e:
-        logger.debug('Error occurred while running shell command', exc_info=e)
+        if not silent_exc_info:
+            logger.debug('Error occurred while running shell command', exc_info=e)
     except subprocess.TimeoutExpired as e:
         logger.debug('Command timed out', exc_info=e)
         raise typer.Abort(f'Command "{command}" timed out') from e
     except Exception as e:
-        logger.debug('Unhandled exception occurred while running shell command', exc_info=e)
+        if not silent_exc_info:
+            logger.debug('Unhandled exception occurred while running shell command', exc_info=e)
+
         raise click.ClickException(f'Unhandled exception: {e}') from e
 
     return None