JoeSecurityV2

from collections.abc import Generator
from copy import deepcopy

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

from CommonServerUserPython import *  # noqa

urllib3.disable_warnings()

""" CONSTANTS """

DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"  # ISO8601 format with UTC, default in XSOAR
API_ERRORS = {
    1: "Quota",
    2: "MissingParameterError",
    3: "InvalidParameterError",
    4: "InvalidApiKeyError",
    5: "ServerOfflineError",
    6: "InternalServerError",
    7: "PermissionError",
    8: "UnknownEndpointError",
}


class Client(jbxapi.JoeSandbox):
    def __init__(
        self,
        apikey: str = "",
        on_premise: bool = True,
        base_url: str = "",
        accept_tac: bool = True,
        verify_ssl: bool = True,
        proxy: bool = False,
        create_relationships: bool = False,
        reliability: str = DBotScoreReliability.C,
    ):
        proxies = {}
        self.reliability = reliability
        self.create_relationships = create_relationships
        self.on_premise = on_premise
        if not proxy:
            demisto.debug("Removing proxy environment variables.")
        else:
            demisto.debug("Handling proxy.")
        proxies = handle_proxy()
        super().__init__(apikey=apikey, apiurl=base_url, accept_tac=accept_tac, verify_ssl=verify_ssl, proxies=proxies)

    def analysis_info_list(self, web_ids: List[str]) -> List[Dict[str, Any]]:
        """
        A wrapper function supporting a list of webids to query.

        Args:
           web_ids: List(str): List of analysis webids to query.

        Returns:
            result: List(Dict(str, any)): List of analysis info result.
        """
        return [self.analysis_info(webid=web_id) for web_id in web_ids]


""" HELPER FUNCTIONS """


def update_metrics(exception: Exception, exe_metrics: ExecutionMetrics):
    """
    Helper function that supports the update of the execution metrics.

    Args:
        exception: Exception: The exception.
        exe_metrics: ExecutionMetrics: Execution metrics object.

    Returns:
        None
    """
    if isinstance(exception, jbxapi.ApiError):
        match API_ERRORS.get(exception.code):
            case "Quota":
                exe_metrics.quota_error += 1
            case "InvalidApiKeyError" | "PermissionError":
                exe_metrics.auth_error += 1
            case "ServerOfflineError" | "InternalServerError":
                exe_metrics.connection_error += 1
            case _:
                exe_metrics.general_error += 1
    else:
        exe_metrics.general_error += 1


def paginate(args: Dict[str, Any], results: Generator) -> List:
    """
    Helper function supporting paginate for results.

    Args:
       args: Dict(str, any): paginate arguments (page, page_size, limit).
       results: (Generator): API results for paginate.

    Returns:
        result: List: The requests pages.
    """
    page: Optional[int] = arg_to_number(args.get("page", None))
    page_size: Optional[int] = arg_to_number(args.get("page_size", None))
    limit: Optional[int] = arg_to_number(args.get("limit", 50))

    # paginate is available only if supplied page and page size.
    if (page and not page_size) or (page_size and not page):
        raise DemistoException("Either `page` or `page_size` was not provided.")
    if page and page <= 0:
        raise ValueError("The 'page' argument value is not valid.")
    if page_size and page_size <= 0:
        raise ValueError("The 'page_size' argument value is not valid.")
    if limit and limit < 0:
        raise ValueError("The 'limit' argument value is not valid.")

    number_of_entries = page * page_size if (page and page_size) else limit if limit else 0
    if page and page_size:
        try:
            all_pages = [next(results) for _ in range(number_of_entries)]
            return all_pages[-page_size:]  # pylint: disable=E1130
        except StopIteration:
            return []
    return list(results)[:limit]


def build_analysis_hr(analysis: Dict[str, Any]) -> Dict[str, Any]:
    """
    Helper function that supports the building of the human-readable output.

    Args:
       analysis: Dict(str, any): Analysis result returned by the API.

    Returns:
        result: Dict(str, any): The analysis human-readable entry.
    """
    file_name = analysis.get("filename")
    sha1 = analysis.get("sha1")
    sha256 = analysis.get("sha256")
    md5 = analysis.get("md5")
    tags = analysis.get("tags")
    hr_analysis = {
        "Id": analysis.get("webid"),
        "SampleName": file_name,
        "Status": analysis.get("status"),
        "Time": analysis.get("time"),
        "MD5": md5,
        "SHA1": sha1,
        "SHA256": sha256,
        "Systems": list({run.get("system") for run in analysis.get("runs", [])}),
        "Result": list({run.get("detection") for run in analysis.get("runs", [])}),
        "Tags": tags,
        "Errors": list({run.get("error") for run in analysis.get("runs", [])}),
        "Comments": analysis.get("comments"),
    }
    return hr_analysis


