Developing Application

Introduction

Imperum applications provide a way to connect external applications and devices. Imperum applications are written in Python.

Application structure

All imperum applications have to implement a class named App that extends soar_core.actions.base_action.BaseAction. You can call this file with any name you want but it has to be pointed in the manifest.json file.

filename: "app.py"

Imperum expects actions to be defined in this class. Actions are defined as async methods with the following signature:

from soar_core.actions.base_action import BaseAction

class App(BaseAction):
    async def action_name(self, payload):
        query = payload['q']
        # await response
        # return response

The action_name is the name of the action that will be called by Imperum.

The payload is a dictionary containing the data that was sent by Imperum according to the schema defined in the manifest.json file.

...
"actions": [
  {
    "name": "action_name",
    "description": "Description of the action",
    "input_params": [
      {
        "name": "q",
        "type": "string",
        "description": "Search query.",
        "placeholder": "Search query",
        "order": 0,
        "required": true
      }
    ]
  }
],
...

The returned data can be reached by the next blocks in the flow. In order to help the user to understand the data, you can define a schema for the returned data in the manifest.json file. The the value of the name key is written in json path as the action may return nested structure.

...
"actions": [
  {
    "name": "action_name",
    "description": "Description of the action",
    "output_params": [
      {
        "name": "items.[*].id",
        "type": "string",
      },
    ]
  }
],
...

Logging

Imperum applications can log messages to the console. To log a message, you need to import the logging module and call the logging.info method.

from soar_core.actions.base_action import BaseAction
import logging

logger = logging.getLogger("colorizedLogger")

class App(BaseAction):

    def action_name(self, payload):
        logger.info("Hello World!")
        pass

Configuration

Imperum applications can specify a configuration schema in manifest.json file. The schema is used to get the configuration from the user.

To reach configuration in the application, you need to call self.get_config() method

Let's assume a configuration schema defined in the manifest.json like the following:

...
"configurations": [
    {
        "name": "api_key",
        "type": "password",
        "description": "API Key",
        "placeholder": "API Key",
        "order": 0,
        "required": true,
        "default": ""
    }
],
...

To get the api_key from the configuration, you need to call self.get_config() method. This method returns the configuration as a dictionary.

from soar_core.actions.base_action import BaseAction

class App(BaseAction):
    def action_name(self, payload):
        config = self.get_config()
        api_key = config['api_key'] # api_key is a string
        pass

Connectors

Developing a connector for an app is optional. Only the apps that have an endpoint that returns events should have a connector script. Connectors script must be under connectors folder and the script name must be matched with the name Connector definition manifest.json file.

Define a class named App that extends base.py App

from datetime import datetime, timedelta
from typing import TypedDict


import aiohttp

from soar_core.functions import initialize_logger

from gatewatcher.base import App as BaseApp

logger = initialize_logger()

class App(BaseApp):
    async def get_alerts(self, params: Payload) -> tuple[list, datetime]:
        return events, last_date

get_alerts method should return a tuple of list and datetime. The list contains events and the datetime keep the last occurred event in the external platform.

Run connector function

Each connector file must contain run_connector function. This function return a dict of partially_mapped_event and raw_event items. After events are fetched the connector parameters need to be updated by App.update_connector_params in order to fetch next events per each cycle.

async def run_connector(config, payload) -> tuple[list, str]:
    app_instance = App(config=config)
    alerts, last_date = await app_instance.get_alerts(payload)
    app_instance.update_connector_params(connector_id=config['connector_id'], update_fields ={"date_from": last_date})
    return map_partially(alerts)

Event fields mapping objects just map entities and artifacts. In each app, event fields such as severity, tactics, techniques, type, external_id must be mapped and added to partially_mapped_event item as follow.

def map_partially(alerts: list):
    raw_and_partially_mapped_events = []
    for alert in alerts:
        tactics, techniques = get_tactics_and_techniques(alert)
        severity = alert["severity"]
        if severity == 1:
            str_severity = "Low"
        elif severity == 2:
            str_severity = "Medium"
        elif severity == 3:
            str_severity = "High"


        raw_and_partially_mapped_events.append(
            {
                "raw_event": alert,
                "partially_mapped_event": {
                    "tactics": tactics,
                    "techniques": techniques,
                    "severity": str_severity,
                    "external_id": alert["uuid"],
                    "external_occured_at": alert["date"],
                    "type": alert["type"]
                }
            }
        )

    return raw_and_partially_mapped_events


def get_tactics_and_techniques(alert):
    tactics = [
        {**mitre_item["tactic"]}
        for mitre_item in alert["mitre"]
    ]
    techniques = []
    for mitre_item in alert["mitre"]:
        techniques.extend(mitre_item["techniques"])

    return tactics, techniques