phantom-dev 0.14.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

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
• Remote app log monitoring
• Asynchronous action handler support
• Dedicated log files for connector logging output
• Cross-platform client
• Support for apps not created with phantom-dev
• Comprehensive Phantom API mocking with type-hinting
• Models for Phantom data types, including Containers and Artifacts

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.

App Versioning

Unless a version is explicitly specified in the app metadata, roboversion will be used to automatically detect the project version from git repository state. To set the version, tag the current commit with a version label (e.g. v1.2.3).

If the current commit doesn't have a version label, roboversion will create a unique development version using the most recent version tag.

Compatibility with Apps Not Created with phantom-dev

Because phantom-dev relies on specific library features to infer action metadata from code implementations, metadata inference will not work with a connector that has been implemented without phantom-dev. Full specification of app metadata must therefore be provided in either metadata.yaml or in the old-style .json file.

If a .json file is used instead of metadata.yaml, phantom-dev will not autogenerate a new .json metadata file. Certain packaging features, such as autopopulation of dependency metadata, will therefore be disabled.

Finally, the debug command will not work unless the app is configured as a debugpy listener when executed as __main__. While phantom-dev apps implement this by default, other apps will have to implement this behaviour or forgo the use of debug.

Aside from these limitations, phantom-dev should work normally with any Phantom app once extracted into a project directory.

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.

Configuring IDE Autocompletion

To enable your IDE to resolve Phantom module methods, configure your IDE to set its PYTHONPATH environment variable to include the location of phantom_dev.dummy.

The dummy file path can be found by running phantom-dev --dummy_path. The path will typically resemble <PYTHON_ENV>/lib/python<VERSION>/site-packages/phantom_dev/dummy, where PYTHON_ENV is your python environment and <VERSION> is your Python version.

Example: VSCode

Find the path to the dummy module:

$ phantom-dev --dummy_path
/home/USER/my_app/.venv/lib/python3.8/site-packages/phantom_dev/dummy

To add this to the PYTHONPATH for VSCode, place the following entry in the .env file in the workspace directory (create the file if it doesn't exist):

PYTHONPATH=/home/USER/my_app/.venv/lib/python3.8/site-packages/phantom_dev/dummy

Extending the App

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

from phantom_dev.action_handler import ActionHandler, SmartConnector
...
class Connector(SmartConnector, main=True):
	...
	@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 SmartConnector, 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, SmartConnector
...

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

...

class Connector(SmartConnector, main=True):
	...
	@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.

Parameters with specific allowed values

To define a set of allowed values for a parameter, define a subclass of Enum with a concrete type that maps to a data_type, and assign those values to members of that class.

Specify this class as the parameter annotation to allow phantom-dev to infer the value_list metadata for this parameter.

For example, for a rule string parameter with allowed values 'allow' and 'block':

@contains('rule')
class Rule(str, Enum):
	ALLOW = 'allow'
	BLOCK = 'block'

...
class Connector(SmartConnector, main=True):
	@ActionHandler(...)
	def set_rule(self, rule: Rule, ...): ...

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/ [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

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.

App Documentation

Aside from the action and parameter descriptions which can be inferred from connector method docstrings, Phantom supports adding more detailed documentation to apps using a readme.html file.

Apps created with phantom-dev will be created with a default readme.html, which can be updated with arbitrary HTML content.

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

SmartConnector 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']`"""

...

class Connector(SmartConnector, main=True):
	...
	@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 parameters 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']`"""

...

class Connector(SmartConnector, main=True):
	...
	@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

Scheduling actions to run simultaneously can be configured with the optional lock parameter given to the ActionHandler decorator.

Within a single action, asynchronous coroutine execution can be configured by implementing the action handler as an async coroutine method.

Asynchronous Action Scheduling

By default, actions are scheduled synchronously with regard to the asset running the action; i.e. only one action will be running on a given asset at at time.

This behaviour can be altered by specifying a value for ActionHandler's lock parameter. If set to None, the action will be scheduled asynchronously with regard to all other actions. If given a str value, the action will be scheduled synchronously with regard to all other actions using the same lock value.

A timeout (in seconds) for lock acquision can be set by specifying a value for ActionHandler's lock_timeout parameter.

Lock Data Paths

Action parameter or configuration values can be specified as data paths for the lock value, by setting the value to parameter.<name> or configuration.<name> respectively. If a datapath for a parameter or configuration value is specified, its value at the time of action execution will be used as the lock value. This allows configurable groups of actions which will be scheduled synchronously with regard to each other, but asynchronously with regard to all other actions.

Asynchronous Coroutine Execution

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.

Data Ingestion

Implementing an on_poll ActionHandler method allows the app to ingest containers based on arbitrary logic according to a configured schedule.

The default connector includes a reference implementation for on_poll. The following example outlines ingestion of a single container with a single artifact:

...
	@ActionHandler(action_type='ingest', read_only=True)
	def on_poll(
			self,
			start_time: int,
			end_time: int,
			container_count: int,
			artifact_count: int,
			container_id: str = None,
	):
		"""
		Callback action for the on_poll ingest functionality.

		:param container_id: Container IDs to limit the ingestion to.
		:param start_time:
			Start of time range, in epoch time (milliseconds)

			If not specified, the default is past 10 days
		:param end_time:
			End of time range, in epoch time (milliseconds)

			If not specified, the default is now"
		:param container_count:
			Maximum number of container records to query for.
		:param artifact_count: Maximum number of artifact records to query for.
		"""
		container = Container.from_fields(name='example container', ...)
		container.add_artifact(name='example artifact', ...)
		return self.save_all_containers([container])
...

Rendering Output

Action output rendering can be customised by adding a render section to the action entry in the metadata.yaml, which takes the same data as specified in the official documentation. If using a custom view, the ActionHandler.custom_view decorator can be used to automatically populate the render section of the action metadata.

Custom Views

Adding a custom view requires registering a custom view function and creating a custom view HTML template.

To register a custom view function, the ActionHandler.custom_view decorator can be used on the function. Alternatively, a custom view can be manually specified by adding a render section to the action entry in the metadata.yaml.

Normally, the return value of the function is a string that contains the path to the HTML template used to render the view. If the decorator is used to register the function and the function doesn't return a value, the template location will be inferred to be views/<function_name>.html.

Refer to the Phantom documentation for more information on using custom views.

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