Agent Packages

 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

from tantoc2.server.crypto_provider import CryptoProviderBase, CryptoSession

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

    def generate_keypair(self) -> tuple[bytes, bytes]:
        """Generate a server-side keypair. Returns (public_key, private_key)."""
        ...

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

        Called when magic bytes match and session_token is all-zeros.

        Args:
            registration_data: Raw registration payload from the agent.
            server_private_key: The server's private key for this agent package.

        Returns:
            A tuple of (new CryptoSession, response bytes to send back).
        """
        ...

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

    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 key exchange (e.g., ECDH).
        Returns an updated session with state "established"."""
        ...

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

1
2
3
4
5

@dataclass
class CryptoSession:
    session_token: bytes            # 16-byte token assigned at registration
    state: str = "handshake"        # "handshake" or "established"
    session_data: dict[str, Any] = field(default_factory=dict)  # provider-specific data

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from tantoc2.server.protocol_codec import ProtocolCodecBase
from tantoc2.server.messages import InternalMessage

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

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

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

1
2
3
4
5
6
7
8

@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)

 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

from tantoc2.server.agent_package import AgentPackageBase, AgentCapabilities

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

    @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:
        """Return the name of the CryptoProvider plugin this agent uses."""
        return "my_crypto"

    @classmethod
    def protocol_codec_name(cls) -> str:
        """Return the name of the ProtocolCodec plugin this agent uses."""
        return "my_codec"

    @classmethod
    def supported_module_formats(cls) -> list[str]:
        """Module formats this agent can load (e.g., ['bof', 'shellcode']).
        Return empty list if agent does not support module loading."""
        return ["bof", "shellcode"]

    @classmethod
    def built_in_commands(cls) -> list[str]:
        """Commands the agent natively supports."""
        return ["survey", "upload", "download", "persist", "unpersist",
                "beacon_config", "kill", "load_module", "unload_module"]

    @classmethod
    def supports_daemonize(cls) -> bool:
        """Whether this agent supports daemonized module loading."""
        return True

    @classmethod
    def supports_relay(cls) -> bool:
        """Whether this agent supports acting as a P2P relay."""
        return False

    @classmethod
    def supported_relay_protocols(cls) -> list[str]:
        """Transport protocols interior agents can use to reach this agent as a relay.
        Only relevant when supports_relay() returns True.
        Example: ["tcp"] means interior agents connect to this relay via TCP.
        Return an empty list if the agent does not support relaying."""
        return []

    @classmethod
    def agent_modules_dir(cls) -> Path | None:
        """Return path to bundled agent modules, or None.
        Agent packages can bundle modules alongside their code.
        The AgentModuleRegistry scans this directory automatically."""
        return None

    @classmethod
    def capabilities(cls) -> AgentCapabilities:
        """Capability metadata for UI/CLI presentation and filtering.
        Default implementation aggregates the other capability methods."""
        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(),
            relay_protocols=cls.supported_relay_protocols(),
        )

 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

from tantoc2.server.agent_package import (
    AgentPackageBase,
    AgentTemplate,
    BuildConfig,
    CryptoMaterial,
)

class MyBuildablePackage(AgentPackageBase):
    # ... magic_bytes, plugin_name, crypto/codec names ...

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

    @classmethod
    def get_templates(cls) -> list[AgentTemplate]:
        return [
            AgentTemplate(
                name="my_beacon",
                platform="linux",
                arch="x86_64",
                format="elf",
                description="Linux ELF beacon agent",
            ),
        ]

    @classmethod
    def get_config_schema(cls) -> dict[str, OptionSchema]:
        """Return config schema for build options."""
        return { ... }

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

 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. "dev_beacon"
    platform: str     # "linux", "windows", "python"
    arch: str         # "x86_64", "any"
    format: str       # "py", "exe", "dll", "elf", "shellcode"
    description: str = ""

@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
    private_key: bytes
    server_public_key: bytes

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

1
2
3
4
5

+--------+--------+------------------+
| Magic  | Session| Payload          |
| 4 bytes| Token  | (variable)       |
|        | 16 bytes|                  |
+--------+--------+------------------+