IoT Cihazlarının Güvenli Yapay Zeka Kontrolü: Yapay Zekanın Verdiği MQTT Komutları için Önlemler

IoT cihazlarının güvenli yapay zeka kontrolü, bir dil modelinin eylem önerebilmesi, ancak bir röleye, vanaya veya motora yalnızca doğrulanmış, hız sınırlı ve kayıt altına alınmış komutların ulaşması anlamına gelir. Bu rehberde bir yapay zeka ajanı ile MQTT cihazlarınız arasında duran bir komut gateway'i tasarlayacak; izin listelerini, aralık kontrollerini, hız sınırlarını, insan onayını ve bir denetim topic'ini yaklaşık 150 satır Python ile uygulayacaksınız.

Yapay Zeka Komutlarının Neden Kendi Güvenlik Katmanına İhtiyacı Var?

Dil modelleri "toplantı odası çok sıcak" ifadesini "oda 3'ün ayar noktasını 21 dereceye ayarla" komutuna dönüştürmekte iyidir. Ancak bir oda numarasını yanlış okuyabilir, bir komutu döngü içinde tekrarlayabilir veya bir sensör etiketine ya da bir e-postaya gizlenmiş bir talimatı izleyebilirler. Bunların hiçbiri olasılıksal bir metin üreteci için olağandışı davranış değildir ve yalnızca prompt'larla bunları dışlamak mümkün değildir.

Çözüm, kontrol mühendislerinin her zaman kullandığı yöntemin aynısıdır: üst akıştaki kaynağın yanılabileceğini varsayın ve her şeyi kontrolünüzdeki bir sınırda doğrulayın. Yapay zeka güvenilmeyen bir client daha olur ve neyin izinli olduğuna deterministik bir program karar verir.

İkinci bir neden daha var. Mükemmel derecede doğru bir model bile bayat bilgiye dayanarak hareket edebilir. On dakikalık bir tank seviyesini okuyup bir doldurma vanasını açarsa, komut "makuldü" ama sonuç yine de bir taşkındır. Harekete geçmeden önce güncelliği ve mevcut durumu kontrol etmenin yeri gateway'dir.

Komut Gateway'i Deseni

Mimarinin tek bir kuralı vardır: yapay zeka hiçbir zaman cihazlara ulaşabilen broker kimlik bilgilerini tutmaz. Bunun yerine:

1. Ajan, IoT için LLM function calling rehberimizde anlatıldığı gibi request_action(device, action, value, reason) benzeri bir aracı çağırır.
2. Araç isteği, sahibi olduğunuz küçük bir servis olan gateway'e iletir.
3. Gateway isteği doğrular, limitleri uygular, isteğe bağlı olarak bir insana sorar ve ancak bundan sonra cihazın komut topic'ine yayın yapar.
4. Kabul edilen veya reddedilen her karar bir denetim topic'ine yayınlanır ve bir log'a yazılır.
5. Cihaz bir state topic'i üzerinden onay verir ve gateway gerçek sonucu ajana bildirir.

Bu önemlidir, çünkü kimlik bilgilerine sahip bir MQTT client'ı genellikle hesabının ulaşabildiği her topic'e yayın yapabilir. ConnectMQ client ID'leri verdiği kimlik bilgilerine bağlar, ancak topic bazında ACL sunmaz; bu yüzden "yapay zeka pompaya doğrudan dokunamaz" kuralını uygulamanın dürüst yolu, cihaz kimlik bilgilerini gateway sürecinin içinde ve ajanın tamamen dışında tutmaktır. Ajan bir MQTT MCP sunucusu olarak çalışıyorsa, yalnızca gateway destekli araçları sunun, asla ham bir publish aracı açmayın.

IoT Cihazlarının Güvenli Yapay Zeka Kontrolü için Önlemler

Her önlem belirli bir hata türünü ele alır:

