Building Agent Packages

Table of Contents

An agent package is the most complex extension point. It defines a complete deployable C2 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 is a fully functional agent package and is the primary reference for all interfaces described here.

Architecture
#

An agent package consists of four server-side components that plug into the teamserver, plus the agent binary that communicates with them.

Component	Interface	Responsibility
`AgentPackageBase`	Magic bytes, component names, capabilities, build pipeline	Routes traffic to the right crypto and codec; builds agent binaries
`CryptoProviderBase`	Key exchange, encrypt/decrypt, anti-replay	Registration handshake and symmetric session crypto
`ProtocolCodecBase`	Encode/decode	Translates between your wire format and `InternalMessage`
Agent binary	Wire protocol client	Implements the check-in loop, task execution, and module loading

Wire Protocol
#

All agent packages use this 20-byte header before every message:

1
2
3
4
+--------+------------------+---------------------------+
| Magic  | Session Token    | Encrypted Payload         |
| 4 bytes| 16 bytes         | variable                  |
+--------+------------------+---------------------------+

Magic (4B): Routes to your AgentPackageBase. Must be unique across all packages.
Session Token (16B): All zeros on registration; assigned 16-byte random token thereafter.
Encrypted Payload: Everything after the 20-byte header. Passed verbatim to CryptoProvider.create_session() (registration) or CryptoProvider.decrypt() (normal messages).

The MessagePipeline reads the header, looks up the package by magic, and dispatches to the correct provider and codec.

Step 1: CryptoProvider
#

The CryptoProviderBase handles key generation, registration handshake, and symmetric encryption.

 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
# tantoc2/src/tantoc2/server/crypto_provider.py

@dataclass
class CryptoSession:
    session_token: bytes           # 16-byte unique token
    state: str = "handshake"       # "handshake" or "established"
    session_data: dict[str, Any] = field(default_factory=dict)  # your keys + state


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

    @abstractmethod
    def generate_keypair(self) -> tuple[bytes, bytes]:
        """Generate (public_key, private_key) for the server.
        Called when stamping a new agent build."""

    @abstractmethod
    def create_session(
        self, registration_data: bytes, server_private_key: bytes
    ) -> tuple[CryptoSession, bytes]:
        """Process an agent registration request.
        Called when session_token is all zeros.
        Returns: (new CryptoSession, response bytes for the agent)"""

    @abstractmethod
    def decrypt(self, session: CryptoSession, data: bytes) -> bytes:
        """Decrypt inbound agent payload using session keys."""

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

    @abstractmethod
    def complete_handshake(self, session: CryptoSession, data: bytes) -> CryptoSession:
        """Finalize a multi-step handshake. Return session with state="established"."""

    @abstractmethod
    def rotate_session_key(self, session: CryptoSession) -> tuple[CryptoSession, bytes]:
        """Rotate the session key. Returns (updated session, message for agent)."""

Reference: Dev Agent CryptoProvider (ECDH + AES-256-GCM)
#

Source: dev_agent/src/tantoc2_dev_agent/crypto.py

 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
