diff --git a/src/commands/device.ts b/src/commands/device.ts index 282e6f1..7a1adbb 100644 --- a/src/commands/device.ts +++ b/src/commands/device.ts @@ -1,5 +1,6 @@ import { defineCommand } from 'citty'; import { existsSync, readFileSync, statSync, writeFileSync } from 'node:fs'; +import * as path from 'node:path'; import { resolve as resolvePath } from 'node:path'; import { apiFlags } from '../config/flags/api.flags.js'; @@ -19,6 +20,7 @@ import { import { platformOfAppFile, uploadLocalApp } from '../utils/app-upload.js'; import { refreshAuth, resolveAuth } from '../utils/auth.js'; import { openBrowser } from '../utils/browser.js'; +import { uploadFlowBundle } from '../utils/device-bundle.js'; import { type DeviceArtifactFile, downloadArtifactFiles, @@ -650,6 +652,10 @@ const screenshotSub = defineCommand({ }, }); +function isDirectory(p: string): boolean { + return existsSync(p) && statSync(p).isDirectory(); +} + /** `--env KEY=VALUE`, given once or several times. */ export function parseEnvFlags(value: unknown): Record | undefined { if (value === undefined || value === false) return undefined; @@ -680,19 +686,23 @@ async function waitForCommand(ctx: Ctx, id: string, commandId: string): Promise< } const runSub = defineCommand({ - meta: { name: 'run', description: 'Run a Maestro flow on the device, whole, and wait for the result' }, + meta: { name: 'run', description: 'Run a Maestro flow (or a directory of flows) on the device, and wait for the result' }, args: { ...commonArgs, id: idArg, flow: { type: 'positional', required: false, - description: 'A flow file (.yaml)', + description: 'A flow file (.yaml), or a directory of flows (uploaded as a bundle)', }, yaml: { type: 'string', description: 'The flow inline, instead of a file', }, + entry: { + type: 'string', + description: 'With a directory: the one flow to run, as a path inside it (default: all of them)', + }, env: { type: 'string', description: 'A variable for the flow, KEY=VALUE, used as ${KEY}. Repeatable. Sent sealed, never stored in plain text', @@ -703,20 +713,21 @@ const runSub = defineCommand({ await runDeviceCommand(json, async () => { let id = args.id as string | undefined; let flow = args.flow as string | undefined; - if (id && !flow && /\.ya?ml$/i.test(id)) [id, flow] = [undefined, id]; + if (id && !flow && (/\.ya?ml$/i.test(id) || isDirectory(id))) [id, flow] = [undefined, id]; const inline = recoverFlagValue('--yaml', args.yaml as string | undefined, rawArgs); if (Boolean(flow) === Boolean(inline)) { - throw new CliError('Give a flow file or --yaml, one of the two.'); + throw new CliError('Give a flow file, a directory of flows, or --yaml: one of the three.'); } - let yaml = inline; - if (flow) { - if (existsSync(flow) && statSync(flow).isDirectory()) { - throw new CliError( - 'Running a directory of flows is not supported yet. Run one flow file at a time.', - ); - } + const dir = flow && isDirectory(flow) ? flow : undefined; + const entry = args.entry as string | undefined; + if (entry && !dir) throw new CliError('--entry picks a flow inside a directory; give a directory.'); + if (entry && (path.isAbsolute(entry) || entry.split(/[\\/]/).includes('..'))) { + throw new CliError('--entry is a path inside the directory, e.g. flows/login.yaml.'); + } + let yaml = inline; + if (flow && !dir) { try { yaml = readFileSync(flow, 'utf8'); } catch (error) { @@ -728,11 +739,23 @@ const runSub = defineCommand({ const env = parseEnvFlags(repeated.length > 0 ? repeated : args.env); const ctx = await context(args); id = await resolveSessionId(ctx.apiUrl, ctx.auth, id); - ctx.progress(ui.running(`Running ${flow ?? 'the flow'} on ${colors.highlight(id)}…`)); - let response = await DeviceGateway.run(ctx.apiUrl, ctx.auth, id, { - yaml: yaml!, - ...(env && { env }), - }); + + let body: Parameters[3]; + if (dir) { + ctx.progress(ui.running(`Uploading the flows in ${colors.highlight(dir)}…`)); + const bundle = await uploadFlowBundle({ apiUrl: ctx.apiUrl, auth: ctx.auth, dir, log: ctx.progress }); + ctx.progress(ui.success(`Uploaded ${bundle.files} files`)); + body = { + flowUploadId: bundle.id, + ...(entry && { flow: entry.split(path.sep).join('/') }), + ...(env && { env }), + }; + } else { + body = { yaml: yaml!, ...(env && { env }) }; + } + + ctx.progress(ui.running(`Running ${entry ?? flow ?? 'the flow'} on ${colors.highlight(id)}…`)); + let response = await DeviceGateway.run(ctx.apiUrl, ctx.auth, id, body); if (response.status === 'running') response = await waitForCommand(ctx, id, response.command_id); const result = response.result ?? { success: false }; diff --git a/src/gateways/device-gateway.ts b/src/gateways/device-gateway.ts index 1b24cea..276db8a 100644 --- a/src/gateways/device-gateway.ts +++ b/src/gateways/device-gateway.ts @@ -1,6 +1,8 @@ import type { AuthContext } from '../types/domain/auth.types.js'; import type { DeviceArtifactsPage, + DeviceBundleUploadUrls, + DeviceFlowBundle, DeviceCatalogue, DeviceCommandResult, DeviceInstalledApp, @@ -120,6 +122,14 @@ export const DeviceGateway = { return request(baseUrl, auth, 'GET', `${sessionPath(id)}/artifacts?${query}`); }, + bundleUploadUrls( + baseUrl: string, + auth: AuthContext, + bytes: number, + ): Promise { + return request(baseUrl, auth, 'POST', '/device/bundles/upload-url', { bytes }); + }, + command( baseUrl: string, auth: AuthContext, @@ -181,11 +191,28 @@ export const DeviceGateway = { return request(baseUrl, auth, 'GET', '/device/models'); }, + registerBundle( + baseUrl: string, + auth: AuthContext, + body: { + backblaze: boolean; + bytes: number; + enc?: unknown; + id: string; + sha: string; + supabase: boolean; + }, + ): Promise<{ bundle: DeviceFlowBundle }> { + return request(baseUrl, auth, 'POST', '/device/bundles', body); + }, + run( baseUrl: string, auth: AuthContext, id: string, - body: { env?: Record; yaml: string }, + body: + | { env?: Record; flow?: string; flowUploadId: string } + | { env?: Record; yaml: string }, ): Promise { return request(baseUrl, auth, 'POST', `${sessionPath(id)}/run`, body); }, diff --git a/src/types/domain/device-session.types.ts b/src/types/domain/device-session.types.ts index 2ac4d7f..5fd2db9 100644 --- a/src/types/domain/device-session.types.ts +++ b/src/types/domain/device-session.types.ts @@ -152,3 +152,19 @@ export interface DeviceArtifactsPage { recording: DeviceArtifactLink | null; session: DeviceSession; } + +/** POST /device/bundles/upload-url: where to PUT the zip (both, when both are given). */ +export interface DeviceBundleUploadUrls { + id: string; + mirror: { path: string; signedUrl: string } | null; + upload: { path: string; signedUrl: string } | null; +} + +/** A registered flow bundle. `run` takes its id as `flowUploadId`. */ +export interface DeviceFlowBundle { + bytes: number; + created_at: string; + encrypted: boolean; + id: string; + sha: string; +} diff --git a/src/utils/device-bundle.ts b/src/utils/device-bundle.ts new file mode 100644 index 0000000..b6dd26d --- /dev/null +++ b/src/utils/device-bundle.ts @@ -0,0 +1,129 @@ +import { createHash } from 'node:crypto'; +import { readdirSync } from 'node:fs'; +import * as path from 'node:path'; + +import { DeviceGateway } from '../gateways/device-gateway.js'; +import { compressFilesFromRelativePath } from '../methods.js'; +import type { AuthContext } from '../types/domain/auth.types.js'; +import type { DeviceFlowBundle } from '../types/domain/device-session.types.js'; +import { detectCiContext } from './ci.js'; +import { CliError, getCliVersion } from './cli.js'; +import { fetchCompatibilityData } from './compatibility.js'; +import { + checkKekForEncryption, + encryptFlowBuffer, + resolveEncryption, + resolveKekPublicKey, +} from './envelope.js'; + +/** The API's limit on a bundle (dcd plan 2.5b). */ +export const MAX_BUNDLE_BYTES = 50 * 1024 * 1024; + +/** Never bundled: version control, dependencies, and other dot-folders. */ +function skipped(relativePath: string): boolean { + return relativePath + .split(path.sep) + .some((part) => part === 'node_modules' || (part.startsWith('.') && part !== '.')); +} + +/** Every file under `dir`, relative to it, in a stable order. */ +export function bundleFiles(dir: string): string[] { + return readdirSync(dir, { recursive: true, withFileTypes: true }) + .filter((entry) => entry.isFile()) + .map((entry) => path.relative(dir, path.join(entry.parentPath, entry.name))) + .filter((file) => !skipped(file)) + .sort(); +} + +/** PUT the zip to one signed URL. The URL is the credential. */ +async function put(url: string, body: Buffer): Promise { + for (let attempt = 1; attempt <= 2; attempt++) { + try { + const res = await fetch(url, { + method: 'PUT', + headers: { 'content-type': 'application/zip' }, + body: new Uint8Array(body), + }); + await res.body?.cancel(); + if (res.ok) return true; + if (res.status >= 400 && res.status < 500 && res.status !== 408 && res.status !== 429) { + return false; + } + } catch { + // A network error: try once more. + } + } + + return false; +} + +export interface UploadFlowBundleOptions { + apiUrl: string; + auth: AuthContext; + dir: string; + log: (message: string) => void; +} + +/** + * Zip a directory of flows and upload it as a flow bundle: ask the API for + * upload URLs, PUT the zip to both copies, then register it with its + * SHA-256. Encrypted like `dcd cloud`'s flow zips when the org's default + * (or DCD_ENCRYPT) says so. Entries are relative to `dir`, so a run's + * `flow` is a path inside it, e.g. `login.yaml`. + */ +export async function uploadFlowBundle( + opts: UploadFlowBundleOptions, +): Promise { + const { apiUrl, auth, dir, log } = opts; + const files = bundleFiles(dir); + if (!files.some((f) => /\.ya?ml$/i.test(f))) { + throw new CliError(`No flow files (.yaml) in ${dir}.`); + } + + const zip = await compressFilesFromRelativePath(dir, files, ''); + + let serverDefault: boolean | undefined; + try { + const ci = detectCiContext(); + const compatibility = await fetchCompatibilityData(apiUrl, auth, { + cliVersion: getCliVersion(), + ciProvider: ci.provider, + ciWrapperVersion: ci.wrapperVersion, + }); + serverDefault = compatibility.encryption?.defaultOn; + } catch { + serverDefault = undefined; + } + + const encryption = checkKekForEncryption(resolveEncryption({ serverDefault }), apiUrl); + if (encryption.notice) log(encryption.notice); + const kek = encryption.enabled ? resolveKekPublicKey(apiUrl) : null; + const sealed = kek ? encryptFlowBuffer(zip, kek) : null; + const body = sealed ? sealed.ciphertext : zip; + + if (body.length > MAX_BUNDLE_BYTES) { + throw new CliError( + `${dir} zips to ${(body.length / 1024 / 1024).toFixed(1)} MB; a bundle can be at most ${MAX_BUNDLE_BYTES / 1024 / 1024} MB.`, + ); + } + + const sha = createHash('sha256').update(body).digest('hex'); + const urls = await DeviceGateway.bundleUploadUrls(apiUrl, auth, body.length); + const [supabase, backblaze] = await Promise.all([ + urls.upload ? put(urls.upload.signedUrl, body) : Promise.resolve(false), + urls.mirror ? put(urls.mirror.signedUrl, body) : Promise.resolve(false), + ]); + if (!supabase && !backblaze) { + throw new Error('The flow bundle could not be uploaded. Please retry.'); + } + + const { bundle } = await DeviceGateway.registerBundle(apiUrl, auth, { + id: urls.id, + sha, + bytes: body.length, + supabase, + backblaze, + ...(sealed && { enc: sealed.enc }), + }); + return { ...bundle, files: files.length }; +} diff --git a/test/unit/device-bundle.test.ts b/test/unit/device-bundle.test.ts new file mode 100644 index 0000000..f15bb8d --- /dev/null +++ b/test/unit/device-bundle.test.ts @@ -0,0 +1,143 @@ +import { expect } from 'chai'; +import { createHash, generateKeyPairSync } from 'node:crypto'; +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import StreamZip from 'node-stream-zip'; + +import { clearCompatibilityCache } from '../../src/utils/compatibility.js'; +import { bundleFiles, uploadFlowBundle } from '../../src/utils/device-bundle.js'; + +const API = 'https://api.devicecloud.dev'; +const auth = { headers: {}, mode: 'apiKey' as const }; + +interface Call { + body?: Uint8Array | string; + method: string; + url: string; +} + +describe('dcd device run : flow bundles', () => { + const realFetch = globalThis.fetch; + const ORIGINAL_ENV = { ...process.env }; + let dir: string; + let calls: Call[]; + let encryptionDefault: boolean; + + beforeEach(() => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'dcd-bundle-test-')); + fs.mkdirSync(path.join(dir, 'flows')); + fs.mkdirSync(path.join(dir, '.git')); + fs.mkdirSync(path.join(dir, 'node_modules')); + fs.writeFileSync(path.join(dir, 'flows', 'login.yaml'), 'appId: x\n---\n- launchApp\n'); + fs.writeFileSync(path.join(dir, 'config.yaml'), 'flows: [flows/*]\n'); + fs.writeFileSync(path.join(dir, '.git', 'HEAD'), 'ref'); + fs.writeFileSync(path.join(dir, 'node_modules', 'x.js'), ''); + delete process.env.DCD_ENCRYPT; + delete process.env.DCD_ENCRYPT_BINARIES; + encryptionDefault = false; + clearCompatibilityCache(); + calls = []; + globalThis.fetch = (async (input: unknown, init?: RequestInit) => { + const url = String(input); + const call: Call = { + url, + method: init?.method ?? 'GET', + body: init?.body as Uint8Array | string | undefined, + }; + calls.push(call); + if (url.includes('/results/compatibility')) { + return new Response( + JSON.stringify({ statusCode: 200, data: { encryption: { defaultOn: encryptionDefault } } }), + ); + } + + if (url.endsWith('/device/bundles/upload-url')) { + return new Response( + JSON.stringify({ + id: 'b-1', + upload: { path: 'p', signedUrl: 'https://sb/put' }, + mirror: { path: 'organizations/p', signedUrl: 'https://b2/put' }, + }), + ); + } + + if (url === 'https://sb/put') return new Response(null, { status: 200 }); + if (url === 'https://b2/put') return new Response(null, { status: 500 }); + if (url.endsWith('/device/bundles')) { + const body = JSON.parse(String(init?.body)); + return new Response( + JSON.stringify({ bundle: { ...body, encrypted: Boolean(body.enc), created_at: 'now' } }), + ); + } + + return new Response('not found', { status: 404 }); + }) as typeof fetch; + }); + + afterEach(() => { + clearCompatibilityCache(); + globalThis.fetch = realFetch; + process.env = { ...ORIGINAL_ENV }; + fs.rmSync(dir, { force: true, recursive: true }); + }); + + it('bundles every file but dot-folders and node_modules', () => { + expect(bundleFiles(dir)).to.deep.equal(['config.yaml', path.join('flows', 'login.yaml')]); + }); + + it('uploads the zip to both copies and registers what landed', async () => { + const bundle = await uploadFlowBundle({ apiUrl: API, auth, dir, log: () => {} }); + expect(bundle.id).to.equal('b-1'); + expect(bundle.files).to.equal(2); + + const put = calls.find((c) => c.url === 'https://sb/put')!; + expect(put.method).to.equal('PUT'); + const zipBytes = Buffer.from(put.body as Uint8Array); + + const register = calls.find((c) => c.url.endsWith('/device/bundles'))!; + const sent = JSON.parse(String(register.body)); + expect(sent).to.deep.equal({ + id: 'b-1', + sha: createHash('sha256').update(zipBytes).digest('hex'), + bytes: zipBytes.length, + supabase: true, + // The B2 PUT failed, so only the Supabase copy is claimed. + backblaze: false, + }); + + // Entries are relative to the directory, so `--entry flows/login.yaml` matches. + const zipPath = path.join(dir, 'out.zip'); + fs.writeFileSync(zipPath, zipBytes); + const zip = new StreamZip.async({ file: zipPath }); + expect(Object.keys(await zip.entries()).sort()).to.deep.equal(['config.yaml', 'flows/login.yaml']); + await zip.close(); + }); + + it("encrypts the zip when the org's default says so", async () => { + const { publicKey } = generateKeyPairSync('x25519'); + const raw = publicKey.export({ type: 'spki', format: 'der' }).subarray(12); + process.env.DCD_BINARY_KEK_PUBLIC = `1:${raw.toString('base64')}`; + encryptionDefault = true; + + const bundle = await uploadFlowBundle({ apiUrl: API, auth, dir, log: () => {} }); + expect(bundle.encrypted).to.equal(true); + const put = calls.find((c) => c.url === 'https://sb/put')!; + expect(Buffer.from(put.body as Uint8Array).subarray(0, 4).toString()).to.equal('DCDE'); + const sent = JSON.parse(String(calls.find((c) => c.url.endsWith('/device/bundles'))!.body)); + expect(sent.enc).to.include.keys('v', 'kek', 'wrapped_key'); + }); + + it('refuses a directory with no flows', async () => { + const empty = fs.mkdtempSync(path.join(os.tmpdir(), 'dcd-bundle-empty-')); + fs.writeFileSync(path.join(empty, 'notes.txt'), ''); + try { + await uploadFlowBundle({ apiUrl: API, auth, dir: empty, log: () => {} }); + expect.fail('should have thrown'); + } catch (error) { + expect((error as Error).message).to.include('No flow files'); + } finally { + fs.rmSync(empty, { force: true, recursive: true }); + } + }); +}); diff --git a/test/unit/device-command.test.ts b/test/unit/device-command.test.ts index d0912c5..835eb25 100644 --- a/test/unit/device-command.test.ts +++ b/test/unit/device-command.test.ts @@ -40,6 +40,16 @@ function session(id: string, extra: Partial = {}): DeviceSession }; } +/** JSON bodies parsed; anything else (a zip being PUT) kept as is. */ +function parseBody(body: unknown): unknown { + if (typeof body !== 'string') return body; + try { + return JSON.parse(body); + } catch { + return body; + } +} + interface Call { body?: unknown; headers: Record; @@ -55,7 +65,7 @@ function stubFetch(handler: (call: Call) => { body: unknown; status?: number }): url: String(input), method: init?.method ?? 'GET', headers: (init?.headers ?? {}) as Record, - body: init?.body ? JSON.parse(String(init.body)) : undefined, + body: parseBody(init?.body), }; calls.push(call); const { body, status = 200 } = handler(call); @@ -368,6 +378,43 @@ describe('dcd device', () => { expect(out.data).to.deep.equal({ command_id: 'c1', status: 'passed', result: { success: true, output: 'ok' } }); }).timeout(10_000); + it('run uploads a bundle and runs it by id', async () => { + const flows = path.join(tempDir, 'flows'); + fs.mkdirSync(path.join(flows, 'sub'), { recursive: true }); + fs.writeFileSync(path.join(flows, 'sub', 'login.yaml'), 'appId: x\n---\n- launchApp\n'); + const calls = stubFetch((call) => { + if (call.url.endsWith('/upload-url')) { + return { body: { id: 'b-1', upload: { path: 'p', signedUrl: 'https://sb/put' }, mirror: null } }; + } + + if (call.url === 'https://sb/put') return { body: {} }; + if (call.url.endsWith('/device/bundles')) { + return { body: { bundle: { id: 'b-1', sha: 's', bytes: 1, encrypted: false, created_at: 'now' } } }; + } + + if (call.url.endsWith('/run')) { + return { body: { command_id: 'c1', status: 'passed', result: { success: true }, session: session('a-b-c') } }; + } + + return { status: 404, body: {} }; + }); + const { exit } = await runJson('run', { id: 'a-b-c', flow: flows, entry: path.join('sub', 'login.yaml') }); + expect(exit).to.equal(0); + const run = calls.find((c) => c.url.endsWith('/run'))!; + expect(run.body).to.deep.equal({ flowUploadId: 'b-1', flow: 'sub/login.yaml' }); + }); + + it('run --entry needs a directory and a path inside it', async () => { + const flows = path.join(tempDir, 'flows'); + fs.mkdirSync(flows); + expect((await runJson('run', { id: 'a-b-c', yaml: '- launchApp', entry: 'x.yaml' })).exit).to.equal( + DEVICE_EXIT.usage, + ); + expect((await runJson('run', { id: 'a-b-c', flow: flows, entry: '../x.yaml' })).exit).to.equal( + DEVICE_EXIT.usage, + ); + }); + it('run exits 2 when the flow fails', async () => { stubFetch(() => ({ body: { command_id: 'c1', status: 'failed', result: { success: false, error: 'Element not found' }, session: session('a-b-c') }, @@ -377,7 +424,7 @@ describe('dcd device', () => { expect(out.error).to.deep.equal({ code: null, message: 'Element not found' }); }); - it('run needs a file or --yaml', async () => { + it('run needs a file, a directory or --yaml', async () => { const { exit } = await runJson('run', { id: 'a-b-c' }); expect(exit).to.equal(DEVICE_EXIT.usage); });