AutoFocus V2 Deprecated

from requests import Response

from CommonServerPython import *

""" IMPORTS """

import socket
import traceback
from collections.abc import Callable

""" GLOBALS/PARAMS """
PARAMS = demisto.params()

API_KEY = AutoFocusKeyRetriever(PARAMS.get("credentials", {}).get("password") or PARAMS.get("api_key")).key

# Remove trailing slash to prevent wrong URL path to service
SERVER = "https://autofocus.paloaltonetworks.com"

# Service base URL
BASE_URL = SERVER + "/api/v1.0"
VENDOR_NAME = "AutoFocus V2"

# Headers to be sent in requests
HEADERS = {"Content-Type": "application/json"}
RELATIONSHIP_TYPE_BY_TAG_CLASS_ID = {
    1: {"entity_b_type": "STIX Threat Actor", "name": "indicator-of"},
    2: {"entity_b_type": "Campaign", "name": "indicator-of"},
    3: {"entity_b_type": "STIX Malware", "name": "indicator-of"},
    5: {"entity_b_type": "STIX Attack Pattern", "name": "indicator-of"},
}
API_PARAM_DICT = {
    "scope": {"Private": "private", "Public": "public", "Global": "global"},
    "order": {"Ascending": "asc", "Descending": "desc"},
    "artifact": "artifactSource",
    "sort": {
        "App Name": "app_name",
        "App Packagename": "app_packagename",
        "File type": "filetype",
        "Size": "size",
        "Finish Date": "finish_date",
        "First Seen (Create Date)": "create_date",
        "Last Updated (Update Date)": "update_date",
        "MD5": "md5",
        "SHA1": "sha1",
        "SHA256": "sha256",
        "Ssdeep Fuzzy Hash": "ssdeep",
        "Application": "app",
        "Device Country": "device_country",
        "Device Country Code": "device_countrycode",
        "Device Hostname": "device_hostname",
        "Device Serial": "device_serial",
        "Device vsys": "vsys",
        "Destination Country": "dst_country",
        "Destination Country Code": "dst_countrycode",
        "Destination IP": "dst_ip",
        "Destination Port": "dst_port",
        "Email Charset": "emailsbjcharset",
        "Industry": "device_industry",
        "Source Country": "src_country",
        "Source Country Code": "src_countrycode",
        "Source IP": "src_ip",
        "Source Port": "src_port",
        "Time": "tstamp",
        "Upload source": "upload_srcPossible",
    },
    "tag_class": {
        "Actor": "actor",
        "Campaign": "campaign",
        "Exploit": "exploit",
        "Malicious Behavior": "malicious_behavior",
        "Malware Family": "malware_family",
    },
    "search_arguments": {
        "file_hash": {"api_name": "alias.hash_lookup", "operator": "is"},
        "domain": {"api_name": "alias.domain", "operator": "contains"},
        "ip": {"api_name": "alias.ip_address", "operator": "contains"},
        "url": {"api_name": "alias.url", "operator": "contains"},
        "wildfire_verdict": {
            "api_name": "sample.malware",
            "operator": "is",
            "translate": {
                "Malware": 1,
                "Grayware": 2,
                "Benign": 3,
                "Phishing": 4,
            },
        },
        "first_seen": {"api_name": "sample.create_date", "operator": "is in the range"},
        "last_updated": {"api_name": "sample.update_date", "operator": "is in the range"},
        "time_range": {"api_name": "session.tstamp", "operator": "is in the range"},
        "time_after": {"api_name": "session.tstamp", "operator": "is after"},
        "time_before": {"api_name": "session.tstamp", "operator": "is before"},
    },
    "file_indicators": {
        "Size": "Size",
        "SHA1": "SHA1",
        "SHA256": "SHA256",
        "FileType": "Type",
        "Tags": "Tags",
        "FileName": "Name",
    },
    "search_results": {
        "sha1": "SHA1",
        "sha256": "SHA256",
        "filetype": "FileType",
        "malware": "Verdict",
        "size": "Size",
        "create_date": "Created",
        "finish_date": "Finished",
        "md5": "MD5",
        "region": "Region",
        "tag": "Tags",
        "_id": "ID",
        "tstamp": "Seen",
        "filename": "FileName",
        "device_industry": "Industry",
        "upload_src": "UploadSource",
        "fileurl": "FileURL",
        "artifact": "Artifact",
    },
}
SAMPLE_ANALYSIS_LINE_KEYS = {
    "behavior": {"display_name": "behavior", "indexes": {"risk": 0, "behavior": -1}},
    "process": {"display_name": "processes", "indexes": {"parent_process": 0, "action": 1}},
    "file": {"display_name": "files", "indexes": {"parent_process": 0, "action": 1}},
    "registry": {"display_name": "registry", "indexes": {"action": 1, "parameters": 2}},
    "dns": {"display_name": "DNS", "indexes": {"query": 0, "response": 1}},
    "http": {"display_name": "HTTP", "indexes": {"host": 0, "method": 1, "url": 2}},
    "connection": {"display_name": "connections", "indexes": {"destination": 2}},
    "mutex": {"display_name": "mutex", "indexes": {"process": 0, "action": 1, "parameters": 2}},
}
SAMPLE_ANALYSIS_COVERAGE_KEYS = {
    "wf_av_sig": {"display_name": "wildfire_signatures", "fields": ["name", "create_date"]},
    "fileurl_sig": {"display_name": "fileurl_signatures", "fields": ["name", "create_date"]},
    "dns_sig": {"display_name": "dns_signatures", "fields": ["name", "create_date"]},
    "url_cat": {"display_name": "url_categories", "fields": ["url", "cat"]},
}

VERDICTS_TO_DBOTSCORE = {"benign": 1, "malware": 3, "grayware": 2, "phishing": 3, "c2": 3}

ERROR_DICT = {
    404: "Invalid URL.",
    408: "Invalid URL.",
    409: "Invalid message or missing parameters.",
    500: "Internal error.",
    503: "Rate limit exceeded.",
}

if PARAMS.get("mark_as_malicious"):
    verdicts = argToList(PARAMS.get("mark_as_malicious"))
    VERDICTS_TO_DBOTSCORE.update(dict.fromkeys(verdicts, 3))

DEFAULT_BUCKET_INFO = dict.fromkeys(
    (
        "minute_points",
        "daily_points",
        "minute_points_remaining",
        "daily_points_remaining",
        "minute_bucket_start",
        "daily_bucket_start",
    ),
    "Unavailable",
)

EXECUTION_METRICS = ExecutionMetrics()
API_POINTS_TABLE = CommandResults(outputs_prefix="AutoFocus.Quota", replace_existing=True)


""" CLIENT CLASS """


class Client(BaseClient):
    def __init__(self, url: str, verify: bool = True, proxy: bool = False):
        super().__init__(base_url=url, verify=verify, proxy=proxy)

    def parse_response(self, resp: requests.Response, err_operation: str | None) -> dict:
        try:
            res_json = resp.json()
            save_api_metrics(res_json)  # type: ignore
            if resp.status_code == 503:
                EXECUTION_METRICS.quota_error += 1
                raise RateLimitExceededError(res_json)

            # Handle error responses gracefully
            if demisto.params().get("handle_error", True) and resp.status_code == 409:
                EXECUTION_METRICS.service_error += 1
                raise Exception("Response status code: 409 \nRequested sample not found")

            resp.raise_for_status()

            if "x-trace-id" in resp.headers:
                # this debug log was request by autofocus team for debugging on their end purposes
                demisto.debug(f'x-trace-id: {resp.headers["x-trace-id"]}')

            EXECUTION_METRICS.success += 1
            return res_json
        # Errors returned from AutoFocus
        except requests.exceptions.HTTPError:
            EXECUTION_METRICS.general_error += 1
            raise DemistoException(f'{err_operation}: {res_json.get("message")}')
        # Unexpected errors (where no json object was received)
        except Exception as err:
            EXECUTION_METRICS.general_error += 1
            raise DemistoException(f"{err_operation}: {err}")

    def http_request(
        self, url_suffix, method="POST", data={}, err_operation=None, ok_codes=(200, 409, 503), headers=HEADERS, params={}
    ):
        """A general http request.
        Args:
            method: str - the method to perform in the http request (GET, POST, etc.)
            url_suffix: str - the url_suffix for the http request.
            data: Any - the data for the request.
            err_operation: str - The string that will be in the exception in case one occurred.
            ok_codes: tuple - a list of status codes that we want to handle in the code and not raise automatic exception about.
            headers: dict - the headers of the request
            params: dict - the parameters of the request
        Returns:
            A response object in case of a GET method (url, ip, domain, file), otherwise a dict.
        """
        # The GET method is used only in search_indicator commands (url, ip, domain, file), and they have the API key in
        # the headers instead of data.
        if method == "POST":
            data.update({"apiKey": API_KEY})
        try:
            demisto.debug("http_request: before the request")
            res = self._http_request(
                method=method,
                url_suffix=url_suffix,
                data=json.dumps(data),
                headers=headers,
                resp_type="response",
                ok_codes=ok_codes,
                retries=3,
                params=params,
            )
            demisto.debug(f"http_request: {res=}")
        # Handle with connection error
        except requests.exceptions.ConnectionError as err:
            demisto.debug("http_request in requests.exceptions.ConnectionError")
            EXECUTION_METRICS.connection_error += 1
            raise DemistoException(f"Error connecting to server. Check your URL/Proxy/Certificate settings: {err}")

        # for the different search_indicator commands (url, ip, domain, file) the calling function handles the different
        # status codes.
        if method == "GET":
            return res

        return self.parse_response(res, err_operation)


