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..8ecede182139 --- /dev/null +++ b/src/main/ssh/ssh-connection-channel-error.test.ts @@ -0,0 +1,63 @@ +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) => { + // 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 + }) + 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-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..b5244ff5f327 --- /dev/null +++ b/src/main/ssh/ssh-connection-disconnect-drain.test.ts @@ -0,0 +1,243 @@ +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) => { + // 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 + }) + 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() } + // 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() + 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() }) + // 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) + 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-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-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..42d61b4d92d3 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 = Object.assign( + new Error('getaddrinfo EAI_AGAIN temporary failure in name resolution'), + { 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..858c264bf2fd --- /dev/null +++ b/src/main/ssh/ssh-connection-transport-closure.test.ts @@ -0,0 +1,179 @@ +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) => { + // 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 + }) + 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() + // 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) + 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() + // 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() + 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..24448fe7b064 --- /dev/null +++ b/src/main/ssh/ssh-connection-work-drain.test.ts @@ -0,0 +1,244 @@ +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) => { + // 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 + }) + 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(() => + // 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') + 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() + // 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') + 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() + // 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' + ) + 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') + // 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') + 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) => { + // 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 + }) + 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..0fee9e369042 100644 --- a/src/main/ssh/ssh-connection.ts +++ b/src/main/ssh/ssh-connection.ts @@ -1,6 +1,34 @@ /* 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, + 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' +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' +import { + openTrackedSshSocket, + trackSshConnectionChannelLifetime +} from './ssh-connection-channel-lifetime' import type { ChildProcess } from 'node:child_process' import type { ClientChannel, @@ -16,11 +44,6 @@ import { getOrcaControlSocketPath, spawnSystemSsh, spawnSystemSshCommand, - downloadFileViaSystemSsh, - uploadDirectoryViaSystemSsh, - uploadFileViaSystemSsh, - writeBufferViaSystemSsh, - writeFileViaSystemSsh, type SystemSshBuildArgsOptions, type SystemSshProcess } from './ssh-system-fallback' @@ -83,18 +106,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 = { @@ -104,12 +118,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 @@ -118,13 +126,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 @@ -200,6 +201,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 +246,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 +269,60 @@ export class SshConnection { getConnectGeneration(): number { return this.connectGeneration } + /** Equals the connect generation; later slices name it by the transport it identifies. */ + 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, this.reportUnhandledChannelError('socket')) + } + forwardOut( + client: SshClient, + localSocket: SshConnectionWorkChannel, + ...args: Parameters + ): void { + assertSshForwardClient(this.client, client) + const report = this.reportUnhandledChannelError('forward') + forwardTrackedSshChannel(this.workLedger, client, localSocket, report, ...args) + } + forwardStreamLocal( + client: SshClient, + ...args: Parameters + ): void { + 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 } @@ -269,7 +357,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() } @@ -284,21 +376,31 @@ 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 ) } /** 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') } @@ -306,16 +408,26 @@ 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') ) ) } - 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) { @@ -328,373 +440,95 @@ 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 { - return new Promise((resolve, reject) => { - type ChannelOpenTerminationError = Error & { sshChannelCloseConfirmed: boolean } - let settled = false - let unconfirmedOpenError: ChannelOpenTerminationError | null = null - const markOpenUnconfirmed = (error: Error): 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 => { - 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() - 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))) - } - }) - } - - async uploadDirectory( + uploadDirectory( 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() - } } - async downloadFile( + downloadFile( 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) + ) } - async openFileUploadSession(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: () => {} - } + openFileUploadSession(options?: SshRemoteFileOptions): Promise { + return openTrackedSshUploadSession(this.workLedger, () => + createSshFileUploadSession(this.fileTransferHost(), options) + ) } - async writeFile( + writeFile( 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() - } } - async writeBuffer( + writeBuffer( 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 + return this.workLedger.run(() => + writeSshBuffer(this.fileTransferHost(), remotePath, contents, options) + ) + } + + 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 { + 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 +536,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 +654,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 +703,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 +719,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 +795,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 +830,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 +880,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 +895,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 +906,7 @@ export class SshConnection { } async reconnect(): Promise { + this.assertConnectionReplacementAllowed() if (this.disposed || this.state.status === 'connecting') { return } @@ -1084,6 +923,7 @@ export class SshConnection { } private async doSystemSshProbe(connectGeneration: number): Promise { + this.unprovenStartupTransport = true this.useSystemSshTransport = true this.client = null this.proxyProcess?.kill() @@ -1244,6 +1084,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 +1175,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, this.reportUnhandledChannelError('command')) + } catch (error) { + work.close(error) + throw error + } this.systemCommandChannels.add(channel) const onAbort = (): void => { channel.close() @@ -1381,6 +1230,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 +1283,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 +1335,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 +1358,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 +1410,7 @@ export class SshConnection { connectGeneration === this.connectGeneration ) { this.hostKeyFingerprint = decision.fingerprint + acceptedFingerprint = decision.fingerprint } if (decision.action === 'reject') { hostKeyRejection = new HostKeyVerificationError( @@ -1647,6 +1509,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 +1580,7 @@ export class SshConnection { if (this.disposed || this.client !== client) { return } + this.workLedger.markTransportUnverifiable() this.client = null this.scheduleReconnect() } @@ -1714,13 +1591,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 +1617,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 +1625,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 +1713,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 +1761,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 +1780,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 +1862,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..ae9bc18ef76d --- /dev/null +++ b/src/main/ssh/ssh-forward-channel-lifetime.test.ts @@ -0,0 +1,102 @@ +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() + // 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 = { + forwardOut: vi.fn((_a, _b, _c, _d, callback) => { + opened = callback + }) + } + const callback = vi.fn() + 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, + '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') + // 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) +}) + +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..902b40b66eed --- /dev/null +++ b/src/main/ssh/ssh-forward-channel-lifetime.ts @@ -0,0 +1,78 @@ +import type { Client } from 'ssh2' +import { + trackSshConnectionChannelLifetime, + type SshChannelErrorReporter +} from './ssh-connection-channel-lifetime' +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 { + 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() + const [socketPath, callback] = args + try { + client.openssh_forwardOutStreamLocal(socketPath, (error, channel) => { + if (error) { + opening.close(error) + } else { + trackSshConnectionChannelLifetime(opening, channel, onUnhandledError) + } + 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: SshConnectionWorkChannel, + onUnhandledError: SshChannelErrorReporter | undefined, + ...args: Parameters +): void { + const opening = ledger.beginChannelOpen() + 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, onUnhandledError) + } + 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..adaa13642ff1 --- /dev/null +++ b/src/main/ssh/ssh-upload-session-lifetime.test.ts @@ -0,0 +1,193 @@ +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' +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(new EventEmitter()) + }) + 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