MetaDefender Sandbox Deprecated

from CommonServerPython import *
from CommonServerUserPython import *

""" IMPORTS """
from typing import Any
import urllib3

# Disable insecure warnings
urllib3.disable_warnings()

"""GLOBALS/PARAMS """
INTEGRATION_NAME = "MetaDefender Sandbox Integration"
INTEGRATION_CONTEXT_NAME = "MetaDefender.Sandbox"


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

        if self.api_key:
            self._headers = {"X-Api-Key": self.api_key}

    def test_module(self) -> dict:
        """
        Return information about the user. (Need API key)
        """
        request_result = self._http_request(
            method="GET",
            ok_codes=([200]),
            url_suffix="/users/me",
        )
        return request_result

    def post_sample(self, args: dict[str, Any]) -> dict[str, Any]:
        data = {}

        if description := args.get("description"):
            data["description"] = description
        if tags := args.get("tags"):
            data["tags"] = tags
        if password := args.get("password"):
            data["password"] = password
        if is_private := args.get("is_private"):
            data["is_private"] = is_private

        if url := args.get("url"):
            data["url"] = url

            return self._http_request(method="POST", url_suffix="/scan/url", ok_codes=([200]), data=data)

        elif entry_id := args.get("entry_id"):
            try:
                file_entry = demisto.getFilePath(entry_id)
            except Exception as e:
                raise DemistoException(f'Failed to find file entry with id:"{entry_id}". got error: {e}')

            with open(file_entry["path"], "rb") as file:
                return self._http_request(
                    method="POST",
                    url_suffix="/scan/file",
                    ok_codes=([200]),
                    data=data,
                    files={"file": file},
                )
        else:
            raise DemistoException("No file or URL was provided.")

    def get_scan_result(self, flow_id: str) -> dict[str, Any]:
        filters = [
            "filter=general",
            "filter=finalVerdict",
            "filter=allTags",
            "filter=overallState",
            "filter=taskReference",
            "filter=subtaskReferences",
            "filter=allSignalGroups",
        ]

        postfix = "&".join(filters)

        url_suffix = f"/scan/{flow_id}/report?{postfix}"

        response = self._http_request(
            method="GET",
            ok_codes=([200]),
            url_suffix=url_suffix,
        )

        return response

    def get_search_query(self, query_string: str, page: int, page_size: int) -> dict[str, Any]:
        return self._http_request(
            method="GET",
            ok_codes=([200]),
            params={"query": query_string, "page_size": page_size, "page": page},
            url_suffix="/reports/search",
        )


""" HELPER FUNCTIONS """


def build_one_reputation_result(report: dict[str, Any]):
    score = Common.DBotScore.NONE

    final_verdict = report.get("finalVerdict", {})
    verdict = final_verdict.get("verdict")
    if verdict.upper() == "BENIGN" or verdict.upper() == "INFORMATIONAL" or verdict.upper() == "NO_THREAT":
        score = Common.DBotScore.GOOD
    elif verdict.upper() == "MALICIOUS" or verdict.upper() == "LIKELY_MALICIOUS":
        score = Common.DBotScore.BAD
    elif verdict.upper() == "SUSPICIOUS":
        score = Common.DBotScore.SUSPICIOUS

    report_file = report.get("file", {})
    report_hash = report_file.get("hash", None)

    dbot_score = Common.DBotScore(
        indicator=report_hash,
        indicator_type=DBotScoreType.FILE,
        integration_name="MetaDefender Sandbox",
        score=score,
    )

    file = Common.File(name=report_file.get("name"), sha256=report_hash, dbot_score=dbot_score)

    tags = [tag.get("tag", {}).get("name") for tag in report.get("allTags", [])]
    subtasks = [subtask.get("name") for subtask in report.get("subtaskReferences", [])]
    human_readable = {
        "FileName": report_file.get("name"),
        "FileHash": report_hash,
        "FileType": report_file.get("type"),
        "FinalVerdict": report.get("finalVerdict", {}).get("verdict"),
        "Tags": tags,
        "SubtaskReferences": subtasks,
    }
    readable_output = tableToMarkdown("Scan Result (digest):", human_readable)

    results = CommandResults(
        outputs_prefix=f"{INTEGRATION_CONTEXT_NAME}.Analysis",
        outputs_key_field="sha256",
        outputs=report,
        raw_response=report,
        readable_output=readable_output,
        indicator=file,
    )
    return results


def build_serach_query_result(analyses: list[dict]) -> CommandResults:
    def build_analysis_hr(analysis: dict[str, Any]) -> dict[str, Any]:
        file_result = analysis.get("file", {})
        hr_analysis = {
            "Id": analysis.get("id"),
            "SampleName": file_result.get("name"),
            "SHA256": file_result.get("sha256"),
            "Verdict": analysis.get("verdict"),
            "State": analysis.get("state"),
            "Date": analysis.get("date"),
            "MIMEType": file_result.get("mime_type"),
            "Type": file_result.get("short_type"),
            "Tags": analysis.get("tags"),
        }
        return hr_analysis

    def build_indicator_object(analysis: dict[str, Any]):
        score = Common.DBotScore.NONE

        verdict = analysis.get("verdict", "UNKNOWN")
        if verdict.upper() == "BENIGN" or verdict.upper() == "INFORMATIONAL" or verdict.upper() == "NO_THREAT":
            score = Common.DBotScore.GOOD
        elif verdict.upper() == "MALICIOUS" or verdict.upper() == "LIKELY_MALICIOUS":
            score = Common.DBotScore.BAD
        elif verdict.upper() == "SUSPICIOUS":
            score = Common.DBotScore.SUSPICIOUS

        analysis_file = analysis.get("file", {})
        dbot_score = Common.DBotScore(
            indicator=analysis_file.get("sha256"),
            indicator_type=DBotScoreType.FILE,
            integration_name="MetaDefender Sandbox",
            score=score,
        )

        file = Common.File(
            name=analysis_file.get("name"),
            sha256=analysis_file.get("sha256"),
            dbot_score=dbot_score,
        )

        results = CommandResults(
            outputs_prefix=f"{INTEGRATION_CONTEXT_NAME}.Analysis",
            outputs_key_field="sha256",
            indicator=file,
        )
        return results

    hr_headers = [
        "Id",
        "SampleName",
        "SHA256",
        "Verdict",
        "State",
        "Date",
        "MIMEType",
        "Type",
        "Tags",
    ]

    hr_analysis_ls = []

    for analysis in analyses:
        hr_analysis_ls.append(build_analysis_hr(analysis))

    command_result = CommandResults(
        outputs=analyses,
        readable_output=tableToMarkdown("Analysis Result:", hr_analysis_ls, hr_headers),
        outputs_prefix=f"{INTEGRATION_CONTEXT_NAME}.Analysis",
    )

    return command_result


def sample_submission(client: Client, args: dict[str, Any]) -> PollResult:
    res = client.post_sample(args)
    partial_res = CommandResults(readable_output=f'Waiting for submission "{res.get("flow_id")}" to finish...')
    return PollResult(
        response=CommandResults(
            outputs=res,
            outputs_prefix=f"{INTEGRATION_CONTEXT_NAME}.Submission",
            outputs_key_field="flow_id",
        ),
        args_for_next_run={"flow_id": res.get("flow_id"), **args},
        continue_to_poll=True,
        partial_result=partial_res,
    )


def build_reputation_result(api_reponse: dict[str, Any]):
    reports = api_reponse.get("reports", [])
    command_res_ls = []
    for report in reports:
        command_res_ls.append(build_one_reputation_result(reports[report]))
    return command_res_ls


def is_valid_pass(api_response: dict[str, Any]):
    if "rejected_files" not in api_response:
        return True
    return all(reject.get("rejected_reason") != "INVALID_PASSWORD" for reject in api_response["rejected_files"])