""" HELPER FUNCTIONS """


class RateLimitExceededError(BaseException):
    def __init__(self, api_res: dict) -> None:
        super().__init__()
        self.api_res = api_res


def return_metrics():
    if EXECUTION_METRICS.metrics is not None and ExecutionMetrics.is_supported():
        return_results(EXECUTION_METRICS.metrics)
    if API_POINTS_TABLE.readable_output:
        return_results(API_POINTS_TABLE)


""" COMMAND FUNCTIONS """


def rerun_command_if_required(api_res: dict, retry_on_rate_limit: bool):
    daily_points_remaining = dict_safe_get(api_res, ("bucket_info", "daily_points_remaining"), 0)
    next_run = int(dict_safe_get(api_res, ("bucket_info", "wait_in_seconds"), 70, (int, float)))  # type: ignore
    if retry_on_rate_limit and daily_points_remaining and next_run < 300:
        results = CommandResults(
            readable_output="API Rate limit exceeded, rerunning command.",
            scheduled_command=ScheduledCommand(
                command=demisto.command(),
                args=(demisto.args() | {"retry_on_rate_limit": "false"}),
                next_run_in_seconds=(next_run + 20),
            ),
        )
    else:
        results = CommandResults(
            readable_output=f'Error in API call to AutoFocus.\nMessage: {api_res.get("message")}',
            entry_type=EntryType.ERROR,
            raw_response=f'Error in API call to AutoFocus.\nMessage: {api_res.get("message")}',
        )
    return_results(results)


def save_api_metrics(res_obj: dict):
    if bucket_info := res_obj.get("bucket_info"):
        demisto.debug(f"save_api_metrics: {bucket_info=}")
        API_POINTS_TABLE.readable_output = tableToMarkdown(
            "Autofocus API Points",
            {
                "Daily points used": "{daily_points_remaining}/{daily_points}",
                "Daily allotment started": "{daily_bucket_start}",
                "Minute points used": "{minute_points_remaining}/{minute_points}",
                "Minute allotment started": "{minute_bucket_start}",
            },
        ).format(**(DEFAULT_BUCKET_INFO | bucket_info))
        API_POINTS_TABLE.outputs = bucket_info
    else:
        demisto.debug("save_api_metrics: no bucket_info")


def run_polling_command(client: Client, args: dict, cmd: str, search_function: Callable, results_function: Callable):
    interval_in_secs = arg_to_number(args.get("interval_in_seconds", 60))
    if "af_cookie" not in args:
        # create new search
        command_results = search_function(client, args)
        outputs = command_results.outputs
        af_cookie = outputs.get("AFCookie")
        if outputs.get("Status") != "complete":
            polling_args = {"af_cookie": af_cookie, "interval_in_seconds": interval_in_secs, "polling": True, **args}
            command_results.scheduled_command = ScheduledCommand(
                command=cmd,
                next_run_in_seconds=interval_in_secs,  # type: ignore
                args=polling_args,
                timeout_in_seconds=600,
            )
            return command_results
        else:
            # continue to look for search results
            args["af_cookie"] = af_cookie
    # get search status
    command_results, status = results_function(client, args)
    if status != "complete":
        # schedule next poll
        polling_args = {"af_cookie": args.get("af_cookie"), "interval_in_seconds": interval_in_secs, "polling": True, **args}
        scheduled_command = ScheduledCommand(
            command=cmd,
            next_run_in_seconds=interval_in_secs,  # type: ignore
            args=polling_args,
            timeout_in_seconds=600,
        )

        # result with scheduled_command only - no update to the war room
        command_results = CommandResults(scheduled_command=scheduled_command)
    return command_results


def validate_sort_and_order_and_artifact(
    sort: Optional[str] = None, order: Optional[str] = None, artifact_source: Optional[str] = None
) -> bool:
    """
    Function that validates the arguments combination.
    sort and order arguments must be defined together.
    Sort and order can't appear with artifact.
    Args:
        sort: variable to sort by.
        order: the order which the results is ordered by.
        artifact_source: true if artifacts are needed and false otherwise.
    Returns:
        true if arguments are valid for the request, false otherwise.
    """
    if artifact_source == "true" and sort:
        raise Exception("Please remove or disable one of sort or artifact, As they are not supported in the api together.")
    elif sort and not order:
        raise Exception("Please specify the order of sorting (Ascending or Descending).")
    elif order and not sort:
        raise Exception("Please specify a field to sort by.")
    return bool(sort and order)


def do_search(
    client: Client,
    search_object: str,
    query: dict,
    scope: Optional[str],
    size: Optional[str] = None,
    sort: Optional[str] = None,
    order: Optional[str] = None,
    err_operation: Optional[str] = None,
    artifact_source: Optional[str] = None,
) -> dict:
    """
    This function created the data to be sent in http request and sends it.
    Args:
        client: The client to make the requests with.
        search_object: Type of search sessions or samples.
        query: Query based on conditions specified within this object.
        scope:  Scope of the search. Only available and required for: samples. e.g. Public, Global, Private.
        size: Number of results to provide.
        sort: Sort based on the provided artifact.
        order: How to display sort results in ascending or descending order.
        err_operation: String error which specificed which command failed.
        artifact_source: Whether artifacts are wanted or not.
    Returns:
        raw response of the http request.
    """
    path = "/samples/search" if search_object == "samples" else "/sessions/search"
    data = {"query": query, "size": size}
    if scope:
        data.update({"scope": API_PARAM_DICT["scope"][scope]})  # type: ignore
    if validate_sort_and_order_and_artifact(sort, order, artifact_source):
        data.update({"sort": {API_PARAM_DICT["sort"][sort]: {"order": API_PARAM_DICT["order"][order]}}})  # type: ignore
    if artifact_source == "true":
        data.update({"artifactSource": "af"})
        data.update({"type": "scan"})
    # Remove nulls
    data = createContext(data, removeNull=True)
    demisto.debug(f"do_search: {path=} {data=} {err_operation=}")
    result = client.http_request(url_suffix=path, data=data, err_operation=err_operation)
    return result


def run_search(
    client: Client,
    search_object: str,
    query: str,
    scope: Optional[str] = None,
    size: str = None,
    sort: str = None,
    order: str = None,
    artifact_source: str = None,
) -> dict:
    """
    This function searches the relevent search and returns search info for result command.
    Args:
        client: The client to make the requests with.
        search_object: Type of search sessions or samples.
        query: Query based on conditions specified within this object.
        scope:  Scope of the search. Only available and required for: samples. e.g. Public, Global, Private.
        size: Number of results to provide.
        sort: Sort based on the provided artifact.
        order: How to display sort results in ascending or descending order.
        artifact_source: Whether artifacts are wanted or not.
    Returns:
        dict of response for result commands.
    """
    result = do_search(
        client,
        search_object,
        query=json.loads(query),
        scope=scope,
        size=size,
        sort=sort,
        order=order,
        artifact_source=artifact_source,
        err_operation="Search operation failed",
    )
    in_progress = result.get("af_in_progress")
    status = "in progress" if in_progress else "complete"
    search_info = {
        "AFCookie": result.get("af_cookie"),
        "Status": status,
        "SessionStart": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
    }
    return search_info


def run_get_search_results(client: Client, search_object, af_cookie):
    path = f"/samples/results/{af_cookie}" if search_object == "samples" else f"/sessions/results/{af_cookie}"
    results = client.http_request(url_suffix=path, err_operation="Fetching search results failed")
    return results


def get_fields_from_hit_object(result_object, response_dict_name):
    new_object = {}
    af_params_dict = API_PARAM_DICT.get(response_dict_name)
    for key, value in result_object.items():
        if key in af_params_dict:  # type: ignore
            new_key = af_params_dict.get(key)  # type: ignore
            new_object[new_key] = value
        else:
            new_object[key] = value
    return new_object


def parse_hits_response(hits, response_dict_name):
    parsed_objects = []  # type: ignore
    if not hits:
        return parsed_objects
    else:
        for hit in hits:
            flattened_obj = {}  # type: ignore
            flattened_obj.update(hit.get("_source"))
            flattened_obj["_id"] = hit.get("_id")
            parsed_obj = get_fields_from_hit_object(flattened_obj, response_dict_name)
            parsed_objects.append(parsed_obj)
        return parsed_objects