class DevCryptoProvider(CryptoProviderBase):
    @classmethod
    def plugin_name(cls) -> str:
        return "dev_crypto"

    def generate_keypair(self) -> tuple[bytes, bytes]:
        """ECDH P-256 keypair in DER format."""
        private_key = ec.generate_private_key(ec.SECP256R1())
        pub = private_key.public_key().public_bytes(
            serialization.Encoding.DER,
            serialization.PublicFormat.SubjectPublicKeyInfo,
        )
        priv = private_key.private_bytes(
            serialization.Encoding.DER,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        )
        return pub, priv

    def create_session(self, registration_data, server_private_key):
        """ECDH key exchange.
        registration_data = agent_pub_key (91B DER) + codec payload
        """
        server_priv = serialization.load_der_private_key(server_private_key, password=None)
        agent_pub = serialization.load_der_public_key(registration_data[:91])

        shared_secret = server_priv.exchange(ec.ECDH(), agent_pub)
        session_key = HKDF(
            algorithm=hashes.SHA256(), length=32, salt=None,
            info=b"tantoc2-dev-agent-session",
        ).derive(shared_secret)

        session_token = os.urandom(16)
        server_pub_bytes = server_priv.public_key().public_bytes(
            serialization.Encoding.DER,
            serialization.PublicFormat.SubjectPublicKeyInfo,
        )
        session = CryptoSession(
            session_token=session_token,
            state="established",
            session_data={
                "session_key": session_key.hex(),
                "send_counter": 0,
                "recv_counter": 0,
            },
        )
        return session, server_pub_bytes   # server_pub sent back to agent

    def decrypt(self, session, data):
        """Wire format: [4B counter BE] [12B nonce] [ciphertext+16B tag]"""
        counter = struct.unpack(">I", data[:4])[0]
        nonce = data[4:16]
        ciphertext = data[16:]
        expected = session.session_data.get("recv_counter", 0)
        if counter < expected:
            raise ValueError(f"Replay: counter {counter} < expected {expected}")
        key = bytes.fromhex(session.session_data["session_key"])
        plaintext = AESGCM(key).decrypt(nonce, ciphertext, None)
        session.session_data["recv_counter"] = counter + 1
        return plaintext

    def encrypt(self, session, data):
        """Wire format: [4B counter BE] [12B nonce] [ciphertext+16B tag]"""
        key = bytes.fromhex(session.session_data["session_key"])
        counter = session.session_data.get("send_counter", 0)
        nonce = os.urandom(12)
        ciphertext = AESGCM(key).encrypt(nonce, data, None)
        session.session_data["send_counter"] = counter + 1
        return struct.pack(">I", counter) + nonce + ciphertext

Key Design Decisions
#

Anti-replay: The dev agent uses a monotonic counter prepended to each ciphertext. The decryption side checks that counter >= recv_counter. You can use nonce-based schemes, but you must ensure replay protection — the pipeline does not enforce this.

session_data: Store all mutable per-session state here. It is a plain dict, persisted as JSON in the engagement database for the session lifetime. It is shared between encrypt() and decrypt(), so use atomic operations on counters.

Thread safety: encrypt() and decrypt() may be called from multiple threads with the same CryptoSession. If session_data is mutated (e.g., incrementing counters), protect it.

Step 2: ProtocolCodec
#

The ProtocolCodecBase translates between raw decrypted bytes and InternalMessage.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
# tantoc2/src/tantoc2/server/protocol_codec.py

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

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

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

InternalMessage Schema
#

All agent communication passes through this canonical structure:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# tantoc2/src/tantoc2/server/messages.py

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 maps between your wire format and these MessageType values. You must handle at minimum: REGISTER, CHECKIN, TASK_RESULT, TASK_ASSIGNMENT, SURVEY_RESULT.

Reference: Dev Agent Codec (JSON + zlib)
#

Source: dev_agent/src/tantoc2_dev_agent/codec.py

 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
class DevProtocolCodec(ProtocolCodecBase):
    @classmethod
    def plugin_name(cls) -> str:
        return "dev_codec"

    def decode(self, data: bytes) -> InternalMessage:
        """zlib → JSON → InternalMessage"""
        decompressed = zlib.decompress(data)
        parsed = json.loads(decompressed)
        return InternalMessage(
            msg_type=MessageType(parsed["type"]),
            agent_id=parsed.get("agent_id"),
            payload=parsed.get("payload", {}),
            timestamp=datetime.fromisoformat(parsed["timestamp"]) if "timestamp" in parsed else datetime.now(UTC),
        )

    def encode(self, message: InternalMessage) -> bytes:
        """InternalMessage → JSON → zlib"""
        data = {
            "type": message.msg_type.value,
            "agent_id": message.agent_id,
            "payload": message.payload,
            "timestamp": message.timestamp.isoformat(),
        }
        return zlib.compress(json.dumps(data, separators=(",", ":")).encode())

