Branch was auto-updated.

This commit is contained in:
srv-rr-gh-researchbt
2023-04-06 07:38:00 -07:00
committed by GitHub
6 changed files with 154 additions and 128 deletions
File diff suppressed because one or more lines are too long
+28 -23
View File
@@ -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
"""
@@ -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:
@@ -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 = [
@@ -411,18 +411,18 @@ 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",
"sandbox": {
"reputation": {
"score": url_object['score'],
"confidence": url_object['confidence']
},
@@ -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
@@ -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 = [
@@ -767,12 +767,12 @@ 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,
"type": "hash",
"sandbox": {
"reputation": {
"score": file_object['score'],
"confidence": file_object['confidence'],
@@ -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:
@@ -839,13 +843,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 +861,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
+2 -3
View File
@@ -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: []
File diff suppressed because one or more lines are too long
+29 -26
View File
@@ -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))
################################################################################
@@ -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 = [
@@ -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,20 +407,20 @@ 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.
################################################################################
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 = [
@@ -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,16 +479,16 @@ 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",
"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:
@@ -548,14 +549,14 @@ 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",
"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:
@@ -591,13 +594,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 +612,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
@@ -13,7 +13,6 @@ app_list:
- urlscan.io
tags:
platform_tags:
- reputation
- url
- domain
- sandbox