mirror of
https://github.com/splunk/security_content
synced 2026-06-08 17:32:49 +00:00
Eric McGinnis
13f06b5e1d
in playbooks that were previously unvalidated. Add a MANUAL_REVIEW section, which is commented out, for clarity and to allow CICD to run and pass on this content. Renamed an existing playbook because it diverges from the name of that playbook elsewhere.
596 lines
29 KiB
Python
596 lines
29 KiB
Python
"""
|
|
Accepts a URL, IP or Domain and does reputation analysis on the objects. Generates a threat level, threat categories and AUP categories that are formatted and added to a container as a note.
|
|
"""
|
|
|
|
|
|
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 'input_filter' block
|
|
input_filter(container=container)
|
|
|
|
return
|
|
|
|
@phantom.playbook_block()
|
|
def input_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("input_filter() called")
|
|
|
|
################################################################################
|
|
# Filter to pass in a url, domain or ip to it's appropriate action
|
|
################################################################################
|
|
|
|
# collect filtered artifact ids and results for 'if' condition 1
|
|
matched_artifacts_1, matched_results_1 = phantom.condition(
|
|
container=container,
|
|
conditions=[
|
|
["playbook_input:url", "!=", ""]
|
|
],
|
|
name="input_filter:condition_1",
|
|
delimiter=None)
|
|
|
|
# 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)
|
|
|
|
# collect filtered artifact ids and results for 'if' condition 2
|
|
matched_artifacts_2, matched_results_2 = phantom.condition(
|
|
container=container,
|
|
conditions=[
|
|
["playbook_input:domain", "!=", ""]
|
|
],
|
|
name="input_filter:condition_2",
|
|
delimiter=None)
|
|
|
|
# call connected blocks if filtered artifacts or results
|
|
if matched_artifacts_2 or matched_results_2:
|
|
domain_reputation(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_2, filtered_results=matched_results_2)
|
|
|
|
# collect filtered artifact ids and results for 'if' condition 3
|
|
matched_artifacts_3, matched_results_3 = phantom.condition(
|
|
container=container,
|
|
conditions=[
|
|
["playbook_input:ip", "!=", ""]
|
|
],
|
|
name="input_filter:condition_3",
|
|
delimiter=None)
|
|
|
|
# call connected blocks if filtered artifacts or results
|
|
if matched_artifacts_3 or matched_results_3:
|
|
ip_reputation(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_3, filtered_results=matched_results_3)
|
|
|
|
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, loop_state_json=None, **kwargs):
|
|
phantom.debug("url_reputation() called")
|
|
|
|
# phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
|
|
|
|
################################################################################
|
|
# Use Talos to get threat data on an url
|
|
################################################################################
|
|
|
|
filtered_input_0_url = phantom.collect2(container=container, datapath=["filtered-data:input_filter:condition_1:playbook_input:url"])
|
|
|
|
parameters = []
|
|
|
|
# build parameters list for 'url_reputation' call
|
|
for filtered_input_0_url_item in filtered_input_0_url:
|
|
if filtered_input_0_url_item[0] is not None:
|
|
parameters.append({
|
|
"url": filtered_input_0_url_item[0],
|
|
})
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
# Write your custom code here...
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.act("url reputation", parameters=parameters, name="url_reputation", assets=["cisco_talos_intelligence"], callback=url_reputation_filter)
|
|
|
|
return
|
|
|
|
|
|
@phantom.playbook_block()
|
|
def domain_reputation(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("domain_reputation() called")
|
|
|
|
# phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
|
|
|
|
################################################################################
|
|
# Use Talos to get threat data on a domain
|
|
################################################################################
|
|
|
|
filtered_input_0_domain = phantom.collect2(container=container, datapath=["filtered-data:input_filter:condition_2:playbook_input:domain"])
|
|
|
|
parameters = []
|
|
|
|
# build parameters list for 'domain_reputation' call
|
|
for filtered_input_0_domain_item in filtered_input_0_domain:
|
|
if filtered_input_0_domain_item[0] is not None:
|
|
parameters.append({
|
|
"domain": filtered_input_0_domain_item[0],
|
|
})
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
# Write your custom code here...
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.act("domain reputation", parameters=parameters, name="domain_reputation", assets=["cisco_talos_intelligence"], callback=domain_reputation_filter)
|
|
|
|
return
|
|
|
|
|
|
@phantom.playbook_block()
|
|
def ip_reputation(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("ip_reputation() called")
|
|
|
|
# phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
|
|
|
|
################################################################################
|
|
# Use Talos to get threat data on an ip
|
|
################################################################################
|
|
|
|
filtered_input_0_ip = phantom.collect2(container=container, datapath=["filtered-data:input_filter:condition_3:playbook_input:ip"])
|
|
|
|
parameters = []
|
|
|
|
# build parameters list for 'ip_reputation' call
|
|
for filtered_input_0_ip_item in filtered_input_0_ip:
|
|
if filtered_input_0_ip_item[0] is not None:
|
|
parameters.append({
|
|
"ip": filtered_input_0_ip_item[0],
|
|
})
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
# Write your custom code here...
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.act("ip reputation", parameters=parameters, name="ip_reputation", assets=["cisco_talos_intelligence"], callback=ip_reputation_filter)
|
|
|
|
return
|
|
|
|
|
|
@phantom.playbook_block()
|
|
def url_reputation_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("url_reputation_filter() called")
|
|
|
|
################################################################################
|
|
# Exclude failing url reputations
|
|
################################################################################
|
|
|
|
# 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="url_reputation_filter:condition_1",
|
|
delimiter=None)
|
|
|
|
# call connected blocks if filtered artifacts or results
|
|
if matched_artifacts_1 or matched_results_1:
|
|
format_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 domain_reputation_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("domain_reputation_filter() called")
|
|
|
|
################################################################################
|
|
# Exclude failing domain reputations
|
|
################################################################################
|
|
|
|
# collect filtered artifact ids and results for 'if' condition 1
|
|
matched_artifacts_1, matched_results_1 = phantom.condition(
|
|
container=container,
|
|
conditions=[
|
|
["domain_reputation:action_result.status", "==", "success"]
|
|
],
|
|
name="domain_reputation_filter:condition_1",
|
|
delimiter=None)
|
|
|
|
# call connected blocks if filtered artifacts or results
|
|
if matched_artifacts_1 or matched_results_1:
|
|
format_2(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 ip_reputation_filter(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("ip_reputation_filter() called")
|
|
|
|
################################################################################
|
|
# Exclude failing ip reputations
|
|
################################################################################
|
|
|
|
# collect filtered artifact ids and results for 'if' condition 1
|
|
matched_artifacts_1, matched_results_1 = phantom.condition(
|
|
container=container,
|
|
conditions=[
|
|
["ip_reputation:action_result.status", "==", "success"]
|
|
],
|
|
name="ip_reputation_filter:condition_1",
|
|
delimiter=None)
|
|
|
|
# call connected blocks if filtered artifacts or results
|
|
if matched_artifacts_1 or matched_results_1:
|
|
format_3(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 format_2(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("format_2() called")
|
|
|
|
################################################################################
|
|
# Format output of domain threat data into an appropriate format for build_domain_output
|
|
# that generates observable objects.
|
|
################################################################################
|
|
|
|
template = """SOAR analyzed Domain using Talos Intelligence. The table below shows a summary of the information gathered.\n\n| Domain | Threat Level | Threat Categories | AUP Categories |\n| --- | --- | --- | --- |\n%%\n| {0} | {1} | {2} | {3}\n%%"""
|
|
|
|
# parameter list for template variable replacement
|
|
parameters = [
|
|
"filtered-data:domain_reputation_filter:condition_1:domain_reputation:action_result.data.*.Observable",
|
|
"filtered-data:domain_reputation_filter:condition_1:domain_reputation:action_result.data.*.Threat_Level",
|
|
"filtered-data:domain_reputation_filter:condition_1:domain_reputation:action_result.data.*.Threat_Categories",
|
|
"filtered-data:domain_reputation_filter:condition_1:domain_reputation:action_result.data.*.AUP"
|
|
]
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
# Write your custom code here...
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.format(container=container, template=template, parameters=parameters, name="format_2")
|
|
|
|
build_domain_output(container=container)
|
|
|
|
return
|
|
|
|
|
|
@phantom.playbook_block()
|
|
def format_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("format_1() called")
|
|
|
|
################################################################################
|
|
# Format output of url threat data into an appropriate format for build_url_output
|
|
# that generates observable objects.
|
|
################################################################################
|
|
|
|
template = """SOAR analyzed URL using Talos Intelligence. The table below shows a summary of the information gathered.\n\n| URL | Threat Level | Threat Categories | AUP Categories |\n| --- | --- | --- | --- |\n%%\n| {0} | {1} | {2} | {3}\n%%"""
|
|
|
|
# parameter list for template variable replacement
|
|
parameters = [
|
|
"filtered-data:url_reputation_filter:condition_1:url_reputation:action_result.data.*.Observable",
|
|
"filtered-data:url_reputation_filter:condition_1:url_reputation:action_result.data.*.Threat_Level",
|
|
"filtered-data:url_reputation_filter:condition_1:url_reputation:action_result.data.*.Threat_Categories",
|
|
"filtered-data:url_reputation_filter:condition_1:url_reputation:action_result.data.*.AUP"
|
|
]
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
# Write your custom code here...
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.format(container=container, template=template, parameters=parameters, name="format_1")
|
|
|
|
build_url_output(container=container)
|
|
|
|
return
|
|
|
|
|
|
@phantom.playbook_block()
|
|
def format_3(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("format_3() called")
|
|
|
|
################################################################################
|
|
# Format output of ip threat data into an appropriate format for build_ip_output
|
|
# that generates observable objects.
|
|
################################################################################
|
|
|
|
template = """SOAR analyzed IP using Talos Intelligence. The table below shows a summary of the information gathered.\n\n| IP | Threat Level | Threat Categories | AUP Categories |\n| --- | --- | --- | --- |\n%%\n| {0} | {1} | {2} | {3}\n%%"""
|
|
|
|
# parameter list for template variable replacement
|
|
parameters = [
|
|
"filtered-data:ip_reputation_filter:condition_1:ip_reputation:action_result.data.*.Observable",
|
|
"filtered-data:ip_reputation_filter:condition_1:ip_reputation:action_result.data.*.Threat_Level",
|
|
"filtered-data:ip_reputation_filter:condition_1:ip_reputation:action_result.data.*.Threat_Categories",
|
|
"filtered-data:ip_reputation_filter:condition_1:ip_reputation:action_result.data.*.AUP"
|
|
]
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
# Write your custom code here...
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.format(container=container, template=template, parameters=parameters, name="format_3")
|
|
|
|
build_ip_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, loop_state_json=None, **kwargs):
|
|
phantom.debug("build_url_output() called")
|
|
|
|
################################################################################
|
|
# Generate an observable dictionary to output into the observables data path.
|
|
################################################################################
|
|
|
|
filtered_result_0_data_url_reputation_filter = phantom.collect2(container=container, datapath=["filtered-data:url_reputation_filter:condition_1:url_reputation:action_result.data.*.Observable","filtered-data:url_reputation_filter:condition_1:url_reputation:action_result.data.*.Threat_Level","filtered-data:url_reputation_filter:condition_1:url_reputation:action_result.data.*.Threat_Categories","filtered-data:url_reputation_filter:condition_1:url_reputation:action_result.data.*.AUP"])
|
|
|
|
filtered_result_0_data___observable = [item[0] for item in filtered_result_0_data_url_reputation_filter]
|
|
filtered_result_0_data___threat_level = [item[1] for item in filtered_result_0_data_url_reputation_filter]
|
|
filtered_result_0_data___threat_categories = [item[2] for item in filtered_result_0_data_url_reputation_filter]
|
|
filtered_result_0_data___aup = [item[3] for item in filtered_result_0_data_url_reputation_filter]
|
|
|
|
build_url_output__observable_array = None
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
from urllib.parse import urlparse
|
|
build_url_output__observable_array = []
|
|
|
|
talos_to_score_mapping = {"unknown": "Unknown", "trusted": "Safe", "favorable": "Probably_Safe", "neutral": "May_not_be_Safe", "questionable": "Suspicious_or_Risky", "unstrusted": "Malicious"}
|
|
score_table = {
|
|
"Unkown": "0",
|
|
"Very_Safe": "1",
|
|
"Safe": "2",
|
|
"Probably_Safe": "3",
|
|
"Leans_Safe": "4",
|
|
"May_not_be_Safe": "5",
|
|
"Exercise_Caution": "6",
|
|
"Suspicious_or_Risky": "7",
|
|
"Possibly_Malicious": "8",
|
|
"Probably_Malicious": "9",
|
|
"Malicious": "10"
|
|
}
|
|
|
|
for url, threat_level, threat_categories, aup in zip(filtered_result_0_data___observable, filtered_result_0_data___threat_level, filtered_result_0_data___threat_categories, filtered_result_0_data___aup):
|
|
parsed_url = urlparse(url)
|
|
score = talos_to_score_mapping.get(threat_level.lower(), "")
|
|
observable_object = {
|
|
"value": url,
|
|
"type": "url",
|
|
"reputation": {
|
|
"threat_level": threat_level,
|
|
"threat_categories": threat_categories,
|
|
"aup_categories": aup,
|
|
"score": score,
|
|
"score_id": score_table.get(score, "")
|
|
},
|
|
"attributes": {
|
|
"hostname": parsed_url.hostname,
|
|
"scheme": parsed_url.scheme
|
|
},
|
|
"source": "Cisco Talos Intelligence",
|
|
}
|
|
if parsed_url.path:
|
|
observable_object['attributes']['path'] = parsed_url.path
|
|
if parsed_url.query:
|
|
observable_object['attributes']['query'] = parsed_url.query
|
|
if parsed_url.port:
|
|
observable_object['attributes']['port'] = parsed_url.port
|
|
|
|
build_url_output__observable_array.append(observable_object)
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.save_run_data(key="build_url_output:observable_array", value=json.dumps(build_url_output__observable_array))
|
|
|
|
return
|
|
|
|
|
|
@phantom.playbook_block()
|
|
def build_domain_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("build_domain_output() called")
|
|
|
|
################################################################################
|
|
# Generate an observable dictionary to output into the observables data path.
|
|
################################################################################
|
|
|
|
filtered_result_0_data_domain_reputation_filter = phantom.collect2(container=container, datapath=["filtered-data:domain_reputation_filter:condition_1:domain_reputation:action_result.data.*.Observable","filtered-data:domain_reputation_filter:condition_1:domain_reputation:action_result.data.*.Threat_Level","filtered-data:domain_reputation_filter:condition_1:domain_reputation:action_result.data.*.Threat_Categories","filtered-data:domain_reputation_filter:condition_1:domain_reputation:action_result.data.*.AUP"])
|
|
|
|
filtered_result_0_data___observable = [item[0] for item in filtered_result_0_data_domain_reputation_filter]
|
|
filtered_result_0_data___threat_level = [item[1] for item in filtered_result_0_data_domain_reputation_filter]
|
|
filtered_result_0_data___threat_categories = [item[2] for item in filtered_result_0_data_domain_reputation_filter]
|
|
filtered_result_0_data___aup = [item[3] for item in filtered_result_0_data_domain_reputation_filter]
|
|
|
|
build_domain_output__observable_array = None
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
build_domain_output__observable_array = []
|
|
|
|
talos_to_score_mapping = {"unknown": "Unknown", "trusted": "Safe", "favorable": "Probably_Safe", "neutral": "May_not_be_Safe", "questionable": "Suspicious_or_Risky", "unstrusted": "Malicious"}
|
|
score_table = {
|
|
"Unkown": "0",
|
|
"Very_Safe": "1",
|
|
"Safe": "2",
|
|
"Probably_Safe": "3",
|
|
"Leans_Safe": "4",
|
|
"May_not_be_Safe": "5",
|
|
"Exercise_Caution": "6",
|
|
"Suspicious_or_Risky": "7",
|
|
"Possibly_Malicious": "8",
|
|
"Probably_Malicious": "9",
|
|
"Malicious": "10"
|
|
}
|
|
|
|
for domain, threat_level, threat_categories, aup in zip(filtered_result_0_data___observable, filtered_result_0_data___threat_level, filtered_result_0_data___threat_categories, filtered_result_0_data___aup):
|
|
score = talos_to_score_mapping.get(threat_level.lower(), "")
|
|
observable_object = {
|
|
"value": domain,
|
|
"type": "domain",
|
|
"reputation": {
|
|
"threat_level": threat_level,
|
|
"threat_categories": threat_categories,
|
|
"aup_categories": aup,
|
|
"score": score,
|
|
"score_id": score_table.get(score, "")
|
|
},
|
|
"source": "Cisco Talos Intelligence"
|
|
}
|
|
build_domain_output__observable_array.append(observable_object)
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.save_run_data(key="build_domain_output:observable_array", value=json.dumps(build_domain_output__observable_array))
|
|
|
|
return
|
|
|
|
|
|
@phantom.playbook_block()
|
|
def build_ip_output(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
|
|
phantom.debug("build_ip_output() called")
|
|
|
|
################################################################################
|
|
# Generate an observable dictionary to output into the observables data path.
|
|
################################################################################
|
|
|
|
filtered_result_0_data_ip_reputation_filter = phantom.collect2(container=container, datapath=["filtered-data:ip_reputation_filter:condition_1:ip_reputation:action_result.data.*.Observable","filtered-data:ip_reputation_filter:condition_1:ip_reputation:action_result.data.*.Threat_Level","filtered-data:ip_reputation_filter:condition_1:ip_reputation:action_result.data.*.Threat_Categories","filtered-data:ip_reputation_filter:condition_1:ip_reputation:action_result.data.*.AUP"])
|
|
|
|
filtered_result_0_data___observable = [item[0] for item in filtered_result_0_data_ip_reputation_filter]
|
|
filtered_result_0_data___threat_level = [item[1] for item in filtered_result_0_data_ip_reputation_filter]
|
|
filtered_result_0_data___threat_categories = [item[2] for item in filtered_result_0_data_ip_reputation_filter]
|
|
filtered_result_0_data___aup = [item[3] for item in filtered_result_0_data_ip_reputation_filter]
|
|
|
|
build_ip_output__observable_array = None
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
import ipaddress
|
|
build_ip_output__observable_array = []
|
|
|
|
talos_to_score_mapping = {"unknown": "Unknown", "trusted": "Safe", "favorable": "Probably_Safe", "neutral": "May_not_be_Safe", "questionable": "Suspicious_or_Risky", "unstrusted": "Malicious"}
|
|
score_table = {
|
|
"Unkown": "0",
|
|
"Very_Safe": "1",
|
|
"Safe": "2",
|
|
"Probably_Safe": "3",
|
|
"Leans_Safe": "4",
|
|
"May_not_be_Safe": "5",
|
|
"Exercise_Caution": "6",
|
|
"Suspicious_or_Risky": "7",
|
|
"Possibly_Malicious": "8",
|
|
"Probably_Malicious": "9",
|
|
"Malicious": "10"
|
|
}
|
|
|
|
for ip, threat_level, threat_categories, aup in zip(filtered_result_0_data___observable, filtered_result_0_data___threat_level, filtered_result_0_data___threat_categories, filtered_result_0_data___aup):
|
|
score = talos_to_score_mapping.get(threat_level.lower(), "")
|
|
observable_object = {
|
|
"value": ip,
|
|
"type": "ipv4",
|
|
"reputation": {
|
|
"threat_level": threat_level,
|
|
"threat_categories": threat_categories,
|
|
"aup_categories": aup,
|
|
"score": score,
|
|
"score_id": score_table.get(score, "")
|
|
},
|
|
"source": "Cisco Talos Intelligence"
|
|
}
|
|
ip_addr = ipaddress.ip_address(ip)
|
|
if isinstance(ip_addr, ipaddress.IPv6Address):
|
|
observable_object["type"] = "ipv6"
|
|
|
|
build_ip_output__observable_array.append(observable_object)
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.save_run_data(key="build_ip_output:observable_array", value=json.dumps(build_ip_output__observable_array))
|
|
|
|
return
|
|
|
|
|
|
@phantom.playbook_block()
|
|
def on_finish(container, summary):
|
|
phantom.debug("on_finish() called")
|
|
|
|
format_1 = phantom.get_format_data(name="format_1")
|
|
format_2 = phantom.get_format_data(name="format_2")
|
|
format_3 = phantom.get_format_data(name="format_3")
|
|
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_domain_output__observable_array = json.loads(_ if (_ := phantom.get_run_data(key="build_domain_output:observable_array")) != "" else "null") # pylint: disable=used-before-assignment
|
|
build_ip_output__observable_array = json.loads(_ if (_ := phantom.get_run_data(key="build_ip_output:observable_array")) != "" else "null") # pylint: disable=used-before-assignment
|
|
|
|
observable_combined_value = phantom.concatenate(build_url_output__observable_array, build_domain_output__observable_array, build_ip_output__observable_array)
|
|
markdown_report_combined_value = phantom.concatenate(format_1, format_2, format_3)
|
|
|
|
output = {
|
|
"observable": observable_combined_value,
|
|
"markdown_report": markdown_report_combined_value,
|
|
}
|
|
|
|
################################################################################
|
|
## Custom Code Start
|
|
################################################################################
|
|
|
|
# Write your custom code here...
|
|
|
|
################################################################################
|
|
## Custom Code End
|
|
################################################################################
|
|
|
|
phantom.save_playbook_output_data(output=output)
|
|
|
|
return |