You can use any serialization format (protobuf, msgpack, custom binary) as long as you can map the message types.

Step 3: AgentPackage
#

The AgentPackageBase ties the magic bytes, crypto provider, and codec together. It also declares capabilities and optionally implements the build pipeline.

 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
# tantoc2/src/tantoc2/server/agent_package.py

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

    # --- Required ---

    @classmethod
    @abstractmethod
    def magic_bytes(cls) -> bytes:
        """4-byte magic identifying this agent type in wire headers.
        MUST be globally unique across all installed packages."""

    @classmethod
    @abstractmethod
    def crypto_provider_name(cls) -> str:
        """plugin_name() of the CryptoProvider plugin to use."""

    @classmethod
    @abstractmethod
    def protocol_codec_name(cls) -> str:
        """plugin_name() of the ProtocolCodec plugin to use."""

    # --- Capabilities (optional — override to enable) ---

    @classmethod
    def supported_module_formats(cls) -> list[str]:
        """Format identifiers this agent can load. Empty = no module loading."""
        return []

    @classmethod
    def built_in_commands(cls) -> list[str]:
        """Built-in command names. Used for UI/CLI help text."""
        return []

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

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

    @classmethod
    def agent_modules_dir(cls) -> Path | None:
        """Path to bundled agent modules, or None."""
        return None

    @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 (optional — override for buildable packages) ---

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

    @classmethod
    def get_templates(cls) -> list[AgentTemplate]:
        return []

    @classmethod
    def get_config_schema(cls) -> dict[str, Any]:
        return {}

    @classmethod
    def stamp(cls, template_name: str, config: BuildConfig, crypto_material: CryptoMaterial) -> bytes:
        raise NotImplementedError(f"{cls.__name__} is not buildable")

Reference Implementation: Dev Agent Package
#

Source: dev_agent/src/tantoc2_dev_agent/package.py

 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
class DevAgentPackage(AgentPackageBase):
    @classmethod
    def plugin_name(cls) -> str:
        return "dev_agent"

    @classmethod
    def magic_bytes(cls) -> bytes:
        return b"\xde\xad\xc2\x01"

    @classmethod
    def crypto_provider_name(cls) -> str:
        return "dev_crypto"     # matches DevCryptoProvider.plugin_name()

    @classmethod
    def protocol_codec_name(cls) -> str:
        return "dev_codec"      # matches DevProtocolCodec.plugin_name()

    @classmethod
    def supported_module_formats(cls) -> list[str]:
        return ["py"]

    @classmethod
    def built_in_commands(cls) -> list[str]:
        return ["kill", "beacon_config", "survey", "ls", "cat", "pwd",
                "cd", "whoami", "env", "ps", "netstat", "upload", "download",
                "load_module", "unload_module"]

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

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

Build Pipeline Dataclasses
#

 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
@dataclass
class AgentTemplate:
    name: str           # e.g., "my_beacon"
    platform: str       # "linux", "windows", "python"
    arch: str           # "x86_64", "any"
    format: str         # "py", "exe", "dll", "elf", "shellcode"
    description: str = ""

@dataclass
class CallbackAddress:
    host: str
    port: int
    protocol: str = "https"   # "http", "https", "tcp", "tcps"

@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 CryptoMaterial:
    public_key: bytes       # agent's public key (embedded in binary)
    private_key: bytes      # agent's private key (embedded in binary)
    server_public_key: bytes  # server's public key (embedded in binary)

stamp() — Config Stamping
#

The stamp() method produces a deployable agent binary with embedded configuration. Implement it however suits your agent format:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
@classmethod
def stamp(
    cls,
    template_name: str,
    config: BuildConfig,
    crypto_material: CryptoMaterial,
) -> bytes:
    """
    Common approaches:
    - Marker replacement: find a 256-byte 0xDEADBEEF block in the binary,
      replace with an encrypted config blob.
    - Append: concatenate encrypted config to the end of the binary.
    - Template rendering: for interpreted agents (Python, PowerShell),
      substitute config variables in a source template.
    """
    template = cls._load_template(template_name)
    config_blob = cls._encrypt_config(config, crypto_material)
    return cls._patch_binary(template, config_blob)

