From a3e2ceec90aa64995528eb818b6f9c01e08ddce2 Mon Sep 17 00:00:00 2001 From: m4air Date: Thu, 1 Oct 2026 08:45:58 -0700 Subject: [PATCH 1/3] feat(ssh): wire SshConnection through the work and transport close ledgers (#16741 T2 P2) Every operation SshConnection admits (exec, shell, sftp, file transfers, upload sessions, forwarded channels and sockets, system-SSH commands) now runs through the connection's work ledger, and every ssh2 client and proxy process it allocates is tracked until it physically closes. Ordinary connect, reconnect and disconnect behavior is unchanged. Adds: - subscribeTransportClosure: one-shot notice once the connection is disposed, every allocated transport has emitted 'close' and tracked work has drained. System-SSH startup is never proven closed from here. - disconnectAndDrain(signal): for owned single-lifetime transports; fences new work, disconnects, and waits for physical close of the client, proxy, every allocated client and all fenced work. Refuses (after cleaning up) when the transport cannot be proven, e.g. system SSH or a connect still in flight. - getExecutionDestination: the ssh2 endpoint, accepted host-key fingerprint and proxy-route digest proven by the current handshake (ssh-connection-destination). - getTransportGeneration, prepareForwardRoute, openForwardSocket, forwardOut, forwardStreamLocal for later forwarding callers. - An automaticReconnect constructor option (default on). Channel close is local lifetime evidence only, never a remote-exit verdict. Porting note (source: #16741 head a68b6f3531, merge-base 277c289bd4): - Taken: the ledger hunks of ssh-connection.ts, ssh-connection-destination, ssh-forward-channel-lifetime, ssh-upload-session-lifetime, the system-SSH facade EOF hunk, and their tests. - Adapted: operation bodies became private *Untracked methods called through the ledger instead of being re-indented; closure gating and the close drain moved to ssh-connection-transport-closure / ssh-connection-close-drain; the destination parser uses type guards instead of a cast. disconnectAndDrain fences through the ledger directly. Main's plain-SSH shell() goes through the ledger too. execCommand takes Pick; its string-stdin/maxOutputBytes hunk is not taken because main already streams stdin. Work-drain tests fence the private ledger until T3 adds the public fence; system-SSH drain cases split into their own file. - Left for later slices: fenceWorkForReset (T3), isEphemeralRuntimeSshOwner (T6), assertProfileLifetimeAdmission (P8b), and the four manager drain cases in ssh-connection-disconnect-drain.test.ts (P3). --- .../ssh/ssh-connection-channel-open.test.ts | 12 +- src/main/ssh/ssh-connection-close-drain.ts | 34 ++ ...ssh-connection-destination-capture.test.ts | 123 ++++++ .../ssh/ssh-connection-destination.test.ts | 57 +++ src/main/ssh/ssh-connection-destination.ts | 74 ++++ .../ssh/ssh-connection-direct-startup.test.ts | 8 +- .../ssh-connection-disconnect-drain.test.ts | 240 ++++++++++++ .../ssh/ssh-connection-file-transfer.test.ts | 7 +- ...h-connection-host-key-store-wiring.test.ts | 5 +- .../ssh-connection-reconnect-ladder.test.ts | 35 ++ .../ssh-connection-system-work-drain.test.ts | 100 +++++ src/main/ssh/ssh-connection-test-fixtures.ts | 9 +- .../ssh-connection-transport-closure.test.ts | 176 +++++++++ .../ssh/ssh-connection-transport-closure.ts | 33 ++ .../ssh/ssh-connection-work-drain.test.ts | 238 ++++++++++++ src/main/ssh/ssh-connection.ts | 352 ++++++++++++++++-- .../ssh/ssh-forward-channel-lifetime.test.ts | 98 +++++ src/main/ssh/ssh-forward-channel-lifetime.ts | 50 +++ .../ssh-multi-factor-authentication.test.ts | 31 +- src/main/ssh/ssh-relay-exec-command.ts | 2 +- src/main/ssh/ssh-relay-exec-input.test.ts | 57 +++ .../ssh-system-transport.integration.test.ts | 6 +- .../ssh/ssh-upload-session-lifetime.test.ts | 192 ++++++++++ src/main/ssh/ssh-upload-session-lifetime.ts | 33 ++ src/main/ssh/system-ssh-command.ts | 4 + 25 files changed, 1917 insertions(+), 59 deletions(-) create mode 100644 src/main/ssh/ssh-connection-close-drain.ts create mode 100644 src/main/ssh/ssh-connection-destination-capture.test.ts create mode 100644 src/main/ssh/ssh-connection-destination.test.ts create mode 100644 src/main/ssh/ssh-connection-destination.ts create mode 100644 src/main/ssh/ssh-connection-disconnect-drain.test.ts create mode 100644 src/main/ssh/ssh-connection-system-work-drain.test.ts create mode 100644 src/main/ssh/ssh-connection-transport-closure.test.ts create mode 100644 src/main/ssh/ssh-connection-transport-closure.ts create mode 100644 src/main/ssh/ssh-connection-work-drain.test.ts create mode 100644 src/main/ssh/ssh-forward-channel-lifetime.test.ts create mode 100644 src/main/ssh/ssh-forward-channel-lifetime.ts create mode 100644 src/main/ssh/ssh-relay-exec-input.test.ts create mode 100644 src/main/ssh/ssh-upload-session-lifetime.test.ts create mode 100644 src/main/ssh/ssh-upload-session-lifetime.ts diff --git a/src/main/ssh/ssh-connection-channel-open.test.ts b/src/main/ssh/ssh-connection-channel-open.test.ts index f0f6454af648..58d4a5633b23 100644 --- a/src/main/ssh/ssh-connection-channel-open.test.ts +++ b/src/main/ssh/ssh-connection-channel-open.test.ts @@ -286,7 +286,7 @@ describe('SshConnection', () => { await expect(outcomePromise).resolves.toBe('AbortError') }) - it('removes the late-channel close listener when abort grace expires', async () => { + it('removes the abort-grace waiter while retaining physical-close tracking', async () => { const conn = new SshConnection(createTarget(), createCallbacks()) await conn.connect() vi.useFakeTimers() @@ -302,12 +302,14 @@ describe('SshConnection', () => { await vi.advanceTimersByTimeAsync(0) controller.abort() pendingSftpCallback?.(undefined, lateSftp) - expect(lateSftp.listenerCount('close')).toBe(1) + expect(lateSftp.listenerCount('close')).toBe(2) expect(() => lateSftp.emit('error', new Error('late SFTP teardown'))).not.toThrow() await vi.advanceTimersByTimeAsync(5_000) await expect(outcomePromise).resolves.toBe('AbortError') + expect(lateSftp.listenerCount('close')).toBe(1) + lateSftp.emit('close') expect(lateSftp.listenerCount('close')).toBe(0) } finally { vi.useRealTimers() @@ -376,7 +378,7 @@ describe('SshConnection', () => { expect(lateSftp.end).toHaveBeenCalledTimes(1) }) - it('removes the late SFTP close listener when the bounded grace expires', async () => { + it('retains SFTP physical-close tracking after the bounded observation expires', async () => { const conn = new SshConnection(createTarget(), createCallbacks()) await conn.connect() ssh2Mock.sftpBehavior = 'pending' @@ -393,11 +395,13 @@ describe('SshConnection', () => { await Promise.resolve() controller.abort() pendingSftpCallback?.(undefined, lateSftp) - expect(lateSftp.listenerCount('close')).toBe(1) + expect(lateSftp.listenerCount('close')).toBe(2) await vi.advanceTimersByTimeAsync(5_000) await expect(outcomePromise).resolves.toBe('AbortError') + expect(lateSftp.listenerCount('close')).toBe(1) + lateSftp.emit('close') expect(lateSftp.listenerCount('close')).toBe(0) expect(lateSftp.end).toHaveBeenCalledTimes(1) } finally { diff --git a/src/main/ssh/ssh-connection-close-drain.ts b/src/main/ssh/ssh-connection-close-drain.ts new file mode 100644 index 000000000000..e905a62ceaab --- /dev/null +++ b/src/main/ssh/ssh-connection-close-drain.ts @@ -0,0 +1,34 @@ +import type { EventEmitter } from 'node:events' +import type { SshConnectionWorkLedger } from './ssh-connection-work-ledger' +import type { SshTransportCloseLedger } from './ssh-transport-close-ledger' +import { observeSshTransportClose } from './ssh-transport-close-observation' + +type WorkFence = ReturnType + +/** + * Disconnects, then waits for the live transport resources, every client ever allocated and all + * fenced work to close. Listeners go on before close starts so a fast 'close' is not missed. + */ +export async function disconnectAndAwaitSshTransportClose(options: { + liveResources: readonly EventEmitter[] + disconnect: () => Promise + transportCloseLedger: SshTransportCloseLedger + fence: WorkFence + signal: AbortSignal +}): Promise { + const observation = observeSshTransportClose(options.liveResources) + const waiting = new AbortController() + const waitSignal = AbortSignal.any([options.signal, waiting.signal]) + try { + await options.disconnect() + await Promise.all([ + observation.wait(waitSignal), + options.transportCloseLedger.drain(waitSignal), + options.fence.drain(waitSignal) + ]) + options.fence.assertDrained() + } finally { + waiting.abort() + observation.dispose() + } +} diff --git a/src/main/ssh/ssh-connection-destination-capture.test.ts b/src/main/ssh/ssh-connection-destination-capture.test.ts new file mode 100644 index 000000000000..6f0ce04a8087 --- /dev/null +++ b/src/main/ssh/ssh-connection-destination-capture.test.ts @@ -0,0 +1,123 @@ +import { createHash } from 'node:crypto' +import { beforeEach, expect, it, vi } from 'vitest' +import { + clientInstances, + nextSshClientCreation, + resetSshConnectionMocks, + resolveWithSshGMock, + ssh2Mock, + VALID_ED25519_HOST_KEY +} from './ssh-connection-test-harness' +import { createCallbacks, createTarget, createResolvedConfig } from './ssh-connection-test-fixtures' +import { SshConnection } from './ssh-connection' + +vi.mock('ssh2', async () => (await import('./ssh-connection-test-harness')).createSsh2Module()) +vi.mock('./system-ssh-binary', async () => + (await import('./ssh-connection-test-harness')).createSystemSshBinaryModule() +) +vi.mock('./ssh-system-fallback', async () => + (await import('./ssh-connection-test-harness')).createSystemFallbackModule() +) +vi.mock('./ssh-control-socket', async () => + (await import('./ssh-connection-test-harness')).createControlSocketModule() +) +vi.mock('./ssh-config-parser', async () => + (await import('./ssh-connection-test-harness')).createSshConfigParserModule() +) + +beforeEach(() => { + resetSshConnectionMocks() +}) + +it('captures only the successful endpoint and accepted key, not the saved alias', async () => { + resolveWithSshGMock.mockResolvedValue( + createResolvedConfig({ + hostname: 'resolved.example', + port: 2222, + user: 'resolved-user', + proxyUseFdpass: false + }) + ) + const conn = new SshConnection(createTarget({ configHost: 'alias' }), createCallbacks()) + expect(conn.getExecutionDestination()).toBeUndefined() + try { + await conn.connect() + const destination = conn.getExecutionDestination() + expect(destination).toEqual({ + version: 1, + transport: 'ssh2', + host: 'resolved.example', + port: 2222, + username: 'resolved-user', + hostKeyFingerprint: `SHA256:${createHash('sha256').update(VALID_ED25519_HOST_KEY).digest('base64').replace(/=+$/, '')}`, + proxyRouteDigest: createHash('sha256').update('null').digest('hex') + }) + expect(Object.isFrozen(destination)).toBe(true) + expect(conn.getExecutionDestination()).toBe(destination) + } finally { + await conn.disconnect() + } + expect(conn.getExecutionDestination()).toBeUndefined() +}) + +it('does not publish an accepted key until authentication succeeds', async () => { + ssh2Mock.connectBehavior = 'pending' + const conn = new SshConnection(createTarget(), createCallbacks()) + const created = nextSshClientCreation() + const connecting = conn.connect() + await created + expect(conn.getExecutionDestination()).toBeUndefined() + const rejected = expect(connecting).rejects.toThrow() + await conn.disconnect() + await rejected + clientInstances[0].emit('ready') + expect(conn.getExecutionDestination()).toBeUndefined() +}) + +it('invalidates identity immediately on transport loss', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + try { + await conn.connect() + expect(conn.getExecutionDestination()).toBeDefined() + clientInstances[0].emit('end') + expect(conn.getExecutionDestination()).toBeUndefined() + } finally { + await conn.disconnect() + } +}) + +it('does not publish an accepted key from failed authentication', async () => { + ssh2Mock.connectBehavior = 'error' + ssh2Mock.connectErrorMessage = 'Authentication failed' + const conn = new SshConnection(createTarget(), createCallbacks()) + try { + await expect(conn.connect()).rejects.toThrow() + expect(ssh2Mock.lastHostKeyAccepted).toBe(true) + expect(conn.getExecutionDestination()).toBeUndefined() + } finally { + await conn.disconnect() + } +}) + +it('never publishes identity from a rejected host key', async () => { + ssh2Mock.presentedHostKey = Buffer.from('invalid host key') + const conn = new SshConnection(createTarget(), createCallbacks()) + try { + await expect(conn.connect()).rejects.toThrow() + expect(conn.getExecutionDestination()).toBeUndefined() + } finally { + await conn.disconnect() + } +}) + +it('does not treat a system SSH probe as command-channel destination evidence', async () => { + const conn = new SshConnection(createTarget({ configHost: 'alias' }), createCallbacks()) + resolveWithSshGMock.mockResolvedValue(createResolvedConfig({ proxyUseFdpass: true })) + try { + await conn.connect() + expect(conn.usesSystemSshTransport()).toBe(true) + expect(conn.getExecutionDestination()).toBeUndefined() + } finally { + await conn.disconnect() + } +}) diff --git a/src/main/ssh/ssh-connection-destination.test.ts b/src/main/ssh/ssh-connection-destination.test.ts new file mode 100644 index 000000000000..7240113e108b --- /dev/null +++ b/src/main/ssh/ssh-connection-destination.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from 'vitest' +import { parseSshConnectionDestination } from './ssh-connection-destination' + +const destination = { + version: 1, + transport: 'ssh2', + host: 'resolved.example', + port: 2222, + username: 'owner', + hostKeyFingerprint: `SHA256:${Buffer.alloc(32, 255).toString('base64').replace(/=+$/, '')}`, + proxyRouteDigest: 'a'.repeat(64) +} + +describe('parseSshConnectionDestination', () => { + it('makes an immutable canonical snapshot without normalizing identity', () => { + const input = { ...destination, host: 'MixedCase.example', ignored: true } + const saved = parseSshConnectionDestination(input) + input.host = 'another.example' + expect(saved).toEqual({ ...destination, host: 'MixedCase.example' }) + expect(Object.isFrozen(saved)).toBe(true) + expect(parseSshConnectionDestination(JSON.parse(JSON.stringify(saved)))).toEqual(saved) + }) + + it.each([ + { version: 2 }, + { transport: 'system-ssh' }, + { host: '' }, + { host: 'host\nother' }, + { host: 'h'.repeat(8193) }, + { username: '' }, + { username: 'user\0name' }, + { username: 3 }, + { port: 0 }, + { port: 65536 }, + { port: 22.5 }, + { port: '22' }, + { port: Number.NaN }, + { hostKeyFingerprint: 'SHA256:short' }, + { hostKeyFingerprint: `SHA256:${'A'.repeat(43)}=` }, + { hostKeyFingerprint: `SHA256:${'A'.repeat(42)}B` }, + { proxyRouteDigest: 'A'.repeat(64) }, + { proxyRouteDigest: 'a'.repeat(63) }, + { proxyRouteDigest: undefined } + ])('refuses malformed binding %j', (change) => { + expect(() => parseSshConnectionDestination({ ...destination, ...change })).toThrow( + 'ssh_connection_destination_invalid' + ) + }) + + it.each([null, undefined, [], 'host', 1])('refuses nonrecords %j', (value) => { + expect(() => parseSshConnectionDestination(value)).toThrow('ssh_connection_destination_invalid') + }) + + it.each([1, 65535])('accepts port boundary %i', (port) => { + expect(parseSshConnectionDestination({ ...destination, port }).port).toBe(port) + }) +}) diff --git a/src/main/ssh/ssh-connection-destination.ts b/src/main/ssh/ssh-connection-destination.ts new file mode 100644 index 000000000000..49bbbe8545a0 --- /dev/null +++ b/src/main/ssh/ssh-connection-destination.ts @@ -0,0 +1,74 @@ +import { createHash } from 'node:crypto' + +/** The execution host a successful ssh2 handshake proved: endpoint, accepted key and proxy route. */ +export type SshConnectionDestination = Readonly<{ + version: 1 + transport: 'ssh2' + host: string + port: number + username: string + hostKeyFingerprint: string + proxyRouteDigest: string +}> + +/** Digest of the resolved proxy route; a changed ProxyCommand/ProxyJump names a different path. */ +export function sshProxyRouteDigest(effectiveProxy: unknown): string { + return createHash('sha256') + .update(JSON.stringify(effectiveProxy ?? null)) + .digest('hex') +} + +function isValidText(text: unknown): text is string { + return ( + typeof text === 'string' && + text.length > 0 && + text.length <= 8192 && + !Array.from(text).some( + (character) => character.charCodeAt(0) < 32 || character.charCodeAt(0) === 127 + ) + ) +} + +function isCanonicalSha256Fingerprint(value: unknown): value is string { + if (typeof value !== 'string' || !/^SHA256:[A-Za-z0-9+/]{43}$/.test(value)) { + return false + } + const encoded = value.slice(7) + // Why: a non-canonical trailing digit decodes to the same bytes but names a different string. + return ( + Buffer.from(encoded, 'base64').toString('base64url') === + encoded.replace(/\+/g, '-').replace(/\//g, '_') + ) +} + +export function parseSshConnectionDestination(value: unknown): SshConnectionDestination { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw new Error('ssh_connection_destination_invalid') + } + const record: Record = { ...value } + const { host, port, username, hostKeyFingerprint, proxyRouteDigest } = record + if ( + record.version !== 1 || + record.transport !== 'ssh2' || + !isValidText(host) || + typeof port !== 'number' || + !Number.isSafeInteger(port) || + port < 1 || + port > 65535 || + !isValidText(username) || + !isCanonicalSha256Fingerprint(hostKeyFingerprint) || + typeof proxyRouteDigest !== 'string' || + !/^[a-f0-9]{64}$/.test(proxyRouteDigest) + ) { + throw new Error('ssh_connection_destination_invalid') + } + return Object.freeze({ + version: 1, + transport: 'ssh2', + host, + port, + username, + hostKeyFingerprint, + proxyRouteDigest + }) +} diff --git a/src/main/ssh/ssh-connection-direct-startup.test.ts b/src/main/ssh/ssh-connection-direct-startup.test.ts index a2344930eeae..09bb39e653b1 100644 --- a/src/main/ssh/ssh-connection-direct-startup.test.ts +++ b/src/main/ssh/ssh-connection-direct-startup.test.ts @@ -56,12 +56,16 @@ describe('SshConnection', () => { await connect expect(channel.close).toHaveBeenCalled() expect(channel.listenerCount('data')).toBe(0) - expect(channel.listenerCount('error')).toBe(1) - expect(channel.listenerCount('close')).toBe(1) + // Lifetime observers remain until physical close, even after the probe timed out. + expect(channel.listenerCount('error')).toBe(2) + expect(channel.listenerCount('close')).toBe(2) expect(channel.stderr.listenerCount('data')).toBe(0) expect( (conn as unknown as { systemCommandChannels: Set }).systemCommandChannels.size ).toBe(0) + channel.emit('close', 0) + expect(channel.listenerCount('close')).toBe(0) + expect(channel.listenerCount('error')).toBe(1) } finally { vi.useRealTimers() } diff --git a/src/main/ssh/ssh-connection-disconnect-drain.test.ts b/src/main/ssh/ssh-connection-disconnect-drain.test.ts new file mode 100644 index 000000000000..b361e6a5ed61 --- /dev/null +++ b/src/main/ssh/ssh-connection-disconnect-drain.test.ts @@ -0,0 +1,240 @@ +import { EventEmitter } from 'node:events' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { SshConnection } from './ssh-connection' +import { createCallbacks, createResolvedConfig, createTarget } from './ssh-connection-test-fixtures' +import { clientInstances, resetSshConnectionMocks, ssh2Mock } from './ssh-connection-test-harness' +import { resolveWithSshG } from './ssh-config-parser' + +vi.mock('ssh2', async () => (await import('./ssh-connection-test-harness')).createSsh2Module()) +vi.mock('./system-ssh-binary', async () => + (await import('./ssh-connection-test-harness')).createSystemSshBinaryModule() +) +vi.mock('./ssh-system-fallback', async () => + (await import('./ssh-connection-test-harness')).createSystemFallbackModule() +) +vi.mock('./ssh-control-socket', async () => + (await import('./ssh-connection-test-harness')).createControlSocketModule() +) +vi.mock('./ssh-config-parser', async () => + (await import('./ssh-connection-test-harness')).createSshConfigParserModule() +) +beforeEach(resetSshConnectionMocks) +afterEach(() => { + vi.useRealTimers() + vi.restoreAllMocks() +}) + +async function fixture(conn = new SshConnection(createTarget(), createCallbacks())) { + await conn.connect() + const client = conn.getClient()! + // The shared SSH mock omits once/removeListener; retain its lifecycle handlers alongside real events. + const observed = new EventEmitter() + const emit = client.emit.bind(client) + Object.assign(client, { + once: observed.once.bind(observed), + removeListener: observed.removeListener.bind(observed), + listenerCount: observed.listenerCount.bind(observed), + emit: (event: string, ...args: unknown[]) => { + emit(event, ...args) + return observed.emit(event, ...args) + } + }) + const end = vi.spyOn(client, 'end').mockImplementation(() => client) + const run = (signal = new AbortController().signal) => conn.disconnectAndDrain(signal) + const forward = () => { + const channel = new EventEmitter() + client.openssh_forwardOutStreamLocal = vi.fn((_path, callback) => { + callback(undefined, channel as never) + return client + }) + conn.forwardStreamLocal(client, '/owned.sock', vi.fn()) + return channel + } + return { conn, client, end, run, forward } +} + +it('waits for failed startup sockets even after another attempt connects successfully', async () => { + const conn = new SshConnection(createTarget(), createCallbacks(), { automaticReconnect: false }) + ssh2Mock.connectSequence = [new Error('startup refused')] + ssh2Mock.destroyErrorMessage = 'socket still closing' + await expect(conn.connect()).rejects.toThrow('startup refused') + const failed = clientInstances[0]! + ssh2Mock.destroyErrorMessage = '' + const f = await fixture(conn) + const done = vi.fn() + const draining = f.run().then(done) + f.client.emit('close') + // Let all resolved drain promises settle; the failed socket remains open. + await new Promise((resolve) => setImmediate(resolve)) + expect(done).not.toHaveBeenCalled() + failed.emit('close') + await draining + expect(done).toHaveBeenCalledOnce() +}) + +it('drains a settled failed exclusive startup only after its socket physically closes', async () => { + const conn = new SshConnection(createTarget(), createCallbacks(), { automaticReconnect: false }) + ssh2Mock.connectSequence = [new Error('startup refused')] + ssh2Mock.destroyErrorMessage = 'socket still closing' + await expect(conn.connect()).rejects.toThrow('startup refused') + const failed = clientInstances[0]! + const done = vi.fn() + const draining = conn.disconnectAndDrain(new AbortController().signal).then(done) + await new Promise((resolve) => setImmediate(resolve)) + expect(done).not.toHaveBeenCalled() + failed.emit('close') + await draining + expect(done).toHaveBeenCalledOnce() + await expect(conn.connect()).rejects.toThrow('ssh_connection_reset_replacement_refused') +}) + +it('remembers failed socket closure across an early ordinary disconnect', async () => { + const conn = new SshConnection(createTarget(), createCallbacks(), { automaticReconnect: false }) + ssh2Mock.connectSequence = [new Error('startup refused')] + await expect(conn.connect()).rejects.toThrow('startup refused') + await conn.disconnect() + await conn.disconnectAndDrain(new AbortController().signal) +}) + +it('allows a fenced exclusive connection with no allocations to finish cleanup', async () => { + const conn = new SshConnection(createTarget(), createCallbacks(), { automaticReconnect: false }) + await conn.disconnectAndDrain(new AbortController().signal) + await expect(conn.connect()).rejects.toThrow('ssh_connection_reset_replacement_refused') +}) + +it('refuses drain while configuration can still allocate a startup transport', async () => { + const resolved = Promise.withResolvers() + vi.mocked(resolveWithSshG).mockReturnValueOnce(resolved.promise) + const conn = new SshConnection(createTarget(), createCallbacks(), { automaticReconnect: false }) + const connecting = conn.connect().catch((error: unknown) => error) + await expect(conn.disconnectAndDrain(new AbortController().signal)).rejects.toThrow( + 'ssh_connection_close_transport_unproven' + ) + resolved.resolve(null) + expect(await connecting).toBeInstanceOf(Error) + await conn.disconnectAndDrain(new AbortController().signal) + expect(clientInstances).toHaveLength(0) +}) + +it('does not treat logical disconnection as physical client closure', async () => { + const f = await fixture() + const done = vi.fn() + const draining = f.run().then(done) + await Promise.resolve() + expect(f.end).toHaveBeenCalledOnce() + expect(f.conn.getState().status).toBe('disconnected') + expect(done).not.toHaveBeenCalled() + f.client.emit('close') + await draining + expect(done).toHaveBeenCalledOnce() +}) + +it.each(['client', 'channel'] as const)( + 'requires both physical client and channel close when %s closes first', + async (first) => { + const f = await fixture() + const channel = f.forward() + const done = vi.fn() + const draining = f.run().then(done) + ;(first === 'client' ? f.client : channel).emit('close') + await Promise.resolve() + await Promise.resolve() + expect(done).not.toHaveBeenCalled() + ;(first === 'client' ? channel : f.client).emit('close') + await draining + expect(done).toHaveBeenCalledOnce() + } +) + +it('subscribes before disconnect can synchronously close the client', async () => { + const f = await fixture() + f.end.mockImplementation(() => { + f.client.emit('close') + return f.client + }) + await f.run() + expect(f.end).toHaveBeenCalledOnce() +}) + +it('aborts waiting for physical close and removes its listener', async () => { + const f = await fixture() + const before = f.client.listenerCount('close') + const controller = new AbortController() + const draining = f.run(controller.signal) + expect(f.client.listenerCount('close')).toBeGreaterThan(before) + controller.abort() + await expect(draining).rejects.toThrow() + expect(f.client.listenerCount('close')).toBe(before) + expect(f.conn.getClient()).toBeNull() +}) + +it('pre-aborted admission does not close the connection', async () => { + const f = await fixture() + await expect(f.run(AbortSignal.abort())).rejects.toThrow() + expect(f.end).not.toHaveBeenCalled() + await f.conn.disconnect() +}) + +it('aborts unresolved channel drain even after the physical client closes', async () => { + const f = await fixture() + const channel = f.forward() + const controller = new AbortController() + const draining = f.run(controller.signal) + f.client.emit('close') + controller.abort() + await expect(draining).rejects.toThrow() + expect(f.client.listenerCount('close')).toBe(0) + channel.emit('close') +}) + +it('cleans owned overlapping clients but refuses unproven drain', async () => { + const f = await fixture() + const pending = { end: vi.fn(), destroy: vi.fn() } + const clients = (f.conn as unknown as { pendingSsh2Clients: Set }).pendingSsh2Clients + clients.add(pending) + await expect(f.run()).rejects.toThrow() + expect(f.end).toHaveBeenCalledOnce() + expect(pending.destroy).toHaveBeenCalledOnce() + clients.delete(pending) + await f.conn.disconnect() +}) + +it('waits for the owned proxy close as well as client close', async () => { + const f = await fixture() + const proxy = Object.assign(new EventEmitter(), { kill: vi.fn() }) + ;(f.conn as unknown as { proxyProcess: typeof proxy }).proxyProcess = proxy + const done = vi.fn() + const draining = f.run().then(done) + expect(proxy.kill).toHaveBeenCalledOnce() + f.client.emit('close') + await Promise.resolve() + expect(done).not.toHaveBeenCalled() + proxy.emit('close') + await draining +}) + +it('never schedules reconnect when a drain closes its client', async () => { + const f = await fixture() + vi.useFakeTimers() + const draining = f.run() + f.client.emit('close') + await draining + await vi.advanceTimersByTimeAsync(60_000) + expect(f.conn.getClient()).toBeNull() + expect(f.conn.getState().reconnectAttempt).toBe(0) + expect(vi.getTimerCount()).toBe(0) +}) + +it('refuses a disconnected connection without claiming drain proof', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await expect(conn.disconnectAndDrain(new AbortController().signal)).rejects.toThrow() +}) + +it('closes owned system SSH but refuses unproven drain', async () => { + vi.mocked(resolveWithSshG).mockResolvedValueOnce(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + await conn.connect() + await expect(conn.disconnectAndDrain(new AbortController().signal)).rejects.toThrow() + expect(conn.getState().status).toBe('disconnected') + await conn.disconnect() +}) diff --git a/src/main/ssh/ssh-connection-file-transfer.test.ts b/src/main/ssh/ssh-connection-file-transfer.test.ts index 0d2e830d6bb7..610d24c08d92 100644 --- a/src/main/ssh/ssh-connection-file-transfer.test.ts +++ b/src/main/ssh/ssh-connection-file-transfer.test.ts @@ -182,11 +182,6 @@ describe('SshConnection', () => { await expect( uploadSession.uploadFile('/tmp/late.txt', '/remote/late.txt') ).rejects.toMatchObject({ name: 'AbortError' }) - expect(uploadFileViaSystemSsh).toHaveBeenCalledWith( - expect.anything(), - '/tmp/late.txt', - '/remote/late.txt', - expect.objectContaining({ signal: expect.objectContaining({ aborted: true }) }) - ) + expect(uploadFileViaSystemSsh).not.toHaveBeenCalled() }) }) diff --git a/src/main/ssh/ssh-connection-host-key-store-wiring.test.ts b/src/main/ssh/ssh-connection-host-key-store-wiring.test.ts index 9dd1e6d22a2a..68d3b6ec19a6 100644 --- a/src/main/ssh/ssh-connection-host-key-store-wiring.test.ts +++ b/src/main/ssh/ssh-connection-host-key-store-wiring.test.ts @@ -16,6 +16,7 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' // Type-only, so it is erased before vi.mock's hoisted factory runs. import type * as SshConfigParser from './ssh-config-parser' +import type * as Ssh2 from 'ssh2' const VALID_ED25519_HOST_KEY = Buffer.from( 'AAAAC3NzaC1lZDI1NTE5AAAAIKqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq', @@ -31,8 +32,8 @@ let eventHandlers: Map void>> let presentedHostKey: Buffer let hostKeyAccepted: boolean | undefined -vi.mock('ssh2', () => { - const utils = { parseKey: vi.fn(() => new Error('parse failed')) } +vi.mock('ssh2', async (importOriginal) => { + const { utils } = await importOriginal() class MockSshClient { setNoDelay = vi.fn() _sock: Socket | undefined = new Socket() diff --git a/src/main/ssh/ssh-connection-reconnect-ladder.test.ts b/src/main/ssh/ssh-connection-reconnect-ladder.test.ts index d62856611bad..40b9451848ef 100644 --- a/src/main/ssh/ssh-connection-reconnect-ladder.test.ts +++ b/src/main/ssh/ssh-connection-reconnect-ladder.test.ts @@ -256,6 +256,41 @@ describe('SshConnection', () => { } }) + it('recovers the same owner after a temporary DNS resolution failure', async () => { + vi.useFakeTimers() + try { + const states: string[] = [] + const conn = new SshConnection( + createTarget(), + createCallbacks({ + onStateChange: vi.fn((_id, state) => states.push(state.status)) + }) + ) + await connectWithFakeTimers(conn) + const connectedGeneration = conn.getTransportGeneration() + const dnsFailure = new Error( + 'getaddrinfo EAI_AGAIN temporary failure in name resolution' + ) as NodeJS.ErrnoException + dnsFailure.code = 'EAI_AGAIN' + ssh2Mock.connectSequence = [dnsFailure, 'ready'] + + emitSshEvent('close') + await advanceToNextSshClient(RECONNECT_BACKOFF_MS[0]) + + expect(conn.getState().status).toBe('reconnecting') + expect(states).not.toContain('error') + expect(states).not.toContain('reconnection-failed') + + await advanceToNextSshClient(RECONNECT_BACKOFF_MS[1]) + + expect(conn.getState().status).toBe('connected') + expect(conn.getTransportGeneration()).toBeGreaterThan(connectedGeneration) + expect(clientInstances).toHaveLength(3) + } finally { + vi.useRealTimers() + } + }) + it('keeps a system-transport target on the ladder after a probe timeout', async () => { vi.useFakeTimers() try { diff --git a/src/main/ssh/ssh-connection-system-work-drain.test.ts b/src/main/ssh/ssh-connection-system-work-drain.test.ts new file mode 100644 index 000000000000..e4785495b204 --- /dev/null +++ b/src/main/ssh/ssh-connection-system-work-drain.test.ts @@ -0,0 +1,100 @@ +import { beforeEach, expect, it, vi } from 'vitest' +import { SshConnection } from './ssh-connection' +import { + createCallbacks, + createResolvedConfig, + createTarget, + fenceSshConnectionWork +} from './ssh-connection-test-fixtures' +import { resetSshConnectionMocks } from './ssh-connection-test-harness' +import { resolveWithSshG } from './ssh-config-parser' +import { + downloadFileViaSystemSsh, + uploadDirectoryViaSystemSsh, + uploadFileViaSystemSsh, + writeFileViaSystemSsh, + writeBufferViaSystemSsh +} from './ssh-system-fallback' + +vi.mock('ssh2', async () => (await import('./ssh-connection-test-harness')).createSsh2Module()) +vi.mock('./system-ssh-binary', async () => + (await import('./ssh-connection-test-harness')).createSystemSshBinaryModule() +) +vi.mock('./ssh-system-fallback', async () => + (await import('./ssh-connection-test-harness')).createSystemFallbackModule() +) +vi.mock('./ssh-control-socket', async () => + (await import('./ssh-connection-test-harness')).createControlSocketModule() +) +vi.mock('./ssh-config-parser', async () => + (await import('./ssh-connection-test-harness')).createSshConfigParserModule() +) + +beforeEach(resetSshConnectionMocks) + +it('holds an admitted system upload session across reset until close and uploads settle', async () => { + vi.mocked(resolveWithSshG).mockResolvedValueOnce(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + await conn.connect() + const session = await conn.openFileUploadSession() + const fence = fenceSshConnectionWork(conn) + await expect(conn.openFileUploadSession()).rejects.toThrow('admission_closed') + await session.uploadFile('/first', '/remote-first') + const pending = Promise.withResolvers() + vi.mocked(uploadFileViaSystemSsh).mockReturnValueOnce(pending.promise) + const upload = session.uploadFile('/second', '/remote-second') + session.close() + await expect(session.uploadFile('/third', '/remote-third')).rejects.toThrow('session_closed') + expect(() => fence.assertDrained()).toThrow('not_drained') + const draining = fence.drain(new AbortController().signal) + expect(conn.getState().status).toBe('connected') + pending.resolve() + await upload + await draining + fence.assertDrained() + expect(uploadFileViaSystemSsh).toHaveBeenCalledTimes(2) +}) + +it.each(['download', 'upload', 'writeFile', 'writeBuffer'])( + 'drains the whole admitted system SSH %s without canceling it', + async (kind) => { + vi.mocked(resolveWithSshG).mockResolvedValueOnce(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + await conn.connect() + const pending = Promise.withResolvers() + const invoke = () => { + if (kind === 'download') { + return conn.downloadFile('/remote', '/local') + } + if (kind === 'upload') { + return conn.uploadDirectory('/local', '/remote') + } + if (kind === 'writeFile') { + return conn.writeFile('/remote', 'text') + } + return conn.writeBuffer('/remote', Buffer.from('bytes')) + } + const operation = + kind === 'download' + ? downloadFileViaSystemSsh + : kind === 'upload' + ? uploadDirectoryViaSystemSsh + : kind === 'writeFile' + ? writeFileViaSystemSsh + : writeBufferViaSystemSsh + vi.mocked(operation).mockReturnValueOnce(pending.promise) + const running = invoke() + const fence = fenceSshConnectionWork(conn) + const done = vi.fn() + const draining = fence.drain(new AbortController().signal).then(done) + await Promise.resolve() + expect(done).not.toHaveBeenCalled() + await expect(invoke()).rejects.toThrow('admission_closed') + expect(operation).toHaveBeenCalledTimes(1) + expect(conn.getState().status).toBe('connected') + pending.resolve() + await running + await draining + fence.assertDrained() + } +) diff --git a/src/main/ssh/ssh-connection-test-fixtures.ts b/src/main/ssh/ssh-connection-test-fixtures.ts index cfed6a010618..dce4a05cb98c 100644 --- a/src/main/ssh/ssh-connection-test-fixtures.ts +++ b/src/main/ssh/ssh-connection-test-fixtures.ts @@ -1,7 +1,8 @@ import { EventEmitter } from 'node:events' import { vi } from 'vitest' import type { Mock } from 'vitest' -import type { SshConnectionCallbacks } from './ssh-connection' +import type { SshConnection, SshConnectionCallbacks } from './ssh-connection' +import type { SshConnectionWorkLedger } from './ssh-connection-work-ledger' import type { SshResolvedConfig } from './ssh-config-parser' import type { SshTarget } from '../../shared/ssh-types' @@ -124,3 +125,9 @@ export function createFailingSystemSshProcess(code: number): MockSystemSshProces }) return proc } + +// The public reset fence lands with T3; until then the connection's own ledger fence drains the same work. +export function fenceSshConnectionWork(conn: SshConnection) { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: SshConnection owns exactly one private workLedger. + return (conn as unknown as { workLedger: SshConnectionWorkLedger }).workLedger.fenceForReset() +} diff --git a/src/main/ssh/ssh-connection-transport-closure.test.ts b/src/main/ssh/ssh-connection-transport-closure.test.ts new file mode 100644 index 000000000000..9581eef7ec05 --- /dev/null +++ b/src/main/ssh/ssh-connection-transport-closure.test.ts @@ -0,0 +1,176 @@ +import { EventEmitter } from 'node:events' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { SshConnection } from './ssh-connection' +import { createCallbacks, createResolvedConfig, createTarget } from './ssh-connection-test-fixtures' +import { resetSshConnectionMocks } from './ssh-connection-test-harness' +import { resolveWithSshG } from './ssh-config-parser' +import type { SshTransportCloseLedger } from './ssh-transport-close-ledger' +import type { SshConnectionWorkLedger } from './ssh-connection-work-ledger' + +vi.mock('ssh2', async () => (await import('./ssh-connection-test-harness')).createSsh2Module()) +vi.mock('./system-ssh-binary', async () => + (await import('./ssh-connection-test-harness')).createSystemSshBinaryModule() +) +vi.mock('./ssh-system-fallback', async () => + (await import('./ssh-connection-test-harness')).createSystemFallbackModule() +) +vi.mock('./ssh-control-socket', async () => + (await import('./ssh-connection-test-harness')).createControlSocketModule() +) +vi.mock('./ssh-config-parser', async () => + (await import('./ssh-connection-test-harness')).createSshConfigParserModule() +) +beforeEach(resetSshConnectionMocks) +afterEach(() => { + vi.restoreAllMocks() + vi.useRealTimers() +}) + +async function fixture() { + const conn = new SshConnection(createTarget(), createCallbacks(), { automaticReconnect: false }) + const closed = vi.fn() + const remove = conn.subscribeTransportClosure(closed) + await conn.connect() + const client = conn.getClient()! + vi.spyOn(client, 'end').mockImplementation(() => client) + return { conn, client, closed, remove } +} + +it('notifies only after disposal and physical client closure, including late subscribers', async () => { + const f = await fixture() + expect(f.closed).not.toHaveBeenCalled() + await f.conn.disconnect() + expect(f.closed).not.toHaveBeenCalled() + f.client.emit('close') + expect(f.closed).toHaveBeenCalledOnce() + const late = vi.fn() + f.conn.subscribeTransportClosure(late) + expect(late).toHaveBeenCalledOnce() + f.client.emit('close') + expect(f.closed).toHaveBeenCalledOnce() + expect(late).toHaveBeenCalledOnce() +}) + +it('remembers physical closure before disposal but does not report it early', async () => { + const f = await fixture() + f.client.emit('close') + expect(f.closed).not.toHaveBeenCalled() + await f.conn.disconnect() + expect(f.closed).toHaveBeenCalledOnce() +}) + +it('holds notification through tracked work and both forwarded resources', async () => { + const f = await fixture() + const pending = Promise.withResolvers() + const operation = f.conn.prepareForwardRoute(() => pending.promise) + const socket = new EventEmitter() + const channel = new EventEmitter() + f.client.forwardOut = vi.fn((_a, _b, _c, _d, callback) => { + callback?.(undefined, channel as never) + return f.client + }) + f.conn.forwardOut(f.client, socket, '127.0.0.1', 0, 'host', 443, vi.fn()) + await f.conn.disconnect() + f.client.emit('close') + channel.emit('close') + expect(f.closed).not.toHaveBeenCalled() + socket.emit('close') + expect(f.closed).not.toHaveBeenCalled() + pending.resolve() + await operation + expect(f.closed).toHaveBeenCalledOnce() +}) + +it('waits for every allocated proxy even after its active pointer is cleared', async () => { + const f = await fixture() + const proxy = new EventEmitter() + const ledger = (f.conn as unknown as { transportCloseLedger: SshTransportCloseLedger }) + .transportCloseLedger + ledger.track(proxy) + await f.conn.disconnect() + f.client.emit('close') + expect(f.closed).not.toHaveBeenCalled() + proxy.emit('close') + expect(f.closed).toHaveBeenCalledOnce() +}) + +it('unsubscribes without changing ordinary disconnection', async () => { + const f = await fixture() + f.remove() + await f.conn.disconnect() + f.client.emit('close') + expect(f.closed).not.toHaveBeenCalled() +}) + +it('isolates observer failures from disconnect and other observers', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + conn.subscribeTransportClosure(() => { + throw new Error('observer') + }) + const observed = vi.fn() + conn.subscribeTransportClosure(observed) + await expect(conn.disconnect()).resolves.toBeUndefined() + expect(observed).toHaveBeenCalledOnce() +}) + +it('never treats unproven system SSH cleanup as physical closure', async () => { + vi.mocked(resolveWithSshG).mockResolvedValueOnce(createResolvedConfig()) + const conn = new SshConnection(createTarget({ configHost: 'fdpass-host' }), createCallbacks()) + const observed = vi.fn() + conn.subscribeTransportClosure(observed) + await conn.connect() + await conn.disconnect() + expect(observed).not.toHaveBeenCalled() +}) + +it.each(['initial', 'reconnect', 'system'] as const)( + 'waits for active %s allocation work', + async (kind) => { + const f = kind === 'reconnect' ? await fixture() : null + const conn = + f?.conn ?? new SshConnection(createTarget(), createCallbacks(), { automaticReconnect: false }) + const observed = vi.fn() + conn.subscribeTransportClosure(observed) + const config = Promise.withResolvers() + vi.mocked(resolveWithSshG).mockReturnValueOnce(config.promise) + const opening = ( + kind === 'initial' + ? conn.connect() + : kind === 'reconnect' + ? conn.reconnect() + : conn.connectViaSystemSsh() + ).catch(() => {}) + await conn.disconnect() + f?.client.emit('close') + expect(observed).not.toHaveBeenCalled() + config.resolve(null) + await opening + expect(observed).toHaveBeenCalledTimes(kind === 'system' ? 0 : 1) + } +) + +it('retains failed ledger drainage rather than signaling successful closure', async () => { + const f = await fixture() + const work = (f.conn as unknown as { workLedger: SshConnectionWorkLedger }).workLedger + const opening = work.beginChannelOpen() + work.fenceForReset() + opening.close(new Error('unproven channel')) + await f.conn.disconnect() + f.client.emit('close') + expect(f.closed).not.toHaveBeenCalled() +}) + +it('refuses fresh forwarding work after notifying physical closure', async () => { + const f = await fixture() + await f.conn.disconnect() + f.client.emit('close') + expect(f.closed).toHaveBeenCalledOnce() + const allocate = vi.fn(() => new EventEmitter()) + const prepare = vi.fn(async () => {}) + expect(() => f.conn.openForwardSocket(allocate)).toThrow( + expect.objectContaining({ name: 'AbortError' }) + ) + await expect(f.conn.prepareForwardRoute(prepare)).rejects.toMatchObject({ name: 'AbortError' }) + expect(allocate).not.toHaveBeenCalled() + expect(prepare).not.toHaveBeenCalled() +}) diff --git a/src/main/ssh/ssh-connection-transport-closure.ts b/src/main/ssh/ssh-connection-transport-closure.ts new file mode 100644 index 000000000000..3a92bc29b424 --- /dev/null +++ b/src/main/ssh/ssh-connection-transport-closure.ts @@ -0,0 +1,33 @@ +/** + * One-shot subscribers told when every transport resource a connection allocated has physically + * closed and its tracked work has drained. That is local lifetime evidence only: it never proves a + * remote process exited. + */ +export class SshTransportClosureSubscribers { + private readonly subscribers = new Set<() => void>() + + constructor(private readonly isPhysicallyClosed: () => boolean) {} + + subscribe(onClosed: () => void): () => void { + this.subscribers.add(onClosed) + this.notify() + return () => { + this.subscribers.delete(onClosed) + } + } + + notify(): void { + if (!this.subscribers.size || !this.isPhysicallyClosed()) { + return + } + const subscribers = [...this.subscribers] + this.subscribers.clear() + for (const onClosed of subscribers) { + try { + onClosed() + } catch { + // Observers must not interrupt physical transport cleanup. + } + } + } +} diff --git a/src/main/ssh/ssh-connection-work-drain.test.ts b/src/main/ssh/ssh-connection-work-drain.test.ts new file mode 100644 index 000000000000..ffe08884390d --- /dev/null +++ b/src/main/ssh/ssh-connection-work-drain.test.ts @@ -0,0 +1,238 @@ +import { EventEmitter } from 'node:events' +import { beforeEach, expect, it, vi } from 'vitest' +import { SshConnection } from './ssh-connection' +import { + createCallbacks, + createTarget, + fenceSshConnectionWork +} from './ssh-connection-test-fixtures' +import { + resetSshConnectionMocks, + pendingSftpCallback, + ssh2Mock +} from './ssh-connection-test-harness' + +vi.mock('ssh2', async () => (await import('./ssh-connection-test-harness')).createSsh2Module()) +vi.mock('./system-ssh-binary', async () => + (await import('./ssh-connection-test-harness')).createSystemSshBinaryModule() +) +vi.mock('./ssh-system-fallback', async () => + (await import('./ssh-connection-test-harness')).createSystemFallbackModule() +) +vi.mock('./ssh-control-socket', async () => + (await import('./ssh-connection-test-harness')).createControlSocketModule() +) +vi.mock('./ssh-config-parser', async () => + (await import('./ssh-connection-test-harness')).createSshConfigParserModule() +) + +beforeEach(resetSshConnectionMocks) + +it('does not automatically replace an explicitly single-lifetime SSH transport', async () => { + const options = { automaticReconnect: false } + const conn = new SshConnection(createTarget(), createCallbacks(), options) + options.automaticReconnect = true + await conn.connect() + const client = conn.getClient()! + vi.useFakeTimers() + try { + client.emit('close') + expect(conn.getClient()).toBeNull() + expect(conn.getState().status).toBe('disconnected') + expect(vi.getTimerCount()).toBe(0) + await vi.advanceTimersByTimeAsync(60_000) + expect(conn.getClient()).toBeNull() + expect(conn.getState().reconnectAttempt).toBe(0) + } finally { + await conn.disconnect() + vi.useRealTimers() + } +}) + +it('accounts for forward-route startup before reset and blocks new socket factories', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + const release = Promise.withResolvers() + const preparing = conn.prepareForwardRoute(async () => { + await release.promise + return conn.openForwardSocket(() => new EventEmitter()) + }) + const fence = fenceSshConnectionWork(conn) + const forbidden = vi.fn(() => new EventEmitter()) + expect(() => conn.openForwardSocket(forbidden)).toThrow('admission_closed') + expect(forbidden).not.toHaveBeenCalled() + expect(() => fence.assertDrained()).toThrow('not_drained') + release.resolve() + const admitted = await preparing + expect(() => fence.assertDrained()).toThrow('not_drained') + admitted.emit('close') + await fence.drain(new AbortController().signal) +}) + +it('routes actual connection forwarding through admission and rejects a replaced client', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + const client = conn.getClient()! + const channel = new EventEmitter() + const socket = new EventEmitter() + client.forwardOut = vi.fn((_a, _b, _c, _d, callback) => { + callback?.(undefined, channel as never) + return client + }) + conn.forwardOut(client, socket, '127.0.0.1', 0, 'remote', 443, vi.fn()) + const fence = fenceSshConnectionWork(conn) + expect(() => + conn.forwardOut(client, new EventEmitter(), '127.0.0.1', 0, 'remote', 443, vi.fn()) + ).toThrow('admission_closed') + expect(() => + conn.forwardOut({} as never, new EventEmitter(), '127.0.0.1', 0, 'remote', 443, vi.fn()) + ).toThrow('client_changed') + channel.emit('close') + expect(() => fence.assertDrained()).toThrow('not_drained') + socket.emit('close') + await fence.drain(new AbortController().signal) + expect(client.forwardOut).toHaveBeenCalledOnce() +}) + +it('tracks exact remote socket forwarding through a late open and physical close', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + const client = conn.getClient()! + let accept: Parameters[1] | undefined + client.openssh_forwardOutStreamLocal = vi.fn((_path, callback) => { + accept = callback + return client + }) + const received = vi.fn() + conn.forwardStreamLocal(client, '/saved/incumbent.sock', received) + expect(client.openssh_forwardOutStreamLocal).toHaveBeenCalledWith( + '/saved/incumbent.sock', + expect.any(Function) + ) + const fence = fenceSshConnectionWork(conn) + expect(() => fence.assertDrained()).toThrow('not_drained') + expect(() => conn.forwardStreamLocal(client, '/other.sock', vi.fn())).toThrow('admission_closed') + const channel = new EventEmitter() + accept!(undefined, channel as never) + expect(received).toHaveBeenCalledWith(undefined, channel) + expect(() => fence.assertDrained()).toThrow('not_drained') + channel.emit('close') + await fence.drain(new AbortController().signal) + expect(client.openssh_forwardOutStreamLocal).toHaveBeenCalledOnce() +}) + +it('rejects stale clients and non-Unix endpoints before remote socket forwarding', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + const client = conn.getClient()! + client.openssh_forwardOutStreamLocal = vi.fn() + expect(() => conn.forwardStreamLocal({} as never, '/saved.sock', vi.fn())).toThrow( + 'client_changed' + ) + for (const endpoint of ['relative.sock', '\\\\.\\pipe\\saved', '/saved\0.sock']) { + expect(() => conn.forwardStreamLocal(client, endpoint, vi.fn())).toThrow('endpoint_invalid') + } + expect(client.openssh_forwardOutStreamLocal).not.toHaveBeenCalled() +}) + +it('does not treat an uncertain remote socket open as drained', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + const client = conn.getClient()! + client.openssh_forwardOutStreamLocal = vi.fn(() => { + throw new Error('uncertain-open') + }) + expect(() => conn.forwardStreamLocal(client, '/saved.sock', vi.fn())).toThrow('uncertain-open') + const fence = fenceSshConnectionWork(conn) + await expect(fence.drain(new AbortController().signal)).rejects.toThrow('uncertain-open') +}) + +it('surfaces remote socket extension refusal without a fallback', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + const client = conn.getClient()! + let accept: Parameters[1] | undefined + client.openssh_forwardOutStreamLocal = vi.fn((_path, callback) => { + accept = callback + return client + }) + const received = vi.fn() + conn.forwardStreamLocal(client, '/saved.sock', received) + const fence = fenceSshConnectionWork(conn) + const refused = new Error('extension unsupported') + accept!(refused, undefined as never) + expect(received).toHaveBeenCalledWith(refused, undefined) + await expect(fence.drain(new AbortController().signal)).rejects.toThrow('extension unsupported') + expect(client.openssh_forwardOutStreamLocal).toHaveBeenCalledOnce() +}) + +it('retains remote socket channel errors through physical close', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + const client = conn.getClient()! + const channel = new EventEmitter() + client.openssh_forwardOutStreamLocal = vi.fn((_path, callback) => { + callback(undefined, channel as never) + return client + }) + conn.forwardStreamLocal(client, '/saved.sock', vi.fn()) + const fence = fenceSshConnectionWork(conn) + channel.emit('error', new Error('connection lost')) + channel.emit('close') + await expect(fence.drain(new AbortController().signal)).rejects.toThrow('connection lost') +}) + +it('keeps the upload session SFTP channel tracked after logical close', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + ssh2Mock.sftpBehavior = 'pending' + const opening = conn.openFileUploadSession() + const fence = fenceSshConnectionWork(conn) + const channel = Object.assign(new EventEmitter(), { end: vi.fn() }) + pendingSftpCallback?.(undefined, channel) + const session = await opening + session.close() + expect(channel.end).toHaveBeenCalledOnce() + expect(() => fence.assertDrained()).toThrow('not_drained') + channel.emit('close') + await fence.drain(new AbortController().signal) + fence.assertDrained() +}) + +it('accounts for an SFTP open begun before fencing until its channel physically closes', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + ssh2Mock.sftpBehavior = 'pending' + const opening = conn.sftp() + const fence = fenceSshConnectionWork(conn) + const done = vi.fn() + const draining = fence.drain(new AbortController().signal).then(done) + const channel = Object.assign(new EventEmitter(), { end: vi.fn() }) + pendingSftpCallback?.(undefined, channel) + await opening + expect(done).not.toHaveBeenCalled() + expect(channel.end).not.toHaveBeenCalled() + await expect(conn.sftp()).rejects.toThrow('admission_closed') + channel.emit('close') + await draining + fence.assertDrained() +}) + +it('retains an opening timeout as unverifiable and never force-closes it to make drain pass', async () => { + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + ssh2Mock.sftpBehavior = 'pending' + vi.useFakeTimers() + try { + const opening = conn.sftp().catch((error: unknown) => error) + const fence = fenceSshConnectionWork(conn) + const draining = expect(fence.drain(new AbortController().signal)).rejects.toThrow('timed out') + await vi.advanceTimersByTimeAsync(30_000) + await opening + await draining + await expect(fence.drain(new AbortController().signal)).rejects.toThrow('timed out') + expect(conn.getState().status).toBe('connected') + } finally { + vi.useRealTimers() + } +}) diff --git a/src/main/ssh/ssh-connection.ts b/src/main/ssh/ssh-connection.ts index f75c4b4f172b..789cd6efdc01 100644 --- a/src/main/ssh/ssh-connection.ts +++ b/src/main/ssh/ssh-connection.ts @@ -1,6 +1,20 @@ /* eslint-disable max-lines -- Why: SSH connection lifecycle, credential retries, reconnect policy, and transport fallback are intentionally co-located so state transitions stay auditable in one file. */ import * as net from 'node:net' import { Client as SshClient } from 'ssh2' +import { sshProxyRouteDigest, type SshConnectionDestination } from './ssh-connection-destination' +import { SshConnectionWorkLedger } from './ssh-connection-work-ledger' +import { SshTransportCloseLedger } from './ssh-transport-close-ledger' +import { SshTransportClosureSubscribers } from './ssh-connection-transport-closure' +import { disconnectAndAwaitSshTransportClose } from './ssh-connection-close-drain' +import { openTrackedSshUploadSession } from './ssh-upload-session-lifetime' +import { + forwardTrackedSshChannel, + forwardTrackedSshStreamLocalChannel +} from './ssh-forward-channel-lifetime' +import { + openTrackedSshSocket, + trackSshConnectionChannelLifetime +} from './ssh-connection-channel-lifetime' import type { ChildProcess } from 'node:child_process' import type { ClientChannel, @@ -200,6 +214,37 @@ export class SshConnection { private useSystemSshTransport = false private credentialAbortController = new AbortController() private readonly pendingSsh2Clients = new Set() + // Local lifetime evidence only: a closed channel or client never proves a remote process exited. + private readonly workLedger = new SshConnectionWorkLedger( + () => this.transportClosure.notify(), + () => { + if (this.disposed) { + throw createSshOperationAbortError() + } + } + ) + private readonly transportCloseLedger = new SshTransportCloseLedger(() => + this.transportClosure.notify() + ) + private readonly transportClosure = new SshTransportClosureSubscribers( + () => + this.disposed && + this.state.status === 'disconnected' && + this.activeConnectCalls === 0 && + this.pendingSsh2Clients.size === 0 && + !this.unprovenStartupTransport && + this.transportCloseLedger.isClosed() && + this.workLedger.isDrained() + ) + private readonly automaticReconnect: boolean + // Set once drained for close; no reconnect or new transport may replace the fenced one. + private resetConnectionFenced = false + private activeConnectCalls = 0 + // Sticky: a system-ssh child's transport closure cannot be proven from here. + private unprovenStartupTransport = false + private executionDestination: + | { client: SshClient; generation: number; destination: SshConnectionDestination } + | undefined private state: SshConnectionState private callbacks: SshConnectionCallbacks private target: SshTarget @@ -214,7 +259,12 @@ export class SshConnection { private hostKeyFingerprint: string | undefined private connectGeneration = 0 - constructor(target: SshTarget, callbacks: SshConnectionCallbacks) { + constructor( + target: SshTarget, + callbacks: SshConnectionCallbacks, + options: { automaticReconnect?: boolean } = {} + ) { + this.automaticReconnect = options.automaticReconnect !== false this.target = target this.callbacks = callbacks this.state = { @@ -232,9 +282,59 @@ export class SshConnection { getConnectGeneration(): number { return this.connectGeneration } + /** Alias for callers that compare transport incarnations rather than connect attempts. */ + getTransportGeneration(): number { + return this.connectGeneration + } + /** The ssh2 destination proven by the current handshake; undefined on system SSH or when stale. */ + getExecutionDestination(): SshConnectionDestination | undefined { + const captured = this.executionDestination + return !this.disposed && + !this.useSystemSshTransport && + this.state.status === 'connected' && + captured?.client === this.client && + captured?.generation === this.connectGeneration + ? captured.destination + : undefined + } + subscribeTransportClosure(onClosed: () => void): () => void { + return this.transportClosure.subscribe(onClosed) + } getClient(): SshClient | null { return this.client } + prepareForwardRoute(prepare: () => Promise): Promise { + return this.workLedger.run(prepare) + } + openForwardSocket(open: () => T): T { + return openTrackedSshSocket(this.workLedger, open) + } + forwardOut( + client: SshClient, + localSocket: object, + ...args: Parameters + ): void { + if (this.client !== client) { + throw new Error('ssh_connection_forward_client_changed') + } + forwardTrackedSshChannel(this.workLedger, client, localSocket, ...args) + } + forwardStreamLocal( + client: SshClient, + ...args: Parameters + ): void { + if (this.client !== client) { + throw new Error('ssh_connection_forward_client_changed') + } + if (this.disposed || this.useSystemSshTransport || this.state.status !== 'connected') { + throw new Error('ssh_connection_streamlocal_transport_unavailable') + } + const [socketPath] = args + if (!socketPath.startsWith('/') || socketPath.includes('\0')) { + throw new Error('ssh_connection_streamlocal_endpoint_invalid') + } + forwardTrackedSshStreamLocalChannel(this.workLedger, client, ...args) + } usesSystemSshTransport(): boolean { return this.useSystemSshTransport } @@ -269,7 +369,11 @@ export class SshConnection { return this.cachedPassphrase != null || this.cachedPassword != null } - async exec(cmd: string, options?: SshExecOptions): Promise { + exec(cmd: string, options?: SshExecOptions): Promise { + return this.workLedger.run(() => this.execUntracked(cmd, options)) + } + + private async execUntracked(cmd: string, options?: SshExecOptions): Promise { if (options?.signal?.aborted) { throw createSshOperationAbortError() } @@ -298,7 +402,15 @@ export class SshConnection { } /** Interactive login shell over a session channel with pty-req; ssh2 transport only. */ - async shell(pty: PseudoTtyOptions, options: ShellOptions = {}): Promise { + shell(pty: PseudoTtyOptions, options: ShellOptions = {}): Promise { + // Plain SSH mode's terminals must drain with the rest of this connection's work. + return this.workLedger.run(() => this.shellUntracked(pty, options)) + } + + private async shellUntracked( + pty: PseudoTtyOptions, + options: ShellOptions + ): Promise { if (this.useSystemSshTransport) { throw new Error('Interactive SSH shells are not available when using system SSH transport') } @@ -315,7 +427,13 @@ export class SshConnection { ) } - async sftp(options?: AbortSignal | { signal?: AbortSignal }): Promise { + sftp(options?: AbortSignal | { signal?: AbortSignal }): Promise { + return this.workLedger.run(() => this.sftpUntracked(options)) + } + + private async sftpUntracked( + options?: AbortSignal | { signal?: AbortSignal } + ): Promise { // Why: relay transfers pass a signal directly, while filesystem factories use an options object. const signal = options && 'aborted' in options ? options : options?.signal if (signal?.aborted) { @@ -382,11 +500,13 @@ export class SshConnection { signal?: AbortSignal, trackRemoteCommandTermination = false ): Promise { + const work = this.workLedger.beginChannelOpen() return new Promise((resolve, reject) => { type ChannelOpenTerminationError = Error & { sshChannelCloseConfirmed: boolean } let settled = false let unconfirmedOpenError: ChannelOpenTerminationError | null = null const markOpenUnconfirmed = (error: Error): Error => { + work.markUnverifiable(error) if (!trackRemoteCommandTermination) { return error } @@ -473,6 +593,12 @@ export class SshConnection { } } const finish = (error: Error | undefined, value?: T): void => { + // Late channels stay tracked until physical close, after the caller has given up. + if (!error && value !== undefined) { + trackSshConnectionChannelLifetime(work, value) + } else { + work.close(error) + } if (settled) { // Why: ssh2 can invoke the open callback after our timeout rejected; close that late channel so it isn't left open with no owner. if (!error && value !== undefined) { @@ -503,6 +629,7 @@ export class SshConnection { if (signal?.aborted) { // No open is in flight yet, so failing fast leaks nothing. cleanup() + work.close() reject(createSshOperationAbortError()) return } @@ -517,7 +644,15 @@ export class SshConnection { }) } - async uploadDirectory( + uploadDirectory( + localDir: string, + remoteDir: string, + options?: SshRemoteFileOptions & { signal?: AbortSignal } + ): Promise { + return this.workLedger.run(() => this.uploadDirectoryUntracked(localDir, remoteDir, options)) + } + + private async uploadDirectoryUntracked( localDir: string, remoteDir: string, options?: SshRemoteFileOptions & { signal?: AbortSignal } @@ -571,7 +706,15 @@ export class SshConnection { } } - async downloadFile( + downloadFile( + remotePath: string, + localPath: string, + options?: SshRemoteFileOptions + ): Promise { + return this.workLedger.run(() => this.downloadFileUntracked(remotePath, localPath, options)) + } + + private async downloadFileUntracked( remotePath: string, localPath: string, options?: SshRemoteFileOptions @@ -593,7 +736,13 @@ export class SshConnection { }) } - async openFileUploadSession(options?: SshRemoteFileOptions): Promise { + openFileUploadSession(options?: SshRemoteFileOptions): Promise { + return openTrackedSshUploadSession(this.workLedger, () => this.createFileUploadSession(options)) + } + + private async createFileUploadSession( + options?: SshRemoteFileOptions + ): Promise { if (!this.useSystemSshTransport) { const sftp = await this.sftp() const { uploadFile } = await import('./sftp-upload') @@ -618,7 +767,15 @@ export class SshConnection { } } - async writeFile( + writeFile( + remotePath: string, + contents: string, + options?: SshRemoteFileOptions & { signal?: AbortSignal } + ): Promise { + return this.workLedger.run(() => this.writeFileUntracked(remotePath, contents, options)) + } + + private async writeFileUntracked( remotePath: string, contents: string, options?: SshRemoteFileOptions & { signal?: AbortSignal } @@ -670,7 +827,15 @@ export class SshConnection { } } - async writeBuffer( + writeBuffer( + remotePath: string, + contents: Buffer, + options?: SshRemoteFileOptions & { append?: boolean; exclusive?: boolean } + ): Promise { + return this.workLedger.run(() => this.writeBufferUntracked(remotePath, contents, options)) + } + + private async writeBufferUntracked( remotePath: string, contents: Buffer, options?: SshRemoteFileOptions & { append?: boolean; exclusive?: boolean } @@ -695,6 +860,23 @@ export class SshConnection { } async connect(): Promise { + this.activeConnectCalls++ + try { + await this.connectInitialAttempts() + } finally { + this.activeConnectCalls-- + this.transportClosure.notify() + } + } + + private assertConnectionReplacementAllowed(): void { + if (this.resetConnectionFenced) { + throw new Error('ssh_connection_reset_replacement_refused') + } + } + + private async connectInitialAttempts(): Promise { + this.assertConnectionReplacementAllowed() if (this.disposed) { throw new Error('Connection disposed') } @@ -702,6 +884,7 @@ export class SshConnection { let lastError: Error | null = null for (let attempt = 0; attempt < INITIAL_RETRY_ATTEMPTS; attempt++) { + this.assertConnectionReplacementAllowed() const connectGeneration = ++this.connectGeneration try { await this.attemptConnect(connectGeneration) @@ -819,6 +1002,7 @@ export class SshConnection { } private async attemptConnect(connectGeneration = ++this.connectGeneration): Promise { + this.executionDestination = undefined this.credentialAbortController.abort() this.credentialAbortController = new AbortController() this.setState('connecting') @@ -867,8 +1051,10 @@ export class SshConnection { // Why: ssh2 doesn't support ProxyCommand/ProxyJump natively; spawn the resolved proxy and pipe its stdin/stdout as config.sock. const effectiveProxy = resolveEffectiveProxy(this.target, resolved) + const proxyRouteDigest = sshProxyRouteDigest(effectiveProxy) if (effectiveProxy) { const proxy = spawnProxyCommand(effectiveProxy, config.host!, config.port!, config.username!) + this.transportCloseLedger.track(proxy.process) this.proxyProcess = proxy.process config.sock = proxy.sock } @@ -881,7 +1067,7 @@ export class SshConnection { } try { - await this.doSsh2Connect(config, connectGeneration) + await this.doSsh2Connect(config, connectGeneration, proxyRouteDigest) } catch (err) { if (!(err instanceof Error)) { this.proxyProcess?.kill() @@ -957,7 +1143,7 @@ export class SshConnection { if (keyConfig.privateKey || keyConfig.password) { this.respawnProxy(keyConfig, effectiveProxy) try { - await this.doSsh2Connect(keyConfig, connectGeneration) + await this.doSsh2Connect(keyConfig, connectGeneration, proxyRouteDigest) return } catch (keyErr) { // Same reason as above: the retry re-runs the handshake, so it can be the attempt that @@ -992,7 +1178,7 @@ export class SshConnection { this.cachedPassphrase = val keyConfig.passphrase = val this.respawnProxy(keyConfig, effectiveProxy) - await this.doSsh2Connect(keyConfig, connectGeneration) + await this.doSsh2Connect(keyConfig, connectGeneration, proxyRouteDigest) return } } @@ -1042,7 +1228,7 @@ export class SshConnection { this.cachedPassphrase = val credentialRetryConfig.passphrase = val this.respawnProxy(credentialRetryConfig, effectiveProxy) - await this.doSsh2Connect(credentialRetryConfig, connectGeneration) + await this.doSsh2Connect(credentialRetryConfig, connectGeneration, proxyRouteDigest) return } } @@ -1057,7 +1243,7 @@ export class SshConnection { this.cachedPassword = val credentialRetryConfig.password = val this.respawnProxy(credentialRetryConfig, effectiveProxy) - await this.doSsh2Connect(credentialRetryConfig, connectGeneration) + await this.doSsh2Connect(credentialRetryConfig, connectGeneration, proxyRouteDigest) return } } @@ -1068,6 +1254,7 @@ export class SshConnection { } async reconnect(): Promise { + this.assertConnectionReplacementAllowed() if (this.disposed || this.state.status === 'connecting') { return } @@ -1084,6 +1271,7 @@ export class SshConnection { } private async doSystemSshProbe(connectGeneration: number): Promise { + this.unprovenStartupTransport = true this.useSystemSshTransport = true this.client = null this.proxyProcess?.kill() @@ -1244,6 +1432,7 @@ export class SshConnection { if (!this.isCurrentConnectAttempt(connectGeneration)) { throw this.createCancelledConnectAttemptError() } + this.unprovenStartupTransport = true const proc = spawnSystemSsh(this.target, this.getSystemSshBuildArgsOptions()) this.systemSsh = proc let settled = false @@ -1334,10 +1523,18 @@ export class SshConnection { options === undefined && Object.keys(buildArgsOptions).length === 0 ? undefined : { ...options, ...buildArgsOptions } - const channel = - commandOptions === undefined - ? spawnSystemSshCommand(this.target, command) - : spawnSystemSshCommand(this.target, command, commandOptions) + const work = this.workLedger.beginChannelOpen() + let channel: ClientChannel + try { + channel = + commandOptions === undefined + ? spawnSystemSshCommand(this.target, command) + : spawnSystemSshCommand(this.target, command, commandOptions) + trackSshConnectionChannelLifetime(work, channel) + } catch (error) { + work.close(error) + throw error + } this.systemCommandChannels.add(channel) const onAbort = (): void => { channel.close() @@ -1381,6 +1578,7 @@ export class SshConnection { } this.proxyProcess?.kill() const p = spawnProxyCommand(proxy, config.host!, config.port!, config.username!) + this.transportCloseLedger.track(p.process) this.proxyProcess = p.process config.sock = p.sock } @@ -1433,7 +1631,16 @@ export class SshConnection { return sources } - private async doSsh2Connect(config: ConnectConfig, connectGeneration: number): Promise { + private async doSsh2Connect( + config: ConnectConfig, + connectGeneration: number, + proxyRouteDigest: string + ): Promise { + const endpoint = { + host: config.host ?? '', + port: config.port ?? 22, + username: config.username ?? '' + } const hostKeyResolved = this.hostKeyResolvedConfig const { host: hostKeyLookupHost, isHostKeyAlias } = resolveKnownHostsLookupHost( hostKeyResolved, @@ -1476,6 +1683,7 @@ export class SshConnection { : undefined return new Promise((resolve, reject) => { const client = new SshClient() + this.transportCloseLedger.track(client) this.pendingSsh2Clients.add(client) let settled = false let startupTimer: ReturnType | null = null @@ -1498,6 +1706,7 @@ export class SshConnection { // the live attempt's error, and substituting a new Error drops ssh2's `code`, so a transient // ECONNRESET would stop being classified as retryable. let hostKeyRejection: HostKeyVerificationError | null = null + let acceptedFingerprint: string | undefined // Why the fingerprint is still recorded: the relay uses the negotiated server key to isolate // shared-home install locks without comparing PIDs from an unrelated SSH host. Its format is @@ -1549,6 +1758,7 @@ export class SshConnection { connectGeneration === this.connectGeneration ) { this.hostKeyFingerprint = decision.fingerprint + acceptedFingerprint = decision.fingerprint } if (decision.action === 'reject') { hostKeyRejection = new HostKeyVerificationError( @@ -1647,6 +1857,20 @@ export class SshConnection { } settled = true this.client = client + // Only this successful handshake can publish recovery destination authority. + this.executionDestination = acceptedFingerprint + ? { + client, + generation: connectGeneration, + destination: Object.freeze({ + version: 1, + transport: 'ssh2', + ...endpoint, + hostKeyFingerprint: acceptedFingerprint, + proxyRouteDigest + }) + } + : undefined this.proxyProcess = null this.setupDisconnectHandler(client) cleanupStartupListeners() @@ -1704,6 +1928,7 @@ export class SshConnection { if (this.disposed || this.client !== client) { return } + this.workLedger.markTransportUnverifiable() this.client = null this.scheduleReconnect() } @@ -1714,13 +1939,18 @@ export class SshConnection { return } console.warn(`[ssh] Connection error for ${this.target.label}: ${err.message}`) + this.workLedger.markTransportUnverifiable() this.client = null this.scheduleReconnect() }) } private scheduleReconnect(): void { - if (this.disposed || this.reconnectTimer) { + if (this.disposed || this.resetConnectionFenced || this.reconnectTimer) { + return + } + if (!this.automaticReconnect) { + this.setState('disconnected') return } const decision = this.reconnectLadder.next(Date.now()) @@ -1735,7 +1965,7 @@ export class SshConnection { ) this.reconnectTimer = setTimeout(async () => { this.reconnectTimer = null - if (this.disposed) { + if (this.disposed || this.resetConnectionFenced) { return } await this.runReconnectAttempt() @@ -1743,6 +1973,19 @@ export class SshConnection { } private async runReconnectAttempt(): Promise { + this.activeConnectCalls++ + try { + await this.runTrackedReconnectAttempt() + } finally { + this.activeConnectCalls-- + this.transportClosure.notify() + } + } + + private async runTrackedReconnectAttempt(): Promise { + if (this.resetConnectionFenced) { + return + } const connectGeneration = ++this.connectGeneration try { // Why: reset before connecting so the 'connected' broadcast carries reconnectAttempt=0, which ssh.ts uses to trigger relay re-establishment. @@ -1818,9 +2061,21 @@ export class SshConnection { } async connectViaSystemSsh(): Promise { + this.activeConnectCalls++ + try { + return await this.connectTrackedViaSystemSsh() + } finally { + this.activeConnectCalls-- + this.transportClosure.notify() + } + } + + private async connectTrackedViaSystemSsh(): Promise { + this.assertConnectionReplacementAllowed() if (this.disposed) { throw new Error('Connection disposed') } + this.unprovenStartupTransport = true const connectGeneration = ++this.connectGeneration this.systemSsh?.kill() this.systemSsh = null @@ -1854,6 +2109,7 @@ export class SshConnection { // Why: register the reconnect handler only after handshake succeeds (the onExit above guards with `settled`). proc.onExit(() => { if (!this.disposed && this.systemSsh === proc) { + this.workLedger.markTransportUnverifiable() this.systemSsh = null this.scheduleReconnect() } @@ -1872,7 +2128,60 @@ export class SshConnection { } } + /** Owned migration transport only; local closure never proves remote process exit. */ + async disconnectAndDrain(signal: AbortSignal): Promise { + signal.throwIfAborted() + // An unconnected non-reconnecting owner has nothing live left to fence. + const settledExclusiveStartup = + !this.automaticReconnect && + this.activeConnectCalls === 0 && + this.pendingSsh2Clients.size === 0 && + !this.unprovenStartupTransport + if ( + this.useSystemSshTransport || + this.unprovenStartupTransport || + this.activeConnectCalls > 0 || + (!this.client && !settledExclusiveStartup) + ) { + await this.disconnect() + throw new Error('ssh_connection_close_transport_unproven') + } + let fence: ReturnType + try { + if ( + this.client && + (this.disposed || this.state.status !== 'connected' || this.pendingSsh2Clients.size > 0) + ) { + throw new Error('ssh_connection_reset_transport_not_connected') + } + fence = this.workLedger.fenceForReset() + this.resetConnectionFenced = true + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer) + this.reconnectTimer = null + } + } catch (error) { + try { + await this.disconnect() + } catch (cleanupError) { + throw new AggregateError([error, cleanupError], 'ssh_connection_close_cleanup_failed') + } + throw error + } + await disconnectAndAwaitSshTransportClose({ + liveResources: [ + ...(this.client ? [this.client] : []), + ...(this.proxyProcess ? [this.proxyProcess] : []) + ], + disconnect: () => this.disconnect(), + transportCloseLedger: this.transportCloseLedger, + fence, + signal + }) + } + async disconnect(): Promise { + this.executionDestination = undefined this.disposed = true this.connectGeneration += 1 if (this.reconnectTimer) { @@ -1901,6 +2210,7 @@ export class SshConnection { this.useSystemSshTransport = false this.reconnectLadder.reset() this.setState('disconnected') + this.transportClosure.notify() } private setState(status: SshConnectionStatus, error?: string): void { diff --git a/src/main/ssh/ssh-forward-channel-lifetime.test.ts b/src/main/ssh/ssh-forward-channel-lifetime.test.ts new file mode 100644 index 000000000000..ed0d6193d420 --- /dev/null +++ b/src/main/ssh/ssh-forward-channel-lifetime.test.ts @@ -0,0 +1,98 @@ +import { EventEmitter } from 'node:events' +import type { Client, ClientChannel } from 'ssh2' +import { expect, it, vi } from 'vitest' +import { SshConnectionWorkLedger } from './ssh-connection-work-ledger' +import { forwardTrackedSshChannel } from './ssh-forward-channel-lifetime' + +function fixture() { + const ledger = new SshConnectionWorkLedger() + const local = new EventEmitter() + const remote = new EventEmitter() as ClientChannel + let opened!: NonNullable[4]> + const client = { + forwardOut: vi.fn((_a, _b, _c, _d, callback) => { + opened = callback + }) + } + const callback = vi.fn() + const start = () => + forwardTrackedSshChannel( + ledger, + client as unknown as Client, + local, + '127.0.0.1', + 0, + 'remote.internal', + 443, + callback + ) + return { ledger, local, remote, client, callback, start, opened: () => opened } +} + +it('tracks a pending open and both physical ends without terminating either', async () => { + const f = fixture() + f.start() + const fence = f.ledger.fenceForReset() + expect(fence.assertDrained).toThrow('not_drained') + f.opened()(undefined, f.remote) + expect(f.callback).toHaveBeenCalledWith(undefined, f.remote) + f.remote.emit('close') + expect(fence.assertDrained).toThrow('not_drained') + f.local.emit('close') + await fence.drain(new AbortController().signal) +}) + +it('retains late channels after the local socket closes', async () => { + const f = fixture() + f.start() + f.local.emit('close') + const fence = f.ledger.fenceForReset() + f.opened()(undefined, f.remote) + expect(fence.assertDrained).toThrow('not_drained') + f.remote.emit('close') + await fence.drain(new AbortController().signal) +}) + +it('refuses fenced opens before forwarding or subscribing to the new socket', () => { + const f = fixture() + f.ledger.fenceForReset() + expect(f.start).toThrow('admission_closed') + expect(f.client.forwardOut).not.toHaveBeenCalled() + expect(f.local.listenerCount('close')).toBe(0) +}) + +it('retains open failures during reset rather than claiming successful drain', async () => { + const f = fixture() + f.start() + const fence = f.ledger.fenceForReset() + const failure = new Error('forward failed') + f.opened()(failure, undefined as never) + f.local.emit('close') + await expect(fence.drain(new AbortController().signal)).rejects.toBe(failure) +}) + +it('retains synchronous open uncertainty even when reset begins later', async () => { + const f = fixture() + const failure = new Error('transport failed') + f.client.forwardOut.mockImplementation(() => { + throw failure + }) + expect(f.start).toThrow(failure) + f.local.emit('close') + await expect(f.ledger.fenceForReset().drain(new AbortController().signal)).rejects.toBe(failure) +}) + +it('aborting observation preserves the admitted sockets and allows another drain', async () => { + const f = fixture() + f.start() + const fence = f.ledger.fenceForReset() + const observer = new AbortController() + const draining = fence.drain(observer.signal).catch((error: unknown) => error) + const failure = new Error('observer stopped') + observer.abort(failure) + expect(await draining).toBe(failure) + f.opened()(undefined, f.remote) + f.local.emit('close') + f.remote.emit('close') + await fence.drain(new AbortController().signal) +}) diff --git a/src/main/ssh/ssh-forward-channel-lifetime.ts b/src/main/ssh/ssh-forward-channel-lifetime.ts new file mode 100644 index 000000000000..5530138fb47b --- /dev/null +++ b/src/main/ssh/ssh-forward-channel-lifetime.ts @@ -0,0 +1,50 @@ +import type { Client } from 'ssh2' +import { trackSshConnectionChannelLifetime } from './ssh-connection-channel-lifetime' +import type { SshConnectionWorkLedger } from './ssh-connection-work-ledger' + +/** The channel remains tracked until physical close, including late open replies. */ +export function forwardTrackedSshStreamLocalChannel( + ledger: SshConnectionWorkLedger, + client: Client, + ...args: Parameters +): void { + const opening = ledger.beginChannelOpen() + const [socketPath, callback] = args + try { + client.openssh_forwardOutStreamLocal(socketPath, (error, channel) => { + if (error) { + opening.close(error) + } else { + trackSshConnectionChannelLifetime(opening, channel) + } + callback(error, channel) + }) + } catch (error) { + opening.markUnverifiable(error instanceof Error ? error : new Error(String(error))) + throw error + } +} + +export function forwardTrackedSshChannel( + ledger: SshConnectionWorkLedger, + client: Client, + localSocket: object, + ...args: Parameters +): void { + const opening = ledger.beginChannelOpen() + trackSshConnectionChannelLifetime(ledger.beginChannelOpen(), localSocket) + const [sourceHost, sourcePort, host, port, callback] = args + try { + client.forwardOut(sourceHost, sourcePort, host, port, (error, channel) => { + if (error) { + opening.close(error) + } else { + trackSshConnectionChannelLifetime(opening, channel) + } + callback?.(error, channel) + }) + } catch (error) { + opening.markUnverifiable(error instanceof Error ? error : new Error(String(error))) + throw error + } +} diff --git a/src/main/ssh/ssh-multi-factor-authentication.test.ts b/src/main/ssh/ssh-multi-factor-authentication.test.ts index 2ee643dea7a9..4b22c9e6157e 100644 --- a/src/main/ssh/ssh-multi-factor-authentication.test.ts +++ b/src/main/ssh/ssh-multi-factor-authentication.test.ts @@ -1,5 +1,6 @@ import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' -import { tmpdir } from 'node:os' +import { homedir, tmpdir } from 'node:os' +import type * as Os from 'node:os' import { join } from 'node:path' import { Client, @@ -10,11 +11,16 @@ import { type KeyboardAuthContext, type PasswordAuthContext } from 'ssh2' -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { SshTarget } from '../../shared/ssh-types' import type { SshResolvedConfig } from './ssh-config-parser' import { buildConnectConfig } from './ssh-connection-utils' +vi.mock('node:os', async (importOriginal) => { + const actual = await importOriginal() + return { ...actual, homedir: vi.fn(() => actual.tmpdir()) } +}) + // OpenSSH's default; a host that burns it disconnects before the MFA stage is reached. const MAX_AUTH_TRIES = 6 const PASSWORD = 'stage-one-password' @@ -186,19 +192,11 @@ function connectWithOrcaConfig( describe('multi-stage SSH authentication', () => { let tempDir: string let keyPaths: string[] - let homeEnv: { HOME?: string; USERPROFILE?: string } beforeEach(() => { tempDir = mkdtempSync(join(tmpdir(), 'orca-mfa-')) - // Why: the cases below pass `resolved: null`, so `resolvePrivateKeys` falls through to - // `findDefaultKeyFile`, which reads `~/.ssh/id_*` through `homedir()`. On a developer - // machine that picks up a real key, and an encrypted one makes ssh2 reject with - // "Cannot parse privateKey" before authentication is exercised at all. Hosted CI has no - // key, so this only ever failed locally. Pointing home at the fixture directory keeps - // default-key discovery inside the test's control on every machine. - homeEnv = { HOME: process.env.HOME, USERPROFILE: process.env.USERPROFILE } - process.env.HOME = tempDir - process.env.USERPROFILE = tempDir + // Default-key discovery must never read the developer's SSH credentials. + vi.mocked(homedir).mockReturnValue(tempDir) keyPaths = ['id_a', 'id_b'].map((name) => { const path = join(tempDir, name) writeFileSync(path, utils.generateKeyPairSync('ecdsa', { bits: 256 }).private) @@ -207,14 +205,7 @@ describe('multi-stage SSH authentication', () => { }) afterEach(() => { - for (const key of ['HOME', 'USERPROFILE'] as const) { - const previous = homeEnv[key] - if (previous === undefined) { - delete process.env[key] - } else { - process.env[key] = previous - } - } + vi.mocked(homedir).mockReturnValue(tmpdir()) rmSync(tempDir, { recursive: true, force: true }) }) diff --git a/src/main/ssh/ssh-relay-exec-command.ts b/src/main/ssh/ssh-relay-exec-command.ts index e7d5fea353ce..10b2b4c37634 100644 --- a/src/main/ssh/ssh-relay-exec-command.ts +++ b/src/main/ssh/ssh-relay-exec-command.ts @@ -47,7 +47,7 @@ export function isUnconfirmedSshCommandTermination( } export async function execCommand( - conn: SshConnection, + conn: Pick, command: string, options?: ExecCommandOptions ): Promise { diff --git a/src/main/ssh/ssh-relay-exec-input.test.ts b/src/main/ssh/ssh-relay-exec-input.test.ts new file mode 100644 index 000000000000..c5b6ffdeeb1a --- /dev/null +++ b/src/main/ssh/ssh-relay-exec-input.test.ts @@ -0,0 +1,57 @@ +import { EventEmitter } from 'node:events' +import { PassThrough, Readable } from 'node:stream' +import { describe, expect, it, vi } from 'vitest' +import { execCommand } from './ssh-relay-exec-command' + +const { spawnMock } = vi.hoisted(() => ({ spawnMock: vi.fn() })) +vi.mock('node:child_process', () => ({ spawn: spawnMock })) +vi.mock('./system-ssh-binary', () => ({ findSystemSsh: () => '/usr/bin/ssh' })) +import { spawnSystemSshCommand } from './system-ssh-command' + +function fakeSshProcess() { + const proc = Object.assign(new EventEmitter(), { + stdin: new PassThrough(), + stdout: new PassThrough(), + stderr: new PassThrough(), + kill: vi.fn() + }) + spawnMock.mockReturnValue(proc) + let input = '' + proc.stdin.on('data', (chunk) => { + input += chunk.toString() + }) + proc.stdin.on('finish', () => { + proc.stdout.write('answer') + proc.emit('close', 0) + }) + return { proc, input: () => input } +} + +const target = { id: 't', label: 't', host: 'example.test', port: 22, username: 'u' } + +describe('system SSH command input', () => { + it('forwards end-of-input written to the facade to the ssh child', async () => { + const { proc, input } = fakeSshProcess() + const channel = spawnSystemSshCommand(target, 'reader') + const finished = new Promise((resolve) => proc.stdin.once('finish', resolve)) + + channel.end('secret') + await finished + + expect(input()).toBe('secret') + expect(proc.stdin.writableFinished).toBe(true) + }) + + it('streams command input through the real facade and returns the response', async () => { + const { proc, input } = fakeSshProcess() + const channel = spawnSystemSshCommand(target, 'reader') + + await expect( + execCommand({ exec: async () => channel, usesSystemSshTransport: () => true }, 'reader', { + stdin: Readable.from(['secret']) + }) + ).resolves.toBe('answer') + expect(input()).toBe('secret') + expect(proc.stdin.writableFinished).toBe(true) + }) +}) diff --git a/src/main/ssh/ssh-system-transport.integration.test.ts b/src/main/ssh/ssh-system-transport.integration.test.ts index 6e45077698a5..ac3ff4f03088 100644 --- a/src/main/ssh/ssh-system-transport.integration.test.ts +++ b/src/main/ssh/ssh-system-transport.integration.test.ts @@ -1,5 +1,5 @@ import { mkdtempSync, writeFileSync, mkdirSync, chmodSync, rmSync, readFileSync } from 'node:fs' -import { join } from 'node:path' +import { dirname, join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' vi.mock('electron', () => ({ @@ -59,7 +59,9 @@ function writeFakeRelay(dir: string): void { // below — so adding one cannot silently fail the completeness probe here. for (const filename of relayArtifactFilenames(false)) { if (filename !== 'relay.js') { - writeFileSync(join(dir, filename), '') + const artifactPath = join(dir, filename) + mkdirSync(dirname(artifactPath), { recursive: true }) + writeFileSync(artifactPath, '') } } writeFileSync( diff --git a/src/main/ssh/ssh-upload-session-lifetime.test.ts b/src/main/ssh/ssh-upload-session-lifetime.test.ts new file mode 100644 index 000000000000..25a27b9caddd --- /dev/null +++ b/src/main/ssh/ssh-upload-session-lifetime.test.ts @@ -0,0 +1,192 @@ +import { expect, it, vi } from 'vitest' +import type { FileUploadSession } from '../providers/filesystem-provider-contract' +import { SshConnectionWorkLedger } from './ssh-connection-work-ledger' +import { openTrackedSshUploadSession } from './ssh-upload-session-lifetime' + +const signal = () => new AbortController().signal +const makeSession = (): FileUploadSession => ({ + uploadFile: vi.fn(async () => undefined), + close: vi.fn() +}) + +it('holds an idle session until explicit close and closes the underlying session once', async () => { + const ledger = new SshConnectionWorkLedger() + const underlying = makeSession() + const session = await openTrackedSshUploadSession(ledger, async () => underlying) + const fence = ledger.fenceForReset() + expect(fence.assertDrained).toThrow('ssh_connection_work_not_drained') + session.close() + session.close() + await fence.drain(signal()) + expect(underlying.close).toHaveBeenCalledTimes(1) + await expect(session.uploadFile('source', 'destination')).rejects.toThrow( + 'ssh_connection_work_session_closed' + ) + expect(underlying.uploadFile).not.toHaveBeenCalled() +}) + +it('waits for every concurrent upload after close', async () => { + const ledger = new SshConnectionWorkLedger() + const first = Promise.withResolvers() + const second = Promise.withResolvers() + const underlying = makeSession() + underlying.uploadFile = vi + .fn() + .mockReturnValueOnce(first.promise) + .mockReturnValueOnce(second.promise) + const session = await openTrackedSshUploadSession(ledger, async () => underlying) + const uploadingFirst = session.uploadFile('a', 'b') + const uploadingSecond = session.uploadFile('c', 'd') + const fence = ledger.fenceForReset() + session.close() + first.resolve() + await uploadingFirst + expect(fence.assertDrained).toThrow('ssh_connection_work_not_drained') + second.resolve() + await uploadingSecond + await fence.drain(signal()) + fence.assertDrained() +}) + +it('allows admitted multi-file sessions to continue after fencing but refuses new sessions', async () => { + const ledger = new SshConnectionWorkLedger() + const underlying = makeSession() + underlying.uploadFile = vi.fn(async () => { + await ledger.run(async () => undefined) + }) + const session = await openTrackedSshUploadSession(ledger, async () => underlying) + await session.uploadFile('a', 'b') + const fence = ledger.fenceForReset() + const unopened = vi.fn(async () => makeSession()) + await expect(openTrackedSshUploadSession(ledger, unopened)).rejects.toThrow( + 'ssh_connection_work_admission_closed' + ) + expect(unopened).not.toHaveBeenCalled() + await session.uploadFile('c', 'd', { exclusive: true }) + expect(underlying.uploadFile).toHaveBeenLastCalledWith('c', 'd', { exclusive: true }) + session.close() + await fence.drain(signal()) +}) + +it('retains pending factory admission across the fence and independently tracks physical channels', async () => { + const ledger = new SshConnectionWorkLedger() + const release = Promise.withResolvers() + let channel!: ReturnType + const opening = openTrackedSshUploadSession(ledger, async () => { + await release.promise + await ledger.run(async () => { + channel = ledger.beginChannelOpen() + channel.bind({}) + }) + return makeSession() + }) + const fence = ledger.fenceForReset() + expect(fence.assertDrained).toThrow('ssh_connection_work_not_drained') + release.resolve() + const session = await opening + session.close() + expect(fence.assertDrained).toThrow('ssh_connection_work_not_drained') + channel.close() + await fence.drain(signal()) +}) + +it('retains failed factories that settle after fencing', async () => { + const ledger = new SshConnectionWorkLedger() + const factory = Promise.withResolvers() + const failure = new Error('factory failed') + const opening = openTrackedSshUploadSession(ledger, () => factory.promise).catch( + (error: unknown) => error + ) + const fence = ledger.fenceForReset() + const draining = fence.drain(signal()).catch((error: unknown) => error) + factory.reject(failure) + expect(await opening).toBe(failure) + expect(await draining).toBe(failure) + expect(fence.assertDrained).toThrow(failure) +}) + +it('wakes reset on upload failure even while the held session remains open', async () => { + const ledger = new SshConnectionWorkLedger() + const upload = Promise.withResolvers() + const underlying = makeSession() + underlying.uploadFile = () => upload.promise + const session = await openTrackedSshUploadSession(ledger, async () => underlying) + const uploading = session.uploadFile('a', 'b').catch((error: unknown) => error) + const fence = ledger.fenceForReset() + const draining = fence.drain(signal()).catch((error: unknown) => error) + const failure = new Error('upload failed') + upload.reject(failure) + expect(await uploading).toBe(failure) + expect(await draining).toBe(failure) + session.close() + expect(fence.assertDrained).toThrow(failure) + await expect(fence.drain(signal())).rejects.toBe(failure) +}) + +it('retains close failure and refuses subsequent uploads or repeated physical close', async () => { + const ledger = new SshConnectionWorkLedger() + const underlying = makeSession() + const failure = new Error('close failed') + underlying.close = vi.fn(() => { + throw failure + }) + const session = await openTrackedSshUploadSession(ledger, async () => underlying) + const fence = ledger.fenceForReset() + expect(session.close).toThrow(failure) + session.close() + expect(underlying.close).toHaveBeenCalledTimes(1) + await expect(session.uploadFile('a', 'b')).rejects.toThrow('ssh_connection_work_session_closed') + await expect(fence.drain(signal())).rejects.toBe(failure) + expect(fence.assertDrained).toThrow(failure) +}) + +it('preserves close uncertainty raised before reset starts', async () => { + const ledger = new SshConnectionWorkLedger() + const underlying = makeSession() + const failure = new Error('close failed before reset') + underlying.close = () => { + throw failure + } + const session = await openTrackedSshUploadSession(ledger, async () => underlying) + expect(session.close).toThrow(failure) + const fence = ledger.fenceForReset() + await expect(fence.drain(signal())).rejects.toBe(failure) + expect(fence.assertDrained).toThrow() +}) + +it('isolates sessions, admission, and failures between connection ledgers', async () => { + const first = new SshConnectionWorkLedger() + const second = new SshConnectionWorkLedger() + const failed = makeSession() + const failure = new Error('first connection failed') + failed.close = () => { + throw failure + } + const firstSession = await openTrackedSshUploadSession(first, async () => failed) + const firstFence = first.fenceForReset() + expect(firstSession.close).toThrow(failure) + const secondSession = await openTrackedSshUploadSession(second, async () => makeSession()) + const secondFence = second.fenceForReset() + await secondSession.uploadFile('a', 'b') + secondSession.close() + await secondFence.drain(signal()) + secondFence.assertDrained() + await expect(firstFence.drain(signal())).rejects.toBe(failure) +}) + +it('does not let stale factory callbacks reuse the held-session admission scope', async () => { + const ledger = new SshConnectionWorkLedger() + const release = Promise.withResolvers() + let stale!: Promise + const session = await openTrackedSshUploadSession(ledger, async () => { + stale = release.promise + .then(() => ledger.run(async () => undefined)) + .catch((error: unknown) => error) + return makeSession() + }) + const fence = ledger.fenceForReset() + release.resolve() + expect(await stale).toMatchObject({ message: 'ssh_connection_work_admission_closed' }) + session.close() + await fence.drain(signal()) +}) diff --git a/src/main/ssh/ssh-upload-session-lifetime.ts b/src/main/ssh/ssh-upload-session-lifetime.ts new file mode 100644 index 000000000000..bbbd6f3119c0 --- /dev/null +++ b/src/main/ssh/ssh-upload-session-lifetime.ts @@ -0,0 +1,33 @@ +import type { FileUploadSession } from '../providers/filesystem-provider-contract' +import type { SshConnectionWorkLedger } from './ssh-connection-work-ledger' + +export async function openTrackedSshUploadSession( + ledger: SshConnectionWorkLedger, + open: () => Promise +): Promise { + const lifetime = ledger.beginSession() + let session: FileUploadSession + try { + session = await lifetime.run(open) + } catch (error) { + lifetime.close(error) + throw error + } + let closed = false + return { + uploadFile: (...args) => lifetime.run(() => session.uploadFile(...args)), + close: () => { + if (closed) { + return + } + closed = true + try { + session.close() + } catch (error) { + lifetime.close(error) + throw error + } + lifetime.close() + } + } +} diff --git a/src/main/ssh/system-ssh-command.ts b/src/main/ssh/system-ssh-command.ts index eab96245ba53..15f099a5f881 100644 --- a/src/main/ssh/system-ssh-command.ts +++ b/src/main/ssh/system-ssh-command.ts @@ -96,6 +96,10 @@ function wrapCommandProcess(proc: ChildProcess): SystemSshCommandChannel { }, write(chunk, encoding, cb) { proc.stdin!.write(chunk, encoding, cb) + }, + final(cb) { + // EOF must reach the remote reader, not just this local facade. + proc.stdin!.end(cb) } }) const channel = duplex as unknown as SystemSshCommandChannel From 15329c7f0ba5e10494e2b43efbc9a807243ae7df Mon Sep 17 00:00:00 2001 From: m4air Date: Thu, 1 Oct 2026 08:51:10 -0700 Subject: [PATCH 2/3] refactor(ssh): shrink SshConnection below main and surface unhandled channel errors ssh-connection.ts no longer grows under its max-lines exemption: it is 1872 lines, below main's 1917. The public API is unchanged. - ssh-channel-open.ts: the channel-open waiter and session-limit retry. - ssh-connection-file-transfers.ts: the uploadDirectory, downloadFile, upload session, writeFile and writeBuffer bodies, reading the connection through a small getter-based host so each read still sees the live transport. - ssh-forward-channel-lifetime.ts: the forward client and stream-local checks. The lifetime tracker's 'error' listener no longer hides errors. When it is a channel's only error listener, the error is reported: SshConnection logs "[ssh] Unhandled channel error for : ", and other callers fall back to a generic [ssh] warning. Nothing throws, so an orphaned channel error still cannot crash the process. --- src/main/ssh/ssh-channel-open.ts | 204 ++++++++ .../ssh/ssh-connection-channel-error.test.ts | 62 +++ .../ssh-connection-channel-lifetime.test.ts | 39 +- .../ssh/ssh-connection-channel-lifetime.ts | 30 +- src/main/ssh/ssh-connection-file-transfers.ts | 219 ++++++++ src/main/ssh/ssh-connection.ts | 495 +++--------------- .../ssh/ssh-forward-channel-lifetime.test.ts | 1 + src/main/ssh/ssh-forward-channel-lifetime.ts | 33 +- 8 files changed, 649 insertions(+), 434 deletions(-) create mode 100644 src/main/ssh/ssh-channel-open.ts create mode 100644 src/main/ssh/ssh-connection-channel-error.test.ts create mode 100644 src/main/ssh/ssh-connection-file-transfers.ts diff --git a/src/main/ssh/ssh-channel-open.ts b/src/main/ssh/ssh-channel-open.ts new file mode 100644 index 000000000000..e46d2c28f620 --- /dev/null +++ b/src/main/ssh/ssh-channel-open.ts @@ -0,0 +1,204 @@ +import { trackSshConnectionChannelLifetime } from './ssh-connection-channel-lifetime' +import { CONNECT_TIMEOUT_MS, createSshOperationAbortError } from './ssh-connection-utils' +import type { SshConnectionWorkLedger } from './ssh-connection-work-ledger' +import { isSshSessionLimitError } from './ssh-session-limit-error' + +type SshChannelOpenWork = ReturnType + +// Upper bound on waiting for an aborted channel's open/close to settle before rejecting anyway. +const ABORTED_CHANNEL_CLOSE_GRACE_MS = 5_000 + +// Why: MaxSessions servers can transiently refuse a channel open; a refused open never ran the command, so retry is safe. +const SESSION_LIMIT_OPEN_RETRIES = 4 +const SESSION_LIMIT_OPEN_RETRY_DELAY_MS = 150 + +export async function openSshSessionChannelWithRetry( + open: () => Promise, + signal?: AbortSignal +): Promise { + let lastError: unknown + for (let attempt = 0; attempt < SESSION_LIMIT_OPEN_RETRIES; attempt++) { + if (attempt > 0) { + // Why: an abort must release the backoff immediately, not after it. + if (!signal?.aborted) { + await new Promise((resolve) => { + const onDelayDone = (): void => { + clearTimeout(delayTimer) + signal?.removeEventListener('abort', onDelayDone) + resolve() + } + const delayTimer = setTimeout(onDelayDone, SESSION_LIMIT_OPEN_RETRY_DELAY_MS) + signal?.addEventListener('abort', onDelayDone, { once: true }) + }) + } + if (signal?.aborted) { + throw createSshOperationAbortError() + } + } + try { + return await open() + } catch (err) { + if (!isSshSessionLimitError(err)) { + throw err + } + lastError = err + } + } + throw lastError +} + +/** + * Waits for an ssh2 channel open, bounded by CONNECT_TIMEOUT_MS and an abort grace. The channel, + * late ones included, stays on `work` until it physically closes. + */ +export function waitForSshChannelOpen( + work: SshChannelOpenWork, + timeoutMessage: string, + register: (callback: (error: Error | undefined, value: T) => void) => void, + cleanupLateValue?: (value: T) => void, + signal?: AbortSignal, + trackRemoteCommandTermination = false, + onUnhandledError?: (error: Error) => void +): Promise { + return new Promise((resolve, reject) => { + type ChannelOpenTerminationError = Error & { sshChannelCloseConfirmed: boolean } + let settled = false + let unconfirmedOpenError: ChannelOpenTerminationError | null = null + const markOpenUnconfirmed = (error: Error): Error => { + work.markUnverifiable(error) + if (!trackRemoteCommandTermination) { + return error + } + unconfirmedOpenError = Object.assign(error, { sshChannelCloseConfirmed: false }) + return unconfirmedOpenError + } + // Why: an in-flight open holds a MaxSessions slot; reject the caller now, then settle from the open callback once the late channel closes. + let abortRequested = false + let abortDeadlineTimer: NodeJS.Timeout | undefined + const cleanup = (): void => { + clearTimeout(timer) + clearTimeout(abortDeadlineTimer) + signal?.removeEventListener('abort', onAbort) + } + const onAbort = (): void => { + abortRequested = true + // Why: a hung socket may never invoke the open callback; bound the aborted caller's wait instead of pinning it for CONNECT_TIMEOUT_MS. + abortDeadlineTimer = setTimeout(() => { + settled = true + cleanup() + reject(markOpenUnconfirmed(createSshOperationAbortError())) + }, ABORTED_CHANNEL_CLOSE_GRACE_MS) + } + const timer = setTimeout(() => { + settled = true + cleanup() + reject( + markOpenUnconfirmed( + abortRequested ? createSshOperationAbortError() : new Error(timeoutMessage) + ) + ) + }, CONNECT_TIMEOUT_MS) + const discardLateValue = (value: T, onClose?: () => void): void => { + const emitter = value as Partial & { + resume?: () => void + stderr?: Partial & { resume?: () => void } + } + const swallowLateError = (): void => {} + emitter.on?.('error', swallowLateError) + emitter.stderr?.on?.('error', swallowLateError) + if (onClose) { + emitter.once?.('close', onClose) + } + // Why: ssh2 withholds CHANNEL_CLOSE while discarded exec streams remain unread, and teardown errors have no other owner. + emitter.resume?.() + emitter.stderr?.resume?.() + try { + cleanupLateValue?.(value) + } catch { + /* best effort */ + } + } + const rejectAfterClose = (value: T): void => { + const abortError = markOpenUnconfirmed(createSshOperationAbortError()) + const emitter = value as Partial & { + resume?: () => void + stderr?: { resume?: () => void } + } + let finished = false + const done = (): void => { + if (finished) { + return + } + finished = true + clearTimeout(closeGraceTimer) + emitter.removeListener?.('close', confirmAndDone) + reject(abortError) + } + const confirmAndDone = (): void => { + if (unconfirmedOpenError === abortError) { + unconfirmedOpenError.sshChannelCloseConfirmed = true + } + done() + } + // Why: bounded so a remote that never confirms the close can't hang the aborted operation forever. + const closeGraceTimer = setTimeout(done, ABORTED_CHANNEL_CLOSE_GRACE_MS) + if (typeof emitter.once === 'function') { + emitter.once('close', confirmAndDone) + } + // Why: ssh2 withholds 'close' until the channel's streams are drained; nobody else will read this discarded channel. + discardLateValue(value) + if (typeof emitter.once !== 'function') { + done() + } + } + const finish = (error: Error | undefined, value?: T): void => { + // Late channels stay tracked until physical close, after the caller has given up. + if (!error && value !== undefined) { + trackSshConnectionChannelLifetime(work, value, onUnhandledError) + } else { + work.close(error) + } + if (settled) { + // Why: ssh2 can invoke the open callback after our timeout rejected; close that late channel so it isn't left open with no owner. + if (!error && value !== undefined) { + discardLateValue(value, () => { + if (unconfirmedOpenError) { + unconfirmedOpenError.sshChannelCloseConfirmed = true + } + }) + } + return + } + settled = true + cleanup() + if (abortRequested) { + if (!error && value !== undefined) { + rejectAfterClose(value) + } else { + reject(createSshOperationAbortError()) + } + return + } + if (error) { + reject(error) + return + } + resolve(value as T) + } + if (signal?.aborted) { + // No open is in flight yet, so failing fast leaks nothing. + cleanup() + work.close() + reject(createSshOperationAbortError()) + return + } + signal?.addEventListener('abort', onAbort, { once: true }) + + try { + // Why: higher-level channel timers start only after ssh2's open callback; a stale SSH socket can otherwise keep exec/sftp stuck. + register(finish) + } catch (error) { + finish(error instanceof Error ? error : new Error(String(error))) + } + }) +} diff --git a/src/main/ssh/ssh-connection-channel-error.test.ts b/src/main/ssh/ssh-connection-channel-error.test.ts new file mode 100644 index 000000000000..9e6179a06f82 --- /dev/null +++ b/src/main/ssh/ssh-connection-channel-error.test.ts @@ -0,0 +1,62 @@ +import { EventEmitter } from 'node:events' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { SshConnection } from './ssh-connection' +import { createCallbacks, createTarget } from './ssh-connection-test-fixtures' +import { resetSshConnectionMocks, ssh2Mock } from './ssh-connection-test-harness' +import * as sshClientMock from './__tests__/ssh-connection-test-client' + +vi.mock('ssh2', async () => (await import('./ssh-connection-test-harness')).createSsh2Module()) +vi.mock('./system-ssh-binary', async () => + (await import('./ssh-connection-test-harness')).createSystemSshBinaryModule() +) +vi.mock('./ssh-system-fallback', async () => + (await import('./ssh-connection-test-harness')).createSystemFallbackModule() +) +vi.mock('./ssh-control-socket', async () => + (await import('./ssh-connection-test-harness')).createControlSocketModule() +) +vi.mock('./ssh-config-parser', async () => + (await import('./ssh-connection-test-harness')).createSshConfigParserModule() +) + +beforeEach(resetSshConnectionMocks) +afterEach(() => { + vi.restoreAllMocks() +}) + +it('logs an exec channel error nothing else handles, naming the target and channel kind', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + ssh2Mock.execBehavior = 'pending' + const opening = conn.exec('true') + await vi.waitFor(() => expect(sshClientMock.pendingExecCallback).not.toBeNull()) + const channel = Object.assign(new EventEmitter(), { close: vi.fn() }) + sshClientMock.pendingExecCallback?.(undefined, channel) + await opening + + expect(() => channel.emit('error', new Error('channel reset'))).not.toThrow() + expect(warn).toHaveBeenCalledWith( + '[ssh] Unhandled exec channel error for Test Server: channel reset' + ) + await conn.disconnect() +}) + +it('does not log a forwarded channel error its owner handles', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const conn = new SshConnection(createTarget(), createCallbacks()) + await conn.connect() + const client = conn.getClient()! + const channel = new EventEmitter() + client.openssh_forwardOutStreamLocal = vi.fn((_path, callback) => { + callback(undefined, channel as never) + return client + }) + const owner = vi.fn() + conn.forwardStreamLocal(client, '/owned.sock', () => channel.on('error', owner)) + + channel.emit('error', new Error('handled')) + expect(owner).toHaveBeenCalledOnce() + expect(warn).not.toHaveBeenCalledWith(expect.stringContaining('Unhandled')) + await conn.disconnect() +}) diff --git a/src/main/ssh/ssh-connection-channel-lifetime.test.ts b/src/main/ssh/ssh-connection-channel-lifetime.test.ts index 56b5a84f92a9..82371f36880f 100644 --- a/src/main/ssh/ssh-connection-channel-lifetime.test.ts +++ b/src/main/ssh/ssh-connection-channel-lifetime.test.ts @@ -1,5 +1,5 @@ import { EventEmitter } from 'node:events' -import { expect, it } from 'vitest' +import { expect, it, vi } from 'vitest' import { trackSshConnectionChannelLifetime } from './ssh-connection-channel-lifetime' import { SshConnectionWorkLedger } from './ssh-connection-work-ledger' @@ -63,3 +63,40 @@ it('runs the injected admission check before admitting work', () => { }) expect(() => ledger.beginChannelOpen()).toThrow('connection disposed') }) + +it('reports a channel error that nothing else handles instead of hiding it', () => { + const report = vi.fn() + const channel = new EventEmitter() + trackSshConnectionChannelLifetime( + new SshConnectionWorkLedger().beginChannelOpen(), + channel, + report + ) + const failure = new Error('channel reset') + expect(() => channel.emit('error', failure)).not.toThrow() + expect(report).toHaveBeenCalledWith(failure) +}) + +it('leaves a channel error to the owner that handles it', () => { + const report = vi.fn() + const channel = new EventEmitter() + trackSshConnectionChannelLifetime( + new SshConnectionWorkLedger().beginChannelOpen(), + channel, + report + ) + const owner = vi.fn() + channel.on('error', owner) + channel.emit('error', new Error('handled')) + expect(owner).toHaveBeenCalledOnce() + expect(report).not.toHaveBeenCalled() +}) + +it('logs an unhandled channel error with the [ssh] prefix when no reporter is given', () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const channel = new EventEmitter() + trackSshConnectionChannelLifetime(new SshConnectionWorkLedger().beginChannelOpen(), channel) + channel.emit('error', new Error('orphaned')) + expect(warn).toHaveBeenCalledWith('[ssh] Unhandled SSH channel error: orphaned') + warn.mockRestore() +}) diff --git a/src/main/ssh/ssh-connection-channel-lifetime.ts b/src/main/ssh/ssh-connection-channel-lifetime.ts index c6ad86100bcf..b09a7d288c9d 100644 --- a/src/main/ssh/ssh-connection-channel-lifetime.ts +++ b/src/main/ssh/ssh-connection-channel-lifetime.ts @@ -21,14 +21,30 @@ function isTrackableChannel(value: unknown): value is TrackableChannel { ) } +export type SshChannelErrorReporter = (error: Error) => void + +const reportOrphanChannelError: SshChannelErrorReporter = (error) => { + console.warn(`[ssh] Unhandled SSH channel error: ${error.message}`) +} + +// The tracker's own listener must not be what keeps an error from surfacing. +function hasOnlyTrackerErrorListener(channel: TrackableChannel): boolean { + return ( + !('listenerCount' in channel) || + typeof channel.listenerCount !== 'function' || + channel.listenerCount('error') <= 1 + ) +} + export function openTrackedSshSocket( ledger: SshConnectionWorkLedger, - open: () => T + open: () => T, + onUnhandledError?: SshChannelErrorReporter ): T { const work = ledger.beginChannelOpen() try { const socket = open() - trackSshConnectionChannelLifetime(work, socket) + trackSshConnectionChannelLifetime(work, socket, onUnhandledError) return socket } catch (error) { work.markUnverifiable(error instanceof Error ? error : new Error(String(error))) @@ -39,7 +55,8 @@ export function openTrackedSshSocket( /** Start tracking before the open callback hands the channel to another owner. */ export function trackSshConnectionChannelLifetime( work: ReturnType, - value: unknown + value: unknown, + onUnhandledError: SshChannelErrorReporter = reportOrphanChannelError ): void { if (!isTrackableChannel(value)) { work.markUnverifiable(new Error('ssh_connection_channel_lifetime_unverifiable')) @@ -51,7 +68,12 @@ export function trackSshConnectionChannelLifetime( work.close() return } - const onError = (error: Error) => work.markUnverifiable(error) + const onError = (error: Error) => { + work.markUnverifiable(error) + if (hasOnlyTrackerErrorListener(channel)) { + onUnhandledError(error) + } + } channel.on('error', onError) channel.once('close', () => { channel.removeListener('error', onError) diff --git a/src/main/ssh/ssh-connection-file-transfers.ts b/src/main/ssh/ssh-connection-file-transfers.ts new file mode 100644 index 000000000000..c9cc36dcdd7c --- /dev/null +++ b/src/main/ssh/ssh-connection-file-transfers.ts @@ -0,0 +1,219 @@ +import type { SFTPWrapper } from 'ssh2' +import type { SshTarget } from '../../shared/ssh-types' +import type { FileUploadSession } from '../providers/types' +import { + resolveSftpTransferPathIfMapped, + type SftpNamespacePathMapping +} from './sftp-namespace-resolution' +import { + createLinkedSshFileTransferSignal, + raceSftpFileTransferWithAbort +} from './ssh-file-transfer-abort' +import type { RemoteHostPlatform } from './ssh-remote-platform' +import { + downloadFileViaSystemSsh, + uploadDirectoryViaSystemSsh, + uploadFileViaSystemSsh, + writeBufferViaSystemSsh, + writeFileViaSystemSsh, + type SystemSshBuildArgsOptions +} from './ssh-system-fallback' + +export type SshRemoteFileOptions = { + hostPlatform?: RemoteHostPlatform + // Only uploadDirectory and writeFile honor this, and only on the non-Windows ssh2 branch. + sftpNamespace?: SftpNamespacePathMapping +} + +/** What a transfer reads from its connection; getters so each read sees the live transport. */ +export type SshFileTransferHost = { + target: SshTarget + usesSystemSshTransport: () => boolean + systemOperationSignal: () => AbortSignal + systemSshBuildArgsOptions: () => SystemSshBuildArgsOptions + sftp: (signal?: AbortSignal) => Promise +} + +export async function uploadSshDirectory( + host: SshFileTransferHost, + localDir: string, + remoteDir: string, + options?: SshRemoteFileOptions & { signal?: AbortSignal } +): Promise { + // Why: relay-deploy timeout and connection teardown are independent owners; either must stop a transfer that could outlive its lock. + const linkedSignal = createLinkedSshFileTransferSignal( + [host.systemOperationSignal(), options?.signal].filter( + (signal): signal is AbortSignal => signal !== undefined + ) + ) + try { + if (!host.usesSystemSshTransport()) { + const sftp = await host.sftp(linkedSignal.signal) + const swallowLateSftpError = (): void => {} + let sftpEndRequested = false + const endSftp = (): void => { + if (!sftpEndRequested) { + sftpEndRequested = true + sftp.end() + } + } + sftp.on('error', swallowLateSftpError) + sftp.once('close', () => sftp.removeListener('error', swallowLateSftpError)) + try { + // Why: resolve on the same session that transfers — a later session is not authoritative for this one's namespace. + const transfer = (async (): Promise => { + const targetDir = await resolveSftpTransferPathIfMapped(sftp, remoteDir, options) + linkedSignal.signal.throwIfAborted() + const { uploadDirectory } = await import('./ssh-relay-deploy-helpers') + await uploadDirectory(sftp, localDir, targetDir, localDir, { + signal: linkedSignal.signal + }) + })() + await raceSftpFileTransferWithAbort(transfer, linkedSignal.signal, (onClose) => { + sftp.once('close', onClose) + endSftp() + return () => sftp.removeListener('close', onClose) + }) + } finally { + endSftp() + } + return + } + await uploadDirectoryViaSystemSsh(host.target, localDir, remoteDir, { + signal: linkedSignal.signal, + hostPlatform: options?.hostPlatform, + ...host.systemSshBuildArgsOptions() + }) + } finally { + linkedSignal.dispose() + } +} + +export async function downloadSshFile( + host: SshFileTransferHost, + remotePath: string, + localPath: string, + options?: SshRemoteFileOptions +): Promise { + if (!host.usesSystemSshTransport()) { + const sftp = await host.sftp() + try { + const { fastGetViaSftp } = await import('../providers/ssh-filesystem-provider-sftp') + await fastGetViaSftp(sftp, remotePath, localPath) + } finally { + sftp.end() + } + return + } + await downloadFileViaSystemSsh(host.target, remotePath, localPath, { + signal: host.systemOperationSignal(), + hostPlatform: options?.hostPlatform, + ...host.systemSshBuildArgsOptions() + }) +} + +export async function createSshFileUploadSession( + host: SshFileTransferHost, + options?: SshRemoteFileOptions +): Promise { + if (!host.usesSystemSshTransport()) { + const sftp = await host.sftp() + const { uploadFile } = await import('./sftp-upload') + return { + uploadFile: (localPath, remotePath, uploadOptions) => + uploadFile(sftp, localPath, remotePath, uploadOptions), + close: () => sftp.end() + } + } + // Why: disconnect replaces the connection controller, so an existing import session must stay bound to the signal and SSH config it opened with. + const signal = host.systemOperationSignal() + const buildArgsOptions = host.systemSshBuildArgsOptions() + return { + uploadFile: (localPath, remotePath, uploadOptions) => + uploadFileViaSystemSsh(host.target, localPath, remotePath, { + signal, + hostPlatform: options?.hostPlatform, + exclusive: uploadOptions?.exclusive, + ...buildArgsOptions + }), + close: () => {} + } +} + +export async function writeSshFile( + host: SshFileTransferHost, + remotePath: string, + contents: string, + options?: SshRemoteFileOptions & { signal?: AbortSignal } +): Promise { + // Keep package/version writes under the same dual cancellation contract as uploads. + const linkedSignal = createLinkedSshFileTransferSignal( + [host.systemOperationSignal(), options?.signal].filter( + (signal): signal is AbortSignal => signal !== undefined + ) + ) + try { + if (!host.usesSystemSshTransport()) { + const sftp = await host.sftp(linkedSignal.signal) + const swallowLateSftpError = (): void => {} + let sftpEndRequested = false + const endSftp = (): void => { + if (!sftpEndRequested) { + sftpEndRequested = true + sftp.end() + } + } + sftp.on('error', swallowLateSftpError) + sftp.once('close', () => sftp.removeListener('error', swallowLateSftpError)) + try { + // Why: resolve on the same session that writes — a later session is not authoritative for this one's namespace. + const write = (async (): Promise => { + const targetPath = await resolveSftpTransferPathIfMapped(sftp, remotePath, options) + linkedSignal.signal.throwIfAborted() + const { writeStringViaSftp } = await import('./sftp-upload') + await writeStringViaSftp(sftp, targetPath, contents) + })() + await raceSftpFileTransferWithAbort(write, linkedSignal.signal, (onClose) => { + sftp.once('close', onClose) + endSftp() + return () => sftp.removeListener('close', onClose) + }) + } finally { + endSftp() + } + return + } + await writeFileViaSystemSsh(host.target, remotePath, contents, { + signal: linkedSignal.signal, + hostPlatform: options?.hostPlatform, + ...host.systemSshBuildArgsOptions() + }) + } finally { + linkedSignal.dispose() + } +} + +export async function writeSshBuffer( + host: SshFileTransferHost, + remotePath: string, + contents: Buffer, + options?: SshRemoteFileOptions & { append?: boolean; exclusive?: boolean } +): Promise { + if (!host.usesSystemSshTransport()) { + const sftp = await host.sftp() + try { + const { uploadBuffer } = await import('./sftp-upload') + await uploadBuffer(sftp, contents, remotePath, options) + } finally { + sftp.end() + } + return + } + await writeBufferViaSystemSsh(host.target, remotePath, contents, { + signal: host.systemOperationSignal(), + hostPlatform: options?.hostPlatform, + append: options?.append, + exclusive: options?.exclusive, + ...host.systemSshBuildArgsOptions() + }) +} diff --git a/src/main/ssh/ssh-connection.ts b/src/main/ssh/ssh-connection.ts index 789cd6efdc01..0436b97b3225 100644 --- a/src/main/ssh/ssh-connection.ts +++ b/src/main/ssh/ssh-connection.ts @@ -8,6 +8,17 @@ import { SshTransportClosureSubscribers } from './ssh-connection-transport-closu import { disconnectAndAwaitSshTransportClose } from './ssh-connection-close-drain' import { openTrackedSshUploadSession } from './ssh-upload-session-lifetime' import { + createSshFileUploadSession, + downloadSshFile, + uploadSshDirectory, + writeSshBuffer, + writeSshFile, + type SshFileTransferHost, + type SshRemoteFileOptions +} from './ssh-connection-file-transfers' +import { + assertSshForwardClient, + assertSshStreamLocalForwardAllowed, forwardTrackedSshChannel, forwardTrackedSshStreamLocalChannel } from './ssh-forward-channel-lifetime' @@ -30,11 +41,6 @@ import { getOrcaControlSocketPath, spawnSystemSsh, spawnSystemSshCommand, - downloadFileViaSystemSsh, - uploadDirectoryViaSystemSsh, - uploadFileViaSystemSsh, - writeBufferViaSystemSsh, - writeFileViaSystemSsh, type SystemSshBuildArgsOptions, type SystemSshProcess } from './ssh-system-fallback' @@ -97,18 +103,9 @@ import { requiresSystemSshForSecurityKey, shouldUseSystemSshTransport } from './ssh-transport-selection' -import type { RemoteHostPlatform } from './ssh-remote-platform' -import { - resolveSftpTransferPathIfMapped, - type SftpNamespacePathMapping -} from './sftp-namespace-resolution' import type { FileUploadSession } from '../providers/types' -import { isSshSessionLimitError } from './ssh-session-limit-error' +import { openSshSessionChannelWithRetry, waitForSshChannelOpen } from './ssh-channel-open' import { withTimeout } from '../../shared/promise-timeout-fallback' -import { - createLinkedSshFileTransferSignal, - raceSftpFileTransferWithAbort -} from './ssh-file-transfer-abort' export type { SshConnectionCallbacks } from './ssh-connection-utils' type HostKeyTrustSources = { @@ -118,12 +115,6 @@ type HostKeyTrustSources = { trustedHostKeys: Awaited> } -type SshRemoteFileOptions = { - hostPlatform?: RemoteHostPlatform - // Only uploadDirectory and writeFile honor this, and only on the non-Windows ssh2 branch. - sftpNamespace?: SftpNamespacePathMapping -} - /** Bounds the trust-source reads that run before the handshake, which nothing else times out. */ const HOST_KEY_SOURCE_READ_TIMEOUT_MS = 5_000 // Counts every INFO_REQUEST of the handshake, so it must cover each partial-success stage the auth @@ -132,13 +123,6 @@ const SSH_KEYBOARD_INTERACTIVE_MAX_ROUNDS = 8 const SSH_KEYBOARD_INTERACTIVE_READY_TIMEOUT_MS = SSH_CREDENTIAL_TIMEOUT_MS + 5_000 const SSH_KEYBOARD_INTERACTIVE_MAX_PROMPTS = 8 -// Upper bound on waiting for an aborted channel's open/close to settle before rejecting anyway. -const ABORTED_CHANNEL_CLOSE_GRACE_MS = 5_000 - -// Why: MaxSessions servers can transiently refuse a channel open; a refused open never ran the command, so retry is safe. -const SESSION_LIMIT_OPEN_RETRIES = 4 -const SESSION_LIMIT_OPEN_RETRY_DELAY_MS = 150 - function cloneResolvedConfig(config: SshResolvedConfig | null): SshResolvedConfig | null { if (!config) { return null @@ -282,7 +266,7 @@ export class SshConnection { getConnectGeneration(): number { return this.connectGeneration } - /** Alias for callers that compare transport incarnations rather than connect attempts. */ + /** Equals the connect generation; later slices name it by the transport it identifies. */ getTransportGeneration(): number { return this.connectGeneration } @@ -307,33 +291,30 @@ export class SshConnection { return this.workLedger.run(prepare) } openForwardSocket(open: () => T): T { - return openTrackedSshSocket(this.workLedger, open) + return openTrackedSshSocket(this.workLedger, open, this.reportUnhandledChannelError('socket')) } - forwardOut( - client: SshClient, - localSocket: object, - ...args: Parameters - ): void { - if (this.client !== client) { - throw new Error('ssh_connection_forward_client_changed') - } - forwardTrackedSshChannel(this.workLedger, client, localSocket, ...args) + forwardOut(client: SshClient, localSocket: object, ...args: Parameters) { + assertSshForwardClient(this.client, client) + const report = this.reportUnhandledChannelError('forward') + forwardTrackedSshChannel(this.workLedger, client, localSocket, report, ...args) } forwardStreamLocal( client: SshClient, ...args: Parameters ): void { - if (this.client !== client) { - throw new Error('ssh_connection_forward_client_changed') - } - if (this.disposed || this.useSystemSshTransport || this.state.status !== 'connected') { - throw new Error('ssh_connection_streamlocal_transport_unavailable') - } - const [socketPath] = args - if (!socketPath.startsWith('/') || socketPath.includes('\0')) { - throw new Error('ssh_connection_streamlocal_endpoint_invalid') - } - forwardTrackedSshStreamLocalChannel(this.workLedger, client, ...args) + assertSshForwardClient(this.client, client) + const usable = + !this.disposed && !this.useSystemSshTransport && this.state.status === 'connected' + assertSshStreamLocalForwardAllowed(usable, args[0]) + const report = this.reportUnhandledChannelError('stream-local forward') + forwardTrackedSshStreamLocalChannel(this.workLedger, client, report, ...args) + } + /** Names the connection and channel kind when a tracked channel's error has no other handler. */ + private reportUnhandledChannelError(kind: string): (error: Error) => void { + return (error) => + console.warn( + `[ssh] Unhandled ${kind} channel error for ${this.target.label}: ${error.message}` + ) } usesSystemSshTransport(): boolean { return this.useSystemSshTransport @@ -388,14 +369,16 @@ export class SshConnection { } const client = this.client const remoteCommand = options?.wrapCommand === false ? cmd : wrapRemoteCommandForPosixShell(cmd) - return this.openSessionChannelWithRetry( + return openSshSessionChannelWithRetry( () => - this.waitForSshCallback( + waitForSshChannelOpen( + this.workLedger.beginChannelOpen(), 'SSH exec channel timed out', (callback) => client.exec(remoteCommand, callback), (channel) => channel.close(), options?.signal, - true + true, + this.reportUnhandledChannelError('exec') ), options?.signal ) @@ -418,11 +401,15 @@ export class SshConnection { throw new Error('Not connected') } const client = this.client - return this.openSessionChannelWithRetry(() => - this.waitForSshCallback( + return openSshSessionChannelWithRetry(() => + waitForSshChannelOpen( + this.workLedger.beginChannelOpen(), 'SSH shell channel timed out', (callback) => client.shell(pty, options, callback), - (channel) => channel.close() + (channel) => channel.close(), + undefined, + false, + this.reportUnhandledChannelError('shell') ) ) } @@ -446,264 +433,29 @@ export class SshConnection { throw new Error('Not connected') } const client = this.client - return this.openSessionChannelWithRetry( + return openSshSessionChannelWithRetry( () => - this.waitForSshCallback( + waitForSshChannelOpen( + this.workLedger.beginChannelOpen(), 'SSH SFTP channel timed out', (callback) => client.sftp(callback), (sftp) => sftp.end(), - signal + signal, + false, + this.reportUnhandledChannelError('sftp') ), signal ) } - private async openSessionChannelWithRetry( - open: () => Promise, - signal?: AbortSignal - ): Promise { - let lastError: unknown - for (let attempt = 0; attempt < SESSION_LIMIT_OPEN_RETRIES; attempt++) { - if (attempt > 0) { - // Why: an abort must release the backoff immediately, not after it. - if (!signal?.aborted) { - await new Promise((resolve) => { - const onDelayDone = (): void => { - clearTimeout(delayTimer) - signal?.removeEventListener('abort', onDelayDone) - resolve() - } - const delayTimer = setTimeout(onDelayDone, SESSION_LIMIT_OPEN_RETRY_DELAY_MS) - signal?.addEventListener('abort', onDelayDone, { once: true }) - }) - } - if (signal?.aborted) { - throw createSshOperationAbortError() - } - } - try { - return await open() - } catch (err) { - if (!isSshSessionLimitError(err)) { - throw err - } - lastError = err - } - } - throw lastError - } - - private waitForSshCallback( - timeoutMessage: string, - register: (callback: (error: Error | undefined, value: T) => void) => void, - cleanupLateValue?: (value: T) => void, - signal?: AbortSignal, - trackRemoteCommandTermination = false - ): Promise { - const work = this.workLedger.beginChannelOpen() - return new Promise((resolve, reject) => { - type ChannelOpenTerminationError = Error & { sshChannelCloseConfirmed: boolean } - let settled = false - let unconfirmedOpenError: ChannelOpenTerminationError | null = null - const markOpenUnconfirmed = (error: Error): Error => { - work.markUnverifiable(error) - if (!trackRemoteCommandTermination) { - return error - } - unconfirmedOpenError = Object.assign(error, { sshChannelCloseConfirmed: false }) - return unconfirmedOpenError - } - // Why: an in-flight open holds a MaxSessions slot; reject the caller now, then settle from the open callback once the late channel closes. - let abortRequested = false - let abortDeadlineTimer: NodeJS.Timeout | undefined - const cleanup = (): void => { - clearTimeout(timer) - clearTimeout(abortDeadlineTimer) - signal?.removeEventListener('abort', onAbort) - } - const onAbort = (): void => { - abortRequested = true - // Why: a hung socket may never invoke the open callback; bound the aborted caller's wait instead of pinning it for CONNECT_TIMEOUT_MS. - abortDeadlineTimer = setTimeout(() => { - settled = true - cleanup() - reject(markOpenUnconfirmed(createSshOperationAbortError())) - }, ABORTED_CHANNEL_CLOSE_GRACE_MS) - } - const timer = setTimeout(() => { - settled = true - cleanup() - reject( - markOpenUnconfirmed( - abortRequested ? createSshOperationAbortError() : new Error(timeoutMessage) - ) - ) - }, CONNECT_TIMEOUT_MS) - const discardLateValue = (value: T, onClose?: () => void): void => { - const emitter = value as Partial & { - resume?: () => void - stderr?: Partial & { resume?: () => void } - } - const swallowLateError = (): void => {} - emitter.on?.('error', swallowLateError) - emitter.stderr?.on?.('error', swallowLateError) - if (onClose) { - emitter.once?.('close', onClose) - } - // Why: ssh2 withholds CHANNEL_CLOSE while discarded exec streams remain unread, and teardown errors have no other owner. - emitter.resume?.() - emitter.stderr?.resume?.() - try { - cleanupLateValue?.(value) - } catch { - /* best effort */ - } - } - const rejectAfterClose = (value: T): void => { - const abortError = markOpenUnconfirmed(createSshOperationAbortError()) - const emitter = value as Partial & { - resume?: () => void - stderr?: { resume?: () => void } - } - let finished = false - const done = (): void => { - if (finished) { - return - } - finished = true - clearTimeout(closeGraceTimer) - emitter.removeListener?.('close', confirmAndDone) - reject(abortError) - } - const confirmAndDone = (): void => { - if (unconfirmedOpenError === abortError) { - unconfirmedOpenError.sshChannelCloseConfirmed = true - } - done() - } - // Why: bounded so a remote that never confirms the close can't hang the aborted operation forever. - const closeGraceTimer = setTimeout(done, ABORTED_CHANNEL_CLOSE_GRACE_MS) - if (typeof emitter.once === 'function') { - emitter.once('close', confirmAndDone) - } - // Why: ssh2 withholds 'close' until the channel's streams are drained; nobody else will read this discarded channel. - discardLateValue(value) - if (typeof emitter.once !== 'function') { - done() - } - } - const finish = (error: Error | undefined, value?: T): void => { - // Late channels stay tracked until physical close, after the caller has given up. - if (!error && value !== undefined) { - trackSshConnectionChannelLifetime(work, value) - } else { - work.close(error) - } - if (settled) { - // Why: ssh2 can invoke the open callback after our timeout rejected; close that late channel so it isn't left open with no owner. - if (!error && value !== undefined) { - discardLateValue(value, () => { - if (unconfirmedOpenError) { - unconfirmedOpenError.sshChannelCloseConfirmed = true - } - }) - } - return - } - settled = true - cleanup() - if (abortRequested) { - if (!error && value !== undefined) { - rejectAfterClose(value) - } else { - reject(createSshOperationAbortError()) - } - return - } - if (error) { - reject(error) - return - } - resolve(value as T) - } - if (signal?.aborted) { - // No open is in flight yet, so failing fast leaks nothing. - cleanup() - work.close() - reject(createSshOperationAbortError()) - return - } - signal?.addEventListener('abort', onAbort, { once: true }) - - try { - // Why: higher-level channel timers start only after ssh2's open callback; a stale SSH socket can otherwise keep exec/sftp stuck. - register(finish) - } catch (error) { - finish(error instanceof Error ? error : new Error(String(error))) - } - }) - } - uploadDirectory( localDir: string, remoteDir: string, options?: SshRemoteFileOptions & { signal?: AbortSignal } ): Promise { - return this.workLedger.run(() => this.uploadDirectoryUntracked(localDir, remoteDir, options)) - } - - private async uploadDirectoryUntracked( - localDir: string, - remoteDir: string, - options?: SshRemoteFileOptions & { signal?: AbortSignal } - ): Promise { - // Why: relay-deploy timeout and connection teardown are independent owners; either must stop a transfer that could outlive its lock. - const linkedSignal = createLinkedSshFileTransferSignal( - [this.systemOperationAbortController.signal, options?.signal].filter( - (signal): signal is AbortSignal => signal !== undefined - ) + return this.workLedger.run(() => + uploadSshDirectory(this.fileTransferHost(), localDir, remoteDir, options) ) - try { - if (!this.useSystemSshTransport) { - const sftp = await this.sftp(linkedSignal.signal) - const swallowLateSftpError = (): void => {} - let sftpEndRequested = false - const endSftp = (): void => { - if (!sftpEndRequested) { - sftpEndRequested = true - sftp.end() - } - } - sftp.on('error', swallowLateSftpError) - sftp.once('close', () => sftp.removeListener('error', swallowLateSftpError)) - try { - // Why: resolve on the same session that transfers — a later session is not authoritative for this one's namespace. - const transfer = (async (): Promise => { - const targetDir = await resolveSftpTransferPathIfMapped(sftp, remoteDir, options) - linkedSignal.signal.throwIfAborted() - const { uploadDirectory } = await import('./ssh-relay-deploy-helpers') - await uploadDirectory(sftp, localDir, targetDir, localDir, { - signal: linkedSignal.signal - }) - })() - await raceSftpFileTransferWithAbort(transfer, linkedSignal.signal, (onClose) => { - sftp.once('close', onClose) - endSftp() - return () => sftp.removeListener('close', onClose) - }) - } finally { - endSftp() - } - return - } - await uploadDirectoryViaSystemSsh(this.target, localDir, remoteDir, { - signal: linkedSignal.signal, - hostPlatform: options?.hostPlatform, - ...this.getSystemSshBuildArgsOptions() - }) - } finally { - linkedSignal.dispose() - } } downloadFile( @@ -711,60 +463,15 @@ export class SshConnection { localPath: string, options?: SshRemoteFileOptions ): Promise { - return this.workLedger.run(() => this.downloadFileUntracked(remotePath, localPath, options)) - } - - private async downloadFileUntracked( - remotePath: string, - localPath: string, - options?: SshRemoteFileOptions - ): Promise { - if (!this.useSystemSshTransport) { - const sftp = await this.sftp() - try { - const { fastGetViaSftp } = await import('../providers/ssh-filesystem-provider-sftp') - await fastGetViaSftp(sftp, remotePath, localPath) - } finally { - sftp.end() - } - return - } - await downloadFileViaSystemSsh(this.target, remotePath, localPath, { - signal: this.systemOperationAbortController.signal, - hostPlatform: options?.hostPlatform, - ...this.getSystemSshBuildArgsOptions() - }) + return this.workLedger.run(() => + downloadSshFile(this.fileTransferHost(), remotePath, localPath, options) + ) } openFileUploadSession(options?: SshRemoteFileOptions): Promise { - return openTrackedSshUploadSession(this.workLedger, () => this.createFileUploadSession(options)) - } - - private async createFileUploadSession( - options?: SshRemoteFileOptions - ): Promise { - if (!this.useSystemSshTransport) { - const sftp = await this.sftp() - const { uploadFile } = await import('./sftp-upload') - return { - uploadFile: (localPath, remotePath, uploadOptions) => - uploadFile(sftp, localPath, remotePath, uploadOptions), - close: () => sftp.end() - } - } - // Why: disconnect replaces the connection controller, so an existing import session must stay bound to the signal and SSH config it opened with. - const signal = this.systemOperationAbortController.signal - const buildArgsOptions = this.getSystemSshBuildArgsOptions() - return { - uploadFile: (localPath, remotePath, uploadOptions) => - uploadFileViaSystemSsh(this.target, localPath, remotePath, { - signal, - hostPlatform: options?.hostPlatform, - exclusive: uploadOptions?.exclusive, - ...buildArgsOptions - }), - close: () => {} - } + return openTrackedSshUploadSession(this.workLedger, () => + createSshFileUploadSession(this.fileTransferHost(), options) + ) } writeFile( @@ -772,59 +479,9 @@ export class SshConnection { contents: string, options?: SshRemoteFileOptions & { signal?: AbortSignal } ): Promise { - return this.workLedger.run(() => this.writeFileUntracked(remotePath, contents, options)) - } - - private async writeFileUntracked( - remotePath: string, - contents: string, - options?: SshRemoteFileOptions & { signal?: AbortSignal } - ): Promise { - // Keep package/version writes under the same dual cancellation contract as uploads. - const linkedSignal = createLinkedSshFileTransferSignal( - [this.systemOperationAbortController.signal, options?.signal].filter( - (signal): signal is AbortSignal => signal !== undefined - ) + return this.workLedger.run(() => + writeSshFile(this.fileTransferHost(), remotePath, contents, options) ) - try { - if (!this.useSystemSshTransport) { - const sftp = await this.sftp(linkedSignal.signal) - const swallowLateSftpError = (): void => {} - let sftpEndRequested = false - const endSftp = (): void => { - if (!sftpEndRequested) { - sftpEndRequested = true - sftp.end() - } - } - sftp.on('error', swallowLateSftpError) - sftp.once('close', () => sftp.removeListener('error', swallowLateSftpError)) - try { - // Why: resolve on the same session that writes — a later session is not authoritative for this one's namespace. - const write = (async (): Promise => { - const targetPath = await resolveSftpTransferPathIfMapped(sftp, remotePath, options) - linkedSignal.signal.throwIfAborted() - const { writeStringViaSftp } = await import('./sftp-upload') - await writeStringViaSftp(sftp, targetPath, contents) - })() - await raceSftpFileTransferWithAbort(write, linkedSignal.signal, (onClose) => { - sftp.once('close', onClose) - endSftp() - return () => sftp.removeListener('close', onClose) - }) - } finally { - endSftp() - } - return - } - await writeFileViaSystemSsh(this.target, remotePath, contents, { - signal: linkedSignal.signal, - hostPlatform: options?.hostPlatform, - ...this.getSystemSshBuildArgsOptions() - }) - } finally { - linkedSignal.dispose() - } } writeBuffer( @@ -832,31 +489,19 @@ export class SshConnection { contents: Buffer, options?: SshRemoteFileOptions & { append?: boolean; exclusive?: boolean } ): Promise { - return this.workLedger.run(() => this.writeBufferUntracked(remotePath, contents, options)) + return this.workLedger.run(() => + writeSshBuffer(this.fileTransferHost(), remotePath, contents, options) + ) } - private async writeBufferUntracked( - remotePath: string, - contents: Buffer, - options?: SshRemoteFileOptions & { append?: boolean; exclusive?: boolean } - ): Promise { - if (!this.useSystemSshTransport) { - const sftp = await this.sftp() - try { - const { uploadBuffer } = await import('./sftp-upload') - await uploadBuffer(sftp, contents, remotePath, options) - } finally { - sftp.end() - } - return + private fileTransferHost(): SshFileTransferHost { + return { + target: this.target, + usesSystemSshTransport: () => this.useSystemSshTransport, + systemOperationSignal: () => this.systemOperationAbortController.signal, + systemSshBuildArgsOptions: () => this.getSystemSshBuildArgsOptions(), + sftp: (signal) => this.sftp(signal) } - await writeBufferViaSystemSsh(this.target, remotePath, contents, { - signal: this.systemOperationAbortController.signal, - hostPlatform: options?.hostPlatform, - append: options?.append, - exclusive: options?.exclusive, - ...this.getSystemSshBuildArgsOptions() - }) } async connect(): Promise { @@ -1530,7 +1175,7 @@ export class SshConnection { commandOptions === undefined ? spawnSystemSshCommand(this.target, command) : spawnSystemSshCommand(this.target, command, commandOptions) - trackSshConnectionChannelLifetime(work, channel) + trackSshConnectionChannelLifetime(work, channel, this.reportUnhandledChannelError('command')) } catch (error) { work.close(error) throw error diff --git a/src/main/ssh/ssh-forward-channel-lifetime.test.ts b/src/main/ssh/ssh-forward-channel-lifetime.test.ts index ed0d6193d420..449d663f0dfe 100644 --- a/src/main/ssh/ssh-forward-channel-lifetime.test.ts +++ b/src/main/ssh/ssh-forward-channel-lifetime.test.ts @@ -20,6 +20,7 @@ function fixture() { ledger, client as unknown as Client, local, + undefined, '127.0.0.1', 0, 'remote.internal', diff --git a/src/main/ssh/ssh-forward-channel-lifetime.ts b/src/main/ssh/ssh-forward-channel-lifetime.ts index 5530138fb47b..af1d59a4f600 100644 --- a/src/main/ssh/ssh-forward-channel-lifetime.ts +++ b/src/main/ssh/ssh-forward-channel-lifetime.ts @@ -1,11 +1,35 @@ import type { Client } from 'ssh2' -import { trackSshConnectionChannelLifetime } from './ssh-connection-channel-lifetime' +import { + trackSshConnectionChannelLifetime, + type SshChannelErrorReporter +} from './ssh-connection-channel-lifetime' import type { SshConnectionWorkLedger } from './ssh-connection-work-ledger' +/** A forward must ride the connection's current client, never one a reconnect replaced. */ +export function assertSshForwardClient(current: Client | null, requested: Client): void { + if (current !== requested) { + throw new Error('ssh_connection_forward_client_changed') + } +} + +/** Remote stream-local forwarding needs a live ssh2 transport and an absolute Unix socket path. */ +export function assertSshStreamLocalForwardAllowed( + transportUsable: boolean, + socketPath: string +): void { + if (!transportUsable) { + throw new Error('ssh_connection_streamlocal_transport_unavailable') + } + if (!socketPath.startsWith('/') || socketPath.includes('\0')) { + throw new Error('ssh_connection_streamlocal_endpoint_invalid') + } +} + /** The channel remains tracked until physical close, including late open replies. */ export function forwardTrackedSshStreamLocalChannel( ledger: SshConnectionWorkLedger, client: Client, + onUnhandledError: SshChannelErrorReporter | undefined, ...args: Parameters ): void { const opening = ledger.beginChannelOpen() @@ -15,7 +39,7 @@ export function forwardTrackedSshStreamLocalChannel( if (error) { opening.close(error) } else { - trackSshConnectionChannelLifetime(opening, channel) + trackSshConnectionChannelLifetime(opening, channel, onUnhandledError) } callback(error, channel) }) @@ -29,17 +53,18 @@ export function forwardTrackedSshChannel( ledger: SshConnectionWorkLedger, client: Client, localSocket: object, + onUnhandledError: SshChannelErrorReporter | undefined, ...args: Parameters ): void { const opening = ledger.beginChannelOpen() - trackSshConnectionChannelLifetime(ledger.beginChannelOpen(), localSocket) + trackSshConnectionChannelLifetime(ledger.beginChannelOpen(), localSocket, onUnhandledError) const [sourceHost, sourcePort, host, port, callback] = args try { client.forwardOut(sourceHost, sourcePort, host, port, (error, channel) => { if (error) { opening.close(error) } else { - trackSshConnectionChannelLifetime(opening, channel) + trackSshConnectionChannelLifetime(opening, channel, onUnhandledError) } callback?.(error, channel) }) From 3e5bb018a8d7ed189beeef2f7ca9d5e7feb5412b Mon Sep 17 00:00:00 2001 From: m4air Date: Thu, 1 Oct 2026 09:10:25 -0700 Subject: [PATCH 3/3] fix(ssh): name the forwarded local socket type and type the upload-session test stub The forwarded local socket now takes SshConnectionWorkChannel, the event surface the ledger tracks, instead of a broad object. The upload-session lifetime test binds an EventEmitter rather than an untyped {}. --- src/main/ssh/ssh-connection-channel-error.test.ts | 1 + src/main/ssh/ssh-connection-disconnect-drain.test.ts | 3 +++ src/main/ssh/ssh-connection-reconnect-ladder.test.ts | 8 ++++---- src/main/ssh/ssh-connection-transport-closure.test.ts | 3 +++ src/main/ssh/ssh-connection-work-drain.test.ts | 6 ++++++ src/main/ssh/ssh-connection.ts | 11 +++++++++-- src/main/ssh/ssh-forward-channel-lifetime.test.ts | 3 +++ src/main/ssh/ssh-forward-channel-lifetime.ts | 7 +++++-- src/main/ssh/ssh-upload-session-lifetime.test.ts | 3 ++- 9 files changed, 36 insertions(+), 9 deletions(-) diff --git a/src/main/ssh/ssh-connection-channel-error.test.ts b/src/main/ssh/ssh-connection-channel-error.test.ts index 9e6179a06f82..8ecede182139 100644 --- a/src/main/ssh/ssh-connection-channel-error.test.ts +++ b/src/main/ssh/ssh-connection-channel-error.test.ts @@ -49,6 +49,7 @@ it('does not log a forwarded channel error its owner handles', async () => { const client = conn.getClient()! const channel = new EventEmitter() client.openssh_forwardOutStreamLocal = vi.fn((_path, callback) => { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the mock ssh2 client hands back this EventEmitter as its channel. callback(undefined, channel as never) return client }) diff --git a/src/main/ssh/ssh-connection-disconnect-drain.test.ts b/src/main/ssh/ssh-connection-disconnect-drain.test.ts index b361e6a5ed61..b5244ff5f327 100644 --- a/src/main/ssh/ssh-connection-disconnect-drain.test.ts +++ b/src/main/ssh/ssh-connection-disconnect-drain.test.ts @@ -44,6 +44,7 @@ async function fixture(conn = new SshConnection(createTarget(), createCallbacks( const forward = () => { const channel = new EventEmitter() client.openssh_forwardOutStreamLocal = vi.fn((_path, callback) => { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the mock ssh2 client hands back this EventEmitter as its channel. callback(undefined, channel as never) return client }) @@ -190,6 +191,7 @@ it('aborts unresolved channel drain even after the physical client closes', asyn it('cleans owned overlapping clients but refuses unproven drain', async () => { const f = await fixture() const pending = { end: vi.fn(), destroy: vi.fn() } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: reads SshConnection's own private pending-client set. const clients = (f.conn as unknown as { pendingSsh2Clients: Set }).pendingSsh2Clients clients.add(pending) await expect(f.run()).rejects.toThrow() @@ -202,6 +204,7 @@ it('cleans owned overlapping clients but refuses unproven drain', async () => { it('waits for the owned proxy close as well as client close', async () => { const f = await fixture() const proxy = Object.assign(new EventEmitter(), { kill: vi.fn() }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: installs a stub proxy process into SshConnection's private field. ;(f.conn as unknown as { proxyProcess: typeof proxy }).proxyProcess = proxy const done = vi.fn() const draining = f.run().then(done) diff --git a/src/main/ssh/ssh-connection-reconnect-ladder.test.ts b/src/main/ssh/ssh-connection-reconnect-ladder.test.ts index 40b9451848ef..42d61b4d92d3 100644 --- a/src/main/ssh/ssh-connection-reconnect-ladder.test.ts +++ b/src/main/ssh/ssh-connection-reconnect-ladder.test.ts @@ -268,10 +268,10 @@ describe('SshConnection', () => { ) await connectWithFakeTimers(conn) const connectedGeneration = conn.getTransportGeneration() - const dnsFailure = new Error( - 'getaddrinfo EAI_AGAIN temporary failure in name resolution' - ) as NodeJS.ErrnoException - dnsFailure.code = 'EAI_AGAIN' + const dnsFailure = Object.assign( + new Error('getaddrinfo EAI_AGAIN temporary failure in name resolution'), + { code: 'EAI_AGAIN' } + ) ssh2Mock.connectSequence = [dnsFailure, 'ready'] emitSshEvent('close') diff --git a/src/main/ssh/ssh-connection-transport-closure.test.ts b/src/main/ssh/ssh-connection-transport-closure.test.ts index 9581eef7ec05..858c264bf2fd 100644 --- a/src/main/ssh/ssh-connection-transport-closure.test.ts +++ b/src/main/ssh/ssh-connection-transport-closure.test.ts @@ -66,6 +66,7 @@ it('holds notification through tracked work and both forwarded resources', async const socket = new EventEmitter() const channel = new EventEmitter() f.client.forwardOut = vi.fn((_a, _b, _c, _d, callback) => { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the mock ssh2 client hands back this EventEmitter as its channel. callback?.(undefined, channel as never) return f.client }) @@ -84,6 +85,7 @@ it('holds notification through tracked work and both forwarded resources', async it('waits for every allocated proxy even after its active pointer is cleared', async () => { const f = await fixture() const proxy = new EventEmitter() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: reads SshConnection's own private transport close ledger. const ledger = (f.conn as unknown as { transportCloseLedger: SshTransportCloseLedger }) .transportCloseLedger ledger.track(proxy) @@ -151,6 +153,7 @@ it.each(['initial', 'reconnect', 'system'] as const)( it('retains failed ledger drainage rather than signaling successful closure', async () => { const f = await fixture() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: reads SshConnection's own private work ledger. const work = (f.conn as unknown as { workLedger: SshConnectionWorkLedger }).workLedger const opening = work.beginChannelOpen() work.fenceForReset() diff --git a/src/main/ssh/ssh-connection-work-drain.test.ts b/src/main/ssh/ssh-connection-work-drain.test.ts index ffe08884390d..24448fe7b064 100644 --- a/src/main/ssh/ssh-connection-work-drain.test.ts +++ b/src/main/ssh/ssh-connection-work-drain.test.ts @@ -76,6 +76,7 @@ it('routes actual connection forwarding through admission and rejects a replaced const channel = new EventEmitter() const socket = new EventEmitter() client.forwardOut = vi.fn((_a, _b, _c, _d, callback) => { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the mock ssh2 client hands back this EventEmitter as its channel. callback?.(undefined, channel as never) return client }) @@ -85,6 +86,7 @@ it('routes actual connection forwarding through admission and rejects a replaced conn.forwardOut(client, new EventEmitter(), '127.0.0.1', 0, 'remote', 443, vi.fn()) ).toThrow('admission_closed') expect(() => + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a stand-in for a different ssh2 client; only its identity is compared. conn.forwardOut({} as never, new EventEmitter(), '127.0.0.1', 0, 'remote', 443, vi.fn()) ).toThrow('client_changed') channel.emit('close') @@ -113,6 +115,7 @@ it('tracks exact remote socket forwarding through a late open and physical close expect(() => fence.assertDrained()).toThrow('not_drained') expect(() => conn.forwardStreamLocal(client, '/other.sock', vi.fn())).toThrow('admission_closed') const channel = new EventEmitter() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the mock ssh2 client hands back this EventEmitter as its channel. accept!(undefined, channel as never) expect(received).toHaveBeenCalledWith(undefined, channel) expect(() => fence.assertDrained()).toThrow('not_drained') @@ -126,6 +129,7 @@ it('rejects stale clients and non-Unix endpoints before remote socket forwarding await conn.connect() const client = conn.getClient()! client.openssh_forwardOutStreamLocal = vi.fn() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a stand-in for a different ssh2 client; only its identity is compared. expect(() => conn.forwardStreamLocal({} as never, '/saved.sock', vi.fn())).toThrow( 'client_changed' ) @@ -160,6 +164,7 @@ it('surfaces remote socket extension refusal without a fallback', async () => { conn.forwardStreamLocal(client, '/saved.sock', received) const fence = fenceSshConnectionWork(conn) const refused = new Error('extension unsupported') + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: ssh2 passes no channel on a refused open; the callback type omits that case. accept!(refused, undefined as never) expect(received).toHaveBeenCalledWith(refused, undefined) await expect(fence.drain(new AbortController().signal)).rejects.toThrow('extension unsupported') @@ -172,6 +177,7 @@ it('retains remote socket channel errors through physical close', async () => { const client = conn.getClient()! const channel = new EventEmitter() client.openssh_forwardOutStreamLocal = vi.fn((_path, callback) => { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the mock ssh2 client hands back this EventEmitter as its channel. callback(undefined, channel as never) return client }) diff --git a/src/main/ssh/ssh-connection.ts b/src/main/ssh/ssh-connection.ts index 0436b97b3225..0fee9e369042 100644 --- a/src/main/ssh/ssh-connection.ts +++ b/src/main/ssh/ssh-connection.ts @@ -2,7 +2,10 @@ import * as net from 'node:net' import { Client as SshClient } from 'ssh2' import { sshProxyRouteDigest, type SshConnectionDestination } from './ssh-connection-destination' -import { SshConnectionWorkLedger } from './ssh-connection-work-ledger' +import { + SshConnectionWorkLedger, + type SshConnectionWorkChannel +} from './ssh-connection-work-ledger' import { SshTransportCloseLedger } from './ssh-transport-close-ledger' import { SshTransportClosureSubscribers } from './ssh-connection-transport-closure' import { disconnectAndAwaitSshTransportClose } from './ssh-connection-close-drain' @@ -293,7 +296,11 @@ export class SshConnection { openForwardSocket(open: () => T): T { return openTrackedSshSocket(this.workLedger, open, this.reportUnhandledChannelError('socket')) } - forwardOut(client: SshClient, localSocket: object, ...args: Parameters) { + forwardOut( + client: SshClient, + localSocket: SshConnectionWorkChannel, + ...args: Parameters + ): void { assertSshForwardClient(this.client, client) const report = this.reportUnhandledChannelError('forward') forwardTrackedSshChannel(this.workLedger, client, localSocket, report, ...args) diff --git a/src/main/ssh/ssh-forward-channel-lifetime.test.ts b/src/main/ssh/ssh-forward-channel-lifetime.test.ts index 449d663f0dfe..ae9bc18ef76d 100644 --- a/src/main/ssh/ssh-forward-channel-lifetime.test.ts +++ b/src/main/ssh/ssh-forward-channel-lifetime.test.ts @@ -7,6 +7,7 @@ import { forwardTrackedSshChannel } from './ssh-forward-channel-lifetime' function fixture() { const ledger = new SshConnectionWorkLedger() const local = new EventEmitter() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the ledger only needs the channel's EventEmitter surface. const remote = new EventEmitter() as ClientChannel let opened!: NonNullable[4]> const client = { @@ -18,6 +19,7 @@ function fixture() { const start = () => forwardTrackedSshChannel( ledger, + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the stub implements only the forwardOut the helper calls. client as unknown as Client, local, undefined, @@ -67,6 +69,7 @@ it('retains open failures during reset rather than claiming successful drain', a f.start() const fence = f.ledger.fenceForReset() const failure = new Error('forward failed') + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: ssh2 passes no channel on a refused open; the callback type omits that case. f.opened()(failure, undefined as never) f.local.emit('close') await expect(fence.drain(new AbortController().signal)).rejects.toBe(failure) diff --git a/src/main/ssh/ssh-forward-channel-lifetime.ts b/src/main/ssh/ssh-forward-channel-lifetime.ts index af1d59a4f600..902b40b66eed 100644 --- a/src/main/ssh/ssh-forward-channel-lifetime.ts +++ b/src/main/ssh/ssh-forward-channel-lifetime.ts @@ -3,7 +3,10 @@ import { trackSshConnectionChannelLifetime, type SshChannelErrorReporter } from './ssh-connection-channel-lifetime' -import type { SshConnectionWorkLedger } from './ssh-connection-work-ledger' +import type { + SshConnectionWorkChannel, + SshConnectionWorkLedger +} from './ssh-connection-work-ledger' /** A forward must ride the connection's current client, never one a reconnect replaced. */ export function assertSshForwardClient(current: Client | null, requested: Client): void { @@ -52,7 +55,7 @@ export function forwardTrackedSshStreamLocalChannel( export function forwardTrackedSshChannel( ledger: SshConnectionWorkLedger, client: Client, - localSocket: object, + localSocket: SshConnectionWorkChannel, onUnhandledError: SshChannelErrorReporter | undefined, ...args: Parameters ): void { diff --git a/src/main/ssh/ssh-upload-session-lifetime.test.ts b/src/main/ssh/ssh-upload-session-lifetime.test.ts index 25a27b9caddd..adaa13642ff1 100644 --- a/src/main/ssh/ssh-upload-session-lifetime.test.ts +++ b/src/main/ssh/ssh-upload-session-lifetime.test.ts @@ -1,3 +1,4 @@ +import { EventEmitter } from 'node:events' import { expect, it, vi } from 'vitest' import type { FileUploadSession } from '../providers/filesystem-provider-contract' import { SshConnectionWorkLedger } from './ssh-connection-work-ledger' @@ -76,7 +77,7 @@ it('retains pending factory admission across the fence and independently tracks await release.promise await ledger.run(async () => { channel = ledger.beginChannelOpen() - channel.bind({}) + channel.bind(new EventEmitter()) }) return makeSession() })