Cortex Attack Surface Management

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401
import urllib3
from typing import Any
from requests import Response  # Used to typing Response as a return from functions

# Disable insecure warnings
urllib3.disable_warnings()

DEFAULT_SEARCH_LIMIT = 100


class NotFoundError(Exception):
    """Exception raised when an error is encountered that does
    not have an error message, like with a waitress error"""


class ProcessingError(Exception):
    """Exception raised when a 500 error is returned from the API
    with a json body containing an error code and message"""


class Client(BaseClient):
    """
    Client class to interact with the service API.
    """

    def __init__(self, base_url, verify, proxy, headers):
        """
        Class initialization.
        """
        super().__init__(base_url, verify=verify, proxy=proxy, headers=headers)

    def list_remediation_rule_request(self, request_data: dict) -> dict[str, Any]:
        """Get a list of all your remediation rules using the 'xpanse_remediation_rules/rules/' endpoint.

        Args:
            request_data (dict): dict of parameters for API call.

        Returns:
            dict: dict containing list of external services.
        """

        response = self._http_request(
            "POST",
            "/xpanse_remediation_rules/rules/",
            json_data=request_data,
            error_handler=get_api_error,
        )

        return response

    def list_external_service_request(self, search_params: list[dict]) -> dict[str, Any]:
        """Get a list of all your external services using the '/assets/get_external_services/' endpoint.

        Args:
            search_params (list): list of search parameters to add to the API call body.

        Returns:
            dict: dict containing list of external services.
        """
        data = {"request_data": {"filters": search_params, "search_to": DEFAULT_SEARCH_LIMIT}}

        response = self._http_request(
            "POST",
            "/assets/get_external_services/",
            json_data=data,
            error_handler=get_api_error,
        )

        return response

    def get_external_service_request(self, service_id_list: list[str]) -> dict[str, Any]:
        """Get service details using the '/assets/get_external_service/' endpoint.

        Args:
            service_id_list (list): single service id in list format.

        Returns:
            dict: dict containing information on single external service.
        """
        data = {"request_data": {"service_id_list": service_id_list}}

        response = self._http_request(
            "POST",
            "/assets/get_external_service",
            json_data=data,
            error_handler=get_api_error,
        )

        return response

    def list_external_ip_address_range_request(self) -> dict[str, Any]:
        """Get a list of all your internet exposure IP ranges using the '/assets/get_external_ip_address_ranges/' endpoint.

        Returns:
            dict: dict containing list of external ip address ranges.
        """
        data = {"request_data": {"search_to": DEFAULT_SEARCH_LIMIT}}

        response = self._http_request(
            "POST",
            "/assets/get_external_ip_address_ranges/",
            json_data=data,
            error_handler=get_api_error,
        )

        return response

    def get_external_ip_address_range_request(self, range_id_list: list[str]) -> dict[str, Any]:
        """Get external IP address range details using the '/assets/get_external_ip_address_range/' endpoint.

        Args:
            range_id_list (list): single range id in list format.

        Returns:
            dict: dict containing information on external ip address range.
        """
        data = {"request_data": {"range_id_list": range_id_list}}

        response = self._http_request(
            "POST",
            "/assets/get_external_ip_address_range/",
            json_data=data,
            error_handler=get_api_error,
        )

        return response

    def get_attack_surface_rule_request(self, search_params: list[dict]) -> dict[str, Any]:
        """Get Attack Surface Rule details for an attack surface rule id using the '/get_attack_surface_rules/' endpoint.

        Args:
            attack_surface_rule_id (str): Coma separated attack surface rule ids.

        Returns:
            dict: dict containing information about Attack surface rule.
        """
        data = {"request_data": {"filters": search_params}}
        response = self._http_request("POST", "/get_attack_surface_rules/", json_data=data, error_handler=get_api_error)
        return response

    def list_asset_internet_exposure_request(self, search_params: list[dict]) -> dict[str, Any]:
        """Get a list of all your internet exposure assets using the '/assets/get_assets_internet_exposure/' endpoint.

        Args:
            search_params (list): list of search parameters to add to the API call body.

        Returns:
            dict: dict containing list of internet exposure assets.
        """
        data = {"request_data": {"filters": search_params, "search_to": DEFAULT_SEARCH_LIMIT}}

        response = self._http_request(
            "POST",
            "/assets/get_assets_internet_exposure/",
            json_data=data,
            error_handler=get_api_error,
        )

        return response

    def get_asset_internet_exposure_request(self, asm_id_list: list[str]) -> dict[str, Any]:
        """Get internet exposure asset details using the '/assets/get_asset_internet_exposure/' endpoint.

        Args:
            asm_id_list (list): single attack surface management id in list format.

        Returns:
            dict: dict containing information on an internet exposure asset.
        """
        data = {"request_data": {"asm_id_list": asm_id_list}}

        response = self._http_request(
            "POST",
            "/assets/get_asset_internet_exposure/",
            json_data=data,
            error_handler=get_api_error,
        )

        return response

    def start_remediation_confirmation_scan(
        self, alert_internal_id: int, service_id: str, attack_surface_rule_id: str
    ) -> Response:
        """Retrieves ID of active (running) scan if it already exists for the given service; otherwise, creates new a scan.

        Args:
            alert_internal_id (str): _description_
            service_id (str): _description_
            attack_surface_rule_id (str): _description_

        Returns:
            Dict[str, Any]: dictionary containing response information that includes a scan ID.
        """

        data = {
            "request_data": {
                "filters": [
                    {
                        "field": "attack_surface_rule_id",
                        "operator": "EQ",
                        "value": attack_surface_rule_id,
                    },
                    {
                        "field": "alert_internal_id",
                        "operator": "EQ",
                        "value": alert_internal_id,
                    },
                    {"field": "service_id", "operator": "EQ", "value": service_id},
                ]
            }
        }

        response = self._http_request(
            method="POST",
            url_suffix="remediation_confirmation_scanning/requests/get_or_create/",
            json_data=data,
            resp_type="response",
            error_handler=get_api_error,
        )

        return response

    def get_remediation_confirmation_scan_status(self, scan_id: str) -> Response:
        """Retrieves ID of active (running) scan if it already exists for the given service; otherwise, creates new a scan.

        Args:
            alert_internal_id (str): _description_
            service_id (str): _description_
            attack_surface_rule_id (str): _description_

        Returns:
            Dict[str, Any]: dictionary containing response information that includes a scan ID.
        """
        data = {"request_data": {"filters": [{"field": "id", "operator": "EQ", "value": scan_id}]}}

        response = self._http_request(
            method="POST",
            url_suffix="/remediation_confirmation_scanning/requests/get/",
            json_data=data,
            resp_type="response",
            error_handler=get_api_error,
        )

        return response