The dev agent uses Python source template stamping — it reads agent.py, strips the CLI entry point, embeds a base64-encoded encrypted config blob, and returns the patched Python source as bytes.

Step 4: Agent-Side Implementation
#

Your agent binary must implement the client side of the registration and check-in protocol.

Registration Handshake
#

1
2
3
4
5
6
7
8
9
1. Agent generates keypair (e.g., ECDH P-256)
2. Agent sends:
     [magic 4B] + [null session token 16B] + [agent_pub_key] + [codec(register_msg)]
3. Server processes: create_session() → CryptoSession, server_pub_key
4. Server responds:
     [magic 4B] + [session_token 16B] + [server_pub_key] + [codec(ack_msg)]
5. Agent derives shared secret: ECDH(agent_priv, server_pub)
6. Agent derives session key: HKDF(shared_secret)
7. Agent stores session_token for all future messages

After step 7 both sides hold the same session key. No further handshake is needed.

Check-in Loop
#

 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
loop:
  # Build check-in
  msg = codec.encode({type: "checkin", payload: {}})
  encrypted = crypto.encrypt(session_key, msg)
  send: [magic 4B] + [session_token 16B] + [encrypted]

  # Receive response
  response = receive()
  plaintext = crypto.decrypt(session_key, response[20:])
  task_msg = codec.decode(plaintext)  # type: task_assignment

  # Execute tasks
  for task in task_msg.payload["tasks"]:
      result = execute(task)
      # Send each result immediately
      result_msg = codec.encode({
          type: "task_result",
          payload: {
              task_id: task.task_id,
              data: result,
          }
      })
      encrypted = crypto.encrypt(session_key, result_msg)
      send: [magic 4B] + [session_token 16B] + [encrypted]

  sleep(interval + random_jitter)

Task Payloads
#

The task_assignment response contains a list of pending tasks:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
{
  "type": "task_assignment",
  "payload": {
    "tasks": [
      {
        "task_id": "uuid-...",
        "task_type": "shell",
        "payload": { "command": "whoami" }
      }
    ]
  }
}

Handle these standard task types:

`task_type`	Expected Behavior
`survey`	Return: OS, hostname, username, arch, IPs, PID
`beacon_config`	Update `beacon_interval` and `beacon_jitter`
`kill`	Self-destruct: zero memory, remove persistence, exit
`upload`	Write `payload.file_data` (base64) to `payload.remote_path`
`download`	Read `payload.remote_path`, return base64 content
`load_module`	Load and execute `payload.module_data` (base64) in format `payload.module_format`
`unload_module`	Clean up managed module by name

Survey Result
#

The survey task expects a survey_result response (not task_result):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
{
  "type": "survey_result",
  "payload": {
    "task_id": "uuid-...",
    "data": {
      "hostname": "WORKSTATION-1",
      "os": "Windows 10 x64",
      "username": "CORP\\jsmith",
      "arch": "x64",
      "pid": 1234,
      "ips": ["10.0.0.5", "172.16.0.1"]
    }
  }
}

Streaming Results
#

For long-running modules, send partial results using streaming: true:

1
2
3
4
5
6
7
8
{
  "type": "task_result",
  "payload": {
    "task_id": "uuid-...",
    "streaming": true,
    "data": { "output": "partial output chunk..." }
  }
}

The pipeline stores these as streaming chunks. Operators see them accumulate in the UI. Send a final non-streaming result to mark completion.

Module Loading (if supported)
#

For format: bof or format: shellcode:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
Managed mode (daemonize=false):
  1. Allocate RWX memory
  2. Copy module_data bytes into memory
  3. Execute (e.g., via BOF loader or shellcode entry point)
  4. Capture output and send as task_result
  5. Respond to unload_module by freeing memory and cleaning up