def build_search_hr(analysis: Dict[str, Any]) -> Dict[str, Any]:
    """
    Helper function that supports the building of the human-readable output for search command.

    Args:
       analysis: Dict(str, any): Analysis result returned by the API.

    Returns:
        result: Dict(str, any): The analysis human-readable entry.
    """
    file_name = analysis.get("filename")
    web_id = analysis.get("webid")
    sha1 = analysis.get("sha1")
    sha256 = analysis.get("sha256")
    md5 = analysis.get("md5")
    tags = analysis.get("tags")
    hr_analysis = {"Id": web_id, "Name": file_name, "MD5": md5, "SHA1": sha1, "SHA256": sha256, "Tags": tags}
    return hr_analysis


def build_quota_hr(res: Dict[str, Any]) -> tuple[Dict[str, Any], List[str]]:
    """
    Helper function that supports the building of the human-readable output for quota command.

    Args:
       res: Dict(str, any): Quota result returned by the API.

    Returns:
        result: Tuple(Dict(str,Any), List(str)): The quota human-readable entry.
    """

    headers = [
        "Quota Type",
        "Daily Quota Current",
        "Daily Quota Limit",
        "Daily Quote Remaining",
        "Monthly Quota Current",
        "Monthly Quota Limit",
        "Monthly Quota Remaining",
    ]
    hr = {
        "Quota Type": res.get("type"),
        "Daily Quota Current": res.get("quota", {}).get("daily", {}).get("current", 0),
        "Daily Quota Limit": res.get("quota", {}).get("daily", {}).get("limit", 0),
        "Daily Quote Remaining": res.get("quota", {}).get("daily", {}).get("remaining", 0),
        "Monthly Quota Current": res.get("quota", {}).get("monthly", {}).get("current", 0),
        "Monthly Quota Limit": res.get("quota", {}).get("monthly", {}).get("limit", 0),
        "Monthly Quota Remaining": res.get("quota", {}).get("monthly", {}).get("remaining", 0),
    }
    return hr, headers


def build_submission_hr(res: Dict[str, Any], analyses: List[Dict[str, Any]]) -> tuple[Dict[str, Any], List[str]]:
    """
       Helper function that supports the building of the human-readable output for submission command.

    Args:
       res: Dict(str, any): Submission result returned by the API.
       analyses: List(Dict(str, any)): List of analysis result returned by the API.
    Returns:
        result: Tuple(Dict(str,Any), List(str)): The submission human-readable entry.
    """
    headers = [
        "Submission Id",
        "Sample Name",
        "Time",
        "Status",
        "Web Id",
        "Encrypted",
        "Analysis Id",
        "Classification",
        "Threat Name",
        "Score",
        "Detection",
    ]
    hr = {
        "Submission Id": res.get("submission_id"),
        "Sample Name": res.get("name"),
        "Time": res.get("time"),
        "Status": res.get("status"),
        "Web Id": analyses[0].get("webid"),
        "Encrypted": analyses[0].get("encrypted"),
        "Analysis Id": analyses[0].get("analysisid"),
        "Classification": analyses[0].get("classification"),
        "Threat Name": analyses[0].get("threatname"),
        "Score": analyses[0].get("score"),
        "Detection": analyses[0].get("detection"),
    }
    if res.get("analyses", [{}])[0].get("sha256"):
        headers.extend(["SHA256", "MD5", "SHA1", "File Name"])
        hr.update(
            {
                "SHA256": analyses[0].get("sha256"),
                "MD5": analyses[0].get("md5"),
                "SHA1": analyses[0].get("sha1"),
                "File Name": analyses[0].get("filename"),
            }
        )
    else:
        headers.append("URL")
        hr.update({"URL": res.get("url")})
    return hr, headers


def build_reputation_hr(analysis: Dict[str, Any], command: str) -> Dict[str, Any]:
    """
    Helper function that supports the building of the human-readable output.

    Args:
       analysis: (Dict(str, any)): Analysis result returned by the API.
       command: (str): The command url or file.

    Returns:
        result: Dict(str, any): The analysis human-readable entry.
    """
    if command == "file":
        file_name = analysis.get("filename")
        sha1 = analysis.get("sha1")
        sha256 = analysis.get("sha256")
        md5 = analysis.get("md5")
        tags = analysis.get("tags")
        hr_analysis = {"File Name": file_name, "Sha1": sha1, "Sha256": sha256, "Md5": md5, "Tags": tags}
        return hr_analysis
    return {"Url": analysis.get("filename")}