""" HELPER FUNCTIONS """


def append_search_param(search_params, field, operator, value):
    """
    Appends a search parameter to the given list of search parameters.

    Args:
        search_params (list): The list of search parameters to append to.
        field (str): The name of the field to search on.
        operator (str): The operator to use for the search (e.g. "eq", "contains", "in").
        value (any): The value to search for.

    Returns:
        None
    """

    search_params.append({"field": field, "operator": operator, "value": value})


def format_asm_id(formatted_response: list[dict]) -> list[dict]:
    """
    Takes the response from the asm-list-asset-internet-exposure command and converts `asm_id` key from list to str

    Args:
        formatted_response (list): response from asm-list-asset-internet-exposure command (json)

    Returns:
        list: list of dictionaries of parsed/formatted json object
    """

    if formatted_response:
        for entry in formatted_response:
            if entry.get("asm_ids"):
                entry["asm_ids"] = entry["asm_ids"][0]

    return formatted_response


def get_api_error(response: Response):
    """Raises a formatted error based on the response from the base_error file from the server.

    Args:
        response: Response object from an API endpoint.
    Raises:
        NotFoundError: Exception for when an API endpoint
            returns an error that does not have a corresponding error message.
        ProcessingError: Exception for when an API endpoint returns an error message.
    """
    error_code, error_message, extra_message, rcs_err_msg = "", "", "", ""
    try:
        json_response = response.json()
        error_code = json_response.get("reply", {}).get("err_code", {})
        error_message = json_response.get("reply", {}).get("err_msg", {})
        extra_message = json_response.get("reply", {}).get("err_extra", {})
        rcs_err_msg = f"{error_message}. {extra_message}"
        response.raise_for_status()
    except requests.exceptions.HTTPError as err:
        if "Forbidden" not in str(err):
            raise ProcessingError(f"{error_code} - Received error message: '{rcs_err_msg}'.")
        else:
            pass
    except (AttributeError, TypeError) as err:
        if "Forbidden" not in str(err):
            raise NotFoundError(f"{type(err).__name__} - {str(err)}")


