StixParser
Parse STIX files to Cortex XSOAR indicators by clicking the **Upload STIX File** button.
- Type
- python
- Pack
- Base
Source
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()
README
Parse STIX files to Cortex XSOAR indicators by clicking the Upload STIX File button. This script is used for the button only.
This script does not support indicators relationships creation.
In order to create indicators from STIX files using an automation, use CreateIndicatorsFromSTIX. This automation supports indicators relationships creation.
Script Data
| Name | Description |
|---|---|
| Script Type | python3 |
| Tags | stix, ioc |
| Cortex XSOAR Version | 5.0.0 |
Inputs
| Argument Name | Description |
|---|---|
| iocXml | IOC XML or JSON in STIX format. |
| entry_id | IOC file entry ID. |
Outputs
There are no outputs for this script.