def build_relationships(threat_name: str, entity: str, entity_type: str) -> EntityRelationship:
    """
    Helper function that creates the relationships table.

    Args:
       threat_name (str): The malware from joe security.
       entity: (str): The main entity value.
       entity_type: Union(File, URL): The main entity type.

    Returns:
        result: List(EntityRelationship): The relationship entry.
    """
    return EntityRelationship(
        name=EntityRelationship.Relationships.INDICATOR_OF,
        entity_a=entity,
        entity_a_type=entity_type,
        entity_b=threat_name,
        entity_b_type=FeedIndicatorType.Malware,
        reverse_name=EntityRelationship.Relationships.INDICATED_BY,
    )


def build_indicator_object(
    client: Client, analysis: Dict[str, Any], analyses: List[Dict[str, Any]], is_reputation: bool = False
) -> CommandResults:
    """
    Helper function that creates the Indicator object.

    Args:
       client (Client): The client class.
       analysis: Dict(str, any): Analysis result returned by the API.
       analyses: List(Dict(str, any)): List of analysis result returned by the API.
       is_reputation: (bool): If the command is reputation command.
    Returns:
        result: (CommandResults) The indicator object including its relationships
    """
    if analysis.get("sha256"):
        return build_file_object(client, analysis, analyses, is_reputation)
    return build_url_object(client, analysis, analyses, is_reputation)


def build_file_object(
    client: Client, analysis: Dict[str, Any], analyses: List[Dict[str, Any]], is_reputation: bool
) -> CommandResults:
    """
    Helper function that creates the File object.

    Args:
       client (Client): The client class.
       analysis: Dict(str, any): Analysis result returned by the API.
       analyses: List(Dict(str, any)): List of analysis result returned by the API.
       is_reputation (bool): Whether the command is reputation command.
    Returns:
        result: (CommandResults) The indicator object including its relationships
    """
    file_name = analysis.get("filename")
    sha1 = analysis.get("sha1")
    sha256 = analysis.get("sha256", "")
    md5 = analysis.get("md5")
    tags = analysis.get("tags")
    threat_name = analysis.get("threatname", "")
    relationships = []
    headers = ["File Name", "Sha1", "Sha256", "Md5"]
    hr = {"File Name": file_name, "Sha1": sha1, "Sha256": sha256, "Md5": md5}
    score, description = max(
        [indicator_calculate_score(entry.get("detection", "")) for entry in analyses if entry.get("sha256") == sha256],
        key=lambda tup: tup[1],
    )  # Find the max dbot score between all the analysis results.
    dbot_score = Common.DBotScore(
        indicator=sha256,
        integration_name="JoeSecurityV2",
        indicator_type=DBotScoreType.FILE,
        reliability=DBotScoreReliability.get_dbot_score_reliability_from_str(client.reliability),
        score=score,
        malicious_description=description,
    )
    if client.create_relationships and threat_name and threat_name.lower() != "unknown":
        relationships.append(build_relationships(threat_name, sha256, FeedIndicatorType.File))
    indicator = Common.File(
        name=file_name, sha1=sha1, sha256=sha256, dbot_score=dbot_score, md5=md5, tags=tags, relationships=relationships
    )
    if is_reputation:
        return CommandResults(
            outputs=hr,
            indicator=indicator,
            relationships=relationships,
            outputs_prefix="Joe.File",
            readable_output=tableToMarkdown("File Result:", hr, headers),
        )
    return CommandResults(
        indicator=indicator, relationships=relationships, readable_output=tableToMarkdown("File Result:", hr, headers)
    )


def build_url_object(
    client: Client, analysis: Dict[str, Any], analyses: List[Dict[str, Any]], is_reputation: bool
) -> CommandResults:
    """
    Helper function that creates the URL object.

    Args:
       client (Client): The client class.
       analysis: Dict(str, any): Analysis result returned by the API.
       analyses: List(Dict(str, any)): Analysis results returned by the API.
       is_reputation (bool): Whether the command is reputation command.
    Returns:
        result: (CommandResults) The indicator object including its relationships
    """
    url = analysis.get("filename", "")
    threat_name = analysis.get("threatname", "")
    relationships = []

    score, description = max(
        [indicator_calculate_score(entry.get("detection", "")) for entry in analyses if entry.get("filename") == url],
        key=lambda tup: tup[1],
    )  # Find the max dbot score between all the analysis results.
    dbot_score = Common.DBotScore(
        indicator=url,
        integration_name="JoeSecurityV2",
        indicator_type=DBotScoreType.URL,
        reliability=DBotScoreReliability.get_dbot_score_reliability_from_str(client.reliability),
        score=score,
        malicious_description=description,
    )

    if client.create_relationships and threat_name and threat_name.lower() != "unknown":
        relationships.append(build_relationships(threat_name, url, FeedIndicatorType.URL))
    indicator = Common.URL(url=url, dbot_score=dbot_score, relationships=relationships)
    if is_reputation:
        return CommandResults(
            outputs=url,
            indicator=indicator,
            relationships=relationships,
            outputs_prefix="Joe.URL",
            readable_output=tableToMarkdown("Url Result:", {"Url": url}),
        )
    return CommandResults(
        indicator=indicator, relationships=relationships, readable_output=tableToMarkdown("Url Result:", {"Url": url})
    )


