Microsoft Defender for Cloud Event Collector

import demistomock as demisto
import urllib3
from CommonServerPython import *
from MicrosoftApiModule import *  # noqa: E402

urllib3.disable_warnings()

""" CONSTANTS """

VENDOR = "microsoft"
PRODUCT = "defender_for_cloud"
API_VERSION = "2022-01-01"
DEFAULT_LIMIT = 50
DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
LOG_PREFIX = "[MicrosoftDefenderForCloudEventCollector]"

""" CLIENT CLASS """


class MsClient:
    """
    Microsoft Client enables authorized access to Azure Security Center.
    """

    def __init__(
        self,
        tenant_id,
        auth_id,
        enc_key,
        server,
        verify,
        proxy,
        self_deployed,
        subscription_id,
        ok_codes,
        certificate_thumbprint,
        private_key,
    ):
        base_url_with_subscription = f"{server}subscriptions/{subscription_id}/"
        self.ms_client = MicrosoftClient(
            tenant_id=tenant_id,
            auth_id=auth_id,
            enc_key=enc_key,
            base_url=base_url_with_subscription,
            verify=verify,
            proxy=proxy,
            self_deployed=self_deployed,
            ok_codes=ok_codes,
            scope="https://management.azure.com/.default",
            certificate_thumbprint=certificate_thumbprint,
            private_key=private_key,
            command_prefix="ms-defender-for-cloud",
        )
        self.server = server
        self.subscription_id = subscription_id

    def get_event_list_basic(self):
        cmd_url = "/providers/Microsoft.Security/alerts"
        params = {"api-version": API_VERSION}
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def get_event_list(self, last_run: dict) -> list:
        """Listing alerts
        Args:
            last_run (str): last run
        Returns:
            tuple: (events, last_run)
        """
        cmd_url = "/providers/Microsoft.Security/alerts"
        params = {"api-version": API_VERSION}
        events: list = []
        demisto.debug(f"{LOG_PREFIX} Fetching events with last_run={last_run}")

        response = self.ms_client.http_request(
            method="GET",
            url_suffix=cmd_url,
            params=params,
            scope="https://management.azure.com/.default",
            resource=Resources.management_azure,
        )
        curr_events = response.get("value", [])
        demisto.debug(f"{LOG_PREFIX} API returned {len(curr_events)} events in first page.")

        curr_filtered_events = filter_out_previosly_digested_events(curr_events, last_run)
        if check_events_were_filtered_out(curr_events, curr_filtered_events):
            demisto.debug(f"{LOG_PREFIX} Events were filtered in first page, returning {len(curr_filtered_events)} events.")
            return curr_filtered_events

        events.extend(curr_filtered_events)

        page_count = 1
        while nextLink := response.get("nextLink", None):
            page_count += 1
            response = self.ms_client.http_request(method="GET", full_url=nextLink)
            curr_events = response.get("value", [])
            demisto.debug(f"{LOG_PREFIX} API returned {len(curr_events)} events in page {page_count}.")
            curr_filtered_events = filter_out_previosly_digested_events(curr_events, last_run)
            events.extend(curr_filtered_events)
            if check_events_were_filtered_out(curr_events, curr_filtered_events):
                demisto.debug(f"{LOG_PREFIX} Events were filtered in page {page_count}, stopping pagination.")
                break

        demisto.debug(f"{LOG_PREFIX} Total events collected: {len(events)} from {page_count} page(s).")
        return events


def get_alert_name(event: dict) -> str:
    """
    Extracts the alert name (GUID) from an event.
    The 'name' field contains just the alert GUID, while 'id' contains the full resource path
    which can differ by location (e.g., centralus vs eastus2) for the same alert.
    Falls back to 'id' if 'name' is not present.

    Args:
        event (dict): The event dictionary from the API

    Returns:
        str: The alert name/GUID, the id, or empty string if neither found
    """
    return event.get("name") or event.get("id", "") or ""


def filter_out_previosly_digested_events(events: list, last_run: dict) -> list:
    """
    Args:
        - events (list): The events list response from the API
        - last_run (dict): The last run dict
    Returns:
        (list): A list with all the duplicates from lastrun filtered out
    """
    if not last_run:
        demisto.debug(f"{LOG_PREFIX} No last_run provided, returning all {len(events)} events without filtering.")
        return events

    last_run_time = last_run.get("last_run", "")
    dup_names = last_run.get("dup_digested_time_id", [])
    demisto.debug(f"{LOG_PREFIX} Filtering events with last_run_time={last_run_time}, dup_names count={len(dup_names)}")

    filtered_events = [
        event
        for event in events
        if event.get("properties", {}).get("startTimeUtc", "") > last_run_time
        or (event.get("properties", {}).get("startTimeUtc", "") == last_run_time and get_alert_name(event) not in dup_names)
    ]

    demisto.debug(f"{LOG_PREFIX} Filtered {len(events)} events down to {len(filtered_events)} events.")
    return filtered_events


def check_events_were_filtered_out(events: list, filtered_events: list) -> bool:
    """
    Args:
        - events (list): The events list response from the API
        - filtered_events (list): The filtered event list (no dups)
    Returns:
        (bool): Whether events were filtered out
    """
    return len(events) > len(filtered_events)


def test_module(client: MsClient):  # pragma: no cover
    """
    Performs basic GET request to check if the API is reachable and authentication is successful.
    Returns ok if successful.
    """
    events_res = client.get_event_list_basic()
    if "value" in events_res:
        demisto.results("ok")


def get_events(client: MsClient, last_run: dict, limit: int) -> tuple:
    """
     Args:
        client (MsClient): The microsoft client.
        last_run (dict): The last run object.
    Returns:
        (tuple): (events_list, CommandResults)
    """
    demisto.debug(f"{LOG_PREFIX} get_events called with last_run={last_run}, limit={limit}")
    events_list = client.get_event_list(last_run)
    demisto.debug(f"{LOG_PREFIX} get_events received {len(events_list)} events from client")

    if limit and len(events_list) > limit:
        demisto.debug(f"{LOG_PREFIX} Limiting events from {len(events_list)} to {limit}")
        events_list = events_list[-limit:]

    outputs = []
    for alert in events_list:
        if properties := alert.get("properties"):
            outputs.append(
                {
                    "DisplayName": properties.get("alertDisplayName"),
                    "CompromisedEntity": properties.get("compromisedEntity"),
                    "DetectedTime": properties.get("timeGeneratedUtc"),
                    "ReportedSeverity": properties.get("severity"),
                    "Description": properties.get("description"),
                    "ID": alert.get("name"),
                }
            )

    md = tableToMarkdown(
        "Microsft Defender For Cloud - List Alerts",
        outputs,
        [
            "DisplayName",
            "CompromisedEntity",
            "DetectedTime",
            "ReportedSeverity",
            "Description",
            "ID",
        ],
        removeNull=True,
    )
    cr = CommandResults(
        outputs_prefix="MicrosoftDefenderForCloud.Alert", outputs=outputs, readable_output=md, raw_response=events_list
    )
    return events_list, cr


def find_next_run(events_list: list, last_run: dict) -> dict:
    """
    Args:
        events (list): The list of events from the API call
        last_run (str): The prevision last run
    Returns:
        The next run for the next fetch-event command.
    """
    demisto.debug(f"{LOG_PREFIX} find_next_run called with {len(events_list)} events, last_run={last_run}")
    if not events_list:
        demisto.debug(f"{LOG_PREFIX} No events in list, returning previous last_run")
        return last_run

    # Sort events by startTimeUtc descending to ensure we get the newest event's timestamp
    # This prevents issues when the API returns events in an unexpected order
    sorted_events = sorted(events_list, key=lambda x: x.get("properties", {}).get("startTimeUtc", ""), reverse=True)

    next_run = sorted_events[0].get("properties", {}).get("startTimeUtc", "")
    # Use alert name (GUID) instead of full id for deduplication
    # This prevents duplicates when the same alert appears from different locations (e.g., centralus vs eastus2)
    names_same_next_run_list = [
        get_alert_name(event) for event in sorted_events if event.get("properties", {}).get("startTimeUtc") == next_run
    ]
    demisto.info(f"{LOG_PREFIX} Setting next run time to {next_run}, alert names with same time are {names_same_next_run_list}.")
    return {"last_run": next_run, "dup_digested_time_id": names_same_next_run_list}


def fetch_events(client: MsClient, last_run: dict) -> list:
    """
    Args:
        client (MsClient): The microsoft client.
        last_run (str): The last run.
    Returns:
        events: The list of fetched events.
    """
    events = client.get_event_list(last_run)
    demisto.info(f"{LOG_PREFIX} Fetched {len(events)} events.")
    return events


