src/app.py

This document will dive deeper into the initial structure of the app.py file when starting working with Apps.

The file consists of a few main parts:

1. Logger initialization

2. Asset definition

3. App initialization

4. Actions definitions

5. App CLI invocation

Here’s an example app.py file which uses a wide variety of the features available in the SDK:

from collections.abc import Iterator
from datetime import UTC, datetime
from zoneinfo import ZoneInfo

from soar_sdk.abstract import SOARClient
from soar_sdk.action_results import ActionOutput, MakeRequestOutput, OutputField
from soar_sdk.app import App
from soar_sdk.asset import AssetField, BaseAsset
from soar_sdk.logging import getLogger
from soar_sdk.models.artifact import Artifact
from soar_sdk.models.attachment_input import AttachmentInput
from soar_sdk.models.container import Container
from soar_sdk.models.finding import Finding
from soar_sdk.params import (
    MakeRequestParams,
    OnESPollParams,
    OnPollParams,
    Param,
    Params,
)

logger = getLogger()

# Test mail template for ES findings
SAMPLE_EMAIL_TEMPLATE = """From: suspicious@example.com
To: {user}
Subject: Suspicious Activity Detected
Date: {date}

This is a suspicious email that triggered the risk threshold.
Event ID: {event_id}
"""

# Test event data CSV for attachments
SAMPLE_EVENTS_CSV = """timestamp,user,action,risk_score
{timestamp},{{user}},login_failed,{{risk_score}}
{timestamp},{{user}},access_denied,{{secondary_score}}
"""


class Asset(BaseAsset):
    base_url: str = AssetField(default="https://example")
    api_key: str = AssetField(sensitive=True, description="API key for authentication")
    key_header: str = AssetField(
        default="Authorization",
        value_list=["Authorization", "X-API-Key"],
        description="Header for API key authentication",
    )
    timezone: ZoneInfo
    timezone_with_default: ZoneInfo = AssetField(default=ZoneInfo("America/Denver"))


app = App(
    asset_cls=Asset,
    name="example_app",
    appid="9b388c08-67de-4ca4-817f-26f8fb7cbf55",
    app_type="sandbox",
    product_vendor="Splunk Inc.",
    logo="logo.svg",
    logo_dark="logo_dark.svg",
    product_name="Example App",
    publisher="Splunk Inc.",
    min_phantom_version="6.2.2.134",
)


@app.test_connectivity()
def test_connectivity(soar: SOARClient, asset: Asset) -> None:
    soar.get("rest/version")
    container_id = soar.get_executing_container_id()
    logger.info(f"current executing container's container_id is: {container_id}")
    asset_id = soar.get_asset_id()
    logger.info(f"current executing container's asset_id is: {asset_id}")
    logger.info(f"testing connectivity against {asset.base_url}")
    logger.debug("hello")
    logger.warning("this is a warning")
    logger.progress("this is a progress message")


class ActionOutputSummary(ActionOutput):
    is_success: bool


@app.action()
def test_summary_with_list_output(
    params: Params, asset: Asset, soar: SOARClient
) -> list[ActionOutput]:
    soar.set_summary(ActionOutputSummary(is_success=True))
    return [ActionOutput(), ActionOutput()]


@app.action()
def test_empty_list_output(
    params: Params, asset: Asset, soar: SOARClient
) -> list[ActionOutput]:
    return []


class JsonOutput(ActionOutput):
    name: str = OutputField(example_values=["John", "Jane", "Jim"], column_name="Name")
    age: int = OutputField(example_values=[25, 30, 35], column_name="Age")


class TableParams(Params):
    company_name: str = Param(column_name="Company Name", default="Splunk")


@app.action(render_as="json")
def test_json_output(params: Params, asset: Asset, soar: SOARClient) -> JsonOutput:
    return JsonOutput(name="John", age=25)


@app.action(render_as="table")
def test_table_output(
    params: TableParams, asset: Asset, soar: SOARClient
) -> JsonOutput:
    return JsonOutput(name="John", age=25)


from .actions.reverse_string import render_reverse_string_view

app.register_action(
    "actions.reverse_string:reverse_string",
    action_type="investigate",
    verbose="Reverses a string.",
    view_template="reverse_string.html",
    view_handler=render_reverse_string_view,
)

from .actions.generate_category import render_statistics_chart

app.register_action(
    "actions.generate_category:generate_statistics",
    action_type="investigate",
    verbose="Generate statistics with pie chart reusable component.",
    view_handler=render_statistics_chart,
)


class MakeRequestParamsCustom(MakeRequestParams):
    endpoint: str = Param(
        description="The endpoint to send the request to. Base url is already included in the endpoint.",
        required=True,
    )


