Como testar ACLs MQTT e permissões de tópicos

Para testar ACLs MQTT, conecte como o usuário que você quer verificar, tente cada publish e subscribe que o dispositivo deve e não deve conseguir fazer, e confira o resultado de fora do cliente. Essa última parte é essencial: no MQTT 3.1.1 um publish negado parece exatamente igual a um bem-sucedido, então a única prova confiável é um segundo cliente observando se a mensagem chegou.

Este guia mostra como cada versão do protocolo informa uma negação, uma matriz de testes de permissões, verificações manuais com mosquitto e um teste automatizado que você pode rodar após cada mudança de ACL.

Como os brokers informam uma ação negada

ACLs (listas de controle de acesso) definem em quais tópicos um usuário pode publicar (escrita) e assinar (leitura). Neste ponto a autenticação já deu certo — o CONNACK foi 0x00 —, então todo o retorno vem em pacotes posteriores.

O comportamento do 3.1.1 não é um bug do seu broker. A especificação diz que, quando o servidor não autoriza um PUBLISH, ele não tem como informar o cliente e deve confirmá-lo segundo as regras normais de QoS ou fechar a conexão de rede. Muitos clientes não registram nada em nenhum dos casos, e é por isso que "minhas mensagens somem" é um sintoma tão comum de ACL — veja ACL MQTT negada em silêncio.

Monte uma matriz de testes de permissões

Liste o que um usuário representativo deve poder fazer e acrescente o que ele não pode fazer. Aqui está um exemplo para um sensor que pode publicar a própria telemetria e ler os próprios comandos, com uma ACL do Mosquitto que implementa isso:

# every device: own telemetry and own commands only
pattern write testmqtt/acl/telemetry/%u/#
pattern read  testmqtt/acl/commands/%u/#

# a monitoring user that may read everything under testmqtt/acl
user acl-observer
topic read testmqtt/acl/#

O caso 9 é onde os padrões pegam as pessoas de surpresa: telemetry/%u/# corresponde ao próprio telemetry/sensor-01, porque # também inclui o nível pai. Cole seus padrões e tópicos no topic matcher para conferir os casos-limite antes de escrever o teste. O caso 8 importa porque mensagens retidas são entregues no momento da assinatura; um broker que esquece de aplicar a ACL de leitura nesse ponto vaza o último valor de cada tópico. Para mais padrões problemáticos, veja segurança de wildcards MQTT.

Verificações manuais com mosquitto

Abra primeiro um observador. Ele usa uma conta com amplo acesso de leitura, então mostra o que realmente chega ao broker:

mosquitto_sub -h broker.connectmq.com -p 1883 \
  -u acl-observer -P "$OBSERVER_PASS" -t 'testmqtt/acl/#' -v

Agora publique como o dispositivo em teste usando MQTT 5 e QoS 1, para que uma negação gere um reason code:

# caso 1: deve aparecer no observador
mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -q 1 \
  -u sensor-01 -P "$MQTT_PASS" -t testmqtt/acl/telemetry/sensor-01/temp -m 21.5

# caso 2: NÃO deve aparecer; o MQTT 5 informa a negação
mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -q 1 \
  -u sensor-01 -P "$MQTT_PASS" -t testmqtt/acl/telemetry/sensor-02/temp -m 99

No publish negado, o cliente mosquitto imprime um aviso como Warning: Publish 1 failed: Not authorized. Rode o mesmo comando com -V mqttv311 e ele termina sem aviso nenhum — o silêncio do observador é sua única evidência.

Assinaturas são mais fáceis de verificar, porque as duas versões respondem no SUBACK:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -d \
  -u sensor-01 -P "$MQTT_PASS" -t 'testmqtt/acl/commands/sensor-02/#'

Com -d você vê Subscribed (mid: 1): 135 para uma negação no MQTT 5 (0x87) ou Subscribed (mid: 1): 128 (0x80) no 3.1.1. Versões recentes do mosquitto então imprimem All subscription requests were denied. e saem. Se o seu broker conceder a assinatura, publique nesse tópico a partir de outra conta e confirme que nada é entregue. Os códigos são explicados em detalhe em SUBACK 0x80 e Not authorized 0x87.

