How to Test MQTT ACLs and Topic Permissions

To test MQTT ACLs, connect as the user you want to check, try every publish and subscribe the device should and should not be able to do, and verify the result from outside the client. That last part matters: in MQTT 3.1.1 a denied publish looks exactly like a successful one, so the only reliable proof is a second client that watches whether the message arrived.

This guide covers how each protocol version reports a denial, a permission test matrix, manual checks with mosquitto and an automated test you can run after every ACL change.

How brokers report a denied action

ACLs (access control lists) decide which topics a user may publish to (write) and subscribe to (read). Authentication already succeeded at this point — the CONNACK was 0x00 — so all feedback comes from later packets.

The 3.1.1 behaviour is not a bug in your broker. The specification says that when a server does not authorize a PUBLISH, it has no way of informing the client and must either acknowledge it according to the normal QoS rules or close the network connection. Many clients log nothing in either case, which is why "my messages disappear" is such a common ACL symptom — see MQTT ACL denied silently.

Build a permission test matrix

List what one representative user should be able to do, then add the things it must not do. Here is an example for a sensor that may publish its own telemetry and read its own commands, with a Mosquitto ACL that implements it:

# every device: own telemetry and own commands only
pattern write testmqtt/acl/telemetry/%u/#
pattern read  testmqtt/acl/commands/%u/#

# a monitoring user that may read everything under testmqtt/acl
user acl-observer
topic read testmqtt/acl/#

Case 9 is where topic patterns surprise people: telemetry/%u/# matches telemetry/sensor-01 itself because # also matches the parent level. Paste your patterns and topics into the topic matcher to check edge cases before you write the test. Case 8 matters because retained messages are delivered at subscribe time; a broker that forgets to apply the read ACL there leaks the last value of every topic. For more patterns that go wrong, see MQTT wildcard security.

Manual checks with mosquitto

Open an observer first. It uses an account with broad read access, so it shows what really reaches the broker:

mosquitto_sub -h broker.connectmq.com -p 1883 \
  -u acl-observer -P "$OBSERVER_PASS" -t 'testmqtt/acl/#' -v

Now publish as the device under test with MQTT 5 and QoS 1, so a denial produces a reason code:

# case 1: should appear in the observer
mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -q 1 \
  -u sensor-01 -P "$MQTT_PASS" -t testmqtt/acl/telemetry/sensor-01/temp -m 21.5

# case 2: should NOT appear; MQTT 5 reports the denial
mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -q 1 \
  -u sensor-01 -P "$MQTT_PASS" -t testmqtt/acl/telemetry/sensor-02/temp -m 99

For the denied publish, the mosquitto client prints a warning like Warning: Publish 1 failed: Not authorized. Run the same command with -V mqttv311 and it exits cleanly with no warning — the observer staying quiet is your only evidence.

Subscriptions are easier to verify because both versions answer in the SUBACK:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -d \
  -u sensor-01 -P "$MQTT_PASS" -t 'testmqtt/acl/commands/sensor-02/#'

With -d you see Subscribed (mid: 1): 135 for an MQTT 5 denial (0x87) or Subscribed (mid: 1): 128 (0x80) in 3.1.1. Recent mosquitto versions then print All subscription requests were denied. and exit. If your broker grants the subscription instead, publish to that topic from another account and confirm nothing is delivered. The codes are explained in detail in SUBACK 0x80 and Not authorized 0x87.

You can do the same interactively in the online MQTT client: connect two sessions with different users, subscribe as the observer in one, publish as the device in the other.

An automated ACL test

This paho-mqtt 2.x script runs the publish and subscribe cases from the matrix. It checks MQTT 5 reason codes and uses the observer as ground truth, so it also works against brokers that silently drop denied messages:

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

HOST, PORT = "broker.connectmq.com", 1883
DEVICE = ("sensor-01", os.environ["MQTT_PASS"])
OBSERVER = ("acl-observer", os.environ["OBSERVER_PASS"])
BASE = "testmqtt/acl"

def client(creds):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                    client_id=f"acl-{uuid.uuid4().hex[:8]}", protocol=mqtt.MQTTv5)
    c.username_pw_set(*creds)
    c.connect(HOST, PORT)
    c.loop_start()
    time.sleep(0.5)
    return c

seen = set()
obs = client(OBSERVER)
obs.on_message = lambda c, u, m: seen.add(m.topic)
obs.subscribe(f"{BASE}/#", qos=1)

dev = client(DEVICE)
acks = {}
dev.on_publish = lambda c, u, mid, rc, props: acks.__setitem__(mid, str(rc))
subacks = {}
dev.on_subscribe = lambda c, u, mid, rcs, props: subacks.__setitem__(mid, [str(r) for r in rcs])

def check_publish(topic, allowed):
    info = dev.publish(topic, "acl-test", qos=1)
    time.sleep(1)
    delivered = topic in seen
    print(f"PUB {topic}: puback={acks.get(info.mid)} delivered={delivered}")
    assert delivered == allowed, f"publish to {topic}: expected allowed={allowed}"
    if not allowed:
        assert acks.get(info.mid) == "Not authorized"

def check_subscribe(topic, allowed):
    _, mid = dev.subscribe(topic, qos=1)
    time.sleep(1)
    granted = not subacks[mid][0].startswith(("Not authorized", "Unspecified"))
    print(f"SUB {topic}: suback={subacks[mid]}")
    assert granted == allowed, f"subscribe to {topic}: expected allowed={allowed}"

check_publish(f"{BASE}/telemetry/sensor-01/temp", True)
check_publish(f"{BASE}/telemetry/sensor-02/temp", False)
check_publish(f"{BASE}/commands/sensor-01/reboot", False)
check_subscribe(f"{BASE}/commands/sensor-01/#", True)
check_subscribe(f"{BASE}/commands/sensor-02/#", False)
print("all ACL checks passed")

Run it after every change to the ACL file or the auth backend, and in CI against a staging broker. When a check fails, the printed reason code and the delivered flag tell you immediately whether the broker rejected the message, dropped it silently or let it through.

Common reasons an ACL test fails

• Everything is denied. The ACL file is not loaded, or the user name in the rule does not match the one in CONNECT. Check the broker log at startup; with Mosquitto, acl_file must be set in the same listener scope as the password file — see Mosquitto authentication.
• Everything is allowed. A broad rule like topic readwrite # earlier in the file, or anonymous access with its own permissive rules.
• Pattern rules don't match. %u is the username and %c the client ID; using the wrong one denies every device whose ID differs from its username.
• Changes have no effect. Most brokers read ACLs at startup or on reload. Send SIGHUP to Mosquitto, or restart, before re-running the test.
• Test passes but production fails. The client publishes with a leading slash (/testmqtt/...) or a different case. Topics are case-sensitive and /a is not a.

Watch ACL test traffic in the online client

An ACL is a security control, and security controls need tests that fail when the control breaks. Keep a matrix of allowed and denied actions per role, verify delivery with an observer instead of trusting the publisher, and prefer MQTT 5 so denials are reported rather than swallowed.

Frequently asked questions