fix(workspaces): align postgres connection diagnostics
This commit is contained in:
@@ -129,8 +129,14 @@ export function createAuthDiagnoser(deps: AuthDiagnoserDependencies): AuthDiagno
|
|||||||
: { ready: false, mode: deps.authMode, checks: result };
|
: { ready: false, mode: deps.authMode, checks: result };
|
||||||
}
|
}
|
||||||
if (deps.authMode === "upstream") {
|
if (deps.authMode === "upstream") {
|
||||||
checks.push(check("auth_config_incomplete", "The deprecated upstream authentication mode is not certifiable."));
|
const result = ordered(checks);
|
||||||
return { ready: false, mode: deps.authMode, checks: ordered(checks) };
|
return result.length === 0
|
||||||
|
? {
|
||||||
|
ready: true,
|
||||||
|
mode: "upstream",
|
||||||
|
checks: [{ level: "info", code: "auth_ready", message: "Authentication is ready." }],
|
||||||
|
}
|
||||||
|
: { ready: false, mode: "upstream", checks: result };
|
||||||
}
|
}
|
||||||
|
|
||||||
let loaded;
|
let loaded;
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
import { readFile } from "node:fs/promises";
|
import { readFile } from "node:fs/promises";
|
||||||
import { Client } from "pg";
|
import { Client, type ClientConfig } from "pg";
|
||||||
import { MAX_WORKSPACE_DIAGNOSTIC_TIMEOUT_MS } from "../config.js";
|
import { MAX_WORKSPACE_DIAGNOSTIC_TIMEOUT_MS } from "../config.js";
|
||||||
import { buildInstallationContract } from "./contracts.js";
|
import { buildInstallationContract } from "./contracts.js";
|
||||||
import type { RuntimeBindings } from "./runtime-renderer.js";
|
import type { RuntimeBindings } from "./runtime-renderer.js";
|
||||||
@@ -111,9 +111,16 @@ export interface DatabaseDiagnosticClientFactory {
|
|||||||
}): Promise<DatabaseDiagnosticClient>;
|
}): Promise<DatabaseDiagnosticClient>;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface PostgreSqlDiagnosticWireClient {
|
||||||
|
connect(): Promise<void>;
|
||||||
|
query(sql: string, values: readonly unknown[]): Promise<{ rows: Array<Record<string, unknown>> }>;
|
||||||
|
end(): Promise<void>;
|
||||||
|
}
|
||||||
|
|
||||||
export interface ConcreteDiagnosticAdapterDependencies {
|
export interface ConcreteDiagnosticAdapterDependencies {
|
||||||
directProtocol?: DirectProtocolFactory;
|
directProtocol?: DirectProtocolFactory;
|
||||||
databaseClient?: DatabaseDiagnosticClientFactory;
|
databaseClient?: DatabaseDiagnosticClientFactory;
|
||||||
|
createPostgresClient?: (config: ClientConfig) => PostgreSqlDiagnosticWireClient;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Adapters retain only diagnostic metadata and never return credential contents or bodies. */
|
/** Adapters retain only diagnostic metadata and never return credential contents or bodies. */
|
||||||
@@ -146,6 +153,15 @@ async function restHeaders(
|
|||||||
export function createConcreteDiagnosticAdapters(
|
export function createConcreteDiagnosticAdapters(
|
||||||
dependencies: ConcreteDiagnosticAdapterDependencies = {},
|
dependencies: ConcreteDiagnosticAdapterDependencies = {},
|
||||||
): DiagnosticAdapters {
|
): DiagnosticAdapters {
|
||||||
|
const createPostgresClient = dependencies.createPostgresClient
|
||||||
|
?? ((config: ClientConfig): PostgreSqlDiagnosticWireClient => {
|
||||||
|
const client = new Client(config);
|
||||||
|
return {
|
||||||
|
connect: async () => { await client.connect(); },
|
||||||
|
query: async (sql, values) => await client.query(sql, [...values]),
|
||||||
|
end: async () => await client.end(),
|
||||||
|
};
|
||||||
|
});
|
||||||
const databaseClient = dependencies.databaseClient ?? {
|
const databaseClient = dependencies.databaseClient ?? {
|
||||||
async connect(request: {
|
async connect(request: {
|
||||||
host: string;
|
host: string;
|
||||||
@@ -157,17 +173,21 @@ export function createConcreteDiagnosticAdapters(
|
|||||||
tlsServername?: string;
|
tlsServername?: string;
|
||||||
signal: AbortSignal;
|
signal: AbortSignal;
|
||||||
}) {
|
}) {
|
||||||
const client = new Client({
|
const tlsConfigured = request.tlsCaFile !== undefined || request.tlsServername !== undefined;
|
||||||
|
const ssl: ClientConfig["ssl"] = tlsConfigured
|
||||||
|
? {
|
||||||
|
...(request.tlsCaFile ? { ca: await readFile(request.tlsCaFile, "utf8") } : {}),
|
||||||
|
...(request.tlsServername ? { servername: request.tlsServername } : {}),
|
||||||
|
rejectUnauthorized: true,
|
||||||
|
}
|
||||||
|
: false;
|
||||||
|
const client = createPostgresClient({
|
||||||
host: request.host,
|
host: request.host,
|
||||||
port: request.port,
|
port: request.port,
|
||||||
database: request.database,
|
database: request.database,
|
||||||
user: request.user,
|
user: request.user,
|
||||||
password: (await readFile(request.credentialFile, "utf8")).trim(),
|
password: (await readFile(request.credentialFile, "utf8")).trim(),
|
||||||
ssl: {
|
ssl,
|
||||||
...(request.tlsCaFile ? { ca: await readFile(request.tlsCaFile, "utf8") } : {}),
|
|
||||||
...(request.tlsServername ? { servername: request.tlsServername } : {}),
|
|
||||||
rejectUnauthorized: true,
|
|
||||||
},
|
|
||||||
connectionTimeoutMillis: 5_000,
|
connectionTimeoutMillis: 5_000,
|
||||||
});
|
});
|
||||||
const abort = () => { void client.end(); };
|
const abort = () => { void client.end(); };
|
||||||
@@ -209,8 +229,18 @@ export function createConcreteDiagnosticAdapters(
|
|||||||
});
|
});
|
||||||
try {
|
try {
|
||||||
const result = await client.query(
|
const result = await client.query(
|
||||||
"SELECT current_database() AS database, current_schema() AS schema",
|
`SELECT
|
||||||
[],
|
current_database() AS database,
|
||||||
|
CASE
|
||||||
|
WHEN pg_catalog.has_schema_privilege(
|
||||||
|
current_user,
|
||||||
|
(SELECT oid FROM pg_catalog.pg_namespace WHERE nspname = $1),
|
||||||
|
'USAGE'
|
||||||
|
)
|
||||||
|
THEN $1
|
||||||
|
ELSE NULL
|
||||||
|
END AS schema`,
|
||||||
|
[schema],
|
||||||
);
|
);
|
||||||
const row = result.rows[0];
|
const row = result.rows[0];
|
||||||
if (row?.database !== database || row.schema !== schema) throw new Error("direct probe failed");
|
if (row?.database !== database || row.schema !== schema) throw new Error("direct probe failed");
|
||||||
|
|||||||
@@ -60,6 +60,40 @@ function registryYaml(role: "user" | "admin", passwordHash = validPasswordHash):
|
|||||||
].join("\n");
|
].join("\n");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
test("reports upstream authentication ready when its protected session root is valid", async () => {
|
||||||
|
const report = await createAuthDiagnoser({
|
||||||
|
authMode: "upstream",
|
||||||
|
authStateRoot: "/safe/auth-state",
|
||||||
|
sessionRootValidator: acceptSessionRoot,
|
||||||
|
}).inspect({ live: true });
|
||||||
|
|
||||||
|
expect(report).toEqual({
|
||||||
|
ready: true,
|
||||||
|
mode: "upstream",
|
||||||
|
checks: [{ level: "info", code: "auth_ready", message: "Authentication is ready." }],
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
test("upstream authentication still fails when its protected session root is invalid", async () => {
|
||||||
|
const sentinel = "synthetic-upstream-session-root-secret";
|
||||||
|
const report = await createAuthDiagnoser({
|
||||||
|
authMode: "upstream",
|
||||||
|
authStateRoot: "/safe/auth-state",
|
||||||
|
sessionRootValidator: async () => { throw new Error(sentinel); },
|
||||||
|
}).inspect({ live: true });
|
||||||
|
|
||||||
|
expect(report).toEqual({
|
||||||
|
ready: false,
|
||||||
|
mode: "upstream",
|
||||||
|
checks: [{
|
||||||
|
level: "error",
|
||||||
|
code: "auth_session_store_invalid",
|
||||||
|
message: "The authentication session store is invalid.",
|
||||||
|
}],
|
||||||
|
});
|
||||||
|
expect(JSON.stringify(report)).not.toContain(sentinel);
|
||||||
|
});
|
||||||
|
|
||||||
test("reports deterministic live OIDC checks and silently ignores unrelated groups", async () => {
|
test("reports deterministic live OIDC checks and silently ignores unrelated groups", async () => {
|
||||||
const oidcDiagnose = vi.fn(async () => undefined);
|
const oidcDiagnose = vi.fn(async () => undefined);
|
||||||
const fetch = vi.fn<typeof globalThis.fetch>(async (input) => {
|
const fetch = vi.fn<typeof globalThis.fetch>(async (input) => {
|
||||||
|
|||||||
@@ -205,8 +205,9 @@ test("concrete DWH direct diagnostics authenticate, verify resource identity, an
|
|||||||
const passwordFile = join(root, "password");
|
const passwordFile = join(root, "password");
|
||||||
await writeFile(passwordFile, "password-value");
|
await writeFile(passwordFile, "password-value");
|
||||||
const end = vi.fn(async () => undefined);
|
const end = vi.fn(async () => undefined);
|
||||||
|
const query = vi.fn(async () => ({ rows: [{ database: "warehouse", schema: "datawarehouse" }] }));
|
||||||
const connect = vi.fn(async () => ({
|
const connect = vi.fn(async () => ({
|
||||||
query: vi.fn(async () => ({ rows: [{ database: "warehouse", schema: "datawarehouse" }] })),
|
query,
|
||||||
end,
|
end,
|
||||||
}));
|
}));
|
||||||
try {
|
try {
|
||||||
@@ -219,6 +220,88 @@ test("concrete DWH direct diagnostics authenticate, verify resource identity, an
|
|||||||
});
|
});
|
||||||
expect(result).toMatchObject({ resolved: true, authenticated: true, tlsVerified: true });
|
expect(result).toMatchObject({ resolved: true, authenticated: true, tlsVerified: true });
|
||||||
expect(connect).toHaveBeenCalledWith(expect.objectContaining({ credentialFile: passwordFile }));
|
expect(connect).toHaveBeenCalledWith(expect.objectContaining({ credentialFile: passwordFile }));
|
||||||
|
expect(query).toHaveBeenCalledWith(
|
||||||
|
expect.stringContaining("pg_catalog.has_schema_privilege"),
|
||||||
|
["datawarehouse"],
|
||||||
|
);
|
||||||
|
expect(query.mock.calls[0]?.[0]).toContain("$1");
|
||||||
|
expect(query.mock.calls[0]?.[0]).not.toContain('"datawarehouse"');
|
||||||
|
expect(end).toHaveBeenCalledOnce();
|
||||||
|
} finally {
|
||||||
|
await rm(root, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("concrete DWH direct diagnostics disable TLS when no TLS binding is declared", async () => {
|
||||||
|
const root = await mkdtemp(join(tmpdir(), "thoth-diagnostic-plain-"));
|
||||||
|
const passwordFile = join(root, "password");
|
||||||
|
await writeFile(passwordFile, "password-value");
|
||||||
|
const connect = vi.fn(async () => undefined);
|
||||||
|
const query = vi.fn(async () => ({ rows: [{ database: "warehouse", schema: "datawarehouse" }] }));
|
||||||
|
const end = vi.fn(async () => undefined);
|
||||||
|
const createPostgresClient = vi.fn(() => ({ connect, query, end }));
|
||||||
|
try {
|
||||||
|
const adapter = createConcreteDiagnosticAdapters({ createPostgresClient });
|
||||||
|
await adapter.probeConnector({
|
||||||
|
role: "dwh", transport: "postgres_direct", host: "127.0.0.1", port: 5432,
|
||||||
|
user: "reader", credentialFile: passwordFile,
|
||||||
|
resource: { database: "warehouse", schema: "datawarehouse" },
|
||||||
|
timeoutMs: 1_000, signal: new AbortController().signal,
|
||||||
|
});
|
||||||
|
expect(createPostgresClient).toHaveBeenCalledWith(expect.objectContaining({ ssl: false }));
|
||||||
|
expect(connect).toHaveBeenCalledOnce();
|
||||||
|
expect(end).toHaveBeenCalledOnce();
|
||||||
|
} finally {
|
||||||
|
await rm(root, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("concrete DWH direct diagnostics use strict TLS only for explicit TLS bindings", async () => {
|
||||||
|
const root = await mkdtemp(join(tmpdir(), "thoth-diagnostic-tls-"));
|
||||||
|
const passwordFile = join(root, "password");
|
||||||
|
const caFile = join(root, "ca.pem");
|
||||||
|
await writeFile(passwordFile, "password-value");
|
||||||
|
await writeFile(caFile, "test-ca");
|
||||||
|
const connect = vi.fn(async () => undefined);
|
||||||
|
const query = vi.fn(async () => ({ rows: [{ database: "warehouse", schema: "datawarehouse" }] }));
|
||||||
|
const end = vi.fn(async () => undefined);
|
||||||
|
const createPostgresClient = vi.fn(() => ({ connect, query, end }));
|
||||||
|
try {
|
||||||
|
const adapter = createConcreteDiagnosticAdapters({ createPostgresClient });
|
||||||
|
await adapter.probeConnector({
|
||||||
|
role: "dwh", transport: "postgres_direct", host: "dwh.example.test", port: 5432,
|
||||||
|
user: "reader", credentialFile: passwordFile, tlsCaFile: caFile,
|
||||||
|
tlsServername: "dwh.example.test",
|
||||||
|
resource: { database: "warehouse", schema: "datawarehouse" },
|
||||||
|
timeoutMs: 1_000, signal: new AbortController().signal,
|
||||||
|
});
|
||||||
|
expect(createPostgresClient).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
|
ssl: { ca: "test-ca", servername: "dwh.example.test", rejectUnauthorized: true },
|
||||||
|
}));
|
||||||
|
expect(connect).toHaveBeenCalledOnce();
|
||||||
|
expect(end).toHaveBeenCalledOnce();
|
||||||
|
} finally {
|
||||||
|
await rm(root, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("concrete DWH direct diagnostics fail closed when the declared schema is inaccessible", async () => {
|
||||||
|
const root = await mkdtemp(join(tmpdir(), "thoth-diagnostic-schema-"));
|
||||||
|
const passwordFile = join(root, "password");
|
||||||
|
await writeFile(passwordFile, "CANARY-DATABASE-SECRET");
|
||||||
|
const end = vi.fn(async () => undefined);
|
||||||
|
const connect = vi.fn(async () => ({
|
||||||
|
query: vi.fn(async () => ({ rows: [{ database: "warehouse", schema: null }] })),
|
||||||
|
end,
|
||||||
|
}));
|
||||||
|
try {
|
||||||
|
const adapter = createConcreteDiagnosticAdapters({ databaseClient: { connect } });
|
||||||
|
await expect(adapter.probeConnector({
|
||||||
|
role: "dwh", transport: "postgres_direct", host: "127.0.0.1", port: 5432,
|
||||||
|
user: "reader", credentialFile: passwordFile,
|
||||||
|
resource: { database: "warehouse", schema: "datawarehouse" },
|
||||||
|
timeoutMs: 1_000, signal: new AbortController().signal,
|
||||||
|
})).rejects.toThrow("direct probe failed");
|
||||||
expect(end).toHaveBeenCalledOnce();
|
expect(end).toHaveBeenCalledOnce();
|
||||||
} finally {
|
} finally {
|
||||||
await rm(root, { recursive: true, force: true });
|
await rm(root, { recursive: true, force: true });
|
||||||
|
|||||||
Reference in New Issue
Block a user