Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

merge: backend resource fetch api

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-08 20:53:12 +09:00
commit 01f94b7591
No known key found for this signature in database
16 changed files with 1346 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -5936,6 +5936,7 @@ dependencies = [
"llm-engine",
"manifest",
"protocol",
"reqwest",
"serde",
"serde_json",
"session-store",

3
crates/worker-runtime/Cargo.toml
View File

@ -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"] }

106
crates/worker-runtime/src/config_bundle.rs
View File

@ -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<ConfigDeclaration>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub profile_source_archive: Option<ProfileSourceArchive>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub profile_source_archive_handle: Option<BackendResourceHandle>,
}
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:<hex>".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"] {

2
crates/worker-runtime/src/http_server.rs
View File

@ -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()
}

1
crates/worker-runtime/src/lib.rs
View File

@ -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;

21
crates/worker-runtime/src/main.rs
View File

@ -78,6 +78,14 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
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<PathBuf>,
worker_metadata_dir: Option<PathBuf>,
worker_runtime_base_dir: Option<PathBuf>,
backend_resource_endpoint: Option<String>,
backend_resource_token: Option<String>,
profile: Option<String>,
}
@ -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 <PATH> Worker session store directory\n\
--worker-metadata-dir <PATH> Worker metadata directory\n\
--worker-runtime-base-dir <PATH> Worker controller runtime directory\n\
--backend-resource-endpoint <URL> Internal Backend resource fetch endpoint for resource handles\n\
--backend-resource-token <TOKEN> Optional bearer token for the Backend resource fetch endpoint\n\
--store <memory|fs> Runtime catalog store selection (default: memory)\n\
--fs-root <PATH> Runtime catalog filesystem store root\n\
--local-token <TOKEN> Minimal local bearer token placeholder\n\

2
crates/worker-runtime/src/profile_archive.rs
View File

