OpenCTI Feed 4.X

from typing import Any

import demistomock as demisto  # noqa: F401
import urllib3
from CommonServerPython import *  # noqa: F401
from pycti import OpenCTIApiClient

# Disable insecure warnings
urllib3.disable_warnings()

# Disable info logging from the api
logging.getLogger().setLevel(logging.ERROR)
OPENCTI_LOGS = "opencti_logs"

XSOAR_TYPES_TO_OPENCTI = {
    "account": "User-Account",
    "domain": "Domain-Name",
    "email": "Email-Addr",
    "file-md5": "StixFile",
    "file-sha1": "StixFile",
    "file-sha256": "StixFile",
    "file": "StixFile",
    "host": "X-OpenCTI-Hostname",
    "ip": "IPv4-Addr",
    "ipv6": "IPv6-Addr",
    "registry key": "Windows-Registry-Key",
    "url": "Url",
}
OPENCTI_TYPES_TO_XSOAR = {
    "User-Account": "Account",
    "Domain-Name": "Domain",
    "Email-Addr": "Email",
    "StixFile": "File",
    "X-OpenCTI-Hostname": "Host",
    "IPv4-Addr": "IP",
    "IPv6-Addr": "IPv6",
    "Windows-Registry-Key": "Registry Key",
    "Url": "URL",
}


def build_indicator_list(indicator_list: list[str]) -> list[str]:
    """Builds an indicator list for the query
    Args:
        indicator_list: List of XSOAR indicators types to return..

    Returns:
        indicators: list of OPENCTI indicators types"""
    result = []
    if "ALL" in indicator_list:
        # Replaces "ALL" for all types supported on XSOAR.
        result = [
            "User-Account",
            "Domain-Name",
            "Email-Addr",
            "StixFile",
            "X-OpenCTI-Hostname",
            "IPv4-Addr",
            "IPv6-Addr",
            "Windows-Registry-Key",
            "Url",
        ]
    else:
        result = [XSOAR_TYPES_TO_OPENCTI.get(indicator.lower(), indicator) for indicator in indicator_list]
    return result


def reset_last_run():
    """
    Reset the last run from the integration context
    """
    demisto.setIntegrationContext({})
    return CommandResults(readable_output="Fetch history deleted successfully")


def get_indicators(
    client: OpenCTIApiClient,
    indicator_types: list[str],
    score: list[str] = None,
    limit: int | None = 500,
    last_run_id: str | None = None,
    tlp_color: str | None = None,
    tags: list[str] = None,
) -> tuple[str, list]:
    """Retrieving indicators from the API

    Args:
        score: Range of scores to filter by.
        client: OpenCTI Client object.
        indicator_types: List of indicators types to return.
        last_run_id: The last id from the previous call to use pagination.
        limit: the max indicators to fetch
        tlp_color: traffic Light Protocol color
        tags: user tags

    Returns:
        new_last_run: the id of the last indicator
        indicators: list of indicators
    """
    indicator_type = build_indicator_list(indicator_types)
    demisto.debug(f"{OPENCTI_LOGS} - in get_indicators - builded indicator type  : {indicator_type}")
    filters: dict[str, Any] = {
        "mode": "and",
        "filters": [{"key": "entity_type", "values": indicator_type, "operator": "eq", "mode": "or"}],
        "filterGroups": [],
    }
    if score:
        filters["filters"].append({"key": "x_opencti_score", "values": score, "operator": "eq", "mode": "or"})
    demisto.debug(f"{OPENCTI_LOGS} - in get_indicators - {filters=}")
    observables = client.stix_cyber_observable.list(filters=filters, after=last_run_id, first=limit, withPagination=True)
    new_last_run = observables.get("pagination").get("endCursor")
    demisto.debug(f"{OPENCTI_LOGS} - in get_indicators - {new_last_run=}")

    indicators = []
    for item in observables.get("entities"):
        indicator = {
            "value": item["observable_value"],
            "type": OPENCTI_TYPES_TO_XSOAR.get(item["entity_type"], item["entity_type"]),
            "rawJSON": item,
            "fields": {
                "tags": [tag.get("value") for tag in item.get("objectLabel")],
                "description": item.get("x_opencti_description"),
            },
        }
        if tags:
            indicator["fields"]["tags"] += tags

        if not tlp_color:
            if object_marking := item.get("objectMarking", []):
                new_tlp_color = object_marking[0].get("definition", "").split(":")
                indicator["fields"]["trafficlightprotocol"] = new_tlp_color[1]
        else:
            indicator["fields"]["trafficlightprotocol"] = tlp_color

        score = item.get("x_opencti_score")
        if score in [*range(21)]:
            indicator["score"] = 1
        elif score in [*range(21, 71)]:
            indicator["score"] = 2
        elif score in [*range(71, 101)]:
            indicator["score"] = 3

        indicators.append(indicator)
    demisto.debug(f"{OPENCTI_LOGS} - in get_indicators - sum of indicators: {len(indicators)}")
    return new_last_run, indicators


