fix(iot): replaceDSNDatabase (смешение тенантов) + guard namespace (Этап 4)

- replaceDSNDatabase: возвращает ОШИБКУ для key=value DSN (раньше молча отдавал
  исходный DSN → телеметрия всех тенантов ушла бы в management-БД). Query
  отделяется первым '?', поэтому sslrootcert=/path больше не ломает поиск БД.
- EnsureTenantDB: guard safeNamespaceRe (defense-in-depth, совпадает с API-валидацией);
  небезопасный namespace → ErrInvalidNamespace → в consumer классифицируется как
  постоянная ошибка (DLQ, не вечный ретрай).
This commit is contained in:
Repinoid
2026-10-10 19:13:35 +03:00
parent c3492cb0f1
commit faba28f824
2 changed files with 41 additions and 16 deletions
+5
View File
@@ -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
+36 -16
View File
@@ -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.