Probar la autenticación MQTT: credenciales válidas, inválidas y ausentes

Probar la autenticación MQTT consiste en conectarse con todos los tipos de credenciales que tu broker puede recibir —válidas, incorrectas, desconocidas, vacías y ausentes— y comprobar que cada una obtiene el código CONNACK correcto. Un broker que acepta la contraseña buena solo está probado a medias; la seguridad depende de cómo rechaza todo lo demás.

Esta guía te da una matriz de pruebas con el resultado esperado en cada caso para MQTT 3.1.1 y MQTT 5, los comandos para ejecutarla a mano y una suite pytest que la ejecuta automáticamente en CI.

Cómo informa MQTT del resultado de la autenticación

La autenticación ocurre una sola vez, en el paquete CONNECT. El cliente envía un usuario y/o una contraseña, y el broker responde con CONNACK. Un código distinto de cero significa que la conexión fue rechazada, y el broker cierra el socket justo después de enviarlo.

La especificación no obliga al broker a elegir entre "contraseña incorrecta" y "no autorizado". Algunos brokers devuelven 0x04/0x86 para una contraseña errónea; Mosquitto devuelve 0x05/0x87 para cualquier fallo de credenciales. No pasa nada: lo importante es que tu broker sea coherente y que tus pruebas fijen su comportamiento real. Consulta cualquier código en la herramienta de reason codes MQTT.

La matriz de pruebas de autenticación

Ejecuta cada fila contra el broker que vas a usar en producción, al menos una vez con MQTT 3.1.1 y otra con MQTT 5. "Rechazada" significa 0x04/0x05 en 3.1.1 y 0x86/0x87 en 5.

Los casos 3 y 9 detectan bugs reales. Si un usuario desconocido recibe un código distinto al de una contraseña incorrecta, el broker filtra qué cuentas existen. Si se acepta un salto de línea final, el fichero de contraseñas o el backend de autenticación está recortando la entrada: normalmente inofensivo, pero conviene saberlo. El caso 10 comprueba que la revocación tiene efecto: en Mosquitto significa que el fichero de contraseñas se recargó (SIGHUP); en un broker gestionado, que el cambio se propagó.

Ejecutar la matriz a mano

mosquitto_pub es la forma más rápida de probar cada caso. Añade -d para ver el CONNACK en bruto:

# 1. credenciales válidas: termina en silencio con código 0
mosquitto_pub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/auth/check -m ok -d

# 2. contraseña incorrecta
mosquitto_pub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "wrong" \
  -t testmqtt/auth/check -m ok -d

# 3. usuario desconocido
mosquitto_pub -h broker.connectmq.com -p 1883 -u "no-such-user" -P "wrong" \
  -t testmqtt/auth/check -m ok -d

# 4. sin credenciales
mosquitto_pub -h broker.connectmq.com -p 1883 -t testmqtt/auth/check -m ok -d

Una conexión rechazada imprime una de estas líneas, que son justo las cadenas que la gente suele buscar:

Connection error: Connection Refused: bad user name or password.
Connection error: Connection Refused: not authorised.

Con -V mqttv5, los mismos casos imprimen el texto del reason code de MQTT 5, por ejemplo Connection error: Not authorized. Para una comprobación visual, el cliente MQTT online muestra el resultado del CONNACK en cada intento; resulta práctico para comparar los listeners WebSocket y TCP, que algunos brokers configuran con backends de autenticación distintos.

Si en su lugar ves Connection Refused: identifier rejected, el problema es el client ID, no las credenciales: consulta identifier rejected. Si el cliente se queda colgado y luego expira, probablemente estás en el puerto equivocado o detrás de un firewall, no fallando la autenticación.

Automatizarlo con pytest

Las pruebas manuales demuestran que el broker funciona hoy. Una suite automatizada demuestra que sigue funcionando tras el próximo cambio de configuración. Esta prueba usa paho-mqtt 2.x y ejecuta cada caso con las dos versiones del protocolo:

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

HOST, PORT = "broker.connectmq.com", 1883
USER, PASS = os.environ["MQTT_USER"], os.environ["MQTT_PASS"]
REFUSED = {"Bad user name or password", "Not authorized"}

