refactoring to use Dependency and DependencyVersion types

Author: sobregosodd

guarddog/cli.py

@@ -5,7 +5,6 @@
 """
 
 from functools import reduce
-import json as js
 import logging
 import os
 import sys
@@ -14,12 +13,12 @@
 
 import click
 from prettytable import PrettyTable
-from termcolor import colored
 
 from guarddog.analyzer.metadata import get_metadata_detectors
 from guarddog.analyzer.sourcecode import get_sourcecode_rules
 from guarddog.ecosystems import ECOSYSTEM
-from guarddog.reporters.sarif import report_verify_sarif
+from guarddog.reporters.reporter_factory import ReporterFactory, ReporterType
+
 from guarddog.scanners import get_package_scanner, get_project_scanner
 from guarddog.utils.archives import safe_extract
 
@@ -127,7 +126,7 @@ def _get_all_rules(ecosystem: ECOSYSTEM) -> set[str]:
 
 def _get_rule_param(
     rules: tuple[str, ...], exclude_rules: tuple[str, ...], ecosystem: ECOSYSTEM
-) -> Optional[set]:
+) -> Optional[set[str]]:
     """
     This function should return None if no rules are provided
     Else a set of rules to be used for scanning
@@ -162,28 +161,20 @@ def _verify(
         log.error(f"Command verify is not supported for ecosystem {ecosystem}")
         exit(1)
 
-    def display_result(result: dict) -> None:
-        identifier = (
-            result["dependency"]
-            if result["version"] is None
-            else f"{result['dependency']} version {result['version']}"
-        )
-        if output_format is None:
-            print_scan_results(result.get("result"), identifier)
-
-        if len(result.get("errors", [])) > 0:
-            print_errors(result.get("error"), identifier)
+    dependencies, results = scanner.scan_local(path=path,rules=rule_param )
 
-    results = scanner.scan_local(path, rule_param, display_result)
-    if output_format == "json":
-        return_value = js.dumps(results)
+    rule_docs = list(rule_param or _get_all_rules(ecosystem=ecosystem))
 
-    if output_format == "sarif":
-        sarif_rules = _get_all_rules(ecosystem)
-        return_value = report_verify_sarif(path, list(sarif_rules), results, ecosystem)
+    reporter = ReporterFactory.create_reporter(ReporterType.from_str(output_format))
+    stdout, stderr = reporter.render_verify(
+        dependency_files=dependencies,
+        rule_names=rule_docs,
+        scan_results=results,
+        ecosystem=ecosystem,
+    )
 
-    if output_format is not None:
-        print(return_value)
+    sys.stdout.write(stdout)
+    sys.stderr.write(stderr)
 
     if exit_non_zero_on_finding:
         exit_with_status_code([result["result"] for result in results])
@@ -231,10 +222,10 @@ def _scan(
         log.error(f"Error occurred while scanning target {identifier}: '{e}'\n")
         sys.exit(1)
 
-    if output_format == "json":
-        print(js.dumps(result))
-    else:
-        print_scan_results(result, result["package"])
+    reporter = ReporterFactory.create_reporter(ReporterType.from_str(output_format))
+    stdout, stderr = reporter.render_scan(result)
+    sys.stdout.write(stdout)
+    sys.stderr.write(stderr)
 
     if exit_non_zero_on_finding:
         exit_with_status_code([result])
@@ -359,83 +350,6 @@ def scan(
         exit_non_zero_on_finding,
         ECOSYSTEM.PYPI,
     )
-
-
-# Pretty prints scan results for the console
-def print_scan_results(results, identifier):
-    num_issues = results.get("issues")
-    errors = results.get("errors", [])
-
-    if num_issues == 0:
-        print(
-            "Found "
-            + colored("0 potentially malicious indicators", "green", attrs=["bold"])
-            + " scanning "
-            + colored(identifier, None, attrs=["bold"])
-        )
-        print()
-    else:
-        print(
-            "Found "
-            + colored(
-                str(num_issues) + " potentially malicious indicators",
-                "red",
-                attrs=["bold"],
-            )
-            + " in "
-            + colored(identifier, None, attrs=["bold"])
-        )
-        print()
-
-        findings = results.get("results", [])
-        for finding in findings:
-            description = findings[finding]
-            if isinstance(description, str):  # package metadata
-                print(colored(finding, None, attrs=["bold"]) + ": " + description)
-                print()
-            elif isinstance(description, list):  # semgrep rule result:
-                source_code_findings = description
-                print(
-                    colored(finding, None, attrs=["bold"])
-                    + ": found "
-                    + str(len(source_code_findings))
-                    + " source code matches"
-                )
-                for finding in source_code_findings:
-                    print(
-                        "  * "
-                        + finding["message"]
-                        + " at "
-                        + finding["location"]
-                        + "\n    "
-                        + format_code_line_for_output(finding["code"])
-                    )
-                print()
-
-    if len(errors) > 0:
-        print_errors(errors, identifier)
-        print("\n")
-
-
-def print_errors(errors, identifier):
-    print(
-        colored("Some rules failed to run while scanning " + identifier + ":", "yellow")
-    )
-    print()
-    for rule in errors:
-        print(f"* {rule}: {errors[rule]}")
-    print()
-
-
-def format_code_line_for_output(code):
-    return "    " + colored(
-        code.strip().replace("\n", "\n    ").replace("\t", "  "),
-        None,
-        "on_red",
-        attrs=["bold"],
-    )
-
-
 # Given the results, exit with the appropriate status code
 def exit_with_status_code(results):
     for result in results:

guarddog/reporters/__init__.py

@@ -0,0 +1,28 @@
+from guarddog.scanners.scanner import DependencyFile
+from typing import List
+from guarddog.ecosystems import ECOSYSTEM
+
+
+class BaseReporter:
+    """
+    Base class for all reporters.
+    """
+
+    @staticmethod
+    def render_scan(scan_results: dict) -> tuple[str, str]:
+        """
+        Report the scans results.
+        """
+        raise NotImplementedError("Subclasses must implement this method.")
+
+    @staticmethod
+    def render_verify(
+        dependency_files: List[DependencyFile],
+        rule_names: list[str],
+        scan_results: list[dict],
+        ecosystem: ECOSYSTEM,
+    ) -> tuple[str, str]:
+        """
+        Report the scans results.
+        """
+        raise NotImplementedError("Subclasses must implement this method.")

guarddog/reporters/human_readable.py

@@ -0,0 +1,138 @@
+from termcolor import colored
+from typing import Generator
+from guarddog.reporters import BaseReporter
+from typing import List
+from guarddog.scanners.scanner import DependencyFile
+from guarddog.ecosystems import ECOSYSTEM
+
+
+class HumanReadableReporter(BaseReporter):
+    """
+    HumanReadableReporter is a class that formats and prints scan results in a human-readable format.
+    """
+
+    @staticmethod
+    def print_errors(identifier: str, results: dict) -> Generator:
+        errors = results.get("errors", [])
+        if not errors:
+            return
+        yield ("")
+        yield (
+            colored(
+                "Some rules failed to run while scanning " + identifier + ":",
+                "yellow",
+            )
+        )
+        yield ("")
+        for rule in errors:
+            yield (f"* {rule}: {errors[rule]}")
+
+    @staticmethod
+    def print_scan_results(identifier:str, results:dict) -> Generator:
+
+        def _format_code_line_for_output(code) -> str:
+            return "    " + colored(
+                code.strip().replace("\n", "\n    ").replace("\t", "  "),
+                None,
+                "on_red",
+                attrs=["bold"],
+            )
+        num_issues = results.get("issues")
+
+        if num_issues == 0:
+            yield (
+                "Found "
+                + colored(
+                    "0 potentially malicious indicators", "green", attrs=["bold"]
+                )
+                + " scanning "
+                + colored(identifier, None, attrs=["bold"])
+            )
+            yield ("")
+        else:
+            yield (
+                "Found "
+                + colored(
+                    str(num_issues) + " potentially malicious indicators",
+                    "red",
+                    attrs=["bold"],
+                )
+                + " in "
+                + colored(identifier, None, attrs=["bold"])
+            )
+            yield ("")
+
+            findings = results.get("results", [])
+            for finding in findings:
+                description = findings[finding]
+                if isinstance(description, str):  # package metadata
+                    yield (
+                        colored(finding, None, attrs=["bold"]) + ": " + description
+                    )
+                    yield ("")
+                elif isinstance(description, list):  # semgrep rule result:
+                    source_code_findings = description
+                    yield (
+                        colored(finding, None, attrs=["bold"])
+                        + ": found "
+                        + str(len(source_code_findings))
+                        + " source code matches"
+                    )
+                    for finding in source_code_findings:
+                        yield (
+                            "  * "
+                            + finding["message"]
+                            + " at "
+                            + finding["location"]
+                            + "\n    "
+                            + _format_code_line_for_output(finding["code"])
+                        )
+                    yield ("")
+
+    @staticmethod
+    def render_scan(scan_results: dict) -> tuple[str, str]:
+        """
+        Report the scans results in a human-readable format.
+
+        Args:
+            scan_results (dict): The scan results to be reported.
+        """
+        return (
+            "\n".join(
+                HumanReadableReporter.print_scan_results(
+                identifier=scan_results["package"], 
+                results=scan_results)),
+            "\n".join(
+                    HumanReadableReporter.print_errors(
+                        identifier=scan_results["package"], 
+                        results=scan_results
+                )
+            ),
+        )
+
+    @staticmethod
+    def render_verify(
+        dependency_files: List[DependencyFile],
+        rule_names: list[str],
+        scan_results: list[dict],
+        ecosystem: ECOSYSTEM,
+    ) -> tuple[str, str]:
+        return (
+            "\n".join(
+                map(lambda s: "\n".join(
+                    HumanReadableReporter.print_scan_results(
+                        identifier=s["dependency"],
+                        results=s["result"]
+                    ),
+                ), scan_results)
+            ),
+            "\n".join(
+                map(lambda s: "\n".join(
+                    HumanReadableReporter.print_errors(
+                        identifier=s["dependency"],
+                        results=s["result"]
+                    ),
+                ), scan_results)
+            )
+        )
+

guarddog/reporters/json.py

@@ -0,0 +1,28 @@
+import json
+from typing import List
+from guarddog.scanners.scanner import DependencyFile
+from guarddog.ecosystems import ECOSYSTEM
+
+from guarddog.reporters import BaseReporter
+
+
+class JsonReporter(BaseReporter):
+    @staticmethod
+    def render_verify(
+        dependency_files: List[DependencyFile],
+        rule_names: list[str],
+        scan_results: list[dict],
+        ecosystem: ECOSYSTEM,
+    ) -> tuple[str, str]:
+        return json.dumps(scan_results), ""
+
+    @staticmethod
+    def render_scan(scan_results: dict) -> tuple[str, str]:
+        """
+        Report the scans results in a json format.
+
+        Args:
+            scan_results (dict): The scan results to be reported.
+        """
+        # this reporter will output the errors in stdout
+        return json.dumps(scan_results), ""