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()