@app.make_request()
def http_action(params: MakeRequestParamsCustom, asset: Asset) -> MakeRequestOutput:
    logger.info(f"HTTP action triggered with params: {params}")
    return MakeRequestOutput(
        status_code=200,
        response_body=f"Base url is {asset.base_url}",
    )


@app.on_poll()
def on_poll(
    params: OnPollParams, soar: SOARClient, asset: Asset
) -> Iterator[Container | Artifact]:
    if params.is_manual_poll():
        logger.info("Manual poll (poll now) detected")
    else:
        logger.info("Scheduled poll detected")

    # Create container first for artifacts
    yield Container(
        name="Network Alerts",
        description="Some network-related alerts",
        severity="medium",
    )

    # Simulate collecting 2 network artifacts that will be put in the network alerts container
    for i in range(1, 3):
        logger.info(f"Processing network artifact {i}")

        alert_id = f"testalert-{datetime.now(UTC).strftime('%Y%m%d')}-{i}"
        artifact = Artifact(
            name=f"Network Alert {i}",
            label="alert",
            severity="medium",
            source_data_identifier=alert_id,
            type="network",
            description=f"Example network alert {i} from polling operation",
            data={
                "alert_id": alert_id,
                "source_ip": f"10.0.0.{i}",
                "destination_ip": "192.168.0.1",
                "protocol": "TCP",
            },
        )

        yield artifact


@app.on_es_poll()
def on_es_poll(
    params: OnESPollParams, soar: SOARClient, asset: Asset
) -> Iterator[tuple[Finding, list[AttachmentInput]]]:
    for i in range(1, 3):
        logger.info(f"Processing ES finding {i}")

        finding = Finding(
            rule_title=f"Risk threshold exceeded for user-{i}",
            rule_description="Risk Threshold Exceeded for an object over a 24 hour period",
            security_domain="threat",
            risk_object=f"user{i}@example.com",
            risk_object_type="user",
            risk_score=75.0 + (i * 10),
            status="New",
            urgency="medium",
        )

        # Attach evidence
        email_evidence = SAMPLE_EMAIL_TEMPLATE.format(
            user=f"user{i}@example.com",
            date=datetime.now(UTC).strftime("%a, %d %b %Y %H:%M:%S +0000"),
            event_id=f"EVT-{i:04d}",
        )

        event_data = SAMPLE_EVENTS_CSV.format(
            timestamp=datetime.now(UTC).isoformat()
        ).format(
            user=f"user{i}@example.com",
            risk_score=75.0 + (i * 10),
            secondary_score=50.0 + (i * 5),
        )

        attachments = [
            AttachmentInput(
                file_content=email_evidence,
                file_name=f"suspicious_email_user{i}.eml",
                metadata={"type": "email_evidence", "source": "investigation_mailbox"},
            ),
            AttachmentInput(
                file_content=event_data,
                file_name=f"risk_events_user{i}.csv",
                metadata={"type": "event_log", "event_count": "2"},
            ),
        ]

        yield (finding, attachments)


app.register_action(
    "actions.async_action:async_process",
    action_type="investigate",
    verbose="Processes a message asynchronously with concurrent HTTP requests.",
)

app.register_action(
    "actions.async_action:sync_process",
    action_type="investigate",
    verbose="Processes a message synchronously with sequential HTTP requests.",
)


class GeneratorActionOutput(ActionOutput):
    iteration: int


class GeneratorActionSummary(ActionOutput):
    total_iterations: int


@app.action(summary_type=GeneratorActionSummary)
def generator_action(
    params: Params, soar: SOARClient[GeneratorActionSummary], asset: Asset
) -> Iterator[GeneratorActionOutput]:
    """Generates a sequence of numbers."""
    logger.info(f"Generator action triggered with params: {params}")
    for i in range(5):
        yield GeneratorActionOutput(iteration=i)
    soar.set_summary(GeneratorActionSummary(total_iterations=5))


@app.action()
def write_state(params: Params, soar: SOARClient, asset: Asset) -> ActionOutput:
    asset.cache_state.clear()
    assert asset.cache_state == {}
    asset.cache_state["value"] = "banana"
    return ActionOutput()


@app.action()
def read_state(params: Params, soar: SOARClient, asset: Asset) -> ActionOutput:
    assert asset.cache_state == {"value": "banana"}
    return ActionOutput()


if __name__ == "__main__":
    app.cli()

Components of the app.py File

Let’s dive deeper into each part of the app.py file above:

Logger Initialization

Logger initialization

from soar_sdk.logging import getLogger
from soar_sdk.models.artifact import Artifact
from soar_sdk.models.attachment_input import AttachmentInput
from soar_sdk.models.container import Container
from soar_sdk.models.finding import Finding
from soar_sdk.params import (
    MakeRequestParams,
    OnESPollParams,
    OnPollParams,
    Param,
    Params,
)

logger = getLogger()

