Skip to main content

Utilities to simplify Phantom app development

Project description

phantom-dev

Utilities for simplifying the development of Phantom apps

Author

David Finn: dfinn@splunk.com

Requirements

  • Splunk>Phantom
  • Python 3.8 or higher

Features

  • Commands for rapid creation, packaging, and deployment of Phantom apps
  • Streamlined Python interface for action definition
  • Automatic bundling of dependencies using standard pip requirements files
  • Pytest integration with automatic mocking and sys path management for local unit testing
  • Debugging remote execution against local source files over a secure SSH tunnel
  • Asynchronous action handler support
  • Dedicated log files for connector logging output

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

The following defines the implementation of a new action called "echo message":

from phantom_dev.action_handler import ActionHandler, smart_connector
...
@smart_connector
class Connector:
	...
	@ActionHandler
	def echo_message(self, message: str):
		"""
		Echo a message

		:param message: The message to be echoed
		"""
		yield {'response': message}
	...

	@echo_message.summary
	def summarise_echo(self, results):
		result, = results
		return {'response': result['response']}
...

If possible, the action description will be inferred from the method docstring- in this case, "Echo a message". Similarly, parameter descriptions will also be inferred if specified in the docstring with appropriate docstring syntax (currently, Sphinx syntax is confirmed to work). In this case, the description for the paramter message will be "The message to be echoed".

The action summary data for the echo message action is created by decorating the summary method with @echo_message.summary. Action summary methods take the action results as an iterable, and create the appropriate summary data.

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, it's still missing output fields and contains information. Unless specfied in metadata.yaml, phantom-dev (and therefore Phantom) has no way of knowing that the output data should contain a response field with a data_type of string. If we want to add contains information, such as setting a contains value of ['text'] for the message and response, we also need to specify these somewhere.

These can be specified as normal using metadata.yaml, but this can also be done directly using appropriate ActionHandler parameters and methods:

from phantom_dev.action_handler import ActionHandler, contains, smart_connector
...

@contains('text')
class Text(str):
	"""A string-based type which contains `['text']`"""

...

@smart_connector
class Connector:
	...
	@ActionHandler(action_type='generic', data_contains={'response': Text})
	def echo_message(self, message: Text):
		"""
		Echo a message

		:param message: The message to be echoed
		"""
		yield {'response': message}

	@echo_message.summary_contains({'response': Text})
	def summarise_echo(self, results):
		result, = results
		return {'response': result['response']}
	...
...

Several things are happening here. First, a new type called Text inherits from str and is decorated with the @contains decorator, mapping it to a contains value of ['text']. This type is used instead of str to annotate the message parameter of the action, which allows phantom-dev to infer both the data_type and the contains value for the parameter.

Next, the data_contains parameter is being used to instantiate the method constructor, and a mapping of 'response' to Text is being specified as the argument. This allows phantom-dev to infer that each data result has a field named 'response', and that its data_type and contains information should be inferred from the Text type the same way it was for the message parameter. Note that we can also set the action type metadata with use of another decorator parameter, action_type.

Finally, the same thing is being done with the echo_message.summary_contains decorator (used instead of summary), though in this case it's the action summary fields rather than the results data fields that are being described.

Use of these decorators is preferred to metadata.yaml specification, as they keep metadata definitions colocated with the data they describe. They also encourage code reuse though the use of common contains-mapped type definitions, shareable between both parameters and output fields.

Dependencies

Any package specified in requirements-whl.txt in the app project directory will be automatically downloaded as a wheel and packaged with the app. requirements-whl.txt should be a normal pip requirements file.

For example, if the app requires the roboversion package, a requirements-whl.txt could be created with the following content:

roboversion>=2

When the app is built, the roboversion wheel will be automatically downloaded and included in the package wheels directory, and the autogenerated app JSON will specify its location for Phantom installation.

If there isn't a compatible wheel for the Phantom platform, requirements-sdist.txt can be used instead of requirements-whl.txt. This will download the package as a source distribution instead of as a wheel. Unlike requirements-whl.txt, subdependencies will not be automatically installed for packages in requirements-sdist.txt, as sdist installation is intended to be used as a backup for when a wheel is unavailable.

Finally, requirements-pypi.txt can be used to specify dependencies that won't be packaged with the app, but that will be downloaded and installed by the Phantom platform itself.

