deploy dev

tools/devsecops_engine_tools/engine_core/src/applications/runner_engine_core.py

@@ -195,11 +195,7 @@ def application_core():
             "local": RuntimeLocal(),
         }.get(args["platform_devops"])
         metrics_manager_gateway = S3Manager()
-
-        if args["tool"] == "engine_risk":
-            printer_table_gateway = PrinterRichTable()
-        else:
-            printer_table_gateway = PrinterPrettyTable()
+        printer_table_gateway = PrinterPrettyTable()
 
         init_engine_core(
             vulnerability_management_gateway,

tools/devsecops_engine_tools/engine_core/src/domain/model/exclusions.py

@@ -15,5 +15,4 @@ def __init__(self, **kwargs):
         self.vm_id = kwargs.get("vm_id", "")
         self.vm_id_url = kwargs.get("vm_id_url", "")
         self.service = kwargs.get("service", "")
-        self.service_url = kwargs.get("service_url", "")
         self.tags = kwargs.get("tags", [])

tools/devsecops_engine_tools/engine_core/src/domain/model/report.py

@@ -29,7 +29,6 @@ def __init__(self, **kwargs):
         self.risk_accepted = kwargs.get("risk_accepted", "")
         self.false_p = kwargs.get("false_p", "")
         self.service = kwargs.get("service", "")
-        self.service_url = kwargs.get("service_url", "")
         self.reason = kwargs.get("reason", "")
         self.component_name = kwargs.get("component_name", "")
         self.component_version = kwargs.get("component_version", "")

tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/defect_dojo/defect_dojo.py

@@ -418,7 +418,6 @@ def _create_report_exclusion(self, finding, date_fn, tool, reason, host_dd):
             vm_id=str(finding.vm_id),
             vm_id_url=f"{host_dd}/finding/{finding.vm_id}",
             service=finding.service,
-            service_url=f"{host_dd}/finding?active=true&service={finding.service}",
             tags=finding.tags,
         )
 
@@ -452,7 +451,6 @@ def _create_report(self, finding, host_dd):
             risk_accepted=finding.risk_accepted,
             false_p=finding.false_p,
             service=finding.service,
-            service_url=f"{host_dd}/finding?active=true&service={finding.service}",
         )
 
     def _format_date_to_dd_format(self, date_string):

tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/printer_pretty_table/printer_pretty_table.py

@@ -10,7 +10,7 @@
     Report,
 )
 from devsecops_engine_tools.engine_core.src.infrastructure.helpers.util import (
-    format_date
+    format_date,
 )
 from prettytable import PrettyTable, DOUBLE_BORDER
 
@@ -63,24 +63,20 @@ def print_table_findings(self, finding_list: "list[Finding]"):
             print(sorted_table)
 
     def print_table_report(self, report_list: "list[Report]"):
-        headers = ["Risk Score", "Severity", "ID", "Tags", "Where", "Service"]
+        headers = ["Risk Score", "ID", "Tags", "Services"]
         table = PrettyTable(headers)
         for report in report_list:
             row_data = [
                 report.risk_score,
-                report.severity.lower(),
-                report.vuln_id_from_tool if report.vuln_id_from_tool else report.id,
-                report.tags,
-                report.where,
-                report.service
+                self._check_spaces_url(report.vm_id, report.vm_id_url),
+                ", ".join(report.tags),
+                self._check_spaces(report.service),
             ]
             table.add_row(row_data)
 
         sorted_table = PrettyTable()
         sorted_table.field_names = table.field_names
-        sorted_table.add_rows(
-            sorted(table._rows, key=lambda row: row[0], reverse=True)
-        )
+        sorted_table.add_rows(sorted(table._rows, key=lambda row: row[0], reverse=True))
 
         for column in table.field_names:
             sorted_table.align[column] = "l"
@@ -90,9 +86,52 @@ def print_table_report(self, report_list: "list[Report]"):
         if len(sorted_table.rows) > 0:
             print(sorted_table)
 
