add nvt statistics per host result

CHANGELOG.md

@@ -6,6 +6,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 ### Added
+- nvt threat information in host result [114](https://github.com/greenbone/pheme/pull/114)
 ### Changed
 - remove pandas due to too old debian version [112](https://github.com/greenbone/pheme/pull/112)
 ### Deprecated

pheme/settings.py

@@ -166,6 +166,7 @@
     'default': {
         'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache',
         'LOCATION': '/tmp/django_cache',
+        'TIMEOUT': 1 * 60 * 2 * 60,  # 2 hours
     }
 }
 

pheme/transformation/scanreport/gvmd.py

@@ -42,13 +42,15 @@
 from pheme.transformation.scanreport.model import (
     CountGraph,
     # Equipment,
-    HostResults,
     Overview,
     Report,
 )
 
 logger = logging.getLogger(__name__)
 
+__threats = ["High", "Medium", "Low"]
+__threat_index_lookup = {v: i for i, v in enumerate(__threats)}
+
 
 def measure_time(func):
     def measure(*args, **kwargs):
@@ -196,12 +198,14 @@ def __tansform_tags(item) -> List[Dict]:
     return None
 
 
-def __return_highest_threat(old: str, new: str) -> str:
-    if old == 'High' or new == 'High':
-        return 'High'
-    if old == 'Medium' or new == 'Medium':
-        return 'Medium'
-    return 'Low'
+def __return_highest_threat(threats: List[int]) -> str:
+    """
+    retuns the highest threat within a list of threats based on __threats
+    """
+    for i, value in enumerate(threats):
+        if value > 0:
+            return __threats[i]
+    return 'NA'
 
 
 def __group_refs(refs: List[Dict]) -> Dict:
@@ -212,14 +216,6 @@ def __group_refs(refs: List[Dict]) -> Dict:
     return refs_ref
 
 
-def __threat_to_index(threat: str) -> int:
-    if threat == 'High':
-        return 0
-    if threat == 'Medium':
-        return 1
-    return 2
-
-
 def __get_hostname_from_result(result) -> str:
     if isinstance(result, str):
         return result
@@ -231,8 +227,54 @@ def __get_hostname_from_result(result) -> str:
     return 'unknown'
 
 
+def __host_nvt_overview(nvt_count: List[int]) -> Dict:
+    """
+    returns nvt statistics mostly used in the per host overview
+    """
+    result = {__threats[i].lower(): value for i, value in enumerate(nvt_count)}
+    result['total'] = sum(nvt_count)
+    result['highest'] = __return_highest_threat(nvt_count)
+    return result
+
+
+def __create_host_information_lookup(report: Dict) -> Dict:
+    """
+    created a lookup table for available host information
+    """
+    # lookup for host information name and dict name
+    information_key = {'best_os_txt': 'os'}
+
+    def filter_per_host(host: Dict) -> Dict:
+        information = {}
+        found = 0
+        details = host.get('detail', [])
+        for detail in details:
+            name = detail.get('name', '')
+            if name in information_key.keys():
+                information[information_key.get(name)] = detail.get('value')
+                found += 1
+            if found == len(information_key.keys()):
+                return information
+        return information
+
+    result = {}
+    hosts = report.get('host')
+    if isinstance(hosts, dict):
+        result[hosts.get('ip', 'unknown')] = filter_per_host(hosts)
+    elif isinstance(hosts, list):
+        # best_os_txt seems to be in the end
+        for host in hosts[::-1]:
+            result[host.get('ip', 'unknown')] = filter_per_host(host)
+    return result
+
+
 @measure_time
-def __create_results_per_host(report: Dict) -> List[HostResults]:
+def __create_results_per_host(report: Dict) -> List[Dict]:
+    """
+    creates the results dict used by a vulnerability-report based on a given
+    gvmd report.
+    """
+    host_information_lookup = __create_host_information_lookup(report)
     results = report.get('results', {}).get('result', [])
     by_host = {}
     host_count = {}
@@ -247,9 +289,6 @@ def per_result(result):
         hostname = __get_hostname_from_result(result)
         host_dict = by_host.get(hostname, {})
         threat = result.get('threat', 'unknown')
-        highest_threat = __return_highest_threat(
-            host_dict.get('threat', ''), threat
-        )
         port = result.get('port')
         nvt = transform_key("nvt", result.get('nvt', {}))
         nvt['nvt_tags_interpreted'] = __tansform_tags(nvt.get('nvt_tags', ''))
@@ -266,23 +305,30 @@ def per_result(result):
         host_results = host_dict.get('results', [])
         host_results.append(new_host_result)
         equipment = host_dict.get('equipment', {})
-        equipment['ports'] = equipment.get('ports', []) + [port]
-        # filter for best_os_cpe
-        equipment['os'] = "unknown"
-        by_host[hostname] = {
-            "host": hostname,
-            "threat": highest_threat,
-            "equipment": equipment,
-            "results": host_results,
-        }
+        equipment['ports'] = list(
+            dict.fromkeys(equipment.get('ports', []) + [port])
+        )
+        if not equipment.get('os'):
+            equipment['os'] = host_information_lookup.get(hostname, {}).get(
+                'os', 'unknown'
+            )
+
         # needs hostname, high, medium, low
         host_threats = host_count.get(hostname, [0, 0, 0])
-        threat_index = __threat_to_index(threat)
+        threat_index = __threat_index_lookup.get(threat, 2)
         host_threats[threat_index] += 1
         host_count[hostname] = host_threats
+
         # needs high, medium, low
         nvt_count[threat_index] += 1
 
+        by_host[hostname] = {
+            "host": hostname,
+            "threats": __host_nvt_overview(host_threats),
+            "equipment": equipment,
+            "results": host_results,
+        }
+
     # lists with just one element can be parsed as dict by xmltodict
     if isinstance(results, dict):
         per_result(results)

pheme/transformation/scanreport/model.py

@@ -33,13 +33,6 @@ class Equipment:
     ports: List[str]  # ports port host and ports port text
 
 
-@dataclass
-class HostResults:
-    host: str
-    equipment: Equipment
-    results: List[Dict]
-
-
 @dataclass
 class NVTCount:
     oid: str
@@ -82,7 +75,7 @@ class Report:
     version: str
     start: str
     overview: Overview
-    results: List[HostResults]
+    results: List[Dict]
 
 
 def describe():
@@ -110,8 +103,15 @@ def describe():
             ),
         ),
         results=[
-            HostResults(
+            dict(
                 host="str; ip address of host",
+                threats={
+                    'high': 'int; amount of high nvts in host',
+                    'medium': 'int; amount of medium nvts in host',
+                    'low': 'int; amount of low nvts in host',
+                    'total': 'int; amount of nvts in host',
+                    'highest': 'str; highest threat within host',
+                },
                 equipment=Equipment(
                     os="str; operating system", ports=["str; open ports"]
                 ),
@@ -134,7 +134,6 @@ def describe():
                     'nvt.solution.type': 'str; nvt.solution.type; optional',
                     'nvt.solution.text': 'str; nvt.solution.text; optional',
                     'port': 'str; port; optional',
-                    'threat': 'str; severity class of nvt',
                     'severity': 'str; severity; optional',
                     'qod.value': 'str; qod.value; optional',
                     'qod.type': 'str; qod.type; optional',

tests/generate_test_data.py

@@ -168,6 +168,7 @@ def gen_report(
                     "ip": res['host']['text'],
                     "detail": [
                         {"name": "best_os_cpe", "value": "rusty kernel"},
+                        {"name": "best_os_txt", "value": "rusty rust rust"},
                         {"name": "something else", "value": "hum"},
                     ],
                 }

tests/test_report_generation.py

@@ -33,6 +33,21 @@ def generate(prefix: str, amount: int) -> List[str]:
     return ["{}_{}".format(prefix, i) for i in range(amount)]
 
 
+def test_report_contains_equipment():
+    client = APIClient()
+    url = reverse('transform')
+    report = {
+        'report': {
+            'report': gen_report(generate('host', 10), generate('oid', 5))
+        }
+    }
+    response = client.post(url, data=report, format='xml')
+    assert response.status_code == 200
+    result = cache.get(response.data)
+    assert result['results'][0]['equipment']['os'] == "rusty rust rust"
+    assert result['results'][0]['equipment']['ports'] is not None
+
+
 def test_report_contains_charts():
     client = APIClient()
     url = reverse('transform')