diff --git a/packages/nodes-base/nodes/Postgres/test/transport/ssh-tunnel.test.ts b/packages/nodes-base/nodes/Postgres/test/transport/ssh-tunnel.test.ts new file mode 100644 index 00000000000..f0dacc4dfcf --- /dev/null +++ b/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>; + +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({ + logger: mock(), + helpers: mock({ + getSSHClient: vi.fn().mockResolvedValue(sshClient), + updateLastUsed: vi.fn(), + }), + }); + +describe('configurePostgres with an SSH tunnel', () => { + afterEach(() => { + ConnectionPoolManager.getInstance(mock()).purgeConnections(); + }); + + it('should reject the query and drop the pool when forwardOut throws', async () => { + const sshClient = mock(); + 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.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(); + }); +}); diff --git a/packages/nodes-base/nodes/Postgres/transport/index.ts b/packages/nodes-base/nodes/Postgres/transport/index.ts index 7dd2b504aad..28295eab17c 100644 --- a/packages/nodes-base/nodes/Postgres/transport/index.ts +++ b/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({