MQTT ile güvenli IoT: Python, TLS ve erişim kontrolü

MQTT, istemcilerin bir aracı sunucu (broker) üzerinden konu (topic) adlarına mesaj yayınladığı ve bu konulara abone olduğu hafif bir yayın-abonelik protokolüdür. Protokol şifrelemeyi ve yetkilendirmeyi zorunlu kılmaz; bir MQTT kurulumunun güvenli olup olmadığını broker yapılandırması ve istemci kodu belirler.

Kamera analitiğini PLC, kapı kontağı ve sensörlerle aynı olay hattına bağlamanın mimarisini IoT ve kamera analitiğini birleştirmek yazısında anlattım. Bu yazı o hattın güvenlik tarafı: Mosquitto’yu TLS, istemci sertifikası ve konu bazlı yetkiyle kurmak, paho-mqtt 2.x ile doğru istemci yazmak. Buradaki yapılandırmayı ve Python kodunu Mosquitto 2.1.2 ve paho-mqtt 2.1.0 ile yerelde çalıştırıp test ettim; test sonuçları aşağıda.

MQTT’nin temelleri

Üç kavram yeterli: broker mesajları dağıtan sunucu, istemci bağlanan her cihaz ya da servis, konu cx/gebze-01/cam-07/telemetry gibi / ile ayrılmış bir adres. Yayıncı aboneyi tanımaz; yeni tüketici eklemek üreticiyi değiştirmez.

Hizmet kalitesi (QoS)

QoS	Garanti	Bedel	Nerede
0	En fazla bir kez; kayıp olabilir	En ucuz	Sık gelen, kaybı önemsiz ölçümler
1	En az bir kez; kopya gelebilir	Onay paketi	Olaylar, alarmlar (tüketici kopyaları ayıklar)
2	Tam bir kez	Dört adımlı el sıkışma	Kopyanın zarar verdiği nadir durumlar

QoS her bağlantı için ayrı işler: yayıncıdan broker’a ve broker’dan aboneye. MQTT 5 spesifikasyonuna göre aboneye giden mesajın QoS’u, yayınlanan QoS ile aboneliğin QoS’unun küçüğüdür. Sahada benim varsayılanım QoS 1 ve tüketici tarafında kopya ayıklama.

Retain, last will ve oturum

• Retain: Broker, konunun son retain’li mesajını saklar ve yeni aboneye hemen gönderir. Cihazın online/offline durumu gibi “son değer” bilgileri için doğru; olaylar ve komutlar için yanlış. Sıfır baytlık retain’li mesaj, saklanan mesajı siler.
• Last will (son vasiyet): İstemci bağlanırken bırakır; bağlantı beklenmedik biçimde koparsa broker onun adına yayınlar. Spesifikasyona göre broker, keep alive süresinin bir buçuk katı boyunca istemciden paket almazsa bağlantıyı kopmuş sayar. Düzgün DISCONNECT ile ayrılan istemci için vasiyet yayınlanmaz.
• Oturum: MQTT 5’te clean_start=False ve SessionExpiryInterval ile broker, istemci koptuğunda aboneliklerini ve bekleyen QoS 1/2 mesajlarını belirtilen süre boyunca saklar.

MQTT 5 ile gelenler

MQTT 5, sahada işe yarayan birkaç özellik ekledi: tüm onay paketlerinde sebep kodu (reason code), oturum ve mesaj süre sonu (session/message expiry), kullanıcı özellikleri (user properties), içerik türü (content type), istek-yanıt için yanıt konusu ve korelasyon verisi, yükü birden fazla tüketiciye bölen paylaşımlı abonelikler ($share/{grup}/{filtre}) ve abonelik seçenekleri (ör. retain’li mesajın ne zaman gönderileceği). Yetki hatalarını istemcide görebilmek için sebep kodları tek başına MQTT 5’i seçmeye yeter; MQTT 3.1.1’de PUBACK paketinde sebep kodu yok.

Konu (topic) tasarımı