def get_search_results(client: Client, search_object, af_cookie):
    results = run_get_search_results(client, search_object, af_cookie)
    retry_count = 0
    # Checking if the query has no results because the server has not fetched them yet.
    # In this case, the complete percentage would be 0 (or lower than 100).
    # In a case where there really aren't results (hits), the af_complete_percentage would be 100.
    while (not results.get("hits") and (results.get("af_complete_percentage", 0) != 100)) and retry_count < 10:
        time.sleep(5)
        results = run_get_search_results(client, search_object, af_cookie)
        retry_count += 1
    parsed_results = parse_hits_response(results.get("hits"), "search_results")
    in_progress = results.get("af_in_progress")
    status = "in progress" if in_progress else "complete"
    return parsed_results, status


def get_session_details(client: Client, session_id):
    path = f"/session/{session_id}"
    result = client.http_request(url_suffix=path, err_operation="Get session failed")
    parsed_result = parse_hits_response(result.get("hits"), "search_results")
    return parsed_result


def validate_if_line_needed(category, info_line):
    line = info_line.get("line")
    line_values = line.split(",")
    category_indexes = SAMPLE_ANALYSIS_LINE_KEYS.get(category).get("indexes")  # type: ignore
    if category == "behavior":
        risk_index = category_indexes.get("risk")  # type: ignore
        risk = line_values[risk_index].strip()
        # only lines with risk higher the informational are considered
        return risk != "informational"
    elif category == "registry":
        action_index = category_indexes.get("action")  # type: ignore
        action = line_values[action_index].strip()
        # Only lines with actions SetValueKey, CreateKey or RegSetValueEx are considered
        return action in ("SetValueKey", "CreateKey", "RegSetValueEx")
    elif category == "file":
        action_index = category_indexes.get("action")  # type: ignore
        action = line_values[action_index].strip()
        benign_count = info_line.get("b") if info_line.get("b") else 0
        malicious_count = info_line.get("m") if info_line.get("m") else 0
        # Only lines with actions Create or CreateFileW where malicious count is grater than benign count are considered
        return action in ("Create", "CreateFileW") and malicious_count > benign_count
    elif category == "process":
        action_index = category_indexes.get("action")  # type: ignore
        action = line_values[action_index].strip()
        # Only lines with actions created, CreateKey or CreateProcessInternalW are considered
        return action in ("created", "CreateProcessInternalW")
    else:
        return True


def get_data_from_line(line, category_name):
    category_indexes = SAMPLE_ANALYSIS_LINE_KEYS.get(category_name).get("indexes")  # type: ignore
    values = line.split(",")
    sub_categories = {}  # type: ignore
    if not category_indexes:
        return sub_categories
    else:
        for sub_category in category_indexes:  # type: ignore
            sub_category_index = category_indexes.get(sub_category)  # type: ignore
            sub_categories.update({sub_category: values[sub_category_index]})
        return sub_categories


def get_data_from_coverage_sub_category(sub_category_name, sub_category_data):
    sub_categories_list = []
    for item in sub_category_data:
        new_sub_category = {}
        fields_to_extract = SAMPLE_ANALYSIS_COVERAGE_KEYS.get(sub_category_name).get("fields")  # type: ignore
        for field in fields_to_extract:  # type: ignore
            new_sub_category[field] = item.get(field)  # type: ignore
        sub_categories_list.append(new_sub_category)
    return sub_categories_list


def parse_coverage_sub_categories(coverage_data):
    new_coverage = {}
    for sub_category_name, sub_category_data in coverage_data.items():
        if sub_category_name in SAMPLE_ANALYSIS_COVERAGE_KEYS and isinstance(sub_category_data, dict):
            new_sub_category_data = get_data_from_coverage_sub_category(sub_category_name, sub_category_data)
            new_sub_category_name = SAMPLE_ANALYSIS_COVERAGE_KEYS.get(sub_category_name).get(  # type: ignore
                "display_name"
            )  # type: ignore
            new_coverage[new_sub_category_name] = new_sub_category_data
    return {"coverage": new_coverage}


def parse_lines_from_os(category_name, data, filter_data_flag):
    new_lines = []
    for info_line in data:
        if not filter_data_flag or validate_if_line_needed(category_name, info_line):
            new_sub_categories = get_data_from_line(info_line.get("line"), category_name)
            new_lines.append(new_sub_categories)
    return new_lines


def parse_sample_analysis_response(resp, filter_data_flag):
    analysis = {}
    for category_name, category_data in resp.items():
        if category_name in SAMPLE_ANALYSIS_LINE_KEYS:
            new_category = {}
            for os_name, os_data in category_data.items():
                os_sanitized_data = parse_lines_from_os(category_name, os_data, filter_data_flag)
                new_category[os_name] = os_sanitized_data

            category_dict = SAMPLE_ANALYSIS_LINE_KEYS.get(category_name)
            analysis.update({category_dict["display_name"]: new_category})  # type: ignore

        elif category_name == "coverage":
            new_category = parse_coverage_sub_categories(category_data)
            analysis.update(new_category)

    return analysis


def sample_analysis(client: Client, sample_id, os, filter_data_flag):
    path = f"/sample/{sample_id}/analysis"
    data = {"coverage": "true"}
    if os:
        data["platforms"] = [os]  # type: ignore

    result = client.http_request(url_suffix=path, data=data, err_operation="Sample analysis failed")
    if "error" in result:
        return demisto.results(result["error"])
    analysis_obj = parse_sample_analysis_response(result, filter_data_flag)

    return analysis_obj


def parse_tag_details_response(resp):
    tag_details = resp.get("tag")
    fields_to_extract_from_tag_details = [
        "public_tag_name",
        "tag_name",
        "customer_name",
        "source",
        "tag_definition_scope",
        "tag_definition_status",
        "tag_class",
        "count",
        "lasthit",
        "description",
    ]
    new_tag_info = {}
    for field in fields_to_extract_from_tag_details:
        new_tag_info[field] = tag_details.get(field)

    tag_group_details = resp.get("tag_groups")
    if tag_group_details:
        new_tag_info["tag_group"] = tag_group_details

    return new_tag_info


def autofocus_tag_details(client: Client, tag_name):
    path = f"/tag/{tag_name}"
    resp = client.http_request(url_suffix=path, err_operation="Tag details operation failed")
    tag_info = parse_tag_details_response(resp)
    return tag_info


def validate_tag_scopes(private, public, commodity, unit42):
    if not any((private, public, commodity, unit42)):
        raise DemistoException("Add at least one Tag scope by setting `commodity`, `private`, `public` or `unit42` to True")


def autofocus_top_tags_search(client: Client, scope, tag_class_display, private, public, commodity, unit42):
    validate_tag_scopes(private, public, commodity, unit42)
    tag_class = API_PARAM_DICT["tag_class"][tag_class_display]  # type: ignore
    query = {"operator": "all", "children": [{"field": "sample.tag_class", "operator": "is", "value": tag_class}]}
    tag_scopes = []
    if private:
        tag_scopes.append("private")
    if public:
        tag_scopes.append("public")
    if commodity:
        tag_scopes.append("commodity")
    if unit42:
        tag_scopes.append("unit42")
    data = {"query": query, "scope": scope, "tagScopes": tag_scopes}
    path = "/top-tags/search/"
    resp = client.http_request(url_suffix=path, data=data, err_operation="Top tags operation failed")
    in_progress = resp.get("af_in_progress")
    status = "in progress" if in_progress else "complete"
    search_info = {"AFCookie": resp.get("af_cookie"), "Status": status}
    return search_info


def parse_top_tags_response(response):
    top_tags_list = []  # type: ignore
    top_tags = response.get("top_tags")
    if not top_tags:
        return top_tags_list
    else:
        for tag in top_tags:
            fields_to_extract_from_top_tags = ["tag_name", "public_tag_name", "count", "lasthit"]
            new_tag = {}
            for field in fields_to_extract_from_top_tags:
                new_tag[field] = tag[field]
            top_tags_list.append(new_tag)
        return top_tags_list


def get_top_tags_results(client: Client, af_cookie):
    path = f"/top-tags/results/{af_cookie}"
    results = client.http_request(url_suffix=path, err_operation="Fetching top tags results failed")
    top_tags = parse_top_tags_response(results)
    in_progress = results.get("af_in_progress")
    status = "in progress" if in_progress else "complete"
    return top_tags, status


def print_hr_by_category(category_name, category_data):
    hr = content = f"### {string_to_table_header(category_name)}:\nNo entries"
    if category_name == "coverage":
        content = category_data
        if category_data:
            hr = tableToMarkdown(
                f"{string_to_table_header(category_name)}:", category_data, headerTransform=string_to_table_header
            )
        else:
            hr = f"### {string_to_table_header(category_name)}:\nNo entries"
        demisto.results({"Type": entryTypes["note"], "ContentsFormat": formats["text"], "Contents": content, "HumanReadable": hr})
    else:
        for os_name, os_data in category_data.items():
            content = os_data
            table_header = f"{category_name}_{os_name}"
            if os_data:
                hr = tableToMarkdown(f"{string_to_table_header(table_header)}:", os_data, headerTransform=string_to_table_header)
            else:
                hr = f"### {string_to_table_header(table_header)}:\nNo entries"
            demisto.results(
                {"Type": entryTypes["note"], "ContentsFormat": formats["text"], "Contents": content, "HumanReadable": hr}
            )


