Cyberint Takedowns

from typing import Any
import urllib3
from CommonServerPython import *

urllib3.disable_warnings()

allowed_reason_values_str = (
    "phishing, brand_abuse, impersonating_application, unofficial_application_distribution, "
    "malicious_content, social_media_impersonation, social_media_employee_impersonation, "
    "fake_job_post, sensitive_file_on_antivirus_repository, instant_messaging_impersonation, other"
)


class Client(BaseClient):
    """
    Client to use in the Cyberint Takedown integration.
    """

    def __init__(
        self,
        base_url: str,
        access_token: str,
        verify: bool = False,
        proxy: bool = False,
    ):
        params = demisto.params()
        self._cookies = {"access_token": access_token}
        self._headers = {
            "X-Integration-Type": "XSOAR",
            "X-Integration-Instance-Name": demisto.integrationInstance(),
            "X-Integration-Instance-Id": "",
            "X-Integration-Customer-Name": params.get("client_name", ""),
            "X-Integration-Version": str(get_pack_version()),
        }
        super().__init__(base_url=base_url, verify=verify, proxy=proxy, headers=self._headers)

    @logger
    def submit_takedown_request(
        self,
        customer: str | None,
        reason: str | None,
        url: str | None,
        brand: str | None,
        original_url: str | None,
        alert_id: int | None,
        note: str | None,
    ) -> dict:
        """
        Retrieve a list of alerts according to parameters.

        Args:
            url (str): Index of page to return.

        Returns:
            response (Response): API response from Cyberint.
        """
        body = {
            "customer": customer,
            "reason": reason,
            "url": url,
            "brand": brand,
            "original_url": original_url,
            "alert_id": alert_id,
            "note": note,
        }
        body = remove_empty_elements(body)
        response = self._http_request(method="POST", json_data=body, cookies=self._cookies, url_suffix="takedown/api/v1/submit")
        return response

    @logger
    def retrieve_takedown_requests(
        self,
        customer_id: str | None = None,
        action: str | None = None,
        alert_id: int | None = None,
        alert_ref_id: str | None = None,
        blocked_date: str | None = None,
        sent_date: str | None = None,
        brand: str | None = None,
        created_date: str | None = None,
        customer: str | None = None,
        domain: str | None = None,
        email_ticket_id: str | None = None,
        hostname: str | None = None,
        id: str | None = None,
        last_action_date: str | None = None,
        last_email_date: str | None = None,
        last_monitored_date: str | None = None,
        last_seen_date: str | None = None,
        last_submit_date: str | None = None,
        metadata_date: str | None = None,
        reason: str | None = None,
        requested_by: str | None = None,
        status: str | None = None,
        url: str | None = None,
    ) -> dict:
        """
        Retrieve a list of alerts according to parameters.

        Args:
            Various filter parameters.

        Returns:
            response (dict): API response from Cyberint.
        """
        filters = {
            "actions": {"action": action} if action else None,
            "alert_id": alert_id,
            "alert_ref_id": alert_ref_id,
            "blocklist_requests": {"blocked_date": blocked_date, "sent_date": sent_date} if blocked_date or sent_date else None,
            "brand": brand,
            "created_date": created_date,
            "customer": customer,
            "domain": domain,
            "email_ticket_id": email_ticket_id,
            "hostname": hostname,
            "id": id,
            "last_action_date": last_action_date,
            "last_email_date": last_email_date,
            "last_monitored_date": last_monitored_date,
            "last_seen_date": last_seen_date,
            "last_submit_date": last_submit_date,
            "metadata": {"date": metadata_date} if metadata_date else None,
            "reason": reason,
            "requested_by": requested_by,
            "status": status,
            "url": url,
        }

        body = {
            "customer_id": customer_id,
            "filters": remove_empty_elements(filters),
        }

        response = self._http_request(method="POST", json_data=body, cookies=self._cookies, url_suffix="takedown/api/v1/request")
        return response