def indicator_calculate_score(detection: str = "") -> tuple[int, str]:
    """
    Calculate the DBot Score based on analysis detection.

    Args:
       detection (str): The analysis detection.

    Returns:
        dbot_score,description (tuple): The DBot Score and the description associated with it.
    """
    if "malicious" in detection:
        return Common.DBotScore.BAD, "This indicator is malicious"
    elif "suspicious" in detection:
        return Common.DBotScore.SUSPICIOUS, ""
    elif "clean" in detection:
        return Common.DBotScore.GOOD, ""
    return Common.DBotScore.NONE, ""


def build_search_command_result(analyses: List[Dict[str, Any]]) -> CommandResults:
    """
    Helper function that parses the search result object.

    Args:
       analyses: (List(Dict(str, any))): The analyses result returned by the API.
    Returns:
        result: (CommandResults) The parsed CommandResults object.
    """
    hr_headers = ["Id", "Name", "SHA256", "SHA1", "MD5", "Tags"]
    hr_analysis_ls = []
    for analysis in analyses:
        hr_analysis_ls.append(build_search_hr(analysis))

    return CommandResults(
        outputs=analyses,
        readable_output=tableToMarkdown("Analysis Result:", hr_analysis_ls, hr_headers),
        outputs_prefix="Joe.Analysis",
        outputs_key_field="Id",
    )


def build_analysis_command_result(client: Client, analyses: List[Dict[str, Any]], full_display: bool) -> List[CommandResults]:
    """
    Helper function that parses the analysis result object.

    Args:
       client (Client): The client class.
       analyses: (List(Dict(str, any))): The analyses result returned by the API.
       full_display (bool): Whether to display the full analysis result or not.
    Returns:
        result: List(CommandResults) The parsed CommandResults object.
    """
    hr_headers = ["Id", "SampleName", "Status", "Time", "MD5", "SHA1", "SHA256", "Systems", "Result", "Errors", "Comments"]
    command_res_ls = []
    hr_analysis_ls = []
    for analysis in analyses:
        hr_analysis_ls.append(build_analysis_hr(analysis))
        if full_display:
            command_res = build_indicator_object(client, analysis, analyses)
            command_res_ls.append(command_res)
    command_res_ls.append(
        CommandResults(
            outputs=analyses,
            readable_output=tableToMarkdown("Analysis Result:", hr_analysis_ls, hr_headers),
            outputs_prefix="Joe.Analysis",
        )
    )
    return command_res_ls


def build_reputiation_command_result(client: Client, analyses: List[Dict[str, Any]]) -> List[CommandResults]:
    """
    Helper function that parses the file or URL result object.

    Args:
       client (Client): The client class.
       analyses: (List[Dict(str, any)]): The analyses file or URL returned by the API.

    Returns:
        result: List(CommandResults): The parsed CommandResults object.
    """
    command_res_ls = []
    for analysis in analyses:
        command_res = build_indicator_object(client, analysis, analyses, is_reputation=True)
        command_res_ls.append(command_res)

    return command_res_ls


def build_submission_command_result(
    client: Client, res: Dict[str, Any], args: Dict[str, Any], exe_metrics: ExecutionMetrics, download_report: bool
) -> List[Union[CommandResults, Dict[str, Any]]]:
    """
    Helper function that parses the submission result object.

     Args:
        client (Client): The client class.
        res: (Dict(str, any)): The submission result returned by the API.
        args: (Dict(str, any)): The submission arguments.
        download_report: (bool): Indicates either to download the report.
        exe_metrics: (ExecutionMetrics): The execution metrics.

     Returns:
        result: (List[Union[CommandResults, Dict[str, Any]]]): A list of CommandResults objects or fileResult data.
    """
    command_results: List[Union[CommandResults, Dict[str, Any]]] = []
    report_type = args.get("report_type")
    full_display = argToBoolean(args.get("full_display", True))
    analyses = res.get("analyses", [])
    hr, headers = build_submission_hr(res, res.get("analyses", []))
    res_copy = deepcopy(res)

    if full_display:
        for analysis in analyses:
            command_res = build_indicator_object(client, analysis, analyses)
            web_id = analysis.get("webid")
            if download_report:
                report = download_report_command(client, {"type": report_type, "webid": web_id})
                command_results.append(report)
            command_results.append(command_res)
    context_entry = {"Joe.Submission": res, "Joe.Analysis": res.pop("analyses")}
    command_results.append(
        CommandResults(
            outputs=context_entry,
            outputs_key_field="submission_id",
            readable_output=tableToMarkdown("Submission Results:", hr, headers=headers),
            execution_metrics=exe_metrics.get_metric_list(),
            raw_response=res_copy,
        )
    )
    return command_results