+    def print_table_report_exlusions(self, exclusions):
+        if exclusions:
+            headers = [
+                "ID",
+                "Tags",
+                "Services",
+                "Created Date",
+                "Expired Date",
+                "Reason",
+            ]
+
+        table = PrettyTable(headers)
+
+        for exclusion in exclusions:
+            row_data = [
+                self._check_spaces_url(exclusion["vm_id"], exclusion["vm_id_url"]),
+                ", ".join(exclusion["tags"]),
+                self._check_spaces(exclusion["service"]),
+                format_date(exclusion["create_date"], "%d%m%Y", "%d/%m/%Y"),
+                (
+                    format_date(exclusion["expired_date"], "%d%m%Y", "%d/%m/%Y")
+                    if exclusion["expired_date"]
+                    and exclusion["expired_date"] != "undefined"
+                    else "NA"
+                ),
+                exclusion["reason"],
+            ]
+            table.add_row(row_data)
+
+        for column in table.field_names:
+            table.align[column] = "l"
+
+        table.set_style(DOUBLE_BORDER)
+        if len(table.rows) > 0:
+            print(table)
+
     def print_table_exclusions(self, exclusions):
-        if (exclusions):
-            headers = ["Severity", "ID", "Where", "Create Date", "Expired Date", "Reason"]
+        if exclusions:
+            headers = [
+                "Severity",
+                "ID",
+                "Where",
+                "Create Date",
+                "Expired Date",
+                "Reason",
+            ]
 
         table = PrettyTable(headers)
 
@@ -102,7 +141,12 @@ def print_table_exclusions(self, exclusions):
                 exclusion["id"],
                 exclusion["where"],
                 format_date(exclusion["create_date"], "%d%m%Y", "%d/%m/%Y"),
-                format_date(exclusion["expired_date"], "%d%m%Y", "%d/%m/%Y") if exclusion["expired_date"] and exclusion["expired_date"] != "undefined" else "NA",
+                (
+                    format_date(exclusion["expired_date"], "%d%m%Y", "%d/%m/%Y")
+                    if exclusion["expired_date"]
+                    and exclusion["expired_date"] != "undefined"
+                    else "NA"
+                ),
                 exclusion["reason"],
             ]
             table.add_row(row_data)
@@ -113,3 +157,22 @@ def print_table_exclusions(self, exclusions):
         table.set_style(DOUBLE_BORDER)
         if len(table.rows) > 0:
             print(table)
+
+    def _check_spaces_url(self, value, url):
+        values = value.split()
+        urls = url.split()
+        new_value = ""
+        if len(values) > 1 or len(urls) > 1:
+            new_value = "\n".join(f"{v}[{u}]" for v, u in zip(values, urls))
+        else:
+            new_value = f"{values[0]}[{urls[0]}]"
+        return new_value
+
+    def _check_spaces(self, value):
+        values = value.split()
+        new_value = ""
+        if len(values) > 1:
+            new_value = "\n".join(values)
+        else:
+            new_value = f"{values[0]}"
+        return new_value

tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/printer_rich_table/printer_rich_table.py

@@ -38,7 +38,7 @@ def print_table_report(self, report_list: "list[Report]"):
                 str(report.risk_score),
                 self._check_spaces(report.vm_id, report.vm_id_url),
                 ", ".join(report.tags),
-                self._check_spaces(report.service, report.service_url),
+                report.service,
             ]
             table.add_row(*row_data)
         console = Console()
