dojo/tools/bugcrowd/parser.py - master - Art86/DefectDojo-DevSecOps

dojo/tools/bugcrowd/parser.py
261 строка9 KB
dogboat
LocationData for parsers (#14395)
02 мар 2026, 09:08
02 мар 2026, 09:08597a1a8
import csv
import hashlib
import io

from dateutil import parser
from django.conf import settings

from dojo.models import Endpoint, Finding
from dojo.tools.locations import LocationData


class BugCrowdParser:
    def get_scan_types(self):
        return ["BugCrowd Scan"]

    def get_label_for_scan_types(self, scan_type):
        return "BugCrowd Scan"

    def get_description_for_scan_types(self, scan_type):
        return "BugCrowd CSV report format"

    def get_findings(self, filename, test):
        if filename is None:
            return ()

        content = filename.read()
        if isinstance(content, bytes):
            content = content.decode("utf-8")
        reader = csv.DictReader(
            io.StringIO(content), delimiter=",", quotechar='"',
        )
        csvarray = []

        for row in reader:
            csvarray.append(row)

        dupes = {}
        for row in csvarray:
            finding = Finding(test=test)

            url = row.get("bug_url", None)
            pre_description = self.split_description(
                row.get("description", None),
            )
            Description = (
                pre_description.get("description", "")
                + "\n\n"
                + pre_description.get("poc", "")
            )
            Description += (
                row.get("extra_info") + "\n\n"
                if row.get("extra_info", None)
                else ""
            )
            Description += (
                "BugCrowd Reference Number: "
                + row.get("reference_number")
                + "\n"
                if row.get("reference_number", None)
                else ""
            )
            Description += "Bug URL: " + url + "\n" if url else ""
            Description += (
                "Bug Source: " + row.get("source") + "\n"
                if row.get("source", None)
                else ""
            )
            Description += (
                "BugCrowd User: " + row.get("username") + "\n"
                if row.get("username", None)
                else ""
            )
            Description += (
                "BugCrowd Payout: " + row.get("amount") + "\n"
                if row.get("amount", None)
                else ""
            )
            Description += (
                "Submitted at: " + row.get("submitted_at") + "\n"
                if row.get("submitted_at", None)
                else ""
            )
            Description += (
                "Validated at: " + row.get("validated_at") + "\n"
                if row.get("validated_at", None)
                else ""
            )
            Description += (
                "Closed at: " + row.get("closed_at") + "\n"
                if row.get("closed_at", None)
                else ""
            )
            Description += (
                "Target name: " + row.get("target_name") + "\n"
                if row.get("target_name", None)
                else ""
            )
            Description += (
                "Target category: " + row.get("target_category") + "\n"
                if row.get("target_category", None)
                else ""
            )
            References = (
                "BugCrowd Reference Number: "
                + row.get("reference_number")
                + "\n"
                if row.get("reference_number", None)
                else ""
            )
            References += row.get("vulnerability_ids", "")

            finding.title = row.get("title", "")
            finding.description = Description
            finding.mitigation = (
                pre_description.get("mitigation", "")
                + "\n"
                + row.get("remediation_advice", "")
            )
            finding.impact = (
                pre_description.get("impact", "")
                + "\n"
                + row.get("vrt_lineage", "")
            )
            finding.steps_to_reproduce = pre_description.get(
                "steps_to_reproduce", None,
            )
            finding.references = References
            finding.severity = self.convert_severity(row.get("priority", 0))

            if row.get("submitted_at"):
                finding.date = parser.parse(row.get("submitted_at"))

            if url:
                if settings.V3_FEATURE_LOCATIONS:
                    finding.unsaved_locations = [self.get_location(url)]
                else:
                    # TODO: Delete this after the move to Locations
                    finding.unsaved_endpoints = [self.get_location(url)]

            if finding is not None:
                if finding.title is None:
                    finding.title = ""
                if finding.description is None:
                    finding.description = ""

                key = hashlib.md5(
                    (finding.title + "|" + finding.description).encode("utf-8"), usedforsecurity=False,
                ).hexdigest()

                if key not in dupes:
                    dupes[key] = finding

        return list(dupes.values())

    def description_parse(self, ret):
        items = ["impact", "steps to reproduce:", "steps to reproduce", "poc"]
        items_in = [i for i in items if i in ret["description"].lower()]
        if len(items_in) == 0:
            return ret

        impact = steps = poc = skip = 0
        lines = [
            line
            for line in ret["description"].replace("#", "").split("\n")
            if line
        ]
        ret["description"] = ""
        for line in lines:
            lower_line = line.lower().strip()
            if lower_line == "impact":
                ret["impact"] = "### Impact\n" + ret.get("impact", "")
                impact = skip = 1
                steps = poc = 0
            elif (
                lower_line in "steps to reproduce:"
                or lower_line == "steps to reproduce"
            ):
                ret[
                    "steps_to_reproduce"
                ] = "### Steps To Reproduce\n" + ret.get(
                    "imsteps_to_reproducepact", "",
                )
                steps = skip = 1
                poc = impact = 0
            elif lower_line == "poc":
                ret["poc"] = "### PoC Code\n" + ret.get("poc", "")
                poc = skip = 1
                impact = steps = 0

            if not skip:
                if steps:
                    ret["steps_to_reproduce"] += line + "\n"
                elif impact:
                    ret["impact"] += line + "\n"
                elif poc:
                    ret["poc"] += line + "\n"
                else:
                    ret["description"] += line + "\n"
            skip = 0
        return ret

    def split_description(self, description):
        ret = {}
        if description is None or not description:
            return ret

        split_des = description.split("---")
        ret["description"] = ""
        for item in split_des:
            lines = [
                line.strip()
                for line in "".join(item.split("#")).splitlines()
                if line
            ]
            if lines:
                first = lines[0].strip()
                if first == "Impact":
                    ret["impact"] = item
                elif first == "Steps to reproduce":
                    ret["steps_to_reproduce"] = item
                elif first in {"How to fix", "Fix"}:
                    ret["mitigation"] = item
                elif first == "PoC code":
                    ret["poc"] = item
                else:
                    ret["description"] += ret["description"] + item

        ret = self.description_parse(ret)

        if not ret["description"]:
            ret["description"] = description

        return ret

    def convert_severity(self, sev_num):
        # Attempt to convert to an int
        try:
            sev_num = int(sev_num)
        except ValueError:
            sev_num = 0
        severity = "Info"
        if sev_num == 1:
            severity = "Critical"
        elif sev_num == 2:
            severity = "High"
        elif sev_num == 3:
            severity = "Medium"
        elif sev_num == 4:
            severity = "Low"
        else:
            # If the arg is an unexpected value, leave it as "Info"
            pass
        return severity

    def get_location(self, url):
        stripped_url = url.strip()
        # is the host full uri?
        # TODO: Delete this after the move to Locations
        if not settings.V3_FEATURE_LOCATIONS:
            return Endpoint.from_uri(stripped_url) if "://" in stripped_url else Endpoint.from_uri("//" + stripped_url)
        return LocationData.url(url=stripped_url) if "://" in stripped_url else LocationData.url(url="//" + stripped_url)