Konu ağacı sonradan değiştirmesi en zor şeylerden biri; ACL’ler, paneller ve entegrasyonlar ona bağlanıyor. Kullandığım yapı:

cx/{site}/{cihaz}/{tür}
cx/gebze-01/cam-07/telemetry     ölçümler
cx/gebze-01/cam-07/status        online/offline (retain)
cx/gebze-01/cam-07/cmd/reboot    cihaza komut

Kurallar:

• Cihaz kimliği sabit bir seviyede olsun; ACL’deki %u deseni ancak böyle çalışır.
• Başta / kullanmayın, boşluk ve Türkçe karakterden kaçının, küçük harf kullanın.
• Konu adına kişisel veri ya da gizli bilgi koymayın; konu adları loglara ve ACL’lere düşer.
• Telemetri, durum ve komutu ayrı dallarda tutun; komut dalına yazma yetkisi yalnızca yetkili servislerde olsun.
• + tek seviye, # kalan tüm seviyeler için joker karakterdir. Cihazların # aboneliği yapmasına izin vermeyin. $ ile başlayan konular ($SYS) broker’a ayrılmıştır.

Adım 1: Mosquitto kurulumu

Debian/Ubuntu’da sudo apt install mosquitto mosquitto-clients yeterli. Yapılandırmayı ana dosyaya değil, /etc/mosquitto/conf.d/ altında ayrı bir dosyaya koymak yaygın yöntem; ana dosyada bu klasörü okuyan include_dir satırının olduğunu kontrol edin.

Sürüm notu: güncel seri 2.1 (2.1.0 Ocak 2026’da, 2.1.2 Şubat 2026’da yayımlandı); dağıtım paketleri 2.0.x olabilir. Aşağıdaki yapılandırma iki seride de geçerli. 2.1 ile password_file ve acl_file seçenekleri aynı kodu içeren eklentiler lehine kullanımdan kaldırılma sürecine girdi ve 3.0’da kaldırılacak; per_listener_settings de aynı durumda. 2.1’de max_packet_size varsayılanı 2.000.000 bayta indi. 2.0’dan beri allow_anonymous varsayılan olarak false; 1.6 ve öncesinde başka güvenlik ayarı yoksa true idi. Eski bir kurulumu devralıyorsanız ilk bakılacak satır bu.

Adım 2: Sertifikalar: CA, broker ve cihaz

Kurum içi bir sertifika otoritesi (CA) kurup broker ve her cihaz için ayrı sertifika imzalıyoruz. Aşağıdaki betik test ortamı içindir; üretimde CA anahtarı broker’da değil, çevrim dışı bir makinede ya da kurumun PKI altyapısında durmalı.

#!/usr/bin/env bash
# Test ortamı için özel CA, broker sertifikası ve bir cihaz (istemci) sertifikası üretir.
set -euo pipefail
BROKER_DNS="${BROKER_DNS:-broker.local}"
DEVICE_ID="${DEVICE_ID:-cam-07}"

# 1) Özel sertifika otoritesi (CA). ca.key'i broker'da değil, çevrim dışı bir yerde saklayın.
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 sertifikası: istemci ana bilgisayar adını SAN alanında doğrular.
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) Cihaz sertifikası: CN = cihaz kimliği (broker bunu kullanıcı adı olarak kullanacak).
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"

İki ayrıntı: istemciler broker’ın adını sertifikanın SAN (Subject Alternative Name) alanında arar; testte sertifikada yalnızca localhost varken 127.0.0.1 ile bağlanınca Python IP address mismatch hatasıyla bağlantıyı reddetti. Broker’a hangi adla bağlanılacaksa SAN’da o olmalı. -copy_extensions seçeneği OpenSSL 3.0 ve üstünü ister. Cihaz sertifikasının CN değeri cihaz kimliğidir (cam-07); broker bunu kullanıcı adı olarak kullanacak.

Adım 3: TLS dinleyicileri ve kimlik doğrulama

İki dinleyici açıyorum: 8883’te servisler ve paneller TLS üzerinden kullanıcı adı/parola ile, 8884’te sahadaki cihazlar karşılıklı TLS (mTLS) ile bağlanıyor. Şifresiz 1883 dinleyicisi hiç tanımlanmıyor.

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

# 8883: servisler ve paneller — TLS + kullanıcı adı/parola
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: sahadaki cihazlar — karşılıklı TLS (mTLS), sertifikadaki CN kullanıcı adı olur
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, izin verilen en düşük sürümü belirtir; ayarlanmazsa TLS 1.3 ve 1.2 kabul edilir. 8884’te require_certificate true, geçerli bir istemci sertifikası olmayan bağlantıyı TLS aşamasında keser. use_identity_as_username true, sertifikadaki CN’yi kullanıcı adı yapar ve bu dinleyicide parola dosyası kullanılmaz. Mosquitto belgesinin uyardığı gibi cafile içindeki her CA bu dinleyici için geçerli istemci sertifikası üretebilir; bu yüzden cihaz CA’sını başka amaçlarla paylaşmayın. İptal edilen cihazlar için crlfile ile sertifika iptal listesi verilebilir.

Servis kullanıcılarının parolalarını mosquitto_passwd ile ekleyin. -b parolayı komut satırından alır ve kabuk geçmişine düşürür; etkileşimli kullanım daha güvenli:

sudo mosquitto_passwd -c /etc/mosquitto/passwd rules-engine   # -c dosyayı yeniden oluşturur
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, bu dosyaların izinleri gevşekse uyarı veriyor ve testte uyarı metni, gelecekteki sürümlerin böyle dosyaları yüklemeyi reddedeceğini söylüyordu.

Adım 4: ACL ile konu bazlı yetki

acl_file tanımlandığında yalnızca listelenen konulara erişilebilir. user satırından sonraki topic satırları o kullanıcıya uygulanır; pattern satırları herkese uygulanır ve %u kullanıcı adıyla, %c istemci kimliğiyle değiştirilir. Belgeye göre yerine koyma deseni, bulunduğu konu seviyesinin tek içeriği olmalı.

# Cihazlar: yalnızca kendi dalına yazar, yalnızca kendi komut konusunu okur.
# %u = kullanıcı adı (mTLS dinleyicisinde sertifikanın CN değeri)
pattern write cx/+/%u/#
pattern read cx/+/%u/cmd/#

# Olay tüketen servis: tüm olayları okur, komut yazabilir
user rules-engine
topic read cx/#
topic write cx/+/+/cmd/#

# Panel: yalnızca okur
user dashboard
topic read cx/#

İki pattern satırı yüzlerce cihazı tek tek yazmadan yönetiyor: her cihaz yalnızca cx/*/kendi-kimliği/... altına yazabiliyor ve yalnızca kendi komut dalını okuyabiliyor. Bu yapıyı test ederken gördüklerim:

Deneme	Sonuç (Mosquitto 2.1.2, MQTT 5)
cam-07 kendi telemetri konusuna yayın	PUBACK: Success
cam-07, cam-08’in konusuna yayın	PUBACK: Not authorized, mesaj dağıtılmadı
dashboard komut konusuna yayın	PUBACK: Not authorized
Yanlış parola	CONNACK: Not authorized
8884’e istemci sertifikasız bağlantı	TLS el sıkışması başarısız, bağlantı kurulmadı
cam-07 ile cx/# aboneliği	SUBACK: Granted QoS 1, ama yalnızca kendi komut konusundaki mesaj teslim edildi

Son satır önemli: bu testte Mosquitto izinsiz abonelikleri (joker içeren cx/# de, başka bir cihazın komut konusu da) SUBACK’te reddetmedi; yetkisiz mesajları teslim aşamasında süzdü. ACL testlerinizi “abonelik kabul edildi mi” diye değil, “yetkisiz konudaki mesaj geldi mi” diye yazın. Cihaz sayısı büyüdüğünde ya da yetkileri çalışma anında değiştirmek gerektiğinde Mosquitto’nun Dynamic Security eklentisi dosya tabanlı yönetimden daha esnek bir seçenek.

Adım 5: Python yayıncı (cihaz)

pip install paho-mqtt ile kurulan 2.x sürümünde Client oluşturulurken geri çağırma API sürümü verilmeli. VERSION2 ile on_connect imzası (client, userdata, flags, reason_code, properties) olur ve reason_code bir nesnedir; is_failure ile hata kontrol edilir.

#!/usr/bin/env python3
"""Cihaz tarafı yayıncı: mTLS ile bağlanır, ölçümü JSON olarak QoS 1 ile yayınlar (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")  # sertifikadaki CN ile aynı olmalı
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,
)
# Varsayılanlar güvenli: sunucu sertifikası CA ile doğrulanır ve ana bilgisayar adı kontrol edilir.
client.tls_set(
    ca_certs=f"{CERT_DIR}/ca.crt",
    certfile=f"{CERT_DIR}/{DEVICE_ID}.crt",
    keyfile=f"{CERT_DIR}/{DEVICE_ID}.key",
)
# Beklenmedik kopuşta broker bu mesajı yayınlar; retain sayesinde yeni abone de son durumu görür.
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("Bağlantı reddedildi: %s", reason_code)  # örn. "Not authorized"
        return
    log.info("Bağlandı (oturum devam ediyor mu: %s)", flags.session_present)
    c.publish(f"{BASE}/status", "online", qos=1, retain=True)
    c.subscribe(f"{BASE}/cmd/#", qos=1)  # yeniden bağlanınca abonelik de yenilenir


def on_disconnect(c, userdata, flags, reason_code, properties):
    log.warning("Bağlantı kapandı: %s", reason_code)  # beklenmedik kopuşta paho kendisi yeniden bağlanır


def on_message(c, userdata, msg):
    log.info("Komut alındı %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  # kopuşta oturum ve QoS 1 kuyruğu 1 saat saklansın
client.connect_async(BROKER, PORT, keepalive=30, clean_start=False, properties=connect_props)
client.loop_start()

stop = threading.Event()  # systemd/Docker SIGTERM gönderir; düzgün kapanış için yakalayın
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  # 5 dk içinde teslim edilemezse anlamsız, broker atsın
    info = client.publish(f"{BASE}/telemetry", json.dumps(message), qos=1, properties=props)
    if info.rc != mqtt.MQTT_ERR_SUCCESS:
        log.warning("Mesaj %d bağlantı yok, bellekte bekliyor (rc=%s)", seq, info.rc)
    stop.wait(5)

# Düzgün kapanışta LWT gönderilmez; "offline" durumunu kendimiz yazıyoruz.
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()

Koddaki kararlar:

• tls_set varsayılanları güvenli: paho kaynak kodunda cert_reqs verilmezse CERT_REQUIRED kullanılıyor ve ana bilgisayar adı doğrulanıyor. “Çalışsın diye” tls_insecure_set(True) ya da CERT_NONE eklemeyin.
• reconnect_delay_set(1, 60): kopuşta 1 saniyeden başlayıp her denemede ikiye katlanan, 60 saniyede sabitlenen bekleme. Varsayılan üst sınır 120 saniye.
• connect_async + loop_start: broker açılışta erişilemese bile arka plandaki döngü bağlanmayı denemeye devam eder.
• Abonelik on_connect içinde: yeniden bağlanmada kendiliğinden yenilenir.
• SessionExpiryInterval=3600 ve clean_start=False: kısa kopuşlarda oturum ve bekleyen mesajlar kaybolmaz. Testte abone servisi durdurup yeniden başlattığımda, kapalıyken yayınlanan ölçümler yeniden bağlanınca teslim edildi.
• MessageExpiryInterval=300: beş dakikadan eski bir ölçüm artık anlamsızsa broker onu atar.
• Düzgün kapanışta last will yayınlanmadığı için offline durumu kapanırken açıkça yazılıyor.

Adım 6: Python abone (servis)

Servis tarafı mesajı alır ve ona güvenmeden önce doğrular.

#!/usr/bin/env python3
"""Servis tarafı abone: TLS + kullanıcı adı/parola ile bağlanır, mesajları doğrular (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"]  # koda gömmeyin; ortam değişkeni ya da gizli anahtar kasası
CA = os.getenv("MQTT_CA", "/etc/cx/certs/ca.crt")
MAX_SKEW_S = 300  # 5 dakikadan eski ya da ileri tarihli mesajları reddet
REQUIRED = {"schema", "device_id", "site", "ts", "seq", "data"}

seen: OrderedDict[tuple[str, int], None] = OrderedDict()  # QoS 1 kopyalarını ayıklamak için


def validate(topic: str, raw: bytes) -> dict | None:
    try:
        msg = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError):
        log.warning("JSON değil, atlandı: %s", topic)
        return None
    if not isinstance(msg, dict) or not REQUIRED.issubset(msg):
        log.warning("Şemaya uymuyor, atlandı: %s", topic)
        return None
    # Konudaki cihaz kimliği ile gövdedeki kimlik aynı olmalı (ACL konuyu korur, gövdeyi değil).
    parts = topic.split("/")  # cx/{site}/{device}/telemetry
    if len(parts) < 4 or parts[2] != msg["device_id"] or parts[1] != msg["site"]:
        log.warning("Kimlik uyuşmazlığı: konu=%s gövde=%s", topic, msg.get("device_id"))
        return None
    try:
        ts = datetime.fromisoformat(msg["ts"].replace("Z", "+00:00"))
    except (AttributeError, ValueError):
        log.warning("Geçersiz zaman damgası: %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("Zaman damgası saat farkı sınırının dışında: %s", msg["ts"])
        return None
    key = (msg["device_id"], msg["seq"])
    if key in seen:  # QoS 1 "en az bir kez" teslim eder; aynı mesaj ikinci kez gelebilir
        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("Bağlantı reddedildi: %s", reason_code)
        return
    # Abonelikleri on_connect içinde yapın: yeniden bağlanınca otomatik yenilenir.
    # $share/rules/... : MQTT 5 paylaşımlı abonelik; aynı gruptaki örnekler yükü bölüşür.
    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:  # reddedilen abonelik (Mosquitto izinsiz konuyu kabul edip teslimde süzebilir)
            log.error("Abonelik reddedildi: %s", rc)


def on_message(c, userdata, msg):
    if msg.topic.endswith("/status"):
        log.info("Durum %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("Geçerli ölçüm %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)  # broker ilk anda kapalıysa da denemeye devam eder

Doğrulamanın en önemli satırı konu ile gövdedeki kimliğin karşılaştırılması. ACL, cam-07’nin başka bir cihazın konusuna yazmasını engeller; ama cam-07’nin kendi konusuna gövdesinde "device_id": "cam-08" yazan bir mesaj göndermesini engellemez. Zaman damgası kontrolü saati kaymış cihazları ve eski mesajların yeniden oynatılmasını yakalar; (device_id, seq) ile kopya ayıklama QoS 1’in “en az bir kez” doğasını telafi eder.

Testte bir ayrıntı daha gördüm: servis yeniden bağlanıp on_connect içinde tekrar abone olunca retain’li durum mesajları yeniden geldi. Bu MQTT’nin olağan davranışı; ya işleyicinizi tekrar gelen mesajdan etkilenmeyecek (idempotent) yazın ya da MQTT 5 abonelik seçeneği retainHandling=SubscribeOptions.RETAIN_SEND_IF_NEW_SUB ile retain’li mesajın yalnızca yeni abonelikte gönderilmesini isteyin.

Mesaj şeması

{
  "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: Gövdenin sürümü; alanın anlamı değişirse sürüm de değişir.
• ts: UTC ve ISO 8601. Yerel saate yalnızca gösterimde çevirin; cihazlar NTP’ye bağlı olsun.
• seq: Cihaz başına artan sayaç. Kopya ayıklamaya ve kayıp mesajı fark etmeye yarar.
• Birim alan adında: temp_c, duration_s.
• Boyut: Mosquitto 2.1’de varsayılan paket sınırı 2.000.000 bayt; görüntü ya da büyük dosya MQTT’den taşınmamalı.
• Kişisel veri yok: Kamera olaylarında görüntü, yüz ya da kimlik bilgisi taşımayın; KVKK açısından en temiz şema, kişiyi tanımlamayan şemadır.

Yeniden bağlanma ve mesaj kaybı

paho, bağlantı yokken QoS 1 ve 2 mesajlarını bellekte tutar ve bağlantı gelince gönderir; testte ilk mesaj rc=4 (bağlantı yok) ile kuyruğa girdi ve bağlantı kurulunca teslim edildi. max_queued_messages_set bu kuyruğa sınır koyar. Bu kuyruk bellektedir, süreç yeniden başlarsa kaybolur. Kaybı kabul edilemeyen veride cihaz tarafında kalıcı bir tampon (ör. SQLite) tutun ve seq ile yeniden gönderin. Broker tarafında max_queued_messages istemci başına kuyruklanan QoS 1/2 mesaj sayısını sınırlar; varsayılan 1000 ve sınır aşıldığında yeni mesajlar sessizce düşer. Uzun süre çevrim dışı kalabilecek abonelerde bu değeri bilinçli seçin.

Sık yapılan güvenlik hataları

1. 1883’ü ağa açmak. Şifresiz dinleyici tanımlamayın; test için bile yalnızca 127.0.0.1’e bağlayın.
2. allow_anonymous true. Eski kurulumlardan kalan en yaygın açık.
3. Tüm cihazlarda aynı kullanıcı ya da sertifika. Biri sızınca hepsi gider ve tek bir cihazı iptal edemezsiniz.
4. Doğrulamayı kapatmak. tls_insecure_set(True), CERT_NONE ya da mosquitto_sub --insecure üretime taşınmamalı.
5. Geniş ACL. topic readwrite # içeren bir kullanıcı, ACL’nin hiç olmaması demektir.
6. Gövdeye güvenmek. ACL konuyu korur; gövdedeki kimlik ve zaman damgası ayrıca doğrulanmalı.
7. Komutu retain’lemek. Retain’li bir “kapıyı aç” komutu, bağlanan her yeni aboneye yeniden gider.
8. CA anahtarını broker’da tutmak, sertifika ömrü ve iptal planı yapmamak.
9. Parolayı kodda, komut satırında ya da logda bırakmak.
10. $SYS ve panelleri herkese açmak. İzleme konularına yalnızca izleme kullanıcısına okuma yetkisi verin.

Aşağıdaki iki komutla kurulumu hızlıca doğrulayabilirsiniz (-P parolayı süreç listesinde görünür kılar, yalnızca testte kullanın):

mosquitto_sub -h broker.local -p 8883 --cafile ca.crt -u dashboard -P '<parola>' -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}'

Kamera analitiği servisinin olayları bu broker’a nasıl yayınladığını Python ve YOLO ile gerçek zamanlı nesne tespiti yazısında gösterdim.

Kontrol listesi

• Şifresiz dinleyici yok; allow_anonymous false.
• Broker sertifikasının SAN alanında istemcilerin kullandığı ad var; CA anahtarı broker dışında.
• Her cihazın kendi sertifikası var, CN = cihaz kimliği; iptal için CRL süreci tanımlı.
• Servis kullanıcılarının parolaları ayrı, dosya izinleri kısıtlı.
• ACL’de cihazlar %u deseniyle yalnızca kendi dalına yazıyor; komut dalına yalnızca yetkili servis yazıyor.
• ACL testleri “yetkisiz mesaj geldi mi” sorusuyla yapıldı.
• İstemciler VERSION2 API ile yazıldı, abonelikler on_connect içinde, TLS doğrulaması açık.
• Abone servis şemayı, konu–gövde kimlik eşleşmesini, zaman damgasını ve kopyaları kontrol ediyor.
• Retain yalnızca durum bilgisinde; komutlar retain’lenmiyor.
• Kuyruk sınırları (istemci ve broker) ve kalıcı tampon ihtiyacı değerlendirildi.