StixCreator

Source

import dateparser
import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401

""" IMPORTS """
import json
from collections.abc import Callable
from typing import Any

from stix2 import (
    AttackPattern,
    Bundle,
    Campaign,
    CourseOfAction,
    ExternalReference,
    KillChainPhase,
    Indicator,
    Infrastructure,
    IntrusionSet,
    Malware,
    Report,
    ThreatActor,
    Tool,
    Vulnerability,
)
from stix2.exceptions import InvalidValueError, MissingPropertiesError

SCOs: dict[str, str] = {  # pragma: no cover
    "md5": "file:hashes.md5",
    "sha1": "file:hashes.sha1",
    "sha256": "file:hashes.sha256",
    "ssdeep": "file:hashes.ssdeep",
    "ip": "ipv4-addr:value",
    "cidr": "ipv4-addr:value",
    "ipv6": "ipv6-addr:value",
    "ipv6cidr": "ipv6-addr:value",
    "url": "url:value",
    "email": "email-message:sender_ref.value",
    "account": "user-account:account_login",
    "domain": "domain-name:value",
    "host": "domain-name:value",
    "registry key": "windows-registry-key:key",
    "asn": "autonomous-system:name",
    "software": "software:name",
}

SDOs: dict[str, Callable] = {  # pragma: no cover
    "malware": Malware,
    "attack pattern": AttackPattern,
    "campaign": Campaign,
    "infrastructure": Infrastructure,
    "tool": Tool,
    "intrusion set": IntrusionSet,
    "report": Report,
    "threat actor": ThreatActor,
    "cve": Vulnerability,
    "course of action": CourseOfAction,
}


def search_related_indicators(value: str) -> list[dict]:  # pragma: no cover
    relationships = demisto.searchRelationships({"entities": [value]}).get("data", [])
    demisto.debug(f"found {len(relationships)} relationships")
    query = ""
    for rel in relationships:
        entity_a = rel.get("entityA", "").lower()
        entity_b = rel.get("entityB", "").lower()
        if entity_a == value.lower():
            query += f'"{entity_b}"'
        elif entity_b == value.lower():
            query += f'"{entity_a}"'
        else:
            demisto.info(f"Relationship: {rel} is not relevant for indicator: {value}")
            continue
        query += " or "
    if not query:
        demisto.info(f"No relevant relationship found for indicator: {value}")
        return []
    query = query[:-4]
    demisto.debug(f"using query: {query}")
    demisto_indicators = demisto.searchIndicators(query=query).get("iocs", [])
    demisto.debug(f"found {len(demisto_indicators)} related indicators")
    return demisto_indicators


def get_indicators_stix_ids(value: str, indicator_type: str, indicators: list[dict]) -> list[str]:
    stix_ids = []
    for indicator in indicators:
        if stix_id := indicator.get("stixid"):
            demisto.debug(f"Found stix id: {stix_id} for indicator: {indicator}")
            stix_ids.append(stix_id)
        else:
            if indicator_type in SDOs:
                stix_type = XSOAR_TYPES_TO_STIX_SDO.get(indicator.get("indicator_type", "indicator"))
                stix_id = XSOAR2STIXParser.create_sdo_stix_uuid(indicator, stix_type, PAWN_UUID, indicator.get("value", ""))
            elif indicator_type in SCOs:
                stix_type = XSOAR_TYPES_TO_STIX_SCO.get(indicator.get("indicator_type", "indicator"), "indicator")
                stix_id = XSOAR2STIXParser.create_sco_stix_uuid(indicator, stix_type, indicator.get("value", ""))
            else:
                demisto.info(f"Indicator type: {indicator_type}, with the value: {value} is unknown.")
                continue
            demisto.debug(f"Created stix id: {stix_id} for indicator: {indicator}")
        stix_ids.append(stix_id)
    return stix_ids


def hash_type(value: str) -> str:  # pragma: no cover
    length = len(value)
    if length == 32:
        return "md5"
    if length == 40:
        return "sha1"
    if length == 64 and ":" in value:
        return "ssdeep"
    elif length == 64:
        return "sha256"
    if length == 128:
        return "sha512"
    return ""


def guess_indicator_type(type_: str, val: str) -> str:
    # try to guess by key
    for sco in SCOs:
        if sco in type_:
            return sco

    # try to auto_detect by value
    return (auto_detect_indicator_type(val) or type_).lower()