İnsanları ve ekipmanı koruyan kilitlemeler (interlock) PLC'de veya cihaz firmware'inde kalır. Gateway rahatsızlığı ve riski azaltır; güvenlik sertifikalı bir sistemin yerini tutmaz.

Korumalı Komutlar için Topic Düzeni

Yapay zekadan gelen trafiği normal trafikten görünür şekilde ayrı tutun. İyi çalışan bir düzen:

<prefix>/ai/requests            AI proposals (JSON), written by the agent tool
<prefix>/ai/pending/<id>        retained while waiting for a human
<prefix>/ai/confirm/<id>        human publishes "approve" or "reject"
<prefix>/ai/audit               every decision, never retained
<prefix>/ai/enabled             retained "true"/"false" kill switch
<prefix>/home/<room>/<device>/set     command topic, only the gateway publishes
<prefix>/home/<room>/<device>/state   retained device state

Akıllı ev MQTT topic yapısı yazımızda anlatılan oda/cihaz/özellik şemasını zaten izliyorsanız, gateway onun yanına yalnızca ai/ alt ağacını ekler.

Gateway'i Python ile Uygulamak

Aşağıdaki gateway paho-mqtt 2.x ve MQTT 5 kullanıyor. Deneyebilmeniz için herkese açık test broker'ına karşı çalışır; öneki kendi rastgele kimliğinizle değiştirin.

import json
import time
import uuid
from collections import defaultdict, deque

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

PREFIX = "connectmq/demo/r8x2m4"   # use your own random id
ALLOW = {
    "livingroom/thermostat": {"set_temp": {"min": 16, "max": 24, "confirm": False}},
    "garage/door":           {"open": {"confirm": True}, "close": {"confirm": False}},
    "garden/valve":          {"open_minutes": {"min": 1, "max": 15, "confirm": True}},
}
RATE_LIMIT = (3, 600)          # max 3 commands per device per 10 minutes
MAX_STATE_AGE = 120            # seconds
history = defaultdict(deque)
last_state = {}
pending = {}
ai_enabled = True

def audit(client, decision, req, reason):
    rec = {"ts": time.time(), "decision": decision, "reason": reason, "request": req}
    client.publish(f"{PREFIX}/ai/audit", json.dumps(rec), qos=1)
    with open("ai_audit.log", "a", encoding="utf-8") as f:
        f.write(json.dumps(rec) + "\n")

def validate(req):
    if not ai_enabled:
        return "AI control disabled"
    rule = ALLOW.get(req.get("device"), {}).get(req.get("action"))
    if rule is None:
        return "device/action not allowlisted"
    if "min" in rule:
        v = req.get("value")
        if not isinstance(v, (int, float)) or not rule["min"] <= v <= rule["max"]:
            return f"value outside {rule['min']}..{rule['max']}"
    seen = last_state.get(req["device"])
    if seen is None or time.time() - seen > MAX_STATE_AGE:
        return "device state unknown or stale"
    window = history[req["device"]]
    while window and time.time() - window[0] > RATE_LIMIT[1]:
        window.popleft()
    if len(window) >= RATE_LIMIT[0]:
        return "rate limit reached"
    return None

def execute(client, req):
    props = Properties(PacketTypes.PUBLISH)
    props.MessageExpiryInterval = 15   # drop if not delivered within 15 s
    payload = {"action": req["action"], "value": req.get("value"), "source": "ai", "id": req["id"]}
    client.publish(f"{PREFIX}/home/{req['device']}/set", json.dumps(payload), qos=1, properties=props)
    history[req["device"]].append(time.time())
    audit(client, "executed", req, "passed all checks")

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe([(f"{PREFIX}/ai/requests", 1), (f"{PREFIX}/ai/confirm/+", 1),
                      (f"{PREFIX}/ai/enabled", 1), (f"{PREFIX}/home/+/+/state", 1)])

def on_message(client, userdata, msg):
    global ai_enabled
    t = msg.topic
    if t.endswith("/state"):
        device = "/".join(t.split("/")[-3:-1])
        last_state[device] = time.time()
    elif t == f"{PREFIX}/ai/enabled":
        ai_enabled = msg.payload.decode().strip().lower() == "true"
    elif t == f"{PREFIX}/ai/requests":
        try:
            req = json.loads(msg.payload)
        except ValueError:
            return
        req["id"] = str(uuid.uuid4())[:8]
        problem = validate(req)
        if problem:
            audit(client, "rejected", req, problem)
            return
        if ALLOW[req["device"]][req["action"]].get("confirm"):
            pending[req["id"]] = req
            client.publish(f"{PREFIX}/ai/pending/{req['id']}", json.dumps(req), qos=1, retain=True)
            audit(client, "awaiting_confirmation", req, "high-impact action")
        else:
            execute(client, req)
    elif "/ai/confirm/" in t:
        req_id = t.rsplit("/", 1)[-1]
        req = pending.pop(req_id, None)
        if req is None:
            return
        client.publish(f"{PREFIX}/ai/pending/{req_id}", b"", qos=1, retain=True)  # clear
        if msg.payload.decode().strip() == "approve":
            problem = validate(req)   # re-check: state may have changed
            if problem:
                audit(client, "rejected", req, "on confirm: " + problem)
            else:
                execute(client, req)
        else:
            audit(client, "rejected", req, "human rejected")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="ai-gateway-r8x2m4",
                     protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Bir terminalden test edin. Önce güncel bir cihaz durumu, ardından çalışması gereken bir istek, sonra da engellenmesi gereken bir istek yayınlayın:

P=connectmq/demo/r8x2m4
mosquitto_pub -h broker.connectmq.com -p 1883 -i t1-r8x2m4 -r -t $P/home/livingroom/thermostat/state -m '{"temp":22}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i t1-r8x2m4 -t $P/ai/requests \
  -m '{"device":"livingroom/thermostat","action":"set_temp","value":21,"reason":"user feels warm"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i t1-r8x2m4 -t $P/ai/requests \
  -m '{"device":"livingroom/thermostat","action":"set_temp","value":35,"reason":"test"}'
mosquitto_sub -h broker.connectmq.com -p 1883 -i t2-r8x2m4 -t "$P/ai/#" -v

İkinci istek denetim topic'inde value outside 16..24 gerekçesiyle rejected olarak görünür. Herkese açık alan herkes tarafından okunabildiğinden, onu yalnızca bunun gibi denemeler için kullanın; gerçek cihaz kontrolünü asla oraya koymayın.

Pratikte Önem Taşıyan Tasarım Notları

Onayda yeniden doğrulayın. Bir insan, bir isteği yapıldıktan beş dakika sonra onaylayabilir. Kod validate fonksiyonunu yeniden çağırır; böylece bu arada ortaya çıkan bir hız sınırı veya bayat durum isteği yine engeller.

Komutların süresini doldurun. MQTT 5 mesaj süresi dolum aralığı, çevrimdışı bir cihaz için kuyruğa alınmış bir komutun, garaj kapısı bir saat sonra yeniden bağlandığında iletilmek yerine atılması anlamına gelir. MQTT 5 özellikleri hakkındaki genel bakışımız, süre dolumunu ve daha zengin onaylar için kullanabileceğiniz response topic desenini ele alıyor.

Gerçek sonucu bildirin. Ajan kullanıcıya "ayarladım" değil, "termostat 21 dereceyi onayladı" demelidir. Yanıt vermeden önce state topic'inin değişmesini bekleyin.

Acil durdurma anahtarını sıkıcı tutun. ai/enabled üzerindeki retained bir false, tüm yapay zeka eylemlerini anında durdurur ve broker abonelik anında onu yeniden ilettiği için gateway yeniden başlatmalarından sonra da geçerli kalır.

Açıklamaları güvenilmeyen girdi olarak ele alın. Modelin okuduğu cihaz adları, notlar ve sensör etiketleri talimatlar içerebilir. Gateway kararlarında reason alanını yok sayar; bu alan yalnızca insanlar için kaydedilir.

Üretim için Güvenli Yapay Zeka Kontrolü Kontrol Listesi

1. Ajanın cihaz kimlik bilgileri yoktur; yalnızca gateway'in vardır.
2. Her cihaz/eylem çifti, aralıklarıyla birlikte açıkça izin listesine alınmıştır.
3. Hem cihaz başına hem de genel düzeyde hız sınırları vardır.
4. Yüksek etkili eylemler insan onayı gerektirir ve onay doğrulamayı yeniden çalıştırır.
5. Komutlar bir süre dolumu ve benzersiz bir kimlik taşır.
6. Her karar bir denetim topic'ine yayınlanır ve ayrı bir abone tarafından kalıcı olarak saklanır.
7. Retained bir acil durdurma anahtarı belgelenmiş ve test edilmiştir.
8. Fiziksel kilitlemeler firmware'de veya PLC'de kalır.
9. Üretim trafiği, MQTT güvenliği en iyi uygulamaları yazımızda önerildiği gibi portaldan alınan kimlik bilgileriyle 8883 portunda TLS üzerinden çalışır.

Demodan Özel Bir Broker'a Geçiş

Bağlantıyı kendi broker'ınıza ve kimlik bilgilerinize geçirin:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID", protocol=mqtt.MQTTv5)
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()
client.connect("broker.connectmq.com", 8883, keepalive=60)

ConnectMQ gibi barındırılan bir MQTT broker size TLS, otomatik oluşturulan kimlik bilgileri, flood koruması ve denetim akışını izlemek için canlı bir topic gezgini sunar; önlem mantığı ise kontrolünüzdeki kodda kalır. Komut gateway'lerinin diğer yapay zeka desenleri arasındaki yerini görmek için yapay zeka ve IoT kapsamlı rehberini, barındırma seçenekleriyle ilgili arka plan için de bulut MQTT broker sayfamızı okuyun. Aynı sistemin doğal dil tarafı için doğal dil ile cihaz kontrolü yazımıza bakın.

ConnectMQ Üzerinde Korumalı Yapay Zeka Otomasyonu Kurun

Herkese açık test broker'ıyla başlayın, ardından gateway'i TLS'li ve kendi kimlik bilgilerine sahip özel bir broker'a taşıyın. Kredi kartı gerekmeden app.connectmq.com adresinde ücretsiz bir Serverless broker oluşturun ve her yapay zeka komutunu ilk günden itibaren bir denetim topic'inde görünür tutun.

Sık Sorulan Sorular

Yapay zekayı güvende tutmak için sistem prompt'una güvenebilir miyim?

Hayır. Prompt'lar hataları azaltır, ancak davranışı garanti edemez. Kuralları modelin atlatamayacağı deterministik kodda uygulayın ve cihaz kimlik bilgilerini ajanın dışında tutun.

ConnectMQ yapay zeka komutlarını benim için doğrular mı?

Hayır. ConnectMQ mesajları güvenli ve güvenilir şekilde iletir; doğrulama, onay ve denetim, ConnectMQ'ya sıradan bir MQTT client'ı olarak bağlanan gateway servisinizde yer alır.

Hangi eylemler her zaman insan onayı gerektirmeli?

Mekanizmaları hareket ettiren, kapıları veya vanaları açan, erişim kilitlerini açan ya da tekrarlandığında hasara yol açabilecek her şey. Dar bir aralıktaki konfor ayarlarını otomatikleştirmek genellikle güvenlidir.

Denetim izini uzun vadede nasıl saklarım?

Denetim topic'inde bir dosyaya veya veritabanına yazan ayrı bir abone çalıştırın. Denetim mesajları broker'da retained tutulmaz; bu yüzden onları yakalamak için abonenin çalışıyor olması gerekir.