SearchIndicatorRelationshipsAgentix

This automation outputs the indicator relationships to context according to the provided query, using the entities, entityTypes, and relationships arguments. All arguments will use the AND operator. For example, using the following arguments entities=8.8.8.8 entities_types=Domain will provide only relationships that the 8.8.8.8 indicator has with indicators of type domain.

Type

python

Pack

Base

Source

import demistomock as demisto
from CommonServerPython import *
from CommonServerUserPython import *
import re


def filter_relationships_by_entity_types(
    entities: list, entities_types: list, relationships: list, limit: int, verbose: str = "false", revoked: str = "false"
) -> list:
    """
    Filters indicator relationships by entity types using pagination.

    Args:
        entities: List of entities to search for relationships
        entities_types: List of entity types to filter by (EntityA or EntityB must match)
        relationships: List of relationship types to search for
        limit: Maximum number of filtered relationships to return

    Returns:
        list: Filtered relationships where EntityA or EntityB type matches entities_types
    """
    filtered_relationships: list = []
    searchAfter = None
    iteration = 0
    while len(filtered_relationships) < limit:
        search_params = {
            "entities": entities,
            "relationships": relationships,
            "revoked": revoked,
            "verbose": verbose,
        }

        if searchAfter:
            search_params["searchAfter"] = searchAfter

        res = demisto.executeCommand("SearchIndicatorRelationships", search_params)

        if not res or len(res) == 0:
            demisto.info("No response received from SearchIndicatorRelationships")
            break

        data = demisto.get(res[0], "Contents") or {}
        demisto.info(f"{iteration=}, Received data: {data}")
        iteration += 1
        if not data or data.get("Relationships") is None:
            demisto.info(f"Invalid or empty data received: {data}")
            break

        for relationship in data.get("Relationships", []):
            # For this case:
            # !SearchIndicatorRelationships Entities = google.com, example.com relationships=relates-to entity_type=DOMAIN
            # Relationships = google.com relates-to example.com
            if (
                relationship["EntityAType"] in entities_types
                and relationship["EntityBType"] in entities_types
                and relationship["EntityA"] in entities
                and relationship["EntityB"] in entities
            ):
                filtered_relationships.append(relationship)
            elif (relationship["EntityAType"] in entities_types and relationship["EntityA"] not in entities) or (
                relationship["EntityBType"] in entities_types and relationship["EntityB"] not in entities
            ):
                filtered_relationships.append(relationship)

                if len(filtered_relationships) >= limit:
                    break

        searchAfter_list = data.get("RelationshipsPagination", [])
        searchAfter = searchAfter_list[0] if searchAfter_list else None
        demisto.info(f"SearchAfter updated: {searchAfter}")

        if not searchAfter or len(filtered_relationships) >= limit:
            break

    demisto.info(f"{filtered_relationships=}")
    return filtered_relationships


def get_relationships(args: dict) -> list:
    """
    Retrieves indicator relationships based on specified entities, entity types, and relationships.

    Args:
        entities (Optional[list[str]]): List of entity values to search for relationships
        entities_types (Optional[list[str]]): List of entity types to filter relationships by
        relationships (Optional[list[str]]): List of relationship types to search for
        limit (int): Maximum number of relationships to return (default: 20)

    Returns:
        list: List of relationships matching the search criteria. If entities_types is provided,
                returns filtered relationships where EntityA or EntityB type matches entities_types.
                Returns empty list if no parameters are provided or no relationships found.
    """
    entities_types = argToList(args.pop("related_entity_types", []))
    entities = argToList(args.get("entities", []))
    relationships = argToList(args.get("relationships", []))
    verbose = args.get("verbose", "false")
    revoked = args.get("revoked", "false")
    limit = int(args.get("limit", "20"))

    if entities_types and entities:
        filtered_relationships = filter_relationships_by_entity_types(
            entities, entities_types, relationships, limit, verbose, revoked
        )
        return filtered_relationships

    search_params = {
        "entities": entities,
        "entities_types": entities_types,
        "relationships": relationships,
        "limit": limit,
        "verbose": verbose,
        "revoked": revoked,
    }
    remove_nulls_from_dictionary(search_params)

    res = demisto.executeCommand("SearchIndicatorRelationships", search_params)
    if not res or len(res) == 0:
        return []

    data = demisto.get(res[0], "Contents") or {}
    return data.get("Relationships", []) or []


def main():
    try:
        args = demisto.args()

        relationships = get_relationships(args)
        hr = tableToMarkdown(
            "Relationships",
            relationships,
            headers=[
                "EntityA",
                "EntityAType",
                "EntityB",
                "EntityBType",
                "Relationship",
            ],
            headerTransform=lambda header: re.sub(r"\B([A-Z])", r" \1", header),
        )
        return_results(
            CommandResults(
                readable_output=hr,
                outputs_prefix="Relationships",
                outputs=relationships,
                outputs_key_field="ID",
            )
        )
    except Exception as ex:
        return_error(f"Failed to execute SearchIndicatorRelationshipsAgentix. Error: {str(ex)}")


if __name__ in ("__main__", "__builtin__", "builtins"):  # pragma: no cover
    main()