def add_file_fields_to_indicator(xsoar_indicator: Dict, value: str) -> Dict:
    """
    Create the hashes dictionary for the indicator object.
    Args:
        xsoar_indicator: Dict - The XSOAR representation of the indicator.
        value: str - The value of the indicator.
    Returns:
        The dictionary with the file hashes.
    """
    hashes_dict = {}
    for hash_kind in ["md5", "sha1", "sha256", "sha512"]:
        if get_hash_type(value) == hash_kind:
            hashes_dict[HASH_TYPE_TO_STIX_HASH_TYPE.get(hash_kind)] = value
        elif hash_kind in xsoar_indicator:
            hashes_dict[HASH_TYPE_TO_STIX_HASH_TYPE.get(hash_kind)] = xsoar_indicator.get(hash_kind, "")
    return hashes_dict


def create_stix_sco_indicator(stix_id: Optional[str], stix_type: Optional[str], value: str, xsoar_indicator: Dict) -> Dict:
    """
    Create stix sco indicator object.
    Args:
        stix_id: Optional[str] - The stix id of the indicator.
        stix_type: str - the stix type of the indicator.
        xsoar_indicator: Dict - The XSOAR representation of the indicator.
        value: str - The value of the indicator.
    Returns:
        The Dictionary representing the stix indicator.
    """
    stix_indicator: Dict[str, Any] = {"type": stix_type, "spec_version": "2.1", "id": stix_id}
    if stix_type == "file":
        stix_indicator["hashes"] = add_file_fields_to_indicator(xsoar_indicator, value)
    elif stix_type == "autonomous-system":
        stix_indicator["number"] = value
        stix_indicator["name"] = xsoar_indicator.get("name", "")
    elif stix_type == "software":
        # Software type requires 'name' as the primary field
        stix_indicator["name"] = value
        # Add optional software fields if available in CustomFields
        custom_fields = xsoar_indicator.get("CustomFields") or {}
        if vendor := custom_fields.get("vendor"):
            stix_indicator["vendor"] = vendor
        if version := custom_fields.get("version"):
            stix_indicator["version"] = version
        if cpe := custom_fields.get("cpe"):
            stix_indicator["cpe"] = cpe
    else:
        stix_indicator["value"] = value
    return stix_indicator


