import collections import copy import tempfile import dateutil.parser import demistomock as demisto # noqa: F401 import pytz from bs4 import BeautifulSoup from CommonServerPython import * # noqa: F401 from lxml import etree from netaddr import IPNetwork from six import string_types EPOCH = datetime.utcfromtimestamp(0).replace(tzinfo=pytz.UTC) SCRIPT_NAME = "STIXParser" # CONSTANTS TAXII_VER_2_0 = "2.0" TAXII_VER_2_1 = "2.1" DFLT_LIMIT_PER_REQUEST = 100 API_USERNAME = "_api_token_key" HEADER_USERNAME = "_header:" XSOAR_TAXII2_SERVER_SCHEMA = ( "https://github.com/demisto/content/blob/4265bd5c71913cd9d9ed47d9c37d0d4d3141c3eb/" "Packs/TAXIIServer/doc_files/XSOAR_indicator_schema.json" ) SYSTEM_FIELDS = [ "id", "version", "modified", "sortValues", "timestamp", "indicator_type", "value", "sourceInstances", "sourceBrands", "investigationIDs", "lastSeen", "firstSeen", "firstSeenEntryID", "score", "insightCache", "moduleToFeedMap", "expirationStatus", "expirationSource", "calculatedTime", "lastReputationRun", "modifiedTime", "aggregatedReliability", ] ERR_NO_COLL = "No collection is available for this user, please make sure you entered the configuration correctly" DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ" # Pattern Regexes - used to extract indicator type and value, spaces are removed before matching the following regexes INDICATOR_OPERATOR_VAL_FORMAT_PATTERN = r"(\w.*?{value}{operator})'(.*?)'" INDICATOR_IN_VAL_PATTERN = r"(\w.*?valueIN)\(+('.*?')\)" INDICATOR_EQUALS_VAL_PATTERN = INDICATOR_OPERATOR_VAL_FORMAT_PATTERN.format(value="value", operator="=") CIDR_ISSUBSET_VAL_PATTERN = INDICATOR_OPERATOR_VAL_FORMAT_PATTERN.format(value="value", operator="ISSUBSET") CIDR_ISUPPERSET_VAL_PATTERN = INDICATOR_OPERATOR_VAL_FORMAT_PATTERN.format(value="value", operator="ISSUPPERSET") HASHES_EQUALS_VAL_PATTERN = INDICATOR_OPERATOR_VAL_FORMAT_PATTERN.format(value=r"hashes\..*?", operator="=") REGISTRY_EQUALS_VAL_PATTERN = INDICATOR_OPERATOR_VAL_FORMAT_PATTERN.format(value="key", operator="=") TAXII_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ" STIX_2_TYPES_TO_CORTEX_TYPES = { "mutex": FeedIndicatorType.MUTEX, "windows-registry-key": FeedIndicatorType.Registry, "user-account": FeedIndicatorType.Account, "email-addr": FeedIndicatorType.Email, "autonomous-system": FeedIndicatorType.AS, "ipv4-addr": FeedIndicatorType.IP, "ipv6-addr": FeedIndicatorType.IPv6, "domain": FeedIndicatorType.Domain, "domain-name": FeedIndicatorType.Domain, "url": FeedIndicatorType.URL, "file": FeedIndicatorType.File, "md5": FeedIndicatorType.File, "sha-1": FeedIndicatorType.File, "sha-256": FeedIndicatorType.File, "file:hashes": FeedIndicatorType.File, "attack-pattern": ThreatIntel.ObjectsNames.ATTACK_PATTERN, "malware": ThreatIntel.ObjectsNames.MALWARE, "tool": ThreatIntel.ObjectsNames.TOOL, "report": ThreatIntel.ObjectsNames.REPORT, "threat-actor": ThreatIntel.ObjectsNames.THREAT_ACTOR, "course-of-action": ThreatIntel.ObjectsNames.COURSE_OF_ACTION, "campaign": ThreatIntel.ObjectsNames.CAMPAIGN, "infrastructure": ThreatIntel.ObjectsNames.INFRASTRUCTURE, "intrusion-set": ThreatIntel.ObjectsNames.INTRUSION_SET, } MITRE_CHAIN_PHASES_TO_DEMISTO_FIELDS = { "build-capabilities": ThreatIntel.KillChainPhases.BUILD_CAPABILITIES, "privilege-escalation": ThreatIntel.KillChainPhases.PRIVILEGE_ESCALATION, "adversary-opsec": ThreatIntel.KillChainPhases.ADVERSARY_OPSEC, "credential-access": ThreatIntel.KillChainPhases.CREDENTIAL_ACCESS, "exfiltration": ThreatIntel.KillChainPhases.EXFILTRATION, "lateral-movement": ThreatIntel.KillChainPhases.LATERAL_MOVEMENT, "defense-evasion": ThreatIntel.KillChainPhases.DEFENSE_EVASION, "persistence": ThreatIntel.KillChainPhases.PERSISTENCE, "collection": ThreatIntel.KillChainPhases.COLLECTION, "impact": ThreatIntel.KillChainPhases.IMPACT, "initial-access": ThreatIntel.KillChainPhases.INITIAL_ACCESS, "discovery": ThreatIntel.KillChainPhases.DISCOVERY, "execution": ThreatIntel.KillChainPhases.EXECUTION, "installation": ThreatIntel.KillChainPhases.INSTALLATION, "delivery": ThreatIntel.KillChainPhases.DELIVERY, "weaponization": ThreatIntel.KillChainPhases.WEAPONIZATION, "act-on-objectives": ThreatIntel.KillChainPhases.ACT_ON_OBJECTIVES, "command-and-control": ThreatIntel.KillChainPhases.COMMAND_AND_CONTROL, } STIX_2_TYPES_TO_CORTEX_CIDR_TYPES = { "ipv4-addr": FeedIndicatorType.CIDR, "ipv6-addr": FeedIndicatorType.IPv6CIDR, } THREAT_INTEL_TYPE_TO_DEMISTO_TYPES = { "campaign": ThreatIntel.ObjectsNames.CAMPAIGN, "attack-pattern": ThreatIntel.ObjectsNames.ATTACK_PATTERN, "report": ThreatIntel.ObjectsNames.REPORT, "malware": ThreatIntel.ObjectsNames.MALWARE, "course-of-action": ThreatIntel.ObjectsNames.COURSE_OF_ACTION, "intrusion-set": ThreatIntel.ObjectsNames.INTRUSION_SET, "tool": ThreatIntel.ObjectsNames.TOOL, "threat-actor": ThreatIntel.ObjectsNames.THREAT_ACTOR, "infrastructure": ThreatIntel.ObjectsNames.INFRASTRUCTURE, } def convert_to_json(string): """Will try to convert given string to json. Args: string: str of stix/json file. may be xml, then function will fail Returns: json object if succeed False if failed """ try: js = json.loads(string) return js except ValueError: return None class STIX2Parser: OBJECTS_TO_PARSE = [ "indicator", "report", "malware", "campaign", "attack-pattern", "course-of-action", "intrusion-set", "tool", "threat-actor", "infrastructure", "autonomous-system", "domain-name", "email-addr", "file", "ipv4-addr", "ipv6-addr", "mutex", "url", "user-account", "windows-registry-key", "relationship", "extension-definition", ] def __init__(self): """ TAXII 2 Client used to poll and parse indicators in XSOAR formar """ self.indicator_regexes = [ re.compile(INDICATOR_EQUALS_VAL_PATTERN), re.compile(INDICATOR_IN_VAL_PATTERN), re.compile(HASHES_EQUALS_VAL_PATTERN), re.compile(REGISTRY_EQUALS_VAL_PATTERN), ] self.cidr_regexes = [ re.compile(CIDR_ISSUBSET_VAL_PATTERN), re.compile(CIDR_ISUPPERSET_VAL_PATTERN), ] self.id_to_object: Dict[str, Any] = {} self.parsed_object_id_to_object: Dict[str, Any] = {} @staticmethod def get_indicator_publication(indicator: Dict[str, Any]): """ Build publications grid field from the indicator external_references field Args: indicator: The indicator with publication field Returns: list. publications grid field """ publications = [] for external_reference in indicator.get("external_references", []): url = external_reference.get("url", "") description = external_reference.get("description", "") source_name = external_reference.get("source_name", "") publications.append({"link": url, "title": description, "source": source_name}) return publications @staticmethod def change_attack_pattern_to_stix_attack_pattern(indicator: Dict[str, Any]): indicator["indicator_type"] = f'STIX {indicator["indicator_type"]}' indicator["customFields"]["stixkillchainphases"] = indicator["customFields"].pop("killchainphases", None) indicator["customFields"]["stixdescription"] = indicator["customFields"].pop("description", None) return indicator @staticmethod def get_ioc_type(indicator: str, id_to_object: Dict[str, Dict[str, Any]]) -> str: """ Get IOC type by extracting it from the pattern field. Args: indicator: the indicator to get information on. id_to_object: a dict in the form of - id: stix_object. Returns: str. the IOC type. """ ioc_type = "" indicator_obj = id_to_object.get(indicator, {}) pattern = indicator_obj.get("pattern", "") for stix_type in STIX_2_TYPES_TO_CORTEX_TYPES: if pattern.startswith(f"[{stix_type}"): ioc_type = STIX_2_TYPES_TO_CORTEX_TYPES.get(stix_type) # type: ignore break return ioc_type @staticmethod def change_ip_to_cidr(indicators): """ Iterates over indicators list and changes IP to CIDR type if needed. :param indicators: list of parsed indicators. :return: changes indicators list in-place. """ for indicator in indicators: if indicator.get("indicator_type") == FeedIndicatorType.IP: value = indicator.get("value") if value.endswith("/32"): pass elif "/" in value: indicator["indicator_type"] = FeedIndicatorType.CIDR """ PARSING FUNCTIONS""" def parse_indicator(self, indicator_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses a single indicator object :param indicator_obj: indicator object :return: indicators extracted from the indicator object in cortex format """ field_map: dict = {} pattern = indicator_obj.get("pattern") indicators = [] if pattern: # this is done in case the server doesn't properly space the operator, # supported indicators have no spaces, so this action shouldn't affect extracted values trimmed_pattern = pattern.replace(" ", "") indicator_groups = self.extract_indicator_groups_from_pattern(trimmed_pattern, self.indicator_regexes) indicators.extend( self.get_indicators_from_indicator_groups( indicator_groups, indicator_obj, STIX_2_TYPES_TO_CORTEX_TYPES, field_map, ) ) cidr_groups = self.extract_indicator_groups_from_pattern(trimmed_pattern, self.cidr_regexes) indicators.extend( self.get_indicators_from_indicator_groups( cidr_groups, indicator_obj, STIX_2_TYPES_TO_CORTEX_CIDR_TYPES, field_map, ) ) self.change_ip_to_cidr(indicators) return indicators @staticmethod def parse_attack_pattern(attack_pattern_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses a single attack pattern object :param attack_pattern_obj: attack pattern object :return: attack pattern extracted from the attack pattern object in cortex format """ publications = STIX2Parser.get_indicator_publication(attack_pattern_obj) kill_chain_mitre = [chain.get("phase_name", "") for chain in attack_pattern_obj.get("kill_chain_phases", [])] kill_chain_phases = [MITRE_CHAIN_PHASES_TO_DEMISTO_FIELDS.get(phase) for phase in kill_chain_mitre] # Extract MitreID from external_references mitre_id = "" for external_ref in attack_pattern_obj.get("external_references", []): if external_ref.get("source_name") == "mitre": mitre_id = external_ref.get("external_id", "") break attack_pattern = { "value": attack_pattern_obj.get("name"), "indicator_type": ThreatIntel.ObjectsNames.ATTACK_PATTERN, "score": ThreatIntel.ObjectsScore.ATTACK_PATTERN, "rawJSON": attack_pattern_obj, } fields = { "mitreid": mitre_id, "stixid": attack_pattern_obj.get("id"), "killchainphases": kill_chain_phases, "firstseenbysource": attack_pattern_obj.get("created"), "modified": attack_pattern_obj.get("modified"), "description": attack_pattern_obj.get("description", ""), "operatingsystemrefs": attack_pattern_obj.get("x_mitre_platforms"), "publications": publications, "tags": list(set(attack_pattern_obj.get("labels", []))), } attack_pattern["customFields"] = fields if not is_demisto_version_ge("6.2.0"): # For versions less than 6.2 - that only support STIX and not the newer types - Malware, Tool, etc. attack_pattern = STIX2Parser.change_attack_pattern_to_stix_attack_pattern(attack_pattern) return [attack_pattern] @staticmethod def parse_report(report_obj: Dict[str, Any]): """ Parses a single report object :param report_obj: report object :return: report extracted from the report object in cortex format """ object_refs = report_obj.get("object_refs", []) new_relationships = [] for obj_id in object_refs: new_relationships.append( { "type": "relationship", "id": "relationship--fakeid", "created": report_obj.get("created"), "modified": report_obj.get("modified"), "relationship_type": "contains", "source_ref": report_obj.get("id"), "target_ref": obj_id, } ) report = { "indicator_type": ThreatIntel.ObjectsNames.REPORT, "value": report_obj.get("name"), "score": ThreatIntel.ObjectsScore.REPORT, "rawJSON": report_obj, } fields = { "stixid": report_obj.get("id"), "firstseenbysource": report_obj.get("created"), "published": report_obj.get("published"), "description": report_obj.get("description", ""), "report_types": report_obj.get("report_types", []), "tags": list(set(report_obj.get("labels", []))), } report["customFields"] = fields return [report], new_relationships @staticmethod def parse_threat_actor(threat_actor_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses a single threat actor object :param threat_actor_obj: report object :return: threat actor extracted from the threat actor object in cortex format """ threat_actor = { "value": threat_actor_obj.get("name"), "indicator_type": ThreatIntel.ObjectsNames.THREAT_ACTOR, "score": ThreatIntel.ObjectsScore.THREAT_ACTOR, "rawJSON": threat_actor_obj, } fields = { "stixid": threat_actor_obj.get("id"), "firstseenbysource": threat_actor_obj.get("created"), "modified": threat_actor_obj.get("modified"), "description": threat_actor_obj.get("description", ""), "aliases": threat_actor_obj.get("aliases", []), "threat_actor_types": threat_actor_obj.get("threat_actor_types", []), "roles": threat_actor_obj.get("roles", []), "goals": threat_actor_obj.get("goals", []), "sophistication": threat_actor_obj.get("sophistication", ""), "resource_level": threat_actor_obj.get("resource_level", ""), "primary_motivation": threat_actor_obj.get("primary_motivation", ""), "secondary_motivations": threat_actor_obj.get("secondary_motivations", []), "tags": list(set(threat_actor_obj.get("labels", []))), } threat_actor["customFields"] = fields return [threat_actor] @staticmethod def parse_infrastructure(infrastructure_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses a single infrastructure object :param infrastructure_obj: infrastructure object :return: infrastructure extracted from the infrastructure object in cortex format """ kill_chain_mitre = [chain.get("phase_name", "") for chain in infrastructure_obj.get("kill_chain_phases", [])] kill_chain_phases = [MITRE_CHAIN_PHASES_TO_DEMISTO_FIELDS.get(phase) for phase in kill_chain_mitre] infrastructure = { "value": infrastructure_obj.get("name"), "indicator_type": ThreatIntel.ObjectsNames.INFRASTRUCTURE, "score": ThreatIntel.ObjectsScore.INFRASTRUCTURE, "rawJSON": infrastructure_obj, } fields = { "stixid": infrastructure_obj.get("id"), "description": infrastructure_obj.get("description", ""), "infrastructure_types": infrastructure_obj.get("infrastructure_types", []), "aliases": infrastructure_obj.get("aliases", []), "kill_chain_phases": kill_chain_phases, "firstseenbysource": infrastructure_obj.get("created"), "modified": infrastructure_obj.get("modified"), "tags": list(set(infrastructure_obj.get("labels", []))), } infrastructure["customFields"] = fields return [infrastructure] @staticmethod def parse_malware(malware_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses a single malware object :param malware_obj: malware object :return: malware extracted from the malware object in cortex format """ kill_chain_mitre = [chain.get("phase_name", "") for chain in malware_obj.get("kill_chain_phases", [])] kill_chain_phases = [MITRE_CHAIN_PHASES_TO_DEMISTO_FIELDS.get(phase) for phase in kill_chain_mitre] malware = { "value": malware_obj.get("name"), "indicator_type": ThreatIntel.ObjectsNames.MALWARE, "score": ThreatIntel.ObjectsScore.MALWARE, "rawJSON": malware_obj, } fields = { "stixid": malware_obj.get("id"), "firstseenbysource": malware_obj.get("created"), "modified": malware_obj.get("modified"), "description": malware_obj.get("description", ""), "malware_types": malware_obj.get("malware_types", []), "is_family": malware_obj.get("is_family", False), "aliases": malware_obj.get("aliases", []), "kill_chain_phases": kill_chain_phases, "os_execution_envs": malware_obj.get("os_execution_envs", []), "architecture_execution_envs": malware_obj.get("architecture_execution_envs", []), "capabilities": malware_obj.get("capabilities", []), "sample_refs": malware_obj.get("sample_refs", []), "tags": list(set(malware_obj.get("labels", []))), } malware["customFields"] = fields return [malware] @staticmethod def parse_tool(tool_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses a single tool object :param tool_obj: tool object :return: tool extracted from the tool object in cortex format """ kill_chain_mitre = [chain.get("phase_name", "") for chain in tool_obj.get("kill_chain_phases", [])] kill_chain_phases = [MITRE_CHAIN_PHASES_TO_DEMISTO_FIELDS.get(phase) for phase in kill_chain_mitre] tool = { "value": tool_obj.get("name"), "indicator_type": ThreatIntel.ObjectsNames.TOOL, "score": ThreatIntel.ObjectsScore.TOOL, "rawJSON": tool_obj, } fields = { "stixid": tool_obj.get("id"), "killchainphases": kill_chain_phases, "firstseenbysource": tool_obj.get("created"), "modified": tool_obj.get("modified"), "tool_types": tool_obj.get("tool_types", []), "description": tool_obj.get("description", ""), "aliases": tool_obj.get("aliases", []), "tool_version": tool_obj.get("tool_version", ""), "tags": list(set(tool_obj.get("labels", []))), } tool["customFields"] = fields return [tool] @staticmethod def parse_course_of_action(coa_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses a single course of action object :param coa_obj: course of action object :return: course of action extracted from the course of action object in cortex format """ publications = STIX2Parser.get_indicator_publication(coa_obj) course_of_action = { "value": coa_obj.get("name"), "indicator_type": ThreatIntel.ObjectsNames.COURSE_OF_ACTION, "score": ThreatIntel.ObjectsScore.COURSE_OF_ACTION, "rawJSON": coa_obj, } fields = { "stixid": coa_obj.get("id"), "firstseenbysource": coa_obj.get("created"), "modified": coa_obj.get("modified"), "description": coa_obj.get("description", ""), "action_type": coa_obj.get("action_type", ""), "publications": publications, "tags": list(set(coa_obj.get("labels", []))), } course_of_action["customFields"] = fields return [course_of_action] @staticmethod def parse_campaign(campaign_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses a single campaign object :param campaign_obj: campaign object :return: campaign extracted from the campaign object in cortex format """ campaign = { "value": campaign_obj.get("name"), "indicator_type": ThreatIntel.ObjectsNames.CAMPAIGN, "score": ThreatIntel.ObjectsScore.CAMPAIGN, "rawJSON": campaign_obj, } fields = { "stixid": campaign_obj.get("id"), "firstseenbysource": campaign_obj.get("created"), "modified": campaign_obj.get("modified"), "description": campaign_obj.get("description", ""), "aliases": campaign_obj.get("aliases", []), "objective": campaign_obj.get("objective", ""), "tags": list(set(campaign_obj.get("labels", []))), } campaign["customFields"] = fields return [campaign] @staticmethod def parse_intrusion_set(intrusion_set_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses a single intrusion set object :param intrusion_set_obj: intrusion set object :return: intrusion set extracted from the intrusion set object in cortex format """ publications = STIX2Parser.get_indicator_publication(intrusion_set_obj) intrusion_set = { "value": intrusion_set_obj.get("name"), "indicator_type": ThreatIntel.ObjectsNames.INTRUSION_SET, "score": ThreatIntel.ObjectsScore.INTRUSION_SET, "rawJSON": intrusion_set_obj, } fields = { "stixid": intrusion_set_obj.get("id"), "firstseenbysource": intrusion_set_obj.get("created"), "modified": intrusion_set_obj.get("modified"), "description": intrusion_set_obj.get("description", ""), "aliases": intrusion_set_obj.get("aliases", []), "goals": intrusion_set_obj.get("goals", []), "resource_level": intrusion_set_obj.get("resource_level", ""), "primary_motivation": intrusion_set_obj.get("primary_motivation", ""), "secondary_motivations": intrusion_set_obj.get("secondary_motivations", []), "publications": publications, "tags": list(set(intrusion_set_obj.get("labels", []))), } intrusion_set["customFields"] = fields return [intrusion_set] @staticmethod def parse_general_sco_indicator(sco_object: Dict[str, Any], value_mapping: str = "value") -> List[Dict[str, Any]]: """ Parses a single SCO indicator. Args: sco_object (dict): indicator as an observable object. value_mapping (str): the key that extracts the value from the indicator response. """ sco_indicator = { "value": sco_object.get(value_mapping), "score": Common.DBotScore.NONE, "rawJSON": sco_object, "indicator_type": STIX_2_TYPES_TO_CORTEX_TYPES.get(sco_object.get("type")), # type: ignore[arg-type] } fields = {"stixid": sco_object.get("id")} sco_indicator["customFields"] = fields return [sco_indicator] @staticmethod def parse_sco_autonomous_system_indicator(autonomous_system_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses autonomous_system indicator type to cortex format. Args: autonomous_system_obj (dict): indicator as an observable object of type autonomous-system. """ autonomous_system_indicator = STIX2Parser.parse_general_sco_indicator(autonomous_system_obj, value_mapping="number") autonomous_system_indicator[0]["customFields"]["name"] = autonomous_system_obj.get("name") return autonomous_system_indicator @staticmethod def parse_sco_file_indicator(file_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses file indicator type to cortex format. Args: file_obj (dict): indicator as an observable object of file type. """ file_hashes = file_obj.get("hashes", {}) value = file_hashes.get("SHA-256") or file_hashes.get("SHA-1") or file_hashes.get("MD5") if not value: return [] file_obj["value"] = value file_indicator = STIX2Parser.parse_general_sco_indicator(file_obj) file_indicator[0]["customFields"].update( { "associatedfilenames": file_obj.get("name"), "size": file_obj.get("size"), "path": file_obj.get("parent_directory_ref"), "md5": file_hashes.get("MD5"), "sha1": file_hashes.get("SHA-1"), "sha256": file_hashes.get("SHA-256"), } ) return file_indicator @staticmethod def parse_sco_mutex_indicator(mutex_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses mutex indicator type to cortex format. Args: mutex_obj (dict): indicator as an observable object of mutex type. """ return STIX2Parser.parse_general_sco_indicator(sco_object=mutex_obj, value_mapping="name") @staticmethod def parse_sco_account_indicator(account_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses account indicator type to cortex format. Args: account_obj (dict): indicator as an observable object of account type. """ account_indicator = STIX2Parser.parse_general_sco_indicator(account_obj, value_mapping="user_id") account_indicator[0]["customFields"].update( {"displayname": account_obj.get("user_id"), "accounttype": account_obj.get("account_type")} ) return account_indicator @staticmethod def parse_sco_windows_registry_key_indicator(registry_key_obj: Dict[str, Any]) -> List[Dict[str, Any]]: """ Parses registry_key indicator type to cortex format. Args: registry_key_obj (dict): indicator as an observable object of registry_key type. """ registry_key_indicator = STIX2Parser.parse_general_sco_indicator(registry_key_obj, value_mapping="key") registry_key_indicator[0]["customFields"].update( { "registryvalue": registry_key_obj.get("values"), "modified_time": registry_key_obj.get("modified_time"), "number_of_subkeys": registry_key_obj.get("number_of_subkeys"), } ) return registry_key_indicator def parse_relationships(self, relationships_lst: List[Dict[str, Any]]) -> Dict[str, Any]: """Parse the Relationships objects retrieved from the feed. Returns: A dict of relationship value to processed relationships as indicator object. """ a_value_to_relationship: Dict[str, Any] = {} for relationships_object in relationships_lst: relationship_type = relationships_object.get("relationship_type") if relationship_type not in EntityRelationship.Relationships.RELATIONSHIPS_NAMES: if relationship_type == "indicates": relationship_type = "indicated-by" else: demisto.debug(f"Invalid relation type: {relationship_type}") continue a_stixid = relationships_object.get("source_ref", "") a_object = self.parsed_object_id_to_object.get(a_stixid, {}) b_stixid = relationships_object.get("target_ref", "") b_object = self.parsed_object_id_to_object.get(b_stixid, {}) if not a_object or not b_object: demisto.debug(f"Cant find {a_object=} or {b_object=}.") continue a_value, a_type = a_object.get("value"), a_object.get("indicator_type") b_value, b_type = b_object.get("value"), b_object.get("indicator_type") if not (a_value and a_type and b_value and b_type): continue mapping_fields = { "lastseenbysource": relationships_object.get("modified"), "firstseenbysource": relationships_object.get("created"), } entity_relation = EntityRelationship( name=relationship_type, entity_a=a_value, entity_a_type=a_type, entity_b=b_value, entity_b_type=b_type, fields=mapping_fields, ) indicator_relationship = entity_relation.to_indicator() if a_value_to_relationship.get(a_value): a_value_to_relationship[a_value].append(indicator_relationship) else: a_value_to_relationship[a_value] = [indicator_relationship] return a_value_to_relationship def parse_stix2(self, js_content) -> List[Dict[str, str]]: """ Polls the taxii server and builds a list of cortex indicators objects from the result :return: Cortex indicators list """ if js_content.get("objects"): envelopes = STIX2Parser.create_envelopes_by_type(js_content["objects"]) else: envelopes = STIX2Parser.create_envelopes_by_type([js_content]) indicators = self.load_stix_objects_from_envelope(envelopes) return indicators def load_stix_objects_from_envelope(self, envelopes: Dict[str, Any]): parse_stix_2_objects = { "indicator": self.parse_indicator, "attack-pattern": self.parse_attack_pattern, "malware": self.parse_malware, "report": self.parse_report, "course-of-action": self.parse_course_of_action, "campaign": self.parse_campaign, "intrusion-set": self.parse_intrusion_set, "tool": self.parse_tool, "threat-actor": self.parse_threat_actor, "infrastructure": self.parse_infrastructure, "domain-name": self.parse_general_sco_indicator, "ipv4-addr": self.parse_general_sco_indicator, "ipv6-addr": self.parse_general_sco_indicator, "email-addr": self.parse_general_sco_indicator, "url": self.parse_general_sco_indicator, "autonomous-system": self.parse_sco_autonomous_system_indicator, "file": self.parse_sco_file_indicator, "mutex": self.parse_sco_mutex_indicator, "user-account": self.parse_sco_account_indicator, "windows-registry-key": self.parse_sco_windows_registry_key_indicator, } indicators = self.parse_dict_envelope(envelopes, parse_stix_2_objects) demisto.debug(f"{SCRIPT_NAME} has extracted {len(indicators)} indicators") return indicators def parse_dict_envelope(self, envelopes: Dict[str, Any], parse_objects_func): indicators = [] relationships_list: List[Dict[str, Any]] = [] xsoar_taxii_server_extensions = self.get_taxii2_extensions_from_envelope(envelopes.get("extension-definition", [])) for obj_type, stix_objects in envelopes.items(): if obj_type == "relationship": relationships_list.extend(stix_objects) else: for obj in stix_objects: # handled separately if obj.get("type") == "extension-definition": continue self.id_to_object[obj.get("id")] = obj if obj.get("type") == "report": result, relationships = self.parse_report(obj) relationships_list.extend(relationships) else: result = parse_objects_func[obj_type](obj) if not result: continue self.update_obj_if_extensions(xsoar_taxii_server_extensions, obj, result) self.parsed_object_id_to_object[obj.get("id")] = result[0] indicators.extend(result) if relationships_list: relationships_mapping = self.parse_relationships(relationships_list) STIX2Parser.add_relationship_to_indicator(relationships_mapping, indicators) return indicators @staticmethod def create_envelopes_by_type(objects) -> dict: """ Creates objects envelops by type """ types_envelopes: dict = {} index = 0 for obj in objects: obj_type = obj.get("type") if obj_type not in STIX2Parser.OBJECTS_TO_PARSE: demisto.debug(f"Cannot parse object of type {obj_type}, skipping.") index += 1 continue if obj_type not in types_envelopes: types_envelopes[obj_type] = [] types_envelopes[obj_type].append(obj) return types_envelopes @staticmethod def get_indicators_from_indicator_groups( indicator_groups: List[tuple[str, str]], indicator_obj: Dict[str, str], indicator_types: Dict[str, str], field_map: Dict[str, str], ) -> List[Dict[str, str]]: """ Get indicators from indicator regex groups :param indicator_groups: caught regex group in pattern of: [`type`, `indicator`] :param indicator_obj: taxii indicator object :param indicator_types: supported indicator types -> cortex types :param field_map: map used to create fields entry ({field_name: field_value}) :return: Indicators list """ indicators = [] if indicator_groups: for term in indicator_groups: for taxii_type in indicator_types: # term should be list with 2 argument parsed with regex - [`type`, `indicator`] if len(term) == 2 and taxii_type in term[0]: type_ = indicator_types[taxii_type] value = term[1] # support added for cases as 'value1','value2','value3' for 3 different indicators for indicator_value in value.split(","): indicator_value = indicator_value.strip("'") indicator = STIX2Parser.create_indicator(indicator_obj, type_, indicator_value.strip("'"), field_map) indicators.append(indicator) break return indicators @staticmethod def create_indicator(indicator_obj, type_, value, field_map): """ Create a cortex indicator from a stix indicator :param indicator_obj: rawJSON value of the indicator :param type_: cortex type of the indicator :param value: indicator value :param field_map: field map used for mapping fields ({field_name: field_value}) :return: Cortex indicator """ ioc_obj_copy = copy.deepcopy(indicator_obj) ioc_obj_copy["value"] = value ioc_obj_copy["type"] = type_ indicator = { "value": value, "indicator_type": type_, "rawJSON": ioc_obj_copy, } fields = {} tags = [] # create tags from labels: for label in ioc_obj_copy.get("labels", []): tags.append(label) # add description if able if "description" in ioc_obj_copy: fields["description"] = ioc_obj_copy["description"] # add field_map fields for field_name, field_path in field_map.items(): if field_path in ioc_obj_copy: fields[field_name] = ioc_obj_copy.get(field_path) # union of tags and labels if "tags" in fields: field_tag = fields.get("tags") if isinstance(field_tag, list): tags.extend(field_tag) else: tags.append(field_tag) fields["tags"] = tags indicator["customFields"] = fields return indicator @staticmethod def extract_indicator_groups_from_pattern(pattern: str, regexes: List) -> List[tuple[str, str]]: """ Extracts indicator [`type`, `indicator`] groups from pattern :param pattern: stix pattern :param regexes: regexes to run to pattern :return: extracted indicators list from pattern """ groups: List[tuple[str, str]] = [] for regex in regexes: find_result = regex.findall(pattern) if find_result: groups.extend(find_result) return groups @staticmethod def add_relationship_to_indicator(relationships_mapping, indicators): """ Adds relationship to right indicator :param relationships_mapping: maps a_value to relationship object :param indicators: all indicators that were fetched from file. """ for indicator in indicators: if (a_value := indicator.get("value")) and (relationships := relationships_mapping.get(a_value)): indicator["relationships"] = relationships @staticmethod def update_obj_if_extensions(xsoar_taxii_server_extensions, obj, result): """ If stix object has extension, check if it xsoar taxii2 server extension, if yes parse it to XSOAR. :param xsoar_taxii_server_extensions: ids of all XSOAR extentions in current bundle. :param obj: stix object :param result: parsed xsoar indicator :return: updated xsoar indicator """ parsed_result = result[0] if extensions := obj.get("extensions"): custom_fields = parsed_result.get("customFields", {}) for ext_id, extension in extensions.items(): if ext_id in xsoar_taxii_server_extensions: extension.pop("extension_type") for field, value in extension.items(): if field in SYSTEM_FIELDS: parsed_result[field] = value elif field.lower() == "customfields": custom_fields.update(value) else: custom_fields[field] = value parsed_result["customFields"] = custom_fields @staticmethod def get_taxii2_extensions_from_envelope(stix_objects): """ :param stix_objects: list of all extension objects. :return: list of xsoar extensions ids. """ xsoar_taxii_server_extensions = [] for obj in stix_objects: if obj.get("schema") == XSOAR_TAXII2_SERVER_SCHEMA: xsoar_taxii_server_extensions.append(obj.get("id")) return xsoar_taxii_server_extensions # STIX 1 Parsing def package_extract_properties(package): """Extracts properties from the STIX package""" result: Dict[str, str] = {} header = package.find_all("STIX_Header") if len(header) == 0: return result # share level mstructures = header[0].find_all("Marking_Structure") for ms in mstructures: type_ = ms.get("xsi:type") if type_ is result: continue color = ms.get("color") if color is result: continue type_ = type_.lower() if "tlpmarkingstructuretype" not in type_: continue result["share_level"] = color.lower() # To keep backward compatibility result["TLP"] = color.upper() # https://www.us-cert.gov/tlp break # decode title title = next((c for c in header[0] if c.name == "Title"), None) if title is not None: result["stix_package_title"] = title.text # decode description description = next((c for c in header[0] if c.name == "Description"), None) if description is not None: result["stix_package_description"] = description.text # decode description sdescription = next((c for c in header[0] if c.name == "Short_Description"), None) if sdescription is not None: result["stix_package_short_description"] = sdescription.text # decode identity name from information_source information_source = next((c for c in header[0] if c.name == "Information_Source"), None) if information_source is not None: identity = next((c for c in information_source if c.name == "Identity"), None) if identity is not None: name = next(c for c in identity if c.name == "Name") if name is not None: result["stix_package_information_source"] = name.text return result def observable_extract_properties(observable): """Extracts properties from observable""" result = {} if id_ref := observable.get("id"): result["indicator_ref"] = id_ref title = next((c for c in observable if c.name == "Title"), None) if title is not None: title = title.text result["stix_title"] = title description = next((c for c in observable if c.name == "Description"), None) if description is not None: description = description.text result["stix_description"] = description return result def indicator_extract_properties(indicator) -> Dict[str, Any]: """Extracts the Indicator properties Args: indicator (bs4.element.Tag): The Indicator content in xml. Returns: dict: The ttp properties in a dict {'property': 'value'}. (The value can be a list) """ result: Dict[str, Any] = {} title = next((c for c in indicator if c.name == "Title"), None) if title is not None: title = title.text result["stix_indicator_name"] = title description = next((c for c in indicator if c.name == "Description"), None) if description is not None: description = description.text result["stix_indicator_description"] = description confidence = next((c for c in indicator if c.name == "Confidence"), None) if confidence is not None: value = next((c for c in confidence if c.name == "Value"), None) if value is not None: value = value.text result["confidence"] = value if indicated_ttp := indicator.find_all("Indicated_TTP"): result["ttp_ref"] = [] # Each indicator can be related to few ttps for ttp_value in indicated_ttp: ttp = next((c for c in ttp_value if c.name == "TTP"), None) if ttp is not None: value = ttp.get("idref") result["ttp_ref"].append(value) return result def ttp_extract_properties(ttp, behavior) -> Dict[str, str]: """Extracts the TTP properties Args: ttp (bs4.element.Tag): The TTP content in xml. behavior (str): The TTP behavior ['Malware', 'Attack Pattern']. Returns: dict: The ttp properties in a dict {'property': 'value'}. """ result = {"type": behavior} if behavior == "Malware": type_ = next((c for c in ttp if c.name == "Type"), None) if type_ is not None: type_ = type_.text result["malware_type"] = type_ name = next((c for c in ttp if c.name == "Name"), None) if name is not None: name = name.text result["indicator"] = name title = next((c for c in ttp if c.name == "Title"), None) if title is not None: title = title.text result["title"] = title if behavior == "Attack Pattern": id_ref = next((c for c in ttp if c.name == "idref"), None) if id_ref is not None: id_ref = id_ref.text result["stix_id_ref"] = id_ref title = next((c for c in ttp if c.name == "Title"), None) if title is not None: title = title.text result["indicator"] = title description = next((c for c in ttp if c.name == "Description"), None) if description is not None: description = description.text result["description"] = description short_description = next((c for c in ttp if c.name == "Short_Description"), None) if short_description is not None: short_description = short_description.text result["short_description"] = short_description return result def create_relationships(indicator): results = [] for relationship in indicator.get("relationships", {}): if relationship.get("type") == "Malware": name = "indicator-of" relationship_type = "Malware" else: name = "related-to" relationship_type = "Attack Pattern" entity_relationship = EntityRelationship( name=name, entity_a=indicator.get("value"), entity_a_type=indicator.get("type"), entity_b=relationship.get("indicator"), entity_b_type=relationship_type, ) results.append(entity_relationship.to_indicator()) return results class AddressObject: """ Implements address object indicator decoding based on: https://stixproject.github.io/data-model/1.2/AddressObj/AddressObjectType/ """ @staticmethod def decode(props, **kwargs): result: List[Dict[str, str]] = [] indicator = props.find("Address_Value") if indicator is None: return result indicator = indicator.string.encode("ascii", "replace").decode() category = props.get("category", None) address_list = indicator.split("##comma##") if category == "e-mail": return [{"indicator": address, "type": "Email"} for address in address_list] try: for address in address_list: ip = IPNetwork(address) if ip.version == 4: if len(address.split("/")) > 1: type_ = "CIDR" else: type_ = "IP" elif ip.version == 6: if len(address.split("/")) > 1: type_ = "IPv6CIDR" else: type_ = "IPv6" else: LOG(f"Unknown ip version: {ip.version!r}") return [] result.append({"indicator": address, "type": type_}) except Exception: return result return result class DomainNameObject: """ Implements domain object indicator decoding based on: https://stixproject.github.io/data-model/1.2/DomainNameObj/DomainNameObjectType/ """ @staticmethod def decode(props, **kwargs): domains = [] dtype = props.get("type", "FQDN") if dtype != "FQDN": return [] if domain_value := props.find("Value"): domain_list = domain_value.string.split("##comma##") for domain in domain_list: domains.append({"indicator": domain, "type": "Domain"}) return domains class FileObject: """ Implements file object indicator decoding based on: https://stixproject.github.io/data-model/1.2/FileObj/FileObjectType/ """ @staticmethod def _decode_basic_props(props): result = {} name = next((c for c in props if c.name == "File_Name"), None) if name is not None: result["stix_file_name"] = name.text size = next((c for c in props if c.name == "File_Size"), None) if size is not None: result["stix_file_size"] = size.text file_format = next((c for c in props if c.name == "File_Format"), None) if file_format is not None: result["stix_file_format"] = file_format.text return result @staticmethod def decode(props, **kwargs): result = [] bprops = FileObject._decode_basic_props(props) hashes = props.find_all("Hash") for h in hashes: value = h.find("Simple_Hash_Value") if value is None: continue value = value.string.lower() value_list = value.split("##comma##") for v in value_list: v = v.strip() if type := detect_file_indicator_type(v): result.append({"indicator": v, "htype": type, "type": "File"}) for r in result: for r2 in result: if r["htype"] == r2["htype"]: continue r[f"stix_file_{r2['htype']}"] = r2["indicator"] r.update(bprops) return result class URIObject: """ Implements URI object indicator decoding based on: https://stixproject.github.io/data-model/1.2/URIObj/URIObjectType/ """ @staticmethod def decode(props, **kwargs): urls = [] utype = props.get("type", "URL") if utype == "URL": type_ = "URL" elif utype == "Domain Name": type_ = "Domain" else: return [] if url_value := props.find("Value"): url_list = url_value.string.split("##comma##") for url in url_list: urls.append({"indicator": url, "type": type_}) return urls class SocketAddressObject: """ Implements socket address object indicator decoding based on: https://stixproject.github.io/data-model/1.2/SocketAddressObj/SocketAddressObjectType/ """ @staticmethod def decode(props, **kwargs): ip = props.get("ip_address", None) if ip: return AddressObject.decode(ip) return [] class LinkObject: """ Implements link object indicator decoding based on: https://stixproject.github.io/data-model/1.2/LinkObj/LinkObjectType/ """ @staticmethod def decode(props, **kwargs): ltype = props.get("type", "URL") if ltype != "URL": LOG(f"Unhandled LinkObjectType type: {ltype}") return [] value = props.get("value", None) if value is None: LOG("no value in observable LinkObject") return [] if not isinstance(value, string_types): value = value.get("value", None) if value is None: LOG("no value in observable LinkObject") return [] return [{"indicator": value, "type": ltype}] class HTTPSessionObject: """ Implements http session object indicator decoding based on: https://stixproject.github.io/data-model/1.2/HTTPSessionObj/HTTPSessionObjectType/ """ @staticmethod def decode(props, **kwargs): if props.get("http_request_response"): tmp = props.get("http_request_response") if len(tmp) == 1: item = tmp[0] http_client_request = item.get("http_client_request", None) if http_client_request is not None: http_request_header = http_client_request.get("http_request_header", None) if http_request_header is not None: raw_header = http_request_header.get("raw_header", None) if raw_header is not None: return [ { "indicator": raw_header.split("\n")[0], "type": "http-session", # we don't support this type natively in demisto "header": raw_header, } ] else: LOG("multiple HTTPSessionObjectTypes not supported") return [] class StixDecode: """ Decode STIX strings formatted as xml, and extract indicators from them """ DECODERS = { "DomainNameObjectType": DomainNameObject.decode, "FileObjectType": FileObject.decode, "WindowsFileObjectType": FileObject.decode, "URIObjectType": URIObject.decode, "AddressObjectType": AddressObject.decode, "SocketAddressObjectType": SocketAddressObject.decode, "LinkObjectType": LinkObject.decode, "HTTPSessionObjectType": HTTPSessionObject.decode, } @staticmethod def object_extract_properties(props, kwargs): type_ = props.get("xsi:type").rsplit(":")[-1] if type_ not in StixDecode.DECODERS: LOG(f"Unhandled cybox Object type: {type_!r} - {props!r}") return [] return StixDecode.DECODERS[type_](props, **kwargs) @staticmethod def _parse_stix_timestamp(stix_timestamp): dt = dateutil.parser.parse(stix_timestamp) if dt.tzinfo is None: dt = dt.replace(tzinfo=pytz.UTC) delta = dt - EPOCH return int(delta.total_seconds() * 1000) @staticmethod def _deduplicate(indicators): result = {} for iv in indicators: result["{}:{}".format(iv["indicator"], iv["type"])] = iv return list(result.values()) @staticmethod def decode(content, **kwargs): observable_result = [] indicator_result: Dict[str, dict] = {} ttp_result: Dict[str, dict] = {} package = BeautifulSoup(content, "xml") timestamp = package.get("timestamp", None) if timestamp is not None: timestamp = StixDecode._parse_stix_timestamp(timestamp) # extract the Observable info if observables := package.find_all("Observable"): pprops = package_extract_properties(package) for o in observables: gprops = observable_extract_properties(o) obj = next((ob for ob in o if ob.name == "Object"), None) # type: ignore if obj is None: continue # main properties properties = next((c for c in obj if c.name == "Properties"), None) # type: ignore if properties is not None: for r in StixDecode.object_extract_properties(properties, kwargs): r.update(gprops) r.update(pprops) observable_result.append(r) # then related objects related = next((c for c in obj if c.name == "Related_Objects"), None) # type: ignore if related is not None: for robj in related: if robj.name != "Related_Object": # type: ignore continue properties = next((c for c in robj if c.name == "Properties"), None) # type: ignore if properties is None: continue for r in StixDecode.object_extract_properties(properties, kwargs): r.update(gprops) r.update(pprops) observable_result.append(r) # extract the Indicator info if (indicators := package.find_all("Indicator")) and observables: indicator_ref = observables[0].get("idref") # type: ignore if indicator_ref: indicator_info = indicator_extract_properties(indicators[0]) indicator_result[indicator_ref] = indicator_info # type: ignore # extract the TTP info if ttp := package.find_all("TTP"): ttp_info: Dict[str, str] = {} id_ref = ttp[0].get("id") # type: ignore title = next((c for c in ttp[0] if c.name == "Title"), None) # type: ignore if title is not None: title = title.text # type: ignore ttp_info["stix_ttp_title"] = title # type: ignore description = next((c for c in ttp[0] if c.name == "Description"), None) # type: ignore if description is not None: description = description.text # type: ignore ttp_info["ttp_description"] = description # type: ignore if behavior := package.find_all("Behavior"): if behavior[0].find_all("Malware"): # type: ignore ttp_info.update(ttp_extract_properties(package.find_all("Malware_Instance")[0], "Malware")) elif behavior[0].find_all("Attack_Patterns"): # type: ignore ttp_info.update(ttp_extract_properties(package.find_all("Attack_Pattern")[0], "Attack Pattern")) ttp_result[id_ref] = ttp_info # type: ignore return timestamp, StixDecode._deduplicate(observable_result), indicator_result, ttp_result def build_observables(file_name): tag_stack = collections.deque() # type: ignore observables = [] indicators = {} ttps = {} for action, element in etree.iterparse( file_name, events=("start", "end"), recover=True, resolve_entities=False, load_dtd=False, no_network=True, ): if action == "start": tag_stack.append(element.tag) else: last_tag = tag_stack.pop() if last_tag != element.tag: raise RuntimeError(f"{SCRIPT_NAME} - error parsing poll response, mismatched tags") if action == "end" and element.tag.endswith("STIX_Package"): for c in element: try: content = etree.tostring(c, encoding="unicode") timestamp, observable, indicator, ttp = StixDecode.decode(content) if observable: observables.extend(observable) if indicator: indicators.update(indicator) if ttp: ttps.update(ttp) except Exception as e: demisto.debug(f"Failed parsing response: {e}") element.clear() for observable in observables: if (indicator_ref := observable.get("indicator_ref")) and (indicator_info := indicators.get(indicator_ref)): observable.update(indicator_info) ttp_ref = observable.get("ttp_ref", []) relationships = [] for reference in ttp_ref: if relationship := ttps.get(reference): relationships.append(relationship) if relationships: observable["relationships"] = relationships return observables, ttps def parse_stix(file_name): """ :param file_name: the file with xml indicators :return: Parsed indicators in XSOAR format """ indicators = [] indicator_custom_fields = { "title": "stix_title", "description": "stix_description", "name": "stix_indicator_name", "stixdescription": "stix_indicator_description", "confidence": "confidence", } # Create the indicators from the observables observables, ttps = build_observables(file_name) for item in observables: if indicator := item.get("indicator"): item["value"] = indicator.strip() indicator_obj = { "value": indicator.strip(), "indicator_type": item.get("type"), "customFields": { xsoar_field: item.get(stix_field) for xsoar_field, stix_field in indicator_custom_fields.items() if item.get(stix_field) }, } if item.get("relationships"): relationships = create_relationships(item) indicator_obj["relationships"] = relationships indicator_obj["rawJSON"] = item indicators.append(indicator_obj) # Create the indicators from the ttps ttps_custom_fields = { "title": "title", "description": "description", "shortdescription": "short_description", "stixdescription": "ttp_description", "stixttptitle": "stix_ttp_title", } for item in ttps.values(): if indicator := item.get("indicator"): item["value"] = indicator.strip() indicator_obj = { "value": indicator.strip(), "indicator_type": item.get("type"), "customFields": { xsoar_field: item.get(stix_field) for xsoar_field, stix_field in ttps_custom_fields.items() if item.get(stix_field) }, } if item.get("type") == "Malware": indicator_obj["score"] = ThreatIntel.ObjectsScore.MALWARE indicator_obj["stixmalwaretypes"] = item.get("malware_type", "").lower().replace(" ", "-") else: indicator_obj["score"] = ThreatIntel.ObjectsScore.ATTACK_PATTERN indicator_obj["rawJSON"] = item indicators.append(indicator_obj) return indicators def main(): # pragma: no cover args = demisto.args() indicator_txt = args.get("iocXml") entry_id = args.get("entry_id") if not indicator_txt and not entry_id: raise Exception("You must enter iocXml or entry_id of the Indicator.") elif entry_id: file_path = demisto.getFilePath(entry_id).get("path") with open(file_path) as f: indicator_txt = f.read() if stix2 := convert_to_json(indicator_txt): stix2_parser = STIX2Parser() observables = stix2_parser.parse_stix2(stix2) else: if "file_path" not in locals(): with tempfile.NamedTemporaryFile() as temp: temp.write(str.encode(indicator_txt)) temp.flush() observables = parse_stix(temp.name) else: observables = parse_stix(file_path) json_data = json.dumps(observables) return_results(json_data) if __name__ in ("__builtin__", "builtins", "__main__"): main()