What are C2 Profiles

Command and Control (C2) profiles are a little different in Mythic than you might be used to. Specifically, C2 profiles live in their own docker containers and act as a translation mechanism between whatever your special sauce C2 protocol is and what the back-end Mythic server understands (HTTP + JSON). Their entire role in life is to get data off the wire from whatever special communications format you're using and forward that to Mythic.

By defining a C2 protocol specification, other payload types can register that they speak that C2 protocol as well and easily hook in without having to do back-end changes. By having C2 protocols divorced from the main Mythic server, you can create entirely new C2 protocols more easily and you can do them in whatever language you want. If you want to do all your work in GoLang, C#, or some other language for the C2 protocol, go for it. It's all encapsulated in the C2's Docker container with whatever environment you desire.

Since there's so much variability possible within a C2 Docker container, there's some required structure and python files similar to how Payload Types are structured. This is covered in C2 Profile Code.

How does a C2 Profile work in Mythic?

When we look at how C2 Profiles work within Mythic, there are two different stages to consider:

1. How is the C2 Profile defined so that Mythic can track all of the parameters and present them to the user when generating payloads.

2. How does the C2 Profile's code run so that it can listen for agent traffic and communicate with Mythic.

Step 1 - Defining your Profile

Just like with Payload Types, C2 Profiles can either run within Docker or on a separate host or within a VM somewhere. This isn't a hard requirement, but makes it easier to share them. The format is the same as for Payload Types (and even Translation Containers) - the only difference is which classes/structs we instantiate. Check out 1. Payload Type Development for the general structure.

Some differences to note:

• Just like how Payload Types have two sections (agent code and Mythic definition files), C2 Profiles have the same sort of thing (agent code and Mythic definition files).

  • Wherever your server code is located, there's a required file, config.json, that the user can edit from the Mythic UI.

    • config.json - this is a JSON file that exposes any configuration parameters that you want to expose to the user (such as which port to open up, do you need SSL, etc).

    • Copy

      server_binary_path = pathlib.Path(".") / "websocket" / "c2_code" / "server"

      server_binary_path - this is the actual program that Mythic executes when you "start" a C2 Profile. This file can be whatever you want as long as it's executable.

  • The Mythic definition files for your profile (what kind of parameters does it take, what's the name, etc) as well as your OPSEC checks and potential redirector generation code.

Once you get all of that created, you'll want to register the new C2 Profile with Mythic. Normally you install C2 Profiles from the web with sudo ./mythic-cli install github https://github.com/C2Profiles/[profile name]. However, since you already have the code and folder structure in your Mythic/InstalledServices folder, we can just 'tell' Mythic that it exists. You can do this via sudo ./mythic-cli add [profile name]. You can then start just that one container with sudo ./mythic-cli start [profile name]. When the container starts, a few things happen:

1. The Docker container kicks off main.py or main depending on Python or GoLang

2. The optional rabbitmq_config.json as well as environment variables passed in are processed and used to start the service. It then processes all of the files within the c2_functions folder to look for your C2 Profile class (You'll notice here that your class extends the C2Profile class). Once it finds that class, it gets a dictionary representation of all of that information (C2 profile name, parameters, etc) and then connects to RabbitMQ to send that data to Mythic.

3. When Mythic gets that synchronization message from the container with all of the dictionary information, it ties to import the C2 Profile. If it is able to successfully import (or update the current instance), then it'll report an event message that the C2 profile successfully synced.

4. Once the sync happens, the Docker container sends periodic heartbeat messages to Mythic to let it know that the container is still up and going. This is how the UI is able to determine if the container is up or down.

Step 2 - Processing messages

The C2 Profile doesn't need to know anything about the actual content of the messages that are coming from agents and in most cases wouldn't be able to read them anyway since they'll be encrypted. Depending on the kind of communications you're planning on doing, your C2 Profile might wrap or break up an agent's message (eg: splitting a message to go across DNS and getting it reassembled), but then once your C2 Profile re-assembles the agent message, it can just forward it along. In most cases, simply sending the agent message as an HTTP POST message to the location specified by your container's http://MythicServerHost:MythicServerPort/agent_message endpoint where MythicServerHost and MythicServerPort are both available via environment variable is good enough. You'll get an immediate result back from that which your C2 profile should hand back to the agent.

Mythic will try to automatically start your server file when the container starts. This same file is what gets executed when you click to "start" the profile in the UI.

Every Docker container has environment variables, MYTHIC_SERVER_HOST which points to 127.0.0.1 by default and MYTHIC_SERVER_PORT which points to 17443 by default. This information is pulled from the main /Mythic/.env file. So, if you change Mythic's main UI to HTTP on port 7444, then each C2 Docker container's MYTHIC_SERVER_PORT environment variable will update. This allows your code within the docker container to always know where to forward requests so that the main Mythic server can process them.

C2 RPC

C2 Profiles can access the same RPC functions that Payload Types can; however, since C2 profiles don't have things like a task_id, there is some functionality they won't be able to leverage.

Payload Type Docker -> C2 Docker

This one is a little less intuitive than the C2 Docker container directly reaching out to the Mythic server for functionality. This functionality allows tasking as an operator to directly manipulate a C2 component. This functionality has no "default" functions, it's all based on the C2 profile itself. Technically, this goes both ways - C2 Profiles can reach back and execute functionality from Payload Types as well.

Payload Types and C2 Profiles can specify an attribute, custom_rpc_functions, which are dictionaries of key-value pairs (much like the completion functions) where the key is the name of the function that a remote services can call, and the value is the actual function itself. These functions have the following format:

async def func_name(incomingMsg: PayloadBuilder.PTOtherServiceRPCMessage) -> PayloadBuilder.PTOtherServiceRPCMessageResponse:
    response = PayloadBuilder.PTOtherServiceRPCMessageResponse(
        Success=True,
        Result={"some dictionary": "with some values", **incomingMsg.ServiceRPCFunctionArguments}
    )

The incoming data is a dictionary in the incomingMsg.ServiceRPCFunctionArguments and the resulting data goes back through the Result key.

This section talks about the different components for creating the server side docker container for a new C2 profile. This piece accepts messages from the agent, decodes them, forwards them off to the Mythic server, and replies back with the response. Specifically, this goes into the following components:

• Docker containers

• Configuration Files

• Server

All C2 profiles are backed by a Docker container or intermediary layer of some sort.

What's the goal of the container?

What do the C2 docker containers do? Why are things broken out this way? In order to make things more modular within Mythic, most services are broken out into their own containers. When it comes to C2 profiles, they simply serve as an intermediary layer that translates between your special sauce C2 mechanism and the normal RESTful interface that Mythic uses. This allows you to create any number of completely disjoint C2 types without having to modify anything in the main Mythic codebase.

Container Information

C2 Profile Contianers, like Payload Type and Translation Containers, all use the same mythic_base_container Docker image. From there you can leverage the github.com/MythicMeta/MythicContainer GoLang package or the mythic_container PyPi package depending on if you want to write the meta information about your C2 profile in GoLang or Python. The actual code that binds to ports and accepts messages can be written in any language.

Container Components

There are a few things needed to make a C2 container. For reference on general code structure options, check out 1. Payload Type Development. The general structure is the same for Payload Types, C2 Profiles, and Translation containers.

Instead of declaring a new class of Payload Type though, you declare a new class of type C2Profile. For example, in Python you can do:

from mythic_container.C2ProfileBase import *

class Websocket(C2Profile):
    name = "websocket"
    description = "Websocket C2 Server for poseidon"
    author = "@xorrior"
    is_p2p = False
    is_server_routed = False
    server_binary_path = pathlib.Path(".") / "websocket" / "c2_code" / "server"
    server_folder_path = pathlib.Path(".") / "websocket" / "c2_code"

The key here for a C2 Profile though is the server_binary_path - this indicates what actually gets executed to start listening for agent messages. This can be whatever you want, in any language you want, but you just need to make sure it's executable and identified here. If you want this to pick up something from the environment (and it's a script), be sure to put it as a #! at the top. For example, the default containers leverage python3, so they have #! /usr/bin/env python3 at the top. This file is always executed via bash, so as a sub-process like ./server

Within the server_folder_path should be a file called config.json, this is what the operator is able to edit through the UI and should contain all of the configuration components. The one piece that doesn't need to be here are if the operator needs to add additional files (like SSL certs).

C2Profile Class

This is a small class that just defines the metadata aspects of the C2 profile. A big piece here is the definition of the parameters array. Each element here is a C2ProfileParameter class instance with a few possible arguments:

class ParameterType(Enum):
    String = "String"
    ChooseOne = "ChooseOne"
    Array = "Array"
    Date = "Date"
    Dictionary = "Dictionary"
    Boolean = "Boolean"


class DictionaryChoice:
    def __init__(self,
                 name: str,
                 default_value: str = "",
                 default_show: bool = True):
        self.name = name
        self.default_show = default_show
        self.default_value = default_value
    def to_json(self):
        return {
            "name": self.name,
            "default_value": self.default_value,
            "default_show": self.default_show
        }

class C2ProfileParameter:
    def __init__(
        self,
        name: str,
        description: str,
        default_value: any = None,
        randomize: bool = False,
        format_string: str = "",
        parameter_type: ParameterType = ParameterType.String,
        required: bool = True,
        verifier_regex: str = "",
        choices: list[str] = None,
        dictionary_choices: list[DictionaryChoice] = None,
        crypto_type: bool = False,
    ):
        self.name = name
        self.description = description
        self.randomize = randomize
        self.format_string = format_string
        self.parameter_type = parameter_type
        self.required = required
        self.verifier_regex = verifier_regex
        self.choices = choices
        self.default_value = default_value
        self.crypto_type = crypto_type
        self.dictionary_choices = dictionary_choices

    def to_json(self):
        return {
            "name": self.name,
            "description": self.description,
            "default_value": self.default_value if self.parameter_type not in [ParameterType.Array, ParameterType.Dictionary] else json.dumps(self.default_value),
            "randomize": self.randomize,
            "format_string": self.format_string,
            "required": self.required,
            "parameter_type": self.parameter_type.value,
            "verifier_regex": self.verifier_regex,
            "crypto_type": self.crypto_type,
            "choices": self.choices,
            "dictionary_choices": [x.to_json() for x in self.dictionary_choices] if self.dictionary_choices is not None else None
        }

C2 Profile Parameter Types

There are a few main values you can supply for parameter_type when defining the parameters of your c2 profile:

• String - This is simply a text box where you can input a string value

• ChooseOne - this is a dropdown choice where the operator makes a choice from a pre-defined list of options

• Array - This allows an operator to input an array of values rather than a single string value. When this is processed on the back-end, it becomes a proper array value like ["val1", "val2"].

• Date - This is a date in the YYYY-MM-DD format. However, when specifying a default value for this, you simply supply an offset of the number of days from the current day. For example, to have a default value for the user be one year from the current date, the default_value would be 365. Similarly, you can supply negative values and it'll be days in the past. This manifests as a date-picker option for the user when generating a payload.

• Dictionary - This one is a bit more complicated, but allows you to specify an arbitrary dictionary for the user to generate and allows you to set some restrictions on the data. Let's take a look at this one more closely:

C2ProfileParameter(
            name="headers",
            description="Customized headers",
            required=False,
            parameter_type=ParameterType.Dictionary,
            dictionary_choices=[
                DictionaryChoice(name="USER_AGENT",
                    default_value="Mozilla/5.0 (Windows NT 6.3; Trident/7.0; rv:11.0) like Gecko",
                    default_show=True,
                ),
                DictionaryChoice(name="host",
                    default_value="",
                    default_show=False,
                ),
            ]
        ),

This is saying that we have a Dictionary called "headers" with a few pre-set options.

Config.json

Every C2 Profile should have a config.json file located in their defined server_folder_path which contains all the fields an operator would need to configure.

This is the main way for an operator to do configurations of the C2 profile without having to connect to the server where Mythic is running. This can be any format in reality, the name just has to stay the same.

OPSEC scripting

When creating payloads, Mythic will send a C2 Profile's parameters to the associated C2 Profile container for an "opsec check". This is a function that you can choose to write (or not) to look over the C2-specific parameter values that an operator selected to see if they pass your risk tolerance. This function is part of your C2 Profile's class definition:

async def opsec(self, request: C2ProfileBase.C2OPSECMessage):
    response = C2ProfileBase.C2OPSECMessageResponse(Success=True)
    return response

OPSECCheckFunction         func(message C2OPSECMessage) C2OPSECMessageResponse 

package c2structs

// C2_OPSEC_CHECKS STRUCTS

type C2OPSECMessage struct {
   C2Parameters
}

type C2OPSECMessageResponse struct {
   Success bool   `json:"success"`
   Error   string `json:"error"`
   Message string `json:"message"`
}

In the end, the function is returning success or error for if the OPSEC check passed or not.

When is this executed?

OPSEC checks for C2 profiles are executed every time a Payload is created. This means when an operator does it through the UI, when somebody scripts it out, and when a payload is automatically generated as part of tasking (such as for lateral movement or spawning new callbacks).

What is this?

Configuration checks are optional checks implemented by a C2 profile to alert the operator if they're generating an agent with a C2 configuration that doesn't match the current C2 docker services.

When does this happen?

This check occurs every time an agent is generated, and this output is added to the payload's build_message. Thus, an operator sees it when generating a payload, but it's always viewable again from the created payloads page.

What does it look like?

The function is part of your C2 Profile's class definition, so it has access to your local config.json file as well as the instance configuration from the agent.

async def config_check(request: C2ProfileBase.C2ConfigCheckMessage) -> C2ProfileBase.C2ConfigCheckMessageResponse:
    return C2ProfileBase.C2ConfigCheckMessageResponse(
        Success=True,
        Message="Some configuration check message",
        )

ConfigCheckFunction        func(message C2ConfigCheckMessage) C2ConfigCheckMessageResponse 

package c2structs

// C2_CONFIG_CHECK STRUCTS

type C2ConfigCheckMessage struct {
   C2Parameters
}

type C2ConfigCheckMessageResponse struct {
   Success bool   `json:"success"`
   Error   string `json:"error"`
   Message string `json:"message"`
}

What is it?

It's often useful to test your C2 redirector setup before your final deployment. It's also tough to know if there's an issue, if it could be with the agent, with a redirector, or with the C2 profile itself. Because of this, it can be very helpful for a C2 profile to generate a "sample message" that should fit all of the criteria based on an agent's configuration that you can either test configurations or even include in a report about how the C2 configuration works.

Where is it?

On the created payloads page, there's an actions dropdown button next to each payload. That dropdown will contain an option to generate a sample message. This request takes that agent's configuration and forwards it along to the C2 profile.

What does it look like?

async def sample_message(self, inputMsg: C2SampleMessageMessage) -> C2SampleMessageMessageResponse:
    """Generate a sample message for this c2 profile based on the configuration specified

    :param inputMsg: Payload's C2 Profile configuration
    :return: C2SampleMessageMessageResponse detailing a sample message
    """
    response = C2SampleMessageMessageResponse(Success=True)
    response.Message = "Not Implemented"
    response.Message += f"\nInput: {json.dumps(inputMsg.to_json(), indent=4)}"
    return response

SampleMessageFunction      func(message C2SampleMessageMessage) C2SampleMessageResponse

package c2structs

// C2_SAMPLE_MESSAGE STRUCTS

// C2SampleMessageMessage - Generate sample C2 Traffic based on this configuration so that the
// operator and developer can more easily troubleshoot
type C2SampleMessageMessage struct {
   C2Parameters
}

// C2SampleMessageResponse - Provide a string representation of the C2 Traffic that the corresponding
// C2SampleMessageMessage configuration would generate
type C2SampleMessageResponse struct {
   Success bool   `json:"success"`
   Error   string `json:"error"`
   Message string `json:"message"`
}

What is it?

There are times you want to host a file or payload through your C2 channel at a custom endpoint. However, every C2 profile works a bit differently; so, there needs to be a way to universally tell the C2 profile to host a file in its own way at a certain endpoint.

What does it look like?

Below is the function definition to include with your C2 profile to host a custom file.

async def host_file(self, inputMsg: C2HostFileMessage) -> C2HostFileMessageResponse:
    """Host a file through a c2 channel

    :param inputMsg: The file UUID to host and which URL to host it at
    :return: C2HostFileMessageResponse detailing success or failure to host the file
    """
    response = C2HostFileMessageResponse(Success=True)
    response.Message = "Not Implemented"
    response.Message += f"\nInput: {json.dumps(inputMsg.to_json(), indent=4)}"
    return response

HostFileFunction           func(message C2HostFileMessage) C2HostFileMessageResponse

package c2structs

type C2_HOST_FILE_STATUS = string

type C2HostFileMessage struct {
   Name     string `json:"c2_profile_name"`
   FileUUID string `json:"file_uuid"`
   HostURL  string `json:"host_url"`
}

type C2HostFileMessageResponse struct {
   Success bool   `json:"success"`
   Error   string `json:"error"`
}

Where is it?

In the Mythic UI, you'll see blue globe icons that open prompts to host those files via any C2. For example, looking at the Payload's table and clicking the information icon you'll see something like the following:

Here you can see the blue globe icon. Click that and supply the endpoint you want to use, let's say /bob along with the c2 profile you want to use. For our example, let's say http, but it could be any egress profile that has this method implemented. The C2 profile should automatically stop and start to ingest the change, but if it doesn't, you might need to toggle the c2 profile off and on again from the payload types and c2 profiles page to make sure the change is picked up by the internal server. From there, you can hit the /bob endpoint at that C2 profile to fetch the file.

What is it?

This is a function operators can manually invoke for a payload to ask the payload's C2 profiles to generate a set of redirection rules for that payload. Nothing in Mythic knows more about a specific C2 profile than the C2 profile itself, so it makes sense that a C2 profile should be able to generate its own redirection rules for a given payload.

These redirection rules are up to the C2 Profile creators, but can include things like Apache mod_rewrite rules, Nginx configurations, and more.

Where is it?

Operationally, users can invoke this function from the created payloads page with a dropdown menu for the payload they're interested in. Functionally, this code lives in the class definition of your C2 Profile.

What does it look like?

This function gets passed the same sort of information that the opsec check and configuration check functions get; namely, information about all of the payload's supplied c2 profile parameter values. This function can also access the C2 Profile's current configuration.

The format of the function is as follows:

async def redirect_rules(request: C2ProfileBase.C2GetRedirectorRulesMessage) -> C2ProfileBase.C2GetRedirectorRulesMessageResponse:
    return C2ProfileBase.C2GetRedirectorRulesMessageResponse(
        Success=True,
        Message="some mod rewrite rules here"
    )

GetRedirectorRulesFunction func(message C2GetRedirectorRuleMessage) C2GetRedirectorRuleMessageResponse

package c2structs

// C2_REDIRECTOR_RULES STRUCTS

type C2_GET_REDIRECTOR_RULE_STATUS = string

type C2GetRedirectorRuleMessage struct {
   C2Parameters
}

type C2GetRedirectorRuleMessageResponse struct {
   Success bool   `json:"success"`
   Error   string `json:"error"`
   Message string `json:"message"`
}

What is it?

Since C2 profiles can vary pretty wildly, it's not always easy to know what potential indicators of compromise exist for any given c2, especially when you consider how it's modified for a very specific agent. The thing that would know best what kinds of IOCs exist for a given agent configuration for a C2 profile would be the C2 profile itse.f

Where is it?

The dropdown actions button for any payload will have an option to generate IOCs from the corresponding built-in C2 profiles.

What does it look like?

async def get_ioc(self, inputMsg: C2GetIOCMessage) -> C2GetIOCMessageResponse:
    """Generate IOCs for the network traffic associated with the specified c2 configuration

    :param inputMsg: Payload's C2 Profile configuration
    :return: C2GetIOCMessageResponse detailing some IOCs
    """
    response = C2GetIOCMessageResponse(Success=True)
    response.IOCs = []
    return response

GetIOCFunction             func(message C2GetIOCMessage) C2GetIOCMessageResponse

package c2structs

// C2_GET_IOC STRUCTS

// C2GetIOCMessage given the following C2 configuration, determine the IOCs that a defender should look for
type C2GetIOCMessage struct {
   C2Parameters
}

// IOC identify the type of ioc with Type and the actual IOC value
// An example could be a Type of URL with the actual IOC value being the configured callback URL with URI parameters
type IOC struct {
   Type string `json:"type" mapstructure:"type"`
   IOC  string `json:"ioc" mapstructure:"ioc"`
}

// C2GetIOCMessageResponse the resulting set of IOCs that a defender should look out for based on the
// C2GetIOCMessage configuration
type C2GetIOCMessageResponse struct {
   Success bool   `json:"success"`
   Error   string `json:"error"`
   IOCs    []IOC  `json:"iocs"`
}

What is it?

Push C2 is a way to do egress C2 channels without requiring the agent to beacon periodically for tasking. Instead, a connection is held open and Mythic "Pushes" tasks and data down to the agent. This is very similar to how many Peer-to-peer (P2P) agents handle connections, just extended to the egress side of things.

How does it work?

For this to work, there needs to be a held open connection between the C2 Profile docker container and the Mythic server itself. This is done via gRPC. As part of this held open connection, the C2 profile identifies itself and forwards along messages.

What does it look like?

In the mythic UI, the last checkin time will change to 1970-01-01 and appear as Streaming Now. The moment the held open gRPC connection disconnects, that time will update to the current UTC time. This makes it very easy to know that a connection is currently held open even if no traffic is going through it.

Below are some simplified examples of working with this gRPC. An example of this Push style C2 as part of websockets is available with the websocket C2 profile.

from mythic_container.grpc.pushC2GRPC_pb2_grpc import PushC2Stub
from mythic_container.grpc import pushC2GRPC_pb2 as grpcFuncs
import grpc.aio


async def handleStreamConnection(client):
    global UUIDToWebsocketConn
    global grpcStream
    try:
        while True:
            grpcStream = client.StartPushC2Streaming()
            # first send a connect message telling Mythic who this is

            logger.info(f"Connected to gRPC for pushC2 Streaming")
            async for request in grpcStream:
                # this is streaming responses from Mythic to go to agents
                try:
                    # do something with request
                except Exception as d:
                    logger.exception(f"Failed to process handleStreamConnection message:\n{d}")

            logger.error(f"disconnected from gRPC for handleStreamConnection")
    except Exception as e:
        logger.exception(f"[-] exception in handleStreamConnection: {e}")


async def handleGrpcStreamingServices():
    maxInt = 2 ** 31 - 1
    while True:
        try:
            logger.info(f"Attempting connection to gRPC for pushC2...")
            channel = grpc.aio.insecure_channel(
                f'127.0.0.1:17444',
                options=[
                    ('grpc.max_send_message_length', maxInt),
                    ('grpc.max_receive_message_length', maxInt),
                ])
            await channel.channel_ready()
            client = PushC2Stub(channel=channel)
            streamConnections = handleStreamConnection(client)
            logger.info(f"[+] Successfully connected to gRPC for pushC2")
            await asyncio.gather(streamConnections)
        except Exception as e:
            logger.exception(f"Translation gRPC services closed for pushC2: {e}")
            
asyncio.run(handleGrpcStreamingServices())

PushConn := mythicGRPC.GetNewPushC2ClientConnection()
grpcClient := services.NewPushC2Client(PushConn)
streamContext, cancel := context.WithCancel(context.Background())
defer func() {
    cancel()
}()
grpcStream, err := grpcClient.StartPushC2Streaming(streamContext)
if err != nil {
	log.Printf("Failed to get new client: %v\n", err)
	return
} else {
	log.Printf("Got new push client")
}
// sending a message from an agent to Mythic
readErr = grpcStream.Send(&services.PushC2MessageFromAgent{
	C2ProfileName: "websocket",
	RemoteIP:      websocketClient.RemoteAddr().String(),
	TaskingSize:   0,
	Message:       nil,
	Base64Message: []byte(fromAgent.Data),
})
if readErr != nil {
	log.Printf("failed to send message to grpc stream: %v\n", readErr)
	grpcStream.CloseSend()
	return
}
// getting a message from Mythic
fromMythic, readErr := grpcStream.Recv()
if readErr != nil {
	log.Printf("Failed to read from grpc stream, closing connections: %v\n", readErr)
	grpcStream.CloseSend()
	return
}
dataContent := fromMythic.GetMessage(

Agent Expectations

How is an agent supposed to work with a Push-style C2 profile? It's the same as working with a Peer-to-peer (P2P) profile:

• If a payload is executed (it's not a callback yet), then reach out to the C2 profile to make a connection. Once a connection is established, start your normal encrypted key exchange or checkin process

• If an existing callback loses connection for some reason, then reach out to the C2 profile to make a connection. Once a connection is established, send your checkin message again to inform Mythic of your existence

• At this point, just wait for messages to come to you (no need to do a get_tasking poll) and as you get any data (socks, edges, alerts, responses, etc) just send them out through your c2 connection.

Types of Push C2

There are two types of Push C2. They both function largely the same, but within the C2 server they function differently.

One-to-One

The first one is probably the most common, one-to-one. This means that for every agent that connects to the C2 server (typically with a held-open connection like a WebSocket), the C2 server also opens one gRPC connection to Mythic. The following is a diagram what this means:

In this case, the C2 server connects via:

grpcClient.StartPushC2Streaming(streamContext)

The first C2 server is just acting like a proxy in this case and forwards the agent's messages along.

grpcStream.Send(&services.PushC2MessageFromAgent{
    C2ProfileName: "websocket",
    RemoteIP:      websocketClient.RemoteAddr().String(),
    Message:       nil,
    Base64Message: []byte(fromAgent.Data),
})

The above snippet shows an example of sending a message from an agent - we specify the c2 profile name, the remote IP, and in this case, we're passing along the direct base64 blob from the agent. If you wanted to, depending on how your C2 functions, you could pass along a base64 decoded version in the Message field instead.

Once Mythic is done processing a message, we can send the response back to the agent:

fromMythic, readErr := grpcStream.Recv()
if readErr != nil {
    log.Printf("Failed to read from grpc stream, closing connections: %v\n", readErr)
    return
}
fromMythic.GetMessage() // do something with the message from Mythic

One-to-Many

The other type of Push C2 is one-to-many. This functions largely the same as one-to-one, except that the C2 server only ever opens up one gRPC connection to Mythic with many agents that go through it. Below is a diagram showing this flow:

Because of this, we need to make sure that the user understands that all agents using this C2 profile are sharing a connection with Mythic. We start this connection by telling Mythic that our C2 profile wants to open a oneToMany stream:

grpcStream = client.StartPushC2StreamingOneToMany()
await grpcStream.write(grpcFuncs.PushC2MessageFromAgent(
    C2ProfileName="websocket"
))

Notice how the function call is StartPushC2StreamingOneToMany() and not StartPushC2Streaming like in the one-to-one example. Before we even get any agent connections, we send a message to the gRPC stream with just the c2 profile name. This let's Mythic know that we have a new one-to-many c2 profile running and the name of that connection.

At this point, everything works the same as before with the one-to-one profile. As agents send the profile messages, the c2 profile should forward them off to Mythic via the stream and read from the stream to send messages back. You're probably wondering though - how is the multiplexing happening between the one connection to Mythic and the many agents on the other side?

This stream utilizes a TrackingID that's supplied by the C2 profile to track these individual streams:

await grpcStream.write(grpcFuncs.PushC2MessageFromAgent(
    C2ProfileName="websocket",
    RemoteIP=str(websocketConn.remote_address),
    Base64Message=jsonMsg["data"].encode("utf-8"),
    TrackingID=connUUID
))

This TrackingID is something generated by the C2 profile when sending messages and is echoed back as part of the responses that Mythic sends. This allows a C2 Profile to do the proper correlation with messages it gets back and which agent to send it to. This data is saved and tracked by Mythic so that it can be used even when Mythic is the one sending the initial piece of data (like a new task). Let's look at a slightly more complete example to see how that works:

import asyncio
import uuid

import websockets
from websockets.server import serve
from mythic_container.logging import logger
from mythic_container.grpc.pushC2GRPC_pb2_grpc import PushC2Stub
from mythic_container.grpc import pushC2GRPC_pb2 as grpcFuncs
import grpc.aio
import json

UUIDToWebsocketConn = {}
grpcStream = None


async def handleStreamConnection(client):
    global UUIDToWebsocketConn
    global grpcStream
    try:
        while True:
            grpcStream = client.StartPushC2StreamingOneToMany()
            await grpcStream.write(grpcFuncs.PushC2MessageFromAgent(
                C2ProfileName="websocket"
            ))
            logger.info(f"Connected to gRPC for pushC2 StreamingOneToMany")
            async for request in grpcStream:
                # this is streaming responses from Mythic to go to agents
                try:
                    if request.TrackingID in UUIDToWebsocketConn:
                        logger.info(f"sending message back to websocket for id: {request.TrackingID}")
                        await UUIDToWebsocketConn[request.TrackingID].send(json.dumps({"data": request.Message.decode()}))
                    else:
                        logger.error(f"tracking ID not tracked: {request.TrackingID} ")
                except Exception as d:
                    logger.exception(f"Failed to process handleStreamConnection message:\n{d}")

            logger.error(f"disconnected from gRPC for handleStreamConnection")
    except Exception as e:
        logger.exception(f"[-] exception in handleStreamConnection: {e}")


async def handleGrpcStreamingServices():
    maxInt = 2 ** 31 - 1
    while True:
        try:
            logger.info(f"Attempting connection to gRPC for pushC2OneToMany...")
            channel = grpc.aio.insecure_channel(
                f'127.0.0.1:17444',
                options=[
                    ('grpc.max_send_message_length', maxInt),
                    ('grpc.max_receive_message_length', maxInt),
                ])
            await channel.channel_ready()
            client = PushC2Stub(channel=channel)
            streamConnections = handleStreamConnection(client)
            logger.info(f"[+] Successfully connected to gRPC for pushC2OneToMany")
            await asyncio.gather(streamConnections)
        except Exception as e:
            logger.exception(f"Translation gRPC services closed for pushC2OneToMany: {e}")


async def handle_connection(websocketConn: websockets.WebSocketServerProtocol):
    global UUIDToWebsocketConn
    global grpcStream
    connUUID = str(uuid.uuid4())
    logger.info(f"New tracking ID created: {connUUID}")
    UUIDToWebsocketConn[connUUID] = websocketConn
    try:
        async for message in websocketConn:
            # get message from agent and send it to grpc stream
            logger.info(f"new websocket msg for id: {connUUID}")
            while True:
                if grpcStream is None:
                    await asyncio.sleep(1)
                    continue
                break
            try:
                jsonMsg = json.loads(message)
                await grpcStream.write(grpcFuncs.PushC2MessageFromAgent(
                    C2ProfileName="websocket",
                    RemoteIP=str(websocketConn.remote_address),
                    Base64Message=jsonMsg["data"].encode("utf-8"),
                    TrackingID=connUUID
                ))
            except Exception as e:
                logger.info(f"Hit exception trying to send websocket message to grpc: {e}")
                await asyncio.sleep(1)
    except Exception as c:
        if grpcStream is not None:
            logger.info(f"websocket connection dead, removing it: {connUUID}")
            try:
                del UUIDToWebsocketConn[connUUID]
                await grpcStream.write(grpcFuncs.PushC2MessageFromAgent(
                    C2ProfileName="websocket",
                    RemoteIP=str(websocketConn.remote_address),
                    TrackingID=connUUID,
                    AgentDisconnected=True
                ))
            except Exception as e:
                logger.error(f"Failed to send message to Mythic that connection dropped: {e}")


async def main():
    logger.info("starting grpc connection server")
    asyncio.create_task(handleGrpcStreamingServices())
    logger.info("starting websocket server")
    async with serve(handle_connection, "127.0.0.1", 8081):
        await asyncio.Future()


asyncio.run(main())

Here we have a version of the websocket profile - this version uses the one-to-many format. On initial WebSocket connection, we generate a UUID and save the UUID + connection information off into a global variable. We then send that data off into the gRPC stream.

When we get data back from the gRPC stream, we use the same TrackingID that's echoed back to look up the corresponding WebSocket stream and send data that way.

There's one additional piece here that hasn't been covered yet though - AgentDisconnected. Since all agents on the other side of this gRPC connection are sharing the one connection to Mythic, if that gRPC connection exits, then Mythic detects that and marks ALL agents that use that C2 profile has having lost that connection. However, what if just one agent on the other end disconnects? The main gRPC connection is still there, so we need a way to inform Mythic that one remote connection is gone. This is where the AgentDisconnected piece comes into play:

await grpcStream.write(grpcFuncs.PushC2MessageFromAgent(
                    C2ProfileName="websocket",
                    RemoteIP=str(websocketConn.remote_address),
                    TrackingID=connUUID,
                    AgentDisconnected=True
                ))

By sending this message with the same TrackingID that the agent was using, Mythic can look up the corresponding Callback and mark it as no longer "Now Streaming...". If at any point the agent re-connects, then you can use a new TrackingID or even the same TrackingID and everything will connect right back up (assuming the agent sends a message like the checkin message) and the callback's last checkin will update back to "Now Streaming...".

Detecting an agent as gone is easy when we have a held open connection like a WebSocket. Sometimes though, it's not nearly as easy. This is up to you, the C2 developer. You could always leave the agents as "Now Streaming...", but that might be a little confusing for operators. Instead, you could have a timeout where if you haven't heard from a particular agent in a certain amount of time, mark them as gone. If they send a message again later, great, if not though, then at least the user has an idea that the agent might not be there anymore.