Teste de Invasão MQTT: Uma Metodologia Prática para os Seus Próprios Brokers

O teste de invasão MQTT (pentest) é uma forma estruturada de encontrar fraquezas no seu próprio broker e nos seus clientes antes que outra pessoa encontre. Este guia apresenta uma metodologia repetível, do escopo e reconhecimento às verificações de autenticação, autorização, TLS e limites de recursos, e termina com um checklist de hardening que você pode entregar à equipe. Tudo aqui se destina a sistemas que são seus ou que você tem permissão por escrito para testar.

As regras de engajamento vêm primeiro

Testar um broker que não é seu é acesso não autorizado na maioria das jurisdições, mesmo que ele "já estivesse aberto". Antes de enviar um único pacote:

• Obtenha autorização por escrito do dono do sistema, indicando hosts, portas, datas e técnicas permitidas.
• Defina o que está fora do escopo, por exemplo dispositivos de produção que acionam equipamentos físicos, brokers de terceiros e brokers públicos de teste.
• Prefira uma cópia de homologação. Um broker com a mesma configuração, mas sem equipamentos reais, permite testar o acesso de escrita sem risco.
• Combine condições de parada. Se um teste causar instabilidade, pare e reporte.

Em um broker MQTT gerenciado, a política de uso aceitável do provedor também se aplica. Teste a configuração do seu próprio tenant e o comportamento dos seus clientes, não a infraestrutura compartilhada do provedor, e nunca execute campanhas de carga ou fuzzing contra um cluster compartilhado ou um servidor MQTT público como o broker.connectmq.com. Os limites por cliente dele existem para mantê-lo utilizável para todos.

A metodologia em resumo

Fase 2: reconhecimento do seu broker

Comece listando todas as formas pelas quais um cliente pode chegar ao broker. As equipes costumam proteger a porta 8883 com cuidado e esquecer um listener WebSocket ou um listener antigo sem criptografia.

# Lista as portas relacionadas a MQTT no seu próprio host
nmap -Pn -p 1883,8883,8083,8084,9001 your-broker.example.com

# Inspeciona o endpoint TLS
openssl s_client -connect your-broker.example.com:8883 -servername your-broker.example.com </dev/null

Registre para cada listener: protocolo (TCP, TLS, WS, WSS), se é acessível pela internet, o subject e a validade do certificado e quais versões de MQTT ele aceita. Se o broker expõe topics $SYS, verifique o que eles revelam a um cliente autenticado comum. Strings de versão e contagem de clientes são úteis para um atacante e raramente necessárias para os dispositivos.

Fase 3: testes de autenticação

Os testes de autenticação respondem a uma pergunta: alguém que não deveria consegue se conectar? Execute cada teste com um identificador que facilite encontrar a tentativa nos logs.

1. Conexão anônima. Conecte sem usuário. Esperado: recusada com o motivo "not authorised".
2. Senha errada. Esperado: recusada. Repita algumas vezes e confirme que falhas repetidas são detectadas ou limitadas.
3. Credenciais válidas, client ID errado. Se o seu broker vincula client IDs a credenciais, uma divergência deve ser recusada.
4. Client ID duplicado. Conecte duas vezes com o mesmo ID e observe a tomada de sessão. Garanta que os seus dispositivos lidam com isso e que um atacante com apenas um client ID não consiga tomar uma sessão.
5. Credenciais revogadas. Revogue uma identidade de teste e confirme que ela não consegue mais se conectar nem se reconectar.

Um pequeno script em Python torna essas verificações repetíveis:

import time
import paho.mqtt.client as mqtt

HOST, PORT = "your-broker.example.com", 8883

CASES = [
    ("anonymous", None, None, "pt-anon-01"),
    ("wrong-password", "YOUR_USERNAME", "not-the-password", "YOUR_CLIENT_ID"),
    ("valid", "YOUR_USERNAME", "YOUR_PASSWORD", "YOUR_CLIENT_ID"),
    ("valid-wrong-id", "YOUR_USERNAME", "YOUR_PASSWORD", "pt-other-id"),
]

def try_connect(name, user, pwd, cid):
    result = {}
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=cid,
                         protocol=mqtt.MQTTv5)
    client.tls_set()  # verifica o certificado do servidor com as CAs do sistema
    if user:
        client.username_pw_set(user, pwd)

    def on_connect(c, userdata, flags, reason_code, properties):
        result["rc"] = str(reason_code)
        c.disconnect()

    client.on_connect = on_connect
    try:
        client.connect(HOST, PORT, keepalive=10)
        client.loop_start()
        time.sleep(5)  # aguarda o CONNACK
        client.loop_stop()
    except Exception as exc:
        result["rc"] = f"exception: {exc}"
    print(f"{name:16} -> {result.get('rc', 'no response')}")

for case in CASES:
    try_connect(*case)

Apenas o caso valid deve informar Success. Todos os outros devem ser recusados com um reason code claro.

Fase 4: testes de autorização

A autenticação prova quem o cliente é; a autorização decide o que ele pode fazer. Monte uma matriz de papéis (sensor, gateway, dashboard, serviço administrativo) versus padrões de topics e teste cada célula com uma conexão real.

Preste atenção em como as negações aparecem. No MQTT 5, uma assinatura recusada retorna um reason code no SUBACK, como 0x87 (not authorised), e uma publicação QoS 1 recusada retorna um PUBACK com reason code. No MQTT 3.1.1, uma publicação negada geralmente é descartada em silêncio, então você precisa verificar com um segundo assinante privilegiado que a mensagem nunca chegou. Em brokers auto-hospedados, o guia de ACL MQTT explica como expressar essas regras. Onde o broker não oferece regras por topic, o controle equivalente é ter brokers ou tenants separados por zona de confiança.

