import { createHash, randomBytes } from "node:crypto"; import { createReadStream, readFileSync } from "node:fs"; import { mkdir, open, readFile, rm } from "node:fs/promises"; import path from "node:path"; import { rootPath } from "./config"; import type { ParsedSshInvocation, ParsedSshRoute, SshCaptureResult } from "./ssh"; export interface SshRemoteCommandStreamHandlers { onStdout(chunk: Buffer): Promise | void; onStderr?(chunk: Buffer): Promise | void; } export interface SshRemoteCommandExecutor { streamInactivityTimeoutMs: number; runRemoteCommand(remoteCommand: string, input?: string, options?: { runtimeTimeoutMs?: number }): Promise; streamRemoteCommand?( remoteCommand: string, handlers: SshRemoteCommandStreamHandlers, input?: string, options?: { inactivityTimeoutMs?: number }, ): Promise; } export interface SshFileTransferCommandBuilders { buildRouteCommand(route: ParsedSshRoute, command: string[], options?: { stdin?: boolean }): string; buildWindowsPowerShellCommand(script: string): string; } type SshFileTransferOperation = | "stat" | "write-b64-argv" | "write-b64-stdin" | "write-b64-begin" | "write-b64-block-stdin" | "write-b64-abort" | "write-b64-commit"; interface SshFileTransferCliOptions { action: "upload" | "download"; localPath: string; remotePath: string; inactivityTimeoutMs?: number; allowTextTransfer: boolean; } interface SshFileTransferStat { bytes: number; sha256: string; } interface SshFileTransferEndpointStat extends SshFileTransferStat { side: "local" | "remote"; path: string; } interface SshFileTransferWriteResult { strategy: "argv" | "stdin" | "parallel-chunked-stdin"; chunks: number; concurrency: number; } interface SshFileTransferDownloadResult { remote: SshFileTransferStat; local: SshFileTransferStat; strategy: "tcp-pool-parallel-range-stream"; transport: "host.ssh.tcp-pool"; sourceRoute: string; sourcePath: string; chunks: number; concurrency: number; elapsedMs: number; throughputBytesPerSecond: number; remainingBytes: number; } export interface SshVerifiedDownloadResult { remotePath: string; localPath: string; bytes: number; sha256: string; verified: true; verification: Record; transfer: { strategy: SshFileTransferDownloadResult["strategy"]; transport: SshFileTransferDownloadResult["transport"]; sourceRoute: string; sourcePath: string; chunks: number; concurrency: number; elapsedMs: number; throughputBytesPerSecond: number; remainingBytes: number; }; } class SshFileTransferError extends Error { constructor(message: string, public readonly details: Record = {}) { super(message); this.name = "SshFileTransferError"; } } const fileTransferWriteB64ArgvLimit = 48_000; const fileTransferWriteRawChunkBytes = 3 * 1024 * 1024; const fileTransferDownloadMinimumRangeBytes = 1024 * 1024; const fileTransferDownloadRangeAlignment = 64 * 1024; const fileTransferConfigPath = "config/unidesk-cli.yaml"; const fileTransferWriteB64ChunkChars = 1_398_104; const fileTransferUploadProgressEveryChunks = 16; const fileTransferDownloadProgressEveryChunks = 256; interface SshFileTransferSettings { concurrency: number; transportOpenRetries: number; transportRetryDelayMs: number; } const knownTextExtensions = new Set([ ".bat", ".bib", ".c", ".cc", ".cfg", ".cmd", ".conf", ".cpp", ".css", ".csv", ".cts", ".cxx", ".diff", ".env", ".go", ".h", ".hpp", ".htm", ".html", ".ini", ".java", ".js", ".json", ".jsonl", ".jsx", ".less", ".log", ".md", ".markdown", ".mts", ".mjs", ".patch", ".ps1", ".py", ".rs", ".scss", ".sh", ".sql", ".svg", ".tex", ".text", ".toml", ".ts", ".tsv", ".tsx", ".txt", ".xml", ".yaml", ".yml", ]); const knownTextBasenames = new Set([ ".editorconfig", ".gitattributes", ".gitignore", "agents.md", "dockerfile", "license", "makefile", "mdtodo", "readme", ]); const fileTransferUsageHint = { code: "prefer-remote-text-operations", message: "已知文本路径会在传输前被拦截;读取使用 cat/head/tail/rg,修改直接使用 apply-patch quoted heredoc。upload/download 仅用于必要的二进制文件或生成物。", preferredOperations: { read: ["cat", "head", "tail", "rg"], edit: ["apply-patch <<'PATCH'"], }, }; function readFileTransferSettings(configPath = rootPath(fileTransferConfigPath)): SshFileTransferSettings { const source = `${fileTransferConfigPath}#trans.fileTransfer`; const root = Bun.YAML.parse(readFileSync(configPath, "utf8")) as unknown; if (typeof root !== "object" || root === null || Array.isArray(root)) { throw new Error(`${fileTransferConfigPath} must be an object`); } const trans = (root as Record).trans; if (typeof trans !== "object" || trans === null || Array.isArray(trans)) { throw new Error(`${fileTransferConfigPath}#trans must be an object`); } const transfer = (trans as Record).fileTransfer; if (typeof transfer !== "object" || transfer === null || Array.isArray(transfer)) { throw new Error(`${fileTransferConfigPath}#trans.fileTransfer must be an object`); } const values = transfer as Record; const concurrency = values.concurrency; if (!Number.isInteger(concurrency) || Number(concurrency) < 1) { throw new Error(`${source}.concurrency must be a positive integer`); } const transportOpenRetries = values.transportOpenRetries; if (!Number.isInteger(transportOpenRetries) || Number(transportOpenRetries) < 0) { throw new Error(`${source}.transportOpenRetries must be a non-negative integer`); } const transportRetryDelayMs = values.transportRetryDelayMs; if (!Number.isInteger(transportRetryDelayMs) || Number(transportRetryDelayMs) < 0) { throw new Error(`${source}.transportRetryDelayMs must be a non-negative integer`); } return { concurrency: Number(concurrency), transportOpenRetries: Number(transportOpenRetries), transportRetryDelayMs: Number(transportRetryDelayMs), }; } export function readFileTransferConcurrency(configPath = rootPath(fileTransferConfigPath)): number { return readFileTransferSettings(configPath).concurrency; } function isPreDispatchTransportFailure(result: SshCaptureResult): boolean { return /timed out waiting for provider session|provider ssh tcp data pool has no idle channel|requested ssh tcp data channel is not ready|provider-data-pool-exhausted/iu.test(result.stderr); } function emitTransportRetry( invocation: ParsedSshInvocation, operation: string, attempt: number, retry: SshFileTransferSettings, ): void { process.stderr.write(`${JSON.stringify({ event: "unidesk.ssh.file-transfer.transport-retry", route: invocation.route.raw, providerId: invocation.providerId, operation, attempt: attempt + 1, nextAttempt: attempt + 2, maxAttempts: retry.transportOpenRetries + 1, delayMs: retry.transportRetryDelayMs, })}\n`); } async function waitForTransportRetry(retry: SshFileTransferSettings): Promise { if (retry.transportRetryDelayMs > 0) await Bun.sleep(retry.transportRetryDelayMs); } export function isSshFileTransferOperation(args: string[]): boolean { const subcommand = args[0] ?? ""; return subcommand === "upload" || subcommand === "download"; } export async function runSshFileTransferOperation( invocation: ParsedSshInvocation, args: string[], executor: SshRemoteCommandExecutor, builders: SshFileTransferCommandBuilders, ): Promise { const options = parseSshFileTransferCliOptions(args); const localPath = path.resolve(options.localPath); const transferPolicy = enforceTextTransferPolicy(invocation, options, localPath); if (options.action === "upload") { const content = await readFile(localPath); const expected = { bytes: content.length, sha256: sha256HexBuffer(content) }; const write = await writeRemoteFileVerified(invocation, executor, builders, options.remotePath, content, options.inactivityTimeoutMs); const remote = await statRemoteFile(invocation, executor, builders, options.remotePath); assertTransferStat("upload final remote verification", options.remotePath, expected, remote); const verification = buildTransferVerification( { side: "local", path: localPath, ...expected }, { side: "remote", path: options.remotePath, ...remote }, ); process.stdout.write(`${JSON.stringify({ ok: true, command: "ssh upload", route: invocation.route.raw, providerId: invocation.providerId, localPath, remotePath: options.remotePath, bytes: expected.bytes, sha256: expected.sha256, verified: true, transferPolicy, hint: fileTransferUsageHint, verification, transfer: write, }, null, 2)}\n`); return 0; } const download = await downloadSshFileVerified(invocation, executor, builders, options.remotePath, localPath, options.inactivityTimeoutMs); process.stdout.write(`${JSON.stringify({ ok: true, command: "ssh download", route: invocation.route.raw, providerId: invocation.providerId, remotePath: options.remotePath, localPath, bytes: download.bytes, sha256: download.sha256, verified: true, transferPolicy, hint: fileTransferUsageHint, verification: download.verification, transfer: download.transfer, }, null, 2)}\n`); return 0; } export async function downloadSshFileVerified( invocation: ParsedSshInvocation, executor: SshRemoteCommandExecutor, builders: SshFileTransferCommandBuilders, remotePath: string, localPath: string, inactivityTimeoutMs?: number, ): Promise { const read = await downloadRemoteFileVerified(invocation, executor, builders, remotePath, localPath, inactivityTimeoutMs); const verification = buildTransferVerification( { side: "remote", path: remotePath, ...read.remote }, { side: "local", path: localPath, ...read.local }, ); return { remotePath, localPath, bytes: read.remote.bytes, sha256: read.remote.sha256, verified: true, verification, transfer: { strategy: read.strategy, transport: read.transport, sourceRoute: read.sourceRoute, sourcePath: read.sourcePath, chunks: read.chunks, concurrency: read.concurrency, elapsedMs: read.elapsedMs, throughputBytesPerSecond: read.throughputBytesPerSecond, remainingBytes: read.remainingBytes, }, }; } function parseSshFileTransferCliOptions(args: string[]): SshFileTransferCliOptions { const action = args[0]; if (action !== "upload" && action !== "download") throw new Error("ssh file transfer requires upload or download"); const positionals: string[] = []; let inactivityTimeoutMs: number | undefined; let allowTextTransfer = false; for (let index = 1; index < args.length; index += 1) { const arg = args[index] ?? ""; if (arg === "--") { positionals.push(...args.slice(index + 1)); break; } if (arg === "--chunk-bytes" || arg === "--block-bytes") { throw new Error(`unsupported ssh ${action} option: ${arg}; downloads use host.ssh.tcp-pool stdout streaming and no longer support the legacy base64 block reader`); } if (arg === "--allow-text-transfer") { allowTextTransfer = true; continue; } if (arg === "--inactivity-timeout-ms" || arg === "--runtime-timeout-ms") { const value = args[index + 1]; if (value === undefined) throw new Error(`ssh ${action} ${arg} requires a positive integer value`); index += 1; inactivityTimeoutMs = parsePositiveRuntimeTimeoutMs(value, `ssh ${action} ${arg}`); continue; } if (arg.startsWith("--inactivity-timeout-ms=")) { inactivityTimeoutMs = parsePositiveRuntimeTimeoutMs(arg.slice("--inactivity-timeout-ms=".length), `ssh ${action} --inactivity-timeout-ms`); continue; } if (arg.startsWith("--runtime-timeout-ms=")) { inactivityTimeoutMs = parsePositiveRuntimeTimeoutMs(arg.slice("--runtime-timeout-ms=".length), `ssh ${action} --runtime-timeout-ms`); continue; } if (arg === "--help" || arg === "-h" || arg === "help") { throw new Error(`ssh ${action} usage: trans ${action} ${action === "upload" ? " " : " "}`); } if (arg.startsWith("-")) throw new Error(`unsupported ssh ${action} option: ${arg}`); positionals.push(arg); } if (positionals.length !== 2) { throw new Error(`ssh ${action} requires exactly two paths: ${action === "upload" ? " " : " "}`); } const [first, second] = positionals; if (!first || !second) throw new Error(`ssh ${action} paths must be non-empty`); return action === "upload" ? { action, localPath: first, remotePath: second, inactivityTimeoutMs, allowTextTransfer } : { action, remotePath: first, localPath: second, inactivityTimeoutMs, allowTextTransfer }; } function enforceTextTransferPolicy( invocation: ParsedSshInvocation, options: SshFileTransferCliOptions, resolvedLocalPath: string, ): Record { const knownTextPaths = [ knownTextPath(options.remotePath), knownTextPath(resolvedLocalPath), ].filter((item): item is { path: string; reason: string } => item !== null); if (knownTextPaths.length > 0 && !options.allowTextTransfer) { const readCommand = `trans ${invocation.route.raw} cat ${JSON.stringify(options.remotePath)}`; const editCommand = `trans ${invocation.route.raw} apply-patch <<'PATCH'`; throw new SshFileTransferError("remote text transfer is discouraged; use the native text operation", { code: "text-transfer-discouraged", action: options.action, route: invocation.route.raw, knownTextPaths, preferred: options.action === "download" ? { operation: "cat", command: readCommand } : { operation: "apply-patch", command: editCommand }, alternatives: options.action === "download" ? [readCommand, `trans ${invocation.route.raw} head -n 80 ${JSON.stringify(options.remotePath)}`, `trans ${invocation.route.raw} rg ${JSON.stringify(options.remotePath)}`] : [editCommand], override: { option: "--allow-text-transfer", purpose: "仅用于确需整体搬运的生成物;成功结果会记录 overrideRequested=true。", }, targetCreated: false, remoteOperationStarted: false, }); } return { code: "binary-or-generated-artifact-transfer", knownTextPaths, overrideRequested: options.allowTextTransfer, overrideUsed: knownTextPaths.length > 0 && options.allowTextTransfer, }; } function knownTextPath(rawPath: string): { path: string; reason: string } | null { const normalized = rawPath.trim().replace(/\\/gu, "/").replace(/\/+$/gu, ""); const basename = normalized.slice(normalized.lastIndexOf("/") + 1).toLowerCase(); if (knownTextBasenames.has(basename)) return { path: rawPath, reason: `basename:${basename}` }; const extension = path.posix.extname(basename); return knownTextExtensions.has(extension) ? { path: rawPath, reason: `extension:${extension}` } : null; } function parsePositiveRuntimeTimeoutMs(value: string, label: string): number { const parsed = Number(value); if (!Number.isFinite(parsed) || parsed <= 0 || !Number.isInteger(parsed)) { throw new Error(`${label} must be a positive integer number of milliseconds`); } return parsed; } async function writeRemoteFileVerified( invocation: ParsedSshInvocation, executor: SshRemoteCommandExecutor, builders: SshFileTransferCommandBuilders, remotePath: string, content: Buffer, runtimeTimeoutMs?: number, ): Promise { const expectedBytes = String(content.length); const expectedSha256 = sha256HexBuffer(content); const encoded = content.length <= fileTransferWriteB64ArgvLimit ? content.toString("base64") : ""; if (invocation.route.plane !== "win" && encoded.length > 0 && encoded.length <= fileTransferWriteB64ArgvLimit) { await checkedFileTransfer(invocation, executor, builders, "write-b64-argv", [remotePath, expectedBytes, expectedSha256, ...chunkString(encoded, fileTransferWriteB64ChunkChars)], undefined, runtimeTimeoutMs); return { strategy: "argv", chunks: encoded.length === 0 ? 0 : Math.ceil(encoded.length / fileTransferWriteB64ChunkChars), concurrency: 1, }; } if (content.length <= fileTransferWriteRawChunkBytes) { try { await checkedFileTransfer(invocation, executor, builders, "write-b64-stdin", [remotePath, expectedBytes, expectedSha256], content.toString("base64"), runtimeTimeoutMs); return { strategy: "stdin", chunks: 1, concurrency: 1 }; } catch { // Fall through to chunked upload with per-block base64 encoding. } } const token = `${process.pid}-${Date.now()}-${randomBytes(4).toString("hex")}-${expectedSha256.slice(0, 12)}`; const startedAtMs = Date.now(); const blocks = byteRanges(content.length, fileTransferWriteRawChunkBytes); const concurrency = readFileTransferConcurrency(); let completedChunks = 0; let completedBytes = 0; await checkedFileTransfer(invocation, executor, builders, "write-b64-begin", [remotePath, token, expectedBytes], undefined, runtimeTimeoutMs); try { await runConcurrent(blocks, concurrency, async ({ offset, length }) => { const block = content.subarray(offset, offset + length); await checkedFileTransfer( invocation, executor, builders, "write-b64-block-stdin", [remotePath, token, String(offset), String(length), sha256HexBuffer(block)], block.toString("base64"), runtimeTimeoutMs, ); completedChunks += 1; completedBytes += length; emitUploadProgress( invocation, remotePath, completedChunks, completedBytes, content.length, length, startedAtMs, ); }); await checkedFileTransfer(invocation, executor, builders, "write-b64-commit", [remotePath, token, expectedBytes, expectedSha256], undefined, runtimeTimeoutMs); } catch (error) { await checkedFileTransfer(invocation, executor, builders, "write-b64-abort", [remotePath, token], undefined, runtimeTimeoutMs).catch(() => undefined); throw error; } emitUploadProgress(invocation, remotePath, Math.max(1, completedChunks), content.length, content.length, 0, startedAtMs, true); return { strategy: "parallel-chunked-stdin", chunks: blocks.length, concurrency: Math.min(concurrency, blocks.length), }; } async function downloadRemoteFileVerified( invocation: ParsedSshInvocation, executor: SshRemoteCommandExecutor, builders: SshFileTransferCommandBuilders, remotePath: string, localPath: string, inactivityTimeoutMs?: number, ): Promise { if (executor.streamRemoteCommand === undefined) { throw new SshFileTransferError("ssh download requires a tcp-pool stdout stream sink", { route: invocation.route.raw, providerId: invocation.providerId, remotePath, localPath, requiredTransport: "host.ssh.tcp-pool", blocker: "streamRemoteCommand unavailable", }); } const streamSource = invocation.route.plane === "win" ? { route: invocation.route, path: resolveWindowsTransferPath(remotePath, invocation.route.workspace), } : { route: invocation.route, path: remotePath }; if (invocation.route.plane === "win") { process.stderr.write(`${JSON.stringify({ event: "unidesk.ssh.download.win-native-stream", at: new Date().toISOString(), route: invocation.route.raw, providerId: invocation.providerId, remotePath, sourceRoute: streamSource.route.raw, sourcePath: streamSource.path, transport: "host.ssh.tcp-pool", })}\n`); } const remote = await statRemoteFile(invocation, executor, builders, remotePath); await mkdir(path.dirname(localPath), { recursive: true }); const concurrency = readFileTransferConcurrency(); const ranges = parallelDownloadRanges(remote.bytes, concurrency); const startedAtMs = Date.now(); const output = await open(localPath, "w", 0o600); await output.truncate(remote.bytes); const retry = readFileTransferSettings(); let actualBytes = 0; let chunkCount = 0; try { await runConcurrent(ranges, concurrency, async (range) => { const remoteCommand = buildStreamDownloadRemoteCommand( streamSource.route, builders, streamSource.path, range.offset, range.length, ); let rangeBytes = 0; let result: SshCaptureResult; for (let attempt = 0; ; attempt += 1) { result = await executor.streamRemoteCommand!(remoteCommand, { onStdout: async (chunk) => { if (rangeBytes + chunk.length > range.length) { throw new SshFileTransferError("remote ssh download range exceeded expected bytes", { remotePath, offset: range.offset, expectedBytes: range.length, actualBytes: rangeBytes + chunk.length, }); } const chunkOffset = rangeBytes; rangeBytes += chunk.length; actualBytes += chunk.length; chunkCount += 1; await writeFileHandleChunk(output, chunk, range.offset + chunkOffset); emitDownloadProgress(invocation, remotePath, chunkCount, actualBytes, remote.bytes, chunk.length, startedAtMs); }, }, undefined, { inactivityTimeoutMs: inactivityTimeoutMs ?? executor.streamInactivityTimeoutMs }); if (result.exitCode === 0 || rangeBytes > 0 || !isPreDispatchTransportFailure(result) || attempt >= retry.transportOpenRetries) break; emitTransportRetry(invocation, "download-range", attempt, retry); await waitForTransportRetry(retry); } if (result.exitCode !== 0 || rangeBytes !== range.length) { throw new SshFileTransferError("remote ssh download range failed", { route: invocation.route.raw, providerId: invocation.providerId, remotePath, localPath, offset: range.offset, expectedBytes: range.length, actualBytes: rangeBytes, exitCode: result.exitCode, stdout: transferTextSnapshot(result.stdout, { headChars: 120, tailChars: 500 }), stderr: transferTextSnapshot(result.stderr, { headChars: 120, tailChars: 1000 }), }); } }); await output.close(); const local = { bytes: actualBytes, sha256: await sha256File(localPath), }; if (actualBytes !== remote.bytes) { throw new SshFileTransferError("remote ssh download ranges returned incomplete data", { remotePath, expectedBytes: remote.bytes, actualBytes, }); } const elapsedMs = Math.max(1, Date.now() - startedAtMs); assertTransferStat("download tcp-pool stdout stream verification", localPath, remote, local); emitDownloadProgress(invocation, remotePath, Math.max(1, chunkCount), actualBytes, remote.bytes, 0, startedAtMs, true); return { remote, local, strategy: "tcp-pool-parallel-range-stream", transport: "host.ssh.tcp-pool", sourceRoute: streamSource.route.raw, sourcePath: streamSource.path, chunks: chunkCount, concurrency: Math.min(concurrency, ranges.length), elapsedMs, throughputBytesPerSecond: Math.round((actualBytes * 1000) / elapsedMs), remainingBytes: Math.max(0, remote.bytes - actualBytes), }; } catch (error) { await output.close().catch(() => undefined); await rm(localPath, { force: true }).catch(() => undefined); throw error; } } function resolveWindowsTransferPath(rawPath: string, basePath: string | null): string { const normalizedRaw = stripWindowsLongPathPrefix(rawPath.trim()).replace(/\//gu, "\\"); if (normalizedRaw.length === 0) { throw new SshFileTransferError("ssh win download requires a non-empty remote path"); } if (/^\\\\/u.test(normalizedRaw)) { throw new SshFileTransferError("ssh win download cannot auto-map UNC paths to WSL /mnt/", { remotePath: rawPath, fallback: "copy the file to a drive-letter path such as D:\\tmp, then retry download", }); } if (/^[A-Za-z]:/u.test(normalizedRaw)) return normalizeWindowsAbsolutePath(normalizedRaw); const base = basePath === null ? "" : stripWindowsLongPathPrefix(basePath.trim()).replace(/\//gu, "\\"); const baseMatch = base.match(/^([A-Za-z]):\\?(.*)$/u); if (baseMatch === null) { throw new SshFileTransferError("ssh win download with a relative path requires a drive-letter route workspace", { remotePath: rawPath, routeWorkspace: basePath, example: "trans D518:win/d/tmp download image.png /tmp/image.png", }); } const drive = baseMatch[1]!; const baseRest = baseMatch[2] ?? ""; return normalizeWindowsAbsolutePath(`${drive}:\\${[baseRest, normalizedRaw].filter((part) => part.length > 0).join("\\")}`); } function stripWindowsLongPathPrefix(value: string): string { if (value.startsWith("\\\\?\\")) return value.slice(4); return value; } function normalizeWindowsAbsolutePath(value: string): string { const match = value.match(/^([A-Za-z]):\\?(.*)$/u); if (match === null) return value; const drive = match[1]!.toUpperCase(); const segments = normalizeWindowsPathSegments((match[2] ?? "").split(/[\\/]+/u)); return `${drive}:\\${segments.join("\\")}`; } function normalizeWindowsPathSegments(segments: string[]): string[] { const normalized: string[] = []; for (const segment of segments) { if (segment.length === 0 || segment === ".") continue; if (segment === "..") { if (normalized.length === 0) throw new SshFileTransferError("ssh win download path escapes the drive root"); normalized.pop(); continue; } normalized.push(segment); } return normalized; } interface FileTransferRange { offset: number; length: number; } function byteRanges(totalBytes: number, rangeBytes: number): FileTransferRange[] { const ranges: FileTransferRange[] = []; for (let offset = 0; offset < totalBytes; offset += rangeBytes) { ranges.push({ offset, length: Math.min(rangeBytes, totalBytes - offset) }); } return ranges; } function parallelDownloadRanges(totalBytes: number, concurrency: number): FileTransferRange[] { if (totalBytes === 0) return []; const rangeCount = Math.min(concurrency, Math.max(1, Math.ceil(totalBytes / fileTransferDownloadMinimumRangeBytes))); const unalignedBytes = Math.ceil(totalBytes / rangeCount); const rangeBytes = Math.ceil(unalignedBytes / fileTransferDownloadRangeAlignment) * fileTransferDownloadRangeAlignment; return byteRanges(totalBytes, rangeBytes); } async function runConcurrent( items: T[], concurrency: number, worker: (item: T) => Promise, ): Promise { let nextIndex = 0; let firstError: unknown = null; const runners = Array.from({ length: Math.min(concurrency, items.length) }, async () => { while (firstError === null) { const index = nextIndex; nextIndex += 1; if (index >= items.length) return; try { await worker(items[index]!); } catch (error) { firstError ??= error; } } }); await Promise.all(runners); if (firstError !== null) throw firstError; } function emitDownloadProgress( invocation: ParsedSshInvocation, remotePath: string, chunkCount: number, actualBytes: number, expectedBytes: number, lastChunkBytes: number, startedAtMs: number, force = false, ): void { if (!force && chunkCount !== 1 && actualBytes < expectedBytes && chunkCount % fileTransferDownloadProgressEveryChunks !== 0) return; const elapsedMs = Math.max(1, Date.now() - startedAtMs); process.stderr.write(`${JSON.stringify({ event: "unidesk.ssh.download.progress", at: new Date().toISOString(), route: invocation.route.raw, providerId: invocation.providerId, remotePath, strategy: "tcp-pool-parallel-range-stream", transport: "host.ssh.tcp-pool", chunks: chunkCount, bytes: actualBytes, totalBytes: expectedBytes, actualBytes, expectedBytes, lastChunkBytes, remainingBytes: Math.max(0, expectedBytes - actualBytes), elapsedMs, throughputBytesPerSecond: Math.round((actualBytes * 1000) / elapsedMs), })}\n`); } function emitUploadProgress( invocation: ParsedSshInvocation, remotePath: string, chunkCount: number, actualBytes: number, expectedBytes: number, lastChunkBytes: number, startedAtMs: number, force = false, ): void { if (!force && chunkCount !== 1 && actualBytes < expectedBytes && chunkCount % fileTransferUploadProgressEveryChunks !== 0) return; const elapsedMs = Math.max(1, Date.now() - startedAtMs); process.stderr.write(`${JSON.stringify({ event: "unidesk.ssh.upload.progress", at: new Date().toISOString(), route: invocation.route.raw, providerId: invocation.providerId, remotePath, strategy: "parallel-chunked-stdin", chunks: chunkCount, bytes: actualBytes, totalBytes: expectedBytes, actualBytes, expectedBytes, lastChunkBytes, remainingBytes: Math.max(0, expectedBytes - actualBytes), elapsedMs, throughputBytesPerSecond: Math.round((actualBytes * 1000) / elapsedMs), })}\n`); } async function statRemoteFile( invocation: ParsedSshInvocation, executor: SshRemoteCommandExecutor, builders: SshFileTransferCommandBuilders, remotePath: string, ): Promise { const stat = await checkedFileTransfer(invocation, executor, builders, "stat", [remotePath]); return parseFileTransferStat(stat.stdout, stat.stderr, remotePath); } function buildStreamDownloadRemoteCommand( route: ParsedSshRoute, builders: SshFileTransferCommandBuilders, remotePath: string, offset: number, length: number, ): string { if (route.plane === "win") { return builders.buildWindowsPowerShellCommand(windowsStreamDownloadScript(route.workspace, remotePath, offset, length)); } return builders.buildRouteCommand(route, [ "sh", "-c", "dd if=\"$1\" bs=65536 skip=\"$2\" count=\"$3\" 2>/dev/null | head -c \"$4\"", "unidesk-download", remotePath, String(offset / fileTransferDownloadRangeAlignment), String(Math.ceil(length / fileTransferDownloadRangeAlignment)), String(length), ]); } function windowsStreamDownloadScript(basePath: string | null, remotePath: string, offset: number, length: number): string { return [ "$ErrorActionPreference = 'Stop';", "$ProgressPreference = 'SilentlyContinue';", `$basePath = ${powerShellSingleQuote(basePath ?? "")};`, `$targetArg = ${powerShellSingleQuote(remotePath)};`, `$offset = [Int64]${offset};`, `$length = [Int64]${length};`, "if ([System.IO.Path]::IsPathRooted($targetArg)) { $target = [System.IO.Path]::GetFullPath($targetArg) } elseif (-not [string]::IsNullOrWhiteSpace($basePath)) { $target = [System.IO.Path]::GetFullPath([System.IO.Path]::Combine($basePath, $targetArg)) } else { $target = [System.IO.Path]::GetFullPath($targetArg) };", "if (-not [System.IO.File]::Exists($target)) { [Console]::Error.WriteLine('file not found: ' + $target); exit 1 };", "$inputStream = [System.IO.File]::Open($target, [System.IO.FileMode]::Open, [System.IO.FileAccess]::Read, [System.IO.FileShare]::ReadWrite);", "try { [void]$inputStream.Seek($offset, [System.IO.SeekOrigin]::Begin); $outputStream = [Console]::OpenStandardOutput(); $buffer = [byte[]]::new(65536); $remaining = $length; while ($remaining -gt 0) { $requested = [Math]::Min([Int64]$buffer.Length, $remaining); $count = $inputStream.Read($buffer, 0, [int]$requested); if ($count -le 0) { break }; $outputStream.Write($buffer, 0, $count); $outputStream.Flush(); $remaining -= $count }; if ($remaining -ne 0) { [Console]::Error.WriteLine('download range ended early: remaining=' + $remaining); exit 26 } } finally { $inputStream.Dispose() }", ].join(" "); } async function writeFileHandleChunk( file: Awaited>, chunk: Buffer, position: number, ): Promise { let written = 0; while (written < chunk.length) { const result = await file.write(chunk, written, chunk.length - written, position + written); if (result.bytesWritten <= 0) throw new Error("local parallel download write made no progress"); written += result.bytesWritten; } } async function sha256File(filePath: string): Promise { const hash = createHash("sha256"); for await (const chunk of createReadStream(filePath)) hash.update(chunk as Buffer); return hash.digest("hex"); } async function checkedFileTransfer( invocation: ParsedSshInvocation, executor: SshRemoteCommandExecutor, builders: SshFileTransferCommandBuilders, operation: SshFileTransferOperation, args: string[], input?: string, runtimeTimeoutMs?: number, ): Promise { const remoteCommand = buildFileTransferRemoteCommand(invocation.route, builders, operation, args, input !== undefined); const retry = readFileTransferSettings(); for (let attempt = 0; ; attempt += 1) { const result = await executor.runRemoteCommand(remoteCommand, input, { runtimeTimeoutMs }); if (result.exitCode === 0) return result; const preDispatchFailure = isPreDispatchTransportFailure(result); if (!preDispatchFailure || attempt >= retry.transportOpenRetries) { throw new SshFileTransferError("remote ssh file transfer operation failed", { route: invocation.route.raw, operation, args: args.slice(0, 4), attempt: attempt + 1, exitCode: result.exitCode, stdout: transferTextSnapshot(result.stdout, { headChars: operation === "stat" ? 500 : 120, tailChars: 500 }), stderr: transferTextSnapshot(result.stderr, { headChars: 120, tailChars: 1000 }), }); } emitTransportRetry(invocation, operation, attempt, retry); await waitForTransportRetry(retry); } } function buildFileTransferRemoteCommand( route: ParsedSshRoute, builders: SshFileTransferCommandBuilders, operation: SshFileTransferOperation, args: string[], hasInput: boolean, ): string { if (route.plane === "win") return buildWindowsFileTransferCommand(route, builders, operation, args); return builders.buildRouteCommand(route, ["sh", "-c", posixFileTransferScript(), "unidesk-file-transfer", operation, ...args], { stdin: hasInput }); } function parseFileTransferStat(stdout: string, stderr: string, remotePath: string): SshFileTransferStat { const lines = stdout.split(/\r?\n/u).map((line) => line.trim()).filter((line) => line.length > 0); for (const line of lines) { const [bytesText, sha256, extra] = line.split(/\s+/u); const bytes = Number(bytesText); if (extra === undefined && Number.isSafeInteger(bytes) && bytes >= 0 && /^[0-9a-f]{64}$/u.test(sha256 ?? "")) { return { bytes, sha256: sha256! }; } } { throw new SshFileTransferError("remote file transfer stat returned invalid metadata", { remotePath, expected: " ", stdout: transferTextSnapshot(stdout, { headChars: 500, tailChars: 500 }), stderr: transferTextSnapshot(stderr, { headChars: 120, tailChars: 1000 }), candidateLines: lines.slice(0, 8).map((line) => line.slice(0, 200)), candidateLineCount: lines.length, }); } } function transferTextSnapshot(value: string, limits: { headChars: number; tailChars: number }): Record { return { bytes: Buffer.byteLength(value, "utf8"), chars: value.length, head: value.slice(0, limits.headChars), tail: value.slice(Math.max(0, value.length - limits.tailChars)), truncated: value.length > limits.headChars + limits.tailChars, }; } function assertTransferStat(label: string, pathName: string, expected: SshFileTransferStat, actual: SshFileTransferStat): void { if (expected.bytes === actual.bytes && expected.sha256 === actual.sha256) return; throw new SshFileTransferError(`${label} failed`, { path: pathName, expected, actual, }); } function buildTransferVerification(source: SshFileTransferEndpointStat, target: SshFileTransferEndpointStat): Record { return { automatic: true, algorithm: "sha256", checked: ["bytes", "sha256"], verified: source.bytes === target.bytes && source.sha256 === target.sha256, source, target, match: { bytes: source.bytes === target.bytes, sha256: source.sha256 === target.sha256, }, }; } function sha256HexBuffer(value: Buffer): string { return createHash("sha256").update(value).digest("hex"); } function chunkString(value: string, chunkSize: number): string[] { const chunks: string[] = []; for (let index = 0; index < value.length; index += chunkSize) { chunks.push(value.slice(index, index + chunkSize)); } return chunks.length > 0 ? chunks : [""]; } function posixFileTransferScript(): string { return [ "set -eu", "sha256_file() {", " if command -v sha256sum >/dev/null 2>&1; then sha256sum -- \"$1\" | awk '{print $1}'; return; fi", " if command -v shasum >/dev/null 2>&1; then shasum -a 256 -- \"$1\" | awk '{print $1}'; return; fi", " if command -v openssl >/dev/null 2>&1; then openssl dgst -sha256 -- \"$1\" | awk '{print $NF}'; return; fi", " printf 'missing sha256 tool\\n' >&2; return 127", "}", "ensure_parent() { case \"$1\" in */*) parent=${1%/*}; mkdir -p -- \"$parent\";; esac; }", "verify_tmp() {", " target=$1; tmp=$2; expected_bytes=$3; expected_sha256=$4", " actual_bytes=$(wc -c < \"$tmp\" | tr -d '[:space:]')", " if [ \"$actual_bytes\" != \"$expected_bytes\" ]; then rm -f -- \"$tmp\"; printf 'transfer byte count mismatch for %s: expected=%s actual=%s\\n' \"$target\" \"$expected_bytes\" \"$actual_bytes\" >&2; exit 23; fi", " actual_sha256=$(sha256_file \"$tmp\")", " if [ \"$actual_sha256\" != \"$expected_sha256\" ]; then rm -f -- \"$tmp\"; printf 'transfer sha256 mismatch for %s: expected=%s actual=%s\\n' \"$target\" \"$expected_sha256\" \"$actual_sha256\" >&2; exit 24; fi", "}", "set_tmp_paths() {", " target=$1; token=$2", " case \"$token\" in ''|*[!a-zA-Z0-9_.-]*) printf 'invalid transfer temp token\\n' >&2; exit 2;; esac", " base=${target##*/}; dir=.", " case \"$target\" in */*) dir=${target%/*};; esac", " tmp=\"$dir/.${base}.unidesk-transfer-${token}.tmp\"", " tmp_b64=\"$tmp.b64\"", "}", "op=$1; shift", "case \"$op\" in", " stat)", " target=$1; bytes=$(wc -c < \"$target\" | tr -d '[:space:]'); digest=$(sha256_file \"$target\"); printf '%s %s\\n' \"$bytes\" \"$digest\"", " ;;", " write-b64-argv)", " target=$1; expected_bytes=$2; expected_sha256=$3; shift 3", " ensure_parent \"$target\"; base=${target##*/}; dir=.; case \"$target\" in */*) dir=${target%/*};; esac", " tmp=\"$dir/.${base}.unidesk-transfer-$$.tmp\"; tmp_b64=\"$tmp.b64\"; : > \"$tmp_b64\"", " for chunk in \"$@\"; do printf '%s' \"$chunk\" >> \"$tmp_b64\"; done", " if ! base64 -d < \"$tmp_b64\" > \"$tmp\"; then rm -f -- \"$tmp\" \"$tmp_b64\"; printf 'transfer base64 decode failed for %s\\n' \"$target\" >&2; exit 22; fi", " rm -f -- \"$tmp_b64\"; verify_tmp \"$target\" \"$tmp\" \"$expected_bytes\" \"$expected_sha256\"; mv -f -- \"$tmp\" \"$target\"", " actual_sha256=$(sha256_file \"$target\"); if [ \"$actual_sha256\" != \"$expected_sha256\" ]; then printf 'transfer final sha256 mismatch for %s\\n' \"$target\" >&2; exit 25; fi", " ;;", " write-b64-stdin)", " target=$1; expected_bytes=$2; expected_sha256=$3", " ensure_parent \"$target\"; base=${target##*/}; dir=.; case \"$target\" in */*) dir=${target%/*};; esac", " tmp=\"$dir/.${base}.unidesk-transfer-$$.tmp\"", " if ! base64 -d > \"$tmp\"; then rm -f -- \"$tmp\"; printf 'transfer base64 decode failed for %s\\n' \"$target\" >&2; exit 22; fi", " verify_tmp \"$target\" \"$tmp\" \"$expected_bytes\" \"$expected_sha256\"; mv -f -- \"$tmp\" \"$target\"", " actual_sha256=$(sha256_file \"$target\"); if [ \"$actual_sha256\" != \"$expected_sha256\" ]; then printf 'transfer final sha256 mismatch for %s\\n' \"$target\" >&2; exit 25; fi", " ;;", " write-b64-begin)", " target=$1; token=$2; expected_bytes=$3; ensure_parent \"$target\"; set_tmp_paths \"$target\" \"$token\"", " if command -v truncate >/dev/null 2>&1; then truncate -s \"$expected_bytes\" \"$tmp\"; else dd if=/dev/zero of=\"$tmp\" bs=1 count=0 seek=\"$expected_bytes\" 2>/dev/null; fi", " ;;", " write-b64-block-stdin)", " target=$1; token=$2; offset=$3; expected_block_bytes=$4; expected_block_sha256=$5; set_tmp_paths \"$target\" \"$token\"; part=\"$tmp.$offset.part\"", " if ! base64 -d > \"$part\"; then rm -f -- \"$part\"; printf 'transfer base64 block decode failed for %s at %s\\n' \"$target\" \"$offset\" >&2; exit 22; fi", " actual_block_bytes=$(wc -c < \"$part\" | tr -d '[:space:]'); actual_block_sha256=$(sha256_file \"$part\")", " if [ \"$actual_block_bytes\" != \"$expected_block_bytes\" ] || [ \"$actual_block_sha256\" != \"$expected_block_sha256\" ]; then rm -f -- \"$part\"; printf 'transfer block verification failed for %s at %s\\n' \"$target\" \"$offset\" >&2; exit 24; fi", " if ! dd if=\"$part\" of=\"$tmp\" bs=1 seek=\"$offset\" conv=notrunc 2>/dev/null; then rm -f -- \"$part\"; printf 'transfer block write failed for %s at %s\\n' \"$target\" \"$offset\" >&2; exit 27; fi", " rm -f -- \"$part\"", " ;;", " write-b64-abort)", " target=$1; token=$2; set_tmp_paths \"$target\" \"$token\"; rm -f -- \"$tmp\" \"$tmp\".*.part", " ;;", " write-b64-commit)", " target=$1; token=$2; expected_bytes=$3; expected_sha256=$4; set_tmp_paths \"$target\" \"$token\"", " verify_tmp \"$target\" \"$tmp\" \"$expected_bytes\" \"$expected_sha256\"; mv -f -- \"$tmp\" \"$target\"", " actual_sha256=$(sha256_file \"$target\"); if [ \"$actual_sha256\" != \"$expected_sha256\" ]; then printf 'transfer final sha256 mismatch for %s\\n' \"$target\" >&2; exit 25; fi", " ;;", " *) printf 'unsupported transfer op: %s\\n' \"$op\" >&2; exit 2;;", "esac", ].join("\n"); } function buildWindowsFileTransferCommand( route: ParsedSshRoute, builders: SshFileTransferCommandBuilders, operation: SshFileTransferOperation, args: string[], ): string { return builders.buildWindowsPowerShellCommand(windowsFileTransferScript(route.workspace, operation, args)); } function windowsFileTransferScript(basePath: string | null, operation: SshFileTransferOperation, args: string[]): string { const target = args[0] ?? ""; const tokenOrBlock = args[1] ?? ""; const third = args[2] ?? ""; const fourth = args[3] ?? ""; const fifth = args[4] ?? ""; const argvChunks = args.slice(3).map(powerShellSingleQuote).join(", "); return [ "$ErrorActionPreference = 'Stop';", "$ProgressPreference = 'SilentlyContinue';", "[Console]::InputEncoding = [System.Text.UTF8Encoding]::new();", "[Console]::OutputEncoding = [System.Text.UTF8Encoding]::new();", "$OutputEncoding = [System.Text.UTF8Encoding]::new();", `$basePath = ${powerShellSingleQuote(basePath ?? "")};`, `$operation = ${powerShellSingleQuote(operation)};`, `$targetArg = ${powerShellSingleQuote(target)};`, `$arg1 = ${powerShellSingleQuote(tokenOrBlock)};`, `$arg2 = ${powerShellSingleQuote(third)};`, `$arg3 = ${powerShellSingleQuote(fourth)};`, `$arg4 = ${powerShellSingleQuote(fifth)};`, `$argvChunks = @(${argvChunks});`, "function Fail([string]$Message, [int]$Code) { [Console]::Error.WriteLine($Message); exit $Code }", "function Resolve-UnideskPath([string]$Raw) { if ([string]::IsNullOrWhiteSpace($Raw)) { Fail 'empty transfer path' 2 }; if ([System.IO.Path]::IsPathRooted($Raw)) { return [System.IO.Path]::GetFullPath($Raw) }; if (-not [string]::IsNullOrWhiteSpace($basePath)) { return [System.IO.Path]::GetFullPath([System.IO.Path]::Combine($basePath, $Raw)) }; return [System.IO.Path]::GetFullPath($Raw) }", "function Ensure-Parent([string]$Target) { $parent = [System.IO.Path]::GetDirectoryName($Target); if (-not [string]::IsNullOrWhiteSpace($parent)) { [System.IO.Directory]::CreateDirectory($parent) | Out-Null } }", "function Get-Sha256([string]$Path) { return (Get-FileHash -LiteralPath $Path -Algorithm SHA256).Hash.ToLowerInvariant() }", "function Set-TmpPaths([string]$Target, [string]$Token) { if ($Token -notmatch '^[A-Za-z0-9_.-]+$') { Fail 'invalid transfer temp token' 2 }; $dir = [System.IO.Path]::GetDirectoryName($Target); if ([string]::IsNullOrWhiteSpace($dir)) { $dir = (Get-Location).ProviderPath }; $base = [System.IO.Path]::GetFileName($Target); $script:tmp = [System.IO.Path]::Combine($dir, '.' + $base + '.unidesk-transfer-' + $Token + '.tmp'); $script:tmpB64 = $script:tmp + '.b64' }", "function Verify-Temp([string]$Target, [string]$Tmp, [Int64]$ExpectedBytes, [string]$ExpectedSha256) { $actualBytes = ([System.IO.FileInfo]$Tmp).Length; if ($actualBytes -ne $ExpectedBytes) { Remove-Item -LiteralPath $Tmp -Force -ErrorAction SilentlyContinue; Fail ('transfer byte count mismatch for ' + $Target + ': expected=' + $ExpectedBytes + ' actual=' + $actualBytes) 23 }; $actualSha256 = Get-Sha256 $Tmp; if ($actualSha256 -ne $ExpectedSha256) { Remove-Item -LiteralPath $Tmp -Force -ErrorAction SilentlyContinue; Fail ('transfer sha256 mismatch for ' + $Target + ': expected=' + $ExpectedSha256 + ' actual=' + $actualSha256) 24 } }", "function Decode-ToTarget([string]$Target, [string]$Encoded, [Int64]$ExpectedBytes, [string]$ExpectedSha256) { Ensure-Parent $Target; Set-TmpPaths $Target ([guid]::NewGuid().ToString('N')); try { $bytes = [Convert]::FromBase64String(($Encoded -replace '\\s','')) } catch { Fail ('transfer base64 decode failed for ' + $Target + ': ' + $_.Exception.Message) 22 }; [System.IO.File]::WriteAllBytes($script:tmp, $bytes); Verify-Temp $Target $script:tmp $ExpectedBytes $ExpectedSha256; Move-Item -LiteralPath $script:tmp -Destination $Target -Force; $actualSha256 = Get-Sha256 $Target; if ($actualSha256 -ne $ExpectedSha256) { Fail ('transfer final sha256 mismatch for ' + $Target) 25 } }", "$target = Resolve-UnideskPath $targetArg;", "switch ($operation) {", " 'stat' { if (-not (Test-Path -LiteralPath $target -PathType Leaf)) { Fail ('file not found: ' + $target) 1 }; $bytes = ([System.IO.FileInfo]$target).Length; $digest = Get-Sha256 $target; [Console]::Out.WriteLine(([string]$bytes) + ' ' + $digest); break }", " 'write-b64-argv' { Decode-ToTarget $target ([string]::Concat($argvChunks)) ([Int64]$arg1) $arg2; break }", " 'write-b64-stdin' { Decode-ToTarget $target ([Console]::In.ReadToEnd()) ([Int64]$arg1) $arg2; break }", " 'write-b64-begin' { Ensure-Parent $target; Set-TmpPaths $target $arg1; $stream = [System.IO.File]::Open($script:tmp, [System.IO.FileMode]::Create, [System.IO.FileAccess]::Write, [System.IO.FileShare]::ReadWrite); try { $stream.SetLength([Int64]$arg2) } finally { $stream.Dispose() }; break }", " 'write-b64-block-stdin' { Set-TmpPaths $target $arg1; try { $bytes = [Convert]::FromBase64String((([Console]::In.ReadToEnd()) -replace '\\s','')) } catch { Fail ('transfer base64 block decode failed for ' + $target + ' at ' + $arg2 + ': ' + $_.Exception.Message) 22 }; $sha = [System.Security.Cryptography.SHA256]::Create(); try { $blockSha256 = ([System.BitConverter]::ToString($sha.ComputeHash($bytes))).Replace('-', '').ToLowerInvariant() } finally { $sha.Dispose() }; if ($bytes.Length -ne [Int64]$arg3 -or $blockSha256 -ne $arg4) { Fail ('transfer block verification failed for ' + $target + ' at ' + $arg2) 24 }; $stream = [System.IO.File]::Open($script:tmp, [System.IO.FileMode]::Open, [System.IO.FileAccess]::Write, [System.IO.FileShare]::ReadWrite); try { [void]$stream.Seek([Int64]$arg2, [System.IO.SeekOrigin]::Begin); $stream.Write($bytes, 0, $bytes.Length); $stream.Flush() } finally { $stream.Dispose() }; break }", " 'write-b64-abort' { Set-TmpPaths $target $arg1; Remove-Item -LiteralPath $script:tmp -Force -ErrorAction SilentlyContinue; break }", " 'write-b64-commit' { Set-TmpPaths $target $arg1; Verify-Temp $target $script:tmp ([Int64]$arg2) $arg3; Move-Item -LiteralPath $script:tmp -Destination $target -Force; $actualSha256 = Get-Sha256 $target; if ($actualSha256 -ne $arg3) { Fail ('transfer final sha256 mismatch for ' + $target) 25 }; break }", " default { Fail ('unsupported transfer op: ' + $operation) 2 }", "}", ].join(" "); } function powerShellSingleQuote(value: string): string { return `'${value.replace(/'/g, "''")}'`; }