@polling_function(
    name=demisto.command(),
    timeout=arg_to_number(demisto.args().get("timeout", 600)),
    interval=1,
    poll_message="Polling result",
    requires_polling_arg=False,
)
def polling_submit_command(args: dict[str, Any], client: Client):
    if flow_id := args.get("flow_id"):
        api_response = client.get_scan_result(flow_id)
        successful_response = False

        if api_response.get("allFinished", False):
            successful_response = True

        if successful_response:
            if not is_valid_pass(api_response):
                raise DemistoException("Invalid password!")

            return PollResult(response=build_reputation_result(api_response), continue_to_poll=False)

        return PollResult(
            response=[
                CommandResults(
                    outputs=api_response,
                    outputs_prefix=f"{INTEGRATION_CONTEXT_NAME}.Submission",
                    outputs_key_field="flow_id",
                    readable_output=f'Waiting for submission "{api_response.get("flow_id")}" to finish...',
                ),
            ],
            continue_to_poll=True,
            args_for_next_run={"flow_id": args.get("flow_id"), **args},
        )
    else:
        return sample_submission(client, args)


""" COMMANDS """


def test_module_command(client: Client, *_) -> str:
    """Performs a basic GET request to check if the API is reachable and authentication is successful."""
    results = client.test_module()
    if "accountId" in results:
        return "ok"
    raise DemistoException(f"\nTest module failed, {results}")


def scan_command(client: Client, args: dict[str, Any]):
    return polling_submit_command(args=args, client=client)


def search_query_command(client: Client, args: dict[str, Any]):
    def validate_args():
        if page_size and page_size not in [5, 10, 20]:
            raise DemistoException("Page size value must be 5, 10 or 20")
        if page and page <= 0:
            raise DemistoException("Page must be an integer and grater than 0")
        if limit and (limit <= 0 or limit > 50):
            raise DemistoException("Limit must be an integer and between 1 and 50")

    items = []
    query_string = args.get("query", "")
    page_size = arg_to_number(args.get("page_size"))
    page = arg_to_number(args.get("page"))
    limit = arg_to_number(args.get("limit")) or 10

    validate_args()

    if page_size and not page:
        page = 1
    elif not page_size and page:
        page_size = 10

    if page_size and page:
        items = client.get_search_query(query_string, page, page_size).get("items", [])
    else:
        page_size = 20
        page = 1
        continue_query = True
        while continue_query:
            response = client.get_search_query(query_string, page, page_size)
            actual_items = response.get("items", [])
            total_available_items = response.get("count", len(items))
            # queried all or reached the limit
            if total_available_items == len(items) or len(items) >= limit:
                continue_query = False
            items += actual_items
            page += 1
        items = items[0:limit]

    if items:
        return build_serach_query_result(items)
    return CommandResults(readable_output="No Results were found.")


""" COMMANDS MANAGER / SWITCH PANEL """


def main():
    params = demisto.params()
    base_url = params.get("url", "")
    api_key = params.get("api_key", {}).get("password")
    verify = not params.get("insecure", False)
    proxy = params.get("proxy", False)

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

    try:
        client = Client(api_key=api_key, base_url=base_url, verify=verify, proxy=proxy)
        handle_proxy()

        if command == "test-module":
            return_results(test_module_command(client))
        elif command == "metadefender-sandbox-scan-url":
            return_results(scan_command(client, args))
        elif command == "metadefender-sandbox-scan-file":
            return_results(scan_command(client, args))
        elif command == "metadefender-sandbox-search-query":
            return_results(search_query_command(client, args))
        else:
            raise NotImplementedError(f"{command} command is not implemented.")
    except Exception as e:
        err_msg = f"Exception in {INTEGRATION_NAME} : Failed to execute {command} command: [{e!r}]"
        return_error(err_msg, error=e)


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