MQTT Kimlik Doğrulamasını Test Etmek: Geçerli, Geçersiz ve Eksik Bilgiler

MQTT kimlik doğrulamasını test etmek; broker'ınızın alabileceği her tür kimlik bilgisiyle — geçerli, yanlış, bilinmeyen, boş ve hiç gönderilmemiş — bağlanmak ve her birinin doğru CONNACK kodunu aldığını doğrulamak demektir. Doğru parolayı kabul eden bir broker yalnızca yarı yarıya test edilmiştir; güvenlik, geri kalan her şeyi nasıl reddettiğinden gelir.

Bu rehber size MQTT 3.1.1 ve MQTT 5 için her durumda beklenen sonucu gösteren bir test matrisi, matrisi elle çalıştırmak için komutlar ve CI'da otomatik çalıştıran bir pytest paketi sunar.

MQTT kimlik doğrulama sonucunu nasıl bildirir

Kimlik doğrulama yalnızca bir kez, CONNECT paketinde yapılır. İstemci bir kullanıcı adı ve/veya parola gönderir, broker CONNACK ile yanıt verir. Sıfırdan farklı bir kod bağlantının reddedildiği anlamına gelir ve broker bu paketi gönderdikten hemen sonra soketi kapatır.

Spesifikasyon, broker'ı "yanlış parola" ile "yetkisiz" arasında seçim yapmaya zorlamaz. Bazı broker'lar yanlış parola için 0x04/0x86 döndürür; Mosquitto ise her kimlik bilgisi hatasında 0x05/0x87 döndürür. Bu sorun değildir — önemli olan broker'ınızın tutarlı olması ve testlerinizin onun gerçek davranışını sabitlemesidir. Gördüğünüz her kodu MQTT reason code aracında bulabilirsiniz.

Kimlik doğrulama test matrisi

Her satırı canlıda kullanacağınız broker'a karşı, en az bir kez MQTT 3.1.1 ve bir kez MQTT 5 ile çalıştırın. "Reddedildi", 3.1.1'de 0x04/0x05, 5'te 0x86/0x87 anlamına gelir.

1. ve 9. durumlar gerçek hataları yakalar. Bilinmeyen bir kullanıcı yanlış paroladan farklı bir kod alıyorsa broker hangi hesapların var olduğunu sızdırıyordur. Sondaki satır sonu kabul ediliyorsa parola dosyanız veya kimlik doğrulama backend'iniz girdiyi kırpıyordur — genellikle zararsızdır ama bilmek istersiniz. 10. durum iptalin gerçekten etkili olduğunu kontrol eder: Mosquitto'da bu, parola dosyasının yeniden yüklendiği (SIGHUP), yönetilen bir broker'da ise değişikliğin yayıldığı anlamına gelir.

Matrisi elle çalıştırmak

Her durumu denemenin en hızlı yolu mosquitto_pub'dır. Ham CONNACK'i görmek için -d ekleyin:

# 1. geçerli bilgiler — sessizce 0 koduyla çıkar
mosquitto_pub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/auth/check -m ok -d

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

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

# 4. hiç kimlik bilgisi yok
mosquitto_pub -h broker.connectmq.com -p 1883 -t testmqtt/auth/check -m ok -d

Reddedilen bir bağlantı aşağıdaki satırlardan birini yazdırır; bunlar insanların en çok aradığı hata metinleridir:

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

-V mqttv5 ile aynı durumlar MQTT 5 reason metnini yazdırır, örneğin Connection error: Not authorized. Görsel bir kontrol için online MQTT istemcisi her denemede CONNACK sonucunu gösterir — bazı broker'ların farklı kimlik doğrulama backend'leriyle yapılandırdığı WebSocket ve TCP listener'larını karşılaştırmak için kullanışlıdır.

Bunun yerine Connection Refused: identifier rejected görüyorsanız sorun kimlik bilgilerinde değil, client ID'dedir — bkz. identifier rejected. İstemci takılıp zaman aşımına uğruyorsa büyük olasılıkla yanlış porttasınız veya bir güvenlik duvarının arkasındasınız; kimlik doğrulamayla ilgisi yoktur.

pytest ile otomatikleştirmek

Elle yapılan kontroller broker'ın bugün çalıştığını kanıtlar. Otomatik bir test paketi ise bir sonraki yapılandırma değişikliğinden sonra da çalıştığını kanıtlar. Bu test paho-mqtt 2.x kullanır ve her durumu iki protokol sürümüyle de çalıştırır:

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):
    """CONNACK reason metnini, CONNACK gelmezse 'closed' döndürür."""
    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")

Birkaç ayrıntı bu testi güvenilir kılar:

• paho-mqtt 2.x, MQTT 3.1.1 return code'larını MQTT 5 reason code'larına çevirir; böylece 0x04 iki sürümde de "Bad user name or password", 0x05 ise "Not authorized" olarak gelir. Tek bir assertion seti ikisini de kapsar.
• Her deneme yeni, rastgele bir client ID kullanır. Aynı ID'yi tekrar kullanmak, başarılı bir testin önceki bağlantıyı düşürmesine ve sonucu karıştırmasına yol açabilir. Client ID üreticisi aynı biçimde ID'ler üretir.
• "closed", CONNACK göndermek yerine TCP bağlantısını kesen broker'ları kapsar; bu bazı hatalarda kurallara uygundur.
• Son test, matrisin 3. durumundaki hesap listeleme (enumeration) kontrolüdür.

MQTT_USER=... MQTT_PASS=... pytest -v test_auth.py ile çalıştırın. CI'da kimlik bilgilerini secret olarak saklayın ve broker yapılandırmasındaki her değişiklikte paketi bir staging broker'a karşı çalıştırın — bkz. GitHub Actions'ta MQTT testleri.

Kimlik doğrulama testlerinin kapsamadıkları

Başarılı bir CONNECT yalnızca kimliği kanıtlar. İstemcinin sonrasında ne yapabileceği hakkında hiçbir şey söylemez. Geçerli kimlik bilgilerine sahip bir sensör yine de başka bir cihazın komut topic'ine publish edememelidir — bu yetkilendirmedir ve farklı şekilde başarısız olur (3.1.1'de sessizce, MQTT 5'te bir reason code ile). Bunu MQTT ACL'lerini test etme yöntemiyle ayrıca test edin.

Ayrıca şunları aklınızda tutun:

• Şifresiz TCP parolaları açığa çıkarır. CONNECT içindeki kimlik bilgileri şifrelenmez. Geçerli bilgi durumunu 8883 portunda TLS üzerinden de çalıştırın ve canlıda şifresiz listener'ın kapalı olduğunu doğrulayın.
• Kaba kuvvet koruması ayrı bir testtir. Bazı broker'lar veya proxy'ler tekrarlanan hatalı denemeleri yavaşlatır ya da engeller. Sizinki bunu yapıyorsa, art arda on yanlış parolanın meşru istemciyi kilitlemediğini — ya da politikanız buysa kilitlediğini — doğrulayan bir test ekleyin.
• Token tabanlı kimlik doğrulama süre testleri ister. Statik parolalar yerine JWT veya OAuth kullanıyorsanız süresi dolmuş, henüz geçerli olmayan ve yanlış imzalanmış token'lar için durumlar ekleyin.

Geçerli ve geçersiz bilgileri online istemcide deneyin

Kimlik doğrulama testi ucuzdur: bir düzine bağlantı denemesi ve birkaç assertion. Matrisi deponuzda broker yapılandırmasının yanında tutun, her değişiklikte çalıştırın; birisi anonim erişimi yeniden açtığında bunu bir saldırgandan önce siz fark edersiniz.

Sıkça sorulan sorular