Testing MQTT Authentication: Valid, Invalid and Missing Credentials

Testing MQTT authentication means connecting with every kind of credential your broker can receive — valid, wrong, unknown, empty and missing — and asserting that each one gets the right CONNACK code. A broker that accepts the good password is only half tested; the security comes from how it rejects everything else.

This guide gives you a test matrix with the expected result per case for MQTT 3.1.1 and MQTT 5, the commands to run it by hand, and a pytest suite that runs it automatically in CI.

How MQTT reports an authentication result

Authentication happens once, in the CONNECT packet. The client sends a username and/or password, and the broker answers with CONNACK. A non-zero code means the connection was refused, and the broker closes the socket right after sending it.

The spec does not force a broker to choose between "bad password" and "not authorized". Some brokers return 0x04/0x86 for a wrong password; Mosquitto returns 0x05/0x87 for every credential failure. That is fine — what matters is that your broker is consistent and that your tests pin down its actual behaviour. Look up any code you see in the MQTT reason codes tool.

The authentication test matrix

Run every row against the broker you ship with, at least once with MQTT 3.1.1 and once with MQTT 5. "Refused" means 0x04/0x05 in 3.1.1 and 0x86/0x87 in 5.

Cases 3 and 9 catch real bugs. If an unknown user gets a different code than a wrong password, the broker leaks which accounts exist. If a trailing newline is accepted, someone's password file or auth backend is trimming input — usually harmless, but you want to know. Case 10 checks that revocation actually takes effect: with Mosquitto that means the password file was reloaded (SIGHUP), with a managed broker that the change propagated.

Running the matrix by hand

mosquitto_pub is the fastest way to try each case. Add -d to see the raw CONNACK:

# 1. valid credentials — exits silently with code 0
mosquitto_pub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/auth/check -m ok -d

# 2. wrong password
mosquitto_pub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "wrong" \
  -t testmqtt/auth/check -m ok -d

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

# 4. no credentials at all
mosquitto_pub -h broker.connectmq.com -p 1883 -t testmqtt/auth/check -m ok -d

A refused connection prints one of these lines, which are the strings people usually search for:

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

With -V mqttv5 the same cases print the MQTT 5 reason string instead, for example Connection error: Not authorized. For a visual check, the online MQTT client shows the CONNACK result for each attempt — handy when you want to compare WebSocket and TCP listeners, which some brokers configure with different auth backends.

If you see Connection Refused: identifier rejected instead, the problem is the client ID, not the credentials — see identifier rejected. If the client hangs and then times out, you are probably on the wrong port or behind a firewall, not failing authentication.

Automating it with pytest

Manual checks prove the broker works today. An automated suite proves it still works after the next configuration change. This test uses paho-mqtt 2.x and runs every case for both protocol versions:

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):
    """Return the CONNACK reason string, or 'closed' if no CONNACK arrived."""
    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")

A few details make this reliable:

• paho-mqtt 2.x maps MQTT 3.1.1 return codes to MQTT 5 reason codes, so 0x04 arrives as "Bad user name or password" and 0x05 as "Not authorized" for both protocol versions. One set of assertions covers both.
• Each attempt uses a fresh random client ID. Reusing one can make a successful test kick out a previous connection and confuse the result. The client ID generator produces IDs in the same format.
• "closed" covers brokers that drop the TCP connection instead of sending a CONNACK, which is legal for some failures.
• The last test is the account-enumeration check from case 3 of the matrix.

Run it with MQTT_USER=... MQTT_PASS=... pytest -v test_auth.py. In CI, store the credentials as secrets and run the suite against a staging broker on every change to the broker configuration — see MQTT testing in GitHub Actions.

What authentication tests do not cover

A successful CONNECT only proves identity. It says nothing about what the client may do afterwards. A sensor with valid credentials should still be unable to publish to another device's command topic — that is authorization, and it fails differently (silently in 3.1.1, with a reason code in MQTT 5). Test it separately with the method in how to test MQTT ACLs.

Also keep in mind:

• Plain TCP exposes passwords. Credentials in CONNECT are not encrypted. Run the valid-credentials case over TLS on port 8883 as well, and confirm the plain listener is disabled in production.
• Brute-force protection is a separate test. Some brokers or proxies slow down or block repeated failures. If yours does, add a test that ten wrong passwords in a row do not lock out the legitimate client — or that they do, if that is your policy.
• Token-based auth needs expiry tests. If you use JWT or OAuth instead of static passwords, add cases for expired, not-yet-valid and wrongly signed tokens.

Try valid and invalid credentials in the online client

Authentication testing is cheap: a dozen connection attempts and a few assertions. Keep the matrix in your repository next to the broker configuration, run it on every change, and you will notice the day someone re-enables anonymous access before an attacker does.

Frequently asked questions