def main():
    user_args = demisto.args().get("indicators", "Unknown")
    doubleBackslash = demisto.args().get("doubleBackslash", True)
    is_sco = argToBoolean(demisto.args().get("sco_flag", False))
    all_args = {}

    if isinstance(user_args, dict):
        all_args = json.loads(json.dumps(user_args))

    else:
        try:
            all_args = json.loads(demisto.args().get("indicators", "Unknown"))
        except:  # noqa: E722
            return_error("indicators argument is invalid json object")

    indicators = []

    for indicator_fields in all_args:
        kwargs: dict[str, Any] = {"allow_custom": True}

        xsoar_indicator = all_args[indicator_fields]
        demisto_indicator_type = xsoar_indicator.get("indicator_type", "Unknown")

        value = xsoar_indicator.get("value", "").replace("\\", "\\\\") if doubleBackslash else xsoar_indicator.get("value", "")

        if demisto_indicator_type in XSOAR_TYPES_TO_STIX_SCO and is_sco:
            stix_type = XSOAR_TYPES_TO_STIX_SCO.get(demisto_indicator_type)
            stix_id = XSOAR2STIXParser.create_sco_stix_uuid(xsoar_indicator, stix_type, value)
            stix_indicator = create_stix_sco_indicator(stix_id, stix_type, value, xsoar_indicator)
            indicators.append(stix_indicator)

        else:
            demisto_score = xsoar_indicator.get("score", "").lower()

            if demisto_score in ["bad", "malicious"]:
                kwargs["score"] = "High"

            elif demisto_score == "suspicious":
                kwargs["score"] = "Medium"

            elif demisto_score in ["good", "benign"]:
                kwargs["score"] = "None"

            else:
                kwargs["score"] = "Not Specified"

            stix_type = XSOAR_TYPES_TO_STIX_SDO.get(demisto_indicator_type, "indicator")
            stix_id = XSOAR2STIXParser.create_sdo_stix_uuid(xsoar_indicator, stix_type, PAWN_UUID, value)
            kwargs["id"] = stix_id

            kwargs["created"] = dateparser.parse(xsoar_indicator.get("timestamp", ""))
            kwargs["modified"] = dateparser.parse(xsoar_indicator.get("lastSeen", f'{kwargs["created"]}'))
            kwargs["labels"] = [demisto_indicator_type.lower()]
            kwargs["description"] = xsoar_indicator.get("description", "")

            kwargs = {k: v for k, v in kwargs.items() if v}  # Removing keys with empty strings

            try:
                indicator_type = demisto_indicator_type.lower().replace("-", "")
                if indicator_type == "file":
                    indicator_type = hash_type(value)
                if indicator_type not in SCOs and indicator_type not in SDOs:
                    indicator_type = guess_indicator_type(indicator_type, value)
                demisto.debug(f"Creating an indicator with the following pattern: [{SCOs[indicator_type]} = '{value}']")

                indicator = Indicator(pattern=f"[{SCOs[indicator_type]} = '{value}']", pattern_type="stix", **kwargs)
                indicators.append(indicator)

            except KeyError:
                demisto.debug(f"{demisto_indicator_type} isn't a SCO, checking other IOC types")

                try:
                    indicator_type = demisto_indicator_type.lower()

                    if indicator_type == "cve":
                        kwargs["external_references"] = [ExternalReference(source_name="cve", external_id=value)]

                    elif indicator_type == "attack pattern":
                        try:
                            mitreid = xsoar_indicator.get("mitreid", "")
                            if mitreid:
                                kwargs["external_references"] = [ExternalReference(source_name="mitre", external_id=mitreid)]

                            # Add kill-chain phases if available
                            kill_chain_phases = xsoar_indicator.get("killchainphases")
                            if kill_chain_phases:
                                # Convert XSOAR kill chain phase names to STIX format
                                # XSOAR stores human-readable names like "Defense Evasion"
                                # STIX requires objects with kill_chain_name and phase_name (lowercase with hyphens)
                                # Docs for this can be found here: https://docs.oasis-open.org/cti/stix/v2.1/os/stix-v2.1-os.html#_i4tjv75ce50h
                                stix_kill_chain_phases = []

                                # Handle both list and comma-separated string formats
                                if isinstance(kill_chain_phases, str):
                                    kill_chain_phases = [phase.strip() for phase in kill_chain_phases.split(",")]

                                for phase in kill_chain_phases:
                                    # Strip whitespace and convert human-readable name to lowercase with hyphens
                                    # e.g., " Defense Evasion" -> "defense-evasion"
                                    phase_name = phase.strip().lower().replace(" ", "-")
                                    stix_kill_chain_phases.append(
                                        KillChainPhase(kill_chain_name="mitre-attack", phase_name=phase_name)
                                    )
                                kwargs["kill_chain_phases"] = stix_kill_chain_phases
                        except KeyError:
                            pass

                    elif indicator_type == "malware":
                        kwargs["is_family"] = argToBoolean(xsoar_indicator.get("ismalwarefamily", "False").lower())

                    if indicator_type == "report":
                        kwargs["published"] = dateparser.parse(xsoar_indicator.get("timestamp", ""))

                        related_indicators = search_related_indicators(value)
                        stix_ids = get_indicators_stix_ids(value, indicator_type, related_indicators)
                        kwargs["object_refs"] = stix_ids
                    demisto.debug(f"Creating {indicator_type} indicator: {value}, with the following kwargs: {kwargs}")
                    indicator = SDOs[indicator_type](name=value, **kwargs)

                    indicators.append(indicator)

                except (KeyError, TypeError):
                    demisto.info(f"Indicator type: {demisto_indicator_type}, with the value: {value} is not STIX compatible")
                    demisto.info(f"Export failure exception: {traceback.format_exc()}")
                    continue

                except (InvalidValueError, MissingPropertiesError):
                    demisto.info(
                        f"Indicator type: {demisto_indicator_type}, with the value: {value} is not STIX compatible. Skipping."
                    )
                    demisto.info(f"Export failure exception: {traceback.format_exc()}")
                    continue

            except (InvalidValueError, MissingPropertiesError):
                demisto.info(
                    f"Indicator type: {demisto_indicator_type}, with the value: {value} is not STIX compatible. Skipping."
                )
                demisto.info(f"Export failure exception: {traceback.format_exc()}")
                continue
    if len(indicators) > 1:
        bundle = Bundle(indicators, allow_custom=True, spec_version="2.1")
        context = {"StixExportedIndicators(val.pattern && val.pattern == obj.pattern)": json.loads(str(bundle))}
        res = CommandResults(readable_output="", outputs=context, raw_response=str(bundle))

    elif len(indicators) == 1:
        bundle = Bundle(indicators, allow_custom=True)
        bundle_obj = bundle.get("objects", [])[0]
        context = {"StixExportedIndicators(val.pattern && val.pattern == obj.pattern)": json.loads(str(bundle_obj))}
        res = CommandResults(readable_output="", outputs=context, raw_response=str(bundle_obj))
    else:
        context = {"StixExportedIndicators": {}}
        res = CommandResults(readable_output="", outputs=context, raw_response={})

    return_results(res)


from TAXII2ApiModule import *  # noqa: E402

if __name__ in ("__builtin__", "builtins", "__main__"):
    main()

README