def build_submission_params(args: Dict[str, Any], on_premise: bool) -> Dict[str, Any]:
    """
    Helper function that builds the submission parameters.

    Args:
       args: (Dict[str, Any]): The command arguments.
    Returns:
        result: (Dict[str, Any]): The submission parameters.
    """
    params = {
        "comments": args.get("comments", None),
        "systems": argToList(args.get("systems")),
        "tags": argToList(args.get("tags")),
        "internet-access": argToBoolean(args.get("internet_access", True)),
        "archive-no-unpack": argToBoolean(args.get("archive_no_unpack", False)),
        "ssl-inspection": argToBoolean(args.get("ssl_inspection", False)),
        "localized-internet-country": args.get("localized_internet_country"),
        "internet-simulation": argToBoolean(args.get("internet_simulation", False)),
        "hybrid-code-analysis": argToBoolean(args.get("hybrid_code_analysis", True)),
        "hybrid-decompilation": argToBoolean(args.get("hybrid_decompilation", False)),
        "vba-instrumentation": argToBoolean(args.get("vba_instrumentation", True)),
        "js-instrumentation": argToBoolean(args.get("js_instrumentation", True)),
        "java-jar-tracing": argToBoolean(args.get("java_jar_tracing", True)),
        "dotnet-tracing": argToBoolean(args.get("dotnet_tracing", True)),
        "amsi-unpacking": argToBoolean(args.get("amsi_unpacking", True)),
        "fast-mode": argToBoolean(args.get("fast_mode", False)),
        "secondary-results": argToBoolean(args.get("secondary_results", False)),
        "report-cache": argToBoolean(args.get("report_cache", False)),
        "command-line-argument": args.get("command_line_argument"),
        "live-interaction": argToBoolean(args.get("live_interaction", False)),
        "document-password": args.get("document_password"),
        "archive-password": args.get("archive_password"),
        "start-as-normal-user": argToBoolean(args.get("start_as_normal_user", False)),
        "language-and-locale": args.get("language_and_locale"),
        "delete-after-days": arg_to_number(args.get("delete_after_days", 30)),
        "encrypt-with-password": args.get("encrypt_with_password"),
        "email-notification": argToBoolean(args.get("email_notification", False)),
    }

    if on_premise:
        params.pop("delete-after-days", None)

    return params


def file_submission(
    client: Client, args: Dict[str, Any], params: Dict[str, Any], file: str, exe_metrics: ExecutionMetrics
) -> PollResult:
    """
    Helper function that submits a file to Joe Sandbox.

     Args:
        client (Client): The client class.
        args: (Dict[str, Any]): The command arguments.
        params: (Dict[str, Any]): The submission parameters.
        file: (str): The file to submit.
        exe_metrics

     Returns:
         result: (PollResult): The parsed PollResult object.
    """
    file_path = demisto.getFilePath(file)
    name = file_path["name"]
    demisto.debug(f"Trying to upload file {name=} from entry= {file_path['path']}")

    with open(file_path["path"], "rb") as f:
        file_to_send = (args.get("file_name") or name, f)
        res = client.submit_sample(sample=file_to_send, params=params, cookbook=args.get("cookbook"))
        exe_metrics.success += 1
        partial_res = CommandResults(readable_output=f'Waiting for submission "{res.get("submission_id")}" to finish...')
        return PollResult(
            response=CommandResults(outputs=res, outputs_prefix="Joe.Submission", outputs_key_field="submission_id"),
            args_for_next_run={"submission_id": res.get("submission_id"), **args},
            continue_to_poll=True,
            partial_result=partial_res,
        )


def url_submission(
    client: Client, args: Dict[str, Any], params: Dict[str, Any], url: str, exe_metrics: ExecutionMetrics
) -> PollResult:
    """
    Helper function that submits a URL to Joe Sandbox.

     Args:
        client (Client): The client class.
        args: (Dict[str, Any]): The command arguments.
        params: (Dict[str, Any]): The submission parameters.
        url: (str): The URL to submit.
        exe_metrics: (ExecutionMetrics): The execution metrics.
     Returns:
         result: (PollResult): The parsed PollResult object.
    """
    res = client.submit_url(url=url, params=params)
    exe_metrics.success += 1
    partial_res = CommandResults(readable_output=f'Waiting for submission "{res.get("submission_id")}" to finish...')
    return PollResult(
        response=CommandResults(outputs=res, outputs_prefix="Joe.Submission", outputs_key_field="submission_id"),
        args_for_next_run={"submission_id": res.get("submission_id"), **args},
        continue_to_poll=True,
        partial_result=partial_res,
    )


