Examples

Complete, runnable examples for common server-side scripting tasks. Each example can be saved as a .py file in the scripts directory.

---

Hello-Bye: Chaining Commands

A minimal alias that runs two shell commands and combines their output into a single result.

import tuoni


class HelloByeAlias:

    def __init__(self):
        self.name = "hello-bye"
        self.description = "Run hello and bye commands, combine output"

    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.type == "SHELLCODE_AGENT" and agent.get_latest_metadata().os == "LINUX"

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        # First command
        hello_cmd = ctx.queue_command("sh", {"command": "echo hello"})
        hello_cmd.wait_for_finish()
        if hello_cmd.is_failed():
            ctx.fail(hello_cmd.get_error_message())
            return

        # Show intermediate result
        ctx.set_result(hello_cmd.get_result())

        # Second command
        bye_cmd = ctx.queue_command("sh", {"command": "echo bye"})
        bye_cmd.wait_for_finish()
        if bye_cmd.is_failed():
            ctx.fail(bye_cmd.get_error_message())
            return

        # Combine results
        ctx.set_result({
            "STDOUT": hello_cmd.get_result()["STDOUT"] + "\n" + bye_cmd.get_result()["STDOUT"],
        })
        ctx.finish()


tuoni.commands.register_dynamic_alias(HelloByeAlias())

Key Patterns

• ctx.set_result() can be called multiple times — each call updates the visible result
• Chain commands sequentially with wait_for_finish() between them
• Always check is_failed() and propagate errors with ctx.fail()

---

Tag / Untag / Note: Metadata Manipulation

Three aliases in a single script that manage agent tags via custom_properties. These aliases do not send any commands to the agent — they only modify server-side metadata.

import tuoni


class TagAlias:

    def __init__(self):
        self.name = "tag"
        self.description = "Add a tag to the agent"

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

    def can_send_to_agent(self, agent):
        return True

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        tag = config.json.as_dict["tag"]
        notes = agent.metadata.custom_properties.get("notes", "")

        tags = notes.split(";")
        if f"#{tag}" not in tags:
            agent.metadata.custom_properties["notes"] = (
                f"{notes};#{tag}" if notes else f"#{tag}"
            )
            ctx.set_result({"STDOUT": "Tag added"})
        else:
            ctx.set_result({"STDOUT": "Tag already exists"})
        ctx.finish()


class UntagAlias:

    def __init__(self):
        self.name = "untag"
        self.description = "Remove a tag from the agent"

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

    def can_send_to_agent(self, agent):
        return True

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        tag = config.json.as_dict["tag"]
        notes = agent.metadata.custom_properties.get("notes", "")

        tags = notes.split(";")
        if f"#{tag}" not in tags:
            ctx.set_result({"STDOUT": "No such tag found"})
        else:
            tags.remove(f"#{tag}")
            agent.metadata.custom_properties["notes"] = ";".join(tags)
            ctx.set_result({"STDOUT": "Tag removed"})
        ctx.finish()


class NoteAlias:

    def __init__(self):
        self.name = "note"
        self.description = "Set a note on the agent"

    def configuration_schema(self):
        return """{
            "$schema": "https://json-schema.org/draft/2020-12/schema",
            "type": "object",
            "properties": {
                "note": {
                    "type": "string",
                    "description": "Overwrites the agent's note"
                }
            },
            "required": ["note"],
            "positional": {
                "note": { "position": 0, "required": true }
            }
        }"""

    def can_send_to_agent(self, agent):
        return True

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        new_note = config.json.as_dict["note"]
        agent.metadata.custom_properties["notes"] = new_note
        ctx.set_result({"STDOUT": "Noted"})
        ctx.finish()


tuoni.commands.register_dynamic_alias(TagAlias())
tuoni.commands.register_dynamic_alias(UntagAlias())
tuoni.commands.register_dynamic_alias(NoteAlias())

Key Patterns

