PANOStoCortexDataLakeMonitoring

Verify that all firewalls successfully pushed logs to the Cortex Data Lake for the last 12 hours. It's an easy way to do monitoring of the FW connection to CDL. You can use either a manual list of FW serials or a Panorama integration to get the list of equipment to monitor.

Type

python

Pack

PANOStoCDLMonitoring

Source

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401

# Cortex Data Lake PAN-OS log monitoring monitoring
# PRE_REQUISITE: Enable Cortex Data Lake integration and/or PAN-OS integration
# (if the automated creation of a FW list is expected)


def check_instance(all_instances: dict, integration_name: str, err_msg: str):
    """
    :param all_instances: All integration instances in the platform.
    :param integration_name: integration name
    :param err_msg: error message string
    """
    instance_found_active: bool = False
    for instance in all_instances:
        if (all_instances[instance]["brand"] == integration_name or instance == integration_name) and all_instances[instance][
            "state"
        ] == "active":
            instance_found_active = True
            break
    if not instance_found_active:
        raise Exception(err_msg)


def get_firewall_serials(pan_os_integration_instance_name: str) -> list:
    """
    :param pan_os_integration_instance_name: Name of the instance of the PAN-OS integration.
    :return: List of the FWs serials associated with this instance.
    """
    fw_monitor_list: list = []
    fw_query = {
        "type": "op",
        "cmd": "<show><devices><all></all></devices></show>",
        "raw-response": "true",
        "using": pan_os_integration_instance_name,
    }

    fw_query_result = demisto.executeCommand("pan-os", fw_query)
    if is_error(fw_query_result):
        raise Exception(
            f'Querying the instance: {pan_os_integration_instance_name} in Cortex Data Lake failed.'
            f' {fw_query_result[0]["Contents"]}'
        )

    if fw_query_result and isinstance(fw_query_result, list):
        for fw in fw_query_result[0]["Contents"]["response"]["result"]["devices"]["entry"]:
            fw_monitor_list.append(fw["serial"])
    else:
        raise Exception("Failed to retrieve Firewalls list from PAn-OS, try to specify manually a list of serials.")

    return fw_monitor_list


def query_cdl(fw_monitor_list: list) -> CommandResults:
    """
    :param fw_monitor_list: list of FWs serials
    :return: CommandResults object containing the serials which sent logs and that did not.
    """
    firewalls_with_logs = []
    firewalls_without_logs = []
    start_time = datetime.utcnow() - timedelta(hours=12)  # Looking for the last 12 hours of logs
    start_time = start_time.strftime("%Y-%m-%d %H:%M:%S")
    query = {"fields": "all", "time_range": "1 day", "limit": str(1), "start_time": start_time}
    for current_fw in fw_monitor_list:
        query["query"] = f"log_source_id = '{current_fw}'"

        query_result = demisto.executeCommand("cdl-query-traffic-logs", query)
        if is_error(query_result):
            raise Exception(f'Querying logs in Cortex Data Lake failed. {query_result[0]["Contents"]}')

        if query_result:
            # HR title comes from the XDR command
            if query_result[0]["HumanReadable"] == "### Logs traffic table\n**No entries.**\n":
                firewalls_without_logs.append(current_fw)
            else:
                firewalls_with_logs.append(current_fw)

    all_results = [{"FirewallsWithLogsSent": firewalls_with_logs, "FirewallsWithoutLogsSent": firewalls_without_logs}]

    return CommandResults(
        outputs_prefix="CDL.monitoring",
        outputs_key_field=["FirewallsWithLogsSent", "FirewallsWithoutLogsSent"],
        ignore_auto_extract=True,
        outputs=all_results,
        raw_response=query_result,
    )


def main():
    try:
        args = demisto.args()
        all_instances = demisto.getModules()  # Gather existing instances

        # Look for active Cortex Data Lake instance
        check_instance(all_instances, "Cortex Data Lake", "No active Cortex Data Lake integration found, please configure one.")

        # The Firewall list must be a comma-separated list of FW serials
        fw_monitor_list = argToList(args.get("fw_serials"))
        if not fw_monitor_list:  # List of FW to monitor is empty, get it from Panorama
            pan_os_integration_instance_name = args.get("pan_os_integration_instance_name")
            if not pan_os_integration_instance_name:
                raise Exception("A Firewall serial list or a PAN-OS integration instance name is needed.")
            # Look for active PAN-OS instance
            check_instance(
                all_instances,
                pan_os_integration_instance_name,
                f"Integration instance {pan_os_integration_instance_name} is not active or is not a PAN-OS integration.",
            )
            # Get FW serials
            fw_monitor_list = get_firewall_serials(pan_os_integration_instance_name)

        # Log the list of firewalls to be monitored
        demisto.debug(f"List of FW serials: {fw_monitor_list}")
        for current_fw in fw_monitor_list:
            if len(current_fw) not in (12, 15):  # VM serial are 15 digits and FW serial are 12 digits
                raise Exception(f"{current_fw} - incorrect Firewall serial number.")
        return_results(query_cdl(fw_monitor_list))

    except Exception as err:
        return_error(str(err), err)


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