# The interval timeout should be set between 30 and 60 seconds for optimal server usage.
@polling_function(
    name=demisto.command(),
    timeout=arg_to_number(demisto.args().get("timeout", 1200)),
    interval=arg_to_number(demisto.args().get("interval_in_seconds", 45)),
    requires_polling_arg=False,
)
def polling_submit_command(
    args: Dict[str, Any], client: Client, params: Dict[str, Any], exe_metrics: ExecutionMetrics
) -> PollResult:
    """
    Helper function that polls the analysis result.

    Args:
       args: (Dict[str, Any]): The command arguments.
       client (Client): The client class.
       params: (Dict[str, Any]): The submission parameters.
       exe_metrics: (ExecutionMetrics): The execution metrics.
    Returns:
        result: (Dict[str, Any]): The analysis result.
    """
    if submission_id := args.get("submission_id"):
        res = client.submission_info(submission_id=submission_id)
        status = res.get("status")
        if status == "finished":
            command_results = build_submission_command_result(client, res, args, exe_metrics, True)
            return PollResult(command_results)
        return PollResult(
            response=None,
            partial_result=CommandResults(
                outputs=res,  # this is what the response will be in case job has finished
                outputs_prefix="Joe.Submission",
                outputs_key_field="submission_id",
                readable_output=f'Waiting for submission "{res.get("submission_id")}" to finish...',
            ),
            continue_to_poll=True,
            args_for_next_run={"submission_id": args.get("submission_id"), **args},
        )
    else:
        if file := args.get("entry_id"):
            return file_submission(client, args, params, file, exe_metrics)
        elif url := args.get("url"):
            return url_submission(client, args, params, url, exe_metrics)
        else:
            raise DemistoException("No file or URL was provided.")


""" COMMAND FUNCTIONS """


def test_module(client: Client) -> str:  # pragma: no cover
    """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.

    :type client: ``JoeSandbox``
    :param JoeSandbox: client to use

    :return: 'ok' if test passed, anything else will fail the test.
    :rtype: ``str``
    """
    try:
        demisto.debug(f"sending request with url: {client.apiurl + '/v2/server/online'}")
        client.server_online()
    except Exception as e:
        if isinstance(e, jbxapi.ApiError):
            match API_ERRORS.get(e.code):
                case "InvalidApiKeyError":
                    return "Invalid Api Key"
                case "PermissionError":
                    return "Permission Error"
                case "ServerOfflineError" | "InternalServerError":
                    return "Server error, please verify the Server URL or check if the server is offline"
        else:
            return f"Server error, please verify the server url or check if the server is online, exact error: {e}"
    return "ok"


def is_online_command(client: Client) -> CommandResults:
    """
    Check if the Joe Sandbox analysis back end is online or in maintenance mode.

    Args:
       client: (Client) The client class.

    Returns:
        result: (CommandResults) The CommandResults object .
    """
    result = client.server_online()
    status = "online" if result.get("online") else "offline"
    return CommandResults(
        outputs_prefix="Joe.ServerStatus.Online", outputs=result.get("online"), readable_output=f"Joe server is {status}"
    )


def analysis_info_command(client: Client, args: Dict[str, Any]) -> Union[CommandResults, List[CommandResults]]:
    """
    Get information about a specific analysis.

     Args:
        client: (Client) The client class.
        args: (Dict(str, any)) The commands arguments.

     Returns:
         result: Union[CommandResults, List[CommandResults]] The CommandResults object.
    """
    try:
        res = client.analysis_info(webid=args.get("webid"))
        command_results = build_analysis_command_result(client, [res], argToBoolean(args.get("full_display")))
        if command_results:
            return command_results
    except jbxapi.InvalidParameterError:
        pass
    return CommandResults(readable_output="No Results were found.")


def list_analysis_command(client: Client, args: Dict[str, Any]) -> Union[CommandResults, List[CommandResults]]:
    """
    List all analyses in descending order, starting with the most recent.
    The result may be sliced by page and page size or by limit. (default 50)

     Args:
        client: (Client) The client class.
        args: (Dict(str, any)) The commands arguments.

     Returns:
         result: Union(CommandResults, List(CommandResults)): The CommandResults object.
    """

    filtered_pages = paginate(args, client.analysis_list_paged())
    analysis_result_ls = client.analysis_info_list(web_ids=[entry.get("webid") for entry in filtered_pages])
    command_results = build_analysis_command_result(client, analysis_result_ls, argToBoolean(args.get("full_display")))
    if command_results:
        return command_results
    return CommandResults(readable_output="No Results were found.")


