update the affected packages to the following versions:
" + description += "The following packages are affected:
" + + for package_name, package_versions in packages.items(): + package_names.append(package_name.split('/')[0]) + + mitigation += "{name}:
".format(name=package_name) + + description += "{name}:
".format(name=package_name) + for versions in package_versions: + mitigation += "{fixed}
".format(fixed=versions.get("fixed_version", "unknown")) + + description += "installed version: {installed}
".format(installed=versions.get("package_version", "unknown")) + description += "fixed version: {fixed}
".format(fixed=versions.get("fixed_version", "unknown")) + + link = vulnerability.get("link") if "link" in vulnerability else "" + + vectors_v3 = vulnerability.get("vectors_v3", "") + + score_v3 = vulnerability.get("score_v3", "") + + published = datetime.fromtimestamp(int(vulnerability.get("published_timestamp", 0))) + + vulnerability_id = vulnerability.get("name") + + # there is nothing like short description, short name or title + package_names_combined = ','.join(sorted(set(package_names), key=str)) + if len(package_names_combined) > 32: + package_names_combined = package_names_combined[-32:] + + title = "{packages}: ({vuln})".format(packages=package_names_combined, vuln=vulnerability.get("name").upper()) + + # create the finding object + finding = Finding( + title=title, + test=test, + description=description, + severity=severity, + mitigation=mitigation, + impact="", + url=link, + cvssv3=vectors_v3, + cvssv3_score=score_v3, + publish_date=published, + ) + + finding.unsaved_vulnerability_ids = [vulnerability_id] + + finding.unsaved_endpoints = [] + + nodes = vulnerability.get("nodes", []) + for node in nodes: + endpoint = Endpoint( + host=node.get("display_name", ""), + ) + finding.unsaved_endpoints.append(endpoint) + + return finding + + # see neuvector/share/types.go def convert_severity(severity): if severity.lower() == "critical": @@ -137,7 +222,7 @@ def get_label_for_scan_types(self, scan_type): return NEUVECTOR_SCAN_NAME def get_description_for_scan_types(self, scan_type): - return "JSON output of /v1/scan/{entity}/{id} endpoint." + return "JSON output of /v1/scan/{entity}/{id} endpoint (vulnerabilities of an endpoint). Or vulnerabilities of several assets (VulnerabilityAsset / ComplianceAsset)." def get_findings(self, filename, test): if filename is None: diff --git a/dojo/tools/neuvector_compliance/parser.py b/dojo/tools/neuvector_compliance/parser.py index 74e5e515fd1..984e7d53e73 100644 --- a/dojo/tools/neuvector_compliance/parser.py +++ b/dojo/tools/neuvector_compliance/parser.py @@ -1,7 +1,7 @@ import hashlib import json -from dojo.models import Finding +from dojo.models import Endpoint,Finding NEUVECTOR_SCAN_NAME = "NeuVector (compliance)" @@ -35,10 +35,10 @@ def get_items(tree, test): # endpoints like /v1/scan/workload/{id}. otherwize, it is an export from # /v1/host/{id}/compliance or similar. thus, we need to support items in a # bit different leafs. - testsTree = None + testsTree = [] if "report" in tree: testsTree = tree.get("report").get("checks", []) - else: + elif "items" in tree: testsTree = tree.get("items", []) for node in testsTree: @@ -51,6 +51,23 @@ def get_items(tree, test): ) unique_key = hashlib.md5(unique_key.encode("utf-8")).hexdigest() items[unique_key] = item + + # asset-style collection with compliance issues of several assets + testsAssetsTree = [] + if "compliance_issues" in tree: + testsAssetsTree = tree.get("compliance_issues", []) + for node in testsAssetsTree: + item = get_asset_item(node, test) + unique_key = ( + node.get("name") + + node.get("category") + + node.get("type") + + node.get("level") + + node.get("profile") + ) + unique_key = hashlib.md5(unique_key.encode("utf-8")).hexdigest() + items[unique_key] = item + return list(items.values()) @@ -117,6 +134,63 @@ def get_item(node, test): return finding +def get_asset_item(comp_issue, test): + test_name = comp_issue.get("name", "unknown") + + test_description = comp_issue.get("description", "").rstrip() + + test_severity = comp_issue.get("level", "unknown") + + mitigation = comp_issue.get("remediation", "").rstrip() + + category = comp_issue.get("category", "unknown") + + test_profile = comp_issue.get("profile", "unknown") + + full_description = "{} ({}), {}:
".format( + test_name, category, test_profile + ) + full_description += "{}
".format(test_description) + full_description += "Audit: {}
".format(test_severity) + full_description += "Mitigation:
" + full_description += "{}
".format(mitigation) + + tags = comp_issue.get("tags", []) + if len(tags) > 0: + full_description += "Applicable compliance standards: {tags}
".format(tags=','.join(sorted(set(tags), key=str))) + + messages = comp_issue.get("message", []) + if len(messages) > 0: + full_description += "Messages:
" + for m in messages: + full_description += "{}
".format(str(m).rstrip()) + + finding = Finding( + title="{name} - {desc}".format(name=test_name, desc=test_description), + test=test, + description=full_description, + severity=convert_severity(test_severity), + mitigation=mitigation, + vuln_id_from_tool="{category}_{name}".format(category=category, name=test_name), + impact="", + static_finding=True, + dynamic_finding=False, + ) + + finding.unsaved_vulnerability_ids = [] + + finding.unsaved_endpoints = [] + + nodes = comp_issue.get("nodes", []) + for node in nodes: + endpoint = Endpoint( + host=node.get("display_name", ""), + ) + finding.unsaved_endpoints.append(endpoint) + + return finding + + # see neuvector/share/clus_apis.go def convert_severity(severity): if severity.lower() == "high":