• custom_properties changes persist immediately — no save call needed
• can_send_to_agent returns True for all agents since no agent-side execution is required
• Multiple aliases can be registered from a single script file
• positional schema allows terminal usage like tag mytag instead of JSON input

---

Process Find & Who: Post-Processing Results

Two aliases that run the built-in ps command and post-process the JSON output.

import tuoni
import json


class PSFindAlias:

    def __init__(self):
        self.name = "psfind"
        self.description = "Search for a process by name or PID"

    def configuration_schema(self):
        return """{
            "$schema": "https://json-schema.org/draft/2020-12/schema",
            "type": "object",
            "properties": {
                "search": {
                    "type": "string",
                    "description": "PID or process name to search for. Defaults to explorer.exe"
                }
            },
            "required": [],
            "positional": {
                "search": { "position": 0, "required": false }
            }
        }"""

    def can_send_to_agent(self, agent):
        return True

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        ps_cmd = ctx.queue_command("ps", {})
        ps_cmd.wait_for_finish()
        if ps_cmd.is_failed():
            ctx.fail(ps_cmd.get_error_message())
            return

        config_dict = config.json.as_dict
        search_str = config_dict.get("search", "explorer.exe")

        # Determine if searching by PID or name
        try:
            int(search_str)
            search_field = "pid"
        except ValueError:
            search_field = "name"

        results = []
        for proc in json.loads(ps_cmd.get_result()["JSON"]):
            if search_field == "pid" and str(search_str) == str(proc["pid"]):
                results.append(self._format_proc(proc))
            elif search_field == "name" and search_str in proc["name"]:
                results.append(self._format_proc(proc))

        ctx.set_result({"STDOUT": "\n".join(results) if results else "No matches found"})
        ctx.finish()

    def _format_proc(self, proc):
        return (
            f"{proc['pid']}: {proc['name']}, "
            f"username: {proc['username']}, "
            f"arch: {proc['architecture']}, "
            f"ppid: {proc['ppid']}"
        )


class PSWhoAlias:

    def __init__(self):
        self.name = "pswho"
        self.description = "List unique usernames from running processes"

    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 True

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        ps_cmd = ctx.queue_command("ps", {})
        ps_cmd.wait_for_finish()
        if ps_cmd.is_failed():
            ctx.fail(ps_cmd.get_error_message())
            return

        users = set()
        for proc in json.loads(ps_cmd.get_result()["JSON"]):
            if proc["username"] != "???":
                users.add(proc["username"])

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


tuoni.commands.register_dynamic_alias(PSFindAlias())
tuoni.commands.register_dynamic_alias(PSWhoAlias())

Key Patterns

• The ps command returns JSON in a result key called "JSON" — parse it with json.loads()
• Optional positional parameters with defaults
• Private helper methods on the alias class work fine

---

BOF Execution: Whoami via Beacon Object File

Execute a Windows Beacon Object File (BOF) from a script-relative path.

import tuoni
import os
import inspect


class WhoamiBofAlias:

    def __init__(self):
        self.name = "whoami-bof"
        self.description = "Run whoami BOF on Windows agent"

    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.type == "SHELLCODE_AGENT"
            and agent.get_latest_metadata().os == "WINDOWS"
        )

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        # Resolve the BOF file relative to this script
        script_path = inspect.getfile(inspect.currentframe())
        script_dir = os.path.dirname(os.path.abspath(script_path))
        bof_path = os.path.join(script_dir, "whoami.bof")

        ctx.set_result({"STDOUT": "[*] Sending BOF..."})

        bof_cmd = ctx.queue_command("bof", {
            "json": {
                "method": "go",
                "inputAsBytes": b""
            },
            "files": {
                "bofFile": open(bof_path, "rb")
            }
        })
        bof_cmd.wait_for_finish()

        if bof_cmd.is_failed():
            ctx.fail(bof_cmd.get_error_message())
            return

        ctx.set_result(bof_cmd.get_result())
        ctx.finish()


