admin/splunk-security_content
1
0
Fork 0
mirror of https://github.com/splunk/security_content synced 2026-06-08 17:32:49 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
develop
splunk-security_content/playbooks/crowdstrike_malware_triage.py
T
Lou Stella 3b58b30516 Adding custom functions & playbooks 
Adding custom functions & playbooks
2021-12-08 14:00:13 -06:00

792 lines
35 KiB
Python
Raw Permalink Blame History

"""
Enrich and respond to a CrowdStrike Falcon detection involving a potentially malicious executable on an endpoint. Check for previous sightings of the same executable, hunt across other endpoints for the file, gather details about all processes associated with the file, and collect all the gathered information into a prompt for an analyst to review. Based on the analyst's choice, the file can be added to the custom indicators list in CrowdStrike with a detection policy of "detect" or "none", and the endpoint can be optionally quarantined from the network.
"""
import phantom.rules as phantom
import json
from datetime import datetime, timedelta
def on_start(container):
phantom.debug('on_start() called')
# call 'if_sha256_exists' block
if_sha256_exists(container=container)
return
"""
List all machines where the file hash has been seen.
"""
def hunt_file_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('hunt_file_1() called')
#phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
# collect data for 'hunt_file_1' call
filtered_artifacts_data_1 = phantom.collect2(container=container, datapath=['filtered-data:filter_main_artifact:condition_1:artifact:*.cef.fileHashSha256', 'filtered-data:filter_main_artifact:condition_1:artifact:*.id'])
parameters = []
# build parameters list for 'hunt_file_1' call
for filtered_artifacts_item_1 in filtered_artifacts_data_1:
if filtered_artifacts_item_1[0]:
parameters.append({
'hash': filtered_artifacts_item_1[0],
'count_only': False,
# context (artifact id) is added to associate results with the artifact
'context': {'artifact_id': filtered_artifacts_item_1[1]},
})
phantom.act(action="hunt file", parameters=parameters, assets=['crowdstrike_oauth'], callback=get_system_info_1, name="hunt_file_1")
return
"""
Ensure that the event has at least one artifact with a SHA256 file hash before attempting to process the event.
"""
def if_sha256_exists(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('if_sha256_exists() called')
# check for 'if' condition 1
matched = phantom.decision(
container=container,
conditions=[
["artifact:*.cef.fileHashSha256", "!=", ""],
])
# call connected blocks if condition 1 matched
if matched:
filter_main_artifact(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
# call connected blocks for 'else' condition 2
ignore_if_no_sha256(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
"""
End the playbook if no SHA256 file hash is found in any of the artifacts.
"""
def ignore_if_no_sha256(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('ignore_if_no_sha256() called')
phantom.comment(container=container, comment="Ignoring alert because no SHA256 file hash was found")
return
"""
Fetch the CrowdStrike indicator for the SHA256 file hash, if there is one. This action will fail if there is no matching indicator in CrowdStrike, but the playbook will check for the failure and continue.
"""
def get_indicator_2(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('get_indicator_2() called')
# collect data for 'get_indicator_2' call
filtered_artifacts_data_1 = phantom.collect2(container=container, datapath=['filtered-data:filter_main_artifact:condition_1:artifact:*.cef.fileHashSha256', 'filtered-data:filter_main_artifact:condition_1:artifact:*.id'])
parameters = []
# build parameters list for 'get_indicator_2' call
for filtered_artifacts_item_1 in filtered_artifacts_data_1:
if filtered_artifacts_item_1[0]:
parameters.append({
'indicator_type': "sha256",
'indicator_value': filtered_artifacts_item_1[0],
# context (artifact id) is added to associate results with the artifact
'context': {'artifact_id': filtered_artifacts_item_1[1]},
})
phantom.act(action="get indicator", parameters=parameters, assets=['crowdstrike_oauth'], callback=if_indicator_exists, name="get_indicator_2")
return
"""
Determine which response to take based on whether an Indicator exists in CrowdStrike for the SHA256 file hash.
"""
def if_indicator_exists(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('if_indicator_exists() called')
# check for 'if' condition 1
matched = phantom.decision(
container=container,
action_results=results,
conditions=[
["Resource Not Found", "in", "get_indicator_2:action_result.message"],
])
# call connected blocks if condition 1 matched
if matched:
hunt_file_1(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
list_processes_with_hash(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
# call connected blocks for 'else' condition 2
indicator_policy_decision(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
"""
Escalate the event because the Indicator policy is "detect", meaning the event is a true positive.
"""
def escalate_severity_to_high(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('escalate_severity_to_high() called')
phantom.set_severity(container=container, severity="High")
return
"""
Format a note to summarize all known information about the event.
"""
def format_repeat_note(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('format_repeat_note() called')
template = """CrowdStrike detected a file on an endpoint which matched a previously detected file hash:
| Field | Value |
|---|---|
| Host | {0} |
| Command Line | {1} |
| SHA 256 | {2} |
| File Path | {3}\\\\{4} |
| CrowdStrike Detection Link | {5} |
---
This event will have the severity escalated to high, and should be investigated further."""
# parameter list for template variable replacement
parameters = [
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.sourceHostName",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.cmdLine",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.fileHashSha256",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.filePath",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.fileName",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.falconHostLink",
]
phantom.format(container=container, template=template, parameters=parameters, name="format_repeat_note")
add_repeat_note(container=container)
return
"""
Add a note to summarize the event information.
"""
def add_repeat_note(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('add_repeat_note() called')
formatted_data_1 = phantom.get_format_data(name='format_repeat_note')
note_title = "Known Malicious File"
note_content = formatted_data_1
note_format = "markdown"
phantom.add_note(container=container, note_type="general", title=note_title, content=note_content, note_format=note_format)
crowdstrike_known_file_quarantine(container=container)
return
"""
Only process the main detection artifact, not any sub event artifacts.
"""
def filter_main_artifact(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('filter_main_artifact() called')
# collect filtered artifact ids for 'if' condition 1
matched_artifacts_1, matched_results_1 = phantom.condition(
container=container,
conditions=[
["artifact:*.label", "==", "event"],
],
name="filter_main_artifact:condition_1")
# call connected blocks if filtered artifacts or results
if matched_artifacts_1 or matched_results_1:
get_indicator_2(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)
return
"""
Handle the Indicator differently if the policy is "detect", "none", or other.
"""
def indicator_policy_decision(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('indicator_policy_decision() called')
# check for 'if' condition 1
matched = phantom.decision(
container=container,
action_results=results,
conditions=[
["get_indicator_2:action_result.data.*.resources.*.policy", "==", "none"],
])
# call connected blocks if condition 1 matched
if matched:
detection_policy_none(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
# check for 'elif' condition 2
matched = phantom.decision(
container=container,
action_results=results,
conditions=[
["get_indicator_2:action_result.data.*.resources.*.policy", "==", "detect"],
])
# call connected blocks if condition 2 matched
if matched:
escalate_severity_to_high(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
format_repeat_note(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
# call connected blocks for 'else' condition 3
comment_unexpected_policy(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
"""
End processing because this playbook only expects "none" or "detect" as the Indicator policy.
"""
def comment_unexpected_policy(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('comment_unexpected_policy() called')
phantom.comment(container=container, comment="The playbook received an unexpected indicator policy and needs to be extended to handle this situation.")
return
"""
Add a comment to explain why the event is being closed.
"""
def detection_policy_none(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('detection_policy_none() called')
phantom.comment(container=container, comment="The file hash indicator has a detection policy of none, so previous investigations have found that the file is not harmful. This playbook will take no further action and the event will be closed.")
close_event(container=container)
return
"""
Fetch additional information about each machine listed in the previous step.
"""
def get_system_info_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('get_system_info_1() called')
#phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
# collect data for 'get_system_info_1' call
results_data_1 = phantom.collect2(container=container, datapath=['hunt_file_1:action_result.data.*.device_id', 'hunt_file_1:action_result.parameter.context.artifact_id'], action_results=results)
parameters = []
# build parameters list for 'get_system_info_1' call
for results_item_1 in results_data_1:
if results_item_1[0]:
parameters.append({
'id': results_item_1[0],
# context (artifact id) is added to associate results with the artifact
'context': {'artifact_id': results_item_1[1]},
})
phantom.act(action="get system info", parameters=parameters, assets=['crowdstrike_oauth'], callback=join_format_prompt, name="get_system_info_1", parent_action=action)
return
"""
Summarize all the gathered information to help the analyst decide a response in the prompt.
"""
def format_prompt(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('format_prompt() called')
template = """CrowdStrike detected the following suspicious activity on an endpoint:
| Field | Value |
|---|---|
| Host | {0} |
| Command Line | {1} |
| SHA 256 | {2} |
| File Path | {3}\\\\{4}
| CrowdStrike Detection Link | {5} |
| Details of processes associated with the file hash | <see \"get process details\" action results> |
| Count of machines that have the file on disk | {6} |
| System information of machines that have the file on disk | <see \"get system info\" action results> |"""
# parameter list for template variable replacement
parameters = [
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.sourceHostName",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.cmdLine",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.fileHashSha256",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.filePath",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.fileName",
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.falconHostLink",
"hunt_file_1:action_result.summary.device_count",
]
phantom.format(container=container, template=template, parameters=parameters, name="format_prompt")
crowdstrike_new_file_detection(container=container)
return
def join_format_prompt(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None):
phantom.debug('join_format_prompt() called')
# if the joined function has already been called, do nothing
if phantom.get_run_data(key='join_format_prompt_called'):
return
# check if all connected incoming playbooks, actions, or custom functions are done i.e. have succeeded or failed
if phantom.completed(action_names=['get_process_details']):
# save the state that the joined function has now been called
phantom.save_run_data(key='join_format_prompt_called', value='format_prompt')
# call connected block "format_prompt"
format_prompt(container=container, handle=handle)
return
"""
Close the event because the Indicator policy is "none", meaning the detection is a false positive.
"""
def close_event(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('close_event() called')
phantom.set_status(container=container, status="Closed")
return
"""
List all processes seen on this host associated with this file hash.
"""
def list_processes_with_hash(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('list_processes_with_hash() called')
#phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
# collect data for 'list_processes_with_hash' call
filtered_artifacts_data_1 = phantom.collect2(container=container, datapath=['filtered-data:filter_main_artifact:condition_1:artifact:*.cef.sensorId', 'filtered-data:filter_main_artifact:condition_1:artifact:*.cef.fileHashSha256', 'filtered-data:filter_main_artifact:condition_1:artifact:*.id'])
parameters = []
# build parameters list for 'list_processes_with_hash' call
for filtered_artifacts_item_1 in filtered_artifacts_data_1:
if filtered_artifacts_item_1[0] and filtered_artifacts_item_1[1]:
parameters.append({
'id': filtered_artifacts_item_1[0],
'ioc': filtered_artifacts_item_1[1],
# context (artifact id) is added to associate results with the artifact
'context': {'artifact_id': filtered_artifacts_item_1[2]},
})
phantom.act(action="list processes", parameters=parameters, assets=['crowdstrike_oauth'], callback=get_process_details, name="list_processes_with_hash")
return
"""
Fetch additional information about each process listed in the previous step.
"""
def get_process_details(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('get_process_details() called')
#phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
# collect data for 'get_process_details' call
results_data_1 = phantom.collect2(container=container, datapath=['list_processes_with_hash:action_result.data.*.falcon_process_id', 'list_processes_with_hash:action_result.parameter.context.artifact_id'], action_results=results)
parameters = []
# build parameters list for 'get_process_details' call
for results_item_1 in results_data_1:
if results_item_1[0]:
parameters.append({
'falcon_process_id': results_item_1[0],
# context (artifact id) is added to associate results with the artifact
'context': {'artifact_id': results_item_1[1]},
})
phantom.act(action="get process detail", parameters=parameters, assets=['crowdstrike_oauth'], callback=join_format_prompt, name="get_process_details", parent_action=action)
return
"""
Prompt the user to determine whether or not to create an Indicator for the file hash and whether or not to quarantine the endpoint.
"""
def crowdstrike_new_file_detection(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('crowdstrike_new_file_detection() called')
# set user and message variables for phantom.prompt call
user = "admin"
message = """{0}"""
# parameter list for template variable replacement
parameters = [
"format_prompt:formatted_data",
]
#responses:
response_types = [
{
"prompt": "Should Phantom create an Indicator in CrowdStrike to track this file hash from now on?",
"options": {
"type": "list",
"choices": [
"No, do not create an Indicator in CrowdStrike at this time.",
"Yes, create a CrowdStrike Indicator to detect and block this file hash from now on. (True Positive)",
"Yes, create a CrowdStrike Indicator to ignore this file hash from now on. (False Positive)",
]
},
},
{
"prompt": "Should Phantom quarantine the endpoint?",
"options": {
"type": "list",
"choices": [
"Yes",
"No",
]
},
},
]
phantom.prompt2(container=container, user=user, message=message, respond_in_mins=30, name="crowdstrike_new_file_detection", parameters=parameters, response_types=response_types, callback=crowdstrike_new_file_detection_callback)
return
def crowdstrike_new_file_detection_callback(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None):
phantom.debug('crowdstrike_new_file_detection_callback() called')
indicator_decision(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
quarantine_decision_1(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
"""
Parse the prompt response to determine how to handle the indicator.
"""
def indicator_decision(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('indicator_decision() called')
# check for 'if' condition 1
matched = phantom.decision(
container=container,
action_results=results,
conditions=[
["No, do not create an Indicator in CrowdStrike at this time.", "==", "crowdstrike_new_file_detection:action_result.summary.responses.0"],
])
# call connected blocks if condition 1 matched
if matched:
comment_no_indicator(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
# check for 'elif' condition 2
matched = phantom.decision(
container=container,
action_results=results,
conditions=[
["Yes, create a CrowdStrike Indicator to detect and block this file hash from now on. (True Positive)", "==", "crowdstrike_new_file_detection:action_result.summary.responses.0"],
])
# call connected blocks if condition 2 matched
if matched:
format_detect_description(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
# check for 'elif' condition 3
matched = phantom.decision(
container=container,
action_results=results,
conditions=[
["Yes, create a CrowdStrike Indicator to ignore this file hash going forward (False Positive)", "==", "crowdstrike_new_file_detection:action_result.summary.responses.0"],
])
# call connected blocks if condition 3 matched
if matched:
format_ignore_description(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
return
"""
Check the quarantine device prompt response.
"""
def quarantine_decision_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('quarantine_decision_1() called')
# check for 'if' condition 1
matched = phantom.decision(
container=container,
action_results=results,
conditions=[
["crowdstrike_new_file_detection:action_result.summary.responses.1", "==", "Yes"],
])
# call connected blocks if condition 1 matched
if matched:
quarantine_device_1(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
# call connected blocks for 'else' condition 2
comment_no_quarantine_1(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
"""
Create an Indicator in CrowdStrike with a policy of "none" to ignore detections based on this file hash in the future.
"""
def create_ignore_indicator(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('create_ignore_indicator() called')
#phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
# collect data for 'create_ignore_indicator' call
filtered_artifacts_data_1 = phantom.collect2(container=container, datapath=['filtered-data:filter_main_artifact:condition_1:artifact:*.cef.fileHashSha256', 'filtered-data:filter_main_artifact:condition_1:artifact:*.id'])
formatted_data_1 = phantom.get_format_data(name='format_ignore_description')
parameters = []
# build parameters list for 'create_ignore_indicator' call
for filtered_artifacts_item_1 in filtered_artifacts_data_1:
if filtered_artifacts_item_1[0]:
parameters.append({
'ioc': filtered_artifacts_item_1[0],
'policy': "none",
'source': "Phantom Playbook crowdstrike_malware_triage",
'expiration': "",
'description': formatted_data_1,
'share_level': "red",
# context (artifact id) is added to associate results with the artifact
'context': {'artifact_id': filtered_artifacts_item_1[1]},
})
phantom.act(action="upload indicator", parameters=parameters, assets=['crowdstrike_oauth'], name="create_ignore_indicator")
return
"""
Format a description to provide when creating an Indicator with a policy of "none".
"""
def format_ignore_description(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('format_ignore_description() called')
template = """This indicator was created by Phantom in the playbook crowdstrike_malware_triage to ignore CrowdStrike detections based on the file hash first seen in {0} and processed in Phantom as {1}"""
# parameter list for template variable replacement
parameters = [
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.falconHostLink",
"container:url",
]
phantom.format(container=container, template=template, parameters=parameters, name="format_ignore_description")
create_ignore_indicator(container=container)
return
"""
Explain in a comment that no Indicator will be created.
"""
def comment_no_indicator(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('comment_no_indicator() called')
phantom.comment(container=container, comment="The analyst decided not to create a custom indicator for the file hash.")
return
"""
Format a description to provide when creating an Indicator with a policy of "detect".
"""
def format_detect_description(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('format_detect_description() called')
template = """This indicator was created by Phantom in the playbook crowdstrike_malware_triage to detect and block process executions based on the file hash first seen in {0} and processed in Phantom as {1}"""
# parameter list for template variable replacement
parameters = [
"filtered-data:filter_main_artifact:condition_1:artifact:*.cef.falconHostLink",
"container:url",
]
phantom.format(container=container, template=template, parameters=parameters, name="format_detect_description")
create_detect_indicator(container=container)
return
"""
Create an Indicator to detect and block this file hash.
"""
def create_detect_indicator(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('create_detect_indicator() called')
#phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
# collect data for 'create_detect_indicator' call
filtered_artifacts_data_1 = phantom.collect2(container=container, datapath=['filtered-data:filter_main_artifact:condition_1:artifact:*.cef.fileHashSha256', 'filtered-data:filter_main_artifact:condition_1:artifact:*.id'])
parameters = []
# build parameters list for 'create_detect_indicator' call
for filtered_artifacts_item_1 in filtered_artifacts_data_1:
if filtered_artifacts_item_1[0]:
parameters.append({
'ioc': filtered_artifacts_item_1[0],
'policy': "detect",
'source': "",
'expiration': "",
'description': "",
'share_level': "red",
# context (artifact id) is added to associate results with the artifact
'context': {'artifact_id': filtered_artifacts_item_1[1]},
})
phantom.act(action="upload indicator", parameters=parameters, assets=['crowdstrike_oauth'], name="create_detect_indicator")
return
"""
Do not quarantine the endpoint because the analyst responded No in the prompt.
"""
def comment_no_quarantine_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('comment_no_quarantine_1() called')
phantom.comment(container=container, comment="The analyst decided not to quarantine the endpoint.")
return
"""
Block the endpoint from everything but the configured allowlist of network addresses while the investigation is ongoing.
"""
def quarantine_device_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('quarantine_device_1() called')
#phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
# collect data for 'quarantine_device_1' call
filtered_artifacts_data_1 = phantom.collect2(container=container, datapath=['filtered-data:filter_main_artifact:condition_1:artifact:*.cef.sensorId', 'filtered-data:filter_main_artifact:condition_1:artifact:*.id'])
parameters = []
# build parameters list for 'quarantine_device_1' call
for filtered_artifacts_item_1 in filtered_artifacts_data_1:
parameters.append({
'hostname': "",
'device_id': filtered_artifacts_item_1[0],
# context (artifact id) is added to associate results with the artifact
'context': {'artifact_id': filtered_artifacts_item_1[1]},
})
phantom.act(action="quarantine device", parameters=parameters, assets=['crowdstrike_oauth'], name="quarantine_device_1")
return
"""
Ask the analyst if the endpoint should be quarantined.
"""
def crowdstrike_known_file_quarantine(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('crowdstrike_known_file_quarantine() called')
# set user and message variables for phantom.prompt call
user = "admin"
message = """{0}
---
Should Phantom quarantine the device?"""
# parameter list for template variable replacement
parameters = [
"format_repeat_note:formatted_data",
]
#responses:
response_types = [
{
"prompt": "",
"options": {
"type": "list",
"choices": [
"Yes",
"No",
]
},
},
]
phantom.prompt2(container=container, user=user, message=message, respond_in_mins=30, name="crowdstrike_known_file_quarantine", parameters=parameters, response_types=response_types, callback=quarantine_decision_2)
return
"""
Check if the analyst responded Yes or No to the quarantine.
"""
def quarantine_decision_2(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('quarantine_decision_2() called')
# check for 'if' condition 1
matched = phantom.decision(
container=container,
action_results=results,
conditions=[
["crowdstrike_known_file_quarantine:action_result.summary.responses.0", "==", "Yes"],
])
# call connected blocks if condition 1 matched
if matched:
quarantine_device_2(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
# call connected blocks for 'else' condition 2
comment_no_quarantine_2(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function)
return
"""
Block the endpoint from everything but the configured allowlist of network addresses while the investigation is ongoing.
"""
def quarantine_device_2(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('quarantine_device_2() called')
#phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
# collect data for 'quarantine_device_2' call
filtered_artifacts_data_1 = phantom.collect2(container=container, datapath=['filtered-data:filter_main_artifact:condition_1:artifact:*.cef.sensorId', 'filtered-data:filter_main_artifact:condition_1:artifact:*.id'])
parameters = []
# build parameters list for 'quarantine_device_2' call
for filtered_artifacts_item_1 in filtered_artifacts_data_1:
parameters.append({
'hostname': "",
'device_id': filtered_artifacts_item_1[0],
# context (artifact id) is added to associate results with the artifact
'context': {'artifact_id': filtered_artifacts_item_1[1]},
})
phantom.act(action="quarantine device", parameters=parameters, assets=['crowdstrike_oauth'], name="quarantine_device_2")
return
"""
Do not quarantine the endpoint because the analyst responded No in the prompt.
"""
def comment_no_quarantine_2(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug('comment_no_quarantine_2() called')
phantom.comment(container=container, comment="The analyst decided not to quarantine the endpoint.")
return
def on_finish(container, summary):
phantom.debug('on_finish() called')
# This function is called after all actions are completed.
# summary of all the action and/or all details of actions
# can be collected here.
# summary_json = phantom.get_summary()
# if 'result' in summary_json:
# for action_result in summary_json['result']:
# if 'action_run_id' in action_result:
# action_results = phantom.get_action_results(action_run_id=action_result['action_run_id'], result_data=False, flatten=False)
# phantom.debug(action_results)
return
Reference in New Issue View Git Blame Copy Permalink