TuoniC2¶

Authentication and migration¶

TuoniC2() accepts self-signed certificates by default (verify=False), preserving compatibility with existing local-server scripts. Enable verification with verify=True for the standard trust store or verify="path/to/ca.pem" for a private CA. REST and WebSocket connections share this setting. HTTP connection/read timeouts default to timeout=(10, 60) seconds and can be set in the constructor.

login() requires v2 authentication. Access tokens refresh automatically; refresh() also allows an explicit rotation. Refreshes are serialized across threads, but the refresh-session deadline does not move. A rejected or ambiguous refresh requires login again: never retry the consumed token. Passwords are not stored. logout() clears local credentials and connections even if remote revocation fails; existing access JWTs remain valid until expiry on the server.

Read requests retry at most once after a 401 and refresh. Mutations and uploads raise their original 401 after refresh so callers can decide whether to retry. 403 responses do not trigger refresh. HTTP requests follow redirects using Requests’ standard redirect handling and limits. Exceptions expose HTTP status, method, URL, response details and requires_login; network and timeout errors have no HTTP status.

Command notes¶

add_command_note(agent_guid, command_id, label, content) and load_command_note(agent_guid, command_id, note_id) return tuoni.TuoniNote objects. load_command_notes() returns a single page as a list, with page=0, pageSize=100, label=None, and include_archived=False by default. Notes require the server’s v2 command-notes endpoints. See TuoniNote for editing, archival, permissions, and filtering behavior.

class tuoni.TuoniC2(verify: str | bool = False, timeout=(10, 60))[source]¶

The primary class for establishing connections to and managing interactions with the Tuoni server. It provides functionality for controlling server operations and facilitating communication.

Parameters:

verify (str | bool) – Defaults to False to accept self-signed certificates. True enables certificate verification; a string selects a CA bundle. The setting applies to both REST and WebSockets.
timeout (float | tuple[float, float]) – HTTP connection/read timeout in seconds. Defaults to (10, 60). A single number applies to both.

add_alias(name, description, command_type, command_conf=None, files=None)[source]¶

Add a new alias.

Parameters:

name (str) – The name of the alias to create.
description (str) – A description of the alias.
command_type (str | TuoniDefaultCommand) – The base command to associate with the alias.
command_conf (dict) – Configuration settings for the alias.
files (dict) – File parameters associated with the alias.

Returns:

The newly created alias.

Return type:

TuoniAlias

Examples

>>> alias1 = tuoni_server.add_alias(
>>>     "ls1",
>>>     "Alias made with python lib based on 'ls' default command class",
>>>     TuoniCommandLs,
>>>     {"depth": 1}
>>> )

>>> alias2 = tuoni_server.add_alias(
>>>     "ls2",
>>>     "Alias made with python lib based on command string name",
>>>     "ls",
>>>     {"depth": 2}
>>> )

>>> alias3 = tuoni_server.add_alias(
>>>     "easm",
>>>     "Alias for execute assembly default command class",
>>>     TuoniCommandexecuteAssembly,
>>>     {},
>>>     files = {
>>>         "executable": ["dotnet.exe",open("dotnet.exe", "rb").read()]
>>>     }
>>> )

>>> alias4 = tuoni_server.add_alias(
>>>     "bof1",
>>>     "Alias for bof based on command string name",
>>>     "bof",
>>>     files = {
>>>         "bofFile": ["bof.o",open("bof.o", "rb").read()]
>>>     }
>>> )

>>> alias5 = tuoni_server.add_alias(
>>>     "bof2",
>>>     "Alias for bof based on command ID",
>>>     "2ac58f33-d35e-4afd-a1ac-00e460ceb9f4",
>>>     files = {
>>>         "bofFile": ["bof.o",open("bof.o", "rb").read()]
>>>     }
>>> )

add_command_note(agent_guid, command_id, label, content)[source]¶

Create a note on a visible command. Requires SEND_COMMANDS.

Parameters:

agent_guid (str) – UUID of the command’s agent.
command_id (int) – Command identifier.
label (str) – Nonblank label, limited to 128 characters. The server trims surrounding whitespace.
content (str) – Nonblank content, limited to 2,000 characters. Whitespace, including newlines, is preserved.