def fetch_indicators_command(
    client: OpenCTIApiClient, indicator_types: list, max_fetch: int, score: list = None, tlp_color=None, tags=None, is_test=False
) -> list:
    """fetch indicators from the OpenCTI

    Args:
        score: Range of scores to filter by.
        client: OpenCTI Client object
        indicator_types(list): List of indicators types to get.
        max_fetch: (int) max indicators to fetch.
        tlp_color: (str)
        tags: (list)
        is_test: (bool) Indicates that it's a test and then does not save the last run.
    Returns:
        list of indicators(list)
    """
    last_run_id = demisto.getIntegrationContext().get("last_run_id")
    demisto.info(f"{OPENCTI_LOGS} - in fetch_indicators_command - get last run = {last_run_id}")

    new_last_run, indicators_list = get_indicators(
        client, indicator_types, limit=max_fetch, last_run_id=last_run_id, tlp_color=tlp_color, score=score, tags=tags
    )

    if new_last_run and not is_test:
        demisto.setIntegrationContext({"last_run_id": new_last_run})
        demisto.debug(f"{OPENCTI_LOGS} - in fetch_indicators_command - set last run = {new_last_run}")
        # we submit the indicators in batches
        for b in batch(indicators_list, batch_size=2000):
            demisto.createIndicators(b)
            demisto.debug(f"{OPENCTI_LOGS} - in fetch_indicators_command - indicators created successfully.")

    return indicators_list


def get_indicators_command(client: OpenCTIApiClient, args: dict) -> CommandResults:
    """Gets indicator from opencti to readable output

    Args:
        client: OpenCTI Client object
        args: demisto.args()

    Returns:
        readable_output, raw_response
    """
    indicator_types = argToList(args.get("indicator_types"))
    last_run_id = args.get("last_run_id")
    limit = arg_to_number(args.get("limit", 50))
    start = arg_to_number(args.get("score_start", 1))
    end = arg_to_number(args.get("score_end", 100)) + 1  # type:ignore
    demisto.debug(f"{OPENCTI_LOGS} - in get_indicators_command - {indicator_types=} {last_run_id=} {limit=} {start=} {end=}")
    score = None
    if start or end:
        score = [str(i) for i in range(start, end)]  # type:ignore
    demisto.debug(f"{OPENCTI_LOGS} - in get_indicators_command -{score=}")
    last_run_id, indicators_list = get_indicators(
        client=client, indicator_types=indicator_types, limit=limit, last_run_id=last_run_id, score=score
    )

    if indicators_list:
        indicators = [
            {
                "type": indicator["type"],
                "value": indicator["value"],
                "id": indicator["rawJSON"]["id"],
                "rawJSON": indicator["rawJSON"],
            }
            for indicator in indicators_list
        ]
        readable_output = tableToMarkdown("Indicators", indicators, headers=["type", "value", "id"], removeNull=True)
        return CommandResults(readable_output=readable_output, raw_response=indicators_list)
    else:
        return CommandResults(readable_output="No indicators")


def main():
    params = demisto.params()
    args = demisto.args()

    credentials = params.get("credentials", {})
    api_key = credentials.get("password")
    base_url = params.get("base_url").strip("/")
    indicator_types = params.get("indicator_types", ["ALL"])
    max_fetch = params.get("max_indicator_to_fetch")
    tlp_color = params.get("tlp_color")
    tags = argToList(params.get("feedTags"))

    if max_fetch:
        max_fetch = arg_to_number(max_fetch)
    else:
        max_fetch = 500
    start = arg_to_number(params.get("score_start", 1))
    end = arg_to_number(params.get("score_end", 100)) + 1  # type:ignore
    score = None
    if start or end:
        score = [str(i) for i in range(start, end)]  # type:ignore
    try:
        client = OpenCTIApiClient(base_url, api_key, ssl_verify=params.get("insecure"), log_level="error", proxies=handle_proxy())
        command = demisto.command()
        demisto.info(f"Command being called is {command}")

        # Switch case
        if command == "fetch-indicators":
            fetch_indicators_command(client, indicator_types, max_fetch, tlp_color=tlp_color, score=score, tags=tags)

        elif command == "test-module":
            """When setting up an OpenCTI Client it is checked that it is valid and allows requests to be sent.
            and if not he immediately sends an error"""
            fetch_indicators_command(client, indicator_types, max_fetch, is_test=True)
            return_results("ok")

        elif command == "opencti-get-indicators":
            return_results(get_indicators_command(client, args))

        elif command == "opencti-reset-fetch-indicators":
            return_results(reset_last_run())

    except Exception as e:
        demisto.error(traceback.format_exc())  # print the traceback
        return_error(f"Error:\n [{e}]")


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