SDK to create Sekoia.io playbook modules
Project description
Sekoia.io Automation Python SDK
SDK to create Sekoia.io playbook modules.
Modules can define:
- Triggers: daemons that create events that will start a playbook run
- Actions: short-lived programs that constitute the main playbook nodes. They take arguments and produce a result.
Create a trigger
Here is how you could define a very basic trigger:
from sekoia_automation.module import Module
from sekoia_automation.trigger import Trigger
class MyTrigger(Trigger):
def run(self):
while True:
# Do some stuff
self.send_event('event_name', {'somekey': 'somevalue'})
# Maybe wait some time
if __name__ == "__main__":
module = Module()
module.register(MyTrigger)
module.run()
You can access the Trigger's configuration with self.configuration
and the module configuration with self.module.configuration
.
Attach files to an event
You can attach files to an event so that these files are available to the playbook runs.
Here is how you could crete a file that should be available to the playbook run:
import os
from sekoia_automation import constants
from sekoia_automation.trigger import Trigger
class MyTrigger(Trigger):
def run(self):
while True:
# Create a directory and a file
directory_name = "test_dir"
dirpath = os.path.join(constants.DATA_STORAGE, directory_name)
os.makedirs(dirpath)
with open(os.path.join(dirpath, "test.txt") "w") as f:
f.write("Hello !")
# Attach the file to the event
self.send_event('event_name', {'file_path': 'test.txt'}, directory_name)
# Maybe wait some time
Please note that:
send_event
's third argument should be the path of a directory, relative toconstants.DATA_STORAGE
- The directory will be the root of the playbook run's storage ("test.txt" will exist, not "test_dir/test.txt")
- You can ask the SDK to automatically remove the directory after it was copied with
remove_directory=True
- You should always do
from sekoia_automation import constants
and useconstants.DATA_STORAGE
so that it is easy to mock
When attaching a single file to a playbook run, you can use the write
function to create the file:
from sekoia_automation.storage import write
from sekoia_automation.trigger import Trigger
class MyTrigger(Trigger):
def run(self):
while True:
# Simple creation of a file
filepath = write('test.txt', {'event': 'data'})
# Attach the file to the event
self.send_event('event_name', {'file_path': os.path.basename(filepath)},
os.path.dirname(directory_name))
# Maybe wait some time
Persisting data to disk
Most of the time, triggers have to maintain some state do to their work properly (such as a cursor). In order to make sure that this data survives a reboot of the Trigger (which can happen with no reason), it is useful to persist it to the trigger's storage.
When the manipulated data is JSON serializable, it is recommended to use the PersistentJSON
class to do
so (instead of shelve
). Used as a context manager, this class will make sure the python dict is properly
synchronised:
from sekoia_automation.trigger import Trigger
from sekoia_automation.storage import PersistentJSON
class MyTrigger(Trigger):
def run(self):
while True:
# Read and update state
with PersistentJSON('cache.json') as cache:
# Use cache as you would use a normal python dict
Create an action
Here is how you could define a very basic action that simply adds its arguments as result:
from sekoia_automation.module import Module
from sekoia_automation.action import Action
class MyAction(Action):
def run(self, arguments):
return arguments # Return value should be a JSON serializable dict
if __name__ == "__main__":
module = Module()
module.register(MyAction)
module.run()
There are a few more things you can do within an Action:
- Access the Module's configuration with
self.module.configuration
- Add log messages with
self.log('message', 'level')
- Activate an output branch with
self.set_output('malicious')
or explicitely disable another withself.set_output('benign', False)
- Raise an error with
self.error('error message')
. Note that raised exceptions that are not catched by your code will be automatically handled by the SDK
Working with files
Actions can read and write files the same way a Trigger can:
from sekoia_automation import constants
filepath = os.path.join(constants.DATA_STORAGE, "test.txt")
It is a common pattern to accept JSON arguments values directly or inside a file. The SDK provides an helper to easily read such arguments:
class MyAction(Action):
def run(self, arguments):
test = self.json_argument("test", arguments)
# Do somehting with test
The value will automatically be fetched from test
if present, or read from the file at test_path
.
The SDK also provides an helper to do the opposite with results:
class MyAction(Action):
def run(self, arguments):
return self.json_result("test", {"some": "value"})
This will create a dict with test_path
by default or test
if the last argument was passed directly.
Same Docker Image for several items
In most cases, it makes sense to define several triggers and / or actions sharing the same code and the same docker image.
In this case, here is how you should define the main:
if __name__ == "__main__":
module = Module()
module.register(Trigger1, "command_trigger1")
module.register(Trigger2, "command_trigger2")
module.register(Action1, "command_action1")
module.register(Action2, "command_action2")
module.run()
The corresponding commands need to be correctly set in the manifests as "docker_parameters".
Use with Pydantic
It is recommended to use Pydantic to develop new modules. This should ease development.
Module Configuration
A pydantic model can be used as self.module.configuration
by adding type hints:
class MyConfigurationModel(BaseModel):
field: str
class MyModule(Module):
configuration: MyConfiguration
class MyAction(Action):
module: MyModule
Triggers
The Trigger configuration can also be a pydantic model by adding a type hint:
class MyTrigger(Trigger):
configuration: MyConfigurationModel
You can also specify the model of created events by setting the results_model
attribute:
class Event(BaseModel):
field: str = "value"
class MyTrigger(Trigger):
results_model = Event
Actions
You can use a pydantic model as action arguments by adding a type hint:
class ActionArguments(BaseModel):
field: str = "value"
class MyAction(Action):
def run(self, arguments: ActionArguments):
...
The model of results can also be specified by setting the results_model
attribute:
class Results(BaseModel):
field: str = "value"
class MyAction(action):
results_model = Results
Automatically generating manifests
When using pydantic models to describe configurations, arguments and results, manifests can be automatically generated:
$ poetry run sekoia-automation generate-files
This will do the following:
- Generate
main.py
- Generate a manifest for each action
- Generate a manifest for each trigger
- Update the module's manifest
For better results, it is recommended to set the name
and description
attributes in Actions
and Triggers.
Project details
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 Distribution
Built Distribution
Hashes for sekoia_automation_sdk-1.17.1.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 72d911684b0ef57c8420045af1d79b0e2511d2074f780b18ac4a68a1a6cc68f8 |
|
MD5 | 2b1569a2dffa2399c2a9a24f34f43200 |
|
BLAKE2b-256 | 2569022322cd48cafcc254fffc27871a08b8eddfef950418a2c8f7655d558b8f |
Hashes for sekoia_automation_sdk-1.17.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f7b267e627445cc3ec0b556a6420e1398410e4a9e534a37f033fc017d23a100b |
|
MD5 | 6084919b3583a4781db8390feb8772e5 |
|
BLAKE2b-256 | 8316c89db7f2b1f1ed31ef89002d824933f91cd192f7f52fdd79195d97c71089 |