Returns:

The created note, including its audit events.

Return type:

TuoniNote

add_datamodel_credential(username, password, host, realm, source, note)[source]¶

Add a credential data model entry.

Parameters:

username (str) – The username.
password (str) – The password.
host (str) – Host where credential works.
realm (str) – Credential realm.
source (str) – Source of the credential.
note (str) – Additional notes.

Returns:

The newly created credential.

Return type:

TuoniDataCredential

Examples

>>> credential = tuoni_server.add_datamodel_credential(
>>>     "rick",
>>>     "NeverGonnaBypassYourEDR",
>>>     "10.20.30.40",
>>>     "", "", ""
>>> )

add_datamodel_host(address, name, note)[source]¶

Add a host data model entry.

Parameters:

address (str) – The host address.
name (str) – Given name to the host.
note (str) – Additional notes.

Returns:

The newly created host.

Return type:

TuoniDataHost

Examples

>>> host = tuoni_server.add_datamodel_host("10.20.30.40", "WS1", "Open windows machine")

add_datamodel_service(address, port, protocol, banner, note)[source]¶

Add a service data model entry.

Parameters:

address (str) – The service address.
port (int) – Port number.
protocol (str) – Service protocol.
banner (str) – Service banner.
note (str) – Additional notes.

Returns:

The newly created service.

Return type:

TuoniDataService

Examples

>>> service = tuoni_server.add_datamodel_service("10.20.30.40", "443", "HTTPS", "", "")

add_hosted(filename, file_content, original_filename=None)[source]¶

Add a hosted file.

Parameters:

filename (str) – The name of the file to host.
file_content (bytes) – The content of the file in bytes.

Returns:

The API URI for the uploaded file.

Return type:

str

Examples

>>> tuoni_server.add_hosted("/hosted/file/here.txt", b"HELLO WORLD")

add_user(username, password, authorities)[source]¶

Add a new user.

Parameters:

username (str) – The username for the new user.
password (str) – The initial password for the user.
authorities (list[str]) – A list of authorities or roles assigned to the user.

Returns:

The newly created user.

Return type:

TuoniUser

Examples

>>> user = tuoni_server.add_user(
>>>     "cool_new_user",
>>>     "cool_new_password",
>>>     [
>>>         "MANAGE_LISTENERS",
>>>         "MANAGE_USERS",
>>>         "SEND_COMMANDS",
>>>         "MANAGE_PAYLOADS",
>>>         "MODIFY_FILES",
>>>         "VIEW_RESOURCES",
>>>         "MANAGE_AGENTS"
>>>     ]
>>> )

connect_websocket(timeout: float = 10, on_error=None)[source]¶

Connect to the Tuoni server’s real-time WebSocket endpoint.

Requires the websocket-client package (pip install websocket-client).

Parameters:

timeout (float) – Seconds to wait for the STOMP handshake to complete.
on_error (callable) – Optional callback(exception) for connection or permission failures, including failures during reconnection.

Returns:

A connected WebSocket client ready for subscriptions.

Return type:

TuoniWebSocket

Raises:

Exception – If the connection cannot be established within timeout seconds.

Examples

>>> ws = tuoni_server.connect_websocket()
>>> results = []
>>> ws.on_command_result(lambda data: results.append(data))

create_payload(payload_template: str, payload_listener: int, payload_conf: dict, encrypted: bool = True, payload_name: str = None)[source]¶

Create a new payload.

Parameters:

payload_template (str) – The payload template to use.
payload_listener (int) – The ID of the listener associated with this payload.
payload_conf (dict) – A dictionary containing the payload’s configuration.
encrypted (bool) – Specifies whether traffic for this payload should be encrypted.
payload_name (str) – The name to assign to the payload.

Returns:

The unique ID of the created payload.

Return type:

int

Examples

>>> payload_id = tuoni_server.create_payload(
>>>     "shelldot.payload.windows-x64",
>>>     listener_id,
>>>     {"type": "executable"}
>>> )

delete_hosted(hosted)[source]¶

Delete a hosted file.

Returns:: None

delete_hosted_by_path(path: str)[source]¶

