Cómo probar las ACL de MQTT y los permisos por topic

Para probar las ACL de MQTT, conéctate como el usuario que quieres comprobar, intenta cada publish y subscribe que el dispositivo debería y no debería poder hacer, y verifica el resultado desde fuera del cliente. Esto último es clave: en MQTT 3.1.1 un publish denegado parece exactamente igual que uno exitoso, así que la única prueba fiable es un segundo cliente que observa si el mensaje llegó.

Esta guía explica cómo informa cada versión del protocolo de una denegación, propone una matriz de pruebas de permisos, comprobaciones manuales con mosquitto y una prueba automática que puedes ejecutar tras cada cambio de ACL.

Cómo informan los brokers de una acción denegada

Las ACL (listas de control de acceso) deciden en qué topics puede publicar (escritura) y a cuáles puede suscribirse (lectura) un usuario. En este punto la autenticación ya fue correcta —el CONNACK fue 0x00—, así que toda la información llega en paquetes posteriores.

El comportamiento de 3.1.1 no es un bug de tu broker. La especificación dice que, cuando un servidor no autoriza un PUBLISH, no tiene forma de informar al cliente y debe confirmarlo según las reglas normales de QoS o cerrar la conexión de red. Muchos clientes no registran nada en ninguno de los dos casos, por eso "mis mensajes desaparecen" es un síntoma tan común de las ACL; consulta ACL de MQTT denegada en silencio.

Construye una matriz de pruebas de permisos

Enumera lo que un usuario representativo debería poder hacer y añade lo que no debe hacer. Este es un ejemplo para un sensor que puede publicar su propia telemetría y leer sus propios comandos, con una ACL de Mosquitto que lo implementa:

# 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/#

El caso 9 es donde los patrones sorprenden: telemetry/%u/# coincide con el propio telemetry/sensor-01, porque # también incluye el nivel padre. Pega tus patrones y topics en el topic matcher para comprobar los casos límite antes de escribir la prueba. El caso 8 importa porque los mensajes retenidos se entregan en el momento de suscribirse; un broker que olvida aplicar ahí la ACL de lectura filtra el último valor de cada topic. Para más patrones problemáticos, consulta seguridad de wildcards MQTT.

Comprobaciones manuales con mosquitto

Abre primero un observador. Usa una cuenta con amplio acceso de lectura, así muestra lo que realmente llega al broker:

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

Ahora publica como el dispositivo bajo prueba con MQTT 5 y QoS 1, para que una denegación genere un reason code:

# caso 1: debe aparecer en el 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: NO debe aparecer; MQTT 5 informa de la denegación
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

En el publish denegado, el cliente mosquitto imprime un aviso como Warning: Publish 1 failed: Not authorized. Ejecuta el mismo comando con -V mqttv311 y terminará sin ningún aviso: el silencio del observador es tu única evidencia.

Las suscripciones son más fáciles de verificar porque ambas versiones responden en el SUBACK:

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

Con -d verás Subscribed (mid: 1): 135 para una denegación en MQTT 5 (0x87) o Subscribed (mid: 1): 128 (0x80) en 3.1.1. Las versiones recientes de mosquitto imprimen después All subscription requests were denied. y salen. Si tu broker concede la suscripción, publica en ese topic desde otra cuenta y confirma que no se entrega nada. Los códigos se explican en detalle en SUBACK 0x80 y Not authorized 0x87.

También puedes hacerlo de forma interactiva en el cliente MQTT online: abre dos sesiones con usuarios distintos, suscríbete como observador en una y publica como dispositivo en la otra.

Una prueba de ACL automatizada

Este script con paho-mqtt 2.x ejecuta los casos de publish y subscribe de la matriz. Comprueba los reason codes de MQTT 5 y usa al observador como fuente de verdad, así que también funciona con brokers que descartan en silencio los mensajes denegados:

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")

Ejecútalo después de cada cambio en el fichero de ACL o en el backend de autenticación, y en CI contra un broker de staging. Cuando una comprobación falla, el reason code impreso y el indicador delivered te dicen al instante si el broker rechazó el mensaje, lo descartó en silencio o lo dejó pasar.

Motivos habituales por los que falla una prueba de ACL

• Todo está denegado. El fichero de ACL no se cargó, o el nombre de usuario de la regla no coincide con el del CONNECT. Revisa el log del broker al arrancar; en Mosquitto, acl_file debe estar en el mismo ámbito de listener que el fichero de contraseñas; consulta autenticación en Mosquitto.
• Todo está permitido. Una regla amplia como topic readwrite # al principio del fichero, o acceso anónimo con sus propias reglas permisivas.
• Las reglas pattern no coinciden. %u es el nombre de usuario y %c el client ID; usar el equivocado deniega a todos los dispositivos cuyo ID difiere de su usuario.
• Los cambios no tienen efecto. La mayoría de brokers leen las ACL al arrancar o al recargar. Envía SIGHUP a Mosquitto, o reinícialo, antes de repetir la prueba.
• La prueba pasa pero producción falla. El cliente publica con una barra inicial (/testmqtt/...) o con otras mayúsculas. Los topics distinguen mayúsculas y /a no es a.

Observa el tráfico de la prueba de ACL en el cliente online

Una ACL es un control de seguridad, y los controles de seguridad necesitan pruebas que fallen cuando el control se rompe. Mantén una matriz de acciones permitidas y denegadas por rol, verifica la entrega con un observador en lugar de fiarte del publicador, y prefiere MQTT 5 para que las denegaciones se notifiquen en vez de tragarse.

Preguntas frecuentes