feat(emqx): PG-auth in EMQX v0.2.4 + EnsureBridgeDevice v0.1.2 (tail 3)

Makefile

@@ -1,7 +1,7 @@
 # Makefile — монолит iot-service (образ naeel/iot-service).
 # Старые k8s-цели — в legacy/Makefile.old.
 
-VERSION ?= v0.1.1
+VERSION ?= v0.1.2
 IMAGE ?= naeel/iot-service
 LDFLAGS = -X main.version=$(VERSION)
 

cmd/iot-service/main.go

@@ -58,6 +58,16 @@ func main() {
 	defer devices.Close()
 	log.Info("devices store ready")
 
+	// Служебная строка бриджа в iot_devices — для PG-auth EMQX (хвост 3).
+	ensureCtx, ensureCancel := context.WithTimeout(ctx, 10*time.Second)
+	err = devices.EnsureBridgeDevice(ensureCtx, cfg.MQTTUsername, cfg.MQTTPassword)
+	ensureCancel()
+	if err != nil {
+		log.Error("ensure bridge device", "err", err)
+		os.Exit(1)
+	}
+	log.Info("bridge device ensured", "username", cfg.MQTTUsername)
+
 	iotStore, err := iotpg.New(cfg.IOTPGDSN, log)
 	if err != nil {
 		log.Error("iotpg store", "err", err)

emqx/Makefile

@@ -1,6 +1,6 @@
 # EMQX-контейнер для IoT (образ naeel/iot-emqx).
 
-VERSION ?= v0.2.1
+VERSION ?= v0.2.4
 IMAGE ?= naeel/iot-emqx
 
 .PHONY: docker-build docker-push

emqx/emqx.conf

@@ -1,7 +1,19 @@
 ## EMQX 5.x — конфиг для контейнера IoT на платформе Nubes.
-## Вшит в образ naeel/iot-emqx. Auth/ACL URL переопределяются env-переменными
+## Вшит в образ naeel/iot-emqx. Auth/ACL — напрямую в PostgreSQL (iot_devices),
-## (EMQX_AUTHENTICATION__1__URL, EMQX_AUTHORIZATION__SOURCES__1__URL) в deck-UI
+## без зависимости от адреса монолита: EMQX можно создать в deck ДО iot-service.
-## или kubectl set env — адрес монолита становится известен после его создания.
+##
+## ⚠ ${VAR}-интерполяция в этом конфиге НЕ работает (проверено на 5.5.1:
+## значения попадают в конфиг буквально). Параметры PG задаются ТОЛЬКО
+## официальными env-переопределениями (deck-UI или kubectl set env):
+##   EMQX_AUTHENTICATION__1__SERVER     — host:port master (внутренний svc)
+##   EMQX_AUTHENTICATION__1__DATABASE   — имя БД (iotdb)
+##   EMQX_AUTHENTICATION__1__USERNAME   — роль (iot_service)
+##   EMQX_AUTHENTICATION__1__PASSWORD   — пароль роли
+##   EMQX_AUTHORIZATION__SOURCES__1__SERVER / DATABASE / USERNAME / PASSWORD
+## SSL обязателен (pg_hba платформы): ssl = {enable=true, verify=verify_none}.
+## ⚠ В authz-postgresql НЕ поддерживаются плейсхолдеры ${action}/${topic} —
+## action/topic возвращаются КОЛОНКАМИ результата запроса (проверено на 5.5.1:
+## ${action} подставлялся как литерал "{action}" и ACL-запрос падал).
 
 node {
   name = "emqx@127.0.0.1"
@@ -9,30 +21,35 @@ node {
   data_dir = "/opt/emqx/data"
 }
 
-## HTTP Auth — монолит проверяет устройства в PostgreSQL (iot_devices).
+## Auth — PostgreSQL: устройства в iot_devices.
+## username устройства = namespace || '_' || device_id; бридж: namespace='__bridge',
+## username = device_id (строка создаётся монолитом EnsureBridgeDevice при старте).
 authentication = [
   {
     mechanism = password_based
-    backend = http
+    backend = postgresql
     enable = true
-    method = post
+    server = "127.0.0.1:5432"
-    url = "http://containerk8s.f1ffb134-7d16-45bd-8bef-69f6ec8ab33c.svc.cluster.local:4100/internal/mqtt/auth"
+    database = "iotdb"
-    body {
+    username = "iot_service"
-      username = "${username}"
+    password = "placeholder"
-      password = "${password}"
+    auto_reconnect = true
-      clientid = "${clientid}"
-    }
-    headers {
-      "content-type" = "application/json"
-    }
-    connect_timeout = 5s
-    request_timeout = 5s
     pool_size = 8
+    ssl {
+      enable = true
+      verify = verify_none
+    }
+    password_hash_algorithm {
+      name = plain
+      salt_position = disable
+    }
+    query = "SELECT mqtt_password AS password_hash FROM iot_devices WHERE enabled = true AND (namespace || '_' || device_id = ${username} OR (namespace = '__bridge' AND ${username} = device_id)) LIMIT 1"
   }
 ]
 
-## ACL — монолит разрешает pub/sub по топикам устройств.
+## ACL — PostgreSQL: устройства pub/sub только в свой топик ns/telemetry/device_id,
-## no_match = deny: недоступность backend = запрет.
+## бридж — subscribe на "+/telemetry/+".
+## no_match = deny: нет строки в iot_devices или устройство выключено — запрет.
 authorization {
   no_match = deny
   deny_action = disconnect
@@ -43,22 +60,19 @@ authorization {
   }
   sources = [
     {
-      type = http
+      type = postgresql
       enable = true
-      method = post
+      server = "127.0.0.1:5432"
-      url = "http://containerk8s.f1ffb134-7d16-45bd-8bef-69f6ec8ab33c.svc.cluster.local:4100/internal/mqtt/acl"
+      database = "iotdb"
-      body {
+      username = "iot_service"
-        username = "${username}"
+      password = "placeholder"
-        clientid = "${clientid}"
+      auto_reconnect = true
-        action = "${action}"
-        topic = "${topic}"
-      }
-      headers {
-        "content-type" = "application/json"
-      }
-      connect_timeout = 5s
-      request_timeout = 5s
       pool_size = 8
+      ssl {
+        enable = true
+        verify = verify_none
+      }
+      query = "SELECT 'allow' AS permission, 'subscribe' AS action, '+/telemetry/+' AS topic FROM iot_devices WHERE enabled = true AND namespace = '__bridge' AND ${username} = device_id UNION ALL SELECT 'allow' AS permission, a.action, namespace || '/telemetry/' || device_id AS topic FROM iot_devices, (VALUES ('publish'), ('subscribe')) AS a(action) WHERE enabled = true AND namespace || '_' || device_id = ${username}"
     }
   ]
 }

internal/service/store/bridge.go

@@ -0,0 +1,32 @@
+package store
+
+import (
+	"context"
+	"fmt"
+)
+
+// bridgeNamespace — служебный namespace строки бриджа в iot_devices.
+// EMQX (postgres auth/ACL) узнаёт бридж по условию
+// namespace = '__bridge' AND username = device_id.
+const bridgeNamespace = "__bridge"
+
+// EnsureBridgeDevice создаёт (или обновляет) служебную строку бриджа в
+// iot_devices. EMQX авторизует MQTT-подключение бриджа по этой строке,
+// поэтому device_id = MQTT username (iot-bridge), пароль = MQTT_PASSWORD.
+//
+// Нужно для PG-auth EMQX (хвост 3): EMQX зависит только от PostgreSQL,
+// монолит при старте гарантирует наличие строки бриджа.
+func (s *DeviceStore) EnsureBridgeDevice(ctx context.Context, username, password string) error {
+	_, err := s.db.ExecContext(ctx, `
+INSERT INTO iot_devices (namespace, name, device_id, enabled, mqtt_password, metadata, phase)
+VALUES ($1, $2, $3, true, $4, '{}'::jsonb, 'Active')
+ON CONFLICT (namespace, name) DO UPDATE
+  SET device_id     = EXCLUDED.device_id,
+      enabled       = true,
+      mqtt_password = EXCLUDED.mqtt_password`,
+		bridgeNamespace, "iot-bridge", username, password)
+	if err != nil {
+		return fmt.Errorf("store: ensure bridge device: %w", err)
+	}
+	return nil
+}