Splunk® SOAR (Cloud)

Python Playbook API Reference for Splunk SOAR (Cloud)

Acrobat logo Download manual as PDF


Acrobat logo Download topic as PDF

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 A string 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 handle object is limited to 4k. For objects bigger than 4k, use the save_data() and get_data() APIs instead.

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 action field replaces the action_name field.

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 file_reputation_cb is called once when both the queries complete. The parameters in the example is a list of dictionaries and contains one IP. But, in one action call the user can specify many different IP addresses.

The results JSON object provides full visibility into the execution of the action on all matching assets using all matching apps for all specified parameters.

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 A string 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

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,
                  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 ], ..]

OPERATOR can be any one of the following characters:

==, !=, <, >, <=, >=, 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.
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.
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.

Generally, CEF fields that are passed into the condition API, if they have commas in the value, cause the value to be treated as a list. If the CEF field is toEmail or fromEmail, then commas do not trigger the list behavior. This occurs because commas frequently appear in the display name portion of an email address.

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 parameter from a custom function callback:

{
    '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 A string 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.

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,
                  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 ], ..]

OPERATOR can be any one of the following characters:

==, !=, <, >, <=, >=, 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.
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.

Generally, CEF fields that are passed into the decision API, if they have commas in the value, cause the value to be treated as a list. One specific exception is if the CEF field is toEmail or fromEmail where commas do not trigger the list behavior because commas frequently appear in the display name portion of an email address.

When passing Boolean values to a decision block in the Visual Playbook Editor, true and false with lowercase letters, are interpreted as strings. while True and False with capital letters, are interpreted as Boolean values. These capitalization conventions match Python, and True and False are built in, but true and false are not.

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,
        )

discontinue

The discontinueAPI allows the users to stop executing active playbooks when a container is being processed against Active playbooks.

The discontinue API is not supported from within a custom function.

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

', '. You can specify an alternate separator using this parameter.

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

html_file_to_pdf

Use the html_file_to_pdf API call to convert an HTML file to a PDF file. This function returns true if the conversion is successful. If an error occurs, the function returns false.

phantom.html_file_to_pdf(<path_to_input_html_file>, <path_to_output_pdf_file>)
Parameter Required Type Description
Source File Path Required Path The location of the HTML file on disc you want to convert to a PDF and save.
Output File Path Required Path The location where the PDF file is saved.

This example shows how you can convert an HTML file into a PDF file, then use the vault_add API to add the PDF file to a container.

# NOTE: The input file path must end in ".html"!
input_html_path = "path/to/file.html"
output_file_path = "/tmp/output.pdf"

# html_file_to_pdf returns False on error, and True on success.
if not phantom.html_file_to_pdf(input_html_path, output_file_path):
    phantom.error("Failed to convert file to a pdf!")

# Add the file as a container attachment using the vault API
phantom.vault_add(container=container, file_location=output_file_path, file_name="output.pdf")

html_string_to_pdf

Use the html_string_to_pdf API call to convert an HTML string to a PDF file. This function returns true if the conversion is successful. If an error occurs, the function returns false.

phantom.html_string_to_pdf(<path_to_input_html_string>, <path_to_output_pdf_file>)
Parameter Required Type Description
Source HTML String Required String The HTML string you want to convert to a PDF and save.
Output File Path Required Path The location where the PDF file is saved.

This example shows how you can convert an HTML string into a PDF file, then use the vault_add API to add the PDF file to a container.

input_html = "<!DOCTYPE html><html lang=\"en\"><body><b>Hello World!</b></body></html>"
# Tip: Use NamedTemporaryFile to prevent concurrent
#      playbooks from writing to the same output file.
output_file_path = "/tmp/output.pdf"

# html_string_to_pdf returns False on error, and True on success.
if not phantom.html_string_to_pdf(input_html, output_file_path):
    phantom.error("Failed to convert file to html!")

# Add the file as a container attachment using the vault API
phantom.vault_add(container=container, file_location=output_file_path, file_name="output.pdf")

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. Be sure both of the following are true:
  • The child playbook that is called must be an input playbook, not an Automation playbook.
  • The inputs you provide must be valid for the child playbook.
name Required An optional parameter unless the callback parameter is specified. This parameter identifies the execution instance of the called playbook. If the code for calling the child playbook is auto-generated, the name of the function is the recommended value for this parameter.

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,
                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, username, or a role. Must be a valid user or 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. 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
Last modified on 30 June, 2022
PREVIOUS
Convert playbooks or custom functions from Python 2 to Python 3
  NEXT
Container automation API

This documentation applies to the following versions of Splunk® SOAR (Cloud): current


Was this documentation topic helpful?


You must be logged into splunk.com in order to post comments. Log in now.

Please try to keep this discussion focused on the content covered in this documentation topic. If you have a more general question about Splunk functionality or are experiencing a difficulty with Splunk, consider posting a question to Splunkbase Answers.

0 out of 1000 Characters