The SDK provides a logging interface via the getLogger() function. This is a standard Python logger which is pre-configured to work with either the local CLI or the Splunk SOAR platform. Within the platform,

• logger.debug() and logger.warning() messages are written to the spawn.log file at DEBUG level.

• logger.error() and logger.critical() messages are written to the spawn.log file at ERROR level.

• logger.info() messages are sent to the Splunk SOAR platform as persistent action progress messages, visible in the UI.

• logger.progress() messages are sent to the Splunk SOAR platform as transient action progress messages, visible in the UI, but overwritten by subsequent progress messages.

When running locally via the CLI, all log messages are printed to the console, in colors corresponding to their log level.

Asset Definition

Asset definition

class Asset(BaseAsset):
    base_url: str = AssetField(default="https://example")
    api_key: str = AssetField(sensitive=True, description="API key for authentication")
    key_header: str = AssetField(
        default="Authorization",
        value_list=["Authorization", "X-API-Key"],
        description="Header for API key authentication",
    )
    timezone: ZoneInfo
    timezone_with_default: ZoneInfo = AssetField(default=ZoneInfo("America/Denver"))

Apps should define an asset class to hold configuration information for the app. The asset class should be a pydantic model that inherits from BaseAsset and defines the app’s configuration fields. Fields requiring metadata should be defined using an instance of AssetField(). The SDK uses this information to generate the asset configuration form in the Splunk SOAR platform UI.

App Initialization

App initialization

app = App(
    asset_cls=Asset,
    name="example_app",
    appid="9b388c08-67de-4ca4-817f-26f8fb7cbf55",
    app_type="sandbox",
    product_vendor="Splunk Inc.",
    logo="logo.svg",
    logo_dark="logo_dark.svg",
    product_name="Example App",
    publisher="Splunk Inc.",
    min_phantom_version="6.2.2.134",
)

This is how you initialize the basic App instance. The app object will be used to register actions, views, and/or webhooks. Keep in mind this object variable and its path are referenced by pyproject.toml so the Splunk SOAR platform knows where the app instance is provided.

Action Definitions

Actions are defined as standalone functions, with a few important rules and recommendations.

Action Metadata

Action definition carry with them important metadata which is used by the Splunk SOAR platform to present the action in the UI, and to generate the app’s manifest. Often, this metadata can be derived automatically from the action function’s signature:

• The action’s “identifier” is, by default, the name of the action function (e.g. my_action).

• The action’s “name” is, by default, the action function’s name with spaces instead of underscores (e.g. my action).

• The action’s “description” is, by default, the action function’s docstring.

• The action’s “type” is, by default, generic unless the action is one of the reserved names like test connectivity or on poll.

Note

By convention, action names should be lowercase, with 2-3 words. Keep action names short but descriptive, and avoid using the name of the app or external service in action names. Where feasible, it’s recommended to consider reusing action names across different apps (e.g. get email) to provide a more consistent user experience.

Action Arguments

There is a magic element, similar to pytest fixtures, in the action arguments. The type hints for the argument definitions of an action function are critical to this mechanism. The rules are as follows:

• The first positional argument of an action function must be the params argument, and its type hint must be a Pydantic model inheriting from Params. The position and type of this argument are required. The name params is a convention, but not strictly required.

• If an action function has any argument named soar, at runtime the SDK will provide an instance of a SOARClient implementation as that argument, which is already authenticated with Splunk SOAR. The type hint for this argument should be SOARClient.

• If an action function has any argument named asset, at runtime the SDK will provide an instance of the app’s asset class, populated with the asset configuration for the current action run. The type hint for this argument should be the app’s asset class.

Note

The special actions which define their own decorators have stricter rules about the type of the params argument. For example, the on poll action must take an OnPollParams instance as its params argument, and test connectivity must take no params argument at all.

Action Returns

An action’s return type annotation is critical for the Splunk SOAR platform to understand, via datapaths, what an action’s output looks like. In practice, this means that you must define a class inheriting from ActionOutput to represent the action’s output, and then return an instance of that class from your action function:

from soar_sdk.action_results import ActionOutput

class MyActionOutput(ActionOutput):
    field1: str
    field2: int

@app.action()
def my_action(params: MyActionParams) -> MyActionOutput:
    # action logic here
    return MyActionOutput(field1="value", field2=42)

Advanced Return Types

For more advanced use cases, an action’s return type can be a list, Iterator, or AsyncGenerator that yields multiple ActionOutput objects:

list

@app.action()
def my_action_list(params: MyActionParams) -> list[MyActionOutput]:
    # action logic here
    return [
        MyActionOutput(field1="value1", field2=1),
        MyActionOutput(field1="value2", field2=2)
    ]

Iterator

from typing import Iterator

