diff --git a/.ci/docker/edgar/scripts/mqtt/manson.py b/.ci/docker/edgar/scripts/mqtt/manson.py
new file mode 100644
index 00000000..1e0d9208
--- /dev/null
+++ b/.ci/docker/edgar/scripts/mqtt/manson.py
@@ -0,0 +1,208 @@
+from __future__ import annotations
+
+import time
+import logging
+
+from serial import Serial
+from serial.tools.list_ports import comports
+
+MANSON_DEFAULT_BR = 9600
+"""Default baudrate for Manson serial ports"""
+
+MANSON_DEVICE_LIST = ((0x10C4, 0xEA60),)
+"""List of known USB VID/PID pairs for Manson power supplies"""
+
+
+class MansonSerialControl:
+    def __init__(self, serial: Serial):
+        """
+        Remote control of a Manson 3304 power supply via Serial UART
+
+        This class provides a simple interface for controlling the Manson 3304 power
+        supply via the Serial UART port. You can either use the constructor and supply
+        it a preconstructed serial interface, or use the convenience
+        method as follows:
+
+        Example::
+            >>> with MansonSerialControl.open_port("COM4") as ctrl:
+            >>>     ctrl.power_cycle(wait_time_s=0.5)
+
+        The class can also search for available interfaces::
+            >>> with MansonSerialControl.open_first_port() as ctrl:
+            >>>     ctrl.power_cycle(wait_time_s=0.5)
+
+        Note: The class instance will automatically close the bound serial port when it
+        goes out of scope.
+
+        :param serial: The open serial port to use for communication with the power supply
+        """
+        self._serial = serial
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+
+    @classmethod
+    def open_port(
+        cls, port_name: str, br: int = MANSON_DEFAULT_BR, debug=False
+    ) -> MansonSerialControl:
+        """Open the specified COM port as a Manson serial interface
+
+        :param port_name: The string name of the port. On Windows, this could be `COM4`.
+            On Linux, it would be something like `/dev/ttyUSB0`.
+        :param br: The baudrate to use for the serial link, the default is
+            `MANSON_DEFAULT_BR`.
+        :return: A new class instance
+        """
+        logger = logging.getLogger()
+        serial = Serial(port_name, br)
+
+        if debug:
+            originalread = serial.read
+
+            def logread(*args, **kwargs):
+                data = originalread(*args, **kwargs)
+                print(f"read({len(data)}): {data}")
+                return data
+
+            serial.read = logread
+
+            originalwrite = serial.write
+
+            def logwrite(data, *args, **kwargs):
+                ret = originalwrite(data, *args, **kwargs)
+                print(f"write({ret}/{len(data)}): {data}")
+                return ret
+
+            serial.write = logwrite
+
+        serial.timeout = 1
+        return cls(serial)
+
+    @classmethod
+    def open_first_port(
+        cls, vid: int = None, pid: int = None, br: int = MANSON_DEFAULT_BR, debug=False
+    ) -> MansonSerialControl:
+        """Search for a serial port with matching USB VID and PID and open it
+
+        The function searches for available serial ports and checks if their USB
+        attributes match a set of known USB vendor IDs / product IDs. It opens the first
+        match and returns a new class instance.
+
+        :param vid: The USB Vendor ID to search for. If this parameter is None, a default
+            set of VIDs and PIDs is used.
+        :param pid: The USB Product ID to search for. If this parameter is None, a default
+            set of VIDs and PIDs is used.
+        :param br: The baudrate to use for the serial link, the default is
+            `MANSON_DEFAULT_BR`.
+        :return: A new class instance
+        """
+        if vid is not None and pid is not None:
+            dev_list = [(vid, pid)]
+        else:
+            dev_list = MANSON_DEVICE_LIST
+
+        for port in comports():
+            if (port.vid, port.pid) in dev_list:
+                return cls.open_port(port.device, br, debug=debug)
+
+        id_list = [f"{vid:04X}:{pid:04X}" for vid, pid in dev_list]
+
+        raise ValueError(
+            "Unable to find Manson with any of these VID / PID: " + ", ".join(id_list)
+        )
+
+    def set_power_state(self, power_on: bool) -> None:
+        """
+        Set the state of the power supply output
+
+        :param power_on: Enable or disable the output of the power supply:
+            True -> Enabled, False -> Disabled
+        """
+        n = "0" if power_on else "1"
+        cmd = f"SOUT{n}\r"
+        self._serial.write(cmd.encode("ascii"))
+
+        r = self._serial.read(size=3)
+        if len(r) < 3:
+            raise IOError(f"Device unresponsive")
+        if r != b"OK\r":
+            raise IOError(f"Incorrect response received from device: {r}")
+
+    def set_voltage(self, voltage):
+        # TODO: error handling, if overvoltage
+        self._serial.write(f"VOLT{int(voltage*10):03d}\r".encode("ascii"))
+        r = self._serial.read(size=3)
+        # print(f"{r}")
+        if len(r) < 3:
+            raise IOError(f"Device unresponsive")
+        if r != b"OK\r":
+            raise IOError(f"Incorrect response received from device: {r}")
+
+    def get_model(self):
+        self._serial.write(f"GMOD\r ".encode("ascii"))
+        r = self._serial.read_until(expected=b"\rOK\r")  # (size=9)
+        self.model = r[:-4]
+        return self.model
+
+    def get_measurements(self):
+        """
+        Read measurements from hardware.
+        Might fail due to flakiness of usb serial -> just retry.
+
+
+        :return:
+        """
+
+        # self._serial.reset_input_buffer()
+
+        # Blocks hardware dials for about 5 seconds. ("REAR CONTROL" LED lights up)
+
+        # VVVVCCCCS\rOK\r
+
+        # self._serial.timeout = 0.5
+        self._write_line("GETD")
+        r = self._read_line()
+
+        if len(r) != 10:
+            raise IOError(f"Incorrect response received from device: {r}")
+        # Returns: VVVVCCCCS\rOK\r Four-digit voltage, e.g. 12.34; four-digit current, e.g. 12.34; status 0=constant voltage, 1=constant current.
+        voltage = float(r[0:4]) / 100
+        current = float(r[4:8]) / 100
+        status = int(r[8:9])
+
+        # starttime = time.time()
+        r = self._read_line()
+
+        if r != b"OK\r":
+            raise IOError(f"Incorrect response received from device: {r}")
+
+        # self._writeline(f"ENDS\r".encode("ascii"))
+
+        return voltage, current, status
+
+    def _write_line(self, command: str):
+        self._serial.write(f"{command}\r".encode("ascii"))
+
+    def _read_line(self):
+        r = self._serial.read_until(expected=b"\r")
+        return r
+
+    def power_cycle(self, wait_time_s: float, post_wait_time_s: float = None) -> None:
+        """
+        Cycle the power supply output, i.e. turn it off and on again
+
+        :param wait_time_s: The wait time in seconds between turning it off and on again
+        :param post_wait_time_s: The wait time after turning power back on
+        """
+        self.set_power_state(power_on=False)
+        time.sleep(wait_time_s)
+        self.set_power_state(power_on=True)
+
+        if post_wait_time_s is not None:
+            time.sleep(post_wait_time_s)
+
+    def close(self):
+        self._serial.close()
diff --git a/.ci/docker/edgar/scripts/mqtt/mockpower.py b/.ci/docker/edgar/scripts/mqtt/mockpower.py
new file mode 100644
index 00000000..7c6bba5e
--- /dev/null
+++ b/.ci/docker/edgar/scripts/mqtt/mockpower.py
@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import time
+import random
+
+
+class MockSerialControl:
+    def __init__(self, serial: Serial):
+        """
+        Remote control of a Manson 3304 power supply via Serial UART
+
+        This class provides a simple interface for controlling the Manson 3304 power
+        supply via the Serial UART port. You can either use the constructor and supply
+        it a preconstructed serial interface, or use the convenience
+        method as follows:
+
+        Example::
+            >>> with MansonSerialControl.open_port("COM4") as ctrl:
+            >>>     ctrl.power_cycle(wait_time_s=0.5)
+
+        The class can also search for available interfaces::
+            >>> with MansonSerialControl.open_first_port() as ctrl:
+            >>>     ctrl.power_cycle(wait_time_s=0.5)
+
+        Note: The class instance will automatically close the bound serial port when it
+        goes out of scope.
+
+        :param serial: The open serial port to use for communication with the power supply
+        """
+        self._serial = serial
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+
+    @classmethod
+    def open_port(cls, port_name: str, br: int = 9600):
+        return cls(port_name)
+
+    @classmethod
+    def open_first_port(cls, vid: int = None, pid: int = None, br: int = 9000):
+        return cls.open_port(True, br)
+
+    def get_model(self):
+        return "Mock-Powersupply"
+
+    def set_power_state(self, power_on: bool) -> None:
+        """
+        Set the state of the power supply output
+
+        :param power_on: Enable or disable the output of the power supply:
+            True -> Enabled, False -> Disabled
+        """
+        n = "0" if power_on else "1"
+
+    def set_voltage(self, voltage, retry=False):
+        pass
+
+    def get_measurements(self):
+        voltage = random.gauss(12.3456, 0.1)
+        current = random.gauss(0.123, 0.1)
+        status = True
+        time.sleep(0.5)
+        return voltage, current, status
+
+    def power_cycle(self, wait_time_s: float, post_wait_time_s: float = None) -> None:
+        """
+        Cycle the power supply output, i.e. turn it off and on again
+
+        :param wait_time_s: The wait time in seconds between turning it off and on again
+        :param post_wait_time_s: The wait time after turning power back on
+        """
+        self.set_power_state(power_on=False)
+        time.sleep(wait_time_s)
+        self.set_power_state(power_on=True)
+
+        if post_wait_time_s is not None:
+            time.sleep(post_wait_time_s)
+
+    def close(self):
+        pass
diff --git a/.ci/docker/edgar/scripts/mqtt/mqttpowerclient.py b/.ci/docker/edgar/scripts/mqtt/mqttpowerclient.py
new file mode 100755
index 00000000..77cf32c4
--- /dev/null
+++ b/.ci/docker/edgar/scripts/mqtt/mqttpowerclient.py
@@ -0,0 +1,185 @@
+#!/usr/bin/python
+import paho.mqtt.client as mqtt
+import argparse
+import manson
+import threading
+from tenacity import Retrying, stop_after_attempt
+import time
+import logging
+
+logging.getLogger().setLevel("DEBUG")
+# setup synchronization and retrying
+
+
+class MqttSerialControl:
+
+    _model = None
+    _voltage = None
+    _current = None
+    _power_state = None
+
+    def on_connect(self, client, userdata, flags, rc, unsure):
+        logging.info("Connected with result code " + str(rc))
+        client.subscribe(self._localtopic + "/model", qos=1)
+        client.subscribe(self._localtopic + "/0/voltage", qos=1)
+        client.subscribe(self._localtopic + "/0/current", qos=1)
+
+    # client.subscribe(localtopic + "/update_interval",qos=1)
+
+    def on_model(self, client, userdata, message):
+        self._model = message.payload.decode()
+        self._model_available.set()
+        logging.info(f"received model {self._model}")
+
+    def on_voltage(self, client, userdata, message):
+        self._voltage = float(message.payload.decode())
+        self._voltage_available.set()
+        logging.info(f"received voltage {self._voltage}")
+
+    def on_current(self, client, userdata, message):
+        self._current = float(message.payload.decode())
+        self._current_available.set()
+        logging.info(f"received current {self._current}")
+
+    def on_message(self, client, userdata, msg):
+        logging.debug(
+            "incoming messages: topic:" + msg.topic + " payload:" + str(msg.payload)
+        )
+
+    def __init__(self, mqtt_server, mqtt_port, localtopic):
+        """
+        Remote control of a power supply via MQTT
+
+        Note: The class instance will automatically close the bound serial port when it
+        goes out of scope.
+
+        :param
+        """
+
+        self._localtopic = localtopic
+        self._mqtt_server = mqtt_server
+        self._mqtt_port = mqtt_port
+        # Generate a unique ID, that is however persistent between restarts.
+        self._client_id = "testclient_" + localtopic.replace("/", "_")
+
+        self._client = mqtt.Client(client_id=self._client_id, protocol=mqtt.MQTTv5)
+        self._client.on_connect = self.on_connect
+        self._client.on_message = self.on_message
+
+        self._client.message_callback_add(localtopic + "/0/voltage", self.on_voltage)
+        self._voltage_available = threading.Event()
+        self._client.message_callback_add(localtopic + "/0/current", self.on_current)
+        self._current_available = threading.Event()
+        self._client.message_callback_add(localtopic + "/model", self.on_model)
+        self._model_available = threading.Event()
+        # client.message_callback_add(localtopic + "/update_interval", on_update_interval)
+        logging.info(f"Connecting to server {mqtt_server}:{mqtt_port}...")
+        self._client.connect(self._mqtt_server, mqtt_port, 60)
+
+        logging.info(f"{self._model=}")
+        # client.publish(topic=localtopic + "/model", payload=model, qos=2, retain=False)
+
+        self._client.loop_start()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+
+    def get_model(self):
+        return self._model
+
+    def set_power_state(self, power_on: bool) -> None:
+        """
+        Set the state of the power supply output
+
+        :param power_on: Enable or disable the output of the power supply:
+            True -> Enabled, False -> Disabled
+        """
+        if power_on == True:
+            payload = 1
+        elif power_on == False:
+            payload = 0
+        msginfo = self._client.publish(
+            topic=self._localtopic + "/0/switch", payload=str(payload), qos=1, retain=True
+        )
+        msginfo.wait_for_publish()
+
+    def set_voltage(self, voltage: float, retry=False):
+        msginfo = self._client.publish(
+            topic=self._localtopic + "/0/voltage_target",
+            payload=voltage,
+            qos=1,
+            retain=True,
+        )
+        msginfo.wait_for_publish()
+
+    def get_measurements(self, timeout=None):
+        # TODO fixme maybe wait for things at the same time
+        if not timeout is None:
+            self._voltage_available.wait(timeout)
+        voltage = self._voltage
+        if not timeout is None:
+            self._current_available.wait(timeout)
+        current = self._current
+        status = True
+        # TODO check how old?
+        return voltage, current, status
+
+    def power_cycle(self, wait_time_s: float, post_wait_time_s: float = None) -> None:
+        """
+        Cycle the power supply output, i.e. turn it off and on again
+
+        :param wait_time_s: The wait time in seconds between turning it off and on again
+        :param post_wait_time_s: The wait time after turning power back on
+        """
+        self.set_power_state(power_on=False)
+        time.sleep(wait_time_s)
+        self.set_power_state(power_on=True)
+
+        if post_wait_time_s is not None:
+            time.sleep(post_wait_time_s)
+
+    def close(self):
+        self._client.loop_stop()
+        pass
+
+
+# client.publish(topic=localtopic + "/update_interval", payload=update_interval, qos=2, retain=False)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        prog="Powersupply MQTT Client",
+        description="connects to a remote powersupply via MQTT",
+    )
+    parser.add_argument("--server", type=str, default="localhost")
+    parser.add_argument("--port", type=int, default=1884)
+    parser.add_argument("--topic", default="defaultpeer/defaultpowersupply")
+    parser.add_argument("--interval", type=float, default=0.5)
+
+    parser.add_argument("--voltage", type=float, default=None)
+    parser.add_argument("--poweron", action='store_true')
+    parser.add_argument("--poweroff", action='store_true')
+    parser.add_argument("--powercycle", action='store_true')
+    parser.add_argument("--getmeasurement", action='store_true')
+    args = parser.parse_args()
+    ps = MqttSerialControl(args.server, args.port, args.topic)
+    if not args.voltage is None:
+        ps.set_voltage(args.voltage)
+
+    if args.poweron:
+        ps.set_power_state(True)
+
+    if args.poweroff:
+        ps.set_power_state(False)
+
+    if args.powercycle:
+        ps.set_power_state(False)
+        time.sleep(1)
+        ps.set_power_state(True)
+
+    if args.getmeasurement:
+        voltage, current, status = ps.get_measurements(timeout=10)
+        print(f"{voltage},{current}")
diff --git a/.ci/docker/edgar/scripts/mqtt/mqttpowerdaemon.py b/.ci/docker/edgar/scripts/mqtt/mqttpowerdaemon.py
new file mode 100755
index 00000000..9e2a04f6
--- /dev/null
+++ b/.ci/docker/edgar/scripts/mqtt/mqttpowerdaemon.py
@@ -0,0 +1,128 @@
+#!/usr/bin/python
+import paho.mqtt.client as mqtt
+import argparse
+import manson
+from threading import Lock
+from tenacity import Retrying, stop_after_attempt
+import time
+import logging
+
+logging.getLogger().setLevel("DEBUG")
+parser = argparse.ArgumentParser(
+    prog="Powersupply MQTT-Daemon",
+    description="connects a Manson powersupply to an MQTT server",
+)
+parser.add_argument("--server")
+parser.add_argument("--port", type=int, default=1884)
+parser.add_argument("--topic", default="defaultpeer/defaultpowersupply")
+parser.add_argument("--interval", type=float, default=0.5)
+# TODO
+parser.add_argument("--device", default="first")
+args = parser.parse_args()
+
+# setup synchronization and retrying
+lock = Lock()
+retryer = Retrying(stop=stop_after_attempt(5), reraise=True)
+
+# setup args
+# Generate a unique ID, that is however persistent between restarts.
+client_id = "powersupply_" + args.topic.replace("/", "_")
+update_interval = args.interval
+localtopic = args.topic
+
+if args.device == "mock":
+    import mockpower
+
+    powersupply = mockpower.MockSerialControl.open_first_port()
+elif args.device == "first":
+    powersupply = manson.MansonSerialControl.open_first_port()
+else:
+    powersupply = manson.MansonSerialControl.open_port(args.device)
+
+
+def on_connect(client, userdata, flags, rc, unsure):
+    logging.info("Connected with result code " + str(rc))
+
+    client.subscribe(localtopic + "/0/switch", qos=1)
+    client.subscribe(localtopic + "/0/voltage_target", qos=1)
+    # client.subscribe(localtopic + "/update_interval",qos=1)
+
+
+def on_update_interval(client, userdata, message):
+    i = float(message.payload.decode())
+    if i > 0 and i < 10:
+        update_interval = i
+
+
+def on_voltage(client, userdata, message):
+    u = float(message.payload.decode())
+    logging.info(f"Set voltage {u}")
+    lock.acquire()
+    try:
+        retryer(powersupply.set_voltage, u)
+    except:
+        logging.exception(f"failed to set voltage {u}")
+        pass
+    finally:
+        logging.debug("release lock")
+        lock.release()
+
+
+def on_switch(client, userdata, message):
+    mpowerstate = message.payload.decode()
+    logging.debug(f'Message Received for switch: "{mpowerstate}"')
+    mpowerstate = int(mpowerstate)
+    if mpowerstate == 1:
+        powerstate = True
+    elif mpowerstate == 0:
+        powerstate = False
+    else:
+        logging.error(f'Invalid power state request:"{mpowerstate}"')
+        return
+    print(f"State: {powerstate}")
+    lock.acquire()
+    retryer(powersupply.set_power_state, powerstate)
+    lock.release()
+
+
+def on_message(client, userdata, msg):
+    logging.debug(
+        "incoming messages: topic:" + msg.topic + " payload:" + str(msg.payload)
+    )
+
+
+client = mqtt.Client(client_id=client_id, protocol=mqtt.MQTTv5)
+client.on_connect = on_connect
+client.on_message = on_message
+client.message_callback_add(localtopic + "/0/switch", on_switch)
+client.message_callback_add(localtopic + "/0/voltage_target", on_voltage)
+# client.message_callback_add(localtopic + "/update_interval", on_update_interval)
+logging.info(f"Connecting to server {args.server}:{args.port}...")
+client.connect(args.server, args.port, 60)
+
+retryer = Retrying(stop=stop_after_attempt(10), reraise=True)
+
+model = retryer(powersupply.get_model)
+logging.info(f"{model=}")
+client.publish(topic=localtopic + "/model", payload=model, qos=1, retain=True)
+
+# client.publish(topic=localtopic + "/update_interval", payload=update_interval, qos=2, retain=False)
+client.loop_start()
+
+while True:
+    lock.acquire()
+    try:
+        voltage, current, status = powersupply.get_measurements()
+    except IOError as e:
+        logging.error(e, exc_info=True)
+        continue
+    finally:
+        lock.release()
+    logging.debug(f"posting measurements: {voltage=}, {current=}, {status=}")
+    client.publish(
+        topic=localtopic + "/0/voltage", payload=voltage, qos=1, retain=False
+    )
+    client.publish(
+        topic=localtopic + "/0/current", payload=current, qos=1, retain=False
+    )
+    time.sleep(update_interval)
diff --git a/.ci/docker/edgar/scripts/mqtt/mqttstats.py b/.ci/docker/edgar/scripts/mqtt/mqttstats.py
new file mode 100755
index 00000000..cd42ffb4
--- /dev/null
+++ b/.ci/docker/edgar/scripts/mqtt/mqttstats.py
@@ -0,0 +1,82 @@
+#!/usr/bin/python
+import paho.mqtt.client as mqtt
+import argparse
+import time
+import logging
+import psutil
+
+logging.getLogger().setLevel("DEBUG")
+parser = argparse.ArgumentParser(
+    prog="Node stats MQTT-Daemon", description="publishes node statistics"
+)
+parser.add_argument("--server")
+parser.add_argument("--port", type=int, default=1884)
+parser.add_argument("--topic", default="defaultpeer/nodestats")
+parser.add_argument("--interval", type=float, default=1)
+# TODO
+parser.add_argument("--device", default="first")
+args = parser.parse_args()
+
+# setup args
+# Generate a unique ID, that is however persistent between restarts.
+client_id = "nodestats_" + args.topic.replace("/", "_")
+update_interval = args.interval
+localtopic = args.topic
+
+
+def on_connect(client, userdata, flags, rc, unsure):
+    logging.info("Connected with result code " + str(rc))
+
+
+client = mqtt.Client(client_id=client_id, protocol=mqtt.MQTTv5)
+client.on_connect = on_connect
+logging.info(f"Connecting to server {args.server}:{args.port}...")
+client.connect(args.server, args.port, 60)
+
+# logging.info(f"{model=}")
+# client.publish(topic=localtopic + "/model", payload=model, qos=2, retain=False)
+
+# client.publish(topic=localtopic + "/update_interval", payload=update_interval, qos=2, retain=False)
+client.loop_start()
+oldtime = time.time()
+oldcounters = newcounters = psutil.net_io_counters(pernic=True)
+while True:
+    client.publish(
+        topic=localtopic + "/cpu%", payload=psutil.cpu_percent(), qos=1, retain=False
+    )
+    client.publish(
+        topic=localtopic + "/mem%",
+        payload=psutil.virtual_memory().percent,
+        qos=1,
+        retain=False,
+    )
+
+    # the net stats are really a lot, and maybe not too interesting, so we send them only at max each 10 seconds.
+    newtime = time.time()
+    if newtime - oldtime >= 10:
+        throughput = dict()
+        # todo depends on what stats we are interested in, we should filter it down:
+        newcounters = psutil.net_io_counters(pernic=True)
+        for ifname, stat in newcounters.items():
+            throughput[ifname] = (stat.bytes_sent - oldcounters[ifname].bytes_sent) / (
+                newtime - oldtime
+            ), (stat.bytes_recv - oldcounters[ifname].bytes_recv) / (newtime - oldtime)
+        # print(throughput)
+        for ifname, value in throughput.items():
+            client.publish(
+                topic=f"{localtopic}/net/{ifname}/sent",
+                payload=round(value[0], 1),
+                qos=1,
+                retain=False,
+            )
+            client.publish(
+                topic=f"{localtopic}/net/{ifname}/recv",
+                payload=round(value[1], 1),
+                qos=1,
+                retain=False,
+            )
+
+        oldcounters = newcounters
+        oldtime = newtime
+
+    time.sleep(update_interval)
diff --git a/.ci/docker/mqtt/docker-compose.yml b/.ci/docker/mqtt/docker-compose.yml
new file mode 100644
index 00000000..24f13b71
--- /dev/null
+++ b/.ci/docker/mqtt/docker-compose.yml
@@ -0,0 +1,20 @@
+version: '3.7'
+services:
+ mqtt:
+    image: bytebeamio/rumqttd
+    command: ["-v", "-c", "/rumqttd.toml"]
+    hostname: mqtt
+    container_name: mqtt
+#    ports:
+#      - 1883:1883
+#      - 1884:1884
+    volumes:
+      - ./rumqttd.toml:/rumqttd.toml
+    restart: always
+    networks:
+      - opendutnet
+
+networks:
+  opendutnet:
+    name: opendut_network
+    external: true  # Use a pre-existing network
diff --git a/.ci/docker/mqtt/expose_ports.yml b/.ci/docker/mqtt/expose_ports.yml
new file mode 100644
index 00000000..e2a36898
--- /dev/null
+++ b/.ci/docker/mqtt/expose_ports.yml
@@ -0,0 +1,7 @@
+---
+version: "3.9"
+services:
+  mqtt:
+    ports:
+      - "1883:1883"
+      - "1884:1884"
diff --git a/.ci/docker/mqtt/localhost_ports.yml b/.ci/docker/mqtt/localhost_ports.yml
new file mode 100644
index 00000000..be0bdd2e
--- /dev/null
+++ b/.ci/docker/mqtt/localhost_ports.yml
@@ -0,0 +1,7 @@
+---
+version: "3.9"
+services:
+  mqtt:
+    ports:
+      - "127.0.0.1:1883:1883"
+      - "127.0.0.1:1884:1884"
diff --git a/.ci/docker/mqtt/rumqttd.toml b/.ci/docker/mqtt/rumqttd.toml
new file mode 100644
index 00000000..ecc741e0
--- /dev/null
+++ b/.ci/docker/mqtt/rumqttd.toml
@@ -0,0 +1,118 @@
+id = 0
+
+# A commitlog read will pull full segment. Make sure that a segment isn't
+# too big as async tcp writes readiness of one connection might affect tail
+# latencies of other connection. Not a problem with preempting runtimes
+[router]
+id = 0
+max_connections = 10010
+max_outgoing_packet_count = 200
+max_segment_size = 104857600
+max_segment_count = 10
+# shared_subscriptions_strategy = "random" # "sticky" | "roundrobin" ( default ) | "random"
+# Any filters that match to configured filter will have custom segment size.
+    # [router.custom_segment.'/office/+/devices/status']
+    # max_segment_size = 102400
+    # max_segment_count = 2
+    # [router.custom_segment.'/home/+/devices/status']
+    # max_segment_size = 51200
+    # max_segment_count = 2
+
+# [bridge]
+# name = "bridge-1"
+# addr = "localhost:1883"
+# qos = 0
+# sub_path = "#"
+# reconnection_delay = 5
+# ping_delay = 5
+# timeout_delay = 5
+#     [bridge.connections]
+#     connection_timeout_ms = 60000
+#     max_payload_size = 20480
+#     max_inflight_count = 500
+#     dynamic_filters = true
+#     [bridge.transport.tls]
+#     ca = "ca.cert.pem"
+#     client_auth = { certs = "test-1.cert.pem", key = "test-1.key.pem" }
+
+# Configuration of server and connections that it accepts
+[v4.1]
+name = "v4-1"
+listen = "0.0.0.0:1883"
+next_connection_delay_ms = 1
+    [v4.1.connections]
+    connection_timeout_ms = 60000
+    max_payload_size = 20480
+    max_inflight_count = 100
+    dynamic_filters = true
+ #   auth = { user1 = "p@ssw0rd", user2 = "password" }
+ #      [v4.1.connections.auth]
+ #      user1 = "p@ssw0rd"
+ #      user2 = "password"
+
+# [v4.2]
+# name = "v4-2"
+# listen = "0.0.0.0:8883"
+# next_connection_delay_ms = 10
+#     # tls config for rustls
+#     [v4.2.tls]
+#     capath = "/etc/tls/ca.cert.pem"
+#     certpath = "/etc/tls/server.cert.pem"
+#     keypath = "/etc/tls/server.key.pem"
+#     # settings for all the connections on this server
+#     [v4.2.connections]
+#     connection_timeout_ms = 60000
+#     throttle_delay_ms = 0
+#     max_payload_size = 20480
+#     max_inflight_count = 100
+#     max_inflight_size = 1024
+
+[v5.1]
+name = "v5-1"
+listen = "0.0.0.0:1884"
+next_connection_delay_ms = 1
+    [v5.1.connections]
+    connection_timeout_ms = 60000
+    max_payload_size = 20480
+    max_inflight_count = 100
+
+[prometheus]
+listen = "127.0.0.1:9042"
+interval = 1
+
+[ws.1]
+name = "ws-1"
+listen = "0.0.0.0:8083"
+next_connection_delay_ms = 1
+    [ws.1.connections]
+    connection_timeout_ms = 60000
+    max_client_id_len = 256
+    throttle_delay_ms = 0
+    max_payload_size = 20480
+    max_inflight_count = 500
+    max_inflight_size = 1024
+
+# [ws.2]
+# name = "ws-2"
+# listen = "0.0.0.0:8081"
+# next_connection_delay_ms = 1
+#     [ws.2.tls]
+#     capath = "/etc/tls/ca.cert.pem"
+#     certpath = "/etc/tls/server.cert.pem"
+#     keypath = "/etc/tls/server.key.pem"
+#     [ws.2.connections]
+#     connection_timeout_ms = 60000
+#     max_client_id_len = 256
+#     throttle_delay_ms = 0
+#     max_payload_size = 20480
+#     max_inflight_count = 500
+#     max_inflight_size = 1024
+
+[console]
+listen = "0.0.0.0:3030"
+
+# [metrics]
+#     [metrics.alerts]
+#     push_interval = 1
+#     [metrics.meters]
+#     push_interval = 1
diff --git a/.ci/docker/theo/src/commands/testenv.rs b/.ci/docker/theo/src/commands/testenv.rs
index 7910a567..6ea67305 100644
--- a/.ci/docker/theo/src/commands/testenv.rs
+++ b/.ci/docker/theo/src/commands/testenv.rs
@@ -72,6 +72,7 @@ impl TestenvCli {
                     docker_compose_up_expose_ports(DockerCoreServices::Firefox.as_str(), expose)?;
                 }
                 docker_compose_up_expose_ports(DockerCoreServices::Keycloak.as_str(), expose)?;
+                docker_compose_up_expose_ports(DockerCoreServices::Mqtt.as_str(), expose)?;
                 crate::core::docker::keycloak::wait_for_keycloak_provisioned()?;
                 if !skip_telemetry {
                     docker_compose_up_expose_ports(DockerCoreServices::Telemetry.as_str(), expose)?;
@@ -94,6 +95,7 @@ impl TestenvCli {
                 docker_compose_down(DockerCoreServices::Netbird.as_str(), false)?;
                 docker_compose_down(DockerCoreServices::Firefox.as_str(), false)?;
                 docker_compose_down(DockerCoreServices::Telemetry.as_str(), false)?;
+                docker_compose_down(DockerCoreServices::Mqtt.as_str(), false)?;
             }
             TaskCli::Network => {
                 crate::core::network::docker_inspect_network()?;
@@ -112,6 +114,7 @@ impl TestenvCli {
                             DockerCoreServices::Netbird => { docker_compose_down(DockerCoreServices::Netbird.as_str(), true) ?; }
                             DockerCoreServices::Firefox => { docker_compose_down(DockerCoreServices::Firefox.as_str(), true) ?; }
                             DockerCoreServices::Telemetry => { docker_compose_down(DockerCoreServices::Telemetry.as_str(), true) ?; }
+                            DockerCoreServices::Mqtt => { docker_compose_down(DockerCoreServices::Mqtt.as_str(), true) ?; }
                         }
                     }
                     None => {
@@ -139,6 +142,7 @@ impl TestenvCli {
         docker_compose_build(DockerCoreServices::Keycloak.as_str())?;
         docker_compose_build(DockerCoreServices::Carl.as_str())?;
         docker_compose_build(DockerCoreServices::Netbird.as_str())?;
+        docker_compose_build(DockerCoreServices::Mqtt.as_str())?;
         Ok(())
     }
 }
diff --git a/.ci/docker/theo/src/core/docker/services.rs b/.ci/docker/theo/src/core/docker/services.rs
index b2e35a46..e5adf08c 100644
--- a/.ci/docker/theo/src/core/docker/services.rs
+++ b/.ci/docker/theo/src/core/docker/services.rs
@@ -12,6 +12,7 @@ pub(crate) enum DockerCoreServices {
     Netbird,
     Firefox,
     Telemetry,
+    Mqtt,
 }
 
 impl DockerCoreServices {
@@ -26,6 +27,7 @@ impl DockerCoreServices {
             DockerCoreServices::Network => "network",
             DockerCoreServices::Firefox => "firefox",
             DockerCoreServices::Telemetry => "telemetry",
+            DockerCoreServices::Mqtt => "mqtt",
         }
     }
 }
diff --git a/.ci/docker/theo/src/core/network.rs b/.ci/docker/theo/src/core/network.rs
index 11e991da..48f88958 100644
--- a/.ci/docker/theo/src/core/network.rs
+++ b/.ci/docker/theo/src/core/network.rs
@@ -39,6 +39,7 @@ enum DockerHostnames {
     NetbirdManagement,
     NetbirdDashboard,
     Firefox,
+    Mqtt,
 }
 
 impl DockerHostnames {
@@ -49,6 +50,7 @@ impl DockerHostnames {
             DockerHostnames::NetbirdManagement => "netbird-management",
             DockerHostnames::NetbirdDashboard => "netbird-dashboard",
             DockerHostnames::Firefox => "firefox",
+            DockerHostnames::Mqtt => "mqtt",
         }
     }
 }
@@ -58,6 +60,7 @@ static CONTAINER_NAME_MAP: phf::Map<&'static str, DockerHostnames> = phf_map! {
     "carl-carl-1" => DockerHostnames::Carl,
     "netbird-management-1" => DockerHostnames::NetbirdManagement,
     "netbird-dashboard-1" => DockerHostnames::NetbirdDashboard,
+    "mqtt" => DockerHostnames::Mqtt,
 };
 
 pub(crate) fn docker_inspect_network() -> crate::Result {