mirror of
https://github.com/splunk/security_content
synced 2026-06-08 17:32:49 +00:00
Branch was auto-updated.
This commit is contained in:
srv-rr-gh-researchbt
File diff suppressed because one or more lines are too long
Binary file not shown.
|
After Width: | Height: | Size: 77 KiB |
@@ -0,0 +1,866 @@
|
||||
"""
|
||||
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/
|
||||
"""
|
||||
|
||||
|
||||
import phantom.rules as phantom
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def on_start(container):
|
||||
phantom.debug('on_start() called')
|
||||
|
||||
# call 'saa_input_filter' block
|
||||
saa_input_filter(container=container)
|
||||
|
||||
return
|
||||
|
||||
@phantom.playbook_block()
|
||||
def saa_input_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("saa_input_filter() called")
|
||||
|
||||
################################################################################
|
||||
# Determine branches based on provided inputs.
|
||||
################################################################################
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 1
|
||||
matched_artifacts_1, matched_results_1 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["playbook_input:url", "!=", ""]
|
||||
],
|
||||
name="saa_input_filter:condition_1")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_1 or matched_results_1:
|
||||
saa_url_detonation(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 2
|
||||
matched_artifacts_2, matched_results_2 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["playbook_input:vault_id", "!=", ""]
|
||||
],
|
||||
name="saa_input_filter:condition_2")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_2 or matched_results_2:
|
||||
saa_file_detonation(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_2, filtered_results=matched_results_2)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def saa_url_detonation(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("saa_url_detonation() called")
|
||||
|
||||
# phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
|
||||
|
||||
################################################################################
|
||||
# Queries SAA for information about the provided URL(s)
|
||||
################################################################################
|
||||
|
||||
playbook_input_url = phantom.collect2(container=container, datapath=["playbook_input:url"])
|
||||
|
||||
parameters = []
|
||||
|
||||
# build parameters list for 'saa_url_detonation' call
|
||||
for playbook_input_url_item in playbook_input_url:
|
||||
if playbook_input_url_item[0] is not None:
|
||||
parameters.append({
|
||||
"url": playbook_input_url_item[0],
|
||||
})
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.act("detonate url", parameters=parameters, name="saa_url_detonation", assets=["splunk attack analyzer"], callback=filter_5)
|
||||
|
||||
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")
|
||||
|
||||
################################################################################
|
||||
# Filters successful url or file detonation results.
|
||||
################################################################################
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 1
|
||||
matched_artifacts_1, matched_results_1 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["saa_file_detonation:action_result.status", "==", "success"]
|
||||
],
|
||||
name="url_detonation_status_filter:condition_1")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_1 or matched_results_1:
|
||||
get_jobid_of_file_detonation_output(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def get_jobid_of_url_detonation_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("get_jobid_of_url_detonation_output() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code for fetching JobID for URL(s) or file(s) detonation.
|
||||
################################################################################
|
||||
|
||||
saa_url_detonation_result_data = phantom.collect2(container=container, datapath=["saa_url_detonation:action_result.data.*.JobID"], action_results=results)
|
||||
|
||||
saa_url_detonation_result_item_0 = [item[0] for item in saa_url_detonation_result_data]
|
||||
|
||||
get_jobid_of_url_detonation_output__jobid = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
get_jobid_of_url_detonation_output__jobid = []
|
||||
|
||||
get_jobid_of_url_detonation_output__jobid.append(saa_url_detonation_result_item_0)
|
||||
#phantom.debug("get_jobid_of_url_detonation_output__jobid: {}".format(get_jobid_of_url_detonation_output__jobid))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="get_jobid_of_url_detonation_output:jobid", value=json.dumps(get_jobid_of_url_detonation_output__jobid))
|
||||
|
||||
ssa_get_job_forensics_output(container=container)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def ssa_get_job_forensics_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("ssa_get_job_forensics_output() called")
|
||||
|
||||
# phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
|
||||
|
||||
################################################################################
|
||||
# Queries SAA Forensics data relative to the JobID of URL(s) or File(s) needs
|
||||
# to be detonated.
|
||||
################################################################################
|
||||
|
||||
get_jobid_of_url_detonation_output__jobid = json.loads(_ if (_ := phantom.get_run_data(key="get_jobid_of_url_detonation_output:jobid")) != "" else "null") # pylint: disable=used-before-assignment
|
||||
|
||||
parameters = []
|
||||
|
||||
if get_jobid_of_url_detonation_output__jobid is not None:
|
||||
parameters.append({
|
||||
"job_id": get_jobid_of_url_detonation_output__jobid,
|
||||
"timeout": 5,
|
||||
})
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
parameters = []
|
||||
for job_ids in get_jobid_of_url_detonation_output__jobid:
|
||||
for job in job_ids:
|
||||
if job is not None:
|
||||
parameters.append({
|
||||
"job_id": job,
|
||||
"timeout": 5,
|
||||
})
|
||||
#phantom.debug(parameters)
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.act("get job forensics", parameters=parameters, name="ssa_get_job_forensics_output", assets=["splunk attack analyzer"], callback=get_jobid_forensic_filter)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def get_jobid_forensic_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("get_jobid_forensic_filter() called")
|
||||
|
||||
################################################################################
|
||||
# Filters successful url or file detonation job forensic results.
|
||||
################################################################################
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 1
|
||||
matched_artifacts_1, matched_results_1 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["ssa_get_job_forensics_output:action_result.status", "==", "success"]
|
||||
],
|
||||
name="get_jobid_forensic_filter:condition_1")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_1 or matched_results_1:
|
||||
normalized_job_forensic_report_output(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def normalized_job_forensic_report_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("normalized_job_forensic_report_output() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code for normalizing score. Adjust the logic as desired
|
||||
# in the documented sections.
|
||||
################################################################################
|
||||
|
||||
ssa_get_job_forensics_output_result_data = phantom.collect2(container=container, datapath=["ssa_get_job_forensics_output:action_result.data.*.URLs.*.URL","ssa_get_job_forensics_output:action_result.data.*.DisplayScore","ssa_get_job_forensics_output:action_result.data.*.Detections.*.Description","ssa_get_job_forensics_output:action_result.data.*.Verdict","ssa_get_job_forensics_output:action_result.data"], action_results=results)
|
||||
|
||||
ssa_get_job_forensics_output_result_item_0 = [item[0] for item in ssa_get_job_forensics_output_result_data]
|
||||
ssa_get_job_forensics_output_result_item_1 = [item[1] for item in ssa_get_job_forensics_output_result_data]
|
||||
ssa_get_job_forensics_output_result_item_2 = [item[2] for item in ssa_get_job_forensics_output_result_data]
|
||||
ssa_get_job_forensics_output_result_item_3 = [item[3] for item in ssa_get_job_forensics_output_result_data]
|
||||
ssa_get_job_forensics_output_result_item_4 = [item[4] for item in ssa_get_job_forensics_output_result_data]
|
||||
|
||||
normalized_job_forensic_report_output__url_score_object = None
|
||||
normalized_job_forensic_report_output__scores = None
|
||||
normalized_job_forensic_report_output__categories = None
|
||||
normalized_job_forensic_report_output__confidence = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
score_id =0
|
||||
score_table = {
|
||||
"0":"Unknown",
|
||||
"10":"Very_Safe",
|
||||
"20":"Safe",
|
||||
"30":"Probably_Safe",
|
||||
"40":"Leans_Safe",
|
||||
"50":"May_not_be_Safe",
|
||||
"60":"Exercise_Caution",
|
||||
"70":"Suspicious_or_Risky",
|
||||
"80":"Possibly_Malicious",
|
||||
"90":"Probably_Malicious",
|
||||
"100":"Malicious"
|
||||
}
|
||||
#phantom.debug("url: {}".format(ssa_get_job_forensics_output_result_item_0))
|
||||
#phantom.debug("DisplayScore: {}".format(ssa_get_job_forensics_output_result_item_1))
|
||||
#phantom.debug("Category: {}".format(ssa_get_job_forensics_output_result_item_2))
|
||||
#phantom.debug("verdict: {}".format(ssa_get_job_forensics_output_result_item_3))
|
||||
#phantom.debug("action_data: {}".format(ssa_get_job_forensics_output_result_item_4))
|
||||
|
||||
|
||||
normalized_job_forensic_report_output__url_score_object = []
|
||||
normalized_job_forensic_report_output__scores = []
|
||||
normalized_job_forensic_report_output__categories = []
|
||||
normalized_job_forensic_report_output__confidence = []
|
||||
|
||||
## normalized NoneType value to avoid enumeration failure
|
||||
url_detonation_param_list = [(i or "") for i in ssa_get_job_forensics_output_result_item_0]
|
||||
url_detonation_threat_score_list = [(i or 0) for i in ssa_get_job_forensics_output_result_item_1]
|
||||
url_detonation_category_list = [(i or "") for i in ssa_get_job_forensics_output_result_item_2]
|
||||
url_detonation_verdict_list = [(i or "") for i in ssa_get_job_forensics_output_result_item_3]
|
||||
|
||||
## get the set() or unique input url parameter.
|
||||
|
||||
index_url_dict = {}
|
||||
set_url_inputs = set(url_detonation_param_list)
|
||||
|
||||
for url_input in set_url_inputs:
|
||||
url_list = []
|
||||
score_list = []
|
||||
display_score_list = []
|
||||
category_list = []
|
||||
|
||||
## getting the index of each detonation phase of the url group the result for each url detonation
|
||||
url_input_index = [indx for indx, url_val in enumerate(url_detonation_param_list) if url_val == url_input]
|
||||
index_url_dict[url_input] = url_input_index
|
||||
|
||||
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)):
|
||||
if _url == url_input and idx in index_url_dict[url_input]:
|
||||
url_list.append(_url)
|
||||
score_list.append(_score)
|
||||
display_score_list.append(_display_score)
|
||||
category_list.append(_category)
|
||||
|
||||
## if score_list is empty or it has one element but empty string, lets score it base on confidence score of its engine detonation
|
||||
#phantom.debug("score_list: {} len: {}".format(score_list, len(score_list)))
|
||||
#phantom.debug("category_list: {} len: {}".format(category_list, len(category_list)))
|
||||
confidence_ = list(set(display_score_list))[0]
|
||||
categories = list(set(category_list))
|
||||
|
||||
if len(score_list) == 0 or (len(set(score_list)) == 1 and score_list[0] == ""):
|
||||
if confidence_ >= 0 and confidence_ < 10:
|
||||
score_id = 0
|
||||
elif confidence_ >= 10 and confidence_ < 20:
|
||||
score_id = 10
|
||||
elif confidence_ >= 20 and confidence_ < 30:
|
||||
score_id = 20
|
||||
elif confidence_ >= 30 and confidence_ < 40:
|
||||
score_id = 30
|
||||
elif confidence_ >= 40 and confidence_ < 50:
|
||||
score_id = 40
|
||||
elif confidence_ >= 50 and confidence_ < 60:
|
||||
score_id = 50
|
||||
elif confidence_ >= 60 and confidence_ < 70:
|
||||
score_id = 60
|
||||
elif confidence_ >= 70 and confidence_ < 80:
|
||||
score_id = 70
|
||||
elif confidence_ >= 80 and confidence_ < 90:
|
||||
score_id = 80
|
||||
elif confidence_ >= 90 and confidence_ < 100:
|
||||
score_id = 90
|
||||
elif confidence_ >= 100:
|
||||
score_id = 100
|
||||
|
||||
score = score_table[str(score_id)]
|
||||
|
||||
else:
|
||||
score = list(set(score_list))[0]
|
||||
|
||||
# Attach final object
|
||||
normalized_job_forensic_report_output__url_score_object.append({'score': score, 'confidence': confidence_, 'categories': categories})
|
||||
normalized_job_forensic_report_output__scores.append(score)
|
||||
normalized_job_forensic_report_output__categories.append(categories)
|
||||
normalized_job_forensic_report_output__confidence.append(confidence_)
|
||||
#phantom.debug("normalized_job_forensic_report_output__url_score_object: {}".format(normalized_job_forensic_report_output__url_score_object))
|
||||
#phantom.debug("normalized_job_forensic_report_output__categories: {}".format(normalized_job_forensic_report_output__categories))
|
||||
#phantom.debug("normalized_job_forensic_report_output__confidence: {}".format(normalized_job_forensic_report_output__confidence))
|
||||
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="normalized_job_forensic_report_output:url_score_object", value=json.dumps(normalized_job_forensic_report_output__url_score_object))
|
||||
phantom.save_run_data(key="normalized_job_forensic_report_output:scores", value=json.dumps(normalized_job_forensic_report_output__scores))
|
||||
phantom.save_run_data(key="normalized_job_forensic_report_output:categories", value=json.dumps(normalized_job_forensic_report_output__categories))
|
||||
phantom.save_run_data(key="normalized_job_forensic_report_output:confidence", value=json.dumps(normalized_job_forensic_report_output__confidence))
|
||||
|
||||
format_url_report(container=container)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def format_url_report(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("format_url_report() called")
|
||||
|
||||
################################################################################
|
||||
# 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"""
|
||||
|
||||
# parameter list for template variable replacement
|
||||
parameters = [
|
||||
"playbook_input:url",
|
||||
"normalized_job_forensic_report_output:custom_function:scores",
|
||||
"normalized_job_forensic_report_output:custom_function:confidence",
|
||||
"normalized_job_forensic_report_output:custom_function:categories",
|
||||
"get_jobid_of_url_detonation_output:custom_function:jobid"
|
||||
]
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
#phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name="format_report_url"))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.format(container=container, template=template, parameters=parameters, name="format_url_report")
|
||||
|
||||
build_url_output(container=container)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def build_url_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("build_url_output() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code to generate an observable dictionary to output into
|
||||
# the observables data path.
|
||||
################################################################################
|
||||
|
||||
playbook_input_url = phantom.collect2(container=container, datapath=["playbook_input:url"])
|
||||
get_jobid_of_url_detonation_output__jobid = json.loads(_ if (_ := phantom.get_run_data(key="get_jobid_of_url_detonation_output:jobid")) != "" else "null") # pylint: disable=used-before-assignment
|
||||
normalized_job_forensic_report_output__url_score_object = json.loads(_ if (_ := phantom.get_run_data(key="normalized_job_forensic_report_output:url_score_object")) != "" else "null") # pylint: disable=used-before-assignment
|
||||
|
||||
playbook_input_url_values = [item[0] for item in playbook_input_url]
|
||||
|
||||
build_url_output__observable_array = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
from urllib.parse import urlparse
|
||||
build_url_output__observable_array = []
|
||||
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))
|
||||
observable_object = {
|
||||
"value": url,
|
||||
"type": "url",
|
||||
"sandbox": {
|
||||
"score": url_object['score'],
|
||||
"confidence": url_object['confidence']
|
||||
},
|
||||
"attributes": {
|
||||
"hostname": parsed_url.hostname,
|
||||
"scheme": parsed_url.scheme
|
||||
},
|
||||
"categories": url_object['categories'],
|
||||
"source": "Splunk Attack Analyzer (SAA)",
|
||||
"source_link": f"https://app.twinwave.io/job/{external_id}"
|
||||
}
|
||||
|
||||
if parsed_url.path:
|
||||
observable_object['attributes']['path'] = parsed_url.path
|
||||
if parsed_url.query:
|
||||
observable_object['attributes']['query'] = parsed_url.query
|
||||
if parsed_url.port:
|
||||
observable_object['attributes']['port'] = parsed_url.port
|
||||
|
||||
build_url_output__observable_array.append(observable_object)
|
||||
#phantom.debug("build_url_output__observable_array: {}".format(build_url_output__observable_array))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="build_url_output:observable_array", value=json.dumps(build_url_output__observable_array))
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def saa_file_detonation(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("saa_file_detonation() called")
|
||||
|
||||
# phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
|
||||
|
||||
################################################################################
|
||||
# Queries SAA for information about the provided vault_id(s)
|
||||
################################################################################
|
||||
|
||||
playbook_input_vault_id = phantom.collect2(container=container, datapath=["playbook_input:vault_id"])
|
||||
|
||||
parameters = []
|
||||
|
||||
# build parameters list for 'saa_file_detonation' call
|
||||
for playbook_input_vault_id_item in playbook_input_vault_id:
|
||||
if playbook_input_vault_id_item[0] is not None:
|
||||
parameters.append({
|
||||
"file": playbook_input_vault_id_item[0],
|
||||
})
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.act("detonate file", parameters=parameters, name="saa_file_detonation", assets=["splunk attack analyzer"], callback=url_detonation_status_filter)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def get_jobid_of_file_detonation_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("get_jobid_of_file_detonation_output() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code for fetching JobID for URL(s) or file(s) detonation.
|
||||
################################################################################
|
||||
|
||||
saa_file_detonation_result_data = phantom.collect2(container=container, datapath=["saa_file_detonation:action_result.data.*.JobID"], action_results=results)
|
||||
|
||||
saa_file_detonation_result_item_0 = [item[0] for item in saa_file_detonation_result_data]
|
||||
|
||||
get_jobid_of_file_detonation_output__jobid = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
get_jobid_of_file_detonation_output__jobid = []
|
||||
|
||||
get_jobid_of_file_detonation_output__jobid.append(saa_file_detonation_result_item_0)
|
||||
#phantom.debug("get_jobid_of_file_detonation_output__jobid: {}".format(get_jobid_of_file_detonation_output__jobid))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="get_jobid_of_file_detonation_output:jobid", value=json.dumps(get_jobid_of_file_detonation_output__jobid))
|
||||
|
||||
saa_get_file_job_forensics_output(container=container)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def saa_get_file_job_forensics_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("saa_get_file_job_forensics_output() called")
|
||||
|
||||
# phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
|
||||
|
||||
################################################################################
|
||||
# Queries SAA Forensics data relative to the JobID of URL(s) or File(s) needs
|
||||
# to be detonated.
|
||||
################################################################################
|
||||
|
||||
get_jobid_of_file_detonation_output__jobid = json.loads(_ if (_ := phantom.get_run_data(key="get_jobid_of_file_detonation_output:jobid")) != "" else "null") # pylint: disable=used-before-assignment
|
||||
|
||||
parameters = []
|
||||
|
||||
if get_jobid_of_file_detonation_output__jobid is not None:
|
||||
parameters.append({
|
||||
"job_id": get_jobid_of_file_detonation_output__jobid,
|
||||
"timeout": 5,
|
||||
})
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
parameters = []
|
||||
for job_ids in get_jobid_of_file_detonation_output__jobid:
|
||||
for job in job_ids:
|
||||
if job is not None:
|
||||
parameters.append({
|
||||
"job_id": job,
|
||||
"timeout": 5,
|
||||
})
|
||||
#phantom.debug(parameters)
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.act("get job forensics", parameters=parameters, name="saa_get_file_job_forensics_output", assets=["splunk attack analyzer"], callback=filter_6)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def normalized_job_forensic_report_output_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("normalized_job_forensic_report_output_1() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code for normalizing score. Adjust the logic as desired
|
||||
# in the documented sections.
|
||||
################################################################################
|
||||
|
||||
playbook_input_vault_id = phantom.collect2(container=container, datapath=["playbook_input:vault_id"])
|
||||
saa_get_file_job_forensics_output_result_data = phantom.collect2(container=container, datapath=["saa_get_file_job_forensics_output:action_result.data.*.DisplayScore","saa_get_file_job_forensics_output:action_result.data.*.Detections.*.Description","saa_get_file_job_forensics_output:action_result.data.*.Verdict","saa_get_file_job_forensics_output:action_result.data"], action_results=results)
|
||||
|
||||
playbook_input_vault_id_values = [item[0] for item in playbook_input_vault_id]
|
||||
saa_get_file_job_forensics_output_result_item_0 = [item[0] for item in saa_get_file_job_forensics_output_result_data]
|
||||
saa_get_file_job_forensics_output_result_item_1 = [item[1] for item in saa_get_file_job_forensics_output_result_data]
|
||||
saa_get_file_job_forensics_output_result_item_2 = [item[2] for item in saa_get_file_job_forensics_output_result_data]
|
||||
saa_get_file_job_forensics_output_result_item_3 = [item[3] for item in saa_get_file_job_forensics_output_result_data]
|
||||
|
||||
normalized_job_forensic_report_output_1__file_score_object = None
|
||||
normalized_job_forensic_report_output_1__scores = None
|
||||
normalized_job_forensic_report_output_1__categories = None
|
||||
normalized_job_forensic_report_output_1__confidence = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
score_id =0
|
||||
score_table = {
|
||||
"0":"Unknown",
|
||||
"10":"Very_Safe",
|
||||
"20":"Safe",
|
||||
"30":"Probably_Safe",
|
||||
"40":"Leans_Safe",
|
||||
"50":"May_not_be_Safe",
|
||||
"60":"Exercise_Caution",
|
||||
"70":"Suspicious_or_Risky",
|
||||
"80":"Possibly_Malicious",
|
||||
"90":"Probably_Malicious",
|
||||
"100":"Malicious"
|
||||
}
|
||||
#phantom.debug("vault_id: {}".format(ssa_get_job_forensics_output_result_item_0))
|
||||
#phantom.debug("DisplayScore: {}".format(ssa_get_job_forensics_output_result_item_1))
|
||||
#phantom.debug("Category: {}".format(ssa_get_job_forensics_output_result_item_2))
|
||||
#phantom.debug("verdict: {}".format(ssa_get_job_forensics_output_result_item_3))
|
||||
#phantom.debug("action_data: {}".format(ssa_get_job_forensics_output_result_item_4))
|
||||
|
||||
|
||||
normalized_job_forensic_report_output_1__file_score_object = []
|
||||
normalized_job_forensic_report_output_1__scores = []
|
||||
normalized_job_forensic_report_output_1__categories = []
|
||||
normalized_job_forensic_report_output_1__confidence = []
|
||||
|
||||
## normalized NoneType value to avoid enumeration failure
|
||||
file_detonation_param_list = [(i or "") for i in playbook_input_vault_id_values]
|
||||
file_detonation_threat_score_list = [(i or 0) for i in saa_get_file_job_forensics_output_result_item_0]
|
||||
file_detonation_category_list = [(i or "") for i in saa_get_file_job_forensics_output_result_item_1]
|
||||
file_detonation_verdict_list = [(i or "") for i in saa_get_file_job_forensics_output_result_item_2]
|
||||
|
||||
## get the set() or unique input url parameter.
|
||||
|
||||
index_file_dict = {}
|
||||
set_file_inputs = set(file_detonation_param_list)
|
||||
|
||||
for file_input in set_file_inputs:
|
||||
vaultid_list = []
|
||||
score_list = []
|
||||
display_score_list = []
|
||||
category_list = []
|
||||
|
||||
## getting the index of each detonation phase of the url group the result for each url detonation
|
||||
file_input_index = [indx for indx, vaultid_val in enumerate(file_detonation_param_list) if vaultid_val == file_input]
|
||||
index_file_dict[file_input] = file_input_index
|
||||
|
||||
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)):
|
||||
if _vaultid == file_input and idx in index_file_dict[file_input]:
|
||||
vaultid_list.append(_vaultid)
|
||||
score_list.append(_score)
|
||||
display_score_list.append(_display_score)
|
||||
category_list.append(_category)
|
||||
|
||||
## if score_list is empty or it has one element but empty string, lets score it base on confidence score of its engine detonation
|
||||
#phantom.debug("score_list: {} len: {}".format(score_list, len(score_list)))
|
||||
#phantom.debug("category_list: {} len: {}".format(category_list, len(category_list)))
|
||||
confidence_ = list(set(display_score_list))[0]
|
||||
categories = list(set(category_list))
|
||||
|
||||
if len(score_list) == 0 or (len(set(score_list)) == 1 and score_list[0] == ""):
|
||||
if confidence_ >= 0 and confidence_ < 10:
|
||||
score_id = 0
|
||||
elif confidence_ >= 10 and confidence_ < 20:
|
||||
score_id = 10
|
||||
elif confidence_ >= 20 and confidence_ < 30:
|
||||
score_id = 20
|
||||
elif confidence_ >= 30 and confidence_ < 40:
|
||||
score_id = 30
|
||||
elif confidence_ >= 40 and confidence_ < 50:
|
||||
score_id = 40
|
||||
elif confidence_ >= 50 and confidence_ < 60:
|
||||
score_id = 50
|
||||
elif confidence_ >= 60 and confidence_ < 70:
|
||||
score_id = 60
|
||||
elif confidence_ >= 70 and confidence_ < 80:
|
||||
score_id = 70
|
||||
elif confidence_ >= 80 and confidence_ < 90:
|
||||
score_id = 80
|
||||
elif confidence_ >= 90 and confidence_ < 100:
|
||||
score_id = 90
|
||||
elif confidence_ >= 100:
|
||||
score_id = 100
|
||||
|
||||
score = score_table[str(score_id)]
|
||||
|
||||
else:
|
||||
score = list(set(score_list))[0]
|
||||
|
||||
# Attach final object
|
||||
normalized_job_forensic_report_output_1__file_score_object.append({'score': score, 'confidence': confidence_, 'categories': categories})
|
||||
normalized_job_forensic_report_output_1__scores.append(score)
|
||||
normalized_job_forensic_report_output_1__categories.append(categories)
|
||||
normalized_job_forensic_report_output_1__confidence.append(confidence_)
|
||||
#phantom.debug("normalized_job_forensic_report_output_1__file_score_object: {}".format(normalized_job_forensic_report_output_1__file_score_object))
|
||||
#phantom.debug("normalized_job_forensic_report_output_1__scores: {}".format(normalized_job_forensic_report_output_1__scores))
|
||||
#phantom.debug("normalized_job_forensic_report_output_1__categories: {}".format(normalized_job_forensic_report_output_1__categories))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="normalized_job_forensic_report_output_1:file_score_object", value=json.dumps(normalized_job_forensic_report_output_1__file_score_object))
|
||||
phantom.save_run_data(key="normalized_job_forensic_report_output_1:scores", value=json.dumps(normalized_job_forensic_report_output_1__scores))
|
||||
phantom.save_run_data(key="normalized_job_forensic_report_output_1:categories", value=json.dumps(normalized_job_forensic_report_output_1__categories))
|
||||
phantom.save_run_data(key="normalized_job_forensic_report_output_1:confidence", value=json.dumps(normalized_job_forensic_report_output_1__confidence))
|
||||
|
||||
format_file_report(container=container)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def format_file_report(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("format_file_report() called")
|
||||
|
||||
################################################################################
|
||||
# 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"""
|
||||
|
||||
# parameter list for template variable replacement
|
||||
parameters = [
|
||||
"playbook_input:vault_id",
|
||||
"normalized_job_forensic_report_output_1:custom_function:scores",
|
||||
"normalized_job_forensic_report_output_1:custom_function:confidence",
|
||||
"normalized_job_forensic_report_output_1:custom_function:categories",
|
||||
"get_jobid_of_file_detonation_output:custom_function:jobid"
|
||||
]
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
#phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name="format_report_file"))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.format(container=container, template=template, parameters=parameters, name="format_file_report")
|
||||
|
||||
build_file_output(container=container)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def build_file_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("build_file_output() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code to generate an observable dictionary to output into
|
||||
# the observables data path.
|
||||
################################################################################
|
||||
|
||||
playbook_input_vault_id = phantom.collect2(container=container, datapath=["playbook_input:vault_id"])
|
||||
get_jobid_of_file_detonation_output__jobid = json.loads(_ if (_ := phantom.get_run_data(key="get_jobid_of_file_detonation_output:jobid")) != "" else "null") # pylint: disable=used-before-assignment
|
||||
normalized_job_forensic_report_output_1__file_score_object = json.loads(_ if (_ := phantom.get_run_data(key="normalized_job_forensic_report_output_1:file_score_object")) != "" else "null") # pylint: disable=used-before-assignment
|
||||
|
||||
playbook_input_vault_id_values = [item[0] for item in playbook_input_vault_id]
|
||||
|
||||
build_file_output__observable_array = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
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))
|
||||
observable_object = {
|
||||
|
||||
"value": _vault_id,
|
||||
"type": "hash",
|
||||
"sandbox": {
|
||||
"score": file_object['score'],
|
||||
"confidence": file_object['confidence'],
|
||||
|
||||
},
|
||||
"enrichment": {
|
||||
"provider": "Splunk Attack Analyzer",
|
||||
"type": "file",
|
||||
|
||||
},
|
||||
"categories": file_object['categories'],
|
||||
"source": "Splunk Attack Analyzer (SAA)",
|
||||
"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))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="build_file_output:observable_array", value=json.dumps(build_file_output__observable_array))
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def 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")
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 1
|
||||
matched_artifacts_1, matched_results_1 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["saa_url_detonation:action_result.status", "==", "success"]
|
||||
],
|
||||
name="filter_5:condition_1")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_1 or matched_results_1:
|
||||
get_jobid_of_url_detonation_output(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def filter_6(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("filter_6() called")
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 1
|
||||
matched_artifacts_1, matched_results_1 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["saa_get_file_job_forensics_output:action_result.status", "==", "success"]
|
||||
],
|
||||
name="filter_6:condition_1")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_1 or matched_results_1:
|
||||
normalized_job_forensic_report_output_1(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def on_finish(container, summary):
|
||||
phantom.debug("on_finish() called")
|
||||
|
||||
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)
|
||||
|
||||
output = {
|
||||
"observable": observable_combined_value,
|
||||
}
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# 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
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_playbook_output_data(output=output)
|
||||
|
||||
return
|
||||
@@ -0,0 +1,25 @@
|
||||
name: SAA Dynamic Analysis
|
||||
id: 2754c122-c63b-4558-9b83-23c1e1b96177
|
||||
version: 1
|
||||
date: '2023-03-24'
|
||||
author: Teoderick Contreras, Splunk
|
||||
type: Investigation
|
||||
description: "Accepts url link, domain or vault_id (hash) to be detonated using Splunk Attacker (SAA) API connector. This playbook produces a normalized output for each user and device."
|
||||
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
|
||||
tags:
|
||||
platform_tags:
|
||||
- reputation
|
||||
- url
|
||||
- domain
|
||||
- sandbox
|
||||
- ip
|
||||
- file_hash
|
||||
playbook_type: Input
|
||||
vpe_type: Modern
|
||||
playbook_fields: []
|
||||
product:
|
||||
- Splunk SOAR
|
||||
File diff suppressed because one or more lines are too long
Binary file not shown.
|
After Width: | Height: | Size: 71 KiB |
@@ -0,0 +1,619 @@
|
||||
"""
|
||||
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/
|
||||
"""
|
||||
|
||||
|
||||
import phantom.rules as phantom
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def on_start(container):
|
||||
phantom.debug('on_start() called')
|
||||
|
||||
# call 'url_input_filter' block
|
||||
url_input_filter(container=container)
|
||||
|
||||
return
|
||||
|
||||
@phantom.playbook_block()
|
||||
def url_input_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("url_input_filter() called")
|
||||
|
||||
################################################################################
|
||||
# Determine branches based on provided inputs.
|
||||
################################################################################
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 1
|
||||
matched_artifacts_1, matched_results_1 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["playbook_input:url", "!=", ""]
|
||||
],
|
||||
name="url_input_filter:condition_1")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_1 or matched_results_1:
|
||||
url_reputation(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def url_reputation(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("url_reputation() called")
|
||||
|
||||
# phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
|
||||
|
||||
################################################################################
|
||||
# Queries urlscan.io for information about the provided URL(s)
|
||||
################################################################################
|
||||
|
||||
playbook_input_url = phantom.collect2(container=container, datapath=["playbook_input:url"])
|
||||
|
||||
parameters = []
|
||||
|
||||
# build parameters list for 'url_reputation' call
|
||||
for playbook_input_url_item in playbook_input_url:
|
||||
if playbook_input_url_item[0] is not None:
|
||||
parameters.append({
|
||||
"url": playbook_input_url_item[0],
|
||||
"private": True,
|
||||
"get_result": True,
|
||||
})
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.act("detonate url", parameters=parameters, name="url_reputation", assets=["urlscan.io"], callback=urlscanio_summary_filter)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def urlscanio_summary_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("urlscanio_summary_filter() called")
|
||||
|
||||
################################################################################
|
||||
# Filters successful url reputation results.
|
||||
################################################################################
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 1
|
||||
matched_artifacts_1, matched_results_1 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["url_reputation:action_result.status", "==", "success"]
|
||||
],
|
||||
name="urlscanio_summary_filter:condition_1")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_1 or matched_results_1:
|
||||
urlscanio_error_code_filter(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def normalize_score_url_with_error_code(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("normalize_score_url_with_error_code() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code for normalizing score. Adjust the logic as desired
|
||||
# in the documented sections.
|
||||
################################################################################
|
||||
|
||||
url_reputation_result_data = phantom.collect2(container=container, datapath=["url_reputation:action_result.parameter.url","url_reputation:action_result.data.*.message","url_reputation:action_result.data.*.status","url_reputation:action_result.data.*.description"], action_results=results)
|
||||
|
||||
url_reputation_parameter_url = [item[0] for item in url_reputation_result_data]
|
||||
url_reputation_result_item_1 = [item[1] for item in url_reputation_result_data]
|
||||
url_reputation_result_item_2 = [item[2] for item in url_reputation_result_data]
|
||||
url_reputation_result_item_3 = [item[3] for item in url_reputation_result_data]
|
||||
|
||||
normalize_score_url_with_error_code__url_score_object = None
|
||||
normalize_score_url_with_error_code__scores = None
|
||||
normalize_score_url_with_error_code__categories = None
|
||||
normalize_score_url_with_error_code__confidence = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
url_reputation_data_message = [str(i or '') for i in url_reputation_result_item_1]
|
||||
url_reputation_data_status = [str(i or '') for i in url_reputation_result_item_2]
|
||||
url_reputation_data_description = [str(i or '') for i in url_reputation_result_item_3]
|
||||
|
||||
|
||||
normalize_score_url_with_error_code__url_score_object = []
|
||||
normalize_score_url_with_error_code__scores = []
|
||||
normalize_score_url_with_error_code__categories = []
|
||||
|
||||
|
||||
#phantom.debug("url_reputation_parameter_url: {}".format(url_reputation_parameter_url))
|
||||
#phantom.debug("url_reputation_data_message: {}".format(url_reputation_data_message))
|
||||
#phantom.debug("url_reputation_data_status: {}".format(url_reputation_data_status))
|
||||
#phantom.debug("url_reputation_data_description: {}".format(url_reputation_data_description))
|
||||
|
||||
|
||||
|
||||
|
||||
urlscan_score_table = {
|
||||
"0":"Legitimate",
|
||||
"1":"Very_Safe",
|
||||
"2":"Safe",
|
||||
"3":"Probably_Safe",
|
||||
"4":"Leans_Safe",
|
||||
"5":"May_not_be_Safe",
|
||||
"6":"Exercise_Caution",
|
||||
"7":"Suspicious_or_Risky",
|
||||
"8":"Possibly_Malicious",
|
||||
"9":"Probably_Malicious",
|
||||
"10":"Malicious",
|
||||
"error_code_query" : "error code return, check the error code descriptions"
|
||||
|
||||
}
|
||||
|
||||
url_scan_io_error_code ={
|
||||
"blacklist" : "Blacklisted URL or Domain",
|
||||
"spam" : "Spammy URL or Domain",
|
||||
"invalid_hostname" : "Invalid Hostname URL or Domain",
|
||||
"missing_url" : "Missing URL OR Domain Property",
|
||||
"auth" : "HTTP basic auth information",
|
||||
"not_be_resolved" : "Non-resolvable hostname (A, AAAA, CNAME)"
|
||||
}
|
||||
## URLSCAN.io return error code especially if the url or domain was already in their blacklist database.
|
||||
## below are the common error code message base on their
|
||||
## - "Blacklisted domains and URLs" : requested to be blacklisted by their respective owners.
|
||||
## - "Spammy submissions" : of URLs known to be used only for spamming this service.
|
||||
## - "Invalid hostnames" : or invalid protocol schemes (FTP etc).
|
||||
## - "Missing URL property" : ... yes, it does happen.
|
||||
## - "Contains HTTP basic auth information" : ... yes, that happens as well.
|
||||
## - "Non-resolvable hostnames (A, AAAA, CNAME)" : which we will not even try to scan.
|
||||
|
||||
blank_result = "--"
|
||||
category = ""
|
||||
score = ""
|
||||
error_message = ""
|
||||
|
||||
## check if there is error code return upon URL reputation query
|
||||
|
||||
for url_descp in url_reputation_data_description:
|
||||
|
||||
for key, value in url_scan_io_error_code.items():
|
||||
if key.replace("_"," ").lower() in url_descp.lower():
|
||||
error_message = url_scan_io_error_code[key]
|
||||
|
||||
# Attach final object
|
||||
normalize_score_url_with_error_code__categories.append(error_message)
|
||||
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))
|
||||
|
||||
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="normalize_score_url_with_error_code:url_score_object", value=json.dumps(normalize_score_url_with_error_code__url_score_object))
|
||||
phantom.save_run_data(key="normalize_score_url_with_error_code:scores", value=json.dumps(normalize_score_url_with_error_code__scores))
|
||||
phantom.save_run_data(key="normalize_score_url_with_error_code:categories", value=json.dumps(normalize_score_url_with_error_code__categories))
|
||||
phantom.save_run_data(key="normalize_score_url_with_error_code:confidence", value=json.dumps(normalize_score_url_with_error_code__confidence))
|
||||
|
||||
error_code_format_report_url(container=container)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def 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("error_code_format_report_url() 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\n"""
|
||||
|
||||
# parameter list for template variable replacement
|
||||
parameters = [
|
||||
"url_reputation:action_result.parameter.url",
|
||||
"normalize_score_url_with_error_code:custom_function:confidence",
|
||||
"normalize_score_url_with_error_code:custom_function:scores",
|
||||
"normalize_score_url_with_error_code:custom_function:categories",
|
||||
"url_reputation:action_result.data.*.task.reportURL"
|
||||
]
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
#phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name="error_code_format_report_url"))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.format(container=container, template=template, parameters=parameters, name="error_code_format_report_url")
|
||||
|
||||
build_url_output_with_error_code(container=container)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def urlscanio_error_code_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("urlscanio_error_code_filter() called")
|
||||
|
||||
################################################################################
|
||||
# This filter is designed to avoid nonetype value in some list and dictionary
|
||||
# object due to error code return value of urlscan.io especially if the URL or
|
||||
# domain is in their blacklist database
|
||||
################################################################################
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 1
|
||||
matched_artifacts_1, matched_results_1 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["filtered-data:urlscanio_summary_filter:condition_1:url_reputation:action_result.data.*.status", "==", 400]
|
||||
],
|
||||
name="urlscanio_error_code_filter:condition_1")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_1 or matched_results_1:
|
||||
normalize_score_url_with_error_code(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)
|
||||
|
||||
# collect filtered artifact ids and results for 'if' condition 2
|
||||
matched_artifacts_2, matched_results_2 = phantom.condition(
|
||||
container=container,
|
||||
conditions=[
|
||||
["filtered-data:urlscanio_summary_filter:condition_1:url_reputation:action_result.data.*.status", "!=", 400]
|
||||
],
|
||||
name="urlscanio_error_code_filter:condition_2")
|
||||
|
||||
# call connected blocks if filtered artifacts or results
|
||||
if matched_artifacts_2 or matched_results_2:
|
||||
normalize_score_url_with_no_error_code(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_2, filtered_results=matched_results_2)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def normalize_score_url_with_no_error_code(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("normalize_score_url_with_no_error_code() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code for normalizing score. Adjust the logic as desired
|
||||
# in the documented sections.
|
||||
################################################################################
|
||||
|
||||
url_reputation_result_data = phantom.collect2(container=container, datapath=["url_reputation:action_result.data.*.verdicts.overall","url_reputation:action_result.data.*.verdicts.urlscan","url_reputation:action_result.data.*.verdicts.engines","url_reputation:action_result.data.*.verdicts.community"], action_results=results)
|
||||
|
||||
url_reputation_result_item_0 = [item[0] for item in url_reputation_result_data]
|
||||
url_reputation_result_item_1 = [item[1] for item in url_reputation_result_data]
|
||||
url_reputation_result_item_2 = [item[2] for item in url_reputation_result_data]
|
||||
url_reputation_result_item_3 = [item[3] for item in url_reputation_result_data]
|
||||
|
||||
normalize_score_url_with_no_error_code__url_score_object = None
|
||||
normalize_score_url_with_no_error_code__scores = None
|
||||
normalize_score_url_with_no_error_code__categories = None
|
||||
normalize_score_url_with_no_error_code__confidence = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
|
||||
normalize_score_url_with_no_error_code__url_score_object = []
|
||||
normalize_score_url_with_no_error_code__scores = []
|
||||
normalize_score_url_with_no_error_code__categories = []
|
||||
|
||||
url_reputation_verdicts_overall_dict = [(i or {}) for i in url_reputation_result_item_0]
|
||||
url_reputation_verdicts_urlscan_dict = [(i or {}) for i in url_reputation_result_item_1]
|
||||
url_reputation_verdicts_engine_dict = [(i or {}) for i in url_reputation_result_item_2]
|
||||
url_reputation_verdicts_community_dict = [(i or {}) for i in url_reputation_result_item_3]
|
||||
|
||||
#phantom.debug("url_reputation_verdicts_overall_dict: {}".format(url_reputation_verdicts_overall_dict))
|
||||
#phantom.debug("url_reputation_verdicts_urlscan_dict: {}".format(url_reputation_verdicts_urlscan_dict))
|
||||
#phantom.debug("url_reputation_verdicts_engine_dict: {}".format(url_reputation_verdicts_engine_dict))
|
||||
#phantom.debug("url_reputation_verdicts_community_dict: {}".format(url_reputation_verdicts_community_dict))
|
||||
|
||||
|
||||
urlscan_score_table = {
|
||||
"0":"Legitimate",
|
||||
"1":"Very_Safe",
|
||||
"2":"Safe",
|
||||
"3":"Probably_Safe",
|
||||
"4":"Leans_Safe",
|
||||
"5":"May_not_be_Safe",
|
||||
"6":"Exercise_Caution",
|
||||
"7":"Suspicious_or_Risky",
|
||||
"8":"Possibly_Malicious",
|
||||
"9":"Probably_Malicious",
|
||||
"10":"Malicious"
|
||||
}
|
||||
|
||||
## if there is no error code, urlscan.io will continue to detonate the URL and query scores in several verdicts object
|
||||
|
||||
## Normalize reputation on a -100 (legitimate) to 100 point scale based on number of malicious and suspicious divided by different urlscan.io verdict objects.
|
||||
## This can be adjusted to include whatever logic is desired.
|
||||
|
||||
for i in range(0,len(url_reputation_verdicts_overall_dict)):
|
||||
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] != {}:
|
||||
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']
|
||||
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']
|
||||
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'])
|
||||
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']
|
||||
|
||||
|
||||
## customized score id calculation
|
||||
|
||||
log_result = (summary_score/4) # avg score from different urlscan.io score object (engine_score, overall_verdict_score, urlscan_verdicts_score and community score)
|
||||
score_id = int(log_result)
|
||||
|
||||
|
||||
if score_id < -50:
|
||||
score_id = "0"
|
||||
elif score_id < 0 and score_id >= -50:
|
||||
score_id = "1"
|
||||
elif score_id >= 0 and score_id <= 10:
|
||||
score_id = "2"
|
||||
elif score_id > 10 and score_id <= 20:
|
||||
score_id = "3"
|
||||
elif score_id > 20 and score_id <= 30:
|
||||
score_id = "4"
|
||||
elif score_id > 30 and score_id <= 40:
|
||||
score_id = "5"
|
||||
elif score_id > 40 and score_id <= 50:
|
||||
score_id = "6"
|
||||
elif score_id > 50 and score_id <= 60:
|
||||
score_id = "7"
|
||||
elif score_id > 70 and score_id <= 80:
|
||||
score_id = "8"
|
||||
elif score_id > 80 and score_id <= 90:
|
||||
score_id = "9"
|
||||
elif score_id > 90 and score_id <= 100:
|
||||
score_id = "10"
|
||||
|
||||
score = urlscan_score_table[str(score_id)]
|
||||
|
||||
malicious_tag_stats = (summary_of_malicious_tag, 4)
|
||||
|
||||
# Attach final object
|
||||
normalize_score_url_with_no_error_code__categories.append(summary_categories)
|
||||
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))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="normalize_score_url_with_no_error_code:url_score_object", value=json.dumps(normalize_score_url_with_no_error_code__url_score_object))
|
||||
phantom.save_run_data(key="normalize_score_url_with_no_error_code:scores", value=json.dumps(normalize_score_url_with_no_error_code__scores))
|
||||
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)
|
||||
|
||||
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")
|
||||
|
||||
################################################################################
|
||||
# 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"""
|
||||
|
||||
# parameter list for template variable replacement
|
||||
parameters = [
|
||||
"url_reputation:action_result.parameter.url",
|
||||
"normalize_score_url_with_no_error_code:custom_function:scores",
|
||||
"normalize_score_url_with_no_error_code:custom_function:confidence",
|
||||
"normalize_score_url_with_no_error_code:custom_function:categories",
|
||||
"url_reputation:action_result.data.*.task.reportURL"
|
||||
]
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
#phantom.debug(phantom.format(container=container, template=template, parameters=parameters, name="no_error_code_format_report_url"))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.format(container=container, template=template, parameters=parameters, name="no_error_code_format_report_url")
|
||||
|
||||
build_url_output_with_no_error_code(container=container)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def build_url_output_with_error_code(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("build_url_output_with_error_code() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code to generate an observable dictionary to output into
|
||||
# the observables data path.
|
||||
################################################################################
|
||||
|
||||
url_reputation_result_data = phantom.collect2(container=container, datapath=["url_reputation:action_result.parameter.url","url_reputation:action_result.data.*.task.reportURL"], action_results=results)
|
||||
normalize_score_url_with_error_code__url_score_object = json.loads(_ if (_ := phantom.get_run_data(key="normalize_score_url_with_error_code:url_score_object")) != "" else "null") # pylint: disable=used-before-assignment
|
||||
|
||||
url_reputation_parameter_url = [item[0] for item in url_reputation_result_data]
|
||||
url_reputation_result_item_1 = [item[1] for item in url_reputation_result_data]
|
||||
|
||||
build_url_output_with_error_code__observable_array = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
# Write your custom code here...
|
||||
from urllib.parse import urlparse
|
||||
build_url_output_with_error_code__observable_array = []
|
||||
|
||||
# Build URL
|
||||
url_scan_io_task_reporturl = [str(i or 'no report url') for i in url_reputation_result_item_1]
|
||||
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_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))
|
||||
observable_object = {
|
||||
"value": url,
|
||||
"type": "url",
|
||||
"sandbox": {
|
||||
"score_id": url_object['score_id'],
|
||||
"score": url_object['score'],
|
||||
"confidence": url_object['confidence']
|
||||
},
|
||||
"attributes": {
|
||||
"hostname": parsed_url.hostname,
|
||||
"scheme": parsed_url.scheme
|
||||
},
|
||||
"categories": url_object['categories'],
|
||||
"description" :url_object['description'],
|
||||
"source": "urlscan.io",
|
||||
"source_link": f"{external_id}"
|
||||
}
|
||||
|
||||
if parsed_url.path:
|
||||
observable_object['attributes']['path'] = parsed_url.path
|
||||
if parsed_url.query:
|
||||
observable_object['attributes']['query'] = parsed_url.query
|
||||
if parsed_url.port:
|
||||
observable_object['attributes']['port'] = parsed_url.port
|
||||
|
||||
|
||||
build_url_output_with_error_code__observable_array.append(observable_object)
|
||||
#phantom.debug("build_url_output_with_error_code__observable_array: {}".format(build_url_output_with_error_code__observable_array))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="build_url_output_with_error_code:observable_array", value=json.dumps(build_url_output_with_error_code__observable_array))
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def build_url_output_with_no_error_code(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
|
||||
phantom.debug("build_url_output_with_no_error_code() called")
|
||||
|
||||
################################################################################
|
||||
# This block uses custom code to generate an observable dictionary to output into
|
||||
# the observables data path.
|
||||
################################################################################
|
||||
|
||||
url_reputation_result_data = phantom.collect2(container=container, datapath=["url_reputation:action_result.parameter.url","url_reputation:action_result.data.*.task.reportURL"], action_results=results)
|
||||
normalize_score_url_with_no_error_code__url_score_object = json.loads(_ if (_ := phantom.get_run_data(key="normalize_score_url_with_no_error_code:url_score_object")) != "" else "null") # pylint: disable=used-before-assignment
|
||||
|
||||
url_reputation_parameter_url = [item[0] for item in url_reputation_result_data]
|
||||
url_reputation_result_item_1 = [item[1] for item in url_reputation_result_data]
|
||||
|
||||
build_url_output_with_no_error_code__observable_array = None
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# Write your custom code here...
|
||||
from urllib.parse import urlparse
|
||||
build_url_output_with_no_error_code__observable_array = []
|
||||
|
||||
# Build 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))
|
||||
observable_object = {
|
||||
"value": url,
|
||||
"type": "url",
|
||||
"sandbox": {
|
||||
"score_id": url_object['score_id'],
|
||||
"score": url_object['score'],
|
||||
"confidence": url_object['confidence']
|
||||
},
|
||||
"attributes": {
|
||||
"hostname": parsed_url.hostname,
|
||||
"scheme": parsed_url.scheme
|
||||
},
|
||||
"categories": url_object['categories'],
|
||||
"description" :url_object['description'],
|
||||
"source": "urlscan.io",
|
||||
"source_link": f"{external_id}"
|
||||
}
|
||||
if parsed_url.path:
|
||||
observable_object['attributes']['path'] = parsed_url.path
|
||||
if parsed_url.query:
|
||||
observable_object['attributes']['query'] = parsed_url.query
|
||||
if parsed_url.port:
|
||||
observable_object['attributes']['port'] = parsed_url.port
|
||||
|
||||
build_url_output_with_no_error_code__observable_array.append(observable_object)
|
||||
#phantom.debug("build_url_output_with_no_error_code__observable_array: {}".format(build_url_output_with_no_error_code__observable_array))
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_run_data(key="build_url_output_with_no_error_code:observable_array", value=json.dumps(build_url_output_with_no_error_code__observable_array))
|
||||
|
||||
return
|
||||
|
||||
|
||||
@phantom.playbook_block()
|
||||
def on_finish(container, summary):
|
||||
phantom.debug("on_finish() called")
|
||||
|
||||
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)
|
||||
|
||||
output = {
|
||||
"observable": observable_combined_value,
|
||||
}
|
||||
|
||||
################################################################################
|
||||
## Custom Code Start
|
||||
################################################################################
|
||||
|
||||
# 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
|
||||
################################################################################
|
||||
## Custom Code End
|
||||
################################################################################
|
||||
|
||||
phantom.save_playbook_output_data(output=output)
|
||||
|
||||
return
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
name: UrlScan IO Dynamic Analysis
|
||||
id: a1173c28-7b33-4a56-9d7f-5dbbca595cb0
|
||||
version: 1
|
||||
date: '2023-03-23'
|
||||
author: Teoderick Contreras, Splunk
|
||||
type: Investigation
|
||||
description: "Accepts a url link, IP, or domain to be detonated using urlscan.io API connector."
|
||||
playbook: UrlScan_IO_Dynamic_Analysis
|
||||
how_to_implement: This input playbook requires the urlscan.io 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:
|
||||
- urlscan.io
|
||||
tags:
|
||||
platform_tags:
|
||||
- reputation
|
||||
- url
|
||||
- domain
|
||||
- sandbox
|
||||
- ip
|
||||
playbook_type: Input
|
||||
vpe_type: Modern
|
||||
playbook_fields: []
|
||||
product:
|
||||
- Splunk SOAR
|
||||
Reference in New Issue
Block a user