def get_files_data_from_results(results):
    """
    Gets a list of results and for each result returns a file object includes all relevant file indicators exists
    in that result
    :param results: a list of dictionaries
    :return: a list of file objects
    """
    files = []
    if results:
        for result in results:
            raw_file = get_fields_from_hit_object(result, "file_indicators")
            file_data = filter_object_entries_by_dict_values(raw_file, "file_indicators")
            files.append(file_data)
    return files


def filter_object_entries_by_dict_values(result_object, response_dict_name):
    """
    Gets a dictionary (result_object) and filters it's keys by the values of another
    dictionary (response_dict_name)
    input: response_dict_name = 'file_indicators' - see API_PARAM_DICT above
           result_object = {
                              "app": "web-browsing",
                              "vsys": 1,
                              "SHA256": "18c9acd34a3aea09121f027857e0004a3ea33a372b213a8361e8a978330f0dc8",
                              "UploadSource": "Firewall",
                              "src_port": 80,
                              "device_serial": "007051000050926",
                              "Seen": "2019-07-24T09:37:04",
                              "Name": "wildfire-test-pe-file.exe",
                              "user_id": "unknown",
                              "src_country": "United States",
                              "src_countrycode": "US",
                              "dst_port": 65168,
                              "device_countrycode": "US",
                              "Industry": "High Tech",
                              "Region": "us",
                              "device_country": "United States",
                              "ID": "179972200903"
                            }
    output: {
                "SHA256": "18c9acd34a3aea09121f027857e0004a3ea33a372b213a8361e8a978330f0dc8",
                "Name": "wildfire-test-pe-file.exe"
            }
    :param result_object: a dictionary representing an object
    :param response_dict_name: a dictionary which it's values are the relevant fields (filters)
    :return: the result_object filtered by the relevant fields
    """
    af_params_dict = API_PARAM_DICT.get(response_dict_name)
    result_object_filtered = {}
    if af_params_dict and isinstance(result_object, dict) and isinstance(af_params_dict, dict):
        for key in result_object:
            if key in af_params_dict.values():  # type: ignore
                result_object_filtered[key] = result_object.get(key)
    return result_object_filtered


def search_samples(
    client: Client,
    query=None,
    scope=None,
    size=None,
    sort=None,
    order=None,
    file_hash=None,
    domain=None,
    ip=None,
    url=None,
    wildfire_verdict=None,
    first_seen=None,
    last_updated=None,
    artifact_source=None,
    use_batches=False,
):
    validate_no_query_and_indicators(query, [file_hash, domain, ip, url, wildfire_verdict, first_seen, last_updated])
    if not query:
        indicator_args_for_query = {"file_hash": file_hash, "domain": domain, "ip": ip, "url": url}
        used_indicator = validate_no_multiple_indicators_for_search(indicator_args_for_query)
        if use_batches:
            search_result = []
            for _batch in batch(indicator_args_for_query[used_indicator], batch_size=100):
                query = build_sample_search_query(used_indicator, _batch, wildfire_verdict, first_seen, last_updated)
                search_result.append(
                    run_search(
                        client,
                        "samples",
                        query=query,
                        scope=scope,
                        size=size,
                        sort=sort,
                        order=order,
                        artifact_source=artifact_source,
                    )
                )
            return search_result
        indicators = indicator_args_for_query[used_indicator]
        query = build_sample_search_query(used_indicator, indicators, wildfire_verdict, first_seen, last_updated)

    return run_search(
        client, "samples", query=query, scope=scope, size=size, sort=sort, order=order, artifact_source=artifact_source
    )


def build_sample_search_query(used_indicator, indicators_values, wildfire_verdict, first_seen, last_updated):
    indicator_list = build_indicator_children_query(used_indicator, indicators_values)
    indicator_query = build_logic_query("OR", indicator_list)
    filtering_args_for_search = {}  # type: ignore
    if wildfire_verdict:
        filtering_args_for_search["wildfire_verdict"] = demisto.get(
            API_PARAM_DICT, f"search_arguments.wildfire_verdict.translate.{wildfire_verdict}"
        )
    if first_seen:
        filtering_args_for_search["first_seen"] = first_seen
    if last_updated:
        filtering_args_for_search["last_updated"] = last_updated
    filters_list = build_children_query(filtering_args_for_search)
    filters_list.append(indicator_query)
    logic_query = build_logic_query("AND", filters_list)
    return json.dumps(logic_query)


def search_sessions(
    client: Client,
    query=None,
    size=None,
    sort=None,
    order=None,
    file_hash=None,
    domain=None,
    ip=None,
    url=None,
    from_time=None,
    to_time=None,
    use_batches=False,
):
    validate_no_query_and_indicators(query, [file_hash, domain, ip, url, from_time, to_time])
    if not query:
        indicator_args_for_query = {"file_hash": file_hash, "domain": domain, "ip": ip, "url": url}
        used_indicator = validate_no_multiple_indicators_for_search(indicator_args_for_query)
        if use_batches:
            search_result = []
            for _batch in batch(indicator_args_for_query[used_indicator], batch_size=100):
                query = build_session_search_query(used_indicator, _batch, from_time, to_time)
                search_result.append(run_search(client, "sessions", query=query, size=size, sort=sort, order=order))
            return search_result

        indicators = indicator_args_for_query[used_indicator]
        query = build_session_search_query(used_indicator, indicators, from_time, to_time)

    return run_search(client, "sessions", query=query, size=size, sort=sort, order=order)


def build_session_search_query(used_indicator, indicators_batch, from_time, to_time):
    indicator_list = build_indicator_children_query(used_indicator, indicators_batch)
    indicator_query = build_logic_query("OR", indicator_list)
    time_filters_for_search = {}  # type: ignore
    if from_time and to_time:
        time_filters_for_search = {"time_range": [from_time, to_time]}
    elif from_time:
        time_filters_for_search = {"time_after": [from_time]}
    elif to_time:
        time_filters_for_search = {"time_before": [to_time]}

    filters_list = build_children_query(time_filters_for_search)
    filters_list.append(indicator_query)
    logic_query = build_logic_query("AND", filters_list)
    return json.dumps(logic_query)


def build_logic_query(logic_operator, condition_list):
    return {"operator": {"AND": "all", "OR": "any"}.get(logic_operator), "children": condition_list}


def build_children_query(args_for_query):
    children_list = []  # type: ignore
    for key, val in args_for_query.items():
        field_api_name = API_PARAM_DICT["search_arguments"][key]["api_name"]  # type: ignore
        operator = API_PARAM_DICT["search_arguments"][key]["operator"]  # type: ignore
        children_list += children_list_generator(field_api_name, operator, [val])
    return children_list


def build_indicator_children_query(used_indicator, indicators_values):
    if indicators_values:
        field_api_name = API_PARAM_DICT["search_arguments"][used_indicator]["api_name"]  # type: ignore
        operator = API_PARAM_DICT["search_arguments"][used_indicator]["operator"]  # type: ignore
        children_list = children_list_generator(field_api_name, operator, indicators_values)
    else:
        children_list = []
        demisto.debug(f"{indicators_values=} -> {children_list=}")
    return children_list


def children_list_generator(field_name, operator, val_list):
    query_list = []
    for value in val_list:
        query_list.append({"field": field_name, "operator": operator, "value": value})
    return query_list


def validate_no_query_and_indicators(query, arg_list):
    if query and any(arg_list):
        raise DemistoException(
            "The search command can either run a search using a custom query or use the builtin arguments, but not both"
        )


def validate_no_multiple_indicators_for_search(arg_dict):
    used_arg = None
    for arg, val in arg_dict.items():
        if val and used_arg:
            raise DemistoException(
                f"The search command can receive one indicator type at a time, two were given: {used_arg}, {arg}."
                " For multiple indicator types use the custom query"
            )
        elif val:
            used_arg = arg
    if not used_arg:
        raise DemistoException("In order to perform a samples/sessions search, a query or an indicator must be given.")
    return used_arg


def search_indicator(client, indicator_type, indicator_value):
    headers = HEADERS | {"apiKey": API_KEY}

    params = {
        "indicatorType": indicator_type,
        "indicatorValue": indicator_value,
        "includeTags": "true",
    }

    demisto.debug(f"search_indicator: {indicator_value=}")
    result = Response()
    try:
        # 404, 409, 503 a list of status codes that we want to handle in the code and not raise automatic exception about.
        ok_codes = (200, 404, 409, 503)
        demisto.debug(f"search_indicator: using the client with {ok_codes=}")
        result = client.http_request(method="GET", url_suffix="/tic", headers=headers, params=params, ok_codes=ok_codes)
        demisto.debug(f"search_indicator: {result.status_code=} {result=}")

        try:
            result_json = result.json()
        except ValueError:
            result_json = {}
        demisto.debug(f"search_indicator {result_json=}")

        save_api_metrics(result_json)
        if result.status_code == 503:
            demisto.debug("search_indicator: result.status_code == 503")
            EXECUTION_METRICS.quota_error += 1
            raise RateLimitExceededError(result_json)

        # Handle error responses gracefully
        result.raise_for_status()

    # Handle with connection error
    except requests.exceptions.ConnectionError as err:
        demisto.debug("search_indicator: in requests.exceptions.ConnectionError")
        EXECUTION_METRICS.connection_error += 1
        raise DemistoException(f"Error connecting to server. Check your URL/Proxy/Certificate settings: {err}")

    # Unexpected errors (where no json object was received)
    except Exception as err:
        demisto.debug(f"search_indicator: Unexpected errors {err}")
        EXECUTION_METRICS.general_error += 1
        try:
            if demisto.params().get("handle_error", True) and (result.status_code == 404 or result.status_code == 409):
                return {
                    "indicator": {
                        "indicatorType": indicator_type,
                        "indicatorValue": indicator_value,
                        "latestPanVerdicts": {"PAN_DB": "UNKNOWN"},
                    }
                }
            text_error = result.json()
        except ValueError:
            text_error = {}
        error_message = text_error.get("message")
        if error_message:
            raise DemistoException(f"Request Failed with status: {result.status_code}.\nReason is: {error_message}.")
        elif result.status_code in ERROR_DICT:
            raise DemistoException(
                f"Request Failed with status: {result.status_code}.\nReason is: {ERROR_DICT[result.status_code]}."
            )
        else:
            err_msg = f"Request Failed with message: {err}."
        raise DemistoException(err_msg)

    EXECUTION_METRICS.success += 1
    return result_json


def parse_indicator_response(res, raw_tags, indicator_type):
    indicator = {}
    indicator["IndicatorValue"] = res.get("indicatorValue", "")
    indicator["IndicatorType"] = res.get("indicatorType", "")
    indicator["LatestPanVerdicts"] = res.get("latestPanVerdicts", "")
    indicator["WildfireRelatedSampleVerdictCounts"] = res.get("wildfireRelatedSampleVerdictCounts", "")
    indicator["SeenBy"] = res.get("seenByDataSourceIds", "")

    first_seen = res.get("firstSeenTsGlobal", "")
    last_seen = res.get("lastSeenTsGlobal", "")

    if first_seen:
        indicator["FirstSeen"] = timestamp_to_datestring(first_seen)
    if last_seen:
        indicator["LastSeen"] = timestamp_to_datestring(last_seen)

    if raw_tags:
        tags = []
        for tag in raw_tags:
            tags.append(
                {
                    "PublicTagName": tag.get("public_tag_name", ""),
                    "TagName": tag.get("tag_name", ""),
                    "CustomerName": tag.get("customer_name", ""),
                    "Source": tag.get("source", ""),
                    "TagDefinitionScopeID": tag.get("tag_definition_scope_id", ""),
                    "TagDefinitionStatusID": tag.get("tag_definition_status_id", ""),
                    "TagClassID": tag.get("tag_class_id", ""),
                    "Count": tag.get("count", ""),
                    "Lasthit": tag.get("lasthit", ""),
                    "Description": tag.get("description", ""),
                }
            )
        indicator["Tags"] = tags

    if indicator_type == "Domain":
        indicator["WhoisAdminCountry"] = res.get("whoisAdminCountry", "")
        indicator["WhoisAdminEmail"] = res.get("whoisAdminEmail", "")
        indicator["WhoisAdminName"] = res.get("whoisAdminName", "")
        indicator["WhoisDomainCreationDate"] = res.get("whoisDomainCreationDate", "")
        indicator["WhoisDomainExpireDate"] = res.get("whoisDomainExpireDate", "")
        indicator["WhoisDomainUpdateDate"] = res.get("whoisDomainUpdateDate", "")
        indicator["WhoisRegistrar"] = res.get("whoisRegistrar", "")
        indicator["WhoisRegistrarUrl"] = res.get("whoisRegistrarUrl", "")
        indicator["WhoisRegistrant"] = res.get("whoisRegistrant", "")

    return indicator


def calculate_dbot_score(indicator_response, indicator_type):
    latest_pan_verdicts = indicator_response["latestPanVerdicts"]
    if not latest_pan_verdicts:
        raise Exception("latestPanVerdicts value is empty in indicator response.")

    pan_db = latest_pan_verdicts.get("PAN_DB")
    wf_sample = latest_pan_verdicts.get("WF_SAMPLE")

    # use WF_SAMPLE value for file indicator and PAN_DB for domain,url and ip indicators
    if indicator_type == "File" and wf_sample:
        return VERDICTS_TO_DBOTSCORE.get(wf_sample.lower(), 0)
    elif pan_db:
        return VERDICTS_TO_DBOTSCORE.get(pan_db.lower(), 0)
    else:
        score = next(iter(latest_pan_verdicts.values()))
        if score:
            return VERDICTS_TO_DBOTSCORE.get(score.lower(), 0)
        return 0


def check_for_ip(indicator):
    if "-" in indicator:
        # check for address range
        ip1, ip2 = indicator.split("-", 1)

        if re.match(ipv4Regex, ip1) and re.match(ipv4Regex, ip2):
            return FeedIndicatorType.IP

        elif re.match(ipv6Regex, ip1) and re.match(ipv6Regex, ip2):
            return FeedIndicatorType.IPv6

        elif re.match(ipv4cidrRegex, ip1) and re.match(ipv4cidrRegex, ip2):
            return FeedIndicatorType.CIDR

        elif re.match(ipv6cidrRegex, ip1) and re.match(ipv6cidrRegex, ip2):
            return FeedIndicatorType.IPv6CIDR

        return None

    if "/" in indicator:
        if re.match(ipv4cidrRegex, indicator):
            return FeedIndicatorType.CIDR

        elif re.match(ipv6cidrRegex, indicator):
            return FeedIndicatorType.IPv6CIDR

        return None

    else:
        if re.match(ipv4Regex, indicator):
            return FeedIndicatorType.IP

        elif re.match(ipv6Regex, indicator):
            return FeedIndicatorType.IPv6

    return None


def find_indicator_type(indicator):
    """Infer the type of the indicator.

    Args:
        indicator(str): The indicator whose type we want to check.

    Returns:
        str. The type of the indicator.
    """
    # trying to catch X.X.X.X:portNum
    if ":" in indicator and "/" not in indicator:
        sub_indicator = indicator.split(":", 1)[0]
        ip_type = check_for_ip(sub_indicator)
        if ip_type:
            return ip_type

    ip_type = check_for_ip(indicator)

    if ip_type:
        # catch URLs of type X.X.X.X/path/url or X.X.X.X:portNum/path/url
        if "/" in indicator and (ip_type not in [FeedIndicatorType.IPv6CIDR, FeedIndicatorType.CIDR]):
            return FeedIndicatorType.URL

        else:
            return ip_type

    elif re.match(sha256Regex, indicator):
        return FeedIndicatorType.File

    # in AutoFocus, URLs include a path while domains do not - so '/' is a good sign for us to catch URLs.
    elif "/" in indicator:
        return FeedIndicatorType.URL

    else:
        return FeedIndicatorType.Domain


def resolve_ip_address(ip):
    if check_for_ip(ip):
        return socket.gethostbyaddr(ip)[0]

    return None


def convert_url_to_ascii_character(url_name):
    def convert_non_ascii_chars(non_ascii):
        # converts non-ASCII chars to IDNA notation
        return str(non_ascii.group(0)).encode("idna").decode("utf-8")

    # Regex to catch all non ascii chars (from 0 to 127 in hexadecimal).
    return re.sub(r"[^\x00-\x7F]+", convert_non_ascii_chars, url_name)


""" COMMANDS"""


def test_module(client: Client):
    """
    Performs basic get request to get item samples
    """
    query = {"operator": "all", "children": [{"field": "sample.malware", "operator": "is", "value": 1}]}
    demisto.debug("test-module")
    do_search(client, "samples", query=query, scope="Public", err_operation="Test module failed")


def search_samples_command(client: Client, args):
    file_hash = argToList(args.get("file_hash"))
    domain = argToList(args.get("domain"))
    ip = argToList(args.get("ip"))
    url = argToList(args.get("url"))
    wildfire_verdict = args.get("wildfire_verdict")
    first_seen = argToList(args.get("first_seen"))
    last_updated = argToList(args.get("last_updated"))
    query = args.get("query")
    scope = args.get("scope").capitalize()
    max_results = args.get("max_results")
    sort = args.get("sort")
    order = args.get("order")
    artifact_source = args.get("artifact")
    use_batches = args.get("use_batches")
    info = search_samples(
        client,
        query=query,
        scope=scope,
        size=max_results,
        sort=sort,
        order=order,
        file_hash=file_hash,
        domain=domain,
        ip=ip,
        url=url,
        wildfire_verdict=wildfire_verdict,
        first_seen=first_seen,
        last_updated=last_updated,
        artifact_source=artifact_source,
        use_batches=use_batches,
    )
    md = tableToMarkdown("Search Samples Info:", info)
    return CommandResults(
        outputs=info, readable_output=md, outputs_key_field="AFCookie", outputs_prefix="AutoFocus.SamplesSearch"
    )


