From 526d2e0833c94f3807b101f65dbf0b6b7a7a24d5 Mon Sep 17 00:00:00 2001 From: tccontre Date: Fri, 31 Mar 2023 11:07:19 +0200 Subject: [PATCH 1/4] minor_update_saa_urlscan_playbook --- playbooks/SAA_Dynamic_Analysis.json | 53 +++++++++++++--------- playbooks/SAA_Dynamic_Analysis.py | 15 +++--- playbooks/SAA_Dynamic_Analysis.yml | 5 +- playbooks/UrlScan_IO_Dynamic_Analysis.json | 45 ++++++++++-------- playbooks/UrlScan_IO_Dynamic_Analysis.py | 42 ++++++++--------- playbooks/UrlScan_IO_Dynamic_Analysis.yml | 1 - 6 files changed, 88 insertions(+), 73 deletions(-) diff --git a/playbooks/SAA_Dynamic_Analysis.json b/playbooks/SAA_Dynamic_Analysis.json index 89876939b0..01e18854e9 100644 --- a/playbooks/SAA_Dynamic_Analysis.json +++ b/playbooks/SAA_Dynamic_Analysis.json @@ -4,7 +4,7 @@ "category": "Dynamic Analysis", "coa": { "data": { - "description": "Accepts a URL or File_Hash and does reputation analysis on the objects. 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:IdentifierReputationAnalysis/", + "description": "Accepts a URL or vault_id and does detonation analysis on the objects. 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\n", "edges": [ { "id": "port_0_to_port_2", @@ -170,7 +170,7 @@ "targetPort": "17_in" } ], - "hash": "4431688d2194725decdd567249661ad0b3d71a42", + "hash": "6b4aa256c537bfb23797087a22c09d3e58ee3082", "nodes": { "0": { "data": { @@ -186,7 +186,7 @@ "type": "start", "warnings": {}, "x": 190, - "y": 0 + "y": -2.5579538487363607e-13 }, "1": { "data": { @@ -200,10 +200,10 @@ "errors": {}, "id": "1", "type": "end", - "userCode": "\n # Write your custom code here...\n format_url_report = phantom.get_format_data(name=\"format_url_report\")\n format_file_report = phantom.get_format_data(name=\"format_file_report\")\n markdown_report_combined_value = phantom.concatenate(format_url_report, format_file_report)\n output['markdown_report'] = markdown_report_combined_value\n", + "userCode": "\n # Write your custom code here...\n #phantom.debug(output)\n", "warnings": {}, - "x": 190, - "y": 1804 + "x": 200, + "y": 1800 }, "10": { "data": { @@ -230,7 +230,7 @@ "errors": {}, "id": "10", "type": "format", - "userCode": "\n # Write your custom code here...\n phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name=\"format_report_url\"))\n", + "userCode": "\n # Write your custom code here...\n #phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name=\"format_report_url\"))\n", "warnings": {}, "x": 0, "y": 1448 @@ -260,7 +260,7 @@ "errors": {}, "id": "11", "type": "code", - "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output__observable_array = []\n phantom.debug(playbook_input_url_values)\n # Build URL\n for jobs_id in get_jobid_of_url_detonation_output__jobid:\n \n for url, external_id, url_object in zip(playbook_input_url_values, jobs_id, normalized_job_forensic_report_output__url_score_object):\n parsed_url = urlparse(url)\n phantom.debug(\"url: {} jobs_id:{}\".format(url, external_id))\n phantom.debug(\"parsed_url: {}, url_object: {}\".format(parsed_url, url_object))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\": f\"https://app.twinwave.io/job/{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 phantom.debug(\"build_url_output__observable_array: {}\".format(build_url_output__observable_array))\n", + "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output__observable_array = []\n phantom.debug(playbook_input_url_values)\n # Build URL\n for jobs_id in get_jobid_of_url_detonation_output__jobid:\n \n for url, external_id, url_object in zip(playbook_input_url_values, jobs_id, normalized_job_forensic_report_output__url_score_object):\n parsed_url = urlparse(url)\n phantom.debug(\"url: {} jobs_id:{}\".format(url, external_id))\n phantom.debug(\"parsed_url: {}, url_object: {}\".format(parsed_url, url_object))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\": f\"https://app.twinwave.io/job/{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 #phantom.debug(\"build_url_output__observable_array: {}\".format(build_url_output__observable_array))\n", "warnings": {}, "x": 0, "y": 1626 @@ -401,10 +401,10 @@ "errors": {}, "id": "17", "type": "code", - "userCode": "\n # Write your custom code here...\n score_id =0\n score_table = {\n \"0\":\"Unknown\",\n \"10\":\"Very_Safe\",\n \"20\":\"Safe\",\n \"30\":\"Probably_Safe\",\n \"40\":\"Leans_Safe\",\n \"50\":\"May_not_be_Safe\",\n \"60\":\"Exercise_Caution\",\n \"70\":\"Suspicious_or_Risky\",\n \"80\":\"Possibly_Malicious\",\n \"90\":\"Probably_Malicious\",\n \"100\":\"Malicious\"\n }\n #phantom.debug(\"vault_id: {}\".format(ssa_get_job_forensics_output_result_item_0))\n #phantom.debug(\"DisplayScore: {}\".format(ssa_get_job_forensics_output_result_item_1))\n #phantom.debug(\"Category: {}\".format(ssa_get_job_forensics_output_result_item_2))\n #phantom.debug(\"verdict: {}\".format(ssa_get_job_forensics_output_result_item_3))\n #phantom.debug(\"action_data: {}\".format(ssa_get_job_forensics_output_result_item_4))\n\n \n normalized_job_forensic_report_output_1__file_score_object = []\n normalized_job_forensic_report_output_1__scores = []\n normalized_job_forensic_report_output_1__categories = []\n normalized_job_forensic_report_output_1__confidence = []\n \n ## normalized NoneType value to avoid enumeration failure\n file_detonation_param_list = [(i or \"\") for i in playbook_input_vault_id_values] \n file_detonation_threat_score_list = [(i or 0) for i in saa_get_file_job_forensics_output_result_item_0] \n file_detonation_category_list = [(i or \"\") for i in saa_get_file_job_forensics_output_result_item_1] \n file_detonation_verdict_list = [(i or \"\") for i in saa_get_file_job_forensics_output_result_item_2] \n \n ## get the set() or unique input url parameter.\n \n index_file_dict = {}\n set_file_inputs = set(file_detonation_param_list)\n \n for file_input in set_file_inputs:\n vaultid_list = []\n score_list = []\n display_score_list = []\n category_list = []\n \n ## getting the index of each detonation phase of the url group the result for each url detonation\n file_input_index = [indx for indx, vaultid_val in enumerate(file_detonation_param_list) if vaultid_val == file_input]\n index_file_dict[file_input] = file_input_index\n \n for idx,(_vaultid, _score, _display_score, _category) in enumerate(zip(file_detonation_param_list, file_detonation_verdict_list, file_detonation_threat_score_list, file_detonation_category_list)):\n if _vaultid == file_input and idx in index_file_dict[file_input]:\n vaultid_list.append(_vaultid)\n score_list.append(_score)\n display_score_list.append(_display_score)\n category_list.append(_category)\n \n ## if score_list is empty or it has one element but empty string, lets score it base on confidence score of its engine detonation\n #phantom.debug(\"score_list: {} len: {}\".format(score_list, len(score_list)))\n #phantom.debug(\"category_list: {} len: {}\".format(category_list, len(category_list)))\n confidence_ = list(set(display_score_list))[0]\n categories = list(set(category_list))\n \n if len(score_list) == 0 or (len(set(score_list)) == 1 and score_list[0] == \"\"):\n if confidence_ >= 0 and confidence_ < 10:\n score_id = 0\n elif confidence_ >= 10 and confidence_ < 20:\n score_id = 10\n elif confidence_ >= 20 and confidence_ < 30:\n score_id = 20\n elif confidence_ >= 30 and confidence_ < 40:\n score_id = 30\n elif confidence_ >= 40 and confidence_ < 50:\n score_id = 40\n elif confidence_ >= 50 and confidence_ < 60:\n score_id = 50\n elif confidence_ >= 60 and confidence_ < 70:\n score_id = 60\n elif confidence_ >= 70 and confidence_ < 80:\n score_id = 70\n elif confidence_ >= 80 and confidence_ < 90:\n score_id = 80\n elif confidence_ >= 90 and confidence_ < 100:\n score_id = 90\n elif confidence_ >= 100:\n score_id = 100\n \n score = score_table[str(score_id)]\n \n else:\n score = list(set(score_list))[0]\n \n # Attach final object\n normalized_job_forensic_report_output_1__file_score_object.append({'score': score, 'confidence': confidence_, 'categories': categories})\n normalized_job_forensic_report_output_1__scores.append(score)\n normalized_job_forensic_report_output_1__categories.append(categories)\n normalized_job_forensic_report_output_1__confidence.append(confidence_)\n phantom.debug(\"normalized_job_forensic_report_output_1__file_score_object: {}\".format(normalized_job_forensic_report_output_1__file_score_object))\n phantom.debug(\"normalized_job_forensic_report_output_1__scores: {}\".format(normalized_job_forensic_report_output_1__scores))\n phantom.debug(\"normalized_job_forensic_report_output_1__categories: {}\".format(normalized_job_forensic_report_output_1__categories))\n", + "userCode": "\n # Write your custom code here...\n score_id =0\n score_table = {\n \"0\":\"Unknown\",\n \"10\":\"Very_Safe\",\n \"20\":\"Safe\",\n \"30\":\"Probably_Safe\",\n \"40\":\"Leans_Safe\",\n \"50\":\"May_not_be_Safe\",\n \"60\":\"Exercise_Caution\",\n \"70\":\"Suspicious_or_Risky\",\n \"80\":\"Possibly_Malicious\",\n \"90\":\"Probably_Malicious\",\n \"100\":\"Malicious\"\n }\n #phantom.debug(\"vault_id: {}\".format(ssa_get_job_forensics_output_result_item_0))\n #phantom.debug(\"DisplayScore: {}\".format(ssa_get_job_forensics_output_result_item_1))\n #phantom.debug(\"Category: {}\".format(ssa_get_job_forensics_output_result_item_2))\n #phantom.debug(\"verdict: {}\".format(ssa_get_job_forensics_output_result_item_3))\n #phantom.debug(\"action_data: {}\".format(ssa_get_job_forensics_output_result_item_4))\n\n \n normalized_job_forensic_report_output_1__file_score_object = []\n normalized_job_forensic_report_output_1__scores = []\n normalized_job_forensic_report_output_1__categories = []\n normalized_job_forensic_report_output_1__confidence = []\n \n ## normalized NoneType value to avoid enumeration failure\n file_detonation_param_list = [(i or \"\") for i in playbook_input_vault_id_values] \n file_detonation_threat_score_list = [(i or 0) for i in saa_get_file_job_forensics_output_result_item_0] \n file_detonation_category_list = [(i or \"\") for i in saa_get_file_job_forensics_output_result_item_1] \n file_detonation_verdict_list = [(i or \"\") for i in saa_get_file_job_forensics_output_result_item_2] \n \n ## get the set() or unique input url parameter.\n \n index_file_dict = {}\n set_file_inputs = set(file_detonation_param_list)\n \n for file_input in set_file_inputs:\n vaultid_list = []\n score_list = []\n display_score_list = []\n category_list = []\n \n ## getting the index of each detonation phase of the url group the result for each url detonation\n file_input_index = [indx for indx, vaultid_val in enumerate(file_detonation_param_list) if vaultid_val == file_input]\n index_file_dict[file_input] = file_input_index\n \n for idx,(_vaultid, _score, _display_score, _category) in enumerate(zip(file_detonation_param_list, file_detonation_verdict_list, file_detonation_threat_score_list, file_detonation_category_list)):\n if _vaultid == file_input and idx in index_file_dict[file_input]:\n vaultid_list.append(_vaultid)\n score_list.append(_score)\n display_score_list.append(_display_score)\n category_list.append(_category)\n \n ## if score_list is empty or it has one element but empty string, lets score it base on confidence score of its engine detonation\n #phantom.debug(\"score_list: {} len: {}\".format(score_list, len(score_list)))\n #phantom.debug(\"category_list: {} len: {}\".format(category_list, len(category_list)))\n confidence_ = list(set(display_score_list))[0]\n categories = list(set(category_list))\n \n if len(score_list) == 0 or (len(set(score_list)) == 1 and score_list[0] == \"\"):\n if confidence_ >= 0 and confidence_ < 10:\n score_id = 0\n elif confidence_ >= 10 and confidence_ < 20:\n score_id = 10\n elif confidence_ >= 20 and confidence_ < 30:\n score_id = 20\n elif confidence_ >= 30 and confidence_ < 40:\n score_id = 30\n elif confidence_ >= 40 and confidence_ < 50:\n score_id = 40\n elif confidence_ >= 50 and confidence_ < 60:\n score_id = 50\n elif confidence_ >= 60 and confidence_ < 70:\n score_id = 60\n elif confidence_ >= 70 and confidence_ < 80:\n score_id = 70\n elif confidence_ >= 80 and confidence_ < 90:\n score_id = 80\n elif confidence_ >= 90 and confidence_ < 100:\n score_id = 90\n elif confidence_ >= 100:\n score_id = 100\n \n score = score_table[str(score_id)]\n \n else:\n score = list(set(score_list))[0]\n \n # Attach final object\n normalized_job_forensic_report_output_1__file_score_object.append({'score': score, 'confidence': confidence_, 'categories': categories})\n normalized_job_forensic_report_output_1__scores.append(score)\n normalized_job_forensic_report_output_1__categories.append(categories)\n normalized_job_forensic_report_output_1__confidence.append(confidence_)\n #phantom.debug(\"normalized_job_forensic_report_output_1__file_score_object: {}\".format(normalized_job_forensic_report_output_1__file_score_object))\n #phantom.debug(\"normalized_job_forensic_report_output_1__scores: {}\".format(normalized_job_forensic_report_output_1__scores))\n #phantom.debug(\"normalized_job_forensic_report_output_1__categories: {}\".format(normalized_job_forensic_report_output_1__categories))\n", "warnings": {}, "x": 340, - "y": 1254 + "y": 1260 }, "18": { "data": { @@ -431,10 +431,10 @@ "errors": {}, "id": "18", "type": "format", - "userCode": "\n # Write your custom code here...\n phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name=\"format_report_file\"))\n", + "userCode": "\n # Write your custom code here...\n #phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name=\"format_report_file\"))\n", "warnings": {}, "x": 340, - "y": 1448 + "y": 1440 }, "19": { "data": { @@ -464,7 +464,7 @@ "userCode": "\n # Write your custom code here...\n build_file_output__observable_array = []\n for jobs_id in get_jobid_of_file_detonation_output__jobid:\n for _vault_id, external_id, file_object in zip(playbook_input_vault_id_values, jobs_id, normalized_job_forensic_report_output_1__file_score_object):\n phantom.debug(\"vault: {} id: {}\".format(_vault_id, external_id))\n observable_object = {\n\n \"value\": _vault_id,\n \"type\": \"hash\",\n \"sandbox\": {\n \"score\": file_object['score'],\n \"confidence\": file_object['confidence'],\n\n },\n \"enrichment\": {\n \"provider\": \"Splunk Attack Analyzer\",\n \"type\": \"file\",\n\n },\n \"categories\": file_object['categories'],\n \"source\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\":f\"https://app.twinwave.io/job/{external_id}\"\n }\n build_file_output__observable_array.append(observable_object)\n phantom.debug(\"build_file_output__observable_array: {}\".format(build_file_output__observable_array))\n", "warnings": {}, "x": 340, - "y": 1626 + "y": 1620 }, "2": { "data": { @@ -789,13 +789,13 @@ "errors": {}, "id": "8", "type": "code", - "userCode": "\n # Write your custom code here...\n score_id =0\n score_table = {\n \"0\":\"Unknown\",\n \"10\":\"Very_Safe\",\n \"20\":\"Safe\",\n \"30\":\"Probably_Safe\",\n \"40\":\"Leans_Safe\",\n \"50\":\"May_not_be_Safe\",\n \"60\":\"Exercise_Caution\",\n \"70\":\"Suspicious_or_Risky\",\n \"80\":\"Possibly_Malicious\",\n \"90\":\"Probably_Malicious\",\n \"100\":\"Malicious\"\n }\n #phantom.debug(\"url: {}\".format(ssa_get_job_forensics_output_result_item_0))\n #phantom.debug(\"DisplayScore: {}\".format(ssa_get_job_forensics_output_result_item_1))\n #phantom.debug(\"Category: {}\".format(ssa_get_job_forensics_output_result_item_2))\n #phantom.debug(\"verdict: {}\".format(ssa_get_job_forensics_output_result_item_3))\n #phantom.debug(\"action_data: {}\".format(ssa_get_job_forensics_output_result_item_4))\n\n \n normalized_job_forensic_report_output__url_score_object = []\n normalized_job_forensic_report_output__scores = []\n normalized_job_forensic_report_output__categories = []\n normalized_job_forensic_report_output__confidence = []\n \n ## normalized NoneType value to avoid enumeration failure\n url_detonation_param_list = [(i or \"\") for i in ssa_get_job_forensics_output_result_item_0] \n url_detonation_threat_score_list = [(i or 0) for i in ssa_get_job_forensics_output_result_item_1] \n url_detonation_category_list = [(i or \"\") for i in ssa_get_job_forensics_output_result_item_2] \n url_detonation_verdict_list = [(i or \"\") for i in ssa_get_job_forensics_output_result_item_3] \n \n ## get the set() or unique input url parameter.\n \n index_url_dict = {}\n set_url_inputs = set(url_detonation_param_list)\n \n for url_input in set_url_inputs:\n url_list = []\n score_list = []\n display_score_list = []\n category_list = []\n \n ## getting the index of each detonation phase of the url group the result for each url detonation\n url_input_index = [indx for indx, url_val in enumerate(url_detonation_param_list) if url_val == url_input]\n index_url_dict[url_input] = url_input_index\n\n for idx,(_url, _score, _display_score, _category) in enumerate(zip(url_detonation_param_list, url_detonation_verdict_list, url_detonation_threat_score_list, url_detonation_category_list)):\n if _url == url_input and idx in index_url_dict[url_input]:\n url_list.append(_url)\n score_list.append(_score)\n display_score_list.append(_display_score)\n category_list.append(_category)\n \n ## if score_list is empty or it has one element but empty string, lets score it base on confidence score of its engine detonation\n #phantom.debug(\"score_list: {} len: {}\".format(score_list, len(score_list)))\n #phantom.debug(\"category_list: {} len: {}\".format(category_list, len(category_list)))\n confidence_ = list(set(display_score_list))[0]\n categories = list(set(category_list))\n\n if len(score_list) == 0 or (len(set(score_list)) == 1 and score_list[0] == \"\"):\n if confidence_ >= 0 and confidence_ < 10:\n score_id = 0\n elif confidence_ >= 10 and confidence_ < 20:\n score_id = 10\n elif confidence_ >= 20 and confidence_ < 30:\n score_id = 20\n elif confidence_ >= 30 and confidence_ < 40:\n score_id = 30\n elif confidence_ >= 40 and confidence_ < 50:\n score_id = 40\n elif confidence_ >= 50 and confidence_ < 60:\n score_id = 50\n elif confidence_ >= 60 and confidence_ < 70:\n score_id = 60\n elif confidence_ >= 70 and confidence_ < 80:\n score_id = 70\n elif confidence_ >= 80 and confidence_ < 90:\n score_id = 80\n elif confidence_ >= 90 and confidence_ < 100:\n score_id = 90\n elif confidence_ >= 100:\n score_id = 100\n \n score = score_table[str(score_id)]\n \n else:\n score = list(set(score_list))[0]\n \n # Attach final object\n normalized_job_forensic_report_output__url_score_object.append({'score': score, 'confidence': confidence_, 'categories': categories})\n normalized_job_forensic_report_output__scores.append(score)\n normalized_job_forensic_report_output__categories.append(categories)\n normalized_job_forensic_report_output__confidence.append(confidence_)\n phantom.debug(\"normalized_job_forensic_report_output__url_score_object: {}\".format(normalized_job_forensic_report_output__url_score_object))\n phantom.debug(\"normalized_job_forensic_report_output__categories: {}\".format(normalized_job_forensic_report_output__categories))\n phantom.debug(\"normalized_job_forensic_report_output__confidence: {}\".format(normalized_job_forensic_report_output__confidence))\n\n", + "userCode": "\n # Write your custom code here...\n score_id =0\n score_table = {\n \"0\":\"Unknown\",\n \"10\":\"Very_Safe\",\n \"20\":\"Safe\",\n \"30\":\"Probably_Safe\",\n \"40\":\"Leans_Safe\",\n \"50\":\"May_not_be_Safe\",\n \"60\":\"Exercise_Caution\",\n \"70\":\"Suspicious_or_Risky\",\n \"80\":\"Possibly_Malicious\",\n \"90\":\"Probably_Malicious\",\n \"100\":\"Malicious\"\n }\n #phantom.debug(\"url: {}\".format(ssa_get_job_forensics_output_result_item_0))\n #phantom.debug(\"DisplayScore: {}\".format(ssa_get_job_forensics_output_result_item_1))\n #phantom.debug(\"Category: {}\".format(ssa_get_job_forensics_output_result_item_2))\n #phantom.debug(\"verdict: {}\".format(ssa_get_job_forensics_output_result_item_3))\n #phantom.debug(\"action_data: {}\".format(ssa_get_job_forensics_output_result_item_4))\n\n \n normalized_job_forensic_report_output__url_score_object = []\n normalized_job_forensic_report_output__scores = []\n normalized_job_forensic_report_output__categories = []\n normalized_job_forensic_report_output__confidence = []\n \n ## normalized NoneType value to avoid enumeration failure\n url_detonation_param_list = [(i or \"\") for i in ssa_get_job_forensics_output_result_item_0] \n url_detonation_threat_score_list = [(i or 0) for i in ssa_get_job_forensics_output_result_item_1] \n url_detonation_category_list = [(i or \"\") for i in ssa_get_job_forensics_output_result_item_2] \n url_detonation_verdict_list = [(i or \"\") for i in ssa_get_job_forensics_output_result_item_3] \n \n ## get the set() or unique input url parameter.\n \n index_url_dict = {}\n set_url_inputs = set(url_detonation_param_list)\n \n for url_input in set_url_inputs:\n url_list = []\n score_list = []\n display_score_list = []\n category_list = []\n \n ## getting the index of each detonation phase of the url group the result for each url detonation\n url_input_index = [indx for indx, url_val in enumerate(url_detonation_param_list) if url_val == url_input]\n index_url_dict[url_input] = url_input_index\n\n for idx,(_url, _score, _display_score, _category) in enumerate(zip(url_detonation_param_list, url_detonation_verdict_list, url_detonation_threat_score_list, url_detonation_category_list)):\n if _url == url_input and idx in index_url_dict[url_input]:\n url_list.append(_url)\n score_list.append(_score)\n display_score_list.append(_display_score)\n category_list.append(_category)\n \n ## if score_list is empty or it has one element but empty string, lets score it base on confidence score of its engine detonation\n #phantom.debug(\"score_list: {} len: {}\".format(score_list, len(score_list)))\n #phantom.debug(\"category_list: {} len: {}\".format(category_list, len(category_list)))\n confidence_ = list(set(display_score_list))[0]\n categories = list(set(category_list))\n\n if len(score_list) == 0 or (len(set(score_list)) == 1 and score_list[0] == \"\"):\n if confidence_ >= 0 and confidence_ < 10:\n score_id = 0\n elif confidence_ >= 10 and confidence_ < 20:\n score_id = 10\n elif confidence_ >= 20 and confidence_ < 30:\n score_id = 20\n elif confidence_ >= 30 and confidence_ < 40:\n score_id = 30\n elif confidence_ >= 40 and confidence_ < 50:\n score_id = 40\n elif confidence_ >= 50 and confidence_ < 60:\n score_id = 50\n elif confidence_ >= 60 and confidence_ < 70:\n score_id = 60\n elif confidence_ >= 70 and confidence_ < 80:\n score_id = 70\n elif confidence_ >= 80 and confidence_ < 90:\n score_id = 80\n elif confidence_ >= 90 and confidence_ < 100:\n score_id = 90\n elif confidence_ >= 100:\n score_id = 100\n \n score = score_table[str(score_id)]\n \n else:\n score = list(set(score_list))[0]\n \n # Attach final object\n normalized_job_forensic_report_output__url_score_object.append({'score': score, 'confidence': confidence_, 'categories': categories})\n normalized_job_forensic_report_output__scores.append(score)\n normalized_job_forensic_report_output__categories.append(categories)\n normalized_job_forensic_report_output__confidence.append(confidence_)\n #phantom.debug(\"normalized_job_forensic_report_output__url_score_object: {}\".format(normalized_job_forensic_report_output__url_score_object))\n #phantom.debug(\"normalized_job_forensic_report_output__categories: {}\".format(normalized_job_forensic_report_output__categories))\n #phantom.debug(\"normalized_job_forensic_report_output__confidence: {}\".format(normalized_job_forensic_report_output__confidence))\n\n", "warnings": {}, "x": 0, "y": 1254 } }, - "notes": "Inputs: url, file_hash\nInteractions: Splunk Attack Analyzer\nActions: url detonation, , file detonation\nOutputs: report, observables" + "notes": "Inputs: url, vault_id\nInteractions: Splunk Attack Analyzer\nActions: url detonation, , file detonation\nOutputs: report, observables" }, "input_spec": [ { @@ -824,6 +824,17 @@ "description": "An array of observable dictionaries with value, type, score, score_id, and categories.", "metadata": {}, "name": "observable" + }, + { + "contains": [], + "datapaths": [ + "format_url_report:formatted_data", + "format_file_report:formatted_data" + ], + "deduplicate": false, + "description": "a report contains value, score, confidence and categories", + "metadata": {}, + "name": "report" } ], "playbook_type": "data", @@ -831,20 +842,18 @@ "schema": "5.0.8", "version": "5.5.0.108488" }, - "create_time": "2023-03-27T09:04:31.268860+00:00", + "create_time": "2023-03-31T08:46:27.522787+00:00", "draft_mode": false, "labels": [ "*" ], "tags": [ "url", - "file_hash", - "D3-IPRA", - "D3-FHRA", - "D3-DNRA", - "D3-URA", "ip", "domain", - "sandbox" + "sandbox", + "D3-DA", + "vault_id", + "splunk_attack_analyzer" ] } \ No newline at end of file diff --git a/playbooks/SAA_Dynamic_Analysis.py b/playbooks/SAA_Dynamic_Analysis.py index dd6127b984..884515f0f2 100644 --- a/playbooks/SAA_Dynamic_Analysis.py +++ b/playbooks/SAA_Dynamic_Analysis.py @@ -1,5 +1,5 @@ """ -Accepts a URL or File_Hash and does reputation analysis on the objects. 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:IdentifierReputationAnalysis/ +Accepts a URL or vault_id and does detonation analysis on the objects. 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\n """ @@ -787,7 +787,7 @@ def build_file_output(action=None, success=None, container=None, results=None, h "source_link":f"https://app.twinwave.io/job/{external_id}" } build_file_output__observable_array.append(observable_object) - #phantom.debug("build_file_output__observable_array: {}".format(build_file_output__observable_array)) + phantom.debug("build_file_output__observable_array: {}".format(build_file_output__observable_array)) ################################################################################ ## Custom Code End ################################################################################ @@ -839,13 +839,17 @@ def filter_6(action=None, success=None, container=None, results=None, handle=Non def on_finish(container, summary): phantom.debug("on_finish() called") + format_url_report = phantom.get_format_data(name="format_url_report") + format_file_report = phantom.get_format_data(name="format_file_report") 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) + report_combined_value = phantom.concatenate(format_url_report, format_file_report) output = { "observable": observable_combined_value, + "report": report_combined_value, } ################################################################################ @@ -853,14 +857,11 @@ def on_finish(container, summary): ################################################################################ # Write your custom code here... - format_url_report = phantom.get_format_data(name="format_url_report") - format_file_report = phantom.get_format_data(name="format_file_report") - markdown_report_combined_value = phantom.concatenate(format_url_report, format_file_report) - output['markdown_report'] = markdown_report_combined_value + #phantom.debug(output) ################################################################################ ## Custom Code End ################################################################################ phantom.save_playbook_output_data(output=output) - return + return \ No newline at end of file diff --git a/playbooks/SAA_Dynamic_Analysis.yml b/playbooks/SAA_Dynamic_Analysis.yml index aaeb0e971f..2c839210a3 100644 --- a/playbooks/SAA_Dynamic_Analysis.yml +++ b/playbooks/SAA_Dynamic_Analysis.yml @@ -9,15 +9,14 @@ playbook: SAA_Dynamic_Analysis how_to_implement: This input playbook requires the SAA 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: - - CrowdStrike OAuth API + - Splunk Attack Analyzer API tags: platform_tags: - - reputation - url - domain - sandbox - ip - - file_hash + - vault_id playbook_type: Input vpe_type: Modern playbook_fields: [] diff --git a/playbooks/UrlScan_IO_Dynamic_Analysis.json b/playbooks/UrlScan_IO_Dynamic_Analysis.json index 259106649b..a4e7507f4e 100644 --- a/playbooks/UrlScan_IO_Dynamic_Analysis.json +++ b/playbooks/UrlScan_IO_Dynamic_Analysis.json @@ -4,7 +4,7 @@ "category": "Dynamic Analysis", "coa": { "data": { - "description": "Accepts a URL for detonation analysis on the objects. 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:IdentifierReputationAnalysis/", + "description": "Accepts a URL for detonation analysis on the objects. 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\n", "edges": [ { "id": "port_0_to_port_2", @@ -111,7 +111,7 @@ "targetPort": "5_in" } ], - "hash": "662e0c1ead292892eb76490c42919b482032f0a2", + "hash": "6579c959dc60d4d44f4d7aee29bd67897a8b913b", "nodes": { "0": { "data": { @@ -127,7 +127,7 @@ "type": "start", "warnings": {}, "x": 190, - "y": -1.2789769243681803e-13 + "y": -3.197442310920451e-13 }, "1": { "data": { @@ -141,7 +141,7 @@ "errors": {}, "id": "1", "type": "end", - "userCode": "\n # Write your custom code here...\n no_error_code_format_report_url = phantom.get_format_data(name=\"no_error_code_format_report_url\")\n error_code_format_report_url = phantom.get_format_data(name=\"error_code_format_report_url\")\n markdown_report_combined_value = phantom.concatenate(no_error_code_format_report_url, error_code_format_report_url)\n output['markdown_report'] = markdown_report_combined_value\n", + "userCode": "\n # Write your custom code here...\n #phantom.debug(output)\n", "warnings": {}, "x": 200, "y": 1440 @@ -171,7 +171,7 @@ "errors": {}, "id": "10", "type": "code", - "userCode": "\n # Write your custom code here...\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output_with_error_code__observable_array = []\n\n # Build URL\n url_scan_io_task_reporturl = [str(i or 'no report url') for i in url_reputation_result_item_1]\n url_scan_io_parameter_url = [str(i or '') for i in url_reputation_parameter_url]\n url_scan_io_url_score_object = normalize_score_url_with_error_code__url_score_object\n \n phantom.debug(\"url_reputation_parameter_url: {}\".format(url_reputation_parameter_url))\n #phantom.debug(\"url_reputation_result_item_1: {}\".format(url_reputation_result_item_1))\n #phantom.debug(\"normalize_score_url_with_error_code__url_score_object: {}\".format(normalize_score_url_with_error_code__url_score_object))\n for url, external_id, url_object in zip(url_scan_io_parameter_url, url_scan_io_task_reporturl, url_scan_io_url_score_object):\n parsed_url = urlparse(url)\n phantom.debug(\"{} {} {} parsed_url: {}\".format(url, external_id, url_object, parsed_url))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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 \"description\" :url_object['description'],\n \"source\": \"urlscan.io\",\n \"source_link\": f\"{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 \n build_url_output_with_error_code__observable_array.append(observable_object)\n phantom.debug(\"build_url_output_with_error_code__observable_array: {}\".format(build_url_output_with_error_code__observable_array))\n", + "userCode": "\n # Write your custom code here...\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output_with_error_code__observable_array = []\n\n # Build URL\n url_scan_io_task_reporturl = [str(i or 'no report url') for i in url_reputation_result_item_1]\n url_scan_io_parameter_url = [str(i or '') for i in url_reputation_parameter_url]\n url_scan_io_url_score_object = normalize_score_url_with_error_code__url_score_object\n \n #phantom.debug(\"url_reputation_parameter_url: {}\".format(url_reputation_parameter_url))\n #phantom.debug(\"url_reputation_result_item_1: {}\".format(url_reputation_result_item_1))\n #phantom.debug(\"normalize_score_url_with_error_code__url_score_object: {}\".format(normalize_score_url_with_error_code__url_score_object))\n for url, external_id, url_object in zip(url_scan_io_parameter_url, url_scan_io_task_reporturl, url_scan_io_url_score_object):\n parsed_url = urlparse(url)\n #phantom.debug(\"{} {} {} parsed_url: {}\".format(url, external_id, url_object, parsed_url))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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 \"description\" :url_object['description'],\n \"source\": \"urlscan.io\",\n \"source_link\": f\"{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 \n build_url_output_with_error_code__observable_array.append(observable_object)\n #phantom.debug(\"build_url_output_with_error_code__observable_array: {}\".format(build_url_output_with_error_code__observable_array))\n", "warnings": {}, "x": 340, "y": 1260 @@ -201,7 +201,7 @@ "errors": {}, "id": "11", "type": "code", - "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output_with_no_error_code__observable_array = []\n\n # Build URL\n phantom.debug(url_reputation_parameter_url)\n for url, external_id, url_object in zip(url_reputation_parameter_url, url_reputation_result_item_1, normalize_score_url_with_no_error_code__url_score_object):\n parsed_url = urlparse(url)\n phantom.debug(\"{} {} {} parsed_url: {}\".format(url, external_id, url_object, parsed_url))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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 \"description\" :url_object['description'],\n \"source\": \"urlscan.io\",\n \"source_link\": f\"{external_id}\"\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_with_no_error_code__observable_array.append(observable_object)\n phantom.debug(\"build_url_output_with_no_error_code__observable_array: {}\".format(build_url_output_with_no_error_code__observable_array))\n", + "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output_with_no_error_code__observable_array = []\n\n # Build URL\n #phantom.debug(url_reputation_parameter_url)\n for url, external_id, url_object in zip(url_reputation_parameter_url, url_reputation_result_item_1, normalize_score_url_with_no_error_code__url_score_object):\n parsed_url = urlparse(url)\n #phantom.debug(\"{} {} {} parsed_url: {}\".format(url, external_id, url_object, parsed_url))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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 \"description\" :url_object['description'],\n \"source\": \"urlscan.io\",\n \"source_link\": f\"{external_id}\"\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_with_no_error_code__observable_array.append(observable_object)\n #phantom.debug(\"build_url_output_with_no_error_code__observable_array: {}\".format(build_url_output_with_no_error_code__observable_array))\n", "warnings": {}, "x": 0, "y": 1260 @@ -358,7 +358,7 @@ "errors": {}, "id": "5", "type": "code", - "userCode": "\n # Write your custom code here...\n url_reputation_data_message = [str(i or '') for i in url_reputation_result_item_1] \n url_reputation_data_status = [str(i or '') for i in url_reputation_result_item_2] \n url_reputation_data_description = [str(i or '') for i in url_reputation_result_item_3] \n\n \n normalize_score_url_with_error_code__url_score_object = []\n normalize_score_url_with_error_code__scores = []\n normalize_score_url_with_error_code__categories = []\n\n \n #phantom.debug(\"url_reputation_parameter_url: {}\".format(url_reputation_parameter_url))\n #phantom.debug(\"url_reputation_data_message: {}\".format(url_reputation_data_message))\n #phantom.debug(\"url_reputation_data_status: {}\".format(url_reputation_data_status))\n #phantom.debug(\"url_reputation_data_description: {}\".format(url_reputation_data_description))\n\n \n\n \n urlscan_score_table = {\n \"0\":\"Legitimate\",\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 \"error_code_query\" : \"error code return, check the error code descriptions\"\n\n }\n \n url_scan_io_error_code ={\n \"blacklist\" : \"Blacklisted URL or Domain\",\n \"spam\" : \"Spammy URL or Domain\",\n \"invalid_hostname\" : \"Invalid Hostname URL or Domain\",\n \"missing_url\" : \"Missing URL OR Domain Property\",\n \"auth\" : \"HTTP basic auth information\",\n \"not_be_resolved\" : \"Non-resolvable hostname (A, AAAA, CNAME)\"\n }\n ## URLSCAN.io return error code especially if the url or domain was already in their blacklist database. \n ## below are the common error code message base on their\n ## - \"Blacklisted domains and URLs\" : requested to be blacklisted by their respective owners.\n ## - \"Spammy submissions\" : of URLs known to be used only for spamming this service.\n ## - \"Invalid hostnames\" : or invalid protocol schemes (FTP etc).\n ## - \"Missing URL property\" : ... yes, it does happen.\n ## - \"Contains HTTP basic auth information\" : ... yes, that happens as well.\n ## - \"Non-resolvable hostnames (A, AAAA, CNAME)\" : which we will not even try to scan.\n \n blank_result = \"--\"\n category = \"\"\n score = \"\"\n error_message = \"\"\n \n ## check if there is error code return upon URL reputation query\n \n for url_descp in url_reputation_data_description:\n \n for key, value in url_scan_io_error_code.items():\n if key.replace(\"_\",\" \").lower() in url_descp.lower():\n error_message = url_scan_io_error_code[key]\n \n # Attach final object\n normalize_score_url_with_error_code__categories.append(error_message) \n normalize_score_url_with_error_code__url_score_object.append({'score': urlscan_score_table['error_code_query'], 'confidence':\"\", 'score_id': \"\", \"malicious_tag_verdicts\": \"\", 'categories': error_message, \"description\": url_descp})\n normalize_score_url_with_error_code__scores.append(urlscan_score_table['error_code_query'])\n \n phantom.debug(\"normalize_score_url_with_error_code__url_score_object: {}\".format(normalize_score_url_with_error_code__url_score_object))\n phantom.debug(\"normalize_score_url_with_error_code__scores: {}\".format(normalize_score_url_with_error_code__scores))\n phantom.debug(\"normalize_score_url_with_error_code__categories: {}\".format(normalize_score_url_with_error_code__categories))\n \n \n", + "userCode": "\n # Write your custom code here...\n url_reputation_data_message = [str(i or '') for i in url_reputation_result_item_1] \n url_reputation_data_status = [str(i or '') for i in url_reputation_result_item_2] \n url_reputation_data_description = [str(i or '') for i in url_reputation_result_item_3] \n\n \n normalize_score_url_with_error_code__url_score_object = []\n normalize_score_url_with_error_code__scores = []\n normalize_score_url_with_error_code__categories = []\n\n \n #phantom.debug(\"url_reputation_parameter_url: {}\".format(url_reputation_parameter_url))\n #phantom.debug(\"url_reputation_data_message: {}\".format(url_reputation_data_message))\n #phantom.debug(\"url_reputation_data_status: {}\".format(url_reputation_data_status))\n #phantom.debug(\"url_reputation_data_description: {}\".format(url_reputation_data_description))\n\n \n\n \n urlscan_score_table = {\n \"0\":\"Legitimate\",\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 \"error_code_query\" : \"error code return, check the error code descriptions\"\n\n }\n \n url_scan_io_error_code ={\n \"blacklist\" : \"Blacklisted URL or Domain\",\n \"spam\" : \"Spammy URL or Domain\",\n \"invalid_hostname\" : \"Invalid Hostname URL or Domain\",\n \"missing_url\" : \"Missing URL OR Domain Property\",\n \"auth\" : \"HTTP basic auth information\",\n \"not_be_resolved\" : \"Non-resolvable hostname (A, AAAA, CNAME)\"\n }\n ## URLSCAN.io return error code especially if the url or domain was already in their blacklist database. \n ## below are the common error code message base on their\n ## - \"Blacklisted domains and URLs\" : requested to be blacklisted by their respective owners.\n ## - \"Spammy submissions\" : of URLs known to be used only for spamming this service.\n ## - \"Invalid hostnames\" : or invalid protocol schemes (FTP etc).\n ## - \"Missing URL property\" : ... yes, it does happen.\n ## - \"Contains HTTP basic auth information\" : ... yes, that happens as well.\n ## - \"Non-resolvable hostnames (A, AAAA, CNAME)\" : which we will not even try to scan.\n \n blank_result = \"--\"\n category = \"\"\n score = \"\"\n error_message = \"\"\n \n ## check if there is error code return upon URL reputation query\n \n for url_descp in url_reputation_data_description:\n \n for key, value in url_scan_io_error_code.items():\n if key.replace(\"_\",\" \").lower() in url_descp.lower():\n error_message = url_scan_io_error_code[key]\n \n # Attach final object\n normalize_score_url_with_error_code__categories.append(error_message) \n normalize_score_url_with_error_code__url_score_object.append({'score': urlscan_score_table['error_code_query'], 'confidence':\"\", 'score_id': \"\", \"malicious_tag_verdicts\": \"\", 'categories': error_message, \"description\": url_descp})\n normalize_score_url_with_error_code__scores.append(urlscan_score_table['error_code_query'])\n \n #phantom.debug(\"normalize_score_url_with_error_code__url_score_object: {}\".format(normalize_score_url_with_error_code__url_score_object))\n #phantom.debug(\"normalize_score_url_with_error_code__scores: {}\".format(normalize_score_url_with_error_code__scores))\n #phantom.debug(\"normalize_score_url_with_error_code__categories: {}\".format(normalize_score_url_with_error_code__categories))\n \n \n", "warnings": {}, "x": 340, "y": 860 @@ -388,7 +388,7 @@ "errors": {}, "id": "6", "type": "format", - "userCode": "\n # Write your custom code here...\n phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name=\"error_code_format_report_url\"))\n", + "userCode": "\n # Write your custom code here...\n #phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name=\"error_code_format_report_url\"))\n", "warnings": {}, "x": 340, "y": 1060 @@ -471,7 +471,7 @@ "errors": {}, "id": "8", "type": "code", - "userCode": "\n # Write your custom code here...\n \n normalize_score_url_with_no_error_code__url_score_object = []\n normalize_score_url_with_no_error_code__scores = []\n normalize_score_url_with_no_error_code__categories = []\n \n url_reputation_verdicts_overall_dict = [(i or {}) for i in url_reputation_result_item_0] \n url_reputation_verdicts_urlscan_dict = [(i or {}) for i in url_reputation_result_item_1] \n url_reputation_verdicts_engine_dict = [(i or {}) for i in url_reputation_result_item_2] \n url_reputation_verdicts_community_dict = [(i or {}) for i in url_reputation_result_item_3] \n \n #phantom.debug(\"url_reputation_verdicts_overall_dict: {}\".format(url_reputation_verdicts_overall_dict))\n #phantom.debug(\"url_reputation_verdicts_urlscan_dict: {}\".format(url_reputation_verdicts_urlscan_dict))\n #phantom.debug(\"url_reputation_verdicts_engine_dict: {}\".format(url_reputation_verdicts_engine_dict))\n #phantom.debug(\"url_reputation_verdicts_community_dict: {}\".format(url_reputation_verdicts_community_dict))\n \n \n urlscan_score_table = {\n \"0\":\"Legitimate\",\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 ## if there is no error code, urlscan.io will continue to detonate the URL and query scores in several verdicts object\n \n ## Normalize reputation on a -100 (legitimate) to 100 point scale based on number of malicious and suspicious divided by different urlscan.io verdict objects.\n ## This can be adjusted to include whatever logic is desired.\n \n for i in range(0,len(url_reputation_verdicts_overall_dict)):\n if url_reputation_verdicts_overall_dict[i] != {} or url_reputation_verdicts_urlscan_dict[i] != {} or url_reputation_verdicts_engine_dict[i] != {} or url_reputation_verdicts_community_dict[i] != {}:\n summary_score = url_reputation_verdicts_overall_dict[i]['score'] + url_reputation_verdicts_urlscan_dict[i]['score'] + url_reputation_verdicts_engine_dict[i]['score'] + url_reputation_verdicts_community_dict[i]['score']\n summary_malicious_verdicts = url_reputation_verdicts_overall_dict[i]['malicious'] or url_reputation_verdicts_urlscan_dict[i]['malicious'] or url_reputation_verdicts_engine_dict[i]['malicious'] or url_reputation_verdicts_community_dict[i]['malicious']\n summary_of_malicious_tag = int(url_reputation_verdicts_overall_dict[i]['malicious']) + int(url_reputation_verdicts_urlscan_dict[i]['malicious']) + int(url_reputation_verdicts_engine_dict[i]['malicious']) + int(url_reputation_verdicts_community_dict[i]['malicious'])\n summary_categories = url_reputation_verdicts_overall_dict[i]['categories'] + url_reputation_verdicts_urlscan_dict[i]['categories'] + url_reputation_verdicts_engine_dict[i]['categories'] + url_reputation_verdicts_community_dict[i]['categories']\n \n \n ## customized score id calculation\n \n log_result = (summary_score/4) # avg score from different urlscan.io score object (engine_score, overall_verdict_score, urlscan_verdicts_score and community score)\n score_id = int(log_result) \n \n\n if score_id < -50:\n score_id = \"0\"\n elif score_id < 0 and score_id >= -50:\n score_id = \"1\"\n elif score_id >= 0 and score_id <= 10:\n score_id = \"2\"\n elif score_id > 10 and score_id <= 20:\n score_id = \"3\"\n elif score_id > 20 and score_id <= 30:\n score_id = \"4\"\n elif score_id > 30 and score_id <= 40:\n score_id = \"5\"\n elif score_id > 40 and score_id <= 50:\n score_id = \"6\"\n elif score_id > 50 and score_id <= 60:\n score_id = \"7\"\n elif score_id > 70 and score_id <= 80:\n score_id = \"8\"\n elif score_id > 80 and score_id <= 90:\n score_id = \"9\"\n elif score_id > 90 and score_id <= 100:\n score_id = \"10\"\n\n score = urlscan_score_table[str(score_id)]\n\n malicious_tag_stats = (summary_of_malicious_tag, 4)\n\n # Attach final object\n normalize_score_url_with_no_error_code__categories.append(summary_categories) \n normalize_score_url_with_no_error_code__url_score_object.append({'score': score, 'confidence':log_result, 'score_id': score_id, \"malicious_tag_verdicts\": summary_malicious_verdicts, \"malicious_tag_stats\": malicious_tag_stats , 'categories': summary_categories, \"description\": \"\"})\n normalize_score_url_with_no_error_code__scores.append(score)\n \n phantom.debug(\"normalize_score_url_with_no_error_code__categories: {}\".format(normalize_score_url_with_no_error_code__categories))\n phantom.debug(\"normalize_score_url_with_no_error_code__url_score_object: {}\".format(normalize_score_url_with_no_error_code__url_score_object))\n phantom.debug(\"normalize_score_url_with_no_error_code__scores: {}\".format(normalize_score_url_with_no_error_code__scores))\n", + "userCode": "\n # Write your custom code here...\n \n normalize_score_url_with_no_error_code__url_score_object = []\n normalize_score_url_with_no_error_code__scores = []\n normalize_score_url_with_no_error_code__categories = []\n \n url_reputation_verdicts_overall_dict = [(i or {}) for i in url_reputation_result_item_0] \n url_reputation_verdicts_urlscan_dict = [(i or {}) for i in url_reputation_result_item_1] \n url_reputation_verdicts_engine_dict = [(i or {}) for i in url_reputation_result_item_2] \n url_reputation_verdicts_community_dict = [(i or {}) for i in url_reputation_result_item_3] \n \n #phantom.debug(\"url_reputation_verdicts_overall_dict: {}\".format(url_reputation_verdicts_overall_dict))\n #phantom.debug(\"url_reputation_verdicts_urlscan_dict: {}\".format(url_reputation_verdicts_urlscan_dict))\n #phantom.debug(\"url_reputation_verdicts_engine_dict: {}\".format(url_reputation_verdicts_engine_dict))\n #phantom.debug(\"url_reputation_verdicts_community_dict: {}\".format(url_reputation_verdicts_community_dict))\n \n \n urlscan_score_table = {\n \"0\":\"Legitimate\",\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 ## if there is no error code, urlscan.io will continue to detonate the URL and query scores in several verdicts object\n \n ## Normalize reputation on a -100 (legitimate) to 100 point scale based on number of malicious and suspicious divided by different urlscan.io verdict objects.\n ## This can be adjusted to include whatever logic is desired.\n \n for i in range(0,len(url_reputation_verdicts_overall_dict)):\n if url_reputation_verdicts_overall_dict[i] != {} or url_reputation_verdicts_urlscan_dict[i] != {} or url_reputation_verdicts_engine_dict[i] != {} or url_reputation_verdicts_community_dict[i] != {}:\n summary_score = url_reputation_verdicts_overall_dict[i]['score'] + url_reputation_verdicts_urlscan_dict[i]['score'] + url_reputation_verdicts_engine_dict[i]['score'] + url_reputation_verdicts_community_dict[i]['score']\n summary_malicious_verdicts = url_reputation_verdicts_overall_dict[i]['malicious'] or url_reputation_verdicts_urlscan_dict[i]['malicious'] or url_reputation_verdicts_engine_dict[i]['malicious'] or url_reputation_verdicts_community_dict[i]['malicious']\n summary_of_malicious_tag = int(url_reputation_verdicts_overall_dict[i]['malicious']) + int(url_reputation_verdicts_urlscan_dict[i]['malicious']) + int(url_reputation_verdicts_engine_dict[i]['malicious']) + int(url_reputation_verdicts_community_dict[i]['malicious'])\n summary_categories = url_reputation_verdicts_overall_dict[i]['categories'] + url_reputation_verdicts_urlscan_dict[i]['categories'] + url_reputation_verdicts_engine_dict[i]['categories'] + url_reputation_verdicts_community_dict[i]['categories']\n \n \n ## customized score id calculation\n \n log_result = (summary_score/4) # avg score from different urlscan.io score object (engine_score, overall_verdict_score, urlscan_verdicts_score and community score)\n score_id = int(log_result) \n \n\n if score_id < -50:\n score_id = \"0\"\n elif score_id < 0 and score_id >= -50:\n score_id = \"1\"\n elif score_id >= 0 and score_id <= 10:\n score_id = \"2\"\n elif score_id > 10 and score_id <= 20:\n score_id = \"3\"\n elif score_id > 20 and score_id <= 30:\n score_id = \"4\"\n elif score_id > 30 and score_id <= 40:\n score_id = \"5\"\n elif score_id > 40 and score_id <= 50:\n score_id = \"6\"\n elif score_id > 50 and score_id <= 60:\n score_id = \"7\"\n elif score_id > 70 and score_id <= 80:\n score_id = \"8\"\n elif score_id > 80 and score_id <= 90:\n score_id = \"9\"\n elif score_id > 90 and score_id <= 100:\n score_id = \"10\"\n\n score = urlscan_score_table[str(score_id)]\n\n malicious_tag_stats = (summary_of_malicious_tag, 4)\n\n # Attach final object\n normalize_score_url_with_no_error_code__categories.append(summary_categories) \n normalize_score_url_with_no_error_code__url_score_object.append({'score': score, 'confidence':log_result, 'score_id': score_id, \"malicious_tag_verdicts\": summary_malicious_verdicts, \"malicious_tag_stats\": malicious_tag_stats , 'categories': summary_categories, \"description\": \"\"})\n normalize_score_url_with_no_error_code__scores.append(score)\n \n #phantom.debug(\"normalize_score_url_with_no_error_code__categories: {}\".format(normalize_score_url_with_no_error_code__categories))\n #phantom.debug(\"normalize_score_url_with_no_error_code__url_score_object: {}\".format(normalize_score_url_with_no_error_code__url_score_object))\n #phantom.debug(\"normalize_score_url_with_no_error_code__scores: {}\".format(normalize_score_url_with_no_error_code__scores))\n", "warnings": {}, "x": 0, "y": 860 @@ -480,13 +480,13 @@ "data": { "advanced": { "customName": "no error code format report url ", - "customNameId": 0, + "customNameId": 1, "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": "no_error_code_format_report_url", + "functionName": "no_error_code_format_report_url_1", "id": "9", "parameters": [ "url_reputation:action_result.parameter.url", @@ -501,7 +501,7 @@ "errors": {}, "id": "9", "type": "format", - "userCode": "\n # Write your custom code here...\n phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name=\"no_error_code_format_report_url\"))\n", + "userCode": "\n # Write your custom code here...\n #phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name=\"no_error_code_format_report_url\"))\n", "warnings": {}, "x": 0, "y": 1060 @@ -531,6 +531,17 @@ "description": "An array of observable dictionaries with value, type, score, score_id, and categories.", "metadata": {}, "name": "observable" + }, + { + "contains": [], + "datapaths": [ + "error_code_format_report_url:formatted_data", + "no_error_code_format_report_url_1:formatted_data" + ], + "deduplicate": false, + "description": "a report contains value, score, confidence and categories", + "metadata": {}, + "name": "report" } ], "playbook_type": "data", @@ -538,7 +549,7 @@ "schema": "5.0.8", "version": "5.5.0.108488" }, - "create_time": "2023-03-27T12:19:43.255588+00:00", + "create_time": "2023-03-31T08:52:15.502467+00:00", "draft_mode": false, "labels": [ "*" @@ -548,11 +559,7 @@ "domain", "ip", "sandbox", - "D3-URA", - "D3-DNRA", - "D3-IPRA", - "D3-IRA", "urlscan.io", - "file_hash" + "D3-DA" ] -} +} \ No newline at end of file diff --git a/playbooks/UrlScan_IO_Dynamic_Analysis.py b/playbooks/UrlScan_IO_Dynamic_Analysis.py index 4c38350830..8ca85d7866 100644 --- a/playbooks/UrlScan_IO_Dynamic_Analysis.py +++ b/playbooks/UrlScan_IO_Dynamic_Analysis.py @@ -1,5 +1,5 @@ """ -Accepts a URL for detonation analysis on the objects. 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:IdentifierReputationAnalysis/ +Accepts a URL for detonation analysis on the objects. 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\n """ @@ -196,9 +196,9 @@ def normalize_score_url_with_error_code(action=None, success=None, container=Non normalize_score_url_with_error_code__url_score_object.append({'score': urlscan_score_table['error_code_query'], 'confidence':"", 'score_id': "", "malicious_tag_verdicts": "", 'categories': error_message, "description": url_descp}) normalize_score_url_with_error_code__scores.append(urlscan_score_table['error_code_query']) - phantom.debug("normalize_score_url_with_error_code__url_score_object: {}".format(normalize_score_url_with_error_code__url_score_object)) - phantom.debug("normalize_score_url_with_error_code__scores: {}".format(normalize_score_url_with_error_code__scores)) - phantom.debug("normalize_score_url_with_error_code__categories: {}".format(normalize_score_url_with_error_code__categories)) + #phantom.debug("normalize_score_url_with_error_code__url_score_object: {}".format(normalize_score_url_with_error_code__url_score_object)) + #phantom.debug("normalize_score_url_with_error_code__scores: {}".format(normalize_score_url_with_error_code__scores)) + #phantom.debug("normalize_score_url_with_error_code__categories: {}".format(normalize_score_url_with_error_code__categories)) ################################################################################ @@ -395,9 +395,9 @@ def normalize_score_url_with_no_error_code(action=None, success=None, container= normalize_score_url_with_no_error_code__url_score_object.append({'score': score, 'confidence':log_result, 'score_id': score_id, "malicious_tag_verdicts": summary_malicious_verdicts, "malicious_tag_stats": malicious_tag_stats , 'categories': summary_categories, "description": ""}) normalize_score_url_with_no_error_code__scores.append(score) - phantom.debug("normalize_score_url_with_no_error_code__categories: {}".format(normalize_score_url_with_no_error_code__categories)) - phantom.debug("normalize_score_url_with_no_error_code__url_score_object: {}".format(normalize_score_url_with_no_error_code__url_score_object)) - phantom.debug("normalize_score_url_with_no_error_code__scores: {}".format(normalize_score_url_with_no_error_code__scores)) + #phantom.debug("normalize_score_url_with_no_error_code__categories: {}".format(normalize_score_url_with_no_error_code__categories)) + #phantom.debug("normalize_score_url_with_no_error_code__url_score_object: {}".format(normalize_score_url_with_no_error_code__url_score_object)) + #phantom.debug("normalize_score_url_with_no_error_code__scores: {}".format(normalize_score_url_with_no_error_code__scores)) ################################################################################ ## Custom Code End ################################################################################ @@ -407,14 +407,14 @@ def normalize_score_url_with_no_error_code(action=None, success=None, container= phantom.save_run_data(key="normalize_score_url_with_no_error_code:categories", value=json.dumps(normalize_score_url_with_no_error_code__categories)) phantom.save_run_data(key="normalize_score_url_with_no_error_code:confidence", value=json.dumps(normalize_score_url_with_no_error_code__confidence)) - no_error_code_format_report_url(container=container) + no_error_code_format_report_url_1(container=container) return @phantom.playbook_block() -def no_error_code_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("no_error_code_format_report_url() called") +def no_error_code_format_report_url_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("no_error_code_format_report_url_1() called") ################################################################################ # Format a summary table with the information gathered from the playbook. @@ -441,7 +441,7 @@ def no_error_code_format_report_url(action=None, success=None, container=None, r ## Custom Code End ################################################################################ - phantom.format(container=container, template=template, parameters=parameters, name="no_error_code_format_report_url") + phantom.format(container=container, template=template, parameters=parameters, name="no_error_code_format_report_url_1") build_url_output_with_no_error_code(container=container) @@ -479,12 +479,12 @@ def build_url_output_with_error_code(action=None, success=None, container=None, url_scan_io_parameter_url = [str(i or '') for i in url_reputation_parameter_url] url_scan_io_url_score_object = normalize_score_url_with_error_code__url_score_object - phantom.debug("url_reputation_parameter_url: {}".format(url_reputation_parameter_url)) + #phantom.debug("url_reputation_parameter_url: {}".format(url_reputation_parameter_url)) #phantom.debug("url_reputation_result_item_1: {}".format(url_reputation_result_item_1)) #phantom.debug("normalize_score_url_with_error_code__url_score_object: {}".format(normalize_score_url_with_error_code__url_score_object)) for url, external_id, url_object in zip(url_scan_io_parameter_url, url_scan_io_task_reporturl, url_scan_io_url_score_object): parsed_url = urlparse(url) - phantom.debug("{} {} {} parsed_url: {}".format(url, external_id, url_object, parsed_url)) + #phantom.debug("{} {} {} parsed_url: {}".format(url, external_id, url_object, parsed_url)) observable_object = { "value": url, "type": "url", @@ -548,10 +548,10 @@ def build_url_output_with_no_error_code(action=None, success=None, container=Non build_url_output_with_no_error_code__observable_array = [] # Build URL - phantom.debug(url_reputation_parameter_url) + #phantom.debug(url_reputation_parameter_url) for url, external_id, url_object in zip(url_reputation_parameter_url, url_reputation_result_item_1, normalize_score_url_with_no_error_code__url_score_object): parsed_url = urlparse(url) - phantom.debug("{} {} {} parsed_url: {}".format(url, external_id, url_object, parsed_url)) + #phantom.debug("{} {} {} parsed_url: {}".format(url, external_id, url_object, parsed_url)) observable_object = { "value": url, "type": "url", @@ -591,13 +591,17 @@ def build_url_output_with_no_error_code(action=None, success=None, container=Non def on_finish(container, summary): phantom.debug("on_finish() called") + error_code_format_report_url = phantom.get_format_data(name="error_code_format_report_url") + no_error_code_format_report_url_1 = phantom.get_format_data(name="no_error_code_format_report_url_1") build_url_output_with_error_code__observable_array = json.loads(_ if (_ := phantom.get_run_data(key="build_url_output_with_error_code:observable_array")) != "" else "null") # pylint: disable=used-before-assignment build_url_output_with_no_error_code__observable_array = json.loads(_ if (_ := phantom.get_run_data(key="build_url_output_with_no_error_code:observable_array")) != "" else "null") # pylint: disable=used-before-assignment observable_combined_value = phantom.concatenate(build_url_output_with_error_code__observable_array, build_url_output_with_no_error_code__observable_array) + report_combined_value = phantom.concatenate(error_code_format_report_url, no_error_code_format_report_url_1) output = { "observable": observable_combined_value, + "report": report_combined_value, } ################################################################################ @@ -605,15 +609,11 @@ def on_finish(container, summary): ################################################################################ # Write your custom code here... - no_error_code_format_report_url = phantom.get_format_data(name="no_error_code_format_report_url") - error_code_format_report_url = phantom.get_format_data(name="error_code_format_report_url") - markdown_report_combined_value = phantom.concatenate(no_error_code_format_report_url, error_code_format_report_url) - output['markdown_report'] = markdown_report_combined_value + #phantom.debug(output) ################################################################################ ## Custom Code End ################################################################################ phantom.save_playbook_output_data(output=output) - return - + return \ No newline at end of file diff --git a/playbooks/UrlScan_IO_Dynamic_Analysis.yml b/playbooks/UrlScan_IO_Dynamic_Analysis.yml index e68eab1e7e..069c34e8e7 100644 --- a/playbooks/UrlScan_IO_Dynamic_Analysis.yml +++ b/playbooks/UrlScan_IO_Dynamic_Analysis.yml @@ -13,7 +13,6 @@ app_list: - urlscan.io tags: platform_tags: - - reputation - url - domain - sandbox From 3b86de8abd355e5dd9738c6e7d255fbe204d5c88 Mon Sep 17 00:00:00 2001 From: tccontre Date: Mon, 3 Apr 2023 10:55:37 +0200 Subject: [PATCH 2/4] minor_update_saa_urlscan_playbook --- playbooks/SAA_Dynamic_Analysis.json | 30 +++++++++++++++------------ playbooks/SAA_Dynamic_Analysis.py | 32 ++++++++++++++++------------- 2 files changed, 35 insertions(+), 27 deletions(-) diff --git a/playbooks/SAA_Dynamic_Analysis.json b/playbooks/SAA_Dynamic_Analysis.json index 01e18854e9..cedd9740fa 100644 --- a/playbooks/SAA_Dynamic_Analysis.json +++ b/playbooks/SAA_Dynamic_Analysis.json @@ -170,7 +170,7 @@ "targetPort": "17_in" } ], - "hash": "6b4aa256c537bfb23797087a22c09d3e58ee3082", + "hash": "c1a9645ef863bec36268198b86a8f917300a51b3", "nodes": { "0": { "data": { @@ -186,7 +186,7 @@ "type": "start", "warnings": {}, "x": 190, - "y": -2.5579538487363607e-13 + "y": -3.197442310920451e-13 }, "1": { "data": { @@ -260,7 +260,7 @@ "errors": {}, "id": "11", "type": "code", - "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output__observable_array = []\n phantom.debug(playbook_input_url_values)\n # Build URL\n for jobs_id in get_jobid_of_url_detonation_output__jobid:\n \n for url, external_id, url_object in zip(playbook_input_url_values, jobs_id, normalized_job_forensic_report_output__url_score_object):\n parsed_url = urlparse(url)\n phantom.debug(\"url: {} jobs_id:{}\".format(url, external_id))\n phantom.debug(\"parsed_url: {}, url_object: {}\".format(parsed_url, url_object))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\": f\"https://app.twinwave.io/job/{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 #phantom.debug(\"build_url_output__observable_array: {}\".format(build_url_output__observable_array))\n", + "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output__observable_array = []\n #phantom.debug(playbook_input_url_values)\n # Build URL\n for jobs_id in get_jobid_of_url_detonation_output__jobid:\n \n for url, external_id, url_object in zip(playbook_input_url_values, jobs_id, normalized_job_forensic_report_output__url_score_object):\n parsed_url = urlparse(url)\n #phantom.debug(\"url: {} jobs_id:{}\".format(url, external_id))\n #phantom.debug(\"parsed_url: {}, url_object: {}\".format(parsed_url, url_object))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\": f\"https://app.twinwave.io/job/{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 #phantom.debug(\"build_url_output__observable_array: {}\".format(build_url_output__observable_array))\n", "warnings": {}, "x": 0, "y": 1626 @@ -461,7 +461,7 @@ "errors": {}, "id": "19", "type": "code", - "userCode": "\n # Write your custom code here...\n build_file_output__observable_array = []\n for jobs_id in get_jobid_of_file_detonation_output__jobid:\n for _vault_id, external_id, file_object in zip(playbook_input_vault_id_values, jobs_id, normalized_job_forensic_report_output_1__file_score_object):\n phantom.debug(\"vault: {} id: {}\".format(_vault_id, external_id))\n observable_object = {\n\n \"value\": _vault_id,\n \"type\": \"hash\",\n \"sandbox\": {\n \"score\": file_object['score'],\n \"confidence\": file_object['confidence'],\n\n },\n \"enrichment\": {\n \"provider\": \"Splunk Attack Analyzer\",\n \"type\": \"file\",\n\n },\n \"categories\": file_object['categories'],\n \"source\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\":f\"https://app.twinwave.io/job/{external_id}\"\n }\n build_file_output__observable_array.append(observable_object)\n phantom.debug(\"build_file_output__observable_array: {}\".format(build_file_output__observable_array))\n", + "userCode": "\n # Write your custom code here...\n build_file_output__observable_array = []\n for jobs_id in get_jobid_of_file_detonation_output__jobid:\n for _vault_id, external_id, file_object in zip(playbook_input_vault_id_values, jobs_id, normalized_job_forensic_report_output_1__file_score_object):\n #phantom.debug(\"vault: {} id: {}\".format(_vault_id, external_id))\n observable_object = {\n\n \"value\": _vault_id,\n \"type\": \"hash\",\n \"sandbox\": {\n \"score\": file_object['score'],\n \"confidence\": file_object['confidence'],\n\n },\n \"enrichment\": {\n \"provider\": \"Splunk Attack Analyzer\",\n \"type\": \"file\",\n\n },\n \"categories\": file_object['categories'],\n \"source\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\":f\"https://app.twinwave.io/job/{external_id}\"\n }\n build_file_output__observable_array.append(observable_object)\n #phantom.debug(\"build_file_output__observable_array: {}\".format(build_file_output__observable_array))\n", "warnings": {}, "x": 340, "y": 1620 @@ -518,7 +518,11 @@ "22": { "data": { "advanced": { - "join": [] + "customName": "url detonation status filter", + "customNameId": 1, + "description": "Filters url detonation results.", + "join": [], + "note": "Filters url detonation results." }, "conditions": [ { @@ -536,7 +540,7 @@ } ], "functionId": 5, - "functionName": "filter_5", + "functionName": "url_detonation_status_filter_1", "id": "22", "type": "filter" }, @@ -545,7 +549,7 @@ "type": "filter", "warnings": {}, "x": 60, - "y": 506 + "y": 500 }, "23": { "data": { @@ -621,11 +625,11 @@ "4": { "data": { "advanced": { - "customName": "url detonation status filter", + "customName": "file detonation status filter", "customNameId": 0, - "description": "Filters successful url or file detonation results.", + "description": "Filters successful file detonation results.", "join": [], - "note": "Filters successful url or file detonation results." + "note": "Filters successful file detonation results." }, "conditions": [ { @@ -643,7 +647,7 @@ } ], "functionId": 2, - "functionName": "url_detonation_status_filter", + "functionName": "file_detonation_status_filter", "id": "4", "type": "filter" }, @@ -652,7 +656,7 @@ "type": "filter", "warnings": {}, "x": 400, - "y": 506 + "y": 500 }, "5": { "data": { @@ -842,7 +846,7 @@ "schema": "5.0.8", "version": "5.5.0.108488" }, - "create_time": "2023-03-31T08:46:27.522787+00:00", + "create_time": "2023-04-03T08:09:23.517027+00:00", "draft_mode": false, "labels": [ "*" diff --git a/playbooks/SAA_Dynamic_Analysis.py b/playbooks/SAA_Dynamic_Analysis.py index 884515f0f2..95bbb9efd8 100644 --- a/playbooks/SAA_Dynamic_Analysis.py +++ b/playbooks/SAA_Dynamic_Analysis.py @@ -83,17 +83,17 @@ def saa_url_detonation(action=None, success=None, container=None, results=None, ## Custom Code End ################################################################################ - phantom.act("detonate url", parameters=parameters, name="saa_url_detonation", assets=["splunk attack analyzer"], callback=filter_5) + phantom.act("detonate url", parameters=parameters, name="saa_url_detonation", assets=["splunk attack analyzer"], callback=url_detonation_status_filter_1) return @phantom.playbook_block() -def url_detonation_status_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): - phantom.debug("url_detonation_status_filter() called") +def file_detonation_status_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): + phantom.debug("file_detonation_status_filter() called") ################################################################################ - # Filters successful url or file detonation results. + # Filters successful file detonation results. ################################################################################ # collect filtered artifact ids and results for 'if' condition 1 @@ -102,7 +102,7 @@ def url_detonation_status_filter(action=None, success=None, container=None, resu conditions=[ ["saa_file_detonation:action_result.status", "==", "success"] ], - name="url_detonation_status_filter:condition_1") + name="file_detonation_status_filter:condition_1") # call connected blocks if filtered artifacts or results if matched_artifacts_1 or matched_results_1: @@ -411,14 +411,14 @@ def build_url_output(action=None, success=None, container=None, results=None, ha # Write your custom code here... from urllib.parse import urlparse build_url_output__observable_array = [] - phantom.debug(playbook_input_url_values) + #phantom.debug(playbook_input_url_values) # Build URL for jobs_id in get_jobid_of_url_detonation_output__jobid: for url, external_id, url_object in zip(playbook_input_url_values, jobs_id, normalized_job_forensic_report_output__url_score_object): parsed_url = urlparse(url) - phantom.debug("url: {} jobs_id:{}".format(url, external_id)) - phantom.debug("parsed_url: {}, url_object: {}".format(parsed_url, url_object)) + #phantom.debug("url: {} jobs_id:{}".format(url, external_id)) + #phantom.debug("parsed_url: {}, url_object: {}".format(parsed_url, url_object)) observable_object = { "value": url, "type": "url", @@ -484,7 +484,7 @@ def saa_file_detonation(action=None, success=None, container=None, results=None, ## Custom Code End ################################################################################ - phantom.act("detonate file", parameters=parameters, name="saa_file_detonation", assets=["splunk attack analyzer"], callback=url_detonation_status_filter) + phantom.act("detonate file", parameters=parameters, name="saa_file_detonation", assets=["splunk attack analyzer"], callback=file_detonation_status_filter) return @@ -767,7 +767,7 @@ def build_file_output(action=None, success=None, container=None, results=None, h build_file_output__observable_array = [] for jobs_id in get_jobid_of_file_detonation_output__jobid: for _vault_id, external_id, file_object in zip(playbook_input_vault_id_values, jobs_id, normalized_job_forensic_report_output_1__file_score_object): - phantom.debug("vault: {} id: {}".format(_vault_id, external_id)) + #phantom.debug("vault: {} id: {}".format(_vault_id, external_id)) observable_object = { "value": _vault_id, @@ -787,7 +787,7 @@ def build_file_output(action=None, success=None, container=None, results=None, h "source_link":f"https://app.twinwave.io/job/{external_id}" } build_file_output__observable_array.append(observable_object) - phantom.debug("build_file_output__observable_array: {}".format(build_file_output__observable_array)) + #phantom.debug("build_file_output__observable_array: {}".format(build_file_output__observable_array)) ################################################################################ ## Custom Code End ################################################################################ @@ -798,8 +798,12 @@ def build_file_output(action=None, success=None, container=None, results=None, h @phantom.playbook_block() -def filter_5(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs): - phantom.debug("filter_5() called") +def url_detonation_status_filter_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_status_filter_1() called") + + ################################################################################ + # Filters url detonation results. + ################################################################################ # collect filtered artifact ids and results for 'if' condition 1 matched_artifacts_1, matched_results_1 = phantom.condition( @@ -807,7 +811,7 @@ def filter_5(action=None, success=None, container=None, results=None, handle=Non conditions=[ ["saa_url_detonation:action_result.status", "==", "success"] ], - name="filter_5:condition_1") + name="url_detonation_status_filter_1:condition_1") # call connected blocks if filtered artifacts or results if matched_artifacts_1 or matched_results_1: From 3a2caa27054cbae216f2d80c6ca923b42e51d7d8 Mon Sep 17 00:00:00 2001 From: tccontre <26181693+tccontre@users.noreply.github.com> Date: Wed, 5 Apr 2023 11:31:12 +0200 Subject: [PATCH 3/4] Update SAA_Dynamic_Analysis.py --- playbooks/SAA_Dynamic_Analysis.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/playbooks/SAA_Dynamic_Analysis.py b/playbooks/SAA_Dynamic_Analysis.py index 95bbb9efd8..2395392f29 100644 --- a/playbooks/SAA_Dynamic_Analysis.py +++ b/playbooks/SAA_Dynamic_Analysis.py @@ -359,7 +359,7 @@ def format_url_report(action=None, success=None, container=None, results=None, h # Format a summary table with the information gathered from the playbook. ################################################################################ - template = """SOAR analyzed URL(s) using Splunk Attack Analyzer. The table below shows a summary of the information gathered.\n\n| URL | Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |https://app.twinwave.io/job/{4} | Splunk Attack Analyzer (SAA){1}{2}{3}{4} |\n%%\n\n\n""" + template = """SOAR analyzed URL(s) using Splunk Attack Analyzer. The table below shows a summary of the information gathered.\n\n| URL | Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |https://app.twinwave.io/job/{4} | Splunk Attack Analyzer (SAA) |\n%%\n\n\n""" # parameter list for template variable replacement parameters = [ @@ -714,7 +714,7 @@ def format_file_report(action=None, success=None, container=None, results=None, # Format a summary table with the information gathered from the playbook. ################################################################################ - template = """SOAR analyzed File(s) using Splunk Attack Analyzer. The table below shows a summary of the information gathered.\n\n| File hash | Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |https://app.twinwave.io/job/{4} | Splunk Attack Analyzer (SAA){1}{2}{3}{4} |\n%%\n\n\n""" + template = """SOAR analyzed File(s) using Splunk Attack Analyzer. The table below shows a summary of the information gathered.\n\n| File hash | Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |https://app.twinwave.io/job/{4} | Splunk Attack Analyzer (SAA) |\n%%\n\n\n""" # parameter list for template variable replacement parameters = [ @@ -868,4 +868,4 @@ def on_finish(container, summary): phantom.save_playbook_output_data(output=output) - return \ No newline at end of file + return From 3ee8c81ece095bdf6f2f4bd6b3f096ca4aa900ef Mon Sep 17 00:00:00 2001 From: tccontre Date: Wed, 5 Apr 2023 19:19:19 +0200 Subject: [PATCH 4/4] minor_update_saa_urlscan_playbook --- playbooks/SAA_Dynamic_Analysis.json | 16 +++---- playbooks/SAA_Dynamic_Analysis.py | 6 +-- playbooks/UrlScan_IO_Dynamic_Analysis.json | 54 +++++++++++----------- playbooks/UrlScan_IO_Dynamic_Analysis.py | 13 ++++-- 4 files changed, 46 insertions(+), 43 deletions(-) diff --git a/playbooks/SAA_Dynamic_Analysis.json b/playbooks/SAA_Dynamic_Analysis.json index cedd9740fa..d6237e0ffd 100644 --- a/playbooks/SAA_Dynamic_Analysis.json +++ b/playbooks/SAA_Dynamic_Analysis.json @@ -170,7 +170,7 @@ "targetPort": "17_in" } ], - "hash": "c1a9645ef863bec36268198b86a8f917300a51b3", + "hash": "e8e4290e9196ed3b9dc21a9395efea9b415a12af", "nodes": { "0": { "data": { @@ -186,7 +186,7 @@ "type": "start", "warnings": {}, "x": 190, - "y": -3.197442310920451e-13 + "y": -3.836930773104541e-13 }, "1": { "data": { @@ -224,7 +224,7 @@ "normalized_job_forensic_report_output:custom_function:categories", "get_jobid_of_url_detonation_output:custom_function:jobid" ], - "template": "SOAR analyzed URL(s) using Splunk Attack Analyzer. The table below shows a summary of the information gathered.\n\n| URL | Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |https://app.twinwave.io/job/{4} | Splunk Attack Analyzer (SAA){1}{2}{3}{4} |\n%%\n\n\n", + "template": "SOAR analyzed URL(s) using Splunk Attack Analyzer. The table below shows a summary of the information gathered.\n\n| URL | Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |https://app.twinwave.io/job/{4} | Splunk Attack Analyzer (SAA) |\n%%\n\n\n", "type": "format" }, "errors": {}, @@ -233,7 +233,7 @@ "userCode": "\n # Write your custom code here...\n #phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name=\"format_report_url\"))\n", "warnings": {}, "x": 0, - "y": 1448 + "y": 1440 }, "11": { "data": { @@ -260,7 +260,7 @@ "errors": {}, "id": "11", "type": "code", - "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output__observable_array = []\n #phantom.debug(playbook_input_url_values)\n # Build URL\n for jobs_id in get_jobid_of_url_detonation_output__jobid:\n \n for url, external_id, url_object in zip(playbook_input_url_values, jobs_id, normalized_job_forensic_report_output__url_score_object):\n parsed_url = urlparse(url)\n #phantom.debug(\"url: {} jobs_id:{}\".format(url, external_id))\n #phantom.debug(\"parsed_url: {}, url_object: {}\".format(parsed_url, url_object))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\": f\"https://app.twinwave.io/job/{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 #phantom.debug(\"build_url_output__observable_array: {}\".format(build_url_output__observable_array))\n", + "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output__observable_array = []\n #phantom.debug(playbook_input_url_values)\n # Build URL\n for jobs_id in get_jobid_of_url_detonation_output__jobid:\n \n for url, external_id, url_object in zip(playbook_input_url_values, jobs_id, normalized_job_forensic_report_output__url_score_object):\n parsed_url = urlparse(url)\n #phantom.debug(\"url: {} jobs_id:{}\".format(url, external_id))\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\": 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\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\": f\"https://app.twinwave.io/job/{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 #phantom.debug(\"build_url_output__observable_array: {}\".format(build_url_output__observable_array))\n", "warnings": {}, "x": 0, "y": 1626 @@ -425,7 +425,7 @@ "normalized_job_forensic_report_output_1:custom_function:categories", "get_jobid_of_file_detonation_output:custom_function:jobid" ], - "template": "SOAR analyzed File(s) using Splunk Attack Analyzer. The table below shows a summary of the information gathered.\n\n| File hash | Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |https://app.twinwave.io/job/{4} | Splunk Attack Analyzer (SAA){1}{2}{3}{4} |\n%%\n\n\n", + "template": "SOAR analyzed File(s) using Splunk Attack Analyzer. The table below shows a summary of the information gathered.\n\n| File hash | Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |https://app.twinwave.io/job/{4} | Splunk Attack Analyzer (SAA) |\n%%\n\n\n", "type": "format" }, "errors": {}, @@ -461,7 +461,7 @@ "errors": {}, "id": "19", "type": "code", - "userCode": "\n # Write your custom code here...\n build_file_output__observable_array = []\n for jobs_id in get_jobid_of_file_detonation_output__jobid:\n for _vault_id, external_id, file_object in zip(playbook_input_vault_id_values, jobs_id, normalized_job_forensic_report_output_1__file_score_object):\n #phantom.debug(\"vault: {} id: {}\".format(_vault_id, external_id))\n observable_object = {\n\n \"value\": _vault_id,\n \"type\": \"hash\",\n \"sandbox\": {\n \"score\": file_object['score'],\n \"confidence\": file_object['confidence'],\n\n },\n \"enrichment\": {\n \"provider\": \"Splunk Attack Analyzer\",\n \"type\": \"file\",\n\n },\n \"categories\": file_object['categories'],\n \"source\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\":f\"https://app.twinwave.io/job/{external_id}\"\n }\n build_file_output__observable_array.append(observable_object)\n #phantom.debug(\"build_file_output__observable_array: {}\".format(build_file_output__observable_array))\n", + "userCode": "\n # Write your custom code here...\n build_file_output__observable_array = []\n for jobs_id in get_jobid_of_file_detonation_output__jobid:\n for _vault_id, external_id, file_object in zip(playbook_input_vault_id_values, jobs_id, normalized_job_forensic_report_output_1__file_score_object):\n #phantom.debug(\"vault: {} id: {}\".format(_vault_id, external_id))\n observable_object = {\n\n \"value\": _vault_id,\n \"type\": \"hash\",\n \"reputation\": {\n \"score\": file_object['score'],\n \"confidence\": file_object['confidence'],\n\n },\n \"enrichment\": {\n \"provider\": \"Splunk Attack Analyzer\",\n \"type\": \"file\",\n\n },\n \"categories\": file_object['categories'],\n \"source\": \"Splunk Attack Analyzer (SAA)\",\n \"source_link\":f\"https://app.twinwave.io/job/{external_id}\"\n }\n build_file_output__observable_array.append(observable_object)\n #phantom.debug(\"build_file_output__observable_array: {}\".format(build_file_output__observable_array))\n", "warnings": {}, "x": 340, "y": 1620 @@ -846,7 +846,7 @@ "schema": "5.0.8", "version": "5.5.0.108488" }, - "create_time": "2023-04-03T08:09:23.517027+00:00", + "create_time": "2023-04-05T16:45:42.058432+00:00", "draft_mode": false, "labels": [ "*" diff --git a/playbooks/SAA_Dynamic_Analysis.py b/playbooks/SAA_Dynamic_Analysis.py index 2395392f29..975f9e46bf 100644 --- a/playbooks/SAA_Dynamic_Analysis.py +++ b/playbooks/SAA_Dynamic_Analysis.py @@ -422,7 +422,7 @@ def build_url_output(action=None, success=None, container=None, results=None, ha observable_object = { "value": url, "type": "url", - "sandbox": { + "reputation": { "score": url_object['score'], "confidence": url_object['confidence'] }, @@ -772,7 +772,7 @@ def build_file_output(action=None, success=None, container=None, results=None, h "value": _vault_id, "type": "hash", - "sandbox": { + "reputation": { "score": file_object['score'], "confidence": file_object['confidence'], @@ -868,4 +868,4 @@ def on_finish(container, summary): phantom.save_playbook_output_data(output=output) - return + return \ No newline at end of file diff --git a/playbooks/UrlScan_IO_Dynamic_Analysis.json b/playbooks/UrlScan_IO_Dynamic_Analysis.json index a4e7507f4e..6cb60fa847 100644 --- a/playbooks/UrlScan_IO_Dynamic_Analysis.json +++ b/playbooks/UrlScan_IO_Dynamic_Analysis.json @@ -32,13 +32,6 @@ "targetNode": "4", "targetPort": "4_in" }, - { - "id": "port_5_to_port_6", - "sourceNode": "5", - "sourcePort": "5_out", - "targetNode": "6", - "targetPort": "6_in" - }, { "conditions": [ { @@ -84,13 +77,6 @@ "targetNode": "1", "targetPort": "1_in" }, - { - "id": "port_6_to_port_10", - "sourceNode": "6", - "sourcePort": "6_out", - "targetNode": "10", - "targetPort": "10_in" - }, { "id": "port_9_to_port_11", "sourceNode": "9", @@ -109,9 +95,23 @@ "sourcePort": "7_out", "targetNode": "5", "targetPort": "5_in" + }, + { + "id": "port_5_to_port_6", + "sourceNode": "5", + "sourcePort": "5_out", + "targetNode": "6", + "targetPort": "6_in" + }, + { + "id": "port_6_to_port_10", + "sourceNode": "6", + "sourcePort": "6_out", + "targetNode": "10", + "targetPort": "10_in" } ], - "hash": "6579c959dc60d4d44f4d7aee29bd67897a8b913b", + "hash": "be20d07e17cf03cfbe4bfd604b5032eb48174f24", "nodes": { "0": { "data": { @@ -127,7 +127,7 @@ "type": "start", "warnings": {}, "x": 190, - "y": -3.197442310920451e-13 + "y": -6.394884621840902e-14 }, "1": { "data": { @@ -143,8 +143,8 @@ "type": "end", "userCode": "\n # Write your custom code here...\n #phantom.debug(output)\n", "warnings": {}, - "x": 200, - "y": 1440 + "x": 190, + "y": 1448 }, "10": { "data": { @@ -171,7 +171,7 @@ "errors": {}, "id": "10", "type": "code", - "userCode": "\n # Write your custom code here...\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output_with_error_code__observable_array = []\n\n # Build URL\n url_scan_io_task_reporturl = [str(i or 'no report url') for i in url_reputation_result_item_1]\n url_scan_io_parameter_url = [str(i or '') for i in url_reputation_parameter_url]\n url_scan_io_url_score_object = normalize_score_url_with_error_code__url_score_object\n \n #phantom.debug(\"url_reputation_parameter_url: {}\".format(url_reputation_parameter_url))\n #phantom.debug(\"url_reputation_result_item_1: {}\".format(url_reputation_result_item_1))\n #phantom.debug(\"normalize_score_url_with_error_code__url_score_object: {}\".format(normalize_score_url_with_error_code__url_score_object))\n for url, external_id, url_object in zip(url_scan_io_parameter_url, url_scan_io_task_reporturl, url_scan_io_url_score_object):\n parsed_url = urlparse(url)\n #phantom.debug(\"{} {} {} parsed_url: {}\".format(url, external_id, url_object, parsed_url))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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 \"description\" :url_object['description'],\n \"source\": \"urlscan.io\",\n \"source_link\": f\"{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 \n build_url_output_with_error_code__observable_array.append(observable_object)\n #phantom.debug(\"build_url_output_with_error_code__observable_array: {}\".format(build_url_output_with_error_code__observable_array))\n", + "userCode": "\n # Write your custom code here...\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output_with_error_code__observable_array = []\n\n # Build URL\n url_scan_io_task_reporturl = [str(i or 'no report url') for i in url_reputation_result_item_1]\n url_scan_io_parameter_url = [str(i or '') for i in url_reputation_parameter_url]\n url_scan_io_url_score_object = normalize_score_url_with_error_code__url_score_object\n \n #phantom.debug(\"url_reputation_parameter_url: {}\".format(url_reputation_parameter_url))\n #phantom.debug(\"url_reputation_result_item_1: {}\".format(url_reputation_result_item_1))\n #phantom.debug(\"normalize_score_url_with_error_code__url_score_object: {}\".format(normalize_score_url_with_error_code__url_score_object))\n for url, external_id, url_object in zip(url_scan_io_parameter_url, url_scan_io_task_reporturl, url_scan_io_url_score_object):\n parsed_url = urlparse(url)\n #phantom.debug(\"{} {} {} parsed_url: {}\".format(url, external_id, url_object, parsed_url))\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 \"description\" :url_object['description'],\n \"source\": \"urlscan.io\",\n \"source_link\": f\"{external_id}\"\n }\n if parsed_url.hostname == None:\n observable_object['attributes']['hostname'] = url.split(\"\\\\\",1)[0]\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 \n build_url_output_with_error_code__observable_array.append(observable_object)\n #phantom.debug(\"build_url_output_with_error_code__observable_array: {}\".format(build_url_output_with_error_code__observable_array))\n", "warnings": {}, "x": 340, "y": 1260 @@ -201,7 +201,7 @@ "errors": {}, "id": "11", "type": "code", - "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output_with_no_error_code__observable_array = []\n\n # Build URL\n #phantom.debug(url_reputation_parameter_url)\n for url, external_id, url_object in zip(url_reputation_parameter_url, url_reputation_result_item_1, normalize_score_url_with_no_error_code__url_score_object):\n parsed_url = urlparse(url)\n #phantom.debug(\"{} {} {} parsed_url: {}\".format(url, external_id, url_object, parsed_url))\n observable_object = {\n \"value\": url,\n \"type\": \"url\",\n \"sandbox\": {\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 \"description\" :url_object['description'],\n \"source\": \"urlscan.io\",\n \"source_link\": f\"{external_id}\"\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_with_no_error_code__observable_array.append(observable_object)\n #phantom.debug(\"build_url_output_with_no_error_code__observable_array: {}\".format(build_url_output_with_no_error_code__observable_array))\n", + "userCode": "\n # Write your custom code here...\n from urllib.parse import urlparse\n build_url_output_with_no_error_code__observable_array = []\n\n # Build URL\n #phantom.debug(url_reputation_parameter_url)\n for url, external_id, url_object in zip(url_reputation_parameter_url, url_reputation_result_item_1, normalize_score_url_with_no_error_code__url_score_object):\n parsed_url = urlparse(url)\n #phantom.debug(\"{} {} {} parsed_url: {}\".format(url, external_id, url_object, parsed_url))\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 \"description\" :url_object['description'],\n \"source\": \"urlscan.io\",\n \"source_link\": f\"{external_id}\"\n }\n if parsed_url.hostname == None:\n observable_object['attributes']['hostname'] = url.split(\"\\\\\",1)[0]\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_with_no_error_code__observable_array.append(observable_object)\n #phantom.debug(\"build_url_output_with_no_error_code__observable_array: {}\".format(build_url_output_with_no_error_code__observable_array))\n", "warnings": {}, "x": 0, "y": 1260 @@ -325,8 +325,8 @@ "id": "4", "type": "filter", "warnings": {}, - "x": 240, - "y": 500 + "x": 230, + "y": 506 }, "5": { "customCode": null, @@ -382,7 +382,7 @@ "normalize_score_url_with_error_code:custom_function:categories", "url_reputation:action_result.data.*.task.reportURL" ], - "template": "SOAR analyzed URL(s) using urlscan.io. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} | {4} | urlscan.io |\n\n", + "template": "SOAR analyzed URL(s) using urlscan.io. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} | {4} | urlscan.io |\n%%\n", "type": "format" }, "errors": {}, @@ -439,8 +439,8 @@ "id": "7", "type": "filter", "warnings": {}, - "x": 240, - "y": 680 + "x": 230, + "y": 686 }, "8": { "data": { @@ -495,7 +495,7 @@ "normalize_score_url_with_no_error_code:custom_function:categories", "url_reputation:action_result.data.*.task.reportURL" ], - "template": "SOAR analyzed URL(s) using urlscan.io. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score |Confidence | Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |{4} | urlscan.io |\n", + "template": "SOAR analyzed URL(s) using urlscan.io. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score |Confidence | Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |{4} | urlscan.io |\n%%", "type": "format" }, "errors": {}, @@ -549,7 +549,7 @@ "schema": "5.0.8", "version": "5.5.0.108488" }, - "create_time": "2023-03-31T08:52:15.502467+00:00", + "create_time": "2023-04-05T17:09:08.932192+00:00", "draft_mode": false, "labels": [ "*" diff --git a/playbooks/UrlScan_IO_Dynamic_Analysis.py b/playbooks/UrlScan_IO_Dynamic_Analysis.py index 8ca85d7866..349b5e7b90 100644 --- a/playbooks/UrlScan_IO_Dynamic_Analysis.py +++ b/playbooks/UrlScan_IO_Dynamic_Analysis.py @@ -223,7 +223,7 @@ def error_code_format_report_url(action=None, success=None, container=None, resu # Format a summary table with the information gathered from the playbook. ################################################################################ - template = """SOAR analyzed URL(s) using urlscan.io. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} | {4} | urlscan.io |\n\n""" + template = """SOAR analyzed URL(s) using urlscan.io. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score | Confidence |Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} | {4} | urlscan.io |\n%%\n""" # parameter list for template variable replacement parameters = [ @@ -420,7 +420,7 @@ def no_error_code_format_report_url_1(action=None, success=None, container=None, # Format a summary table with the information gathered from the playbook. ################################################################################ - template = """SOAR analyzed URL(s) using urlscan.io. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score |Confidence | Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |{4} | urlscan.io |\n""" + template = """SOAR analyzed URL(s) using urlscan.io. The table below shows a summary of the information gathered.\n\n| URL | Normalized Score |Confidence | Categories | Report Link | Source |\n| --- | --- | --- | --- | --- |\n%%\n| `{0}` | {1} | {2} | {3} |{4} | urlscan.io |\n%%""" # parameter list for template variable replacement parameters = [ @@ -488,7 +488,7 @@ def build_url_output_with_error_code(action=None, success=None, container=None, observable_object = { "value": url, "type": "url", - "sandbox": { + "reputation": { "score_id": url_object['score_id'], "score": url_object['score'], "confidence": url_object['confidence'] @@ -502,7 +502,8 @@ def build_url_output_with_error_code(action=None, success=None, container=None, "source": "urlscan.io", "source_link": f"{external_id}" } - + if parsed_url.hostname == None: + observable_object['attributes']['hostname'] = url.split("\\",1)[0] if parsed_url.path: observable_object['attributes']['path'] = parsed_url.path if parsed_url.query: @@ -555,7 +556,7 @@ def build_url_output_with_no_error_code(action=None, success=None, container=Non observable_object = { "value": url, "type": "url", - "sandbox": { + "reputation": { "score_id": url_object['score_id'], "score": url_object['score'], "confidence": url_object['confidence'] @@ -569,6 +570,8 @@ def build_url_output_with_no_error_code(action=None, success=None, container=Non "source": "urlscan.io", "source_link": f"{external_id}" } + if parsed_url.hostname == None: + observable_object['attributes']['hostname'] = url.split("\\",1)[0] if parsed_url.path: observable_object['attributes']['path'] = parsed_url.path if parsed_url.query: