mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
refactor(ssh): keep ssh2 types out of the orcad host transport (#27689)
The orcad/remote-install transport now exposes exec as a neutral RemoteCommandChannel (a Duplex with stdin, stderr, close and exit/close events) instead of ssh2's ClientChannel, and drops sftp(): the only consumer was a fallback for doubles without writeFile/uploadDirectory, which the transport type already requires. SshConnection still satisfies the type structurally, so the SSH adapter keeps returning the connection itself and identity-keyed caches are unchanged. The stdio bridge provider is split: the loopback listener (admission, queueing, socket piping) now takes a channel factory and knows nothing of SSH; the SSH provider resolves and checks the bridge over the host transport and hands the listener a factory. No behavior or wire change.
This commit is contained in:
@@ -0,0 +1,169 @@
|
||||
/**
|
||||
* The local half of the stdio bridge: a loopback listener whose every accepted socket rides its
|
||||
* own bridge channel from a channel factory. Owns admission, queueing and teardown; knows nothing
|
||||
* of the host transport the channels run over.
|
||||
*/
|
||||
import { createServer, type Socket } from 'node:net'
|
||||
import type { OrcadStdioBridgeMode } from './orcad-stdio-bridge-script'
|
||||
import {
|
||||
OrcadStdioBridgeBase64Encoder,
|
||||
OrcadStdioBridgeDecoder,
|
||||
type OrcadStdioBridgeSignal
|
||||
} from './orcad-stdio-bridge-stream'
|
||||
import type { RemoteCommandChannel } from './remote-install-transport'
|
||||
import type { PortForwardStartOptions, StartedPortForward } from './ssh-port-forward-provider'
|
||||
import { listenForPortForward } from './ssh2-port-forward-provider'
|
||||
|
||||
/**
|
||||
* Each bridge is one sshd session, and OpenSSH's MaxSessions defaults to 10 per connection. The
|
||||
* client holds one shared control socket and one per stream (terminals, each browser), so 8
|
||||
* covers it and leaves room for the exec work sharing the connection. Past it, sockets queue.
|
||||
*/
|
||||
export const ORCAD_STDIO_BRIDGE_MAX_CHANNELS = 8
|
||||
export const ORCAD_STDIO_BRIDGE_QUEUE_TIMEOUT_MS = 30_000
|
||||
|
||||
/** Opens one bridge channel for one accepted socket; a rejection just drops that socket. */
|
||||
export type OrcadStdioBridgeChannelFactory = () => Promise<{
|
||||
channel: RemoteCommandChannel
|
||||
mode: OrcadStdioBridgeMode
|
||||
}>
|
||||
|
||||
export async function listenOrcadStdioBridge(
|
||||
options: PortForwardStartOptions,
|
||||
openChannel: OrcadStdioBridgeChannelFactory
|
||||
): Promise<StartedPortForward> {
|
||||
const sockets = new Set<Socket>()
|
||||
const channels = new Set<RemoteCommandChannel>()
|
||||
const waiting: { socket: Socket; timer: ReturnType<typeof setTimeout> }[] = []
|
||||
let open = 0
|
||||
let closed = false
|
||||
|
||||
const admitNext = (): void => {
|
||||
while (open < ORCAD_STDIO_BRIDGE_MAX_CHANNELS) {
|
||||
const next = waiting.shift()
|
||||
if (!next) {
|
||||
return
|
||||
}
|
||||
clearTimeout(next.timer)
|
||||
void bridgeSocket(next.socket)
|
||||
}
|
||||
}
|
||||
const release = (): void => {
|
||||
open -= 1
|
||||
admitNext()
|
||||
}
|
||||
|
||||
const bridgeSocket = async (socket: Socket): Promise<void> => {
|
||||
if (closed || socket.destroyed) {
|
||||
return
|
||||
}
|
||||
open += 1
|
||||
let opened: Awaited<ReturnType<OrcadStdioBridgeChannelFactory>>
|
||||
try {
|
||||
opened = await openChannel()
|
||||
} catch {
|
||||
socket.destroy()
|
||||
release()
|
||||
return
|
||||
}
|
||||
const { channel, mode } = opened
|
||||
channel.once('close', release)
|
||||
if (closed || socket.destroyed) {
|
||||
closeChannel(channel)
|
||||
return
|
||||
}
|
||||
channels.add(channel)
|
||||
channel.once('close', () => channels.delete(channel))
|
||||
pipeSocketThroughBridge(socket, channel, mode)
|
||||
}
|
||||
|
||||
const server = createServer((socket) => {
|
||||
sockets.add(socket)
|
||||
socket.on('error', () => socket.destroy())
|
||||
const entry = {
|
||||
socket,
|
||||
timer: setTimeout(() => socket.destroy(), ORCAD_STDIO_BRIDGE_QUEUE_TIMEOUT_MS)
|
||||
}
|
||||
socket.on('close', () => {
|
||||
sockets.delete(socket)
|
||||
clearTimeout(entry.timer)
|
||||
const index = waiting.indexOf(entry)
|
||||
if (index !== -1) {
|
||||
waiting.splice(index, 1)
|
||||
}
|
||||
})
|
||||
waiting.push(entry)
|
||||
admitNext()
|
||||
})
|
||||
|
||||
await listenForPortForward(server, options.localHost, options.localPort)
|
||||
const address = server.address()
|
||||
const localPort = typeof address === 'object' && address ? address.port : options.localPort
|
||||
const entry = {
|
||||
id: options.id,
|
||||
connectionId: options.connectionId,
|
||||
localPort,
|
||||
remoteHost: options.remoteHost,
|
||||
remotePort: options.remotePort,
|
||||
label: options.label
|
||||
}
|
||||
|
||||
const close = (): Promise<void> => {
|
||||
if (closed) {
|
||||
return Promise.resolve()
|
||||
}
|
||||
closed = true
|
||||
for (const { timer } of waiting.splice(0)) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
for (const channel of channels) {
|
||||
closeChannel(channel)
|
||||
}
|
||||
for (const socket of sockets) {
|
||||
socket.destroy()
|
||||
}
|
||||
return new Promise((resolve) => server.close(() => resolve()))
|
||||
}
|
||||
return { entry, close, dispose: () => void close() }
|
||||
}
|
||||
|
||||
function pipeSocketThroughBridge(
|
||||
socket: Socket,
|
||||
channel: RemoteCommandChannel,
|
||||
mode: OrcadStdioBridgeMode
|
||||
): void {
|
||||
const decoder = new OrcadStdioBridgeDecoder(mode)
|
||||
// A refused dial is what the forward's failed channel open is: the socket just closes.
|
||||
decoder.once('signal', (signal: OrcadStdioBridgeSignal) => {
|
||||
if (signal !== 'ready') {
|
||||
socket.destroy()
|
||||
closeChannel(channel)
|
||||
}
|
||||
})
|
||||
decoder.on('error', () => {
|
||||
socket.destroy()
|
||||
closeChannel(channel)
|
||||
})
|
||||
channel.pipe(decoder).pipe(socket)
|
||||
if (mode === 'base64') {
|
||||
socket.pipe(new OrcadStdioBridgeBase64Encoder()).pipe(channel)
|
||||
} else {
|
||||
socket.pipe(channel)
|
||||
}
|
||||
channel.stderr.resume()
|
||||
channel.on('error', () => socket.destroy())
|
||||
channel.on('close', () => {
|
||||
if (!socket.writableEnded) {
|
||||
socket.destroy()
|
||||
}
|
||||
})
|
||||
socket.on('close', () => closeChannel(channel))
|
||||
}
|
||||
|
||||
function closeChannel(channel: RemoteCommandChannel): void {
|
||||
try {
|
||||
channel.close()
|
||||
} catch {
|
||||
// Late callbacks after the transport dropped.
|
||||
}
|
||||
}
|
||||
@@ -3,18 +3,16 @@ import { connect, type Server, type Socket } from 'node:net'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { PassThrough } from 'node:stream'
|
||||
import type { ClientChannel } from 'ssh2'
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest'
|
||||
import { OrcadStdioBridgeUnavailableError } from './orcad-host-unavailable'
|
||||
import type { OrcadStdioBridge } from './orcad-stdio-bridge'
|
||||
import {
|
||||
ORCAD_STDIO_BRIDGE_MAX_CHANNELS,
|
||||
OrcadStdioBridgePortForwardProvider
|
||||
} from './orcad-stdio-bridge-provider'
|
||||
import { ORCAD_STDIO_BRIDGE_MAX_CHANNELS } from './orcad-stdio-bridge-listener'
|
||||
import { OrcadStdioBridgePortForwardProvider } from './orcad-stdio-bridge-provider'
|
||||
import { spawnLocalBridgeChannel, startEchoServer } from './orcad-stdio-bridge-test-channel'
|
||||
import { ORCAD_WINDOWS_HOST_SCRIPT } from './orcad-windows-host-script'
|
||||
import type { SshConnection } from './ssh-connection'
|
||||
import type { StartedPortForward } from './ssh-port-forward-provider'
|
||||
import type { RemoteCommandChannel } from './remote-install-transport'
|
||||
|
||||
let echo: { server: Server; port: number }
|
||||
let script: string
|
||||
@@ -46,7 +44,7 @@ const conn = { getClient: () => ({}) } as unknown as SshConnection
|
||||
const bridge: OrcadStdioBridge = { command: 'bridge', mode: 'base64', wrapCommand: false }
|
||||
|
||||
/** The real bridge, run locally through the Windows host script's base64 framing. */
|
||||
const localBridge = (): ClientChannel =>
|
||||
const localBridge = (): RemoteCommandChannel =>
|
||||
spawnLocalBridgeChannel(process.execPath, [script, 'stdio-bridge', String(echo.port)])
|
||||
|
||||
async function startProvider(
|
||||
@@ -138,7 +136,7 @@ describe('the managed tunnel over the stdio bridge', () => {
|
||||
const channel = Object.assign(new PassThrough(), { stderr: new PassThrough(), close() {} })
|
||||
channels.push(channel)
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the provider uses only the stream, `stderr`, `close`, and events.
|
||||
return channel as unknown as ClientChannel
|
||||
return channel as unknown as RemoteCommandChannel
|
||||
})
|
||||
const forward = await startProvider({ openChannel })
|
||||
const lost = await dial(forward.entry.localPort)
|
||||
@@ -158,7 +156,7 @@ describe('the managed tunnel over the stdio bridge', () => {
|
||||
})
|
||||
channels.push(channel)
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the provider uses only the stream, `stderr`, `close`, and events.
|
||||
return channel as unknown as ClientChannel
|
||||
return channel as unknown as RemoteCommandChannel
|
||||
})
|
||||
const forward = await startProvider({ openChannel })
|
||||
const sockets = await Promise.all(
|
||||
|
||||
@@ -6,40 +6,28 @@
|
||||
* Disconnects match the forward: a lost channel drops its socket and the listener stays, so the
|
||||
* tunnel's own reconnect logic decides, and a lost transport never reads as a dead server.
|
||||
*/
|
||||
import { createServer, type Socket } from 'node:net'
|
||||
import type { ClientChannel } from 'ssh2'
|
||||
import { OrcadStdioBridgeUnavailableError } from './orcad-host-unavailable'
|
||||
import type { OrcadHostTransport } from './orcad-host-transport'
|
||||
import {
|
||||
checkOrcadStdioBridge,
|
||||
openOrcadStdioBridgeChannel,
|
||||
resolveOrcadStdioBridge,
|
||||
type OrcadStdioBridge
|
||||
} from './orcad-stdio-bridge'
|
||||
import {
|
||||
OrcadStdioBridgeBase64Encoder,
|
||||
OrcadStdioBridgeDecoder,
|
||||
type OrcadStdioBridgeSignal
|
||||
} from './orcad-stdio-bridge-stream'
|
||||
import { listenOrcadStdioBridge } from './orcad-stdio-bridge-listener'
|
||||
import { sshOrcadHostTransport } from './orcad-ssh-host-transport'
|
||||
import type { RemoteCommandChannel } from './remote-install-transport'
|
||||
import type { SshConnection } from './ssh-connection'
|
||||
import type {
|
||||
PortForwardStartOptions,
|
||||
SshPortForwardProvider,
|
||||
StartedPortForward
|
||||
} from './ssh-port-forward-provider'
|
||||
import { listenForPortForward } from './ssh2-port-forward-provider'
|
||||
|
||||
/**
|
||||
* Each bridge is one sshd session, and OpenSSH's MaxSessions defaults to 10 per connection. The
|
||||
* client holds one shared control socket and one per stream (terminals, each browser), so 8
|
||||
* covers it and leaves room for the exec work sharing the connection. Past it, sockets queue.
|
||||
*/
|
||||
export const ORCAD_STDIO_BRIDGE_MAX_CHANNELS = 8
|
||||
export const ORCAD_STDIO_BRIDGE_QUEUE_TIMEOUT_MS = 30_000
|
||||
|
||||
type OrcadStdioBridgeDependencies = {
|
||||
resolveBridge: (conn: SshConnection, port: number) => Promise<OrcadStdioBridge>
|
||||
checkBridge: (conn: SshConnection, bridge: OrcadStdioBridge) => Promise<unknown>
|
||||
openChannel: (conn: SshConnection, bridge: OrcadStdioBridge) => Promise<ClientChannel>
|
||||
resolveBridge: (conn: OrcadHostTransport, port: number) => Promise<OrcadStdioBridge>
|
||||
checkBridge: (conn: OrcadHostTransport, bridge: OrcadStdioBridge) => Promise<unknown>
|
||||
openChannel: (conn: OrcadHostTransport, bridge: OrcadStdioBridge) => Promise<RemoteCommandChannel>
|
||||
}
|
||||
|
||||
const defaultDependencies: OrcadStdioBridgeDependencies = {
|
||||
@@ -57,161 +45,36 @@ export class OrcadStdioBridgePortForwardProvider implements SshPortForwardProvid
|
||||
return conn.getClient() !== null
|
||||
}
|
||||
|
||||
async start(conn: SshConnection, options: PortForwardStartOptions): Promise<StartedPortForward> {
|
||||
const deps = this.dependencies
|
||||
let bridge: Promise<OrcadStdioBridge> | null = null
|
||||
// Memoized once it resolves; a failed resolve is retried by the next socket.
|
||||
const resolveBridge = (): Promise<OrcadStdioBridge> => {
|
||||
bridge ??= deps.resolveBridge(conn, options.remotePort).catch((error: unknown) => {
|
||||
bridge = null
|
||||
throw error
|
||||
})
|
||||
return bridge
|
||||
}
|
||||
try {
|
||||
await deps.checkBridge(conn, await resolveBridge())
|
||||
} catch (error) {
|
||||
if (error instanceof OrcadStdioBridgeUnavailableError) {
|
||||
throw error
|
||||
}
|
||||
// A slow or dropped check is loss of contact; sockets resolve the bridge again.
|
||||
console.warn('[ssh] Orca stdio bridge check was unverifiable:', error)
|
||||
}
|
||||
start(conn: SshConnection, options: PortForwardStartOptions): Promise<StartedPortForward> {
|
||||
return startOrcadStdioBridge(sshOrcadHostTransport(conn), options, this.dependencies)
|
||||
}
|
||||
}
|
||||
|
||||
const sockets = new Set<Socket>()
|
||||
const channels = new Set<ClientChannel>()
|
||||
const waiting: { socket: Socket; timer: ReturnType<typeof setTimeout> }[] = []
|
||||
let open = 0
|
||||
let closed = false
|
||||
|
||||
const admitNext = (): void => {
|
||||
while (open < ORCAD_STDIO_BRIDGE_MAX_CHANNELS) {
|
||||
const next = waiting.shift()
|
||||
if (!next) {
|
||||
return
|
||||
}
|
||||
clearTimeout(next.timer)
|
||||
void bridgeSocket(next.socket)
|
||||
}
|
||||
}
|
||||
const release = (): void => {
|
||||
open -= 1
|
||||
admitNext()
|
||||
}
|
||||
|
||||
const bridgeSocket = async (socket: Socket): Promise<void> => {
|
||||
if (closed || socket.destroyed) {
|
||||
return
|
||||
}
|
||||
open += 1
|
||||
let resolved: OrcadStdioBridge
|
||||
let channel: ClientChannel
|
||||
try {
|
||||
resolved = await resolveBridge()
|
||||
channel = await deps.openChannel(conn, resolved)
|
||||
} catch {
|
||||
socket.destroy()
|
||||
release()
|
||||
return
|
||||
}
|
||||
channel.once('close', release)
|
||||
if (closed || socket.destroyed) {
|
||||
closeChannel(channel)
|
||||
return
|
||||
}
|
||||
channels.add(channel)
|
||||
channel.once('close', () => channels.delete(channel))
|
||||
pipeSocketThroughBridge(socket, channel, resolved.mode)
|
||||
}
|
||||
|
||||
const server = createServer((socket) => {
|
||||
sockets.add(socket)
|
||||
socket.on('error', () => socket.destroy())
|
||||
const entry = {
|
||||
socket,
|
||||
timer: setTimeout(() => socket.destroy(), ORCAD_STDIO_BRIDGE_QUEUE_TIMEOUT_MS)
|
||||
}
|
||||
socket.on('close', () => {
|
||||
sockets.delete(socket)
|
||||
clearTimeout(entry.timer)
|
||||
const index = waiting.indexOf(entry)
|
||||
if (index !== -1) {
|
||||
waiting.splice(index, 1)
|
||||
}
|
||||
})
|
||||
waiting.push(entry)
|
||||
admitNext()
|
||||
async function startOrcadStdioBridge(
|
||||
host: OrcadHostTransport,
|
||||
options: PortForwardStartOptions,
|
||||
deps: OrcadStdioBridgeDependencies
|
||||
): Promise<StartedPortForward> {
|
||||
let bridge: Promise<OrcadStdioBridge> | null = null
|
||||
// Memoized once it resolves; a failed resolve is retried by the next socket.
|
||||
const resolveBridge = (): Promise<OrcadStdioBridge> => {
|
||||
bridge ??= deps.resolveBridge(host, options.remotePort).catch((error: unknown) => {
|
||||
bridge = null
|
||||
throw error
|
||||
})
|
||||
|
||||
await listenForPortForward(server, options.localHost, options.localPort)
|
||||
const address = server.address()
|
||||
const localPort = typeof address === 'object' && address ? address.port : options.localPort
|
||||
const entry = {
|
||||
id: options.id,
|
||||
connectionId: options.connectionId,
|
||||
localPort,
|
||||
remoteHost: options.remoteHost,
|
||||
remotePort: options.remotePort,
|
||||
label: options.label
|
||||
}
|
||||
|
||||
const close = (): Promise<void> => {
|
||||
if (closed) {
|
||||
return Promise.resolve()
|
||||
}
|
||||
closed = true
|
||||
for (const { timer } of waiting.splice(0)) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
for (const channel of channels) {
|
||||
closeChannel(channel)
|
||||
}
|
||||
for (const socket of sockets) {
|
||||
socket.destroy()
|
||||
}
|
||||
return new Promise((resolve) => server.close(() => resolve()))
|
||||
}
|
||||
return { entry, close, dispose: () => void close() }
|
||||
return bridge
|
||||
}
|
||||
}
|
||||
|
||||
function pipeSocketThroughBridge(
|
||||
socket: Socket,
|
||||
channel: ClientChannel,
|
||||
mode: OrcadStdioBridge['mode']
|
||||
): void {
|
||||
const decoder = new OrcadStdioBridgeDecoder(mode)
|
||||
// A refused dial is what the forward's failed channel open is: the socket just closes.
|
||||
decoder.once('signal', (signal: OrcadStdioBridgeSignal) => {
|
||||
if (signal !== 'ready') {
|
||||
socket.destroy()
|
||||
closeChannel(channel)
|
||||
}
|
||||
})
|
||||
decoder.on('error', () => {
|
||||
socket.destroy()
|
||||
closeChannel(channel)
|
||||
})
|
||||
channel.pipe(decoder).pipe(socket)
|
||||
if (mode === 'base64') {
|
||||
socket.pipe(new OrcadStdioBridgeBase64Encoder()).pipe(channel)
|
||||
} else {
|
||||
socket.pipe(channel)
|
||||
}
|
||||
channel.stderr.resume()
|
||||
channel.on('error', () => socket.destroy())
|
||||
channel.on('close', () => {
|
||||
if (!socket.writableEnded) {
|
||||
socket.destroy()
|
||||
}
|
||||
})
|
||||
socket.on('close', () => closeChannel(channel))
|
||||
}
|
||||
|
||||
function closeChannel(channel: ClientChannel): void {
|
||||
try {
|
||||
channel.close()
|
||||
} catch {
|
||||
// Late callbacks after the transport dropped.
|
||||
await deps.checkBridge(host, await resolveBridge())
|
||||
} catch (error) {
|
||||
if (error instanceof OrcadStdioBridgeUnavailableError) {
|
||||
throw error
|
||||
}
|
||||
// A slow or dropped check is loss of contact; sockets resolve the bridge again.
|
||||
console.warn('[ssh] Orca stdio bridge check was unverifiable:', error)
|
||||
}
|
||||
return listenOrcadStdioBridge(options, async () => {
|
||||
const resolved = await resolveBridge()
|
||||
return { channel: await deps.openChannel(host, resolved), mode: resolved.mode }
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1,14 +1,14 @@
|
||||
/** Test-only: a bridge run as a local child, shaped like the SSH exec channel that runs it. */
|
||||
/** Test-only: a bridge run as a local child, shaped like the host exec channel that runs it. */
|
||||
import { createServer, type Server } from 'node:net'
|
||||
import { Duplex } from 'node:stream'
|
||||
import type { ClientChannel } from 'ssh2'
|
||||
import { spawnProcess } from '@orca/process-host'
|
||||
import type { RemoteCommandChannel } from './remote-install-transport'
|
||||
|
||||
export function spawnLocalBridgeChannel(
|
||||
program: string,
|
||||
args: readonly string[],
|
||||
env: NodeJS.ProcessEnv = process.env
|
||||
): ClientChannel {
|
||||
): RemoteCommandChannel {
|
||||
const child = spawnProcess({ program, args, env })
|
||||
// Why emitClose off: like ssh2, `close` comes once, after the exit status.
|
||||
const channel = new Duplex({
|
||||
@@ -34,7 +34,7 @@ export function spawnLocalBridgeChannel(
|
||||
child.on('close', () => channel.emit('close'))
|
||||
Object.assign(channel, { stderr: child.stderr, close: () => child.kill() })
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the bridge paths use only the stream, `stderr`, `close`, and the `exit`/`close` events assigned above.
|
||||
return channel as unknown as ClientChannel
|
||||
return channel as unknown as RemoteCommandChannel
|
||||
}
|
||||
|
||||
/** A loopback echo server standing in for orcad. */
|
||||
|
||||
@@ -4,7 +4,6 @@ import { join } from 'node:path'
|
||||
import type { Server } from 'node:net'
|
||||
import type { EventEmitter } from 'node:events'
|
||||
import { PassThrough } from 'node:stream'
|
||||
import type { ClientChannel } from 'ssh2'
|
||||
import { afterAll, beforeAll, describe, expect, it } from 'vitest'
|
||||
import { OrcadStdioBridgeUnavailableError } from './orcad-host-unavailable'
|
||||
import { checkOrcadStdioBridge, type OrcadStdioBridge } from './orcad-stdio-bridge'
|
||||
@@ -16,6 +15,7 @@ import { OrcadStdioBridgeBase64Encoder, OrcadStdioBridgeDecoder } from './orcad-
|
||||
import { spawnLocalBridgeChannel, startEchoServer } from './orcad-stdio-bridge-test-channel'
|
||||
import { ORCAD_WINDOWS_HOST_SCRIPT } from './orcad-windows-host-script'
|
||||
import type { OrcadHostTransport } from './orcad-host-transport'
|
||||
import type { RemoteCommandChannel } from './remote-install-transport'
|
||||
|
||||
const POSIX = process.platform !== 'win32'
|
||||
let echo: { server: Server; port: number }
|
||||
@@ -34,7 +34,10 @@ afterAll(() => {
|
||||
/** Every byte value, so a text layer anywhere in the path would show. */
|
||||
const PAYLOAD = Buffer.from(Array.from({ length: 4096 }, (_, index) => index % 256))
|
||||
|
||||
async function roundTrip(channel: ClientChannel, mode: OrcadStdioBridgeMode): Promise<Buffer> {
|
||||
async function roundTrip(
|
||||
channel: RemoteCommandChannel,
|
||||
mode: OrcadStdioBridgeMode
|
||||
): Promise<Buffer> {
|
||||
const decoder = new OrcadStdioBridgeDecoder(mode)
|
||||
channel.pipe(decoder)
|
||||
const input = mode === 'base64' ? new OrcadStdioBridgeBase64Encoder() : new PassThrough()
|
||||
@@ -67,7 +70,7 @@ function posixHome(withRuntime: boolean): string {
|
||||
return home
|
||||
}
|
||||
|
||||
function connectionRunning(channel: () => ClientChannel): OrcadHostTransport {
|
||||
function connectionRunning(channel: () => RemoteCommandChannel): OrcadHostTransport {
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the check calls only exec.
|
||||
return { exec: async () => channel() } as unknown as OrcadHostTransport
|
||||
}
|
||||
@@ -164,13 +167,13 @@ describe('reading a bridge channel', () => {
|
||||
})
|
||||
|
||||
/** A channel the test drives by hand. */
|
||||
function scriptedChannel(): ClientChannel & EventEmitter & { stderr: PassThrough } {
|
||||
function scriptedChannel(): RemoteCommandChannel & EventEmitter & { stderr: PassThrough } {
|
||||
const channel = Object.assign(new PassThrough(), {
|
||||
stderr: new PassThrough(),
|
||||
close: () => {}
|
||||
})
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the check uses only the stream, `stderr`, `close`, and events.
|
||||
return channel as unknown as ClientChannel & EventEmitter & { stderr: PassThrough }
|
||||
return channel as unknown as RemoteCommandChannel & EventEmitter & { stderr: PassThrough }
|
||||
}
|
||||
|
||||
describe('checking a bridge before the tunnel relies on it', () => {
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
/** Which command runs the stdio bridge on a host, and whether it can run there at all. */
|
||||
import type { ClientChannel } from 'ssh2'
|
||||
import {
|
||||
OrcadHostUnsupportedError,
|
||||
OrcadStdioBridgeUnavailableError
|
||||
@@ -12,6 +11,7 @@ import {
|
||||
} from './orcad-stdio-bridge-script'
|
||||
import { OrcadStdioBridgeDecoder, type OrcadStdioBridgeSignal } from './orcad-stdio-bridge-stream'
|
||||
import type { OrcadHostTransport } from './orcad-host-transport'
|
||||
import type { RemoteCommandChannel } from './remote-install-transport'
|
||||
import { isWindowsRemoteHost } from './ssh-remote-platform'
|
||||
import { detectRemoteHostPlatform } from './ssh-remote-platform-detection'
|
||||
|
||||
@@ -49,11 +49,11 @@ export async function resolveOrcadStdioBridge(
|
||||
export function openOrcadStdioBridgeChannel(
|
||||
conn: OrcadHostTransport,
|
||||
bridge: OrcadStdioBridge
|
||||
): Promise<ClientChannel> {
|
||||
): Promise<RemoteCommandChannel> {
|
||||
return conn.exec(bridge.command, { wrapCommand: bridge.wrapCommand })
|
||||
}
|
||||
|
||||
function closeQuietly(channel: ClientChannel): void {
|
||||
function closeQuietly(channel: RemoteCommandChannel): void {
|
||||
try {
|
||||
channel.close()
|
||||
} catch {
|
||||
@@ -70,7 +70,7 @@ export async function checkOrcadStdioBridge(
|
||||
bridge: OrcadStdioBridge,
|
||||
timeoutMs = ORCAD_STDIO_BRIDGE_CHECK_TIMEOUT_MS
|
||||
): Promise<'running' | 'unverifiable'> {
|
||||
let channel: ClientChannel
|
||||
let channel: RemoteCommandChannel
|
||||
try {
|
||||
channel = await openOrcadStdioBridgeChannel(conn, bridge)
|
||||
} catch {
|
||||
|
||||
@@ -1,16 +1,25 @@
|
||||
import type { ClientChannel, SFTPWrapper } from 'ssh2'
|
||||
import type { Duplex, Readable, Writable } from 'node:stream'
|
||||
import type { SshRemoteFileOptions } from './ssh-connection-file-transfers'
|
||||
import type { SshExecOptions } from './ssh-connection-utils'
|
||||
|
||||
type RemoteFileWriteOptions = SshRemoteFileOptions & { signal?: AbortSignal }
|
||||
|
||||
/**
|
||||
* A command running on the host: reads are its stdout, writes its stdin. Emits `exit` with the
|
||||
* exit code when the host reports one, then `close` (with that code) once the channel is gone.
|
||||
*/
|
||||
export type RemoteCommandChannel = Duplex & {
|
||||
readonly stdin: Writable
|
||||
readonly stderr: Readable
|
||||
close(): void
|
||||
}
|
||||
|
||||
/** What remote install, lock, GC and host-probe code uses from a host connection. */
|
||||
export type RemoteInstallTransport = {
|
||||
exec(command: string, options?: SshExecOptions): Promise<ClientChannel>
|
||||
exec(command: string, options?: SshExecOptions): Promise<RemoteCommandChannel>
|
||||
usesSystemSshTransport(): boolean
|
||||
/** Changes whenever the underlying transport is replaced; scopes per-transport verdicts. */
|
||||
getConnectGeneration(): number
|
||||
sftp(options?: AbortSignal | { signal?: AbortSignal }): Promise<SFTPWrapper>
|
||||
writeFile(remotePath: string, contents: string, options?: RemoteFileWriteOptions): Promise<void>
|
||||
uploadDirectory(
|
||||
localDir: string,
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
* chroot makes a late `SSH_FX_NO_SUCH_FILE` the normal answer, so the listener has to
|
||||
* outlive the transfer rather than the other way round.
|
||||
*
|
||||
* `runSftpFallbackTransfer` already does this for the session emitter; this is the same
|
||||
* `latchLateSftpSessionErrors` does this for the session emitter; this is the same
|
||||
* guarantee one level down, on the streams.
|
||||
*/
|
||||
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import type { Readable } from 'node:stream'
|
||||
import type { ClientChannel } from 'ssh2'
|
||||
import type { SshConnection } from './ssh-connection'
|
||||
import type { RemoteCommandChannel, RemoteInstallTransport } from './remote-install-transport'
|
||||
import { createSshOperationAbortError, type SshExecOptions } from './ssh-connection-utils'
|
||||
import {
|
||||
redactRelayInstallMarkerError,
|
||||
@@ -74,7 +73,7 @@ export function sshCommandExitError(
|
||||
}
|
||||
|
||||
export async function execCommand(
|
||||
conn: Pick<SshConnection, 'exec' | 'usesSystemSshTransport'>,
|
||||
conn: Pick<RemoteInstallTransport, 'exec' | 'usesSystemSshTransport'>,
|
||||
command: string,
|
||||
options?: ExecCommandOptions
|
||||
): Promise<string> {
|
||||
@@ -90,7 +89,7 @@ export async function execCommand(
|
||||
// (an `echo`, a title escape) would otherwise be read as the command's answer.
|
||||
const fence = execOptions.wrapCommand === false ? null : buildPosixStdoutFence(command, 'SSH')
|
||||
const readPayload = (stdout: string): string => fence?.readStdout(stdout) ?? stdout
|
||||
let channel: ClientChannel
|
||||
let channel: RemoteCommandChannel
|
||||
try {
|
||||
channel = await conn.exec(fence?.command ?? command, execOptions)
|
||||
} catch (error) {
|
||||
|
||||
@@ -1,18 +1,8 @@
|
||||
// Relay-install SFTP writes. Each helper prefers the SshConnection transfer
|
||||
// method and otherwise drives one SFTP session itself, because deploy and
|
||||
// native-dependency tests pass partial connection doubles. Both routes share the
|
||||
// same namespace resolution, abort race, and one-shot session teardown. When the
|
||||
// host definitively refuses SFTP, POSIX hosts fall back to exec-channel stdin.
|
||||
// Relay-install file writes over the host transport's own transfer methods (SFTP on SSH). When
|
||||
// the host definitively refuses SFTP, POSIX hosts fall back to exec-channel stdin.
|
||||
|
||||
import type { SFTPWrapper } from 'ssh2'
|
||||
import type { RemoteInstallTransport } from './remote-install-transport'
|
||||
import { writeStringViaSftp } from './sftp-upload'
|
||||
import { uploadDirectory } from './ssh-relay-deploy-helpers'
|
||||
import { raceSftpFileTransferWithAbort } from './ssh-file-transfer-abort'
|
||||
import {
|
||||
resolveSftpTransferPathIfMapped,
|
||||
type SftpNamespacePathMapping
|
||||
} from './sftp-namespace-resolution'
|
||||
import type { SftpNamespacePathMapping } from './sftp-namespace-resolution'
|
||||
import { isWindowsRemoteHost, type RemoteHostPlatform } from './ssh-remote-platform'
|
||||
import {
|
||||
uploadDirectoryViaExecStdin,
|
||||
@@ -21,8 +11,7 @@ import {
|
||||
import { isUnconfirmedSshCommandTermination } from './ssh-relay-exec-command'
|
||||
import {
|
||||
describeSandboxedSftpFailure,
|
||||
isSandboxedSftpNamespaceError,
|
||||
latchLateSftpSessionErrors
|
||||
isSandboxedSftpNamespaceError
|
||||
} from './sftp-stream-late-error'
|
||||
import { isSftpHandshakeCorruption } from './sftp-handshake-corruption'
|
||||
|
||||
@@ -43,7 +32,12 @@ export async function uploadRelayDirectory(
|
||||
hostPlatform,
|
||||
shellRemoteDir,
|
||||
options?.signal,
|
||||
() => uploadRelayDirectoryTransfer(conn, localRelayDir, shellRemoteDir, hostPlatform, options),
|
||||
() =>
|
||||
conn.uploadDirectory(localRelayDir, shellRemoteDir, {
|
||||
hostPlatform,
|
||||
signal: options?.signal,
|
||||
sftpNamespace: options?.sftpNamespace
|
||||
}),
|
||||
() =>
|
||||
uploadDirectoryViaExecStdin(conn, localRelayDir, shellRemoteDir, hostPlatform, {
|
||||
signal: options?.signal
|
||||
@@ -51,33 +45,6 @@ export async function uploadRelayDirectory(
|
||||
)
|
||||
}
|
||||
|
||||
async function uploadRelayDirectoryTransfer(
|
||||
conn: RemoteInstallTransport,
|
||||
localRelayDir: string,
|
||||
shellRemoteDir: string,
|
||||
hostPlatform: RemoteHostPlatform,
|
||||
options?: RelayTransferOptions
|
||||
): Promise<void> {
|
||||
if (typeof conn.uploadDirectory === 'function') {
|
||||
await conn.uploadDirectory(localRelayDir, shellRemoteDir, {
|
||||
hostPlatform,
|
||||
signal: options?.signal,
|
||||
sftpNamespace: options?.sftpNamespace
|
||||
})
|
||||
return
|
||||
}
|
||||
await runSftpFallbackTransfer(conn, options, async (sftp) => {
|
||||
const targetDir = await resolveSftpTransferPathIfMapped(sftp, shellRemoteDir, {
|
||||
hostPlatform,
|
||||
sftpNamespace: options?.sftpNamespace
|
||||
})
|
||||
options?.signal?.throwIfAborted()
|
||||
await uploadDirectory(sftp, localRelayDir, targetDir, localRelayDir, {
|
||||
signal: options?.signal
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
export async function writeRelayFile(
|
||||
conn: RemoteInstallTransport,
|
||||
hostPlatform: RemoteHostPlatform,
|
||||
@@ -90,65 +57,16 @@ export async function writeRelayFile(
|
||||
hostPlatform,
|
||||
shellRemotePath,
|
||||
options?.signal,
|
||||
() => writeRelayFileTransfer(conn, hostPlatform, shellRemotePath, contents, options),
|
||||
() =>
|
||||
conn.writeFile(shellRemotePath, contents, {
|
||||
hostPlatform,
|
||||
signal: options?.signal,
|
||||
sftpNamespace: options?.sftpNamespace
|
||||
}),
|
||||
() => writeStringViaExecStdin(conn, shellRemotePath, contents, { signal: options?.signal })
|
||||
)
|
||||
}
|
||||
|
||||
async function writeRelayFileTransfer(
|
||||
conn: RemoteInstallTransport,
|
||||
hostPlatform: RemoteHostPlatform,
|
||||
shellRemotePath: string,
|
||||
contents: string,
|
||||
options?: RelayTransferOptions
|
||||
): Promise<void> {
|
||||
if (typeof conn.writeFile === 'function') {
|
||||
await conn.writeFile(shellRemotePath, contents, {
|
||||
hostPlatform,
|
||||
signal: options?.signal,
|
||||
sftpNamespace: options?.sftpNamespace
|
||||
})
|
||||
return
|
||||
}
|
||||
await runSftpFallbackTransfer(conn, options, async (sftp) => {
|
||||
const targetPath = await resolveSftpTransferPathIfMapped(sftp, shellRemotePath, {
|
||||
hostPlatform,
|
||||
sftpNamespace: options?.sftpNamespace
|
||||
})
|
||||
options?.signal?.throwIfAborted()
|
||||
await writeStringViaSftp(sftp, targetPath, contents)
|
||||
})
|
||||
}
|
||||
|
||||
async function runSftpFallbackTransfer(
|
||||
conn: RemoteInstallTransport,
|
||||
options: RelayTransferOptions | undefined,
|
||||
transfer: (sftp: SFTPWrapper) => Promise<void>
|
||||
): Promise<void> {
|
||||
const sftp = await conn.sftp(options?.signal)
|
||||
let sftpEndRequested = false
|
||||
const endSftp = (): void => {
|
||||
if (!sftpEndRequested) {
|
||||
sftpEndRequested = true
|
||||
sftp.end()
|
||||
}
|
||||
}
|
||||
latchLateSftpSessionErrors(sftp)
|
||||
try {
|
||||
await raceSftpFileTransferWithAbort(
|
||||
transfer(sftp),
|
||||
options?.signal ?? new AbortController().signal,
|
||||
(onClose) => {
|
||||
sftp.once('close', onClose)
|
||||
endSftp()
|
||||
return () => sftp.removeListener('close', onClose)
|
||||
}
|
||||
)
|
||||
} finally {
|
||||
endSftp()
|
||||
}
|
||||
}
|
||||
|
||||
export type SftpExecFallbackReason = 'sftp-unavailable' | 'sftp-sandboxed'
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user