Delete a hosted file by its path.

Parameters:: path (str) – The hosted path of the file (e.g. /somedir/file.txt).

Examples

>>> tuoni_server.delete_hosted_by_path("/payloads/agent.exe")

download_payload(payload_id: int, file_name: str)[source]¶

Download a payload.

Parameters:

payload_id (int) – The unique ID of the payload to download.
file_name (str) – The name of the file to save the downloaded payload.

let_it_run()[source]¶: Block execution and wait indefinitely if any callback functions are initialized, or until all monitoring threads have completed.

load_agents(active=True, unactive=False)[source]¶

Retrieve a list of agents.

Parameters:

active (bool) – Should active agents be returned
unactive (bool) – Should unactive agents be returned

Returns:

A list of agents.

Return type:

list[TuoniAgent]

load_aliases()[source]¶

Retrieve a list of aliases.

Returns:: A list of aliases.
Return type:: list[TuoniAlias]

load_command_note(agent_guid, command_id, note_id)[source]¶

Retrieve a command note, including an archived note.

Parameters:

agent_guid (str) – UUID of the command’s agent.
command_id (int) – Command identifier.
note_id (str) – Note UUID.

Returns:

The note. Requires VIEW_RESOURCES; a missing or: mismatched command/note raises an HTTP 404 request exception.

Return type:

TuoniNote

load_command_notes(agent_guid, command_id, page=0, pageSize=100, label=None, include_archived=False)[source]¶

Load one page of notes for a visible command, ordered oldest first.

Requires the server’s v2 command-notes API and VIEW_RESOURCES.

Parameters:

agent_guid (str) – UUID of the command’s agent.
command_id (int) – Command identifier.
page (int) – Zero-based page number. Defaults to 0.
pageSize (int) – Notes per page, from 1 to 1000. Defaults to 100.
label (str | None) – Exact, case-sensitive label filter. None omits it.
include_archived (bool) – Include archived notes. Defaults to False.

Returns:

Notes in the requested page, or an empty list.: Pagination metadata is not returned; pages are not fetched automatically.

Return type:

list[TuoniNote]

load_command_plugins()[source]¶

Retrieve a list of command plugins.

Returns:: A list of available command plugins.
Return type:: list[TuoniCommandPlugin]

load_command_template(guid)[source]¶

Retrieve a specific command template by its unique identifier.

Parameters:: guid (str) – The unique identifier of the command template to retrieve.
Returns:: The requested command template, or None if not found.
Return type:: TuoniCommandTemplate

load_command_templates()[source]¶

Retrieve a list of command templates.

Returns:: A list of available command templates.
Return type:: list[TuoniCommandTemplate]

load_datamodel_credentials(page=0, pageSize=256, filter=None)[source]¶

Retrieve a list of credentials.

Parameters:

page (int) – Page of the result.
pageSize (int) – How many results per page.
filter (str) – Additional filters.

Returns:

A list of credentials.

Return type:

list[TuoniDataCredential]

load_datamodel_hosts(page=0, pageSize=256, filter=None)[source]¶

Retrieve a list of hosts.

Parameters:

page (int) – Page of the result.
pageSize (int) – How many results per page.
filter (str) – Additional filters.

Returns:

A list of hosts.

Return type:

list[TuoniDataHost]

load_datamodel_services(page=0, pageSize=256, filter=None)[source]¶

Retrieve a list of services.

Parameters:

page (int) – Page of the result.
pageSize (int) – How many results per page.
filter (str) – Additional filters.

Returns:

A list of services.

Return type:

list[TuoniDataService]

load_hosted()[source]¶

Retrieve a list of hosted files.

Returns:: A list of objects containing the details of hosted files.
Return type:: list[TuoniFile]

load_hosted_by_path(path: str)[source]¶

Retrieve a hosted file by its path.

Parameters:: path (str) – The hosted path of the file (e.g. /somedir/file.txt).
Returns:: The hosted file.
Return type:: TuoniFile

Examples

>>> f = tuoni_server.load_hosted_by_path("/payloads/agent.exe")

load_jobs(page=0, pageSize=256, sort_col='id', sort_order='asc', inactives=False)[source]¶