def test_module(client):
    """Test module for Cyberint integration."""
    try:
        client.retrieve_takedown_requests(customer_id="Cyberint", url="https://cyberint.com")
        return "ok"
    except DemistoException as exc:
        error_msg = str(exc)

        if "[403] - Forbidden" in error_msg:
            return "ok"
        elif "[401] - Unauthorized" in error_msg:
            raise DemistoException("Authentication failed: Invalid API token")
        else:
            # Re-raise the original exception for all other cases
            raise exc


def submit_takedown_request_command(
    client: Client,
    args: dict[str, Any],
) -> CommandResults:
    """
    Submits a takedown request and formats the output.

    Args:
        client: Cyberint API Client.
        args: Command arguments.

    Returns:
        CommandResults: Formatted takedown request.
    """
    # Define allowed reason values
    allowed_reasons = [
        "phishing",
        "brand_abuse",
        "impersonating_application",
        "unofficial_application_distribution",
        "malicious_content",
        "social_media_impersonation",
        "social_media_employee_impersonation",
        "fake_job_post",
        "sensitive_file_on_antivirus_repository",
        "instant_messaging_impersonation",
        "other",
    ]

    # Validate reason value
    reason = args.get("reason")
    if reason:
        reason = reason.lower()
        if reason and reason not in allowed_reasons:
            errormsg = f"Invalid reason provided: {reason}. Allowed values are: {allowed_reason_values_str}"
            demisto.error(errormsg)
            raise DemistoException(errormsg)

    response = client.submit_takedown_request(
        customer=args.get("customer"),
        reason=reason,
        url=args.get("url"),
        brand=args.get("brand"),
        original_url=args.get("original_url"),
        alert_id=arg_to_number(args.get("alert_id"), arg_name="alert_id"),
        note=args.get("note"),
    )

    data = response.get("data", {})
    takedown_request = data.get("takedown_request", {})

    formatted_request = [
        {
            "reason": takedown_request.get("reason"),
            "url": takedown_request.get("url"),
            "original_url": takedown_request.get("original_url"),
            "customer": takedown_request.get("customer"),
            "status": takedown_request.get("status"),
            "brand": takedown_request.get("brand"),
            "alert_ref_id": takedown_request.get("alert_ref_id"),
            "alert_id": takedown_request.get("alert_id"),
            "hosting_providers": takedown_request.get("hosting_providers"),
            "name_servers": takedown_request.get("name_servers"),
            "escalation_actions": takedown_request.get("escalation_actions"),
            "last_escalation_date": takedown_request.get("last_escalation_date"),
            "last_status_change_date": takedown_request.get("last_status_change_date"),
            "last_seen_date": takedown_request.get("last_seen_date"),
            "created_date": takedown_request.get("created_date"),
            "status_reason": takedown_request.get("status_reason"),
            "id": takedown_request.get("id"),
        }
    ]

    human_readable = tableToMarkdown(
        "Takedown Request",
        formatted_request,
        headers=[
            "reason",
            "url",
            "original_url",
            "customer",
            "status",
            "brand",
            "alert_ref_id",
            "alert_id",
            "hosting_providers",
            "name_servers",
            "escalation_actions",
            "last_escalation_date",
            "last_status_change_date",
            "last_seen_date",
            "created_date",
            "status_reason",
            "id",
        ],
        headerTransform=takedown_response_header_transformer,
        removeNull=True,
    )

    return CommandResults(
        readable_output=human_readable,
        outputs_prefix="Cyberint.takedowns_submit",
        outputs_key_field="id",
        raw_response=takedown_request,
        outputs=takedown_request,
    )


