diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue index 1f5540798..17ed1f51e 100644 --- a/apps/desktop/src/components/connection/ConnectionDialog.vue +++ b/apps/desktop/src/components/connection/ConnectionDialog.vue @@ -39,6 +39,7 @@ import { prestoSqlBuiltinDriverPaths } from "@/lib/database/prestoSqlBuiltinDriv import { SQLITE_DATABASE_FILE_EXTENSIONS } from "@/lib/database/databaseFileDetection"; import { connectionAttemptOriginalErrorMessage, connectionAttemptTimeoutMessage, connectionAttemptTimeoutMs } from "@/lib/connection/connectionAttemptTimeout"; import { appendConnectionErrorHints } from "@/lib/connection/connectionErrorHints"; +import { normalizeKafkaBootstrapServers } from "@/lib/connection/kafkaBootstrapServers"; import { driverInstallProgressPercent, type DriverInstallProgress } from "@/lib/connection/driverInstallProgressUi"; import { ArrowLeft, ArrowDown, ArrowUp, CheckSquare, ChevronRight, CircleHelp, Copy, ExternalLink, FilePlus2, FolderOpen, GripVertical, Grid3X3, KeyRound, Link2, List, ListFilter, Loader2, Pencil, Pipette, Plus, Search, ShieldCheck, Square, Trash2 } from "@lucide/vue"; import { buildDraftVisibleDatabasesConnectionId, connectionCanChooseVisibleDatabases, initialVisibleDatabaseSelection, visibleDatabaseSelectionIsStale } from "@/lib/connection/connectionVisibleDatabases"; @@ -795,32 +796,6 @@ function requireMqField(value: string, message: string): string { return trimmed; } -function normalizeMqKafkaBootstrapServer(server: string): string { - if (server.includes("://")) { - throw new Error("Kafka bootstrap servers must be host:port values without a URL scheme"); - } - let parsed: URL; - try { - parsed = new URL(`kafka://${server}`); - } catch { - throw new Error("Kafka bootstrap servers are invalid"); - } - if (!parsed.hostname || parsed.username || parsed.password || parsed.search || parsed.hash || (parsed.pathname && parsed.pathname !== "/")) { - throw new Error("Kafka bootstrap servers are invalid"); - } - return server; -} - -function normalizeMqKafkaBootstrapServers(value: string): string { - const servers = requireMqField(value, "Kafka bootstrap servers are required") - .split(",") - .map((server) => server.trim()) - .filter(Boolean) - .map(normalizeMqKafkaBootstrapServer); - if (!servers.length) throw new Error("Kafka bootstrap servers are required"); - return servers.join(","); -} - function buildMqAuth(): MqAuth { switch (mqAuthKind.value) { case "token": @@ -862,7 +837,7 @@ function buildMqTokenSigning() { function buildMqAdminConfig(): MqAdminConfig { const systemKind = mqSystemKind.value; if (systemKind === "kafka") { - const bootstrapServers = normalizeMqKafkaBootstrapServers(mqKafkaBootstrapServers.value); + const bootstrapServers = normalizeKafkaBootstrapServers(mqKafkaBootstrapServers.value); const extra: Record = { bootstrapServers }; const securityProtocol = mqKafkaSecurityProtocol.value === MQ_KAFKA_SECURITY_PROTOCOL_AUTO ? "" : mqKafkaSecurityProtocol.value.trim(); const saslMechanism = mqKafkaSaslMechanism.value.trim(); @@ -1107,7 +1082,7 @@ function applyMqAdminUrl(config: LegacyConnectionConfig, adminUrl: string) { } function applyMqKafkaBootstrapServers(config: LegacyConnectionConfig, bootstrapServers: string, securityProtocol?: string) { - const first = normalizeMqKafkaBootstrapServers(bootstrapServers).split(",")[0]; + const first = normalizeKafkaBootstrapServers(bootstrapServers).split(",")[0]; if (!first) throw new Error("Kafka bootstrap servers are required"); let parsed: URL; try { diff --git a/apps/desktop/src/lib/__tests__/connection/kafkaBootstrapServers.spec.ts b/apps/desktop/src/lib/__tests__/connection/kafkaBootstrapServers.spec.ts new file mode 100644 index 000000000..f102d3276 --- /dev/null +++ b/apps/desktop/src/lib/__tests__/connection/kafkaBootstrapServers.spec.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from "vitest"; +import { normalizeKafkaBootstrapServers } from "@/lib/connection/kafkaBootstrapServers"; + +describe("Kafka bootstrap servers", () => { + it("keeps comma-separated bootstrap servers", () => { + expect(normalizeKafkaBootstrapServers("broker1:9092, broker2:9092")).toBe("broker1:9092,broker2:9092"); + }); + + it("normalizes common cluster address separators to commas", () => { + expect(normalizeKafkaBootstrapServers("broker1:9092;broker2:9092,broker3:9092\nbroker4:9092 broker5:9092")).toBe("broker1:9092,broker2:9092,broker3:9092,broker4:9092,broker5:9092"); + }); + + it("keeps IPv6 bootstrap servers", () => { + expect(normalizeKafkaBootstrapServers("[::1]:9092;[2001:db8::1]:9092")).toBe("[::1]:9092,[2001:db8::1]:9092"); + }); + + it("rejects bootstrap servers with URL schemes", () => { + expect(() => normalizeKafkaBootstrapServers("PLAINTEXT://broker1:9092,broker2:9092")).toThrow("Kafka bootstrap servers must be host:port values without a URL scheme"); + }); + + it("rejects invalid bootstrap server values", () => { + expect(() => normalizeKafkaBootstrapServers("broker1:9092/path,broker2:9092")).toThrow("Kafka bootstrap servers are invalid"); + }); +}); diff --git a/apps/desktop/src/lib/connection/kafkaBootstrapServers.ts b/apps/desktop/src/lib/connection/kafkaBootstrapServers.ts new file mode 100644 index 000000000..1b836aff7 --- /dev/null +++ b/apps/desktop/src/lib/connection/kafkaBootstrapServers.ts @@ -0,0 +1,33 @@ +const KAFKA_BOOTSTRAP_SERVER_SEPARATOR = /[\s,;,;]+/u; + +function requireKafkaBootstrapServers(value: string): string { + const trimmed = value.trim(); + if (!trimmed) throw new Error("Kafka bootstrap servers are required"); + return trimmed; +} + +function normalizeKafkaBootstrapServer(server: string): string { + if (server.includes("://")) { + throw new Error("Kafka bootstrap servers must be host:port values without a URL scheme"); + } + let parsed: URL; + try { + parsed = new URL(`kafka://${server}`); + } catch { + throw new Error("Kafka bootstrap servers are invalid"); + } + if (!parsed.hostname || parsed.username || parsed.password || parsed.search || parsed.hash || (parsed.pathname && parsed.pathname !== "/")) { + throw new Error("Kafka bootstrap servers are invalid"); + } + return server; +} + +export function normalizeKafkaBootstrapServers(value: string): string { + const servers = requireKafkaBootstrapServers(value) + .split(KAFKA_BOOTSTRAP_SERVER_SEPARATOR) + .map((server) => server.trim()) + .filter(Boolean) + .map(normalizeKafkaBootstrapServer); + if (!servers.length) throw new Error("Kafka bootstrap servers are required"); + return servers.join(","); +}