Skip to content

API Reference

Import the Tuoni 0.15.0 scripting API as:

1
import tuoni_script_engine as tse

The tuoni-script-engine-stubs package is the exhaustive source for signatures and return types. This page covers the interfaces most commonly used by scripts.

Module entry points

Object Purpose
tse.agents Query agents and change their status
tse.commands Queue, inspect, cancel, and extend commands
tse.command_templates Query command templates
tse.command_aliases Create and query static aliases
tse.discovery Manage discovered hosts, services, and credentials
tse.events Subscribe to and query audit events
tse.files Upload and query Tuoni files
tse.listeners Create and manage listeners
tse.payloads Create and generate payloads
tse.payload_templates Query payload templates
tse.plugins Query loaded plugins
tse.jobs Query jobs and invoke supported actions
tse.settings Read server and plugin settings
tse.ips List addresses on which Tuoni can be reached

Enums

The API uses string enums instead of untyped string constants. Important values include:

Enum Values
AgentStatus ACTIVE, INACTIVE, BLOCKED
AgentType SHELLCODE_AGENT, GENERIC_SHELL_AGENT, EXTERNAL_AGENT
OperatingSystem WINDOWS, LINUX, BSD, MAC
Architecture X86, X64, ARM32, ARM64
ExecUnitType SHELLCODE_NATIVE, NATIVE_LIB, DOTNET_DLL, DOTNET_EXE
CommandStatus CREATED, SENT, ONGOING, CANCELED, FAILED, COMPLETE
CommandResultStatus FAILED, ONGOING, SUCCESS, UNKNOWN

Compare enum values directly:

1
2
if agent.metadata.operating_system is tse.OperatingSystem.MAC:
    print("macOS agent")

Configuration

Commands, listeners, and payloads accept one of three explicit configuration types.

JsonConfiguration

1
2
3
4
5
6
7
config = tse.JsonConfiguration({
    "command": "whoami",
    "timeout": 30,
})

print(config.to_dict())
print(config.to_json())

BinaryConfiguration

1
2
config = tse.BinaryConfiguration(b"\x00\x01\x02")
print(config.bytes)

MultipartConfiguration

Use multipart configuration for a JSON object accompanied by named files:

1
2
3
4
config = tse.MultipartConfiguration(
    json={"filepath": "/tmp/tool.bin"},
    files={"file": b"file contents"},
)

File values may be text, bytes-like data, an existing tse.File, or a readable binary stream. A returned multipart configuration exposes .json and .files.

Agents

tse.agents

Method Description
list(...) Iterate over agents, optionally filtering by status, OS, architecture, listener, or registration/callback time
get(guid) Return one agent or None
bulk_update_status(guids, status) Change several agents atomically 
1
2
3
4
5
for agent in tse.agents.list(
    status=tse.AgentStatus.ACTIVE,
    metadata_os=tse.OperatingSystem.LINUX,
):
    print(agent.guid, agent.metadata.hostname)

Agent

Important properties are guid, type, metadata, status, and listeners. Use methods for state-changing operations:

Method Description
update_status(status) Set ACTIVE, INACTIVE, or BLOCKED
commands() Iterate over commands sent to this agent
available_command_templates() Return templates compatible with this agent
queue_command(template, config, *, exec_config=None) Queue a command
clear_command_queue() Cancel commands that have not been delivered

AgentMetadata

Metadata properties are read-only views. Update them with update(...) or the matching update_*() method.

Property Type
username, process_name, working_directory, hostname, integrity str \| None
pid, os_major, os_minor, ansi_code_page, agent_version int \| None
operating_system OperatingSystem \| None
process_architecture, os_architecture Architecture \| None
reported_addresses_raw, reported_agent_type str \| None
features Sequence[str]
payload_id int \| None
listener_properties, custom_properties read-only mappings 
1
2
3
4
5
6
7
8
metadata = agent.metadata
metadata.update_hostname("workstation-01")
metadata.update({
    "operating_system": tse.OperatingSystem.MAC,
    "os_architecture": tse.Architecture.ARM64,
})
metadata.update_custom_properties({"team": "red", "reviewed": True})
metadata.remove_custom_property("obsolete")