@app.action()
def my_action_iterator(params: MyActionParams) -> Iterator[MyActionOutput]:
    # action logic here
    yield MyActionOutput(field1="value1", field2=1)
    yield MyActionOutput(field1="value2", field2=2)

AsyncGenerator

from typing import AsyncGenerator

@app.action()
async def my_action_async_generator(
    params: MyActionParams,
    asset: Asset,
) -> AsyncGenerator[MyActionOutput]:
    async with client = httpx.AsyncClient() as client:
        async for i in range(10):
            response = await client.get(
                f"{asset.base_url}/data",
                params={"page": i}
            )
            yield MyActionOutput(**response.json())

test connectivity Action

Test connectivity action definition

@app.test_connectivity()
def test_connectivity(soar: SOARClient, asset: Asset) -> None:
    soar.get("rest/version")
    container_id = soar.get_executing_container_id()
    logger.info(f"current executing container's container_id is: {container_id}")
    asset_id = soar.get_asset_id()
    logger.info(f"current executing container's asset_id is: {asset_id}")
    logger.info(f"testing connectivity against {asset.base_url}")
    logger.debug("hello")
    logger.warning("this is a warning")
    logger.progress("this is a progress message")

All apps must register exactly one test connectivity action in order to be considered valid by Splunk SOAR. This action takes no parameters, and is used to verify that the app and its associated asset configuration are working correctly. Running test connectivity on the Splunk SOAR platform should answer the questions:

• Can the app connect to the external service?

• Can the app authenticate with the external service?

• Does the app have the necessary permissions to perform its actions?

A successful test connectivity action should return None, and a failure should raise an ActionFailure with a descriptive error message.

on poll Action

on poll action definition

@app.on_poll()
def on_poll(
    params: OnPollParams, soar: SOARClient, asset: Asset
) -> Iterator[Container | Artifact]:
    if params.is_manual_poll():
        logger.info("Manual poll (poll now) detected")
    else:
        logger.info("Scheduled poll detected")

    # Create container first for artifacts
    yield Container(
        name="Network Alerts",
        description="Some network-related alerts",
        severity="medium",
    )

    # Simulate collecting 2 network artifacts that will be put in the network alerts container
    for i in range(1, 3):
        logger.info(f"Processing network artifact {i}")

        alert_id = f"testalert-{datetime.now(UTC).strftime('%Y%m%d')}-{i}"
        artifact = Artifact(
            name=f"Network Alert {i}",
            label="alert",
            severity="medium",
            source_data_identifier=alert_id,
            type="network",
            description=f"Example network alert {i} from polling operation",
            data={
                "alert_id": alert_id,
                "source_ip": f"10.0.0.{i}",
                "destination_ip": "192.168.0.1",
                "protocol": "TCP",
            },
        )

        yield artifact

on poll is another special action that apps may choose to implement. This action always takes an OnPollParams instance as its parameter. If defined, this action will be called in order to ingest new data into the Splunk SOAR platform. The action should yield Container and/or Artifact instances representing the new data to be ingested. The SDK will handle actually creating the containers and artifacts in the platform.

Make Request Action

Make request action definition

@app.make_request()
def http_action(params: MakeRequestParamsCustom, asset: Asset) -> MakeRequestOutput:
    logger.info(f"HTTP action triggered with params: {params}")
    return MakeRequestOutput(
        status_code=200,
        response_body=f"Base url is {asset.base_url}",
    )

Apps may define a special “make request” action, which can be used to interact with the underlying external service’s REST API directly. Having this action available can be useful when there are parts of the REST API that don’t have dedicated actions implemented in the app.

We create an action by decorating a function with the app.action decorator. The default action_type is generic, so usually you will not have to provide this argument for the decorator. This is not the case for the test action type though, so we provide this type here explicitly.

Custom Actions

Actions can be registered one of two ways:

@app.action()

Using the action() decorator to decorate a standalone function.

decorated action definition

@app.action(summary_type=GeneratorActionSummary)
def generator_action(
    params: Params, soar: SOARClient[GeneratorActionSummary], asset: Asset
) -> Iterator[GeneratorActionOutput]:
    """Generates a sequence of numbers."""
    logger.info(f"Generator action triggered with params: {params}")
    for i in range(5):
        yield GeneratorActionOutput(iteration=i)
    soar.set_summary(GeneratorActionSummary(total_iterations=5))

app.register_action()

Using the register_action() method to register a function which may be defined in another module.

The two methods are functionally equivalent. The decorator method is often more convenient for simple actions, while the registration method may be preferable for larger apps where actions are defined in separate modules. Apps may use either or both methods to register their actions.

App CLI Invocation

App CLI invocation

if __name__ == "__main__":
    app.cli()

A generic invocation to the app’s cli() method, which enables running the app actions directly from command line. The app template created by soarapps init includes this snippet by default, and it is recommended to keep it in order to facilitate local testing and debugging of your app actions.