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

HISTORY/2026-08-16-session-log.md

@@ -905,3 +905,104 @@ EMQX_LOG__CONSOLE_HANDLER__LEVEL=debug (диагностика)
 - EMQX-логика e2e-диагностики: grep EMQX-логов по clientid → CONNECT/auth →
   authorization_permission_allowed → publish_to; монолит: auth/acl 200,
   bridge subscribed; PG: API telemetry.
+
+---
+
+## 27. Tail 3 выполнен: PG-auth в EMQX (v0.2.4) + EnsureBridgeDevice (v0.1.2) (15:20 GMT+03)
+
+### 27.1 Цель и итог
+EMQX авторизует устройства и бридж НАПРЯМУЮ в PostgreSQL (таблица
+iot_devices) — без HTTP-вызовов монолита и без kubectl на чужом кластере.
+Порядок деплоя: EMQX первым (нужны только PG-env), монолит вторым.
+Итог: живой e2e PASS (строка id=4, 15:17:40 GMT+03), негативные auth-тесты PASS.
+
+### 27.2 Монолит v0.1.2 — EnsureBridgeDevice
+- `internal/service/store/bridge.go` (новый): EnsureBridgeDevice upsert-ит
+  строку бриджа в iot_devices: namespace='__bridge', name='iot-bridge',
+  device_id=<MQTT username>, enabled=true, mqtt_password=<MQTT_PASSWORD>.
+  ON CONFLICT (namespace,name) DO UPDATE device_id/password/enabled.
+- `cmd/iot-service/main.go`: вызов после store.Open (timeout 10s), при ошибке
+  os.Exit(1). Лог "bridge device ensured" username=iot-bridge (11:29:06).
+- Makefile VERSION → v0.1.2; образ naeel/iot-service:v0.1.2 (digest
+  e269796c8f81) запушен и задеплоен. HTTP-хендлеры /internal/mqtt/auth|acl
+  ОСТАВЛЕНЫ в коде (fallback/локальные тесты), EMQX их больше не зовёт.
+
+### 27.3 EMQX PG-auth — ВЫСТРАДАННАЯ СХЕМА 5.5.1 (важно для будущего!)
+OSS-образ emqx/emqx:5.5.1 СОДЕРЖИТ postgres authn/authz
+(emqx_auth_postgresql-0.1.1), НО:
+1. Тип в конфиге — `postgresql`, НЕ `postgres`
+   (`backend = postgresql`, `type = postgresql`). `postgres` → валидация
+   падает: authentication.1 unsupported_mechanism / authorization
+   unknown_authz_type. Проверено: v0.2.2 (crash-loop на платформе).
+2. `${VAR}`-интерполяция в emqx.conf НЕ работает (значения попадают
+   буквально). Только официальные env-переопределения:
+   EMQX_AUTHENTICATION__1__{SERVER,DATABASE,USERNAME,PASSWORD},
+   EMQX_AUTHORIZATION__SOURCES__1__{SERVER,DATABASE,USERNAME,PASSWORD}.
+   В конфиге — плейсхолдеры. Проверено: emqx ctl conf show — env
+   перекрывает конфиг (v0.2.3, локальный docker-тест).
+3. Оставшиеся от HTTP-auth env `EMQX_AUTHENTICATION__1__URL` /
+   `EMQX_AUTHORIZATION__SOURCES__1__URL` → unknown_fields "url" →
+   CrashLoop. Удалены kubectl set env VAR-.
+4. В authz-postgresql НЕ поддерживаются плейсхолдеры ${action}/${topic}
+   (${action} подставлялся литералом "{action}", epgsql падал
+   lists:zip function_clause). action/topic возвращаются КОЛОНКАМИ
+   результата: permission + action + topic. Устройствам — ДВЕ строки
+   (publish и subscribe) через (VALUES ('publish'),('subscribe')).
+5. password_hash_algorithm = {name=plain, salt_position=disable} — пароли в
+   iot_devices хранятся открытым hex, сравнение прямое.
+6. SSL обязателен: ssl { enable=true, verify=verify_none } (pg_hba платформы).
+7. ENV для смены уровня лога: EMQX_LOG__CONSOLE_HANDLER__LEVEL=warning
+   (дефолт в проде; debug — только диагностика).
+
+Итоговый authn-запрос:
+  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
+Итоговый authz-запрос:
+  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}
+
+### 27.4 Версии и деплой
+- v0.2.2 (типы postgres) — CrashLoop: unsupported_mechanism/unknown_authz_type.
+- v0.2.3 (типы postgresql + ${VAR} env) — CrashLoop: unknown_fields "url"
+  (старые URL-env). После удаления URL-env под поднялся, НО ACL-запрос
+  с ${action}/${topic} падал (см. 27.3.4) → бридж отваливался close 1005.
+- v0.2.4 (ACL через колонки результата) — РАБОТАЕТ. digest cf80ec73973a.
+- EMQX env на деплойменте (kubectl set env): EMQX_AUTHENTICATION__1__SERVER=
+  postgresqlk8s-master.509145c3-….svc.cluster.local:5432, DATABASE=iotdb,
+  USERNAME=iot_service, PASSWORD=<из IOT_PG_DSN монолита>; то же для
+  AUTHORIZATION__SOURCES__1__*; URL-env удалены; лог=warning.
+- EMQX CPU-квота 2/1Gi сохранилась после всех деплоев (загрузка ~2 мин).
+
+### 27.5 Проверка authn (paho, НЕ по rc connect()!)
+⚠ paho connect() возвращает 0 ДАЖЕ при CONNACK reason≠0 — результат только
+через on_connect rc! Первые «все приняты» — артефакт теста, не баг EMQX.
+Корректные результаты (on_connect):
+- test_dev-001 + неверный пароль → CONNACK rc=4 (bad username/password) — отказ.
+- nosuch_user → CONNACK rc=5 (not authorized) — отказ.
+- test_dev-001 + верный пароль → rc=0 — приём.
+- EMQX-лог: authentication_result {ok,{error,not_authorized}} →
+  CONNACK ReasonCode=5 → websocket_terminated not_authorized.
+
+### 27.6 e2e и бридж
+- Бридж (username iot-bridge, пароль из строки __bridge) подключился и
+  подписался на +/telemetry/+ при PG-auth (12:13:11, повторно после каждого
+  рестарта EMQX — фикс resubscribe из секции 26 работает).
+- Финальный e2e: e2e-final-1 CONNACK rc=0 → publish test/telemetry/dev-001
+  {"e2e_pg_final":true,"temp":42.0} → API: id=4 ts=2026-08-16T15:17:40+03:00.
+- Цепочка: устройство → wss → EMQX (authn+authz=PG) → bridge → SQS →
+  consumer → PG → API. Без HTTP-зависимости EMQX↔монолит.
+
+### 27.7 Неожиданные события платформы (задокументированы)
+- ~11:33-11:40 kubectl с ВМ: "You must be logged in to the server
+  (Unauthorized)" на ВСЕ вызовы (get ns тоже). Самовосстановилось.
+  Диагностика велась через внешние каналы (curl/wss/paho) и локальный docker.
+- Во время окна недоступности бридж завис в одной попытке дозвона (тишина
+  в логах) → вылечено rollout restart монолита. Для будущего: в paho
+  выставить SetConnectTimeout явно (кандидат на hardening).

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
+}