replace_custom_properties() replaces the entire custom-property mapping; update_custom_properties() merges keys.

Commands

tse.commands

Method Description
register_dynamic_alias(alias) Register an alias that has a name property
register_dynamic_alias(name, alias) Register an alias under an explicit name
get(command_id) Return one command or None
list() Iterate over visible commands
cancel(command_id) Cancel and return a command
queue_command(agent_guid, template, config, *, exec_config=None) Queue a command
queue_update(command_id, config) Send a configuration update 
1
2
3
4
5
6
command = tse.commands.queue_command(
    agent.guid,
    "ps",
    tse.JsonConfiguration({}),
)
command.wait_for_completion()

Inside a dynamic alias, use ctx.queue_command() instead; the alias context already identifies the agent.

Command

Property or method Description
id, command_template_id, agent_guid Command identifiers
status Current CommandStatus
result Current CommandResult or None
wait_for_completion(timeout=None) Wait for a final state; return False if a finite timeout expires
is_failed(), is_ongoing(), is_success() Inspect the current result state
cancel() Cancel this command
queue_update(config) Send a command update

A result exposes status, error_message, receive_time, and entries. Entries are a read-only mapping whose values are text, bytes, or tse.File objects.

1
2
3
4
5
6
if not command.wait_for_completion(timeout=30):
    command.cancel()
elif command.is_success() and command.result is not None:
    print(command.result.entries)
elif command.result is not None:
    print(command.result.error_message or "Command failed")

Execution configurations

Pass execution configuration with the exec_config keyword:

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

Available classes are:

  • SelfExecutionConfiguration(exec_unit_type=None)
  • NewExecutionConfiguration(executable=None, suspended=None, ppid=None, username=None, password=None, exec_unit_type=None)
  • ExistingExecutionConfiguration(pid=None, exec_unit_type=None)

Dynamic aliases

An alias implements:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
class DynamicAlias:
    def configuration_schema(self):
        ...

    def can_send_to_agent(self, agent):
        ...

    def validate_config(self, config, agent):
        ...

    def execute(self, ctx, config, agent):
        ...

AliasContext provides:

Method Description
queue_command(template, config, *, exec_config=None) Queue a child command
set_result(entries) Publish ongoing result entries
finish() Complete the alias successfully
fail(message) Complete the alias as failed

Every execution path must eventually call finish() or fail(). See Dynamic aliases for a complete example.

Discovery

Each discovery collection provides list(), search(), get(), archive_many(), and restore_many().

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
host = tse.discovery.hosts.create(
    "192.168.1.10",
    name="web-server",
    note="created by a script",
)

service = tse.discovery.services.create(
    "192.168.1.10",
    443,
    protocol="tcp",
    banner="HTTPS",
)

credential = tse.discovery.credentials.create(
    "operator",
    "secret",
    host="192.168.1.10",
    source="script",
)

The returned objects expose update methods for editable fields plus archive() and restore().

Events

Subscribe with an EventType or ReferenceType. Callbacks receive an Event and a sequence of referenced objects.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def on_agents(event, agents):
    for registered_agent in agents:
        print(event.time, registered_agent.guid)


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

# Stop receiving callbacks later:
# task.stop()

Use events.get(event_id) for one event and events.get_by_reference(reference_type, reference_id) for an item's history. See Event system for event categories and lifecycle details.

Files and other managers

Files can be reused without reading and re-uploading their bytes:

1
2
3
4
5
stored = tse.files.upload(b"report contents", name="report.txt")
print(stored.id, stored.name, stored.size)

with stored.open("rb") as stream:
    print(stream.read())

The remaining managers follow the same query-object-action pattern:

  • listeners.list/get/create; a Listener can be renamed, reconfigured, started, stopped, or deleted.
  • payloads.list/get/create; a Payload can be renamed, generated, or archived.
  • command_templates, payload_templates, and plugins expose schemas, examples, support status, and plugin metadata.
  • jobs.list/get returns jobs whose supported actions can be invoked.
  • settings.list/get is read-only.