Common API calls used by the Visual Playbook Editor

The Visual Playbook Editor uses the phantom.act API calls to perform actions in a playbook. The phantom.collect API calls are used to collect input for actions.

Run actions using the phantom.act API calls

phantom.act is the call for running an action. The following code shows an example of a simple act call. In this act call, the parameters are static as shown by the 1.1.1.1 ip. This means that regardless of which container you run against, the playbook will work on the same input data. In the following act call, there is no callback.

phantom.act(action="geolocate ip", parameters=parameters, assets=['maxmind'], name="geolocate_ip_1")

It is important to note that in the call to phantom.act, the callback parameter isn't set, and you can see that in the logs for the playbook run and the log output.

View the simplest phantom.act call.

"""
"""

import phantom.rules as phantom
import json
from datetime import datetime, timedelta
def on_start(container):
    phantom.debug('on_start() called')
    
    # call 'geolocate_ip_1' block
    geolocate_ip_1(container=container)

    return

def geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None):
    phantom.debug('geolocate_ip_1() called')

    # collect data for 'geolocate_ip_1' call

    parameters = []
    
    # build parameters list for 'geolocate_ip_1' call
    parameters.append({
        'ip': "1.1.1.1",
    })

    phantom.act(action="geolocate ip", parameters=parameters, assets=['maxmind'], name="geolocate_ip_1")

    return

def on_finish(container, summary):
    phantom.debug('on_finish() called')
    # This function is called after all actions are completed.
    # summary of all the action and/or all details of actions
    # can be collected here.

    # summary_json = phantom.get_summary()
    # if 'result' in summary_json:
        # for action_result in summary_json['result']:
            # if 'action_run_id' in action_result:
                # action_results = phantom.get_action_results(action_run_id=action_result['action_run_id'], result_data=False, flatten=False)
                # phantom.debug(action_results)

    return

Using the callback parameter

Use the callback function to chain playbook blocks together that rely on data from the previous blocks. If you'd like a later playbook block to rely on the output of your act call, use a callback function. In the following example, the filter block relies on the output of the geolocate ip command.

View a phantom.act call that uses a callback.

"""
"""

import phantom.rules as phantom
import json
from datetime import datetime, timedelta
def on_start(container):
    phantom.debug('on_start() called')
    
    # call 'geolocate_ip_1' block
    geolocate_ip_1(container=container)

    return

def geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None):
    phantom.debug('geolocate_ip_1() called')

    # collect data for 'geolocate_ip_1' call

    parameters = []
    
    # build parameters list for 'geolocate_ip_1' call
    parameters.append({
        'ip': "1.1.1.1",
    })

    phantom.act(action="geolocate ip", parameters=parameters, assets=['maxmind'], callback=filter_1, name="geolocate_ip_1")

    return

def filter_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None):
    phantom.debug('filter_1() called')

    # collect filtered artifact ids for 'if' condition 1
    matched_artifacts_1, matched_results_1 = phantom.condition(
        container=container,
        action_results=results,
        conditions=[
            ["geolocate_ip_1:action_result.data.*.country_iso_code", "==", "AU"],
        ],
        name="filter_1:condition_1")

    # call connected blocks if filtered artifacts or results
    if matched_artifacts_1 or matched_results_1:
        lookup_ip_1(action=action, success=success, container=container, results=results, handle=handle, custom_function=custom_function, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)

    return

def lookup_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None):
    phantom.debug('lookup_ip_1() called')
    
    #phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
    
    # collect data for 'lookup_ip_1' call
    filtered_results_data_1 = phantom.collect2(container=container, datapath=["filtered-data:filter_1:condition_1:geolocate_ip_1:action_result.parameter.ip", "filtered-data:filter_1:condition_1:geolocate_ip_1:action_result.parameter.context.artifact_id"])

    parameters = []
    
    # build parameters list for 'lookup_ip_1' call
    for filtered_results_item_1 in filtered_results_data_1:
        if filtered_results_item_1[0]:
            parameters.append({
                'ip': filtered_results_item_1[0],
                # context (artifact id) is added to associate results with the artifact
                'context': {'artifact_id': filtered_results_item_1[1]},
            })

    phantom.act(action="lookup ip", parameters=parameters, assets=['google_dns'], name="lookup_ip_1")

    return

def on_finish(container, summary):
    phantom.debug('on_finish() called')
    # This function is called after all actions are completed.
    # summary of all the action and/or all details of actions
    # can be collected here.

    # summary_json = phantom.get_summary()
    # if 'result' in summary_json:
        # for action_result in summary_json['result']:
            # if 'action_run_id' in action_result:
                # action_results = phantom.get_action_results(action_run_id=action_result['action_run_id'], result_data=False, flatten=False)
                # phantom.debug(action_results)

    return