tuoni.commands.register_dynamic_alias(WhoamiBofAlias())

Key Patterns

• Use inspect.getfile(inspect.currentframe()) to get the script's own path
• Place BOF files alongside the script in the scripts directory
• The "bof" command expects "bofFile" in the files configuration
• "method" is the BOF entry point (typically "go")
• "inputAsBytes" passes binary arguments to the BOF

---

Download & Upload: File Transfer Pipeline

Download a file from an agent, then upload it back with a different name. Demonstrates file handling between commands.

import tuoni


class DownloadUploadAlias:

    def __init__(self):
        self.name = "download-upload"
        self.description = "Download a file and re-upload it with .new extension"

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

    def can_send_to_agent(self, agent):
        return agent.type == "SHELLCODE_AGENT"

    def validate_config(self, config, agent):
        pass

    def execute(self, ctx, config, agent):
        filepath = config.json.as_dict["path"]

        # Step 1: Download the file
        download_cmd = ctx.queue_command("download", {"filepath": filepath})
        download_cmd.wait_for_finish()
        if download_cmd.is_failed():
            ctx.fail(download_cmd.get_error_message())
            return

        # Get the downloaded file from results
        result_file = list(download_cmd.get_result().items())[0][1]

        # Step 2: Upload the same file with a new name
        # Using file_id avoids re-uploading the bytes — it references the
        # file already stored in Tuoni's file store
        upload_cmd = ctx.queue_command("upload", {
            "json": {"filepath": filepath + ".new"},
            "files": {
                "file": {"file_id": result_file.file_id}
            }
        })
        upload_cmd.wait_for_finish()
        if upload_cmd.is_failed():
            ctx.fail(upload_cmd.get_error_message())
            return

        # Return the downloaded file as the alias result
        ctx.set_result({result_file.file_name: result_file})
        ctx.finish()


tuoni.commands.register_dynamic_alias(DownloadUploadAlias())

File Passing Alternatives

There are several ways to pass a file between commands:

# Reference by file_id (most efficient — no re-upload)
"file": {"file_id": result_file.file_id}

# Pass the File object directly (also uses file_id internally)
"file": result_file

# Read and re-upload the bytes
"file": result_file.open(mode="rb").read()

# Read via vfs_path
"file": io.BytesIO(open(result_file.vfs_path, mode="rb").read())

---

Nmap: External Tool Integration with Discovery

Run nmap on the agent and automatically add discovered services to Tuoni's discovery database.

import tuoni


class NmapAlias:

    def __init__(self):
        self.name = "nmap"
        self.description = "Run nmap scan and add results to discovery"

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

    def can_send_to_agent(self, agent):
        return (
            agent.type == "SHELLCODE_AGENT"
            and agent.get_latest_metadata().os == "LINUX"
        )

    def validate_config(self, config, agent):
        if "addresses" not in config.json.as_dict:
            raise Exception("addresses is required")

    def execute(self, ctx, config, agent):
        cfg = config.json.as_dict
        ports = cfg.get("ports", [])
        ports_part = f"-p{','.join(ports)}" if ports else ""
        addresses_part = " ".join(cfg["addresses"])

        # Run nmap via shell command
        sh_cmd = ctx.queue_command(
            "shelldot.commands.native:global:sh",
            {"json": {"command": f"nmap {ports_part} {addresses_part}".strip()}}
        )
        sh_cmd.wait_for_finish()

        if sh_cmd.is_failed():
            ctx.fail(sh_cmd.get_error_message())
            return

        # Parse nmap output and add to discovery
        stdout = sh_cmd.get_result().get("STDOUT", "")
        results = []

        for block in stdout.split("Nmap scan report for")[1:]:
            lines = block.splitlines()
            addr_line = lines[0]
            address = (
                addr_line.split("(")[-1].rstrip(")")
                if "(" in addr_line
                else addr_line.strip()
            )

            for line in lines:
                if "open" in line:
                    parts = line.split()
                    port, protocol = parts[0].split("/")
                    service = parts[2]
                    results.append(f"{address}:{port}")

                    tuoni.discovery.services.add(
                        address=address,
                        port=int(port),
                        protocol=protocol,
                        banner=service,
                        note="discovered via nmap script"
                    )

        ctx.set_result({
            "nmap": f"Found {len(results)} open ports: {results}"
        })
        ctx.finish()


