Transport Plugins

TransportConfig

#

All transport configuration is bundled into a TransportConfig dataclass that is passed to the constructor:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from dataclasses import dataclass, field
from typing import Any

@dataclass
class TransportConfig:
    host: str = "0.0.0.0"
    port: int = 443
    tls_enabled: bool = False
    tls_cert_file: str | None = None
    tls_key_file: str | None = None
    options: dict[str, Any] = field(default_factory=dict)

Transport Interface

#

 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
27
28
29
30
31
32
33

from tantoc2.server.transports_module import TransportBase, TransportConfig

class MyTransport(TransportBase):
    @classmethod
    def plugin_name(cls) -> str:
        return "my_transport"

    def __init__(self, config: TransportConfig, on_message: Callable[..., Any]) -> None:
        """Initialize the transport.

        Args:
            config: Bind address, port, TLS settings, and extra options.
            on_message: Callback invoked when data arrives from an agent.
                        Signature: (client_id: str, data: bytes) -> None
        """
        self._config = config
        self._on_message = on_message

    def start(self) -> None:
        """Start listening for connections (no parameters)."""
        ...

    def stop(self) -> None:
        """Stop the listener and release the port."""
        ...

    def is_running(self) -> bool:
        """Whether the listener is actively accepting connections."""
        ...

    def send(self, client_id: str, data: bytes) -> None:
        """Send data to a connected client."""
        ...

The plugin_type() classmethod is already implemented by TransportBase and returns "transport". You only need to override plugin_name().

Key Concepts

#

Message Callback

#

The on_message callback is provided at construction time and connects the transport to the message pipeline. When data arrives from an agent:

1. Derive a client_id string identifying the remote connection (e.g. "192.168.1.10:54321")
2. Call on_message(client_id, data) with the client identifier and raw bytes
3. The pipeline processes the data (magic routing, crypto, codec, handler)

The callback returns None. To send response data back to the agent, the pipeline calls send(client_id, data) on the transport, which queues the response for delivery.

Sending Responses

#

Use send(client_id, data) to queue response bytes for a specific client. How the data reaches the agent depends on the transport:

• HTTP – queued data is returned in the body of the client’s next POST response.
• TCP – queued data is sent as a length-prefixed message after the current request is processed.

Port Management

#

The transport must release its port when stop() is called. The listener manager calls stop() on shutdown and when an operator explicitly stops a listener.

TLS Support

#

TLS settings are part of TransportConfig:

• tls_enabled – whether TLS is requested
• tls_cert_file – path to PEM certificate
• tls_key_file – path to PEM private key

The transport decides how to apply TLS. If tls_enabled is true but no certificate files are specified, the built-in transports generate self-signed certificates automatically.

Built-in Transports

#

Deployment

#

Place the file in plugins/transports/my_transport.py. It is discovered on the next server start or plugin refresh.