To learn more about the callback function, see callback in the Python Playbook API Reference for .

Collect input using phantom.collect API calls

Collect APIs are used to gather data from artifacts, containers, and action results. The data is then used in filtering, or used as input to act calls. The input to this phantom.act call is determined by the container data. In the following example, the geolocate ip action is using the input container artifacts for its data input.

Phantom.act calls also often use collect2 output. With collect2, the generated code loops through the list of results from collect2, and then appends a dictionary to the list parameters, with each row containing the ip and context parameters. This parameter list is passed to the phantom.act call.

def geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None):
    phantom.debug('geolocate_ip_1() called')

    # collect data for 'geolocate_ip_1' call
    container_data = phantom.collect2(container=container, datapath=['artifact:*.cef.sourceAddress', 'artifact:*.id'])

    parameters = []
    
    # build parameters list for 'geolocate_ip_1' call
    for container_item in container_data:
        if container_item[0]:
            parameters.append({
                'ip': container_item[0],
                # context (artifact id) is added to associate results with the artifact
                'context': {'artifact_id': container_item[1]},
            })

    phantom.act("geolocate ip", parameters=parameters, assets=['maxmind'], name="geolocate_ip_1")

    return

The purpose of the collect call is to extract the requested set of data from the container. In this example, the collect call requested the sourceAddress CEF artifacts. To get that set of artifacts, we use a datapath with the collect call.

Using phantom.act calls with collect2 output

The collect2 call can pre-filter results. The Visual Playbook Editor (VPE) generates code of this type when you use a decision block. This is shown in this filter block called by on_start:

def filter_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None):
    phantom.debug('filter_1() called')

    # collect filtered artifact ids for 'if' condition 1
    matched_artifacts_1, matched_results_1 = phantom.condition(
        container=container,
        conditions=[
            ["artifact:*.cef.sourceAddress", "<", "10.0.0.0"],
            ["artifact:*.cef.sourceAddress", ">", "10.255.255.255"],
        ],
        logical_operator='or',
        name="filter_1:condition_1")

    # call connected blocks if filtered artifacts or results
    if matched_artifacts_1 or matched_results_1:
        geolocate_ip_1(action=action, success=success, container=container, results=results, handle=handle, filtered_artifacts=matched_artifacts_1, filtered_results=matched_results_1)

    return

This is passed to the geolocate block as a parameter, and then geolocate block uses it in the phantom.collect2 call as shown here:

def geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None):
    phantom.debug('geolocate_ip_1() called')

    # collect data for 'geolocate_ip_1' call
    container_data = phantom.collect2(container=container, datapath=['artifact:*.cef.sourceAddress', 'artifact:*.id'])

    parameters = []
    
    # build parameters list for 'geolocate_ip_1' call
    for container_item in container_data:
        if container_item[0]:
            parameters.append({
                'ip': container_item[0],
                # context (artifact id) is added to associate results with the artifact
                'context': {'artifact_id': container_item[1]},
            })

    phantom.act("geolocate ip", parameters=parameters, assets=['maxmind'], name="geolocate_ip_1")

    return

The datapath for the geolocate ip command starts with "filtered-artifacts:", so you are using the results of the filtered set, instead of all artifacts.

For more information on the phantom.act and phantom.collect API calls, see API Reference.

phantom.debug call

If you aren't getting the results you expected with your API calls, add a phantom.debug call below the collect2 call, as shown in the following example:

container_data = phantom.collect2(container=container, datapath=['artifact:*.cef.sourceAddress', 'artifact:*.id'])
phantom.debug(container_data)

Once the collect2 call is debugged, the additional output in the debug window shows the sourceAddress artifacts and their artifact IDs.

Wed Mar 21 2018 20:32:37 GMT-0700 (PDT): 
[
    [
        "10.10.10.10",
        2
    ],
    [
        "175.45.176.1",
        3
    ]
]

phantom.debug prints the Python representation of the object that you pass to it.

container_data = phantom.collect2(container=container, datapath=['artifact:*.cef', 'artifact:*.id'])
phantom.debug(container_data)

The debug output is this:

Wed Mar 21 2018 20:35:43 GMT-0700 (PDT): 
[
    [
        {
            "destinationAddress": "192.168.10.10",
            "sourceAddress": "10.10.10.10"
        },
        2
    ],
    [
        {
            "destinationAddress": "192.168.10.11",
            "sourceAddress": "175.45.176.1"
        },
        3
    ]
]