Sichere KI-Steuerung von IoT-Geräten: Leitplanken für KI-generierte MQTT-Befehle

Sichere KI-Steuerung von IoT-Geräten bedeutet, dass ein Sprachmodell Aktionen vorschlagen kann, aber nur validierte, ratenbegrenzte und protokollierte Befehle jemals ein Relais, ein Ventil oder einen Motor erreichen. In diesem Leitfaden entwerfen Sie ein Befehls-Gateway, das zwischen einem KI-Agenten und Ihren MQTT-Geräten sitzt, und implementieren Allowlists, Bereichsprüfungen, Ratenbegrenzung, menschliche Bestätigung und ein Audit-Topic in rund 150 Zeilen Python.

Warum KI-Befehle eine eigene Sicherheitsschicht brauchen

Sprachmodelle sind gut darin, aus „Im Besprechungsraum ist es zu warm“ ein „Setze den Sollwert von Raum 3 auf 21 Grad“ zu machen. Sie sind aber auch in der Lage, eine Raumnummer falsch zu lesen, einen Befehl in einer Schleife zu wiederholen oder einer Anweisung zu folgen, die in einer Sensorbeschriftung oder einer E-Mail versteckt ist. Nichts davon ist ungewöhnlich für einen probabilistischen Textgenerator, und Prompts allein können es nicht ausschließen.

Die Lösung ist dieselbe, die Steuerungstechniker schon immer nutzen: Gehen Sie davon aus, dass die vorgelagerte Quelle falsch liegen kann, und validieren Sie alles an einer Grenze, die Sie kontrollieren. Die KI wird zu einem weiteren nicht vertrauenswürdigen Client, und ein deterministisches Programm entscheidet, was erlaubt ist.

Es gibt einen zweiten Grund. Selbst ein vollkommen korrektes Modell kann auf Basis veralteter Informationen handeln. Hat es einen zehn Minuten alten Tankfüllstand gelesen und öffnet ein Füllventil, war der Befehl „vernünftig“ – und das Ergebnis ist trotzdem eine Überschwemmung. Das Gateway ist der Ort, an dem Aktualität und aktueller Zustand vor dem Handeln geprüft werden.

Das Muster des Befehls-Gateways

Die Architektur hat eine Regel: Die KI besitzt niemals Broker-Zugangsdaten, mit denen sie Geräte erreichen kann. Stattdessen:

1. Der Agent ruft ein Tool wie request_action(device, action, value, reason) auf, wie in unserem Leitfaden zu LLM Function Calling für IoT beschrieben.
2. Das Tool übergibt die Anfrage an das Gateway, einen kleinen Dienst, den Sie selbst betreiben.
3. Das Gateway validiert die Anfrage, wendet Grenzwerte an, fragt optional einen Menschen und veröffentlicht erst dann auf dem Befehls-Topic des Geräts.
4. Jede Entscheidung, ob angenommen oder abgelehnt, wird auf einem Audit-Topic veröffentlicht und in ein Log geschrieben.
5. Das Gerät bestätigt auf einem Zustands-Topic, und das Gateway meldet das tatsächliche Ergebnis an den Agenten zurück.

Das ist wichtig, weil ein MQTT-Client mit Zugangsdaten in der Regel auf jedes Topic veröffentlichen kann, das sein Konto erreicht. ConnectMQ bindet Client-IDs an die ausgestellten Zugangsdaten, bietet aber keine ACLs pro Topic. Der ehrliche Weg, „die KI kann die Pumpe nicht direkt ansprechen“ durchzusetzen, besteht daher darin, die Geräte-Zugangsdaten im Gateway-Prozess zu halten und vollständig vom Agenten fernzuhalten. Läuft der Agent als MCP-Server für MQTT, stellen Sie nur Gateway-gestützte Tools bereit, niemals ein rohes publish.

Leitplanken für die sichere KI-Steuerung von IoT-Geräten

Jede Leitplanke adressiert einen bestimmten Fehlermodus:

Verriegelungen, die Menschen und Anlagen schützen, bleiben in der SPS oder der Gerätefirmware. Das Gateway reduziert Störungen und Risiken; es ist kein Ersatz für ein sicherheitsbewertetes System. In Produktionsanlagen bedeutet das konkret: Funktionale Sicherheit nach ISO 13849 oder IEC 61508 bleibt in der Sicherheitssteuerung, und die KI darf höchstens Sollwerte vorschlagen, die die SPS ohnehin über eine definierte Schnittstelle annimmt und selbst plausibilisiert.

Topic-Struktur für abgesicherte Befehle

Halten Sie KI-initiierten Verkehr sichtbar getrennt vom normalen Verkehr. Eine bewährte Struktur:

<prefix>/ai/requests            AI proposals (JSON), written by the agent tool
<prefix>/ai/pending/<id>        retained while waiting for a human
<prefix>/ai/confirm/<id>        human publishes "approve" or "reject"
<prefix>/ai/audit               every decision, never retained
<prefix>/ai/enabled             retained "true"/"false" kill switch
<prefix>/home/<room>/<device>/set     command topic, only the gateway publishes
<prefix>/home/<room>/<device>/state   retained device state

Wenn Sie bereits einem Raum/Gerät/Eigenschaft-Schema folgen, wie in unserem Artikel zur Smart-Home-MQTT-Topic-Struktur beschrieben, fügt das Gateway einfach den Teilbaum ai/ daneben hinzu.

In einer Fabrik folgt die Struktur stattdessen dem Unified Namespace (UNS): Die ISA-95-Hierarchie (Enterprise → Site → Area → Line → Cell) bildet die Topic-Struktur, und der ai/-Teilbaum hängt auf der passenden Ebene, etwa acme/werk-stuttgart/montage/linie-1/ai/requests neben acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur. So sehen Schichtleitung, MES und Audit-Werkzeuge KI-Anfragen genau dort, wo die betroffene Anlage im UNS steht. Mehr zur Struktur im Beitrag Sparkplug B und Unified Namespace.

Das Gateway in Python implementieren

Das folgende Gateway nutzt paho-mqtt 2.x und MQTT 5. Es läuft gegen den öffentlichen MQTT-Broker, damit Sie es ausprobieren können; ändern Sie das Präfix auf Ihre eigene zufällige ID.

import json
import time
import uuid
from collections import defaultdict, deque

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

PREFIX = "connectmq/demo/r8x2m4"   # eigene Zufalls-ID verwenden
ALLOW = {
    "livingroom/thermostat": {"set_temp": {"min": 16, "max": 24, "confirm": False}},
    "garage/door":           {"open": {"confirm": True}, "close": {"confirm": False}},
    "garden/valve":          {"open_minutes": {"min": 1, "max": 15, "confirm": True}},
}
RATE_LIMIT = (3, 600)          # max. 3 Befehle pro Gerät je 10 Minuten
MAX_STATE_AGE = 120            # Sekunden
history = defaultdict(deque)
last_state = {}
pending = {}
ai_enabled = True

def audit(client, decision, req, reason):
    rec = {"ts": time.time(), "decision": decision, "reason": reason, "request": req}
    client.publish(f"{PREFIX}/ai/audit", json.dumps(rec), qos=1)
    with open("ai_audit.log", "a", encoding="utf-8") as f:
        f.write(json.dumps(rec) + "\n")

def validate(req):
    if not ai_enabled:
        return "AI control disabled"
    rule = ALLOW.get(req.get("device"), {}).get(req.get("action"))
    if rule is None:
        return "device/action not allowlisted"
    if "min" in rule:
        v = req.get("value")
        if not isinstance(v, (int, float)) or not rule["min"] <= v <= rule["max"]:
            return f"value outside {rule['min']}..{rule['max']}"
    seen = last_state.get(req["device"])
    if seen is None or time.time() - seen > MAX_STATE_AGE:
        return "device state unknown or stale"
    window = history[req["device"]]
    while window and time.time() - window[0] > RATE_LIMIT[1]:
        window.popleft()
    if len(window) >= RATE_LIMIT[0]:
        return "rate limit reached"
    return None

def execute(client, req):
    props = Properties(PacketTypes.PUBLISH)
    props.MessageExpiryInterval = 15   # verwerfen, wenn nicht innerhalb von 15 s zugestellt
    payload = {"action": req["action"], "value": req.get("value"), "source": "ai", "id": req["id"]}
    client.publish(f"{PREFIX}/home/{req['device']}/set", json.dumps(payload), qos=1, properties=props)
    history[req["device"]].append(time.time())
    audit(client, "executed", req, "passed all checks")

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe([(f"{PREFIX}/ai/requests", 1), (f"{PREFIX}/ai/confirm/+", 1),
                      (f"{PREFIX}/ai/enabled", 1), (f"{PREFIX}/home/+/+/state", 1)])

def on_message(client, userdata, msg):
    global ai_enabled
    t = msg.topic
    if t.endswith("/state"):
        device = "/".join(t.split("/")[-3:-1])
        last_state[device] = time.time()
    elif t == f"{PREFIX}/ai/enabled":
        ai_enabled = msg.payload.decode().strip().lower() == "true"
    elif t == f"{PREFIX}/ai/requests":
        try:
            req = json.loads(msg.payload)
        except ValueError:
            return
        req["id"] = str(uuid.uuid4())[:8]
        problem = validate(req)
        if problem:
            audit(client, "rejected", req, problem)
            return
        if ALLOW[req["device"]][req["action"]].get("confirm"):
            pending[req["id"]] = req
            client.publish(f"{PREFIX}/ai/pending/{req['id']}", json.dumps(req), qos=1, retain=True)
            audit(client, "awaiting_confirmation", req, "high-impact action")
        else:
            execute(client, req)
    elif "/ai/confirm/" in t:
        req_id = t.rsplit("/", 1)[-1]
        req = pending.pop(req_id, None)
        if req is None:
            return
        client.publish(f"{PREFIX}/ai/pending/{req_id}", b"", qos=1, retain=True)  # löschen
        if msg.payload.decode().strip() == "approve":
            problem = validate(req)   # erneut prüfen: Zustand kann sich geändert haben
            if problem:
                audit(client, "rejected", req, "on confirm: " + problem)
            else:
                execute(client, req)
        else:
            audit(client, "rejected", req, "human rejected")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="ai-gateway-r8x2m4",
                     protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Testen Sie es in einem Terminal. Veröffentlichen Sie zuerst einen frischen Gerätezustand, dann eine Anfrage, die ausgeführt werden sollte, und anschließend eine, die blockiert werden sollte:

P=connectmq/demo/r8x2m4
mosquitto_pub -h broker.connectmq.com -p 1883 -i t1-r8x2m4 -r -t $P/home/livingroom/thermostat/state -m '{"temp":22}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i t1-r8x2m4 -t $P/ai/requests \
  -m '{"device":"livingroom/thermostat","action":"set_temp","value":21,"reason":"user feels warm"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i t1-r8x2m4 -t $P/ai/requests \
  -m '{"device":"livingroom/thermostat","action":"set_temp","value":35,"reason":"test"}'
mosquitto_sub -h broker.connectmq.com -p 1883 -i t2-r8x2m4 -t "$P/ai/#" -v

Die zweite Anfrage erscheint auf dem Audit-Topic als rejected mit value outside 16..24. Da der öffentliche MQTT-Broker für jeden lesbar ist, nutzen Sie ihn nur für Tests und Prototypen wie diesen; legen Sie niemals echte Gerätesteuerung oder Produktionsdaten darauf.

Designhinweise, die in der Praxis zählen

Bei der Bestätigung erneut validieren. Ein Mensch gibt eine Anfrage vielleicht erst fünf Minuten nach ihrer Erstellung frei. Der Code ruft validate erneut auf, sodass ein inzwischen erreichtes Ratenlimit oder ein veralteter Zustand sie trotzdem blockiert.

Befehle ablaufen lassen. Das MQTT 5 Message Expiry Interval sorgt dafür, dass ein Befehl, der für ein Offline-Gerät in der Warteschlange steht, verworfen wird, statt eine Stunde später zugestellt zu werden, wenn sich das Garagentor wieder verbindet. Unser Überblick über die Funktionen von MQTT 5 behandelt Expiry und das Response-Topic-Muster, das Sie für ausführlichere Bestätigungen nutzen können.

Das tatsächliche Ergebnis melden. Der Agent sollte dem Benutzer sagen „Das Thermostat hat 21 Grad bestätigt“, nicht „Ich habe es eingestellt“. Warten Sie auf die Änderung des Zustands-Topics, bevor Sie antworten.

Den Not-Aus-Schalter langweilig gestalten. Ein retained false auf ai/enabled stoppt sofort alle KI-Aktionen und übersteht Gateway-Neustarts, weil der Broker es beim Abonnieren erneut ausliefert.

Beschreibungen als nicht vertrauenswürdige Eingabe behandeln. Gerätenamen, Notizen und Sensorbeschriftungen, die das Modell liest, können Anweisungen enthalten. Das Gateway ignoriert das Feld reason bei Entscheidungen; es wird nur für Menschen protokolliert.

Checkliste für sichere KI-Steuerung in der Produktion

1. Der Agent hat keine Geräte-Zugangsdaten; nur das Gateway hat sie.
2. Jedes Paar aus Gerät und Aktion ist ausdrücklich in der Allowlist, mit Wertebereichen.
3. Ratenbegrenzungen gibt es pro Gerät und global.
4. Aktionen mit großer Wirkung erfordern menschliche Bestätigung, und die Freigabe führt die Validierung erneut aus.
5. Befehle tragen eine Ablaufzeit und eine eindeutige ID.
6. Jede Entscheidung wird auf einem Audit-Topic veröffentlicht und von einem separaten Subscriber dauerhaft gespeichert. Diese Nachvollziehbarkeit erleichtert auch Nachweise gegenüber Auditoren, etwa im Rahmen von IEC 62443 oder der menschlichen Aufsicht, die der EU AI Act für bestimmte KI-Systeme verlangt.
7. Ein retained Not-Aus-Schalter ist dokumentiert und getestet.
8. Physische Verriegelungen verbleiben in der Firmware oder der SPS.
9. Produktiver Verkehr läuft über TLS auf Port 8883 mit Zugangsdaten aus dem Portal, wie in unseren Best Practices zur MQTT-Sicherheit empfohlen.

Von der Demo zu einem privaten Broker

Stellen Sie die Verbindung auf Ihren eigenen Broker und Ihre Zugangsdaten um:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID", protocol=mqtt.MQTTv5)
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()
client.connect("broker.connectmq.com", 8883, keepalive=60)

Ein verwalteter MQTT-Broker wie ConnectMQ bietet Ihnen TLS, generierte Zugangsdaten, Flood-Schutz und einen Live-Topic-Explorer, um den Audit-Datenstrom zu beobachten, während die Logik der Leitplanken in Code bleibt, den Sie kontrollieren. Wo Befehls-Gateways unter anderen KI-Mustern einzuordnen sind, zeigt der vollständige Leitfaden zu KI und IoT, und Hintergründe zu Hosting-Optionen finden Sie auf unserer Seite zum Cloud-MQTT-Broker. Die Seite der natürlichen Sprache desselben Systems behandelt der Beitrag zur Gerätesteuerung per natürlicher Sprache.

Abgesicherte KI-Automatisierung auf ConnectMQ aufbauen

Beginnen Sie mit dem öffentlichen MQTT-Broker und verlagern Sie das Gateway dann auf einen privaten Broker mit TLS und eigenen Zugangsdaten. Erstellen Sie einen kostenlosen Serverless-Broker unter app.connectmq.com, ohne Kreditkarte, und halten Sie jeden KI-Befehl vom ersten Tag an auf einem Audit-Topic sichtbar.

Häufig gestellte Fragen

Kann ich mich darauf verlassen, dass der System-Prompt die KI sicher hält?

Nein. Prompts reduzieren Fehler, können aber kein Verhalten garantieren. Setzen Sie Regeln in deterministischem Code durch, den das Modell nicht umgehen kann, und halten Sie Geräte-Zugangsdaten vom Agenten fern.

Validiert ConnectMQ KI-Befehle für mich?

Nein. ConnectMQ stellt Nachrichten sicher und zuverlässig zu; Validierung, Bestätigung und Audit liegen in Ihrem Gateway-Dienst, der sich als gewöhnlicher MQTT-Client mit ConnectMQ verbindet.

Welche Aktionen sollten immer eine menschliche Bestätigung erfordern?

Alles, was Mechanik bewegt, Türen oder Ventile öffnet, Zugänge entriegelt oder bei Wiederholung Schaden anrichten könnte. Komfortanpassungen innerhalb eines engen Bereichs lassen sich meist gefahrlos automatisieren.

Wie speichere ich den Audit-Trail langfristig?

Betreiben Sie einen separaten Subscriber auf dem Audit-Topic, der in eine Datei oder Datenbank schreibt. Audit-Nachrichten werden auf dem Broker nicht retained, daher muss der Subscriber laufen, um sie zu erfassen.