def try_connect(username=None, password=None, protocol=mqtt.MQTTv5, timeout=5):
    """Devuelve el texto del reason code del CONNACK, o 'closed' si no llegó CONNACK."""
    result, done = {}, threading.Event()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                    client_id=f"auth-test-{uuid.uuid4().hex[:8]}", protocol=protocol)
    if username is not None or password is not None:
        c.username_pw_set(username, password)

    def on_connect(client, userdata, flags, reason_code, properties):
        result["rc"] = str(reason_code)
        done.set()

    c.on_connect = on_connect
    c.connect(HOST, PORT, keepalive=10)
    c.loop_start()
    done.wait(timeout)
    c.disconnect()
    c.loop_stop()
    return result.get("rc", "closed")

PROTOCOLS = [mqtt.MQTTv311, mqtt.MQTTv5]

@pytest.mark.parametrize("proto", PROTOCOLS)
def test_valid_credentials(proto):
    assert try_connect(USER, PASS, proto) == "Success"

@pytest.mark.parametrize("proto", PROTOCOLS)
@pytest.mark.parametrize("user,pw", [
    (USER, "wrong-password"),
    ("no-such-user", "wrong-password"),
    (USER, ""),
    (USER, PASS + "\n"),
])
def test_bad_credentials_refused(proto, user, pw):
    assert try_connect(user, pw, proto) in REFUSED

def test_username_is_case_sensitive():
    if USER.upper() == USER:
        pytest.skip("username has no lowercase letters")
    assert try_connect(USER.upper(), PASS) in REFUSED

@pytest.mark.parametrize("proto", PROTOCOLS)
def test_anonymous_refused(proto):
    assert try_connect(protocol=proto) in REFUSED | {"closed"}

def test_unknown_user_looks_like_wrong_password():
    assert try_connect(USER, "wrong") == try_connect("no-such-user", "wrong")

Algunos detalles hacen que sea fiable:

• paho-mqtt 2.x traduce los return codes de MQTT 3.1.1 a reason codes de MQTT 5, así que 0x04 llega como "Bad user name or password" y 0x05 como "Not authorized" en ambas versiones. Un único conjunto de aserciones cubre las dos.
• Cada intento usa un client ID aleatorio nuevo. Reutilizar uno puede hacer que una prueba exitosa expulse una conexión anterior y confunda el resultado. El generador de client IDs produce IDs con el mismo formato.
• "closed" cubre los brokers que cortan la conexión TCP en lugar de enviar un CONNACK, algo legal en algunos fallos.
• La última prueba es la comprobación de enumeración de cuentas del caso 3 de la matriz.

Ejecútala con MQTT_USER=... MQTT_PASS=... pytest -v test_auth.py. En CI, guarda las credenciales como secrets y ejecuta la suite contra un broker de staging en cada cambio de la configuración del broker: consulta pruebas MQTT en GitHub Actions.

Lo que las pruebas de autenticación no cubren

Un CONNECT exitoso solo demuestra la identidad. No dice nada de lo que el cliente puede hacer después. Un sensor con credenciales válidas no debería poder publicar en el topic de comandos de otro dispositivo: eso es autorización, y falla de otra manera (en silencio en 3.1.1, con un reason code en MQTT 5). Pruébala por separado con el método de cómo probar ACLs MQTT.

Ten en cuenta además:

• TCP sin cifrar expone las contraseñas. Las credenciales del CONNECT no van cifradas. Ejecuta también el caso de credenciales válidas por TLS en el puerto 8883 y confirma que el listener sin cifrar está desactivado en producción.
• La protección contra fuerza bruta es otra prueba. Algunos brokers o proxies ralentizan o bloquean los fallos repetidos. Si el tuyo lo hace, añade una prueba de que diez contraseñas incorrectas seguidas no bloquean al cliente legítimo, o de que sí lo hacen, si esa es tu política.
• La autenticación con tokens necesita pruebas de expiración. Si usas JWT u OAuth en lugar de contraseñas estáticas, añade casos para tokens caducados, todavía no válidos y mal firmados.

Prueba credenciales válidas e inválidas en el cliente online

Probar la autenticación es barato: una docena de intentos de conexión y unas pocas aserciones. Guarda la matriz en tu repositorio junto a la configuración del broker, ejecútala en cada cambio y te darás cuenta del día en que alguien reactiva el acceso anónimo antes que un atacante.

Preguntas frecuentes