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:
- Generate a test action JSON file on the remote Phantom host
- Open an SSH tunnel to the remote Phantom host, forwarding the local port to the remote debugging port
- 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
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
Built Distribution
Hashes for phantom_dev-0.8.3-py3-none-any.whl
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | acf55b0b056aa2bc71ad3221c2ce10a2b6d508d4ffca44c367ce66ba48a5c70e |
|
| MD5 | 48d5f75989120811524a7de8b54851ac |
|
| BLAKE2b-256 | 0c931f2840776290edfe822b59eec4ef46f6ac12c5c4bead8f9b9b49a1c37575 |
