AggregatedCommandApiModule
Common Aggregated Command API Module, provides generic Infrastructure for different kinds of aggregated commands.
python · ApiModules
Source
import time from abc import ABC from collections import defaultdict from dataclasses import dataclass from enum import Enum from functools import cached_property from typing import Any from datetime import datetime, timedelta from CommonServerPython import * import demistomock as demisto # Type alias for complex dictionary structures to improve readability ContextResult = dict[str, Any] DBotScoreList = list[ContextResult] CommandProcessResults = list[tuple[ContextResult, str, str]] # Calculating time interval for indicators freshness STATUS_FRESHNESS_WINDOW = timedelta(weeks=1) try: TIM_BRAND = "Threat Intel" if demisto.caller() == "agentix" else "TIM" # type: ignore[attr-defined] # pylint: disable=no-member except Exception: TIM_BRAND = "TIM" DBOT_SCORE_TO_VERDICT = { 0: "Unknown", 1: "Benign", 2: "Suspicious", 3: "Malicious", } # --- Core Enumerations and Data Classes --- class Status(Enum): """Enum for command status.""" SUCCESS = "Success" FAILURE = "Failure" class IndicatorStatus(Enum): """Enum for indicator status. - FRESH: If the indicator modifiedTime is within the freshness window (default is one week). - STALE: If the indicator modifiedTime is outside the freshness window. - MANUAL: If the indicator was manually added. - ERROR: If the indicator enrichment pipeline failed. """ FRESH = "Fresh" STALE = "Stale" MANUAL = "Manual" ERROR = "Error" class EntryResult: """ Captures a single command's summarized outcome for the final HR table. Attributes: command_name (str): Executed command name. (Not visible in HR) args (Any): The args passed (dict or string); kept for HR visibility. brand (str): Integration brand (or TIM/Core/VirusTotal). status (Status): Success/Failure. score_field_name (str): Score field name. For CVE Enrichment "TIM CVSS", for all other "TIM Score". score (float): The CVSS/Score value. message (str): Short message (e.g., error text or brand summary). """ def __init__( self, command_name: str, args: Any, brand: str, status: Status, message: str, score_field_name: str = "", score: float | None = None, ): self.command_name = command_name self.args = args self.brand = brand self.status = status self.score = score self.score_field_name = score_field_name self.message = message def to_entry(self) -> dict[str, Any]: if self.brand == TIM_BRAND: return { "Arguments": self.args, "Brand": self.brand, "Status": self.status.value, self.score_field_name: self.score if self.score is not None else "", "Message": self.message, } return { "Arguments": self.args, "Brand": self.brand, "Status": self.status.value, "Message": self.message, } @dataclass class IndicatorSchema: """ Indicator schema + mapping rules. Attributes: type (str): Indicator type (e.g., "url", "ip", "cve"). value_field (str | list[str]): Field name/ list of field name holding the indicator's value (e.g., "Data", "Address", ["MD5","SHA256"]). context_path_prefix (str): Context key prefix to extract (e.g., "URL(" or "IP("). context_output_mapping (dict[str, str]): Mapping rules from source context to target; the separator is `".."`. - {"Name": "Value"} -> flat key rename. - {"Name..Value": "Value"} -> nested extract {"Name":{"Value":x}} -> {"Value":x}. - Append "[]" to make the destination a list. - If "Score" is among mapped keys, we compute MaxScore/MaxVerdict/TIMScore. - If "CVSS" is among mapped keys, we Extract TIMCVSS. """ type: str value_field: str | list[str] context_path_prefix: str context_output_mapping: dict[str, str] def get_all_values_from(self, data: dict[str, Any]) -> dict[str, Any]: """ Collects all value fields from the given context entry based on this object's `value_field` definition. Supports single-field (e.g., "Address") and multi-field types (e.g., File hashes: ["MD5","SHA256"]). - For a single value_field (string), returns {"Value": <extracted_value>} using that key. - For multiple value_fields (list), returns only the matching fields with their original names. Example: get_all_values_from({"MD5": "aaa", "SHA256": "bbb", "Address": "ccc"}) self.value_field = ["MD5", "SHA256"] → {"MD5": "aaa", "SHA256": "bbb"} get_all_values_from({"MD5": "aaa", "SHA256": "bbb", "Address": "ccc"}) self.value_field = "Address" → {"Value": "ccc"} Args: data: The context dictionary to extract from. Returns: A dictionary of all found value fields. """ out: dict[str, Any] = {} if isinstance(self.value_field, str): val = data.get(self.value_field) if val: out["Value"] = val return out lower_data = {k.lower(): v for k, v in data.items()} for candidate in self.value_field: if val := lower_data.get(candidate.lower()): out[candidate] = val return out @dataclass class IndicatorInstance: """ Represents a single indicator undergoing the full enrichment pipeline. This object tracks the state of one input value across all stages of the enrichment flow (extract → create → enrich → TIM → final context). It centralizes the status, errors, enrichment results, and final outcome, making it easier to build the final context entries and human-readable output. Attributes: raw_input (str): The original input string provided by the user. extracted_value (str | None): The indicator value extracted by `extractIndicators`. None if extraction failed or returned no value of the expected type. created (bool): Whether the indicator was successfully created by `CreateNewIndicatorsOnly`. enriched (bool): Whether external enrichment (`enrichIndicators` or internal commands) succeeded for this indicator. tim_context (list[ContextResult] | None): List of TIM results (standardized enrichment results) returned from TIM search. None if no TIM results were found. hr_message (str | None): Message for human readable. error_message (str | None): Accumulated error message from enrichment pipeline. context_message (str | None): Small message for Context, relevant for failures. final_status (Status): The computed status of the indicator after all enrichment stages. Typically SUCCESS unless any stage failed or an invalid input was detected. indicator_score (float | None): The final Score/CVSS from the TIM data relevant for final Human Readable. """ raw_input: str extracted_value: str | None = None created: bool = False enriched: bool = False tim_context: list[ContextResult] | None = None hr_message: str | None = None context_message: str | None = None error_message: str | None = None final_status: Status = Status.SUCCESS indicator_score: float | None = None def _is_valid(self) -> bool: return bool(self.extracted_value) def _tim_found(self) -> bool: return bool(self.tim_context) def _build_failure_context(self) -> str | None: """Builds a consistent failure context message.""" messages: list[str] = [] if not self.created: messages.append("Failed To Create Indicator using CreateNewIndicatorsOnly.") if not self.enriched: messages.append("Failed to Enrich using enrichIndicator.") if not self._tim_found(): messages.append("Failed to extract from TIM using findIndicator.") return " | ".join(messages) if messages else None def compute_status(self) -> None: """ Evaluates the instance's internal state to determine the final verdict (SUCCESS/FAILURE) and generates the appropriate Human Readable/ Context message. """ # Case 1 – Invalid indicator (highest priority) if not self._is_valid(): self.final_status = Status.FAILURE self.context_message = self.context_message or "Invalid" # Case 2 – Valid + Successful enrichment pipeline elif self.enriched and self._tim_found(): self.final_status = Status.SUCCESS # Case 3 – Enrichment Pipeline failed else: self.final_status = Status.FAILURE self.context_message = self._build_failure_context() class CommandType(Enum): """ Execution policy category: - INTERNAL: runs when no brands specified, or if its brand is requested (e.g., wildfire-get-verdict). - EXTERNAL: runs only if `external_enrichment` is True or specific brands were provided (e.g., enrichIndicator). - BUILTIN: server builtins (e.g., createNewIndicator) always run regardless of flags. """ INTERNAL = "Internal" EXTERNAL = "External" BUILTIN = "Builtin" class Command: """ A single command spec for batch execution. Args: name (str): The command name (e.g., 'enrichIndicators', 'url'). args (dict[str, Any]): Arguments for the command. brand (str): The specific integration brand to use. leave empty to exclude the command from the human readable entries. command_type (CommandType): The type of the command. ignore_using_brand (bool): Whether to add the using-brand parameter to the args. is_multi_input (bool): Whether the command accepts multiple inputs. Relevant for HumanReadable output. is_aggregated_output (bool): Whether the command return one output for multiple inputs. Relevant for HumanReadable output. When is_multi_input is True and not is_aggregated_output is True, will split the data list one per result entry. context_output_mapping (dict[str, str]): Mapping rules from source context to target; the separator is `".."`. - {"Name": "Value"} -> flat key rename. - {"Name..Value": "Value"} -> nested extract {"Name":{"Value":x}} -> {"Value":x}. - Append "[]" to make the destination a list. - If "Score" is among mapped keys, we compute MaxScore/MaxVerdict/TIMScore. - If "CVSS" is among mapped keys, we extract TIMCVSS. """ def __init__( self, name: str, args: dict | None = None, brand: str = "", command_type: CommandType = CommandType.EXTERNAL, context_output_mapping: dict[str, str] | None = None, ignore_using_brand: bool = False, is_multi_input: bool = False, is_aggregated_output: bool = False, ): self.name = name self.args = args self.brand = brand self.command_type = command_type self.context_output_mapping = context_output_mapping self.ignore_using_brand = ignore_using_brand self.is_multi_input = is_multi_input self.is_aggregated_output = False if not is_multi_input else is_aggregated_output def __str__(self) -> str: """ Returns a string representation of the Command object. """ return f"Command(name='{self.name}', args={self.args}, type={self.command_type.value})" def to_batch_item(self, brands_to_run: list[str] | None = None) -> dict[str, dict[str, Any]]: """ Converts the command object to the format required by `executeCommandBatch`. inject using-brand to args if brands_to_run is not empty and ignore_using_brand is False. """ final_args = self.args.copy() if self.args else {} if brands_to_run and not self.ignore_using_brand: final_args["using-brand"] = ",".join(brands_to_run) return {self.name: final_args} def execute(self) -> list[dict[str, Any]]: """Executes the command and returns the results.""" demisto.debug(f"[Command.execute] Executing command {self.name} with args: {self.args}") is_failed, results = execute_command(self.name, self.args, fail_on_error=False) if is_failed: demisto.debug(f"[Command.execute] Command {self.name} execution failed with error: {results}") demisto.debug(f"[Command.execute] Command {self.name} execution completed with {len(results)} results") return results class BatchExecutor: """ Executes commands in batch/batches and performs minimal, uniform result processing. """ def process_results( self, results: list[list[ContextResult]], commands_list: list[Command], verbose: bool = False, ) -> list[CommandProcessResults]: """ Normalize command results into `(result, hr_output, error_message)` tuples, skipping debug entries. Args: results (list[list[ContextResult]]): The results of the batch of commands. commands_list (list[Command]): The list of commands that were executed. verbose (bool): Whether to print verbose output. Returns: list[CommandProcessResults]: List of lists of tuples (result, hr_output, error_message). """ demisto.debug("[BatchExecutor.process_results]") final_results = [] for results_list, command in zip(results, commands_list): demisto.debug(f"Processing results for command {command.name} with {len(results_list)} results") processed_command_results = [] for i, result in enumerate(results_list): if is_debug_entry(result): demisto.debug(f"Result #{i+1} is debug") continue hr_output = "" error_message = "" brand = result.get("Metadata", {}).get("brand") or command.brand or "Unknown" if is_error(result): demisto.debug(f"Result #{i+1} is error") error_message = f"Error in {command.name}. " + get_error(result) demisto.debug(f"Command {command.name} with args: {command.args} failed with error {error_message}") if verbose: hr_output = ( f"#### Error for name={command.name} args={command.args} current brand={brand}\n{error_message}" ) if human_readable := result.get("HumanReadable"): hr_output += f"\n\n{human_readable}" elif verbose: if human_readable := result.get("HumanReadable"): hr_output = ( f"#### Result for name={command.name} args={command.args} current brand={brand}\n{human_readable}" ) demisto.debug(f"Result #{i+1} processed") processed_command_results.append((result, hr_output, error_message)) final_results.append(processed_command_results) return final_results def execute_batch( self, commands: list[Command], brands_to_run: list[str] | None = None, verbose: bool = False, ) -> list[CommandProcessResults]: """ Execute one batch (list of Command). Returns a list aligned to `commands`, where each item is a list of `(result, hr_output, error_message)` tuples. Args: commands (list[Command]): List of commands to execute. brands_to_run (list[str]): List of brands to run on. verbose (bool): Whether to print verbose output. Returns: list[CommandProcessResults]: List of lists of tuples (result, hr_output, error_message). """ brands_to_run = brands_to_run or [] commands_to_execute = [command.to_batch_item(brands_to_run) for command in commands] command_names = [command.name for command in commands] demisto.debug( f"[Timing][execute_batch] Executing batch: {len(commands_to_execute)} commands " f"({command_names}); using-brands={brands_to_run or 'all'}" ) batch_start = time.perf_counter() results = demisto.executeCommandBatch(commands_to_execute) # Results is list of lists, for each command list of results batch_elapsed = time.perf_counter() - batch_start demisto.debug( f"[Timing][execute_batch] demisto.executeCommandBatch for commands {command_names} " f"took {batch_elapsed:.3f}s" ) demisto.debug("Batch returned [" + ", ".join(str(len(r)) for r in results) + "] results before processing") return self.process_results(results, commands, verbose) def execute_list_of_batches( self, list_of_batches: list[list[Command]], brands_to_run: list[str] | None = None, verbose: bool = False, ) -> list[list[CommandProcessResults]]: """ Execute batches in order. See `execute_batch` for inner tuple shape. Args: list_of_batches (list[list[Command]]): A list of batches to execute. each batch is list of commands. brands_to_run (list[str]): A list of brands to run on. verbose (bool): Whether to print verbose output. Returns: List of corresponding results for each batch. Foreach batch list of results list per commands. Foreach command list of tuples (ContextResult, str, str) corresponding to (result, hr_output, error_message). Example: [ [ # batch 0 [ (result, hr_output, error), ... ], # command 0 results [ (result, hr_output, error), ... ], # command 1 results ], [ # batch 1 ... ] ] """ out: list[list[list[tuple[ContextResult, str, str]]]] = [] for i, batch in enumerate(list_of_batches): if not batch: demisto.debug(f"Skipping empty batch #{i+1} (no commands).") out.append([]) # keep alignment; process_results will just see nothing continue demisto.debug(f"Executing batch #{i+1} with {len(batch)} commands") out.append(self.execute_batch(batch, brands_to_run or [], verbose)) return out # --- Context Builder --- class ContextBuilder: """ A builder class to handle the aggregation and merging of context data. TIM data will be added under final_context_path(val.Value && val.Value == obj.Value) after some enrichments. Other commands results will be added to context as is. Args: indicator_schema (IndicatorSchema): The indicator schema definition. final_context_path (str): The final context path where the aggregated data will be stored. """ def __init__(self, indicator_schema: IndicatorSchema, final_context_path: str): self.indicator_schema = indicator_schema self.final_context_path = final_context_path self.indicator_instances: list[IndicatorInstance] = [] self.other_context: ContextResult = {} def add_indicator_instances(self, indicator_instances: list[IndicatorInstance]): """ Registers a list of indicator instances to be included in the final context build. These instances hold the lifecycle state (Created, Enriched, TIM Data) of specific indicator values. Args: indicator_instances (list[IndicatorInstance]): The list of indicator instances to add. """ self.indicator_instances.extend(indicator_instances) def add_other_commands_results(self, commands_ctx: ContextResult): """ Adds context from non enrichment commands. Will be added to the final context as is. Args: commands_ctx (ContextResult): The commands context. """ self.other_context.update(commands_ctx or {}) def build(self) -> ContextResult: """ Builds the final context by merging 'other' context and constructing the aggregated indicator objects. The Context will be constructed as follows: { self.final_context_path(val.Value && val.Value == obj.Value):[ { "Value":"Example", "MaxScore":1, //Optional "MaxVerdict":"Malicious", //Optional "TIMScore":1, //Optional "TIMCVSS":1, //Optional "Status": "Stale"/"Fresh"/"Manual", "ModifiedTime": "2022-01-01T00:00:00Z", "Results": [ # From Enrichment Brands { "Brand": "Brand1", "Score": 1, "Data": "Example", ... } ] } ], ... (Other keys from add_other_commands_results) } Returns: ContextResult: The final cleaned context dictionary. """ final_context: ContextResult = {} if self.indicator_instances: indicator_list = self.build_indicators_context() self.enrich_final_indicator(indicator_list) final_context[f"{self.final_context_path}(val.Value && val.Value == obj.Value)"] = indicator_list final_context.update(self.other_context) return remove_empty_elements_with_exceptions(final_context, exceptions={"TIMCVSS", "Status", "ModifiedTime"}) def build_indicators_context(self) -> list[dict]: """ Iterates over the stored IndicatorInstances and constructs the standardized dictionary structure. It processes both successful (found in TIM) and failed (Error/Invalid) instances. Returns: list[dict]: A list of dictionaries representing the final state of each indicator. """ results: list[dict] = [] for indicator_instance in self.indicator_instances: indicator_instance.compute_status() if indicator_instance.final_status == Status.FAILURE: results.append(self._build_failure_indicator_context(indicator_instance)) else: results.append(self._build_success_indicator_context(indicator_instance)) return results def _build_failure_indicator_context(self, instance: IndicatorInstance) -> dict[str, Any]: """ Builds context entry for an indicator that failed processing. Args: instance (IndicatorInstance): The instance. Return: dict: A dict represents failed entry context for the final context. """ return { "Value": instance.extracted_value or instance.raw_input, "Status": IndicatorStatus.ERROR.value, "Message": instance.context_message, } def _build_success_indicator_context(self, instance: IndicatorInstance) -> dict[str, Any]: """ Builds the context entry for a successfully processed indicator. Extracts metadata (Status, ModifiedTime, Scores) from the TIM result. Example: Input Instance: instance.extracted_value = "1.1.1.1" instance.tim_context = [ {"Brand": "TIM", "Status": "Fresh", "ModifiedTime": "2023-01-01T00:00:00Z", "Score": 3}, {"Brand": "VirusTotal", "Score": 2} ] Output: { "Value": "1.1.1.1", "Status": "Fresh", "ModifiedTime": "2023-01-01T00:00:00Z", "TIMScore": 3, "Results": [ # Note: Status and ModifiedTime are popped from the TIM entry {"Brand": "TIM", "Score": 3}, {"Brand": "VirusTotal", "Score": 2} ] } """ value = instance.extracted_value current_indicator: dict[str, Any] = {"Value": value} # TIM section if tim_obj := self._get_tim_entry(instance): # Extract File hashing if the indicator is file if self.indicator_schema.type == "file": current_indicator["Hashes"] = self._get_file_hashes(instance, tim_obj) # Main Indicator Fields current_indicator["Status"] = pop_dict_value(tim_obj, "Status") current_indicator["ModifiedTime"] = pop_dict_value(tim_obj, "ModifiedTime") if "Score" in self.indicator_schema.context_output_mapping: current_indicator["TIMScore"] = tim_obj.get("Score") if "CVSS" in self.indicator_schema.context_output_mapping: current_indicator["TIMCVSS"] = tim_obj.get("CVSS") current_indicator["Results"] = instance.tim_context return current_indicator def _get_tim_entry(self, instance: IndicatorInstance) -> dict[str, Any] | None: """ Returns the first TIM result from tim_context with Brand='TIM' (only one should exists per indicatorInstance). Args: instance (IndicatorInstance): The instance. Return: dict: The TIM results from tim_context. """ if not instance.tim_context: return None return next((i for i in instance.tim_context if i.get("Brand") == TIM_BRAND), None) def _get_file_hashes(self, instance: IndicatorInstance, tim_obj: dict) -> dict: """ Returns file hashes dict from TIM or builds a default hash dict. Extract all hashes values from the tim_obj if exists. Otherwise (Can happen when no enrichment happen so the CustomFields was empty) will create Hash dict from the value itself. Example: 1) tim_obj = { "SHA256":"AAA", "MD5":"BBB", "Other":"CCC" } Returns = { "SHA256":"AAA", "MD5":"BBB" } 2) tim_obj = { "Other":"CCC" } instance.extracted_value = "AAA" # Some SHA256 Value Return = { "SHA256": "AAA" } Args: instance (IndicatorInstance): The instance. Returns: dict[str, str]: Mapping of hash type → hash value derived from the TIM object or the indicator value. """ return self.indicator_schema.get_all_values_from(tim_obj) or build_hash_dict(instance.extracted_value) def enrich_final_indicator(self, indicator_list: list[dict]): """ Calculates aggregate fields (MaxScore, MaxVerdict) for the final indicator list. Args: indicator_list (list[dict]): The list of indicator dictionaries to enrich in-place. """ for indicator in indicator_list: if "Score" in self.indicator_schema.context_output_mapping and indicator.get("Status") != IndicatorStatus.ERROR.value: all_scores = [res.get("Score", 0) for res in indicator.get("Results", [])] + [indicator.get("TIMScore", 0)] max_score = max(all_scores or [0]) indicator["MaxScore"] = max_score indicator["MaxVerdict"] = DBOT_SCORE_TO_VERDICT.get(max_score, "Unknown") class BrandManager: """ Centralizes brand management. - requested: brands the user asked for - enabled: brands with active instances - to_run: [] => all brands, else intersection(requested, enabled) - missing: requested but not enabled - unsupported_external(...): requested brands that are only meaningful via EXTERNAL enrichment but currently not enabled """ def __init__(self, requested: list[str] | None): self.requested = requested demisto.debug(f"[BrandManager] requested={self.requested}") @cached_property def enabled(self) -> set[str]: """Return set of active integration brands.""" return BrandManager.enabled_brands() @staticmethod def enabled_brands() -> set[str]: """Return set of active integration brands.""" all_modules = demisto.getModules() enabled_brands = {module.get("brand") for module in all_modules.values() if module.get("state") == "active"} demisto.debug(f"[BrandManager] enabled={enabled_brands}") return enabled_brands @staticmethod def get_brands_by_type(command_batches: list[list[Command]] | None, command_type: CommandType) -> list[str]: """ Returns all unique brands that appear in the given command batches for the specified command type. Help getting all Internal command brands to inject them to the brand variable. For example when WildFire-v2 is internal enrichment brand, we will add it to brands (when external_enrichment is false and no brands is given). in order to not skip other Internal commands such as core-get-hash-analytics-prevalence, we will call this function to append the internal brands to run them as well. Args: command_batches: A list of batches containing `Command` objects. command_type: The command type to filter by. Returns: A list of unique brand names. Example: Input: All command batches, one of them is Internal with brand 'Cortex Core-IR'. Calling: get_brands_by_type(batches, CommandType.INTERNAL) Returns: ['Cortex Core-IR'] """ if not command_batches: return [] brands: set[str] = set() for batch in command_batches: for command in batch: if command.command_type == command_type: brands.add(command.brand) return list(brands) @cached_property def to_run(self) -> list[str]: """ Active brands to run based on user input. If none provided, returns empty -> means "all". """ if not self.requested: demisto.debug("No specific brands provided; will run on all available brands.") return [] brands_to_execute = list(set(self.requested) & set(self.enabled)) if not brands_to_execute: raise DemistoException( "None of the provided brands correspond to an enabled integration instance. " "Please ensure brand names are correct and instances are enabled." ) demisto.debug(f"[BrandManager] to_run={brands_to_execute}") return brands_to_execute @cached_property def missing(self) -> list[str]: """ Returns a list of missing brands from the given brands. If no brands are given, returns empty list. Caches the result to avoid redundant calculations. """ if not self.requested: return [] missing_brands = list(set(self.requested) - self.enabled) demisto.debug(f"[BrandManager] missing={missing_brands}") return missing_brands def unsupported_external(self, commands: list[list["Command"]]) -> list[str]: """ Returns a list of unsupported enrichment brands to run on from the given brands. If no brands are given, returns empty list. """ if not self.requested: demisto.debug("No specific brands provided; will run on all available brands.") return [] non_external_brands = { command.brand for command_list in commands for command in command_list if command.command_type != CommandType.EXTERNAL and command.command_type != CommandType.BUILTIN } demisto.debug(f"[BrandManager] non_external_brands={non_external_brands}") external_brands = set(self.requested) - non_external_brands demisto.debug(f"[BrandManager] external_brands={external_brands}") return list(external_brands - set(self.enabled)) # --- Main Framework Abstraction --- class AggregatedCommand(ABC): def __init__( self, args: dict, brands: list[str], verbose: bool, commands: list[list[Command]] | None = None, ): """ Initializes the module. Args: args (dict[str, Any]): The arguments from `demisto.args()`. brands (list[str]): A list of specific integration brands to run. verbose (bool): If True, include detailed command outputs. commands (list[list[Command]]): List of batches of commands to run. """ self.args = args self.brands = brands self.verbose = verbose self.commands = commands or [] self.brand_manager = BrandManager(brands) class ReputationAggregatedCommand(AggregatedCommand): def __init__( self, args: dict[str, Any], brands: list[str], indicator_schema: IndicatorSchema, indicator_instances: list[IndicatorInstance], final_context_path: str, external_enrichment: bool = False, additional_fields: bool = False, internal_enrichment_brands: list[str] | None = None, verbose: bool = False, commands: list[list[Command]] | None = None, verbose_outputs: list[str] | None = None, redundant_error_raising: bool = False, ): """ Initializes the reputation aggregated command. Args: args (dict[str, Any]): The arguments from `demisto.args()`. brands (list[str]): List of brands to run on. indicator_schema (IndicatorSchema): IndicatorSchema object to use for reputation. indicator_instances (list[IndicatorInstance]): list of all indicator we are going to enrich, both valid and invalid. will hold all information on the enrichment steps. data (list[str]): Data to enrich Example: ["https://example.com"]. final_context_path (str): Path to the context to extract to for the indicators, will be added to the path ExampleEnrichment(val.Value && val.Value == obj.Value). external_enrichment (bool): Whether to run external enrichment (e.g enrichIndicators). additional_fields (bool): Whether to include additional fields in the output. internal_enrichment_brands (list[str]): A list of internal brands to use for enrichment when no brands are provided and external_enrichment is False. Applied only if those brands are available; ignored otherwise. verbose (bool): Whether to add verbose outputs. commands (list[list[Command]]): List of batches of commands to run. verbose_outputs (list[str]): verbose outputs to add to this current verbose. redundant_error_raising (bool): will return error results only if all commands failed - for Agentix action only. """ self.external_enrichment = external_enrichment self.final_context_path = final_context_path self.additional_fields = additional_fields self.indicator_instances = indicator_instances self.redundant_error_raising = redundant_error_raising # Help to find the instance from the value itself, relevant only for extracted (valid) which we will enrich. self.indicator_mapping: dict[str, list[IndicatorInstance]] = {} for indicator_instance in indicator_instances: if indicator_instance.extracted_value: self.indicator_mapping.setdefault(indicator_instance.extracted_value.lower(), []).append(indicator_instance) self.valid_inputs = [ indicator_instance.extracted_value for indicator_instance in indicator_instances if indicator_instance.extracted_value ] self.indicator_schema = indicator_schema self.internal_enrichment_brands = internal_enrichment_brands or [] self.entry_results: list[EntryResult] = [] self.verbose_outputs = verbose_outputs or [] # If no brands and external_enrichment is false, will insert internal enrichment if available # Addes internal command brands as well to make sure they also will run if not brands and not external_enrichment: active_internal_enrichment_brands = list(set(self.internal_enrichment_brands) & BrandManager.enabled_brands()) if active_internal_enrichment_brands: brands = active_internal_enrichment_brands + BrandManager.get_brands_by_type(commands, CommandType.INTERNAL) demisto.debug("External enrichment false with no brands asked; using internal enrichment + commands") super().__init__(args, brands, verbose, commands) @cached_property def unsupported_enrichment_brands(self) -> list[str]: """Returns a list of brands that are not enabled but are required for external enrichment.""" if not self.brands: return [] return self.brand_manager.unsupported_external(self.commands) def run(self) -> CommandResults: """ Main execution loop for the reputation aggregation. """ demisto.debug("Aggregated reputation run: start") batch_results: list[list[list[tuple[ContextResult, str, str]]]] = [] context_result: ContextResult = {} batch_verbose_outputs: list[str] = [] demisto.debug("Step 1: Executing batch commands.") commands_to_execute = self.prepare_commands_batches(self.external_enrichment) if commands_to_execute: demisto.debug( f"Executing {sum(len(b) for b in commands_to_execute)} commands in " f"{len(commands_to_execute)} batch(es)" ) batch_executor = BatchExecutor() batch_results = batch_executor.execute_list_of_batches(commands_to_execute, self.brand_manager.to_run, self.verbose) if batch_results: context_result, batch_verbose_outputs, entry_results = self.process_batch_results( batch_results, commands_to_execute ) self.verbose_outputs += batch_verbose_outputs self.entry_results += entry_results else: demisto.debug("No batch results.") demisto.debug("Step 2: Finding indicators.") self.get_indicators_from_tim() self.update_indicator_instances_status() demisto.debug("Step 3: Building final context.") context_builder = ContextBuilder(self.indicator_schema, self.final_context_path) context_builder.add_other_commands_results(context_result) context_builder.add_indicator_instances(self.indicator_instances) final_context = context_builder.build() demisto.debug("Step 4: Summarizing command results.") return self.summarize_command_results(final_context) def update_indicator_instances_status(self): """ Updates each IndicatorInstance based on the EntryResults of CreateNewIndicatorsOnly and enrichIndicator Commands. The following IndicatorInstance attributes will be updated: - created (bool): If CreateNewIndicatorsOnly succeeded/failed. - enriched (bool) If enrichIndicator succeeded/failed. if enrichIndicator does not exits (wasn't called) we will look at it as succeeded. - error_message (str): If one of the command failed will add the error message from the EntryResult Object. - final_status based on errors. """ create_entry = next( (e for e in self.entry_results if e.command_name == "CreateNewIndicatorsOnly"), None, ) if create_entry: demisto.debug(f"Create Entry: Status={create_entry.status.value}, args={create_entry.args}") enrich_entry = next( (e for e in self.entry_results if e.command_name == "enrichIndicators"), None, ) if enrich_entry: demisto.debug(f"Enrich Entry: Status={enrich_entry.status.value}, args={enrich_entry.args}") errors: list[str] = [] # Determine CreateNewIndicatorsOnly status and error message is_created = bool(create_entry and create_entry.status == Status.SUCCESS) if not is_created and create_entry: errors.append(create_entry.message or "Creating indicator failed.") # Determine enrichIndicator status and error message # enrich_entry is None → enrichment stage not used → treat as success if enrich_entry is None: is_enriched = True else: is_enriched = enrich_entry.status == Status.SUCCESS if not is_enriched: msg = enrich_entry.message or "Enrichment failed." errors.append(msg) for inst in self.indicator_instances: if inst.extracted_value: inst.created = is_created inst.enriched = is_enriched if errors: inst.error_message = " | ".join(errors) if errors else None # trigger the logic internally inst.compute_status() def prepare_commands_batches(self, external_enrichment: bool = False) -> list[list[Command]]: """ The commands that will be added to execution are filtered as follow: 1. external_enrichment=False, brands=[] → INTERNAL + BUILTIN. 2. external_enrichment=True, brands=[] → INTERNAL + EXTERNAL + BUILTIN. 3. external_enrichment=False, brands != []: - INTERNAL only if command.brand in self.brands. - EXTERNAL: all external commands are included (using-brand: {brands} injected to the command later). - BUILTIN included. Args: external_enrichment (bool): Flag to determine if external commands should run. Return: list[list[Command]]: The command batches after filtering. """ demisto.debug(f"Preparing commands. External enrichment: {external_enrichment}") prepared_commands: list[list[Command]] = [] for command_list in self.commands: current_command_list: list[Command] = [] for command in command_list: if command.command_type == CommandType.INTERNAL and (command.brand in self.brands or not self.brands): # If Command is internal and brand is in brands or brands is empty, add it to the list. demisto.debug(f"Adding internal command {command}") current_command_list.append(command) elif command.command_type == CommandType.EXTERNAL and (external_enrichment or self.brands): # If Command is external and external_enrichment is True or brands is not empty, add it to the list. demisto.debug(f"Adding external command {command}") current_command_list.append(command) # added using-brand argument on execution elif command.command_type == CommandType.BUILTIN: # If Command is built-in, add it to the list. demisto.debug(f"Adding Builtin command {command}") current_command_list.append(command) else: demisto.debug(f"Skipping command {command} | {command.command_type} | {self.brands}") prepared_commands.append(current_command_list) return prepared_commands def get_indicators_from_tim(self): """ Searches TIM for indicators and processes the results. Updates the relevant Fields of each of the indicator instances. """ iocs = self.search_indicators_in_tim() if not iocs: demisto.debug("No Search Results") return self.process_tim_results(iocs) def search_indicators_in_tim(self) -> list[ContextResult]: """ Performs the actual search against TIM using the IndicatorsSearcher class. Returns: list[ContextResult]: The search results. """ indicator_values = " ".join( { f'"{indicator_instance.extracted_value}"' for indicator_instance in self.indicator_instances if indicator_instance.extracted_value } ) query = f"type:{self.indicator_schema.type} and (value:({indicator_values}))" try: demisto.debug(f"Executing TIM search with query: {query}") searcher_start = time.perf_counter() searcher = IndicatorsSearcher(query=query) iocs = flatten_list([res.get("iocs", []) for res in searcher]) searcher_elapsed = time.perf_counter() - searcher_start demisto.debug( f"[Timing][search_indicators_in_tim] IndicatorsSearcher returned {len(iocs)} raw IOCs " f"in {searcher_elapsed:.3f}s" ) if not iocs: return [] return iocs except Exception as e: msg = f"Error searching TIM: {str(e)}" demisto.debug(msg) for indicator_instance in self.indicator_instances: if indicator_instance.extracted_value: indicator_instance.hr_message = msg return [] def process_tim_results(self, iocs: list[dict[str, Any]]): """Processes raw IOCs from a TIM search into structured context. Updates the indicator instances tim_context with the relevant extracted context. Args: iocs (list[dict[str, Any]]): The IOC objects from the TIM search. """ demisto.debug(f"Processing {len(iocs)} IOCs from TIM.") for i, ioc in enumerate(iocs): demisto.debug(f"Processing #{i+1} TIM result") parsed_indicators, indicator_score, value, message = self._process_single_tim_ioc(ioc) instances = self.indicator_mapping.get(value.lower(), []) for indicator_instance in instances: indicator_instance.tim_context = parsed_indicators indicator_instance.indicator_score = indicator_score demisto.debug(f"Score2: {indicator_score}, {indicator_instance.indicator_score} ") indicator_instance.hr_message = message def _process_single_tim_ioc(self, ioc: dict[str, Any]) -> tuple[list[dict], float | None, str, str]: """ Processes a single IOC object returned from a TIM search. Extract Score and brand and add them to parsed indicators. Args: ioc (dict[str, Any]): The IOC object to process. Returns: tuple[list[dict], float, str, str]: - The parsed indicators. - Indicator score/cvss. - Indicator value. - HR message for final table. """ all_parsed_indicators: list[ContextResult] = [] tim_indicator, indicator_score = self.create_tim_indicator(ioc) all_parsed_indicators.append(tim_indicator) value = tim_indicator.get("Value", "") found_brands = [] demisto.debug(f"Extracting per brand information for {value}") for brand, brand_data in ioc.get("insightCache", {}).get("scores", {}).items(): demisto.debug(f"Processing TIM indicators from brand: {brand}") score = brand_data.get("score", 0) context = brand_data.get("context", {}) reliability = brand_data.get("reliability", "") parsed_indicators = self.parse_indicator(context, brand, reliability, score) if parsed_indicators: found_brands.append(brand) all_parsed_indicators.extend(parsed_indicators) message = f"Found indicator from brands: {', '.join(found_brands)}." if found_brands else "No matching indicators found." return all_parsed_indicators, indicator_score, value, message def create_tim_indicator(self, ioc: dict[str, Any]) -> tuple[dict[str, Any], float | None]: """ Creates a TIM indicator from a TIM IOC CustomFields and Main Fields. Relevant for extracting the CVSS/Score/Status/ModifiedTime from the TIM IOC to the main context. Args: ioc (dict[str, Any]): The TIM IOC to create a TIM indicator from. Returns: dict[str, Any]: The TIM indicator. float | None: The Final TIM Score/CVSS. """ demisto.debug("Extracting Custom Fields") customFields = ioc.get("CustomFields", {}) # all Keys under CustomFields are lowercase while the mapping are CamelCase, this insure we will find the right keys lower_mapping = {k.lower(): v for k, v in self.indicator_schema.context_output_mapping.items()} mapped_indicator = self.map_command_context(customFields.copy(), lower_mapping, is_indicator=True) indicator_score: float | None = 0 if "Score" in self.indicator_schema.context_output_mapping: indicator_score = ioc.get("score", Common.DBotScore.NONE) mapped_indicator.update({"Score": indicator_score}) if "CVSS" in self.indicator_schema.context_output_mapping: indicator_score = customFields.get("cvssscore") mapped_indicator.update({"CVSS": indicator_score}) mapped_indicator.update( { "Brand": TIM_BRAND, "Status": self.get_indicator_status_from_ioc(ioc), "ModifiedTime": ioc.get("modifiedTime"), } ) if self.indicator_schema.type == "file": value = map_back_to_input( values=self.valid_inputs, mapping=self.indicator_schema.get_all_values_from(mapped_indicator) ) or ioc.get("value") else: value = ioc.get("value", "") mapped_indicator.update({"Value": value}) demisto.debug(f"Created TIM Indicator for {value}") return mapped_indicator, indicator_score def get_indicator_status_from_ioc(self, ioc: dict) -> str | None: """ Determine the status of a dict based on manual edits and modification time. Rules: - If "Score" is in manuallyEditedFields → "manual" - Else if modifiedTime is less than STATUS_FRESHNESS_WINDOW ago → "fresh" - Else → "stale" """ manually_edited_fields = ioc.get("manuallyEditedFields", {}) if "Score" in manually_edited_fields or "score" in manually_edited_fields: return IndicatorStatus.MANUAL.value modified_time_str = ioc.get("modifiedTime") if modified_time_str: try: modified_time = datetime.fromisoformat(modified_time_str.replace("Z", "+00:00")) except ValueError: return None if modified_time >= datetime.now(modified_time.tzinfo) - STATUS_FRESHNESS_WINDOW: return IndicatorStatus.FRESH.value else: return IndicatorStatus.STALE.value return None def process_batch_results( self, all_results: list[list[list[tuple[ContextResult, str, str]]]], commands_to_execute: list[list[Command]], ) -> tuple[ContextResult, list[str], list[EntryResult]]: """ Processes the results from the batch executor. runs through the execution results and processes each result according to the command parameters. Construct entry results and context results. Args: all_results (list[list[ContextResult]]): The results from the batch executor for all commands. commands_to_execute (list[Command]): The commands to execute. Returns: tuple[ ContextResult, The non-reputation context output. list[str], The verbose command results. list[EntryResult], The entry results. ] """ verbose_outputs: list[str] = [] entry_results: list[EntryResult] = [] context_result: ContextResult = defaultdict(lambda: defaultdict(list)) demisto.debug(f"Processing {len(all_results)} batches.") for j, (command_batch, results_batch) in enumerate(zip(commands_to_execute, all_results)): demisto.debug(f"Processing batch {j}.") for command, processed_results_list in zip(command_batch, results_batch): demisto.debug(f"Processing result for command: {command} len: {len(processed_results_list)}") for i, result_tuple in enumerate(processed_results_list): entry, mapped_ctx, verbose = self._process_single_command_result(result_tuple, command) if not command.is_aggregated_output and command.is_multi_input and i < len(self.valid_inputs): # Only if the command input is list and the command return many Command results. entry.args = self.valid_inputs[i] entry_results.append(entry) deep_merge_in_place(context_result, mapped_ctx) if verbose: verbose_outputs.append(verbose) demisto.debug(f"Command {command} processed.") return context_result, verbose_outputs, entry_results def _process_single_command_result( self, result: tuple[ContextResult, str, str], command: Command ) -> tuple[EntryResult, ContextResult, str | None]: """ Processes a single result from a batch command execution. Args: result (tuple[ContextResult, str, str]): The result from the batch command execution. command (Command): The command to execute. Returns: tuple[EntryResult, ContextResult, str | None]: The entry result, context result, and verbose output. """ raw_result, hr_output, error = result entry = EntryResult( command_name=command.name, args=command.args, brand=command.brand, status=Status.SUCCESS if not error else Status.FAILURE, message=error or "No matching indicators found.", ) mapped_context: ContextResult = {} if cmd_context := raw_result.get("EntryContext", {}): entry.message = error if error else "" demisto.debug(f"EntryContext found for command {command.name}") mapped_context = self.map_command_context(cmd_context.copy(), command.context_output_mapping) verbose_output = hr_output if self.verbose else None return entry, mapped_context, verbose_output def map_command_context( self, entry_context: dict[str, Any], context_output_mapping: dict[str, str] | None, is_indicator: bool = False ) -> dict[str, Any]: """ Maps the entry context item to the final context using the mapping. Can add [] to transform the final path value to list. If mapping is empty, return entry_context. If mapping is None, return empty dict. Args: entry_context (dict[str, Any]): The entry context item. mapping (dict[str, str]): The mapping to use. Example1: mapping = {"result..value": "final_context..value"} {"results":{"value":value}} -> {"final_context":{"value":value}} Example2: mapping = {"result..value": "final_context..value[]"} {"results":{"value":value}} -> {"final_context":{"value":[value]}} Returns: dict[str, Any]: The mapped context. """ if context_output_mapping is None: demisto.debug("Mapping is None, return Empty Dict.") return {} if not context_output_mapping: demisto.debug("Mapping is empty, return entry_context.") return entry_context if not entry_context: demisto.debug("No entry context provided, returning empty context.") return {} mapped_context: ContextResult = defaultdict() demisto.debug(f"Starting context mapping with {len(context_output_mapping)} rules. is_indicator: {is_indicator}") for src_path, dst_path in context_output_mapping.items(): if (value := pop_dict_value(entry_context, src_path)) is not None: set_dict_value(mapped_context, dst_path, value) if self.additional_fields and is_indicator and entry_context: demisto.debug(f"Adding {len(entry_context)} remaining fields to AdditionalFields.") set_dict_value(mapped_context, "AdditionalFields", entry_context) return dict(mapped_context) def parse_indicator( self, entry_context: ContextResult, brand: str, reliability: str, score: int = Common.DBotScore.NONE, ) -> list[ContextResult]: """ Parse the indicator context and complete missing fields such as brand, score, verdict if needed. self.indicator.context_output_mapping is used to map the indicator context to the final context. What is not mapped is added to the AdditionalFields if AdditionalFields is enabled. Final indicator is saved under the following structure: {indicator_value: {brand: [indicator]}} {"https://example.com":[ {"indicator_value":"https://example.com", "brand":"brand", "score": 1, "verdict": "Good"}, ] } Args: entry_context (ContextResult): The entry context item straight from the command result.entry_context. brand (str): The brand from the result.metadata.brand. score (int, optional): The score. Defaults to Common.DBotScore.NONE. Returns: list[ContextResult]: The parsed result. """ demisto.debug(f"Starting parsing indicators from brand '{brand}'.") indicators_context: list[ContextResult] = [] indicator_entries = flatten_list( [v for k, v in entry_context.items() if k.startswith(self.indicator_schema.context_path_prefix)] ) demisto.debug(f"Extracted {len(indicator_entries)} indicators from {brand} entry context.") for indicator_data in indicator_entries: indicator_value = self.indicator_schema.get_all_values_from(indicator_data) demisto.debug(f"Parsing indicator: {indicator_value}") mapped_indicator = self.map_command_context( indicator_data, self.indicator_schema.context_output_mapping, is_indicator=True ) if "Score" in self.indicator_schema.context_output_mapping: mapped_indicator["Score"] = score mapped_indicator["Verdict"] = DBOT_SCORE_TO_VERDICT.get(score, "Unknown") mapped_indicator["Brand"] = brand mapped_indicator["Reliability"] = reliability indicators_context.append(mapped_indicator) demisto.debug(f"Parsed indicator '{indicator_value}' from brand '{brand}'") return indicators_context def _inject_unsupported_brands(self): """ If any unsupported enrichment brands exist, append a summary entry for each of them. """ if not self.unsupported_enrichment_brands: return # Add Entry with all requested unsupported brands. demisto.debug(f"Missing brands: {self.unsupported_enrichment_brands}") for unsupported_enrichment_brand in self.unsupported_enrichment_brands: self.entry_results.append( EntryResult( command_name=self.indicator_schema.type, args="", brand=unsupported_enrichment_brand, status=Status.FAILURE, message="Unsupported Command: Verify you have proper integrations enabled to support it", ) ) def _build_final_human_readable(self, final_entries: list[EntryResult]) -> str: """ Builds the final human-readable markdown table and appends verbose outputs if enabled. Args: final_entries (list[EntryResult]): The entries to include in the final table. Returns: str: The human-readable markdown output. """ human_readable = tableToMarkdown( "Final Results", t=[entry.to_entry() for entry in final_entries], headers=[ "Brand", "Arguments", "Status", "TIM CVSS" if self.indicator_schema.type == "cve" else "TIM Score", "Message", ], ) if self.verbose and self.verbose_outputs: demisto.debug("Adding verbose outputs to human readable.") human_readable += "\n\n".join(self.verbose_outputs) return human_readable def _is_final_result_error(self, entries: list[EntryResult]) -> bool: """ Determines whether the overall result should be treated as an error. Rules: - Return True only if: * there was at least one FAILURE, and * all entries are either FAILURE or "No matching indicators found." - Otherwise, return False (success or partial success). Args: entries (list[EntryResult]): The final entries that will appear in the HR table. Returns: bool: True if the result should be treated as an error, False otherwise. """ if not entries: return False if self.redundant_error_raising: all_failed_or_no_match = all(entry.status == Status.FAILURE for entry in entries) else: all_failed_or_no_match = all( entry.status == Status.FAILURE or entry.message == "No matching indicators found." for entry in entries ) any_failure = any(entry.status == Status.FAILURE for entry in entries) return all_failed_or_no_match and any_failure def summarize_command_results(self, final_context: dict[str, Any]) -> CommandResults: """ Construct the final Command Result with the appropriate readable output and context. Summarizes the human readable output. Adds verbose messages from all commands if verbose is True. If all commands failed, return an error message. If no indicator found and at least one command failed, return error message. If all commands succeeded with no indicators found, return a success message. If at least one command succeeded with indicators found, return a success message. Args: final_context (ContextResult): The final context. Returns: CommandResults: The command results. """ demisto.debug(f"Summarizing final results from {len(self.entry_results)} command entries.") self.create_indicators_entry_results() self._inject_unsupported_brands() # Remove Entries from non brands command such as CreateNewIndicator and EnrichIndicator final_entries = [entry for entry in self.entry_results if entry.brand != ""] human_readable = self._build_final_human_readable(final_entries) # Return an error only if there were no successes AND at least one of those was a hard failure. if self._is_final_result_error(final_entries): demisto.debug("All commands failed or no indicators found. Returning an error entry.") raise DemistoException( "Error: All commands failed or no indicators found.\n" + human_readable, ) demisto.debug("Returning a success entry (at least one command succeeded or no hard failures).") return CommandResults( readable_output=human_readable, outputs=final_context, raw_response=flatten_context_values(final_context), ) def create_indicators_entry_results(self): """ Create one EntryResult Object for each of the indicator instances, based on the commands and enrichment process. Relevant for the final HR table. """ entry_results: list[EntryResult] = [] for indicator_instance in self.indicator_instances: # If enrichment pipeline failed (status == failure) we will add the error message if exists. # If status == failure and no error message will add the context message generated under compute_status function # If status == success hr message will be the one generated under process _tim_result (Found indicator from brands:..) message = ( indicator_instance.error_message or indicator_instance.context_message if indicator_instance.final_status == Status.FAILURE else indicator_instance.hr_message ) entry_results.append( EntryResult( command_name="", brand=TIM_BRAND, status=indicator_instance.final_status, args=indicator_instance.extracted_value or indicator_instance.raw_input, score_field_name="TIM CVSS" if self.indicator_schema.type == "cve" else "TIM Score", score=indicator_instance.indicator_score, message=message or "", ) ) self.entry_results[:0] = entry_results """HELPER FUNCTIONS""" def build_hash_dict(value: str | None) -> dict[str, str]: """ Constructs a dictionary mapping the hash type to the hash value. Args: value (str): The hash string to process. Returns: dict[str, str]: A dictionary where the key is the uppercase hash type (e.g., "MD5") and the value is the original hash string. """ if value: return {get_hash_type(value).upper(): value} return {} def map_back_to_input(values: list[str], mapping: dict[str, str]) -> str: """ Find the original input value that matches one of the mapped hash values. This compares `values` to `mapping.values()` case-insensitively and returns the first input value that appears in the mapping (e.g., to map back from TIM hash fields to the hash the user originally provided). Args: values: Original input values (e.g., hashes provided by the user). mapping: Hash field → value mapping (e.g., {"MD5": "...", "SHA256": "..."}). Returns: The matching original value, or an empty string if no match is found. """ lower_mapping_values = [v.lower() for v in mapping.values()] for v in values: if v.lower() in lower_mapping_values: return v return "" def create_and_extract_indicators( data: list[str], indicator_type: str, mark_mismatched_type_as_invalid: bool = False, ) -> tuple[list[IndicatorInstance], str]: """ Extract indicators from the provided input list for a specific indicator type, using the `extractIndicators` command. Create IndicatorInstance foreach of the inputs and if is the right type will update the extracted field. """ if not data: raise ValueError("No data provided to enrich") indicators_instances: list[IndicatorInstance] = [] full_hr: list[str] = [] valid_set: set[str] = set() invalid_set: set[str] = set() expected_type_lower = indicator_type.lower() total_start = time.perf_counter() for raw in data: if raw in valid_set or raw in invalid_set: continue single_start = time.perf_counter() instances_for_raw, hr = _process_single_input( raw=raw, expected_type_lower=expected_type_lower, mark_mismatched_type_as_invalid=mark_mismatched_type_as_invalid, valid_set=valid_set, invalid_set=invalid_set, ) single_elapsed = time.perf_counter() - single_start demisto.debug( f"[Timing][create_and_extract_indicators] extractIndicators for input '{raw}' " f"took {single_elapsed:.3f}s" ) indicators_instances.extend(instances_for_raw) full_hr.append(hr) total_elapsed = time.perf_counter() - total_start demisto.debug( f"[Timing][create_and_extract_indicators] Total one-by-one extractIndicators for " f"{len(data)} inputs took {total_elapsed:.3f}s" ) if not valid_set: raise ValueError("No valid indicators found in the input data.") demisto.debug(f"Valid Inputs: {valid_set}") demisto.debug(f"Invalid Inputs: {invalid_set}") return indicators_instances, "".join(full_hr) # This function is relevant only for the Agentix enrichment scripts def create_and_extract_indicators_batch(data: list[str], indicator_type: str) -> tuple[list[IndicatorInstance], str]: """ Extract and validate a batch of indicator values using a single `extractIndicators` call, then build IndicatorInstance objects for every valid indicator of the requested type. - Processes the entire list in a single `extractIndicators` call (one round-trip). - Does NOT raise when no valid indicators are found – it returns an empty list of instances along with an explanatory human-readable string. Args: data: List of raw string values to validate. indicator_type: The expected indicator type (e.g., "IP", "URL", "Domain"). Returns: A tuple of: - list[IndicatorInstance]: Default-SUCCESS-state IndicatorInstance objects for every valid extracted indicator matching `indicator_type`. Empty when none are found. - str: Human-readable markdown describing the batch extraction result. Raises: DemistoException: If extractIndicators raises an exception (actual failure). """ hr_header = f"\n\n### Result for name=extractIndicators (batch) args='text': {data}\n\n" if not data: demisto.debug("[create_and_extract_indicators_batch] Empty data provided, returning empty list.") return [], hr_header + "No input data provided." demisto.debug(f"[create_and_extract_indicators_batch] Validating {len(data)} values for type '{indicator_type}'.") try: batch_extract_start = time.perf_counter() results = execute_command("extractIndicators", {"text": data}, extract_contents=False) batch_extract_elapsed = time.perf_counter() - batch_extract_start demisto.debug( f"[Timing][create_and_extract_indicators_batch] extractIndicators batch call for " f"{len(data)} inputs took {batch_extract_elapsed:.3f}s" ) except Exception as ex: raise DemistoException(f"Failed to validate input using extractIndicators: {ex}") from ex if not results: demisto.debug( "[create_and_extract_indicators_batch] extractIndicators returned empty/None results, returning empty list." ) return [], hr_header + "extractIndicators returned no results." extracted_ctx: dict = results[0].get("EntryContext", {}).get("ExtractedIndicators", {}) or {} demisto.debug(f"[create_and_extract_indicators_batch] Extracted context keys: {list(extracted_ctx.keys())}") expected_type_lower = indicator_type.lower() matched_indicators, _ = _split_expected_and_other_types(extracted_ctx, expected_type_lower) hr = hr_header + tableToMarkdown(name="Extracted Indicators", t=extracted_ctx) if not matched_indicators: demisto.debug("[create_and_extract_indicators_batch] extractIndicators No valid indicators returned.") return [], hr indicator_instances = [IndicatorInstance(raw_input=value, extracted_value=value) for value in matched_indicators] demisto.debug( f"[create_and_extract_indicators_batch] Found {len(indicator_instances)} valid indicators " f"of type '{indicator_type}'." ) return indicator_instances, hr def _process_single_input( raw: str, expected_type_lower: str, mark_mismatched_type_as_invalid: bool, valid_set: set[str], invalid_set: set[str], ) -> tuple[list[IndicatorInstance], str]: """ Handle extractIndicators + IndicatorInstance creation for a single raw input. Mutates valid_set / invalid_set to keep global bookkeeping. """ demisto.debug(f"Validating input '{raw}' using extractIndicators") instances: list[IndicatorInstance] = [] extracted_ctx, hr_fragment, error_message = _execute_extraction(raw) # If extraction failed, mark as invalid and return if error_message is not None: if raw not in invalid_set: instances.append( IndicatorInstance( raw_input=raw, context_message="extractIndicators Failed", error_message=error_message, final_status=Status.FAILURE, ) ) invalid_set.add(raw) return instances, hr_fragment expected_indicators, has_other_types = _split_expected_and_other_types(extracted_ctx, expected_type_lower) demisto.debug(f"expected for '{raw}': {expected_indicators}, other types exists={has_other_types}") # Extracted something, but none are of the expected type -> invalid # Or expected type + other types and we treat that as invalid if (not expected_indicators and raw not in invalid_set) or ( mark_mismatched_type_as_invalid and has_other_types and raw not in invalid_set ): demisto.debug("Invalid input") instances.append( IndicatorInstance( raw_input=raw, hr_message="Invalid", final_status=Status.FAILURE, ) ) invalid_set.add(raw) return instances, hr_fragment # Valid input – create IndicatorInstance per unique expected indicator demisto.debug("Valid input") for expected_indicator in expected_indicators: if expected_indicator not in valid_set: instances.append( IndicatorInstance( raw_input=raw, extracted_value=expected_indicator, ) ) valid_set.add(expected_indicator) return instances, hr_fragment def _execute_extraction(raw: str) -> tuple[dict, str, str | None]: """ Execute the extractIndicators command for a single raw input and build the HR fragment. Returns: extracted_ctx (dict): ExtractedIndicators context (may be empty). hr_fragment (str): Markdown describing the result for this input. error_message (str | None): Error message if extraction failed, otherwise None. """ hr: str = f"\n\n### Result for name=extractIndicators args='text': {raw}\n\n" try: results = execute_command("extractIndicators", {"text": raw}, extract_contents=False) extracted_ctx = results[0].get("EntryContext", {}).get("ExtractedIndicators", {}) or {} demisto.debug(f"extractIndicators context for '{raw}': {extracted_ctx}") hr += tableToMarkdown(name="Extracted Indicators", t=extracted_ctx) return extracted_ctx, hr, None except Exception as ex: msg = str(ex) demisto.debug(f"extractIndicators failed for '{raw}': {msg}") hr += f"Error Message: {msg}" # Return empty ctx + error message; caller decides how to mark invalid return {}, hr, msg def _split_expected_and_other_types(extracted_ctx: dict, expected_type_lower: str) -> tuple[list[str], bool]: """ From the ExtractedIndicators context, separate: - the indicators of the expected type - whether there are any indicators of other types """ if not extracted_ctx: return [], False expected_indicators: list[str] = [] has_other_types = False for key, indicators in extracted_ctx.items(): if key.lower() == expected_type_lower: expected_indicators = indicators else: has_other_types = True return expected_indicators, has_other_types def deep_merge_in_place(dst: dict, src: dict) -> None: """ Recursively merges src into dst. If a key exists in both and the value is a list, it extends the list. For nested dicts, it recurses. This function modifies dst in place. Args: dst (dict): The first dictionary to merge to. src (dict): The second dictionary to merge from. """ if not src: return for k, v in src.items(): if k in dst and isinstance(dst[k], dict) and isinstance(v, dict): deep_merge_in_place(dst[k], v) elif k in dst and isinstance(dst[k], list) and isinstance(v, list): dst[k].extend(v) else: dst[k] = v def flatten_list(nested_list: list[Any]) -> list[Any]: """ Recursively flattens a nested list of lists. Args: nested_list (list): A list of that could have lists nested inside. Returns: list: A flattened list. """ flattened: list[Any] = [] for item in nested_list: if isinstance(item, list): flattened.extend(flatten_list(item)) # Recursive call for nested lists else: flattened.append(item) return flattened def set_dict_value(d: dict[str, Any], path: str, value: Any) -> None: """ Sets a value in a nested dictionary using a '..' separated path. Appends to a list if the path ends with '[]'. Args: d (dict[str, Any]): Dictionary to set nested key in. path (str): A double dot-separated key path (e.g "Signature.Copyright") value (Any): Value to set in the dictionary. """ if path is None: return is_list = path.endswith("[]") if is_list: path = path[:-2] parts = path.split("..") current = d for part in parts[:-1]: current = current.setdefault(part, {}) last_part = parts[-1] if is_list: current.setdefault(last_part, []).append(value) else: current[last_part] = value def pop_dict_value(d: dict[str, Any], path: str) -> Any: """ Retrieves a value from a nested dictionary given a ".." separated path. after getting remove the item from the dict. Returns `None` if any key along the path doesn’t exist. Args: d (Mapping[str, Any]): Dictionary to get nested key from. path (str): A ".." separated key path (e.g. "Signature..Copyright"). """ keys = path.split("..") current = d for key in keys[:-1]: if isinstance(current, dict) and key in current: current = current[key] else: return None # Path not found last_key = keys[-1] if isinstance(current, dict) and last_key in current: return current.pop(last_key) return None def is_debug_entry(execute_command_result) -> bool: """ Check if the given execute_command_result is a debug entry. Args: execute_command_result: Demisto entry (required) or result of demisto.executeCommand() Returns: bool: True if the execute_command_result is a debug entry, false otherwise """ if execute_command_result is None: return False if isinstance(execute_command_result, list) and len(execute_command_result) > 0: for entry in execute_command_result: if isinstance(entry, dict) and entry["Type"] == entryTypes["debug"]: return True return isinstance(execute_command_result, dict) and execute_command_result["Type"] == entryTypes["debug"] def remove_empty_elements_with_exceptions(d, exceptions: set[str] | None = None) -> Any: """ Recursively remove empty lists, empty dicts, or None elements from a dictionary, unless their key is in the `exceptions` set. Args: d (dict): The dictionary to remove empty elements from. exceptions (set[str], optional): A set of keys to keep even if their values are empty. Defaults to set(). Returns: dict: The dictionary with empty elements removed. """ exceptions: set[str] = exceptions or set() def empty(k, v) -> bool: """Check if a value is considered empty, unless the key is in exceptions.""" if isinstance(v, dict | list): return not v # empty dict or list return v is None and (k not in exceptions if exceptions else True) if isinstance(d, list): return [v for v in (remove_empty_elements_with_exceptions(v, exceptions) for v in d) if v not in (None, {}, [])] elif isinstance(d, dict): result = {} for k, v in d.items(): cleaned = remove_empty_elements_with_exceptions(v, exceptions) if not empty(k, cleaned) or k in exceptions: result[k] = cleaned return result else: return d def flatten_context_values(data: dict[str, Any]) -> list[dict]: """ Flattens a dictionary whose values are either: - A dictionary (e.g., {"Value": "1"}), or - A list of dictionaries (e.g., [{"Value": "2"}, {"Value": "3"}]). This is useful when you want to build a "raw" response list that contains all output items without preserving the original top-level keys. Examples: >>> flatten_context_values({ ... "IPEnrichment(val.Value && val.Value == obj.Value)": {"Value": "1"}, ... "EndpointData(val.Brand && val.Brand = ... )": [{"Value": "2"}, {"Value": "3"}], ... }) [{'Value': '1'}, {'Value': '2'}, {'Value': '3'}] Returns: A single flat list of dictionaries collected from all values. """ out: list[dict] = [] for v in data.values(): if isinstance(v, dict): out.append(v) elif isinstance(v, list): out.extend(item for item in v if isinstance(item, dict)) return out