Shared Agent binary HTTP server

[mssalvatore]

Problem

Infection Monkey has multiple exploiters that require exploited hosts to execute a download command (e.g. wget, curl). Each of these exploiters starts its own HTTP server to serve agent binaries. A new HTTP server is spawned per target host. Therefore, if 2 exploiters attempt to exploit 10 different hosts, up to 20 HTTP servers may be started and stopped. This is noisy, resource intensive, and leads to some duplicated code and error handling within exploiters plugins.

Solution

The Agent should start one single HTTP server for serving agent binaries and allow exploiter plugins to use it.

Benefits

• Some exploiters can be decoupled from certain components, such as TCPPortSelector and IAgentBinaryRepository
• Reduce code duplication between exploiters
• Drastically reduce the number of starting/stopping servers (stealthier)
• HTTP servers are more likely to be hosted on an HTTP-related port (80, 443, 8080, etc.)
• Reduced latency (no longer need to wait for multiple HTTP servers to start)

Rough draft

Below is a rough draft. Not all of the parts and pieces are implemented and none of it has been tested. Large portions of the code have been lifted from other components, such as infection_monkey.exploit.tools.HTTPBytesServer.

# WARNING: Locks and Events need to be multiprocessing- and context-aware. As much as possible, we
# want to avoid leaking as multiprocessing-specific details into these classes.


class RequestType(Enum):
    AGENT_BINARY = "agent_binary"
    DROPPER_SCRIPT = "dropper_script"


RequestID = int  # Or maybe UUID?


# TODO: Can we come up with a better name than "Request"?
class Request:
    id: RequestID
    type: RequestType
    operating_system: OperatingSystem
    download_url: URL
    bytes_downloaded: Event


class HTTPAgentBinaryServer:
    def __init__(
        self,
        tcp_port_selector: TCPPortSelector,
        agent_binary_repository: IAgentBinaryRepository,
        poll_interval: float = 0.5,
    ):
        self._tcp_port_selector = tcp_port_selector
        AgentBinaryHTTPHandler.agent_binary_repository = agent_binary_repository

        self._start_lock = Lock()
        self._server_thread: Optional[Thread] = None

    def register_request(
        self,
        operating_system: OperatingSystem,
        request_type: RequestType,
        requestor_ip: IPv4Address,
    ) -> Request:
        request_id = random_id()
        url = self._build_request_url(
            request_id, request_type, operating_system, requestor_ip
        )
        # NOTE: Generate the Event from a multiprocessing context
        new_request = Request(random_id(), request_type, operating_system, url, Event())

        AgentBinaryHTTPHandler.register_request(new_request)

        with self._start_lock:
            if self._server_thread is None:
                self.start()

    def _build_request_url(
        self,
        request_id: RequestID,
        request_type: RequestType,
        operating_system: OperatingSystem,
        requestor_ip: IPv4Address,
    ) -> URL:
        server_ip = get_interface_to_target(requestor_ip)
        return (
            f"http://{server_ip}:{self._port}/{str(operating_system)}/{request_type}/{request_id}",
        )

    def deregister_request(self, request_id: RequestID):
        AgentBinaryHTTPHandler.deregister_request(request_id)

    def start(self):
        """
        Runs the HTTP server in the background and blocks until the server has successfully started
        """
        # The agent (monkey.py) does not need to start the server. The server will be started when
        # the first request is registered. This prevents the server running unnecessarily if we're
        # only using exploiters that don't require it (or no exploiters at all)

        chosen_port = None
        preferences = [443, 80, 8080, 8008, 8000]
        port = int(tcp_port_selector.select_port(preferences=preferences))

        self._server = http.server.HTTPServer(
            ("0.0.0.0", port), AgentBinaryHTTPHandler
        )
        self._server_thread = create_daemon_thread(
            target=self._server.serve_forever,
            name="HTTPAgentBinaryServer",
            args=(self._poll_interval,),
        )
        self._server_thread.start()

    def stop(self, timeout: Optional[float] = None):
        """
        Stops the HTTP server.

        :param timeout: A floating point number of seconds to wait for the server to stop. If this
                        argument is None (the default), the method blocks until the HTTP server
                        terminates. If `timeout` is a positive floating point number, this method
                        blocks for at most `timeout` seconds.
        """
        if self._server_thread is None:
            return

        if self._server_thread.is_alive():
            logger.debug("Stopping the HTTP server")
            self._server.shutdown()
            self._server_thread.join(timeout)

        if self._server_thread.is_alive():
            logger.warning("Timed out while waiting for the HTTP server to stop")
        else:
            logger.debug("The HTTP server has stopped")



