Building Agent Packages

An agent package is the most complex extension point. It defines a complete deployable agent with its own cryptographic protocol, wire format, capability declarations, and build pipeline.

Before building a custom agent, study the dev agent reference implementation at dev_agent/src/tantoc2_dev_agent/. It implements every interface described here.

Architecture
#

An agent package consists of three server-side components and one client-side component:

graph TB
    subgraph "Server-Side (Python Plugins)"
        AP[AgentPackageBase] --> CP[CryptoProvider]
        AP --> PC[ProtocolCodec]
        AP --> BT[Build Templates]
        AP --> CAP[Capability Declarations]
    end
    subgraph "Client-Side (Your Agent Binary)"
        Agent[Agent Binary] --> CC[Crypto Implementation]
        Agent --> WP[Wire Protocol]
        Agent --> CMD[Built-in Commands]
        Agent --> ML[Module Loader]
    end
    AP -.->|"stamp()"| Agent

The server-side plugins handle registration, decryption, decoding, and building. The client-side agent handles communication, command execution, and module loading.

Step 1: CryptoProvider
#

The CryptoProvider handles key exchange and message encryption. Implement CryptoProviderBase:

 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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
from tantoc2.server.crypto_provider import CryptoProviderBase, CryptoSession

class MyCryptoProvider(CryptoProviderBase):
    @classmethod
    def plugin_name(cls) -> str:
        return "my_crypto"

    @classmethod
    def plugin_type(cls) -> str:
        return "crypto_provider"

    def generate_keypair(self) -> tuple[bytes, bytes]:
        """Generate (public_key, private_key) for a new agent build.

        Called by the build system when stamping a new agent binary.
        The public key is embedded in the agent; the private key is
        stored server-side.
        """
        ...

    def create_session(
        self, registration_data: bytes, server_private_key: bytes
    ) -> tuple[CryptoSession, bytes]:
        """Process an agent's registration request.

        Called when the pipeline receives a message with a null session
        token (all zeros). Perform the key exchange (e.g., ECDH) and
        derive a shared session key.

        Args:
            registration_data: Raw bytes after the 20-byte wire header.
            server_private_key: This agent package's server private key.

        Returns:
            (CryptoSession with derived keys, response bytes for the agent)
        """
        # 1. Parse the agent's public key from registration_data
        # 2. Derive shared secret (e.g., ECDH)
        # 3. Derive session key (e.g., HKDF)
        # 4. Generate a 16-byte session token
        # 5. Return CryptoSession + response bytes

        session = CryptoSession(
            session_token=os.urandom(16),
            state="established",
            session_data={
                "session_key": derived_key.hex(),
                "send_counter": 0,
                "recv_counter": 0,
            },
        )
        return session, response_bytes

    def decrypt(self, session: CryptoSession, data: bytes) -> bytes:
        """Decrypt inbound data using the session key.

        Must handle anti-replay (e.g., monotonic counter validation).
        """
        ...

    def encrypt(self, session: CryptoSession, data: bytes) -> bytes:
        """Encrypt outbound data for the agent."""
        ...

    def complete_handshake(
        self, session: CryptoSession, data: bytes
    ) -> CryptoSession:
        """Finalize a multi-step handshake if needed.

        For single-step handshakes (like ECDH), this can simply
        return the session unchanged with state="established".
        """
        return CryptoSession(
            session_token=session.session_token,
            state="established",
            session_data=session.session_data,
        )

    def rotate_session_key(
        self, session: CryptoSession
    ) -> tuple[CryptoSession, bytes]:
        """Rotate the session key.

        Called by the background key rotation service when enabled.

        Returns:
            (updated session with new key, message bytes to send to agent)
        """
        ...

CryptoSession
#

The CryptoSession dataclass holds per-agent session state:

1
2
3
4
5
@dataclass
class CryptoSession:
    session_token: bytes    # 16-byte unique token assigned at registration
    state: str = "handshake"  # "handshake" or "established"
    session_data: dict[str, Any] = field(default_factory=dict)  # Provider-specific

