Testando a autenticação MQTT: credenciais válidas, inválidas e ausentes

Testar a autenticação MQTT significa conectar com todo tipo de credencial que o seu broker pode receber — válida, errada, desconhecida, vazia e ausente — e verificar que cada uma recebe o código CONNACK correto. Um broker que aceita a senha certa está testado só pela metade; a segurança vem da forma como ele rejeita todo o resto.

Este guia traz uma matriz de testes com o resultado esperado em cada caso para MQTT 3.1.1 e MQTT 5, os comandos para executá-la manualmente e uma suíte pytest que a executa automaticamente no CI.

Como o MQTT informa o resultado da autenticação

A autenticação acontece uma única vez, no pacote CONNECT. O cliente envia um usuário e/ou uma senha, e o broker responde com CONNACK. Um código diferente de zero significa que a conexão foi recusada, e o broker fecha o socket logo depois de enviá-lo.

A especificação não obriga o broker a escolher entre "senha errada" e "não autorizado". Alguns brokers retornam 0x04/0x86 para uma senha errada; o Mosquitto retorna 0x05/0x87 para qualquer falha de credencial. Tudo bem: o que importa é que o seu broker seja consistente e que os seus testes fixem o comportamento real dele. Consulte qualquer código na ferramenta de reason codes MQTT.

A matriz de testes de autenticação

Execute cada linha contra o broker que você vai usar em produção, pelo menos uma vez com MQTT 3.1.1 e uma vez com MQTT 5. "Recusada" significa 0x04/0x05 no 3.1.1 e 0x86/0x87 no 5.

Os casos 3 e 9 pegam bugs reais. Se um usuário desconhecido recebe um código diferente de uma senha errada, o broker vaza quais contas existem. Se uma quebra de linha no final é aceita, o arquivo de senhas ou o backend de autenticação está cortando a entrada — geralmente inofensivo, mas é bom saber. O caso 10 verifica se a revogação realmente tem efeito: no Mosquitto isso significa que o arquivo de senhas foi recarregado (SIGHUP); num broker gerenciado, que a alteração foi propagada.

Executando a matriz manualmente

O mosquitto_pub é a forma mais rápida de testar cada caso. Adicione -d para ver o CONNACK bruto:

# 1. credenciais válidas — termina em silêncio com 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. senha errada
mosquitto_pub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "wrong" \
  -t testmqtt/auth/check -m ok -d

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

# 4. sem nenhuma credencial
mosquitto_pub -h broker.connectmq.com -p 1883 -t testmqtt/auth/check -m ok -d

Uma conexão recusada imprime uma destas linhas, que são exatamente as mensagens que as pessoas costumam pesquisar:

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

Com -V mqttv5, os mesmos casos imprimem o texto do reason code do MQTT 5, por exemplo Connection error: Not authorized. Para uma verificação visual, o cliente MQTT online mostra o resultado do CONNACK a cada tentativa — útil para comparar os listeners WebSocket e TCP, que alguns brokers configuram com backends de autenticação diferentes.

Se aparecer Connection Refused: identifier rejected, o problema é o client ID, não as credenciais — veja identifier rejected. Se o cliente trava e depois dá timeout, provavelmente você está na porta errada ou atrás de um firewall, e não falhando na autenticação.

Automatizando com pytest

Testes manuais provam que o broker funciona hoje. Uma suíte automatizada prova que ele continua funcionando depois da próxima mudança de configuração. Este teste usa paho-mqtt 2.x e executa cada caso nas duas versões do 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):
    """Retorna o texto do reason code do CONNACK, ou 'closed' se nenhum CONNACK chegou."""
    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")

Alguns detalhes tornam o teste confiável:

• O paho-mqtt 2.x converte os return codes do MQTT 3.1.1 em reason codes do MQTT 5, então 0x04 chega como "Bad user name or password" e 0x05 como "Not authorized" nas duas versões. Um único conjunto de asserções cobre ambas.
• Cada tentativa usa um client ID aleatório novo. Reutilizar um pode fazer um teste bem-sucedido derrubar uma conexão anterior e confundir o resultado. O gerador de client IDs cria IDs no mesmo formato.
• "closed" cobre brokers que derrubam a conexão TCP em vez de enviar um CONNACK, o que é permitido em algumas falhas.
• O último teste é a verificação de enumeração de contas do caso 3 da matriz.

Execute com MQTT_USER=... MQTT_PASS=... pytest -v test_auth.py. No CI, guarde as credenciais como secrets e rode a suíte contra um broker de staging a cada mudança na configuração do broker — veja testes MQTT no GitHub Actions.

O que os testes de autenticação não cobrem

Um CONNECT bem-sucedido só prova a identidade. Ele não diz nada sobre o que o cliente pode fazer depois. Um sensor com credenciais válidas ainda não deve conseguir publicar no topic de comandos de outro dispositivo — isso é autorização, e ela falha de outro jeito (em silêncio no 3.1.1, com reason code no MQTT 5). Teste separadamente com o método de como testar ACLs MQTT.

Tenha também em mente:

• TCP sem criptografia expõe senhas. As credenciais no CONNECT não são criptografadas. Execute o caso de credenciais válidas também via TLS na porta 8883 e confirme que o listener sem criptografia está desativado em produção.
• Proteção contra força bruta é outro teste. Alguns brokers ou proxies desaceleram ou bloqueiam falhas repetidas. Se o seu faz isso, adicione um teste de que dez senhas erradas seguidas não bloqueiam o cliente legítimo — ou de que bloqueiam, se essa for a sua política.
• Autenticação por token precisa de testes de expiração. Se você usa JWT ou OAuth em vez de senhas estáticas, adicione casos para tokens expirados, ainda não válidos e com assinatura errada.

Teste credenciais válidas e inválidas no cliente online

Testar autenticação é barato: uma dúzia de tentativas de conexão e algumas asserções. Mantenha a matriz no seu repositório junto com a configuração do broker, execute-a a cada mudança e você vai perceber o dia em que alguém reativar o acesso anônimo antes de um atacante.

Perguntas frequentes