CM-23720 - Add table output for all scan types

README.md

@@ -211,12 +211,12 @@ repos:
 
 The following are the options and commands available with the Cycode CLI application:
 
-| Option                  | Description                                               |
-|-------------------------|-----------------------------------------------------------|
-| `--output [text\|json]` | Specify the output (`text`/`json`). The default is `text` |
-| `-v`, `--verbose`       | Show detailed logs                                        |
-| `--version`             | Show the version and exit.                                |
-| `--help`                | Show options for given command.                           |
+| Option                         | Description                                                       |
+|--------------------------------|-------------------------------------------------------------------|
+| `--output [text\|json\|table]` | Specify the output (`text`/`json`/`table`). The default is `text` |
+| `-v`, `--verbose`              | Show detailed logs                                                |
+| `--version`                    | Show the version and exit.                                        |
+| `--help`                       | Show options for given command.                                   |
 
 | Command                             | Description |
 |-------------------------------------|-------------|

cycode/cli/main.py

@@ -65,10 +65,10 @@
 @click.option('--output', default=None,
               help="""
               \b
-              Specify the results output (text/json), 
+              Specify the results output (text/json/table), 
               the default is text
               """,
-              type=click.Choice(['text', 'json']))
+              type=click.Choice(['text', 'json', 'table']))
 @click.option('--severity-threshold',
               default=None,
               help='Show only violations at the specified level or higher (supported for SCA scan type only).',
@@ -142,8 +142,8 @@ def finalize(context: click.Context, *args, **kwargs):
 @click.option(
     '--output',
     default='text',
-    help='Specify the output (text/json), the default is text',
-    type=click.Choice(['text', 'json'])
+    help='Specify the output (text/json/table), the default is text',
+    type=click.Choice(['text', 'json', 'table'])
 )
 @click.option(
     '--user-agent',

cycode/cli/printers/base_printer.py

@@ -7,7 +7,9 @@
 
 
 class BasePrinter(ABC):
-    context: click.Context
+    RED_COLOR_NAME = 'red'
+    WHITE_COLOR_NAME = 'white'
+    GREEN_COLOR_NAME = 'green'
 
     def __init__(self, context: click.Context):
         self.context = context

cycode/cli/printers/console_printer.py

@@ -1,10 +1,10 @@
 import click
 from typing import List, TYPE_CHECKING
 
-from cycode.cli.consts import SCA_SCAN_TYPE
 from cycode.cli.exceptions.custom_exceptions import CycodeError
 from cycode.cli.models import DocumentDetections, CliResult, CliError
 from cycode.cli.printers.table_printer import TablePrinter
+from cycode.cli.printers.sca_table_printer import SCATablePrinter
 from cycode.cli.printers.json_printer import JsonPrinter
 from cycode.cli.printers.text_printer import TextPrinter
 
@@ -16,11 +16,15 @@ class ConsolePrinter:
     _AVAILABLE_PRINTERS = {
         'text': TextPrinter,
         'json': JsonPrinter,
-        'text_sca': TablePrinter
+        'table': TablePrinter,
+        # overrides
+        'table_sca': SCATablePrinter,
+        'text_sca': SCATablePrinter,
     }
 
     def __init__(self, context: click.Context):
         self.context = context
+        self.scan_type = self.context.obj.get('scan_type')
         self.output_type = self.context.obj.get('output')
 
         self._printer_class = self._AVAILABLE_PRINTERS.get(self.output_type)
@@ -32,11 +36,11 @@ def print_scan_results(self, detections_results_list: List[DocumentDetections])
         printer.print_scan_results(detections_results_list)
 
     def _get_scan_printer(self) -> 'BasePrinter':
-        scan_type = self.context.obj.get('scan_type')
-
         printer_class = self._printer_class
-        if scan_type == SCA_SCAN_TYPE and self.output_type == 'text':
-            printer_class = TablePrinter
+
+        composite_printer = self._AVAILABLE_PRINTERS.get(f'{self.output_type}_{self.scan_type}')
+        if composite_printer:
+            printer_class = composite_printer
 
         return printer_class(self.context)
 

cycode/cli/printers/sca_table_printer.py

@@ -0,0 +1,166 @@
+from collections import defaultdict
+from typing import List, Dict
+
+import click
+from texttable import Texttable
+
+from cycode.cli.consts import LICENSE_COMPLIANCE_POLICY_ID, PACKAGE_VULNERABILITY_POLICY_ID
+from cycode.cli.models import DocumentDetections, Detection, CliError, CliResult
+from cycode.cli.printers.base_printer import BasePrinter
+
+SEVERITY_COLUMN = 'Severity'
+LICENSE_COLUMN = 'License'
+UPGRADE_COLUMN = 'Upgrade'
+REPOSITORY_COLUMN = 'Repository'
+CVE_COLUMN = 'CVE'
+
+PREVIEW_DETECTIONS_COMMON_HEADERS = [
+    'File Path',
+    'Ecosystem',
+    'Dependency Name',
+    'Direct Dependency',
+    'Development Dependency'
+]
+
+
+class SCATablePrinter(BasePrinter):
+    def __init__(self, context: click.Context):
+        super().__init__(context)
+        self.scan_id = context.obj.get('scan_id')
+
+    def print_result(self, result: CliResult) -> None:
+        raise NotImplemented
+
+    def print_error(self, error: CliError) -> None:
+        raise NotImplemented
+
+    def print_scan_results(self, results: List[DocumentDetections]):
+        click.secho(f"Scan Results: (scan_id: {self.scan_id})")
+
+        if not results:
+            click.secho("Good job! No issues were found!!! 👏👏👏", fg=self.GREEN_COLOR_NAME)
+            return
+
+        detections_per_detection_type_id = self._extract_detections_per_detection_type_id(results)
+
+        self._print_detection_per_detection_type_id(detections_per_detection_type_id)
+
+        report_url = self.context.obj.get('report_url')
+        if report_url:
+            click.secho(f'Report URL: {report_url}')
+
+    @staticmethod
+    def _extract_detections_per_detection_type_id(results: List[DocumentDetections]) -> Dict[str, List[Detection]]:
+        detections_per_detection_type_id = defaultdict(list)
+
+        for document_detection in results:
+            for detection in document_detection.detections:
+                detections_per_detection_type_id[detection.detection_type_id].append(detection)
+
+        return detections_per_detection_type_id
+
+    def _print_detection_per_detection_type_id(
+            self, detections_per_detection_type_id: Dict[str, List[Detection]]
+    ) -> None:
+        for detection_type_id in detections_per_detection_type_id:
+            detections = detections_per_detection_type_id[detection_type_id]
+            headers = self._get_table_headers()
+
+            title = None
+            rows = []
+
+            if detection_type_id == PACKAGE_VULNERABILITY_POLICY_ID:
+                title = "Dependencies Vulnerabilities"
+
+                headers = [SEVERITY_COLUMN] + headers
+                headers.extend(PREVIEW_DETECTIONS_COMMON_HEADERS)
+                headers.append(CVE_COLUMN)
+                headers.append(UPGRADE_COLUMN)
+
+                for detection in detections:
+                    rows.append(self._get_upgrade_package_vulnerability(detection))
+            elif detection_type_id == LICENSE_COMPLIANCE_POLICY_ID:
+                title = "License Compliance"
+
+                headers.extend(PREVIEW_DETECTIONS_COMMON_HEADERS)
+                headers.append(LICENSE_COLUMN)
+
+                for detection in detections:
+                    rows.append(self._get_license(detection))
+
+            if rows:
+                self._print_table_detections(detections, headers, rows, title)
+
+    def _get_table_headers(self) -> list:
+        if self._is_git_repository():
+            return [REPOSITORY_COLUMN]
+
+        return []
+
+    def _print_table_detections(
+            self, detections: List[Detection], headers: List[str], rows, title: str
+    ) -> None:
+        self._print_summary_issues(detections, title)
+        text_table = Texttable()
+        text_table.header(headers)
+
+        self.set_table_width(headers, text_table)
+
+        for row in rows:
+            text_table.add_row(row)
+
+        click.echo(text_table.draw())
+
+    @staticmethod
+    def set_table_width(headers: List[str], text_table: Texttable) -> None:
+        header_width_size_cols = []
+        for header in headers:
+            header_len = len(header)
+            if header == CVE_COLUMN:
+                header_width_size_cols.append(header_len * 5)
+            elif header == UPGRADE_COLUMN:
+                header_width_size_cols.append(header_len * 2)
+            else:
+                header_width_size_cols.append(header_len)
+        text_table.set_cols_width(header_width_size_cols)
+
+    @staticmethod
+    def _print_summary_issues(detections: List, title: str) -> None:
+        click.echo(f'⛔ Found {len(detections)} issues of type: {click.style(title, bold=True)}')
+
+    def _get_common_detection_fields(self, detection: Detection) -> List[str]:
+        row = [
+            detection.detection_details.get('file_name'),
+            detection.detection_details.get('ecosystem'),
+            detection.detection_details.get('package_name'),
+            detection.detection_details.get('is_direct_dependency_str'),
+            detection.detection_details.get('is_dev_dependency_str')
+        ]
+
+        if self._is_git_repository():
+            row = [detection.detection_details.get('repository_name')] + row
+
+        return row
+
+    def _is_git_repository(self) -> bool:
+        return self.context.obj.get("remote_url") is not None
+
+    def _get_upgrade_package_vulnerability(self, detection: Detection) -> List[str]:
+        alert = detection.detection_details.get('alert')
+        row = [
+            detection.detection_details.get('advisory_severity'),
+            *self._get_common_detection_fields(detection),
+            detection.detection_details.get('vulnerability_id')
+        ]
+
+        upgrade = ''
+        if alert.get("first_patched_version"):
+            upgrade = f'{alert.get("vulnerable_requirements")} -> {alert.get("first_patched_version")}'
+        row.append(upgrade)
+
+        return row
+
+    def _get_license(self, detection: Detection) -> List[str]:
+        row = self._get_common_detection_fields(detection)
+        row.append(f'{detection.detection_details.get("license")}')
+        return row