def search_samples_with_polling_command(client: Client, args):
    return run_polling_command(client, args, "autofocus-search-samples", search_samples_command, samples_search_results_command)


def search_sessions_with_polling_command(client: Client, args):
    return run_polling_command(
        client, args, "autofocus-search-sessions", search_sessions_command, sessions_search_results_command
    )


def search_sessions_command(client: Client, args):
    file_hash = argToList(args.get("file_hash"))
    domain = argToList(args.get("domain"))
    ip = argToList(args.get("ip"))
    url = argToList(args.get("url"))
    from_time = args.get("time_after")
    to_time = args.get("time_before")
    time_range = args.get("time_range")
    query = args.get("query")
    max_results = args.get("max_results")
    sort = args.get("sort")
    order = args.get("order")
    use_batches = args.get("use_batches")

    if time_range:
        if from_time or to_time:
            raise DemistoException(
                "The 'time_range' argument cannot be specified with neither 'time_after' nor 'time_before' arguments."
            )
        else:
            from_time, to_time = time_range.split(",")

    info = search_sessions(
        client,
        query=query,
        size=max_results,
        sort=sort,
        order=order,
        file_hash=file_hash,
        domain=domain,
        ip=ip,
        url=url,
        from_time=from_time,
        to_time=to_time,
        use_batches=use_batches,
    )
    md = tableToMarkdown("Search Sessions Info:", info)
    cmd_results = CommandResults(
        outputs_prefix="AutoFocus.SessionsSearch", outputs_key_field="AFCookie", outputs=info, readable_output=md
    )
    return cmd_results


def samples_search_results_command(client: Client, args):
    af_cookie = args.get("af_cookie")
    results, status = get_search_results(client, "samples", af_cookie)
    files = get_files_data_from_results(results)
    hr = ""
    if not results or len(results) == 0:
        hr = "No entries found that match the query" if status == "complete" else f"Search Sessions Results is {status}"
    context = {
        "AutoFocus.SamplesResults(val.ID === obj.ID)": results,
        "AutoFocus.SamplesSearch(val.AFCookie === obj.AFCookie)": {"Status": status, "AFCookie": af_cookie},
        outputPaths["file"]: files,
    }
    if not results:
        return_outputs(readable_output=hr, outputs=context, raw_response={})
    else:
        # for each result a new entry will be set with two tables, one of the result and one of its artifacts
        for result in results:
            if "Artifact" in result:
                hr = samples_search_result_hr(result, status)
                return_outputs(readable_output=hr, outputs=context, raw_response=results)
            else:
                hr = tableToMarkdown(f"Search Samples Result is {status}", result)
                hr += tableToMarkdown("Artifacts for Sample: ", [])
                return_outputs(readable_output=hr, outputs=context, raw_response=results)
    return None, status


def samples_search_result_hr(result: dict, status: str) -> str:
    """
    Creates human readable output for a specific entry which contains two tables, one for the result's
    and another for the artifacts that are related to it.
    Args:
        result: one result of the search sample command.
        status: status of result command.
    Returns:
        human readable of two tables for this result.
    """
    artifact = result.pop("Artifact")
    updated_artifact = []
    for indicator in artifact:
        # Filter on returned indicator types, as we do not support Mutex and User Agent.
        if "Mutex" not in indicator.get("indicator_type") and "User Agent" not in indicator.get("indicator_type"):
            updated_artifact.append(indicator)
    hr = "\n\n".join(
        (
            tableToMarkdown(f"Search Samples Result is {status}", result),
            tableToMarkdown(
                "Artifacts for Sample: ", updated_artifact, headers=["b", "g", "m", "indicator_type", "confidence", "indicator"]
            ),
        )
    )
    return hr


def sessions_search_results_command(client: Client, args):
    af_cookie = args.get("af_cookie")
    results, status = get_search_results(client, "sessions", af_cookie)
    files = get_files_data_from_results(results)
    if not results or len(results) == 0:
        md = results = "No entries found that match the query"
    else:
        md = tableToMarkdown(f"Search Samples Results is {status}", results)
    context = {
        "AutoFocus.SessionsResults(val.ID === obj.ID)": results,
        "AutoFocus.SessionsSearch(val.AFCookie === obj.AFCookie)": {"Status": status, "AFCookie": af_cookie},
        outputPaths["file"]: files,
    }
    return CommandResults(outputs=context, raw_response=results, readable_output=md), status


def get_session_details_command(client: Client):
    args = demisto.args()
    session_id = args.get("session_id")
    result = get_session_details(client, session_id)
    files = get_files_data_from_results(result)
    md = tableToMarkdown(f"Session {session_id}:", result)
    context = {"AutoFocus.Sessions(val.ID === obj.ID)": result, outputPaths["file"]: files}
    demisto.results(
        {
            "Type": entryTypes["note"],
            "ContentsFormat": formats["text"],
            "Contents": result,
            "EntryContext": context,
            "HumanReadable": md,
        }
    )


def sample_analysis_command(client: Client):
    args = demisto.args()
    sample_id = args.get("sample_id")
    os = args.get("os")
    filter_data = args.get("filter_data") != "False"
    analysis = sample_analysis(client, sample_id, os, filter_data)
    context = createContext(analysis, keyTransform=string_to_context_key)
    demisto.results(
        {
            "Type": entryTypes["note"],
            "ContentsFormat": formats["text"],
            "Contents": {"ID": sample_id, "Analysis": analysis},
            "HumanReadable": f"### Sample Analysis results for {sample_id}:",
            "EntryContext": {"AutoFocus.SampleAnalysis(val.ID == obj.ID)": {"ID": sample_id, "Analysis": context}},
        }
    )
    for category_name, category_data in analysis.items():
        print_hr_by_category(category_name, category_data)


def tag_details_command(client: Client):
    args = demisto.args()
    tag_name = args.get("tag_name")
    result = autofocus_tag_details(client, tag_name)
    md = tableToMarkdown(f"Tag {tag_name} details:", result, headerTransform=string_to_table_header)
    context = createContext(result, keyTransform=string_to_context_key)
    demisto.results(
        {
            "Type": entryTypes["note"],
            "ContentsFormat": formats["text"],
            "Contents": result,
            "EntryContext": {"AutoFocus.Tag(val.ID == obj.ID)": context},
            "HumanReadable": md,
        }
    )


def top_tags_search_command(client: Client, args):
    scope = args.get("scope")
    tag_class = args.get("class")
    private = args.get("private") == "True"
    public = args.get("public") == "True"
    commodity = args.get("commodity") == "True"
    unit42 = args.get("unit42") == "True"
    info = autofocus_top_tags_search(client, scope, tag_class, private, public, commodity, unit42)
    md = tableToMarkdown("Top tags search Info:", info)
    return CommandResults(
        outputs_prefix="AutoFocus.TopTagsSearch", outputs_key_field="AFCookie", outputs=info, readable_output=md
    )


def top_tags_results_command(client: Client, args) -> tuple[CommandResults, str]:
    af_cookie = args.get("af_cookie")
    results, status = get_top_tags_results(client, af_cookie)
    md = tableToMarkdown(f"Search Top Tags Results is {status}:", results, headerTransform=string_to_table_header)
    context = createContext(results, keyTransform=string_to_context_key)
    outputs = {
        "AutoFocus.TopTagsResults(val.PublicTagName === obj.PublicTagName)": context,
        "AutoFocus.TopTagsSearch(val.AFCookie === obj.AFCookie)": {"Status": status, "AFCookie": af_cookie},
    }
    return CommandResults(outputs=outputs, raw_response=results, readable_output=md), status


def top_tags_with_polling_command(client: Client, args):
    return run_polling_command(client, args, "autofocus-top-tags-search", top_tags_search_command, top_tags_results_command)


