How can I find the number of Panther alerts sent to a specific destination?

QUESTION

I'd like to determine how many alerts were sent to a specific destination (like a Slackbot integration or custom webhook). Is there a way to do this?

ANSWER

We don't have this feature built into the product yet, but there is a workaround using our API.

With our API, you can query all of the alerts generated within a specific time period, then use a script to filter the results based on the destination ID. Here's an end-to-end example which uses Python. Be sure to adjust the time window and set the appropriate destination ID on lines 68-70. Also note that this script assumes that your API credentials are saved in your local environment.

# pip install gql aiohttp

from datetime import datetime
from os import environ
from typing import List

from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport

transport = AIOHTTPTransport(
  url=environ.get('PANTHER_API_HOST'),
  headers={"X-API-Key": environ.get('PANTHER_API_TOKEN')}
)

client = Client(transport=transport, fetch_schema_from_transport=True)

def find_alerts(start: datetime, end: datetime) -> List[dict]:
    # `FindAlerts` is a nickname for the query. You can fully omit it.
    find_alerts = gql(
        """
        query FindAlerts($input: AlertsInput!) {
        alerts(input: $input) {
            edges {
            node {
                id
                deliveries {
                outputId
                }
            }
            }
            pageInfo {
            hasNextPage
            endCursor
            }
        }
        }
        """
    )

    # an accumulator that holds all alerts that we fetch all pages
    all_alerts = []
    # a helper to know when to exit the loop
    has_more = True
    # the pagination cursor
    cursor = None

    # Keep fetching pages until there are no more left
    while has_more:
        query_data = client.execute(
            find_alerts,
            variable_values={
                "input": {
                    "severities": ["HIGH", "MEDIUM", "LOW"],
                    "createdAtBefore": end.strftime('%Y-%m-%dT%H:%M:%SZ'),
                    "createdAtAfter": start.strftime('%Y-%m-%dT%H:%M:%SZ'),
                    "cursor": cursor
                }
            }
        )

        all_alerts.extend([edge["node"] for edge in query_data["alerts"]["edges"]])
        has_more = query_data["alerts"]["pageInfo"]["hasNextPage"]
        cursor = query_data["alerts"]["pageInfo"]["endCursor"]

    return all_alerts

if __name__ == "__main__":
    start = datetime(2023, 3, 1)
    end = datetime(2023, 4, 1)
    dest_id = 'YOUR_DESTINATION_ID'

    alerts = []
    for alert in find_alerts(start, end):
        for delivery in alert.get('deliveries', []):
            if delivery.get('outputId', '') == dest_id:
                alerts.append(alert)
    
    print(f"{len(alerts)} alerts were delivered to {dest_id}. These are the IDs:")
    for alert in alerts:
        print(alert.get('id'))