CM-24116 - Migrate SCA tables to the new table management system

Author: MarshalX

cycode/cli/printers/console_printer.py

@@ -5,23 +5,23 @@
 from cycode.cli.exceptions.custom_exceptions import CycodeError
 from cycode.cli.models import CliError, CliResult
 from cycode.cli.printers.json_printer import JsonPrinter
-from cycode.cli.printers.sca_table_printer import SCATablePrinter
-from cycode.cli.printers.table_printer import TablePrinter
+from cycode.cli.printers.tables.sca_table_printer import ScaTablePrinter
+from cycode.cli.printers.tables.table_printer import TablePrinter
 from cycode.cli.printers.text_printer import TextPrinter
 
 if TYPE_CHECKING:
     from cycode.cli.models import LocalScanResult
-    from cycode.cli.printers.base_printer import BasePrinter
+    from cycode.cli.printers.tables.table_printer_base import PrinterBase
 
 
 class ConsolePrinter:
-    _AVAILABLE_PRINTERS: ClassVar[Dict[str, 'BasePrinter']] = {
+    _AVAILABLE_PRINTERS: ClassVar[Dict[str, 'PrinterBase']] = {
         'text': TextPrinter,
         'json': JsonPrinter,
         'table': TablePrinter,
         # overrides
-        'table_sca': SCATablePrinter,
-        'text_sca': SCATablePrinter,
+        'table_sca': ScaTablePrinter,
+        'text_sca': ScaTablePrinter,
     }
 
     def __init__(self, context: click.Context) -> None:
@@ -37,7 +37,7 @@ def print_scan_results(self, local_scan_results: List['LocalScanResult']) -> Non
         printer = self._get_scan_printer()
         printer.print_scan_results(local_scan_results)
 
-    def _get_scan_printer(self) -> 'BasePrinter':
+    def _get_scan_printer(self) -> 'PrinterBase':
         printer_class = self._printer_class
 
         composite_printer = self._AVAILABLE_PRINTERS.get(f'{self.output_type}_{self.scan_type}')

cycode/cli/printers/json_printer.py

@@ -4,14 +4,14 @@
 import click
 
 from cycode.cli.models import CliError, CliResult
-from cycode.cli.printers.base_printer import BasePrinter
+from cycode.cli.printers.printer_base import PrinterBase
 from cycode.cyclient.models import DetectionSchema
 
 if TYPE_CHECKING:
     from cycode.cli.models import LocalScanResult
 
 
-class JsonPrinter(BasePrinter):
+class JsonPrinter(PrinterBase):
     def print_result(self, result: CliResult) -> None:
         result = {'result': result.success, 'message': result.message}
 

cycode/cli/printers/base_printer.py renamed to cycode/cli/printers/printer_base.py

@@ -9,7 +9,7 @@
     from cycode.cli.models import LocalScanResult
 
 
-class BasePrinter(ABC):
+class PrinterBase(ABC):
     RED_COLOR_NAME = 'red'
     WHITE_COLOR_NAME = 'white'
     GREEN_COLOR_NAME = 'green'

cycode/cli/printers/sca_table_printer.py

@@ -0,0 +1,140 @@
+from collections import defaultdict
+from typing import TYPE_CHECKING, Dict, List
+
+import click
+
+from cycode.cli.consts import LICENSE_COMPLIANCE_POLICY_ID, PACKAGE_VULNERABILITY_POLICY_ID
+from cycode.cli.models import Detection
+from cycode.cli.printers.tables.table import Table
+from cycode.cli.printers.tables.table_models import ColumnInfoBuilder, ColumnWidths
+from cycode.cli.printers.tables.table_printer_base import TablePrinterBase
+from cycode.cli.utils.string_utils import shortcut_dependency_paths
+
+if TYPE_CHECKING:
+    from cycode.cli.models import LocalScanResult
+
+
+column_builder = ColumnInfoBuilder()
+
+# Building must have strict order. Represents the order of the columns in the table (from left to right)
+SEVERITY_COLUMN = column_builder.build(name='Severity')
+REPOSITORY_COLUMN = column_builder.build(name='Repository')
+
+FILE_PATH_COLUMN = column_builder.build(name='File Path')
+ECOSYSTEM_COLUMN = column_builder.build(name='Ecosystem')
+DEPENDENCY_NAME_COLUMN = column_builder.build(name='Dependency Name')
+DIRECT_DEPENDENCY_COLUMN = column_builder.build(name='Direct Dependency')
+DEVELOPMENT_DEPENDENCY_COLUMN = column_builder.build(name='Development Dependency')
+DEPENDENCY_PATHS_COLUMN = column_builder.build(name='Dependency Paths')
+
+CVE_COLUMNS = column_builder.build(name='CVE')
+UPGRADE_COLUMN = column_builder.build(name='Upgrade')
+LICENSE_COLUMN = column_builder.build(name='License')
+
+COLUMN_WIDTHS_CONFIG: ColumnWidths = {
+    REPOSITORY_COLUMN: 2,
+    FILE_PATH_COLUMN: 3,
+    CVE_COLUMNS: 5,
+    UPGRADE_COLUMN: 3,
+    LICENSE_COLUMN: 2,
+}
+
+
+class ScaTablePrinter(TablePrinterBase):
+    def _print_results(self, local_scan_results: List['LocalScanResult']) -> None:
+        detections_per_detection_type_id = self._extract_detections_per_policy_id(local_scan_results)
+        for policy_id, detections in detections_per_detection_type_id.items():
+            table = self._get_table(policy_id)
+            table.set_cols_width(COLUMN_WIDTHS_CONFIG)
+
+            for detection in detections:
+                self._enrich_table_with_values(table, detection)
+
+            self._print_summary_issues(len(detections), self._get_title(policy_id))
+            click.echo(table.get_table().draw())
+
+        self._print_report_urls(local_scan_results)
+
+    @staticmethod
+    def _get_title(policy_id: str) -> str:
+        if policy_id == PACKAGE_VULNERABILITY_POLICY_ID:
+            return 'Dependencies Vulnerabilities'
+        if policy_id == LICENSE_COMPLIANCE_POLICY_ID:
+            return 'License Compliance'
+
+        return 'Unknown'
+
+    def _get_table(self, policy_id: str) -> Table:
+        table = Table()
+
+        if policy_id == PACKAGE_VULNERABILITY_POLICY_ID:
+            table.add(SEVERITY_COLUMN)
+            table.add(CVE_COLUMNS)
+            table.add(UPGRADE_COLUMN)
+        elif policy_id == LICENSE_COMPLIANCE_POLICY_ID:
+            table.add(LICENSE_COLUMN)
+
+        if self._is_git_repository():
+            table.add(REPOSITORY_COLUMN)
+
+        table.add(FILE_PATH_COLUMN)
+        table.add(ECOSYSTEM_COLUMN)
+        table.add(DEPENDENCY_NAME_COLUMN)
+        table.add(DIRECT_DEPENDENCY_COLUMN)
+        table.add(DEVELOPMENT_DEPENDENCY_COLUMN)
+        table.add(DEPENDENCY_PATHS_COLUMN)
+
+        return table
+
+    @staticmethod
+    def _enrich_table_with_values(table: Table, detection: Detection) -> None:
+        detection_details = detection.detection_details
+
+        table.set(REPOSITORY_COLUMN, detection_details.get('repository_name'))
+
+        table.set(FILE_PATH_COLUMN, detection_details.get('file_name'))
+        table.set(ECOSYSTEM_COLUMN, detection_details.get('ecosystem'))
+        table.set(DEPENDENCY_NAME_COLUMN, detection_details.get('package_name'))
+        table.set(DIRECT_DEPENDENCY_COLUMN, detection_details.get('is_direct_dependency_str'))
+        table.set(DEVELOPMENT_DEPENDENCY_COLUMN, detection_details.get('is_dev_dependency_str'))
+
+        dependency_paths = 'N/A'
+        dependency_paths_raw = detection_details.get('dependency_paths')
+        if dependency_paths_raw:
+            dependency_paths = shortcut_dependency_paths(dependency_paths_raw)
+        table.set(DEPENDENCY_PATHS_COLUMN, dependency_paths)
+
+        upgrade = ''
+        alert = detection_details.get('alert')
+        if alert and alert.get('first_patched_version'):
+            upgrade = f'{alert.get("vulnerable_requirements")} -> {alert.get("first_patched_version")}'
+        table.set(UPGRADE_COLUMN, upgrade)
+
+        table.set(SEVERITY_COLUMN, detection_details.get('advisory_severity'))
+        table.set(CVE_COLUMNS, detection_details.get('vulnerability_id'))
+        table.set(LICENSE_COLUMN, detection_details.get('license'))
+
+    @staticmethod
+    def _print_report_urls(local_scan_results: List['LocalScanResult']) -> None:
+        click.echo('Report URLs:')
+        for local_scan_result in local_scan_results:
+            report_url = local_scan_result.report_url if local_scan_result.report_url else 'N/A'
+            click.echo(f'- {report_url}')
+
+    @staticmethod
+    def _print_summary_issues(detections_count: int, title: str) -> None:
+        click.echo(f'⛔ Found {detections_count} issues of type: {click.style(title, bold=True)}')
+
+    @staticmethod
+    def _extract_detections_per_policy_id(
+        local_scan_results: List['LocalScanResult'],
+    ) -> Dict[str, List[Detection]]:
+        detections_to_policy_id = defaultdict(list)
+
+        for local_scan_result in local_scan_results:
+            for document_detection in local_scan_result.document_detections:
+                for detection in document_detection.detections:
+                    detections_to_policy_id[detection.detection_type_id].append(detection)
+
+        # sort dict by keys (policy id) to make persist output order
+        return dict(sorted(detections_to_policy_id.items(), reverse=True))

cycode/cli/printers/tables/__init__.py

@@ -3,7 +3,7 @@
 from texttable import Texttable
 
 if TYPE_CHECKING:
-    from cycode.cli.printers.table_models import ColumnInfo, ColumnWidths
+    from cycode.cli.printers.tables.table_models import ColumnInfo, ColumnWidths
 
 
 class Table:

cycode/cli/printers/tables/sca_table_printer.py

@@ -2,12 +2,12 @@
 
 
 class ColumnInfoBuilder:
-    _index = 0
+    def __init__(self) -> None:
+        self._index = 0
 
-    @staticmethod
-    def build(name: str) -> 'ColumnInfo':
-        column_info = ColumnInfo(name, ColumnInfoBuilder._index)
-        ColumnInfoBuilder._index += 1
+    def build(self, name: str) -> 'ColumnInfo':
+        column_info = ColumnInfo(name, self._index)
+        self._index += 1
         return column_info