CM-41217 - Implement SBT restore support for SCA

README.md

@@ -30,6 +30,8 @@ This guide will guide you through both installation and usage.
         6. [Commit History Scan](#commit-history-scan)
             1. [Commit Range Option](#commit-range-option)
         7. [Pre-Commit Scan](#pre-commit-scan)
+        8. [Lock Restore Options](#lock-restore-options)
+            1. [SBT Scan](#sbt-scan)
     2. [Scan Results](#scan-results)
         1. [Show/Hide Secrets](#showhide-secrets)
         2. [Soft Fail](#soft-fail)
@@ -496,6 +498,17 @@ After your install the pre-commit hook and, you may, on occasion, wish to skip s
 
 `SKIP=cycode git commit -m <your commit message>`
 
+### Lock Restore Options
+
+#### SBT Scan
+
+We use sbt-dependency-lock plugin to restore the lock file for SBT projects.  
+To disable lock restore in use `--no-restore` option.
+
+Prerequisites
+* sbt-dependency-lock Plugin: Install the plugin by adding the following line to `project/plugins.sbt`:
+`addSbtPlugin("software.purpledragon" % "sbt-dependency-lock" % "1.5.1")`
+
 ## Scan Results
 
 Each scan will complete with a message stating if any issues were found or not.

cycode/cli/commands/scan/repository/repository_command.py

@@ -48,8 +48,15 @@ def repository_command(context: click.Context, path: str, branch: str) -> None:
             # FIXME(MarshalX): probably file could be tree or submodule too. we expect blob only
             progress_bar.update(ScanProgressBarSection.PREPARE_LOCAL_FILES)
 
-            file_path = file.path if monitor else get_path_by_os(os.path.join(path, file.path))
-            documents_to_scan.append(Document(file_path, file.data_stream.read().decode('UTF-8', errors='replace')))
+            absolute_path = get_path_by_os(os.path.join(path, file.path))
+            file_path = file.path if monitor else absolute_path
+            documents_to_scan.append(
+                Document(
+                    file_path,
+                    file.data_stream.read().decode('UTF-8', errors='replace'),
+                    absolute_path=absolute_path,
+                )
+            )
 
         documents_to_scan = exclude_irrelevant_documents_to_scan(scan_type, documents_to_scan)
 

cycode/cli/files_collector/sca/base_restore_dependencies.py

@@ -14,10 +14,14 @@ def build_dep_tree_path(path: str, generated_file_name: str) -> str:
 
 
 def execute_command(
-    command: List[str], file_name: str, command_timeout: int, dependencies_file_name: Optional[str] = None
+    command: List[str],
+    file_name: str,
+    command_timeout: int,
+    dependencies_file_name: Optional[str] = None,
+    working_directory: Optional[str] = None,
 ) -> Optional[str]:
     try:
-        dependencies = shell(command=command, timeout=command_timeout)
+        dependencies = shell(command=command, timeout=command_timeout, working_directory=working_directory)
         # Write stdout output to the file if output_file_path is provided
         if dependencies_file_name:
             with open(dependencies_file_name, 'w') as output_file:
@@ -51,18 +55,26 @@ def get_manifest_file_path(self, document: Document) -> str:
     def try_restore_dependencies(self, document: Document) -> Optional[Document]:
         manifest_file_path = self.get_manifest_file_path(document)
         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_command(
-                self.get_command(manifest_file_path), manifest_file_path, self.command_timeout, output_file_path
+                self.get_command(manifest_file_path),
+                manifest_file_path,
+                self.command_timeout,
+                output_file_path,
+                working_directory_path,
             )
             restore_file_content = get_file_content(restore_file_path)
 
         return Document(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

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

@@ -0,0 +1,25 @@
+import os
+from typing import List, Optional
+
+from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies
+from cycode.cli.models import Document
+
+SBT_PROJECT_FILE_EXTENSIONS = ['sbt']
+SBT_LOCK_FILE_NAME = 'build.sbt.lock'
+
+
+class RestoreSbtDependencies(BaseRestoreDependencies):
+    def is_project(self, document: Document) -> bool:
+        return any(document.path.endswith(ext) for ext in SBT_PROJECT_FILE_EXTENSIONS)
+
+    def get_command(self, manifest_file_path: str) -> List[str]:
+        return ['sbt', 'dependencyLockWrite', '--verbose']
+
+    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/files_collector/sca/sca_code_scanner.py

@@ -7,8 +7,7 @@
 from cycode.cli.files_collector.sca.base_restore_dependencies import BaseRestoreDependencies
 from cycode.cli.files_collector.sca.maven.restore_gradle_dependencies import RestoreGradleDependencies
 from cycode.cli.files_collector.sca.maven.restore_maven_dependencies import RestoreMavenDependencies
-from cycode.cli.files_collector.sca.npm.restore_npm_dependencies import RestoreNpmDependencies
-from cycode.cli.files_collector.sca.nuget.restore_nuget_dependencies import RestoreNugetDependencies
+from cycode.cli.files_collector.sca.sbt.restore_sbt_dependencies import RestoreSbtDependencies
 from cycode.cli.models import Document
 from cycode.cli.utils.git_proxy import git_proxy
 from cycode.cli.utils.path_utils import get_file_content, get_file_dir, get_path_from_context, join_paths
@@ -17,9 +16,7 @@
 if TYPE_CHECKING:
     from git import Repo
 
-BUILD_GRADLE_DEP_TREE_TIMEOUT = 180
-BUILD_NUGET_DEP_TREE_TIMEOUT = 180
-BUILD_NPM_DEP_TREE_TIMEOUT = 180
+BUILD_DEP_TREE_TIMEOUT = 180
 
 
 def perform_pre_commit_range_scan_actions(
@@ -132,10 +129,9 @@ def add_dependencies_tree_document(
 
 def restore_handlers(context: click.Context, is_git_diff: bool) -> List[BaseRestoreDependencies]:
     return [
-        RestoreGradleDependencies(context, is_git_diff, BUILD_GRADLE_DEP_TREE_TIMEOUT),
-        RestoreMavenDependencies(context, is_git_diff, BUILD_GRADLE_DEP_TREE_TIMEOUT),
-        RestoreNugetDependencies(context, is_git_diff, BUILD_NUGET_DEP_TREE_TIMEOUT),
-        RestoreNpmDependencies(context, is_git_diff, BUILD_NPM_DEP_TREE_TIMEOUT),
+        RestoreGradleDependencies(context, is_git_diff, BUILD_DEP_TREE_TIMEOUT),
+        RestoreMavenDependencies(context, is_git_diff, BUILD_DEP_TREE_TIMEOUT),
+        RestoreSbtDependencies(context, is_git_diff, BUILD_DEP_TREE_TIMEOUT),
     ]
 
 

cycode/cli/models.py

@@ -7,12 +7,18 @@
 
 class Document:
     def __init__(
-        self, path: str, content: str, is_git_diff_format: bool = False, unique_id: Optional[str] = None
+        self,
+        path: str,
+        content: str,
+        is_git_diff_format: bool = False,
+        unique_id: Optional[str] = None,
+        absolute_path: Optional[str] = None,
     ) -> None:
         self.path = path
         self.content = content
         self.is_git_diff_format = is_git_diff_format
         self.unique_id = unique_id
+        self.absolute_path = absolute_path
 
     def __repr__(self) -> str:
         return 'path:{0}, content:{1}'.format(self.path, self.content)

cycode/cli/utils/shell_executor.py

@@ -8,12 +8,16 @@
 _SUBPROCESS_DEFAULT_TIMEOUT_SEC = 60
 
 
-def shell(command: Union[str, List[str]], timeout: int = _SUBPROCESS_DEFAULT_TIMEOUT_SEC) -> Optional[str]:
+def shell(
+    command: Union[str, List[str]],
+    timeout: int = _SUBPROCESS_DEFAULT_TIMEOUT_SEC,
+    working_directory: Optional[str] = None,
+) -> Optional[str]:
     logger.debug('Executing shell command: %s', command)
 
     try:
         result = subprocess.run(  # noqa: S603
-            command, timeout=timeout, check=True, capture_output=True
+            command, cwd=working_directory, timeout=timeout, check=True, capture_output=True
         )
 
         return result.stdout.decode('UTF-8').strip()