Merge 4547a731f7 into ec074519d9

packages/nodes-base/nodes/Postgres/test/transport/ssh-tunnel.test.ts

@@ -0,0 +1,69 @@
+import type { IExecuteFunctions, Logger } from 'n8n-workflow';
+import { mock } from 'vitest-mock-extended';
+
+import { ConnectionPoolManager } from '@utils/connection-pool-manager';
+
+import { configurePostgres } from '../../transport';
+import type { PostgresNodeCredentials } from '../../v2/helpers/interfaces';
+
+type SSHClient = Awaited<ReturnType<IExecuteFunctions['helpers']['getSSHClient']>>;
+
+const credentials = (): PostgresNodeCredentials => ({
+	host: 'db.example.com',
+	port: 5432,
+	database: 'postgres',
+	user: 'postgres',
+	password: 'password',
+	maxConnections: 10,
+	ssl: 'disable',
+	sshTunnel: true,
+	sshAuthenticateWith: 'password',
+	sshHost: 'bastion.example.com',
+	sshPort: 22,
+	sshUser: 'tunnel',
+	sshPassword: 'password',
+});
+
+const createContext = (sshClient: SSHClient) =>
+	mock<IExecuteFunctions>({
+		logger: mock<Logger>(),
+		helpers: mock<IExecuteFunctions['helpers']>({
+			getSSHClient: vi.fn().mockResolvedValue(sshClient),
+			updateLastUsed: vi.fn(),
+		}),
+	});
+
+describe('configurePostgres with an SSH tunnel', () => {
+	afterEach(() => {
+		ConnectionPoolManager.getInstance(mock<Logger>()).purgeConnections();
+	});
+
+	it('should reject the query and drop the pool when forwardOut throws', async () => {
+		const sshClient = mock<SSHClient>();
+		sshClient.forwardOut.mockImplementation(() => {
+			throw new Error('Not connected');
+		});
+		const context = createContext(sshClient);
+
+		const { db } = await configurePostgres.call(context, credentials(), { nodeVersion: 2.6 });
+
+		await expect(db.one('SELECT 1')).rejects.toThrow();
+
+		// The pool was dropped, so the next call builds a new one with a fresh SSH client
+		await configurePostgres.call(context, credentials(), { nodeVersion: 2.6 });
+		expect(context.helpers.getSSHClient).toHaveBeenCalledTimes(2);
+	});
+
+	it('should reject the query when the SSH server refuses the forward', async () => {
+		const sshClient = mock<SSHClient>();
+		sshClient.forwardOut.mockImplementation((_srcIP, _srcPort, _dstIP, _dstPort, callback) => {
+			callback?.(new Error('(SSH) Channel open failure: open failed'), undefined as never);
+			return sshClient;
+		});
+		const context = createContext(sshClient);
+
+		const { db } = await configurePostgres.call(context, credentials(), { nodeVersion: 2.6 });
+
+		await expect(db.one('SELECT 1')).rejects.toThrow();
+	});
+});

packages/nodes-base/nodes/Postgres/transport/index.ts

@@ -160,21 +160,34 @@ export async function configurePostgres(
 			});
 
 			proxy.on('connection', (localSocket) => {
-				sshClient.forwardOut(
-					LOCALHOST,
-					localSocket.remotePort!,
-					credentials.host,
-					credentials.port,
-					(error, clientChannel) => {
-						if (error) {
-							this.logger.error('SSH Client: Port forwarding encountered an error', { error });
-							abortController.abort();
-						} else {
-							localSocket.pipe(clientChannel);
-							clientChannel.pipe(localSocket);
-						}
-					},
-				);
+				// Closing the proxy server does not close sockets it already accepted, so
+				// destroy this one too. Otherwise pg waits on it forever.
+				const onForwardError = (error: unknown) => {
+					this.logger.error('SSH Client: Port forwarding encountered an error', { error });
+					localSocket.destroy();
+					abortController.abort();
+				};
+
+				try {
+					sshClient.forwardOut(
+						LOCALHOST,
+						localSocket.remotePort!,
+						credentials.host,
+						credentials.port,
+						(error, clientChannel) => {
+							if (error) {
+								onForwardError(error);
+							} else {
+								localSocket.pipe(clientChannel);
+								clientChannel.pipe(localSocket);
+							}
+						},
+					);
+				} catch (error) {
+					// `forwardOut` throws synchronously once the SSH client has been closed,
+					// e.g. when another pool sharing it was cleaned up.
+					onForwardError(error);
+				}
 			});
 
 			const db = pgp({