Retrieve a list of jobs.

Parameters:

page (int) – Page of the result (only used when inactives=True).
pageSize (int) – How many results per page (only used when inactives=True).
sort_col (str) – Based of what field to sort
sort_order (str) – Order of sort
inactives (bool) – Should also inactive jobs be returned

Returns:

A list of jobs.

Return type:

list[TuoniJob]

load_listener_plugins()[source]¶

Retrieve a list of listener plugins.

Returns:: A list of available listener plugins.
Return type:: list[TuoniListenerPlugin]

Examples

>>> http_listener_plugin = tuoni_server.load_listener_plugins()[
>>>     "shelldot.listener.agent-reverse-http"
>>> ]
>>> conf = http_listener_plugin.conf_examples["default"]
>>> conf["sleep"] = 2
>>> conf["instantResponses"] = True
>>> listener = http_listener_plugin.create(conf)

load_listeners()[source]¶

Retrieve a list of listeners.

Returns:: A list of active listeners.
Return type:: list[TuoniListener]

load_payload(payload_id: int)[source]¶

Retrieve a single payload by its ID.

Parameters:: payload_id (int) – The unique ID of the payload.
Returns:: The requested payload.
Return type:: TuoniPayload

Examples

>>> payload = tuoni_server.load_payload(42)
>>> print(payload.status)

load_payload_plugins()[source]¶

Retrieve a list of payload plugins.

Returns:: A list of available payload plugins.
Return type:: list[TuoniPayloadPlugin]

load_payloads(include_non_active: bool = False)[source]¶

Retrieve a list of payloads.

Parameters:: include_non_active (bool) – If True, archived and unavailable payloads are included.
Returns:: A list of payloads.
Return type:: list[TuoniPayload]

Examples

>>> payloads = tuoni_server.load_payloads()
>>> for p in payloads:
>>>     print(p.payload_id, p.name)

load_setting(key: str)[source]¶

Retrieve a single server setting by key.

Parameters:: key (str) – The setting key.
Returns:: The setting object including currentValue, defaultValue, type, etc.
Return type:: dict

Examples

>>> s = tuoni_server.load_setting("tuoni.websocket.command-result-max-size")
>>> print(s["currentValue"])

load_settings()[source]¶

Retrieve all server settings.

Returns:: A dict with server and plugin keys, each containing setting objects.
Return type:: dict

Examples

>>> settings = tuoni_server.load_settings()
>>> for key, s in settings["server"].items():
>>>     print(key, s["currentValue"])

load_users()[source]¶

Retrieve a list of users.

Returns:: A list of users.
Return type:: list[TuoniUser]

login(url: str, username: str, password: str)[source]¶

Access tokens refresh automatically before expiry. Refresh sessions have a fixed server-provided deadline, after which login is required again. A successful login disconnects WebSockets belonging to the previous login.

Parameters:

url (str) – The URL of the Tuoni server to connect to.
username (str) – The username for authentication.
password (str) – The password for authentication.

Examples

>>> tuoni_server = TuoniC2()
# Enable SSL verification using the standard trust store
>>> tuoni_server = TuoniC2(verify=True)
# Set path for CA bundle
>>> tuoni_server = TuoniC2(verify="/path/to/ca_bundle.pem")
# Login to the server
>>> tuoni_server.login("https://localhost:8443", "my_user", "S3cr37")

logout()[source]¶

Revoke the refresh session, clear local tokens and disconnect WebSockets.

Local logout completes even if the server request fails; that failure is still raised. Already-issued access JWTs remain valid until their expiry.

on_new_agent(function, interval: int = 1)[source]¶

Set a callback function to be triggered when a new agent connects.

Parameters:

function (func) – The function to execute when a new agent connects.
interval (int) – The interval, in seconds, to check for new connections.

Examples

>>> def new_agent_callback(agent):
>>>     print(
>>>         f"We got outselves a new agent {agent.guid} from {agent.metadata['hostname']}"
>>>     )
>>>
>>> tuoni_server.on_new_agent(new_agent_callback)

refresh()[source]¶

Explicitly refresh, for example after permissions have been granted.

