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()