Controle Seguro de Dispositivos IoT com IA: Proteções para Comandos MQTT Emitidos por IA

O controle seguro de dispositivos IoT com IA significa que um modelo de linguagem pode propor ações, mas apenas comandos validados, com limite de taxa e registrados chegam de fato a um relé, uma válvula ou um motor. Neste guia você vai projetar um gateway de comandos que fica entre um agente de IA e seus dispositivos MQTT, e implementar listas de permissões, verificações de faixa, limites de taxa, confirmação humana e um topic de auditoria em cerca de 150 linhas de Python.

Por que comandos de IA precisam de uma camada de segurança própria

Modelos de linguagem são bons em transformar "está quente demais na sala de reunião" em "ajustar o setpoint da sala 3 para 21 graus". Eles também são capazes de ler errado o número de uma sala, repetir um comando em loop ou seguir uma instrução escondida no rótulo de um sensor ou em um e-mail. Nada disso é um comportamento incomum para um gerador de texto probabilístico, e prompts sozinhos não conseguem descartá-lo.

A solução é a mesma que os engenheiros de controle sempre usaram: presumir que a fonte anterior pode estar errada e validar tudo em uma fronteira que você controla. A IA passa a ser mais um client não confiável, e um programa determinístico decide o que é permitido.

Há um segundo motivo. Mesmo um modelo perfeitamente correto pode agir com base em informações desatualizadas. Se ele leu um nível de tanque de dez minutos atrás e abre uma válvula de enchimento, o comando foi "razoável" e o resultado continua sendo uma inundação. O gateway é o lugar para verificar a atualidade dos dados e o estado atual antes de agir.

O padrão de gateway de comandos

A arquitetura tem uma única regra: a IA nunca detém credenciais do broker capazes de alcançar os dispositivos. Em vez disso:

1. O agente chama uma ferramenta como request_action(device, action, value, reason), conforme descrito em nosso guia sobre function calling de LLM para IoT.
2. A ferramenta entrega o pedido ao gateway, um pequeno serviço que é seu.
3. O gateway valida o pedido, aplica os limites, opcionalmente consulta um humano e só então publica no topic de comando do dispositivo.
4. Cada decisão, aceita ou rejeitada, é publicada em um topic de auditoria e gravada em um log.
5. O dispositivo confirma em um topic de estado, e o gateway informa ao agente o resultado real.

Isso importa porque um client MQTT com credenciais normalmente pode publicar em qualquer topic que a sua conta alcance. O ConnectMQ vincula os client IDs às credenciais que emite, mas não oferece ACLs por topic; portanto, a forma honesta de garantir que "a IA não pode tocar a bomba diretamente" é manter as credenciais dos dispositivos dentro do processo do gateway e totalmente fora do agente. Se o agente rodar como um servidor MCP para MQTT, exponha apenas ferramentas apoiadas pelo gateway, nunca um publish bruto.

Proteções para o controle seguro de dispositivos IoT com IA

Cada proteção trata de um modo de falha específico:

Os intertravamentos que protegem pessoas e equipamentos continuam no CLP ou no firmware do dispositivo e, em máquinas industriais no Brasil, seguem atendendo à NR-12. O gateway reduz incômodos e riscos; ele não substitui um sistema com classificação de segurança.

Organização de topics para comandos protegidos

Mantenha o tráfego originado pela IA visivelmente separado do tráfego normal. Uma organização que funciona bem:

<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

Se você já segue um esquema de cômodo/dispositivo/propriedade como o descrito em nosso artigo sobre estrutura de topics MQTT para casa inteligente, o gateway simplesmente adiciona a subárvore ai/ ao lado.

Implementando o gateway em Python

O gateway abaixo usa paho-mqtt 2.x e MQTT 5. Ele roda no servidor MQTT público (gratuito) para que você possa experimentá-lo; troque o prefixo pelo seu próprio ID aleatório.

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 seu próprio id aleatório
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)          # no máximo 3 comandos por dispositivo a 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   # descarta se não for entregue em 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)  # limpa
        if msg.payload.decode().strip() == "approve":
            problem = validate(req)   # verifica de novo: o estado pode ter mudado
            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()

Teste a partir de um terminal. Primeiro publique um estado atualizado do dispositivo, depois um pedido que deve ser executado e então um que deve ser bloqueado:

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

