phantom-dev 0.1.2

Utilities to simplify Phantom app development

phantom-dev

Utilities for simplifying the development of Phantom apps

Author

David Finn: [e-mail]

Requirements

• Splunk>Phantom
• Python 3.6 or higher

Installation

pip install phantom-dev

Description

phantom-dev is a command-line utility for creating, building, and deploying Phantom apps.

App packages are built from project directories containing a YAML metadata file and a connector implemented in Python. Any other files in the project directory will be packaged and included with the app.

App Metadata

The metadata.yaml file in the project directory will be used to generate the app JSON expected by the Phantom platform. Minor alterations to the JSON structure have been made to assist with readability and maintainability; related keys have been grouped under a common prefix key (e.g. JSON project-* keys are defined under the project object in the YAML) and lists of uniquely-identifiable objects have been converted to objects (e.g. JSON actions is now a mapping of action names to action data, rather than a list of action objects with unique and potentially conflicting names).

For information on the generated app metadata, see the official Phantom documentation.

Connector Implementation

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.

Action handler methods defined using the phantom_dev.action_handler.ActionHandler decorator will be used to dynamically infer action metadata unless overridden in the metadata file. Action names, parameter names, and parameter types can all be inferred from the implementation assuming parameters are type-annotated in the code.

The phantom_dev.action_handler module will be automatically embedded in the dependencies directory when the app is built, allowing the developer to make full use of the module without being concerned with managing it as a dependency.

Edge cases and use of more specialised BaseConnector methods should be dealt with as normal, in accordance with the official Phantom documentation.

Quickstart

Running phantom-dev create will prompt the user for the set of details required to define a Phantom app.

$ phantom-dev create "My Special App"
Product Vendor: Special Vendor
Product Name: Special Product
Description: My special little app
Publisher: David Finn
License: Special license

A new project directory will be created and populated with a metadata.yaml and a connector.py. The metadata YAML will contain the details provided by the user, and the connector module will define an example connector implementation. The metadata will also be populated with action information required by the example connector implementation.

Extending the App

Given the following metadata snipped describing a new echo message action:

...
actions:
	...
	echo_message:
		parameters:
			message:
				description: The message to be echoed
	...
...

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

from phantom_dev.action_handler import ActionHandler, smart_connector
...
@smart_connector
class Connector:
	...
	@ActionHandler
	def echo_message(self, message: str, context=None):
		yield message
	...
...

Thanks to smart_connector, all methods of the BaseConnector class are available for use in the action handler logic implementation.

Although the app will successfully install and run given the above metadata, other fields including output should be defined in metadata.yaml before production use.

Deploying the App

Once the app is ready to install, assuming a Phantom server location of phantom.example.com:

$ phantom-dev push my_special_app/ [e-mail]

Note: The Phantom server must be a known host; SSH to it first to confirm credentials and connectivity.

The user will be prompted for the SSH password. Once supplied, the app will be automatically packaged, sent to the Phantom server, and installed.

The SSH password can also be provided as part of the command:

$ phantom-dev push my_special_app/ root:[e-mail]

If certificate authentication is used, an empty password can be also be specified:

$ phantom-dev push my_special_app/ root:@phantom.example.com

Other commands

For information on the other phantom-dev subcommands including package and deploy, run:

$ phantom-dev --help
$ phantom-dev <subcommand> --help

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 provided by smart_connector automatically delegates incoming actions to the correct method based on the action identifier.

smart_connector also wraps the functionality of a the main_connector decorator. main_connector 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:

...
@smart_connector
class Connector:
	@ActionHandler
	def echo_message(self, message: str, 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:

...
@smart_connector
class Connector:
	@ActionHandler
	def echo_message(self, message: str, context=None):
		yield message

	@echo_message.summary
	def summarise_echo_message(self, results):
		message, = results
		return {'message': 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.
		# Unfortunately, this will require manual specification of more
		# parameter metadata.
		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}

In the example, parameter packing with **param was used instead of describing and annotating the paramters for echo fail. This is possible but not recommended, because now the user must manually specify more parameter information in metadata.yaml:

...
actions:
	...
	echo_message:
		parameters:
			message:
				description: The message to be echoed
	echo_fail:
		parameters:
			message:
				data_type: string
				description: The message to be echoed
				required: true
	...
...

Actions with no results

...
actions:
	...
	test_connectivity:
		read_only: true
	...
...

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:

...
@smart_connector
class Connector:
	@ActionHandler
	def echo_message(self, message: str, context=None):
		yield message

	@echo_message.summary
	def summarise_echo_message(self, results):
		message, = results
		return {'message': 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.
		# Unfortunately, this will require manual specification of more
		# parameter metadata.
		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.

Icons made by Freepik from www.flaticon.com