From 5ce544750fe7910bba9f58e42c1a5811c69b2bb8 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Thu, 30 Jul 2026 04:52:37 -0700 Subject: [PATCH] fix(dispatch): close result-pull gaps left by the first pass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four loose ends in the result-return path shipped in #1893, found by re-reading that change rather than by a failure. **An older target failed with clap's own error.** `workspace_result_bundle` is advertised and deliberately kept out of `REQUIRED_DISPATCH_CAPABILITIES` so a CLI that predates it stays fully usable for running jobs — but nothing checked it, so pulling from such a target surfaced `unrecognized subcommand '__workspace_result'`, which says nothing about what to do. Probe for the capability first and fail with the actual remedy. Checked against the live protocol rather than anything cached, so it cannot go stale. **The staging directory did not follow its own neighbour's rule.** Result bundles land beside `.workspace-uploads`, which hardens itself to 0700 even though the outbound root is already owner-only. `.results` was created with a bare `create_dir_all` and its bundle and summary written under the process umask. The parent's 0700 does block traversal today, so this was defence in depth rather than an open door — but the bundle carries the user's source, including the ignored files the snapshot deliberately shipped, and the file beside it records which paths changed. Harden the directory and create both files 0600 before writing, so their contents are never briefly umask-governed. **Pulled bundles were never collected.** `remove_workspace_snapshot` clears only `.workspace-uploads`, so terminal jobs expired after 30 days while their bundles accumulated forever. Add `remove_result_bundle` and call it from the same retention sweep. Kept separate from the snapshot removal on purpose: that one runs as soon as the target durably owns the job, long before the user has had a chance to look at the results. **Account devices could receive a snapshot but never return one.** Device dispatch has a full chunked upload path, yet `pull_result` bailed with "requires an SSH target", leaving device-dispatched snapshot jobs with no way home. SSH pulls the bundle over SFTP; a device transport carries JSON only, so add a bounded read-side verb and stream the same bytes back in base64 chunks — the mirror of the upload. The verb never rebuilds the bundle, so the digest the target reported stays the digest the controller verifies, and the reassembled bytes are checked against it before anything is staged. Both transports record the same durable summary, so the apply step stays transport-blind. Verified through the real CLI: the chunked read reassembles a bundle across five 64-byte chunks with a matching digest, and rejects a zero length, an oversized length, and an offset past the end. Device streaming and its digest-mismatch refusal are covered by tests over a stub RPC. Note: `cargo clippy --all-targets` on services-integrations reports three pre-existing `octal_escapes` errors in manager.rs:6037, unrelated to this change and present on main; only `--all-targets` surfaces them. --- src/apps/cli/src/dispatch/mod.rs | 6 +- src/apps/cli/src/dispatch/protocol.rs | 22 ++ src/apps/cli/src/dispatch/workspace.rs | 36 ++- src/apps/cli/src/main.rs | 2 + src/apps/cli/src/peer_host/dispatch.rs | 2 + src/apps/cli/src/root_handlers.rs | 1 + src/apps/desktop/src/api/dispatch_api.rs | 17 +- src/apps/desktop/src/api/dispatch_host.rs | 2 + .../core/src/service/dispatch/controller.rs | 41 ++- .../src/service/dispatch/device_controller.rs | 250 +++++++++++++++++- .../assembly/core/src/service/dispatch/mod.rs | 68 ++++- .../services-core/src/dispatch_workspace.rs | 3 +- .../src/remote_ssh/dispatch_ssh.rs | 145 +++++++++- 13 files changed, 577 insertions(+), 18 deletions(-) diff --git a/src/apps/cli/src/dispatch/mod.rs b/src/apps/cli/src/dispatch/mod.rs index e98c66a3e9..31d6097de1 100644 --- a/src/apps/cli/src/dispatch/mod.rs +++ b/src/apps/cli/src/dispatch/mod.rs @@ -19,7 +19,7 @@ use protocol::{ DispatchListRequest, DispatchProbeRequest, DispatchProbeResponse, DispatchStatusRequest, DispatchStatusResponse, DispatchSubmitRequest, DispatchSubmitResponse, DispatchWorkspaceBeginRequest, DispatchWorkspaceChunkRequest, DispatchWorkspaceCommitRequest, - DispatchWorkspaceResultRequest, + DispatchWorkspaceResultChunkRequest, DispatchWorkspaceResultRequest, DispatchWorkspaceProbe, DISPATCH_PROTOCOL_VERSION, MAX_DISPATCH_TEXT_BYTES, }; use store::{CreateJobOutcome, DispatchStateRecord, DispatchStore}; @@ -76,6 +76,10 @@ pub(crate) async fn run_dispatch_verb( DispatchWorkspaceResultRequest, >(input)?)?) .context("encode workspace result response"), + "workspace-result-chunk" => serde_json::to_value(workspace::result_chunk(parse::< + DispatchWorkspaceResultChunkRequest, + >(input)?)?) + .context("encode workspace result chunk response"), _ => bail!("unsupported dispatch verb: {verb}"), } } diff --git a/src/apps/cli/src/dispatch/protocol.rs b/src/apps/cli/src/dispatch/protocol.rs index 4537f49714..a20007fbba 100644 --- a/src/apps/cli/src/dispatch/protocol.rs +++ b/src/apps/cli/src/dispatch/protocol.rs @@ -206,6 +206,28 @@ pub(crate) struct DispatchWorkspaceResultResponse { pub(crate) summary: WorkspaceResultSummary, } +/// Read a slice of an already-built result bundle. +/// +/// Exists for transports with no file channel of their own: SSH pulls the +/// bundle over SFTP, but an account device can only carry JSON, so it streams +/// the same bytes back in chunks — the mirror of the upload path. +#[derive(Clone, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct DispatchWorkspaceResultChunkRequest { + pub(crate) job_id: String, + pub(crate) offset: u64, + pub(crate) length: u64, +} + +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct DispatchWorkspaceResultChunkResponse { + pub(crate) offset: u64, + pub(crate) data_base64: String, + /// True once this chunk reaches the end of the bundle. + pub(crate) eof: bool, +} + #[derive(Clone, Debug, Deserialize, PartialEq, Eq)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub(crate) struct DispatchCancelRequest { diff --git a/src/apps/cli/src/dispatch/workspace.rs b/src/apps/cli/src/dispatch/workspace.rs index c42c0457c0..c234a61567 100644 --- a/src/apps/cli/src/dispatch/workspace.rs +++ b/src/apps/cli/src/dispatch/workspace.rs @@ -14,7 +14,8 @@ use serde::{Deserialize, Serialize}; use super::protocol::{ DispatchWorkspaceBeginRequest, DispatchWorkspaceBeginResponse, DispatchWorkspaceChunkRequest, DispatchWorkspaceChunkResponse, DispatchWorkspaceCommitRequest, - DispatchWorkspaceCommitResponse, DispatchWorkspaceResultRequest, + DispatchWorkspaceCommitResponse, DispatchWorkspaceResultChunkRequest, + DispatchWorkspaceResultChunkResponse, DispatchWorkspaceResultRequest, DispatchWorkspaceResultResponse, DISPATCH_PROTOCOL_VERSION, }; use super::store::{ @@ -341,6 +342,39 @@ pub(crate) fn result( }) } +/// Stream back a slice of the bundle `result` already produced. +/// +/// Read-only and bounded: it never rebuilds the bundle, so the digest the +/// controller verified stays the digest it receives. +pub(crate) fn result_chunk( + request: DispatchWorkspaceResultChunkRequest, +) -> Result { + if request.length == 0 || request.length > MAX_CHUNK_BYTES as u64 { + bail!("workspace result chunk length must be between 1 and {MAX_CHUNK_BYTES} bytes"); + } + let store = DispatchStore::open_default()?; + let upload_dir = store.workspace_upload_dir(&request.job_id)?; + let bundle_path = upload_dir.join(RESULT_BUNDLE_FILE); + let mut file = fs::File::open(&bundle_path) + .context("build the dispatch result bundle before reading it")?; + let size = file.metadata()?.len(); + if request.offset > size { + bail!("workspace result chunk offset is past the end of the bundle"); + } + file.seek(SeekFrom::Start(request.offset))?; + let remaining = size - request.offset; + let take = request.length.min(remaining) as usize; + let mut buffer = vec![0_u8; take]; + file.read_exact(&mut buffer) + .context("read dispatch result bundle")?; + let next_offset = request.offset + take as u64; + Ok(DispatchWorkspaceResultChunkResponse { + offset: next_offset, + data_base64: base64::engine::general_purpose::STANDARD.encode(&buffer), + eof: next_offset >= size, + }) +} + /// Detached target-side materialization. The short `workspace-commit` RPC /// starts this process and subsequent commit calls poll the durable record, so /// extraction is not bounded by an SSH or Relay request timeout. diff --git a/src/apps/cli/src/main.rs b/src/apps/cli/src/main.rs index f90d7b6a22..b34420a4d1 100644 --- a/src/apps/cli/src/main.rs +++ b/src/apps/cli/src/main.rs @@ -606,6 +606,8 @@ pub(crate) enum DispatchAction { WorkspaceCommit, #[command(name = "__workspace_result", hide = true)] WorkspaceResult, + #[command(name = "__workspace_result_chunk", hide = true)] + WorkspaceResultChunk, #[command(name = "__workspace_materialize", hide = true)] WorkspaceMaterialize { #[arg(long)] diff --git a/src/apps/cli/src/peer_host/dispatch.rs b/src/apps/cli/src/peer_host/dispatch.rs index 3866b43edc..c483f8abd4 100644 --- a/src/apps/cli/src/peer_host/dispatch.rs +++ b/src/apps/cli/src/peer_host/dispatch.rs @@ -131,6 +131,8 @@ fn dispatch_target_verb(command: &str) -> Option<&'static str> { "dispatch_target_workspace_begin" => Some("workspace-begin"), "dispatch_target_workspace_chunk" => Some("workspace-chunk"), "dispatch_target_workspace_commit" => Some("workspace-commit"), + "dispatch_target_workspace_result" => Some("workspace-result"), + "dispatch_target_workspace_result_chunk" => Some("workspace-result-chunk"), _ => None, } } diff --git a/src/apps/cli/src/root_handlers.rs b/src/apps/cli/src/root_handlers.rs index f161b6a1f3..5bced4b5ee 100644 --- a/src/apps/cli/src/root_handlers.rs +++ b/src/apps/cli/src/root_handlers.rs @@ -60,6 +60,7 @@ pub(crate) async fn handle_dispatch_action(action: DispatchAction) -> Result<()> DispatchAction::WorkspaceChunk => "workspace-chunk", DispatchAction::WorkspaceCommit => "workspace-commit", DispatchAction::WorkspaceResult => "workspace-result", + DispatchAction::WorkspaceResultChunk => "workspace-result-chunk", }; let result = async { use std::io::{IsTerminal, Read}; diff --git a/src/apps/desktop/src/api/dispatch_api.rs b/src/apps/desktop/src/api/dispatch_api.rs index 53ab8c09e6..0f4331122c 100644 --- a/src/apps/desktop/src/api/dispatch_api.rs +++ b/src/apps/desktop/src/api/dispatch_api.rs @@ -14,7 +14,8 @@ use bitfun_core::service::dispatch::{ cancel_device_dispatch, cancel_dispatch, cancel_dispatch_cli_install, get_device_dispatch_status, get_dispatch_status, list_device_dispatch_jobs, list_dispatch_jobs, list_dispatch_targets, poll_dispatch_cli_install, probe_device_dispatch_target, - probe_dispatch_target, pull_dispatch_result, start_dispatch_cli_install, + probe_dispatch_target, pull_device_dispatch_result, pull_dispatch_result, + start_dispatch_cli_install, start_dispatch_cli_source_build, submit_device_dispatch, submit_dispatch, sync_dispatch_model_config, DeviceDispatchRpc, DispatchAnswerRequest, DispatchAppendRequest, DispatchConnectionRequest, @@ -277,6 +278,20 @@ pub async fn dispatch_pull_result( request: DispatchJobRequest, ) -> Result { let store = OutboundDispatchStore::new(path_manager.as_ref()); + // Both transports stage the bundle and its summary identically, so the + // apply step below is transport-blind. + if matches!( + store + .get(&request.job_id) + .await + .map_err(|error| error.to_string())? + .map(|record| record.target), + Some(DispatchTarget::Device { .. }) + ) { + return pull_device_dispatch_result(&AccountDeviceDispatchRpc, &store, request) + .await + .map_err(|error| error.to_string()); + } let manager = state .get_ssh_manager_async() .await diff --git a/src/apps/desktop/src/api/dispatch_host.rs b/src/apps/desktop/src/api/dispatch_host.rs index 48484e7083..206e7850a4 100644 --- a/src/apps/desktop/src/api/dispatch_host.rs +++ b/src/apps/desktop/src/api/dispatch_host.rs @@ -42,6 +42,8 @@ fn target_cli_verb(command: &str) -> Option<&'static str> { "dispatch_target_workspace_begin" => Some("__workspace_begin"), "dispatch_target_workspace_chunk" => Some("__workspace_chunk"), "dispatch_target_workspace_commit" => Some("__workspace_commit"), + "dispatch_target_workspace_result" => Some("__workspace_result"), + "dispatch_target_workspace_result_chunk" => Some("__workspace_result_chunk"), _ => None, } } diff --git a/src/crates/assembly/core/src/service/dispatch/controller.rs b/src/crates/assembly/core/src/service/dispatch/controller.rs index 2d364fb821..768478f686 100644 --- a/src/crates/assembly/core/src/service/dispatch/controller.rs +++ b/src/crates/assembly/core/src/service/dispatch/controller.rs @@ -499,28 +499,49 @@ pub async fn pull_result( .await? .ok_or_else(|| anyhow::anyhow!("Outbound dispatch job was not found"))?; let DispatchTarget::Ssh { connection_id, .. } = &record.target else { - anyhow::bail!("Pulling dispatch results requires an SSH target"); + anyhow::bail!("SSH dispatch result pull requires an SSH target"); }; let destination = result_bundle_path(store, &request.job_id); let response = dispatch_ssh::pull_result(manager, connection_id, &request.job_id, &destination).await?; - // Persist the summary next to the bundle so applying reads both from disk. - // The digests that decide whether a local file may be overwritten must come - // from the verified pull, not from whatever the caller hands back later. + record_result_summary(store, &request.job_id, &response)?; + Ok(response) +} + +/// Persist the summary next to the bundle so applying reads both from disk. +/// +/// The digests that decide whether a local file may be overwritten must come +/// from the verified pull, not from whatever the caller hands back later. +pub(super) fn record_result_summary( + store: &OutboundDispatchStore, + job_id: &str, + response: &Value, +) -> anyhow::Result<()> { if let Some(summary) = response.get("summary") { - let summary_path = result_summary_path(store, &request.job_id); - std::fs::write(&summary_path, serde_json::to_vec(summary)?) + // Owner-only like the bundle beside it: this records which paths of the + // user's workspace changed. + let summary_path = result_summary_path(store, job_id); + dispatch_ssh::write_private_file(&summary_path, &serde_json::to_vec(summary)?) .with_context(|| format!("record result summary {}", summary_path.display()))?; } - Ok(response) + Ok(()) } -fn result_bundle_path(store: &OutboundDispatchStore, job_id: &str) -> std::path::PathBuf { - store.root().join(".results").join(format!("{job_id}.tar.gz")) +pub(super) fn result_bundle_path( + store: &OutboundDispatchStore, + job_id: &str, +) -> std::path::PathBuf { + store + .root() + .join(super::OUTBOUND_RESULTS_DIR) + .join(format!("{job_id}.tar.gz")) } fn result_summary_path(store: &OutboundDispatchStore, job_id: &str) -> std::path::PathBuf { - store.root().join(".results").join(format!("{job_id}.json")) + store + .root() + .join(super::OUTBOUND_RESULTS_DIR) + .join(format!("{job_id}.json")) } /// Apply a pulled result bundle to a local workspace. diff --git a/src/crates/assembly/core/src/service/dispatch/device_controller.rs b/src/crates/assembly/core/src/service/dispatch/device_controller.rs index 0c15504170..c0f3a9c04e 100644 --- a/src/crates/assembly/core/src/service/dispatch/device_controller.rs +++ b/src/crates/assembly/core/src/service/dispatch/device_controller.rs @@ -3,7 +3,10 @@ use std::path::Path; use anyhow::{anyhow, Context}; use async_trait::async_trait; use base64::Engine as _; -use bitfun_services_integrations::remote_ssh::dispatch_ssh::{self, DispatchSshProbe}; +use bitfun_services_core::dispatch_workspace::sha256_bytes; +use bitfun_services_integrations::remote_ssh::dispatch_ssh::{ + self, harden_result_directory, DispatchSshProbe, +}; use serde_json::{json, Value}; use tokio::io::{AsyncReadExt, AsyncSeekExt}; @@ -20,6 +23,9 @@ use super::{ }; const DEVICE_WORKSPACE_CHUNK_BYTES: usize = 256 * 1024; +/// A result bundle carries only changed files, and the device transport +/// reassembles it in memory, so it is bounded well below a full snapshot. +const MAX_DEVICE_RESULT_BUNDLE_BYTES: u64 = 256 * 1024 * 1024; const DEVICE_WORKSPACE_COMMIT_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(750); const DEVICE_WORKSPACE_COMMIT_WAIT: std::time::Duration = std::time::Duration::from_secs(15 * 60); @@ -357,6 +363,109 @@ async fn resolve_device_workspace( } } +/// Pull a finished job's result bundle back from an account device. +/// +/// The device transport carries JSON only, so the bundle streams back in +/// base64 chunks — the mirror of `upload_device_workspace`. The digest the +/// target reported is verified over the reassembled bytes before anything is +/// staged, so a truncated or altered stream cannot reach the apply step. +pub async fn pull_device_result( + rpc: &dyn DeviceDispatchRpc, + store: &OutboundDispatchStore, + request: DispatchJobRequest, +) -> anyhow::Result { + let destination = super::controller::result_bundle_path(store, &request.job_id); + let destination = destination.as_path(); + let record = store + .get(&request.job_id) + .await? + .ok_or_else(|| anyhow!("Outbound dispatch job was not found"))?; + let DispatchTarget::Device { device_id, .. } = &record.target else { + anyhow::bail!("Device dispatch result pull requires a device target"); + }; + + let response = rpc + .invoke( + device_id, + "dispatch_target_workspace_result", + json!({ "jobId": request.job_id }), + ) + .await?; + let summary = response + .get("summary") + .ok_or_else(|| anyhow!("Device dispatch target returned no result summary"))?; + let expected_size = summary + .get("archiveSize") + .and_then(Value::as_u64) + .ok_or_else(|| anyhow!("Device dispatch target returned no result bundle size"))?; + let expected_digest = summary + .get("archiveSha256") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("Device dispatch target returned no result bundle digest"))? + .to_string(); + if expected_size > MAX_DEVICE_RESULT_BUNDLE_BYTES { + anyhow::bail!( + "Device dispatch result bundle exceeds the {} MB safety limit", + MAX_DEVICE_RESULT_BUNDLE_BYTES / (1024 * 1024) + ); + } + + let mut bytes = Vec::with_capacity(expected_size as usize); + while (bytes.len() as u64) < expected_size { + let chunk = rpc + .invoke( + device_id, + "dispatch_target_workspace_result_chunk", + json!({ + "jobId": request.job_id, + "offset": bytes.len() as u64, + "length": DEVICE_WORKSPACE_CHUNK_BYTES as u64, + }), + ) + .await?; + let encoded = chunk + .get("dataBase64") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("Device dispatch target returned no result chunk data"))?; + let decoded = base64::engine::general_purpose::STANDARD + .decode(encoded) + .context("decode dispatch result chunk")?; + if decoded.is_empty() { + anyhow::bail!( + "Device dispatch result bundle ended at {} of {expected_size} bytes", + bytes.len() + ); + } + bytes.extend_from_slice(&decoded); + if bytes.len() as u64 > expected_size { + anyhow::bail!("Device dispatch target returned more result bytes than it declared"); + } + } + + let actual_digest = sha256_bytes(&bytes); + if !actual_digest.eq_ignore_ascii_case(&expected_digest) { + anyhow::bail!("Device dispatch result bundle does not match the reported digest"); + } + + if let Some(parent) = destination.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("create result staging {}", parent.display()))?; + harden_result_directory(parent)?; + } + dispatch_ssh::write_private_file(destination, &bytes)?; + + let mut response = response; + if let Some(object) = response.as_object_mut() { + object.insert( + "localBundlePath".to_string(), + Value::String(destination.to_string_lossy().to_string()), + ); + } + // Same durable summary the SSH path records, so applying is transport-blind. + super::controller::record_result_summary(store, &request.job_id, &response)?; + Ok(response) +} + async fn upload_device_workspace( rpc: &dyn DeviceDispatchRpc, device_id: &str, @@ -483,6 +592,9 @@ async fn load_device_record( #[cfg(test)] mod tests { + use super::*; + use std::sync::Mutex; + #[test] fn target_command_names_are_separate_from_outbound_commands() { for command in [ @@ -496,9 +608,145 @@ mod tests { "dispatch_target_workspace_begin", "dispatch_target_workspace_chunk", "dispatch_target_workspace_commit", + "dispatch_target_workspace_result", + "dispatch_target_workspace_result_chunk", ] { assert!(command.starts_with("dispatch_target_")); assert_ne!(command, "dispatch_submit"); } } + + /// Serves a fixed bundle back in chunks, like a real device would. + struct BundleRpc { + bundle: Vec, + declared_digest: String, + calls: Mutex>, + } + + #[async_trait] + impl DeviceDispatchRpc for BundleRpc { + async fn invoke( + &self, + _device_id: &str, + command: &str, + args: Value, + ) -> anyhow::Result { + self.calls.lock().unwrap().push(command.to_string()); + match command { + "dispatch_target_workspace_result" => Ok(json!({ + "bundlePath": "/home/u/.bitfun/dispatch/workspaces/job-1/result.tar.gz", + "workspacePath": "/home/u/.bitfun/dispatch/workspaces/job-1/current", + "summary": { + "added": ["new.txt"], + "modified": [], + "deleted": [], + "baselineSha256": {}, + "archiveSize": self.bundle.len() as u64, + "archiveSha256": self.declared_digest, + } + })), + "dispatch_target_workspace_result_chunk" => { + let offset = args.get("offset").and_then(Value::as_u64).unwrap() as usize; + let length = args.get("length").and_then(Value::as_u64).unwrap() as usize; + let end = (offset + length).min(self.bundle.len()); + Ok(json!({ + "offset": end as u64, + "dataBase64": base64::engine::general_purpose::STANDARD + .encode(&self.bundle[offset..end]), + "eof": end >= self.bundle.len(), + })) + } + other => anyhow::bail!("unexpected command {other}"), + } + } + } + + async fn device_store(root: &Path) -> OutboundDispatchStore { + let store = OutboundDispatchStore::new_in_root_for_tests(root.to_path_buf()); + let record = OutboundDispatchRecord::new( + "job-1".to_string(), + DispatchTarget::Device { + device_id: "device-a".to_string(), + workspace_path: "/w".to_string(), + display_name: "Phone".to_string(), + }, + "session-1".to_string(), + "/w".to_string(), + "prompt", + "succeeded", + ) + .expect("record"); + store.bind_if_absent(&record).await.expect("bind"); + store + } + + #[tokio::test] + async fn a_device_streams_its_result_bundle_back_and_it_is_verified() { + let temp = tempfile::tempdir().expect("temp"); + let store = device_store(temp.path()).await; + // Larger than one chunk, so the loop is genuinely exercised. + let bundle = vec![7_u8; DEVICE_WORKSPACE_CHUNK_BYTES + 1234]; + let rpc = BundleRpc { + declared_digest: sha256_bytes(&bundle), + bundle: bundle.clone(), + calls: Mutex::new(Vec::new()), + }; + + let response = pull_device_result( + &rpc, + &store, + DispatchJobRequest { + job_id: "job-1".to_string(), + }, + ) + .await + .expect("pull"); + + let staged = response + .get("localBundlePath") + .and_then(Value::as_str) + .expect("staged path"); + assert_eq!(std::fs::read(staged).expect("read staged"), bundle); + let chunk_calls = rpc + .calls + .lock() + .unwrap() + .iter() + .filter(|c| c.as_str() == "dispatch_target_workspace_result_chunk") + .count(); + assert!(chunk_calls >= 2, "a multi-chunk bundle must loop"); + // The summary must be recorded for the apply step, as on the SSH path. + assert!(temp.path().join(".results/job-1.json").is_file()); + } + + #[tokio::test] + async fn a_tampered_device_stream_never_reaches_the_apply_step() { + let temp = tempfile::tempdir().expect("temp"); + let store = device_store(temp.path()).await; + let bundle = vec![3_u8; 4096]; + let rpc = BundleRpc { + // Declares a digest the streamed bytes do not match. + declared_digest: sha256_bytes(b"something else entirely"), + bundle, + calls: Mutex::new(Vec::new()), + }; + + let error = pull_device_result( + &rpc, + &store, + DispatchJobRequest { + job_id: "job-1".to_string(), + }, + ) + .await + .expect_err("a digest mismatch must fail the pull"); + assert!( + error.to_string().contains("does not match the reported digest"), + "{error}" + ); + assert!( + !temp.path().join(".results/job-1.tar.gz").exists(), + "nothing may be staged when the stream does not verify" + ); + } } diff --git a/src/crates/assembly/core/src/service/dispatch/mod.rs b/src/crates/assembly/core/src/service/dispatch/mod.rs index 3b4109912a..d6e99b5d21 100644 --- a/src/crates/assembly/core/src/service/dispatch/mod.rs +++ b/src/crates/assembly/core/src/service/dispatch/mod.rs @@ -46,13 +46,17 @@ pub use controller::{ pub use device_controller::{ answer_device as answer_device_dispatch, append_device as append_device_dispatch, cancel_device as cancel_device_dispatch, list_device_jobs as list_device_dispatch_jobs, - probe_device as probe_device_dispatch_target, status_device as get_device_dispatch_status, + probe_device as probe_device_dispatch_target, + pull_device_result as pull_device_dispatch_result, + status_device as get_device_dispatch_status, submit_device as submit_device_dispatch, DeviceDispatchRpc, }; pub use target::{DispatchTarget, DispatchTargetRequest, DispatchWorkspaceDeliveryRequest}; const PROMPT_PREVIEW_CHARS: usize = 160; const OUTBOUND_WORKSPACE_UPLOADS_DIR: &str = ".workspace-uploads"; +/// Where pulled result bundles are staged before the user applies them. +pub(super) const OUTBOUND_RESULTS_DIR: &str = ".results"; const TERMINAL_OUTBOUND_RETENTION_DAYS: i64 = 30; #[derive(Debug, Clone)] @@ -286,6 +290,16 @@ impl OutboundDispatchStore { .num_days() >= TERMINAL_OUTBOUND_RETENTION_DAYS => { + // Best effort: a stranded bundle is disk waste, not a + // correctness problem, and must not keep the expired record + // alive forever. + if let Err(error) = self.remove_result_bundle(&record.job_id).await { + log::warn!( + "Failed to remove expired dispatch result bundle: job_id={} error={}", + record.job_id, + error + ); + } if let Err(error) = self.remove_workspace_snapshot(&record.job_id).await { log::warn!( "Failed to remove expired outbound dispatch snapshot: job_id={} error={}", @@ -439,6 +453,27 @@ impl OutboundDispatchStore { }) } + /// Drop a pulled result bundle and its summary. + /// + /// Separate from `remove_workspace_snapshot` on purpose: that one runs as + /// soon as the target durably owns the job, which is long before the user + /// has had a chance to look at the results. + pub async fn remove_result_bundle(&self, job_id: &str) -> anyhow::Result<()> { + validate_id(job_id)?; + let results = self.root.join(OUTBOUND_RESULTS_DIR); + for path in [ + results.join(format!("{job_id}.tar.gz")), + results.join(format!("{job_id}.json")), + ] { + match fs::remove_file(&path).await { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + } + Ok(()) + } + pub async fn remove_workspace_snapshot(&self, job_id: &str) -> anyhow::Result<()> { validate_id(job_id)?; let uploads = self.root.join(OUTBOUND_WORKSPACE_UPLOADS_DIR); @@ -726,6 +761,37 @@ mod tests { } } + #[tokio::test] + async fn expired_jobs_do_not_strand_their_result_bundles() { + let temp = tempfile::tempdir().expect("temp dir"); + let store = OutboundDispatchStore::new_in_root_for_tests(temp.path().to_path_buf()); + let results = temp.path().join(OUTBOUND_RESULTS_DIR); + fs::create_dir_all(&results).await.expect("results dir"); + let bundle = results.join("job-1.tar.gz"); + let summary = results.join("job-1.json"); + fs::write(&bundle, b"bundle").await.expect("bundle"); + fs::write(&summary, b"{}").await.expect("summary"); + // A second job's bundle must survive the first job's cleanup. + let other = results.join("job-2.tar.gz"); + fs::write(&other, b"other").await.expect("other"); + + store.remove_result_bundle("job-1").await.expect("remove"); + assert!(!bundle.exists(), "expired bundle must be removed"); + assert!(!summary.exists(), "expired summary must be removed"); + assert!(other.exists(), "an unrelated job must be untouched"); + + // Removing twice is how GC behaves after a partial failure. + store + .remove_result_bundle("job-1") + .await + .expect("removing an absent bundle is not an error"); + + assert!( + store.remove_result_bundle("../escape").await.is_err(), + "job ids must stay validated on this path too" + ); + } + #[tokio::test] async fn rejects_path_traversal_job_ids() { let temp = tempfile::tempdir().expect("tempdir"); diff --git a/src/crates/services/services-core/src/dispatch_workspace.rs b/src/crates/services/services-core/src/dispatch_workspace.rs index ef3f26f6db..50a5125290 100644 --- a/src/crates/services/services-core/src/dispatch_workspace.rs +++ b/src/crates/services/services-core/src/dispatch_workspace.rs @@ -1013,7 +1013,8 @@ fn valid_sha256(value: &str) -> bool { value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit()) } -fn sha256_bytes(bytes: &[u8]) -> String { +/// Digest of an in-memory buffer, the counterpart of [`sha256_file`]. +pub fn sha256_bytes(bytes: &[u8]) -> String { format!("{:x}", Sha256::digest(bytes)) } diff --git a/src/crates/services/services-integrations/src/remote_ssh/dispatch_ssh.rs b/src/crates/services/services-integrations/src/remote_ssh/dispatch_ssh.rs index c0166ff8ac..b9db59cda8 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/dispatch_ssh.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/dispatch_ssh.rs @@ -1209,6 +1209,23 @@ pub async fn pull_result( let cli_path = target.cli_path.as_deref().ok_or_else(|| { anyhow!("BitFun CLI is not installed on the SSH target; confirm installation first") })?; + + // Returning results is an optional capability, so a target that predates it + // is a normal situation rather than a fault. Ask before invoking the verb: + // otherwise the only signal is clap's `unrecognized subcommand`, which says + // nothing about what the user should do. + let protocol = invoke_json_at_path( + manager, + connection_id, + &target.home, + cli_path, + "probe", + &serde_json::json!({}), + ) + .await + .context("probe the dispatch target before pulling results")?; + ensure_result_bundle_capability(&protocol)?; + let response = invoke_json_at_path( manager, connection_id, @@ -1239,12 +1256,15 @@ pub async fn pull_result( MAX_RESULT_BUNDLE_BYTES / (1024 * 1024) )); } + // The bundle carries the user's source, including the ignored files the + // snapshot deliberately shipped. The outbound root is already owner-only, + // but harden this level too rather than relying on a parent one layer up. if let Some(parent) = destination.parent() { std::fs::create_dir_all(parent) .with_context(|| format!("create result staging {}", parent.display()))?; + harden_result_directory(parent)?; } - std::fs::write(destination, &bytes) - .with_context(|| format!("store result bundle {}", destination.display()))?; + write_private_file(destination, &bytes)?; let mut response = response; if let Some(object) = response.as_object_mut() { @@ -1256,6 +1276,58 @@ pub async fn pull_result( Ok(response) } +pub fn harden_result_directory(path: &std::path::Path) -> Result<()> { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)) + .with_context(|| format!("restrict result staging {}", path.display()))?; + } + #[cfg(not(unix))] + let _ = path; + Ok(()) +} + +/// Create owner-only before writing, so the contents are never briefly governed +/// by the process umask. +pub fn write_private_file(path: &std::path::Path, bytes: &[u8]) -> Result<()> { + let mut options = std::fs::OpenOptions::new(); + options.write(true).create(true).truncate(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let mut file = options + .open(path) + .with_context(|| format!("create {}", path.display()))?; + std::io::Write::write_all(&mut file, bytes) + .with_context(|| format!("write {}", path.display()))?; + Ok(()) +} + +/// Optional capability: a target without it still runs jobs, it just cannot +/// hand their results back. Deliberately absent from +/// `REQUIRED_DISPATCH_CAPABILITIES` so an older CLI stays fully usable. +pub const WORKSPACE_RESULT_CAPABILITY: &str = "workspace_result_bundle"; + +fn ensure_result_bundle_capability(protocol: &Value) -> Result<()> { + let advertises = protocol + .get("capabilities") + .and_then(Value::as_array) + .is_some_and(|capabilities| { + capabilities + .iter() + .any(|capability| capability.as_str() == Some(WORKSPACE_RESULT_CAPABILITY)) + }); + if !advertises { + return Err(anyhow!( + "this target's BitFun CLI cannot return job results; update it to a release that supports {WORKSPACE_RESULT_CAPABILITY}" + )); + } + Ok(()) +} + /// A result bundle may only be read from the managed directory of the job it /// belongs to. fn validate_managed_result_path(home: &str, job_id: &str, bundle_path: &str) -> Result<()> { @@ -2455,6 +2527,75 @@ mod tests { } } + #[test] + fn a_target_without_the_result_capability_is_told_what_to_do() { + // Optional capability: the failure must name the fix, not surface + // clap's "unrecognized subcommand" from the verb invocation. + let without = serde_json::json!({ + "capabilities": ["persistent_jobs", "cursor_events"] + }); + let error = ensure_result_bundle_capability(&without) + .expect_err("a target that cannot return results must say so"); + assert!( + error.to_string().contains("cannot return job results"), + "{error}" + ); + + let with = serde_json::json!({ + "capabilities": ["persistent_jobs", WORKSPACE_RESULT_CAPABILITY] + }); + assert!(ensure_result_bundle_capability(&with).is_ok()); + + // A malformed probe must fail closed rather than assume support. + assert!(ensure_result_bundle_capability(&serde_json::json!({})).is_err()); + } + + #[test] + fn the_optional_result_capability_is_never_required_for_ordinary_dispatch() { + // Requiring it would make every older target unusable for jobs it can + // still run perfectly well. + assert!( + !REQUIRED_DISPATCH_CAPABILITIES.contains(&WORKSPACE_RESULT_CAPABILITY), + "returning results must stay optional" + ); + } + + #[cfg(unix)] + #[test] + fn staged_result_bundles_are_owner_only() { + use std::os::unix::fs::PermissionsExt; + + let temp = tempfile::tempdir().expect("temp dir"); + let staging = temp.path().join(".results"); + std::fs::create_dir_all(&staging).expect("staging"); + // Simulate a permissive umask having created it. + std::fs::set_permissions(&staging, std::fs::Permissions::from_mode(0o755)) + .expect("loosen"); + harden_result_directory(&staging).expect("harden"); + assert_eq!( + std::fs::metadata(&staging).expect("stat").permissions().mode() & 0o777, + 0o700, + "the staging directory holds user source and must not be world-readable" + ); + + let bundle = staging.join("job-1.tar.gz"); + write_private_file(&bundle, b"bundle bytes").expect("write"); + assert_eq!( + std::fs::metadata(&bundle).expect("stat").permissions().mode() & 0o777, + 0o600, + "the bundle itself must be owner-only" + ); + assert_eq!(std::fs::read(&bundle).expect("read"), b"bundle bytes"); + + // Rewriting must not widen the mode or leave a stale tail. + write_private_file(&bundle, b"short").expect("rewrite"); + assert_eq!( + std::fs::metadata(&bundle).expect("stat").permissions().mode() & 0o777, + 0o600 + ); + assert_eq!(std::fs::read(&bundle).expect("read"), b"short"); + } + #[test] fn a_result_bundle_is_only_read_from_its_own_managed_directory() { // The path is chosen by the target, so a compromised or buggy one must