import demistomock as demisto # noqa: F401 import urllib3 from CommonServerPython import * # noqa: F401 # disable insecure warnings urllib3.disable_warnings() class Client(BaseClient): """Client class to interact with the service API This Client implements API calls, and does not contain any Demisto logic. Should only do requests and return data. It inherits from BaseClient defined in CommonServer Python. Most calls use _http_request() that handles proxy, SSL verification, etc. The _http_request() has to allow redirects to support the snort IP blocklist """ def build_iterator(self) -> list: """Retrieves all entries from the feed. Returns: A list of objects, containing the indicators. """ result = [] res = self._http_request("GET", url_suffix="", full_url=self._base_url, resp_type="text", allow_redirects=True) # In that case the feed output is in text format, so extracting the indicators from the response requires # iterating over it's lines solely: try: indicators = res.splitlines() for indicator in indicators: # Infer the type of the indicator using 'auto_detect_indicator_type(indicator)' function # (defined in CommonServerPython). if indicator_type := auto_detect_indicator_type(indicator): result.append({"value": indicator, "type": indicator_type, "FeedURL": self._base_url}) except ValueError as err: demisto.debug(str(err)) raise ValueError(f"Could not parse returned data as indicator. \n\nError massage: {err}") return result def test_module(client: Client) -> str: """Builds the iterator to check that the feed is accessible. Args: client: Client object. Returns: Outputs. """ fetch_indicators(client=client, limit=1) return "ok" def fetch_indicators( client: Client, tlp_color: str | None = None, feed_tags: list = [], limit: int = -1, ) -> list[dict]: """Retrieves indicators from the feed Args: client (Client): Client object with request tlp_color (str): Traffic Light Protocol color feed_tags (list): tags to assign fetched indicators limit (int): limit the results Returns: Indicators. """ iterator = client.build_iterator() indicators = [] if limit > 0: iterator = iterator[:limit] # extract values from iterator for item in iterator: value_ = item.get("value") type_ = item.get("type") raw_data = { "value": value_, "type": type_, } # Create an indicator object for each value. # The object consists of a dictionary with required and optional keys and values, as described below. for key, value in item.items(): raw_data.update({key: value}) indicator_obj = { # The indicator value. "value": value_, # The indicator type as defined in Cortex XSOAR. # One can use the FeedIndicatorType class under CommonServerPython to populate this field. "type": type_, # The name of the service supplying this feed. "service": "Snort IP Blocklist", # A dictionary that maps values to existing indicator fields defined in Cortex XSOAR. # One can use this section in order to map custom indicator fields previously defined # in Cortex XSOAR to their values. "fields": {}, # A dictionary of the raw data returned from the feed source about the indicator. "rawJSON": raw_data, } if feed_tags: indicator_obj["fields"]["tags"] = feed_tags if tlp_color: indicator_obj["fields"]["trafficlightprotocol"] = tlp_color indicators.append(indicator_obj) return indicators def get_indicators_command(client: Client, params: dict[str, str], args: dict[str, str]) -> CommandResults: """Wrapper for retrieving indicators from the feed to the war-room. Args: client: Client object with request params: demisto.params() args: demisto.args() Returns: Outputs. """ limit = arg_to_number(args.get("limit", "10"), arg_name="limit") or 10 tlp_color = params.get("tlp_color") feed_tags = argToList(params.get("feedTags", "")) indicators = fetch_indicators(client, tlp_color, feed_tags, limit) human_readable = tableToMarkdown( f"Indicators from Snort IP Blocklist Feed: (first {limit} indicators)", indicators, headers=["value", "type"], headerTransform=string_to_table_header, removeNull=True, ) return CommandResults( readable_output=human_readable, outputs_prefix="", outputs_key_field="", raw_response=indicators, outputs={}, ) def fetch_indicators_command(client: Client, params: dict[str, str]) -> list[dict]: """Wrapper for fetching indicators from the feed to the Indicators tab. Args: client: Client object with request params: demisto.params() Returns: Indicators. """ feed_tags = argToList(params.get("feedTags", "")) tlp_color = params.get("tlp_color") indicators = fetch_indicators(client, tlp_color, feed_tags) return indicators def main(): """ main function, parses params and runs command functions """ params = demisto.params() # Get the service API url base_url = params.get("url") insecure = not params.get("insecure", False) proxy = params.get("proxy", False) command = demisto.command() args = demisto.args() demisto.debug(f"Command being called is {command}") try: client = Client( base_url=base_url, verify=insecure, proxy=proxy, ) if command == "test-module": # This is the call made when pressing the integration Test button. return_results(test_module(client)) elif command == "snort-get-ip-blocklist-indicators": # This is the command that fetches a limited number of indicators from the feed source # and displays them in the war room. return_results(get_indicators_command(client, params, args)) elif command == "fetch-indicators": # This is the command that initiates a request to the feed endpoint and create new indicators objects from # the data fetched. If the integration instance is configured to fetch indicators, then this is the command # that will be executed at the specified feed fetch interval. indicators = fetch_indicators_command(client, params) for iter_ in batch(indicators, batch_size=2000): demisto.createIndicators(iter_) else: raise NotImplementedError(f"Command {command} is not implemented.") # Log exceptions and return errors except Exception as e: return_error(f"Failed to execute {command} command.\nError:\n{e!s}") if __name__ in ["__main__", "builtin", "builtins"]: main()