Secure IoT with MQTT: Python, TLS and access control

MQTT is a lightweight publish/subscribe protocol in which clients publish messages to topic names and subscribe to those topics through an intermediary server, the broker. The protocol does not mandate encryption or authorisation; whether an MQTT deployment is secure is decided by the broker configuration and the client code.

I described the architecture of putting camera analytics on the same event bus as PLCs, door contacts and sensors in combining IoT and camera analytics. This article is the security side of that bus: setting up Mosquitto with TLS, client certificates and per-topic authorisation, and writing correct clients with paho-mqtt 2.x. I ran and tested the configuration and Python code below locally with Mosquitto 2.1.2 and paho-mqtt 2.1.0; the test results are included.

MQTT basics

Three concepts are enough: the broker is the server that distributes messages, a client is every device or service that connects, and a topic is a /-separated address such as cx/gebze-01/cam-07/telemetry. A publisher does not know its subscribers; adding a new consumer does not change the producer.

Quality of service (QoS)

QoS	Guarantee	Cost	Where
0	At most once; may be lost	Cheapest	Frequent readings where a loss does not matter
1	At least once; duplicates possible	An acknowledgement packet	Events, alarms (consumer drops duplicates)
2	Exactly once	Four-step handshake	Rare cases where a duplicate does harm

QoS applies per hop: publisher to broker, and broker to subscriber. According to the MQTT 5 specification, the QoS of a message sent to a subscriber is the lower of the published QoS and the subscription QoS. My default in the field is QoS 1 plus deduplication on the consumer side.

Retain, last will and sessions

• Retain: The broker stores the last retained message of a topic and sends it immediately to new subscribers. Right for “last value” information such as a device’s online/offline state; wrong for events and commands. A zero-byte retained message deletes the stored one.
• Last will: Left by the client when it connects; if the connection drops unexpectedly, the broker publishes it on the client’s behalf. Per the specification, the broker treats the connection as failed if it receives no packet from the client within one and a half times the keep-alive period. The will is not published for a client that leaves with a clean DISCONNECT.
• Session: In MQTT 5, with clean_start=False and SessionExpiryInterval, the broker keeps a disconnected client’s subscriptions and pending QoS 1/2 messages for the given time.

What MQTT 5 added

MQTT 5 added several features that matter in the field: reason codes on all acknowledgement packets, session and message expiry, user properties, content type, response topic and correlation data for request/response, shared subscriptions that split the load across several consumers ($share/{group}/{filter}), and subscription options (for example, when retained messages are sent). Reason codes alone are a good reason to choose MQTT 5, because they let clients see authorisation errors; in MQTT 3.1.1 the PUBACK packet carries no reason code.

Topic design

The topic tree is one of the hardest things to change later; ACLs, dashboards and integrations all hang off it. The structure I use:

cx/{site}/{device}/{kind}
cx/gebze-01/cam-07/telemetry     readings
cx/gebze-01/cam-07/status        online/offline (retained)
cx/gebze-01/cam-07/cmd/reboot    command to the device

Rules:

• Keep the device ID at a fixed level; the %u pattern in the ACL only works that way.
• No leading /, avoid spaces and non-ASCII characters, use lower case.
• Never put personal data or secrets into topic names; topic names end up in logs and ACLs.
• Keep telemetry, status and commands in separate branches; only authorised services should be able to write to the command branch.
• + is the single-level wildcard, # matches all remaining levels. Do not let devices subscribe to #. Topics starting with $ ($SYS) are reserved for the broker.

Step 1: Install Mosquitto

On Debian/Ubuntu, sudo apt install mosquitto mosquitto-clients is enough. A common practice is to put the configuration into a separate file under /etc/mosquitto/conf.d/ rather than the main file; check that the main file has an include_dir line that reads that folder.

A note on versions: the current series is 2.1 (2.1.0 was released in January 2026, 2.1.2 in February 2026), while distribution packages may still ship 2.0.x. The configuration below works on both. In 2.1, the password_file and acl_file options were deprecated in favour of plugins containing the same code and will be removed in 3.0; per_listener_settings is in the same position. In 2.1 the default max_packet_size dropped to 2,000,000 bytes. Since 2.0, allow_anonymous has defaulted to false; in 1.6 and earlier it defaulted to true unless another security option was set. If you inherit an old installation, that is the first line to check.

Step 2: Certificates: CA, broker and device

We set up an internal certificate authority (CA) and sign a separate certificate for the broker and for each device. The script below is for a test environment; in production the CA key belongs on an offline machine or in the organisation’s PKI, not on the broker.

#!/usr/bin/env bash
# Creates a private CA, a broker certificate and one device (client) certificate for a test setup.
set -euo pipefail
BROKER_DNS="${BROKER_DNS:-broker.local}"
DEVICE_ID="${DEVICE_ID:-cam-07}"

# 1) Private certificate authority (CA). Keep ca.key offline, not on the broker.
openssl req -x509 -new -newkey rsa:4096 -sha256 -days 3650 -nodes \
  -keyout ca.key -out ca.crt -subj "/CN=CX IoT Test CA"

# 2) Broker certificate: clients verify the hostname against the SAN field.
openssl req -new -newkey rsa:2048 -nodes -keyout server.key -out server.csr \
  -subj "/CN=${BROKER_DNS}" -addext "subjectAltName=DNS:${BROKER_DNS}"
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial \
  -out server.crt -days 825 -sha256 -copy_extensions copy

# 3) Device certificate: CN = device ID (the broker will use it as the username).
openssl req -new -newkey rsa:2048 -nodes -keyout "${DEVICE_ID}.key" -out "${DEVICE_ID}.csr" \
  -subj "/CN=${DEVICE_ID}"
openssl x509 -req -in "${DEVICE_ID}.csr" -CA ca.crt -CAkey ca.key -CAcreateserial \
  -out "${DEVICE_ID}.crt" -days 365 -sha256

chmod 600 ./*.key
openssl verify -CAfile ca.crt server.crt "${DEVICE_ID}.crt"

Two details: clients look for the broker’s name in the certificate’s SAN (Subject Alternative Name) field; in my test, with only localhost in the certificate, connecting via 127.0.0.1 made Python reject the connection with an IP address mismatch error. The SAN must contain whatever name clients use to reach the broker. The -copy_extensions option requires OpenSSL 3.0 or newer. The CN of the device certificate is the device ID (cam-07); the broker will use it as the username.

Step 3: TLS listeners and authentication

I open two listeners: on 8883, services and dashboards connect over TLS with a username and password; on 8884, field devices connect with mutual TLS (mTLS). No plain-text 1883 listener is defined at all.

# /etc/mosquitto/conf.d/cx.conf
allow_anonymous false
password_file /etc/mosquitto/passwd
acl_file /etc/mosquitto/acl

# 8883: services and dashboards — TLS + username/password
listener 8883
cafile /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key
tls_version tlsv1.2

# 8884: field devices — mutual TLS (mTLS), the certificate CN becomes the username
listener 8884
cafile /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key
tls_version tlsv1.2
require_certificate true
use_identity_as_username true

tls_version sets the minimum version allowed; if unset, TLS 1.3 and 1.2 are accepted. On 8884, require_certificate true cuts off any connection without a valid client certificate at the TLS stage. use_identity_as_username true makes the certificate CN the username, and the password file is not used on that listener. As the Mosquitto documentation warns, any CA in cafile can issue client certificates that are valid for this listener, so do not share the device CA for other purposes. For revoked devices, a certificate revocation list can be supplied with crlfile.

Add passwords for service users with mosquitto_passwd. -b takes the password on the command line and leaves it in the shell history; interactive use is safer:

sudo mosquitto_passwd -c /etc/mosquitto/passwd rules-engine   # -c recreates the file
sudo mosquitto_passwd /etc/mosquitto/passwd dashboard
sudo chown mosquitto:mosquitto /etc/mosquitto/passwd /etc/mosquitto/acl
sudo chmod 0700 /etc/mosquitto/passwd /etc/mosquitto/acl
sudo systemctl restart mosquitto

Mosquitto warns when these files have loose permissions, and in my test the warning said future versions will refuse to load such files.

Step 4: Per-topic authorisation with an ACL

Once acl_file is set, only the listed topics are accessible. topic lines following a user line apply to that user; pattern lines apply to everyone, with %u replaced by the username and %c by the client ID. According to the documentation, the substitution must be the only text at its topic level.

# Devices: write only to their own branch, read only their own command topic.
# %u = username (the certificate CN on the mTLS listener)
pattern write cx/+/%u/#
pattern read cx/+/%u/cmd/#

# Event-consuming service: reads all events, may write commands
user rules-engine
topic read cx/#
topic write cx/+/+/cmd/#

# Dashboard: read only
user dashboard
topic read cx/#

The two pattern lines manage hundreds of devices without listing them one by one: each device can write only under cx/*/its-own-id/... and read only its own command branch. What I saw while testing this setup:

Attempt	Result (Mosquitto 2.1.2, MQTT 5)
cam-07 publishes to its own telemetry topic	PUBACK: Success
cam-07 publishes to cam-08’s topic	PUBACK: Not authorized, message not delivered
dashboard publishes to a command topic	PUBACK: Not authorized
Wrong password	CONNACK: Not authorized
Connecting to 8884 without a client certificate	TLS handshake fails, no connection
cam-07 subscribes to cx/#	SUBACK: Granted QoS 1, but only messages on its own command topic were delivered

The last row matters: in this test Mosquitto did not reject any forbidden subscription in the SUBACK (neither the wildcard cx/# nor another device’s command topic); it filtered unauthorised messages at delivery time. Write your ACL tests as “did a message on a forbidden topic arrive?”, not “was the subscription accepted?”. When the device count grows or permissions need to change at run time, Mosquitto’s Dynamic Security plugin is a more flexible option than file-based management.

Step 5: Python publisher (device)

With the 2.x release installed by pip install paho-mqtt, you must pass the callback API version when creating a Client. With VERSION2, the on_connect signature is (client, userdata, flags, reason_code, properties) and reason_code is an object; check for errors with is_failure.

#!/usr/bin/env python3
"""Device-side publisher: connects with mTLS, publishes readings as JSON with QoS 1 (paho-mqtt 2.x)."""
from __future__ import annotations

import json
import logging
import os
import random
import signal
import threading
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("device")

BROKER = os.getenv("MQTT_HOST", "broker.local")
PORT = int(os.getenv("MQTT_PORT", "8884"))
SITE = os.getenv("SITE", "gebze-01")
DEVICE_ID = os.getenv("DEVICE_ID", "cam-07")  # must match the CN in the certificate
CERT_DIR = os.getenv("CERT_DIR", "/etc/cx/certs")
BASE = f"cx/{SITE}/{DEVICE_ID}"

client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id=DEVICE_ID,
    protocol=mqtt.MQTTv5,
)
# Defaults are secure: the server certificate is verified against the CA and the hostname is checked.
client.tls_set(
    ca_certs=f"{CERT_DIR}/ca.crt",
    certfile=f"{CERT_DIR}/{DEVICE_ID}.crt",
    keyfile=f"{CERT_DIR}/{DEVICE_ID}.key",
)
# On an unexpected drop the broker publishes this; retain lets new subscribers see the last state.
client.will_set(f"{BASE}/status", "offline", qos=1, retain=True)
client.reconnect_delay_set(min_delay=1, max_delay=60)
client.max_queued_messages_set(500)


def on_connect(c, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        log.error("Connection refused: %s", reason_code)  # e.g. "Not authorized"
        return
    log.info("Connected (session present: %s)", flags.session_present)
    c.publish(f"{BASE}/status", "online", qos=1, retain=True)
    c.subscribe(f"{BASE}/cmd/#", qos=1)  # re-subscribed automatically after a reconnect


def on_disconnect(c, userdata, flags, reason_code, properties):
    log.warning("Connection closed: %s", reason_code)  # paho reconnects by itself after an unexpected drop


def on_message(c, userdata, msg):
    log.info("Command received %s: %s", msg.topic, msg.payload[:200])


client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_message = on_message

connect_props = Properties(PacketTypes.CONNECT)
connect_props.SessionExpiryInterval = 3600  # keep session and QoS 1 queue for 1 hour after a drop
client.connect_async(BROKER, PORT, keepalive=30, clean_start=False, properties=connect_props)
client.loop_start()

stop = threading.Event()  # systemd/Docker send SIGTERM; catch it for a clean shutdown
signal.signal(signal.SIGTERM, lambda *_: stop.set())
signal.signal(signal.SIGINT, lambda *_: stop.set())

seq = 0
while not stop.is_set():
    seq += 1
    message = {
        "schema": "cx.telemetry/1",
        "device_id": DEVICE_ID,
        "site": SITE,
        "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z"),
        "seq": seq,
        "data": {"temp_c": round(random.uniform(20, 30), 2)},
    }
    props = Properties(PacketTypes.PUBLISH)
    props.ContentType = "application/json"
    props.MessageExpiryInterval = 300  # useless if not delivered within 5 min; let the broker drop it
    info = client.publish(f"{BASE}/telemetry", json.dumps(message), qos=1, properties=props)
    if info.rc != mqtt.MQTT_ERR_SUCCESS:
        log.warning("Message %d: no connection, held in memory (rc=%s)", seq, info.rc)
    stop.wait(5)

# A clean shutdown does not trigger the LWT; publish "offline" ourselves.
try:
    client.publish(f"{BASE}/status", "offline", qos=1, retain=True).wait_for_publish(timeout=3)
except (RuntimeError, ValueError):
    pass
client.disconnect()
client.loop_stop()

The decisions in the code:

• tls_set defaults are secure: in paho’s source, if cert_reqs is not given, CERT_REQUIRED is used and the hostname is verified. Do not add tls_insecure_set(True) or CERT_NONE “to make it work”.
• reconnect_delay_set(1, 60): after a drop, wait 1 second and double on every attempt, capped at 60 seconds. The default cap is 120 seconds.
• connect_async + loop_start: even if the broker is unreachable at start-up, the background loop keeps trying to connect.
• Subscribing inside on_connect: renewed automatically after a reconnect.
• SessionExpiryInterval=3600 and clean_start=False: the session and pending messages survive short drops. In my test, when I stopped and restarted the subscribing service, readings published while it was down were delivered after it reconnected.
• MessageExpiryInterval=300: if a reading older than five minutes is useless, the broker drops it.
• Because the last will is not published on a clean shutdown, the offline state is written explicitly while closing.

Step 6: Python subscriber (service)

The service side receives the message and validates it before trusting it.

#!/usr/bin/env python3
"""Service-side subscriber: connects with TLS + username/password and validates messages (paho-mqtt 2.x)."""
from __future__ import annotations

import json
import logging
import os
from collections import OrderedDict
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("rules-engine")

BROKER = os.getenv("MQTT_HOST", "broker.local")
PORT = int(os.getenv("MQTT_PORT", "8883"))
USER = os.getenv("MQTT_USER", "rules-engine")
PASSWORD = os.environ["MQTT_PASS"]  # never hard-code; use an env variable or a secrets vault
CA = os.getenv("MQTT_CA", "/etc/cx/certs/ca.crt")
MAX_SKEW_S = 300  # reject messages more than 5 minutes old or in the future
REQUIRED = {"schema", "device_id", "site", "ts", "seq", "data"}

seen: OrderedDict[tuple[str, int], None] = OrderedDict()  # to drop QoS 1 duplicates


def validate(topic: str, raw: bytes) -> dict | None:
    try:
        msg = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError):
        log.warning("Not JSON, skipped: %s", topic)
        return None
    if not isinstance(msg, dict) or not REQUIRED.issubset(msg):
        log.warning("Schema mismatch, skipped: %s", topic)
        return None
    # The device ID in the topic must match the one in the body (ACL protects the topic, not the body).
    parts = topic.split("/")  # cx/{site}/{device}/telemetry
    if len(parts) < 4 or parts[2] != msg["device_id"] or parts[1] != msg["site"]:
        log.warning("Identity mismatch: topic=%s body=%s", topic, msg.get("device_id"))
        return None
    try:
        ts = datetime.fromisoformat(msg["ts"].replace("Z", "+00:00"))
    except (AttributeError, ValueError):
        log.warning("Invalid timestamp: %s", msg.get("ts"))
        return None
    if ts.tzinfo is None or abs((datetime.now(timezone.utc) - ts).total_seconds()) > MAX_SKEW_S:
        log.warning("Timestamp outside the allowed clock skew: %s", msg["ts"])
        return None
    key = (msg["device_id"], msg["seq"])
    if key in seen:  # QoS 1 is "at least once"; the same message may arrive twice
        return None
    seen[key] = None
    if len(seen) > 10_000:
        seen.popitem(last=False)
    return msg


def on_connect(c, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        log.error("Connection refused: %s", reason_code)
        return
    # Subscribe inside on_connect so subscriptions are renewed after a reconnect.
    # $share/rules/... : MQTT 5 shared subscription; instances in the group split the load.
    c.subscribe([("$share/rules/cx/+/+/telemetry", 1), ("cx/+/+/status", 1)])


def on_subscribe(c, userdata, mid, reason_code_list, properties):
    for rc in reason_code_list:
        if rc.is_failure:  # rejected subscription (Mosquitto may accept a forbidden topic and filter on delivery)
            log.error("Abonelik reddedildi: %s", rc)


def on_message(c, userdata, msg):
    if msg.topic.endswith("/status"):
        log.info("Status %s = %s (retain=%s)", msg.topic, msg.payload.decode(), msg.retain)
        return
    data = validate(msg.topic, msg.payload)
    if data is not None:
        log.info("Valid reading %s seq=%s %s", data["device_id"], data["seq"], data["data"])


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="rules-engine-1",
                     protocol=mqtt.MQTTv5)
client.tls_set(ca_certs=CA)
client.username_pw_set(USER, PASSWORD)
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.on_message = on_message
client.reconnect_delay_set(min_delay=1, max_delay=60)

props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = 3600
client.connect_async(BROKER, PORT, keepalive=30, clean_start=False, properties=props)
client.loop_forever(retry_first_connection=True)  # keeps retrying even if the broker is down at start

The most important line of the validation compares the identity in the topic with the one in the body. The ACL stops cam-07 from writing to another device’s topic, but it does not stop cam-07 from sending a message to its own topic with "device_id": "cam-08" in the body. The timestamp check catches devices with drifting clocks and replays of old messages; deduplication on (device_id, seq) compensates for the “at least once” nature of QoS 1.

I noticed one more detail in testing: when the service reconnected and subscribed again inside on_connect, the retained status messages arrived again. That is normal MQTT behaviour; either make your handler idempotent, or use the MQTT 5 subscription option retainHandling=SubscribeOptions.RETAIN_SEND_IF_NEW_SUB so retained messages are sent only for a new subscription.

Message schema

{
  "schema": "cx.telemetry/1",
  "device_id": "cam-07",
  "site": "gebze-01",
  "ts": "2026-09-25T09:14:03.120Z",
  "seq": 1842,
  "data": {"temp_c": 24.61}
}

• schema: The body version; if a field’s meaning changes, the version changes.
• ts: UTC in ISO 8601. Convert to local time only for display; devices should be synchronised with NTP.
• seq: A per-device counter. Useful for deduplication and for noticing lost messages.
• Units in field names: temp_c, duration_s.
• Size: The default packet limit in Mosquitto 2.1 is 2,000,000 bytes; images and large files should not travel over MQTT.
• No personal data: Do not carry images, faces or identities in camera events. Under KVKK (Turkey’s Personal Data Protection Law No. 6698, broadly comparable to the GDPR), the cleanest schema is one that does not identify people.

Reconnection and message loss

While disconnected, paho keeps QoS 1 and 2 messages in memory and sends them once the connection returns; in my test the first message was queued with rc=4 (no connection) and delivered after the connection was established. max_queued_messages_set caps that queue. It lives in memory, so it is lost if the process restarts. For data where loss is unacceptable, keep a persistent buffer on the device (for example SQLite) and resend using seq. On the broker, max_queued_messages limits the number of QoS 1/2 messages queued per client; the default is 1000, and messages beyond the limit are silently dropped. Choose this value deliberately for subscribers that may stay offline for long periods.

Common security mistakes

1. Exposing 1883 to the network. Do not define a plain-text listener; even for testing, bind it to 127.0.0.1 only.
2. allow_anonymous true. The most common hole left over from old installations.
3. The same username or certificate on every device. One leak exposes them all, and you cannot revoke a single device.
4. Turning off verification. tls_insecure_set(True), CERT_NONE or mosquitto_sub --insecure must not reach production.
5. Broad ACLs. A user with topic readwrite # is the same as having no ACL.
6. Trusting the body. The ACL protects the topic; the identity and timestamp in the body must be validated separately.
7. Retaining commands. A retained “open the door” command is re-sent to every new subscriber.
8. Keeping the CA key on the broker, with no plan for certificate lifetime and revocation.
9. Leaving passwords in code, on the command line or in logs.
10. Opening $SYS and dashboards to everyone. Give read access to monitoring topics to the monitoring user only.

You can quickly verify the setup with the two commands below (-P makes the password visible in the process list, so use it only for testing):

mosquitto_sub -h broker.local -p 8883 --cafile ca.crt -u dashboard -P '<password>' -t 'cx/#' -v
mosquitto_pub -h broker.local -p 8884 --cafile ca.crt --cert cam-07.crt --key cam-07.key \
  -t 'cx/gebze-01/cam-07/telemetry' -q 1 -m '{"test": 1}'

I showed how the camera analytics service publishes its events to this broker in real-time object detection with Python and YOLO.

Checklist

• No plain-text listener; allow_anonymous false.
• The broker certificate’s SAN contains the name clients use; the CA key is kept off the broker.
• Every device has its own certificate with CN = device ID; a CRL process for revocation is defined.
• Service users have separate passwords; file permissions are restricted.
• In the ACL, devices write only to their own branch via the %u pattern; only the authorised service writes to the command branch.
• ACL tests ask “did a forbidden message arrive?”.
• Clients use the VERSION2 API, subscribe inside on_connect, and keep TLS verification on.
• The subscribing service checks the schema, topic–body identity match, timestamp and duplicates.
• Retain is used only for state; commands are never retained.
• Queue limits (client and broker) and the need for a persistent buffer have been assessed.