From 67bb4f6ef92d50c16adbdd17447675ad54b0414d Mon Sep 17 00:00:00 2001 From: mptyl Date: Tue, 4 Aug 2026 00:05:33 +0200 Subject: [PATCH] fix: complete workspace diagnostic adapters --- backend/src/workspaces/diagnostics.ts | 63 ++++++++++++++++----- backend/test/workspaces-diagnostics.test.ts | 62 ++++++++++++++++++++ 2 files changed, 111 insertions(+), 14 deletions(-) diff --git a/backend/src/workspaces/diagnostics.ts b/backend/src/workspaces/diagnostics.ts index bf3da193..305a907f 100644 --- a/backend/src/workspaces/diagnostics.ts +++ b/backend/src/workspaces/diagnostics.ts @@ -1,7 +1,9 @@ import { randomUUID } from "node:crypto"; import { readFile, realpath } from "node:fs/promises"; import { createConnection } from "node:net"; +import { createServer } from "node:net"; import { once } from "node:events"; +import { spawn } from "node:child_process"; import { Client } from "pg"; import { MAX_WORKSPACE_DIAGNOSTIC_TIMEOUT_MS } from "../config.js"; import { buildInstallationContract } from "./contracts.js"; @@ -127,6 +129,7 @@ export interface WriteDiagnosticRecordRequest { timeoutMs: number; signal: AbortSignal; credentialFile?: string; + tlsCaFile?: string; baseUrl?: string; diagnostic?: RestDiagnosticRequest; } @@ -157,6 +160,8 @@ export interface ConcreteDiagnosticAdapterDependencies { directProtocol?: DirectProtocolFactory; sshProcess?: SshProcessFactory; databaseClient?: DatabaseDiagnosticClientFactory; + sshSpawn?: (args: readonly string[]) => { kill(signal?: NodeJS.Signals): boolean }; + reserveLoopbackPort?: () => Promise; } /** @@ -211,6 +216,21 @@ async function restHeaders( return diagnostic.auth === "bearer" ? { authorization: `Bearer ${secret}` } : { "x-api-key": secret }; } +async function reserveLoopbackPort(): Promise { + const server = createServer(); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + try { + const address = server.address(); + if (!address || typeof address === "string") throw new Error("SSH tunnel port unavailable"); + return address.port; + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +} + /** * Concrete production adapters deliberately retain only probe metadata. Protocol failures and * response bodies are discarded at this boundary; callers receive fixed diagnostics instead. @@ -218,6 +238,22 @@ async function restHeaders( export function createConcreteDiagnosticAdapters( dependencies: ConcreteDiagnosticAdapterDependencies = {}, ): DiagnosticAdapters { + const spawnSsh = dependencies.sshSpawn ?? ((args: readonly string[]) => spawn("ssh", [...args], { stdio: "ignore" })); + const reserveSshPort = dependencies.reserveLoopbackPort ?? reserveLoopbackPort; + const sshProcess = dependencies.sshProcess ?? { + async start(request: SshTunnelRequest, args: readonly string[]) { + const port = await reserveSshPort(); + const resolvedArgs = args.map((argument) => argument === `127.0.0.1:0:${request.targetHost}:${request.targetPort}` + ? `127.0.0.1:${port}:${request.targetHost}:${request.targetPort}` : argument); + const child = spawnSsh(resolvedArgs); + const abort = () => { child.kill("SIGTERM"); }; + request.signal.addEventListener("abort", abort, { once: true }); + return { + tunnel: { host: "127.0.0.1" as const, port }, + async close() { request.signal.removeEventListener("abort", abort); child.kill("SIGTERM"); }, + }; + }, + }; const databaseClient = dependencies.databaseClient ?? { async connect(request: { host: string; port: number; database: string; user: string; credentialFile: string; tlsCaFile: string; signal: AbortSignal }) { const client = new Client({ @@ -295,14 +331,13 @@ export function createConcreteDiagnosticAdapters( if (!request.knownHostsFile || !(await secretPresent(request.privateKeyFile))) { throw new Error("SSH probe failed"); } - if (!dependencies.sshProcess) throw new Error("SSH tunnel process is unavailable"); const args = [ "-N", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=yes", "-o", `UserKnownHostsFile=${request.knownHostsFile}`, "-i", request.privateKeyFile, "-p", String(request.sshPort), "-L", `127.0.0.1:0:${request.targetHost}:${request.targetPort}`, `${request.sshUser}@${request.sshHost}`, ]; - const tunnel = await dependencies.sshProcess.start(request, args); + const tunnel = await sshProcess.start(request, args); try { return await probe(tunnel.tunnel); } finally { @@ -322,7 +357,7 @@ export function createConcreteDiagnosticAdapters( }); try { const metadata = await client.query( - "SELECT a.atttypmod - 4 AS dimensions, CASE WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_cosine_ops%' THEN 'cosine' WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_l2_ops%' THEN 'l2' WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_ip_ops%' THEN 'inner_product' END AS distance FROM pg_attribute a JOIN pg_class c ON c.oid = a.attrelid JOIN pg_namespace n ON n.oid = c.relnamespace LEFT JOIN pg_index i ON i.indrelid = c.oid WHERE n.nspname = $1 AND c.relname = $2 AND a.attnum > 0 AND NOT a.attisdropped AND a.atttypid = (SELECT oid FROM pg_type WHERE typname = 'vector') LIMIT 1", + "SELECT a.atttypmod - 4 AS dimensions, CASE WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_cosine_ops%' THEN 'cosine' WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_l2_ops%' THEN 'l2' WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_ip_ops%' THEN 'inner_product' END AS distance FROM pg_attribute a JOIN pg_class c ON c.oid = a.attrelid JOIN pg_namespace n ON n.oid = c.relnamespace JOIN pg_index i ON i.indrelid = c.oid AND a.attnum = ANY(i.indkey) WHERE n.nspname = $1 AND c.relname = $2 AND a.attnum > 0 AND NOT a.attisdropped AND a.atttypid = (SELECT oid FROM pg_type WHERE typname = 'vector') ORDER BY i.indexrelid LIMIT 1", [resource.schema, request.collection], ); const row = metadata.rows[0]; @@ -332,11 +367,12 @@ export function createConcreteDiagnosticAdapters( await client.end().catch(() => undefined); } } - if (request.transport !== "rest_api" || !request.baseUrl || !request.credentialFile || !request.diagnostic - || !(await secretPresent(request.credentialFile))) throw new Error("vector metadata adapter is unavailable"); + if (request.transport !== "rest_api" || !request.baseUrl || !request.diagnostic || request.tlsCaFile) { + throw new Error("vector metadata adapter is unavailable"); + } const response = await fetch(resolveDiagnosticUrl(request.baseUrl, request.diagnostic.path).toString(), { method: request.diagnostic.method, - headers: { authorization: `Bearer ${(await readFile(request.credentialFile, "utf8")).trim()}` }, + headers: await restHeaders(request.diagnostic, request.credentialFile), signal: request.signal, redirect: "error", }); @@ -353,10 +389,10 @@ export function createConcreteDiagnosticAdapters( }; }, async probeEmbedding(request) { - if (!request.diagnostic || !(await secretPresent(request.credentialFile ?? ""))) throw new Error("embedding probe failed"); + if (!request.diagnostic || request.tlsCaFile) throw new Error("embedding probe failed"); const response = await fetch(resolveDiagnosticUrl(request.baseUrl, request.diagnostic.path).toString(), { method: request.diagnostic.method, - headers: { authorization: `Bearer ${await readFile(request.credentialFile!, "utf8")}` }, + headers: await restHeaders(request.diagnostic, request.credentialFile), signal: request.signal, redirect: "error", }); @@ -366,12 +402,11 @@ export function createConcreteDiagnosticAdapters( return { available: true, dimensions: payload[request.diagnostic.response.dimensions] as number }; }, async writeDiagnosticRecord(request) { - if (!request.baseUrl || !request.credentialFile || !request.diagnostic - || !(await secretPresent(request.credentialFile))) throw new Error("vector write adapter is unavailable"); + if (!request.baseUrl || !request.diagnostic || request.tlsCaFile) throw new Error("vector write adapter is unavailable"); const response = await fetch(resolveDiagnosticUrl(request.baseUrl, request.diagnostic.path).toString(), { method: request.diagnostic.method, headers: { - authorization: `Bearer ${(await readFile(request.credentialFile, "utf8")).trim()}`, + ...await restHeaders(request.diagnostic, request.credentialFile), "content-type": "application/json", }, body: JSON.stringify({ operation: "create", id: request.id, collection: request.collection, dimensions: request.dimensions }), @@ -381,12 +416,11 @@ export function createConcreteDiagnosticAdapters( if (!response.ok) throw new Error("vector write adapter is unavailable"); }, async removeDiagnosticRecord(request) { - if (!request.baseUrl || !request.credentialFile || !request.diagnostic - || !(await secretPresent(request.credentialFile))) throw new Error("vector write adapter is unavailable"); + if (!request.baseUrl || !request.diagnostic || request.tlsCaFile) throw new Error("vector write adapter is unavailable"); const response = await fetch(resolveDiagnosticUrl(request.baseUrl, request.diagnostic.path).toString(), { method: request.diagnostic.method, headers: { - authorization: `Bearer ${(await readFile(request.credentialFile, "utf8")).trim()}`, + ...await restHeaders(request.diagnostic, request.credentialFile), "content-type": "application/json", }, body: JSON.stringify({ operation: "remove", id: request.id, collection: request.collection }), @@ -728,6 +762,7 @@ export function createWorkspaceDiagnoser( timeoutMs: vectorTimeout, signal: new AbortController().signal, credentialFile, + tlsCaFile: bindings.vector.values[bindingName(canonical, "VECTOR", "TLS_CA_FILE")], baseUrl: bindings.vector.values[bindingName(canonical, "VECTOR", "BASE_URL")], diagnostic: canonical.diagnostics.vector_rest.reversible_probe, }; diff --git a/backend/test/workspaces-diagnostics.test.ts b/backend/test/workspaces-diagnostics.test.ts index 8750c5fb..5d46c6a3 100644 --- a/backend/test/workspaces-diagnostics.test.ts +++ b/backend/test/workspaces-diagnostics.test.ts @@ -326,6 +326,24 @@ test("uses strict known-host SSH arguments and always closes the temporary tunne } }); +test("provides a default bounded SSH factory through injected spawn and loopback allocation", async () => { + const directory = await mkdtemp(join(tmpdir(), "thothii-diagnostic-")); + const privateKeyFile = join(directory, "ssh-key"); + await writeFile(privateKeyFile, "test-key\n", { mode: 0o600 }); + const kill = vi.fn(() => true); + const sshSpawn = vi.fn(() => ({ kill })); + try { + const adapter = createConcreteDiagnosticAdapters({ sshSpawn, reserveLoopbackPort: async () => 45432 } as any); + await adapter.withSshTunnel({ sshHost: "bastion.example.test", sshPort: 22, sshUser: "tunnel", privateKeyFile, knownHostsFile: "/run/secrets/known-hosts", targetHost: "dwh.internal", targetPort: 5432, localHost: "127.0.0.1", localPort: 0, timeoutMs: 5000, signal: new AbortController().signal }, async () => undefined); + expect(sshSpawn).toHaveBeenCalledWith(expect.arrayContaining([ + "StrictHostKeyChecking=yes", "UserKnownHostsFile=/run/secrets/known-hosts", "-L", "127.0.0.1:45432:dwh.internal:5432", + ])); + expect(kill).toHaveBeenCalledWith("SIGTERM"); + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + test("requires a matching embedding model vector and removes its unique write probe", async () => { const adapters = successfulAdapters(); @@ -454,6 +472,25 @@ test("requires an authenticated TLS database query before direct diagnostics suc } }); +test("selects the vector index containing the declared vector column for direct metadata", async () => { + const directory = await mkdtemp(join(tmpdir(), "thothii-diagnostic-")); + const passwordFile = join(directory, "password"); + const caFile = join(directory, "ca.pem"); + await Promise.all([writeFile(passwordFile, "password\n"), writeFile(caFile, "test-ca\n")]); + const query = vi.fn(async () => ({ rows: [{ dimensions: 768, distance: "cosine" }] })); + const connect = vi.fn(async () => ({ query, end: vi.fn(async () => undefined) })); + try { + const result = await createConcreteDiagnosticAdapters({ databaseClient: { connect } } as any).inspectVector({ + transport: "pgvector_direct", host: "127.0.0.1", port: 5432, user: "reader", credentialFile: passwordFile, tlsCaFile: caFile, + resource: { database: "postgres", schema: "vectors" }, collection: "clinical_documents", timeoutMs: 5000, signal: new AbortController().signal, + }); + expect(query).toHaveBeenCalledWith(expect.stringContaining("a.attnum = ANY(i.indkey)"), ["vectors", "clinical_documents"]); + expect(result).toMatchObject({ dimensions: 768, distance: "cosine" }); + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + test("honors a declared unauthenticated REST diagnostic without reading a credential", async () => { const fetchSpy = vi.fn(async () => new Response(JSON.stringify({ database: "warehouse", schema: "datawarehouse" }), { status: 200, headers: { "content-type": "application/json" }, @@ -472,6 +509,31 @@ test("honors a declared unauthenticated REST diagnostic without reading a creden } }); +test("applies declared auth modes and rejects private CA files across vector REST paths", async () => { + const directory = await mkdtemp(join(tmpdir(), "thothii-diagnostic-")); + const keyFile = join(directory, "api-key"); + const caFile = join(directory, "ca.pem"); + await Promise.all([writeFile(keyFile, "writer-key\n", { mode: 0o600 }), writeFile(caFile, "private-ca\n")]); + const fetchSpy = vi.fn(async () => new Response(JSON.stringify({ collection: "clinical_documents", dimensions: 768, distance: "cosine", model: "embed" }), { status: 200, headers: { "content-type": "application/json" } })); + vi.stubGlobal("fetch", fetchSpy); + const adapter = createConcreteDiagnosticAdapters(); + const signal = new AbortController().signal; + try { + await adapter.inspectVector({ transport: "rest_api", baseUrl: "https://vector.example.test", collection: "clinical_documents", timeoutMs: 1, signal, diagnostic: { method: "GET", path: "/metadata", auth: "none", response: { collection: "collection", dimensions: "dimensions", distance: "distance" } } }); + await adapter.probeEmbedding({ baseUrl: "https://embed.example.test", model: "embed", timeoutMs: 1, signal, credentialFile: keyFile, diagnostic: { method: "POST", path: "/embed", auth: "x-api-key", response: { model: "model", dimensions: "dimensions" } } }); + await adapter.writeDiagnosticRecord({ baseUrl: "https://vector.example.test", credentialFile: keyFile, collection: "clinical_documents", dimensions: 768, id: "diagnostic:test", timeoutMs: 1, signal, diagnostic: { method: "POST", path: "/probe", auth: "none" } }); + expect(fetchSpy.mock.calls[0]?.[1]).toMatchObject({ headers: {} }); + expect(fetchSpy.mock.calls[1]?.[1]).toMatchObject({ headers: { "x-api-key": "writer-key" } }); + expect(fetchSpy.mock.calls[2]?.[1]).toMatchObject({ headers: expect.not.objectContaining({ authorization: expect.anything() }) }); + await expect(adapter.inspectVector({ transport: "rest_api", baseUrl: "https://vector.example.test", credentialFile: keyFile, tlsCaFile: caFile, collection: "clinical_documents", timeoutMs: 1, signal, diagnostic: { method: "GET", path: "/metadata", auth: "bearer", response: { collection: "collection", dimensions: "dimensions", distance: "distance" } } })).rejects.toThrow("vector metadata adapter is unavailable"); + await expect(adapter.probeEmbedding({ baseUrl: "https://embed.example.test", credentialFile: keyFile, tlsCaFile: caFile, model: "embed", timeoutMs: 1, signal, diagnostic: { method: "POST", path: "/embed", auth: "bearer", response: { model: "model", dimensions: "dimensions" } } })).rejects.toThrow("embedding probe failed"); + await expect(adapter.removeDiagnosticRecord({ baseUrl: "https://vector.example.test", credentialFile: keyFile, tlsCaFile: caFile, collection: "clinical_documents", id: "diagnostic:test", dimensions: 768, timeoutMs: 1, signal, diagnostic: { method: "POST", path: "/probe", auth: "bearer" } })).rejects.toThrow("vector write adapter is unavailable"); + } finally { + vi.unstubAllGlobals(); + await rm(directory, { recursive: true, force: true }); + } +}); + test("constructs the production diagnoser with the configured timeout and injected adapters", async () => { const adapters = successfulAdapters();