import demistomock as demisto # noqa: F401 import urllib3 from CommonServerPython import * # noqa: F401 # disable insecure warnings urllib3.disable_warnings() FEED_NAME = [ "dga-dns", "autorun-registry", "banking-dns", "dll-hijacking-dns", "doc-net-com-dns", "downloaded-pe-dns", "dynamic-dns", "irc-dns", "modified-hosts-dns", "parked-dns", "public-ip-check-dns", "ransomware-dns", "rat-dns", "scheduled-tasks", "sinkholed-ip-dns", "stolen-cert-dns", ] class Client(BaseClient): def __init__( self, api_key, verify, proxy, first_fetch, feed_name, feed_tags, tlp_color, create_relationships, base_url="https://panacea.threatgrid.com/api/v3/feeds/", ): """Implements class for the feed. Args: api_key: API Key. verify: boolean, if *false* feed HTTPS server certificate is verified. Default: *false* first_fetch: The date from which to start fetching feeds feed_name: The feed names to fetch feed_tags: feed tags. tlp_color: Traffic Light Protocol color. create_relationships: boolean, if *false* no relationships are created. Default: *true* """ super().__init__(base_url=base_url, verify=verify, proxy=proxy) self.first_fetch = first_fetch self.feed_name = feed_name self.feed_tags = feed_tags self.tlp_color = tlp_color self.create_relationships = create_relationships self._api_key = api_key def get_indicators(self, feed_name: str, time_stamp: Optional[str]): """Retrieves all indicators from the feed. Args: :param feed_name: the feed name to fetch :param time_stamp: Optional - fetch feeds only from this day """ suffix = f"{feed_name}_{time_stamp}.json" if time_stamp else f"{feed_name}.json" try: return self._http_request("GET", suffix, params={"api_key": self._api_key}, timeout=70) except Exception as error: if "Unauthorized" in str(error): raise DemistoException("Authorization Error: make sure the API Key is correctly set") else: raise DemistoException(error) def create_related_ips(obj: dict, feed_name: str, tlp_color: str) -> list[dict]: """creates list indicators of the related ips. Args: :param obj: from the response :param feed_name: the feed That it was fetch there :param tlp_color: Traffic Light Protocol color :return: indicators list """ return [ ( { "value": ip, "type": "IP", "rawJSON": obj, "fields": { "Tags": [feed_name], "reportedby": "CiscoSMA", "Description": obj.get("description"), "FirstSeenBySource": obj.get("timestamp"), "trafficlightprotocol": tlp_color, }, } ) for ip in obj.get("ips") # type:ignore ] def create_related_files(obj: dict, feed_name: str, tlp_color: str) -> dict: """creates dict of the related file indicator. Args: :param obj: from the response :param feed_name: the feed That it was fetch there :param tlp_color: Traffic Light Protocol color :return: indicator """ return { "value": obj.get("sample_sha256"), "type": "File", "rawJSON": obj, "fields": { "Tags": [feed_name], "reportedby": "CiscoSMA", "Description": obj.get("description"), "FirstSeenBySource": obj.get("timestamp"), "MD5": obj.get("sample_md5"), "SHA1": obj.get("sample_sha1"), "trafficlightprotocol": tlp_color, }, } def create_entity_relationships(feed_related_indicators: list, current_domain: str) -> list: """ Args: feed_related_indicators: Indicators related to the domain returned from fetch_indicators. current_domain: Returned indicator from fetch_indicators. Returns: relationships: EntityRelationships list. """ relationships = [] for relation in feed_related_indicators: if relation.get("value") != "": entity_relation = EntityRelationship( name=EntityRelationship.Relationships.RELATED_TO, entity_a=current_domain, entity_a_type="Domain", entity_b=relation.get("value"), entity_b_type=relation.get("type"), ) relationships.append(entity_relation.to_indicator()) return relationships def test_module(client: Client) -> str: """Goes through all the code of the fetch. Args: client: Client object with request Returns: str: ok. """ fetch_indicators(client) return "ok" def fetch_indicators(client: Client, time_to_stop_fetch: Optional[datetime] = None) -> tuple[list, bool]: """Retrieves indicators from the feed Args: client: Client object with request time_to_stop_fetch: Time to stop the fetch before it falls on timeout Returns: List. Processed indicator from feed. """ fetch_time_stamp = None files = ["sample_sha256", "sample_sha1", "sample_md5"] if client.first_fetch.date() < arg_to_datetime("now").date(): # type:ignore fetch_time_stamp = client.first_fetch.date().isoformat() else: client.first_fetch = arg_to_datetime("now") indicators = [] domains_map: dict[str, dict] = {} for name in client.feed_name: if time_to_stop_fetch and arg_to_datetime("now") > time_to_stop_fetch: # type:ignore return [], False result = client.get_indicators(feed_name=name, time_stamp=fetch_time_stamp) for obj in result: feed_related_indicators = [{"type": "IP", "value": ip} for ip in obj.get("ips")] feed_related_indicators.extend([{"type": "File", "value": obj.get(file_hash)} for file_hash in files]) current_domain = obj.get("domain") relationships = create_entity_relationships(feed_related_indicators, current_domain) if current_domain in domains_map: if name not in domains_map[current_domain]["fields"]["Tags"]: domains_map[current_domain]["fields"]["Tags"].append(name) domains_map[current_domain]["fields"]["FeedRelatedIndicators"].extend( related for related in feed_related_indicators if related not in domains_map[current_domain]["fields"]["FeedRelatedIndicators"] ) existing_relations = [relation.get("entityB") for relation in domains_map[current_domain]["relationships"]] new_relations = filter(lambda relation: relation.get("entityB") not in existing_relations, relationships) domains_map[current_domain]["relationships"].extend(new_relations) else: domains_map[current_domain] = { "value": current_domain, "type": "Domain", "rawJSON": obj, "relationships": relationships, "fields": { "Tags": [name], "reportedby": "CiscoSMA", "FeedRelatedIndicators": feed_related_indicators, "Description": obj.get("description"), "FirstSeenBySource": obj.get("timestamp"), }, } if client.tlp_color: domains_map[current_domain]["fields"]["trafficlightprotocol"] = client.tlp_color if client.feed_tags: domains_map[current_domain]["fields"]["Tags"] += client.feed_tags indicators += create_related_ips(obj, name, client.tlp_color) indicators.append(create_related_files(obj, name, client.tlp_color)) indicators += list(domains_map.values()) return indicators, True def fetch_indicators_command(client: Client, time_to_stop_fetch: datetime = None): """Retrieves indicators from the feed Args: client: Client object with request time_to_stop_fetch: Time to stop the fetch before it falls on the docker timeout Returns: List. Processed indicator from feed. """ current_date = arg_to_datetime("now").date() # type:ignore while client.first_fetch.date() <= current_date: indicators, status = fetch_indicators(client, time_to_stop_fetch) if not status: return for iter_ in batch(indicators, batch_size=2000): demisto.createIndicators(iter_) # Each request returns one day's indicators, so we'll keep the date of the next day in IntegrationContext. next_fetch = dateparser.parse("tomorrow", settings={"RELATIVE_BASE": client.first_fetch}) # type:ignore if client.first_fetch.date() < current_date: demisto.setIntegrationContext({"last_fetch": next_fetch.isoformat()}) # type:ignore client.first_fetch = next_fetch demisto.debug(f"{len(indicators)} XSOAR Indicators were created.") return def get_indicators_command(client: Client, args: dict[str, str]) -> CommandResults: """Wrapper for retrieving indicators from the feed to the war-room. Args: client: Client object with request args: demisto.args() Returns: Demisto Outputs. """ limit = int(args.get("limit", 50)) indicators, status = fetch_indicators(client) limit_indicators = indicators[:limit] readable_output = tableToMarkdown("CiscoSMA Indicators:", t=limit_indicators, headers=["value", "type"]) command_results = CommandResults( outputs_prefix="CiscoSMA", outputs_key_field="value", outputs=limit_indicators, readable_output=readable_output, raw_response=indicators, ) return command_results def reset_last_run(): """ Reset the last run from the integration context """ demisto.setIntegrationContext({}) return CommandResults(readable_output="Fetch history deleted successfully") def main(): # pragma: no cover """ PARSE AND VALIDATE FEED PARAMS """ try: time_to_stop_fetch = arg_to_datetime(arg="in 25 minutes") # type:ignore # To stop before docker timeout. params = demisto.params() args = demisto.args() base_url = params["base_url"] api_key = params.get("api_key", {}).get("password") if not api_key: raise DemistoException("API Key was not given. Please insert your API Key.") verify = not params.get("insecure", False) proxy = params.get("proxy", False) feed_name = params.get("feed_name", "") if "all" in feed_name: feed_name = FEED_NAME feed_tags = argToList(params.get("feedTags")) tlp_color = params.get("tlp_color") create_relationships = params.get("create_relationships", True) if fetch_date_string := demisto.getIntegrationContext().get("last_fetch"): first_fetch = arg_to_datetime(fetch_date_string) # type:ignore else: if not (first_fetch_param := params.get("first_fetch")): first_fetch_param = "today" first_fetch = arg_to_datetime(arg=first_fetch_param, arg_name="First fetch") # type:ignore command = demisto.command() demisto.debug(f"Command being called is: {command}") client = Client( base_url=base_url, api_key=api_key, verify=verify, proxy=proxy, feed_name=feed_name, first_fetch=first_fetch, feed_tags=feed_tags, tlp_color=tlp_color, create_relationships=create_relationships, ) if command == "test-module": return_results(test_module(client)) elif command == "fetch-indicators": fetch_indicators_command(client, time_to_stop_fetch) elif command == "cisco-sma-get-indicators": return_results(get_indicators_command(client, args)) elif command == "cisco-sma-reset-fetch-indicators": return_results(reset_last_run()) else: raise NotImplementedError(f'Command "{command}" is not implemented.') except Exception as err: return_error(f"Failed to execute {demisto.command()} command.\nError:\n{err!s}") if __name__ in ("__main__", "__builtin__", "builtins"): main()