Testing

phantom-dev can automatically manage pytest execution, including adding the app directory to the Python path and mocking imports of the phantom module, using the phantom-dev test command:

$ phantom-dev test <app_directory> [<pytest arguments>...]

Apps created using phantom-dev create will be initialised with a tests directory containing a basic example of a pytest script.

If working from the app directory, pytest will automatically locate and run the tests when it is invoked:

$ phantom-dev test

pytest also accepts a test directory location:

$ phantom-dev test my/special/test/location

If working from a directory other than the app directory, the app directory can also be specified:

$ phantom-dev test --app-directory my_special_app

Positional arguments will be passed to pytest, allowing us to specify test locations as well:

$ phantom-dev test --app-directory my_special_app my/special/test/location

As phantom-dev test is a wrapper around pytest invocation, it supports all of pytest's features and behaviour. To avoid capturing optional parameters before passing them to pytest, use the pseudo-argument --:

$ phantom-dev test --app-directory my_special_app -- --show-locals my/special/test/location

Refer to the pytest documentation for details.

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/ root@phantom.example.com

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:PASSWORD@phantom.example.com

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

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

Remote Debugging

phantom-dev streamlines the debugging process by automating the creation of test action data and configuring a debugger connection through a secure SSH tunnel. To open a remote debugging session:

$ phantom-dev debug my_special_app/ root:@phantom.example.com "dummy action"

This command will:

  1. Generate a test action JSON file on the remote Phantom host
  2. Open an SSH tunnel to the remote Phantom host, forwarding the local port to the remote debugging port
  3. Wait for a debugger connection, then run the connector script with the correct Python path using the generated action JSON

A local debugging client (e.g. Visual Studio Code) will then be able to connect and step through action execution.

Visual Studio Code Debugging Configuration

Visual Studio Code debugging is configured through entries in launch.json.

To use the SSH tunnel provided by the phantom-dev debug command, create an entry configured to connect to localhost on the specified debugger port:

		...
		{
			"name": "Remote Debuggable Phantom App",
			"type": "python",
			"request": "attach",
			"connect": {
				"host": "localhost",
				"port": 8869
			},
			"pathMappings": [
				{
					"localRoot": "${workspaceFolder}/my_special_app/connector.py",
					"remoteRoot": "/opt/phantom/apps/myspecialapp_<SOME UUID>/connector.py"
				}
			]
		}
		...

The pathMappings entry should be configured to map the connector module to its location in the installed app on the remote Phantom host. The installed app folder will be under <PHANTOM_HOME>/apps.

Once configured, a debugging session can be run against the remote action execution with full debugger functionality.

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 (orActionHandler.data_contains) 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 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.

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:

...
@contains('text')
class Text(str):
	"""A string-based type which contains `['text']`"""

...

@smart_connector
class Connector:
	...
	@ActionHandler(action_type='generic', data_contains={'response': Text})
	def echo_message(self, message: Text):
		"""
		Echo a message

		:param message: The message to be echoed
		"""
		yield {'response': message}

	@echo_message.summary_contains({'response': Text})
	def summarise_echo(self, results):
		result, = results
		return {'response': result['response']}
	...

	@ActionHandler(action_type='generic', data_contains={'response': Text})
	def echo_fail(self, **param):
		"""
		Echo the message as normal, then fail
		"""
		# 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_contains({'response': Text})
	@echo_message.summary_contains({'response': Text})
	def summarise_echo(self, results):
		result, = results
		return {'response': result['response']}

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_fail:
		parameters:
			message:
				data_type: string
				description: The message to be echoed
				required: true
				contains:
					- text
	...
...

Actions with no results

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:

...
@contains('text')
class Text(str):
	"""A string-based type which contains `['text']`"""

...

@smart_connector
class Connector:
	...
	@ActionHandler(action_type='generic', data_contains={'response': Text})
	def echo_message(self, message: Text):
		"""
		Echo a message

		:param message: The message to be echoed
		"""
		yield {'response': message}

	@echo_message.summary_contains({'response': Text})
	def summarise_echo(self, results):
		result, = results
		return {'response': result['response']}
	...

	@ActionHandler(action_type='generic', data_contains={'response': Text})
	def echo_fail(self, **param):
		"""
		Echo the message as normal, then fail
		"""
		# 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_contains({'response': Text})
	@echo_message.summary_contains({'response': Text})
	def summarise_echo(self, results):
		result, = results
		return {'response': result['response']}

	@ActionHandler(action_type='generic')
	def test_connectivity(self):
		"""
		Check `echo message` and `echo fail`
		"""
		# The test connectivity action is a special case that does not
		# receive a param dictionary at all, so there are no arguments to
		# unpack
		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.

