mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
feat(wsl): orcad host transport over wsl.exe (WSL-via-orcad C1) (#27792)
* feat(wsl): WslOrcadHostTransport — orcad host transport over wsl.exe * fix(wsl): key the WSL orcad identity to the runtime environment
This commit is contained in:
@@ -7,8 +7,9 @@ import { pipeline } from 'node:stream/promises'
|
||||
import { createGunzip } from 'node:zlib'
|
||||
import { SKILL_PACKAGE_MAX_COMPRESSED_BYTES } from '../../shared/skill-package-manifest'
|
||||
import { createDeterministicGzipStore } from './skill-package-deterministic-gzip'
|
||||
import { buildUstarEntryHeader, USTAR_BLOCK_BYTES, ustarPaddingBytes } from '../ustar-entry-header'
|
||||
|
||||
const TAR_BLOCK_BYTES = 512
|
||||
const TAR_BLOCK_BYTES = USTAR_BLOCK_BYTES
|
||||
const TAR_EXPANDED_LIMIT = 40 * 1024 * 1024
|
||||
|
||||
export type SkillTarWriteEntry = {
|
||||
@@ -25,54 +26,11 @@ export type SkillTarReadEntry = {
|
||||
executable: boolean
|
||||
}
|
||||
|
||||
function writeField(header: Buffer, offset: number, length: number, value: string): void {
|
||||
const bytes = Buffer.from(value, 'utf8')
|
||||
if (bytes.length > length) {
|
||||
throw new Error('skill-package-tar-path-limit')
|
||||
}
|
||||
bytes.copy(header, offset)
|
||||
}
|
||||
|
||||
function writeOctal(header: Buffer, offset: number, length: number, value: number): void {
|
||||
const encoded = value.toString(8).padStart(length - 1, '0')
|
||||
if (encoded.length > length - 1) {
|
||||
throw new Error('skill-package-tar-number-limit')
|
||||
}
|
||||
writeField(header, offset, length, `${encoded}\0`)
|
||||
}
|
||||
|
||||
function splitTarPath(path: string): { name: string; prefix: string } {
|
||||
if (Buffer.byteLength(path, 'utf8') <= 100) {
|
||||
return { name: path, prefix: '' }
|
||||
}
|
||||
for (let index = path.lastIndexOf('/'); index > 0; index = path.lastIndexOf('/', index - 1)) {
|
||||
const prefix = path.slice(0, index)
|
||||
const name = path.slice(index + 1)
|
||||
if (Buffer.byteLength(prefix, 'utf8') <= 155 && Buffer.byteLength(name, 'utf8') <= 100) {
|
||||
return { name, prefix }
|
||||
}
|
||||
}
|
||||
throw new Error('skill-package-tar-path-limit')
|
||||
}
|
||||
|
||||
function tarHeader(entry: SkillTarWriteEntry): Buffer {
|
||||
const header = Buffer.alloc(TAR_BLOCK_BYTES)
|
||||
const { name, prefix } = splitTarPath(entry.path)
|
||||
writeField(header, 0, 100, name)
|
||||
writeOctal(header, 100, 8, entry.executable ? 0o755 : 0o644)
|
||||
writeOctal(header, 108, 8, 0)
|
||||
writeOctal(header, 116, 8, 0)
|
||||
writeOctal(header, 124, 12, entry.size)
|
||||
writeOctal(header, 136, 12, 0)
|
||||
header.fill(0x20, 148, 156)
|
||||
header[156] = '0'.charCodeAt(0)
|
||||
writeField(header, 257, 6, 'ustar')
|
||||
writeField(header, 263, 2, '00')
|
||||
writeField(header, 345, 155, prefix)
|
||||
const checksum = header.reduce((sum, byte) => sum + byte, 0)
|
||||
const checksumText = checksum.toString(8).padStart(6, '0')
|
||||
writeField(header, 148, 8, `${checksumText}\0 `)
|
||||
return header
|
||||
return buildUstarEntryHeader(
|
||||
{ path: entry.path, size: entry.size, mode: entry.executable ? 0o755 : 0o644, type: 'file' },
|
||||
'skill-package'
|
||||
)
|
||||
}
|
||||
|
||||
async function writeWithBackpressure(stream: Writable, bytes: Buffer): Promise<void> {
|
||||
@@ -127,7 +85,7 @@ export async function writeSkillTarGzip(
|
||||
} else {
|
||||
throw new Error('skill-package-source-required')
|
||||
}
|
||||
const padding = (TAR_BLOCK_BYTES - (entry.size % TAR_BLOCK_BYTES)) % TAR_BLOCK_BYTES
|
||||
const padding = ustarPaddingBytes(entry.size)
|
||||
if (padding > 0) {
|
||||
await writeWithBackpressure(gzip, Buffer.alloc(padding))
|
||||
}
|
||||
|
||||
@@ -1,8 +1,16 @@
|
||||
import type { SshTarget } from '../../shared/ssh-types'
|
||||
import type { WslOrcadHostTarget } from '../wsl/wsl-orcad-host-target'
|
||||
import type { RemoteInstallTransport } from './remote-install-transport'
|
||||
|
||||
/** The host an orcad transport reaches; only `id` and `label` mean the same for every kind. */
|
||||
export type OrcadHostTarget = SshTarget | WslOrcadHostTarget
|
||||
|
||||
export function isWslOrcadHostTarget(target: OrcadHostTarget): target is WslOrcadHostTarget {
|
||||
return 'kind' in target && target.kind === 'wsl'
|
||||
}
|
||||
|
||||
/** The host connection orcad deploy, activation, serving and lifecycle code runs over. */
|
||||
export type OrcadHostTransport = RemoteInstallTransport & {
|
||||
/** Keys per-host state (remote context, wake dedupe) to the target it reaches. */
|
||||
getTarget(): SshTarget
|
||||
getTarget(): OrcadHostTarget
|
||||
}
|
||||
|
||||
@@ -10,7 +10,7 @@ import {
|
||||
type OrcadStdioBridgeMode
|
||||
} from './orcad-stdio-bridge-script'
|
||||
import { OrcadStdioBridgeDecoder, type OrcadStdioBridgeSignal } from './orcad-stdio-bridge-stream'
|
||||
import type { OrcadHostTransport } from './orcad-host-transport'
|
||||
import { isWslOrcadHostTarget, 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'
|
||||
@@ -37,8 +37,13 @@ export async function resolveOrcadStdioBridge(
|
||||
if (!isWindowsRemoteHost(host)) {
|
||||
return { command: orcadPosixStdioBridgeCommand(port), mode: 'raw', wrapCommand: true }
|
||||
}
|
||||
const target = conn.getTarget()
|
||||
// A WSL distro is a Linux host, so a Windows answer from it is not one this path can serve.
|
||||
if (isWslOrcadHostTarget(target)) {
|
||||
throw new OrcadHostUnsupportedError('This WSL distro reported a Windows host platform.')
|
||||
}
|
||||
// Stages this client's pinned node.exe and host script, which an app update may have changed.
|
||||
const context = await resolveOrcadRemoteContext(conn.getTarget(), conn, undefined, host)
|
||||
const context = await resolveOrcadRemoteContext(target, conn, undefined, host)
|
||||
return {
|
||||
command: orcadWindowsHomeOpCommand(host, context.remoteHome, 'stdio-bridge', [String(port)]),
|
||||
mode: 'base64',
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
export const USTAR_BLOCK_BYTES = 512
|
||||
|
||||
export type UstarEntry = {
|
||||
path: string
|
||||
size: number
|
||||
mode: number
|
||||
type: 'file' | 'directory'
|
||||
}
|
||||
|
||||
/** `errorPrefix` keeps each caller's failure codes (`<prefix>-tar-path-limit`). */
|
||||
function writeField(
|
||||
header: Buffer,
|
||||
offset: number,
|
||||
length: number,
|
||||
value: string,
|
||||
errorPrefix: string
|
||||
): void {
|
||||
const bytes = Buffer.from(value, 'utf8')
|
||||
if (bytes.length > length) {
|
||||
throw new Error(`${errorPrefix}-tar-path-limit`)
|
||||
}
|
||||
bytes.copy(header, offset)
|
||||
}
|
||||
|
||||
function writeOctal(
|
||||
header: Buffer,
|
||||
offset: number,
|
||||
length: number,
|
||||
value: number,
|
||||
errorPrefix: string
|
||||
): void {
|
||||
const encoded = value.toString(8).padStart(length - 1, '0')
|
||||
if (encoded.length > length - 1) {
|
||||
throw new Error(`${errorPrefix}-tar-number-limit`)
|
||||
}
|
||||
writeField(header, offset, length, `${encoded}\0`, errorPrefix)
|
||||
}
|
||||
|
||||
function splitTarPath(path: string, errorPrefix: string): { name: string; prefix: string } {
|
||||
if (Buffer.byteLength(path, 'utf8') <= 100) {
|
||||
return { name: path, prefix: '' }
|
||||
}
|
||||
for (let index = path.lastIndexOf('/'); index > 0; index = path.lastIndexOf('/', index - 1)) {
|
||||
const prefix = path.slice(0, index)
|
||||
const name = path.slice(index + 1)
|
||||
if (Buffer.byteLength(prefix, 'utf8') <= 155 && Buffer.byteLength(name, 'utf8') <= 100) {
|
||||
return { name, prefix }
|
||||
}
|
||||
}
|
||||
throw new Error(`${errorPrefix}-tar-path-limit`)
|
||||
}
|
||||
|
||||
/** A ustar header with root ownership and a zero mtime, so equal inputs archive identically. */
|
||||
export function buildUstarEntryHeader(entry: UstarEntry, errorPrefix: string): Buffer {
|
||||
const header = Buffer.alloc(USTAR_BLOCK_BYTES)
|
||||
const { name, prefix } = splitTarPath(entry.path, errorPrefix)
|
||||
writeField(header, 0, 100, name, errorPrefix)
|
||||
writeOctal(header, 100, 8, entry.mode & 0o7777, errorPrefix)
|
||||
writeOctal(header, 108, 8, 0, errorPrefix)
|
||||
writeOctal(header, 116, 8, 0, errorPrefix)
|
||||
writeOctal(header, 124, 12, entry.type === 'directory' ? 0 : entry.size, errorPrefix)
|
||||
writeOctal(header, 136, 12, 0, errorPrefix)
|
||||
header.fill(0x20, 148, 156)
|
||||
header[156] = (entry.type === 'directory' ? '5' : '0').charCodeAt(0)
|
||||
writeField(header, 257, 6, 'ustar', errorPrefix)
|
||||
writeField(header, 263, 2, '00', errorPrefix)
|
||||
writeField(header, 345, 155, prefix, errorPrefix)
|
||||
const checksum = header.reduce((sum, byte) => sum + byte, 0)
|
||||
const checksumText = checksum.toString(8).padStart(6, '0')
|
||||
writeField(header, 148, 8, `${checksumText}\0 `, errorPrefix)
|
||||
return header
|
||||
}
|
||||
|
||||
/** Zero bytes that round `size` up to a whole block. */
|
||||
export function ustarPaddingBytes(size: number): number {
|
||||
return (USTAR_BLOCK_BYTES - (size % USTAR_BLOCK_BYTES)) % USTAR_BLOCK_BYTES
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
import { Duplex } from 'node:stream'
|
||||
import type { PipedChildProcess } from '@orca/process-host/process-spec'
|
||||
import type { RemoteCommandChannel } from '../ssh/remote-install-transport'
|
||||
|
||||
/**
|
||||
* A `wsl.exe` child shaped like an SSH exec channel: reads are its stdout, writes its stdin.
|
||||
* Emits `exit` with the code, then `close` once its stdio is gone — never `close` first.
|
||||
*/
|
||||
export function wrapWslCommandProcess(child: PipedChildProcess): RemoteCommandChannel {
|
||||
// Why emitClose off: like ssh2, `close` comes once, after the exit status, from the child.
|
||||
const duplex = new Duplex({
|
||||
emitClose: false,
|
||||
read() {
|
||||
child.stdout.resume()
|
||||
},
|
||||
write(chunk: Buffer, _encoding, done) {
|
||||
child.stdin.write(chunk, done)
|
||||
},
|
||||
final(done) {
|
||||
// EOF must reach the guest reader, not just this facade.
|
||||
child.stdin.end(done)
|
||||
}
|
||||
})
|
||||
let exitCode: number | null = null
|
||||
const onStdoutData = (chunk: Buffer): void => {
|
||||
// Backpressure: pause wsl.exe instead of buffering a slow consumer's lag in main.
|
||||
if (!duplex.push(chunk)) {
|
||||
child.stdout.pause()
|
||||
}
|
||||
}
|
||||
const onExit = (code: number | null, signal: NodeJS.Signals | null): void => {
|
||||
exitCode = code
|
||||
duplex.emit('exit', code, signal)
|
||||
}
|
||||
const onClose = (code: number | null, signal: NodeJS.Signals | null): void => {
|
||||
cleanup()
|
||||
duplex.emit('close', exitCode ?? code, signal)
|
||||
}
|
||||
const onError = (error: Error): void => {
|
||||
duplex.destroy(error)
|
||||
}
|
||||
// A guest that exits before draining stdin is reported by its exit code, not EPIPE.
|
||||
const ignoreStreamError = (): void => {}
|
||||
const cleanup = (): void => {
|
||||
child.stdout.off('data', onStdoutData)
|
||||
child.off('exit', onExit)
|
||||
child.off('close', onClose)
|
||||
child.off('error', onError)
|
||||
child.stdout.off('error', onError)
|
||||
}
|
||||
child.stdout.on('data', onStdoutData)
|
||||
child.stdout.on('end', () => duplex.push(null))
|
||||
child.stdout.on('error', onError)
|
||||
child.stdin.on('error', ignoreStreamError)
|
||||
// Consumers add their own; this only keeps an unread stderr's error from crashing main.
|
||||
child.stderr.on('error', ignoreStreamError)
|
||||
child.on('exit', onExit)
|
||||
child.on('close', onClose)
|
||||
child.on('error', onError)
|
||||
Object.assign(duplex, {
|
||||
stdin: child.stdin,
|
||||
stderr: child.stderr,
|
||||
close: () => {
|
||||
if (child.exitCode === null && child.signalCode === null) {
|
||||
child.kill()
|
||||
}
|
||||
}
|
||||
})
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: `stdin`, `stderr` and `close` were assigned above and `exit`/`close` are emitted by the listeners.
|
||||
return duplex as unknown as RemoteCommandChannel
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
import { execFileSync } from 'node:child_process'
|
||||
import {
|
||||
chmodSync,
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
mkdtempSync,
|
||||
readdirSync,
|
||||
readFileSync,
|
||||
rmSync,
|
||||
statSync,
|
||||
writeFileSync
|
||||
} from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
|
||||
import { spawnProcess } from '@orca/process-host'
|
||||
import { uploadDirectoryIntoWsl, wslGuestSourcePath } from './wsl-directory-upload'
|
||||
import { createWslOrcadHostTransport } from './wsl-orcad-host-transport'
|
||||
|
||||
function hasSha256sum(): boolean {
|
||||
try {
|
||||
execFileSync('sh', ['-c', 'command -v sha256sum'], { stdio: 'ignore' })
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
describe('wslGuestSourcePath', () => {
|
||||
const originalPlatform = process.platform
|
||||
afterEach(() => {
|
||||
Object.defineProperty(process, 'platform', { configurable: true, value: originalPlatform })
|
||||
})
|
||||
|
||||
it('maps a drive-letter directory onto its /mnt automount', () => {
|
||||
expect(wslGuestSourcePath('C:\\Users\\me\\orcad', 'Ubuntu')).toBe('/mnt/c/Users/me/orcad')
|
||||
})
|
||||
|
||||
it('leaves a source with no drive letter to the tar stream', () => {
|
||||
expect(wslGuestSourcePath('/opt/orcad', 'Ubuntu')).toBeNull()
|
||||
expect(wslGuestSourcePath('\\\\server\\share\\orcad', 'Ubuntu')).toBeNull()
|
||||
})
|
||||
|
||||
it('reads the same distro directly but not another distro', () => {
|
||||
Object.defineProperty(process, 'platform', { configurable: true, value: 'win32' })
|
||||
expect(wslGuestSourcePath('\\\\wsl.localhost\\ubuntu\\home\\me\\o', 'Ubuntu')).toBe(
|
||||
'/home/me/o'
|
||||
)
|
||||
expect(wslGuestSourcePath('\\\\wsl.localhost\\Debian\\home\\me\\o', 'Ubuntu')).toBeNull()
|
||||
})
|
||||
})
|
||||
|
||||
// Runs the generated scripts through a real POSIX shell standing in for the distro.
|
||||
describe.skipIf(process.platform === 'win32' || !hasSha256sum())('uploadDirectoryIntoWsl', () => {
|
||||
let root: string
|
||||
let source: string
|
||||
let dest: string
|
||||
let commands: string[]
|
||||
|
||||
const transport = createWslOrcadHostTransport({
|
||||
target: { kind: 'wsl', id: 'env-test', distro: 'test', label: 'test' },
|
||||
spawn: (spec) => {
|
||||
const args = spec.args ?? []
|
||||
const command = args[args.indexOf('--exec') + 3] ?? ''
|
||||
commands.push(command)
|
||||
return spawnProcess({ program: 'sh', args: ['-c', command], stdio: 'pipe' })
|
||||
}
|
||||
})
|
||||
|
||||
beforeEach(() => {
|
||||
root = mkdtempSync(join(tmpdir(), 'wsl-upload-'))
|
||||
source = join(root, 'source')
|
||||
dest = join(root, 'dest')
|
||||
commands = []
|
||||
mkdirSync(join(source, 'bin'), { recursive: true })
|
||||
mkdirSync(join(source, 'empty'), { recursive: true })
|
||||
writeFileSync(join(source, 'bin', 'tool'), '#!/bin/sh\necho ran\n')
|
||||
chmodSync(join(source, 'bin', 'tool'), 0o755)
|
||||
writeFileSync(join(source, "it's here.txt"), 'quoted')
|
||||
writeFileSync(join(source, 'large.bin'), Buffer.alloc(3 * 1024 * 1024 + 7, 9))
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
rmSync(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
const expectUploaded = (): void => {
|
||||
expect(readFileSync(join(dest, 'bin', 'tool'), 'utf8')).toBe('#!/bin/sh\necho ran\n')
|
||||
expect(statSync(join(dest, 'bin', 'tool')).mode & 0o111).not.toBe(0)
|
||||
expect(readFileSync(join(dest, "it's here.txt"), 'utf8')).toBe('quoted')
|
||||
expect(statSync(join(dest, 'large.bin')).size).toBe(3 * 1024 * 1024 + 7)
|
||||
expect(existsSync(join(dest, 'empty'))).toBe(true)
|
||||
expect(readdirSync(dest).filter((name) => name.startsWith('.orca-upload-'))).toEqual([])
|
||||
}
|
||||
|
||||
it('streams a tar when the guest cannot see the source', async () => {
|
||||
await uploadDirectoryIntoWsl(transport, source, null, dest)
|
||||
expectUploaded()
|
||||
expect(commands.some((command) => command.includes('tar -x'))).toBe(true)
|
||||
})
|
||||
|
||||
it('copies from the guest-visible path without a tar stream', async () => {
|
||||
await uploadDirectoryIntoWsl(transport, source, source, dest)
|
||||
expectUploaded()
|
||||
expect(commands.some((command) => command.includes('tar -x'))).toBe(false)
|
||||
})
|
||||
|
||||
it('falls back to the tar stream when the mount is missing', async () => {
|
||||
await uploadDirectoryIntoWsl(transport, source, join(root, 'not-mounted'), dest)
|
||||
expectUploaded()
|
||||
expect(commands.some((command) => command.includes('tar -x'))).toBe(true)
|
||||
})
|
||||
|
||||
it('merges into an existing destination, replacing changed files', async () => {
|
||||
mkdirSync(dest, { recursive: true })
|
||||
writeFileSync(join(dest, 'keep.txt'), 'kept')
|
||||
writeFileSync(join(dest, 'large.bin'), 'stale')
|
||||
await uploadDirectoryIntoWsl(transport, source, null, dest)
|
||||
expectUploaded()
|
||||
expect(readFileSync(join(dest, 'keep.txt'), 'utf8')).toBe('kept')
|
||||
})
|
||||
|
||||
it('refuses a staged copy whose bytes differ from the Windows-side digests', async () => {
|
||||
const other = join(root, 'other')
|
||||
execFileSync('cp', ['-R', source, other])
|
||||
writeFileSync(join(other, 'large.bin'), 'tampered')
|
||||
await expect(uploadDirectoryIntoWsl(transport, source, other, dest)).rejects.toThrow(
|
||||
/digest check/
|
||||
)
|
||||
expect(existsSync(join(dest, 'large.bin'))).toBe(false)
|
||||
expect(readdirSync(dest).filter((name) => name.startsWith('.orca-upload-'))).toEqual([])
|
||||
})
|
||||
|
||||
it('writes a file atomically through stdin', async () => {
|
||||
await transport.writeFile(join(dest, 'nested', 'record.json'), '{"ok":true}')
|
||||
expect(readFileSync(join(dest, 'nested', 'record.json'), 'utf8')).toBe('{"ok":true}')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,204 @@
|
||||
/**
|
||||
* Upload a Windows directory into a WSL distro (design D1). The guest copies it straight off its
|
||||
* `/mnt/<drive>` mount when it can see the source, and otherwise reads a tar from stdin. Either way
|
||||
* the tree is staged inside the destination, checked against digests taken on the Windows side,
|
||||
* then renamed into place file by file under `flock`, so no reader sees a partially written file.
|
||||
*/
|
||||
import { createHash, randomBytes } from 'node:crypto'
|
||||
import { constants } from 'node:fs'
|
||||
import { open } from 'node:fs/promises'
|
||||
import { Readable } from 'node:stream'
|
||||
import { parseWslUncPath, toLinuxPath } from '../../shared/wsl-paths'
|
||||
import type { RemoteInstallTransport } from '../ssh/remote-install-transport'
|
||||
import { shellEscape } from '../ssh/ssh-connection-utils'
|
||||
import { execCommand } from '../ssh/ssh-relay-exec-command'
|
||||
import { RELAY_DEPLOY_TIMEOUT_MS } from '../ssh/ssh-relay-deploy-timing'
|
||||
import { getRemoteHostPlatform } from '../ssh/ssh-remote-platform'
|
||||
import { collectLocalUploadPlan, type LocalUploadPlan } from '../ssh/system-ssh-file-transfer'
|
||||
import { buildUstarEntryHeader, USTAR_BLOCK_BYTES, ustarPaddingBytes } from '../ustar-entry-header'
|
||||
|
||||
export const WSL_GUEST_SOURCE_UNAVAILABLE = 'ORCA_WSL_GUEST_SOURCE_UNAVAILABLE'
|
||||
export const WSL_UPLOAD_PROMOTED = 'ORCA_WSL_UPLOAD_PROMOTED'
|
||||
|
||||
type UploadOptions = { signal?: AbortSignal }
|
||||
|
||||
/** Where the guest sees `localDir`, or null when only a stream can carry it there. */
|
||||
export function wslGuestSourcePath(localDir: string, distro: string): string | null {
|
||||
const unc = process.platform === 'win32' ? parseWslUncPath(localDir) : null
|
||||
if (unc) {
|
||||
// Another distro's files are not mounted in this one.
|
||||
return unc.distro.toLowerCase() === distro.toLowerCase() ? unc.linuxPath : null
|
||||
}
|
||||
// Drive letters only: a UNC share or mapped network drive has no /mnt automount.
|
||||
return /^[A-Za-z]:[\\/]/.test(localDir) ? toLinuxPath(localDir) : null
|
||||
}
|
||||
|
||||
// Success paths never `exit`: execCommand's stdout fence must still print its end marker.
|
||||
function stageFromGuestPathCommand(source: string, stage: string): string {
|
||||
return [
|
||||
`source=${shellEscape(source)}; stage=${shellEscape(stage)}`,
|
||||
'if [ -d "$source" ]; then',
|
||||
' { mkdir -p "$stage" && cp -R "$source/." "$stage/"; } || { rm -rf "$stage"; false; }',
|
||||
'else',
|
||||
` echo ${WSL_GUEST_SOURCE_UNAVAILABLE}`,
|
||||
'fi'
|
||||
].join('\n')
|
||||
}
|
||||
|
||||
function stageFromTarCommand(stage: string): string {
|
||||
return [
|
||||
`stage=${shellEscape(stage)}`,
|
||||
// `-o`: files belong to whoever extracts them, not to the archive's uid 0.
|
||||
'{ mkdir -p "$stage" && tar -x -o -f - -C "$stage"; } || { rm -rf "$stage"; false; }'
|
||||
].join('\n')
|
||||
}
|
||||
|
||||
/** Reads the digest manifest from stdin; the stage is removed whatever the outcome. */
|
||||
function verifyAndPromoteCommand(stage: string, dest: string, hasFiles: boolean): string {
|
||||
return [
|
||||
`stage=${shellEscape(stage)}; dest=${shellEscape(dest)}`,
|
||||
`trap 'rm -rf "$stage"' EXIT`,
|
||||
// GNU sha256sum rejects an empty manifest.
|
||||
...(hasFiles
|
||||
? [
|
||||
'(cd "$stage" && sha256sum -c - >/dev/null) ||',
|
||||
" { echo 'staged upload failed its digest check' >&2; exit 1; }"
|
||||
]
|
||||
: []),
|
||||
// Serialises concurrent uploads into one destination; each rename is atomic regardless.
|
||||
'if command -v flock >/dev/null 2>&1; then exec 9<"$dest" && flock 9 || exit 1; fi',
|
||||
'(cd "$stage" && find . -mindepth 1 -type d) | while IFS= read -r p; do',
|
||||
' mkdir -p "$dest/$p" || exit 1',
|
||||
'done || exit 1',
|
||||
'(cd "$stage" && find . ! -type d) | while IFS= read -r p; do',
|
||||
' mv -f "$stage/$p" "$dest/$p" || exit 1',
|
||||
'done || exit 1',
|
||||
`echo ${WSL_UPLOAD_PROMOTED}`
|
||||
].join('\n')
|
||||
}
|
||||
|
||||
function assertManifestSafePath(path: string): void {
|
||||
// sha256sum escapes these, and the line-based promote loop would split on a newline.
|
||||
if (/[\n\r\\]/.test(path)) {
|
||||
throw new Error(`Unsupported upload path: ${JSON.stringify(path)}`)
|
||||
}
|
||||
}
|
||||
|
||||
async function openPlannedFile(localPath: string): ReturnType<typeof open> {
|
||||
const handle = await open(localPath, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0))
|
||||
const stat = await handle.stat()
|
||||
if (!stat.isFile()) {
|
||||
await handle.close()
|
||||
throw new Error(`Unsupported upload source: ${localPath}`)
|
||||
}
|
||||
return handle
|
||||
}
|
||||
|
||||
async function digestManifest(plan: LocalUploadPlan, signal?: AbortSignal): Promise<string> {
|
||||
const lines: string[] = []
|
||||
for (const file of plan.files) {
|
||||
signal?.throwIfAborted()
|
||||
assertManifestSafePath(file.remotePath)
|
||||
const handle = await openPlannedFile(file.localPath)
|
||||
try {
|
||||
const hash = createHash('sha256')
|
||||
for await (const chunk of handle.createReadStream({ autoClose: false })) {
|
||||
hash.update(chunk)
|
||||
}
|
||||
lines.push(`${hash.digest('hex')} ${file.remotePath}\n`)
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
}
|
||||
return lines.join('')
|
||||
}
|
||||
|
||||
const TAR_ERROR_PREFIX = 'wsl-upload'
|
||||
|
||||
async function* tarStream(plan: LocalUploadPlan, signal?: AbortSignal): AsyncGenerator<Buffer> {
|
||||
for (const directory of plan.directories) {
|
||||
if (directory !== '.') {
|
||||
yield buildUstarEntryHeader(
|
||||
{ path: directory, size: 0, mode: 0o755, type: 'directory' },
|
||||
TAR_ERROR_PREFIX
|
||||
)
|
||||
}
|
||||
}
|
||||
for (const file of plan.files) {
|
||||
signal?.throwIfAborted()
|
||||
const handle = await openPlannedFile(file.localPath)
|
||||
try {
|
||||
const stat = await handle.stat()
|
||||
// Windows has no executable bit; 0755 matches what a /mnt copy of the same file gets.
|
||||
const mode = process.platform === 'win32' ? 0o755 : stat.mode & 0o777
|
||||
yield buildUstarEntryHeader(
|
||||
{ path: file.remotePath, size: stat.size, mode, type: 'file' },
|
||||
TAR_ERROR_PREFIX
|
||||
)
|
||||
let written = 0
|
||||
if (stat.size > 0) {
|
||||
for await (const chunk of handle.createReadStream({
|
||||
autoClose: false,
|
||||
start: 0,
|
||||
end: stat.size - 1
|
||||
})) {
|
||||
written += chunk.length
|
||||
yield chunk
|
||||
}
|
||||
}
|
||||
// A file that shrank mid-read would shift every later header.
|
||||
if (written !== stat.size) {
|
||||
throw new Error(`File changed during upload: ${file.localPath}`)
|
||||
}
|
||||
const padding = ustarPaddingBytes(stat.size)
|
||||
if (padding > 0) {
|
||||
yield Buffer.alloc(padding)
|
||||
}
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
}
|
||||
yield Buffer.alloc(USTAR_BLOCK_BYTES * 2)
|
||||
}
|
||||
|
||||
/** `guestSource` is `wslGuestSourcePath(localDir, distro)`: where the guest may read it directly. */
|
||||
export async function uploadDirectoryIntoWsl(
|
||||
conn: RemoteInstallTransport,
|
||||
localDir: string,
|
||||
guestSource: string | null,
|
||||
remoteDir: string,
|
||||
options?: UploadOptions
|
||||
): Promise<void> {
|
||||
const signal = options?.signal
|
||||
const plan = await collectLocalUploadPlan(
|
||||
localDir,
|
||||
'.',
|
||||
getRemoteHostPlatform('linux-x64'),
|
||||
signal
|
||||
)
|
||||
const manifest = await digestManifest(plan, signal)
|
||||
const stage = `${remoteDir.replace(/\/+$/, '')}/.orca-upload-${randomBytes(8).toString('hex')}`
|
||||
// Why the deploy bound: a runtime tree over a cold distro outlasts the 30 s probe default.
|
||||
const execOptions = { signal, timeoutMs: RELAY_DEPLOY_TIMEOUT_MS }
|
||||
const stagedFromGuest =
|
||||
guestSource !== null &&
|
||||
(await execCommand(conn, stageFromGuestPathCommand(guestSource, stage), execOptions)).trim() !==
|
||||
WSL_GUEST_SOURCE_UNAVAILABLE
|
||||
if (!stagedFromGuest) {
|
||||
await execCommand(conn, stageFromTarCommand(stage), {
|
||||
...execOptions,
|
||||
stdin: Readable.from(tarStream(plan, signal))
|
||||
})
|
||||
}
|
||||
const promoted = await execCommand(
|
||||
conn,
|
||||
verifyAndPromoteCommand(stage, remoteDir, plan.files.length > 0),
|
||||
{
|
||||
...execOptions,
|
||||
stdin: Readable.from([Buffer.from(manifest)])
|
||||
}
|
||||
)
|
||||
if (promoted.trim().split('\n').at(-1) !== WSL_UPLOAD_PROMOTED) {
|
||||
throw new Error(`The distro did not confirm the upload into ${remoteDir}`)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
/**
|
||||
* The identity a WSL distro's orcad transport reports; `kind` is absent from `SshTarget`, which is
|
||||
* what tells the two apart. `id` and `distro` are the WSL deployment record's `environmentId` and
|
||||
* `distro`; `label` is the environment's name.
|
||||
*/
|
||||
export type WslOrcadHostTarget = {
|
||||
kind: 'wsl'
|
||||
id: string
|
||||
distro: string
|
||||
label: string
|
||||
}
|
||||
@@ -0,0 +1,175 @@
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { PassThrough, Readable } from 'node:stream'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { PipedChildProcess, PipedProcessSpec } from '@orca/process-host/process-spec'
|
||||
import { execCommand } from '../ssh/ssh-relay-exec-command'
|
||||
import { isWslOrcadHostTarget } from '../ssh/orcad-host-transport'
|
||||
import { createWslOrcadHostTransport } from './wsl-orcad-host-transport'
|
||||
|
||||
type FakeChild = EventEmitter & {
|
||||
stdin: PassThrough
|
||||
stdout: PassThrough
|
||||
stderr: PassThrough
|
||||
exitCode: number | null
|
||||
signalCode: NodeJS.Signals | null
|
||||
kill: ReturnType<typeof vi.fn>
|
||||
received: () => Buffer
|
||||
finish: (code: number | null, signal?: NodeJS.Signals | null) => void
|
||||
}
|
||||
|
||||
function fakeChild(options: { spawnError?: Error } = {}): FakeChild {
|
||||
const emitter = new EventEmitter()
|
||||
const stdin = new PassThrough()
|
||||
const received: Buffer[] = []
|
||||
stdin.on('data', (chunk: Buffer) => received.push(chunk))
|
||||
const child: FakeChild = Object.assign(emitter, {
|
||||
stdin,
|
||||
stdout: new PassThrough(),
|
||||
stderr: new PassThrough(),
|
||||
exitCode: null,
|
||||
signalCode: null,
|
||||
kill: vi.fn(() => {
|
||||
child.finish(null, 'SIGTERM')
|
||||
return true
|
||||
}),
|
||||
received: () => Buffer.concat(received),
|
||||
finish: (code: number | null, signal: NodeJS.Signals | null = null) => {
|
||||
child.exitCode = code
|
||||
child.signalCode = signal
|
||||
child.emit('exit', code, signal)
|
||||
child.stdout.end()
|
||||
child.stderr.end()
|
||||
setImmediate(() => child.emit('close', code, signal))
|
||||
}
|
||||
})
|
||||
process.nextTick(() =>
|
||||
options.spawnError ? child.emit('error', options.spawnError) : child.emit('spawn')
|
||||
)
|
||||
return child
|
||||
}
|
||||
|
||||
function transportWith(children: FakeChild[]) {
|
||||
const specs: PipedProcessSpec[] = []
|
||||
const transport = createWslOrcadHostTransport({
|
||||
target: { kind: 'wsl', id: 'env-1', distro: 'Ubuntu-24.04', label: 'My Ubuntu' },
|
||||
spawn: (spec) => {
|
||||
specs.push(spec)
|
||||
const next = children.shift()
|
||||
if (!next) {
|
||||
throw new Error('no fake child left')
|
||||
}
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the channel reads only the streams, exit state, kill and events the fake provides.
|
||||
return next as unknown as PipedChildProcess
|
||||
}
|
||||
})
|
||||
return { transport, specs }
|
||||
}
|
||||
|
||||
describe('createWslOrcadHostTransport', () => {
|
||||
it('runs the command through wsl.exe --exec sh -c in the chosen distro', async () => {
|
||||
const child = fakeChild()
|
||||
const { transport, specs } = transportWith([child])
|
||||
const channel = await transport.exec('echo $HOME')
|
||||
expect(specs[0]?.program).toMatch(/wsl(\.exe)?$/i)
|
||||
expect(specs[0]?.args).toEqual(['-d', 'Ubuntu-24.04', '--exec', 'sh', '-c', 'echo $HOME'])
|
||||
expect(specs[0]?.env?.WSL_UTF8).toBe('1')
|
||||
channel.close()
|
||||
})
|
||||
|
||||
it('emits exit before close and reports the exit code on both', async () => {
|
||||
const child = fakeChild()
|
||||
const { transport } = transportWith([child])
|
||||
const channel = await transport.exec('true')
|
||||
const events: string[] = []
|
||||
channel.on('exit', (code: number) => events.push(`exit:${code}`))
|
||||
const closed = new Promise<void>((resolve) =>
|
||||
channel.on('close', (code: number) => {
|
||||
events.push(`close:${code}`)
|
||||
resolve()
|
||||
})
|
||||
)
|
||||
channel.resume()
|
||||
child.finish(0)
|
||||
await closed
|
||||
expect(events).toEqual(['exit:0', 'close:0'])
|
||||
})
|
||||
|
||||
it('pipes stdin to the child and ends it', async () => {
|
||||
const child = fakeChild()
|
||||
const { transport } = transportWith([child])
|
||||
const ended = new Promise<void>((resolve) => child.stdin.on('end', resolve))
|
||||
const pending = execCommand(transport, 'cat >/dev/null', {
|
||||
stdin: Readable.from([Buffer.from('payload-'), Buffer.from('bytes')])
|
||||
})
|
||||
await ended
|
||||
child.finish(0)
|
||||
await pending
|
||||
expect(child.received().toString()).toBe('payload-bytes')
|
||||
})
|
||||
|
||||
it('rejects a nonzero exit with its stderr', async () => {
|
||||
const child = fakeChild()
|
||||
const { transport } = transportWith([child])
|
||||
const pending = execCommand(transport, 'false')
|
||||
await new Promise((resolve) => setImmediate(resolve))
|
||||
child.stderr.write('boom\n')
|
||||
child.finish(3)
|
||||
await expect(pending).rejects.toThrow(/exit 3[\s\S]*boom/)
|
||||
})
|
||||
|
||||
it('refuses an already-aborted exec without spawning', async () => {
|
||||
const { transport, specs } = transportWith([])
|
||||
const controller = new AbortController()
|
||||
controller.abort()
|
||||
await expect(transport.exec('true', { signal: controller.signal })).rejects.toMatchObject({
|
||||
name: 'AbortError'
|
||||
})
|
||||
expect(specs).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('kills the child when an exec is aborted', async () => {
|
||||
const child = fakeChild()
|
||||
const { transport } = transportWith([child])
|
||||
const controller = new AbortController()
|
||||
const pending = execCommand(transport, 'sleep 300', { signal: controller.signal })
|
||||
await new Promise((resolve) => setImmediate(resolve))
|
||||
controller.abort()
|
||||
await expect(pending).rejects.toMatchObject({ name: 'AbortError' })
|
||||
expect(child.kill).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('rejects exec when wsl.exe cannot be started', async () => {
|
||||
const { transport } = transportWith([fakeChild({ spawnError: new Error('spawn ENOENT') })])
|
||||
await expect(transport.exec('true')).rejects.toThrow('spawn ENOENT')
|
||||
})
|
||||
|
||||
it('streams large stdout intact under backpressure', async () => {
|
||||
const child = fakeChild()
|
||||
const { transport } = transportWith([child])
|
||||
const channel = await transport.exec('cat big')
|
||||
const chunk = Buffer.alloc(1024 * 1024, 7)
|
||||
const received: Buffer[] = []
|
||||
const closed = new Promise<void>((resolve) => channel.on('close', () => resolve()))
|
||||
channel.on('data', (data: Buffer) => received.push(data))
|
||||
for (let i = 0; i < 20; i++) {
|
||||
child.stdout.write(chunk)
|
||||
}
|
||||
child.finish(0)
|
||||
await closed
|
||||
expect(Buffer.concat(received).length).toBe(20 * 1024 * 1024)
|
||||
})
|
||||
|
||||
it('reports a stable WSL identity and never claims the system SSH transport', () => {
|
||||
const { transport } = transportWith([])
|
||||
const target = transport.getTarget()
|
||||
expect(target).toEqual({
|
||||
kind: 'wsl',
|
||||
id: 'env-1',
|
||||
distro: 'Ubuntu-24.04',
|
||||
label: 'My Ubuntu'
|
||||
})
|
||||
expect(isWslOrcadHostTarget(target)).toBe(true)
|
||||
expect(transport.usesSystemSshTransport()).toBe(false)
|
||||
expect(transport.getConnectGeneration()).toBe(transport.getConnectGeneration())
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,82 @@
|
||||
import { once } from 'node:events'
|
||||
import { spawnProcess } from '@orca/process-host'
|
||||
import type { PipedProcessSpawner } from '@orca/process-host/process-spec'
|
||||
import { buildWslExecArgs } from '../../shared/wsl-login-shell-command'
|
||||
import { commandLineLength, MAX_COMMAND_LINE_CHARS } from '../../shared/windows-command-line-budget'
|
||||
import type { OrcadHostTransport } from '../ssh/orcad-host-transport'
|
||||
import type { RemoteCommandChannel } from '../ssh/remote-install-transport'
|
||||
import { createSshOperationAbortError, type SshExecOptions } from '../ssh/ssh-connection-utils'
|
||||
import { writeStringViaExecStdin } from '../ssh/ssh-exec-stdin-file-transfer'
|
||||
import { resolveWslInteropSpawnCwd } from '../wsl-interop-spawn-directory'
|
||||
import { wrapWslCommandProcess } from './wsl-command-channel'
|
||||
import { uploadDirectoryIntoWsl, wslGuestSourcePath } from './wsl-directory-upload'
|
||||
import { resolveWslExecutablePath } from './wsl-executable-path'
|
||||
import type { WslOrcadHostTarget } from './wsl-orcad-host-target'
|
||||
|
||||
export type WslOrcadHostTransport = OrcadHostTransport & {
|
||||
getTarget(): WslOrcadHostTarget
|
||||
}
|
||||
|
||||
type WslOrcadHostTransportOptions = {
|
||||
target: WslOrcadHostTarget
|
||||
/** Test seam; production spawns through process-host. */
|
||||
spawn?: PipedProcessSpawner
|
||||
}
|
||||
|
||||
/**
|
||||
* orcad's host connection to a WSL distro: every exec is its own `wsl.exe --exec sh -c` child.
|
||||
* The distro is never a Windows host and has no SFTP, so every transfer runs over exec.
|
||||
*/
|
||||
export function createWslOrcadHostTransport(
|
||||
options: WslOrcadHostTransportOptions
|
||||
): WslOrcadHostTransport {
|
||||
const spawn = options.spawn ?? spawnProcess
|
||||
const { target } = options
|
||||
const transport: WslOrcadHostTransport = {
|
||||
// `wrapCommand` is SSH-only: there is no login shell between wsl.exe and `sh -c`.
|
||||
async exec(command: string, execOptions?: SshExecOptions): Promise<RemoteCommandChannel> {
|
||||
if (execOptions?.signal?.aborted) {
|
||||
throw createSshOperationAbortError()
|
||||
}
|
||||
const program = resolveWslExecutablePath()
|
||||
const args = buildWslExecArgs(target.distro, ['sh', '-c', command])
|
||||
if (commandLineLength([program, ...args]) > MAX_COMMAND_LINE_CHARS) {
|
||||
throw new Error('WSL command exceeds the Windows command-line limit')
|
||||
}
|
||||
const child = spawn({
|
||||
program,
|
||||
args,
|
||||
cwd: resolveWslInteropSpawnCwd(),
|
||||
// Why WSL_UTF8: otherwise wsl.exe writes its own errors as UTF-16LE.
|
||||
env: { ...process.env, WSL_UTF8: '1' }
|
||||
})
|
||||
const channel = wrapWslCommandProcess(child)
|
||||
// Like an SSH channel open: a child that never started rejects here, not as a stream error.
|
||||
const spawnError = (error: Error): void => {
|
||||
channel.destroy(error)
|
||||
}
|
||||
channel.on('error', spawnError)
|
||||
try {
|
||||
await once(child, 'spawn')
|
||||
} finally {
|
||||
channel.off('error', spawnError)
|
||||
}
|
||||
return channel
|
||||
},
|
||||
usesSystemSshTransport: () => false,
|
||||
// One transport, one wsl.exe lane: nothing underneath is ever replaced.
|
||||
getConnectGeneration: () => 0,
|
||||
getTarget: () => target,
|
||||
writeFile: (remotePath, contents, fileOptions) =>
|
||||
writeStringViaExecStdin(transport, remotePath, contents, { signal: fileOptions?.signal }),
|
||||
uploadDirectory: (localDir, remoteDir, fileOptions) =>
|
||||
uploadDirectoryIntoWsl(
|
||||
transport,
|
||||
localDir,
|
||||
wslGuestSourcePath(localDir, target.distro),
|
||||
remoteDir,
|
||||
{ signal: fileOptions?.signal }
|
||||
)
|
||||
}
|
||||
return transport
|
||||
}
|
||||
Reference in New Issue
Block a user