tuoni.commands.register_dynamic_alias(NmapAlias())

Key Patterns

• Use the fully-qualified command template ID "shelldot.commands.native:global:sh" for the shell command (short form "sh" also works)
• tuoni.discovery.services.add() is available globally, not just inside alias execute() methods
• Parse stdout text to extract structured data from external tools

---

Auto-Tag on Agent Connect: Event-Driven Automation

Automatically tag agents based on their IP address when they first connect. This script runs indefinitely, listening for new agent registrations.

import tuoni


def handle_new_agent(event, agent):
    """Auto-tag agents based on their IP subnet."""
    print(f"New agent registered: {agent.guid}")

    # Mark agent as auto-processed
    agent.metadata.custom_properties["auto_tagged"] = True

    ips = agent.metadata.ips or ""
    for ip in ips.split():
        octets = ip.split(".")
        if len(octets) != 4:
            continue
        try:
            third_octet = int(octets[2])
            team_tag = f"#Team{third_octet:02d}"

            notes = agent.metadata.custom_properties.get("notes", "")
            tags = notes.split(";")
            if team_tag not in tags:
                agent.metadata.custom_properties["notes"] = (
                    f"{notes};{team_tag}" if notes else team_tag
                )

            agent.metadata.custom_properties["team"] = third_octet
            print(f"Tagged agent {agent.guid} as {team_tag} (from IP: {ip})")
            break
        except ValueError:
            continue


# Register the event listener — script stays alive while this is active
event_task = tuoni.events.register_callback_by_event_type(
    "REGISTER_AGENT", handle_new_agent
)
print("Auto-tagging script loaded. Listening for new agents...")

Long-Running Script

This script never calls event_task.stop(), so it runs indefinitely. It will be stopped when the script file is deleted, modified (triggering a restart), or the server shuts down.

---

Direct Command Sending: Automated Reconnaissance

Send commands to agents from event handlers without going through an alias. Uses tuoni.commands.queue_command() instead of ctx.queue_command().

import tuoni


def recon_new_agent(event, agent):
    """Run basic recon on every new agent."""
    print(f"New agent: {agent.guid}, running recon...")

    # Send a shell command directly (not through an alias)
    cmd = tuoni.commands.queue_command(
        agent.guid,
        "sh",
        {"command": "id && hostname && cat /etc/os-release 2>/dev/null || systeminfo"}
    )
    cmd.wait_for_finish()

    if cmd.is_success():
        stdout = cmd.get_result().get("STDOUT", "")
        print(f"Recon result for {agent.guid}:\n{stdout}")

        # Add the host to discovery
        hostname = agent.metadata.hostname or "unknown"
        tuoni.discovery.hosts.add(
            address=agent.metadata.ips.split()[0] if agent.metadata.ips else "unknown",
            name=hostname,
            note=f"auto-discovered via agent {agent.guid}"
        )
    else:
        print(f"Recon failed for {agent.guid}: {cmd.get_error_message()}")


task = tuoni.events.register_callback_by_event_type("REGISTER_AGENT", recon_new_agent)
print("Automated recon script active.")

tuoni.commands.queue_command vs ctx.queue_command

• tuoni.commands.queue_command(agent_guid, template, config) — Use anywhere in a script (event handlers, top-level code). Requires the agent GUID as the first argument.
• ctx.queue_command(template, config) — Only available inside an alias execute() method. The agent is implicit from the alias context.