session_data is where you store your session keys, counters, nonces, etc. It’s persisted in the pipeline for the lifetime of the session and serialized to the database as JSON.

Thread Safety
#

encrypt() and decrypt() may be called from multiple threads with the same session. If you use counters or mutable state in session_data, ensure atomic updates.

Step 2: ProtocolCodec
#

The ProtocolCodec translates between wire bytes and InternalMessage:

 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
34
35
36
37
from tantoc2.server.protocol_codec import ProtocolCodecBase
from tantoc2.server.messages import InternalMessage, MessageType

class MyProtocolCodec(ProtocolCodecBase):
    @classmethod
    def plugin_name(cls) -> str:
        return "my_codec"

    @classmethod
    def plugin_type(cls) -> str:
        return "protocol_codec"

    def decode(self, data: bytes) -> InternalMessage:
        """Decode decrypted bytes into an InternalMessage.

        The data has already been decrypted by the CryptoProvider.
        Parse it into the canonical InternalMessage format.
        """
        # Parse your wire format (JSON, protobuf, msgpack, custom, etc.)
        parsed = your_deserialize(data)
        return InternalMessage(
            msg_type=MessageType(parsed["type"]),
            payload=parsed.get("payload", {}),
        )

    def encode(self, message: InternalMessage) -> bytes:
        """Encode an InternalMessage into bytes for encryption.

        The returned bytes will be encrypted by the CryptoProvider
        before being sent on the wire.
        """
        obj = {
            "type": message.msg_type.value,
            "agent_id": message.agent_id,
            "payload": message.payload,
        }
        return your_serialize(obj)

InternalMessage
#

All agent communication passes through this canonical format:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
class MessageType(StrEnum):
    REGISTER = "register"
    CHECKIN = "checkin"
    TASK_RESULT = "task_result"
    TASK_ASSIGNMENT = "task_assignment"
    BEACON_CONFIG = "beacon_config"
    KILL = "kill"
    SURVEY_RESULT = "survey_result"
    KEY_ROTATION_REQUEST = "key_rotation_request"
    KEY_ROTATION_RESPONSE = "key_rotation_response"
    RELAY_FORWARD = "relay_forward"
    PLUGIN_MESSAGE = "plugin_message"

@dataclass
class InternalMessage:
    msg_type: MessageType
    agent_id: str | None = None
    engagement_id: str | None = None
    payload: dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
    metadata: dict[str, Any] = field(default_factory=dict)

Your codec must map your wire format’s message types to MessageType values.

Step 3: AgentPackage
#

The AgentPackageBase ties everything together:

  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
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
from tantoc2.server.agent_package import (
    AgentPackageBase,
    AgentCapabilities,
    AgentTemplate,
    BuildConfig,
    CryptoMaterial,
)