Você também pode fazer isso de forma interativa no cliente MQTT online: abra duas sessões com usuários diferentes, assine como observador em uma e publique como dispositivo na outra.

Um teste de ACL automatizado

Este script com paho-mqtt 2.x executa os casos de publish e subscribe da matriz. Ele verifica os reason codes do MQTT 5 e usa o observador como fonte da verdade, então funciona também com brokers que descartam em silêncio as mensagens negadas:

import os, threading, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
DEVICE = ("sensor-01", os.environ["MQTT_PASS"])
OBSERVER = ("acl-observer", os.environ["OBSERVER_PASS"])
BASE = "testmqtt/acl"

def client(creds):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                    client_id=f"acl-{uuid.uuid4().hex[:8]}", protocol=mqtt.MQTTv5)
    c.username_pw_set(*creds)
    c.connect(HOST, PORT)
    c.loop_start()
    time.sleep(0.5)
    return c

seen = set()
obs = client(OBSERVER)
obs.on_message = lambda c, u, m: seen.add(m.topic)
obs.subscribe(f"{BASE}/#", qos=1)

dev = client(DEVICE)
acks = {}
dev.on_publish = lambda c, u, mid, rc, props: acks.__setitem__(mid, str(rc))
subacks = {}
dev.on_subscribe = lambda c, u, mid, rcs, props: subacks.__setitem__(mid, [str(r) for r in rcs])

def check_publish(topic, allowed):
    info = dev.publish(topic, "acl-test", qos=1)
    time.sleep(1)
    delivered = topic in seen
    print(f"PUB {topic}: puback={acks.get(info.mid)} delivered={delivered}")
    assert delivered == allowed, f"publish to {topic}: expected allowed={allowed}"
    if not allowed:
        assert acks.get(info.mid) == "Not authorized"

def check_subscribe(topic, allowed):
    _, mid = dev.subscribe(topic, qos=1)
    time.sleep(1)
    granted = not subacks[mid][0].startswith(("Not authorized", "Unspecified"))
    print(f"SUB {topic}: suback={subacks[mid]}")
    assert granted == allowed, f"subscribe to {topic}: expected allowed={allowed}"

check_publish(f"{BASE}/telemetry/sensor-01/temp", True)
check_publish(f"{BASE}/telemetry/sensor-02/temp", False)
check_publish(f"{BASE}/commands/sensor-01/reboot", False)
check_subscribe(f"{BASE}/commands/sensor-01/#", True)
check_subscribe(f"{BASE}/commands/sensor-02/#", False)
print("all ACL checks passed")

Rode-o após cada mudança no arquivo de ACL ou no backend de autenticação, e no CI contra um broker de staging. Quando uma verificação falha, o reason code impresso e a flag delivered mostram na hora se o broker rejeitou a mensagem, descartou em silêncio ou deixou passar.

Motivos comuns para um teste de ACL falhar

• Tudo é negado. O arquivo de ACL não foi carregado, ou o nome de usuário na regra não bate com o do CONNECT. Confira o log do broker na inicialização; no Mosquitto, acl_file precisa estar no mesmo escopo de listener que o arquivo de senhas — veja autenticação no Mosquitto.
• Tudo é permitido. Uma regra ampla como topic readwrite # no início do arquivo, ou acesso anônimo com regras próprias permissivas.
• As regras pattern não batem. %u é o nome de usuário e %c o client ID; usar o errado nega todos os dispositivos cujo ID é diferente do usuário.
• As mudanças não têm efeito. A maioria dos brokers lê as ACLs na inicialização ou no reload. Envie SIGHUP ao Mosquitto, ou reinicie, antes de repetir o teste.
• O teste passa, mas a produção falha. O cliente publica com uma barra inicial (/testmqtt/...) ou com outra capitalização. Tópicos diferenciam maiúsculas de minúsculas e /a não é a.

Acompanhe o tráfego do teste de ACL no cliente online

Uma ACL é um controle de segurança, e controles de segurança precisam de testes que falhem quando o controle quebra. Mantenha uma matriz de ações permitidas e negadas por papel, verifique a entrega com um observador em vez de confiar no publicador e prefira MQTT 5 para que as negações sejam informadas em vez de engolidas.

Perguntas frequentes