""" COMMAND FUNCTIONS """


def list_remediation_rule_command(args: dict[str, Any], client: Client) -> CommandResults:
    """
    asm-list-remediation-rule command: Returns list of remediation path rules.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()``.
            ``args['asm_rule_id']`` A string representing the ASM Rule ID you want to get association
            remediation path rules for.
            ``args['sort_by_creation_time']`` optional - enum (asc,desc).

    Returns:
        CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
        that contains list of remediation path rules.
    """
    asm_rule_id = str(args.get("asm_rule_id"))
    sort_by_creation_time = args.get("sort_by_creation_time")

    # create list of search parameters or pass empty list.
    search_params = []
    if asm_rule_id:
        search_params.append({"field": "attack_surface_rule_id", "operator": "eq", "value": asm_rule_id})
    if sort_by_creation_time:
        request_data = {
            "request_data": {
                "filters": search_params,
                "search_from": 0,
                "search_to": DEFAULT_SEARCH_LIMIT,
                "sort": {"field": "created_at", "keyword": sort_by_creation_time},
            }
        }
    else:
        request_data = {
            "request_data": {
                "filters": search_params,
                "search_from": 0,
                "search_to": DEFAULT_SEARCH_LIMIT,
            }
        }

    response = client.list_remediation_rule_request(request_data)
    parsed = response.get("reply", {}).get("remediation_rules")
    markdown = tableToMarkdown(
        "Remediation Rules",
        parsed,
        removeNull=True,
        headerTransform=string_to_table_header,
    )
    command_results = CommandResults(
        outputs_prefix="ASM.RemediationRule",
        outputs_key_field="rule_id",
        outputs=parsed,
        raw_response=response,
        readable_output=markdown,
    )

    return command_results


def list_external_service_command(args: dict[str, Any], client: Client) -> CommandResults:
    """
    asm-list-external-service command: Returns list of external services.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()``.
            ``args['ip_address']`` IP Address to search on.
            ``args['domain']`` Domain to search on.
            ``args['is_active']`` If the service active or not.
            ``args['discovery_type']`` how service was discovered.

    Returns:
        CommandResults: A ``CommandResults`` object that is then passed to ``return_results``, that contains external
        services.
    """
    ip_address = args.get("ip_address")
    domain = args.get("domain")
    is_active = args.get("is_active")
    discovery_type = args.get("discovery_type")
    # create list of search parameters or pass empty list.
    search_params = []
    if ip_address:
        search_params.append({"field": "ip_address", "operator": "eq", "value": ip_address})
    if domain:
        search_params.append({"field": "domain", "operator": "contains", "value": domain})
    if is_active:
        search_params.append({"field": "is_active", "operator": "in", "value": [is_active]})
    if discovery_type:
        search_params.append({"field": "discovery_type", "operator": "in", "value": [discovery_type]})

    response = client.list_external_service_request(search_params)
    parsed = response.get("reply", {}).get("external_services")
    markdown = tableToMarkdown(
        "External Services",
        parsed,
        removeNull=True,
        headerTransform=string_to_table_header,
    )
    command_results = CommandResults(
        outputs_prefix="ASM.ExternalService",
        outputs_key_field="service_id",
        outputs=parsed,
        raw_response=response,
        readable_output=markdown,
    )

    return command_results