All refreshes are serialized. A failed or ambiguous rotation invalidates the local session; log in again instead of retrying the old refresh token.

rename_payload(payload_id: int, new_name: str)[source]¶

Rename a payload.

Parameters:

payload_id (int) – The unique ID of the payload to rename.
new_name (str) – The new name for the payload.

request_delete(uri: str, json_data: dict = None)[source]¶

Send a DELETE request to the Tuoni server.

Parameters:

uri (str) – The URI endpoint to send the request to.
json_data (dict) – A dictionary containing the JSON payload to include in the DELETE request.

Returns:

The server’s response as a dictionary.

Return type:

dict

request_get(uri: str, result_as_json: bool = True, result_as_bytes: bool = False)[source]¶

Send a GET request to the Tuoni server.

Parameters:

uri (str) – The URI endpoint to send the request to.
result_as_json (bool) – If True, the server’s response is treated as JSON and converted to a dictionary.
result_as_bytes (bool) – If True, the server’s response is returned as a bytes.

Returns:

The server’s response, either as a raw string or a dictionary if result_as_json is True or as bytes if result_as_bytes is True.

Return type:

str | dict | bytes

request_get_file(uri: str, file_name: str)[source]¶

Send a GET request to the Tuoni server and save the result to the filesystem.

Parameters:

uri (str) – The URI endpoint to send the request to.
file_name (str) – The name of the file where the response will be saved.

request_patch(uri: str, json_data: dict = None, files: dict = None)[source]¶

Send a PATCH request to the Tuoni server.

Parameters:

uri (str) – The URI endpoint to send the request to.
json_data (dict) – A dictionary containing the JSON payload to include in the POST request.
files (dict) – A dictionary of files to upload with the POST request.

Returns:

The server’s response as a dictionary.

Return type:

dict

request_post(uri: str, json_data: dict = None, files: dict = None)[source]¶

Send a POST request to the Tuoni server.

Parameters:

uri (str) – The URI endpoint to send the request to.
json_data (dict) – A dictionary containing the JSON payload to include in the POST request.
files (dict) – A dictionary of files to upload with the POST request.

Returns:

The server’s response as a dictionary.

Return type:

dict

Examples

>>> tuoni = TuoniC2()
>>> tuoni_server.request_post(
>>>     "/api/v1/command-alias",
>>>     {
>>>         "name": "bofX",
>>>         "description": "Example",
>>>         "baseTemplate": "bof",
>>>         "fixedConfiguration": {"method": "go"}
>>>     },
>>>     {"bofFile" : ["some_bof.o", open("some_bof", "rb").read()]}
>>> )

request_put(uri: str, json_data: dict = None)[source]¶

Send a PUT request to the Tuoni server.

Parameters:

uri (str) – The URI endpoint to send the request to.
json_data (dict) – A dictionary containing the JSON payload to include in the PUT request.

Returns:

The server’s response as a dictionary.

Return type:

dict

request_put_multipart(uri: str, files: dict)[source]¶

Send a PUT request with multipart/form-data to the Tuoni server.

Parameters:

uri (str) – The URI endpoint to send the request to.
files (dict) – A dictionary of files to upload. Each key is the form field name and the value is a list of [filename, file_content].

Returns:

The server’s response as a dictionary, or None.

Return type:

dict

update_setting(key: str, value)[source]¶

Update a single server setting.

Parameters:

key (str) – The setting key.
value – The new value for the setting (must be JSON-serialisable).

Returns:

The updated setting object.

Return type:

dict

Examples

>>> tuoni_server.update_setting("tuoni.agent.default-sleep", 5)

update_settings(settings: dict)[source]¶

Update multiple server settings at once.

Parameters:: settings (dict) – A mapping of setting key to new value. Follows the same structure as AllSettingsUpdateRequest on the server.
Returns:: The updated settings.
Return type:: dict

wait_new_agent(interval: int = 1, max_wait: int = 0)[source]¶

Wait for a new agent to connect.

Parameters:

interval (int) – The interval, in seconds, to check for new connections.
max_wait (int) – The maximum time to wait, in seconds. A value of 0 means to wait indefinitely.

Returns:

The newly connected agent.

Return type:

TuoniAgent