Tuoni 0.15.0: Tuoni Script Engine

Tuoni has included the experimental Tuoni Script Engine for some time. In 0.15.0, the engine is enabled by default and the API has been expanded to cover more of the normal operator workflow.

Scripts run in GraalPy inside the Tuoni server. Put a Python file in the configured scripts directory (/srv/tuoni/data/scripts by default) and Tuoni loads it, reports its logs, and reloads it when the file changes. New scripts should import tuoni_script_engine. The older tuoni module is deprecated, but remains available so existing scripts continue to run.

Setup

Tuoni provides the runtime module inside the server. For editor completion, install the stub package locally with Python 3.12 or newer.

python -m pip install tuoni-script-engine-stubs

If you would also like to verify the types, install and run Pyright:

python -m pip install pyright
python -m pyright /srv/tuoni/data/scripts/example.py

React to new agents

Scripts can subscribe to Tuoni events and act on the referenced resources. This example watches for newly registered Windows agents, checks that they registered through an HTTP listener, and changes their sleep time to ten seconds.

import tuoni_script_engine as tse

HTTP_PLUGIN_ID = "shelldot.listener.agent-reverse-http"

def send_sleep_to_windows_agent(agent):
    is_windows = (
        agent.metadata.operating_system is tse.OperatingSystem.WINDOWS
    )
    uses_http = any(
        listener.plugin_id == HTTP_PLUGIN_ID
        for listener in agent.listeners
    )
    if is_windows and uses_http:
        agent.queue_command(
            "sleep",
            tse.JsonConfiguration({"sleep": 10, "sleepRandom": 0}),
        )

def on_register_agents(agents):
    for agent in agents:
        send_sleep_to_windows_agent(agent)

tse.events.subscribe_by_event_type(
    tse.EventType.REGISTER_AGENT,
    lambda _event, agents: on_register_agents(agents),
)

Event callbacks receive the referenced agents directly. They can also update metadata, queue other command templates, or hand work to a background job.

Turn a BOF into a command

A script can register a dynamic alias that appears as a command template in Tuoni. The alias defines its configuration schema, limits the command to compatible agents, and maps the operator's input to another command. Here, dir.x64.o and dir.x86.o live in a dir directory next to the script.

import inspect
import json
import os

import tuoni_script_engine as tse

def get_bof_path():
    """Return the directory containing this script."""
    frame = inspect.currentframe()
    return os.path.dirname(os.path.abspath(inspect.getfile(frame)))

class DirAlias:
    description = "List a directory with a BOF"

    def configuration_schema(self):
        return json.dumps({
            "type": "object",
            "properties": {
                "directory": {
                    "type": "string",
                    "default": ".\\",
                },
                "recursive": {
                    "type": "boolean",
                    "default": False,
                },
            },
            "additionalProperties": False,
        })

    def can_send_to_agent(self, agent):
        return (
            agent.type is tse.AgentType.SHELLCODE_AGENT
            and agent.metadata.operating_system is tse.OperatingSystem.WINDOWS
        )

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        directory = config.to_dict().get("directory", ".\\")
        recursive = 1 if config.to_dict().get("recursive", False) else 0
        architecture = agent.metadata.process_architecture

        if architecture is None:
            ctx.fail("Agent architecture is unknown")
            return

        bof_path = os.path.join(
            get_bof_path(),
            "dir",
            f"dir.{architecture.lower()}.o",
        )
        ctx.set_result({"STDOUT": "[!] Sending BOF..."})

        with open(bof_path, "rb") as bof_file:
            command = ctx.queue_command(
                "bof",
                tse.MultipartConfiguration(
                    files={"bofFile": bof_file},
                    json={
                        "method": "go",
                        "pack_format": "zs",
                        "pack_args": [directory, recursive],
                    },
                ),
            )

        command.wait_for_completion()
        if command.is_failed():
            error = command.result.error_message if command.result else None
            ctx.fail(error or "BOF command failed")
            return

        if command.result:
            ctx.set_result(command.result.entries)
        ctx.finish()

tse.commands.register_dynamic_alias("dir", DirAlias())

The alias is available through the same GUI, terminal, and REST command flows as plugin-provided command templates. Its result is copied from the underlying bof command.

Save discovered data

Scripts can write findings directly to Tuoni's discovery records. This is useful when parsing command output, importing results from another tool, or enriching data from an internal service.

import tuoni_script_engine as tse

tse.discovery.services.create(
    "10.10.20.15",
    445,
    protocol="tcp",
    banner="SMB",
    note="Found during subnet scan",
)

tse.discovery.credentials.create(
    "svc_backup",
    "example-password",
    host="10.10.20.15",
    realm="LAB",
    source="Imported by audit script",
)

The discovery API also supports hosts, searches, updates, archiving, and restoring. Records created by a script use the same persistence and event history as records created through the REST API or a plugin.

More than one-off scripts

The 0.15.0 API also covers files, events, jobs, listeners, payloads, plugins, and settings. This makes it possible to keep small automation close to the server without writing and packaging a Java plugin.