diff --git a/playbooks/VirusTotal_v3_Dynamic_Analysis.json b/playbooks/VirusTotal_v3_Dynamic_Analysis.json new file mode 100755 index 0000000000..dc8fdd2e2b --- /dev/null +++ b/playbooks/VirusTotal_v3_Dynamic_Analysis.json @@ -0,0 +1,599 @@ +{ + "blockly": false, + "blockly_xml": "", + "category": "Dynamic Analysis", + "coa": { + "data": { + "customCode": null, + "description": "Accepts a URL or vault_id and detonates the object in VirusTotal's sandbox. Generates a global report and a per observable sub-report and normalized score. The score can be customized based on a variety of factors.\n\nRef: https://d3fend.mitre.org/technique/d3f:DynamicAnalysis", + "edges": [ + { + "id": "port_0_to_port_2", + "sourceNode": "0", + "sourcePort": "0_out", + "targetNode": "2", + "targetPort": "2_in" + }, + { + "conditions": [ + { + "index": 0 + } + ], + "id": "port_2_to_port_3", + "sourceNode": "2", + "sourcePort": "2_out", + "targetNode": "3", + "targetPort": "3_in" + }, + { + "conditions": [ + { + "index": 1 + } + ], + "id": "port_2_to_port_4", + "sourceNode": "2", + "sourcePort": "2_out", + "targetNode": "4", + "targetPort": "4_in" + }, + { + "id": "port_7_to_port_8", + "sourceNode": "7", + "sourcePort": "7_out", + "targetNode": "8", + "targetPort": "8_in" + }, + { + "id": "port_8_to_port_1", + "sourceNode": "8", + "sourcePort": "8_out", + "targetNode": "1", + "targetPort": "1_in" + }, + { + "id": "port_4_to_port_10", + "sourceNode": "4", + "sourcePort": "4_out", + "targetNode": "10", + "targetPort": "10_in" + }, + { + "id": "port_12_to_port_13", + "sourceNode": "12", + "sourcePort": "12_out", + "targetNode": "13", + "targetPort": "13_in" + }, + { + "id": "port_13_to_port_1", + "sourceNode": "13", + "sourcePort": "13_out", + "targetNode": "1", + "targetPort": "1_in" + }, + { + "conditions": [ + { + "index": 0 + } + ], + "id": "port_10_to_port_11", + "sourceNode": "10", + "sourcePort": "10_out", + "targetNode": "11", + "targetPort": "11_in" + }, + { + "id": "port_6_to_port_7", + "sourceNode": "6", + "sourcePort": "6_out", + "targetNode": "7", + "targetPort": "7_in" + }, + { + "id": "port_3_to_port_15", + "sourceNode": "3", + "sourcePort": "3_out", + "targetNode": "15", + "targetPort": "15_in" + }, + { + "conditions": [ + { + "index": 0 + } + ], + "id": "port_15_to_port_6", + "sourceNode": "15", + "sourcePort": "15_out", + "targetNode": "6", + "targetPort": "6_in" + }, + { + "id": "port_11_to_port_12", + "sourceNode": "11", + "sourcePort": "11_out", + "targetNode": "12", + "targetPort": "12_in" + } + ], + "globalCustomCode": "\n\n\nfrom math import log", + "hash": "c62a83f2b885857ecb0ed931a92f20293e4829a0", + "nodes": { + "0": { + "data": { + "advanced": { + "join": [] + }, + "functionName": "on_start", + "id": "0", + "type": "start" + }, + "errors": {}, + "id": "0", + "type": "start", + "warnings": {}, + "x": 180, + "y": 0 + }, + "1": { + "data": { + "advanced": { + "join": [] + }, + "functionName": "on_finish", + "id": "1", + "type": "end" + }, + "errors": {}, + "id": "1", + "type": "end", + "warnings": {}, + "x": 190, + "y": 1220 + }, + "10": { + "data": { + "advanced": { + "customName": "file detonate filter", + "customNameId": 0, + "description": "Filters successful file detonation results.", + "join": [], + "note": "Filters successful file detonation results." + }, + "conditions": [ + { + "comparisons": [ + { + "conditionIndex": 0, + "op": "==", + "param": "file_detonation:action_result.status", + "value": "success" + } + ], + "conditionIndex": 0, + "customName": "success_vt_query", + "logic": "and" + } + ], + "functionId": 3, + "functionName": "file_detonate_filter", + "id": "10", + "type": "filter" + }, + "errors": {}, + "id": "10", + "type": "filter", + "warnings": {}, + "x": 400, + "y": 506 + }, + "11": { + "data": { + "advanced": { + "customName": "normalize score file", + "customNameId": 0, + "description": "This block uses custom code for normalizing score. Adjust the logic as desired in the documented sections.", + "join": [], + "note": "This block uses custom code for normalizing score. Adjust the logic as desired in the documented sections." + }, + "functionId": 3, + "functionName": "normalize_score_file", + "id": "11", + "inputParameters": [ + "filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.summary" + ], + "outputVariables": [ + "file_score_object", + "scores" + ], + "type": "code" + }, + "errors": {}, + "id": "11", + "type": "code", + "userCode": " # Reference for scores: https://schema.ocsf.io/objects/reputation\n #phantom.debug(\"filtered_result_0_summary: {}\".format(filtered_result_0_summary))\n #phantom.debug(\"filtered_result_0_data: {}\".format(filtered_result_0_data))\n #phantom.debug(\"filtered_result_0_data___scans: {}\".format(filtered_result_0_data___scans))\n #phantom.debug(\"filtered_result_0_data___attributes___category: {}\".format(filtered_result_0_data___attributes___category))\n #phantom.debug(\"vault_id_detonation_result_item_0: {}\".format(vault_id_detonation_result_item_0))\n \n\n score_table = {\n \"0\":\"Unknown\",\n \"1\":\"Very_Safe\",\n \"2\":\"Safe\",\n \"3\":\"Probably_Safe\",\n \"4\":\"Leans_Safe\",\n \"5\":\"May_not_be_Safe\",\n \"6\":\"Exercise_Caution\",\n \"7\":\"Suspicious_or_Risky\",\n \"8\":\"Possibly_Malicious\",\n \"9\":\"Probably_Malicious\",\n \"10\":\"Malicious\"\n }\n \n file_summary_list = filtered_result_0_summary\n normalize_score_file__file_score_object = []\n normalize_score_file__scores = []\n normalize_score_file__categories = []\n \n for summary_data in file_summary_list:\n # Set confidence based on percentage of vendors undetected\n # Reduce the confidence by percentage of vendors undetected.\n vendors = summary_data['harmless'] + summary_data['undetected'] + summary_data['malicious'] + summary_data['suspicious']\n confidence = 100 - int((summary_data['undetected']/vendors) * 100)\n\n # Normalize reputation on a 10 point scale based on number of malicious and suspicious divided by harmless vendors\n # This can be adjusted to include whatever logic is desired.\n suspect = summary_data['malicious'] + summary_data['suspicious']\n # If there are only harmless verdicts and no suspicious entries, set score_id to 1.\n if summary_data['harmless'] and not suspect:\n score_id = 1\n # If there are only harmless verdicts and no suspicious entries, set score_id to 1.\n elif not summary_data['harmless'] and not suspect:\n score_id = 0\n else:\n # customize score calculation as desired\n log_result = log((suspect/vendors) * 100, 100) # log imported from math in global code block\n score_id = int(log_result * 10) + 3\n\n phantom.debug(\"log_result: {}\".format(log_result))\n if score_id > 10:\n score_id = 10\n\n score = score_table[str(score_id)]\n\n normalize_score_file__file_score_object.append({'score': score, 'score_id': score_id, 'confidence': confidence})\n normalize_score_file__scores.append(score)\n #phantom.debug(\"normalize_score_file__file_score_object: {}\".format(normalize_score_file__file_score_object))\n #phantom.debug(\"normalize_score_file__scores: {}\".format(normalize_score_file__scores))\n\n", + "warnings": {}, + "x": 340, + "y": 686 + }, + "12": { + "data": { + "advanced": { + "customName": "format report file", + "customNameId": 0, + "description": "Format a summary table with the information gathered from the playbook.", + "join": [], + "note": "Format a summary table with the information gathered from the playbook." + }, + "functionId": 2, + "functionName": "format_report_file", + "id": "12", + "parameters": [ + "file_detonation:action_result.parameter.vault_id", + "normalize_score_file:custom_function:scores", + "filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.data.*.attributes.meaningful_name", + "filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.data.*.attributes.magic" + ], + "template": "SOAR analyzed file(s) using VirusTotal. The table below shows a summary of the information gathered.\n\n| File | VT Name | VT Decription | Normalized Score | Report Link | Source |\n| --- | --- | --- | --- | --- | --- |\n%%\n| `{0}` | {2} | {3} | {1} | https://www.virustotal.com/gui/file/{0} | VirusTotal v3 |\n%%\n", + "type": "format" + }, + "errors": {}, + "id": "12", + "type": "format", + "warnings": {}, + "x": 340, + "y": 860 + }, + "13": { + "data": { + "advanced": { + "customName": "build file output", + "customNameId": 0, + "description": "This block uses custom code to generate an observable dictionary to output into the observables data path.", + "join": [], + "note": "This block uses custom code to generate an observable dictionary to output into the observables data path." + }, + "functionId": 4, + "functionName": "build_file_output", + "id": "13", + "inputParameters": [ + "filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.parameter.vault_id", + "filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.data", + "normalize_score_file:custom_function:file_score_object" + ], + "outputVariables": [ + "observable_array" + ], + "type": "code" + }, + "errors": {}, + "id": "13", + "type": "code", + "userCode": "\n # Write your custom code here...\n build_file_output__observable_array = []\n # Build file_hash\n for file_hash, file_data, file_object in zip(filtered_result_0_parameter_vault_id, filtered_result_0_data , normalize_score_file__file_score_object):\n observable_object = {\n \"value\": file_hash,\n \"type\": \"hash\",\n \"reputation\": {\n \"score_id\": file_object['score_id'],\n \"score\": file_object['score'],\n \"confidence\": file_object['confidence']\n },\n \"attributes\": {\n \"type\": file_data[0]['attributes'].get('type_extension'),\n \"name\": file_data[0]['attributes'].get('meaningful_name'),\n \"size\": file_data[0]['attributes'].get('size'),\n \"md5\": file_data[0]['attributes'].get('md5'),\n \"sha1\": file_data[0]['attributes'].get('sha1'),\n \"sha256\": file_data[0]['attributes'].get('sha256')\n },\n \"source\": \"VirusTotal v3\",\n \"source_link\": f\"https://www.virustotal.com/gui/file/{file_hash}\"\n }\n \n build_file_output__observable_array.append(observable_object)\n \n", + "warnings": {}, + "x": 340, + "y": 1040 + }, + "15": { + "data": { + "advanced": { + "customName": "url detonate filter", + "customNameId": 0, + "description": "Filters successful url reputation results.", + "join": [], + "note": "Filters successful url reputation results." + }, + "conditions": [ + { + "comparisons": [ + { + "conditionIndex": 0, + "op": "==", + "param": "url_detonation_1:action_result.status", + "value": "success" + } + ], + "conditionIndex": 0, + "customName": "success_url_vt_query", + "logic": "and" + } + ], + "functionId": 4, + "functionName": "url_detonate_filter", + "id": "15", + "type": "filter" + }, + "errors": {}, + "id": "15", + "type": "filter", + "warnings": {}, + "x": 60, + "y": 506 + }, + "2": { + "data": { + "advanced": { + "customName": "input filter", + "customNameId": 0, + "description": "Determine branches based on provided inputs.", + "join": [], + "note": "Determine branches based on provided inputs." + }, + "conditions": [ + { + "comparisons": [ + { + "conditionIndex": 0, + "op": "!=", + "param": "playbook_input:url", + "value": "" + } + ], + "conditionIndex": 0, + "customName": "url_input", + "logic": "and" + }, + { + "comparisons": [ + { + "conditionIndex": 1, + "op": "!=", + "param": "playbook_input:vault_id", + "value": "" + } + ], + "conditionIndex": 1, + "customName": "vauld_id_input", + "logic": "and" + } + ], + "functionId": 1, + "functionName": "input_filter", + "id": "2", + "type": "filter" + }, + "errors": {}, + "id": "2", + "type": "filter", + "warnings": { + "config": [ + "Reconfigure invalid datapath." + ] + }, + "x": 220, + "y": 140 + }, + "3": { + "data": { + "action": "detonate url", + "actionType": "investigate", + "advanced": { + "customName": "url detonation", + "customNameId": 1, + "description": "Queries VirusTotal for information about the provided URL(s)", + "join": [], + "note": "Queries VirusTotal for information about the provided URL(s)" + }, + "connector": "VirusTotal v3", + "connectorConfigs": [ + "virus_total" + ], + "connectorId": "3fe4875d-a4a7-47d3-9ef1-f9e63a6653a4", + "connectorVersion": "v1", + "functionId": 1, + "functionName": "url_detonation_1", + "id": "3", + "parameters": { + "url": "filtered-data:input_filter:condition_1:playbook_input:url" + }, + "requiredParameters": [ + { + "data_type": "string", + "field": "url" + } + ], + "type": "action" + }, + "errors": {}, + "id": "3", + "type": "action", + "userCode": " \n # Write your custom code here...\n\n", + "warnings": {}, + "x": 0, + "y": 328 + }, + "4": { + "data": { + "action": "detonate file", + "actionType": "investigate", + "advanced": { + "customName": "file detonation", + "customNameId": 0, + "description": "Queries VirusTotal for information about the provided vault_id(s)", + "join": [], + "note": "Queries VirusTotal for information about the provided vault_id(s)" + }, + "connector": "VirusTotal v3", + "connectorConfigs": [ + "virus_total" + ], + "connectorId": "3fe4875d-a4a7-47d3-9ef1-f9e63a6653a4", + "connectorVersion": "v1", + "functionId": 1, + "functionName": "file_detonation", + "id": "4", + "parameters": { + "vault_id": "filtered-data:input_filter:condition_2:playbook_input:vault_id" + }, + "requiredParameters": [ + { + "data_type": "string", + "field": "vault_id" + } + ], + "type": "action" + }, + "errors": {}, + "id": "4", + "type": "action", + "userCode": "\n # Write your custom code here...\n \n\n", + "warnings": {}, + "x": 340, + "y": 328 + }, + "6": { + "customCode": null, + "data": { + "advanced": { + "customName": "normalize score url", + "customNameId": 0, + "description": "This block uses custom code for normalizing score. Adjust the logic as desired in the documented sections.", + "join": [], + "note": "This block uses custom code for normalizing score. Adjust the logic as desired in the documented sections." + }, + "functionId": 1, + "functionName": "normalize_score_url", + "id": "6", + "inputParameters": [ + "filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.data.*.attributes.categories", + "filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.summary" + ], + "outputVariables": [ + "url_score_object", + "score", + "categories" + ], + "type": "code" + }, + "errors": {}, + "id": "6", + "type": "code", + "userCode": "\n # Write your custom code here...\n #phantom.debug(\"filtered_result_0_data___attributes_categories: {}\".format(filtered_result_0_data___data_attributes_results___category))\n #phantom.debug(\"filtered_result_0_summary: {}\".format(filtered_result_0_summary))\n #phantom.debug(\"filtered_result_1_data___scans: {}\".format(filtered_result_1_data___scans))\n #phantom.debug(\"url_detonation_result_item_0: {}\".format(url_detonation_result_item_0))\n score_table = {\n \"0\":\"Unknown\",\n \"1\":\"Very_Safe\",\n \"2\":\"Safe\",\n \"3\":\"Probably_Safe\",\n \"4\":\"Leans_Safe\",\n \"5\":\"May_not_be_Safe\",\n \"6\":\"Exercise_Caution\",\n \"7\":\"Suspicious_or_Risky\",\n \"8\":\"Possibly_Malicious\",\n \"9\":\"Probably_Malicious\",\n \"10\":\"Malicious\"\n }\n \n url_categories_list = filtered_result_0_data___attributes_categories\n url_summary_list = filtered_result_0_summary\n normalize_score_url__url_score_object = []\n normalize_score_url__score = []\n normalize_score_url__categories = []\n\n #for category, summary_data in zip(url_categories_list, url_summary_list):\n for category, summary_data in zip(url_categories_list, url_summary_list):\n \n # Set confidence based on percentage of vendors undetected\n # Reduce the confidence by percentage of vendors undetected.\n vendors = summary_data['harmless'] + summary_data['undetected'] + summary_data['malicious'] + summary_data['suspicious']\n confidence = 100 - int((summary_data['undetected']/vendors) * 100)\n \n #phantom.debug(\"vendors: {}\".format(vendors))\n #phantom.debug(\"confidence: {}\".format(confidence))\n\n # Normalize reputation on a 10 point scale based on number of malicious and suspicious divided by harmless vendors\n # This can be adjusted to include whatever logic is desired.\n suspect = summary_data['malicious'] + summary_data['suspicious']\n # If there are only harmless verdicts and no suspicious entries, set score_id to 1.\n if summary_data['harmless'] and not suspect:\n score_id = 1\n else:\n # customize score calculation as desired\n log_result = log((suspect/vendors) * 100, 100) # log imported from math in global code block\n score_id = int(log_result * 10) + 3\n #log_result = (suspect/vendors) # log imported from math in global code block\n #score_id = int(log_result * 100)\n #phantom.debug(\"log_result: {}\".format(log_result))\n \n if score_id > 10:\n score_id = 10\n \n if category != None:\n categories = [cat.lower() for cat in category.values()]\n categories = list(set(categories))\n else:\n categories = []\n \n score = score_table[str(score_id)]\n\n # Attach final object\n normalize_score_url__url_score_object.append({'score': score, 'score_id': score_id, 'confidence': confidence, 'categories': categories})\n normalize_score_url__score.append(score)\n normalize_score_url__categories.append(categories)\n #phantom.debug(\"normalize_score_url__url_score_object: {}\".format(normalize_score_url__url_score_object))\n #phantom.debug(\"normalize_score_url__score: {}\".format(normalize_score_url__score))\n #phantom.debug(\"normalize_score_url__categories: {}\".format(normalize_score_url__categories))\n\n\n", + "warnings": {}, + "x": 0, + "y": 686 + }, + "7": { + "data": { + "advanced": { + "customName": "format report url", + "customNameId": 0, + "description": "Format a summary table with the information gathered from the playbook.", + "join": [], + "note": "Format a summary table with the information gathered from the playbook." + }, + "functionId": 1, + "functionName": "format_report_url", + "id": "7", + "parameters": [ + "filtered-data:input_filter:condition_1:playbook_input:url", + "normalize_score_url:custom_function:score", + "normalize_score_url:custom_function:categories", + "filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.data.*.id" + ], + "template": "SOAR analyzed URL(s) using VirusTotal. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score | Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | https://www.virustotal.com/gui/url/{3} | VirusTotal v3 |\n%%", + "type": "format" + }, + "errors": {}, + "id": "7", + "type": "format", + "userCode": "\n # Write your custom code here...\n\n", + "warnings": {}, + "x": 0, + "y": 864 + }, + "8": { + "data": { + "advanced": { + "customName": "build url output", + "customNameId": 0, + "description": "This block uses custom code to generate an observable dictionary to output into the observables data path.", + "join": [], + "note": "This block uses custom code to generate an observable dictionary to output into the observables data path." + }, + "functionId": 2, + "functionName": "build_url_output", + "id": "8", + "inputParameters": [ + "filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.parameter.url", + "filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.data.*.id", + "normalize_score_url:custom_function:url_score_object" + ], + "outputVariables": [ + "observable_array" + ], + "type": "code" + }, + "errors": {}, + "id": "8", + "type": "code", + "userCode": " from urllib.parse import urlparse\n build_url_output__observable_array = []\n \n # Build URL\n for url, external_id, url_object in zip(filtered_result_0_parameter_url, filtered_result_0_data___id, normalize_score_url__url_score_object):\n parsed_url = urlparse(url)\n phantom.debug(\"parsed_url: {}, url_object: {}\".format(parsed_url, url_object))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"reputation\": {\n \"score_id\": url_object['score_id'],\n \"score\": url_object['score'],\n \"confidence\": url_object['confidence']\n },\n \"attributes\": {\n \"hostname\": parsed_url.hostname,\n \"scheme\": parsed_url.scheme\n },\n \"categories\": url_object['categories'],\n \"source\": \"VirusTotal v3\",\n \"source_link\": f\"https://www.virustotal.com/gui/url/{external_id}\"\n }\n \n if parsed_url.path:\n observable_object['attributes']['path'] = parsed_url.path\n if parsed_url.query:\n observable_object['attributes']['query'] = parsed_url.query\n if parsed_url.port:\n observable_object['attributes']['port'] = parsed_url.port\n \n build_url_output__observable_array.append(observable_object)\n \n \n", + "warnings": {}, + "x": 0, + "y": 1042 + } + }, + "notes": "Inputs: url, vault_id\nInteractions: VirusTotal v3\nActions: detonate url, detonate file\nOutputs: observable, markdown_report" + }, + "input_spec": [ + { + "contains": [ + "url" + ], + "description": "A URL provided for reputation analysis - Virustotal", + "name": "url" + }, + { + "contains": [ + "vault id" + ], + "description": "A file provided for reputation analysis", + "name": "vault_id" + } + ], + "output_spec": [ + { + "contains": [], + "datapaths": [ + "build_url_output:custom_function:observable_array", + "build_file_output:custom_function:observable_array" + ], + "deduplicate": false, + "description": "An array of observable dictionaries with value, type, score, score_id, and categories.", + "metadata": {}, + "name": "observable" + }, + { + "contains": [], + "datapaths": [ + "format_report_file:formatted_data", + "format_report_url:formatted_data" + ], + "deduplicate": false, + "description": "An array of reports. One report per detonation type.", + "metadata": {}, + "name": "markdown_report" + } + ], + "playbook_type": "data", + "python_version": "3", + "schema": "5.0.9", + "version": "6.0.0.114895" + }, + "create_time": "2023-04-12T11:31:47.902551+00:00", + "draft_mode": false, + "labels": [ + "*" + ], + "tags": [ + "url", + "VirusTotal_v3", + "sandbox", + "D3-DA", + "vault_id", + "D3-FA" + ] +} \ No newline at end of file diff --git a/playbooks/VirusTotal_v3_Dynamic_Analysis.png b/playbooks/VirusTotal_v3_Dynamic_Analysis.png new file mode 100644 index 0000000000..a15b277c36 Binary files /dev/null and b/playbooks/VirusTotal_v3_Dynamic_Analysis.png differ diff --git a/playbooks/VirusTotal_v3_Dynamic_Analysis.py b/playbooks/VirusTotal_v3_Dynamic_Analysis.py new file mode 100755 index 0000000000..a6e370baa3 --- /dev/null +++ b/playbooks/VirusTotal_v3_Dynamic_Analysis.py @@ -0,0 +1,597 @@ +""" +Accepts a URL or vault_id and detonates the object in VirusTotal's sandbox. Generates a global report and a per observable sub-report and normalized score. The score can be customized based on a variety of factors.\n\nRef: https://d3fend.mitre.org/technique/d3f:DynamicAnalysis +""" + + +import phantom.rules as phantom +import json +from datetime import datetime, timedelta + + +################################################################################ +## Global Custom Code Start +################################################################################ + + + +from math import log +################################################################################ +## Global Custom Code End +################################################################################ + +@phantom.playbook_block() +def on_start(container): + phantom.debug('on_start() called') + + # call 'input_filter' block + input_filter(container=container) + + return + +@phantom.playbook_block() +def input_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("input_filter() called") + + ################################################################################ + # Determine branches based on provided inputs. + ################################################################################ + + # collect filtered artifact ids and results for 'if' condition 1 + matched_artifacts_1, matched_results_1 = phantom.condition( + container=container, + conditions=[ + ["playbook_input:url", "!=", ""] + ], + name="input_filter:condition_1") + + # call connected blocks if filtered artifacts or results + if matched_artifacts_1 or matched_results_1: + url_detonation_1(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1) + + # collect filtered artifact ids and results for 'if' condition 2 + matched_artifacts_2, matched_results_2 = phantom.condition( + container=container, + conditions=[ + ["playbook_input:vault_id", "!=", ""] + ], + name="input_filter:condition_2") + + # call connected blocks if filtered artifacts or results + if matched_artifacts_2 or matched_results_2: + file_detonation(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_2, filtered_results=matched_results_2) + + return + + +@phantom.playbook_block() +def url_detonation_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("url_detonation_1() called") + + # phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED'))) + + ################################################################################ + # Queries VirusTotal for information about the provided URL(s) + ################################################################################ + + filtered_input_0_url = phantom.collect2(container=container, datapath=["filtered-data:input_filter:condition_1:playbook_input:url"]) + + parameters = [] + + # build parameters list for 'url_detonation_1' call + for filtered_input_0_url_item in filtered_input_0_url: + if filtered_input_0_url_item[0] is not None: + parameters.append({ + "url": filtered_input_0_url_item[0], + }) + + ################################################################################ + ## Custom Code Start + ################################################################################ + + # Write your custom code here... + + ################################################################################ + ## Custom Code End + ################################################################################ + + phantom.act("detonate url", parameters=parameters, name="url_detonation_1", assets=["virus_total"], callback=url_detonate_filter) + + return + + +@phantom.playbook_block() +def file_detonation(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("file_detonation() called") + + # phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED'))) + + ################################################################################ + # Queries VirusTotal for information about the provided vault_id(s) + ################################################################################ + + filtered_input_0_vault_id = phantom.collect2(container=container, datapath=["filtered-data:input_filter:condition_2:playbook_input:vault_id"]) + + parameters = [] + + # build parameters list for 'file_detonation' call + for filtered_input_0_vault_id_item in filtered_input_0_vault_id: + if filtered_input_0_vault_id_item[0] is not None: + parameters.append({ + "vault_id": filtered_input_0_vault_id_item[0], + }) + + ################################################################################ + ## Custom Code Start + ################################################################################ + + # Write your custom code here... + + + ################################################################################ + ## Custom Code End + ################################################################################ + + phantom.act("detonate file", parameters=parameters, name="file_detonation", assets=["virus_total"], callback=file_detonate_filter) + + return + + +@phantom.playbook_block() +def normalize_score_url(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("normalize_score_url() called") + + ################################################################################ + # This block uses custom code for normalizing score. Adjust the logic as desired + # in the documented sections. + ################################################################################ + + filtered_result_0_data_url_detonate_filter = phantom.collect2(container=container, datapath=["filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.data.*.attributes.categories","filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.summary"]) + + filtered_result_0_data___attributes_categories = [item[0] for item in filtered_result_0_data_url_detonate_filter] + filtered_result_0_summary = [item[1] for item in filtered_result_0_data_url_detonate_filter] + + normalize_score_url__url_score_object = None + normalize_score_url__score = None + normalize_score_url__categories = None + + ################################################################################ + ## Custom Code Start + ################################################################################ + + # Write your custom code here... + #phantom.debug("filtered_result_0_data___attributes_categories: {}".format(filtered_result_0_data___data_attributes_results___category)) + #phantom.debug("filtered_result_0_summary: {}".format(filtered_result_0_summary)) + #phantom.debug("filtered_result_1_data___scans: {}".format(filtered_result_1_data___scans)) + #phantom.debug("url_detonation_result_item_0: {}".format(url_detonation_result_item_0)) + score_table = { + "0":"Unknown", + "1":"Very_Safe", + "2":"Safe", + "3":"Probably_Safe", + "4":"Leans_Safe", + "5":"May_not_be_Safe", + "6":"Exercise_Caution", + "7":"Suspicious_or_Risky", + "8":"Possibly_Malicious", + "9":"Probably_Malicious", + "10":"Malicious" + } + + url_categories_list = filtered_result_0_data___attributes_categories + url_summary_list = filtered_result_0_summary + normalize_score_url__url_score_object = [] + normalize_score_url__score = [] + normalize_score_url__categories = [] + + #for category, summary_data in zip(url_categories_list, url_summary_list): + for category, summary_data in zip(url_categories_list, url_summary_list): + + # Set confidence based on percentage of vendors undetected + # Reduce the confidence by percentage of vendors undetected. + vendors = summary_data['harmless'] + summary_data['undetected'] + summary_data['malicious'] + summary_data['suspicious'] + confidence = 100 - int((summary_data['undetected']/vendors) * 100) + + #phantom.debug("vendors: {}".format(vendors)) + #phantom.debug("confidence: {}".format(confidence)) + + # Normalize reputation on a 10 point scale based on number of malicious and suspicious divided by harmless vendors + # This can be adjusted to include whatever logic is desired. + suspect = summary_data['malicious'] + summary_data['suspicious'] + # If there are only harmless verdicts and no suspicious entries, set score_id to 1. + if summary_data['harmless'] and not suspect: + score_id = 1 + else: + # customize score calculation as desired + log_result = log((suspect/vendors) * 100, 100) # log imported from math in global code block + score_id = int(log_result * 10) + 3 + #log_result = (suspect/vendors) # log imported from math in global code block + #score_id = int(log_result * 100) + #phantom.debug("log_result: {}".format(log_result)) + + if score_id > 10: + score_id = 10 + + if category != None: + categories = [cat.lower() for cat in category.values()] + categories = list(set(categories)) + else: + categories = [] + + score = score_table[str(score_id)] + + # Attach final object + normalize_score_url__url_score_object.append({'score': score, 'score_id': score_id, 'confidence': confidence, 'categories': categories}) + normalize_score_url__score.append(score) + normalize_score_url__categories.append(categories) + #phantom.debug("normalize_score_url__url_score_object: {}".format(normalize_score_url__url_score_object)) + #phantom.debug("normalize_score_url__score: {}".format(normalize_score_url__score)) + #phantom.debug("normalize_score_url__categories: {}".format(normalize_score_url__categories)) + + + ################################################################################ + ## Custom Code End + ################################################################################ + + phantom.save_run_data(key="normalize_score_url:url_score_object", value=json.dumps(normalize_score_url__url_score_object)) + phantom.save_run_data(key="normalize_score_url:score", value=json.dumps(normalize_score_url__score)) + phantom.save_run_data(key="normalize_score_url:categories", value=json.dumps(normalize_score_url__categories)) + + format_report_url(container=container) + + return + + +@phantom.playbook_block() +def format_report_url(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("format_report_url() called") + + ################################################################################ + # Format a summary table with the information gathered from the playbook. + ################################################################################ + + template = """SOAR analyzed URL(s) using VirusTotal. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score | Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | https://www.virustotal.com/gui/url/{3} | VirusTotal v3 |\n%%""" + + # parameter list for template variable replacement + parameters = [ + "filtered-data:input_filter:condition_1:playbook_input:url", + "normalize_score_url:custom_function:score", + "normalize_score_url:custom_function:categories", + "filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.data.*.id" + ] + + ################################################################################ + ## Custom Code Start + ################################################################################ + + # Write your custom code here... + + ################################################################################ + ## Custom Code End + ################################################################################ + + phantom.format(container=container, template=template, parameters=parameters, name="format_report_url") + + build_url_output(container=container) + + return + + +@phantom.playbook_block() +def build_url_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("build_url_output() called") + + ################################################################################ + # This block uses custom code to generate an observable dictionary to output into + # the observables data path. + ################################################################################ + + filtered_result_0_data_url_detonate_filter = phantom.collect2(container=container, datapath=["filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.parameter.url","filtered-data:url_detonate_filter:condition_1:url_detonation_1:action_result.data.*.id"]) + normalize_score_url__url_score_object = json.loads(_ if (_ := phantom.get_run_data(key="normalize_score_url:url_score_object")) != "" else "null") # pylint: disable=used-before-assignment + + filtered_result_0_parameter_url = [item[0] for item in filtered_result_0_data_url_detonate_filter] + filtered_result_0_data___id = [item[1] for item in filtered_result_0_data_url_detonate_filter] + + build_url_output__observable_array = None + + ################################################################################ + ## Custom Code Start + ################################################################################ + from urllib.parse import urlparse + build_url_output__observable_array = [] + + # Build URL + for url, external_id, url_object in zip(filtered_result_0_parameter_url, filtered_result_0_data___id, normalize_score_url__url_score_object): + parsed_url = urlparse(url) + phantom.debug("parsed_url: {}, url_object: {}".format(parsed_url, url_object)) + observable_object = { + "value": url, + "type": "url", + "reputation": { + "score_id": url_object['score_id'], + "score": url_object['score'], + "confidence": url_object['confidence'] + }, + "attributes": { + "hostname": parsed_url.hostname, + "scheme": parsed_url.scheme + }, + "categories": url_object['categories'], + "source": "VirusTotal v3", + "source_link": f"https://www.virustotal.com/gui/url/{external_id}" + } + + if parsed_url.path: + observable_object['attributes']['path'] = parsed_url.path + if parsed_url.query: + observable_object['attributes']['query'] = parsed_url.query + if parsed_url.port: + observable_object['attributes']['port'] = parsed_url.port + + build_url_output__observable_array.append(observable_object) + + + ################################################################################ + ## Custom Code End + ################################################################################ + + phantom.save_run_data(key="build_url_output:observable_array", value=json.dumps(build_url_output__observable_array)) + + return + + +@phantom.playbook_block() +def file_detonate_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("file_detonate_filter() called") + + ################################################################################ + # Filters successful file detonation results. + ################################################################################ + + # collect filtered artifact ids and results for 'if' condition 1 + matched_artifacts_1, matched_results_1 = phantom.condition( + container=container, + conditions=[ + ["file_detonation:action_result.status", "==", "success"] + ], + name="file_detonate_filter:condition_1") + + # call connected blocks if filtered artifacts or results + if matched_artifacts_1 or matched_results_1: + normalize_score_file(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1) + + return + + +@phantom.playbook_block() +def normalize_score_file(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("normalize_score_file() called") + + ################################################################################ + # This block uses custom code for normalizing score. Adjust the logic as desired + # in the documented sections. + ################################################################################ + + filtered_result_0_data_file_detonate_filter = phantom.collect2(container=container, datapath=["filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.summary"]) + + filtered_result_0_summary = [item[0] for item in filtered_result_0_data_file_detonate_filter] + + normalize_score_file__file_score_object = None + normalize_score_file__scores = None + + ################################################################################ + ## Custom Code Start + ################################################################################ + # Reference for scores: https://schema.ocsf.io/objects/reputation + #phantom.debug("filtered_result_0_summary: {}".format(filtered_result_0_summary)) + #phantom.debug("filtered_result_0_data: {}".format(filtered_result_0_data)) + #phantom.debug("filtered_result_0_data___scans: {}".format(filtered_result_0_data___scans)) + #phantom.debug("filtered_result_0_data___attributes___category: {}".format(filtered_result_0_data___attributes___category)) + #phantom.debug("vault_id_detonation_result_item_0: {}".format(vault_id_detonation_result_item_0)) + + + score_table = { + "0":"Unknown", + "1":"Very_Safe", + "2":"Safe", + "3":"Probably_Safe", + "4":"Leans_Safe", + "5":"May_not_be_Safe", + "6":"Exercise_Caution", + "7":"Suspicious_or_Risky", + "8":"Possibly_Malicious", + "9":"Probably_Malicious", + "10":"Malicious" + } + + file_summary_list = filtered_result_0_summary + normalize_score_file__file_score_object = [] + normalize_score_file__scores = [] + normalize_score_file__categories = [] + + for summary_data in file_summary_list: + # Set confidence based on percentage of vendors undetected + # Reduce the confidence by percentage of vendors undetected. + vendors = summary_data['harmless'] + summary_data['undetected'] + summary_data['malicious'] + summary_data['suspicious'] + confidence = 100 - int((summary_data['undetected']/vendors) * 100) + + # Normalize reputation on a 10 point scale based on number of malicious and suspicious divided by harmless vendors + # This can be adjusted to include whatever logic is desired. + suspect = summary_data['malicious'] + summary_data['suspicious'] + # If there are only harmless verdicts and no suspicious entries, set score_id to 1. + if summary_data['harmless'] and not suspect: + score_id = 1 + # If there are only harmless verdicts and no suspicious entries, set score_id to 1. + elif not summary_data['harmless'] and not suspect: + score_id = 0 + else: + # customize score calculation as desired + log_result = log((suspect/vendors) * 100, 100) # log imported from math in global code block + score_id = int(log_result * 10) + 3 + + phantom.debug("log_result: {}".format(log_result)) + if score_id > 10: + score_id = 10 + + score = score_table[str(score_id)] + + normalize_score_file__file_score_object.append({'score': score, 'score_id': score_id, 'confidence': confidence}) + normalize_score_file__scores.append(score) + #phantom.debug("normalize_score_file__file_score_object: {}".format(normalize_score_file__file_score_object)) + #phantom.debug("normalize_score_file__scores: {}".format(normalize_score_file__scores)) + + ################################################################################ + ## Custom Code End + ################################################################################ + + phantom.save_run_data(key="normalize_score_file:file_score_object", value=json.dumps(normalize_score_file__file_score_object)) + phantom.save_run_data(key="normalize_score_file:scores", value=json.dumps(normalize_score_file__scores)) + + format_report_file(container=container) + + return + + +@phantom.playbook_block() +def format_report_file(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("format_report_file() called") + + ################################################################################ + # Format a summary table with the information gathered from the playbook. + ################################################################################ + + template = """SOAR analyzed file(s) using VirusTotal. The table below shows a summary of the information gathered.\n\n| File | VT Name | VT Decription | Normalized Score | Report Link | Source |\n| --- | --- | --- | --- | --- | --- |\n%%\n| `{0}` | {2} | {3} | {1} | https://www.virustotal.com/gui/file/{0} | VirusTotal v3 |\n%%\n""" + + # parameter list for template variable replacement + parameters = [ + "file_detonation:action_result.parameter.vault_id", + "normalize_score_file:custom_function:scores", + "filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.data.*.attributes.meaningful_name", + "filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.data.*.attributes.magic" + ] + + ################################################################################ + ## Custom Code Start + ################################################################################ + + # Write your custom code here... + + ################################################################################ + ## Custom Code End + ################################################################################ + + phantom.format(container=container, template=template, parameters=parameters, name="format_report_file") + + build_file_output(container=container) + + return + + +@phantom.playbook_block() +def build_file_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("build_file_output() called") + + ################################################################################ + # This block uses custom code to generate an observable dictionary to output into + # the observables data path. + ################################################################################ + + filtered_result_0_data_file_detonate_filter = phantom.collect2(container=container, datapath=["filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.parameter.vault_id","filtered-data:file_detonate_filter:condition_1:file_detonation:action_result.data"]) + normalize_score_file__file_score_object = json.loads(_ if (_ := phantom.get_run_data(key="normalize_score_file:file_score_object")) != "" else "null") # pylint: disable=used-before-assignment + + filtered_result_0_parameter_vault_id = [item[0] for item in filtered_result_0_data_file_detonate_filter] + filtered_result_0_data = [item[1] for item in filtered_result_0_data_file_detonate_filter] + + build_file_output__observable_array = None + + ################################################################################ + ## Custom Code Start + ################################################################################ + + # Write your custom code here... + build_file_output__observable_array = [] + # Build file_hash + for file_hash, file_data, file_object in zip(filtered_result_0_parameter_vault_id, filtered_result_0_data , normalize_score_file__file_score_object): + observable_object = { + "value": file_hash, + "type": "hash", + "reputation": { + "score_id": file_object['score_id'], + "score": file_object['score'], + "confidence": file_object['confidence'] + }, + "attributes": { + "type": file_data[0]['attributes'].get('type_extension'), + "name": file_data[0]['attributes'].get('meaningful_name'), + "size": file_data[0]['attributes'].get('size'), + "md5": file_data[0]['attributes'].get('md5'), + "sha1": file_data[0]['attributes'].get('sha1'), + "sha256": file_data[0]['attributes'].get('sha256') + }, + "source": "VirusTotal v3", + "source_link": f"https://www.virustotal.com/gui/file/{file_hash}" + } + + build_file_output__observable_array.append(observable_object) + + ################################################################################ + ## Custom Code End + ################################################################################ + + phantom.save_run_data(key="build_file_output:observable_array", value=json.dumps(build_file_output__observable_array)) + + return + + +@phantom.playbook_block() +def url_detonate_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("url_detonate_filter() called") + + ################################################################################ + # Filters successful url reputation results. + ################################################################################ + + # collect filtered artifact ids and results for 'if' condition 1 + matched_artifacts_1, matched_results_1 = phantom.condition( + container=container, + conditions=[ + ["url_detonation_1:action_result.status", "==", "success"] + ], + name="url_detonate_filter:condition_1") + + # call connected blocks if filtered artifacts or results + if matched_artifacts_1 or matched_results_1: + normalize_score_url(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1) + + return + + +@phantom.playbook_block() +def on_finish(container, summary): + phantom.debug("on_finish() called") + + format_report_file = phantom.get_format_data(name="format_report_file") + format_report_url = phantom.get_format_data(name="format_report_url") + build_url_output__observable_array = json.loads(_ if (_ := phantom.get_run_data(key="build_url_output:observable_array")) != "" else "null") # pylint: disable=used-before-assignment + build_file_output__observable_array = json.loads(_ if (_ := phantom.get_run_data(key="build_file_output:observable_array")) != "" else "null") # pylint: disable=used-before-assignment + + observable_combined_value = phantom.concatenate(build_url_output__observable_array, build_file_output__observable_array) + markdown_report_combined_value = phantom.concatenate(format_report_file, format_report_url) + + output = { + "observable": observable_combined_value, + "markdown_report": markdown_report_combined_value, + } + + ################################################################################ + ## Custom Code Start + ################################################################################ + + # Write your custom code here... + + ################################################################################ + ## Custom Code End + ################################################################################ + + phantom.save_playbook_output_data(output=output) + + return \ No newline at end of file diff --git a/playbooks/VirusTotal_v3_Dynamic_Analysis.yml b/playbooks/VirusTotal_v3_Dynamic_Analysis.yml new file mode 100644 index 0000000000..8b514e204c --- /dev/null +++ b/playbooks/VirusTotal_v3_Dynamic_Analysis.yml @@ -0,0 +1,26 @@ +name: VirusTotal V3 Dynamic Analysis +id: 388ed434-a498-4d55-8de4-b2657825cb67 +version: 1 +date: '2023-03-23' +author: Teoderick Contreras, Splunk +type: Investigation +description: "Accepts a url link, domain or vault_id (hash) to be detonated using Virustotal V3 connector." +playbook: VirusTotal_v3_Dynamic_Analysis +how_to_implement: This input playbook requires the Virustotal V3 API connector to be configured. + It is designed to work in conjunction with the Dynamic Attribute Lookup playbook or other playbooks in the same style. +references: [] +app_list: + - virustotal v3 +tags: + platform_tags: + - url + - domain + - sandbox + - ip + - file_hash + - virustotal_v3 + playbook_type: Input + vpe_type: Modern + playbook_fields: [] + product: + - Splunk SOAR