Skip to content

Dynamic Aliases

Dynamic aliases are script-defined commands. They appear alongside built-in commands in the Tuoni UI and can validate input, queue child commands, combine results, and update server-side data.

Alias contract

An alias implements four methods:

Method Result Purpose
configuration_schema() str Return raw JSON Schema text for the alias input
can_send_to_agent(agent) bool Report whether the alias supports an agent
validate_config(config, agent) None Raise tse.ValidationError for invalid input
execute(ctx, config, agent) None Run the alias and finish or fail its context

The same alias instance can execute concurrently. Do not store per-execution state on the instance unless access to it is thread-safe.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import tuoni_script_engine as tse


class MyAlias:
    name = "my-alias"

    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):
        ctx.set_result({"STDOUT": "Completed"})
        ctx.finish()


tse.commands.register_dynamic_alias(MyAlias())

Register an instance, not the class. An alias without a name property can be registered with an explicit name:

1
tse.commands.register_dynamic_alias("my-alias", MyAlias())

Configuration schemas

configuration_schema() returns JSON Schema text. Tuoni also recognizes files for named file inputs and positional for terminal arguments.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def configuration_schema(self):
    return """{
        "$schema": "https://json-schema.org/draft/2020-12/schema",
        "type": "object",
        "properties": {
            "target": {
                "type": "string",
                "description": "Target address"
            },
            "port": {
                "type": "integer",
                "minimum": 1,
                "maximum": 65535
            }
        },
        "required": ["target"],
        "files": {
            "payload": {}
        },
        "positional": {
            "target": {"position": 0, "required": true},
            "port": {"position": 1, "required": false}
        }
    }"""

The alias receives one of the public configuration types. JSON-only input is a JsonConfiguration; input with named files is a MultipartConfiguration.

1
2
3
4
5
6
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 {}

Use tse.ValidationError for expected validation failures:

1
2
3
4
def validate_config(self, config, agent):
    values = json_values(config)
    if not values.get("target"):
        raise tse.ValidationError("target is required")

Agent compatibility

Use typed properties and enums instead of string comparisons:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
def can_send_to_agent(self, agent):
    return (
        agent.status is tse.AgentStatus.ACTIVE
        and agent.type is tse.AgentType.SHELLCODE_AGENT
        and agent.metadata.operating_system is tse.OperatingSystem.MAC
        and agent.metadata.os_architecture in {
            tse.Architecture.X64,
            tse.Architecture.ARM64,
        }
    )

Queueing commands

AliasContext.queue_command() accepts a command template name or object and an explicit configuration:

1
2
3
4
command = ctx.queue_command(
    "sh",
    tse.JsonConfiguration({"command": "id"}),
)

For a command with file input, use MultipartConfiguration:

1
2
3
4
5
6
7
command = ctx.queue_command(
    "upload",
    tse.MultipartConfiguration(
        json={"filepath": "/tmp/tool.bin"},
        files={"file": b"file contents"},
    ),
)

Pass execution settings through the keyword-only exec_config argument:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
command = ctx.queue_command(
    "portscan",
    tse.JsonConfiguration({
        "ips": ["10.0.0.1"],
        "ports": [22, 443],
    }),
    exec_config=tse.SelfExecutionConfiguration(
        exec_unit_type=tse.ExecUnitType.NATIVE_LIB,
    ),
)

Available execution configurations are:

  • SelfExecutionConfiguration
  • NewExecutionConfiguration
  • ExistingExecutionConfiguration

When exec_config is omitted, Tuoni chooses the normal agent context and a compatible ExecUnit format.

Waiting for and reading results

wait_for_completion() returns False only when a finite timeout expires. After a final result, inspect command.result.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def execute(self, ctx, config, agent):
    command = ctx.queue_command(
        "sh",
        tse.JsonConfiguration({"command": "whoami"}),
    )

    if not command.wait_for_completion(timeout=30):
        command.cancel()
        ctx.fail("The child command timed out")
        return

    if command.result is None:
        ctx.fail("The child command returned no result")
        return

    if command.is_failed():
        ctx.fail(command.result.error_message or "The child command failed")
        return

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

Result entry values can be str, bytes, or tse.File. Reuse a returned file directly in another multipart configuration:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
returned_file = next(
    value
    for value in command.result.entries.values()
    if isinstance(value, tse.File)
)

upload = ctx.queue_command(
    "upload",
    tse.MultipartConfiguration(
        json={"filepath": "/tmp/copied-file"},
        files={"file": returned_file},
    ),
)

Completing an alias

Every execution path must call exactly one of:

  • ctx.finish() to complete successfully
  • ctx.fail(message) to complete with an error

ctx.set_result() may be called more than once to publish progress, but it does not finish the alias.

1
2
3
ctx.set_result({"status": "Halfway complete"})
ctx.set_result({"status": "Complete", "report": b"result bytes"})
ctx.finish()

Returning without finishing or failing leaves the alias ongoing.

Updating metadata

Metadata properties and custom-property mappings are read-only views. Use the provided update methods:

1
2
3
4
5
6
7
8
properties = agent.metadata.custom_properties
notes = str(properties.get("notes", ""))

agent.metadata.update_custom_properties({
    "notes": f"{notes};#reviewed" if notes else "#reviewed",
})
ctx.set_result({"STDOUT": "Agent tagged"})
ctx.finish()

Use replace_custom_properties() to replace all custom values and remove_custom_property(name) to remove one.

Error handling

Raise tse.ValidationError only for expected input errors during validation. Catch operational errors when the alias can add useful context; otherwise Tuoni treats an exception from execute() as a failed parent command.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
def execute(self, ctx, config, agent):
    try:
        command = ctx.queue_command(
            "ps",
            tse.JsonConfiguration({}),
        )
        command.wait_for_completion()
        if command.result is None or not command.is_success():
            message = (
                command.result.error_message
                if command.result is not None
                else "No result"
            )
            ctx.fail(message or "Process listing failed")
            return
        ctx.set_result(command.result.entries)
        ctx.finish()
    except tse.TuoniError as error:
        ctx.fail(f"Tuoni API error: {error}")