def get_external_service_command(args: dict[str, Any], client: Client) -> CommandResults:
    """
    asm-get-external-service command: Returns details of single external service.
    Returns error if more than one service_id was provided in comma separated format.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()``.
            ``args['service_id']`` A string representing the service ID you want to get details for.

    Returns:
        CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
        that contains external service information.
    """
    # assume that only one service_id was passed in or fail.
    service_id = str(args.get("service_id"))
    service_id_list = service_id.split(",")
    if len(service_id_list) > 1:
        raise ValueError("This command only supports one service_id at this time")

    response = client.get_external_service_request(service_id_list)
    parsed = response.get("reply", {}).get("details")
    markdown = tableToMarkdown(
        "External Service",
        parsed,
        removeNull=True,
        headerTransform=string_to_table_header,
    )
    command_results = CommandResults(
        outputs_prefix="ASM.ExternalService",
        outputs_key_field="service_id",
        outputs=parsed,
        raw_response=response,
        readable_output=markdown,
    )

    return command_results


def list_external_ip_address_range_command(args: dict[str, Any], client: Client) -> CommandResults:
    """
    asm-list-external-ip-address-range command: Returns list of external ip ranges.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()`` (not used in this function).

    Returns:
        CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
        that contains external IP address ranges.
    """
    response = client.list_external_ip_address_range_request()
    parsed = response.get("reply", {}).get("external_ip_address_ranges")
    markdown = tableToMarkdown(
        "External IP Address Ranges",
        parsed,
        removeNull=True,
        headerTransform=string_to_table_header,
    )
    command_results = CommandResults(
        outputs_prefix="ASM.ExternalIpAddressRange",
        outputs_key_field="range_id",
        outputs=parsed,
        raw_response=response,
        readable_output=markdown,
    )

    return command_results


def get_external_ip_address_range_command(args: dict[str, Any], client: Client) -> CommandResults:
    """
    asm-get-external-ip-address-range command: Returns details of single external ip range.
    Returns error if more than one range_id was provided in comma separated format.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()``.
            ``args['range_id']`` A string representing the range ID for which you want to get the details for.

    Returns:
        CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
        that contains external ip range information.
    """
    # assume that only one range_id was passed in or fail.
    range_id = str(args.get("range_id"))
    range_id_list = range_id.split(",")
    if len(range_id_list) > 1:
        raise ValueError("This command only supports one range_id at this time")

    response = client.get_external_ip_address_range_request(range_id_list)
    parsed = response.get("reply", {}).get("details")
    markdown = tableToMarkdown(
        "External IP Address Range",
        parsed,
        removeNull=True,
        headerTransform=string_to_table_header,
    )
    command_results = CommandResults(
        outputs_prefix="ASM.ExternalIpAddressRange",
        outputs_key_field="range_id",
        outputs=parsed,
        raw_response=response,
        readable_output=markdown,
    )

    return command_results


def get_attack_surface_rule_command(args: dict[str, Any], client: Client) -> CommandResults:
    """
    asm-get-attack-surface-rule command: Returns attack surface rule details.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()``

    Returns:
        CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
        that contains Remediation guidance information.
    """
    attack_surface_rule_id = args.get("attack_surface_rule_id")
    enabled_status = args.get("enabled_status")
    priority = args.get("priority")
    category = args.get("category")

    search_params = []
    if attack_surface_rule_id:
        search_params.append({"field": "attack_surface_rule_id", "operator": "in", "value": attack_surface_rule_id.split(",")})
    if enabled_status:
        search_params.append({"field": "enabled_status", "operator": "in", "value": enabled_status.split(",")})
    if priority:
        search_params.append({"field": "priority", "operator": "in", "value": priority.split(",")})
    if category:
        search_params.append({"field": "category", "operator": "in", "value": category.split(",")})

    response = client.get_attack_surface_rule_request(search_params)
    parsed = response.get("reply", {}).get("attack_surface_rules")
    command_results = CommandResults(
        outputs_prefix="ASM.AttackSurfaceRule",
        outputs_key_field="attack_surface_rule",
        outputs=parsed,
        raw_response=response,
    )

    return command_results


