Rating from vendor: {rating}
".format(rating=feed_rating) + published_date = None + published_ts = vulnerability.get("published_timestamp", 0) + if published_ts > 0: + published_date = datetime.fromtimestamp(int(published_ts)) - mitigation = "" + reference = vulnerability.get("link", "not provided") - package_names = [] + affected_packages = "" packages = vulnerability.get("packages", {}) if len(packages.values()) > 0: - mitigation += "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) + affected_packages += f"*{package_name}*\n" - 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", "") + installed=versions.get("package_version", "unknown") + fixed=versions.get("fixed_version", "unknown") + affected_packages += f" installed version: {installed}\n" + affected_packages += f" fixed version: {fixed}\n" - published_date = None - published_ts = vulnerability.get("published_timestamp", 0) - if published_ts > 0: - published_date = datetime.fromtimestamp(int(published_ts)) - - 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:] + affected_packages += "\n" - title = "{packages}: ({vuln})".format(packages=package_names_combined, vuln=vulnerability.get("name").upper()) + nodes = vulnerability.get("nodes", []) + workloads = vulnerability.get("workloads", []) + images = vulnerability.get("images", []) + platforms = vulnerability.get("platforms", []) + + # the same information is saved as Endpoint(s), however, DefectDojo + # Endpoint lacks many metadata fields, thus, difficult to read. + affected_systems = "" + + for asset in nodes: + display_name = asset.get("display_name", "") + domains = ','.join(asset.get("domains", [])) + affected_systems += f"*Node {display_name}*\n" + affected_systems += f" domains: {domains}\n" + + for asset in platforms: + display_name = asset.get("display_name", "") + domains = ','.join(asset.get("domains", [])) + affected_systems += f"*Platform {display_name}*\n" + affected_systems += f" domains: {domains}\n" + + for asset in images: + display_name = asset.get("display_name", "") + domains = ','.join(asset.get("domains", [])) + affected_systems += f"*Image {display_name}*\n" + affected_systems += f" domains: {domains}\n" + + for asset in workloads: + display_name = asset.get("display_name", "") + domains = ','.join(asset.get("domains", [])) + service = asset.get("service", "") + image = asset.get("image", "") + affected_systems += f"*Workload {display_name}*\n" + affected_systems += f" domains: {domains}\n" + affected_systems += f" service: {service}\n" + affected_systems += f" image: {image}\n" + + description = ASSET_FINDING_DESCRIPTION_TEMPLATE.format( + title=vulnerability_id, + description=vuln_description, + feed_rating=vulnerability.get("feed_rating", "not provided"), + published_date=published_date, + reference=reference, + affected_packages=affected_packages, + affected_systems=affected_systems, + ) # create the finding object finding = Finding( - title=title, + title=vulnerability_id, test=test, description=description, severity=severity, - mitigation=mitigation, impact="", - url=link, - cvssv3=vectors_v3, - cvssv3_score=score_v3, + references=reference, + cvssv3=vulnerability.get("vectors_v3", ""), + cvssv3_score=vulnerability.get("score_v3", ""), publish_date=published_date, ) @@ -195,16 +232,65 @@ def get_asset_item(vulnerability, test): finding.unsaved_endpoints = [] - nodes = vulnerability.get("nodes", []) - for node in nodes: - endpoint = Endpoint( - host=node.get("display_name", ""), - ) - finding.unsaved_endpoints.append(endpoint) + for asset in nodes: + endpoints = endpoints_from_asset("node", asset) + finding.unsaved_endpoints += endpoints + + for asset in workloads: + endpoints = endpoints_from_asset("workload", asset) + finding.unsaved_endpoints += endpoints + + for asset in images: + endpoints = endpoints_from_asset("image", asset) + finding.unsaved_endpoints += endpoints + + for asset in platforms: + endpoints = endpoints_from_asset("platform", asset) + finding.unsaved_endpoints += endpoints return finding +def endpoints_from_asset(kind, asset): + endpoints = [] + + # usually, there is only one namespace (domain, as NeuVector name it) + namespaces = asset.get("domains", []) + + name = asset.get("display_name", "") + + if kind == "workload": + # only workload assets have 'service' field + service = asset.get("service", "unknown_service") + name += f"/{service}" + + # in principle, naming follows the approach chosen for trivy parser + endpoints.append(Endpoint( + # host needs to comply with domain name syntax, we just expect that + # there will be only one namespace + host='-'.join(namespaces), + # we abuse path to have as much details as possible + path=f"{kind}/{name}", + )) + + # if it is a workload and it has an associated image, add image as a + # separate endpoint + if kind == "workload" and asset.get("image", "") != "": + image = asset.get("image", "unknown_image") + # image value example: + # someregistry.com/bitnami/postgresql:11.21.0-debian-11-r58 + artifact_and_tag = image.split("/")[-1] + # extracting only image name, without tag or digest + artifact_name = artifact_and_tag.split("@")[0] + artifact_name = artifact_name.split(":")[0] + + endpoints.append(Endpoint( + host=f"{artifact_name}", + path=f"{image}", + )) + + return endpoints + # see neuvector/share/types.go def convert_severity(severity): if severity.lower() == "critical":