|
|
@@ -1,9 +1,8 @@
|
|
|
/** Private POSIX SSH helper; filesystem and process effects use the installed local providers. */
|
|
|
-import { createHash } from 'node:crypto'
|
|
|
+import { createHash, randomUUID } from 'node:crypto'
|
|
|
import { readFileSync } from 'node:fs'
|
|
|
import { mkdtemp, rm } from 'node:fs/promises'
|
|
|
-import { fileURLToPath } from 'node:url'
|
|
|
-import { spawnSync } from 'node:child_process'
|
|
|
+import type { Readable, Writable } from 'node:stream'
|
|
|
import { Context } from '@deepseek-ai/cordis'
|
|
|
import { FsError, type FsTarget, type FsEditRequest, type FsWriteIntent, type FsVersion } from '@deepseek-ai/dsh-fs'
|
|
|
import { SandboxedFileSystem } from '@deepseek-ai/dsh-fs-sandbox'
|
|
|
@@ -13,7 +12,7 @@ import { SandboxPolicyService } from '@deepseek-ai/dsh-sandbox-policy'
|
|
|
import { SessionProjectionRegistry } from '@deepseek-ai/dsh-session-projection'
|
|
|
import type { SandboxExecutionPolicy, SandboxPolicy } from '@deepseek-ai/dsh-sandbox'
|
|
|
import { z } from 'zod'
|
|
|
-import { SshRpcPeer, SSH_PROTOCOL_VERSION } from './protocol.ts'
|
|
|
+import { SshRpcPeer, SSH_MAX_PROCESS_HANDLES, SSH_MAX_TEXT_STREAMS, SSH_PROTOCOL_VERSION } from './protocol.ts'
|
|
|
import { RemoteProcesses } from './helper-processes.ts'
|
|
|
import { editSchema, environmentSchema, intentSchema, policySchema, remotePath, targetSchema } from './schemas.ts'
|
|
|
|
|
|
@@ -32,52 +31,75 @@ async function services() {
|
|
|
return { ctx, close: async () => { for (const fiber of fibers.reverse()) await fiber.dispose() } }
|
|
|
}
|
|
|
|
|
|
-/** Run the private helper until the SSH channel closes or its client lease expires. */
|
|
|
-export async function runSshHelper(): Promise<void> {
|
|
|
+/** The private helper's authenticated transport and process lifetime. */
|
|
|
+export interface HelperTransport {
|
|
|
+ /** Authenticated request bytes from OpenSSH. */
|
|
|
+ input: Readable
|
|
|
+ /** Responses carried only by OpenSSH exec stdout. */
|
|
|
+ output: Writable
|
|
|
+ /** Installed entry whose digest is compared by the client. */
|
|
|
+ entryPath: string
|
|
|
+ /** Process termination joins the same cleanup as transport loss. */
|
|
|
+ signal: AbortSignal
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * Run a helper until its channel closes or its client lease expires.
|
|
|
+ * @param transport - private process streams, entry identity, and cancellation.
|
|
|
+ */
|
|
|
+export async function runSshHelper(transport: HelperTransport): Promise<void> {
|
|
|
if (process.platform !== 'linux' && process.platform !== 'darwin') throw new Error('SSH helper requires a POSIX host')
|
|
|
+ transport.signal.throwIfAborted()
|
|
|
const runtime = await services()
|
|
|
const { ctx } = runtime
|
|
|
const root = await mkdtemp('/tmp/dsh-ssh-')
|
|
|
- const processes = new RemoteProcesses(ctx, root, 128, 30_000)
|
|
|
- const iterators = new Map<string, AsyncIterator<string>>()
|
|
|
+ const processes = new RemoteProcesses(ctx, root, SSH_MAX_PROCESS_HANDLES, 30_000)
|
|
|
+ const lifetime = new AbortController()
|
|
|
+ const iterators = new Map<string, { iterator: AsyncIterator<string>; controller: AbortController }>()
|
|
|
let lease: NodeJS.Timeout | undefined
|
|
|
let leaseMs = 30_000
|
|
|
let initialized = false
|
|
|
+ let workspace = process.cwd()
|
|
|
let cleanup: Promise<void> | undefined
|
|
|
const close = (): Promise<void> => {
|
|
|
cleanup ??= (async () => {
|
|
|
if (lease !== undefined) clearTimeout(lease)
|
|
|
- await Promise.allSettled([...iterators.values()].map(iterator => iterator.return?.()))
|
|
|
- await processes.close()
|
|
|
- await runtime.close()
|
|
|
- await rm(root, { recursive: true, force: true })
|
|
|
+ lifetime.abort(new Error('SSH helper is closing'))
|
|
|
+ for (const record of iterators.values()) record.controller.abort(lifetime.signal.reason)
|
|
|
+ await Promise.allSettled([...iterators.values()].map(record => record.iterator.return?.()))
|
|
|
+ iterators.clear()
|
|
|
+ try { await processes.close() }
|
|
|
+ finally {
|
|
|
+ try { await runtime.close() }
|
|
|
+ finally { await rm(root, { recursive: true, force: true }) }
|
|
|
+ }
|
|
|
})()
|
|
|
return cleanup
|
|
|
}
|
|
|
const touchLease = (): void => {
|
|
|
if (lease !== undefined) clearTimeout(lease)
|
|
|
- lease = setTimeout(() => { peer.close(new Error('SSH helper client lease expired')); void close().catch(() => { process.exitCode = 1 }) }, leaseMs)
|
|
|
+ lease = setTimeout(() => { peer.close(new Error('SSH helper client lease expired')) }, leaseMs)
|
|
|
}
|
|
|
- const policy = async (raw: unknown): Promise<SandboxExecutionPolicy> => {
|
|
|
+ const policy = async (raw: unknown, signal: AbortSignal): Promise<SandboxExecutionPolicy> => {
|
|
|
const parsed = policySchema.parse(raw)
|
|
|
- const target = await ctx.fs.resolve(parsed.workspaceRoot)
|
|
|
+ const target = await ctx.fs.resolve(parsed.workspaceRoot, { signal })
|
|
|
return { ...parsed, workspaceRoot: ctx.fs.processPath(target) } as SandboxExecutionPolicy
|
|
|
}
|
|
|
const asTarget = (raw: unknown): FsTarget => targetSchema.parse(raw) as FsTarget
|
|
|
- const peer = new SshRpcPeer(process.stdin, process.stdout, MAX_FRAME_BYTES, 128, async (method, raw, signal) => {
|
|
|
+ const peer = new SshRpcPeer(transport.input, transport.output, MAX_FRAME_BYTES, 128, async (method, raw, requestSignal) => {
|
|
|
+ const signal = AbortSignal.any([requestSignal, lifetime.signal])
|
|
|
if (method === 'hello') {
|
|
|
if (initialized) throw new Error('SSH helper handshake already completed')
|
|
|
const input = z.object({
|
|
|
protocol: z.literal(SSH_PROTOCOL_VERSION), workspace: remotePath, leaseMs: z.number().int().min(3000).max(600_000),
|
|
|
bootstrapPath: remotePath.optional(),
|
|
|
}).strict().parse(raw)
|
|
|
- const workspace = ctx.fs.processPath(await ctx.fs.resolve(input.workspace, { signal }))
|
|
|
- process.chdir(workspace)
|
|
|
+ workspace = ctx.fs.processPath(await ctx.fs.resolve(input.workspace, { signal }))
|
|
|
leaseMs = input.leaseMs
|
|
|
initialized = true
|
|
|
touchLease()
|
|
|
return {
|
|
|
- protocol: SSH_PROTOCOL_VERSION, hash: createHash('sha256').update(readFileSync(fileURLToPath(import.meta.url))).digest('hex'),
|
|
|
+ protocol: SSH_PROTOCOL_VERSION, hash: createHash('sha256').update(readFileSync(transport.entryPath)).digest('hex'),
|
|
|
platform: process.platform, nodeVersion: process.version, node: process.execPath, root, workspace,
|
|
|
...(input.bootstrapPath === undefined ? {} : { bootstrapHash: createHash('sha256').update(readFileSync(input.bootstrapPath)).digest('hex') }),
|
|
|
}
|
|
|
@@ -103,26 +125,36 @@ export async function runSshHelper(): Promise<void> {
|
|
|
}
|
|
|
if (method === 'sandbox') {
|
|
|
const input = z.object({ argv: z.array(z.string()).min(1), policy: policySchema }).strict().parse(raw)
|
|
|
- const resolved = await policy(input.policy)
|
|
|
+ const resolved = await policy(input.policy, signal)
|
|
|
if (resolved.mode === 'danger-full-access') throw new Error('Unconfined argv does not need a sandbox wrapper')
|
|
|
- return ctx.sandbox.confine(input.argv, resolved as SandboxPolicy)
|
|
|
+ return ctx.sandbox.confine(input.argv, resolved as SandboxPolicy, signal)
|
|
|
}
|
|
|
if (method === 'fs.resolve' || method === 'fs.lstat') {
|
|
|
const input = z.object({ path: z.string(), cwd: remotePath.optional() }).strict().parse(raw)
|
|
|
- return method === 'fs.resolve' ? ctx.fs.resolve(input.path, { cwd: input.cwd ?? process.cwd(), signal }) : await ctx.fs.lstat(input.path, { cwd: input.cwd ?? process.cwd() }, signal) ?? null
|
|
|
+ return method === 'fs.resolve' ? ctx.fs.resolve(input.path, { cwd: input.cwd ?? workspace, signal }) : await ctx.fs.lstat(input.path, { cwd: input.cwd ?? workspace }, signal) ?? null
|
|
|
}
|
|
|
if (method === 'fs.stat' || method === 'fs.list' || method === 'fs.readText' || method === 'fs.stream') {
|
|
|
const target = asTarget(z.object({ target: targetSchema }).strict().parse(raw).target)
|
|
|
if (method === 'fs.stat') return await ctx.fs.stat(target, signal) ?? null
|
|
|
if (method === 'fs.list') return ctx.fs.listDir(target, signal)
|
|
|
- const stream = await ctx.fs.streamText(target, signal)
|
|
|
if (method === 'fs.stream') {
|
|
|
- const { randomUUID } = await import('node:crypto')
|
|
|
- const id = randomUUID()
|
|
|
- if (iterators.size >= 128) throw new Error('SSH text stream limit reached')
|
|
|
- iterators.set(id, stream[Symbol.asyncIterator]())
|
|
|
- return id
|
|
|
+ if (iterators.size >= SSH_MAX_TEXT_STREAMS) throw new Error('SSH text stream limit reached')
|
|
|
+ const controller = new AbortController()
|
|
|
+ const stream = await ctx.fs.streamText(target, AbortSignal.any([signal, controller.signal]))
|
|
|
+ const iterator = stream[Symbol.asyncIterator]()
|
|
|
+ try {
|
|
|
+ signal.throwIfAborted()
|
|
|
+ if (iterators.size >= SSH_MAX_TEXT_STREAMS) throw new Error('SSH text stream limit reached')
|
|
|
+ const id = randomUUID()
|
|
|
+ iterators.set(id, { iterator, controller })
|
|
|
+ return id
|
|
|
+ } catch (error) {
|
|
|
+ controller.abort(error)
|
|
|
+ await iterator.return?.()
|
|
|
+ throw error
|
|
|
+ }
|
|
|
}
|
|
|
+ const stream = await ctx.fs.streamText(target, signal)
|
|
|
let text = ''
|
|
|
let bytes = 0
|
|
|
for await (const chunk of stream) {
|
|
|
@@ -134,12 +166,27 @@ export async function runSshHelper(): Promise<void> {
|
|
|
}
|
|
|
if (method === 'fs.next' || method === 'fs.streamClose') {
|
|
|
const { id } = idRequest.parse(raw)
|
|
|
- const iterator = iterators.get(id)
|
|
|
- if (iterator === undefined) throw new Error('Unknown SSH text stream')
|
|
|
- if (method === 'fs.streamClose') { iterators.delete(id); await iterator.return?.(); return null }
|
|
|
- const next = await iterator.next()
|
|
|
- if (next.done) iterators.delete(id)
|
|
|
- return { done: next.done ?? false, value: next.value ?? '' }
|
|
|
+ const record = iterators.get(id)
|
|
|
+ if (record === undefined) throw new Error('Unknown SSH text stream')
|
|
|
+ if (method === 'fs.streamClose') {
|
|
|
+ iterators.delete(id)
|
|
|
+ record.controller.abort(new Error('SSH text stream closed'))
|
|
|
+ await record.iterator.return?.()
|
|
|
+ return null
|
|
|
+ }
|
|
|
+ const abort = (): void => { record.controller.abort(signal.reason) }
|
|
|
+ signal.addEventListener('abort', abort, { once: true })
|
|
|
+ try {
|
|
|
+ signal.throwIfAborted()
|
|
|
+ const next = await record.iterator.next()
|
|
|
+ if (next.done) iterators.delete(id)
|
|
|
+ return { done: next.done ?? false, value: next.value ?? '' }
|
|
|
+ } catch (error) {
|
|
|
+ iterators.delete(id)
|
|
|
+ record.controller.abort(error)
|
|
|
+ await record.iterator.return?.()
|
|
|
+ throw error
|
|
|
+ } finally { signal.removeEventListener('abort', abort) }
|
|
|
}
|
|
|
if (method === 'fs.readBytes' || method === 'fs.readRange') {
|
|
|
const input = z.object({
|
|
|
@@ -161,7 +208,7 @@ export async function runSshHelper(): Promise<void> {
|
|
|
expected: z.union([intentSchema, z.object({ version: z.string() }).strict()]).optional(), policy: policySchema,
|
|
|
}).strict().parse(raw)
|
|
|
const target = asTarget(input.target)
|
|
|
- const resolved = await policy(input.policy)
|
|
|
+ const resolved = await policy(input.policy, signal)
|
|
|
if (method === 'fs.write') return ctx.fs.writeText(
|
|
|
target, z.string().parse(input.content),
|
|
|
input.expected === undefined ? undefined : intentSchema.parse(input.expected) as FsWriteIntent, signal, resolved,
|
|
|
@@ -175,49 +222,16 @@ export async function runSshHelper(): Promise<void> {
|
|
|
}
|
|
|
throw new Error(`Unknown SSH helper operation: ${method}`)
|
|
|
})
|
|
|
- peer.once('closed', () => { void close().catch(() => { process.exitCode = 1 }) })
|
|
|
- const onSignal = (): void => { peer.close(new Error('SSH helper transport was terminated')) }
|
|
|
- process.once('SIGTERM', onSignal)
|
|
|
- process.once('SIGHUP', onSignal)
|
|
|
- process.once('SIGINT', onSignal)
|
|
|
- touchLease()
|
|
|
- try {
|
|
|
- await new Promise<void>((resolve) => { peer.once('closed', () => { resolve() }) })
|
|
|
- await close()
|
|
|
- } finally {
|
|
|
- process.off('SIGTERM', onSignal)
|
|
|
- process.off('SIGHUP', onSignal)
|
|
|
- process.off('SIGINT', onSignal)
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-async function confine(): Promise<void> {
|
|
|
- const runtime = await services()
|
|
|
- try {
|
|
|
- const request = z.object({
|
|
|
- policy: policySchema, runner: z.string(), enforcement: z.enum(['full', 'partial']), denialSignatures: z.array(z.string()),
|
|
|
- }).strict().parse(JSON.parse(process.argv[3] ?? ''))
|
|
|
- const { policy } = request
|
|
|
- if (policy.mode === 'danger-full-access' || process.argv[4] !== '--') throw new Error('Invalid remote sandbox invocation')
|
|
|
- const workspaceRoot = runtime.ctx.fs.processPath(await runtime.ctx.fs.resolve(policy.workspaceRoot))
|
|
|
- const wrapped = await runtime.ctx.sandbox.confine(process.argv.slice(5), { ...policy, workspaceRoot } as SandboxPolicy)
|
|
|
- if (wrapped.argv[0] !== request.runner || wrapped.enforcement !== request.enforcement
|
|
|
- || JSON.stringify(wrapped.denialSignatures) !== JSON.stringify(request.denialSignatures)) {
|
|
|
- throw new Error('Remote sandbox backend changed; this connection cannot attest the requested confinement')
|
|
|
- }
|
|
|
- const control = process.env.DSH_SUBPROCESS_CONTROL === 'pipe'
|
|
|
- const result = spawnSync(wrapped.argv[0] as string, wrapped.argv.slice(1), {
|
|
|
- stdio: control ? ['inherit', 'inherit', 'inherit', 'ignore', 'ignore', 'ignore', 'ignore', 7] : 'inherit',
|
|
|
- env: process.env,
|
|
|
- })
|
|
|
- if (result.error !== undefined) throw result.error
|
|
|
- process.exitCode = result.status ?? 128
|
|
|
- } finally { await runtime.close() }
|
|
|
-}
|
|
|
-
|
|
|
-if (import.meta.main) {
|
|
|
- void (process.argv[2] === '--confine' ? confine() : runSshHelper()).catch((error) => {
|
|
|
- process.stderr.write(`dsh-ssh-sandbox: ${error instanceof Error ? error.message : String(error)}\n`)
|
|
|
- process.exitCode = 127
|
|
|
+ const closed = Promise.withResolvers<undefined>()
|
|
|
+ peer.once('closed', () => {
|
|
|
+ void close().catch(() => {})
|
|
|
+ closed.resolve(undefined)
|
|
|
})
|
|
|
+ const onAbort = (): void => { peer.close(new Error('SSH helper transport was terminated')) }
|
|
|
+ transport.signal.addEventListener('abort', onAbort, { once: true })
|
|
|
+ if (transport.signal.aborted) onAbort()
|
|
|
+ else touchLease()
|
|
|
+ try { await closed.promise; await close() }
|
|
|
+ finally { transport.signal.removeEventListener('abort', onAbort) }
|
|
|
+
|
|
|
}
|