def retrieve_takedown_requests_command(
    client: Client,
    args: dict[str, Any],
) -> CommandResults:
    """
    Retrieves takedown requests and formats the output.

    Args:
        client: Cyberint API Client.
        args: Command arguments.

    Returns:
        CommandResults: Formatted takedown requests.
    """
    # Define allowed reason values
    allowed_reasons = [
        "phishing",
        "brand_abuse",
        "impersonating_application",
        "unofficial_application_distribution",
        "malicious_content",
        "social_media_impersonation",
        "social_media_employee_impersonation",
        "fake_job_post",
        "sensitive_file_on_antivirus_repository",
        "instant_messaging_impersonation",
        "other",
    ]

    # Validate reason value
    reason = args.get("reason")
    if reason:
        reason = reason.lower()
        if reason and reason not in allowed_reasons:
            errormsg = f"Invalid reason provided: {reason}. Allowed values are: {allowed_reason_values_str}"
            demisto.error(errormsg)
            raise DemistoException(errormsg)

    response = client.retrieve_takedown_requests(**args)
    data = response.get("data", {})
    takedown_requests = data.get("takedown_requests", [])

    formatted_requests = [
        {
            "reason": reason,
            "url": req.get("url"),
            "original_url": req.get("original_url"),
            "customer": req.get("customer"),
            "status": req.get("status"),
            "brand": req.get("brand"),
            "alert_ref_id": req.get("alert_ref_id"),
            "alert_id": req.get("alert_id"),
            "hosting_providers": req.get("hosting_providers"),
            "name_servers": req.get("name_servers"),
            "escalation_actions": req.get("escalation_actions"),
            "last_escalation_date": req.get("last_escalation_date"),
            "last_status_change_date": req.get("last_status_change_date"),
            "last_seen_date": req.get("last_seen_date"),
            "created_date": req.get("created_date"),
            "status_reason": req.get("status_reason"),
            "id": req.get("id"),
        }
        for req in takedown_requests
    ]

    human_readable = tableToMarkdown(
        "Takedown Requests",
        formatted_requests,
        headers=[
            "reason",
            "url",
            "original_url",
            "customer",
            "status",
            "brand",
            "alert_ref_id",
            "alert_id",
            "hosting_providers",
            "name_servers",
            "escalation_actions",
            "last_escalation_date",
            "last_status_change_date",
            "last_seen_date",
            "created_date",
            "status_reason",
            "id",
        ],
        headerTransform=takedown_response_header_transformer,
        removeNull=False,
    )

    return CommandResults(
        readable_output=human_readable,
        outputs_prefix="Cyberint.takedowns_list",
        outputs_key_field="id",
        raw_response=takedown_requests,
        outputs=takedown_requests,
    )


@logger
def main():
    """
    PARSE AND VALIDATE INTEGRATION PARAMS
    """
    params = demisto.params()
    args = demisto.args()

    base_url = params.get("url")
    access_token = params.get("access_token").get("password")
    insecure = not params.get("insecure", False)
    proxy = params.get("proxy", False)

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

    try:
        client = Client(
            base_url=base_url,
            access_token=access_token,
            verify=insecure,
            proxy=proxy,
        )

        if command == "test-module":
            return_results(test_module(client))

        elif command == "cyberint-takedown-url":
            return_results(submit_takedown_request_command(client, args))

        elif command == "cyberint-retrieve-takedowns":
            return_results(retrieve_takedown_requests_command(client, args))

        else:
            raise NotImplementedError(f"Command {command} is not implemented.")

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


def takedown_response_header_transformer(header: str) -> str:
    """
    Returns a correct header.
    Args:
        header (Str): header.
    Returns:
        header (Str).
    """
    if header == "customer_id":
        return "Customer ID"
    if header == "actions":
        return "Actions"
    if header == "alert_id":
        return "Alert ID"
    if header == "alert_ref_id":
        return "Alert Ref ID"
    if header == "blocked_date":
        return "Blocklist Blocked Date"
    if header == "sent_date":
        return "Blocklist Sent Date"
    if header == "brand":
        return "Brand"
    if header == "created_date":
        return "Created Date"
    if header == "customer":
        return "Customer"
    if header == "domain":
        return "Domain"
    if header == "email_ticket_id":
        return "Email Ticket ID"
    if header == "hostname":
        return "Hostname"
    if header == "id":
        return "ID"
    if header == "last_action_date":
        return "Last Action Date"
    if header == "last_email_date":
        return "Last Email Date"
    if header == "last_monitored_date":
        return "Last Monitored Date"
    if header == "last_seen_date":
        return "Last Seen Date"
    if header == "last_submit_date":
        return "Last Submit Date"
    if header == "metadata":
        return "Metadata"
    if header == "reason":
        return "Reason"
    if header == "requested_by":
        return "Requested By"
    if header == "status":
        return "Status"
    if header == "url":
        return "URL"
    return string_to_table_header(header)


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