def list_asset_internet_exposure_command(args: dict[str, Any], client: Client) -> CommandResults:
    """
    asm-list-asset-internet-exposure command: Returns list of external internet exposures.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()``.
            ``args['ip_address']`` IP Address to search on.
            ``args['name']`` name of asset to search on.
            ``args['type']`` type of external service.
            ``args['has_active_external_services']`` if the internet exposure have an active external service.

    Returns:
        CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
        that contains external internet exposures.
    """
    ip_address = args.get("ip_address")
    name = args.get("name")
    asm_type = args.get("type")
    has_active_external_services = args.get("has_active_external_services")
    asm_id_list = args.get("asm_id_list")
    ipv6_address = args.get("ipv6_address")
    gcp_cloud_tags = args.get("gcp_cloud_tags")
    azure_cloud_tags = args.get("azure_cloud_tags")
    aws_cloud_tags = args.get("aws_cloud_tags")
    has_xdr_agent = args.get("has_xdr_agent")
    externally_detected_providers = args.get("externally_detected_providers")
    externally_inferred_cves = args.get("externally_inferred_cves")
    business_units_list = args.get("business_units_list")
    has_bu_overrides = args.get("has_bu_overrides")
    mac_addresses = args.get("mac_addresses")
    # create list of search parameters or pass empty list.
    search_params: list[dict[str, Any]] = []

    if ip_address:
        append_search_param(search_params, "ip_address", "eq", ip_address)

    if name:
        append_search_param(search_params, "name", "contains", name)

    if asm_type:
        append_search_param(search_params, "type", "in", [asm_type])

    if has_active_external_services:
        append_search_param(search_params, "has_active_external_services", "in", [has_active_external_services])

    if asm_id_list:
        append_search_param(search_params, "asm_id_list", "in", str(asm_id_list).split(","))

    if ipv6_address:
        append_search_param(search_params, "ipv6_address", "eq", str(ipv6_address))

    if aws_cloud_tags:
        append_search_param(search_params, "aws_cloud_tags", "in", str(aws_cloud_tags).split(","))

    if gcp_cloud_tags:
        append_search_param(search_params, "gcp_cloud_tags", "in", str(gcp_cloud_tags).split(","))

    if azure_cloud_tags:
        append_search_param(search_params, "azure_cloud_tags", "in", str(azure_cloud_tags).split(","))

    if has_xdr_agent:
        append_search_param(search_params, "has_xdr_agent", "in", str(has_xdr_agent).split(","))

    if externally_detected_providers:
        append_search_param(search_params, "externally_detected_providers", "contains", externally_detected_providers)

    if externally_inferred_cves:
        append_search_param(search_params, "externally_inferred_cves", "contains", str(externally_inferred_cves))

    if business_units_list:
        append_search_param(search_params, "business_units_list", "in", str(business_units_list).split(","))

    if has_bu_overrides:
        append_search_param(search_params, "has_bu_overrides", "eq", has_bu_overrides.lower() != "false")

    if mac_addresses:
        append_search_param(search_params, "mac_addresses", "contains", mac_addresses)

    response = client.list_asset_internet_exposure_request(search_params)
    formatted_response = response.get("reply", {}).get("assets_internet_exposure", [])
    parsed = format_asm_id(formatted_response)
    markdown = tableToMarkdown(
        "Asset Internet Exposures",
        parsed,
        removeNull=True,
        headerTransform=string_to_table_header,
    )
    command_results = CommandResults(
        outputs_prefix="ASM.AssetInternetExposure",
        outputs_key_field="asm_ids",
        outputs=parsed,
        raw_response=response,
        readable_output=markdown,
    )

    return command_results