# TODO: Consider generating this class dynamically so that more than one HTTPAgentBinaryServer can
# be instantiated if desired. See infection_monkey.exploit.tools.http_bytes_server.py for an
# example.
class AgentBinaryHTTPHandler(BaseHTTPRequestHandler):
    agent_binary_repository: IAgentBinaryRepository
    # These dicts must be shared across multiple processes
    requests: Dict[RequestID, Request] = {}
    locks: Dict[RequestID, Lock] = {}

    def do_GET(self):
        cls = self.__class__
        request_id = int(self.path.split("/")[-1])  # Parse request from the URL

        try:
            lock = cls.locks[request_id]
        except KeyError:
            self.send_response(404)
            self.end_headers()
            return

        with lock:
            request = cls.requests[request_id]
            if request.bytes_downloaded.is_set():
                self.send_error(
                    HTTPStatus.TOO_MANY_REQUESTS,
                    "A download has already been requested",
                )
                return

            logger.info("Received a GET request!")

            self.send_response(HTTPStatus.OK)
            self.send_header("Content-type", "application/octet-stream")
            self.end_headers()

            logger.info("Sending the bytes to the requester")
            agent_binary = cls.agent_binary_repository.get_agent_binary(
                request.operating_system
            )

            if request.type == RequestType.AGENT_BINARY:
                bytes_to_send = agent_binary
            else:
                bytes_to_send = build_dropper_script(
                    request.operating_system, agent_binary
                )

            self.wfile.write(bytes_to_send)
            self.bytes_downloaded.set()

    @classmethod
    def register_request(cls, request: Request) -> Request:
        # NOTE: Generate the Lock from a multiprocessing context
        cls.locks[request.id] = Lock()

    @classmethod
    def deregister_request(cls, request_id: RequestID):
        del_key(cls.lock, request_id)


# This is an interface that limits a plugin's access to the HTTPAgentBinaryServer. Specifically, we
# don't want to allow plugins to start/stop the common HTTPAgentBinaryServer instance.
class IHTTPAgentBinaryServerRegistrar(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def register_request(
        self, operating_system: OperatingSystem, request_type: RequestType
    ) -> Request:
        pass

    @abc.abstractmethod
    def deregister_request(self, request_id: RequestID):
        pass


# This is the concrete object that gets sent to the plugins.
class HTTPAgentBinaryServerRegistrar(IHTTPAgentBinaryServerRegistrar):
    def __init__(self, server: HTTPAgentBinaryServer):
        self._server = server

    def register_request(
        self, operating_system: OperatingSystem, request_type: RequestType
    ) -> Request:
        return self._server.register_request(operating_system, request_type)

    def deregister_request(self, request_id: RequestID):
        self._server.deregister_request(request_id)

Tasks

• Add "preferences" to TCPPortSelector.select_port() (0d) @cakekoa
• Implement the HTTPAgentBinaryServer (0d) @cakekoa
• Create the registrar interface and implement it (0d) @ilija-lazoroski
• Pass an instance of HTTPAgentBinaryServerRegistrar to plugins (0d) @ilija-lazoroski
• Modify plugins to use HTTPAgentBinaryServerRegistrar
  • log4shell (0d) @cakekoa
  • hadoop (0d) - @shreyamalviya
  • mssql (0d) @ilija-lazoroski
  • snmp (0d) @cakekoa
• Remove any now disused HTTP server classes/utilities (HTTPLockedTransfer, etc.) (0d) @cakekoa