def search_ip_command(client, ip, reliability, create_relationships):
    indicator_type = "IP"
    ip_list = argToList(ip)

    command_results = []
    relationships = []

    for ip_address in ip_list:
        ip_type = "ipv6_address" if is_ipv6_valid(ip_address) else "ipv4_address"
        raw_res = search_indicator(client, ip_type, ip_address)

        indicator = raw_res.get("indicator")
        if indicator:
            raw_tags = raw_res.get("tags")

            score = calculate_dbot_score(indicator, indicator_type)
            dbot_score = Common.DBotScore(
                indicator=ip_address,
                indicator_type=DBotScoreType.IP,
                integration_name=VENDOR_NAME,
                score=score,
                reliability=reliability,
            )
            if create_relationships:
                relationships = create_relationships_list(
                    entity_a=ip_address, entity_a_type=indicator_type, tags=raw_tags, reliability=reliability
                )
            ip = Common.IP(
                ip=ip_address,
                dbot_score=dbot_score,
                malware_family=get_tags_for_tags_and_malware_family_fields(raw_tags, True),
                tags=get_tags_for_tags_and_malware_family_fields(raw_tags),
                relationships=relationships,
            )

            autofocus_ip_output = parse_indicator_response(indicator, raw_tags, indicator_type)

            # create human readable markdown for ip
            tags = autofocus_ip_output.get("Tags")
            table_name = f"{VENDOR_NAME} {indicator_type} reputation for: {ip_address}"
            if tags:
                indicators_data = autofocus_ip_output.copy()
                del indicators_data["Tags"]
                md = tableToMarkdown(table_name, indicators_data)
                md += tableToMarkdown("Indicator Tags:", tags)
            else:
                md = tableToMarkdown(table_name, autofocus_ip_output)
        else:
            dbot_score = Common.DBotScore(
                indicator=ip_address,
                indicator_type=DBotScoreType.IP,
                integration_name=VENDOR_NAME,
                score=0,
                reliability=reliability,
            )
            ip = Common.IP(
                ip=ip_address,
                dbot_score=dbot_score,
            )
            md = f"### The IP indicator: {ip_address} was not found in AutoFocus"
            autofocus_ip_output = {"IndicatorValue": ip_address}

        command_results.append(
            CommandResults(
                outputs_prefix="AutoFocus.IP",
                outputs_key_field="IndicatorValue",
                outputs=autofocus_ip_output,
                readable_output=md,
                raw_response=raw_res,
                indicator=ip,
                relationships=relationships,
            )
        )

    return command_results


def search_domain_command(client, domain, reliability, create_relationships):
    indicator_type = "Domain"
    domain_name_list = argToList(domain)

    command_results = []
    relationships = []

    for domain_name in domain_name_list:
        raw_res = search_indicator(client, "domain", domain_name)
        indicator = raw_res.get("indicator")

        if indicator:
            raw_tags = raw_res.get("tags")
            score = calculate_dbot_score(indicator, indicator_type)
            dbot_score = Common.DBotScore(
                indicator=domain_name,
                indicator_type=DBotScoreType.DOMAIN,
                integration_name=VENDOR_NAME,
                score=score,
                reliability=reliability,
            )
            if create_relationships:
                relationships = create_relationships_list(
                    entity_a=domain_name, entity_a_type=indicator_type, tags=raw_tags, reliability=reliability
                )
            domain = Common.Domain(
                domain=domain_name,
                dbot_score=dbot_score,
                # Converting date format from YYYY-MM-DD to DD-MM-YYYY due to a parsing problem on the server later
                creation_date="-".join((indicator.get("whoisDomainCreationDate") or "").split("-")[::-1]),
                expiration_date="-".join((indicator.get("whoisDomainExpireDate") or "").split("-")[::-1]),
                updated_date="-".join((indicator.get("whoisDomainUpdateDate") or "").split("-")[::-1]),
                admin_email=indicator.get("whoisAdminEmail"),
                admin_name=indicator.get("whoisAdminName"),
                admin_country=indicator.get("whoisAdminCountry"),
                registrar_name=indicator.get("whoisRegistrar"),
                registrant_name=indicator.get("whoisRegistrant"),
                malware_family=get_tags_for_tags_and_malware_family_fields(raw_tags, True),
                tags=get_tags_for_tags_and_malware_family_fields(raw_tags),
                relationships=relationships,
            )
            autofocus_domain_output = parse_indicator_response(indicator, raw_tags, indicator_type)
            # create human readable markdown for ip
            tags = autofocus_domain_output.get("Tags")
            table_name = f"{VENDOR_NAME} {indicator_type} reputation for: {domain_name}"
            if tags:
                indicators_data = autofocus_domain_output.copy()
                del indicators_data["Tags"]
                md = tableToMarkdown(table_name, indicators_data)
                md += tableToMarkdown("Indicator Tags:", tags)
            else:
                md = tableToMarkdown(table_name, autofocus_domain_output)
        else:
            dbot_score = Common.DBotScore(
                indicator=domain_name,
                indicator_type=DBotScoreType.DOMAIN,
                integration_name=VENDOR_NAME,
                score=0,
                reliability=reliability,
            )
            domain = Common.Domain(domain=domain_name, dbot_score=dbot_score)
            md = f"### The Domain indicator: {domain_name} was not found in AutoFocus"
            autofocus_domain_output = {"IndicatorValue": domain_name}

        command_results.append(
            CommandResults(
                outputs_prefix="AutoFocus.Domain",
                outputs_key_field="IndicatorValue",
                outputs=autofocus_domain_output,
                readable_output=md,
                raw_response=raw_res,
                indicator=domain,
                relationships=relationships,
            )
        )
    return command_results


def search_url_command(client, url, reliability, create_relationships, separator=","):
    indicator_type = "URL"
    url_list = argToList(url, separator)

    command_results = []
    relationships = []

    for url_name in url_list:
        raw_res = search_indicator(client, "url", convert_url_to_ascii_character(url_name))

        indicator = raw_res.get("indicator")
        if indicator:
            indicator["indicatorValue"] = url_name
            raw_tags = raw_res.get("tags")

            score = calculate_dbot_score(indicator, indicator_type)

            dbot_score = Common.DBotScore(
                indicator=url_name,
                indicator_type=DBotScoreType.URL,
                integration_name=VENDOR_NAME,
                score=score,
                reliability=reliability,
            )
            if create_relationships:
                relationships = create_relationships_list(
                    entity_a=url_name, entity_a_type=indicator_type, tags=raw_tags, reliability=reliability
                )
            url = Common.URL(
                url=url_name,
                dbot_score=dbot_score,
                malware_family=get_tags_for_tags_and_malware_family_fields(raw_tags, True),
                tags=get_tags_for_tags_and_malware_family_fields(raw_tags),
                relationships=relationships,
            )

            autofocus_url_output = parse_indicator_response(indicator, raw_tags, indicator_type)
            autofocus_url_output = {k: v for k, v in autofocus_url_output.items() if v}

            tags = autofocus_url_output.get("Tags")
            table_name = f"{VENDOR_NAME} {indicator_type} reputation for: {url_name}"
            if tags:
                indicators_data = autofocus_url_output.copy()
                del indicators_data["Tags"]
                md = tableToMarkdown(table_name, indicators_data)
                md += tableToMarkdown("Indicator Tags:", tags)
            else:
                md = tableToMarkdown(table_name, autofocus_url_output)
        else:
            dbot_score = Common.DBotScore(
                indicator=url_name,
                indicator_type=DBotScoreType.URL,
                integration_name=VENDOR_NAME,
                score=0,
                reliability=reliability,
            )
            url = Common.URL(url=url_name, dbot_score=dbot_score)
            md = f"### The URL indicator: {url_name} was not found in AutoFocus"
            autofocus_url_output = {"IndicatorValue": url_name}

        command_results.append(
            CommandResults(
                outputs_prefix="AutoFocus.URL",
                outputs_key_field="IndicatorValue",
                outputs=autofocus_url_output,
                readable_output=md,
                raw_response=raw_res,
                indicator=url,
                relationships=relationships,
            )
        )

    return command_results


def search_file_command(client, file, reliability, create_relationships):
    indicator_type = "File"
    file_list = argToList(file)

    command_results = []
    relationships = []

    for file_hash in file_list:
        raw_res = search_indicator(client, "filehash", file_hash.lower())

        indicator = raw_res.get("indicator")
        if indicator:
            raw_tags = raw_res.get("tags")

            score = calculate_dbot_score(indicator, indicator_type)
            dbot_score = Common.DBotScore(
                indicator=file_hash,
                indicator_type=DBotScoreType.FILE,
                integration_name=VENDOR_NAME,
                score=score,
                reliability=reliability,
            )
            if create_relationships:
                relationships = create_relationships_list(
                    entity_a=file_hash, entity_a_type=indicator_type, tags=raw_tags, reliability=reliability
                )
            autofocus_file_output = parse_indicator_response(indicator, raw_tags, indicator_type)

            tags = autofocus_file_output.get("Tags")
            table_name = f"{VENDOR_NAME} {indicator_type} reputation for: {file_hash}"
            if tags:
                indicators_data = autofocus_file_output.copy()
                del indicators_data["Tags"]
                md = tableToMarkdown(table_name, indicators_data)
                md += tableToMarkdown("Indicator Tags:", tags)
            else:
                md = tableToMarkdown(table_name, autofocus_file_output)

            hash_type = get_hash_type(file_hash)

            file = Common.File(
                md5=file_hash if hash_type == "md5" else None,
                sha1=file_hash if hash_type == "sha1" else None,
                sha256=file_hash if hash_type == "sha256" else None,
                dbot_score=dbot_score,
                malware_family=get_tags_for_tags_and_malware_family_fields(raw_tags, True),
                tags=get_tags_for_tags_and_malware_family_fields(raw_tags),
                relationships=relationships,
            )
        else:
            dbot_score = Common.DBotScore(
                indicator=file_hash,
                indicator_type=DBotScoreType.FILE,
                integration_name=VENDOR_NAME,
                score=0,
                reliability=reliability,
            )
            hash_type = get_hash_type(file_hash)
            hash_val_arg = {hash_type: file_hash}
            file = Common.File(dbot_score=dbot_score, **hash_val_arg)
            md = f"### The File indicator: {file_hash} was not found in AutoFocus"
            autofocus_file_output = {"IndicatorValue": file_hash}

        command_results.append(
            CommandResults(
                outputs_prefix="AutoFocus.File",
                outputs_key_field="IndicatorValue",
                outputs=autofocus_file_output,
                readable_output=md,
                raw_response=raw_res,
                indicator=file,
                relationships=relationships,
            )
        )

    return command_results