@@ -57,7 +57,7 @@ def print_table_exclusions(self, exclusions_list):
             row_data = [
                 self._check_spaces(exclusion["vm_id"], exclusion["vm_id_url"]),
                 ", ".join(exclusion["tags"]),
-                self._check_spaces(exclusion["service"], exclusion["service_url"]),
+                exclusion["service"],
                 format_date(exclusion["create_date"], "%d%m%Y", "%d/%m/%Y"),
                 (
                     format_date(exclusion["expired_date"], "%d%m%Y", "%d/%m/%Y")

tools/devsecops_engine_tools/engine_core/test/infrastructure/driven_adapters/printer_pretty_table/test_printer_pretty_table.py

@@ -103,7 +103,16 @@ def test_print_table_without_findings(self, mock_print):
     @patch("builtins.print")
     def test_print_table_exclusions(self, mock_print):
         # Arrange
-        exclusions = [{"severity": "severity" ,"id": "id", "where": "path", "create_date": "01042023", "expired_date": "04032023", "reason": "reason"}]
+        exclusions = [
+            {
+                "severity": "severity",
+                "id": "id",
+                "where": "path",
+                "create_date": "01042023",
+                "expired_date": "04032023",
+                "reason": "reason",
+            }
+        ]
         printer = PrinterPrettyTable()
 
         # Act
@@ -119,13 +128,14 @@ def test_print_table_report(self, mock_print):
         report_list = [
             Report(
                 risk_score=1,
-                id="id2",
-                date="21022024",
+                vm_id="id1 id2",
+                vm_id_url="url1 url2",
                 status="stat2",
                 where="path",
                 tags=["tag1"],
                 severity="low",
                 active=True,
+                service="service1",
             ),
         ]
         printer = PrinterPrettyTable()
@@ -135,6 +145,25 @@ def test_print_table_report(self, mock_print):
 
         # Assert
         assert mock_print.called
-        # Add more assertions to validate the output
 
+    @patch("builtins.print")
+    def test_print_table_report_exlusions(self, mock_print):
+        # Arrange
+        exclusions = [
+            {
+                "vm_id": "id",
+                "vm_id_url": "url",
+                "tags": ["tag1"],
+                "service": "service1",
+                "create_date": "01042023",
+                "expired_date": "04032023",
+                "reason": "reason",
+            }
+        ]
+        printer = PrinterPrettyTable()
+
+        # Act
+        printer.print_table_report_exlusions(exclusions)
 
+        # Assert
+        assert mock_print.called

tools/devsecops_engine_tools/engine_core/test/infrastructure/driven_adapters/printer_rich_table/test_printer_rich_table.py

@@ -60,7 +60,6 @@ def test_print_table_exclusions(self, mock_console):
                 "expired_date": "02012021",
                 "reason": "reason1",
                 "vm_id_url": "url1",
-                "service_url": "url2",
             }
         ]
         printer = PrinterRichTable()

tools/devsecops_engine_tools/engine_risk/src/domain/usecases/break_build.py

@@ -13,7 +13,6 @@
 
 from collections import Counter
 import copy
-from rich.console import Console
 
 
 class BreakBuild:
@@ -166,7 +165,6 @@ def _map_applied_exclusion(self, exclusions: "list[Exclusions]"):
                 "vm_id": exclusion.vm_id,
                 "vm_id_url": exclusion.vm_id_url,
                 "service": exclusion.service,
-                "service_url": exclusion.service_url,
                 "tags": exclusion.tags,
             }
             for exclusion in exclusions
@@ -185,13 +183,13 @@ def _apply_exclusions(self, report_list: "list[Report]"):
                         and report.vuln_id_from_tool == exclusion.id
                     )
                     or (report.id and report.id == exclusion.id)
+                    or (report.vm_id and exclusion.id in report.vm_id)
                 ) and ((exclusion.where in report.where) or (exclusion.where == "all")):
                     exclude = True
                     exclusion_copy = copy.deepcopy(exclusion)
                     exclusion_copy.vm_id = report.vm_id
                     exclusion_copy.vm_id_url = report.vm_id_url
                     exclusion_copy.service = report.service
-                    exclusion_copy.service_url = report.service_url
                     exclusion_copy.tags = report.tags
                     applied_exclusions.append(exclusion_copy)
                     break
@@ -206,7 +204,6 @@ def _tag_blacklist_control(self, report_list: "list[Report]"):
         if report_list:
             tag_blacklist = set(remote_config["THRESHOLD"]["TAG_BLACKLIST"])
             tag_age_threshold = remote_config["THRESHOLD"]["TAG_MAX_AGE"]
