After the future removal of the classic playbook editor, your existing classic playbooks will continue to run, However, you will no longer be able to visualize or modify existing classic playbooks.
For details, see:
Common API calls used by the Visual Playbook Editor
The Visual Playbook Editor uses the phantom.act
API calls to perform actions in a playbook. The phantom.collect
API calls are used to collect input for actions.
Run actions using the phantom.act API calls
phantom.act
is the call for running an action. The following code shows an example of a simple act call. In this act call, the parameters are static as shown by the 1.1.1.1 ip. This means that regardless of which container you run against, the playbook will work on the same input data. In the following act call, there is no callback.
phantom.act(action="geolocate ip", parameters=parameters, assets=['maxmind'], name="geolocate_ip_1")
It is important to note that in the call to phantom.act, the callback parameter isn't set, and you can see that in the logs for the playbook run and the log output.
View the simplest phantom.act call.
""" """ import phantom.rules as phantom import json from datetime import datetime, timedelta def on_start(container): phantom.debug('on_start() called') # call 'geolocate_ip_1' block geolocate_ip_1(container=container) return def geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None): phantom.debug('geolocate_ip_1() called') # collect data for 'geolocate_ip_1' call parameters = [] # build parameters list for 'geolocate_ip_1' call parameters.append({ 'ip': "1.1.1.1", }) phantom.act(action="geolocate ip", parameters=parameters, assets=['maxmind'], name="geolocate_ip_1") 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
Using the callback parameter
Use the callback function to chain playbook blocks together that rely on data from the previous blocks. If you'd like a later playbook block to rely on the output of your act call, use a callback function. In the following example, the filter block relies on the output of the geolocate ip command.
View a phantom.act call that uses a callback.
""" """ import phantom.rules as phantom import json from datetime import datetime, timedelta def on_start(container): phantom.debug('on_start() called') # call 'geolocate_ip_1' block geolocate_ip_1(container=container) return def geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None): phantom.debug('geolocate_ip_1() called') # collect data for 'geolocate_ip_1' call parameters = [] # build parameters list for 'geolocate_ip_1' call parameters.append({ 'ip': "1.1.1.1", }) phantom.act(action="geolocate ip", parameters=parameters, assets=['maxmind'], callback=filter_1, name="geolocate_ip_1") return def filter_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None): phantom.debug('filter_1() called') # collect filtered artifact ids for 'if' condition 1 matched_artifacts_1, matched_results_1 = phantom.condition( container=container, action_results=results, conditions=[ ["geolocate_ip_1:action_result.data.*.country_iso_code", "==", "AU"], ], name="filter_1:condition_1") # call connected blocks if filtered artifacts or results if matched_artifacts_1 or matched_results_1: lookup_ip_1(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 def lookup_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None): phantom.debug('lookup_ip_1() called') #phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED'))) # collect data for 'lookup_ip_1' call filtered_results_data_1 = phantom.collect2(container=container, datapath=["filtered-data:filter_1:condition_1:geolocate_ip_1:action_result.parameter.ip", "filtered-data:filter_1:condition_1:geolocate_ip_1:action_result.parameter.context.artifact_id"]) parameters = [] # build parameters list for 'lookup_ip_1' call for filtered_results_item_1 in filtered_results_data_1: if filtered_results_item_1[0]: parameters.append({ 'ip': filtered_results_item_1[0], # context (artifact id) is added to associate results with the artifact 'context': {'artifact_id': filtered_results_item_1[1]}, }) phantom.act(action="lookup ip", parameters=parameters, assets=['google_dns'], name="lookup_ip_1") 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
To learn more about the callback function, see callback in the Python Playbook API Reference for .
Collect input using phantom.collect API calls
Collect APIs are used to gather data from artifacts, containers, and action results. The data is then used in filtering, or used as input to act calls. The input to this phantom.act call is determined by the container data. In the following example, the geolocate ip action is using the input container artifacts for its data input.
Phantom.act calls also often use collect2 output. With collect2, the generated code loops through the list of results from collect2, and then appends a dictionary to the list parameters, with each row containing the ip and context parameters. This parameter list is passed to the phantom.act call.
def geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None): phantom.debug('geolocate_ip_1() called') # collect data for 'geolocate_ip_1' call container_data = phantom.collect2(container=container, datapath=['artifact:*.cef.sourceAddress', 'artifact:*.id']) parameters = [] # build parameters list for 'geolocate_ip_1' call for container_item in container_data: if container_item[0]: parameters.append({ 'ip': container_item[0], # context (artifact id) is added to associate results with the artifact 'context': {'artifact_id': container_item[1]}, }) phantom.act("geolocate ip", parameters=parameters, assets=['maxmind'], name="geolocate_ip_1") return
The purpose of the collect call is to extract the requested set of data from the container. In this example, the collect call requested the sourceAddress CEF artifacts. To get that set of artifacts, we use a datapath with the collect call.
Using phantom.act calls with collect2 output
The collect2 call can pre-filter results. The Visual Playbook Editor (VPE) generates code of this type when you use a decision block. This is shown in this filter block called by on_start:
def filter_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None): phantom.debug('filter_1() called') # collect filtered artifact ids for 'if' condition 1 matched_artifacts_1, matched_results_1 = phantom.condition( container=container, conditions=[ ["artifact:*.cef.sourceAddress", "<", "10.0.0.0"], ["artifact:*.cef.sourceAddress", ">", "10.255.255.255"], ], logical_operator='or', name="filter_1:condition_1") # call connected blocks if filtered artifacts or results if matched_artifacts_1 or matched_results_1: geolocate_ip_1(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1) return
This is passed to the geolocate block as a parameter, and then geolocate block uses it in the phantom.collect2 call as shown here:
def geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None): phantom.debug('geolocate_ip_1() called') # collect data for 'geolocate_ip_1' call container_data = phantom.collect2(container=container, datapath=['artifact:*.cef.sourceAddress', 'artifact:*.id']) parameters = [] # build parameters list for 'geolocate_ip_1' call for container_item in container_data: if container_item[0]: parameters.append({ 'ip': container_item[0], # context (artifact id) is added to associate results with the artifact 'context': {'artifact_id': container_item[1]}, }) phantom.act("geolocate ip", parameters=parameters, assets=['maxmind'], name="geolocate_ip_1") return
The datapath for the geolocate ip command starts with "filtered-artifacts:", so you are using the results of the filtered set, instead of all artifacts.
For more information on the phantom.act and phantom.collect API calls, see API Reference.
phantom.debug call
If you aren't getting the results you expected with your API calls, add a phantom.debug call below the collect2 call, as shown in the following example:
container_data = phantom.collect2(container=container, datapath=['artifact:*.cef.sourceAddress', 'artifact:*.id']) phantom.debug(container_data)
Once the collect2 call is debugged, the additional output in the debug window shows the sourceAddress artifacts and their artifact IDs.
Wed Mar 21 2018 20:32:37 GMT-0700 (PDT): [ [ "10.10.10.10", 2 ], [ "175.45.176.1", 3 ] ]
phantom.debug
prints the Python representation of the object that you pass to it.
container_data = phantom.collect2(container=container, datapath=['artifact:*.cef', 'artifact:*.id']) phantom.debug(container_data)
The debug output is this:
Wed Mar 21 2018 20:35:43 GMT-0700 (PDT): [ [ { "destinationAddress": "192.168.10.10", "sourceAddress": "10.10.10.10" }, 2 ], [ { "destinationAddress": "192.168.10.11", "sourceAddress": "175.45.176.1" }, 3 ] ]
Python Playbook Tutorial for overview | Tutorial: Create a simple playbook in |
This documentation applies to the following versions of Splunk® SOAR (On-premises): 5.1.0, 5.2.1, 5.3.1, 5.3.2, 5.3.3, 5.3.4, 5.3.5, 5.3.6, 5.4.0, 5.5.0, 6.0.0, 6.0.1, 6.0.2, 6.1.0, 6.1.1, 6.2.0, 6.2.1, 6.2.2, 6.3.0
Feedback submitted, thanks!