def add_time_key_to_events(events: list) -> list:
    """
    Adds the _time key to the events.
    Args:
        events: list, the events to add the time key to.
    Returns:
        list: The events with the _time key.
    """
    demisto.debug(f"{LOG_PREFIX} Adding _time key to {len(events)} events")
    for event in events:
        time_generated = event.get("properties", {}).get("timeGeneratedUtc")
        event["_time"] = time_generated or event.get("properties", {}).get("startTimeUtc", "")
    return events


def handle_last_run(first_fetch_time: str) -> dict:
    """
    In the first run init with the first_fetch_time after use the previos last_run.
    Args:
        - first_fetch_time(str): The first fetch time arg
    Returns:
        The last run object.
    """
    existing_last_run = demisto.getLastRun()
    if existing_last_run:
        demisto.debug(f"{LOG_PREFIX} Using existing last_run from previous execution: {existing_last_run}")
        return existing_last_run
    demisto.debug(f"{LOG_PREFIX} No existing last_run found, initializing with first_fetch_time: {first_fetch_time}")
    return {
        "last_run": first_fetch_time,
        "dup_digested_time_id": [],
    }


def main() -> None:  # pragma: no cover
    """
    main function, parses params and runs command functions
    """
    args = demisto.args()
    command = demisto.command()
    params: dict = demisto.params()
    server = params.get("server_url", "").rstrip("/") + "/"
    tenant = params.get("tenant_id", {}).get("password")
    client_id = params.get("client_id", {}).get("password")
    enc_key = params.get("enc_key", {}).get("password")
    use_ssl = not params.get("unsecure", False)
    proxy = params.get("proxy", False)
    subscription_id = params.get("sub_id", {}).get("password")
    ok_codes = (200, 201, 202, 204)
    certificate_thumbprint = params.get("certificate_thumbprint", {}).get("password")
    private_key = params.get("private_key")
    first_fetch_time = params.get("first_fetch", "3 days")

    if not enc_key and not (certificate_thumbprint and private_key):
        raise DemistoException(
            "Key or Certificate Thumbprint and Private Key must be provided."
            "For further information see "
            "https://xsoar.pan.dev/docs/reference/articles/microsoft-integrations---authentication"
        )

    first_fetch_time: datetime = arg_to_datetime(  # type: ignore
        arg=first_fetch_time,
        arg_name="First fetch time",
        required=True,
    )

    first_fetch_time_strftime = first_fetch_time.strftime(DATE_FORMAT)

    demisto.debug(f"{LOG_PREFIX} Command being called is {command}")
    try:
        client = MsClient(
            tenant_id=tenant,
            auth_id=client_id,
            enc_key=enc_key,
            proxy=proxy,
            server=server,
            verify=use_ssl,
            self_deployed=True,
            subscription_id=subscription_id,
            ok_codes=ok_codes,
            certificate_thumbprint=certificate_thumbprint,
            private_key=private_key,
        )

        last_run = handle_last_run(first_fetch_time_strftime)

        if command == "test-module":
            # This is the call made when pressing the integration Test button.
            test_module(client)

        elif command == "ms-defender-for-cloud-auth-reset":
            return_results(reset_auth())

        elif command in ("ms-defender-for-cloud-get-events", "fetch-events"):
            if command == "ms-defender-for-cloud-get-events":
                limit = arg_to_number(args.get("limit", DEFAULT_LIMIT))
                should_push_events = argToBoolean(args.pop("should_push_events", False))
                events, results = get_events(client, last_run, limit=limit)  # type: ignore
                return_results(results)

            else:  # command == 'fetch-events':
                should_push_events = True
                events = fetch_events(client=client, last_run=last_run)

            demisto.debug(f"{LOG_PREFIX} Before adding _time key, events count: {len(events)}")
            events = add_time_key_to_events(events)

            if should_push_events:
                next_run = find_next_run(events, last_run)
                demisto.debug(f"{LOG_PREFIX} Sending {len(events)} events to XSIAM")
                send_events_to_xsiam(events, vendor=VENDOR, product=PRODUCT)
                # saves next_run only after events are successfully sent to prevent data loss
                demisto.debug(f"{LOG_PREFIX} Setting next_run to: {next_run}")
                demisto.setLastRun(next_run)

    # Log exceptions and return errors
    except Exception as e:
        return_error(f"Failed to execute {command} command.\nError:\n{e!s}")


""" ENTRY POINT """

if __name__ in ("__main__", "__builtin__", "builtins"):
    main()