Daemonized mode (daemonize=true):
  1. Launch payload as independent process/thread
  2. Do not wait for results
  3. If payload registers: new agent appears with parent_agent_id set

P2P Relay (if supported)
#

1
2
3
4
5
6
7
1. Listen on local port for interior agent connections
2. For each interior message received:
     send relay_forward { inner_message: base64(raw_wire_bytes) }
3. On checkin response:
     if relay_responses in task_assignment:
       forward each response to the corresponding interior agent
       (identified by target_token)

Step 5: Packaging
#

File Structure
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
my-agent/
  pyproject.toml
  src/my_agent/
    __init__.py
    package.py      # AgentPackageBase subclass
    crypto.py       # CryptoProviderBase subclass
    codec.py        # ProtocolCodecBase subclass
    agent.py        # (optional) client-side agent source
    template.py     # (optional) build template helpers
    my_module/      # (optional) bundled agent modules
      manifest.yaml
      my_module.py

`pyproject.toml` Entry Points
#

Agent packages require three entry points — one per plugin component:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "my-agent"
version = "1.0.0"
requires-python = ">=3.11"
dependencies = ["tantoc2", "cryptography>=42.0"]

[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"

[tool.hatch.build.targets.wheel]
packages = ["src/my_agent"]

The entry point keys must match the corresponding plugin_name() return values.

Pre-Deployment Checklist
#

Before deploying your agent package:

magic_bytes() returns exactly 4 bytes, unique across all installed packages
crypto_provider_name() matches your CryptoProviderBase.plugin_name()
protocol_codec_name() matches your ProtocolCodecBase.plugin_name()
All three entry points are declared in pyproject.toml
Registration handshake completes: agent registers and receives session token
encrypt / decrypt roundtrip: decrypt(session, encrypt(session, plaintext)) == plaintext
encode / decode roundtrip: decode(encode(msg)).msg_type == msg.msg_type
Anti-replay correctly rejects counter < recv_counter
Agent enforces kill date before contacting the server
stamp() produces a working binary with the embedded config
capabilities() accurately reflects what the agent supports
Bundled modules appear in agent-modules list after install

Common Pitfalls
#

Magic collision — if two packages share magic bytes, the pipeline routes all traffic to whichever was registered first. Choose 4 bytes that are unlikely to collide and document them. The dev agent uses \xde\xad\xc2\x01.

create_session() parsing — registration data is raw bytes starting right after the 20-byte header. The pipeline passes the full remainder to create_session(). For the dev agent pattern (agent_pub || codec_payload), the pipeline also tries to decode at offset 91 (P-256 DER key size). If you use a different key format, your create_session() must parse correctly from offset 0.

Thread-unsafe session mutation — session_data is mutated by both encrypt() and decrypt(). If the same session is used from multiple threads, use locks or atomic operations.

Missing capabilities() declaration — the built_in_commands list drives the UI command palette. If you declare commands the agent doesn’t implement, operators will see errors. If you don’t declare them, they won’t appear as options.

stamp() not idempotent — calling stamp() multiple times with the same inputs should produce functionally equivalent (not necessarily byte-identical) binaries. Avoid side effects.

See Plugin Packaging for the full packaging and distribution reference.

Architecture#

Wire Protocol#

Step 1: CryptoProvider#

Reference: Dev Agent CryptoProvider (ECDH + AES-256-GCM)#

Key Design Decisions#

Step 2: ProtocolCodec#

InternalMessage Schema#

Reference: Dev Agent Codec (JSON + zlib)#

Step 3: AgentPackage#

Reference Implementation: Dev Agent Package#

Build Pipeline Dataclasses#

stamp() — Config Stamping#

Step 4: Agent-Side Implementation#

Registration Handshake#

Check-in Loop#

Task Payloads#

Survey Result#

Streaming Results#

Module Loading (if supported)#

P2P Relay (if supported)#

Step 5: Packaging#

File Structure#

pyproject.toml Entry Points#

Pre-Deployment Checklist#

Common Pitfalls#