def get_tags_for_generic_context(tags: Optional[list]):
    if not tags:
        return None
    results = []
    keys = ["TagGroups", "Aliases", "PublicTagName", "TagName"]
    sub_keys = ["TagGroupName"]
    for item in tags:
        generic_context_tags = {key: item.get(key) for key in keys}
        generic_context_tags["tagGroups"] = {key: item.get(key) for key in sub_keys}
        results.append(remove_empty_elements(generic_context_tags))
    return results


def get_tags_for_tags_and_malware_family_fields(tags: Optional[list], is_malware_family=False):
    """get specific tags for the tags and malware_family fields
    Args
        tags (Optional[list]): tags from the response
        is_malware_family (bool): indicating whether it is for the malware_family field
    return:
        List[str]: list of tags without duplicates and empty elements
    """
    if not tags:
        return None
    results = []
    for item in tags:
        results.append(item.get("tag_name"))
        results.append(item.get("public_tag_name"))
        for alias in item.get("aliases", []):
            results.append(alias)
        if not is_malware_family:
            for group in item.get("tagGroups", [{}]):
                results.append(group.get("tag_group_name"))
    # Returns a list without duplicates and empty elements
    return list(set(filter(None, results)))


def create_relationships_list(entity_a, entity_a_type, tags, reliability):
    """
    Create a list of relationships objects from the tags.

    entity_a (str): the entity a of the relation which is the current indicator.
    entity_a_type (str): the entity a type which is the type of the current indicator (IP/Domain/URL/File)
    tags (list): list of tags returned from the api.
    reliability (str): reliability of the source.

    return:
    list of EntityRelationship objects containing all the relationships from the enricher.
    """
    if not tags:
        return []
    relationships = []
    for tag in tags:
        tag_class = tag.get("tag_class_id")
        entity_b = tag.get("tag_name")
        relation_by_type = RELATIONSHIP_TYPE_BY_TAG_CLASS_ID.get(tag_class)
        if entity_b and relation_by_type:
            relationships.append(
                EntityRelationship(
                    relation_by_type.get("name"),
                    entity_a=entity_a,
                    entity_a_type=entity_a_type,
                    entity_b=entity_b,
                    entity_b_type=FeedIndicatorType.indicator_type_by_server_version(relation_by_type.get("entity_b_type")),
                    source_reliability=reliability,
                    brand=VENDOR_NAME,
                )
            )

    return relationships


def get_export_list_command(client: Client, args):
    # the label is the name of the export list we want to fetch.
    # panosFormatted is a flag stating that only indicators should be returned in the list.
    data = {"label": args.get("label"), "panosFormatted": True, "apiKey": ""}

    results = client.http_request(
        url_suffix="/export", method="POST", data=data, err_operation=f"Failed to fetch export list: {args.get('label')}"
    )

    indicators = []
    context_ip = []
    context_url = []
    context_domain = []
    context_file = []
    for indicator_value in results.get("export_list"):
        indicator_type = find_indicator_type(indicator_value)
        if indicator_type in [FeedIndicatorType.IP, FeedIndicatorType.IPv6, FeedIndicatorType.IPv6CIDR, FeedIndicatorType.CIDR]:
            if "-" in indicator_value:
                context_ip.append({"Address": indicator_value.split("-")[0]})
                context_ip.append({"Address": indicator_value.split("-")[1]})

            elif ":" in indicator_value:
                context_ip.append({"Address": indicator_value.split(":", 1)[0]})

            else:
                context_ip.append({"Address": indicator_value})

        elif indicator_type == FeedIndicatorType.Domain:
            context_domain.append({"Name": indicator_value})

        elif indicator_type == FeedIndicatorType.File:
            context_file.append({"SHA256": indicator_value})

        elif indicator_type == FeedIndicatorType.URL:
            if ":" in indicator_value:
                resolved_address = resolve_ip_address(indicator_value.split(":", 1)[0])
                semicolon_suffix = indicator_value.split(":", 1)[1]
                slash_suffix = None

            else:
                resolved_address = resolve_ip_address(indicator_value.split("/", 1)[0])
                slash_suffix = indicator_value.split("/", 1)[1]
                semicolon_suffix = None

            if resolved_address:
                if semicolon_suffix:
                    indicator_value = resolved_address + ":" + semicolon_suffix

                else:
                    indicator_value = resolved_address + "/" + slash_suffix

            context_url.append(
                {
                    "Data": indicator_value,
                }
            )

        indicators.append(
            {
                "Type": indicator_type,
                "Value": indicator_value,
            }
        )

    hr = tableToMarkdown(f"Export list {args.get('label')}", indicators, headers=["Type", "Value"])

    return_outputs(
        hr,
        {
            "AutoFocus.Indicator(val.Value == obj.Value && val.Type == obj.Type)": indicators,
            "IP(obj.Address == val.Address)": context_ip,
            "URL(obj.Data == val.Data)": context_url,
            "File(obj.SHA256 == val.SHA256)": context_file,
            "Domain(obj.Name == val.Name)": context_domain,
        },
        results,
    )


def main():
    if is_demisto_version_ge("8.12.0") and datetime.now() > datetime(2025, 12, 1):
        return_error("Palo Alto Networks AutoFocus v2 integration is deprecated. Please use Unit 42 Intelligence integration instead.")
    else:
        return_warning("AutoFocus v2 is deprecated and will stop working after December 1, 2025. Please migrate to Unit 42 Intelligence integration.")
    command = demisto.command()
    demisto.debug(f"Command being called is {command}")
    reliability = PARAMS.get("integrationReliability", "B - Usually reliable")
    create_relationships = PARAMS.get("create_relationships", True)
    proxy = PARAMS.get("proxy", False)
    USE_SSL = not PARAMS.get("insecure", False)  # Should we use SSL
    if DBotScoreReliability.is_valid_type(reliability):
        reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(reliability)
    else:
        raise Exception("AutoFocus error: Please provide a valid value for the Source Reliability parameter")

    client = Client(url=BASE_URL, verify=USE_SSL, proxy=proxy)
    args = demisto.args() | {
        "reliability": reliability,
        "create_relationships": create_relationships,
    }

    try:
        if command == "test-module":
            # This is the call made when pressing the integration test button.
            test_module(client)
            demisto.results("ok")
        elif command == "autofocus-search-samples":
            if args.get("polling") == "true":
                cmd_res = search_samples_with_polling_command(client, args)
                if cmd_res is not None:
                    return_results(cmd_res)
            else:
                args["use_batches"] = True
                return_results(search_samples_command(client, args))
        elif command == "autofocus-search-sessions":
            if args.get("polling") == "true":
                return_results(search_sessions_with_polling_command(client, args))
            else:
                args["use_batches"] = True
                return_results(search_sessions_command(client, args))
        elif command == "autofocus-samples-search-results":
            samples_search_results_command(client, args)
        elif command == "autofocus-sessions-search-results":
            return_results(sessions_search_results_command(client, args)[0])  # first result is CommandResults
        elif command == "autofocus-get-session-details":
            get_session_details_command(client)
        elif command == "autofocus-sample-analysis":
            sample_analysis_command(client)
        elif command == "autofocus-tag-details":
            tag_details_command(client)
        elif command == "autofocus-top-tags-search":
            if args.get("polling") == "true":
                return_results(top_tags_with_polling_command(client, args))
            else:
                return_results(top_tags_search_command(client, args))
        elif command == "autofocus-top-tags-results":
            return_results(top_tags_results_command(client, args)[0])
        elif command == "autofocus-get-export-list-indicators":
            get_export_list_command(client, args)
        elif command == "ip":
            return_results(search_ip_command(client, **args))
        elif command == "domain":
            return_results(search_domain_command(client, **args))
        elif command == "url":
            return_results(search_url_command(client, **args))
        elif command == "file":
            return_results(search_file_command(client, **args))
        else:
            raise NotImplementedError(f"Command {command!r} is not implemented.")

    except RateLimitExceededError as e:
        rerun_command_if_required(e.api_res, argToBoolean(args.get("retry_on_rate_limit", False)))

    except Exception as e:
        return_error(f"Unexpected error: {e}.\ntraceback: {traceback.format_exc()}")

    finally:
        return_metrics()


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