class MyAgentPackage(AgentPackageBase):
    @classmethod
    def plugin_name(cls) -> str:
        return "my_agent"

    @classmethod
    def plugin_type(cls) -> str:
        return "agent_package"

    # --- Required ---

    @classmethod
    def magic_bytes(cls) -> bytes:
        """4-byte magic identifier. MUST be unique across all packages."""
        return b"\xca\xfe\xba\xbe"

    @classmethod
    def crypto_provider_name(cls) -> str:
        """Name of the CryptoProvider plugin to use."""
        return "my_crypto"

    @classmethod
    def protocol_codec_name(cls) -> str:
        """Name of the ProtocolCodec plugin to use."""
        return "my_codec"

    # --- Capabilities ---

    @classmethod
    def supported_module_formats(cls) -> list[str]:
        """Module formats this agent can load.
        Empty list = no module loading support."""
        return ["bof", "shellcode"]

    @classmethod
    def built_in_commands(cls) -> list[str]:
        """Commands the agent natively supports."""
        return ["ls", "cat", "pwd", "whoami", "ps", "upload", "download",
                "load_module", "unload_module"]

    @classmethod
    def supports_daemonize(cls) -> bool:
        return True

    @classmethod
    def supports_relay(cls) -> bool:
        return False

    @classmethod
    def capabilities(cls) -> AgentCapabilities:
        return AgentCapabilities(
            module_formats=cls.supported_module_formats(),
            built_in_commands=cls.built_in_commands(),
            supports_daemonize=cls.supports_daemonize(),
            supports_relay=cls.supports_relay(),
        )

    # --- Build Pipeline ---

    @classmethod
    def is_buildable(cls) -> bool:
        return True

    @classmethod
    def get_templates(cls) -> list[AgentTemplate]:
        return [
            AgentTemplate(
                name="my_beacon",
                platform="windows",
                arch="x86_64",
                format="exe",
                description="Windows x64 beacon executable",
            ),
            AgentTemplate(
                name="my_shellcode",
                platform="windows",
                arch="x86_64",
                format="shellcode",
                description="Windows x64 shellcode",
            ),
        ]

    @classmethod
    def get_config_schema(cls) -> dict[str, Any]:
        """Options shown to operators during build."""
        from tantoc2.server.module_base import OptionSchema
        return {
            "callbacks": OptionSchema(
                name="callbacks", type="str",
                description="Callback addresses", required=True,
            ),
            "kill_date": OptionSchema(
                name="kill_date", type="str",
                description="Kill date (YYYY-MM-DD)", required=True,
            ),
            "beacon_interval": OptionSchema(
                name="beacon_interval", type="int",
                description="Check-in interval (seconds)",
                required=False, default=60,
            ),
            "beacon_jitter": OptionSchema(
                name="beacon_jitter", type="int",
                description="Jitter percentage",
                required=False, default=10,
            ),
        }

    @classmethod
    def stamp(
        cls,
        template_name: str,
        config: BuildConfig,
        crypto_material: CryptoMaterial,
    ) -> bytes:
        """Stamp config and crypto material into an agent binary.

        This is YOUR build logic. Common approaches:
        - Marker replacement (find 0xDEADBEEF block, replace with encrypted config)
        - Append encrypted config blob to binary
        - Template rendering (for interpreted agents)

        Args:
            template_name: Which template to build (e.g., "my_beacon")
            config: BuildConfig with callbacks, kill_date, intervals
            config.callbacks: list[CallbackAddress] with host, port, protocol
            config.kill_date: datetime
            config.beacon_interval: int
            config.beacon_jitter: int
            crypto_material: CryptoMaterial with public_key, private_key, server_public_key

        Returns:
            The complete agent binary as bytes.
        """
        template = cls._load_template(template_name)
        encrypted_config = cls._encrypt_config(config, crypto_material)
        return cls._patch_binary(template, encrypted_config)

Build Dataclasses
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
@dataclass
class BuildConfig:
    callbacks: list[CallbackAddress]
    kill_date: datetime
    beacon_interval: int = 60
    beacon_jitter: int = 10
    extra: dict[str, Any] = field(default_factory=dict)

@dataclass
class CallbackAddress:
    host: str
    port: int
    protocol: str = "https"

@dataclass
class CryptoMaterial:
    public_key: bytes      # Agent's public key
    private_key: bytes     # Agent's private key (embedded in agent)
    server_public_key: bytes  # Server's public key (embedded in agent)

Step 4: Wire Protocol
#

All agent packages use this header:

1
2
3
4
5
+--------+----------+------------------+
| Magic  | Session  | Payload          |
| 4 bytes| Token    | (variable)       |
|        | 16 bytes |                  |
+--------+----------+------------------+

Magic bytes (4B): Routes to the correct CryptoProvider and ProtocolCodec
Session token (16B): All zeros for registration; assigned token thereafter
Payload: Everything after the header is passed to CryptoProvider.decrypt() (or create_session() for registration)

Step 5: Agent-Side Implementation
#

Your agent binary must implement:

Registration
#

Generate a keypair
Send: [magic] + [null_token (16 zeros)] + [public_key + codec(register_msg)]
Receive: [magic] + [session_token] + [server_public_key + codec(register_response)]
Derive shared session key using the server’s public key
Store session token for future messages

