Control seguro de dispositivos IoT con IA y MQTT

El control seguro de dispositivos IoT con IA significa que un modelo de lenguaje puede proponer acciones, pero solo los comandos validados, con límite de frecuencia y registrados llegan alguna vez a un relé, una válvula o un motor. En esta guía diseñarás un gateway de comandos que se sitúa entre un agente de IA y tus dispositivos MQTT, e implementarás listas permitidas, comprobaciones de rango, límites de frecuencia, confirmación humana y un topic de auditoría en unas 150 líneas de Python.

Por qué los comandos de la IA necesitan su propia capa de seguridad

Los modelos de lenguaje convierten muy bien "hace demasiado calor en la sala de reuniones" en "fija la consigna de la sala 3 en 21 grados". También son capaces de leer mal un número de sala, repetir un comando en bucle o seguir una instrucción oculta en la etiqueta de un sensor o en un correo. Nada de eso es un comportamiento extraño para un generador de texto probabilístico, y los prompts por sí solos no pueden descartarlo.

La solución es la misma que los ingenieros de control han usado siempre: suponer que la fuente anterior puede equivocarse y validarlo todo en una frontera que tú controlas. La IA se convierte en un cliente no fiable más, y un programa determinista decide qué está permitido.

Hay un segundo motivo. Incluso un modelo perfectamente correcto puede actuar con información obsoleta. Si leyó un nivel de depósito de hace diez minutos y abre una válvula de llenado, el comando era "razonable" y el resultado sigue siendo una inundación. El gateway es el lugar donde comprobar la frescura y el estado actual antes de actuar.

El patrón de gateway de comandos

La arquitectura tiene una única regla: la IA nunca tiene credenciales de broker que puedan llegar a los dispositivos. En su lugar:

El agente llama a una herramienta como request_action(device, action, value, reason), como se describe en nuestra guía sobre function calling de LLM para IoT.
La herramienta entrega la petición al gateway, un pequeño servicio que es tuyo.
El gateway valida la petición, aplica límites, opcionalmente pregunta a una persona y solo entonces publica en el topic de comandos del dispositivo.
Cada decisión, aceptada o rechazada, se publica en un topic de auditoría y se escribe en un log.
El dispositivo confirma en un topic de estado, y el gateway comunica el resultado real al agente.

Esto importa porque un cliente MQTT con credenciales normalmente puede publicar en cualquier topic al que llegue su cuenta. ConnectMQ vincula los client IDs a las credenciales que emite, pero no ofrece ACL por topic, así que la forma honesta de garantizar que "la IA no puede tocar la bomba directamente" es mantener las credenciales de los dispositivos dentro del proceso del gateway y completamente fuera del agente. Si el agente se ejecuta como un servidor MCP para MQTT, expón solo herramientas respaldadas por el gateway, nunca un publish directo.

Salvaguardas para el control seguro de dispositivos IoT con IA

Cada salvaguarda aborda un modo de fallo concreto:

Salvaguarda	Qué evita	Implementación típica
Lista permitida de dispositivos y acciones	Comandos a equipos que la IA nunca debería tocar	Diccionario estático en el gateway, revisado como el código
Rangos de valores	Consignas inseguras como 90 grados en un termostato	Mínimo, máximo y paso por acción
Límites de frecuencia	Bucles y oscilaciones	Máximo de comandos por dispositivo y ventana
Comprobaciones de frescura	Actuar con lecturas obsoletas	Rechazar si la última actualización de estado es demasiado antigua
Confirmación humana	Acciones de alto impacto ejecutadas a ciegas	Cola de pendientes, aprobación mediante un topic de confirmación
Caducidad de comandos	Comandos antiguos ejecutados tras una reconexión	Intervalo de caducidad de mensajes de MQTT 5
Rastro de auditoría	Acciones imposibles de rastrear	Topic de auditoría más log de solo anexado
Interruptor de emergencia	Automatización desbocada	Un único indicador retenido que desactiva todas las acciones de la IA

Los enclavamientos que protegen a las personas y los equipos se quedan en el PLC o en el firmware del dispositivo. El gateway reduce molestias y riesgos; no sustituye a un sistema con certificación de seguridad.

Estructura de topics para comandos protegidos

Mantén el tráfico originado por la IA claramente separado del tráfico normal. Una estructura que funciona 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 ya sigues un esquema habitación/dispositivo/propiedad como el descrito en nuestro artículo sobre la estructura de topics MQTT para el hogar inteligente, el gateway simplemente añade el subárbol ai/ a su lado.

Implementar el gateway en Python

El gateway siguiente usa paho-mqtt 2.x y MQTT 5. Se ejecuta contra el servidor MQTT público gratuito en broker.connectmq.com, pensado para pruebas y no para datos de producción, para que puedas probarlo; cambia el prefijo por tu propio ID aleatorio.

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"   # usa tu propio id aleatorio
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)          # máximo 3 comandos por dispositivo cada 10 minutos
MAX_STATE_AGE = 120            # segundos
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   # descartar si no se entrega en 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)  # limpiar
        if msg.payload.decode().strip() == "approve":
            problem = validate(req)   # volver a comprobar: el estado puede haber cambiado
            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()

