OpenVAS endpoint and severity improvements (#11955)

* openvas: trim/strip hostname

* openvas: trim/strip hostname

* openvas: parse endpoints from xml

* openvas: use string based severity in xml parser

* openvas: add ip to description

* cleanup

* cleanup

Author: valentijnscholten

dojo/tools/openvas/csv_parser.py

@@ -118,8 +118,9 @@ def __init__(self):
     def map_column_value(self, finding, column_value):
         if not finding.unsaved_endpoints[
             0
-        ].host:  # process only if host is not already defined (by field hostname)
-            finding.unsaved_endpoints[0].host = column_value
+        ].host and column_value is not None:  # process only if host is not already defined (by field hostname)
+            # strip due to https://github.com/greenbone/gvmd/issues/2378
+            finding.unsaved_endpoints[0].host = column_value.strip()
 
 
 class HostnameColumnMappingStrategy(ColumnMappingStrategy):
@@ -129,7 +130,8 @@ def __init__(self):
 
     def map_column_value(self, finding, column_value):
         if column_value:  # do not override IP if hostname is empty
-            finding.unsaved_endpoints[0].host = column_value
+            # strip due to https://github.com/greenbone/gvmd/issues/2378
+            finding.unsaved_endpoints[0].host = column_value.strip()
 
 
 class SeverityColumnMappingStrategy(ColumnMappingStrategy):
@@ -278,13 +280,21 @@ def get_findings(self, filename, test):
             finding = Finding(test=test)
             finding.unsaved_vulnerability_ids = []
             finding.unsaved_endpoints = [Endpoint()]
+            ip = None
             if row_number == 0:
                 column_names = self.read_column_names(row)
                 continue
             for column_number, column in enumerate(row):
                 chain.process_column(
                     column_names[column_number], column, finding,
                 )
+                # due to the way this parser is implemented we have to do this stuff to retrieve a value for later use
+                if column_names[column_number].lower() == "ip":
+                    ip = column
+
+            if ip:
+                finding.description += f"\n**IP**: {ip}"
+
             if finding is not None and row_number > 0:
                 if finding.title is None:
                     finding.title = ""

dojo/tools/openvas/xml_parser.py

@@ -1,8 +1,9 @@
+import contextlib
 from xml.dom import NamespaceErr
 
 from defusedxml import ElementTree as ET
 
-from dojo.models import Finding
+from dojo.models import Endpoint, Finding
 
 
 class OpenVASXMLParser:
@@ -17,26 +18,41 @@ def get_findings(self, filename, test):
         results = report.find("results")
         for result in results:
             script_id = None
-            for finding in result:
-                if finding.tag == "name":
-                    title = finding.text
-                    description = [f"**Name**: {finding.text}"]
-                if finding.tag == "host":
-                    title = title + "_" + finding.text
-                    description.append(f"**Host**: {finding.text}")
-                if finding.tag == "port":
-                    title = title + "_" + finding.text
-                    description.append(f"**Port**: {finding.text}")
-                if finding.tag == "nvt":
-                    description.append(f"**NVT**: {finding.text}")
-                    script_id = finding.get("oid") or finding.text
-                if finding.tag == "severity":
-                    severity = self.convert_cvss_score(finding.text)
-                    description.append(f"**Severity**: {finding.text}")
-                if finding.tag == "qod":
-                    description.append(f"**QOD**: {finding.text}")
-                if finding.tag == "description":
-                    description.append(f"**Description**: {finding.text}")
+            unsaved_endpoint = Endpoint()
+            for field in result:
+                if field.tag == "name":
+                    title = field.text
+                    description = [f"**Name**: {field.text}"]
+                if field.tag == "hostname":
+                    title = title + "_" + field.text
+                    description.append(f"**Hostname**: {field.text}")
+                    if field.text:
+                        unsaved_endpoint.host = field.text.strip()  # strip due to https://github.com/greenbone/gvmd/issues/2378
+                if field.tag == "host":
+                    title = title + "_" + field.text
+                    description.append(f"**Host**: {field.text}")
+                    if not unsaved_endpoint.host and field.text:
+                        unsaved_endpoint.host = field.text.strip()  # strip due to https://github.com/greenbone/gvmd/issues/2378
+                if field.tag == "port":
+                    title = title + "_" + field.text
+                    description.append(f"**Port**: {field.text}")
+                    if field.text:
+                        port_str, protocol = field.text.split("/")
+                        with contextlib.suppress(ValueError):
+                            unsaved_endpoint.port = int(port_str)
+                        unsaved_endpoint.protocol = protocol
+                if field.tag == "nvt":
+                    description.append(f"**NVT**: {field.text}")
+                    script_id = field.get("oid") or field.text
+                if field.tag == "severity":
+                    description.append(f"**Severity**: {field.text}")
+                if field.tag == "threat":
+                    description.append(f"**Threat**: {field.text}")
+                    severity = field.text if field.text in {"Info", "Low", "Medium", "High", "Critical"} else "Info"
+                if field.tag == "qod":
+                    description.append(f"**QOD**: {field.text}")
+                if field.tag == "description":
+                    description.append(f"**Description**: {field.text}")
 
             finding = Finding(
                 title=str(title),
@@ -47,6 +63,7 @@ def get_findings(self, filename, test):
                 static_finding=False,
                 vuln_id_from_tool=script_id,
             )
+            finding.unsaved_endpoints = [unsaved_endpoint]
             findings.append(finding)
         return findings