@ -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() {

336
crates/worker-runtime/src/resource.rs Normal file
View File

@ -0,0 +1,336 @@
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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub runtime_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub worker_id: Option<String>,
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<u64>,
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<crate::profile_archive::ProfileSourceGraphSummary>,
}
#[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<String>,
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<u8>,
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<BackendResourceFetchResponse, BackendResourceError>;
}
#[cfg(feature = "http-server")]
#[derive(Clone, Debug)]
pub struct HttpBackendResourceClient {
endpoint: String,
bearer_token: Option<String>,
client: reqwest::Client,
}
#[cfg(feature = "http-server")]
impl HttpBackendResourceClient {
pub fn new(endpoint: impl Into<String>, bearer_token: Option<String>) -> 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<BackendResourceFetchResponse, BackendResourceError> {
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::<BackendResourceFetchResponse>()
.await
.map_err(|err| BackendResourceError::InvalidResponse {
message: err.to_string(),
})
} else {
let status = response.status();
match response.json::<BackendResourceError>().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<HashMap<String, ProfileSourceArchive>>,
}
impl ProfileSourceArchiveCache {
pub fn get(&self, digest: &str) -> Option<ProfileSourceArchive> {
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<ProfileSourceArchive, BackendResourceError> {
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(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::profile_archive::ProfileSourceGraphSummary;
use std::collections::BTreeMap;
fn graph() -> ProfileSourceGraphSummary {
ProfileSourceGraphSummary {
source_count: 1,
total_source_bytes: 4,
entrypoints: BTreeMap::from([(
"default".to_string(),
"profiles/default.dcdl".to_string(),
)]),
import_count: 0,
}
}
fn handle_for(bytes: &[u8]) -> BackendResourceHandle {
BackendResourceHandle {
kind: BackendResourceKind::ProfileSourceArchive,
workspace_id: "workspace-test".to_string(),
scope_id: Some("workspace-profile-source".to_string()),
runtime_id: Some("runtime-test".to_string()),
worker_id: Some("worker-test".to_string()),
resource_id: "profile-source-archive:test".to_string(),
digest: sha256_hex(bytes),
operation: BackendResourceOperation::FetchArchive,
expires_at_unix_seconds: 4_102_444_800,
nonce: "nonce-test".to_string(),
revision: sha256_hex(bytes),
generation: Some(1),
max_bytes: 1024,
content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
redaction: ResourceRedactionPolicy::RuntimeInternalOnly,
audit_correlation_id: "audit-test".to_string(),
profile_source_graph: Some(graph()),
}
}
#[test]
fn response_verification_detects_digest_mismatch() {
let bytes = b"archive-bytes";
let handle = handle_for(bytes);
let error = profile_source_archive_from_response(
&handle,
BackendResourceFetchResponse {
kind: BackendResourceKind::ProfileSourceArchive,
resource_id: handle.resource_id.clone(),
digest: handle.digest.clone(),
content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
bytes: b"tampered-bytes".to_vec(),
audit_correlation_id: handle.audit_correlation_id.clone(),
},
)
.unwrap_err();
assert!(matches!(error, BackendResourceError::DigestMismatch { .. }));
}
#[test]
fn response_verification_rejects_oversized_bytes() {
let bytes = b"archive-bytes";
let mut handle = handle_for(bytes);
handle.max_bytes = 2;
let error = profile_source_archive_from_response(
&handle,
BackendResourceFetchResponse {
kind: BackendResourceKind::ProfileSourceArchive,
resource_id: handle.resource_id.clone(),
digest: handle.digest.clone(),
content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
bytes: bytes.to_vec(),
audit_correlation_id: handle.audit_correlation_id.clone(),
},
)
.unwrap_err();
assert!(matches!(error, BackendResourceError::Oversized { .. }));
}
}

1
crates/worker-runtime/src/runtime.rs
View File

@ -1628,6 +1628,7 @@ mod tests {
reference: "capability:read".to_string(),
}],
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}

229
crates/worker-runtime/src/worker_backend.rs
View File

@ -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<PathBuf>,
profile: Option<String>,
runtime_base_dir: Option<PathBuf>,
resource_client: Option<Arc<dyn BackendResourceClient>>,
profile_archive_cache: Arc<ProfileSourceArchiveCache>,
}
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<dyn BackendResourceClient>) -> Self {
self.resource_client = Some(resource_client);
self
}
fn store_dir(&self) -> Result<PathBuf, String> {
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<crate::profile_archive::VerifiedProfileSourceArchive, String> {
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
)
})?;
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 fetched_archive = profile_source_archive_from_response(&handle, response)
.map_err(format_backend_resource_error)?;
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}"));
}
self.profile_archive_cache.insert(fetched_archive.clone());
fetched_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}"))?;
@ -643,6 +701,7 @@ where
#[cfg(test)]
mod tests {
use super::*;
use std::collections::{BTreeMap, VecDeque};
use std::fs;
use std::pin::Pin;
use std::process::Command;
@ -653,6 +712,7 @@ mod tests {
ConfigBundleRef, CreateWorkerRequest, DirtyStatePolicy, MaterializerKind, ProfileSelector,
RepositorySelector, WorkingDirectoryRepository, WorkingDirectoryRequest,
};
use crate::execution::WorkerExecutionContext;
use crate::identity::RuntimeId;
use crate::management::RuntimeOptions;
use crate::observation::{TranscriptQuery, TranscriptRole};
@ -791,10 +851,130 @@ mod tests {
}],
declarations: Vec::new(),
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}
fn sample_profile_archive() -> crate::profile_archive::ProfileSourceArchive {
let entrypoints =
BTreeMap::from([("default".to_string(), "profiles/default.dcdl".to_string())]);
let sources = BTreeMap::from([(
"profiles/default.dcdl".to_string(),
r#"{
slug = "default";
description = "Default";
scope = "workspace_read";
}"#
.to_string(),
)]);
crate::profile_archive::ProfileSourceArchive::build(
crate::profile_archive::ProfileSourceArchiveInput {
id: "profile-source-archive:test".to_string(),
entrypoints,
imports: BTreeMap::new(),
sources,
},
)
.unwrap()
}
fn handle_for_archive(
archive: &crate::profile_archive::ProfileSourceArchive,
) -> crate::resource::BackendResourceHandle {
crate::resource::BackendResourceHandle {
kind: crate::resource::BackendResourceKind::ProfileSourceArchive,
workspace_id: "workspace-test".to_string(),
scope_id: Some("workspace-profile-source".to_string()),
runtime_id: Some("runtime-test".to_string()),
worker_id: Some("worker-test".to_string()),
resource_id: archive.reference.id.clone(),
digest: archive.reference.digest.clone(),
operation: crate::resource::BackendResourceOperation::FetchArchive,
expires_at_unix_seconds: 4_102_444_800,
nonce: "nonce-test".to_string(),
revision: archive.reference.digest.clone(),
generation: Some(1),
max_bytes: crate::resource::DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES,
content_type: crate::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
redaction: crate::resource::ResourceRedactionPolicy::RuntimeInternalOnly,
audit_correlation_id: "audit-test".to_string(),
profile_source_graph: Some(archive.reference.source_graph.clone()),
}
}
fn response_for_archive(
handle: &crate::resource::BackendResourceHandle,
archive: &crate::profile_archive::ProfileSourceArchive,
) -> crate::resource::BackendResourceFetchResponse {
crate::resource::BackendResourceFetchResponse {
kind: crate::resource::BackendResourceKind::ProfileSourceArchive,
resource_id: handle.resource_id.clone(),
digest: handle.digest.clone(),
content_type: crate::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
bytes: archive.content.clone(),
audit_correlation_id: handle.audit_correlation_id.clone(),
}
}
#[derive(Clone)]
struct SequencedResourceClient {
responses: Arc<
Mutex<
VecDeque<
Result<
crate::resource::BackendResourceFetchResponse,
crate::resource::BackendResourceError,
>,
>,
>,
>,
call_count: Arc<AtomicUsize>,
}
#[async_trait]
impl crate::resource::BackendResourceClient for SequencedResourceClient {
async fn fetch_resource(
&self,
_request: crate::resource::BackendResourceFetchRequest,
) -> Result<
crate::resource::BackendResourceFetchResponse,
crate::resource::BackendResourceError,
> {
self.call_count.fetch_add(1, Ordering::SeqCst);
self.responses
.lock()
.unwrap()
.pop_front()
.expect("missing sequenced resource response")
}
}
fn spawn_request_with_bundle(
bundle: crate::config_bundle::ConfigBundle,
) -> WorkerExecutionSpawnRequest {
let worker_ref = crate::identity::WorkerRef::new(
RuntimeId::new("runtime-test").unwrap(),
crate::identity::WorkerId::new("worker-test").unwrap(),
);
WorkerExecutionSpawnRequest {
worker_ref: worker_ref.clone(),
request: CreateWorkerRequest {
profile: ProfileSelector::RuntimeDefault,
config_bundle: ConfigBundleRef {
id: bundle.metadata.id.clone(),
digest: bundle.metadata.digest.clone(),
},
initial_input: None,
working_directory_request: None,
working_directory: None,
},
context: WorkerExecutionContext::new(worker_ref, Arc::new(|_, _| panic!("unused"))),
working_directory: None,
config_bundle: Some(bundle),
}
}
fn create_request(_name: &str) -> CreateWorkerRequest {
let bundle = test_bundle();
CreateWorkerRequest {
@ -847,6 +1027,37 @@ mod tests {
}
}
#[tokio::test]
async fn cached_profile_archive_still_requires_backend_authorization() {
let archive = sample_profile_archive();
let handle = handle_for_archive(&archive);
let call_count = Arc::new(AtomicUsize::new(0));
let client = SequencedResourceClient {
responses: Arc::new(Mutex::new(VecDeque::from([
Ok(response_for_archive(&handle, &archive)),
Err(crate::resource::BackendResourceError::Expired),
]))),
call_count: call_count.clone(),
};
let factory = ProfileRuntimeWorkerFactory::new(tempfile::tempdir().unwrap().path())
.with_resource_client(Arc::new(client));
let mut bundle = test_bundle();
bundle.profile_source_archive_handle = Some(handle);
bundle = bundle.with_computed_digest();
factory
.resolve_profile_source_archive(&bundle, &spawn_request_with_bundle(bundle.clone()))
.await
.expect("first fetch should authorize and cache archive");
let err = factory
.resolve_profile_source_archive(&bundle, &spawn_request_with_bundle(bundle.clone()))
.await
.unwrap_err();
assert!(err.contains("expired"), "unexpected error: {err}");
assert_eq!(call_count.load(Ordering::SeqCst), 2);
}
#[test]
fn builtin_profile_selector_is_not_double_prefixed() {
assert_eq!(

1
crates/workspace-server/src/companion.rs
View File

@ -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()
}

102
crates/workspace-server/src/hosts.rs
View File

@ -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<str>, 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<ConfigBundle, String> {
fn default_embedded_config_bundle(
profile: &ProfileSelector,
workspace_id: &str,
runtime_id: Option<&EmbeddedRuntimeId>,
resource_broker: &BackendResourceBroker,
) -> Result<ConfigBundle, String> {
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<ConfigBun
label: embedded_profile_label(profile),
}],
declarations: Vec::new(),
profile_source_archive: Some(default_profile_source_archive(profile)?),
profile_source_archive: None,
profile_source_archive_handle: Some(handle),
}
.with_computed_digest())
}
@ -2967,6 +3008,8 @@ mod tests {
#[test]
fn embedded_builtin_decodal_profiles_resolve_through_archive() {
let root = tempfile::tempdir().unwrap();
let broker = BackendResourceBroker::default();
let runtime_id = EmbeddedRuntimeId::new("runtime-test".to_string()).unwrap();
for selector in [
ProfileSelector::RuntimeDefault,
ProfileSelector::Builtin("builtin:companion".to_string()),
@ -2975,13 +3018,30 @@ mod tests {
ProfileSelector::Builtin("builtin:coder".to_string()),
ProfileSelector::Builtin("builtin:reviewer".to_string()),
] {
let bundle = default_embedded_config_bundle(&selector).unwrap();
let archive = bundle
.profile_source_archive
.as_ref()
.unwrap()
.verify()
let bundle = default_embedded_config_bundle(
&selector,
"workspace-test",
Some(&runtime_id),
&broker,
)
.unwrap();
let handle = bundle.profile_source_archive_handle.as_ref().unwrap();
assert!(bundle.profile_source_archive.is_none());
let response = broker
.fetch_profile_source_archive(
worker_runtime::resource::BackendResourceFetchRequest {
handle: handle.clone(),
runtime_id: runtime_id.as_str().to_string(),
worker_id: None,
audit_correlation_id: handle.audit_correlation_id.clone(),
},
)
.unwrap();
let archive =
worker_runtime::resource::profile_source_archive_from_response(handle, response)
.unwrap()
.verify()
.unwrap();
let selector_key = match &selector {
ProfileSelector::RuntimeDefault => "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()
}

1
crates/workspace-server/src/lib.rs
View File

@ -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;

336
crates/workspace-server/src/resource_broker.rs Normal file
View File

@ -0,0 +1,336 @@
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<Mutex<HashMap<String, StoredResource>>>,
}
#[derive(Clone)]
struct StoredResource {
runtime_id: Option<String>,
worker_id: Option<String>,
handle: BackendResourceHandle,
archive: ProfileSourceArchive,
}
impl BackendResourceBroker {
pub fn issue_profile_source_archive_handle(
&self,
workspace_id: impl Into<String>,
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 {
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<BackendResourceFetchResponse, BackendResourceError> {
verify_handle_shape(&request.handle)?;
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)?;
verify_handle_shape(&stored.handle)?;
if stored.handle.expires_at_unix_seconds < Utc::now().timestamp() {
return Err(BackendResourceError::Expired);
}
let actual_bytes = stored.archive.content.len() as u64;
if actual_bytes > stored.handle.max_bytes {
return Err(BackendResourceError::Oversized {
max_bytes: stored.handle.max_bytes,
actual_bytes,
});
}
if request.handle != stored.handle {
return Err(BackendResourceError::Unauthorized {
message: "resource handle does not match broker-issued handle".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(),
});
}
}
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<BackendResourceFetchResponse, BackendResourceError> {
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, WorkerId};
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,
}
}
fn archive_with_len(len: usize) -> ProfileSourceArchive {
let content = vec![b'x'; len];
let mut entrypoints = BTreeMap::new();
entrypoints.insert("default".to_string(), "profiles/default.dcdl".to_string());
ProfileSourceArchive {
reference: ProfileSourceArchiveRef {
id: format!("profile-source-archive:test-{len}"),
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,
}
}
fn request(
handle: BackendResourceHandle,
runtime_id: &RuntimeId,
worker_id: Option<&WorkerId>,
) -> BackendResourceFetchRequest {
BackendResourceFetchRequest {
audit_correlation_id: handle.audit_correlation_id.clone(),
handle,
runtime_id: runtime_id.as_str().to_string(),
worker_id: worker_id.map(|id| id.as_str().to_string()),
}
}
#[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 runtime_a = RuntimeId::new("runtime-a").unwrap();
let handle = broker.issue_profile_source_archive_handle(
"workspace-test",
Some(&runtime_a),
None,
archive(),
);
let err = broker
.fetch_profile_source_archive(request(
handle,
&RuntimeId::new("runtime-b").unwrap(),
None,
))
.unwrap_err();
assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
}
#[test]
fn broker_rejects_worker_mismatch() {
let broker = BackendResourceBroker::default();
let runtime_id = RuntimeId::new("runtime-test").unwrap();
let worker_a = WorkerId::new("worker-a").unwrap();
let worker_b = WorkerId::new("worker-b").unwrap();
let handle = broker.issue_profile_source_archive_handle(
"workspace-test",
Some(&runtime_id),
Some(&worker_a),
archive(),
);
let err = broker
.fetch_profile_source_archive(request(handle, &runtime_id, Some(&worker_b)))
.unwrap_err();
assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
}
#[test]
fn broker_rejects_expiry_extension_from_request_handle() {
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(),
);
broker
.resources
.lock()
.unwrap()
.get_mut(&handle.nonce)
.unwrap()
.handle
.expires_at_unix_seconds = 1;
let mut extended = handle;
extended.expires_at_unix_seconds = 4_102_444_800;
let err = broker
.fetch_profile_source_archive(request(extended, &runtime_id, None))
.unwrap_err();
assert!(matches!(err, BackendResourceError::Expired));
}
#[test]
fn broker_rejects_policy_tampered_request_handle() {
let broker = BackendResourceBroker::default();
let runtime_id = RuntimeId::new("runtime-test").unwrap();
let mut handle = broker.issue_profile_source_archive_handle(
"workspace-test",
Some(&runtime_id),
None,
archive(),
);
handle.scope_id = Some("tampered-scope".to_string());
let err = broker
.fetch_profile_source_archive(request(handle, &runtime_id, None))
.unwrap_err();
assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
}
#[test]
fn broker_uses_stored_max_bytes_when_request_handle_is_tampered() {
let broker = BackendResourceBroker::default();
let runtime_id = RuntimeId::new("runtime-test").unwrap();
let archive = archive_with_len((DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES + 1) as usize);
let mut handle = broker.issue_profile_source_archive_handle(
"workspace-test",
Some(&runtime_id),
None,
archive,
);
handle.max_bytes = DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES + 1024;
let err = broker
.fetch_profile_source_archive(request(handle, &runtime_id, None))
.unwrap_err();
assert!(matches!(err, BackendResourceError::Oversized { .. }));
}
}

247
crates/workspace-server/src/server.rs
View File

@ -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<RuntimeRegistry>,
companion: Arc<CompanionConsole>,
observation_proxy: BackendObservationProxy,
resource_broker: BackendResourceBroker,
working_directory_materializer: Arc<LocalGitWorktreeMaterializer>,
}
@ -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<dyn ControlPlaneStore>,
execution_backend: Arc<dyn worker_runtime::execution::WorkerExecutionBackend>,
) -> Result<Self> {
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<dyn ControlPlaneStore>,
execution_backend: Arc<dyn worker_runtime::execution::WorkerExecutionBackend>,
resource_broker: BackendResourceBroker,
) -> Result<Self> {
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<WorkspaceApi>,
Json(request): Json<BackendResourceFetchRequest>,
) -> std::result::Result<
Json<worker_runtime::resource::BackendResourceFetchResponse>,
(StatusCode, Json<BackendResourceError>),
> {
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<WorkspaceApi>,
) -> ApiResult<Json<CompanionStatusResponse>> {
@ -3203,6 +3269,7 @@ mod tests {
use tokio_tungstenite::connect_async;
use tokio_tungstenite::tungstenite::Message;
use tower::ServiceExt;
use worker_runtime::resource::BackendResourceClient;
use crate::hosts::{
TicketWorkerRole, WorkerInputKind, WorkerOperationState, WorkerSpawnAcceptanceRequirement,
@ -3375,16 +3442,161 @@ mod tests {
}
}
async fn test_app(workspace_root: impl Into<PathBuf>) -> Router {
async fn test_api(workspace_root: impl Into<PathBuf>) -> WorkspaceApi {
let store = SqliteWorkspaceStore::in_memory().unwrap();
let api = WorkspaceApi::new_with_execution_backend(
WorkspaceApi::new_with_execution_backend(
test_server_config(workspace_root),
Arc::new(store),
Arc::new(DeterministicExecutionBackend::default()),
)
.await
.unwrap();
build_router(api)
.unwrap()
}
async fn test_app(workspace_root: impl Into<PathBuf>) -> Router {
build_router(test_api(workspace_root).await)
}
fn test_profile_archive() -> worker_runtime::profile_archive::ProfileSourceArchive {
use worker_runtime::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveInput};
ProfileSourceArchive::build(ProfileSourceArchiveInput {
id: "profile-source-archive:server-test".to_string(),
entrypoints: std::collections::BTreeMap::from([(
"default".to_string(),
"profiles/default.dcdl".to_string(),
)]),
imports: std::collections::BTreeMap::new(),
sources: std::collections::BTreeMap::from([(
"profiles/default.dcdl".to_string(),
r#"{
slug = "default";
description = "Default";
scope = "workspace_read";
}"#
.to_string(),
)]),
})
.unwrap()
}
fn missing_resource_handle() -> worker_runtime::resource::BackendResourceHandle {
worker_runtime::resource::BackendResourceHandle {
kind: worker_runtime::resource::BackendResourceKind::ProfileSourceArchive,
workspace_id: "workspace-test".to_string(),
scope_id: Some("workspace-profile-source".to_string()),
runtime_id: Some("runtime-test".to_string()),
worker_id: None,
resource_id: "profile-source-archive:missing".to_string(),
digest: "sha256:0000000000000000000000000000000000000000000000000000000000000000"
.to_string(),
operation: worker_runtime::resource::BackendResourceOperation::FetchArchive,
expires_at_unix_seconds: 4_102_444_800,
nonce: "missing-nonce".to_string(),
revision: "missing-revision".to_string(),
generation: None,
max_bytes: worker_runtime::resource::DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES,
content_type: worker_runtime::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
redaction: worker_runtime::resource::ResourceRedactionPolicy::RuntimeInternalOnly,
audit_correlation_id: "audit-missing".to_string(),
profile_source_graph: Some(
worker_runtime::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: 0,
},
),
}
}
#[tokio::test]
async fn internal_resource_fetch_rest_returns_typed_missing_resource() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let app = test_app(workspace.path()).await;
let handle = missing_resource_handle();
let response = app
.oneshot(
Request::post("/internal/runtime/resources/fetch")
.header("content-type", "application/json")
.body(Body::from(
serde_json::to_vec(
&worker_runtime::resource::BackendResourceFetchRequest {
audit_correlation_id: handle.audit_correlation_id.clone(),
runtime_id: "runtime-test".to_string(),
worker_id: None,
handle,
},
)
.unwrap(),
))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let error: worker_runtime::resource::BackendResourceError =
serde_json::from_slice(&bytes).unwrap();
assert!(matches!(
error,
worker_runtime::resource::BackendResourceError::MissingResource
));
}
#[tokio::test]
async fn remote_http_resource_fetch_uses_backend_resource_contract() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let broker = api.resource_broker.clone();
let archive = test_profile_archive();
let runtime_id = worker_runtime::identity::RuntimeId::new("runtime-test").unwrap();
let handle = broker.issue_profile_source_archive_handle(
"workspace-test",
Some(&runtime_id),
None,
archive,
);
let app = build_router(api);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = worker_runtime::resource::HttpBackendResourceClient::new(
format!("http://{addr}/internal/runtime/resources/fetch"),
None,
);
let response = client
.fetch_resource(worker_runtime::resource::BackendResourceFetchRequest {
audit_correlation_id: handle.audit_correlation_id.clone(),
runtime_id: runtime_id.as_str().to_string(),
worker_id: None,
handle: handle.clone(),
})
.await
.expect("remote HTTP resource fetch succeeds");
assert_eq!(response.digest, handle.digest);
let mut tampered = handle;
tampered.scope_id = Some("tampered".to_string());
let error = client
.fetch_resource(worker_runtime::resource::BackendResourceFetchRequest {
audit_correlation_id: tampered.audit_correlation_id.clone(),
runtime_id: runtime_id.as_str().to_string(),
worker_id: None,
handle: tampered,
})
.await
.unwrap_err();
assert!(matches!(
error,
worker_runtime::resource::BackendResourceError::Unauthorized { .. }
));
server.abort();
}
fn runtime_test_bundle() -> worker_runtime::config_bundle::ConfigBundle {
@ -3406,6 +3618,7 @@ mod tests {
}],
declarations: Vec::new(),
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}

2
package.nix
View File

@ -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,