vulnerabilityToStix2.py

import datetime

import stix2
from pycti import Identity, StixCoreRelationship, Vulnerability  # type: ignore
from services.utils import APP_VERSION, ConfigCVE  # type: ignore

from ..client import CVEVulnerability  # type: ignore


class CVEConverter:
    def __init__(self, helper):
        self.config = ConfigCVE()
        self.helper = helper
        self.client_api = CVEVulnerability(
            api_key=self.config.api_key,
            helper=self.helper,
            header=f"OpenCTI-cve/{APP_VERSION}",
        )
        self.author = self._create_author()

    def send_bundle(self, cve_params: dict, work_id: str) -> None:
        """
        Send bundle to API
        :param cve_params: Dict of params
        :param work_id: work id in string
        :return:
        """
        vulnerabilities_objects = self.vulnerabilities_to_stix2(cve_params)

        if len(vulnerabilities_objects) != 0:
            vulnerabilities_objects.append(self.author)
            vulnerabilities_bundle = self._to_stix_bundle(vulnerabilities_objects)
            vulnerabilities_to_json = self._to_json_bundle(vulnerabilities_bundle)

            # Retrieve the author object for the info message
            info_msg = (
                f"[CONVERTER] Sending bundle to server with {len(vulnerabilities_bundle)} objects, "
                f"concerning {len(vulnerabilities_objects) - 1} vulnerabilities"
            )
            self.helper.log_info(info_msg)

            self.helper.send_stix2_bundle(
                vulnerabilities_to_json,
                work_id=work_id,
            )
        else:
            pass

    def vulnerabilities_to_stix2(self, cve_params: dict) -> list:
        """
        Retrieve all CVEs from NVD to convert into STIX2 format
        :param cve_params: Dict of params
        :return: List of data converted into STIX2
        """
        vulnerabilities = self.client_api.get_vulnerabilities(cve_params)

        vulnerabilities_to_stix2 = []

        for vulnerability in vulnerabilities:
            # Getting different fields
            name = vulnerability["cve"]["id"]
            description = vulnerability["cve"]["descriptions"][0]["value"]
            created_date = datetime.datetime.strptime(
                vulnerability["cve"]["published"], "%Y-%m-%dT%H:%M:%S.%f"
            )
            modified_date = datetime.datetime.strptime(
                vulnerability["cve"]["lastModified"], "%Y-%m-%dT%H:%M:%S.%f"
            )

            # Create external references
            external_reference = stix2.ExternalReference(
                source_name="NIST NVD", url=f"https://nvd.nist.gov/vuln/detail/{name}"
            )
            external_references = [external_reference]

            if "references" in vulnerability["cve"]:
                for reference in vulnerability["cve"]["references"]:
                    external_reference = stix2.ExternalReference(
                        source_name=reference["source"], url=reference["url"]
                    )
                    external_references.append(external_reference)

            # Getting metrics based on CVSS V3.1
            cvss_metrics = vulnerability["cve"]["metrics"]["cvssMetricV31"][0][
                "cvssData"
            ]
            attack_vector = cvss_metrics["attackVector"]
            availability_impact = cvss_metrics["availabilityImpact"]
            base_score = cvss_metrics["baseScore"]
            base_severity = cvss_metrics["baseSeverity"]
            confidentiality_impact = cvss_metrics["confidentialityImpact"]
            integrity_impact = cvss_metrics["integrityImpact"]

            # Creating the vulnerability with the extracted fields
            vulnerability_to_stix2 = stix2.Vulnerability(
                id=Vulnerability.generate_id(name),
                name=name,
                created=created_date,
                modified=modified_date,
                description=description,
                created_by_ref=self.author,
                confidence=(
                    100 if description is not None and len(description) > 0 else 60
                ),
                external_references=external_references,
                custom_properties={
                    "x_opencti_base_score": base_score,
                    "x_opencti_base_severity": base_severity,
                    "x_opencti_attack_vector": attack_vector,
                    "x_opencti_integrity_impact": integrity_impact,
                    "x_opencti_availability_impact": availability_impact,
                    "x_opencti_confidentiality_impact": confidentiality_impact,
                },
            )

            # Adding the vulnerability to the list of vulnerabilities
            vulnerabilities_to_stix2.append(vulnerability_to_stix2)

        return vulnerabilities_to_stix2

    def _create_relationship(self, from_id: str, to_id: str, relation):
        """
        :param from_id: From id in string
        :param to_id: To id in string
        :param relation:
        :return: Relationship STIX object
        """
        return stix2.Relationship(
            id=StixCoreRelationship.generate_id(relation, from_id, to_id),
            relationship_type=relation,
            source_ref=from_id,
            target_ref=to_id,
            created_by_ref=self.author.id,
        )

    @staticmethod
    def _create_author():
        """
        :return: CVEs' default author
        """
        return stix2.Identity(
            id=Identity.generate_id("The MITRE Corporation", "organization"),
            name="The MITRE Corporation",
            identity_class="organization",
        )

    @staticmethod
    def _to_stix_bundle(stix_objects):
        """
        :return: STIX objects as a Bundle
        """
        return stix2.Bundle(objects=stix_objects, allow_custom=True)

    @staticmethod
    def _to_json_bundle(stix_bundle):
        """
        :return: STIX bundle as JSON format
        """
        return stix_bundle.serialize()