def get_asset_internet_exposure_command(args: dict[str, Any], client: Client) -> CommandResults:
    """
    asm-get-asset-internet-exposure command: Returns details of single external internet exposure.
    Returns error if more than one asm_id was provided in comma separated format.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()``.
            ``args['asm_id']`` A string representing the asset ID for which you want to get the details for.

    Returns:
        CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
        that contains internet exposure information.
    """
    # assume that only one asm_id was passed in or fail.
    asm_id = str(args.get("asm_id"))
    asm_id_list = asm_id.split(",")
    if len(asm_id_list) > 1:
        raise ValueError("This command only supports one asm_id at this time")

    response = client.get_asset_internet_exposure_request(asm_id_list)
    parsed = response.get("reply", {}).get("details")
    markdown = tableToMarkdown(
        "Asset Internet Exposure",
        parsed,
        removeNull=True,
        headerTransform=string_to_table_header,
    )
    command_results = CommandResults(
        outputs_prefix="ASM.AssetInternetExposure",
        outputs_key_field="asm_ids",
        outputs=parsed,
        raw_response=response,
        readable_output=markdown,
    )

    return command_results


def start_remediation_confirmation_scan_command(args: dict[str, Any], client: Client) -> CommandResults:
    """
    asm-start-remediation-confirmation-scan command: Starts a new scan or gets an existing scan ID.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()`` (not used in this function).

    Returns:
        CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
        that contains the ID of the Remediation Confirmation Scan.
    """
    service_id = str(args.get("service_id"))
    attack_surface_rule_id = str(args.get("attack_surface_rule_id"))
    alert_internal_id = int(args.get("alert_internal_id", ""))
    if alert_internal_id < 0:
        raise ValueError(f"Expected a non-negative integer, but got {alert_internal_id}.")

    response = client.start_remediation_confirmation_scan(
        alert_internal_id=alert_internal_id,
        service_id=service_id,
        attack_surface_rule_id=attack_surface_rule_id,
    )

    demisto.debug(response.status_code)

    json_response = response.json()
    formatted_outputs = json_response.get("reply", {})

    if response.status_code == 201:
        formatted_outputs.update({"scan_creation_status": "created"})
    elif response.status_code == 200:
        formatted_outputs.update({"scan_creation_status": "existing"})

    markdown = tableToMarkdown(
        "Creation of Remediation Confirmation Scan",
        formatted_outputs,
        removeNull=True,
        headerTransform=string_to_table_header,
    )
    command_results = CommandResults(
        outputs_prefix="ASM.RemediationScan",
        outputs_key_field="",
        outputs=formatted_outputs,
        raw_response=json_response,
        readable_output=markdown,
    )
    return command_results


@polling_function(
    name=demisto.command(),
    interval=arg_to_number(demisto.args().get("interval_in_seconds", 600)),
    timeout=arg_to_number(demisto.args().get("timeout_in_seconds", 11000)),
    requires_polling_arg=False,  # This means it will always be default to poll, poll=true
)
def get_remediation_confirmation_scan_status_command(args: dict[str, Any], client: Client):
    """
    asm-get-remediation-confirmation-scan-status command: Polls for status of an existing remediation confirmation scan.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.
        args (dict): all command arguments, usually passed from ``demisto.args()`` (not used in this function).

    Returns:
        PollResult: A ``PollResult`` object that is then passed to ``return_results``,
        that contains the ID of the Remediation Confirmation Scan and a success or failure message.
    """
    scan_id = str(args.get("scan_id"))
    response = client.get_remediation_confirmation_scan_status(scan_id=scan_id)
    json_response = response.json()
    scan_status = json_response.get("reply").get("status")

    if scan_status == "IN_PROGRESS":
        return PollResult(
            response=None,
            partial_result=CommandResults(
                outputs_prefix="ASM.RemediationScan",
                outputs_key_field="scan_id",
                readable_output="Waiting for remediation confirmation scan to finish...",
            ),
            continue_to_poll=True,
            args_for_next_run={"scan_id": scan_id, **args},
        )
    elif scan_status == "SUCCESS":
        formatted_outputs = json_response.get("reply", {})
        markdown = tableToMarkdown(
            "Status of Remediation Confirmation Scan",
            formatted_outputs,
            removeNull=True,
            headerTransform=string_to_table_header,
        )
        command_results = CommandResults(
            outputs_prefix="ASM.RemediationScan",
            outputs_key_field="",
            outputs=formatted_outputs,
            raw_response=json_response,
            readable_output=markdown,
        )
        return PollResult(response=command_results, continue_to_poll=False)
    elif scan_status == "FAILED_TIMEOUT" or scan_status == "FAILED_ERROR":
        formatted_outputs = json_response.get("reply", {})
        command_results = CommandResults(
            outputs_prefix="ASM.RemediationScan",
            outputs_key_field="",
            outputs=formatted_outputs,
            raw_response=json_response,
            readable_output="The remediation confirmation scan timed out or failed.",
        )
        return PollResult(response=command_results, continue_to_poll=False)
    else:
        formatted_outputs = json_response.get("reply", {})
        command_results = CommandResults(
            outputs_prefix="ASM.RemediationScan",
            outputs_key_field="",
            outputs=formatted_outputs,
            raw_response=json_response,
            readable_output="The remediation confirmation scan timed out or failed.",
        )
        return PollResult(response=command_results, continue_to_poll=False)


