dojo/tools/generic/json_parser.py - master - Art86/DefectDojo-DevSecOps

dojo/tools/generic/json_parser.py
180 строк7 KB
dogboat
LocationData for parsers (#14395)
02 мар 2026, 09:08
02 мар 2026, 09:08597a1a8
import base64

import dateutil
from django.conf import settings
from django.core.files.base import ContentFile

from dojo.models import Endpoint, FileUpload, Finding
from dojo.tools.locations import LocationData
from dojo.tools.parser_test import ParserTest


class GenericJSONParser:
    ID = "Generic Findings Import"

    def _get_test_json(self, data):
        test_internal = ParserTest(
            name=data.get("name", self.ID),
            parser_type=data.get("type", self.ID),
            version=data.get("version"),
            description=data.get("description"),
            dynamic_tool=data.get("dynamic_tool"),
            static_tool=data.get("static_tool"),
            soc=data.get("soc"),
        )
        test_internal.findings = []
        for item in data.get("findings", []):
            # remove endpoints from the dictionary
            unsaved_locations = None
            if "endpoints" in item:
                unsaved_locations = item["endpoints"]
                del item["endpoints"]
            # remove files from the dictionary
            unsaved_files = None
            if "files" in item:
                unsaved_files = item["files"]
                del item["files"]
            # remove tags from the dictionary
            unsaved_tags = None
            if "tags" in item:
                unsaved_tags = item["tags"]
                del item["tags"]
            # remove vulnerability_ids from the dictionary
            unsaved_vulnerability_ids = None
            if "vulnerability_ids" in item:
                unsaved_vulnerability_ids = item["vulnerability_ids"]
                del item["vulnerability_ids"]
            # check for required keys
            required = {"title", "severity", "description"}

            if "date" in item:
                item["date"] = dateutil.parser.parse(item["date"]).date()

            if "mitigated" in item:
                item["mitigated"] = dateutil.parser.parse(item["mitigated"])

            missing = sorted(required.difference(item))
            if missing:
                msg = f"Required fields are missing: {missing}"
                raise ValueError(msg)

            # check for allowed keys
            allowed = {
                "date",
                "cwe",
                "cve",
                "epss_score",
                "epss_percentile",
                "cvssv3",
                "cvssv3_score",
                "cvssv4",
                "cvssv4_score",
                "mitigation",
                "impact",
                "steps_to_reproduce",
                "severity_justification",
                "references",
                "active",
                "verified",
                "false_p",
                "out_of_scope",
                "risk_accepted",
                "under_review",
                "is_mitigated",
                "thread_id",
                "mitigated",
                "numerical_severity",
                "param",
                "payload",
                "line",
                "file_path",
                "component_name",
                "component_version",
                "static_finding",
                "dynamic_finding",
                "scanner_confidence",
                "unique_id_from_tool",
                "vuln_id_from_tool",
                "sast_source_object",
                "sast_sink_object",
                "sast_source_line",
                "sast_source_file_path",
                "nb_occurences",
                "publish_date",
                "service",
                "planned_remediation_date",
                "planned_remediation_version",
                "effort_for_fixing",
                "tags",
                "kev_date",
                "known_exploited",
                "ransomware_used",
                "fix_available",
                "fix_version",
            }.union(required)
            not_allowed = sorted(set(item).difference(allowed))
            if not_allowed:
                msg = f"Not allowed fields are present: {not_allowed}"
                raise ValueError(msg)
            finding = Finding(**item)

            # manage endpoints
            if unsaved_locations:
                if settings.V3_FEATURE_LOCATIONS:
                    for location_item in unsaved_locations:
                        if isinstance(location_item, str):
                            if "://" in location_item:  # is the host full uri?
                                location = LocationData.url(url=location_item)
                            else:
                                location = LocationData.url(url="//" + location_item)
                        else:
                            location = LocationData.url(**location_item)
                        finding.unsaved_locations.append(location)
                else:
                    # TODO: Delete this after the move to Locations
                    for endpoint_item in unsaved_locations:
                        if isinstance(endpoint_item, str):
                            if "://" in endpoint_item:  # is the host full uri?
                                endpoint = Endpoint.from_uri(endpoint_item)
                                # can raise exception if the host is not valid URL
                            else:
                                endpoint = Endpoint.from_uri("//" + endpoint_item)
                                # can raise exception if there is no way to parse
                                # the host
                        else:
                            endpoint = Endpoint(**endpoint_item)
                        finding.unsaved_endpoints.append(endpoint)
            if settings.V3_FEATURE_LOCATIONS:
                component_name = item.get("component_name")
                component_version = item.get("component_version")
                file_path = item.get("file_path")
                if component_name or component_version or file_path:
                    finding.unsaved_locations.append(
                        LocationData.dependency(
                            name=component_name,
                            version=component_version,
                            file_path=file_path,
                        ),
                    )
            if unsaved_files:
                for unsaved_file in unsaved_files:
                    data = base64.b64decode(unsaved_file.get("data"))
                    title = unsaved_file.get("title", "<No title>")
                    FileUpload(title=title, file=ContentFile(data)).clean()

                finding.unsaved_files = unsaved_files
            if unsaved_tags:
                finding.unsaved_tags = unsaved_tags
            if finding.cve:
                finding.unsaved_vulnerability_ids = [finding.cve]
            if unsaved_vulnerability_ids:
                if finding.unsaved_vulnerability_ids:
                    finding.unsaved_vulnerability_ids.append(
                        unsaved_vulnerability_ids,
                    )
                else:
                    finding.unsaved_vulnerability_ids = (
                        unsaved_vulnerability_ids
                    )
            test_internal.findings.append(finding)
        return test_internal