Splunk® SOAR (On-premises)

Python Playbook API Reference for Splunk SOAR (On-premises)

The classic playbook editor will be deprecated in early 2025. Convert your classic playbooks to modern mode.
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:

Understanding loop state

Loop state is used in playbook logic loops, to retry actions, custom functions, and child playbooks within a playbook. Looping is helpful for the following scenarios:

  • Repeating the same action until it succeeds. For example, checking that a time-intensive process like a virus detonation has completed successfully.
  • Attempting an action repeatedly, waiting between tries, until you achieve a specific result. For example, checking that a work ticket like Jira has been updated.

For details on using logic loops in the Visual Playbook Editor, refer to Repeat actions with logic loops.

The loop state is exposed to the platform as a Python object. It contains all necessary fields and methods to execute playbook blocks (action, playbook, and custom function blocks) repetitively until an exit condition is reached. The daemons treat this object as a JSON string and methods exist to convert them back to a Python object.

Example

Here is a sample action playbook block with a loop state object. This example shows a geolocate_ip action playbook block that runs in a loop, up to 2 times, before calling a filter playbook block.

  • The geolocate_ip_1 method initializes the loop and runs the geolocate_ip action.
  • The loop_geolocate_ip_1 method is used to determine if the loop needs to continue running geolocate_ip or go into the filter block.

Fields are described in the following section.

@phantom.playbook_block()
def loop_geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
    phantom.debug("loop_geolocate_ip_1() called")
    loop_state = phantom.LoopState(state=loop_state_json)
    if loop_state.should_continue(container=container, results=results): # should_continue evaluates iteration/timeout/conditions
        loop_state.increment() # increments iteration count
        geolocate_ip_1(container=container, loop_state_json=loop_state.to_json())
    else:
        filter_1(container=container)
    return
@phantom.playbook_block()
def geolocate_ip_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None, custom_function=None, loop_state_json=None, **kwargs):
    phantom.debug("geolocate_ip_1() called")
    # phantom.debug('Action: {0} {1}'.format(action['name'], ('SUCCEEDED' if success else 'FAILED')))
    parameters = []
    parameters.append({
        "ip": "1.2.3.4",
    })
    if not loop_state_json:
        # Loop state is empty. We are creating a new one from the inputs
        loop_state_json = {
            # Looping configs
            "current_iteration": 1,
            "max_iterations": 2,
            "conditions": None,
            "max_ttl": 600,
            "delay_time": 1,
        }
    # Load state from the JSON passed to it
    loop_state = phantom.LoopState(state=loop_state_json)
    ################################################################################
    ## Custom Code Start
    ################################################################################
    # Write your custom code here...
    ################################################################################
    ## Custom Code End
    ################################################################################
    phantom.act("geolocate ip", parameters=parameters, name="geolocate_ip_1", assets=["maxmind"], callback=loop_geolocate_ip_1, loop_state=loop_state.to_json())
    return

Fields

Fields are described in the following table:

Field Default value Description
current_iteration 0 Current iteration running in the loop. Incremented when an iteration is about to run.
max_iterations 3 Maximum number of iterations allowed to run.
exit_reason "" Reason for exiting the loop. Available as a data path in downstream blocks.
conditions empty list Exit conditions for the loop. The loop exits when the condition resolves to true.
logical_operator or Operator to join multiple conditions. Possible values are or and and.
start_time current time Time when the loop started. Used with max_ttl, to determine if the loop has reached the time-out value.
last_iteration_time current time Time when the last iteration started.
max_ttl 600 Time, in seconds, to wait for the loop to complete before stopping due to timeout. Used with start_time, to determine if the loop has reached the time-out value.
delay_time 3 Time, in seconds, to pause between iterations of a loop.

Methods

The loop state object has the following public methods.

LoopState

The constructor method that returns a new Loop state object.

Parameter Required? Description
state no Accepts either a python dictionary or a JSON string. It is used to "reload" the object from a string/dict representation

Example

loop_state_json = {
            # Looping configs
            "current_iteration": 1,
            "max_iterations": 2,
            "conditions": None,
            "max_ttl": 600,
            "delay_time": 1,
        }
    # Load state from the JSON passed to it
    loop_state = phantom.LoopState(state=loop_state_json)

to_json

Used to convert the loop state object to a JSON string and to transfer loop state between daemons.

This method does not require any arguments.

Example

# Example to save state to JSON string and load it as an object again 
json_str = old_loop_state_obj.to_json()
# Some custom code
LoopState(state=json_str)


increment

Used by the loop_* methods to increment the current_iteration field and updates the last_iteration_time field with the current time.

This method does not require any arguments.

Example

loop_state = phantom.LoopState(state=loop_state_json)
...
    loop_state.increment() # increments iteration count
...

should_continue

Used to determine whether a loop needs to continue.

  • Returns true when exit conditions have not been met and the loop must continue
  • Returns false when exit conditions are met and the loop must exit
Parameter Required? Description
container yes The container JSON object to pass to run the playbook on. This is the same container JSON object that you get in on_start() or any other callback function.
results yes This JSON object contains all of the details and results of the previous iteration.
The should_continue method uses this information to evaluate exit conditions that depend on the results of previous iteration to match given criteria. For example, if the loop will exit when an API response is successful, the should_continue method evaluates the results object that contains the API response of the previous iteration.

Example

loop_state = phantom.LoopState(state=loop_state_json)
# Run geolocate if loop needs to continue. Else go the filter block
if loop_state.should_continue(container=container, results=results): # should_continue evaluates iteration/timeout/conditions
    loop_state.increment() # increments iteration count
    geolocate_ip_1(container=container, loop_state_json=loop_state.to_json())
else:
    filter_1(container=container)
Last modified on 09 May, 2024
Understanding callbacks   Convert playbooks or custom functions from Python 2 to Python 3

This documentation applies to the following versions of Splunk® SOAR (On-premises): 6.2.0, 6.2.1, 6.2.2, 6.3.0


Was this topic useful?







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