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:
Playbook automation API
The playbook automation API allows security operations teams to develop detailed automation strategies. Playbooks serve many purposes, ranging from automating small investigative tasks that can speed up analysis to large-scale responses to a security breach. The following APIs are supported to leverage the capabilities of the platform using playbooks.
act
The act API can be called from on_start() or the callback of any phantom.act() call. If multiple phantom.act() calls are called within the same function, they execute actions in parallel. If the action is executed on an asset that has primary approvers assigned or a reviewer specified, the action is not executed unless the primary approvers or reviewer approves the action.
The act API is not supported from within a custom function.
phantom.act(action, parameters=[], assets=None, tags=None,
callback=None, reviewer=None, handle=None,
start_time=None, name=None, asset_type=None,
app=None)
| Parameter | Required? | Description |
|---|---|---|
| action | Required | The name of the action that the user intends to run. Actions include block IP, list VM, or file reputation that are supported by the apps installed on the platform. |
| parameters | Optional | A list of dictionaries that contain the parameters expected by the action. The name of the keys are specific to the action being taken. |
| assets | Optional | A list of assets on which the action is run. If the user intends to take the action on a specific asset, it must be specified in this parameter. Assets are a list of asset IDs, as specified when an asset is configured. If the assets are configured with primary and secondary owners, the owners are required to approve an action before it can be run. If the asset is not specified, the action is run on all possible assets on which the action can be run. If multiple apps provide the same action for the same product, the system automatically uses the latest installed app.
If new assets or apps are added to the Splunk SOAR platform, they might run actions that you hadn't intended to run. For example, if you begin your deployment with a simple network-based topology and configure a perimeter firewall that supports block IP, and then add an active directory (AD) server which has an associated app that also reads block IP, that action is run on both the firewall and AD server. Setting appropriate approvals on assets can help to minimize this risk. |
| tags | Optional | A list of asset tags that help specify certain assets to be used for executing the action. You can assign assets a tag when they are configured. For example, if the asset is tagged critical and the action is block IP, the action is run only on assets that are tagged as critical. If tags and assets are both specified, then the action is run only on assets tagged with the matching tag. |
| callback | Optional | A specified callback function to be called upon completion of the action. Use the callback to evaluate the outcome of one action and then take more actions. Use the callback function to either serialize actions where you intend to run the actions one after the other, or where the subsequent action is dependent on the outcome or results of the first action. |
| reviewer | Optional | A username, email address, or group that receives an approval request to review the action before it is run. The user receives an approval request with all of the details of the action and its parameters. If is provided a comma-separated list or group, only one approval by any member of the list is required. SLA escalation settings affect how long the action is held for approval. |
| handle | Optional | An object that, when specified, is passed on to the callback. Users can save any Python object that the user needs to access in the context of the callback from the action called. Handle is always saved with the action and passed to the callback. It is best to use handles to pass objects from action to callbacks instead of global variables.
The size of the |
| start_time | Optional | The time when the action is scheduled for execution. This value is a datetime object.
params=[]
params.append({'ip':'1.1.1.1'})
# schedule 'geolocate ip' 60 seconds from now
when = datetime.now()+timedelta(seconds=60)
phantom.act("geolocate ip",
parameters=params,
start_time=when)
|
| name | Optional | A name the user can give to an instance of an action that is run. |
| asset_type | Optional | Use the asset_type parameter to limit the action on assets of the specified type. This parameter can be a string or a list of strings. |
| app | Optional | The specific app used to run the action. Specify the app as a Python dictionary: {"name":"some_app_name", "version":"x.x.x"}. "name" is case insensitive.
app_data={}
app_data['name']='MaxMind'
app_data['version']='1.1.0'
phantom.act('geolocate ip',
parameters=[{ "ip" : "1.1.1.1" }],
assets=["maxmind"],
callback=geolocate_ip_cb,
app=app_data)
or phantom.act('geolocate ip',
parameters=[{ "ip" : "1.1.1.1" }],
assets=["maxmind"],
callback=geolocate_ip_cb,
app={'name':'MaxMind', 'version':'1.1.0' })
|
This sample playbook uses the phantom.act() API.
import phantom.rules as phantom
import json
def geolocate_ip_cb(action, success, container, results, handle):
phantom.debug(results)
if not success:
phantom.debug('Action '+json.dumps(action)+' : FAILED')
return
return
def on_start(container):
ips = set(phantom.collect(container, 'artifact:*.cef.sourceAddress'))
parameters = []
for ip in ips:
parameters.append({ "ip" : ip })
if parameters:
phantom.act('geolocate ip', parameters=parameters, assets=["maxmind"], callback=geolocate_ip_cb)
return
def on_finish(container, summary):
return
callback
Callback functions are specified as parameters in phantom.act():
phantom.act('geolocate ip', parameters=parameters, assets=["maxmind"], callback=geolocate_ip_cb)Callback functions are called when the phantom.act() action has completed, regardless if the action succeeded or failed.
The phantom.act API takes a callback function, which accepts the following keyword arguments:
- action
- success
- container
- results
- handle
The simplest no-op callback looks like this:
def do_nothing(action=None, success=None, container=None, results=None, handle=None):
pass
Such a callback is functionally equivalent to not providing a callback parameter to phantom.act at all.
| Parameter | Description |
|---|---|
| action | A JSON object that specifies the action name and the action run ID. The action run ID uniquely identifies that instance of action execution and can be used to retrieve the details of the action execution. The following example is an example of an action parameter from a callback:
{
"action_run_id": 205,
"action_name": "whois ip",
"action" : "whois ip",
"name": "my_whois_ip_action"
}
The |
| success | Either true or false. An action is considered failed only if the action has failed on all assets and with all specified parameters. If any part of the action succeeds, it is not considered failed. Use the utility functions like parse_errors() or parse_success() to get a flat listing of all errors and success. These utility functions parse the results to give the user a different view of the overall results.
|
| container | The container JSON object. Here is an example of container parameter from a callback:
{
"sensitivity": "amber",
"create_time": "2016-01-14 18:25:55.921199+00",
"owner": "admin",
"id": 7,
"close_time": "",
"severity": "medium",
"label": "incident",
"due_time": "2016-01-15 06:24:00+00",
"version": "1",
"current_rule_run_id": 1,
"status": "open",
"owner_name": "",
"hash": "093d1d4d22cab1c5931bbfb1b16ce12c",
"description": "this is my test incident",
"tags": ["red-network"],
"start_time": "2016-01-14 18:25:55.926468+00",
"asset_name": "",
"artifact_update_time": "2016-01-14 18:26:33.55643+00",
"container_update_time": "2016-01-14 18:28:43.859814+00",
"kill_chain": "",
"name": "test",
"ingest_app_id": "",
"source_data_identifier":
"48e4ab9c-2ec1-44a5-9d05-4e83bec05f87",
"end_time": "",
"artifact_count": 1
}
|
| results | A JSON object that has the full details and status of the complete action on all assets for each parameter. Here is an example where the geolocate ip action is run:
parameters = []
parameters.append({ "ip" : '1.1.1.1' })
phantom.act('geolocate ip', parameters=parameters,
assets=["maxmind"], callback=geolocate_ip_cb,
name='my_geolocate_action')
This simple action can result in various execution strategies and outcomes, depending on how the system is configured. In this simple form, one app supports the geolocate IP action and there is one Maxmind asset that is configured, which results in one IP being queried once on one asset. In a more complex example, if there are two apps, both of which support file reputation, then this one simple action results in a file hash queried on both of the assets. The single action file reputation results in the hash being queried twice, once on each asset. The two queries still constitute one action, so the callback The Here are the results of the geolocate IP action. [
{
"asset_id": 63,
"status": "success",
"name": "geolocate_ip_1",
"app": "MaxMind",
"action_results": [
{
"status": "success",
"data": [
{
"state_name": "Victoria",
"latitude": -37.7,
"country_iso_code": "AU",
"time_zone": "Australia/Melbourne",
"longitude": 145.1833,
"state_iso_code": "VIC",
"city_name": "Research",
"country_name": "Australia",
"continent_name": "Oceania",
"postal_code": "3095"
}
],
"message": "City: Research, State: VIC,
Country: Australia",
"parameter": {
"ip": "1.1.1.1",
"context": {
"guid": "f42fd73f-...-8194aaa9bc11",
"artifact_id": 0,
"parent_action_run": []
}
},
"summary": {
"city": "Research",
"state": "VIC",
"country": "Australia"
}
}
],
"app_id": 83,
"summary": {
"total_objects": 1,
"total_objects_successful": 1
},
"asset": "maxmind",
"action": "geolocate ip",
"message": "'geolocate_ip_1' on asset 'maxmind': 1
action succeeded. (1)For Parameter:
{\"ip\":\"1.1.1.1\"} Message: \"City:
Research, State: VIC, Country: Australia\"",
"app_run_id": 51,
"action_run_id": 11
}
]
|
| handle | An object that was specified in the action phantom.act() call for passing data between action and callbacks. |
completed
The completed API checks if all of the provided runnables have finished running. Runnables are defined as actions, synchronous child playbooks, and custom functions. A runnable is finished running if its status is either succeeded or failed. Succeeded or failed implies that the action is done. If any combination of the action names, playbook names, or custom function names are not completed, then the function returns False. Use the completed API in the join function where certain blocks are run in parallel but the next block has to be called only when all the joining blocks have completed executing.
The completed API is not supported from within a custom function.
phantom.completed(action_names=None, playbook_names=None, custom_function_names=None, trace=False)
| Parameter | Required? | Description |
|---|---|---|
| action_names | Optional | A list of names given to an action through the phantom.act() API in the parameter name.
|
| playbook_names | Optional | A list of names given to a playbook execution using phantom.playbook() API in the parameter name.
|
| custom_function_names | Optional | A list of names given to a custom function using the phantom.custom_function API in the parameter name.
|
This sample uses the completed API.
def join_add_tag_1(
action=None,
success=None,
container=None,
results=None,
handle=None,
filtered_artifacts=None,
filtered_results=None,
):
# Continue if all of the dependent blocks have completed
if phantom.completed(
action_names=['whois_ip_1'],
playbook_names=['playbook_send_precautionary_email_1'],
custom_function=['get_subnet_1'],
):
# call subsequent block "add_tag_1"
add_tag_1(container=container, handle=handle)
return
concatenate
The completed API combines all arguments together into a list. If an argument is, itself, a list, the API will concatenate that list to the rest of the arguments, or append it to the list the API is creating. In this case, the final list will not have embedded lists. This is a shallow flattening - if an argument is a list of lists, the final resulting list will include some values as lists.
The completed API is not supported from within a custom function.
phantom.concatenate(*args, **kwargs):
| Parameter | Required? | Description |
|---|---|---|
| dedup | Optional | A Boolean value that removes duplicate argument values, specifically ignoring values that are not JSON serializable. |
Here are several example commands and their corresponding results.
concatenate("1", "2", "3", "3")
Result: ["1", "2", "3", "3"]
concatenate(["1", "2", "3"], ["4", "5"])
Result: ["1", "2", "3", "4", "5"]
concatenate({"val": 1}, {"val": 2}, {"val": 3})
Result: [{"val": 1}, {"val": 2}, {"val": 3}]
concatenate(["1", "2", "2", "3"], ["2", "3", "4", "5"], dedup=True)
Result: ["1", "2", "3", "4", "5"]
concatenate([1, [2, 3], [2, 3], 6], [[4, 5], 6, [7, 8], 1], dedup=True)
Result: [1, [2, 3], 6,[4, 5],[7, 8]],
condition
The condition API implements the decision block in the visual playbook editor (VPE). It evaluates expressions and returns matching artifacts and actions results that evaluate as true. Each filter block you create in the VPE calls condition.
The condition API is not supported from within a custom function.
phantom.condition(container=None,
action_results=None,
conditions=[],
logical_operator='or',
scope='new',
filtered_artifacts=[],
filtered_results=[],
limit=100,
name=None,
trace=False,
case_sensitive=True,
delimiter=","
auto=True)
| Parameter | Required? | Description |
|---|---|---|
| container | Required | The container dictionary object that is passed into the calling function. |
| action_results | Optional | The action results passed into any callback function or a subset of action results that had been filtered from a condition call. When you pass action results, you can also pass in custom function results. In other words, action results can be both action results and custom function results. |
| conditions | Required | A list of one or more and or or expressions to be evaluated. Matching artifacts or matching action results are returned.
The following example shows the expression format: [ [ LHS, OPERATOR, RHS ], [ LHS, OPERATOR, RHS ], ..]
==, !=, <, >, <=, >=, in, not in, is true, is false, is none, matches regex, is empty, is not empty, is list, is not list LHS and RHS values can be a value, artifact datapath, action result datapath, custom function result datapath, or a custom list datapath. |
| logical_operator | Optional | Expresses the relationship between conditions. Valid logical operators are and or or. Defaults to or.
|
| scope | Optional | See collect. Possible values include new, all, or an artifact ID.
|
| filtered_artifacts | Optional | Filtered artifacts that were returned from a preceding condition block. |
| filtered_results | Optional | Filtered results that were returned from a preceding condition block. |
| limit | Optional | See collect. |
| name | Optional | Specify a unique name to save the filtered action results and filtered artifacts to retrieve using either the collect2() API or phantom.get_filtered_data() API. |
| trace | Optional | Trace is a flag related to the level of logging. If trace is on (True), more logging is enabled. When set to True, more detailed output is displayed in debug output. |
| case_sensitive | Optional | Default is True. Set to False for evaluating conditions in a case-insensitive manner. |
| delimiter | Optional | Default is ",". For fields in artifacts (CEF fields). Set this value to any string to split the artifact's value by that string and treat the results as a list. Set to None to disable the splitting behavior. Default behavior of splitting on commas is deprecated. Avoid comma separation, because some CEF fields might contain commas and cause unexpected results in a list.
|
| auto | Optional | A Boolean value where the default is True. When this value is True, remove the database record associated with the filtered data once the playbook run has finished. |
The condition API returns a list of artifact IDs and a list of action result objects. These are the artifacts, actions results, and custom function results that match the conditions expressed. If you don't specify a filter statement about action results, no filtered action results or custom function results are returned and the VPE UI doesn't show that as a selectable option in subsequent blocks.
When using the VPE, you can select to connect various UI blocks. Each of these blocks implements a function in the auto-generated Python code. These functions have various parameters like container, results, filtered_artifacts, and filtered_results. The expressions used for the conditions can be either a constant or a datapath to specify what you need to retrieve and operate on. These datapaths can point to either a field in the artifact, action_result, filtered-artifacts, filtered results, or a constant.
To learn more about the datapaths used in this API, see Understanding datapaths.
If two strings that can be converted to a numeric type are being compared with one of the following operators, they are converted to numeric types before the comparison occurs:
==, !=, <, >, <=, >=
Example of condition
Here is some sample code that uses phantom.condition.
def filter_1(
action=None,
success=None,
container=None,
results=None,
handle=None,
filtered_artifacts=None,
filtered_results=None,
custom_function=None,
):
action_results = [
{
'name':
'normalize_ip_1',
'action_results': [{
'data': [{
'ip': '3.3.3.3',
}],
'parameter': {
'ip': '3.3.3.3\n',
},
}, ],
},
{
'name':
'normalize_ip_1',
'action_results': [{
'data': [{
'ip': '2.2.2.2',
}],
'parameter': {
'ip': '2.2.2.2\n',
},
}, ],
},
]
conditions = [
[
'normalize_ip_1:action_result.data.*.ip',
'==',
'2.2.2.2',
],
[
'normalize_ip_1:action_result.data.*.ip',
'==',
'0.0.0.0',
],
]
# Call phantom.condition
matched_artifacts_1, matched_results_1 = phantom.condition(container=container, action_results=action_results, conditions=conditions, logical_operator='or')
# The value of matched_results_1 is
assert matched_results_1 == [
{
'name':
'normalize_ip_1',
'action_results': [{
'data': [{
'ip': '2.2.2.2',
}],
'parameter': {
'ip': '2.2.2.2\n',
},
}, ],
},
]
# The value of matched_artifacts_1 is
assert matched_artifacts_1 == []
# Call the callback
# The other parameters come from inputs into the block
if matched_artifacts_1 or matched_results_1:
domain_reputation_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,
)
custom_function
Use the custom_function API to call a custom function from a playbook. The following table describes the parameters used in this function.
The custom_function API is not supported from within a custom function.
def custom_function(custom_function=None, parameters=None, callback=None, name=None):
| Parameter | Description |
|---|---|
| custom_function | The custom function identifier. The Visual Playbook Editor (VPE) generates this identifier for you if you select your custom function through the configuration panel. Otherwise, use the following format:
# The custom function identifier specification <repository_name>/drafts?/<custom_function_name> # Valid identifier: <repository_name>/<custom_function_name> local/make_upper community/combine_datapaths # Valid identifier: <repository_name>/drafts/<custom_function_name> local/drafts/first_ten # Invalid identifier: missing a repository my_custom_function # Valid identifier, but drafts will be interpreted as a repository name drafts/my_custom_function |
| parameters | A list of dictionaries containing the inputs to pass to the custom function callback. The shape of the dictionaries that are in the parameters list depends on what custom function you are calling. |
| name | The name of the custom function block. This is autogenerated by the VPE, but you can specify your own name from the configuration panel for the block using Advanced Settings > General Settings > Name. If you are not using the VPE, be aware that the name must be unique amongst all of the names in your playbook. For example, you cannot use the same name as an action elsewhere in the playbook. |
| callback | A callable object with a certain function signature. It is typically a function or possibly any Python callable. Invoke the object that you provide as the callback parameter as follows:
callback(
container=container,
results=result,
handle=handle,
custom_function=custom_function,
success=success
)
Your callable object must be able to accept these keyword arguments. |
Example custom_function results object
The results parameter passed to the callback looks like this:
[
{
'custom_function_name': 'to_upper',
'custom_function_results': [
{
'data': {
'upper': 'HELLO',
},
'parameter': {
'value': 'hello',
},
},
],
'custom_function_run_id': 14,
'message': '',
'name': 'to_upper_1',
'status': 'success',
},
]
The length of the list corresponding to the custom_function_results key is the same as the length of the parameters list that was passed to the custom_function API.
callback
Callback functions are specified as parameters in the custom_function API:
phantom.custom_function('local/to_upper', parameters=[{'value': 'hello world'}], callback=decision_1)The custom_function API takes a callback function, which accepts the following keyword arguments: custom_function, success, container, results, and handle. Although Python allows callers to pass keyword arguments in any order, customized callback functions accept the keyword arguments in the same order as previously listed, since Python also allows keyword arguments to be passed by position.
The simplest callback looks like this:
def do_nothing(custom_function=None, success=None, container=None, results=None, handle=None):
pass
This callback is equivalent to not providing a callback parameter to phantom.custom_function at all.
Callback functions are called when the phantom.custom_function() action has completed, irrespective of action success or failure.
| Parameter | Description |
|---|---|
| custom_function | A JSON object that specifies the metadata about the custom function that triggered the callback. The custom_function_run_id value corresponds to the object in the database that contains the data for the custom function run. You can give this ID to the phantom.get_custom_function_results API in order to retrieve the custom function results synchronously. The name value is the same as the name value passed to the triggering call of the API custom_function. It uniquely identifies the block name of the calling custom function. The custom_function_name parameter corresponds to the name of the triggering custom function.
Here is an example {
'custom_function_run_id': 22,
'custom_function_name': 'phtest_cf_to_upper',
'name': 'to_upper_1',
}
|
| success | Returns as either true or false. A custom function always has a status of success unless it raises an uncaught exception. |
| container | The container JSON object. Here is an example of a container parameter from a callback:
{
"sensitivity": "amber",
"create_time": "2016-01-14 18:25:55.921199+00",
"owner": "admin",
"id": 7,
"close_time": "",
"severity": "medium",
"label": "incident",
"due_time": "2016-01-15 06:24:00+00",
"version": "1",
"current_rule_run_id": 1,
"status": "open",
"owner_name": "",
"hash": "093d1d4d22cab1c5931bbfb1b16ce12c",
"description": "this is my test incident",
"tags": ["red-network"],
"start_time": "2016-01-14 18:25:55.926468+00",
"asset_name": "",
"artifact_update_time": "2016-01-14 18:26:33.55643+00",
"container_update_time": "2016-01-14 18:28:43.859814+00",
"kill_chain": "",
"name": "test",
"ingest_app_id": "",
"source_data_identifier":
"48e4ab9c-2ec1-44a5-9d05-4e83bec05f87",
"end_time": "",
"artifact_count": 1
}
|
| results | A JSON object that contains all of the custom function results produced by the triggering call to phantom.custom_function. Here is an example where the phtest_to_upper action is run:
[
{
"custom_function_name": "phtest_cf_to_upper",
"custom_function_results": [
{
"data": {
"upper": "hello world",
},
"parameter": {
"value": "hello world",
},
},
],
"custom_function_run_id":22,
"message": "",
"name": "to_upper_1",
"status": "success",
},
]
|
| handle | An object that is specified in the action phantom.custom_function() call for passing data between custom functions and callbacks. |
| status | The status of the custom function that was run. Status returns as either success or fail. |
debug
When logging is enabled, the debug API lets the author debug as the playbook is being developed and tested. This is similar to a print() statement. The parameter for the call is a string type object and the contents are shown in the debug console in cyan text so that you can distinguish your text from the system text.
The debug API is supported from within a custom function.
phantom.debug(message)
The following example shows the debug API:
def on_start(container):
phantom.debug('in on_start() of playbook')
return
The response looks something like this:
2016-02-13T01:32:52.583000+00:00: calling on_start(): on incident 'test', id: 107. 2016-02-13T01:32:52.608695+00:00: in on_start() of playbook
The debug and error APIs encode arguments to UTF-8 before printing them. If debug is passed a Python list or a dictionary at any level of nesting, it decodes any unicode strings within that mutable object. This means that calling debug or error can mutate the argument passed to those functions. To work around this behavior, do a deep copy of the object that you want to debug and pass the copy to phantom.debug as shown in the following example:
from copy import deepcopy names = [u'José', u'María', u'Rosa'] names_copy = deepcopy(names) # Debug print names_copy, thus preserving names phantom.debug(names_copy) # Due to the bug, names_copy has been mutated assert names != names_copy assert names == [u'José', u'María', u'Rosa']
decision
Decision blocks in the Visual Playbook Editor generate calls to the decision API. The decision API returns a Boolean value to indicate decision success or failure. You can have up to five such True/False outcomes.
The decision API is a mechanism of control flow so it can't be called from within a custom function.
phantom.decision(container=None,
action_results=None,
conditions=[],
logical_operator='or',
scope='new',
filtered_artifacts=[],
filtered_results=[],
limit=100,
name=None,
trace=False,
case_sensitive=True,
delimiter=","
auto=True)
| Parameter | Required? | Description |
|---|---|---|
| container | Required | The container dictionary object that is passed into the calling function. |
| action_results | Optional | The action results passed into any callback function or a subset of action results that were filtered from a phantom.condition() call. When you pass in action results, you can also pass in custom function results. Action results can be both action results and custom function results. |
| conditions | Required | A list of one or more and or or expressions to be evaluated. Matching artifacts or matching action results are returned.
The following example shows the expression format: [ [ LHS, OPERATOR, RHS ], [ LHS, OPERATOR, RHS ], ..]
==, !=, <, >, <=, >=, in, not in LHS and RHS values can be a value, artifact datapath, action result datapath, custom function result datapath, or a custom list datapath. |
| logical_operator | Optional | Expresses the relationship between conditions. Valid logical operators are and or or. Defaults to or. If the logical operator is and then each expression passed to the condition must be true on the same result, if the expression relates to a result, for decision to return true. Potential result types are artifacts, action results, or custom function results.
|
| scope | Optional | See the collect API documentation. Possible values include new, all, or an artifact ID.
|
| filtered_artifacts | Optional | Filtered artifacts that were returned from a preceding phantom.condition() block. |
| filtered_results | Optional | Filtered results that were returned from a preceding phantom.condition() block. |
| limit | Optional | This enforces the maximum number of artifacts that can be retrieved in this call. The default is 100. |
| name | Optional | Specify a unique name to save the filtered action results and filtered artifacts which can be retrieved using either the collect2() API or the phantom.get_filtered_data() API. |
| trace | Optional | Trace is a flag related to the level of logging. If trace is on (True), more logging is enabled. When set to True, more detailed output is displayed in debug output. |
| delimiter | Optional | Default is ",". For fields in artifacts (CEF fields). Set this value to any string to split the artifact's value by that string and treat the results as a list. Set to None to disable the splitting behavior. Default behavior of splitting on commas is deprecated. Avoid comma separation, because some CEF fields might contain commas and cause unexpected results in a list.
|
| case_sensitive | Optional | Default is True. Set to False for evaluating conditions in a case-insensitive manner. |
| auto | Optional | A Boolean value where the default is True. When this value is True, remove the database record associated with the filtered data once the playbook run has finished. |
automatically converts true and false strings to their Boolean values.
If two strings that can be converted to a numeric type are being compared with one of the following operators, they are converted to numeric types before the comparison occurs:
==, !=, <, >, <=, >=
Example of decision
Here is some sample code that uses phantom.decision.
def decision_2(
action=None,
success=None,
container=None,
results=None,
handle=None,
filtered_artifacts=None,
filtered_results=None,
custom_function=None,
):
action_results = [
{
'name':
'normalize_ip_1',
'action_results': [{
'data': [{
'ip': '3.3.3.3',
}],
'parameter': {
'ip': '3.3.3.3\n',
},
}, ],
},
{
'name':
'normalize_ip_1',
'action_results': [{
'data': [{
'ip': '2.2.2.2',
}],
'parameter': {
'ip': '2.2.2.2\n',
},
}, ],
},
]
conditions = [
[
'normalize_ip_1:action_result.data.*.ip',
'==',
'2.2.2.2',
],
[
'normalize_ip_1:action_result.data.*.ip',
'==',
'0.0.0.0',
],
]
# Call to the phantom.decision API
# With logical_operator set to 'or'
matched = phantom.decision(container=container, action_results=action_results, conditions=conditions, logical_operator='or')
# Return value is True
assert matched is True
# Call to the phantom.decision API
# With logical_operator set to 'and'
phantom.decision(container=container, action_results=action_results, conditions=conditions, logical_operator='and')
# Return value is False
assert matched is False
# Call to the phantom.decision API
# with literal conditions
phantom.decision(container=container, action_results=action_results, conditions=[[4, '==', 4], [True, '!=', False], logical_operator='and')
# Return value is True
assert matched is True
# Call the callback function
if matched:
send_email_1(
action=action,
success=success,
container=container,
results=results,
handle=handle,
custom_function=custom_function,
)
Understanding the "in" operator
Use the optional "in" operator with strings, lists, dictionaries, the delimiter and CEF strings, and the delimiter and toEmail / fromEmail CEF.
See the following table for more information about using the "in" operator with the various data types.
| Type | Can you use the "in" operator? | Description of results |
|---|---|---|
| Integer | No | Not applicable. |
| Strings | Yes | The "in" operator checks if the operator on the left-hand side is a substring of the string on the right-hand side. |
| Lists | Yes | The operator checks if the element is in the list, and provides an output similar to the following.
found_match_1 = phantom.decision(
container=container,
conditions=[
["Hello", "in", ["Hello", "Goodbye"]]
],
delimiter=None)
phantom.debug(f"Output: {found_match_1}")
assert found_match_1 == True, "Output should be True"
|
| Dictionaries | Yes | The operator only works with dictionary keys. |
| Delimiter and CEF | Yes, but this only applies to strings. | When the delimiter option is specified, string CEF fields will be split at the delimiter and converted to lists. See the following examples.
# act = "test1, test2", app = "test2, test1"
found_match_1 = phantom.decision(
container=container,
conditions=[
["artifact:*.cef.act", "in", "artifact:*.cef.app"]
],
delimiter=",")
assert found_match_1 == True, "Output should be True"
# act = "string", app = "string 1, string 2"
found_match_1 = phantom.decision(
container=container,
conditions=[
["artifact:*.cef.act", "in", "artifact:*.cef.app"]
],
delimiter=",")
assert found_match_1 == False, "Output should be False"
|
| Delimiter and ToEmail / fromEmail CEF | Yes | For toEmail and fromEmail CEF types, if the delimiter is a comma, the delimiter will not be used to split the string if the delimiter is quoted with double quotes. See the following examples.
# toEmail cef = `"Alex, Lastname" <alex@buttercupgames.com>`
found_match_1 = phantom.decision(
container=container,
conditions=[
["splunk.com", "in", "artifact:*.cef.toEmail"]
],
delimiter=",") # notice that delimiter is no longer None
assert found_match_1 == True, "Output should be True"
# toEmail cef = `Alex, Lastname <alex@buttercupgames.com>`
found_match_1 = phantom.decision(
container=container,
conditions=[
["splunk.com", "in", "artifact:*.cef.toEmail"]
],
delimiter=",") # notice that delimiter is no longer None
assert found_match_1 == False, "Output should be False"
|
discontinue
Use the discontinue API in a playbook to cancel runs for subsequent playbooks in the playbook batch.
- Use this API only inside an on_start() block.
- If called during an ingestion event, this API stops later playbooks in the execution order from starting.
- Using phantom.discontinue does not affect the current playbook. It only affects future playbooks.
- Using this API by calling it directly on its own or in debug mode does nothing, as there are no future playbooks to discontinue.
generally uses a label, on the Administration page, to determine whether to run a playbook. If you have a batch of playbooks in a container with the same label, runs them all, in sequence. If you add phantom.discontinue to any playbook of the batch, it will cancel all subsequent, pending playbook runs.
phantom.discontinue()
Example:
def on_start(container):
phantom.discontinue()
error
The error API lets the author debug or print log messages as the playbook is run with logging disabled. This is similar to a print() statement. The parameter for the call is a string type object and the contents are shown in the playbook debug console in red text so that you can distinguish your text from the system text.
The error API is supported from within a custom function.
phantom.error(message)
The following example shows the error API:
def on_start(container):
phantom.error('in on_start() of playbook')
return
The response looks like something like this:
2016-02-13T01:32:52.583000+00:00: calling on_start(): on incident 'test', id: 107. 2016-02-13T01:32:52.608695+00:00: in on_start() of playbook
The error and debug APIs encode arguments to UTF-8 before printing them. If phantom.debug is passed a Python list or a dictionary at any level of nesting, it decodes any unicode strings within that mutable object. This means that calling debug or error can mutate the argument passed to those functions. To work around this behavior, do a deep copy of the object that you want like to debug and pass the copy to debug as shown in the following example:
from copy import deepcopy names = [u'José', u'María', u'Rosa'] names_copy = deepcopy(names) # Debug print names_copy, thus preserving names phantom.debug(names_copy) # Due to the bug, names_copy has been mutated assert names != names_copy assert names == [u'José', u'María', u'Rosa']
format
The format API formats text with values that are extracted using datapaths for other complex objects such as artifacts or action results.
The format API is supported from within a custom function.
phantom.format(container=None,
template=None,
parameters=None,
scope='new',
name=None,
trace=False,
separator=None,
drop_none=False):
| Parameter | Required? | Description |
|---|---|---|
| container | Required | The container object passed into the action callback or on_start.
|
| template | Required | The format string where positional arguments are substituted with values. The arguments are expressed and passed as a list of datapaths in the parameters argument. If the datapath returns a list of items, the positional argument is replaced by a comma-separated value of the items. The format string uses positional arguments that are the same as Python string formatting.
|
| parameters | Required | A list of datapaths with a corresponding datapath for each positional format argument used in the template string. |
| scope | Optional | See collect for more information. The default value for scope is new, but the values can be either all or new.
|
| name | Optional | The name used to save the resulting formatted data. Use this name to retrieve this parameter through the get_format_data() API. If this parameter is not specified, the data is not saved. |
| trace | Optional | Trace is a flag related to the level of logging. If trace is on (True), more logging is enabled. When set to True, more detailed output is displayed in debug output. |
| separator | Optional | If a datapath response contains a list of strings or numbers, but not Python objects, the default output separator is
|
| drop_none | Optional | The default value is False. By default None values are included in the resulting string, but the user can filter None type values in the resulting string through this parameter.
|
Example request
This sample uses the phantom.format API.
def on_start(container):
template = "Host '{0}' transferred in '{1}' bytes"
datapaths = ['artifact:*.cef.sourceAddress','artifact:*.cef.bytesIn']
formatted_data = phantom.format(container=container,
template=template,
parameters=datapaths,
name='formatted_1')
phantom.debug("Formatted data is: {}".format(formatted_data))
# use the same name that was provided for key in phantom.format()
formatted_data = phantom.get_format_data(name="format_1")
phantom.debug("Retrieved formatted data is: {}".format(formatted_data))
return
Example response
The following is an example of what the format API returns.
Starting playbook 'format_test' on 'incident' id '13' with playbook run id '7'.
calling on_start() on incident 'test_incident'(id: 13).
phantom.collect2(): called for datapath['artifact:*.cef.sourceAddress']
phantom.collect2(): called for datapath['artifact:*.cef.bytesIn']
save_run_data() called
Formatted data is: Host '1.1.1.1' transferred in '999' bytes
get_run_data() called
Retrieved formatted data is: Host '1.1.1.1' transferred in '999' bytes
No actions were executed
calling on_finish()
Playbook 'format_test' (playbook id: 6) executed (playbook run id: 7) on
incident 'test_incident'(container id: 13).
Playbook execution status is 'success'
Total actions executed: 0
{"message":"No actions were executed","playbook_run_id":7,"result":[],
"status":"success"}
In the previous example, if there are multiple artifacts artifact:*.cef.sourceAddress refers to a list of IPs, and the output looks like the following:
Retrieved formatted data is: Host '1.1.1.1', '8.8.8.8', '8.8.4.4' transferred in '999', '888', '777' bytes
For each pair of sourceAddress and bytesIn to have their own line in this output, wrap sections of the format text in %%, as shown in this sample.
def on_start(container):
template = """\
%%
Host '{0}' transferred in '{1}' bytes"
%%"""
datapaths = ['artifact:*.cef.sourceAddress','artifact:*.cef.bytesIn']
formatted_data = phantom.format(container=container,
template=template,
parameters=datapaths,
name='formatted_1')
phantom.debug("Formatted data is: {}".format(formatted_data))
# use the same name that was provided for key in phantom.format()
formatted_data = phantom.get_format_data(name="format_1")
phantom.debug("Retrieved formatted data is: {}".format(formatted_data))
return
This creates the following string:
Retrieved formatted data is: Host '1.1.1.1' transferred in '999' bytes Host '8.8.8.8' transferred in '888' bytes Host '8.8.4.4' transferred in '777' bytes
When creating this string, each section is saved into a list. When using the VPE, you can get this list for the format block, and then each item in this list will be passed a parameter to the action as shown in the following example.
def on_start(container):
template = """\
%%
Host '{0}' transferred in '{1}' bytes"
%%"""
datapaths = ['artifact:*.cef.sourceAddress','artifact:*.cef.bytesIn']
# formatted_data will always be a string
formatted_data = phantom.format(container=container,
template=template,
parameters=datapaths,
name='formatted_1')
# Add __as_list to the end of the name to retrieve the list instead of the string
formatted_data_list = phantom.get_format_data(name="format_1__as_list")
return
playbook
The playbook API enables users to call another playbook from within the current playbook. If there are two or more playbooks with the same name from different repositories, the call fails. As such, use the format "repo_name/playbook_name" to be specific. The playbook API returns the playbook_run_id that can be used to query corresponding playbook execution details and report.
The playbook API is not supported from within a custom function.
phantom.playbook(playbook=None, container=None, handle=None, show_debug=False, callback=None, inherit_scope=True, inputs=None, name=None)
| Parameter | Required? | Description |
|---|---|---|
| playbook | Required | The playbook name to run. Use the format "repo_name/playbook_name". |
| container | Required | The container JSON object that needs to be passed to run the playbook on. This is the same container JSON object that you get in on_start() or any other callback function.
|
| handle | Optional | An object that you can pass to the API that is passed back to the callback when the playbook finishes execution. |
| show_debug | Optional | The default for this parameter is False, but if you set it to True, the debug messages of the launched playbook is shown in the debug window when you debug the caller playbook. |
| callback | Optional | If this parameter is specified, the playbook is launched in a synchronous fashion. When the child playbook finishes, the specified callback function is called with playbook execution results. When child playbooks are launched synchronously, the parent playbook is not considered completed until the called child playbook has finished executing. If this parameter is specified, you must also specify the name parameter.
|
| inherit_scope | Optional | Default is True. This parameter implies that the child playbook inherits the scope settings from the parent when called. If set to false, the child playbook runs with the default playbook scope. |
| inputs | Optional | If specified, this value must be a JSON-serializable dictionary of one or more key/value pairs, passed as an input for the child playbook that is called. Find the input specifications in the input_spec field, either in the user interface or in the REST API. Be sure both of the following are true:
|
| name | Required | An optional parameter, unless the callback parameter is specified. The name can be any unique identifier for this playbook run. If the code for calling the child playbook is auto-generated, the name of the playbook block that called this playbook populates automatically. |
This sample playbook shows calling a playbook from a playbook.
"""
This sample playbook shows calling a playbook from a playbook
"""
import json
import phantom.rules as phantom
def playbook_file_analysis_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, **kwargs):
phantom.debug("playbook_file_analysis_1() called")
inputs = {
"file_hash": "09ca7e4eaa6e8ae9c7d261167129184883644d07dfba7cbfbc4c8a2e08360d5b",
"file_name": "hello.txt",
}
# call playbook "local/file_analysis", returns the playbook_run_id
playbook_run_id = phantom.playbook("local/file_analysis", container=container, inputs=inputs)
return
prompt
Using phantom.prompt() results in a message sent to the specified approvers.
- Approvers can be users or roles.
- A notification is sent to the notification bell icon in the upper right corner of the screen for the specified approvers.
- If has been configured with an SMTP asset, and the approvers have valid email addresses in their account settings, the approvers are sent an email.
An email notification sent using phantom.prompt or phantom.prompt2 cannot be disabled by Splunk SOAR users by disabling notifications in their account settings.
Pending notifications can be accessed by clicking the bell icon in the top right corner of the UI. Delegation for approvals is possible. Either the approver or a delegate can complete the task. When the task is completed, the prompt() callback is called with the final response included in the action results.
The structure of the callback function and all the parameters is consistent with an action callback. Using prompt only allows the user to complete a task. The playbook can contain a callback function and use the prompt response, found in the result object in the callback, to change playbook behavior.
The prompt API is not supported from within a custom function.
phantom.prompt(user=None,
message='',
respond_in_mins=30,
callback=None,
name=None,
options=None,
parameters=None,
container=None,
scope='new',
trace=False,
separator=None,
drop_none=False)
| Parameter | Required? | Description |
|---|---|---|
| container | Required | The object that is associated with the current playbook execution. This object is available to all action callbacks and other playbook execution functions. |
| user | Required | The recipient in the form of a user email address, username, or a role. Must be a valid user or role in . |
| message | Required | The message content to send. |
| respond_in_mins | Optional | The time the user is given to respond. Default is 30 minutes. If the user does not respond in the specified time, the prompt fails and a failed status is sent to the callback. |
| callback | Optional | This parameter the same prototype as action callbacks. Status indicates success when the user has responded to the action and is failure only when the user does not respond in the specified time. The results JSON has the same format as any action results. Handle is not used and is an empty object. |
| name | Optional | The name of the action. |
| options | Optional | A JSON dictionary. Allows the user response to display with programmed choices. |
| parameters | Optional | A list of datapaths whose values are used to format the message. Recognized datapaths are used to retrieve data, and the data is used to populate the curly brackets in the message. The first parameter replaces {0}, the second replaces {1}, and so on. |
| scope | Optional | The scope can either be new or all. Default value is new. See collect for more information.
|
| trace | Optional | Trace is a flag related to the level of logging. If trace is on (True), more logging is enabled. When set to True, more detailed output is displayed in debug output. |
| separator | Optional | Specify an alternate separator using this parameter. If a datapath response contains a list, the default output separator is ', '.
|
| drop_none | Optional | By default, the None values are included in the resulting string.
|
This sample playbook shows calling a phantom.prompt() from a playbook.
"""
This sample playbook shows calling a phantom.prompt() from a playbook
"""
import phantom.rules as phantom
import json
def on_start(container):
phantom.prompt(container=container,
user="user@company.com",
message="proceed with blocking these ips on FW?",
respond_in_mins=10,
callback=prompt_cb,
options={ 'type': 'list', 'choices': ['yes', 'no', 'maybe'] },
name="prompt_to_block_ips")
return
def prompt_cb(action, success, container, results, handle):
phantom.debug(results)
return
def on_finish(container, summary):
return
The following shows the output of the playbook:
Fri Apr 29 2016 19:38:25 GMT-0700 (PDT): Starting playbook 'manual_action' testing on 'incident' id: '215'...
Fri Apr 29 2016 19:38:25 GMT-0700 (PDT): calling on_start(): on incident 'test', id: 215.
Fri Apr 29 2016 19:38:25 GMT-0700 (PDT): phantom.act(): Warning: For action 'prompt' no assets were specified. The action shall execute on all assets the app (supporting the action) can be executed on
Fri Apr 29 2016 19:38:25 GMT-0700 (PDT): phantom.act(): action 'prompt' shall be executed with parameters: '[{"to": "user@company.com", "message": "please make sure xyz is ok ...", "mins_to_act": 10}]', assets: '', callback function: 'prompt_cb', with no action approver, no delay to execute the action, no user provided name for the action, no tags, no asset type
Fri Apr 29 2016 19:38:25 GMT-0700 (PDT): Request sent for action'automated action 'prompt' of 'manual_action' playbook'
Fri Apr 29 2016 19:38:50 GMT-0700 (PDT): Manual action was completed by the user. User message: yes I am OK..
Fri Apr 29 2016 19:38:50 GMT-0700 (PDT): calling action callback function: prompt_cb
Fri Apr 29 2016 19:38:50 GMT-0700 (PDT):
[
{
"asset_id": 0,
"status": "success",
"name": "prompt_to_block_ips",
"app": "",
"action_results": [
{
"status": "success",
"data": [
{
"response": "maybe"
}
],
"message": "proceed with blocking these ips on FW?",
"parameter": {
"message": "proceed with blocking these ips on FW?"
},
"summary": {
"response": "maybe"
}
}
],
"app_id": 0,
"app_run_id": 0,
"asset": "",
"action": "prompt",
"message": "1 action succeeded",
"summary": {},
"action_run_id": 57
}
]
Fri Apr 29 2016 19:38:50 GMT-0700 (PDT): successfully called action callback 'prompt_cb()' in rule: manual_action(id:182)
Fri Apr 29 2016 19:38:50 GMT-0700 (PDT): calling on_finish()
Fri Apr 29 2016 19:38:50 GMT-0700 (PDT):
Playbook 'manual_action (id: 182)' executed (playbook_run_id: 195) on incident 'test'(id: 215).
Playbook execution status is:'success'
No actions were executed for this playbook and 'incident'
Fri Apr 29 2016 19:38:50 GMT-0700 (PDT): *** Playbook execution has completed with status: SUCCESS ***
Fri Apr 29 2016 19:38:51 GMT-0700 (PDT): Playbook execution report:
{"message":"","playbook_run_id":195,"result":[{"action":"prompt","app_runs":null,"close_time":"2016-04-30T02:38:50.844839+00:00","create_time":"2016-04-30T02:38:25.731+00:00","id":156,"message":"yes I am OK.. ","name":"automated action 'prompt' of 'manual_action' playbook","status":"success","type":"manual"}],"status":"success"}
Option Parameter Examples
Type: list
Shows the items in choices as the availabe responses.
{ 'type': 'list', 'choices': ['High', 'Medium', 'Low'] }
Type: range
Shows an input that requires a response within the given range of integers, i.e. 1-10.
{ 'type': 'range', 'min': 1, 'max': 100 }
Type: message
Shows a text area input in which a free form response can be entered.
{ 'type': 'message'}
prompt2
The prompt2 API is similar to the prompt API, but with prompt2 you can create a prompt with multiple user input fields. In prompt2, the options parameter is replaced by the response_types parameter. The other parameters are the same as in the prompt API. See prompt.
The prompt2 API is not supported from within a custom function.
phantom.prompt2(user=None,
role="Automation",
message='',
respond_in_mins=30,
response_types=None,
callback=None,
name=None,
parameters=None,
container=None,
scope='new',
trace=False,
separator=None,
drop_none=False)
| Parameter | Required? | Description |
|---|---|---|
| container | Required | The container object associated with the current playbook execution. This object is available to all action callbacks and other playbook execution functions. |
| user | Required | The recipient in the form of a user email address or username. Must be a valid user in . |
| role | Required | The recipient in the form of a role. For automated playbooks, set the role to Automation. Must be a valid role in . |
| message | Required | The message content to send. |
| response_types | Required | The list of JSON dictionaries describing each input field in the prompt. |
| respond_in_mins | Optional | The time the user is given to respond. The default is 30 minutes. If the user does not respond in the specified time, the prompt fails and a failed status is sent to the callback. |
| callback | Optional | This parameter has the same prototype as action callbacks. Status indicates success when the user has responded to the action and is failure only when the user does not respond in the specified time. The results JSON has the same format as any action results. Handle is not used and is an empty object. |
| name | Optional | The name of the prompt. |
| parameters | Optional | A list of datapaths whose values are used to format the message. Recognized datapaths are used to retrieve data, and the data is used to populate the curly brackets in the message. The first parameter replaces {0}, the second replaces {1}, and so on. |
| scope | Optional | Can either be new or all. Default value is new. See collect.
|
| trace | Optional | Trace is a flag related to the level of logging. If trace is on (True), more logging is enabled. When set to True, more detailed output is displayed in debug output. |
| separator | Optional | Specify an alternate separator using this parameter. If a datapath response contains a list, the default output separator is ', '.
|
| drop_none | Optional | By default, the None values are included in the resulting string.
|
Example of prompt2 response types
Response types are a list of JSON objects. Each object represents one input field in the prompt to be created. Each object needs to have a value with the key prompt, and has an optional key, options. prompt is a message for that input field. options is a dictionary with the same structure as a dictionary for the options in the prompt API. No options being specified results in a normal message type prompt.
[
{'prompt': 'Select a number in this range', 'options': {'type': 'range', 'min': 1, 'max': 50}},
{'prompt': 'Describe the event'}
]
The message parameter is still used in the prompt2 API. The message is displayed at the top of the created prompt, before the input fields.
render_template
The render_template API accepts a Django 1.11 template and fills the variable fields with contextual information from a provided dictionary. The template must follow the template language format and it can render any of the text-based formats such as HTML, XML, CSV, and so on. Common uses of the template are for user prompts or case management updates. Additional information about Django 1.11 templates can be found by searching on the Django Project home page.
The render_template API is supported from within a custom function.
phantom.render_template(template, context)
| Parameter | Required? | Description |
|---|---|---|
| template | Required | The Django 1.11 template. |
| context | Required | Dictionary of values used to populate variable fields in the Django template. |
This sample demonstrates the addition and population of a template in a playbook.
phantom.render_template(
"<html>
<head>
<title>Report for {{ report_name }}</title>
</head>
<body>Hi {{ subject }}, here are a list of IPs you should look at! <ul>{% for ip in ip_list %}
<li>{{ ip }}</li>
{% endfor %} </ul>
</body>
</html>",
{
'report_name': 'Task for {}: {}'.format(container['id'],
container['name']),
'subject': container['owner_name'],
'ip_list': ips_affected
}
)
task
The task API is a specialization of a manual action to ask a user or a role to perform work in the course of a response workflow or playbook.
The task API is not supported from within a custom function.
phantom.task(user=None, message=None, respond_in_mins=0, callback=None, name=None)
| Parameter | Required? | Description |
|---|---|---|
| user | Required | The person or a role to whom the task is assigned. |
| message | Required | The text that has the information or details of the task. |
| respond_in_mins | Required | The time given to the user to perform the task, after which the task fails and the status is expressed in the callback if it was specified. |
| callback | Optional | A callback function to be called when the task completes. |
| name | Required | A unique name to distinguish this action from other actions |
| Convert playbooks or custom functions from Python 2 to Python 3 | Container automation API |
This documentation applies to the following versions of Splunk® SOAR (On-premises): 6.2.2, 6.3.0, 6.3.1
Download manual
Feedback submitted, thanks!