def download_report_command(client: Client, args: Dict[str, Any]) -> Dict[str, Any]:
    """
    Download a resource belonging to a report. This can be the full report, dropped binaries, etc.

     Args:
        client: (Client): The client class.
        args: (Dict(str, any)): The commands arguments.

     Returns:
         result: (Dict(str, Any)): The fileResult object.
    """
    web_id = args.get("webid")
    report_type = args.get("type")

    result = client.analysis_download(webid=web_id, type=report_type)[1]
    return fileResult(f"{web_id}_report.{report_type}", result, EntryType.ENTRY_INFO_FILE)


def download_sample_command(client: Client, args: Dict[str, Any]) -> Dict[str, Any]:
    """
    Download a sample.

     Args:
        client: (Client): The client class.
        args: (Dict(str, any)): The commands arguments.

     Returns:
         result: (Dict(str, Any)): The fileResult object.
    """
    web_id = args.get("webid")

    result = client.analysis_download(webid=web_id, type="sample")[1]
    return fileResult(f"{web_id}.dontrun", result)


def search_command(client: Client, args: Dict[str, Any]) -> Union[CommandResults, List[CommandResults]]:
    """
    Search through all analyses.

     Args:
        client: (Client) The client class.
        args: (Dict(str, any)) The commands arguments.

     Returns:
         result: (CommandResults) The CommandResults object.
    """
    query = args.get("query")
    result = client.analysis_search(query=query)
    if result:
        return build_analysis_command_result(client, result, argToBoolean(args.get("full_display", False)))
    return CommandResults(readable_output="No Results were found.")


def file_command(client: Client, args: Dict[str, Any]) -> Union[CommandResults, List[CommandResults]]:
    """
    The file reputation command.

     Args:
        client: (Client) The client class.
        args: (Dict(str, any)) The commands arguments.

     Returns:
         result: (CommandResults) The CommandResults object.
    """
    analyses = []
    files = set(argToList(args.get("file", "")))

    for file in files:
        res = client.analysis_search(query=file)
        if res:
            analyses.append(res[0])
    if analyses:
        return build_reputiation_command_result(client, analyses)
    return CommandResults(readable_output="No Results were found.")


def url_command(client: Client, args: Dict[str, Any]) -> Union[CommandResults, List[CommandResults]]:
    """
    The url reputation command.

     Args:
        client: (Client) The client class.
        args: (Dict(str, any)) The commands arguments.

     Returns:
         result: (CommandResults) The CommandResults object.
    """
    analyses = []
    urls = set(argToList(args.get("url", "")))

    for url in urls:
        res = client.analysis_search(query=url)
        if res:
            analyses.append(res[0])
    if analyses:
        return build_reputiation_command_result(client, analyses)
    return CommandResults(readable_output="No Results were found.")


def list_lia_countries_command(client: Client) -> CommandResults:
    """
    Retrieve a list of localized internet anonymization countries.

     Args:
        client: (Client) The client class.

     Returns:
         result: (CommandResults) The CommandResults object.
    """

    res = client.server_lia_countries()
    if res:
        data = [country.get("name") for country in res]
        return CommandResults(
            outputs_prefix="Joe.LIACountry", outputs=data, readable_output=tableToMarkdown("Results:", {"Name": data})
        )
    return CommandResults(readable_output="No Results were found.")


def list_lang_locales_command(client: Client) -> CommandResults:
    """
    Retrieve a list of available language and locale combinations.

     Args:
        client: (Client) The client class.

     Returns:
         result: (CommandResults) The CommandResults object.
    """

    res = client.server_languages_and_locales()
    if res:
        data = [lang.get("name") for lang in res]
        return CommandResults(
            outputs_prefix="Joe.LangLocale", outputs=data, readable_output=tableToMarkdown("Results:", {"Name": data})
        )
    return CommandResults(readable_output="No Results were found.")


def get_account_quota_command(client: Client) -> CommandResults:
    """
    Retrieve the quota for the current account.

     Args:
        client: (Client) The client class.

     Returns:
         result: (CommandResults) The CommandResults object.
    """
    res = client.account_info()
    if res:
        hr, headers = build_quota_hr(res)
        return CommandResults(
            outputs_prefix="Joe.AccountQuota", outputs=res, readable_output=tableToMarkdown("Results:", hr, headers=headers)
        )
    return CommandResults(readable_output="No Results were found.")


