TAXII2ApiModule
Common TAXII 2 code that will be appended into each TAXII 2 integration when it's deployed.
- Type
- python
- Pack
- ApiModules
Source
import copy
import logging
import re
import tempfile
import traceback
import types
import uuid
# pylint: disable=E9010, E9011
import demistomock as demisto # noqa: F401
import urllib3
from CommonServerPython import * # noqa: F401
from requests.exceptions import HTTPError
from requests.sessions import CaseInsensitiveDict, merge_setting
from stix2patterns.pattern import Pattern
from taxii2client import v20, v21
from taxii2client.common import TokenAuth, _HTTPConnection
from taxii2client.exceptions import InvalidJSONError
# disable insecure warnings
urllib3.disable_warnings()
class XsoarSuppressWarningFilter(logging.Filter): # pragma: no cover
def filter(self, record):
# Suppress all logger records, but send the important ones to demisto logger
if record.levelno == logging.WARNING:
demisto.debug(record.getMessage())
elif record.levelno in [logging.ERROR, logging.CRITICAL]:
demisto.error(record.getMessage())
return False
# Make sure we have only one XsoarSuppressWarningFilter
v21_logger = logging.getLogger("taxii2client.v21")
demisto.debug(f"Logging Filters before cleaning: {v21_logger.filters=}")
for current_filter in list(v21_logger.filters): # pragma: no cover
if "XsoarSuppressWarningFilter" in type(current_filter).__name__:
v21_logger.removeFilter(current_filter)
v21_logger.addFilter(XsoarSuppressWarningFilter())
demisto.debug(f"Logging Filters: {v21_logger.filters=}")
# 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:"
ALLOWED_VERSIONS = [TAXII_VER_2_0, TAXII_VER_2_1]
ERR_NO_COLL = "No collection is available for this user, please make sure you entered the configuration correctly"
# Pattern Regexes - used to extract indicator type and value
INDICATOR_OPERATOR_VAL_FORMAT_PATTERN = r"(\w.*?{value}{operator})'(.*?)'"
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="ISUPPERSET")
HASHES_EQUALS_VAL_PATTERN = INDICATOR_OPERATOR_VAL_FORMAT_PATTERN.format(value=r"hashes\..*?", operator="=")
TAXII_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
TAXII_TIME_FORMAT_NO_MS = "%Y-%m-%dT%H:%M:%SZ"
STIX_2_TYPES_TO_CORTEX_TYPES = { # pragma: no cover
"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,
"sha-512": 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,
"identity": FeedIndicatorType.Identity,
"location": FeedIndicatorType.Location,
"vulnerability": FeedIndicatorType.CVE,
"x509-certificate": FeedIndicatorType.X509,
"software": FeedIndicatorType.Software,
}
STIX_SUPPORTED_TYPES = {
"url": ("value",),
"ip": ("value",),
"domain-name": ("value",),
"email-addr": ("value",),
"ipv4-addr": ("value",),
"ipv6-addr": ("value",),
"attack-pattern": ("name",),
"campaign": ("name",),
"identity": ("name",),
"infrastructure": ("name",),
"intrusion-set": ("name",),
"malware": ("name",),
"report": ("name",),
"threat-actor": ("name",),
"tool": ("name",),
"vulnerability": ("name",),
"mutex": ("name",),
"software": ("name",),
"autonomous-system": ("number",),
"file": ("hashes",),
"user-account": ("user_id",),
"location": ("name", "country"),
"x509-certificate": ("serial_number", "issuer"),
"windows-registry-key": ("key", "values"),
}
MITRE_CHAIN_PHASES_TO_DEMISTO_FIELDS = { # pragma: no cover
"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 = { # pragma: no cover
"ipv4-addr": FeedIndicatorType.CIDR,
"ipv6-addr": FeedIndicatorType.IPv6CIDR,
}
THREAT_INTEL_TYPE_TO_DEMISTO_TYPES = { # pragma: no cover
"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,
}
# marking definitions of TLPs are constant (marking definitions of statements can vary)
MARKING_DEFINITION_TO_TLP = {
"marking-definition--613f2e26-407d-48c7-9eca-b8e91df99dc9": "WHITE",
"marking-definition--34098fce-860f-48ae-8e50-ebd3cc5e41da": "GREEN",
"marking-definition--f88d31f6-486f-44da-b317-01333bde0b82": "AMBER",
"marking-definition--5e57c739-391a-4eb3-b6be-7d15ca92d5ed": "RED",
}
# country codes are in ISO-2 format
COUNTRY_CODES_TO_NAMES = {
"AD": "Andorra",
"AE": "United Arab Emirates", # pragma: no cover
"AF": "Afghanistan",
"AG": "Antigua and Barbuda",
"AI": "Anguilla",
"AL": "Albania",
"AM": "Armenia",
"AO": "Angola",
"AQ": "Antarctica",
"AR": "Argentina",
"AS": "American Samoa",
"AT": "Austria",
"AU": "Australia",
"AW": "Aruba",
"AX": "Aland Islands",
"AZ": "Azerbaijan",
"BA": "Bosnia and Herzegovina",
"BB": "Barbados",
"BD": "Bangladesh",
"BE": "Belgium",
"BF": "Burkina Faso",
"BG": "Bulgaria",
"BH": "Bahrain",
"BI": "Burundi",
"BJ": "Benin",
"BL": "Saint Barthelemy",
"BM": "Bermuda",
"BN": "Brunei",
"BO": "Bolivia",
"BQ": "Bonaire, Saint Eustatius and Saba ",
"BR": "Brazil",
"BS": "Bahamas",
"BT": "Bhutan",
"BV": "Bouvet Island",
"BW": "Botswana",
"BY": "Belarus",
"BZ": "Belize",
"CA": "Canada",
"CC": "Cocos Islands",
"CD": "Democratic Republic of the Congo",
"CF": "Central African Republic",
"CG": "Republic of the Congo",
"CH": "Switzerland",
"CI": "Ivory Coast",
"CK": "Cook Islands",
"CL": "Chile",
"CM": "Cameroon",
"CN": "China",
"CO": "Colombia",
"CR": "Costa Rica",
"CU": "Cuba",
"CV": "Cape Verde",
"CW": "Curacao",
"CX": "Christmas Island",
"CY": "Cyprus",
"CZ": "Czech Republic",
"DE": "Germany",
"DJ": "Djibouti",
"DK": "Denmark",
"DM": "Dominica",
"DO": "Dominican Republic",
"DZ": "Algeria",
"EC": "Ecuador",
"EE": "Estonia",
"EG": "Egypt",
"EH": "Western Sahara",
"ER": "Eritrea",
"ES": "Spain",
"ET": "Ethiopia",
"FI": "Finland",
"FJ": "Fiji",
"FK": "Falkland Islands",
"FM": "Micronesia",
"FO": "Faroe Islands",
"FR": "France",
"GA": "Gabon",
"GB": "United Kingdom",
"GD": "Grenada",
"GE": "Georgia",
"GF": "French Guiana",
"GG": "Guernsey",
"GH": "Ghana",
"GI": "Gibraltar",
"GL": "Greenland",
"GM": "Gambia",
"GN": "Guinea",
"GP": "Guadeloupe",
"GQ": "Equatorial Guinea",
"GR": "Greece",
"GS": "South Georgia and the South Sandwich Islands",
"GT": "Guatemala",
"GU": "Guam",
"GW": "Guinea-Bissau",
"GY": "Guyana",
"HK": "Hong Kong",
"HM": "Heard Island and McDonald Islands",
"HN": "Honduras",
"HR": "Croatia",
"HT": "Haiti",
"HU": "Hungary",
"ID": "Indonesia",
"IE": "Ireland",
"IL": "Israel",
"IM": "Isle of Man",
"IN": "India",
"IO": "British Indian Ocean Territory",
"IQ": "Iraq",
"IR": "Iran",
"IS": "Iceland",
"IT": "Italy",
"JE": "Jersey",
"JM": "Jamaica",
"JO": "Jordan",
"JP": "Japan",
"KE": "Kenya",
"KG": "Kyrgyzstan",
"KH": "Cambodia",
"KI": "Kiribati",
"KM": "Comoros",
"KN": "Saint Kitts and Nevis",
"KP": "North Korea",
"KR": "South Korea",
"KW": "Kuwait",
"KY": "Cayman Islands",
"KZ": "Kazakhstan",
"LA": "Laos",
"LB": "Lebanon",
"LC": "Saint Lucia",
"LI": "Liechtenstein",
"LK": "Sri Lanka",
"LR": "Liberia",
"LS": "Lesotho",
"LT": "Lithuania",
"LU": "Luxembourg",
"LV": "Latvia",
"LY": "Libya",
"MA": "Morocco",
"MC": "Monaco",
"MD": "Moldova",
"ME": "Montenegro",
"MF": "Saint Martin",
"MG": "Madagascar",
"MH": "Marshall Islands",
"MK": "Macedonia",
"ML": "Mali",
"MM": "Myanmar",
"MN": "Mongolia",
"MO": "Macao",
"MP": "Northern Mariana Islands",
"MQ": "Martinique",
"MR": "Mauritania",
"MS": "Montserrat",
"MT": "Malta",
"MU": "Mauritius",
"MV": "Maldives",
"MW": "Malawi",
"MX": "Mexico",
"MY": "Malaysia",
"MZ": "Mozambique",
"NA": "Namibia",
"NC": "New Caledonia",
"NE": "Niger",
"NF": "Norfolk Island",
"NG": "Nigeria",
"NI": "Nicaragua",
"NL": "Netherlands",
"NO": "Norway",
"NP": "Nepal",
"NR": "Nauru",
"NU": "Niue",
"NZ": "New Zealand",
"OM": "Oman",
"PA": "Panama",
"PE": "Peru",
"PF": "French Polynesia",
"PG": "Papua New Guinea",
"PH": "Philippines",
"PK": "Pakistan",
"PL": "Poland",
"PM": "Saint Pierre and Miquelon",
"PN": "Pitcairn",
"PR": "Puerto Rico",
"PS": "Palestinian Territory",
"PT": "Portugal",
"PW": "Palau",
"PY": "Paraguay",
"QA": "Qatar",
"RE": "Reunion",
"RO": "Romania",
"RS": "Serbia",
"RU": "Russia",
"RW": "Rwanda",
"SA": "Saudi Arabia",
"SB": "Solomon Islands",
"SC": "Seychelles",
"SD": "Sudan",
"SE": "Sweden",
"SG": "Singapore",
"SH": "Saint Helena",
"SI": "Slovenia",
"SJ": "Svalbard and Jan Mayen",
"SK": "Slovakia",
"SL": "Sierra Leone",
"SM": "San Marino",
"SN": "Senegal",
"SO": "Somalia",
"SR": "Suriname",
"SS": "South Sudan",
"ST": "Sao Tome and Principe",
"SV": "El Salvador",
"SX": "Sint Maarten",
"SY": "Syria",
"SZ": "Swaziland",
"TC": "Turks and Caicos Islands",
"TD": "Chad",
"TF": "French Southern Territories",
"TG": "Togo",
"TH": "Thailand",
"TJ": "Tajikistan",
"TK": "Tokelau",
"TL": "East Timor",
"TM": "Turkmenistan",
"TN": "Tunisia",
"TO": "Tonga",
"TR": "Turkey",
"TT": "Trinidad and Tobago",
"TV": "Tuvalu",
"TW": "Taiwan",
"TZ": "Tanzania",
"UA": "Ukraine",
"UG": "Uganda",
"UM": "United States Minor Outlying Islands",
"US": "United States",
"UY": "Uruguay",
"UZ": "Uzbekistan",
"VA": "Vatican",
"VC": "Saint Vincent and the Grenadines",
"VE": "Venezuela",
"VG": "British Virgin Islands",
"VI": "U.S. Virgin Islands",
"VN": "Vietnam",
"VU": "Vanuatu",
"WF": "Wallis and Futuna",
"WS": "Samoa",
"XK": "Kosovo",
"YE": "Yemen",
"YT": "Mayotte",
"ZA": "South Africa",
"ZM": "Zambia",
"ZW": "Zimbabwe",
}
STIX2_TYPES_TO_XSOAR: dict[str, Union[str, tuple[str, ...]]] = { # pragma: no cover
"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,
"vulnerability": FeedIndicatorType.CVE,
"ipv4-addr": FeedIndicatorType.IP,
"ipv6-addr": FeedIndicatorType.IPv6,
"domain-name": (FeedIndicatorType.DomainGlob, FeedIndicatorType.Domain),
"user-account": FeedIndicatorType.Account,
"email-addr": FeedIndicatorType.Email,
"url": FeedIndicatorType.URL,
"file": FeedIndicatorType.File,
"windows-registry-key": FeedIndicatorType.Registry,
"indicator": (
FeedIndicatorType.IP,
FeedIndicatorType.IPv6,
FeedIndicatorType.DomainGlob,
FeedIndicatorType.Domain,
FeedIndicatorType.Account,
FeedIndicatorType.Email,
FeedIndicatorType.URL,
FeedIndicatorType.File,
FeedIndicatorType.Registry,
),
"software": FeedIndicatorType.Software,
"autonomous-system": FeedIndicatorType.AS,
"x509-certificate": FeedIndicatorType.X509,
}
PAWN_UUID = uuid.uuid5(uuid.NAMESPACE_URL, "https://www.paloaltonetworks.com")
XSOAR_TYPES_TO_STIX_SDO = { # pragma: no cover
ThreatIntel.ObjectsNames.ATTACK_PATTERN: "attack-pattern",
ThreatIntel.ObjectsNames.CAMPAIGN: "campaign",
ThreatIntel.ObjectsNames.COURSE_OF_ACTION: "course-of-action",
ThreatIntel.ObjectsNames.INFRASTRUCTURE: "infrastructure",
ThreatIntel.ObjectsNames.INTRUSION_SET: "intrusion-set",
ThreatIntel.ObjectsNames.REPORT: "report",
ThreatIntel.ObjectsNames.THREAT_ACTOR: "threat-actor",
ThreatIntel.ObjectsNames.TOOL: "tool",
ThreatIntel.ObjectsNames.MALWARE: "malware",
FeedIndicatorType.CVE: "vulnerability",
FeedIndicatorType.Identity: "identity",
FeedIndicatorType.Location: "location",
}
XSOAR_TYPES_TO_STIX_SCO = { # pragma: no cover
FeedIndicatorType.CIDR: "ipv4-addr",
FeedIndicatorType.DomainGlob: "domain-name",
FeedIndicatorType.IPv6: "ipv6-addr",
FeedIndicatorType.IPv6CIDR: "ipv6-addr",
FeedIndicatorType.Account: "user-account",
FeedIndicatorType.Domain: "domain-name",
FeedIndicatorType.Email: "email-addr",
FeedIndicatorType.IP: "ipv4-addr",
FeedIndicatorType.Registry: "windows-registry-key",
FeedIndicatorType.File: "file",
FeedIndicatorType.URL: "url",
FeedIndicatorType.Software: "software",
FeedIndicatorType.AS: "autonomous-system",
FeedIndicatorType.X509: "x509-certificate",
}
HASH_TYPE_TO_STIX_HASH_TYPE = { # pragma: no cover
"md5": "MD5",
"sha1": "SHA-1",
"sha256": "SHA-256",
"sha512": "SHA-512",
}
STIX_DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
SCO_DET_ID_NAMESPACE = uuid.UUID("00abedb4-aa42-466c-9c01-fed23315a9b7")
def reached_limit(limit: int, element_count: int):
return element_count >= limit > -1
PatternComparisons = dict[str, list[tuple[list[str], str, str]]]
class XSOAR2STIXParser:
def __init__(self, namespace_uuid, fields_to_present, types_for_indicator_sdo, server_version=TAXII_VER_2_1):
self.server_version = server_version
if server_version not in ALLOWED_VERSIONS:
raise Exception(f'Wrong TAXII 2 Server version: {server_version}. Possible values: {", ".join(ALLOWED_VERSIONS)}.')
self.namespace_uuid = namespace_uuid
self.fields_to_present = fields_to_present
self.has_extension = fields_to_present != {"name", "type"}
self.types_for_indicator_sdo = types_for_indicator_sdo or []
def create_indicators(self, indicator_searcher: IndicatorsSearcher, is_manifest: bool):
"""
Args:
indicator_searcher: indicators list
is_manifest: whether this call is for manifest or indicators
Returns: Created indicators and its extensions.
"""
total = 0
extensions_dict: dict = {}
iocs = []
extensions = []
for ioc in indicator_searcher:
found_indicators = ioc.get("iocs") or []
total = ioc.get("total")
for xsoar_indicator in found_indicators:
xsoar_type = xsoar_indicator.get("indicator_type")
if is_manifest:
manifest_entry = self.create_manifest_entry(xsoar_indicator, xsoar_type)
if manifest_entry:
iocs.append(manifest_entry)
else:
stix_ioc, extension_definition, extensions_dict = self.create_stix_object(
xsoar_indicator, xsoar_type, extensions_dict
)
if XSOAR_TYPES_TO_STIX_SCO.get(xsoar_type) in self.types_for_indicator_sdo:
stix_ioc = self.convert_sco_to_indicator_sdo(stix_ioc, xsoar_indicator)
if self.has_extension and stix_ioc:
iocs.append(stix_ioc)
if extension_definition:
extensions.append(extension_definition)
elif stix_ioc:
iocs.append(stix_ioc)
demisto.info(f"T2API: indicators count: {len(iocs)}")
if (
not is_manifest
and iocs
and is_demisto_version_ge("6.6.0")
and (relationships := self.create_relationships_objects(iocs, extensions))
):
demisto.info(f"T2API: indicators count: {len(relationships)}")
total += len(relationships)
iocs.extend(relationships)
iocs = sorted(iocs, key=lambda k: k["modified"])
return iocs, extensions, total
def create_manifest_entry(self, xsoar_indicator: dict, xsoar_type: str) -> dict:
"""
Args:
xsoar_indicator: to create manifest entry from
xsoar_type: type of indicator in xsoar system
Returns:
manifest entry for given indicator.
"""
if stix_type := XSOAR_TYPES_TO_STIX_SCO.get(xsoar_type):
stix_id = self.create_sco_stix_uuid(xsoar_indicator, stix_type)
elif stix_type := XSOAR_TYPES_TO_STIX_SDO.get(xsoar_type):
stix_id = self.create_sdo_stix_uuid(xsoar_indicator, stix_type, self.namespace_uuid)
else:
demisto.debug(f"No such indicator type: {xsoar_type} in stix format.")
return {}
date_added = arg_to_datetime(xsoar_indicator.get("timestamp"))
version = arg_to_datetime(xsoar_indicator.get("modified"))
demisto.debug(f"T2API: create_manifest_entry {xsoar_indicator.get('timestamp')=} {xsoar_indicator.get('modified')=}")
entry = {
"id": stix_id,
"date_added": date_added.strftime(STIX_DATE_FORMAT) if date_added else "",
}
if self.server_version == TAXII_VER_2_1:
entry["version"] = version.strftime(STIX_DATE_FORMAT) if version else ""
demisto.debug(f"T2API: create_manifest_entry {entry=}")
return entry
def create_stix_object(self, xsoar_indicator: dict, xsoar_type: str, extensions_dict: dict = {}) -> tuple[dict, dict, dict]:
"""
Args:
xsoar_indicator: to create stix object entry from
xsoar_type: type of indicator in xsoar system
extensions_dict: dict contains all object types that already have their extension defined
Returns:
Stix object entry for given indicator, and extension. Format described here:
(https://docs.google.com/document/d/1wE2JibMyPap9Lm5-ABjAZ02g098KIxlNQ7lMMFkQq44/edit#heading=h.naoy41lsrgt0)
extensions_dict: dict contains all object types that already have their extension defined
"""
is_sdo = False
if stix_type := XSOAR_TYPES_TO_STIX_SCO.get(xsoar_type):
stix_id = self.create_sco_stix_uuid(xsoar_indicator, stix_type)
object_type = stix_type
elif stix_type := XSOAR_TYPES_TO_STIX_SDO.get(xsoar_type):
stix_id = self.create_sdo_stix_uuid(xsoar_indicator, stix_type, self.namespace_uuid)
object_type = stix_type
is_sdo = True
else:
demisto.debug(f"No such indicator type: {xsoar_type} in stix format.")
return {}, {}, {}
indicator_value = xsoar_indicator.get("value")
if (stix_type == "file") and (get_hash_type(indicator_value) == "Unknown"):
demisto.debug(f"Skip indicator of type 'file' with value: '{indicator_value}', as it is not a valid hash.")
return {}, {}, {}
demisto.debug(f"T2API: {xsoar_indicator=}")
timestamp_datetime = arg_to_datetime(xsoar_indicator.get("timestamp", ""))
created_parsed = timestamp_datetime.strftime(STIX_DATE_FORMAT) if timestamp_datetime else ""
demisto.debug(f"T2API: {created_parsed=}")
try:
modified_datetime = arg_to_datetime(xsoar_indicator.get("modified", ""))
modified_parsed = modified_datetime.strftime(STIX_DATE_FORMAT) if modified_datetime else ""
demisto.debug(f"T2API: {modified_parsed=}")
except Exception:
modified_parsed = ""
# Properties required for STIX objects in all versions: id, type, created, modified.
stix_object: Dict[str, Any] = {
"id": stix_id,
"type": object_type,
"spec_version": self.server_version,
"created": created_parsed,
"modified": modified_parsed,
}
demisto.debug(f"T2API: {stix_object=}")
if xsoar_type == ThreatIntel.ObjectsNames.REPORT:
stix_object["object_refs"] = [
ref["objectstixid"] for ref in xsoar_indicator["CustomFields"].get("reportobjectreferences", [])
]
if is_sdo:
stix_object["name"] = indicator_value
stix_object = self.add_sdo_required_field_2_1(stix_object, xsoar_indicator)
stix_object = self.add_sdo_required_field_2_0(stix_object, xsoar_indicator)
else:
stix_object = self.build_sco_object(stix_object, xsoar_indicator)
xsoar_indicator_to_return = {}
# filter only requested fields
if self.has_extension and self.fields_to_present:
# if Server fields_to_present is None - no filters, return all. If Existing fields - filter
for field in self.fields_to_present:
value = xsoar_indicator.get(field)
if not value:
value = (xsoar_indicator.get("CustomFields") or {}).get(field)
xsoar_indicator_to_return[field] = value
else:
xsoar_indicator_to_return = xsoar_indicator
extension_definition = {}
if self.has_extension:
stix_object, extension_definition, extensions_dict = self.create_extension_definition(
object_type, extensions_dict, xsoar_type, created_parsed, modified_parsed, stix_object, xsoar_indicator_to_return
)
if is_sdo:
stix_object["description"] = (xsoar_indicator.get("CustomFields") or {}).get("description", "")
demisto.debug(f"T2API: at the end of the function create_stix_object {stix_object=}")
return stix_object, extension_definition, extensions_dict
def handle_report_relationships(self, relationships: list[dict[str, Any]], stix_iocs: list[dict[str, Any]]):
"""Handle specific behavior of report relationships.
Args:
relationships (list[dict[str, Any]]): the created relationships list.
stix_iocs (list[dict[str, Any]]): the ioc objects.
"""
id_to_report_objects = {stix_ioc.get("id"): stix_ioc for stix_ioc in stix_iocs if stix_ioc.get("type") == "report"}
for relationship in relationships:
if source_report := id_to_report_objects.get(relationship.get("source_ref")):
object_refs = source_report.get("object_refs", [])
object_refs.extend([relationship.get("target_ref"), relationship.get("id")])
source_report["object_refs"] = sorted(object_refs)
if target_report := id_to_report_objects.get(relationship.get("target_ref")):
object_refs = target_report.get("object_refs", [])
object_refs.extend([relationship.get("source_ref"), relationship.get("id")])
target_report["object_refs"] = sorted(object_refs)
@staticmethod
def get_stix_object_value(stix_ioc):
demisto.debug(f"{stix_ioc=}")
if stix_ioc.get("type") == "file":
for hash_type in ["SHA-256", "MD5", "SHA-1", "SHA-512"]:
if hash_value := stix_ioc.get("hashes", {}).get(hash_type):
return hash_value
return None
else:
return stix_ioc.get("value") or stix_ioc.get("name")
def create_extension_definition(
self, object_type, extensions_dict, xsoar_type, created_parsed, modified_parsed, stix_object, xsoar_indicator_to_return
):
"""
Args:
object_type: the type of the stix_object.
xsoar_type: type of indicator in xsoar system.
extensions_dict: dict contains all object types that already have their extension defined.
created_parsed: the stix object creation time.
modified_parsed: the stix object last modified time.
stix_object: Stix object entry.
xsoar_indicator_to_return: the xsoar indicator to return.
Create an extension definition and update the stix object and extensions dict accordingly.
Returns:
the updated Stix object, its extension and updated extensions_dict.
"""
extension_definition = {}
xsoar_indicator_to_return["extension_type"] = "property_extension"
extension_id = f"extension-definition--{uuid.uuid4()}"
if object_type not in extensions_dict:
extension_definition = {
"id": extension_id,
"type": "extension-definition",
"spec_version": self.server_version,
"name": f"Cortex XSOAR TIM {xsoar_type}",
"description": "This schema adds TIM data to the object",
"created": created_parsed,
"modified": modified_parsed,
"created_by_ref": f"identity--{PAWN_UUID!s}",
"schema": "https://github.com/demisto/content/blob/4265bd5c71913cd9d9ed47d9c37d0d4d3141c3eb/"
"Packs/TAXIIServer/doc_files/XSOAR_indicator_schema.json",
"version": "1.0",
"extension_types": ["property-extension"],
}
extensions_dict[object_type] = True
stix_object["extensions"] = {extension_id: xsoar_indicator_to_return}
return stix_object, extension_definition, extensions_dict
def convert_sco_to_indicator_sdo(self, stix_object: dict, xsoar_indicator: dict) -> dict:
"""
Create a STIX domain object of 'indicator' type from a STIX Cyber Observable Objects.
Args:
stix_object: The STIX Cyber Observable Object
xsoar_indicator: The stix object entry from which the 'stix_object' has been created.
Returns:
Stix indicator domain object for given indicator. Format described here:
https://docs.oasis-open.org/cti/stix/v2.1/cs01/stix-v2.1-cs01.html#_muftrcpnf89v
"""
try:
demisto.debug(f"T2API: convert_sco_to_indicator_sdo {xsoar_indicator.get('expiration')=}")
expiration_datetime = arg_to_datetime(xsoar_indicator.get("expiration"))
expiration_parsed = expiration_datetime.strftime(STIX_DATE_FORMAT) if expiration_datetime else ""
demisto.debug(f"T2API: convert_sco_to_indicator_sdo {expiration_parsed=}")
except Exception:
expiration_parsed = ""
indicator_value = xsoar_indicator.get("value")
if isinstance(indicator_value, str):
indicator_pattern_value: Any = indicator_value.replace("'", "\\'")
else:
indicator_pattern_value = json.dumps(indicator_value)
object_type = stix_object["type"]
stix_type = "indicator"
pattern = ""
if object_type == "file":
hash_type = HASH_TYPE_TO_STIX_HASH_TYPE.get(get_hash_type(indicator_value), "Unknown")
pattern = f"[file:hashes.'{hash_type}' = '{indicator_pattern_value}']"
else:
pattern = f"[{object_type}:value = '{indicator_pattern_value}']"
score_labels = self.get_labels_for_indicator(xsoar_indicator.get("score")) or []
custom_tags = (xsoar_indicator.get("CustomFields") or {}).get("tags") or []
merged_labels = list({*score_labels, *[str(t).lower().replace(" ", "-") for t in custom_tags]})
labels = merged_labels or score_labels
stix_domain_object: Dict[str, Any] = assign_params(
type=stix_type,
id=self.create_sdo_stix_uuid(xsoar_indicator, stix_type, self.namespace_uuid),
pattern=pattern,
valid_from=stix_object["created"],
valid_until=expiration_parsed,
description=(xsoar_indicator.get("CustomFields") or {}).get("description", ""),
pattern_type="stix",
labels=labels,
)
return dict(
{k: v for k, v in stix_object.items() if k in ("spec_version", "created", "modified", "extensions")},
**stix_domain_object,
)
@staticmethod
def create_sdo_stix_uuid(
xsoar_indicator: dict, stix_type: str | None, uuid_value: uuid.UUID, value: str | None = None
) -> str:
"""
Create uuid for SDO objects.
Args:
xsoar_indicator: dict - The XSOAR representation of the indicator.
stix_type: Optional[str] - The indicator type according to STIX.
value: str - The value of the indicator.
Returns:
The uuid that represents the indicator according to STIX.
"""
if stixid := (xsoar_indicator.get("CustomFields") or {}).get("stixid"):
return stixid
value = value if value else xsoar_indicator.get("value")
if stix_type == "attack-pattern":
if mitre_id := xsoar_indicator.get("CustomFields", {}).get("mitreid"):
unique_id = uuid.uuid5(uuid_value, f"{stix_type}:{mitre_id}")
else:
unique_id = uuid.uuid5(uuid_value, f"{stix_type}:{value}")
else:
unique_id = uuid.uuid5(uuid_value, f"{stix_type}:{value}")
return f"{stix_type}--{unique_id}"
@staticmethod
def create_sco_stix_uuid(xsoar_indicator: dict, stix_type: str | None, value: str | None = None) -> str:
"""
Create uuid for sco objects.
"""
if stixid := (xsoar_indicator.get("CustomFields") or {}).get("stixid"):
return stixid
if not value:
value = xsoar_indicator.get("value")
if stix_type == "user-account":
account_type = (xsoar_indicator.get("CustomFields") or {}).get("accounttype")
user_id = (xsoar_indicator.get("CustomFields") or {}).get("userid")
unique_id = uuid.uuid5(
SCO_DET_ID_NAMESPACE, f'{{"account_login":"{value}","account_type":"{account_type}","user_id":"{user_id}"}}'
)
elif stix_type == "windows-registry-key":
unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"key":"{value}"}}')
elif stix_type == "file":
if get_hash_type(value) == "md5":
unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"hashes":{{"MD5":"{value}"}}}}')
elif get_hash_type(value) == "sha1":
unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"hashes":{{"SHA-1":"{value}"}}}}')
elif get_hash_type(value) == "sha256":
unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"hashes":{{"SHA-256":"{value}"}}}}')
elif get_hash_type(value) == "sha512":
unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"hashes":{{"SHA-512":"{value}"}}}}')
else:
unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"value":"{value}"}}')
else:
unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"value":"{value}"}}')
stix_id = f"{stix_type}--{unique_id}"
return stix_id
def create_entity_b_stix_objects(self, relationships: list[dict[str, Any]], iocs_value_to_id: dict, extensions: list) -> list:
"""
Generates a list of STIX objects for the 'entityB' values in the provided 'relationships' list.
:param relationships: A list of dictionaries representing relationships between entities
:param iocs_value_to_id: A dictionary mapping IOC values to their corresponding ID values.
:param extensions: A list of dictionaries representing extension properties to include in the generated STIX objects.
:return: A list of dictionaries representing STIX objects for the 'entityB' values
"""
entity_b_objects: list[dict[str, Any]] = []
entity_b_values = ""
for relationship in relationships:
if relationship:
if (relationship.get("CustomFields") or {}).get("revoked", False):
continue
if (entity_b_value := relationship.get("entityB")) and entity_b_value not in iocs_value_to_id:
iocs_value_to_id[entity_b_value] = ""
entity_b_values += f'"{entity_b_value}" '
else:
demisto.debug(f"relationship is empty {relationship=}")
if not entity_b_values:
return entity_b_objects
try:
found_indicators = demisto.searchIndicators(query=f"value:({entity_b_values})").get("iocs") or []
except AttributeError:
demisto.debug(f"Could not find indicators from using query value:({entity_b_values})")
found_indicators = []
extensions_dict: dict = {}
for xsoar_indicator in found_indicators:
if xsoar_indicator:
xsoar_type = xsoar_indicator.get("indicator_type")
stix_ioc, extension_definition, extensions_dict = self.create_stix_object(
xsoar_indicator, xsoar_type, extensions_dict
)
if XSOAR_TYPES_TO_STIX_SCO.get(xsoar_type) in self.types_for_indicator_sdo:
stix_ioc = self.convert_sco_to_indicator_sdo(stix_ioc, xsoar_indicator)
demisto.debug(f"T2API: create_entity_b_stix_objects {stix_ioc=}")
if self.has_extension and stix_ioc:
entity_b_objects.append(stix_ioc)
if extension_definition:
extensions.append(extension_definition)
elif stix_ioc:
entity_b_objects.append(stix_ioc)
else:
demisto.debug(f"{xsoar_indicator=} is emtpy")
iocs_value_to_id[(self.get_stix_object_value(stix_ioc))] = stix_ioc.get("id") if stix_ioc else None
demisto.debug(f"Generated {len(entity_b_objects)} STIX objects for 'entityB' values.")
return entity_b_objects
def create_relationships_objects(self, stix_iocs: list[dict[str, Any]], extensions: list) -> list[dict[str, Any]]:
"""
Create entries for the relationships returned by the searchRelationships command.
:param stix_iocs: Entries for the Stix objects associated with given indicators
:param extensions: A list of dictionaries representing extension properties to include in the generated STIX objects.
:return: A list of dictionaries representing the relationships objects, including entityBs objects
"""
relationships_list: list[dict[str, Any]] = []
iocs_value_to_id = {self.get_stix_object_value(stix_ioc): stix_ioc.get("id") for stix_ioc in stix_iocs}
search_relationships = demisto.searchRelationships({"entities": list(iocs_value_to_id.keys())}).get("data") or []
demisto.debug(f"Found {len(search_relationships)} relationships for {len(iocs_value_to_id)} Stix IOC values.")
relationships_list.extend(self.create_entity_b_stix_objects(search_relationships, iocs_value_to_id, extensions))
for relationship in search_relationships:
if demisto.get(relationship, "CustomFields.revoked"):
continue
if not iocs_value_to_id.get(relationship.get("entityB")):
demisto.debug(f'TAXII: {iocs_value_to_id=} When {relationship.get("entityB")=}')
demisto.debug(
f"WARNING: Invalid entity B - Relationships will not be created to entity A:"
f" {relationship.get('entityA')} with relationship name {relationship.get('name')}"
)
continue
try:
demisto.debug(f"T2API: in create_relationships_objects {relationship=}")
created_datetime = arg_to_datetime(relationship.get("createdInSystem"))
modified_datetime = arg_to_datetime(relationship.get("modified"))
created_parsed = created_datetime.strftime(STIX_DATE_FORMAT) if created_datetime else ""
modified_parsed = modified_datetime.strftime(STIX_DATE_FORMAT) if modified_datetime else ""
demisto.debug(f"T2API: {created_parsed=} {modified_parsed=}")
except Exception as e:
created_parsed, modified_parsed = "", ""
demisto.debug(f"Error parsing dates for relationship {relationship.get('id')}: {e}")
relationship_unique_id = uuid.uuid5(self.namespace_uuid, f'relationship:{relationship.get("id")}')
relationship_stix_id = f"relationship--{relationship_unique_id}"
relationship_object: dict[str, Any] = {
"type": "relationship",
"spec_version": self.server_version,
"id": relationship_stix_id,
"created": created_parsed,
"modified": modified_parsed,
"relationship_type": relationship.get("name"),
"source_ref": iocs_value_to_id.get(relationship.get("entityA")),
"target_ref": iocs_value_to_id.get(relationship.get("entityB")),
}
if description := demisto.get(relationship, "CustomFields.description"):
relationship_object["Description"] = description
relationships_list.append(relationship_object)
self.handle_report_relationships(relationships_list, stix_iocs)
return relationships_list
def add_sdo_required_field_2_1(self, stix_object: Dict[str, Any], xsoar_indicator: Dict[str, Any]) -> Dict[str, Any]:
"""
Args:
stix_object: A stix object from
xsoar_type: indicator from xsoar system
Returns:
Stix object entry for given indicator
"""
if self.server_version == TAXII_VER_2_1:
custom_fields = xsoar_indicator.get("CustomFields", {}) or {}
stix_type = stix_object["type"]
if stix_type == "malware":
stix_object["is_family"] = custom_fields.get("ismalwarefamily", False)
elif stix_type == "report" and (published := custom_fields.get("published")):
stix_object["published"] = published
if stix_type in {"indicator", "malware", "report", "threat-actor", "tool"}:
tags = custom_fields.get("tags") or [stix_object["type"]]
stix_object["labels"] = [str(x).lower().replace(" ", "-") for x in tags]
return stix_object
def add_sdo_required_field_2_0(self, stix_object: Dict[str, Any], xsoar_indicator: Dict[str, Any]) -> Dict[str, Any]:
"""
Args:
stix_object: A stix object from
xsoar_type: indicator from xsoar system
Returns:
Stix object entry for given indicator
"""
if self.server_version == TAXII_VER_2_0:
custom_fields = xsoar_indicator.get("CustomFields", {}) or {}
stix_type = stix_object["type"]
if stix_type in {"indicator", "malware", "report", "threat-actor", "tool"}:
tags = custom_fields.get("tags", []) if custom_fields.get("tags", []) != [] else [stix_object["type"]]
stix_object["labels"] = [x.lower().replace(" ", "-") for x in tags]
if stix_type == "identity" and (identity_class := custom_fields.get("identityclass", "unknown")):
stix_object["identity_class"] = identity_class
return stix_object
def create_x509_certificate_subject_issuer(self, list_dict_values: list) -> str:
"""
Args:
dict_values: A dict with keys and values for subject/issuer
Example: [{'title': 'title', 'data': 'data'}, {'title': 'title1', 'data': 'data1'}]
Returns:
A string
Example: 'title=data, title1=data1'
"""
string_to_return = ""
if list_dict_values:
for dict_values in list_dict_values:
title = dict_values.get("title")
data = dict_values.get("data")
if data is not None:
string_to_return += f"{title}={data}, "
string_to_return = string_to_return.rstrip(", ")
return string_to_return
return ""
def create_x509_certificate_object(self, stix_object: Dict[str, Any], xsoar_indicator: Dict[str, Any]) -> dict:
"""
Builds a correct JSON object for specific x509 certificate.
Args:
stix_object (Dict[str, Any]): A JSON object of a STIX indicator
xsoar_indicator (Dict[str, Any]): A JSON object of an XSOAR indicator
Returns:
Dict[str, Any]: A JSON object of a STIX indicator.
"""
custom_fields = xsoar_indicator.get("CustomFields") or {}
stix_object["validity_not_before"] = custom_fields.get("validitynotbefore")
stix_object["validity_not_after"] = custom_fields.get("validitynotafter")
stix_object["serial_number"] = xsoar_indicator.get("value")
stix_object["subject"] = self.create_x509_certificate_subject_issuer(custom_fields.get("subject", []))
stix_object["issuer"] = self.create_x509_certificate_subject_issuer(custom_fields.get("issuer", []))
remove_nulls_from_dictionary(stix_object)
return stix_object
def build_sco_object(self, stix_object: Dict[str, Any], xsoar_indicator: Dict[str, Any]) -> Dict[str, Any]:
"""
Builds a correct JSON object for specific SCO types
Args:
stix_object (Dict[str, Any]): A JSON object of a STIX indicator
xsoar_indicator (Dict[str, Any]): A JSON object of an XSOAR indicator
Returns:
Dict[str, Any]: A JSON object of a STIX indicator
"""
custom_fields = xsoar_indicator.get("CustomFields") or {}
if stix_object["type"] == "autonomous-system":
# number is the only required field for autonomous-system
stix_object["number"] = xsoar_indicator.get("value", "")
stix_object["name"] = custom_fields.get("name", "")
elif stix_object["type"] == "file":
# hashes is the only required field for file
value = xsoar_indicator.get("value")
stix_object["hashes"] = {HASH_TYPE_TO_STIX_HASH_TYPE[get_hash_type(value)]: value}
for hash_type in ("md5", "sha1", "sha256", "sha512"):
try:
stix_object["hashes"][HASH_TYPE_TO_STIX_HASH_TYPE[hash_type]] = custom_fields[hash_type]
except KeyError:
pass
elif stix_object["type"] == "windows-registry-key":
# key is the only required field for windows-registry-key
stix_object["key"] = xsoar_indicator.get("value")
stix_object["values"] = []
for keyvalue in custom_fields["keyvalue"]:
if keyvalue:
stix_object["values"].append(keyvalue)
stix_object["values"][-1]["data_type"] = stix_object["values"][-1]["type"]
del stix_object["values"][-1]["type"]
else:
pass
elif stix_object["type"] in ("mutex", "software"):
stix_object["name"] = xsoar_indicator.get("value")
# user_id is the only required field for user-account
elif stix_object["type"] == "user-account":
user_id = (xsoar_indicator.get("CustomFields") or {}).get("userid")
if user_id:
stix_object["user_id"] = user_id
elif stix_object["type"] == "x509-certificate":
self.create_x509_certificate_object(stix_object, xsoar_indicator)
# ipv4-addr or ipv6-addr or URL
else:
stix_object["value"] = xsoar_indicator.get("value")
return stix_object
@staticmethod
def get_labels_for_indicator(score):
"""Get indicator label based on the DBot score"""
return {0: [""], 1: ["benign"], 2: ["anomalous-activity"], 3: ["malicious-activity"]}.get(int(score))
class STIX2XSOARParser(BaseClient):
def __init__(
self,
id_to_object: dict[str, Any],
verify: bool = True,
base_url: str | None = None,
proxy: bool = False,
tlp_color: str | None = None,
field_map: dict | None = None,
skip_complex_mode: bool = False,
tags: list | None = None,
update_custom_fields: bool = False,
enrichment_excluded: bool = False,
):
super().__init__(base_url=base_url, verify=verify, proxy=proxy)
self.skip_complex_mode = skip_complex_mode
self.indicator_regexes = [
re.compile(INDICATOR_EQUALS_VAL_PATTERN),
re.compile(HASHES_EQUALS_VAL_PATTERN),
]
self.tlp_color = tlp_color
self.id_to_object = id_to_object
self.cidr_regexes = [
re.compile(CIDR_ISSUBSET_VAL_PATTERN),
re.compile(CIDR_ISUPPERSET_VAL_PATTERN),
]
self.field_map = field_map or {}
self.update_custom_fields = update_custom_fields
self.tags = tags or []
self.last_fetched_indicator__modified = None
self.enrichment_excluded = enrichment_excluded
@staticmethod
def get_pattern_comparisons(pattern: str, supported_only: bool = True) -> PatternComparisons | None:
"""
Parses a pattern and comparison and extracts the comparisons as a dictionary.
If the pattern is invalid, the return value will be "None".
For Example:
>>> STIX2XSOARParser.get_pattern_comparisons(
>>> "[ipv4-addr:value = '1.1.1.1/32' "
>>> "OR ipv4-addr:value = '8.8.8.8/32' "
>>> "AND domain-name:value = 'example.com' "
>>> "OR file:hashes.'SHA-256' = '13987239847...']"
>>> )
{
'ipv4-addr': [(['value'], '=', "'1.1.1.1/32'"), (['value'], '=', "'8.8.8.8/32'")],
'domain-name': [(['value'], '=', "'example.com'")],
'file': [(['hashes', 'SHA-256'], '=', "'13987239847...'")]
}
Args:
pattern: the pattern to extract the value from.
supported_only: Whether to remove comparisons that are not supported by Cortex XSOAR.
Returns:
Optional[PatternComparisons]. the value in the pattern.
"""
try:
comparisons = cast(PatternComparisons, Pattern(pattern).inspect().comparisons)
return STIX2XSOARParser.get_supported_pattern_comparisons(comparisons) if supported_only else comparisons
except Exception as error:
demisto.debug(f"Unable to parse {pattern=}, {error=}")
return None
@staticmethod
def get_supported_pattern_comparisons(comparisons: PatternComparisons) -> PatternComparisons:
"""
Get only the patterns supported by XSOAR from a parsed pattern.
Args:
comparisons: The comparisons of the pattern to extract the supported values from.
Returns:
PatternComparisons. the value in the pattern.
"""
def get_comparison_field(comparison: tuple[list[str], str, str]) -> str:
"""retrieves the field of a STIX comparison."""
return cast(str, dict_safe_get(comparison, [0, 0]))
supported_comparisons: PatternComparisons = {}
for indicator_type, comps in comparisons.items():
if indicator_type in STIX_SUPPORTED_TYPES:
field_comparisons = [
comp for comp in comps if (get_comparison_field(comp) in STIX_SUPPORTED_TYPES[indicator_type])
]
if field_comparisons:
supported_comparisons[indicator_type] = field_comparisons
return supported_comparisons
@staticmethod
def get_indicator_publication(indicator: dict[str, Any], ignore_external_id: bool = False):
"""
Build publications grid field from the indicator external_references field
Args:
indicator: The indicator with publication field.
ignore_external_id: Whether to ignore external_id or not.
Returns:
list. publications grid field
"""
publications = []
for external_reference in indicator.get("external_references", []):
if ignore_external_id and external_reference.get("external_id"):
continue
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["type"] = f'STIX {indicator["type"]}'
indicator["fields"]["stixkillchainphases"] = indicator["fields"].pop("killchainphases", None)
indicator["fields"]["stixdescription"] = indicator["fields"].pop("description", None)
return indicator
@staticmethod
def get_entity_b_type_and_value(
related_obj: str, id_to_object: dict[str, dict[str, Any]], is_unit42_report: bool = False
) -> tuple:
"""
Gets the type and value of the indicator in entity_b.
Args:
related_obj: the indicator to get information on.
id_to_object: a dict in the form of - id: stix_object.
is_unit42_report: represents whether unit42 report or not.
Returns:
tuple. the indicator type and value.
"""
indicator_obj = id_to_object.get(related_obj, {})
entity_b_value = indicator_obj.get("name", "")
entity_b_obj_type = STIX_2_TYPES_TO_CORTEX_TYPES.get(
indicator_obj.get("type", ""), STIX2XSOARParser.get_ioc_type(related_obj, id_to_object)
)
if indicator_obj.get("type") == "indicator":
entity_b_value = STIX2XSOARParser.get_single_pattern_value(id_to_object.get(related_obj, {}).get("pattern", ""))
elif indicator_obj.get("type") == "attack-pattern" and is_unit42_report:
_, entity_b_value = STIX2XSOARParser.get_mitre_attack_id_and_value_from_name(indicator_obj)
elif indicator_obj.get("type") == "report" and is_unit42_report:
entity_b_value = f"[Unit42 ATOM] {indicator_obj.get('name')}"
return entity_b_obj_type, entity_b_value
@staticmethod
def get_mitre_attack_id_and_value_from_name(attack_indicator):
"""
Split indicator name into MITRE ID and indicator value: 'T1108: Redundant Access' -> MITRE ID = T1108,
indicator value = 'Redundant Access'.
"""
ind_name = attack_indicator.get("name")
separator = ":"
try:
partition_result = ind_name.partition(separator)
if partition_result[1] != separator:
raise DemistoException(f"Failed parsing attack indicator {ind_name}")
except ValueError:
raise DemistoException(f"Failed parsing attack indicator {ind_name}")
ind_id = partition_result[0]
value = partition_result[2].strip()
if attack_indicator.get("x_mitre_is_subtechnique"):
value = attack_indicator.get("x_panw_parent_technique_subtechnique")
return ind_id, value
@staticmethod
def parse_report_relationships(
report_obj: dict[str, Any],
id_to_object: dict[str, dict[str, Any]],
relationships_prefix: str = "",
ignore_reports_relationships: bool = False,
is_unit42_report: bool = False,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
obj_refs = report_obj.get("object_refs", [])
relationships: list[dict[str, Any]] = []
obj_refs_excluding_relationships_prefix = []
for related_obj in obj_refs:
# relationship-- objects ref handled in parse_relationships
if not related_obj.startswith("relationship--"):
if ignore_reports_relationships and related_obj.startswith("report--"):
continue
obj_refs_excluding_relationships_prefix.append(related_obj)
if id_to_object.get(related_obj):
entity_b_obj_type, entity_b_value = STIX2XSOARParser.get_entity_b_type_and_value(
related_obj, id_to_object, is_unit42_report
)
if not entity_b_obj_type:
demisto.debug(f"Could not find the type of {related_obj} skipping.")
continue
relationships.append(
EntityRelationship(
name="related-to",
entity_a=f"{relationships_prefix}{report_obj.get('name')}",
entity_a_type=ThreatIntel.ObjectsNames.REPORT,
entity_b=entity_b_value,
entity_b_type=entity_b_obj_type,
).to_indicator()
)
return relationships, obj_refs_excluding_relationships_prefix
@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}"):
if STIX2XSOARParser.is_supported_iocs_type(pattern):
ioc_type = STIX_2_TYPES_TO_CORTEX_TYPES.get(stix_type, "") # type: ignore
break
demisto.debug(f"Indicator {indicator_obj.get('id')} is not supported indicator.")
return ioc_type
@staticmethod
def is_supported_iocs_type(pattern: str):
"""
Get pattern and check if the type is supported by XSOAR.
Args:
pattern: the indicator pattern.
Returns:
bool.
"""
return any(
any(pattern.startswith(f"[{key}:{field}") for field in STIX_SUPPORTED_TYPES[key]) for key in STIX_SUPPORTED_TYPES
)
@staticmethod
def get_tlp(indicator_json: dict) -> str:
object_marking_definition_list = indicator_json.get("object_marking_refs", "")
tlp_color: str = ""
for object_marking_definition in object_marking_definition_list:
if tlp := MARKING_DEFINITION_TO_TLP.get(object_marking_definition):
tlp_color = tlp
break
return tlp_color
def set_default_fields(self, obj_to_parse):
fields = {
"stixid": obj_to_parse.get("id", ""),
"firstseenbysource": obj_to_parse.get("created", ""),
"modified": obj_to_parse.get("modified", ""),
"description": obj_to_parse.get("description", ""),
}
tlp_from_marking_refs = self.get_tlp(obj_to_parse)
tlp_color = tlp_from_marking_refs if tlp_from_marking_refs else self.tlp_color
if obj_to_parse.get("confidence", ""):
fields["confidence"] = obj_to_parse.get("confidence", "")
if obj_to_parse.get("lang", ""):
fields["languages"] = obj_to_parse.get("lang", "")
if tlp_color:
fields["trafficlightprotocol"] = tlp_color
obj_to_parse["trafficlightprotocol"] = tlp_color
return fields
@staticmethod
def parse_custom_fields(extensions):
custom_fields = {}
score = None
for key, value in extensions.items():
if key.startswith("extension-definition--"):
custom_fields = value.get("CustomFields", {})
if not custom_fields:
custom_fields = value
score = value.get("score", None)
break
return custom_fields, score
@staticmethod
def get_single_pattern_value(pattern: str) -> str | None:
"""
Parses a pattern with a single comparison and extracts the right hand value of the comparison.
If the pattern is invalid, the pattern itself will be returned.
For Example:
>>> STIX2XSOARParser.get_single_pattern_value("[domain-name:value = 'www.example.com']")
'www.example.com'
Args:
pattern: the pattern to extract the value from.
Returns:
str. the value in the pattern.
"""
comparisons = STIX2XSOARParser.get_pattern_comparisons(pattern) or {}
if comparisons:
return dict_safe_get(tuple(comparisons.values()), [0, 0, -1], "", str).strip("'") or None
return None
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 = self.field_map or {}
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,
)
)
return indicators
def parse_attack_pattern(self, attack_pattern_obj: dict[str, Any], ignore_external_id: bool = False) -> 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 = self.get_indicator_publication(attack_pattern_obj, ignore_external_id)
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]
attack_pattern = {
"value": attack_pattern_obj.get("name"),
"type": ThreatIntel.ObjectsNames.ATTACK_PATTERN,
"score": ThreatIntel.ObjectsScore.ATTACK_PATTERN,
"rawJSON": attack_pattern_obj,
}
fields = self.set_default_fields(attack_pattern_obj)
fields.update(
{
"killchainphases": kill_chain_phases,
"operatingsystemrefs": attack_pattern_obj.get("x_mitre_platforms"),
"publications": publications,
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(attack_pattern_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
attack_pattern["score"] = score
fields["tags"] = list(set(attack_pattern_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", []))))
attack_pattern_obj["tags"] = fields["tags"]
attack_pattern["fields"] = 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 = self.change_attack_pattern_to_stix_attack_pattern(attack_pattern)
if self.enrichment_excluded:
attack_pattern["enrichmentExcluded"] = self.enrichment_excluded
return [attack_pattern]
def create_obj_refs_list(self, obj_refs_list: list):
"""
Creates a list of object references for a STIX report type and organize it for an XSOAR "object refs" grid field.
:param obj_refs_list: A list of obj refs
:return: A list of dicts.
"""
# remove duplicates
obj_refs_list_result = []
obj_refs_list_without_dup = list(dict.fromkeys(obj_refs_list))
omitted_object_number = len(obj_refs_list) - len(obj_refs_list_without_dup)
demisto.debug(f"Omitting {omitted_object_number} object ref form the report")
if obj_refs_list:
obj_refs_list_result.extend([{"objectstixid": object} for object in obj_refs_list_without_dup])
return obj_refs_list_result
def parse_report(
self,
report_obj: dict[str, Any],
relationships_prefix: str = "",
ignore_reports_relationships: bool = False,
is_unit42_report: bool = False,
) -> list[dict[str, Any]]:
"""
Parses a single report object
:param report_obj: report object
:return: report extracted from the report object in cortex format
"""
report = {
"type": ThreatIntel.ObjectsNames.REPORT,
"value": report_obj.get("name"),
"score": ThreatIntel.ObjectsScore.REPORT,
"rawJSON": report_obj,
}
fields = self.set_default_fields(report_obj)
fields.update(
{
"published": report_obj.get("published"),
"report_types": report_obj.get("report_types", []),
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(report_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
report["score"] = score
fields["tags"] = list(set(report_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", []))))
report_obj["tags"] = fields["tags"]
relationships, obj_refs_excluding_relationships_prefix = self.parse_report_relationships(
report_obj, self.id_to_object, relationships_prefix, ignore_reports_relationships, is_unit42_report
)
report["relationships"] = relationships
if obj_refs_excluding_relationships_prefix:
fields["Report Object References"] = self.create_obj_refs_list(obj_refs_excluding_relationships_prefix)
report["fields"] = fields
if self.enrichment_excluded:
report["enrichmentExcluded"] = self.enrichment_excluded
return [report]
def parse_threat_actor(self, 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"),
"type": ThreatIntel.ObjectsNames.THREAT_ACTOR,
"score": ThreatIntel.ObjectsScore.THREAT_ACTOR,
"rawJSON": threat_actor_obj,
}
fields = self.set_default_fields(threat_actor_obj)
fields.update(
{
"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", []),
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(threat_actor_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
threat_actor["score"] = score
fields["tags"] = list(set(threat_actor_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", []))))
threat_actor_obj["tags"] = fields["tags"]
threat_actor["fields"] = fields
if self.enrichment_excluded:
threat_actor["enrichmentExcluded"] = self.enrichment_excluded
return [threat_actor]
def parse_infrastructure(self, 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"),
"type": ThreatIntel.ObjectsNames.INFRASTRUCTURE,
"score": ThreatIntel.ObjectsScore.INFRASTRUCTURE,
"rawJSON": infrastructure_obj,
}
fields = self.set_default_fields(infrastructure_obj)
fields.update(
{
"infrastructure_types": infrastructure_obj.get("infrastructure_types", []),
"aliases": infrastructure_obj.get("aliases", []),
"kill_chain_phases": kill_chain_phases,
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(infrastructure_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
infrastructure["score"] = score
fields["tags"] = list(set(list(fields.get("tags", [])) + self.tags))
infrastructure_obj["tags"] = fields["tags"]
infrastructure["fields"] = fields
if self.enrichment_excluded:
infrastructure["enrichmentExcluded"] = self.enrichment_excluded
return [infrastructure]
def parse_malware(self, 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"),
"type": ThreatIntel.ObjectsNames.MALWARE,
"score": ThreatIntel.ObjectsScore.MALWARE,
"rawJSON": malware_obj,
}
fields = self.set_default_fields(malware_obj)
fields.update(
{
"malware_types": malware_obj.get("malware_types", []),
"ismalwarefamily": 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": malware_obj.get("architecture_execution_envs", []),
"capabilities": malware_obj.get("capabilities", []),
"samples": malware_obj.get("sample_refs", []),
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(malware_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
malware["score"] = score
fields["tags"] = list(set(malware_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", []))))
malware_obj["tags"] = fields["tags"]
malware["fields"] = fields
if self.enrichment_excluded:
malware["enrichmentExcluded"] = self.enrichment_excluded
return [malware]
def parse_tool(self, 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"),
"type": ThreatIntel.ObjectsNames.TOOL,
"score": ThreatIntel.ObjectsScore.TOOL,
"rawJSON": tool_obj,
}
fields = self.set_default_fields(tool_obj)
fields.update(
{
"killchainphases": kill_chain_phases,
"tool_types": tool_obj.get("tool_types", []),
"aliases": tool_obj.get("aliases", []),
"tool_version": tool_obj.get("tool_version", ""),
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(tool_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
tool["score"] = score
fields["tags"] = list(set(tool_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", []))))
tool_obj["tags"] = fields["tags"]
tool["fields"] = fields
if self.enrichment_excluded:
tool["enrichmentExcluded"] = self.enrichment_excluded
return [tool]
def parse_course_of_action(self, coa_obj: dict[str, Any], ignore_external_id: bool = False) -> 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 = self.get_indicator_publication(coa_obj, ignore_external_id)
course_of_action = {
"value": coa_obj.get("name"),
"type": ThreatIntel.ObjectsNames.COURSE_OF_ACTION,
"score": ThreatIntel.ObjectsScore.COURSE_OF_ACTION,
"rawJSON": coa_obj,
}
fields = self.set_default_fields(coa_obj)
fields.update(
{
"action_type": coa_obj.get("action_type", ""),
"publications": publications,
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(coa_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
course_of_action["score"] = score
fields["tags"] = list(set(list(fields.get("tags", [])) + self.tags))
coa_obj["tags"] = fields["tags"]
course_of_action["fields"] = fields
if self.enrichment_excluded:
course_of_action["enrichmentExcluded"] = self.enrichment_excluded
return [course_of_action]
def parse_campaign(self, 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"),
"type": ThreatIntel.ObjectsNames.CAMPAIGN,
"score": ThreatIntel.ObjectsScore.CAMPAIGN,
"rawJSON": campaign_obj,
}
fields = self.set_default_fields(campaign_obj)
fields.update(
{
"aliases": campaign_obj.get("aliases", []),
"objective": campaign_obj.get("objective", ""),
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(campaign_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
campaign["score"] = score
fields["tags"] = list(set(campaign_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", []))))
campaign_obj["tags"] = fields["tags"]
campaign["fields"] = fields
if self.enrichment_excluded:
campaign["enrichmentExcluded"] = self.enrichment_excluded
return [campaign]
def parse_intrusion_set(self, intrusion_set_obj: dict[str, Any], ignore_external_id: bool = False) -> 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 = self.get_indicator_publication(intrusion_set_obj, ignore_external_id)
intrusion_set = {
"value": intrusion_set_obj.get("name"),
"type": ThreatIntel.ObjectsNames.INTRUSION_SET,
"score": ThreatIntel.ObjectsScore.INTRUSION_SET,
"rawJSON": intrusion_set_obj,
}
fields = self.set_default_fields(intrusion_set_obj)
fields.update(
{
"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,
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(intrusion_set_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
intrusion_set["score"] = score
fields["tags"] = list(set(intrusion_set_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", []))))
intrusion_set_obj["tags"] = fields["tags"]
if self.enrichment_excluded:
intrusion_set["enrichmentExcluded"] = self.enrichment_excluded
intrusion_set["fields"] = fields
return [intrusion_set]
def parse_general_sco_indicator(self, 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,
"type": STIX_2_TYPES_TO_CORTEX_TYPES.get(sco_object.get("type")), # type: ignore[arg-type]
}
fields = self.set_default_fields(sco_object)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(sco_object.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
sco_indicator["score"] = score
fields["tags"] = list(set(list(fields.get("tags", [])) + self.tags))
sco_object["tags"] = fields["tags"]
sco_indicator["fields"] = fields
if self.enrichment_excluded:
sco_indicator["enrichmentExcluded"] = self.enrichment_excluded
return [sco_indicator]
def parse_sco_autonomous_system_indicator(self, 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.
"""
if isinstance(autonomous_system_obj, dict) and "number" in autonomous_system_obj:
autonomous_system_obj["number"] = str(autonomous_system_obj.get("number", ""))
autonomous_system_indicator = self.parse_general_sco_indicator(autonomous_system_obj, value_mapping="number")
autonomous_system_indicator[0]["fields"]["name"] = autonomous_system_obj.get("name")
return autonomous_system_indicator
def parse_sco_file_indicator(self, 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 = self.parse_general_sco_indicator(file_obj)
file_indicator[0]["fields"].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
def parse_sco_mutex_indicator(self, 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 self.parse_general_sco_indicator(sco_object=mutex_obj, value_mapping="name")
def parse_sco_software_indicator(self, software_obj: dict[str, Any]) -> list[dict[str, Any]]:
"""
Parses software indicator type to cortex format.
Args:
software_obj (dict): indicator as an observable object of software type.
"""
return self.parse_general_sco_indicator(sco_object=software_obj, value_mapping="name")
def parse_sco_account_indicator(self, 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 = self.parse_general_sco_indicator(account_obj, value_mapping="user_id")
account_indicator[0]["fields"].update(
{"displayname": account_obj.get("user_id"), "accounttype": account_obj.get("account_type")}
)
return account_indicator
def create_keyvalue_dict(self, registry_key_obj_values: list[dict[str, Any]]) -> list:
"""
Creates a grid field related to the keyvalue field of the registry key.
Args:
registry_key_obj_values (dict[str, Any]): A list of dict from the stix object.
Returns:
list: The return value. A list of dict.
"""
returned_grid = []
for stix_values_entry in registry_key_obj_values:
returned_grid.append(
{
"name": stix_values_entry.get("name", ""),
"type": stix_values_entry.get("data_type"),
"data": stix_values_entry.get("data"),
}
)
return returned_grid
def parse_sco_windows_registry_key_indicator(self, 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 = self.parse_general_sco_indicator(registry_key_obj, value_mapping="key")
registry_key_indicator[0]["fields"].update(
{
"keyvalue": self.create_keyvalue_dict(registry_key_obj.get("values", [])),
"modified_time": registry_key_obj.get("modified_time"),
"numberofsubkeys": registry_key_obj.get("number_of_subkeys"),
}
)
return registry_key_indicator
def parse_identity(self, identity_obj: dict[str, Any]) -> list[dict[str, Any]]:
"""
Parses a single identity object
:param identity_obj: identity object
:return: identity extracted from the identity object in cortex format
"""
identity = {
"value": identity_obj.get("name"),
"type": FeedIndicatorType.Identity,
"score": Common.DBotScore.NONE,
"rawJSON": identity_obj,
}
fields = self.set_default_fields(identity_obj)
fields.update(
{"identityclass": identity_obj.get("identity_class", ""), "industrysectors": identity_obj.get("sectors", [])}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(identity_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
identity["score"] = score
fields["tags"] = list(set(identity_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", []))))
identity_obj["tags"] = fields["tags"]
identity["fields"] = fields
if self.enrichment_excluded:
identity["enrichmentExcluded"] = self.enrichment_excluded
return [identity]
def parse_location(self, location_obj: dict[str, Any]) -> list[dict[str, Any]]:
"""
Parses a single location object
:param location_obj: location object
:return: location extracted from the location object in cortex format
"""
country_name = COUNTRY_CODES_TO_NAMES.get(str(location_obj.get("country", "")).upper(), "")
location = {
"value": location_obj.get("name") or country_name,
"type": FeedIndicatorType.Location,
"score": Common.DBotScore.NONE,
"rawJSON": location_obj,
}
fields = self.set_default_fields(location_obj)
fields.update(
{
"countrycode": location_obj.get("country", ""),
}
)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(location_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
location["score"] = score
fields["tags"] = list(set(location_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", []))))
location_obj["tags"] = fields["tags"]
location["fields"] = fields
if self.enrichment_excluded:
location["enrichmentExcluded"] = self.enrichment_excluded
return [location]
def parse_vulnerability(self, vulnerability_obj: dict[str, Any]) -> list[dict[str, Any]]:
"""
Parses a single vulnerability object
:param vulnerability_obj: vulnerability object
:return: vulnerability extracted from the vulnerability object in cortex format
"""
name = ""
for external_reference in vulnerability_obj.get("external_references", []):
if external_reference.get("source_name") == "cve":
name = external_reference.get("external_id")
break
cve = {"value": name, "type": FeedIndicatorType.CVE, "score": Common.DBotScore.NONE, "rawJSON": vulnerability_obj}
fields = self.set_default_fields(vulnerability_obj)
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(vulnerability_obj.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
cve["score"] = score
fields["tags"] = list(
set(vulnerability_obj.get("labels", [])).union(set(self.tags), set(fields.get("tags", [])), {name} if name else {})
)
vulnerability_obj["tags"] = fields["tags"]
cve["fields"] = fields
if self.enrichment_excluded:
cve["enrichmentExcluded"] = self.enrichment_excluded
return [cve]
def create_x509_certificate_grids(self, string_object: str | None) -> list:
"""
Creates a grid field related to the subject and issuer field of the x509 certificate object.
Args:
string_object (Optional[str]): A str in format of C=ZA, ST=Western Cape, L=Cape Town, O=Thawte.
Returns:
list: The return value. A list of dict [{"title": "C", "data": "ZA"}].
"""
result_grid_list = []
if string_object:
key_value_pairs = string_object.split(", ")
for pair in key_value_pairs:
result_grid = {}
key, value = pair.split("=", 1)
result_grid["title"] = key
result_grid["data"] = value
result_grid_list.append(result_grid)
return result_grid_list
def parse_x509_certificate(self, x509_certificate_obj: dict[str, Any]):
"""
Parses a single x509_certificate object
:param x509_certificate_obj: x509_certificate object
:return: x509_certificate extracted from the x509_certificate object in cortex format.
"""
if x509_certificate_obj.get("serial_number"):
x509_certificate = {
"value": x509_certificate_obj.get("serial_number"),
"type": FeedIndicatorType.X509,
"score": Common.DBotScore.NONE,
"rawJSON": x509_certificate_obj,
}
fields = {
"stixid": x509_certificate_obj.get("id", ""),
"validitynotbefore": x509_certificate_obj.get("validity_not_before"),
"validitynotafter": x509_certificate_obj.get("validity_not_after"),
"subject": self.create_x509_certificate_grids(x509_certificate_obj.get("subject")),
"issuer": self.create_x509_certificate_grids(x509_certificate_obj.get("issuer")),
}
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(x509_certificate.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
x509_certificate["score"] = score
fields["tags"] = list(set(list(fields.get("tags", [])) + self.tags))
x509_certificate_obj["tags"] = fields["tags"]
x509_certificate["fields"] = fields
if self.enrichment_excluded:
x509_certificate["enrichmentExcluded"] = self.enrichment_excluded
return [x509_certificate]
return []
def parse_relationships(self, relationships_lst: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Parse the Relationships objects retrieved from the feed.
Returns:
A list of processed relationships an indicator object.
"""
relationships_list = []
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_threat_intel_type = relationships_object.get("source_ref", "").split("--")[0]
a_type = THREAT_INTEL_TYPE_TO_DEMISTO_TYPES.get(a_threat_intel_type, "") or STIX_2_TYPES_TO_CORTEX_TYPES.get(
a_threat_intel_type, ""
) # type: ignore
if a_threat_intel_type == "indicator":
id = relationships_object.get("source_ref", "")
a_type = self.get_ioc_type(id, self.id_to_object)
b_threat_intel_type = relationships_object.get("target_ref", "").split("--")[0]
b_type = THREAT_INTEL_TYPE_TO_DEMISTO_TYPES.get(b_threat_intel_type, "") or STIX_2_TYPES_TO_CORTEX_TYPES.get(
b_threat_intel_type, ""
) # type: ignore
if b_threat_intel_type == "indicator":
b_type = self.get_ioc_type(relationships_object.get("target_ref", ""), self.id_to_object)
if not a_type or not b_type:
continue
mapping_fields = {
"lastseenbysource": relationships_object.get("modified"),
"firstseenbysource": relationships_object.get("created"),
}
entity_a = self.get_ioc_value(relationships_object.get("source_ref"), self.id_to_object)
entity_b = self.get_ioc_value(relationships_object.get("target_ref"), self.id_to_object)
entity_relation = EntityRelationship(
name=relationship_type,
entity_a=entity_a,
entity_a_type=a_type,
entity_b=entity_b,
entity_b_type=b_type,
fields=mapping_fields,
)
relationships_list.append(entity_relation.to_indicator())
dummy_indicator = {"value": "$$DummyIndicator$$", "relationships": relationships_list}
return [dummy_indicator] if dummy_indicator else []
@staticmethod
def extract_indicators_from_stix_objects(
stix_objs: list[dict[str, str]], required_objects: list[str]
) -> list[dict[str, str]]:
"""
Extracts indicators from taxii objects
:param stix_objs: taxii objects
:return: indicators in json format
"""
extracted_objs = [item for item in stix_objs if item.get("type") in required_objects] # retrieve only required type
demisto.debug(
f"Extracted {len(list(extracted_objs))} out of {len(list(stix_objs))} Stix objects with the types: "
f"{required_objects}"
)
return extracted_objs
def get_indicators_from_indicator_groups(
self,
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]
indicator = self.create_indicator(indicator_obj, type_, value, field_map)
indicators.append(indicator)
break
if self.skip_complex_mode and len(indicators) > 1:
# we managed to pull more than a single indicator - indicating complex relationship
return []
return indicators
def create_indicator(self, 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,
"type": type_,
"rawJSON": ioc_obj_copy,
}
fields = self.set_default_fields(indicator_obj)
tags = list(self.tags)
if self.update_custom_fields:
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)
if not fields.get("trafficlightprotocol"):
tlp_from_marking_refs = self.get_tlp(ioc_obj_copy)
fields["trafficlightprotocol"] = tlp_from_marking_refs if tlp_from_marking_refs else self.tlp_color
if fields.get("trafficlightprotocol"):
ioc_obj_copy["trafficlightprotocol"] = fields["trafficlightprotocol"]
if self.update_custom_fields:
custom_fields, score = self.parse_custom_fields(ioc_obj_copy.get("extensions", {}))
fields.update(assign_params(**custom_fields))
if score:
indicator["score"] = score
# 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"] = list(set(tags))
ioc_obj_copy["tags"] = fields["tags"]
indicator["fields"] = fields
fields["publications"] = self.get_indicator_publication(indicator_obj)
if self.enrichment_excluded:
indicator["enrichmentExcluded"] = self.enrichment_excluded
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 stix_time_to_datetime(s_time):
"""
Converts datetime to str in "%Y-%m-%dT%H:%M:%S.%fZ" format
:param s_time: time in string format
:return: datetime
"""
try:
return datetime.strptime(s_time, TAXII_TIME_FORMAT)
except ValueError:
return datetime.strptime(s_time, TAXII_TIME_FORMAT_NO_MS)
@staticmethod
def get_ioc_value(ioc, id_to_obj):
"""
Get IOC value from the indicator name/value/pattern field.
Uses extract_ioc_value to extract values for all supported indicator types from STIX patterns.
Args:
ioc: the indicator to get information on.
id_to_obj: a dict in the form of - id: stix_object.
Returns:
str. the IOC value.
if its attack pattern remove the id from the name.
"""
ioc_obj = id_to_obj.get(ioc)
if ioc_obj:
for key in ("name", "value", "pattern"):
field_value = ioc_obj.get(key, "")
if (
key in ioc_obj
and field_value
and field_value.startswith("[")
and field_value.endswith("]")
and (ioc_value := STIX2XSOARParser.extract_ioc_value(ioc_obj, key))
):
return ioc_value
return ioc_obj.get("name") or ioc_obj.get("value")
return None
@staticmethod
def extract_ioc_value(ioc_obj, key: str = "name"):
"""
Extract indicator values from a specific key; the default key is name.
Extracts values for all supported indicator types from STIX patterns.
Examples:
- "[file:hashes.'SHA-256' = '1111']" -> "1111" (SHA-256 hash)
- "[ipv4-addr:value = '1.2.3.4']" -> "1.2.3.4" (IP address)
- "[domain-name:value = 'example.com']" -> "example.com" (domain)
- "[windows-registry-key:key = 'HKEY_LOCAL_MACHINE\\Software\\Malware']" -> "HKEY_LOCAL_MACHINE\\Software\\Malware"
- "[windows-registry-key:value = 'MalwareValue']" -> "MalwareValue"
Returns the first found indicator value based on a priority order:
1. file:hashes.'SHA-256'
2. ipv4-addr:value
3. domain-name:value
4. url:value
5. Other supported indicator types defined in STIX_SUPPORTED_TYPES
If multiple indicators of the same type are found, returns the first one.
Args:
ioc_obj: The indicator object containing the pattern
key: The key in the object containing the pattern (default: "name")
Returns:
str: The extracted indicator value, or None if no supported indicators found
"""
ioc_value = ioc_obj.get(key, "")
comparisons = STIX2XSOARParser.get_pattern_comparisons(ioc_value) or {}
# Define the structure of comparison tuples for better readability
# Each comparison is a tuple of (field_path, operator, value)
# where field_path is a list of field names, operator is a string, and value is the indicator value
FIELD_PATH = 0
VALUE = 2
# Due to backward compatibility, First check for SHA-256 hash specifically
sha256 = next(
(
str(comparison[VALUE]).strip("'")
for comparison in comparisons.get("file", [])
if ["hashes", "SHA-256"] in comparison
),
None,
)
if sha256:
return sha256
# Define priority order for indicator types
priority_order = ["file", "ipv4-addr", "domain-name", "url", "email-addr", "mutex", "windows-registry-key"]
# Then check other indicator types in priority order
for indicator_type in priority_order:
if comparisons.get(indicator_type):
# For file type, look for other hash types if SHA-256 wasn't found
if indicator_type == "file":
# Check for other hash types (MD5, SHA-1, etc.)
for comparison in comparisons.get("file", []):
field_path = comparison[FIELD_PATH]
if len(comparison) >= 3 and field_path and len(field_path) >= 2 and field_path[0] == "hashes":
return str(comparison[VALUE]).strip("'")
# For all other types, check if the field is in STIX_SUPPORTED_TYPES
else:
# Get supported fields for this indicator type
supported_fields = STIX_SUPPORTED_TYPES.get(indicator_type, ())
for comparison in comparisons.get(indicator_type, []):
field_path = comparison[FIELD_PATH]
# Check if the comparison field is in the supported fields
if field_path and len(field_path) > 0 and field_path[0] in supported_fields and comparison[VALUE]:
return str(comparison[VALUE]).strip("'")
# If no indicators found in priority list, check any other supported types
for indicator_type, comparison_list in comparisons.items():
if indicator_type not in priority_order and comparison_list and indicator_type in STIX_SUPPORTED_TYPES:
supported_fields = STIX_SUPPORTED_TYPES.get(indicator_type, ())
for comparison in comparison_list:
field_path = comparison[FIELD_PATH]
if field_path and len(field_path) > 0 and field_path[0] in supported_fields and comparison[VALUE]:
return str(comparison[VALUE]).strip("'")
return None
def update_last_modified_indicator_date(self, indicator_modified_str: str):
if not indicator_modified_str:
return
if self.last_fetched_indicator__modified is None:
self.last_fetched_indicator__modified = indicator_modified_str # type: ignore[assignment]
else:
last_datetime = self.stix_time_to_datetime(self.last_fetched_indicator__modified)
indicator_created_datetime = self.stix_time_to_datetime(indicator_modified_str)
if indicator_created_datetime > last_datetime:
self.last_fetched_indicator__modified = indicator_modified_str
def load_stix_objects_from_envelope(self, envelopes: types.GeneratorType, limit: int = -1):
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,
"software": self.parse_sco_software_indicator,
"user-account": self.parse_sco_account_indicator,
"windows-registry-key": self.parse_sco_windows_registry_key_indicator,
"identity": self.parse_identity,
"location": self.parse_location,
"vulnerability": self.parse_vulnerability,
"x509-certificate": self.parse_x509_certificate,
}
indicators, relationships_lst = self.parse_generator_type_envelope(envelopes, parse_stix_2_objects, limit)
if relationships_lst:
indicators.extend(self.parse_relationships(relationships_lst))
demisto.debug(f"TAXII 2 Feed has extracted {len(list(indicators))} indicators")
return indicators
def increase_count(self, counter: Dict[str, int], id: str):
if id in counter:
counter[id] = counter[id] + 1
else:
counter[id] = 1
def parse_generator_type_envelope(self, envelopes: types.GeneratorType, parse_objects_func, limit: int = -1):
indicators = []
relationships_lst = []
# Used mainly for logging
parsed_objects_counter: Dict[str, int] = {}
try:
for envelope in envelopes:
self.increase_count(parsed_objects_counter, "envelope")
try:
stix_objects = envelope.get("objects")
if not stix_objects:
# no fetched objects
self.increase_count(parsed_objects_counter, "not-parsed-envelope-not-stix")
break
except Exception as e:
demisto.info(f"Exception trying to get envelope objects: {e}, {traceback.format_exc()}")
self.increase_count(parsed_objects_counter, "exception-envelope-get-objects")
continue
# we should build the id_to_object dict before iteration as some object reference each other
self.id_to_object.update(
{
obj.get("id"): obj
for obj in stix_objects
if obj.get("type") not in ["extension-definition", "relationship"]
}
)
# now we have a list of objects, go over each obj, save id with obj, parse the obj
for obj in stix_objects:
try:
obj_type = obj.get("type")
except Exception as e:
demisto.info(f"Exception trying to get stix_object-type: {e}, {traceback.format_exc()}")
self.increase_count(parsed_objects_counter, "exception-stix-object-type")
continue
# we currently don't support extension object
if obj_type == "extension-definition":
demisto.debug(f'There is no parsing function for object type "extension-definition", for object {obj}.')
self.increase_count(parsed_objects_counter, "not-parsed-extension-definition")
continue
elif obj_type == "relationship":
relationships_lst.append(obj)
self.increase_count(parsed_objects_counter, "not-parsed-relationship")
continue
if not parse_objects_func.get(obj_type):
demisto.debug(f"There is no parsing function for object type {obj_type}, for object {obj}.")
self.increase_count(parsed_objects_counter, f"not-parsed-{obj_type}")
continue
try:
if result := parse_objects_func[obj_type](obj):
indicators.extend(result)
self.update_last_modified_indicator_date(obj.get("modified"))
except Exception as e:
demisto.info(f"Exception parsing stix_object-type {obj_type}: {e}, {traceback.format_exc()}")
self.increase_count(parsed_objects_counter, f"exception-parsing-{obj_type}")
continue
self.increase_count(parsed_objects_counter, f"parsed-{obj_type}")
if reached_limit(limit, len(indicators)):
demisto.debug(
f"Reached the limit ({limit}) of indicators to fetch. Indicators len: {len(indicators)}."
f" Got {len(indicators)} indicators and {len(list(relationships_lst))} relationships."
f" Objects counters: {parsed_objects_counter}"
)
return indicators, relationships_lst
except Exception as e:
demisto.info(f"Exception trying to parse envelope: {e}, {traceback.format_exc()}")
if len(indicators) == 0:
demisto.debug("No Indicator were parsed")
raise e
demisto.debug(f"Failed while parsing envelopes, succeeded to retrieve {len(indicators)} indicators.")
demisto.debug(
f"Finished parsing all objects. Got {len(list(indicators))} indicators "
f"and {len(list(relationships_lst))} relationships. Objects counters: {parsed_objects_counter}"
)
return indicators, relationships_lst
class Taxii2FeedClient(STIX2XSOARParser):
def __init__(
self,
url: str,
collection_to_fetch,
proxies,
verify: bool,
objects_to_fetch: list[str],
skip_complex_mode: bool = False,
username: str | None = None,
password: str | None = None,
field_map: dict | None = None,
tags: list | None = None,
tlp_color: str | None = None,
limit_per_request: int = DFLT_LIMIT_PER_REQUEST,
certificate: str = None,
key: str = None,
default_api_root: str = None,
update_custom_fields: bool = False,
enrichment_excluded: bool = False,
):
"""
TAXII 2 Client used to poll and parse indicators in XSOAR formar
:param url: discovery service URL
:param collection_to_fetch: Collection to fetch objects from
:param proxies: proxies used in request
:param skip_complex_mode: if set to True will skip complex observations
:param verify: verify https
:param username: username used for basic authentication OR api_key used for authentication
:param password: password used for basic authentication
:param field_map: map used to create fields entry ({field_name: field_value})
:param tags: custom tags to be added to the created indicator
:param limit_per_request: Limit the objects requested per poll request
:param tlp_color: Traffic Light Protocol color
:param certificate: TLS Certificate
:param key: TLS Certificate key
:param default_api_root: The default API Root to use
"""
super().__init__(
tlp_color=tlp_color,
id_to_object={},
field_map=field_map if field_map else {},
skip_complex_mode=skip_complex_mode,
update_custom_fields=update_custom_fields,
tags=tags if tags else [],
enrichment_excluded=enrichment_excluded,
)
self._conn = None
self.server = None
self.api_root = None
self.collections = None
self.collection_to_fetch = collection_to_fetch
if not limit_per_request:
limit_per_request = DFLT_LIMIT_PER_REQUEST
self.limit_per_request = limit_per_request
self.base_url = url
self.proxies = proxies
self.verify = verify
self.auth = None
self.auth_header = None
self.auth_key = None
self.crt = None
if username and password:
# authentication methods:
# 1. API Token
# 2. Authentication Header
# 3. Basic
if username == API_USERNAME:
self.auth = TokenAuth(key=password)
elif username.startswith(HEADER_USERNAME):
self.auth_header = username.split(HEADER_USERNAME)[1]
self.auth_key = password
else:
self.auth = requests.auth.HTTPBasicAuth(username, password)
if (certificate and not key) or (not certificate and key):
raise DemistoException("Both certificate and key should be provided or neither should be.")
if certificate and key:
self.crt = (self.build_certificate(certificate), self.build_certificate(key))
self.objects_to_fetch = objects_to_fetch
self.default_api_root = default_api_root
def init_server(self, version=TAXII_VER_2_1):
"""
Initializes a server in the requested version
:param version: taxii version key (either 2.0 or 2.1)
"""
server_url = urljoin(self.base_url)
self._conn = _HTTPConnection(verify=self.verify, proxies=self.proxies, version=version, auth=self.auth, cert=self.crt)
if self.auth_header:
# add auth_header to the session object
self._conn.session.headers = merge_setting( # type: ignore[attr-defined]
self._conn.session.headers, # type: ignore[attr-defined]
{self.auth_header: self.auth_key},
dict_class=CaseInsensitiveDict,
)
if version is TAXII_VER_2_0:
self.server = v20.Server(
server_url,
verify=self.verify,
proxies=self.proxies,
conn=self._conn,
)
else:
self.server = v21.Server(
server_url,
verify=self.verify,
proxies=self.proxies,
conn=self._conn,
)
def init_roots(self):
"""
Initializes the api roots (used to get taxii server objects)
"""
if not self.server:
self.init_server()
try:
# disable logging as we might receive client error and try 2.0
logging.disable(logging.ERROR)
# try TAXII 2.1
self.set_api_root()
# (TAXIIServiceException, HTTPError) should suffice, but sometimes it raises another type of HTTPError
except HTTPError as e:
if e.response.status_code != 406 and "version=2.0" not in str(e):
raise e
# switch to TAXII 2.0
self.init_server(version=TAXII_VER_2_0)
self.set_api_root()
except Exception as e:
if "406 Client Error" not in str(e) and "version=2.0" not in str(e):
raise e
# switch to TAXII 2.0
self.init_server(version=TAXII_VER_2_0)
self.set_api_root()
finally:
# enable logging
logging.disable(logging.NOTSET)
def set_api_root(self):
roots_to_api = {}
for api_root in self.server.api_roots: # type: ignore[attr-defined]
# ApiRoots are initialized with wrong _conn because we are not providing auth or cert to Server
# closing wrong unused connections
api_root_name = str(api_root.url).split("/")[-2]
demisto.debug(f"closing api_root._conn for {api_root_name}")
api_root._conn.close()
roots_to_api[api_root_name] = api_root
if self.default_api_root:
if not roots_to_api.get(self.default_api_root):
raise DemistoException(
f"The given default API root {self.default_api_root} doesn't exist. "
f"Available API roots are {list(roots_to_api.keys())}."
)
self.api_root = roots_to_api.get(self.default_api_root)
elif server_default := self.server.default: # type: ignore[attr-defined]
self.api_root = server_default
else:
self.api_root = self.server.api_roots[0] # type: ignore[attr-defined]
# override _conn - api_root isn't initialized with the right _conn
self.api_root._conn = self._conn # type: ignore[union-attr]
def init_collections(self):
"""
Collects available taxii collections
"""
self.collections = list(self.api_root.collections) # type: ignore[union-attr, attr-defined, assignment]
def init_collection_to_fetch(self, collection_to_fetch=None):
"""
Tries to initialize `collection_to_fetch` if possible
"""
if not collection_to_fetch and isinstance(self.collection_to_fetch, str):
# self.collection_to_fetch will be changed from str -> Union[v20.Collection, v21.Collection]
collection_to_fetch = self.collection_to_fetch
if not self.collections:
raise DemistoException(ERR_NO_COLL)
if collection_to_fetch:
collection_found = False
for collection in self.collections:
if collection.title == collection_to_fetch:
self.collection_to_fetch = collection
collection_found = True
break
if not collection_found:
raise DemistoException(
f"Could not find the provided Collection name {collection_to_fetch} in the available collections. "
"Please make sure you entered the name correctly."
)
def initialise(self):
self.init_server()
self.init_roots()
self.init_collections()
self.init_collection_to_fetch()
@staticmethod
def build_certificate(cert_var):
var_list = cert_var.split("-----")
# replace spaces with newline characters
certificate_fixed = "-----".join(var_list[:2] + [var_list[2].replace(" ", "\n")] + var_list[3:])
cf = tempfile.NamedTemporaryFile(delete=False)
cf.write(certificate_fixed.encode())
cf.flush()
return cf.name
def build_iterator(self, limit: int = -1, **kwargs) -> list[dict[str, str]]:
"""
Polls the taxii server and builds a list of cortex indicators objects from the result
:param limit: max amount of indicators to fetch
:return: Cortex indicators list
"""
if not isinstance(self.collection_to_fetch, v20.Collection | v21.Collection):
raise DemistoException("Could not find a collection to fetch from. Please make sure you provided a collection.")
if limit is None:
limit = -1
page_size = self.get_page_size(limit, limit)
if page_size <= 0:
return []
try:
demisto.debug(f"Fetching {page_size} objects from TAXII server")
envelopes = self.poll_collection(page_size, **kwargs) # got data from server
indicators = self.load_stix_objects_from_envelope(envelopes, limit)
except InvalidJSONError as e:
demisto.debug(f"Excepted InvalidJSONError, continuing with empty result.\nError: {e}, {traceback.format_exc()}")
# raised when the response is empty, because {} is parsed into '筽'
indicators = []
return indicators
def poll_collection(self, page_size: int, **kwargs) -> types.GeneratorType:
"""
Polls a taxii collection
:param page_size: size of the request page
"""
get_objects = self.collection_to_fetch.get_objects
if self.objects_to_fetch:
if (
"relationship" not in self.objects_to_fetch and len(self.objects_to_fetch) > 1
): # when fetching one type no need to fetch relationship
self.objects_to_fetch.append("relationship")
kwargs["type"] = self.objects_to_fetch
if isinstance(self.collection_to_fetch, v20.Collection):
demisto.debug(f"Collection is a v20 type collction, {self.collection_to_fetch}")
return v20.as_pages(get_objects, per_request=page_size, **kwargs)
demisto.debug(f"Collection is a v21 type collction, {self.collection_to_fetch}")
return v21.as_pages(get_objects, per_request=page_size, **kwargs)
def get_page_size(self, max_limit: int, cur_limit: int) -> int:
"""
Get a page size given the limit on entries `max_limit` and the limit on the current poll
:param max_limit: max amount of entries allowed overall
:param cur_limit: max amount of entries allowed in a page
:return: page size
"""
return min(self.limit_per_request, cur_limit) if max_limit > -1 else self.limit_per_request
README
To use the common TAXII 2 Server logic (authentication and API requests), run the following command to import the TAXII2ApiModule.
def main():
...
from TAXII2ApiModule import * # noqa: E402
if __name__ in ["builtins", "__main__"]:
main()
Then, the TAXII2ApiModule will be available for usage. For examples, see the TAXII 2 Feed integration.