def test_module(client: Client) -> None:
    """
    Tests API connectivity and authentication'
    When 'ok' is returned it indicates the integration works like it is supposed to and connection to the service is
    successful.
    Raises exceptions if something goes wrong.

    Args:
        client (Client): CortexAttackSurfaceManagment client to use.

    Returns:
        str: 'ok' if test passed, anything else will raise an exception and will fail the test.
    """
    try:
        client.list_external_service_request([])
    except DemistoException as e:
        if "Forbidden" in str(e):
            raise DemistoException("Authorization Error: make sure API Key is correctly set")
        else:
            raise e
    return_results("ok")


def main() -> None:
    """
    main function
    """
    params: dict[str, Any] = demisto.params()
    args: dict[str, Any] = demisto.args()

    command = demisto.command()
    demisto.debug(f"Command being called is {command}")

    try:
        creds = params.get("credentials", {})
        api = creds.get("password", "")
        auth_id = creds.get("identifier", "")
        headers = {
            "Authorization": f"{api}",
            "x-xdr-auth-id": f"{auth_id}",
            "Content-Type": "application/json",
        }

        proxy = params.get("proxy", False)
        handle_proxy()
        verify_certificate = not params.get("insecure", False)

        url_suffix = "/public_api/v1"
        url = params.get("url", "")
        add_sensitive_log_strs(api)
        base_url = urljoin(url, url_suffix)
        client = Client(base_url=base_url, verify=verify_certificate, headers=headers, proxy=proxy)

        commands = {
            "asm-list-external-service": list_external_service_command,
            "asm-get-external-service": get_external_service_command,
            "asm-list-external-ip-address-range": list_external_ip_address_range_command,
            "asm-get-external-ip-address-range": get_external_ip_address_range_command,
            "asm-get-attack-surface-rule": get_attack_surface_rule_command,
            "asm-list-asset-internet-exposure": list_asset_internet_exposure_command,
            "asm-get-asset-internet-exposure": get_asset_internet_exposure_command,
            "asm-list-remediation-rule": list_remediation_rule_command,
            "asm-start-remediation-confirmation-scan": start_remediation_confirmation_scan_command,
            "asm-get-remediation-confirmation-scan-status": get_remediation_confirmation_scan_status_command,
        }

        if command == "test-module":
            test_module(client)
        elif command in commands:
            return_results(commands[command](args, client))
        else:
            raise NotImplementedError(f"{command} command is not implemented.")

    except Exception as e:
        return_error(f"Failed to execute {command} command.\nError:\n{str(e)}")


""" ENTRY POINT """

if __name__ in ["__main__", "builtin", "builtins"]:
    main()