Fase 5: testes de segurança de transporte

Verifique os dois lados do TLS:

• Lado do servidor: apenas versões modernas do protocolo (TLS 1.2 e 1.3), um certificado de uma CA confiável que corresponda ao hostname e nenhum listener sem criptografia acessível pela internet.
• Lado do cliente: os dispositivos devem rejeitar um certificado errado. Aponte um dispositivo de teste para um host de homologação com certificado autoassinado ou divergente e confirme que a conexão falha. Um cliente que se conecta mesmo assim está, na prática, sem criptografia diante de um atacante ativo.

O artigo de configuração de TLS no MQTT cobre as configurações corretas do cliente, e analisar tráfego MQTT com o Wireshark é uma boa forma de provar que os payloads não ficam mais visíveis na rede.

Fase 6: robustez e limites (apenas em homologação)

O teste de robustez verifica como o broker e os seus clientes se comportam com entradas excessivas ou malformadas. Execute-o apenas em um broker de homologação que você controla, nunca em infraestrutura compartilhada ou de produção.

• Payloads grandes demais: publique uma mensagem maior que o máximo configurado. Espere uma recusa, não uma queda.
• Limites de taxa: publique mais rápido do que a sua política permite e confirme que o broker limita ou desconecta. Nos planos do ConnectMQ, exceder uma cota é sinalizado com o reason code 0x97 (quota exceeded) do MQTT 5, que os seus clientes devem tratar de forma elegante.
• Inundação de conexões: abra muitas conexões a partir de uma única origem e verifique se os limites se aplicam por cliente ou por IP.
• Fuzzing de protocolo: ferramentas que alteram pacotes MQTT podem revelar bugs de parsing. Mantenha-as em um broker de teste isolado e reporte as quedas ao fornecedor do broker.

Para contexto sobre como projetar esses limites, veja limitação de taxa e proteção contra DoS no MQTT.

Fase 7: revisão de clientes e dispositivos

Muitas descobertas do mundo real estão no dispositivo, e não no broker:

• Credenciais fixas no firmware ou enviadas para um repositório.
• Uma única senha compartilhada por todos os dispositivos de uma frota.
• Topics de depuração que publicam configurações ou segredos.
• Topics de comando que executam payloads sem validação.
• Verificação de TLS desativada "para funcionar".

Trate cada dispositivo como potencialmente extraído e inspecionado por um atacante. Cada um deve ter credenciais únicas que possam ser revogadas individualmente.

Checklist de hardening a partir das descobertas

Converta as descobertas em resultados que a equipe possa verificar:

1. Todo listener exige autenticação; o acesso anônimo só existe onde é intencional e documentado.
2. Apenas listeners TLS são acessíveis pela internet, com um certificado válido.
3. Todo dispositivo e serviço tem credenciais únicas e revogáveis.
4. Cada identidade só pode publicar e assinar os topics de que precisa.
5. Os clientes rejeitam certificados inválidos.
6. Limites de tamanho de mensagem, taxa e conexões estão configurados, e os clientes tratam as recusas.
7. Falhas de login repetidas são limitadas ou bloqueadas.
8. O software do broker recebe atualizações periódicas.
9. Topics retidos e de depuração não contêm segredos.
10. Novos testes são agendados após grandes mudanças.

Vários itens desta lista ficam a cargo da plataforma quando você usa um broker MQTT gerenciado. No ConnectMQ, brokers privados exigem credenciais geradas, os client IDs são vinculados a essas credenciais, o TLS usa um certificado Let's Encrypt, os tenants são isolados, e a proteção contra flood e o bloqueio de IP após repetidas falhas de login já vêm incluídos. Assim, o seu esforço de teste pode se concentrar no comportamento dos clientes e no design dos topics. O artigo de boas práticas de segurança MQTT é um bom complemento para os itens do lado do cliente.

Teste em um broker que você controla

A forma mais segura de praticar esta metodologia é em um broker que é seu. Crie um broker privado do ConnectMQ no plano Serverless gratuito em app.connectmq.com (sem cartão de crédito), execute as verificações de autenticação e TLS com as suas próprias credenciais e mantenha os testes de carga e fuzzing em um broker de homologação local. Para experimentos rápidos e sem dados sensíveis, você ainda pode usar o servidor MQTT público (gratuito), dentro dos limites dele e nunca com dados de produção.

Perguntas frequentes

Posso fazer um teste de invasão em um broker MQTT público?

Não sem a permissão explícita do operador. Brokers públicos de teste são serviços compartilhados, e testes de carga ou fuzzing os degradam para todos. Teste os seus próprios brokers ou uma cópia de homologação local.

Quais ferramentas são comumente usadas em testes de segurança MQTT?

Scanners de rede para descobrir listeners, openssl s_client para verificações de TLS, mosquitto_pub e mosquitto_sub para testes rápidos e clientes programados, como o paho-mqtt, para matrizes repetíveis de autenticação e autorização.

Com que frequência devemos testar novamente o nosso ambiente MQTT?

Após cada mudança significativa em listeners, credenciais, estrutura de topics ou firmware e, fora isso, em uma periodicidade regular. Automatizar as verificações de autenticação e autorização torna barato testar com frequência.

Qual é a descoberta mais comum?

Credenciais compartilhadas ou fixas nos dispositivos e verificação de TLS desativada nos clientes são descobertas frequentes, junto com listeners que aceitam conexões anônimas.