# Sekoia.io Automation Python SDK
[](https://github.com/SEKOIA-IO/sekoia-automation-sdk/actions/workflows/ci.yml)
[](https://codecov.io/github/SEKOIA-IO/sekoia-automation-sdk)
[](https://pypi.org/project/sekoia-automation-sdk/)
[](https://pypi.org/project/sekoia-automation-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:
```python
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:
```python
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 to `constants.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 use `constants.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:
```python
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:
```python
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:
```python
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 with `self.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:
```python
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:
```python
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:
```python
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:
```python
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:
```python
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:
```python
class MyTrigger(Trigger):
configuration: MyConfigurationModel
```
You can also specify the model of created events by setting the `results_model` attribute:
```python
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:
```python
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:
```python
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.
Raw data
{
"_id": null,
"home_page": "https://sekoia.io/",
"name": "sekoia-automation-sdk",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.10",
"maintainer_email": null,
"keywords": "SDK, Sekoia.io, automation, playbook",
"author": "Sekoia.io",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/2f/52/80699bf9dd05b9e25fbd38b77b1ad9f63d84a4e229764bb13cc059f2740f/sekoia_automation_sdk-1.18.1.tar.gz",
"platform": null,
"description": "# Sekoia.io Automation Python SDK\n\n[](https://github.com/SEKOIA-IO/sekoia-automation-sdk/actions/workflows/ci.yml)\n[](https://codecov.io/github/SEKOIA-IO/sekoia-automation-sdk)\n[](https://pypi.org/project/sekoia-automation-sdk/)\n[](https://pypi.org/project/sekoia-automation-sdk/)\n\nSDK to create Sekoia.io playbook modules.\n\nModules can define:\n\n* Triggers: daemons that create events that will start a playbook run\n* Actions: short-lived programs that constitute the main playbook nodes. They take arguments and produce a result.\n\n## Create a trigger\n\nHere is how you could define a very basic trigger:\n\n```python\nfrom sekoia_automation.module import Module\nfrom sekoia_automation.trigger import Trigger\n\n\nclass MyTrigger(Trigger):\n def run(self):\n while True:\n # Do some stuff\n self.send_event('event_name', {'somekey': 'somevalue'})\n # Maybe wait some time\n\n\nif __name__ == \"__main__\":\n module = Module()\n\n module.register(MyTrigger)\n module.run()\n```\n\nYou can access the Trigger's configuration with `self.configuration` and the module configuration with `self.module.configuration`.\n\n### Attach files to an event\n\nYou can attach files to an event so that these files are available to the playbook runs.\n\nHere is how you could crete a file that should be available to the playbook run:\n\n```python\nimport os\n\nfrom sekoia_automation import constants\nfrom sekoia_automation.trigger import Trigger\n\n\nclass MyTrigger(Trigger):\n def run(self):\n while True:\n # Create a directory and a file\n directory_name = \"test_dir\"\n dirpath = os.path.join(constants.DATA_STORAGE, directory_name)\n os.makedirs(dirpath)\n\n with open(os.path.join(dirpath, \"test.txt\") \"w\") as f:\n f.write(\"Hello !\")\n\n # Attach the file to the event\n self.send_event('event_name', {'file_path': 'test.txt'}, directory_name)\n\n # Maybe wait some time\n```\n\nPlease note that:\n\n* `send_event`'s third argument should be the path of a directory, relative to `constants.DATA_STORAGE`\n* The directory will be the root of the playbook run's storage (\"test.txt\" will exist, not \"test_dir/test.txt\")\n* You can ask the SDK to automatically remove the directory after it was copied with `remove_directory=True`\n* You should always do `from sekoia_automation import constants` and use `constants.DATA_STORAGE` so that it is easy to mock\n\nWhen attaching a single file to a playbook run, you can use the `write` function to create the file:\n\n```python\nfrom sekoia_automation.storage import write\nfrom sekoia_automation.trigger import Trigger\n\n\nclass MyTrigger(Trigger):\n def run(self):\n while True:\n # Simple creation of a file\n filepath = write('test.txt', {'event': 'data'})\n\n # Attach the file to the event\n self.send_event('event_name', {'file_path': os.path.basename(filepath)},\n os.path.dirname(directory_name))\n\n # Maybe wait some time\n```\n\n### Persisting data to disk\n\nMost of the time, triggers have to maintain some state do to their work properly (such as a cursor).\nIn order to make sure that this data survives a reboot of the Trigger (which can happen with no reason),\nit is useful to persist it to the trigger's storage.\n\nWhen the manipulated data is JSON serializable, it is recommended to use the `PersistentJSON` class to do\nso (instead of `shelve`). Used as a context manager, this class will make sure the python dict is properly\nsynchronised:\n\n```python\nfrom sekoia_automation.trigger import Trigger\nfrom sekoia_automation.storage import PersistentJSON\n\n\nclass MyTrigger(Trigger):\n def run(self):\n while True:\n # Read and update state\n with PersistentJSON('cache.json') as cache:\n # Use cache as you would use a normal python dict\n```\n\n## Create an action\n\nHere is how you could define a very basic action that simply adds its arguments as result:\n\n```python\nfrom sekoia_automation.module import Module\nfrom sekoia_automation.action import Action\n\n\nclass MyAction(Action):\n def run(self, arguments):\n return arguments # Return value should be a JSON serializable dict\n\n\nif __name__ == \"__main__\":\n module = Module()\n\n module.register(MyAction)\n module.run()\n```\n\nThere are a few more things you can do within an Action:\n\n* Access the Module's configuration with `self.module.configuration`\n* Add log messages with `self.log('message', 'level')`\n* Activate an output branch with `self.set_output('malicious')` or explicitely disable another with `self.set_output('benign', False)`\n* 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\n\n### Working with files\n\nActions can read and write files the same way a Trigger can:\n\n```python\nfrom sekoia_automation import constants\n\nfilepath = os.path.join(constants.DATA_STORAGE, \"test.txt\")\n```\n\nIt is a common pattern to accept JSON arguments values directly or inside a file. The SDK provides an helper to easily read such arguments:\n\n```python\nclass MyAction(Action):\n\n def run(self, arguments):\n test = self.json_argument(\"test\", arguments)\n\n # Do somehting with test\n```\n\nThe value will automatically be fetched from `test` if present, or read from the file at `test_path`.\n\nThe SDK also provides an helper to do the opposite with results:\n\n```python\nclass MyAction(Action):\n\n def run(self, arguments):\n return self.json_result(\"test\", {\"some\": \"value\"})\n```\n\nThis will create a dict with `test_path` by default or `test` if the last argument was passed directly.\n\n## Same Docker Image for several items\n\nIn most cases, it makes sense to define several triggers and / or actions sharing the same code and the same docker image.\n\nIn this case, here is how you should define the main:\n\n```python\nif __name__ == \"__main__\":\n module = Module()\n\n module.register(Trigger1, \"command_trigger1\")\n module.register(Trigger2, \"command_trigger2\")\n module.register(Action1, \"command_action1\")\n module.register(Action2, \"command_action2\")\n module.run()\n```\n\nThe corresponding commands need to be correctly set in the manifests as \"docker_parameters\".\n\n## Use with Pydantic\n\nIt is recommended to use Pydantic to develop new modules. This should ease development.\n\n### Module Configuration\n\nA pydantic model can be used as `self.module.configuration` by adding type hints:\n\n```python\nclass MyConfigurationModel(BaseModel):\n field: str\n\nclass MyModule(Module):\n configuration: MyConfiguration\n\nclass MyAction(Action):\n module: MyModule\n```\n\n### Triggers\n\nThe Trigger configuration can also be a pydantic model by adding a type hint:\n\n```python\nclass MyTrigger(Trigger):\n configuration: MyConfigurationModel\n```\n\nYou can also specify the model of created events by setting the `results_model` attribute:\n\n```python\nclass Event(BaseModel):\n field: str = \"value\"\n\nclass MyTrigger(Trigger):\n results_model = Event\n```\n\n### Actions\n\nYou can use a pydantic model as action arguments by adding a type hint:\n\n```python\nclass ActionArguments(BaseModel):\n field: str = \"value\"\n\nclass MyAction(Action):\n def run(self, arguments: ActionArguments):\n ...\n```\n\nThe model of results can also be specified by setting the `results_model` attribute:\n\n```python\nclass Results(BaseModel):\n field: str = \"value\"\n\nclass MyAction(action):\n results_model = Results\n```\n\n### Automatically generating manifests\n\nWhen using pydantic models to describe configurations, arguments and results, manifests\ncan be automatically generated:\n\n```\n$ poetry run sekoia-automation generate-files\n```\n\nThis will do the following:\n\n* Generate `main.py`\n* Generate a manifest for each action\n* Generate a manifest for each trigger\n* Update the module's manifest\n\nFor better results, it is recommended to set the `name` and `description` attributes in Actions\nand Triggers.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "SDK to create Sekoia.io playbook modules",
"version": "1.18.1",
"project_urls": {
"Documentation": "https://docs.sekoia.io/",
"Homepage": "https://sekoia.io/",
"Repository": "https://github.com/SEKOIA-IO/sekoia-automation-sdk"
},
"split_keywords": [
"sdk",
" sekoia.io",
" automation",
" playbook"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "d7e68f51a43cfe8fb107be105df80a42187cd5022ce052ad7ebebde100c19b11",
"md5": "353390714c2cf98989edb4c083d83648",
"sha256": "1affa65ec46414bf67912e49b2295476279cf86c3cb34ed70ce6392d42c0b919"
},
"downloads": -1,
"filename": "sekoia_automation_sdk-1.18.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "353390714c2cf98989edb4c083d83648",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.10",
"size": 72483,
"upload_time": "2024-12-04T09:11:33",
"upload_time_iso_8601": "2024-12-04T09:11:33.203160Z",
"url": "https://files.pythonhosted.org/packages/d7/e6/8f51a43cfe8fb107be105df80a42187cd5022ce052ad7ebebde100c19b11/sekoia_automation_sdk-1.18.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "2f5280699bf9dd05b9e25fbd38b77b1ad9f63d84a4e229764bb13cc059f2740f",
"md5": "ebe3089603a493348c44443f4b8f7513",
"sha256": "6c534ac1356aba404d5caf45390dd74a5d4aa3fb5c32bf9700e4d05b495b76d5"
},
"downloads": -1,
"filename": "sekoia_automation_sdk-1.18.1.tar.gz",
"has_sig": false,
"md5_digest": "ebe3089603a493348c44443f4b8f7513",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.10",
"size": 54974,
"upload_time": "2024-12-04T09:11:35",
"upload_time_iso_8601": "2024-12-04T09:11:35.072185Z",
"url": "https://files.pythonhosted.org/packages/2f/52/80699bf9dd05b9e25fbd38b77b1ad9f63d84a4e229764bb13cc059f2740f/sekoia_automation_sdk-1.18.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-04 09:11:35",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "SEKOIA-IO",
"github_project": "sekoia-automation-sdk",
"travis_ci": false,
"coveralls": true,
"github_actions": true,
"lcname": "sekoia-automation-sdk"
}
JSON
