mirror of
https://github.com/stablyai/orca.git
synced 2026-10-07 16:02:29 +00:00
* feat(ssh): wire SshConnection through the work and transport close ledgers (#16741 T2 P2) Every operation SshConnection admits (exec, shell, sftp, file transfers, upload sessions, forwarded channels and sockets, system-SSH commands) now runs through the connection's work ledger, and every ssh2 client and proxy process it allocates is tracked until it physically closes. Ordinary connect, reconnect and disconnect behavior is unchanged. Adds: - subscribeTransportClosure: one-shot notice once the connection is disposed, every allocated transport has emitted 'close' and tracked work has drained. System-SSH startup is never proven closed from here. - disconnectAndDrain(signal): for owned single-lifetime transports; fences new work, disconnects, and waits for physical close of the client, proxy, every allocated client and all fenced work. Refuses (after cleaning up) when the transport cannot be proven, e.g. system SSH or a connect still in flight. - getExecutionDestination: the ssh2 endpoint, accepted host-key fingerprint and proxy-route digest proven by the current handshake (ssh-connection-destination). - getTransportGeneration, prepareForwardRoute, openForwardSocket, forwardOut, forwardStreamLocal for later forwarding callers. - An automaticReconnect constructor option (default on). Channel close is local lifetime evidence only, never a remote-exit verdict. Porting note (source: #16741 heada68b6f3531, merge-base277c289bd4): - Taken: the ledger hunks of ssh-connection.ts, ssh-connection-destination, ssh-forward-channel-lifetime, ssh-upload-session-lifetime, the system-SSH facade EOF hunk, and their tests. - Adapted: operation bodies became private *Untracked methods called through the ledger instead of being re-indented; closure gating and the close drain moved to ssh-connection-transport-closure / ssh-connection-close-drain; the destination parser uses type guards instead of a cast. disconnectAndDrain fences through the ledger directly. Main's plain-SSH shell() goes through the ledger too. execCommand takes Pick<SshConnection, 'exec' | 'usesSystemSshTransport'>; its string-stdin/maxOutputBytes hunk is not taken because main already streams stdin. Work-drain tests fence the private ledger until T3 adds the public fence; system-SSH drain cases split into their own file. - Left for later slices: fenceWorkForReset (T3), isEphemeralRuntimeSshOwner (T6), assertProfileLifetimeAdmission (P8b), and the four manager drain cases in ssh-connection-disconnect-drain.test.ts (P3). * refactor(ssh): shrink SshConnection below main and surface unhandled channel errors ssh-connection.ts no longer grows under its max-lines exemption: it is 1872 lines, below main's 1917. The public API is unchanged. - ssh-channel-open.ts: the channel-open waiter and session-limit retry. - ssh-connection-file-transfers.ts: the uploadDirectory, downloadFile, upload session, writeFile and writeBuffer bodies, reading the connection through a small getter-based host so each read still sees the live transport. - ssh-forward-channel-lifetime.ts: the forward client and stream-local checks. The lifetime tracker's 'error' listener no longer hides errors. When it is a channel's only error listener, the error is reported: SshConnection logs "[ssh] Unhandled <kind> channel error for <target>: <message>", and other callers fall back to a generic [ssh] warning. Nothing throws, so an orphaned channel error still cannot crash the process. * fix(ssh): name the forwarded local socket type and type the upload-session test stub The forwarded local socket now takes SshConnectionWorkChannel, the event surface the ledger tracks, instead of a broad object. The upload-session lifetime test binds an EventEmitter rather than an untyped {}. --------- Co-authored-by: m4air <m4air@m4airs-Air.localdomain>
173 lines
4.8 KiB
TypeScript
173 lines
4.8 KiB
TypeScript
import { spawn, type ChildProcess } from 'node:child_process'
|
|
import { Duplex } from 'node:stream'
|
|
import type { ClientChannel } from 'ssh2'
|
|
import type { SshTarget } from '../../shared/ssh-types'
|
|
import { wrapRemoteCommandForPosixShell, type SshExecOptions } from './ssh-connection-utils'
|
|
import { buildSshArgs, type SystemSshBuildArgsOptions } from './system-ssh-args'
|
|
import { findSystemSsh } from './system-ssh-binary'
|
|
|
|
export type SystemSshProcess = {
|
|
stdin: NodeJS.WritableStream
|
|
stdout: NodeJS.ReadableStream
|
|
stderr: NodeJS.ReadableStream
|
|
kill: () => void
|
|
onExit: (cb: (code: number | null) => void) => void
|
|
pid: number | undefined
|
|
}
|
|
|
|
export type SystemSshCommandChannel = ClientChannel & {
|
|
_process?: ChildProcess
|
|
_closeRequested?: boolean
|
|
}
|
|
|
|
type SystemSshCommandOptions = SshExecOptions & SystemSshBuildArgsOptions
|
|
|
|
/**
|
|
* Spawn a system ssh process connecting to the given target.
|
|
* Used when ssh2 cannot handle the auth method (FIDO2, ControlMaster).
|
|
*
|
|
* The returned process's stdin/stdout are used as the transport for
|
|
* the relay's JSON-RPC protocol, exactly like an ssh2 channel.
|
|
*/
|
|
export function spawnSystemSsh(
|
|
target: SshTarget,
|
|
options?: SystemSshBuildArgsOptions
|
|
): SystemSshProcess {
|
|
const sshPath = findSystemSsh()
|
|
if (!sshPath) {
|
|
throw new Error(
|
|
'No system ssh binary found. Install OpenSSH to use FIDO2 keys or ControlMaster.'
|
|
)
|
|
}
|
|
|
|
const args = buildSshArgs(target, options)
|
|
const proc = spawn(sshPath, args, {
|
|
stdio: ['pipe', 'pipe', 'pipe'],
|
|
windowsHide: true
|
|
})
|
|
|
|
return wrapChildProcess(proc)
|
|
}
|
|
|
|
export function spawnSystemSshCommand(
|
|
target: SshTarget,
|
|
command: string,
|
|
options?: SystemSshCommandOptions
|
|
): SystemSshCommandChannel {
|
|
const sshPath = findSystemSsh()
|
|
if (!sshPath) {
|
|
throw new Error(
|
|
'No system ssh binary found. Install OpenSSH to use ProxyUseFdpass, FIDO2 keys, or ControlMaster.'
|
|
)
|
|
}
|
|
|
|
const remoteCommand =
|
|
options?.wrapCommand === false ? command : wrapRemoteCommandForPosixShell(command)
|
|
const proc = spawn(sshPath, [...buildSshArgs(target, options), remoteCommand], {
|
|
stdio: ['pipe', 'pipe', 'pipe'],
|
|
windowsHide: true
|
|
})
|
|
return wrapCommandProcess(proc)
|
|
}
|
|
|
|
function wrapChildProcess(proc: ChildProcess): SystemSshProcess {
|
|
return {
|
|
stdin: proc.stdin!,
|
|
stdout: proc.stdout!,
|
|
stderr: proc.stderr!,
|
|
pid: proc.pid,
|
|
kill: () => {
|
|
try {
|
|
proc.kill('SIGTERM')
|
|
} catch {
|
|
// Process may already be dead
|
|
}
|
|
},
|
|
onExit: (cb) => {
|
|
proc.on('exit', (code) => cb(code))
|
|
}
|
|
}
|
|
}
|
|
|
|
function wrapCommandProcess(proc: ChildProcess): SystemSshCommandChannel {
|
|
const duplex = new Duplex({
|
|
read() {
|
|
proc.stdout?.resume()
|
|
},
|
|
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
|
|
|
|
const mutableChannel = channel as unknown as {
|
|
stdin: NodeJS.WritableStream
|
|
stderr: NodeJS.ReadableStream
|
|
_process?: ChildProcess
|
|
_closeRequested?: boolean
|
|
close: () => void
|
|
}
|
|
mutableChannel.stdin = proc.stdin!
|
|
mutableChannel.stderr = proc.stderr!
|
|
mutableChannel._process = proc
|
|
mutableChannel.close = () => {
|
|
mutableChannel._closeRequested = true
|
|
try {
|
|
proc.kill('SIGTERM')
|
|
} catch {
|
|
// Process may already be dead
|
|
}
|
|
}
|
|
|
|
const cleanupProcessListeners = (): void => {
|
|
proc.stdout!.off('data', onStdoutData)
|
|
proc.stdout!.off('end', onStdoutEnd)
|
|
proc.off('exit', onExit)
|
|
proc.off('close', onClose)
|
|
proc.off('error', onProcessError)
|
|
proc.stdin!.off('error', onStreamError)
|
|
proc.stdout!.off('error', onStreamError)
|
|
}
|
|
const fail = (err: Error): void => {
|
|
cleanupProcessListeners()
|
|
duplex.destroy(err)
|
|
}
|
|
const onStdoutData = (data: Buffer): void => {
|
|
// Why: file downloads can outpace the local destination; pause OpenSSH
|
|
// instead of buffering the producer-consumer lag in the main process.
|
|
if (!duplex.push(data)) {
|
|
proc.stdout!.pause()
|
|
}
|
|
}
|
|
const onStdoutEnd = (): void => {
|
|
duplex.push(null)
|
|
}
|
|
const onExit = (code: number | null, signal?: NodeJS.Signals | null): void => {
|
|
channel.emit('exit', code, signal)
|
|
}
|
|
const onClose = (code: number | null, signal?: NodeJS.Signals | null): void => {
|
|
cleanupProcessListeners()
|
|
channel.emit('close', code, signal)
|
|
}
|
|
const onProcessError = (err: Error): void => {
|
|
fail(err)
|
|
}
|
|
const onStreamError = (err: Error): void => {
|
|
fail(err)
|
|
}
|
|
|
|
proc.stdout!.on('data', onStdoutData)
|
|
proc.stdout!.on('end', onStdoutEnd)
|
|
proc.on('exit', onExit)
|
|
proc.on('close', onClose)
|
|
proc.on('error', onProcessError)
|
|
proc.stdin!.on('error', onStreamError)
|
|
proc.stdout!.on('error', onStreamError)
|
|
|
|
return channel
|
|
}
|