Check-in Loop
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
loop:
    message = codec.encode({type: "checkin", payload: {}})
    encrypted = crypto.encrypt(session_key, message)
    send: [magic] + [session_token] + [encrypted]

    response = receive()
    decrypted = crypto.decrypt(session_key, response[20:])
    tasks = codec.decode(decrypted)

    for task in tasks.payload.tasks:
        result = execute_task(task)
        send_result(result)

    sleep(interval + random_jitter)

Task Execution
#

Handle standard task types:

Task Type	Expected Behavior
`survey`	Return system info (hostname, OS, arch, username, IPs)
`beacon_config`	Update check-in interval and jitter
`kill`	Self-destruct (zero memory, remove persistence, exit)
`upload`	Receive file content, write to disk
`download`	Read file, return content
`load_module`	Load and execute a compiled payload
`unload_module`	Clean up a managed module

Module Loading (if supported)
#

For agents that support module loading:

Managed mode (daemonize=false): Load payload, execute, return results through your check-in channel
Daemonized mode (daemonize=true): Launch payload independently; if it’s an agent, it registers on its own

P2P Relay (if supported)
#

For agents that support relay:

Listen on a local port for interior agent connections
Forward their raw wire bytes to the teamserver as relay_forward messages
Buffer and forward responses back to interior agents

Step 6: Packaging
#

File Structure
#

1
2
3
4
5
6
7
8
9
my_agent/
  pyproject.toml          # Package config with entry points
  src/my_agent/
    __init__.py
    package.py            # AgentPackageBase implementation
    crypto.py             # CryptoProviderBase implementation
    codec.py              # ProtocolCodecBase implementation
    agent.py              # Client-side agent logic
    template.py           # Build template data/logic

Entry Points (pyproject.toml)
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
[project]
name = "my-agent"
version = "1.0.0"
dependencies = ["tantoc2"]

[project.entry-points."tantoc2.agent_packages"]
my_agent = "my_agent.package:MyAgentPackage"

[project.entry-points."tantoc2.crypto_providers"]
my_crypto = "my_agent.crypto:MyCryptoProvider"

[project.entry-points."tantoc2.protocol_codecs"]
my_codec = "my_agent.codec:MyProtocolCodec"

Installation
#

1
2
3
4
pip install ./my_agent
# or
pip install my-agent-1.0.0.whl
# or drop the .whl in the plugin inbox

Reference: Dev Agent
#

The dev agent (dev_agent/src/tantoc2_dev_agent/) is the complete reference implementation:

File	Component	Details
`package.py`	AgentPackageBase	Magic: `\xde\xad\xc2\x01`, two templates (beacon/session), full stamping logic
`crypto.py`	CryptoProviderBase	ECDH P-256 + HKDF-SHA256 + AES-256-GCM, counter-based anti-replay
`codec.py`	ProtocolCodecBase	JSON + zlib compression
`agent.py`	Client agent	Beacon/session modes, built-in commands, module loading, P2P relay
`template.py`	Build stamping	Python source patching with encrypted config embedding

Checklist
#

Before deploying your agent package:

magic_bytes() returns exactly 4 bytes, unique across all packages
crypto_provider_name() matches your CryptoProvider’s plugin_name()
protocol_codec_name() matches your ProtocolCodec’s plugin_name()
Registration handshake completes successfully
Encrypt/decrypt roundtrip produces correct plaintext
Codec encode/decode roundtrip preserves all InternalMessage fields
Anti-replay rejects duplicate messages
Kill date is enforced by the agent
stamp() produces a working binary with embedded config
Entry points are correctly declared in pyproject.toml
capabilities() accurately reflects what the agent supports

Architecture#

Step 1: CryptoProvider#

CryptoSession#

Thread Safety#

Step 2: ProtocolCodec#

InternalMessage#

Step 3: AgentPackage#

Build Dataclasses#

Step 4: Wire Protocol#

Step 5: Agent-Side Implementation#

Registration#

Check-in Loop#

Task Execution#

Module Loading (if supported)#

P2P Relay (if supported)#

Step 6: Packaging#

File Structure#

Entry Points (pyproject.toml)#

Installation#

Reference: Dev Agent#

Checklist#