def submission_info_command(
    client: Client, args: Dict[str, Any], exe_metrics: ExecutionMetrics
) -> List[Union[CommandResults, Dict[str, Any]]]:
    """
    Retrieve information about a submission.
     Args:
        client: (Client) The client class.
        args: (Dict(str, any)) The commands arguments.
        exe_metrics: (ExecutionMetrics) The execution metrics object.
     Returns:
         result: (CommandResults) The CommandResults object.
    """
    submission_ids = argToList(args.get("submission_ids"))
    command_results = []
    for submission_id in submission_ids:
        res = client.submission_info(submission_id=submission_id)
        if res:
            command_results.extend(build_submission_command_result(client, res, args, exe_metrics, False))
        else:
            command_results.append(CommandResults(readable_output=f"No Results were found for submission {submission_id}."))
    return command_results


def submit_sample_command(client: Client, args: Dict[str, Any], exe_metrics: ExecutionMetrics) -> PollResult:
    """
    Upload a sample to Joe server.
     Args:
        client: (Client) The client class.
        args: (Dict(str, any)) The commands arguments.
        exe_metrics: (ExecutionMetrics) The execution metrics.
     Returns:
         result: (CommandResults) The CommandResults object.
    """
    params = build_submission_params(args, client.on_premise)
    try:
        return polling_submit_command(args, client, params, exe_metrics)
    except jbxapi.InvalidParameterError as e:
        if e.message == "Unknown parameter delete-after-days.":
            raw = {
                "code": 3,
                "message": (
                    "The parameter delete-after-days is only available in the cloud version.\n"
                    'Check the "On-Premise" option in the integration settings for on-premise setups to avoid this error'
                ),
            }
            raise jbxapi.ApiError(raw)
        else:
            raise


def submit_url_command(client: Client, args: Dict[str, Any], exe_metrics: ExecutionMetrics) -> PollResult:
    """
    Upload a URL to Joe server.
     Args:
        client: (Client) The client class.
        args: (Dict(str, any)) The commands arguments.
        exe_metrics: (ExecutionMetrics) The execution metrics.
     Returns:
         result: (CommandResults) The CommandResults object.
    """
    params = build_submission_params(args, client.on_premise)
    params.update({"url-reputation": argToBoolean(args.get("url_reputation", False))})
    return polling_submit_command(args, client, params, exe_metrics)


""" MAIN FUNCTION """


def main() -> None:  # pragma: no cover
    """main function, parses params and runs command functions

    :return:
    :rtype:
    """
    api_key = demisto.get(demisto.params(), "credentials.password")
    base_url = demisto.params().get("url")

    demisto.debug(f"base url before handling it: {base_url}")
    if "api" not in base_url:
        base_url = urljoin(base_url, "api/")
    demisto.debug(f"base url after handling it: {base_url}")

    verify_certificate = not demisto.params().get("insecure", False)
    proxy = demisto.params().get("proxy", False)
    reliability = demisto.params().get("Reliability", DBotScoreReliability.C)
    create_relationships = demisto.params().get("create_relationships", False)
    on_premise = demisto.params().get("onprem", False)
    command = demisto.command()
    args = demisto.args()
    exe_metrics = ExecutionMetrics()

    demisto.debug(f"Command being called is {demisto.command()}")
    try:
        client = Client(
            apikey=api_key,
            on_premise=on_premise,
            base_url=base_url,
            accept_tac=True,
            verify_ssl=verify_certificate,
            proxy=proxy,
            reliability=reliability,
            create_relationships=create_relationships,
        )

        if command == "test-module":
            return_results(test_module(client))
        elif command == "joe-is-online":
            return_results(is_online_command(client))
        elif command == "joe-analysis-info":
            return_results(analysis_info_command(client, args))
        elif command == "joe-list-analysis":
            return_results(list_analysis_command(client, args))
        elif command == "joe-download-report":
            demisto.results(download_report_command(client, args))
        elif command == "joe-download-sample":
            demisto.results(download_sample_command(client, args))
        elif command == "joe-search":
            return_results(search_command(client, args))
        elif command == "file":
            return_results(file_command(client, args))
        elif command == "url":
            return_results(url_command(client, args))
        elif command == "joe-list–lia-countries":  # noqa: RUF001 (backwards compatibility)
            return_results(list_lia_countries_command(client))
        elif command == "joe-list-lang-locales":
            return_results(list_lang_locales_command(client))
        elif command == "joe-get-account-quota":
            return_results(get_account_quota_command(client))
        elif command == "joe-submission-info":
            return_results(submission_info_command(client, args, exe_metrics))
        elif command == "joe-submit-sample":
            return_results(submit_sample_command(client, args, exe_metrics))
        elif command == "joe-submit-url":
            return_results(submit_url_command(client, args, exe_metrics))
        else:
            raise NotImplementedError(f"{command} command is not implemented.")

    except Exception as e:
        update_metrics(e, exe_metrics)
        return_results(CommandResults(execution_metrics=exe_metrics.get_metric_list()))
        return_error(f"Failed to execute {command} command.\nError:\n{e!s}")


""" ENTRY POINT """

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