Pruébalo desde una terminal. Primero publica un estado reciente del dispositivo, luego una petición que debería ejecutarse y después una que debería bloquearse:

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 segunda petición aparece en el topic de auditoría como rejected con value outside 16..24. Como el área pública es legible por cualquiera, úsala solo para experimentos como este; nunca pongas en ella el control de dispositivos reales.

Notas de diseño que importan en la práctica

Vuelve a validar al confirmar. Una persona puede aprobar una petición cinco minutos después de que se hiciera. El código vuelve a llamar a validate, así que un límite de frecuencia o un estado obsoleto que haya aparecido entretanto sigue bloqueándola.

Haz que los comandos caduquen. El intervalo de caducidad de mensajes de MQTT 5 hace que un comando en cola para un dispositivo desconectado se descarte en lugar de entregarse una hora después, cuando la puerta del garaje se vuelva a conectar. Nuestro repaso de las características de MQTT 5 cubre la caducidad y el patrón de response topic que puedes usar para confirmaciones más ricas.

Comunica el resultado real. El agente debe decirle al usuario "el termostato ha confirmado 21 grados", no "lo he fijado". Espera a que cambie el topic de estado antes de responder.

Haz que el interruptor de emergencia sea aburrido. Un false retenido en ai/enabled detiene de inmediato todas las acciones de la IA y sobrevive a los reinicios del gateway, porque el broker lo vuelve a entregar al suscribirse.

Trata las descripciones como entrada no fiable. Los nombres de dispositivos, notas y etiquetas de sensores que lee el modelo pueden contener instrucciones. El gateway ignora el campo reason para tomar decisiones; solo se registra para las personas.

Checklist de control seguro con IA para producción

El agente no tiene credenciales de dispositivos; solo las tiene el gateway.
Cada par dispositivo/acción está explícitamente en la lista permitida, con sus rangos.
Existen límites de frecuencia por dispositivo y globales.
Las acciones de alto impacto requieren confirmación humana, y la aprobación vuelve a ejecutar la validación.
Los comandos llevan una caducidad y un ID único.
Cada decisión se publica en un topic de auditoría y la almacena de forma duradera un suscriptor independiente.
Hay un interruptor de emergencia retenido documentado y probado.
Los enclavamientos físicos permanecen en el firmware o en el PLC.
El tráfico de producción va por TLS en el puerto 8883 con credenciales del portal, como se recomienda en nuestras buenas prácticas de seguridad MQTT.

Pasar de la demo a un broker privado

Cambia la conexión a tu propio broker y tus credenciales:

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 gestionado como ConnectMQ te da TLS, credenciales generadas, protección contra inundaciones y un explorador de topics en vivo para observar el flujo de auditoría, mientras la lógica de salvaguardas permanece en código que tú controlas. Para ver dónde encajan los gateways de comandos entre otros patrones de IA, lee la guía completa de IA e IoT, y para conocer las opciones de alojamiento consulta nuestra página sobre el broker MQTT en la nube. Para la parte de lenguaje natural del mismo sistema, consulta control de dispositivos con lenguaje natural.

Construye automatización con IA protegida sobre ConnectMQ

Empieza con el servidor MQTT público gratuito y después lleva el gateway a un broker privado con TLS y sus propias credenciales. Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y mantén cada comando de la IA visible en un topic de auditoría desde el primer día.

Preguntas frecuentes

¿Puedo confiar en el prompt del sistema para mantener segura la IA?

No. Los prompts reducen los errores, pero no garantizan el comportamiento. Impón las reglas en código determinista que el modelo no pueda eludir y mantén las credenciales de los dispositivos fuera del agente.

¿Valida ConnectMQ los comandos de la IA por mí?

No. ConnectMQ entrega los mensajes de forma segura y fiable; la validación, la confirmación y la auditoría viven en tu servicio de gateway, que se conecta a ConnectMQ como un cliente MQTT corriente.

¿Qué acciones deberían necesitar siempre confirmación humana?

Cualquier cosa que mueva mecanismos, abra puertas o válvulas, desbloquee accesos o pueda causar daños si se repite. Los ajustes de confort dentro de un rango estrecho suelen poder automatizarse con seguridad.

¿Cómo guardo el rastro de auditoría a largo plazo?

Ejecuta un suscriptor independiente sobre el topic de auditoría que escriba en un archivo o en una base de datos. Los mensajes de auditoría no se retienen en el broker, así que el suscriptor debe estar en marcha para capturarlos.

Control seguro de dispositivos IoT con IA: salvaguardas para comandos MQTT emitidos por IA