Contrôle sécurisé des appareils IoT par l'IA via MQTT

Le contrôle sécurisé des appareils IoT par l'IA signifie qu'un modèle de langage peut proposer des actions, mais que seules des commandes validées, limitées en débit et journalisées atteignent un relais, une vanne ou un moteur. Dans ce guide, vous allez concevoir une passerelle de commandes placée entre un agent IA et vos appareils MQTT, et implémenter listes autorisées, contrôles de plage, limites de débit, confirmation humaine et topic d'audit en 150 lignes de Python environ.

Pourquoi les commandes de l'IA ont besoin de leur propre couche de sécurité

Les modèles de langage savent très bien transformer « il fait trop chaud dans la salle de réunion » en « régler la consigne de la salle 3 à 21 degrés ». Ils sont aussi capables de mal lire un numéro de salle, de répéter une commande en boucle ou de suivre une instruction cachée dans l'étiquette d'un capteur ou dans un e-mail. Rien de tout cela n'est inhabituel pour un générateur de texte probabiliste, et les prompts seuls ne peuvent pas l'exclure.

La solution est celle que les automaticiens ont toujours appliquée : partir du principe que la source amont peut se tromper et tout valider à une frontière que vous maîtrisez. L'IA devient un client non fiable de plus, et un programme déterministe décide de ce qui est autorisé.

Il existe une seconde raison. Même un modèle parfaitement correct peut agir sur des informations périmées. S'il a lu un niveau de cuve vieux de dix minutes et ouvre une vanne de remplissage, la commande était « raisonnable » et le résultat reste une inondation. La passerelle est l'endroit où vérifier la fraîcheur des données et l'état actuel avant d'agir.

Le pattern de la passerelle de commandes

L'architecture repose sur une règle : l'IA ne détient jamais d'identifiants de broker permettant d'atteindre les appareils. À la place :

L'agent appelle un outil tel que request_action(device, action, value, reason), comme décrit dans notre guide sur le function calling LLM pour l'IoT.
L'outil transmet la demande à la passerelle, un petit service qui vous appartient.
La passerelle valide la demande, applique les limites, sollicite éventuellement un humain, et ce n'est qu'ensuite qu'elle publie sur le topic de commande de l'appareil.
Chaque décision, acceptée ou rejetée, est publiée sur un topic d'audit et écrite dans un journal.
L'appareil confirme sur un topic d'état, et la passerelle rapporte le résultat réel à l'agent.

C'est important, car un client MQTT disposant d'identifiants peut généralement publier sur n'importe quel topic accessible à son compte. ConnectMQ associe les client IDs aux identifiants qu'il délivre, mais ne propose pas d'ACL par topic : la manière honnête d'imposer « l'IA ne peut pas toucher directement à la pompe » est donc de garder les identifiants des appareils à l'intérieur du processus de la passerelle, entièrement hors de l'agent. Si l'agent fonctionne comme un serveur MCP pour MQTT, n'exposez que des outils adossés à la passerelle, jamais un publish brut.

Garde-fous pour le contrôle sécurisé des appareils IoT par l'IA

Chaque garde-fou répond à un mode de défaillance précis :

Garde-fou	Ce qu'il empêche	Mise en œuvre typique
Liste autorisée d'appareils et d'actions	Des commandes vers des équipements que l'IA ne doit jamais toucher	Dictionnaire statique dans la passerelle, relu comme du code
Plages de valeurs	Des consignes dangereuses, comme 90 degrés sur un thermostat	Minimum, maximum et pas par action
Limites de débit	Boucles et oscillations	Nombre maximal de commandes par appareil et par fenêtre
Contrôles de fraîcheur	Agir sur des mesures périmées	Rejet si la dernière mise à jour d'état est trop ancienne
Confirmation humaine	Des actions à fort impact exécutées à l'aveugle	File d'attente, approbation via un topic de confirmation
Expiration des commandes	D'anciennes commandes exécutées après une reconnexion	Intervalle d'expiration des messages de MQTT 5
Piste d'audit	Des actions intraçables	Topic d'audit plus journal en ajout seul
Arrêt d'urgence logiciel	Une automatisation qui s'emballe	Un unique indicateur en retained message qui désactive toutes les actions de l'IA

Les verrouillages qui protègent les personnes et les équipements restent dans l'automate ou le firmware de l'appareil. La passerelle réduit les nuisances et les risques ; elle ne remplace pas un système de sécurité certifié.

Organisation des topics pour des commandes encadrées

Séparez visiblement le trafic issu de l'IA du trafic normal. Une organisation qui fonctionne bien :

<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

Si vous suivez déjà un schéma pièce/appareil/propriété comme décrit dans notre article sur la structure des topics MQTT pour la maison connectée, la passerelle ajoute simplement la sous-arborescence ai/ à côté.

Implémenter la passerelle en Python

La passerelle ci-dessous utilise paho-mqtt 2.x et MQTT 5. Elle fonctionne sur le serveur MQTT public gratuit, réservé aux tests et aux prototypes, pour que vous puissiez l'essayer ; remplacez le préfixe par votre propre identifiant aléatoire.

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"   # use your own random id
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 commands per device per 10 minutes
MAX_STATE_AGE = 120            # seconds
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   # drop if not delivered within 15 s
    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)  # clear
        if msg.payload.decode().strip() == "approve":
            problem = validate(req)   # re-check: state may have changed
            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()

Testez-la depuis un terminal. Publiez d'abord un état d'appareil récent, puis une demande qui doit s'exécuter, puis une autre qui doit être bloquée :

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

La deuxième demande apparaît sur le topic d'audit comme rejected avec value outside 16..24. La zone publique étant lisible par tous, réservez-la à des expériences de ce type ; n'y faites jamais transiter le contrôle de vrais appareils.

Notes de conception qui comptent en pratique

Revalidez lors de la confirmation. Un humain peut approuver une demande cinq minutes après qu'elle a été émise. Le code rappelle validate : une limite de débit atteinte ou un état devenu périmé entre-temps bloque donc toujours la commande.

Faites expirer les commandes. Grâce à l'intervalle d'expiration des messages de MQTT 5, une commande en attente pour un appareil hors ligne est abandonnée au lieu d'être livrée une heure plus tard, quand la porte de garage se reconnecte. Notre présentation des fonctionnalités de MQTT 5 couvre l'expiration et le pattern du response topic, utilisable pour des accusés de réception plus riches.

Rapportez le résultat réel. L'agent doit dire à l'utilisateur « le thermostat a confirmé 21 degrés », et non « je l'ai réglé ». Attendez que le topic d'état change avant de répondre.

Gardez un arrêt d'urgence logiciel simple. Un false en retained message sur ai/enabled stoppe immédiatement toutes les actions de l'IA et survit aux redémarrages de la passerelle, car le broker le renvoie lors de l'abonnement.

Traitez les descriptions comme des entrées non fiables. Les noms d'appareils, notes et étiquettes de capteurs lus par le modèle peuvent contenir des instructions. La passerelle ignore le champ reason pour ses décisions ; il n'est journalisé que pour les humains.

Check-list du contrôle sécurisé par l'IA en production

L'agent ne possède aucun identifiant d'appareil ; seule la passerelle en dispose.
Chaque couple appareil/action est explicitement autorisé, avec ses plages.
Des limites de débit existent par appareil et globalement.
Les actions à fort impact exigent une confirmation humaine, et l'approbation relance la validation.
Les commandes portent une expiration et un identifiant unique.
Chaque décision est publiée sur un topic d'audit et stockée durablement par un abonné distinct.
Un arrêt d'urgence logiciel en retained message est documenté et testé.
Les verrouillages physiques restent dans le firmware ou l'automate.
Le trafic de production passe par TLS sur le port 8883 avec des identifiants issus du portail, comme le recommandent nos bonnes pratiques de sécurité MQTT.

De la démo à un broker privé

Basculez la connexion vers votre propre broker et vos identifiants :

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)

Un broker MQTT managé comme ConnectMQ vous fournit TLS, des identifiants générés, une protection contre le flood et un explorateur de topics en direct pour suivre le flux d'audit, tandis que la logique des garde-fous reste dans du code que vous maîtrisez. Pour situer les passerelles de commandes parmi les autres patterns d'IA, lisez le guide complet IA et IoT, et pour les choix d'hébergement, consultez notre page sur le broker MQTT cloud. Pour le volet langage naturel du même système, voyez le contrôle d'appareils en langage naturel.

Construisez une automatisation IA encadrée sur ConnectMQ

Commencez avec le serveur MQTT public gratuit, puis déplacez la passerelle vers un broker privé avec TLS et ses propres identifiants. Créez un broker Serverless (Free) sur app.connectmq.com, sans carte bancaire, et gardez chaque commande de l'IA visible sur un topic d'audit dès le premier jour.

Questions fréquentes

Puis-je compter sur le prompt système pour sécuriser l'IA ?

Non. Les prompts réduisent les erreurs mais ne peuvent pas garantir un comportement. Imposez les règles dans un code déterministe que le modèle ne peut pas contourner, et gardez les identifiants des appareils hors de l'agent.

ConnectMQ valide-t-il les commandes de l'IA à ma place ?

Non. ConnectMQ achemine les messages de façon sécurisée et fiable ; la validation, la confirmation et l'audit résident dans votre service de passerelle, qui se connecte à ConnectMQ comme un client MQTT ordinaire.

Quelles actions doivent toujours exiger une confirmation humaine ?

Tout ce qui déplace des mécanismes, ouvre des portes ou des vannes, déverrouille un accès ou pourrait causer des dégâts en cas de répétition. Les réglages de confort dans une plage étroite peuvent généralement être automatisés sans risque.

Comment conserver la piste d'audit sur le long terme ?

Exécutez un abonné distinct sur le topic d'audit qui écrit dans un fichier ou une base de données. Les messages d'audit ne sont pas conservés en retained message sur le broker : l'abonné doit donc être actif pour les capturer.

Contrôle sécurisé des appareils IoT par l'IA : garde-fous pour les commandes MQTT émises par une IA