-            console = Console()
 
             filtered_reports_above_threshold = [
                 (report, tag)
@@ -224,13 +221,19 @@ def _tag_blacklist_control(self, report_list: "list[Report]"):
 
             for report, tag in filtered_reports_above_threshold:
                 report.reason = "Blacklisted"
-                console.print(
-                    f"[red]Report [link={report.vm_id_url}]{report.vm_id}[/link] with tag {tag} is blacklisted and age {report.age} is above threshold {tag_age_threshold}[/red]"
+                print(
+                    self.devops_platform_gateway.message(
+                        "error",
+                        f"Report {report.vm_id}[{report.vm_id_url}] with tag {tag} is blacklisted and age {report.age} is above threshold {tag_age_threshold}",
+                    )
                 )
 
             for report, tag in filtered_reports_below_threshold:
-                console.print(
-                    f"[yellow]Report [link={report.vm_id_url}]{report.vm_id}[/link] with tag {tag} is blacklisted but age {report.age} is below threshold {tag_age_threshold}[/yellow]"
+                print(
+                    self.devops_platform_gateway.message(
+                        "warning",
+                        f"Report {report.vm_id}[{report.vm_id_url}] with tag {tag} is blacklisted but age {report.age} is below threshold {tag_age_threshold}",
+                    )
                 )
 
             if filtered_reports_above_threshold:
@@ -301,7 +304,7 @@ def _print_exclusions(self, applied_exclusions: "list[Exclusions]"):
                     "warning", "Bellow are all findings that were excepted"
                 )
             )
-            self.printer_table_gateway.print_table_exclusions(applied_exclusions)
+            self.printer_table_gateway.print_table_report_exlusions(applied_exclusions)
             for reason, total in Counter(
                 map(lambda x: x["reason"], applied_exclusions)
             ).items():

tools/devsecops_engine_tools/engine_risk/src/domain/usecases/handle_filters.py

@@ -1,5 +1,4 @@
 import copy
-from rich.console import Console
 
 
 class HandleFilters:
@@ -21,11 +20,6 @@ def filter_duplicated(self, findings):
                     for s in finding.service.split()
                     if s not in existing_finding.service.split()
                 ]
-                combined_services_url = existing_finding.service_url.split() + [
-                    s_url
-                    for s_url in finding.service_url.split()
-                    if s_url not in existing_finding.service_url.split()
-                ]
                 combined_vm_ids = existing_finding.vm_id.split() + [
                     vm
                     for vm in finding.vm_id.split()
@@ -39,13 +33,11 @@ def filter_duplicated(self, findings):
                 if finding.age >= existing_finding.age:
                     new_finding = copy.deepcopy(finding)
                     new_finding.service = " ".join(combined_services)
-                    new_finding.service_url = " ".join(combined_services_url)
                     new_finding.vm_id = " ".join(combined_vm_ids)
                     new_finding.vm_id_url = " ".join(combined_vm_id_urls)
                     findings_map[key] = new_finding
                 else:
                     existing_finding.service = " ".join(combined_services)
-                    existing_finding.service_url = " ".join(combined_services_url)
                     new_finding.vm_id = " ".join(combined_vm_ids)
                     new_finding.vm_id_url = " ".join(combined_vm_id_urls)
             else:
@@ -54,18 +46,20 @@ def filter_duplicated(self, findings):
         unique_findings = list(findings_map.values())
         return unique_findings
 
-    def filter_tags_days(self, remote_config, findings):
+    def filter_tags_days(self, devops_platform_gateway, remote_config, findings):
         tag_exclusion_days = remote_config["TAG_EXCLUSION_DAYS"]
         filtered_findings = []
-        console = Console()
 
         for finding in findings:
             exclude = False
             for tag in finding.tags:
                 if tag in tag_exclusion_days and finding.age < tag_exclusion_days[tag]:
                     exclude = True
-                    console.print(
-                        f"[yellow]Report [link={finding.vm_id_url}]{finding.vm_id}[/link] with tag '{tag}' and age {finding.age} days is being excluded. It will be considered in {tag_exclusion_days[tag] - finding.age} days.[/yellow]"
+                    print(
+                        devops_platform_gateway.message(
+                            "warning",
+                            f"Report {finding.vm_id}[{finding.vm_id_url}] with tag '{tag}' and age {finding.age} days is being excluded. It will be considered in {tag_exclusion_days[tag] - finding.age} days.",
+                        )
                     )
                     break
             if not exclude: