From faba28f824ac2c5e3c05d3f3d59b7b676000acbd Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Sat, 10 Oct 2026 19:13:35 +0300
Subject: [PATCH] =?UTF-8?q?fix(iot):=20replaceDSNDatabase=20(=D1=81=D0=BC?=
 =?UTF-8?q?=D0=B5=D1=88=D0=B5=D0=BD=D0=B8=D0=B5=20=D1=82=D0=B5=D0=BD=D0=B0?=
 =?UTF-8?q?=D0=BD=D1=82=D0=BE=D0=B2)=20+=20guard=20namespace=20(=D0=AD?=
 =?UTF-8?q?=D1=82=D0=B0=D0=BF=204)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- replaceDSNDatabase: возвращает ОШИБКУ для key=value DSN (раньше молча отдавал
  исходный DSN → телеметрия всех тенантов ушла бы в management-БД). Query
  отделяется первым '?', поэтому sslrootcert=/path больше не ломает поиск БД.
- EnsureTenantDB: guard safeNamespaceRe (defense-in-depth, совпадает с API-валидацией);
  небезопасный namespace → ErrInvalidNamespace → в consumer классифицируется как
  постоянная ошибка (DLQ, не вечный ретрай).
---
 internal/service/consumer/processor.go        |  5 ++
 internal/storage/iotpg/iot_telemetry_store.go | 52 +++++++++++++------
 2 files changed, 41 insertions(+), 16 deletions(-)

diff --git a/internal/service/consumer/processor.go b/internal/service/consumer/processor.go
index abf0c24..e87f6b8 100644
--- a/internal/service/consumer/processor.go
+++ b/internal/service/consumer/processor.go
@@ -39,6 +39,11 @@ func processTelemetry(ctx context.Context, body string, store *iotpg.IoTPostgres
 
 	// EnsureTenantDB идемпотентен, кэшируется после первого вызова.
 	if err := store.EnsureTenantDB(ctx, envelope.Namespace); err != nil {
+		if errors.Is(err, iotpg.ErrInvalidNamespace) {
+			// Небезопасный namespace — постоянная ошибка → в DLQ (не ретраить).
+			log.Warn("consumer: invalid namespace, routing to DLQ", "namespace", envelope.Namespace)
+			return errInvalidEnvelope
+		}
 		return err
 	}
 	// Время приёма сохраняем в ts (порядок телеметрии). Битый/пустой received_at
diff --git a/internal/storage/iotpg/iot_telemetry_store.go b/internal/storage/iotpg/iot_telemetry_store.go
index 6064b02..4a32fa1 100644
--- a/internal/storage/iotpg/iot_telemetry_store.go
+++ b/internal/storage/iotpg/iot_telemetry_store.go
@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"log/slog"
 	"os"
+	"regexp"
 	"strings"
 	"sync"
 	"time"
@@ -67,6 +68,14 @@ const telemetryBatchSize = 100
 // telemetryFlushInterval — период фонового flush неполных батчей.
 const telemetryFlushInterval = 200 * time.Millisecond
 
+// ErrInvalidNamespace — namespace небезопасен для имени БД tenant_<ns>/username.
+var ErrInvalidNamespace = errors.New("invalid namespace")
+
+// safeNamespaceRe — безопасный namespace: [a-zA-Z0-9-], не с дефиса, ≤56 символов
+// (имя БД tenant_<ns> ≤ 63 байт PG). Совпадает с валидацией API (identRe).
+// Defense-in-depth: основной барьер — валидация при создании устройства.
+var safeNamespaceRe = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9-]{0,55}$`)
+
 // TelemetryRow — одна запись телеметрии из таблицы iot_telemetry.
 type TelemetryRow struct {
 	ID       int64           `json:"id"`
@@ -269,6 +278,10 @@ created_at  TIMESTAMPTZ DEFAULT now()
 // EnsureTenantDB создаёт DATABASE, USER и таблицу iot_telemetry для namespace.
 // Идемпотентен — повторный вызов безопасен (кэш ensured пропускает проверки).
 func (s *IoTPostgresStore) EnsureTenantDB(ctx context.Context, namespace string) error {
+	// Guard (defense-in-depth): небезопасный namespace → постоянная ошибка.
+	if !safeNamespaceRe.MatchString(namespace) {
+		return fmt.Errorf("%w: %q", ErrInvalidNamespace, namespace)
+	}
 	if _, ok := s.ensured.Load(namespace); ok {
 		return nil
 	}
@@ -616,7 +629,10 @@ func (s *IoTPostgresStore) getTenantDB(ctx context.Context, namespace string) (*
 	if cached, ok := s.tenants.Load(namespace); ok {
 		return cached.(*sql.DB), nil
 	}
-	dsn := replaceDSNDatabase(s.adminDSN, tenantDBName(namespace))
+	dsn, err := replaceDSNDatabase(s.adminDSN, tenantDBName(namespace))
+	if err != nil {
+		return nil, err
+	}
 	db, err := sql.Open("postgres", dsn)
 	if err != nil {
 		return nil, fmt.Errorf("iotpg: open tenant DB %s: %w", tenantDBName(namespace), err)
@@ -639,26 +655,30 @@ func (s *IoTPostgresStore) getTenantDB(ctx context.Context, namespace string) (*
 	return db, nil
 }
 
-// replaceDSNDatabase заменяет имя базы данных в DSN.
-// Вход:  "postgresql://user:pass@host:5432/iot_platform?sslmode=disable"
-// Выход: "postgresql://user:pass@host:5432/tenant_abc?sslmode=disable"
-func replaceDSNDatabase(dsn, newDBName string) string {
+// replaceDSNDatabase заменяет имя базы данных в URL-DSN.
+// Вход:  "postgresql://user:pass@host:5432/iot_platform?sslmode=require"
+// Выход: "postgresql://user:pass@host:5432/tenant_abc?sslmode=require"
+//
+// Возвращает ОШИБКУ для не-URL (key=value) DSN: иначе функция молча вернула бы
+// исходный DSN, и телеметрия ВСЕХ тенантов ушла бы в management-БД — смешение
+// тенантов (ревью 2026-10-10). Query отделяется ПЕРВЫМ '?', поэтому '/' внутри
+// query (напр. sslrootcert=/path) не ломает поиск имени БД.
+func replaceDSNDatabase(dsn, newDBName string) (string, error) {
 	schemeEnd := strings.Index(dsn, "://")
 	if schemeEnd < 0 {
-		return dsn
+		return "", fmt.Errorf("iotpg: replaceDSNDatabase: неподдерживаемый DSN (нет «://»); нужен URL вида postgresql://user:pass@host:port/db")
 	}
-	hostPart := dsn[schemeEnd+3:]
-	slashIdx := strings.LastIndex(hostPart, "/")
-	if slashIdx < 0 {
-		return dsn
+	rest := dsn[schemeEnd+3:]
+	query := ""
+	if q := strings.Index(rest, "?"); q >= 0 {
+		query = rest[q:]
+		rest = rest[:q]
 	}
-	afterSlash := hostPart[slashIdx+1:]
-	suffix := ""
-	if qIdx := strings.Index(afterSlash, "?"); qIdx >= 0 {
-		suffix = afterSlash[qIdx:]
+	slash := strings.LastIndex(rest, "/")
+	if slash < 0 {
+		return "", fmt.Errorf("iotpg: replaceDSNDatabase: в DSN нет имени базы данных")
 	}
-	prefix := dsn[:schemeEnd+3+slashIdx+1]
-	return prefix + newDBName + suffix
+	return dsn[:schemeEnd+3] + rest[:slash+1] + newDBName + query, nil
 }
 
 // tenantDBName возвращает имя Postgres DATABASE для namespace.