CM-20753 - flag for excluding detection in deleted lines (#70)

Author: MichalBor

cli/code_scanner.py

@@ -234,7 +234,7 @@ def scan_documents(context: click.Context, documents_to_scan: List[Document], is
     cycode_client = context.obj["client"]
     scan_type = context.obj["scan_type"]
     severity_threshold = context.obj["severity_threshold"]
-    scan_command_type = context.info_name
+    command_scan_type = context.info_name
     error_message = None
     all_detections_count = 0
     output_detections_count = 0
@@ -246,7 +246,7 @@ def scan_documents(context: click.Context, documents_to_scan: List[Document], is
         scan_result = perform_scan(cycode_client, zipped_documents, scan_type, scan_id, is_git_diff, is_commit_range,
                                    scan_parameters)
         all_detections_count, output_detections_count = \
-            handle_scan_result(context, scan_result, scan_command_type, scan_type, severity_threshold,
+            handle_scan_result(context, scan_result, command_scan_type, scan_type, severity_threshold,
                                documents_to_scan)
         scan_completed = True
     except Exception as e:
@@ -259,7 +259,7 @@ def scan_documents(context: click.Context, documents_to_scan: List[Document], is
                  {'all_violations_count': all_detections_count, 'relevant_violations_count': output_detections_count,
                   'scan_id': str(scan_id), 'zip_file_size': zip_file_size})
     _report_scan_status(context, scan_type, str(scan_id), scan_completed, output_detections_count,
-                        all_detections_count, len(documents_to_scan), zip_file_size, scan_command_type, error_message)
+                        all_detections_count, len(documents_to_scan), zip_file_size, command_scan_type, error_message)
 
 
 def scan_commit_range_documents(context: click.Context, from_documents_to_scan: List[Document],
@@ -309,10 +309,10 @@ def should_scan_documents(from_documents_to_scan: List[Document], to_documents_t
     return len(from_documents_to_scan) > 0 or len(to_documents_to_scan) > 0
 
 
-def handle_scan_result(context, scan_result, scan_command_type, scan_type, severity_threshold, to_documents_to_scan):
+def handle_scan_result(context, scan_result, command_scan_type, scan_type, severity_threshold, to_documents_to_scan):
     document_detections_list = enrich_scan_result(scan_result, to_documents_to_scan)
     relevant_document_detections_list = exclude_irrelevant_scan_results(document_detections_list, scan_type,
-                                                                        scan_command_type, severity_threshold)
+                                                                        command_scan_type, severity_threshold)
     context.obj['report_url'] = scan_result.report_url
     print_results(context, relevant_document_detections_list)
     context.obj['issue_detected'] = len(relevant_document_detections_list) > 0
@@ -424,10 +424,10 @@ def enrich_scan_result(scan_result: ZippedFileScanResult, documents_to_scan: Lis
 
 
 def exclude_irrelevant_scan_results(document_detections_list: List[DocumentDetections], scan_type: str,
-                                    scan_command_type: str, severity_threshold: str) -> List[DocumentDetections]:
+                                    command_scan_type: str, severity_threshold: str) -> List[DocumentDetections]:
     relevant_document_detections_list = []
     for document_detections in document_detections_list:
-        relevant_detections = exclude_irrelevant_detections(scan_type, scan_command_type, severity_threshold,
+        relevant_detections = exclude_irrelevant_detections(scan_type, command_scan_type, severity_threshold,
                                                             document_detections.detections)
         if relevant_detections:
             relevant_document_detections_list.append(DocumentDetections(document=document_detections.document,
@@ -540,9 +540,9 @@ def exclude_irrelevant_files(context: click.Context, filenames: List[str]) -> Li
     return [filename for filename in filenames if _is_relevant_file_to_scan(scan_type, filename)]
 
 
-def exclude_irrelevant_detections(scan_type: str, scan_command_type: str, severity_threshold: str, detections) -> List:
+def exclude_irrelevant_detections(scan_type: str, command_scan_type: str, severity_threshold: str, detections) -> List:
     relevant_detections = exclude_detections_by_exclusions_configuration(scan_type, detections)
-    relevant_detections = exclude_detections_by_scan_command_type(scan_command_type, relevant_detections)
+    relevant_detections = exclude_detections_by_scan_type(scan_type, command_scan_type, relevant_detections)
     relevant_detections = exclude_detections_by_severity(scan_type, severity_threshold, relevant_detections)
 
     return relevant_detections
@@ -557,14 +557,18 @@ def exclude_detections_by_severity(scan_type: str, severity_threshold: str, dete
                                                     severity_threshold)]
 
 
-def exclude_detections_by_scan_command_type(scan_command_type: str, detections) -> List:
-    if scan_command_type != PRE_COMMIT_SCAN_COMMAND_TYPE:
-        return detections
+def exclude_detections_by_scan_type(scan_type: str, command_scan_type: str, detections) -> List:
+    if command_scan_type == PRE_COMMIT_COMMAND_SCAN_TYPE:
+        return exclude_detections_in_deleted_lines(detections)
 
-    return exclude_detections_for_pre_commit_scan_command_type(detections)
+    if command_scan_type in COMMIT_RANGE_BASED_COMMAND_SCAN_TYPES \
+            and scan_type == SECRET_SCAN_TYPE\
+            and configuration_manager.get_should_exclude_detections_in_deleted_lines(command_scan_type):
+        return exclude_detections_in_deleted_lines(detections)
+    return detections
 
 
-def exclude_detections_for_pre_commit_scan_command_type(detections) -> List:
+def exclude_detections_in_deleted_lines(detections) -> List:
     return [detection for detection in detections if detection.detection_details.get('line_type') != 'Removed']
 
 
@@ -795,7 +799,7 @@ def _handle_exception(context: click.Context, e: Exception):
 
 def _report_scan_status(context: click.Context, scan_type: str, scan_id: str, scan_completed: bool,
                         output_detections_count: int, all_detections_count: int, files_to_scan_count: int,
-                        zip_size: int, scan_command_type: str, error_message: Optional[str]):
+                        zip_size: int, command_scan_type: str, error_message: Optional[str]):
     try:
         cycode_client = context.obj["client"]
         end_scan_time = time.time()
@@ -806,7 +810,7 @@ def _report_scan_status(context: click.Context, scan_type: str, scan_id: str, sc
             'all_detections_count': all_detections_count,
             'scannable_files_count': files_to_scan_count,
             'status': 'Completed' if scan_completed else 'Error',
-            'scan_command_type': scan_command_type,
+            'scan_command_type': command_scan_type,
             'operation_system': platform(),
             'error_message': error_message
         }

cli/consts.py

@@ -1,4 +1,6 @@
-PRE_COMMIT_SCAN_COMMAND_TYPE = 'pre_commit'
+PRE_COMMIT_COMMAND_SCAN_TYPE = 'pre_commit'
+PRE_RECEIVE_COMMAND_SCAN_TYPE = 'pre_receive'
+COMMIT_HISTORY_COMMAND_SCAN_TYPE = 'commit_history'
 
 SECRET_SCAN_TYPE = 'secret'
 INFRA_CONFIGURATION_SCAN_TYPE = 'iac'
@@ -46,6 +48,8 @@
 
 COMMIT_RANGE_SCAN_SUPPORTED_SCAN_TYPES = [SECRET_SCAN_TYPE, SCA_SCAN_TYPE]
 
+COMMIT_RANGE_BASED_COMMAND_SCAN_TYPES = [PRE_RECEIVE_COMMAND_SCAN_TYPE, COMMIT_HISTORY_COMMAND_SCAN_TYPE]
+
 DEFAULT_CYCODE_API_URL = "https://api.cycode.com"
 DEFAULT_CYCODE_APP_URL = "https://app.cycode.com"
 
@@ -98,6 +102,9 @@
 Learn how to: https://cycode.com/dont-let-hardcoded-secrets-compromise-your-security-4-effective-remediation-techniques 
 """
 
+EXCLUDE_DETECTIONS_IN_DELETED_LINES_ENV_VAR_NAME = 'EXCLUDE_DETECTIONS_IN_DELETED_LINES'
+DEFAULT_EXCLUDE_DETECTIONS_IN_DELETED_LINES = True
+
 # scan statuses
 SCAN_STATUS_COMPLETED = 'Completed'
 SCAN_STATUS_ERROR = 'Error'

cli/printers/text_printer.py

@@ -4,7 +4,7 @@
 from cli.printers.base_printer import BasePrinter
 from cli.models import DocumentDetections, Detection, Document
 from cli.config import config
-from cli.consts import SECRET_SCAN_TYPE
+from cli.consts import SECRET_SCAN_TYPE, COMMIT_RANGE_BASED_COMMAND_SCAN_TYPES
 from cli.utils.string_utils import obfuscate_text
 
 
@@ -57,7 +57,7 @@ def _print_detection_summary(self, detection: Detection, document_path: str):
             f'{detection_sha_message}{detection_commit_id_message}  ⛔ ')
 
     def _print_detection_code_segment(self, detection: Detection, document: Document, code_segment_size: int):
-        if document.is_git_diff_format:
+        if self._is_git_diff_based_scan():
             self._print_detection_from_git_diff(detection, document)
             return
 
@@ -169,3 +169,6 @@ def _print_detection_from_git_diff(self, detection: Detection, document: Documen
         self._print_detection_line(document, detection_line, detection_line_number_in_original_file,
                                    detection_position_in_line, violation_length)
         click.echo()
+
+    def _is_git_diff_based_scan(self):
+        return self.command_scan_type in COMMIT_RANGE_BASED_COMMAND_SCAN_TYPES and self.scan_type == SECRET_SCAN_TYPE

cli/user_settings/config_file_manager.py

@@ -18,7 +18,7 @@ class ConfigFileManager(BaseFileManager):
 
     MAX_COMMITS_FIELD_NAME: str = 'max_commits'
     COMMAND_TIMEOUT_FIELD_NAME: str = 'command_timeout'
-
+    EXCLUDE_DETECTIONS_IN_DELETED_LINES: str = 'exclude_detections_in_deleted_lines'
 
     def __init__(self, path):
         self.path = path
@@ -43,6 +43,10 @@ def get_max_commits(self, command_scan_type) -> Optional[int]:
     def get_command_timeout(self, command_scan_type) -> Optional[int]:
         return self._get_value_from_command_scan_type_configuration(command_scan_type, self.COMMAND_TIMEOUT_FIELD_NAME)
 
+    def get_exclude_detections_in_deleted_lines(self, command_scan_type) -> Optional[bool]:
+        return self._get_value_from_command_scan_type_configuration(command_scan_type,
+                                                                    self.EXCLUDE_DETECTIONS_IN_DELETED_LINES)
+
     def update_base_url(self, base_url: str):
         update_data = {
             self.ENVIRONMENT_SECTION_NAME: {

cli/user_settings/configuration_manager.py

@@ -117,6 +117,24 @@ def get_pre_receive_command_timeout(self, command_scan_type: str) -> int:
 
         return DEFAULT_PRE_RECEIVE_COMMAND_TIMEOUT_IN_SECONDS
 
+    def get_should_exclude_detections_in_deleted_lines(self, command_scan_type: str) -> bool:
+        exclude_detections_in_deleted_lines = self._get_value_from_environment_variables(
+            EXCLUDE_DETECTIONS_IN_DELETED_LINES_ENV_VAR_NAME)
+        if exclude_detections_in_deleted_lines is not None:
+            return exclude_detections_in_deleted_lines.lower() in ('true', '1')
+
+        exclude_detections_in_deleted_lines = self.local_config_file_manager \
+            .get_exclude_detections_in_deleted_lines(command_scan_type)
+        if exclude_detections_in_deleted_lines is not None:
+            return exclude_detections_in_deleted_lines
+
+        exclude_detections_in_deleted_lines = self.global_config_file_manager\
+            .get_exclude_detections_in_deleted_lines(command_scan_type)
+        if exclude_detections_in_deleted_lines is not None:
+            return exclude_detections_in_deleted_lines
+
+        return DEFAULT_EXCLUDE_DETECTIONS_IN_DELETED_LINES
+
     @staticmethod
     def _get_value_from_environment_variables(env_var_name, default=None):
         return os.getenv(env_var_name, default)