phantom-action-handler 0.0.1a9

Utilities to simplify Phantom app development

phantom-action-handler

Utilities for simplifying the development of Phantom apps

Author

David Finn: [e-mail]

Requirements

• Splunk>Phantom
• Python 3.6 or higher

Installation

This library should generally be specified as a Phantom app pip dependency, and therefore not require manual installation.

If manual installation on the Phantom instance is desirable:

phenv pip3 install phantom-action-handler

Description

The phantom_dev.action_handler module greatly simplifies the implementation of a phantom.base_connector.BaseConnector subclass, which is the basis for Phantom app development.

Edge cases and

Quickstart

Given the following app JSON describing a hypothetical echo message action:

{
	...
	"pip_dependencies": {
        "pypi": [
			...
			{"module": "phantom-action-handler"},
			...
		],
		...
    },
	...
	"actions": [
		{
			"action": "echo message",
			"identifier": "echo_message",
			"description": "Return the input message",
			"verbose": "",
			"type": "test",
			"read_only": true,
			"parameters": {
				"message": {
					"description": "The message to be echoed",
					"data_type": "string",
					"contains": ["text"],
					"required": true,
					"primary": true
				},
			},
			"output": [
				{
					"data_path": "action_result.data.*",
					"data_type": "string",
					"contains": ["text"]
				},
			],
			"versions": "EQ(*)"
		},
	],
	...
}

The corresponding phantom.base_connector.BaseConnector could be implemented as follows:

from phantom.base_connector import BaseConnector
from phantom_dev.action_handler import (
	ActionHandler, HandlerMixin, main_connector)


@main_connector
class MyConnector(HandlerMixin, BaseConnector):
	@ActionHandler
	def echo_message(self, message, context=None):
		yield message

All methods of the BaseConnector class are available for use in the action handler logic implementation.

Details

In the above example, use of the ActionHandler decorator wraps the decorated echo_message method in the logic required for error handling and results reporting. The param dictionary is automatically unpacked as keyword arguments to handler method, allowing for quick and explicit argument validation and intuitive access to action parameters. param contains a context entry (except for the test connectivity action) and the parameters described in the app JSON.

Handler methods such as echo_message are expected to return iterables of results data. The items from this iterable are added as data objects to the ActionResult. Implementing handler methods as generators is highly convenient, as this allows custom logic to be run any time before or after data is yielded, but methods can also be implemented as normal functions that return iterable objects.

The HandlerMixin superclass automatically delegates incoming actions to the correct method based on the action identifier.

The main_connector class decorator simply calls the class's main method if the class is defined in the __main__ module, reproducing the testing functionality provided by autogenerated app wizard code.

Summaries

To add summary information to the result, the ActionHandler.summary decorator can be used:

...
@main_connector
class MyConnector(HandlerMixin, BaseConnector):
	@ActionHandler
	def echo_message(self, message, context=None):
		yield message

	@echo_message.summary
	def summarise_echo_message(self, results):
		message, = results
		return {'message': message}

This will insert the result of the summary method as the action result summary object.

Signaling Failure

Failure is signaled through raising exceptions. If the handler executes without raising an exception, the action is treated as a success.

To implement an echo fail action that does the same thing as echo message, but always fails after producing results (assuming the correct entries are added to the app JSON):

...
@main_connector
class MyConnector(HandlerMixin, BaseConnector):
	@ActionHandler
	def echo_message(self, message, context=None):
		yield message

	@ActionHandler
	def echo_fail(self, **param):
		# Demonstration of re-packing param; this will be the same as the
		# original param dictionary, which we can then unpack for the call
		# to echo_message
		yield from self.echo_message(**param)
		raise RuntimeError('Failed on purpose')

	# The same summary method can be decorated multiple times for different
	# handlers to duplicate functionality
	@echo_fail.summary
	@echo_message.summary
	def summarise_echo_message(self, results):
		message, = results
		return {'message': message}

Actions with no results

		...
		{
			"action": "test connectivity",
			"identifier": "test_connectivity",
			"description": "Validate the asset configuration for connectivity using supplied configuration",
			"verbose": "",
			"type": "test",
			"read_only": true,
			"parameters": {},
			"output": [],
			"versions": "EQ(*)"
		},
		...

test connectivity is an example of an action which produces no results. The handler method needs only to return an empty iterable, which is easily accomplished by returning an empty collection rather than implementing a generator:

...
@main_connector
class MyConnector(HandlerMixin, BaseConnector):
	@ActionHandler
	def echo_message(self, message, context=None):
		yield message

	@ActionHandler
	def echo_fail(self, **param):
		# Demonstration of re-packing param; this will be the same as the
		# original param dictionary, which we can then unpack for the call
		# to echo_message
		yield from self.echo_message(**param)
		raise RuntimeError('Failed on purpose')

	# The same summary method can be decorated multiple times for different
	# handlers to duplicate functionality
	@echo_fail.summary
	@echo_message.summary
	def summarise_echo_message(self, results):
		message, = results
		return {'message': message}

	@ActionHandler
	def test_connectivity(self):
		# The test connectivity action is a special case that does not
		# receive a param dictionary at all, so there are no arguments to
		# unpack (not even context)
		test_value = 'SOME TEST MESSAGE'
		results = []
		try:
			for result in self.echo_fail(test_value):
				results.append(result)
		except RuntimeError:
			pass
		else:
			raise RuntimeError('echo fail failed to fail')

		message, = results
		if message != test_value:
			raise ValueError('echo fail failed to echo')

		return []

It would also be possible to achieve this with a return statement before a yield statement in a generator, or by failing before any results are yielded.