Symantec Blue Coat Content and Malware Analysis

import urllib3
from CommonServerPython import *

# disable insecure warnings
urllib3.disable_warnings()

BASE_URL = demisto.params()["url"]
USE_SSL = not demisto.params().get("insecure", False)


def http_request(url_suffix, data=None, files=None, parse_json=True):
    """
    Generic request to BlueCoat Malware Analysis
    """
    data = {} if data is None else data
    url_params = {}  # type:dict
    full_url = BASE_URL + url_suffix
    api_key = demisto.params().get("api_key_creds", {}).get("password") or demisto.params().get("api_key", False)
    headers = {"X-API-TOKEN": api_key}

    try:
        if files:
            result = requests.post(full_url, verify=USE_SSL, headers=headers, params=url_params, data=data, files=files)
        else:
            result = requests.get(full_url, verify=USE_SSL, headers=headers, params=url_params)
        if result.status_code == 200:
            if parse_json:
                return result.json()
            return result.content
        if result.status_code == 404:
            raise Exception("Resource Not Available or Does not Exist")
        demisto.debug(f"result is: {result.text}")
        raise Exception(f"Your request failed with the following error: {result.reason}.\n{result.text}")

    except requests.exceptions.ConnectTimeout as exception:
        err_msg = (
            "Connection Timeout Error - potential reasons might be that the Server URL parameter"
            " is incorrect or that the Server is not accessible from your host."
        )
        raise DemistoException(err_msg, exception)
    except requests.exceptions.SSLError as exception:
        err_msg = (
            "SSL Certificate Verification Failed - try selecting 'Trust any certificate' checkbox in"
            " the integration configuration."
        )
        raise DemistoException(err_msg, exception)
    except requests.exceptions.ProxyError as exception:
        err_msg = (
            "Proxy Error - if the 'Use system proxy' checkbox in the integration configuration is"
            " selected, try clearing the checkbox."
        )
        raise DemistoException(err_msg, exception)
    except Exception as exception:
        raise Exception(str(exception))


def url_analysis_to_entry(title, analysis):
    """
    Args:
        title: entry title
        analysis: url analysis data

    Returns:
        an entry
    """
    context = []
    table = []
    dbot_scores = []

    analysis_info = {
        "ID": analysis["id"],  # for detonate generic polling
        "SampleName": analysis["results"]["metadata"]["sample"]["label"].replace("hxxp", "http"),
        "Time": analysis["results"]["metadata"]["sample"]["date_added"],
        "URL": analysis["results"]["metadata"]["sample"]["url"],
        "Score": analysis["results"]["summary"]["risk_score"],
        "Result": analysis["results"]["summary"],
        "status": "done",
    }
    risk_score = analysis["results"]["summary"]["risk_score"]
    analysis_context = dict(analysis_info)
    analysis_table = dict(analysis_info)
    malicious = None
    suspicious = None

    if risk_score > 9:
        dbot_score = 3
        malicious = {
            "Vendor": "Symantec Content and Malware Analysis",
            "SHA256": analysis["results"]["metadata"]["sample"]["hashes"]["sha256"],
        }
    elif 7 < risk_score < 9:
        dbot_score = 2
        malicious = {
            "Vendor": "Symantec Content and Malware Analysis",
            "SHA256": analysis["results"]["metadata"]["sample"]["hashes"]["sha256"],
        }
        suspicious = True
    else:
        dbot_score = 1

    dbot_scores.append(
        {
            "Vendor": "Symantec Content and Malware Analysis",
            "Indicator": analysis["results"]["metadata"]["sample"]["label"].replace("hxxp", "http"),
            "Type": "url",
            "Score": dbot_score,
            "Malicious": malicious,
            "Suspicious": suspicious,
        }
    )
    context.append(analysis_context)
    table.append(analysis_table)

    entry = {
        "ContentsFormat": formats["json"],
        "Type": entryTypes["note"],
        "Contents": context,
        "ReadableContentsFormat": formats["markdown"],
        "HumanReadable": tableToMarkdown(title, table, removeNull=True),
        "EntryContext": {
            "Symantec.Analysis(val.ID && val.ID == obj.ID)": createContext(context, removeNull=True),
            "DBotScore": createContext(dbot_scores, removeNull=True),
        },
    }

    return entry


def file_analysis_to_entry(title, analysis):
    """
    Args:
        title: entry title
        analysis: file analysis data

    Returns:
        an entry
    """
    context = []
    table = []
    dbot_scores = []

    analysis_info = {
        "ID": analysis["id"],  # for detonate generic polling
        "SampleName": analysis["results"]["metadata"]["sample"]["label"],
        "Time": analysis["results"]["metadata"]["sample"]["date_added"],
        "MD5": analysis["results"]["metadata"]["sample"]["hashes"]["md5"],
        "SHA256": analysis["results"]["metadata"]["sample"]["hashes"]["sha256"],
        "Score": analysis["results"]["summary"]["risk_score"],
        "Result": analysis["results"]["summary"],
        "status": "done",
    }

    risk_score = analysis["results"]["summary"]["risk_score"]
    analysis_context = dict(analysis_info)
    analysis_table = dict(analysis_info)
    malicious = None
    suspicious = None

    if risk_score > 9:
        dbot_score = 3
        malicious = {
            "Vendor": "Symantec Content and Malware Analysis",
            "SHA256": analysis["results"]["metadata"]["sample"]["hashes"]["md5"],
        }
    elif 7 < risk_score < 9:
        dbot_score = 2
        malicious = {
            "Vendor": "Symantec Content and Malware Analysis",
            "SHA256": analysis["results"]["metadata"]["sample"]["hashes"]["md5"],
        }
        suspicious = True
    else:
        dbot_score = 1

    dbot_scores.append(
        {
            "Vendor": "Symantec Content and Malware Analysis",
            "Indicator": analysis["results"]["metadata"]["sample"]["hashes"]["md5"],
            "Type": "file",
            "Score": dbot_score,
            "Malicious": malicious,
            "Suspicious": suspicious,
        }
    )
    context.append(analysis_context)
    table.append(analysis_table)

    entry = {
        "ContentsFormat": formats["json"],
        "Type": entryTypes["note"],
        "Contents": context,
        "ReadableContentsFormat": formats["markdown"],
        "HumanReadable": tableToMarkdown(title, table, removeNull=True),
        "EntryContext": {
            "Symantec.Analysis(val.ID && val.ID == obj.ID)": createContext(context, removeNull=True),
            "DBotScore": createContext(dbot_scores[0], removeNull=True),
        },
    }
    return entry