O segundo pedido aparece no topic de auditoria como rejected, com value outside 16..24. Como a área pública pode ser lida por qualquer pessoa, use-a apenas para experimentos como este; nunca coloque o controle de dispositivos reais nela.

Observações de design que importam na prática

Valide de novo na confirmação. Um humano pode aprovar um pedido cinco minutos depois de ele ter sido feito. O código chama validate novamente, então um limite de taxa ou um estado obsoleto que tenha surgido nesse meio-tempo ainda o bloqueia.

Faça os comandos expirarem. O intervalo de expiração de mensagens do MQTT 5 faz com que um comando enfileirado para um dispositivo offline seja descartado, em vez de entregue uma hora depois, quando a porta da garagem se reconectar. Nossa visão geral dos recursos do MQTT 5 aborda a expiração e o padrão de response topic, que você pode usar para confirmações mais ricas.

Informe o resultado real. O agente deve dizer ao usuário "o termostato confirmou 21 graus", e não "eu ajustei". Espere o topic de estado mudar antes de responder.

Deixe o kill switch sem surpresas. Um false retido em ai/enabled interrompe imediatamente todas as ações da IA e sobrevive a reinicializações do gateway, porque o broker o reenvia no momento da assinatura.

Trate descrições como entrada não confiável. Nomes de dispositivos, notas e rótulos de sensores que o modelo lê podem conter instruções. O gateway ignora o campo reason nas decisões; ele é registrado apenas para humanos.

Checklist de controle seguro por IA para produção

1. O agente não tem credenciais de dispositivos; apenas o gateway tem.
2. Cada par dispositivo/ação está explicitamente na lista de permissões, com faixas.
3. Existem limites de taxa por dispositivo e globais.
4. Ações de alto impacto exigem confirmação humana, e a aprovação executa a validação de novo.
5. Os comandos carregam uma expiração e um ID exclusivo.
6. Cada decisão é publicada em um topic de auditoria e armazenada de forma durável por um subscriber separado.
7. Um kill switch retido está documentado e testado.
8. Os intertravamentos físicos permanecem no firmware ou no CLP.
9. O tráfego de produção passa por TLS na porta 8883 com credenciais do portal, como recomendado em nossas boas práticas de segurança MQTT.

Da demonstração para um broker privado

Mude a conexão para o seu próprio broker e as suas credenciais:

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)

Um broker MQTT gerenciado como o ConnectMQ oferece TLS, credenciais geradas, proteção contra flood e um explorador de topics ao vivo para acompanhar o fluxo de auditoria, enquanto a lógica das proteções fica em código que você controla. Para ver onde os gateways de comandos se encaixam entre outros padrões de IA, leia o guia completo de IA e IoT e, para entender as opções de hospedagem, veja nossa página sobre o broker MQTT na nuvem. Para o lado de linguagem natural do mesmo sistema, veja controle de dispositivos por linguagem natural.

Construa automação com IA protegida no ConnectMQ

Comece com o servidor MQTT público, apenas para testes e protótipos, e depois mova o gateway para um broker MQTT gerenciado e privado com TLS e credenciais próprias. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, e mantenha cada comando da IA visível em um topic de auditoria desde o primeiro dia.

Perguntas frequentes

Posso confiar no prompt de sistema para manter a IA segura?

Não. Os prompts reduzem erros, mas não garantem o comportamento. Imponha as regras em código determinístico que o modelo não consiga contornar e mantenha as credenciais dos dispositivos fora do agente.

O ConnectMQ valida os comandos da IA para mim?

Não. O ConnectMQ entrega as mensagens com segurança e confiabilidade; a validação, a confirmação e a auditoria ficam no seu serviço de gateway, que se conecta ao ConnectMQ como um client MQTT comum.

Quais ações devem sempre exigir confirmação humana?

Qualquer coisa que mova mecanismos, abra portas ou válvulas, libere acessos ou possa causar danos se repetida. Ajustes de conforto dentro de uma faixa estreita geralmente são seguros para automatizar.

Como armazeno a trilha de auditoria a longo prazo?

Execute um subscriber separado no topic de auditoria que grave em um arquivo ou banco de dados. As mensagens de auditoria não são retidas no broker, então o subscriber precisa estar em execução para capturá-las.