Added support for Warn & Monitor policy modes (#13)

Author: dacoburn

socketsecurity/__init__.py

@@ -1,2 +1,2 @@
 __author__ = 'socket.dev'
-__version__ = '0.0.99'
+__version__ = '1.0.0'

socketsecurity/core/__init__.py

@@ -637,30 +637,16 @@ def compare_issue_alerts(new_scan_alerts: dict, head_scan_alerts: dict, alerts:
             if alert_key not in head_scan_alerts:
                 new_alerts = new_scan_alerts[alert_key]
                 for alert in new_alerts:
-                    if Core.is_error(alert):
+                    if alert.error or alert.warn:
                         alerts.append(alert)
             else:
                 new_alerts = new_scan_alerts[alert_key]
                 head_alerts = head_scan_alerts[alert_key]
                 for alert in new_alerts:
-                    if alert not in head_alerts and Core.is_error(alert):
+                    if alert not in head_alerts and (alert.error or alert.warn):
                         alerts.append(alert)
         return alerts
 
-    @staticmethod
-    def is_error(alert: Alert):
-        """
-        Compare the current alert against the Security Policy to determine if it should be included. Can be overridden
-        with all_new_alerts Global setting if desired to return all alerts and not just the error category from the
-        security policy.
-        :param alert:
-        :return:
-        """
-        if all_new_alerts or (alert.type in security_policy and security_policy[alert.type]['action'] == "error"):
-            return True
-        else:
-            return False
-
     @staticmethod
     def create_issue_alerts(package: Package, alerts: dict, packages: dict) -> dict:
         """
@@ -704,6 +690,9 @@ def create_issue_alerts(package: Package, alerts: dict, packages: dict) -> dict:
                 purl=package.purl,
                 url=package.url
             )
+            if alert.type in security_policy:
+                action = security_policy[alert.type]['action']
+                setattr(issue_alert, action, True)
             if issue_alert.key not in alerts:
                 alerts[issue_alert.key] = [issue_alert]
             else:

socketsecurity/core/classes.py

@@ -140,7 +140,10 @@ class Issue:
     pkg_id: str
     props: dict
     key: str
-    is_error: bool
+    error: bool
+    warn: bool
+    ignore: bool
+    monitor: bool
     description: str
     title: str
     emoji: str
@@ -162,6 +165,14 @@ def __init__(self, **kwargs):
             self.introduced_by = []
         if not hasattr(self, "manifests"):
             self.manifests = ""
+        if not hasattr(self, "error"):
+            self.error = False
+        if not hasattr(self, "warn"):
+            self.warn = False
+        if not hasattr(self, "monitor"):
+            self.monitor = False
+        if not hasattr(self, "ignore"):
+            self.ignore = False
 
     def __str__(self):
         return json.dumps(self.__dict__)

socketsecurity/core/messages.py

@@ -9,10 +9,13 @@ class Messages:
 
     @staticmethod
     def create_security_comment_json(diff: Diff) -> dict:
+        scan_failed = False
         if len(diff.new_alerts) == 0:
-            scan_failed = False
-        else:
-            scan_failed = True
+            for alert in diff.new_alerts:
+                alert: Issue
+                if alert.error:
+                    scan_failed = True
+                    break
         output = {
             "scan_failed": scan_failed,
             "new_alerts": []
@@ -22,7 +25,6 @@ def create_security_comment_json(diff: Diff) -> dict:
             output["new_alerts"].append(json.loads(str(alert)))
         return output
 
-
     @staticmethod
     def security_comment_template(diff: Diff) -> str:
         """
@@ -130,7 +132,8 @@ def create_security_alert_table(diff: Diff, md: MdUtils) -> (MdUtils, list, dict
             "Alert",
             "Package",
             "Introduced by",
-            "Manifest File"
+            "Manifest File",
+            "CI"
         ]
         num_of_alert_columns = len(alert_table)
         next_steps = {}
@@ -147,11 +150,16 @@ def create_security_alert_table(diff: Diff, md: MdUtils) -> (MdUtils, list, dict
                 ignore_commands.append(ignore)
             manifest_str, sources = Messages.create_sources(alert, "console")
             purl_url = f"[{alert.purl}]({alert.url})"
+            if alert.error:
+                emoji = ':no_entry_sign:'
+            else:
+                emoji = ':warning:'
             row = [
                 alert.title,
                 purl_url,
                 ", ".join(sources),
-                manifest_str
+                manifest_str,
+                emoji
             ]
             if row not in alert_table:
                 alert_table.extend(row)
@@ -262,17 +270,27 @@ def create_console_security_alert_table(diff: Diff) -> PrettyTable:
                 "Alert",
                 "Package",
                 "Introduced by",
-                "Manifest File"
+                "Manifest File",
+                "CI Status"
             ]
         )
         for alert in diff.new_alerts:
             alert: Issue
             manifest_str, sources = Messages.create_sources(alert, "console")
+            if alert.error:
+                state = "block"
+            elif alert.warn:
+                state = "warn"
+            elif alert.monitor:
+                state = "monitor"
+            else:
+                state = "ignore"
             row = [
                 alert.title,
                 alert.url,
                 ", ".join(sources),
-                manifest_str
+                manifest_str,
+                state
             ]
             alert_table.add_row(row)
         return alert_table

socketsecurity/socketcli.py

@@ -1,7 +1,7 @@
 import argparse
 import json
 from socketsecurity.core import Core, __version__
-from socketsecurity.core.classes import FullScanParams, Diff, Package
+from socketsecurity.core.classes import FullScanParams, Diff, Package, Alert
 from socketsecurity.core.messages import Messages
 from socketsecurity.core.scm_comments import Comments
 from socketsecurity.core.git_interface import Git
@@ -146,9 +146,10 @@
 def output_console_comments(diff_report: Diff, sbom_file_name: str = None) -> None:
     console_security_comment = Messages.create_console_security_alert_table(diff_report)
     save_sbom_file(diff_report, sbom_file_name)
-    if len(diff_report.new_alerts) > 0:
+    if not report_pass(diff_report):
         log.info("Security issues detected by Socket Security")
-        log.info(console_security_comment)
+        msg = f"\n{console_security_comment}"
+        log.info(msg)
         sys.exit(1)
     else:
         log.info("No New Security issues detected by Socket Security")
@@ -158,14 +159,26 @@ def output_console_json(diff_report: Diff, sbom_file_name: str = None) -> None:
     console_security_comment = Messages.create_security_comment_json(diff_report)
     save_sbom_file(diff_report, sbom_file_name)
     print(json.dumps(console_security_comment))
-    if len(diff_report.new_alerts) > 0:
+    if not report_pass(diff_report):
         sys.exit(1)
 
 
+def report_pass(diff_report: Diff) -> bool:
+    report_passed = True
+    if len(diff_report.new_alerts) > 0:
+        for alert in diff_report.new_alerts:
+            alert: Alert
+            if report_passed and alert.error:
+                report_passed = False
+                break
+    return report_passed
+
+
 def save_sbom_file(diff_report: Diff, sbom_file_name: str = None):
     if diff_report is not None and sbom_file_name is not None:
         Core.save_file(sbom_file_name, json.dumps(Core.create_sbom_output(diff_report)))
 
+
 def cli():
     try:
         main_code()