From 57e96d3be8c7a66476ce2ac50e4f28027a3bfc26 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 8 Jul 2026 20:18:38 +0900 Subject: [PATCH] feat: add backend resource fetch api --- Cargo.lock | 1 + crates/worker-runtime/Cargo.toml | 3 +- crates/worker-runtime/src/config_bundle.rs | 106 +++++++- crates/worker-runtime/src/http_server.rs | 2 + crates/worker-runtime/src/lib.rs | 1 + crates/worker-runtime/src/main.rs | 21 ++ crates/worker-runtime/src/profile_archive.rs | 2 +- crates/worker-runtime/src/resource.rs | 256 ++++++++++++++++++ crates/worker-runtime/src/runtime.rs | 1 + crates/worker-runtime/src/worker_backend.rs | 77 +++++- crates/workspace-server/src/companion.rs | 1 + crates/workspace-server/src/hosts.rs | 102 ++++++- crates/workspace-server/src/lib.rs | 1 + .../workspace-server/src/resource_broker.rs | 231 ++++++++++++++++ crates/workspace-server/src/server.rs | 93 ++++++- package.nix | 2 +- 16 files changed, 859 insertions(+), 41 deletions(-) create mode 100644 crates/worker-runtime/src/resource.rs create mode 100644 crates/workspace-server/src/resource_broker.rs diff --git a/Cargo.lock b/Cargo.lock index e23dd6c8..c90d9941 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5936,6 +5936,7 @@ dependencies = [ "llm-engine", "manifest", "protocol", + "reqwest", "serde", "serde_json", "session-store", diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index a010ed23..8cf715a3 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -14,7 +14,7 @@ required-features = ["ws-server", "fs-store"] [features] default = [] fs-store = [] -http-server = ["dep:axum", "dep:tower"] +http-server = ["dep:axum", "dep:tower", "dep:reqwest"] ws-server = ["http-server", "axum/ws", "dep:futures", "tokio/sync"] [dependencies] @@ -28,6 +28,7 @@ serde = { workspace = true, features = ["derive"] } session-store.workspace = true sha2.workspace = true serde_json.workspace = true +reqwest = { version = "0.13", optional = true, default-features = false, features = ["json", "rustls"] } tar.workspace = true thiserror = { workspace = true } tokio = { workspace = true, features = ["net", "rt", "sync", "time"] } diff --git a/crates/worker-runtime/src/config_bundle.rs b/crates/worker-runtime/src/config_bundle.rs index beafad32..a098bca7 100644 --- a/crates/worker-runtime/src/config_bundle.rs +++ b/crates/worker-runtime/src/config_bundle.rs @@ -4,6 +4,7 @@ use crate::profile_archive::{ ProfileArchiveError, ProfileSourceArchive, ProfileSourceArchiveRef, VerifiedProfileSourceArchive, }; +use crate::resource::{BackendResourceHandle, validate_resource_handle_text}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -24,6 +25,8 @@ pub struct ConfigBundle { pub declarations: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub profile_source_archive: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub profile_source_archive_handle: Option, } impl ConfigBundle { @@ -75,6 +78,27 @@ impl ConfigBundle { lines.push(format!("profile_archive_entrypoint\0{selector}\0{path}")); } } + if let Some(handle) = &self.profile_source_archive_handle { + lines.push(format!( + "profile_archive_handle\0{}\0{}\0{}\0{}\0{}", + handle.workspace_id, + handle.resource_id, + handle.digest, + handle.revision, + handle.max_bytes + )); + for (selector, path) in handle + .profile_source_graph + .as_ref() + .map(|graph| &graph.entrypoints) + .into_iter() + .flatten() + { + lines.push(format!( + "profile_archive_handle_entrypoint\0{selector}\0{path}" + )); + } + } lines.sort(); let mut hasher = Sha256::new(); @@ -105,7 +129,12 @@ impl ConfigBundle { profile_source_archive: self .profile_source_archive .as_ref() - .map(|archive| archive.reference.source_graph.clone()), + .map(|archive| archive.reference.source_graph.clone()) + .or_else(|| { + self.profile_source_archive_handle + .as_ref() + .and_then(|handle| handle.profile_source_graph.clone()) + }), } } @@ -253,6 +282,33 @@ pub(crate) fn validate_config_bundle(bundle: &ConfigBundle) -> Result<(), Runtim RuntimeError::InvalidRequest(format!("invalid profile source archive: {err}")) })?; } + if let Some(handle) = &bundle.profile_source_archive_handle { + for (label, value) in [ + ("resource handle workspace id", handle.workspace_id.as_str()), + ("resource handle resource id", handle.resource_id.as_str()), + ("resource handle digest", handle.digest.as_str()), + ("resource handle nonce", handle.nonce.as_str()), + ("resource handle revision", handle.revision.as_str()), + ("resource handle content type", handle.content_type.as_str()), + ( + "resource handle audit correlation id", + handle.audit_correlation_id.as_str(), + ), + ] { + validate_resource_handle_text(label, value).map_err(RuntimeError::InvalidRequest)?; + } + if !handle.digest.starts_with("sha256:") { + return Err(RuntimeError::InvalidRequest( + "resource handle digest must use sha256:".to_string(), + )); + } + if handle.profile_source_graph.is_none() { + return Err(RuntimeError::InvalidRequest( + "profile source archive resource handle must include source graph summary" + .to_string(), + )); + } + } Ok(()) } @@ -529,6 +585,7 @@ mod tests { reference: reference.to_string(), }], profile_source_archive: None, + profile_source_archive_handle: None, } .with_computed_digest() } @@ -560,6 +617,53 @@ mod tests { validate_config_bundle(&bundle_with_declaration("vault:team.api-key")).unwrap(); } + #[test] + fn bundle_summary_redacts_runtime_internal_resource_handle() { + let mut bundle = bundle_with_declaration("secret:github-token"); + let source_graph = crate::profile_archive::ProfileSourceGraphSummary { + entrypoints: std::collections::BTreeMap::from([( + "default".to_string(), + "profiles/default.dcdl".to_string(), + )]), + source_count: 1, + import_count: 0, + total_source_bytes: 16, + }; + bundle.profile_source_archive_handle = Some(crate::resource::BackendResourceHandle { + kind: crate::resource::BackendResourceKind::ProfileSourceArchive, + workspace_id: "workspace-1".to_string(), + scope_id: Some("scope-1".to_string()), + runtime_id: Some("runtime-1".to_string()), + worker_id: Some("worker-1".to_string()), + resource_id: "profile-source-archive:test".to_string(), + digest: "sha256:0123456789abcdef".to_string(), + operation: crate::resource::BackendResourceOperation::FetchArchive, + expires_at_unix_seconds: 4_102_444_800, + nonce: "nonce-not-for-browser".to_string(), + revision: "rev-1".to_string(), + generation: Some(1), + max_bytes: 128, + content_type: crate::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(), + redaction: crate::resource::ResourceRedactionPolicy::RuntimeInternalOnly, + audit_correlation_id: "audit-1".to_string(), + profile_source_graph: Some(source_graph), + }); + let rendered = serde_json::to_string(&bundle.summary()).unwrap(); + assert!(rendered.contains("profile_source_archive")); + for forbidden in [ + "nonce-not-for-browser", + "profile-source-archive:test", + "sha256:0123456789abcdef", + "runtime-1", + "worker-1", + ] { + assert!( + !rendered.contains(forbidden), + "leaked {forbidden}: {rendered}" + ); + } + } + #[test] fn rejects_unsafe_bundle_ids_and_refs() { for id in ["bundle/1", "bundle?x", "bundle&x", "bundle#x", " bundle"] { diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index e01dc9b3..f73f2b16 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -850,6 +850,7 @@ mod tests { }], declarations: Vec::new(), profile_source_archive: None, + profile_source_archive_handle: None, } .with_computed_digest() } @@ -1137,6 +1138,7 @@ mod ws_tests { }], declarations: Vec::new(), profile_source_archive: None, + profile_source_archive_handle: None, } .with_computed_digest() } diff --git a/crates/worker-runtime/src/lib.rs b/crates/worker-runtime/src/lib.rs index 405aab74..c897f79c 100644 --- a/crates/worker-runtime/src/lib.rs +++ b/crates/worker-runtime/src/lib.rs @@ -20,6 +20,7 @@ pub mod interaction; pub mod management; pub mod observation; pub mod profile_archive; +pub mod resource; mod runtime; pub mod worker_backend; pub mod working_directory; diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index 54112219..1f0f723c 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -78,6 +78,14 @@ fn build_runtime(config: &ProcessConfig) -> Result { if let Some(runtime_base_dir) = config.worker_runtime_base_dir.clone() { factory = factory.with_runtime_base_dir(runtime_base_dir); } + if let Some(endpoint) = config.backend_resource_endpoint.clone() { + factory = factory.with_resource_client(Arc::new( + worker_runtime::resource::HttpBackendResourceClient::new( + endpoint, + config.backend_resource_token.clone(), + ), + )); + } if let Some(profile) = config.profile.clone() { factory = factory.with_profile(profile); } @@ -173,6 +181,13 @@ where config.worker_runtime_base_dir = Some(PathBuf::from(take_value(&flag, inline_value, &mut args)?)); } + "--backend-resource-endpoint" => { + config.backend_resource_endpoint = + Some(take_value(&flag, inline_value, &mut args)?); + } + "--backend-resource-token" => { + config.backend_resource_token = Some(take_value(&flag, inline_value, &mut args)?); + } "--profile" => { config.profile = Some(take_value(&flag, inline_value, &mut args)?); } @@ -317,6 +332,8 @@ struct ProcessConfig { worker_store_dir: Option, worker_metadata_dir: Option, worker_runtime_base_dir: Option, + backend_resource_endpoint: Option, + backend_resource_token: Option, profile: Option, } @@ -330,6 +347,8 @@ impl ProcessConfig { worker_store_dir: None, worker_metadata_dir: None, worker_runtime_base_dir: None, + backend_resource_endpoint: None, + backend_resource_token: None, profile: None, }) } @@ -405,6 +424,8 @@ Options:\n\ --worker-store-dir Worker session store directory\n\ --worker-metadata-dir Worker metadata directory\n\ --worker-runtime-base-dir Worker controller runtime directory\n\ + --backend-resource-endpoint Internal Backend resource fetch endpoint for resource handles\n\ + --backend-resource-token Optional bearer token for the Backend resource fetch endpoint\n\ --store Runtime catalog store selection (default: memory)\n\ --fs-root Runtime catalog filesystem store root\n\ --local-token Minimal local bearer token placeholder\n\ diff --git a/crates/worker-runtime/src/profile_archive.rs b/crates/worker-runtime/src/profile_archive.rs index cc7a6d09..b759c9d6 100644 --- a/crates/worker-runtime/src/profile_archive.rs +++ b/crates/worker-runtime/src/profile_archive.rs @@ -472,7 +472,7 @@ fn validate_archive_path(path: &str) -> Result<(), ProfileArchiveError> { Ok(()) } -fn sha256_hex(data: &[u8]) -> String { +pub fn sha256_hex(data: &[u8]) -> String { let digest = Sha256::digest(data); let mut out = String::from("sha256:"); for byte in digest.as_slice() { diff --git a/crates/worker-runtime/src/resource.rs b/crates/worker-runtime/src/resource.rs new file mode 100644 index 00000000..8335fc5c --- /dev/null +++ b/crates/worker-runtime/src/resource.rs @@ -0,0 +1,256 @@ +use crate::identity::{RuntimeId, WorkerId}; +use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef, sha256_hex}; +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::Mutex; + +pub const PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE: &str = + "application/vnd.yoi.profile-source-archive+tar"; +pub const DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES: u64 = 2 * 1024 * 1024; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum BackendResourceKind { + ProfileSourceArchive, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum BackendResourceOperation { + FetchArchive, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackendResourceHandle { + pub kind: BackendResourceKind, + pub workspace_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scope_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub runtime_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_id: Option, + pub resource_id: String, + pub digest: String, + pub operation: BackendResourceOperation, + pub expires_at_unix_seconds: i64, + pub nonce: String, + pub revision: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub generation: Option, + pub max_bytes: u64, + pub content_type: String, + pub redaction: ResourceRedactionPolicy, + pub audit_correlation_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub profile_source_graph: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ResourceRedactionPolicy { + RuntimeInternalOnly, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackendResourceFetchRequest { + pub handle: BackendResourceHandle, + pub runtime_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_id: Option, + pub audit_correlation_id: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackendResourceFetchResponse { + pub kind: BackendResourceKind, + pub resource_id: String, + pub digest: String, + pub content_type: String, + pub bytes: Vec, + pub audit_correlation_id: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)] +#[serde(tag = "code", rename_all = "snake_case")] +pub enum BackendResourceError { + #[error("backend resource handle is expired")] + Expired, + #[error("backend resource handle is unauthorized: {message}")] + Unauthorized { message: String }, + #[error("backend resource kind is unsupported")] + UnsupportedKind, + #[error("backend resource is missing")] + MissingResource, + #[error("backend resource digest mismatch: expected {expected}, got {actual}")] + DigestMismatch { expected: String, actual: String }, + #[error("backend resource response is oversized: limit {max_bytes}, actual {actual_bytes}")] + Oversized { max_bytes: u64, actual_bytes: u64 }, + #[error("backend resource content type mismatch: expected {expected}, got {actual}")] + ContentTypeMismatch { expected: String, actual: String }, + #[error("backend resource transport failed: {message}")] + Transport { message: String }, + #[error("backend resource response is invalid: {message}")] + InvalidResponse { message: String }, +} + +#[async_trait] +pub trait BackendResourceClient: Send + Sync + 'static { + async fn fetch_resource( + &self, + request: BackendResourceFetchRequest, + ) -> Result; +} + +#[cfg(feature = "http-server")] +#[derive(Clone, Debug)] +pub struct HttpBackendResourceClient { + endpoint: String, + bearer_token: Option, + client: reqwest::Client, +} + +#[cfg(feature = "http-server")] +impl HttpBackendResourceClient { + pub fn new(endpoint: impl Into, bearer_token: Option) -> Self { + Self { + endpoint: endpoint.into(), + bearer_token, + client: reqwest::Client::new(), + } + } +} + +#[cfg(feature = "http-server")] +#[async_trait] +impl BackendResourceClient for HttpBackendResourceClient { + async fn fetch_resource( + &self, + request: BackendResourceFetchRequest, + ) -> Result { + let builder = self.client.post(&self.endpoint).json(&request); + let builder = if let Some(token) = self.bearer_token.as_deref() { + builder.bearer_auth(token) + } else { + builder + }; + let response = builder + .send() + .await + .map_err(|err| BackendResourceError::Transport { + message: err.to_string(), + })?; + if response.status().is_success() { + response + .json::() + .await + .map_err(|err| BackendResourceError::InvalidResponse { + message: err.to_string(), + }) + } else { + let status = response.status(); + match response.json::().await { + Ok(error) => Err(error), + Err(err) => Err(BackendResourceError::Transport { + message: format!("backend resource fetch failed with HTTP {status}: {err}"), + }), + } + } + } +} + +#[derive(Default, Debug)] +pub struct ProfileSourceArchiveCache { + archives: Mutex>, +} + +impl ProfileSourceArchiveCache { + pub fn get(&self, digest: &str) -> Option { + self.archives.lock().ok()?.get(digest).cloned() + } + + pub fn insert(&self, archive: ProfileSourceArchive) { + if let Ok(mut archives) = self.archives.lock() { + archives.insert(archive.reference.digest.clone(), archive); + } + } +} + +pub fn build_profile_source_archive_fetch_request( + handle: BackendResourceHandle, + runtime_id: &RuntimeId, + worker_id: Option<&WorkerId>, +) -> BackendResourceFetchRequest { + let audit_correlation_id = handle.audit_correlation_id.clone(); + BackendResourceFetchRequest { + handle, + runtime_id: runtime_id.as_str().to_string(), + worker_id: worker_id.map(|id| id.as_str().to_string()), + audit_correlation_id, + } +} + +pub fn profile_source_archive_from_response( + handle: &BackendResourceHandle, + response: BackendResourceFetchResponse, +) -> Result { + if handle.kind != BackendResourceKind::ProfileSourceArchive + || response.kind != BackendResourceKind::ProfileSourceArchive + { + return Err(BackendResourceError::UnsupportedKind); + } + if handle.operation != BackendResourceOperation::FetchArchive { + return Err(BackendResourceError::Unauthorized { + message: "resource handle operation is not fetch_archive".to_string(), + }); + } + if response.content_type != handle.content_type { + return Err(BackendResourceError::ContentTypeMismatch { + expected: handle.content_type.clone(), + actual: response.content_type, + }); + } + let actual_bytes = response.bytes.len() as u64; + if actual_bytes > handle.max_bytes { + return Err(BackendResourceError::Oversized { + max_bytes: handle.max_bytes, + actual_bytes, + }); + } + let actual_digest = sha256_hex(&response.bytes); + if actual_digest != handle.digest || response.digest != handle.digest { + return Err(BackendResourceError::DigestMismatch { + expected: handle.digest.clone(), + actual: if response.digest != handle.digest { + response.digest + } else { + actual_digest + }, + }); + } + Ok(ProfileSourceArchive { + reference: ProfileSourceArchiveRef { + id: handle.resource_id.clone(), + digest: handle.digest.clone(), + size_bytes: actual_bytes, + source_graph: handle.profile_source_graph.clone().ok_or_else(|| { + BackendResourceError::InvalidResponse { + message: "profile source archive handle omitted source graph summary" + .to_string(), + } + })?, + }, + content: response.bytes, + }) +} + +pub fn validate_resource_handle_text(label: &str, value: &str) -> Result<(), String> { + if value.trim().is_empty() { + return Err(format!("{label} must not be empty")); + } + if value.len() > 256 || value.contains('\0') || value.contains('\n') || value.contains('\r') { + return Err(format!("{label} contains unsupported boundary text")); + } + Ok(()) +} diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 8665d6f3..e9d7b87c 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1628,6 +1628,7 @@ mod tests { reference: "capability:read".to_string(), }], profile_source_archive: None, + profile_source_archive_handle: None, } .with_computed_digest() } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index a5e62d4e..6863ae43 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -20,6 +20,10 @@ use crate::execution::{ WorkerExecutionRunState, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, }; use crate::interaction::{WorkerInput, WorkerInputKind}; +use crate::resource::{ + BackendResourceClient, BackendResourceError, ProfileSourceArchiveCache, + build_profile_source_archive_fetch_request, profile_source_archive_from_response, +}; use crate::working_directory::{WorkingDirectoryBinding, WorkingDirectoryMaterializer}; use async_trait::async_trait; use manifest::paths; @@ -46,7 +50,7 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static { /// Production factory that resolves a normal Worker profile and spawns it under /// `WorkerController`. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct ProfileRuntimeWorkerFactory { profile_base_dir: PathBuf, cwd: PathBuf, @@ -54,6 +58,8 @@ pub struct ProfileRuntimeWorkerFactory { worker_metadata_dir: Option, profile: Option, runtime_base_dir: Option, + resource_client: Option>, + profile_archive_cache: Arc, } impl ProfileRuntimeWorkerFactory { @@ -66,6 +72,8 @@ impl ProfileRuntimeWorkerFactory { worker_metadata_dir: None, profile: None, runtime_base_dir: None, + resource_client: None, + profile_archive_cache: Arc::new(ProfileSourceArchiveCache::default()), } } @@ -96,6 +104,11 @@ impl ProfileRuntimeWorkerFactory { self } + pub fn with_resource_client(mut self, resource_client: Arc) -> Self { + self.resource_client = Some(resource_client); + self + } + fn store_dir(&self) -> Result { self.store_dir .clone() @@ -152,6 +165,56 @@ impl ProfileRuntimeWorkerFactory { } Self::runtime_profile_value(&request.request.profile) } + async fn resolve_profile_source_archive( + &self, + bundle: &crate::config_bundle::ConfigBundle, + request: &WorkerExecutionSpawnRequest, + ) -> Result { + if let Some(archive) = verified_profile_source_archive(bundle) + .map_err(|err| format!("failed to verify profile source archive: {err}"))? + { + return Ok(archive); + } + let handle = bundle + .profile_source_archive_handle + .clone() + .ok_or_else(|| { + format!( + "config bundle {} does not contain a ProfileSourceArchive resource handle", + bundle.metadata.id + ) + })?; + if let Some(cached) = self.profile_archive_cache.get(&handle.digest) { + return cached + .verify() + .map_err(|err| format!("failed to verify cached profile source archive: {err}")); + } + let client = self.resource_client.as_ref().ok_or_else(|| { + format!( + "config bundle {} requires a Backend resource client for profile source archive fetch", + bundle.metadata.id + ) + })?; + let fetch_request = build_profile_source_archive_fetch_request( + handle.clone(), + &request.worker_ref.runtime_id, + Some(&request.worker_ref.worker_id), + ); + let response = client + .fetch_resource(fetch_request) + .await + .map_err(format_backend_resource_error)?; + let archive = profile_source_archive_from_response(&handle, response) + .map_err(format_backend_resource_error)?; + self.profile_archive_cache.insert(archive.clone()); + archive + .verify() + .map_err(|err| format!("failed to verify fetched profile source archive: {err}")) + } +} + +fn format_backend_resource_error(error: BackendResourceError) -> String { + format!("backend resource fetch failed: {error}") } #[async_trait] @@ -174,14 +237,9 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .unwrap_or_else(|| self.cwd.clone()); let (mut manifest, loader) = if let Some(bundle) = request.config_bundle.as_ref() { let selector = profile.as_deref().unwrap_or("builtin:default"); - let archive = verified_profile_source_archive(bundle) - .map_err(|err| format!("failed to verify profile source archive: {err}"))? - .ok_or_else(|| { - format!( - "config bundle {} does not contain a ProfileSourceArchive", - bundle.metadata.id - ) - })?; + let archive = self + .resolve_profile_source_archive(bundle, &request) + .await?; let manifest = archive .resolve_profile(selector, &worker_root, &worker_name) .map_err(|err| format!("failed to resolve profile source archive: {err}"))?; @@ -791,6 +849,7 @@ mod tests { }], declarations: Vec::new(), profile_source_archive: None, + profile_source_archive_handle: None, } .with_computed_digest() } diff --git a/crates/workspace-server/src/companion.rs b/crates/workspace-server/src/companion.rs index 27a00ed4..6f97f26e 100644 --- a/crates/workspace-server/src/companion.rs +++ b/crates/workspace-server/src/companion.rs @@ -455,6 +455,7 @@ fn companion_config_bundle() -> ConfigBundle { }], declarations: Vec::new(), profile_source_archive: Some(companion_profile_archive()), + profile_source_archive_handle: None, } .with_computed_digest() } diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 290657d3..978a10b1 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1,4 +1,5 @@ use crate::Error; +use crate::resource_broker::BackendResourceBroker; use chrono::Utc; use reqwest::StatusCode; use reqwest::blocking::{Client as BlockingHttpClient, RequestBuilder}; @@ -1032,6 +1033,7 @@ pub struct EmbeddedWorkerRuntime { host_id: String, runtime: worker_runtime::Runtime, execution_enabled: bool, + resource_broker: BackendResourceBroker, } fn embedded_runtime_options() -> EmbeddedRuntimeOptions { @@ -1082,6 +1084,11 @@ impl EmbeddedWorkerRuntime { Ok(embedded) } + pub fn with_resource_broker(mut self, resource_broker: BackendResourceBroker) -> Self { + self.resource_broker = resource_broker; + self + } + pub fn from_runtime(workspace_id: impl AsRef, runtime: worker_runtime::Runtime) -> Self { let runtime_id = runtime .runtime_id() @@ -1093,6 +1100,7 @@ impl EmbeddedWorkerRuntime { host_id: host_id_for_embedded_workspace(workspace_id.as_ref()), runtime, execution_enabled: false, + resource_broker: BackendResourceBroker::default(), } } @@ -1339,7 +1347,14 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { .profile .clone() .unwrap_or_else(|| embedded_profile_selector(&request.intent)); - let config_bundle = match default_embedded_config_bundle(&profile).and_then(|bundle| { + let runtime_id = EmbeddedRuntimeId::new(self.runtime_id.clone()); + let config_bundle = match default_embedded_config_bundle( + &profile, + &self.host_id, + runtime_id.as_ref(), + &self.resource_broker, + ) + .and_then(|bundle| { self.runtime .store_config_bundle(bundle) .map_err(|err| err.to_string()) @@ -1717,6 +1732,7 @@ pub struct RemoteWorkerRuntime { cached_capabilities: RuntimeCapabilitySummary, cached_status: String, host_id: String, + resource_broker: BackendResourceBroker, http: BlockingHttpClient, } @@ -1740,10 +1756,16 @@ impl RemoteWorkerRuntime { bearer_token: config.bearer_token, cached_capabilities: config.cached_capabilities, cached_status: config.cached_status, + resource_broker: BackendResourceBroker::default(), http, }) } + pub fn with_resource_broker(mut self, resource_broker: BackendResourceBroker) -> Self { + self.resource_broker = resource_broker; + self + } + fn endpoint(&self, path: &str) -> String { format!("{}{}", self.base_url, path) } @@ -2025,7 +2047,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { .profile .clone() .unwrap_or_else(|| embedded_profile_selector(&request.intent)); - let sync = match default_embedded_config_bundle(&profile) { + let runtime_id = EmbeddedRuntimeId::new(self.runtime_id.clone()); + let sync = match default_embedded_config_bundle( + &profile, + &self.host_id, + runtime_id.as_ref(), + &self.resource_broker, + ) { Ok(bundle) => self.sync_config_bundle(bundle), Err(error) => ConfigBundleSyncResult { state: WorkerOperationState::Rejected, @@ -2365,19 +2393,31 @@ fn embedded_worker_execution_status_label( } } -fn default_embedded_config_bundle(profile: &ProfileSelector) -> Result { +fn default_embedded_config_bundle( + profile: &ProfileSelector, + workspace_id: &str, + runtime_id: Option<&EmbeddedRuntimeId>, + resource_broker: &BackendResourceBroker, +) -> Result { let id = format!( "workspace-runtime-{}", embedded_profile_label(profile) .unwrap_or_else(|| "default".to_string()) .replace([':', '/', ' '], "-") ); + let archive = default_profile_source_archive(profile)?; + let handle = resource_broker.issue_profile_source_archive_handle( + workspace_id.to_string(), + runtime_id, + None, + archive, + ); Ok(ConfigBundle { metadata: ConfigBundleMetadata { id, digest: String::new(), revision: "workspace-runtime-v0".to_string(), - workspace_id: "workspace-server".to_string(), + workspace_id: workspace_id.to_string(), created_at: "runtime-generated".to_string(), provenance: ConfigBundleProvenance { source: "workspace-server".to_string(), @@ -2389,7 +2429,8 @@ fn default_embedded_config_bundle(profile: &ProfileSelector) -> Result "default".to_string(), ProfileSelector::Builtin(name) => name.clone(), @@ -2996,14 +3056,25 @@ mod tests { #[test] fn embedded_archive_rejects_unknown_selectors() { + let broker = BackendResourceBroker::default(); + let runtime_id = EmbeddedRuntimeId::new("runtime-test".to_string()).unwrap(); assert!( - default_embedded_config_bundle(&ProfileSelector::Builtin( - "builtin:missing".to_string() - )) + default_embedded_config_bundle( + &ProfileSelector::Builtin("builtin:missing".to_string()), + "workspace-test", + Some(&runtime_id), + &broker, + ) .is_err() ); assert!( - default_embedded_config_bundle(&ProfileSelector::Named("custom".to_string())).is_err() + default_embedded_config_bundle( + &ProfileSelector::Named("custom".to_string()), + "workspace-test", + Some(&runtime_id), + &broker, + ) + .is_err() ); } @@ -3030,6 +3101,7 @@ mod tests { reference: "capability:read".to_string(), }], profile_source_archive: None, + profile_source_archive_handle: None, } .with_computed_digest() } diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index d279a8d2..5507755d 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -11,6 +11,7 @@ pub mod identity; pub mod observation; pub mod records; pub mod repositories; +pub mod resource_broker; pub mod server; pub mod store; diff --git a/crates/workspace-server/src/resource_broker.rs b/crates/workspace-server/src/resource_broker.rs new file mode 100644 index 00000000..bfe4a4ca --- /dev/null +++ b/crates/workspace-server/src/resource_broker.rs @@ -0,0 +1,231 @@ +use async_trait::async_trait; +use chrono::{Duration, Utc}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use uuid::Uuid; +use worker_runtime::identity::{RuntimeId, WorkerId}; +use worker_runtime::profile_archive::ProfileSourceArchive; +use worker_runtime::resource::{ + BackendResourceClient, BackendResourceError, BackendResourceFetchRequest, + BackendResourceFetchResponse, BackendResourceHandle, BackendResourceKind, + BackendResourceOperation, DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES, + PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE, ResourceRedactionPolicy, +}; + +#[derive(Clone, Default)] +pub struct BackendResourceBroker { + resources: Arc>>, +} + +#[derive(Clone)] +struct StoredResource { + workspace_id: String, + runtime_id: Option, + worker_id: Option, + handle: BackendResourceHandle, + archive: ProfileSourceArchive, +} + +impl BackendResourceBroker { + pub fn issue_profile_source_archive_handle( + &self, + workspace_id: impl Into, + runtime_id: Option<&RuntimeId>, + worker_id: Option<&WorkerId>, + archive: ProfileSourceArchive, + ) -> BackendResourceHandle { + let workspace_id = workspace_id.into(); + let nonce = Uuid::now_v7().to_string(); + let audit_correlation_id = format!("resource-fetch-{nonce}"); + let expires_at = Utc::now() + Duration::minutes(15); + let handle = BackendResourceHandle { + kind: BackendResourceKind::ProfileSourceArchive, + workspace_id: workspace_id.clone(), + scope_id: Some("workspace-profile-source".to_string()), + runtime_id: runtime_id.map(|id| id.as_str().to_string()), + worker_id: worker_id.map(|id| id.as_str().to_string()), + resource_id: archive.reference.id.clone(), + digest: archive.reference.digest.clone(), + operation: BackendResourceOperation::FetchArchive, + expires_at_unix_seconds: expires_at.timestamp(), + nonce: nonce.clone(), + revision: archive.reference.digest.clone(), + generation: None, + max_bytes: DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES, + content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(), + redaction: ResourceRedactionPolicy::RuntimeInternalOnly, + audit_correlation_id, + profile_source_graph: Some(archive.reference.source_graph.clone()), + }; + let stored = StoredResource { + workspace_id, + runtime_id: runtime_id.map(|id| id.as_str().to_string()), + worker_id: worker_id.map(|id| id.as_str().to_string()), + handle: handle.clone(), + archive, + }; + if let Ok(mut resources) = self.resources.lock() { + resources.insert(nonce, stored); + } + handle + } + + pub fn fetch_profile_source_archive( + &self, + request: BackendResourceFetchRequest, + ) -> Result { + verify_handle_shape(&request.handle)?; + if request.handle.expires_at_unix_seconds < Utc::now().timestamp() { + return Err(BackendResourceError::Expired); + } + let stored = self + .resources + .lock() + .map_err(|_| BackendResourceError::Transport { + message: "resource broker lock poisoned".to_string(), + })? + .get(&request.handle.nonce) + .cloned() + .ok_or(BackendResourceError::MissingResource)?; + if stored.workspace_id != request.handle.workspace_id + || stored.runtime_id != request.handle.runtime_id + || stored.worker_id != request.handle.worker_id + || stored.handle.resource_id != request.handle.resource_id + || stored.handle.digest != request.handle.digest + || stored.handle.revision != request.handle.revision + { + return Err(BackendResourceError::Unauthorized { + message: "resource handle metadata does not match broker record".to_string(), + }); + } + if let Some(expected_runtime_id) = stored.runtime_id.as_deref() { + if expected_runtime_id != request.runtime_id { + return Err(BackendResourceError::Unauthorized { + message: "runtime id does not match resource handle".to_string(), + }); + } + } + if let Some(expected_worker_id) = stored.worker_id.as_deref() { + if Some(expected_worker_id) != request.worker_id.as_deref() { + return Err(BackendResourceError::Unauthorized { + message: "worker id does not match resource handle".to_string(), + }); + } + } + let actual_bytes = stored.archive.content.len() as u64; + if actual_bytes > request.handle.max_bytes { + return Err(BackendResourceError::Oversized { + max_bytes: request.handle.max_bytes, + actual_bytes, + }); + } + Ok(BackendResourceFetchResponse { + kind: BackendResourceKind::ProfileSourceArchive, + resource_id: stored.archive.reference.id, + digest: stored.archive.reference.digest, + content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(), + bytes: stored.archive.content, + audit_correlation_id: request.audit_correlation_id, + }) + } +} + +#[async_trait] +impl BackendResourceClient for BackendResourceBroker { + async fn fetch_resource( + &self, + request: BackendResourceFetchRequest, + ) -> Result { + self.fetch_profile_source_archive(request) + } +} + +fn verify_handle_shape(handle: &BackendResourceHandle) -> Result<(), BackendResourceError> { + if handle.kind != BackendResourceKind::ProfileSourceArchive { + return Err(BackendResourceError::UnsupportedKind); + } + if handle.operation != BackendResourceOperation::FetchArchive { + return Err(BackendResourceError::Unauthorized { + message: "resource handle operation is not fetch_archive".to_string(), + }); + } + if handle.content_type != PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE { + return Err(BackendResourceError::ContentTypeMismatch { + expected: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(), + actual: handle.content_type.clone(), + }); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeMap; + use worker_runtime::identity::RuntimeId; + use worker_runtime::profile_archive::{ + ProfileSourceArchive, ProfileSourceArchiveRef, ProfileSourceGraphSummary, sha256_hex, + }; + + fn archive() -> ProfileSourceArchive { + let content = b"archive-content".to_vec(); + let mut entrypoints = BTreeMap::new(); + entrypoints.insert("default".to_string(), "profiles/default.dcdl".to_string()); + ProfileSourceArchive { + reference: ProfileSourceArchiveRef { + id: "profile-source-archive:test".to_string(), + digest: sha256_hex(&content), + size_bytes: content.len() as u64, + source_graph: ProfileSourceGraphSummary { + entrypoints, + source_count: 1, + import_count: 0, + total_source_bytes: content.len() as u64, + }, + }, + content, + } + } + + #[test] + fn broker_issues_and_verifies_profile_source_archive_handles() { + let broker = BackendResourceBroker::default(); + let runtime_id = RuntimeId::new("runtime-test").unwrap(); + let handle = broker.issue_profile_source_archive_handle( + "workspace-test", + Some(&runtime_id), + None, + archive(), + ); + let response = broker + .fetch_profile_source_archive(BackendResourceFetchRequest { + handle: handle.clone(), + runtime_id: runtime_id.as_str().to_string(), + worker_id: None, + audit_correlation_id: handle.audit_correlation_id.clone(), + }) + .expect("fetch succeeds"); + assert_eq!(response.digest, handle.digest); + assert_eq!(response.content_type, PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE); + } + + #[test] + fn broker_rejects_runtime_mismatch() { + let broker = BackendResourceBroker::default(); + let handle = broker.issue_profile_source_archive_handle( + "workspace-test", + Some(&RuntimeId::new("runtime-a").unwrap()), + None, + archive(), + ); + let err = broker + .fetch_profile_source_archive(BackendResourceFetchRequest { + handle: handle.clone(), + runtime_id: RuntimeId::new("runtime-b").unwrap().as_str().to_string(), + worker_id: None, + audit_correlation_id: handle.audit_correlation_id.clone(), + }) + .unwrap_err(); + assert!(matches!(err, BackendResourceError::Unauthorized { .. })); + } +} diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index b0f1ec80..6100f7c9 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -12,7 +12,8 @@ use chrono::{SecondsFormat, Utc}; use futures::StreamExt; use serde::{Deserialize, Serialize}; use tokio::net::TcpListener; -use worker_runtime::worker_backend::WorkerRuntimeExecutionBackend; +use worker_runtime::resource::{BackendResourceError, BackendResourceFetchRequest}; +use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend}; use worker_runtime::working_directory::{ LocalGitWorktreeMaterializer, WorkingDirectoryDiagnostic, WorkingDirectoryMaterializer, }; @@ -42,6 +43,7 @@ use crate::repositories::{ ConfiguredRepository, RepositoryListProjection, RepositoryLogRead, RepositoryLookupError, RepositoryRegistryReader, RepositorySummary, }; +use crate::resource_broker::BackendResourceBroker; use crate::store::{ControlPlaneStore, WorkspaceRecord}; use crate::{Error, Result}; use worker_runtime::catalog::{ @@ -154,6 +156,7 @@ pub struct WorkspaceApi { runtime: Arc, companion: Arc, observation_proxy: BackendObservationProxy, + resource_broker: BackendResourceBroker, working_directory_materializer: Arc, } @@ -162,21 +165,46 @@ impl WorkspaceApi { let materializer = Arc::new(LocalGitWorktreeMaterializer::new( config.embedded_runtime_store_root.clone(), )); - let execution_backend = - WorkerRuntimeExecutionBackend::from_workspace(config.workspace_root.clone()) - .map_err(|err| { - crate::Error::Store(format!( - "failed to initialize embedded Worker backend: {err}" - )) - })? - .with_working_directory_materializer((*materializer).clone()); - Self::new_with_execution_backend(config, store, Arc::new(execution_backend)).await + let resource_broker = BackendResourceBroker::default(); + let execution_backend = WorkerRuntimeExecutionBackend::new( + ProfileRuntimeWorkerFactory::new(config.workspace_root.clone()) + .with_resource_client(Arc::new(resource_broker.clone())), + ) + .map_err(|err| { + crate::Error::Store(format!( + "failed to initialize embedded Worker backend: {err}" + )) + })? + .with_working_directory_materializer((*materializer).clone()); + Self::new_with_execution_backend_and_broker( + config, + store, + Arc::new(execution_backend), + resource_broker, + ) + .await } + #[cfg(test)] async fn new_with_execution_backend( config: ServerConfig, store: Arc, execution_backend: Arc, + ) -> Result { + Self::new_with_execution_backend_and_broker( + config, + store, + execution_backend, + BackendResourceBroker::default(), + ) + .await + } + + async fn new_with_execution_backend_and_broker( + config: ServerConfig, + store: Arc, + execution_backend: Arc, + resource_broker: BackendResourceBroker, ) -> Result { store .upsert_workspace(&WorkspaceRecord { @@ -193,13 +221,17 @@ impl WorkspaceApi { config.embedded_runtime_store_root.clone(), execution_backend, ) + .map(|runtime| runtime.with_resource_broker(resource_broker.clone())) .map_err(|err| { crate::Error::Store(format!("invalid embedded Worker backend: {err}")) })?, ); for remote_config in config.remote_runtime_sources.iter().cloned() { - runtime - .register(RemoteWorkerRuntime::new(remote_config).map_err(|err| err.into_error())?); + runtime.register( + RemoteWorkerRuntime::new(remote_config) + .map(|host| host.with_resource_broker(resource_broker.clone())) + .map_err(|err| err.into_error())?, + ); } let runtime = Arc::new(runtime); let companion = Arc::new(CompanionConsole::new(runtime.clone())); @@ -214,6 +246,7 @@ impl WorkspaceApi { runtime, companion, observation_proxy, + resource_broker, working_directory_materializer, }) } @@ -328,6 +361,10 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/settings/runtime-connections/remotes/{runtime_id}/test", post(scoped_test_remote_runtime_connection), ) + .route( + "/internal/runtime/resources/fetch", + post(post_internal_runtime_resource_fetch), + ) .route("/api/companion/status", get(get_companion_status)) .route( "/api/w/{workspace_id}/companion/status", @@ -1445,7 +1482,9 @@ async fn add_remote_runtime_connection( vec![diagnostic], ) })?; - let active_runtime = RemoteWorkerRuntime::new(active_config).map_err(|err| err.into_error())?; + let active_runtime = RemoteWorkerRuntime::new(active_config) + .map(|host| host.with_resource_broker(api.resource_broker.clone())) + .map_err(|err| err.into_error())?; local_config.runtimes.remote.push(remote_config); write_workspace_backend_config_for_settings(&api, &local_config)?; api.runtime.register_or_replace(active_runtime); @@ -1696,6 +1735,33 @@ async fn create_workspace_worker( })) } +async fn post_internal_runtime_resource_fetch( + State(api): State, + Json(request): Json, +) -> std::result::Result< + Json, + (StatusCode, Json), +> { + api.resource_broker + .fetch_profile_source_archive(request) + .map(Json) + .map_err(|error| (backend_resource_error_status(&error), Json(error))) +} + +fn backend_resource_error_status(error: &BackendResourceError) -> StatusCode { + match error { + BackendResourceError::Expired => StatusCode::GONE, + BackendResourceError::Unauthorized { .. } => StatusCode::UNAUTHORIZED, + BackendResourceError::MissingResource => StatusCode::NOT_FOUND, + BackendResourceError::UnsupportedKind + | BackendResourceError::DigestMismatch { .. } + | BackendResourceError::Oversized { .. } + | BackendResourceError::ContentTypeMismatch { .. } + | BackendResourceError::InvalidResponse { .. } => StatusCode::BAD_REQUEST, + BackendResourceError::Transport { .. } => StatusCode::BAD_GATEWAY, + } +} + async fn get_companion_status( State(api): State, ) -> ApiResult> { @@ -3406,6 +3472,7 @@ mod tests { }], declarations: Vec::new(), profile_source_archive: None, + profile_source_archive_handle: None, } .with_computed_digest() } diff --git a/package.nix b/package.nix index 0b196b78..abe5476c 100644 --- a/package.nix +++ b/package.nix @@ -43,7 +43,7 @@ rustPlatform.buildRustPackage rec { filter = sourceFilter; }; - cargoHash = "sha256-g9Bh9qQ1fGZtzsjO2/vc4jZdrH5YoCR0rOgs//xUDrU="; + cargoHash = "sha256-Pk6pRrtQlMAw2Shl+nQyX81FHQsHlKYAsxtg0M4Ya8Y="; depsExtraArgs = { # Older fetchCargoVendor utilities used crates.io's API download endpoint,