# IMPORTS import cabby import dateutil import dateutil.parser import demistomock as demisto import pytz import requests import urllib3 from bs4 import BeautifulSoup from CommonServerPython import * from netaddr import IPAddress, IPNetwork, iprange_to_cidrs from six import string_types # type: ignore from CommonServerUserPython import * # Disable insecure warnings urllib3.disable_warnings() SOURCE_NAME = "Alien Vault OTX TAXII" # disable insecure warnings urllib3.disable_warnings() EPOCH = datetime.utcfromtimestamp(0).replace(tzinfo=pytz.UTC) """ Helper Methods """ def package_extract_properties(package): """Extracts properties from the STIX package""" result: dict[str, str] = {} header = package.find_all("STIX_Header") if len(header) == 0: return result # share level mstructures = header[0].find_all("Marking_Structure") for ms in mstructures: type_ = ms.get("xsi:type") if type_ is result: continue color = ms.get("color") if color is result: continue type_ = type_.lower() if "tlpmarkingstructuretype" not in type_: continue result["share_level"] = color.lower() # https://www.us-cert.gov/tlp break # decode title title = next((c for c in header[0] if c.name == "Title"), None) if title is not None: result["stix_package_title"] = title.text # decode description description = next((c for c in header[0] if c.name == "Description"), None) if description is not None: result["stix_package_description"] = description.text # decode description sdescription = next((c for c in header[0] if c.name == "Short_Description"), None) if sdescription is not None: result["stix_package_short_description"] = sdescription.text # decode identity name from information_source information_source = next((c for c in header[0] if c.name == "Information_Source"), None) if information_source is not None: identity = next((c for c in information_source if c.name == "Identity"), None) if identity is not None: name = next(c for c in identity if c.name == "Name") if name is not None: result["stix_package_information_source"] = name.text return result def observable_extract_properties(observable): """Extracts properties from observable""" result = {} title = next((c for c in observable if c.name == "Title"), None) if title is not None: title = title.text result["stix_title"] = title description = next((c for c in observable if c.name == "Description"), None) if description is not None: description = description.text result["stix_description"] = description return result """ TAXII STIX DECODE """ class AddressObject: """ Implements address object indicator decoding based on: https://stixproject.github.io/data-model/1.2/AddressObj/AddressObjectType/ """ @staticmethod def decode(props, **kwargs): indicator = props.find("Address_Value") if indicator is None: return [] indicator = indicator.string.encode("ascii", "replace").decode() acategory = props.get("category", None) if acategory is None: try: # if ',' in the ip address then we have a range (10.0.0.0,10.0.0.255) # and we want to change it to a CIDR object (10.0.0.0/24) if "," in indicator: ips = indicator.split(",") indicator = str(iprange_to_cidrs(ips[0], ips[1])[0].cidr) cidr = IPNetwork(indicator) if cidr.version == 4: type_ = "CIDR" elif cidr.version == 6: type_ = "IPv6CIDR" else: LOG(f"Unknown ip version: {cidr.version!r}") return [] else: ip = IPAddress(indicator) if ip.version == 4: type_ = "IP" elif ip.version == 6: type_ = "IPv6" else: LOG(f"Unknown ip version: {ip.version!r}") return [] except Exception: return [] elif acategory == "ipv4-addr": # if ',' in the ip address then we have a range (10.0.0.0,10.0.0.255) # and we want to change it to a CIDR object (10.0.0.0/24) if "," in indicator: ips = indicator.split(",") indicator = str(iprange_to_cidrs(ips[0], ips[1])[0].cidr) type_ = "CIDR" else: type_ = "IP" elif acategory == "ipv6-addr": # Same logic as above just for ipv6 if "," in indicator: ips = indicator.split(",") indicator = str(iprange_to_cidrs(ips[0], ips[1])[0].cidr) type_ = "IPv6CIDR" else: type_ = "IPv6" elif acategory == "e-mail": type_ = "Email" else: LOG(f"Unknown AddressObjectType category: {acategory!r}") return [] return [{"indicator": indicator, "type": type_}] class DomainNameObject: """ Implements domain object indicator decoding based on: https://stixproject.github.io/data-model/1.2/DomainNameObj/DomainNameObjectType/ """ @staticmethod def decode(props, **kwargs): dtype = props.get("type", "FQDN") if dtype != "FQDN": return [] domain = props.find("Value") if domain is None: return [] return [{"indicator": domain.string.encode("ascii", "replace").decode(), "type": "Domain"}] class FileObject: """ Implements file object indicator decoding based on: https://stixproject.github.io/data-model/1.2/FileObj/FileObjectType/ """ @staticmethod def _decode_basic_props(props): result = {} name = next((c for c in props if c.name == "File_Name"), None) if name is not None: result["stix_file_name"] = name.text size = next((c for c in props if c.name == "File_Size"), None) if size is not None: result["stix_file_size"] = size.text format = next((c for c in props if c.name == "File_Format"), None) if format is not None: result["stix_file_format"] = format.text return result @staticmethod def decode(props, **kwargs): result = [] bprops = FileObject._decode_basic_props(props) hashes = props.find_all("Hash") for h in hashes: htype = h.find("Type") if htype is None: continue htype = htype.string.lower() if htype not in ["md5", "sha1", "sha256", "ssdeep"]: continue value = h.find("Simple_Hash_Value") if value is None: continue value = value.string.lower() result.append({"indicator": value, "htype": htype, "type": "File"}) for r in result: for r2 in result: if r["htype"] == r2["htype"]: continue r["stix_file_{}".format(r2["htype"])] = r2["indicator"] r.update(bprops) return result class URIObject: """ Implements URI object indicator decoding based on: https://stixproject.github.io/data-model/1.2/URIObj/URIObjectType/ """ @staticmethod def decode(props, **kwargs): utype = props.get("type", "URL") if utype == "URL": type_ = "URL" elif utype == "Domain Name": type_ = "Domain" else: return [] url = props.find("Value") if url is None: return [] return [{"indicator": url.string.encode("utf8", "replace").decode(), "type": type_}] class SocketAddressObject: """ Implements socket address object indicator decoding based on: https://stixproject.github.io/data-model/1.2/SocketAddressObj/SocketAddressObjectType/ """ @staticmethod def decode(props, **kwargs): ip = props.get("ip_address", None) if ip: return AddressObject.decode(ip) return [] class LinkObject: """ Implements link object indicator decoding based on: https://stixproject.github.io/data-model/1.2/LinkObj/LinkObjectType/ """ @staticmethod def decode(props, **kwargs): ltype = props.get("type", "URL") if ltype != "URL": LOG(f"Unhandled LinkObjectType type: {ltype}") return [] value = props.get("value", None) if value is None: LOG("no value in observable LinkObject") return [] if not isinstance(value, string_types): value = value.get("value", None) if value is None: LOG("no value in observable LinkObject") return [] return [{"indicator": value, "type": ltype}] class HTTPSessionObject: """ Implements http session object indicator decoding based on: https://stixproject.github.io/data-model/1.2/HTTPSessionObj/HTTPSessionObjectType/ """ @staticmethod def decode(props, **kwargs): if "http_request_response" in props: tmp = props["http_request_response"] if len(tmp) == 1: item = tmp[0] http_client_request = item.get("http_client_request", None) if http_client_request is not None: http_request_header = http_client_request.get("http_request_header", None) if http_request_header is not None: raw_header = http_request_header.get("raw_header", None) if raw_header is not None: return [ { "indicator": raw_header.split("\n")[0], "type": "http-session", # we don't support this type natively in demisto "header": raw_header, } ] else: LOG("multiple HTTPSessionObjectTypes not supported") return [] class StixDecode: """ Decode STIX strings formatted as xml, and extract indicators from them """ DECODERS = { "DomainNameObjectType": DomainNameObject.decode, "FileObjectType": FileObject.decode, "WindowsFileObjectType": FileObject.decode, "URIObjectType": URIObject.decode, "AddressObjectType": AddressObject.decode, "SocketAddressObjectType": SocketAddressObject.decode, "LinkObjectType": LinkObject.decode, "HTTPSessionObjectType": HTTPSessionObject.decode, } @staticmethod def object_extract_properties(props, kwargs): type_ = props.get("xsi:type").rsplit(":")[-1] if type_ not in StixDecode.DECODERS: LOG(f"Unhandled cybox Object type: {type_!r} - {props!r}") return [] return StixDecode.DECODERS[type_](props, **kwargs) @staticmethod def _parse_stix_timestamp(stix_timestamp): dt = dateutil.parser.parse(stix_timestamp) if dt.tzinfo is None: dt = dt.replace(tzinfo=pytz.UTC) delta = dt - EPOCH return int(delta.total_seconds() * 1000) @staticmethod def _deduplicate(indicators): result = {} for iv in indicators: result["{}:{}".format(iv["indicator"], iv["type"])] = iv return result.values() @staticmethod def decode(content, **kwargs): result = [] package = BeautifulSoup(content, "xml") if package.contents[0].name != "STIX_Package": # type: ignore[attr-defined] return None, [] package = package.contents[0] timestamp = package.get("timestamp", None) # type: ignore[attr-defined] if timestamp is not None: timestamp = StixDecode._parse_stix_timestamp(timestamp) pprops = package_extract_properties(package) indicators = package.find_all("Indicator") # type: ignore[attr-defined] for ind in indicators: observables = ind.find_all("Observable") for o in observables: gprops = observable_extract_properties(o) obj = next((ob for ob in o if ob.name == "Object"), None) if obj is None: continue # main properties properties = next((c for c in obj if c.name == "Properties"), None) if properties is not None: for r in StixDecode.object_extract_properties(properties, kwargs): r.update(gprops) r.update(pprops) r.update({"added_time": ind.get("timestamp") + "Z"}) result.append(r) # then related objects related = next((c for c in obj if c.name == "Related_Objects"), None) if related is not None: for robj in related: if robj.name != "Related_Object": continue properties = next((c for c in robj if c.name == "Properties"), None) if properties is None: continue for r in StixDecode.object_extract_properties(properties, kwargs): r.update(gprops) r.update(pprops) result.append(r) return timestamp, StixDecode._deduplicate(result) """ Alien Vault OTX TAXII Client """ class Client: """Client for AlienVault OTX Feed - gets indicator lists from collections using TAXII client Attributes: api_key(str): The API key for AlienVault OTX. collection(str): The collections on which to run the feed. insecure(bool): Use SSH on http request. proxy(str): Use system proxy. all_collections(bool): Whether to run on all active collections. """ def __init__( self, api_key: str, collection: str, insecure: bool = False, proxy: bool = False, all_collections: bool = False, tags: list = [], tlp_color: str | None = None, ): taxii_client = cabby.create_client(discovery_path="https://otx.alienvault.com/taxii/discovery") taxii_client.set_auth(username=str(api_key), password="foo", verify_ssl=not insecure) if proxy: taxii_client.set_proxies(handle_proxy()) self.taxii_client = taxii_client self.tags = tags self.tlp_color = tlp_color self.all_collections = all_collections if all_collections: self.collections = self.get_all_collections() else: if collection is None or collection == "": return_error(f"No collection set. Here is a list of all accessible collections: {self.get_all_collections()}") self.collections = collection.split(",") def get_all_collections(self): """Gets a list of all collections listed in the AlienVault OTX instance. Returns: list. A list of all collection names in AlienVault OTX. """ collections = self.taxii_client.get_collections() full_collection_list = [] for collection in collections: full_collection_list.append(collection.name) return full_collection_list def build_iterator(self, collection, begin_date=None): """Returns a list of all XML elements from the given collection. Args: collection(str): The collection name to fetch the elements from. begin_date(datetime): what is the first date to fetch indicators from. Returns: list. A list of XML elements (strings). """ if not begin_date: begin_date, _ = parse_date_range(demisto.params().get("initial_interval")) if begin_date.tzinfo is None: begin_date = begin_date.replace(tzinfo=pytz.UTC) # bring all the indicators created from the last date until now. return list(self.taxii_client.poll(collection_name=collection, begin_date=begin_date)) def decode_indicators(self, response): """Decode the XML response given using STIXDecode class. Args: response(str): An XML response from a collection of AlienVault. Returns: list. A list of indicators (dicts) decoded from the response. """ return StixDecode.decode(response) def module_test_command(client: Client, args: dict): """Test module for the integration will run on all the collections given and check for a response. if all_collections is checked will return an error only in case no collection returned a response. if all_collections is not checked will return an error for the collections that did not respond. Args: client(Client): The AlienVault OTX client. args(dict): empty dictionary. Returns: str,dict,dict. ok if passed - will raise an Exception otherwise. """ passed_collections = [] # type:List failed_collections = [] # type:List for collection in client.collections: try: client.build_iterator(collection) passed_collections.append(collection) except Exception as e: if e.__class__ is requests.exceptions.SSLError: raise Exception("SSL Connection failed - try marking the Trust Any Certificate checkbox.") else: if not client.all_collections: failed_collections.append(collection) continue if not client.all_collections and len(failed_collections) > 0: raise Exception( f"Unable to poll from the collections {failed_collections!s} check the collection names and " f"configuration on Alien Vault" ) if len(passed_collections) == 0: raise Exception("Unable to poll from any collection - please check the configuration on Alien Vault") return "ok", {}, {} def get_indicators_command(client: Client, args: dict): """Runs fetch indicators and return the indicators. Args: client(Client): The AlienVault OTX client. args(dict): The command arguments Returns: str,dict,dict. The human readable, and rawJSON from the command - no context created. """ limit = int(args.get("limit", 50)) if args.get("begin_date"): begin_date, _ = parse_date_range(args.get("begin_date")) else: begin_date = None indicator_list, _ = fetch_indicators_command(client, limit=limit, begin_date=begin_date) human_readable = tableToMarkdown("Indicators from AlienVault OTX TAXII:", indicator_list, removeNull=True) return human_readable, {}, indicator_list def parse_indicators(sub_indicator_list, full_indicator_list, tags, tlp_color): """Gets a decoded indicator list and returns a parsed version of the indicator with accordance with Demisto's Feed indicator standards. Args: tags(list): The tags to add to the indicator. sub_indicator_list(list): A list of STIXDecoded indicators full_indicator_list(list): A list of all the indicators fetched to this point - used to prevent duplications. tlp_color(str): Traffic Light Protocol color. Returns: list,list. A list of parsed indicators and an updated list of all indicators polled """ parsed_indicator_list = [] # type: List for indicator in sub_indicator_list: # If the indicator was already seen, skip it. if indicator["indicator"] in full_indicator_list: continue indicator["value"] = indicator["indicator"] indicator["fields"] = { "description": indicator["stix_package_short_description"], "tags": tags, "firstseenbysource": indicator.get("added_time"), } if tlp_color: indicator["fields"]["trafficlightprotocol"] = tlp_color temp_copy = indicator.copy() del indicator["indicator"] indicator["rawJSON"] = temp_copy parsed_indicator_list.append(indicator) full_indicator_list.append(indicator["value"]) return parsed_indicator_list, full_indicator_list def get_latest_indicator_time(indicators_list): if not indicators_list: return None latest_indicator_time = dateutil.parser.parse(indicators_list[0].get("added_time", "1970-01-01T00:00:00Z")) for ind in indicators_list: indicator_time = dateutil.parser.parse(ind.get("added_time", "1970-01-01T00:00:00Z")) if indicator_time > latest_indicator_time: latest_indicator_time = indicator_time return latest_indicator_time def fetch_indicators_command(client: Client, limit=None, begin_date=None): """Fetch indicators from AlienVault OTX. Args: client(Client): The AlienVault OTX client. limit(any): How many XML elements to parse, None if all should be parsed. begin_date(datetime): what is the first date to fetch indicators from. Returns: list. A list of indicators. """ indicator_list = [] # type:List for collection in client.collections: try: taxii_iter = client.build_iterator(collection, begin_date=begin_date) except Exception as e: if not client.all_collections: raise Exception(e) else: continue only_indicator_list = [] # type:List for raw_response in taxii_iter: _, res = client.decode_indicators(raw_response.content) # the only_indicator_list is a list containing only the indicators themselves. # it is used to prevent duplicated indicators from being created in the system. # this is because AlienVault OTX can return the same indicator several times from the same collection. parsed_list, only_indicator_list = parse_indicators(res, only_indicator_list, client.tags, client.tlp_color) indicator_list.extend(parsed_list) if limit is not None and limit <= len(indicator_list): indicator_list = indicator_list[:limit] break latest_indicator_time = get_latest_indicator_time(indicator_list) return indicator_list, latest_indicator_time def main(): params = demisto.params() tags = argToList(params.get("feedTags")) tlp_color = params.get("tlp_color") api_key = params.get("credentials", {}).get("password") or params.get("api_key") client = Client( api_key, params.get("collections"), params.get("insecure"), params.get("proxy"), params.get("all_collections"), tags=tags, tlp_color=tlp_color, ) command = demisto.command() demisto.info(f"Command being called is {command}") # Switch case commands = {"test-module": module_test_command, "alienvaultotx-get-indicators": get_indicators_command} try: integration_cache = demisto.getIntegrationContext() begin_date = integration_cache.get("begin_date") if demisto.command() == "fetch-indicators": if not begin_date: begin_date = demisto.params().get("initial_interval") if begin_date: begin_date, _ = parse_date_range(begin_date) else: begin_date = dateutil.parser.parse(begin_date) indicators, latest_indicator_time = fetch_indicators_command(client, begin_date=begin_date) # we submit the indicators in batches if indicators: # if new indicators were found - submit them in batches, set the next begin_date # to the time of the last indicator found for b in batch(indicators, batch_size=2000): demisto.createIndicators(b) demisto.setIntegrationContext({"begin_date": str(latest_indicator_time)}) else: # if no indicators entered - dont change begin date. demisto.setIntegrationContext({"begin_date": str(begin_date)}) else: readable_output, outputs, raw_response = commands[command](client, demisto.args()) return_outputs(readable_output, outputs, raw_response) except Exception as e: raise Exception(f"Error in {SOURCE_NAME} Integration [{e}]") if __name__ == "__builtin__" or __name__ == "builtins" or __name__ == "__main__": main()