Action Context

The hidden phantom action context parameter can be accessed as the context member of the connector object during action execution. This data contains entries for artifact_id, guid, and parent_action_run; see 'Understanding Datapaths' in the offical phantom documentation for details.

Logging

The logger member of the connector object is a standard Python logging.Logger object that can be used in the normal way.

All log messages of ERROR and above will be logged to <syslog>/phantom/spawn.log. <syslog> will typically be /var/log on privileged installations and <PHANTOM_HOME>/var/log on unprivileged installations.

If the system is configured for DEBUG logging, all messages of DEBUG and above will also be logged. Additionally, all log messages of level INFO and above will be reported using BaseConnector.save_progress regardless of Phantom debug configuration.

The connector will also keep logfiles in the app's state directory containing logging records created by calls to the connector's logger. This allows convenient consumption of connector logging without having to filter through the content of spawn.log. The logfile can be found at <PHANTOM_HOME>/local_data/app_states/<APP_ID>/logs/connector.log.

Vault Files

The get_vault_path method can be used to retrieve a pathlib.Path object from a vault ID:

vault_path = connector.get_vault_path('<VAULT_ID>')
with vault_path.open() as vault_file:
	...

This is implemented to work on both the Phantom 4.10 and 4.8 Vault interfaces.

Persistent State

The state connector property can be used to access and save persistent state. The property will automatically load the peristent state dict from the filesystem on the first access, and save the current state to the filesystem at the end of action handling if such access has been performed. This property is intended to replace the variety of methods available on the base connector for loading, accessing, and saving state data.

More sophisticated state management can be performed using the open_state method. Providing a subpath of the app's state directory to the method will allow IO to the specified file. If no subpath is specified, the default state file containing the state dict will be opened.

Asynchronous Actions

While the Phantom app API specification only allows execution of a single action at once, it can still be beneficial to implement handler methods asynchronously.

If a handler method is reused by other handler methods, an asynchronous implementation will allow the method to be run concurrently with other coroutines. Additionally, implementing a handler as an asynchronous coroutine function will cause the connector to automatically handle the event loop, meaning that await statements can be made in the function implementation without manual event loop configuration.

Consider:

@ActionHandler
def slow_action(self):
	"""
	Two long IO operations
	"""
	expensive_call()
	expensive_call()

@ActionHandler
def even_slower_action(self):
	"""
	Two slow actions (4 long IO operations)
	"""
	self.slow_action()
	self.slow_action()

Executing a single even_slower_action action will cause expensive_call to be invoked 4 times in sequence.

Assuming a significant portion of expensive_call's execution time is waiting for IO operations, an asynchronous implementation called async_call will allow other coroutines to run while it waits. Assuming access to async_call, we can refactor our actions using asynchronous coroutine functions:

@ActionHandler
async def efficient_action(self):
	"""
	Two long IO operations run concurrently
	"""
	await asyncio.gather(async_call(), async_call())

@ActionHandler
async def even_more_efficient_action(self):
	"""
	Two efficient actions run concurrently
	(4 long IO operations run concurrently)
	"""
	await asyncio.gather(self.efficient_action(), self.efficient_action())

Under normal circumstances, even_slower_action will take almost 4 times as long to run as even_more_efficient_action while producing the same results. This efficiency will scale over the number of actions that can be run concurrently.

App Logo

The app logo can be customised by including an icon file in the app directory with the name logo.<extension>. logo_light.<extension> and logo_dark.<extension> can also be used to differentiate between light-mode and dark-mode logos. Arbitrary filenames and paths can also be used; specify a path for logo in the app metadata, or paths for light and dark under logo to configure the respecitve themes.

Default Logo

Icons made by Freepik from www.flaticon.com

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distributions

No source distribution files available for this release.See tutorial on generating distribution archives.

Built Distribution

phantom_dev-0.8.3-py3-none-any.whl (32.2 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page