def create_task(sample_id):
    """
    Args:
        sample_id: smaple id for the url

    Returns:
        the task data
    """
    files = {"sample_id": str(sample_id), "env": demisto.params()["environment"]}
    new_task = http_request("/rapi/tasks", data=files, files=files)
    return new_task


def upload_url(url):
    """
    Args:
        url: url to upload

    Returns:
        an entry
    """
    res = http_request("/rapi/samples/url", data={"url": url}, files={"url": url})
    sample_id = res["results"][0]["samples_url_sample_id"]

    if sample_id == 0:
        raise Exception("Unable to upload file to Symantec Content Analysis - no sample id was returned")

    task_details = create_task(sample_id)["results"][0]
    samples_label = task_details["samples_label"]
    samples_owner = task_details["samples_owner"]
    samples_sample_id = task_details["samples_sample_id"]
    tasks_task_id = task_details["tasks_task_id"]
    detonate_result = {
        "Label": samples_label,
        "Owner": samples_owner,
        "Sample ID": samples_sample_id,
        "ID": tasks_task_id,
        "status": "submitted",
    }
    return {
        "Type": entryTypes["note"],
        "Contents": task_details,
        "ContentsFormat": formats["json"],
        "HumanReadable": tableToMarkdown("Symantec URL Detonation", [detonate_result]),
        "ReadableContentsFormat": formats["markdown"],
        "EntryContext": {"Symantec.Analysis(val.ID && val.ID == obj.ID)": detonate_result},
    }


def upload_file(file_entry):
    """
    Args:
        file_entry: entry_id of the file to upload

    Returns:
        an entry
    """
    with open(demisto.getFilePath(file_entry)["path"], "rb") as file:
        result = http_request("/rapi/samples/basic", data={}, files={"file": file})

    sample_id = result["results"][0]["samples_basic_sample_id"]
    if sample_id == 0:
        raise Exception("Unable to upload file to Symantec Content Analysis - no sample id was returned")

    task_details = create_task(sample_id)["results"][0]
    samples_label = task_details["samples_label"]
    samples_owner = task_details["samples_owner"]
    samples_sample_id = task_details["samples_sample_id"]
    tasks_task_id = task_details["tasks_task_id"]
    detonate_result = {
        "Label": samples_label.replace("hxxp", "http"),
        "Owner": samples_owner,
        "Sample ID": samples_sample_id,
        "ID": tasks_task_id,
        "status": "submitted",
    }
    return {
        "Type": entryTypes["note"],
        "Contents": task_details,
        "ContentsFormat": formats["json"],
        "HumanReadable": tableToMarkdown("Symantec File Detonation", [detonate_result]),
        "ReadableContentsFormat": formats["markdown"],
        "EntryContext": {"Symantec.Analysis(val.ID && val.ID == obj.ID)": detonate_result},
    }


def get_report(task_id):
    """
    Args:
        task_id: task_id to retrieve the report upon

    Returns:
        Report data for a given task_id
    """
    report = http_request(f"/rapi/search/report/{task_id!s}", data={}, files={})
    if "Entry" not in report:
        report["id"] = int(task_id)
        if "sample_type" in report["results"]["metadata"]["sample"]:  # noqa: RET503
            if report["results"]["metadata"]["sample"]["sample_type"] == "url":
                return url_analysis_to_entry("URL Detonation Report", report)
            return file_analysis_to_entry("File Detonation Report", report)
    else:
        return {
            "Type": entryTypes["note"],
            "Contents": {},
            "ContentsFormat": formats["json"],
            "HumanReadable": "Scan in  pending or does not exist",
            "ReadableContentsFormat": formats["markdown"],
            "EntryContext": {
                "Symantec.Analysis(val.ID && val.ID == obj.ID)": {"ID": task_id, "status": "pending/does_not_exist"}
            },
        }


def main():
    """
    EXECUTE INTEGRATION PARAM
    """
    try:
        command = demisto.command()
        LOG(f"command is {command}")
        handle_proxy()
        if command == "test-module":
            demisto.results("ok")
        elif command == "symantec-cma-upload-file":
            demisto.results(upload_file(demisto.args()["file_id"]))
        elif command == "symantec-cma-upload-url":
            demisto.results(upload_url(demisto.args()["url"]))
        elif command == "symantec-cma-get-report":
            demisto.results(get_report(demisto.args()["task_id"]))
        else:
            raise NotImplementedError(f'Command "{command}" is not implemented.')

    except Exception as err:
        return_error(f"An error has occurred in the Symantec Blue Coat Malware Analysis integration: {err!s}")


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