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