ThreatExchange v2

"""
An integration module for the ThreatExchange V2 API.
API Documentation:
    https://developers.facebook.com/docs/threat-exchange/reference/apis
"""

import collections

import urllib3
from CommonServerPython import *  # noqa # pylint: disable=unused-wildcard-import

from CommonServerUserPython import *  # noqa

# Disable insecure warnings #
urllib3.disable_warnings()

DEFAULT_LIMIT = 20
COMMAND_PREFIX = "threatexchange"
VENDOR_NAME = "ThreatExchange v2"
CONTEXT_PREFIX = "ThreatExchange"
THREAT_DESCRIPTORS_SUFFIX = "threat_descriptors"
MALWARE_ANALYSES_SUFFIX = "malware_analyses"
THREAT_TAGS_SUFFIX = "threat_tags"
TAGGED_OBJECTS_SUFFIX = "tagged_objects"
THREAT_EXCHANGE_MEMBERS_SUFFIX = "threat_exchange_members"
TIMEOUT_FOR_LIST_CALLS = 30
DEFAULT_DESCRIPTION_FOR_MALICIOUS_INDICATOR = "Indicator was classified as malicious by more than {}% of detection engines."


class ThreatExchangeV2Status:
    UNKNOWN = "UNKNOWN"
    NON_MALICIOUS = "NON_MALICIOUS"
    SUSPICIOUS = "SUSPICIOUS"
    MALICIOUS = "MALICIOUS"


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

    This Client implements API calls, and does not contain any XSOAR logic.
    Should only do requests and return data.
    It inherits from BaseClient defined in CommonServer Python.
    Most calls use _http_request() that handles proxy, SSL verification, etc.
    For this implementation, no special attributes defined
    """

    def __init__(self, base_url, access_token, verify=True, proxy=False):
        super().__init__(base_url, verify, proxy)
        self.access_token = access_token

    def ip(self, ip: str, since: Optional[int], until: Optional[int], limit: Optional[int] = DEFAULT_LIMIT) -> Dict:
        """
        See Also:
            https://developers.facebook.com/docs/threat-exchange/reference/apis/threat-descriptors
        Args:
            ip: ip address
            since: Returns malware collected after a timestamp
            until: Returns malware collected before a timestamp
            limit: Defines the maximum size of a page of results. The maximum is 1,000

        Returns: The API call response
        """
        response = self._http_request(
            "GET",
            THREAT_DESCRIPTORS_SUFFIX,
            params={
                "access_token": self.access_token,
                "type": "IP_ADDRESS",
                "text": ip,
                "strict_text": True,
                "since": since,
                "until": until,
                "limit": limit,
            },
        )
        return response

    def file(self, file: str, since: Optional[int], until: Optional[int], limit: Optional[int] = DEFAULT_LIMIT) -> Dict:
        """
        See Also:
            https://developers.facebook.com/docs/threat-exchange/reference/apis/malware-analyses/v10.0
        Args:
            file: Hash of a file
            since: Returns malware collected after a timestamp
            until: Returns malware collected before a timestamp
            limit: Defines the maximum size of a page of results. The maximum is 1,000

        Returns: The API call response
        """

        response = self._http_request(
            "GET",
            MALWARE_ANALYSES_SUFFIX,
            params=assign_params(
                access_token=self.access_token, text=file, strict_text=True, since=since, until=until, limit=limit
            ),
        )
        return response

    def domain(
        self, domain: str, since: Optional[int], until: Optional[int], share_level: str, limit: Optional[int] = DEFAULT_LIMIT
    ) -> Dict:
        """
        See Also:
            https://developers.facebook.com/docs/threat-exchange/reference/apis/threat-descriptors
        Args:
            domain: Domain
            since: Returns malware collected after a timestamp
            until: Returns malware collected before a timestamp
            share_level: A designation of how the indicator may be shared, based on the US-CERT's Traffic Light Protocol.
            limit: Defines the maximum size of a page of results. The maximum is 1,000

        Returns: The API call response
        """

        response = self._http_request(
            "GET",
            THREAT_DESCRIPTORS_SUFFIX,
            params=assign_params(
                access_token=self.access_token,
                type="DOMAIN",
                text=domain,
                strict_text=True,
                since=since,
                until=until,
                limit=limit,
                share_level=share_level,
            ),
        )
        return response

    def url(
        self, url: str, since: Optional[int], until: Optional[int], share_level: str, limit: Optional[int] = DEFAULT_LIMIT
    ) -> Dict:
        """
        See Also:
            https://developers.facebook.com/docs/threat-exchange/reference/apis/threat-descriptors
        Args:
            url: URL
            since: Returns malware collected after a timestamp
            until: Returns malware collected before a timestamp
            share_level: A designation of how the indicator may be shared, based on the US-CERT's Traffic Light Protocol.
            limit: Defines the maximum size of a page of results. The maximum is 1,000

        Returns: The API call response
        """

        response = self._http_request(
            "GET",
            THREAT_DESCRIPTORS_SUFFIX,
            params=assign_params(
                access_token=self.access_token,
                type="URI",
                text=url,
                strict_text=True,
                since=since,
                until=until,
                limit=limit,
                share_level=share_level,
            ),
        )
        return response

    def members(self) -> Dict:
        """
        See Also:
            https://developers.facebook.com/docs/threat-exchange/reference/apis/threat-exchange-members/v10.0

        Returns: The API call response
        """

        response = self._http_request(
            "GET", THREAT_EXCHANGE_MEMBERS_SUFFIX, params={"access_token": self.access_token}, timeout=TIMEOUT_FOR_LIST_CALLS
        )
        return response

    def query(
        self,
        text: str,
        descriptor_type: str,
        since: Optional[int],
        until: Optional[int],
        limit: Optional[int] = DEFAULT_LIMIT,
        strict_text: Optional[bool] = False,
        before: Optional[str] = None,
        after: Optional[str] = None,
    ) -> Dict:
        """
        See Also:
            https://developers.facebook.com/docs/threat-exchange/reference/apis/threat-descriptors
        Args:
            text: Freeform text field with a value to search for
            descriptor_type: The type of descriptor to search for
            since: Returns malware collected after a timestamp
            until: Returns malware collected before a timestamp
            limit: Defines the maximum size of a page of results. The maximum is 1,000
            strict_text: When set to 'true', the API will not do approximate matching on the value in text
            before: Returns results collected before this cursor
            after: Returns results collected after this cursor

        Returns: The API call response
        """

        response = self._http_request(
            "GET",
            THREAT_DESCRIPTORS_SUFFIX,
            params=assign_params(
                access_token=self.access_token,
                type=descriptor_type,
                text=text,
                strict_text=strict_text,
                since=since,
                until=until,
                limit=limit,
                before=before,
                after=after,
            ),
            timeout=TIMEOUT_FOR_LIST_CALLS,
        )
        return response

    def tags_search(self, text: str, before: Optional[str] = None, after: Optional[str] = None) -> Dict:
        """
        See also:
            https://developers.facebook.com/docs/threat-exchange/reference/apis/threat-tags/v10.0
        Args:
            text: Freeform text field with a value to search for.
                  This value should describe a broader type or class of attack you are interested in.
            before: Returns results collected before this cursor
            after:  Returns results collected after this cursor

        Returns: The API call response
        """
        response = self._http_request(
            "GET",
            THREAT_TAGS_SUFFIX,
            params=assign_params(access_token=self.access_token, text=text, before=before, after=after),
            timeout=TIMEOUT_FOR_LIST_CALLS,
        )
        return response

    def tagged_objects_list(
        self,
        tag_id: str,
        tagged_since: Optional[int],
        tagged_until: Optional[int],
        before: Optional[str] = None,
        after: Optional[str] = None,
    ) -> Dict:
        """
        See also:
            https://developers.facebook.com/docs/threat-exchange/reference/apis/threattags/v10.0
        Args:
            tag_id: ThreatTag ID to get it's related tagged objects
            tagged_since:  Fetches all objects that have been tagged since this time (inclusive)
            tagged_until: Fetches all objects that have been tagged until this time (inclusive)
            before: Returns results collected before this cursor
            after:  Returns results collected after this cursor

        Returns: The API call response
        """
        url_suffix = f"{tag_id}/{TAGGED_OBJECTS_SUFFIX}"
        response = self._http_request(
            "GET",
            url_suffix,
            params=assign_params(
                access_token=self.access_token, tagged_since=tagged_since, tagged_until=tagged_until, before=before, after=after
            ),
            timeout=TIMEOUT_FOR_LIST_CALLS,
        )
        return response

    def object_get_by_id(self, object_id: str) -> Dict:
        """
        Gets ThreatExchange object by ID
        Args:
            object_id: ID of a ThreatExchange object

        Returns: The API call response
        """
        response = self._http_request("GET", object_id, params={"access_token": self.access_token})
        return response


def get_reputation_data_statuses(reputation_data: List) -> List[str]:
    """
    collects reported statuses of reputation data
    Args:
        reputation_data: returned data list of a certain reputation command

    Returns: a list of reported statuses
    """
    reputation_statuses = [status for data_entry in reputation_data if (status := data_entry.get("status"))]
    return reputation_statuses


def calculate_dbot_score(reputation_data: List, params: Dict[str, Any]) -> int:
    """
    Calculates the Dbot score of the given reputation command data, by the following logic:
    MALICIOUS > malicious threshold (50%) = Malicious
    MALICIOUS <= malicious threshold (50%) = Suspicious
    SUSPICIOUS > suspicious threshold (1) = Suspicious
    NON_MALICIOUS > non malicious threshold (50%) = Good
    else Unknown
    Args:
        reputation_data: returned data list of a certain reputation command
        params: parameters of the integration
    Returns: the calculated Dbot score
    """
    # get user's thresholds:
    malicious_threshold = arg_to_number(params.get("malicious_threshold", 50), arg_name="malicious_threshold")
    if malicious_threshold is None:
        malicious_threshold = 50
    suspicious_threshold = arg_to_number(params.get("suspicious_threshold", 1), arg_name="suspicious_threshold")
    if suspicious_threshold is None:
        suspicious_threshold = 1
    non_malicious_threshold = arg_to_number(params.get("non_malicious_threshold", 50), arg_name="non_malicious_threshold")
    if non_malicious_threshold is None:
        non_malicious_threshold = 50

    # collect and count reported statuses:
    reputation_statuses = get_reputation_data_statuses(reputation_data)
    num_of_statuses = len(reputation_statuses)
    occurrences = collections.Counter(reputation_statuses)

    # calculate Dbot score:
    num_of_malicious = occurrences.get(ThreatExchangeV2Status.MALICIOUS, 0)
    num_of_suspicious = occurrences.get(ThreatExchangeV2Status.SUSPICIOUS, 0)
    num_of_non_malicious = occurrences.get(ThreatExchangeV2Status.NON_MALICIOUS, 0)
    if num_of_statuses == 0:  # no reported statuses
        score = Common.DBotScore.NONE
    elif num_of_malicious >= 1:  # at least one malicious status was reported
        if ((num_of_malicious / num_of_statuses) * 100) > malicious_threshold:
            score = Common.DBotScore.BAD
        else:  # num_of_malicious <= malicious_threshold
            score = Common.DBotScore.SUSPICIOUS
    elif num_of_suspicious > suspicious_threshold:  # number of suspicious statuses is above threshold
        score = Common.DBotScore.SUSPICIOUS
    elif ((num_of_non_malicious / num_of_statuses) * 100) > non_malicious_threshold:
        # number of non malicious statuses is above threshold
        score = Common.DBotScore.GOOD
    else:  # there isn't enough information - Dbot score is defined as unknown
        score = Common.DBotScore.NONE

    return score


def calculate_engines(reputation_data: List) -> tuple[int, int]:
    """
    Calculates the number of engines that scanned the indicator, and how many of them are positive
     - i.e returned malicious status.
    Args:
        reputation_data: returned data list of a certain reputation command

    Returns: number of engines, number of positive engines
    """
    num_of_engines = len(reputation_data)
    reputation_statuses = get_reputation_data_statuses(reputation_data)
    occurrences = collections.Counter(reputation_statuses)
    num_of_positive_engines = occurrences.get(ThreatExchangeV2Status.MALICIOUS, 0)

    return num_of_engines, num_of_positive_engines


def flatten_outputs_paging(raw_response: Dict) -> Dict:
    """
    flatten the paging section of the raw_response - i.e removes 'cursors' key.
    Args:
        raw_response: response of an API call

    Returns: outputs dict

    """
    paging: Dict
    paging = raw_response.get("paging", {})
    outputs = raw_response.copy()
    cursor_before = paging.get("cursors", {}).get("before")
    cursor_after = paging.get("cursors", {}).get("after")
    outputs.pop("paging", None)
    outputs["paging"] = {
        "before": cursor_before,
        "after": cursor_after,
    }
    return outputs


def get_malicious_description(score: int, reputation_data: List[Dict], params: Dict[str, Any]) -> Optional[str]:
    """
    Gets the malicious description of certain indicator.
    If the indicator was classified as malicious, description is defined as default malicious description.
    If the indicator wasn't classified as malicious, description will be None (and won't be added to context).
    Args:
        score: calculated dbot score of the indicator
        reputation_data: returned data of a certain reputation command
        params: integration's parameters

    Returns: malicious description

    """
    malicious_description: Union[str, None]
    if score == Common.DBotScore.BAD:
        malicious_threshold = arg_to_number(params.get("malicious_threshold", 50))
        default_description = DEFAULT_DESCRIPTION_FOR_MALICIOUS_INDICATOR.format(malicious_threshold)
        malicious_description = default_description

    else:  # dbot-score isn't malicious
        malicious_description = None

    return malicious_description


def convert_string_to_epoch_time(date: Optional[str], arg_name: Optional[str] = None) -> Optional[int]:
    """
    Converts a string representing a date into epoch time format
    Args:
        date: date string
        arg_name: name of the date argument

    Returns: date in epoch time format (if an error occurred, or date in None returns None)
    """
    if date:
        if date.isdigit():  # date is an epoch time format string
            return int(date)

        date_obj = dateparser.parse(date)  # date is a string in a time format such as: iso 8601, free text, etc
        if isinstance(date_obj, datetime):
            epoch_time = date_obj.timestamp()
            return int(epoch_time)
        else:  # date was given in a wrong format
            if arg_name:
                raise ValueError(f'Invalid date: "{arg_name}"="{date}"')

    return None


def test_module(client: Client) -> str:
    """
    Tests API connectivity and authentication.

    Returning 'ok' indicates that the integration works like it is supposed to.
    Connection to the service is successful.
    Raises exceptions if something goes wrong.

    Args:
        client: client to use

    Returns: 'ok' if test passed, anything else will fail the test
    """

    client.ip(ip="8.8.8.8", since=None, until=None)
    return "ok"


def ip_command(client: Client, args: Dict[str, Any], params: Dict[str, Any]) -> List[CommandResults]:
    """
    Returns IP's reputation
    """
    ips = argToList(args.get("ip"))
    since = convert_string_to_epoch_time(args.get("since"), arg_name="since")
    until = convert_string_to_epoch_time(args.get("until"), arg_name="until")
    limit = arg_to_number(args.get("limit"), arg_name="limit")
    headers = argToList(args.get("headers"))
    reliability = params.get("feedReliability")
    results: List[CommandResults] = []

    for ip in ips:
        if not is_ip_valid(ip, accept_v6_ips=True):  # check IP's validity
            raise ValueError(f'IP "{ip}" is not valid')
        try:
            raw_response = client.ip(ip, since, until, limit)
        except Exception as exception:
            # If anything happens, handle like there are no results
            err_msg = f'Could not process IP: "{ip}"\n {exception!s}'
            demisto.debug(err_msg)
            raw_response = {}
        if data := raw_response.get("data"):
            score = calculate_dbot_score(reputation_data=data, params=params)
            num_of_engines, num_of_positive_engines = calculate_engines(reputation_data=data)
            malicious_description = get_malicious_description(score, data, params)
            dbot_score = Common.DBotScore(
                indicator=ip,
                indicator_type=DBotScoreType.IP,
                integration_name=VENDOR_NAME,
                score=score,
                reliability=reliability,
                malicious_description=malicious_description,
            )
            if not headers:
                headers = [
                    "description",
                    "owner",
                    "status",
                    "type",
                    "raw_indicator",
                    "share_level",
                    "confidence",
                    "severity",
                    "added_on",
                    "last_updated",
                    "review_status",
                    "id",
                ]
            readable_output = tableToMarkdown(f"{CONTEXT_PREFIX} Result for IP {ip}", data, headers=headers)
            ip_indicator = Common.IP(
                ip=ip, dbot_score=dbot_score, detection_engines=num_of_engines, positive_engines=num_of_positive_engines
            )

        else:  # no data
            dbot_score = Common.DBotScore(
                indicator=ip,
                indicator_type=DBotScoreType.IP,
                integration_name=VENDOR_NAME,
                score=Common.DBotScore.NONE,
                reliability=reliability,
            )
            readable_output = f"{CONTEXT_PREFIX} does not have details about IP: {ip} \n"
            ip_indicator = Common.IP(
                ip=ip,
                dbot_score=dbot_score,
            )

        result = CommandResults(
            outputs_prefix=f"{CONTEXT_PREFIX}.IP",
            outputs_key_field="id",
            outputs=data,
            indicator=ip_indicator,
            readable_output=readable_output,
            raw_response=raw_response,
        )
        results.append(result)

    return results


def file_command(client: Client, args: Dict[str, Any], params: Dict[str, Any]) -> List[CommandResults]:
    """
    Returns file's reputation
    """
    files = argToList(args.get("file"))
    since = convert_string_to_epoch_time(args.get("since"), arg_name="since")
    until = convert_string_to_epoch_time(args.get("until"), arg_name="until")
    limit = arg_to_number(args.get("limit"), arg_name="limit")
    headers = argToList(args.get("headers"))
    reliability = params.get("feedReliability")
    results: List[CommandResults] = []

    for file in files:
        if get_hash_type(file) not in ("sha256", "sha1", "md5"):  # check file's validity
            raise ValueError(f'Hash "{file}" is not of type SHA-256, SHA-1 or MD5')
        try:
            raw_response = client.file(file, since, until, limit)
        except Exception as exception:
            # If anything happens, handle like there are no results
            err_msg = f'Could not process file: "{file}"\n {exception!s}'
            demisto.debug(err_msg)
            raw_response = {}
        if data := raw_response.get("data"):
            score = calculate_dbot_score(reputation_data=data, params=params)
            malicious_description = get_malicious_description(score, data, params)
            dbot_score = Common.DBotScore(
                indicator=file,
                indicator_type=DBotScoreType.FILE,
                integration_name=VENDOR_NAME,
                score=score,
                reliability=reliability,
                malicious_description=malicious_description,
            )
            if not headers:
                headers = [
                    "description",
                    "status",
                    "share_level",
                    "added_on",
                    "review_status",
                    "id",
                    "password",
                    "sample_size",
                    "sample_size_compressed",
                    "sample_type",
                    "victim_count",
                    "md5",
                    "sha1",
                    "sha256",
                    "sha3_384",
                    "ssdeep",
                ]
            readable_output = tableToMarkdown(f"{CONTEXT_PREFIX} Result for file hash {file}", data, headers=headers)
            data_entry = data[0]
            file_indicator = Common.File(
                dbot_score=dbot_score,
                file_type=data_entry.get("sample_type"),
                size=data_entry.get("sample_size"),
                md5=data_entry.get("md5"),
                sha1=data_entry.get("sha1"),
                sha256=data_entry.get("sha256"),
                ssdeep=data_entry.get("ssdeep"),
                tags=data_entry.get("tags"),
            )

        else:  # no data
            dbot_score = Common.DBotScore(
                indicator=file,
                indicator_type=DBotScoreType.FILE,
                integration_name=VENDOR_NAME,
                score=Common.DBotScore.NONE,
                reliability=reliability,
            )
            readable_output = f"{CONTEXT_PREFIX} does not have details about file: {file} \n"
            file_indicator = Common.File(dbot_score=dbot_score)

        result = CommandResults(
            outputs_prefix=f"{CONTEXT_PREFIX}.File",
            outputs_key_field="id",
            outputs=data,
            indicator=file_indicator,
            readable_output=readable_output,
            raw_response=raw_response,
        )
        results.append(result)

    return results


def domain_command(client: Client, args: Dict[str, Any], params: Dict[str, Any]) -> List[CommandResults]:
    """
    Returns domain's reputation
    """
    domains = argToList(args.get("domain"))
    since = convert_string_to_epoch_time(args.get("since"), arg_name="since")
    until = convert_string_to_epoch_time(args.get("until"), arg_name="until")
    limit = arg_to_number(args.get("limit"), arg_name="limit")
    headers = argToList(args.get("headers"))
    reliability = params.get("feedReliability")
    share_level = args.get("share_level", params.get("share_level", "RED"))
    demisto.debug(f"Setting share level to {share_level}")
    results: List[CommandResults] = []

    for domain in domains:
        try:
            raw_response = client.domain(domain, since, until, share_level, limit)
        except Exception as exception:
            # If anything happens, handle like there are no results
            err_msg = f'Could not process domain: "{domain}"\n {exception!s}'
            demisto.debug(err_msg)
            raw_response = {}
            readable_output = f'Processing domain "{domain}" resulted in an exception. See logs for the exact error.'
            result = CommandResults(outputs={}, readable_output=readable_output, raw_response=raw_response)
            results.append(result)
            continue
        if data := raw_response.get("data"):
            score = calculate_dbot_score(reputation_data=data, params=params)
            num_of_engines, num_of_positive_engines = calculate_engines(reputation_data=data)
            malicious_description = get_malicious_description(score, data, params)
            dbot_score = Common.DBotScore(
                indicator=domain,
                indicator_type=DBotScoreType.DOMAIN,
                integration_name=VENDOR_NAME,
                score=score,
                reliability=reliability,
                malicious_description=malicious_description,
            )
            if not headers:
                headers = [
                    "description",
                    "owner",
                    "status",
                    "type",
                    "raw_indicator",
                    "share_level",
                    "confidence",
                    "severity",
                    "added_on",
                    "last_updated",
                    "review_status",
                    "id",
                ]
            readable_output = tableToMarkdown(f"{CONTEXT_PREFIX} Result for domain {domain}", data, headers=headers)
            domain_indicator = Common.Domain(
                domain=domain,
                dbot_score=dbot_score,
                detection_engines=num_of_engines,
                positive_detections=num_of_positive_engines,
            )
        else:  # no data
            dbot_score = Common.DBotScore(
                indicator=domain,
                indicator_type=DBotScoreType.DOMAIN,
                integration_name=VENDOR_NAME,
                score=Common.DBotScore.NONE,
                reliability=reliability,
            )
            readable_output = f"{CONTEXT_PREFIX} does not have details about domain: {domain} \n"
            domain_indicator = Common.Domain(domain=domain, dbot_score=dbot_score)

        result = CommandResults(
            outputs_prefix=f"{CONTEXT_PREFIX}.Domain",
            outputs_key_field="id",
            outputs=data,
            indicator=domain_indicator,
            readable_output=readable_output,
            raw_response=raw_response,
        )
        results.append(result)

    return results


def url_command(client: Client, args: Dict[str, Any], params: Dict[str, Any]) -> List[CommandResults]:
    """
    Returns URL's reputation
    """
    urls = argToList(args.get("url"))
    since = convert_string_to_epoch_time(args.get("since"), arg_name="since")
    until = convert_string_to_epoch_time(args.get("until"), arg_name="until")
    limit = arg_to_number(args.get("limit"), arg_name="limit")
    headers = argToList(args.get("headers"))
    reliability = params.get("feedReliability")
    share_level = args.get("share_level", params.get("share_level", "RED"))
    demisto.debug(f"Setting share level to {share_level}")
    results: List[CommandResults] = []
    for url in urls:
        try:
            raw_response = client.url(url, since, until, share_level, limit)
        except Exception as exception:
            # If anything happens, handle like there are no results
            err_msg = f'Could not process URL: "{url}"\n {exception!s}'
            demisto.debug(err_msg)
            raw_response = {}
            readable_output = f'Processing URL "{url}" resulted in an exception. See logs for the exact error.'
            result = CommandResults(outputs={}, readable_output=readable_output, raw_response=raw_response)
            results.append(result)
            continue
        if data := raw_response.get("data"):
            score = calculate_dbot_score(reputation_data=data, params=params)
            num_of_engines, num_of_positive_engines = calculate_engines(reputation_data=data)
            malicious_description = get_malicious_description(score, data, params)
            dbot_score = Common.DBotScore(
                indicator=url,
                indicator_type=DBotScoreType.URL,
                integration_name=VENDOR_NAME,
                score=score,
                reliability=reliability,
                malicious_description=malicious_description,
            )
            if not headers:
                headers = [
                    "description",
                    "owner",
                    "status",
                    "type",
                    "raw_indicator",
                    "share_level",
                    "confidence",
                    "severity",
                    "added_on",
                    "last_updated",
                    "review_status",
                    "id",
                ]
            readable_output = tableToMarkdown(f"{CONTEXT_PREFIX} Result for URL {url}", data, headers=headers)
            url_indicator = Common.URL(
                url=url, dbot_score=dbot_score, detection_engines=num_of_engines, positive_detections=num_of_positive_engines
            )
        else:  # no data
            dbot_score = Common.DBotScore(
                indicator=url,
                indicator_type=DBotScoreType.URL,
                integration_name=VENDOR_NAME,
                score=Common.DBotScore.NONE,
                reliability=reliability,
            )
            readable_output = f"{CONTEXT_PREFIX} does not have details about URL: {url} \n"
            url_indicator = Common.URL(url=url, dbot_score=dbot_score)

        result = CommandResults(
            outputs_prefix=f"{CONTEXT_PREFIX}.URL",
            outputs_key_field="id",
            outputs=data,
            indicator=url_indicator,
            readable_output=readable_output,
            raw_response=raw_response,
        )
        results.append(result)

    return results


def members_command(client: Client) -> CommandResults:
    """
    Returns a list of current members of the ThreatExchange, alphabetized by application name.
    Each application may also include an optional contact email address.
    See Also:
          https://developers.facebook.com/docs/threat-exchange/reference/apis/threat-exchange-members/v10.0

    """
    raw_response = client.members()
    if data := raw_response.get("data"):
        headers = ["id", "name", "email"]
        readable_output = tableToMarkdown(f"{CONTEXT_PREFIX} Members: ", data, headers=headers, removeNull=True)
    else:  # no data
        readable_output = f"{CONTEXT_PREFIX} does not have any members \n"

    result = CommandResults(
        outputs_prefix=f"{CONTEXT_PREFIX}.Member",
        outputs_key_field="id",
        outputs=data,
        readable_output=readable_output,
        raw_response=raw_response,
    )
    return result


def query_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    """
    Searches for subjective opinions on indicators of compromise stored in ThreatExchange.
    """
    text = str(args.get("text"))
    descriptor_type = str(args.get("type"))
    since = convert_string_to_epoch_time(args.get("since"), arg_name="since")
    until = convert_string_to_epoch_time(args.get("until"), arg_name="until")
    limit = arg_to_number(args.get("limit"), arg_name="limit")
    strict_text = argToBoolean(args.get("strict_text", False))
    headers = argToList(args.get("headers"))
    before = args.get("before")
    after = args.get("after")

    raw_response = client.query(text, descriptor_type, since, until, limit, strict_text, before, after)
    try:  # removes 'next' field to prevent access token uncovering
        del raw_response["paging"]["next"]
    except KeyError:  # for no paging cases
        pass

    if data := raw_response.get("data"):
        readable_output = tableToMarkdown(f"{CONTEXT_PREFIX} Query Result:", data, headers=headers)
        if raw_response.get("paging"):  # if paging exist - flatten the output
            outputs = flatten_outputs_paging(raw_response)
            readable_output += tableToMarkdown("Pagination:", outputs.get("paging"))
        else:  # no paging
            outputs = raw_response

    else:  # no data
        readable_output = f"{CONTEXT_PREFIX} does not have details about {descriptor_type}: {text} \n"
        outputs = raw_response

    outputs["text"] = text
    outputs["type"] = descriptor_type

    result = CommandResults(
        outputs_prefix=f"{CONTEXT_PREFIX}.Query",
        outputs_key_field=["text", "type"],
        outputs=outputs,
        readable_output=readable_output,
        raw_response=raw_response,
    )

    return result


def tags_search_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    """
    Enables searching for tags in ThreatExchange.
    With this call you can search for ThreatTag objects by text.
    See Also:
        https://developers.facebook.com/docs/threat-exchange/reference/apis/threattags/v10.0
    """
    text = str(args.get("text"))
    before = args.get("before")
    after = args.get("after")

    raw_response = client.tags_search(text, before, after)
    try:  # removes 'next' field to prevent access token uncovering
        del raw_response["paging"]["next"]
    except KeyError:  # for no paging cases
        pass

    if data := raw_response.get("data"):
        readable_output = tableToMarkdown(f"{CONTEXT_PREFIX} Tags: ", data, removeNull=True)
        if raw_response.get("paging"):  # if paging exist - flatten the output
            outputs = flatten_outputs_paging(raw_response)
            readable_output += tableToMarkdown("Pagination:", outputs.get("paging"))
        else:  # no paging
            outputs = raw_response

    else:  # no data
        readable_output = f"{CONTEXT_PREFIX} does not have any tags for text: {text} \n"
        outputs = raw_response

    outputs["text"] = text

    result = CommandResults(
        outputs_prefix=f"{CONTEXT_PREFIX}.Tag",
        outputs_key_field="text",
        outputs=outputs,
        readable_output=readable_output,
        raw_response=raw_response,
    )
    return result


def tagged_objects_list_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    """
    Gets a list of tagged objects for a specific ThreatTag.
    See Also:
        https://developers.facebook.com/docs/threat-exchange/reference/apis/threat-tags/v10.0
    """
    tag_id = str(args.get("tag_id"))
    tagged_since = arg_to_number(args.get("tagged_since"), arg_name="tagged_since")
    tagged_until = arg_to_number(args.get("tagged_until"), arg_name="tagged_until")
    before = args.get("before")
    after = args.get("after")

    raw_response = client.tagged_objects_list(tag_id, tagged_since, tagged_until, before, after)
    try:  # removes 'next' field to prevent access token uncovering
        del raw_response["paging"]["next"]
    except KeyError:  # for no paging cases
        pass

    if data := raw_response.get("data"):
        readable_output = tableToMarkdown(f"{CONTEXT_PREFIX} Tagged Objects for ThreatTag: {tag_id}", data, removeNull=True)
        if raw_response.get("paging"):  # if paging exist - flatten the output
            outputs = flatten_outputs_paging(raw_response)
            readable_output += tableToMarkdown("Pagination:", outputs.get("paging"))
        else:  # no paging
            outputs = raw_response

    else:  # no data
        readable_output = f"{CONTEXT_PREFIX} does not have any tagged objects for ThreatTag: {tag_id} \n"
        outputs = raw_response

    outputs["tag_id"] = tag_id

    result = CommandResults(
        outputs_prefix=f"{CONTEXT_PREFIX}.TaggedObject",
        outputs_key_field="tag_id",
        outputs=outputs,
        readable_output=readable_output,
        raw_response=raw_response,
    )
    return result


def object_get_by_id_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    """
    Gets ThreatExchange object by ID.
    """
    object_id = str(args.get("object_id"))

    raw_response = client.object_get_by_id(object_id)
    if raw_response:
        readable_output = tableToMarkdown(f"{CONTEXT_PREFIX} Object {object_id}:", raw_response, removeNull=True)
    else:  # no data
        readable_output = f"{CONTEXT_PREFIX} does not have any object with ID: {object_id} \n"

    result = CommandResults(
        outputs_prefix=f"{CONTEXT_PREFIX}.Object",
        outputs_key_field="id",
        outputs=raw_response,
        readable_output=readable_output,
        raw_response=raw_response,
    )
    return result


def main():
    """
    main function, parses params and runs command functions
    """
    command = demisto.command()
    params = demisto.params()
    args = demisto.args()

    app_id_obj = params.get("app_id")
    app_id = app_id_obj["identifier"]
    app_secret = app_id_obj["password"]
    access_token = f"{app_id}|{app_secret}"
    base_url = "https://graph.facebook.com/v3.2"
    verify_certificate = not argToBoolean(params.get("insecure", False))
    proxy = argToBoolean(params.get("proxy", False))
    handle_proxy()

    demisto.debug(f"Command being called is {command}")
    try:
        client = Client(base_url=base_url, access_token=access_token, verify=verify_certificate, proxy=proxy)
        result: Union[str, CommandResults, List[CommandResults]]
        if command == "test-module":
            result = test_module(client)
        elif command == "ip":
            result = ip_command(client, args, params)
        elif command == "file":
            result = file_command(client, args, params)
        elif command == "domain":
            result = domain_command(client, args, params)
        elif command == "url":
            result = url_command(client, args, params)
        elif command == f"{COMMAND_PREFIX}-members":
            result = members_command(client)
        elif command == f"{COMMAND_PREFIX}-query":
            result = query_command(client, args)
        elif command == f"{COMMAND_PREFIX}-tags-search":
            result = tags_search_command(client, args)
        elif command == f"{COMMAND_PREFIX}-tagged-objects-list":
            result = tagged_objects_list_command(client, args)
        elif command == f"{COMMAND_PREFIX}-object-get-by-id":
            result = object_get_by_id_command(client, args)
        else:
            raise NotImplementedError(f"Command {command} is not implemented")
        return_results(result)
    # Log exceptions and return errors
    except Exception as e:
        demisto.error(traceback.format_exc())  # print the traceback
        return_error(f"Failed to execute {command} command.\nError:\n{e!s}")


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