Criptografia de Payload MQTT Ponta a Ponta com AES-GCM

Criptografia de payload MQTT significa criptografar o próprio corpo da mensagem, de modo que apenas o publicador e os assinantes previstos consigam lê-lo, independentemente de quem opera o broker. Neste guia você vai aprender quando vale a pena adicionar criptografia ponta a ponta além do TLS, como projetar o formato da mensagem com segurança e como implementá-la em Python com AES-GCM sobre uma conexão real com um broker.

TLS vs criptografia de payload MQTT

O TLS criptografa a conexão entre cada cliente e o broker. Isso é essencial e você deve sempre usá-lo; veja nosso guia de configuração de TLS no MQTT. Mas o TLS termina no broker. Dentro do broker, a mensagem está em texto puro: precisa estar, para que o broker possa roteá-la, armazená-la como retained message ou enfileirá-la para uma sessão offline.

A criptografia de payload acrescenta uma segunda camada que viaja com a mensagem do publicador até o assinante.

Repare nas duas linhas com "Não" que as pessoas costumam ignorar. Os topics continuam legíveis, porque o broker precisa deles para o roteamento, então nunca coloque segredos em nomes de topic. E qualquer recurso do broker que leia payloads deixa de funcionar para mensagens criptografadas. Por exemplo, o explorador de topics ao vivo, as regras de alerta por limite e a decodificação de Sparkplug B da ConnectMQ só conseguem mostrar ou avaliar dados que conseguem interpretar, então payloads criptografados aparecerão ali como bytes opacos.

Quando a criptografia ponta a ponta faz sentido

Adicione criptografia de payload quando:

• Regulamentos ou contratos exigirem que os provedores de infraestrutura não possam ler os dados.
• As mensagens passarem por brokers, bridges ou gateways compartilhados ou públicos que você não controla totalmente.
• Retained messages ou sessões enfileiradas guardarem valores sensíveis, como dados pessoais (sujeitos à LGPD) ou códigos de acesso.
• Você quiser defesa em profundidade caso vazem credenciais de alguma parte do sistema.

Dispense-a quando precisar de visibilidade no lado do broker (dashboards no portal do broker, regras de alerta sobre valores) e confiar no operador, ou quando dispositivos com recursos limitados não comportarem o código extra e a gestão de chaves. Muitas equipes criptografam apenas alguns topics sensíveis, como .../commands/unlock, e deixam a telemetria em texto puro sobre TLS.

Escolhendo um algoritmo

Use um algoritmo de criptografia autenticada. Ele criptografa o payload e adiciona uma tag que detecta qualquer modificação. Duas boas opções:

• AES-256-GCM: amplamente disponível, com aceleração por hardware em muitas CPUs e no ESP32 (via mbedTLS). Exige um nonce de 96 bits único por mensagem sob a mesma chave.
• XChaCha20-Poly1305 (libsodium crypto_aead_xchacha20poly1305_ietf): rápido em software, e seu nonce de 192 bits pode ser gerado aleatoriamente sem risco realista de colisão.

Não use AES-CBC sem MAC, modo ECB de qualquer tipo, nem "criptografia" com XOR ou Base64. Eles dão uma falsa sensação de segurança.

Projetando o formato da mensagem criptografada

Um formato simples e robusto para AES-GCM é:

| version (1 byte) | nonce (12 bytes) | ciphertext + 16-byte tag |

Três regras de projeto importam mais do que o algoritmo:

1. Nunca reutilize um nonce com a mesma chave. No AES-GCM, a reutilização pode expor o texto puro e permitir falsificações. Um nonce aleatório de 12 bytes é aceitável para volumes moderados de mensagens por chave; faça a rotação das chaves periodicamente em publicadores de alto volume.
2. Vincule o texto cifrado ao seu topic. Passe o topic como dados autenticados adicionais (AAD). Caso contrário, um atacante com acesso ao broker poderia reproduzir uma mensagem "unlock" criptografada no topic de outro dispositivo.
3. Inclua um timestamp ou contador dentro do texto puro para que os assinantes possam rejeitar mensagens antigas reproduzidas depois.

Exemplo em Python: publicação e assinatura criptografadas

O exemplo abaixo usa paho-mqtt 2.x e o pacote cryptography. Instale-os primeiro:

pip install "paho-mqtt>=2.0" cryptography

Gere uma chave uma única vez e compartilhe-a com o publicador e o assinante por meio do seu repositório de segredos ou do ambiente, e não via MQTT:

python -c "import os,base64;print(base64.b64encode(os.urandom(32)).decode())"

Salve o código a seguir como secure_mqtt.py:

import base64
import json
import os
import sys
import time

import paho.mqtt.client as mqtt
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.exceptions import InvalidTag

BROKER = "broker.connectmq.com"
PORT = 8883
TOPIC = "connectmq/demo/e2e-91c4/door/command"
VERSION = b"\x01"
MAX_AGE_SECONDS = 30

KEY = base64.b64decode(os.environ["MQTT_PAYLOAD_KEY"])
aead = AESGCM(KEY)


def encrypt(topic: str, data: dict) -> bytes:
    data["ts"] = int(time.time())
    plaintext = json.dumps(data).encode()
    nonce = os.urandom(12)
    ciphertext = aead.encrypt(nonce, plaintext, topic.encode())
    return VERSION + nonce + ciphertext


def decrypt(topic: str, blob: bytes) -> dict:
    if len(blob) < 1 + 12 + 16 or blob[:1] != VERSION:
        raise ValueError("unknown message format")
    nonce, ciphertext = blob[1:13], blob[13:]
    plaintext = aead.decrypt(nonce, ciphertext, topic.encode())
    data = json.loads(plaintext)
    if abs(time.time() - data["ts"]) > MAX_AGE_SECONDS:
        raise ValueError("stale or replayed message")
    return data


def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)
    if userdata == "sub":
        client.subscribe(TOPIC, qos=1)


def on_message(client, userdata, msg):
    try:
        print("Decrypted:", decrypt(msg.topic, msg.payload))
    except (InvalidTag, ValueError) as exc:
        print("Rejected message:", exc)


def make_client(role: str) -> mqtt.Client:
    client = mqtt.Client(
        mqtt.CallbackAPIVersion.VERSION2,
        client_id=f"e2e-{role}-91c4",
        userdata=role,
    )
    client.tls_set()  # o TLS continua protegendo credenciais e metadados
    # Broker privado da ConnectMQ:
    # client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(BROKER, PORT, keepalive=60)
    return client


if __name__ == "__main__":
    role = sys.argv[1] if len(sys.argv) > 1 else "sub"
    client = make_client(role)
    if role == "pub":
        client.loop_start()
        time.sleep(1)
        payload = encrypt(TOPIC, {"action": "unlock", "door": "lab-2"})
        client.publish(TOPIC, payload, qos=1).wait_for_publish()
        print("Published", len(payload), "encrypted bytes")
        client.loop_stop()
        client.disconnect()
    else:
        client.loop_forever()

Execute o assinante e o publicador em dois terminais com a mesma chave:

export MQTT_PAYLOAD_KEY="paste-the-base64-key-here"
python secure_mqtt.py sub      # terminal 1
python secure_mqtt.py pub      # terminal 2

Se você assinar o mesmo topic com mosquitto_sub, verá apenas ruído binário, e essa é a ideia. Altere um byte de um payload, publique-o em outro topic ou espere mais de 30 segundos antes de descriptografar, e o assinante o rejeita. Como o exemplo usa o servidor MQTT público (gratuito), que serve apenas para testes e protótipos, qualquer pessoa pode ver o texto cifrado, mas ninguém sem a chave consegue lê-lo ou falsificá-lo.

Gestão de chaves: a parte difícil

O código de criptografia é curto; é na gestão de chaves que os projetos dão certo ou fracassam.

1. Restrinja o escopo das chaves. Use uma chave por dispositivo ou por grupo de dispositivos, e não uma chave para a frota inteira. Uma chave vazada deve expor o mínimo possível.
2. Armazene as chaves com segurança nos dispositivos. Use flash criptografada ou um secure element em microcontroladores e um keystore do sistema operacional ou gerenciador de segredos em servidores. Nosso guia sobre como armazenar credenciais MQTT em dispositivos se aplica igualmente às chaves de criptografia.
3. Inclua um ID de chave. Estenda o cabeçalho com um identificador de chave para que os assinantes escolham a chave certa e você possa fazer a rotação sem indisponibilidade.
4. Faça a rotação das chaves. Introduza uma nova chave, deixe os assinantes aceitarem ambas, troque os publicadores e então aposente a chave antiga.
5. Considere esquemas assimétricos para fluxos muitos-para-um. Com as sealed boxes da libsodium, milhares de dispositivos podem criptografar para a chave pública de um backend, enquanto só o backend detém a chave privada.

Dispositivos embarcados

No ESP32, o mbedTLS (já presente para o TLS) oferece mbedtls_gcm_crypt_and_tag e mbedtls_gcm_auth_decrypt, então a criptografia de payload acrescenta pouco ao tamanho do código. Use o gerador de números aleatórios por hardware (esp_fill_random) para os nonces e garanta que o Wi-Fi ou o Bluetooth esteja ativo, ou que uma fonte de entropia adequada esteja habilitada, para que a saída do RNG seja realmente aleatória.

Erros comuns de implementação

Ao revisar código de criptografia de payload em projetos IoT, os mesmos erros aparecem repetidamente:

• Um nonce fixo ou um contador que zera ao reiniciar. Um dispositivo que começa o contador do zero a cada ciclo de energia vai reutilizar nonces. Use nonces aleatórios ou persista o contador na flash.
• Ignorar falhas de descriptografia. Um assinante que registra um erro InvalidTag e depois processa a mensagem mesmo assim não tem proteção alguma. Trate qualquer falha como uma mensagem descartada.
• Comprimir depois de criptografar. Texto cifrado não se comprime, então comprima o JSON antes se o tamanho importar.
• Misturar formatos sem um byte de versão. Quando você muda de algoritmo ou adiciona um ID de chave, dispositivos antigos e novos precisam ser distinguíveis. O byte de versão do exemplo existe exatamente por esse motivo.
• Colocar a chave no mesmo repositório do código. Chaves pertencem a um repositório de segredos, a variáveis de ambiente ou ao armazenamento seguro do dispositivo, nunca ao Git.

Teste deliberadamente os caminhos negativos: publique uma mensagem truncada, uma mensagem com um bit invertido e uma mensagem antiga reproduzida, e confirme que cada uma é rejeitada.

A criptografia não substitui a segurança do broker

A criptografia de payload protege a confidencialidade e a integridade dos dados, mas não impede que alguém se conecte, inunde topics ou assine metadados. Você ainda precisa de TLS, credenciais fortes por dispositivo e proteções no lado do broker. Um broker MQTT gerenciado como a ConnectMQ oferece TLS, credenciais vinculadas a client IDs, cotas por plano, proteção contra inundação e bloqueio de IP após logins com falha repetidos, enquanto a criptografia ponta a ponta mantém os próprios dados privados. Juntos, eles cobrem o checklist das nossas melhores práticas de segurança MQTT.

Comece a usar a ConnectMQ

Crie um broker gratuito em app.connectmq.com no plano Serverless, sem cartão de crédito, e execute o exemplo acima com suas próprias credenciais na porta 8883. Como broker MQTT gerenciado, a ConnectMQ cuida de TLS, autenticação e escalabilidade, enquanto sua aplicação mantém as chaves dos payloads.

Perguntas frequentes

O MQTT tem criptografia de payload embutida?

Não. O MQTT 3.1.1 e o 5.0 não definem criptografia de payload; os payloads são bytes opacos. A criptografia é feita pelo TLS na conexão e, opcionalmente, pela sua aplicação no payload.

Posso criptografar também o nome do topic?

Não sem quebrar o roteamento, porque o broker compara as assinaturas pelos nomes de topic. Mantenha os topics livres de informações sensíveis e use IDs de dispositivo opacos em vez de nomes ou endereços.

A criptografia de payload aumenta o tamanho da mensagem?

Um pouco. O formato AES-GCM acima acrescenta 29 bytes por mensagem (versão, nonce e tag), mais a sobrecarga de codificação se você converter o resultado para Base64.

Ainda devo usar TLS se os payloads estiverem criptografados?

Sim. O TLS protege suas credenciais MQTT, client IDs, nomes de topic e metadados do protocolo, além de autenticar o broker, e a criptografia de payload não cobre nada disso.