diff --git a/crates/worker-runtime/src/error.rs b/crates/worker-runtime/src/error.rs index 6392f2ce..1de44f83 100644 --- a/crates/worker-runtime/src/error.rs +++ b/crates/worker-runtime/src/error.rs @@ -39,6 +39,15 @@ pub enum RuntimeError { #[error("invalid request: {0}")] InvalidRequest(String), + #[error( + "workspace `{workspace_id}` is owned by Runtime server `{owner_server_id}`, not `{requester_server_id}`" + )] + WorkspaceOwnerMismatch { + workspace_id: String, + owner_server_id: String, + requester_server_id: String, + }, + #[error(transparent)] WorkingDirectory(#[from] WorkingDirectoryDiagnostic), diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 6ec27e70..6e67181c 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -292,6 +292,7 @@ pub(crate) struct PersistedRuntimeState { pub(crate) next_event_id: u64, pub(crate) next_diagnostic_id: u64, pub(crate) workers: BTreeMap, + pub(crate) workspace_owners: BTreeMap, pub(crate) config_bundles: BTreeMap, pub(crate) events: Vec, pub(crate) diagnostics: Vec, @@ -319,6 +320,8 @@ struct RuntimeSnapshot { next_diagnostic_id: u64, #[serde(default)] config_bundles: BTreeMap, + #[serde(default)] + workspace_owners: BTreeMap, diagnostics: Vec, } @@ -350,6 +353,7 @@ impl RuntimeSnapshot { next_event_id: state.next_event_id, next_diagnostic_id: state.next_diagnostic_id, config_bundles: state.config_bundles.clone(), + workspace_owners: state.workspace_owners.clone(), diagnostics: state.diagnostics.clone(), } } @@ -389,6 +393,7 @@ impl RuntimeSnapshot { next_diagnostic_id: self.next_diagnostic_id, workers, config_bundles: self.config_bundles, + workspace_owners: self.workspace_owners, events, diagnostics: self.diagnostics, } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 548344cc..fdb68012 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -6,7 +6,6 @@ //! Runtime process directly; a backend is expected to own any browser-facing //! credentials, registration, and policy. -use crate::Runtime; use crate::auth::{ RuntimeAuthContext, RuntimeAuthError, RuntimeHttpAuthConfig, unix_now_seconds, verify_capability_token, @@ -22,6 +21,7 @@ use crate::interaction::{WorkerInput, WorkerInteractionAck}; use crate::management::{RuntimeLimits, RuntimeSummary, WorkerDeleteResult}; #[cfg(feature = "ws-server")] use crate::observation::WorkerObservationCursor; +use crate::{Runtime, RuntimeWorkspaceScope}; use axum::body::{Body, Bytes}; use axum::extract::rejection::{JsonRejection, QueryRejection}; #[cfg(feature = "ws-server")] @@ -400,15 +400,15 @@ async fn list_workers( query: Result, QueryRejection>, ) -> RestResult { let Query(query) = query.map_err(RuntimeHttpRestError::query_rejection)?; - let workspace_id = auth_workspace_id(&state, auth.as_ref())?; - let workers = match (query.status, workspace_id) { - (Some(RuntimeHttpWorkerStatusFilter::Stopped), Some(workspace_id)) => { - state.runtime.list_stopped_workers_scoped(workspace_id) + let scope = auth_workspace_scope(&state, auth.as_ref())?; + let workers = match (query.status, scope.as_ref()) { + (Some(RuntimeHttpWorkerStatusFilter::Stopped), Some(scope)) => { + state.runtime.list_stopped_workers_scoped(scope) } (Some(RuntimeHttpWorkerStatusFilter::Stopped), None) => { state.runtime.list_stopped_workers() } - (None, Some(workspace_id)) => state.runtime.list_workers_scoped(workspace_id), + (None, Some(scope)) => state.runtime.list_workers_scoped(scope), (None, None) => state.runtime.list_workers(), } .map_err(RuntimeHttpRestError::runtime)?; @@ -473,10 +473,8 @@ async fn get_worker( Path(worker_id): Path, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let worker = match auth_workspace_id(&state, auth.as_ref())? { - Some(workspace_id) => state - .runtime - .worker_detail_scoped(workspace_id, &worker_ref), + let worker = match auth_workspace_scope(&state, auth.as_ref())? { + Some(scope) => state.runtime.worker_detail_scoped(&scope, &worker_ref), None => state.runtime.worker_detail(&worker_ref), } .map_err(RuntimeHttpRestError::runtime)?; @@ -489,10 +487,8 @@ async fn delete_worker( Path(worker_id): Path, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let worker = match auth_workspace_id(&state, auth.as_ref())? { - Some(workspace_id) => state - .runtime - .delete_worker_scoped(workspace_id, &worker_ref), + let worker = match auth_workspace_scope(&state, auth.as_ref())? { + Some(scope) => state.runtime.delete_worker_scoped(&scope, &worker_ref), None => state.runtime.delete_worker(&worker_ref), } .map_err(RuntimeHttpRestError::runtime)?; @@ -505,8 +501,8 @@ async fn create_worker( body: Result, JsonRejection>, ) -> RestResult { let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?; - let worker = match auth_workspace_id(&state, auth.as_ref())? { - Some(workspace_id) => state.runtime.create_worker_scoped(workspace_id, request), + let worker = match auth_workspace_scope(&state, auth.as_ref())? { + Some(scope) => state.runtime.create_worker_scoped(&scope, request), None => state.runtime.create_worker(request), } .map_err(RuntimeHttpRestError::runtime)?; @@ -519,10 +515,8 @@ async fn restore_worker( Path(worker_id): Path, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let worker = match auth_workspace_id(&state, auth.as_ref())? { - Some(workspace_id) => state - .runtime - .restore_worker_scoped(workspace_id, &worker_ref), + let worker = match auth_workspace_scope(&state, auth.as_ref())? { + Some(scope) => state.runtime.restore_worker_scoped(&scope, &worker_ref), None => state.runtime.restore_worker(&worker_ref), } .map_err(RuntimeHttpRestError::runtime)?; @@ -538,18 +532,18 @@ async fn worker_protocol_ws( ws: WebSocketUpgrade, ) -> Result { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let workspace_id = auth_workspace_id(&state, auth.as_ref())?.map(ToOwned::to_owned); - match workspace_id.as_deref() { - Some(workspace_id) => state + let scope = auth_workspace_scope(&state, auth.as_ref())?; + match scope.as_ref() { + Some(scope) => state .runtime - .worker_detail_scoped(workspace_id, &worker_ref) + .worker_detail_scoped(scope, &worker_ref) .map(|_| ()), None => state.runtime.worker_detail(&worker_ref).map(|_| ()), } .map_err(RuntimeHttpRestError::runtime)?; Ok(ws .on_upgrade(move |socket| { - worker_protocol_ws_session(state.runtime, workspace_id, worker_ref, query, socket) + worker_protocol_ws_session(state.runtime, scope, worker_ref, query, socket) }) .into_response()) } @@ -557,7 +551,7 @@ async fn worker_protocol_ws( #[cfg(feature = "ws-server")] async fn worker_protocol_ws_session( runtime: Runtime, - workspace_id: Option, + scope: Option, worker_ref: WorkerRef, query: RuntimeWorkerEventsWsQuery, mut socket: WebSocket, @@ -642,12 +636,10 @@ async fn worker_protocol_ws_session( match inbound { Some(Ok(WsMessage::Text(text))) => match decode_method(&text) { Ok(method) => { - let result = match workspace_id.as_deref() { - Some(workspace_id) => runtime.send_protocol_method_scoped( - workspace_id, - &worker_ref, - method, - ), + let result = match scope.as_ref() { + Some(scope) => { + runtime.send_protocol_method_scoped(scope, &worker_ref, method) + } None => runtime.send_protocol_method(&worker_ref, method), }; match result { @@ -746,10 +738,8 @@ async fn send_worker_input( ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; let Json(input) = body.map_err(RuntimeHttpRestError::json_rejection)?; - let ack = match auth_workspace_id(&state, auth.as_ref())? { - Some(workspace_id) => state - .runtime - .send_input_scoped(workspace_id, &worker_ref, input), + let ack = match auth_workspace_scope(&state, auth.as_ref())? { + Some(scope) => state.runtime.send_input_scoped(&scope, &worker_ref, input), None => state.runtime.send_input(&worker_ref, input), } .map_err(RuntimeHttpRestError::runtime)?; @@ -764,9 +754,9 @@ async fn worker_completions( ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?; - let entries = match auth_workspace_id(&state, auth.as_ref())? { - Some(workspace_id) => state.runtime.worker_completions_scoped( - workspace_id, + let entries = match auth_workspace_scope(&state, auth.as_ref())? { + Some(scope) => state.runtime.worker_completions_scoped( + &scope, &worker_ref, request.kind, &request.prefix, @@ -791,12 +781,10 @@ async fn stop_worker( ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; let request = parse_optional_lifecycle_request(body)?; - let ack = match auth_workspace_id(&state, auth.as_ref())? { - Some(workspace_id) => { - state - .runtime - .stop_worker_scoped(workspace_id, &worker_ref, request.reason) - } + let ack = match auth_workspace_scope(&state, auth.as_ref())? { + Some(scope) => state + .runtime + .stop_worker_scoped(&scope, &worker_ref, request.reason), None => state.runtime.stop_worker(&worker_ref, request.reason), } .map_err(RuntimeHttpRestError::runtime)?; @@ -811,12 +799,10 @@ async fn cancel_worker( ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; let request = parse_optional_lifecycle_request(body)?; - let ack = match auth_workspace_id(&state, auth.as_ref())? { - Some(workspace_id) => { - state - .runtime - .cancel_worker_scoped(workspace_id, &worker_ref, request.reason) - } + let ack = match auth_workspace_scope(&state, auth.as_ref())? { + Some(scope) => state + .runtime + .cancel_worker_scoped(&scope, &worker_ref, request.reason), None => state.runtime.cancel_worker(&worker_ref, request.reason), } .map_err(RuntimeHttpRestError::runtime)?; @@ -928,10 +914,10 @@ fn runtime_auth_error_response(error: RuntimeAuthError) -> RuntimeHttpRestError } } -fn auth_workspace_id<'a>( +fn auth_workspace_scope( state: &RuntimeHttpState, - auth: Option<&'a Extension>, -) -> Result, RuntimeHttpRestError> { + auth: Option<&Extension>, +) -> Result, RuntimeHttpRestError> { let Some(Extension(context)) = auth else { if state.auth.is_some() || state.local_token.is_some() { return Err(RuntimeHttpRestError::new( @@ -950,7 +936,15 @@ fn auth_workspace_id<'a>( "Runtime worker operation requires a non-empty workspace scope", )); } - Ok(Some(workspace_id)) + let server_id = context.server_id.trim(); + if server_id.is_empty() { + return Err(RuntimeHttpRestError::new( + StatusCode::FORBIDDEN, + "server_scope_required", + "Runtime worker operation requires a non-empty server scope", + )); + } + Ok(Some(RuntimeWorkspaceScope::new(workspace_id, server_id))) } fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static str> { @@ -1052,6 +1046,7 @@ fn status_for_runtime_error(error: &RuntimeError) -> StatusCode { | RuntimeError::WorkerExecutionUnavailable { .. } | RuntimeError::ExecutionBackendUnavailable { .. } | RuntimeError::WorkerExecutionRejected { .. } => StatusCode::CONFLICT, + RuntimeError::WorkspaceOwnerMismatch { .. } => StatusCode::FORBIDDEN, RuntimeError::LimitTooLarge { .. } | RuntimeError::InvalidRequest(_) | RuntimeError::InvalidInitialInputKind { .. } @@ -1077,6 +1072,7 @@ fn code_for_runtime_error(error: &RuntimeError) -> String { "execution_backend_unavailable".to_string() } RuntimeError::WorkerExecutionRejected { .. } => "worker_execution_rejected".to_string(), + RuntimeError::WorkspaceOwnerMismatch { .. } => "workspace_owner_mismatch".to_string(), RuntimeError::LimitTooLarge { .. } => "limit_too_large".to_string(), RuntimeError::InvalidRequest(_) => "invalid_request".to_string(), RuntimeError::WorkingDirectory(diagnostic) => diagnostic.code.clone(), @@ -1175,6 +1171,35 @@ mod tests { (auth, signer) } + fn auth_config_and_two_signers() -> ( + RuntimeHttpAuthConfig, + CapabilityTokenSigner, + CapabilityTokenSigner, + ) { + let identity_a = RuntimeIdentityMaterial::generate("server-a").unwrap(); + let identity_b = RuntimeIdentityMaterial::generate("server-b").unwrap(); + let signer_a = + CapabilityTokenSigner::new(identity_a.identity_id.clone(), identity_a.private_key); + let signer_b = + CapabilityTokenSigner::new(identity_b.identity_id.clone(), identity_b.private_key); + let auth = RuntimeHttpAuthConfig { + runtime_id: "runtime-test".to_string(), + trusted_servers: vec![ + TrustedServerKey { + server_id: identity_a.identity_id, + public_key: identity_a.public_key, + display_name: None, + }, + TrustedServerKey { + server_id: identity_b.identity_id, + public_key: identity_b.public_key, + display_name: None, + }, + ], + }; + (auth, signer_a, signer_b) + } + fn token_for_workspace(signer: &CapabilityTokenSigner, workspace_id: &str) -> String { token_for_workspace_with_permissions( signer, @@ -1302,6 +1327,42 @@ mod tests { assert_eq!(response.status(), StatusCode::NOT_FOUND); } + #[tokio::test] + async fn capability_workspace_owner_binding_rejects_other_trusted_server() { + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(AcceptingBackend)) + .unwrap(); + let (auth, signer_a, signer_b) = auth_config_and_two_signers(); + let token_a = token_for_workspace(&signer_a, "workspace-a"); + let token_b = token_for_workspace(&signer_b, "workspace-a"); + let app = runtime_http_router_with_auth(runtime, None, auth); + + let create_a = scoped_task_request("a", "workspace-a"); + let response = app + .clone() + .oneshot(bearer_request( + Method::POST, + "/v1/workers", + &token_a, + serde_json::to_vec(&create_a).unwrap(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let create_b = scoped_task_request("b", "workspace-a"); + let response = app + .oneshot(bearer_request( + Method::POST, + "/v1/workers", + &token_b, + serde_json::to_vec(&create_b).unwrap(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + } + #[tokio::test] async fn capability_token_without_workspace_scope_is_forbidden() { let runtime = @@ -1787,7 +1848,10 @@ mod ws_tests { .store_config_bundle(ws_test_bundle(ProfileSelector::RuntimeDefault)) .unwrap(); let worker = runtime - .create_worker_scoped("local", ws_create_request()) + .create_worker_scoped( + &RuntimeWorkspaceScope::new("local", "local-token"), + ws_create_request(), + ) .unwrap(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); diff --git a/crates/worker-runtime/src/lib.rs b/crates/worker-runtime/src/lib.rs index e08c2d86..e6d3adec 100644 --- a/crates/worker-runtime/src/lib.rs +++ b/crates/worker-runtime/src/lib.rs @@ -29,4 +29,4 @@ pub mod working_directory; #[cfg(feature = "fs-store")] pub use fs_store::{FsRuntimeStore, FsRuntimeStoreOptions}; pub use management::{RuntimeLimits, RuntimeOptions}; -pub use runtime::Runtime; +pub use runtime::{Runtime, RuntimeWorkspaceScope}; diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 8ea6695a..8bf620cb 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -39,6 +39,22 @@ use std::sync::{Arc, Mutex, MutexGuard}; #[cfg(feature = "ws-server")] use tokio::sync::broadcast; +/// Workspace-scoped Runtime authorization context supplied by a trusted backend. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RuntimeWorkspaceScope { + pub workspace_id: String, + pub server_id: String, +} + +impl RuntimeWorkspaceScope { + pub fn new(workspace_id: impl Into, server_id: impl Into) -> Self { + Self { + workspace_id: workspace_id.into(), + server_id: server_id.into(), + } + } +} + /// Concrete embedded Runtime domain entity. /// /// The default implementation is memory-backed and tools/provider-less by @@ -327,22 +343,28 @@ impl Runtime { /// Create a Worker scoped to a workspace authorization context. pub fn create_worker_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, request: CreateWorkerRequest, ) -> Result { - self.create_worker_with_workspace(request, Some(workspace_id)) + self.create_worker_with_workspace(request, Some(scope)) } fn create_worker_with_workspace( &self, request: CreateWorkerRequest, - workspace_id: Option<&str>, + scope: Option<&RuntimeWorkspaceScope>, ) -> Result { let (backend, worker_ref, spawn_request) = { let mut state = self.lock()?; state.ensure_running()?; validate_create_worker_request(&request)?; - validate_create_workspace_scope(&request, workspace_id)?; + validate_create_workspace_scope( + &request, + scope.map(|scope| scope.workspace_id.as_str()), + )?; + if let Some(scope) = scope { + state.ensure_workspace_owner(scope, true)?; + }; state.validate_worker_config_boundary(&request)?; if let Some(working_directory_id) = requested_primary_workdir_id(&request) { if let Some(owner_worker_id) = @@ -372,7 +394,7 @@ impl Runtime { worker_ref: worker_ref.clone(), worker_id: worker_id.clone(), status: WorkerStatus::Stopped, - workspace_id: workspace_id.map(ToOwned::to_owned), + workspace_id: scope.map(|scope| scope.workspace_id.clone()), request: request.clone(), working_directory: None, execution_handle: None, @@ -468,13 +490,18 @@ impl Runtime { /// List Workers visible to a workspace-scoped Runtime authorization context. pub fn list_workers_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, ) -> Result, RuntimeError> { - let state = self.lock()?; + let mut state = self.lock()?; + let visible_workspace = state.ensure_workspace_owner_for_existing_workers(scope)?; + if !visible_workspace { + return Ok(Vec::new()); + } + state.persist_runtime_snapshot()?; Ok(state .workers .values() - .filter(|worker| worker.belongs_to_workspace(workspace_id)) + .filter(|worker| worker.belongs_to_workspace(&scope.workspace_id)) .map(WorkerRecord::summary) .collect()) } @@ -482,14 +509,20 @@ impl Runtime { /// List stopped Workers visible to a workspace-scoped Runtime authorization context. pub fn list_stopped_workers_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, ) -> Result, RuntimeError> { - let state = self.lock()?; + let mut state = self.lock()?; + let visible_workspace = state.ensure_workspace_owner_for_existing_workers(scope)?; + if !visible_workspace { + return Ok(Vec::new()); + } + state.persist_runtime_snapshot()?; Ok(state .workers .values() .filter(|worker| { - worker.status == WorkerStatus::Stopped && worker.belongs_to_workspace(workspace_id) + worker.status == WorkerStatus::Stopped + && worker.belongs_to_workspace(&scope.workspace_id) }) .map(WorkerRecord::summary) .collect()) @@ -498,17 +531,19 @@ impl Runtime { /// Fetch Worker detail through a workspace-scoped Runtime authorization context. pub fn worker_detail_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, ) -> Result { - let state = self.lock()?; + let mut state = self.lock()?; let worker = state.worker(worker_ref)?; - if !worker.belongs_to_workspace(workspace_id) { + if !worker.belongs_to_workspace(&scope.workspace_id) { return Err(RuntimeError::WorkerNotFound { worker_id: worker_ref.worker_id, }); } - Ok(worker.detail()) + state.ensure_workspace_owner(scope, true)?; + state.persist_runtime_snapshot()?; + Ok(state.worker(worker_ref)?.detail()) } /// Fetch Worker detail. The supplied [`WorkerRef`] must match this Runtime. @@ -520,27 +555,28 @@ impl Runtime { fn ensure_worker_in_workspace( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, ) -> Result<(), RuntimeError> { - let state = self.lock()?; + let mut state = self.lock()?; let worker = state.worker(worker_ref)?; - if worker.belongs_to_workspace(workspace_id) { - Ok(()) - } else { - Err(RuntimeError::WorkerNotFound { + if !worker.belongs_to_workspace(&scope.workspace_id) { + return Err(RuntimeError::WorkerNotFound { worker_id: worker_ref.worker_id, - }) + }); } + state.ensure_workspace_owner(scope, true)?; + state.persist_runtime_snapshot()?; + Ok(()) } /// Attach a live execution through a workspace-scoped Runtime authorization context. pub fn restore_worker_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, ) -> Result { - self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.ensure_worker_in_workspace(scope, worker_ref)?; self.restore_worker(worker_ref) } @@ -635,11 +671,11 @@ impl Runtime { /// Accept input into a Worker through a workspace-scoped Runtime authorization context. pub fn send_input_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, input: WorkerInput, ) -> Result { - self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.ensure_worker_in_workspace(scope, worker_ref)?; self.send_input(worker_ref, input) } @@ -717,12 +753,12 @@ impl Runtime { /// Return live completion entries through a workspace-scoped Runtime authorization context. pub fn worker_completions_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, kind: protocol::CompletionKind, prefix: &str, ) -> Result, RuntimeError> { - self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.ensure_worker_in_workspace(scope, worker_ref)?; self.worker_completions(worker_ref, kind, prefix) } @@ -751,11 +787,11 @@ impl Runtime { /// Accept a protocol method through a workspace-scoped Runtime authorization context. pub fn send_protocol_method_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, method: Method, ) -> Result, RuntimeError> { - self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.ensure_worker_in_workspace(scope, worker_ref)?; self.send_protocol_method(worker_ref, method) } @@ -845,6 +881,10 @@ impl Runtime { fn rollback_failed_create(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> { let mut state = self.lock()?; if let Some(record) = state.workers.remove(&worker_ref.worker_id) { + let workspace_id = record.workspace_id.clone(); + if let Some(workspace_id) = workspace_id.as_deref() { + state.forget_workspace_owner_if_unused(workspace_id); + } state.events.retain(|event| { event.id != record.last_event_id || event.worker_ref.as_ref() != Some(worker_ref) }); @@ -911,11 +951,11 @@ impl Runtime { /// Stop a Worker through a workspace-scoped Runtime authorization context. pub fn stop_worker_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, reason: Option, ) -> Result { - self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.ensure_worker_in_workspace(scope, worker_ref)?; self.stop_worker(worker_ref, reason) } @@ -937,11 +977,11 @@ impl Runtime { /// Cancel a Worker through a workspace-scoped Runtime authorization context. pub fn cancel_worker_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, reason: Option, ) -> Result { - self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.ensure_worker_in_workspace(scope, worker_ref)?; self.cancel_worker(worker_ref, reason) } @@ -963,10 +1003,10 @@ impl Runtime { /// Delete a non-running Worker through a workspace-scoped Runtime authorization context. pub fn delete_worker_scoped( &self, - workspace_id: &str, + scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, ) -> Result { - self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.ensure_worker_in_workspace(scope, worker_ref)?; self.delete_worker(worker_ref) } @@ -990,6 +1030,9 @@ impl Runtime { worker_id: worker_ref.worker_id, } })?; + if let Some(workspace_id) = removed.workspace_id.as_deref() { + state.forget_workspace_owner_if_unused(workspace_id); + } #[cfg(feature = "ws-server")] state .observation_events @@ -1373,6 +1416,7 @@ struct RuntimeState { #[cfg(feature = "fs-store")] next_diagnostic_id: u64, workers: BTreeMap, + workspace_owners: BTreeMap, config_bundles: BTreeMap, events: Vec, diagnostics: Vec, @@ -1398,6 +1442,7 @@ impl RuntimeState { #[cfg(feature = "fs-store")] next_diagnostic_id: 1, workers: BTreeMap::new(), + workspace_owners: BTreeMap::new(), config_bundles: BTreeMap::new(), events: Vec::new(), diagnostics: Vec::new(), @@ -1428,6 +1473,7 @@ impl RuntimeState { #[cfg(feature = "fs-store")] next_diagnostic_id: 1, workers: BTreeMap::new(), + workspace_owners: BTreeMap::new(), config_bundles: BTreeMap::new(), events: Vec::new(), diagnostics: Vec::new(), @@ -1476,6 +1522,7 @@ impl RuntimeState { next_diagnostic_id, workers, config_bundles: persisted.config_bundles, + workspace_owners: persisted.workspace_owners, events: persisted.events, diagnostics, #[cfg(feature = "ws-server")] @@ -1502,6 +1549,7 @@ impl RuntimeState { .map(|(worker_id, worker)| (worker_id.clone(), worker.persisted_record())) .collect(), config_bundles: self.config_bundles.clone(), + workspace_owners: self.workspace_owners.clone(), events: self.events.clone(), diagnostics: self.diagnostics.clone(), } @@ -1644,6 +1692,59 @@ impl RuntimeState { Ok(()) } + fn ensure_workspace_owner( + &mut self, + scope: &RuntimeWorkspaceScope, + claim_if_missing: bool, + ) -> Result { + if scope.workspace_id.trim().is_empty() { + return Err(RuntimeError::InvalidRequest( + "Runtime auth workspace_id must not be empty".to_string(), + )); + } + if scope.server_id.trim().is_empty() { + return Err(RuntimeError::InvalidRequest( + "Runtime auth server_id must not be empty".to_string(), + )); + } + match self.workspace_owners.get(&scope.workspace_id) { + Some(owner_server_id) if owner_server_id == &scope.server_id => Ok(true), + Some(owner_server_id) => Err(RuntimeError::WorkspaceOwnerMismatch { + workspace_id: scope.workspace_id.clone(), + owner_server_id: owner_server_id.clone(), + requester_server_id: scope.server_id.clone(), + }), + None if claim_if_missing => { + self.workspace_owners + .insert(scope.workspace_id.clone(), scope.server_id.clone()); + Ok(true) + } + None => Ok(false), + } + } + + fn ensure_workspace_owner_for_existing_workers( + &mut self, + scope: &RuntimeWorkspaceScope, + ) -> Result { + let has_workspace_worker = self + .workers + .values() + .any(|worker| worker.belongs_to_workspace(&scope.workspace_id)); + self.ensure_workspace_owner(scope, has_workspace_worker) + } + + fn forget_workspace_owner_if_unused(&mut self, workspace_id: &str) -> bool { + if self + .workers + .values() + .any(|worker| worker.belongs_to_workspace(workspace_id)) + { + return false; + } + self.workspace_owners.remove(workspace_id).is_some() + } + fn worker(&self, worker_ref: &WorkerRef) -> Result<&WorkerRecord, RuntimeError> { self.ensure_worker_ref(worker_ref)?; self.workers @@ -2064,6 +2165,10 @@ mod tests { request } + fn scope(workspace_id: &str, server_id: &str) -> RuntimeWorkspaceScope { + RuntimeWorkspaceScope::new(workspace_id, server_id) + } + fn test_bundle_for_profile(profile: ProfileSelector) -> ConfigBundle { ConfigBundle { metadata: ConfigBundleMetadata { @@ -2245,17 +2350,23 @@ mod tests { fn scoped_worker_access_hides_other_workspace_workers() { let runtime = runtime_with_backend(); let workspace_a = runtime - .create_worker_scoped("workspace-a", scoped_task_request("a", "workspace-a")) + .create_worker_scoped( + &scope("workspace-a", "server-a"), + scoped_task_request("a", "workspace-a"), + ) .unwrap(); let workspace_b = runtime - .create_worker_scoped("workspace-b", scoped_task_request("b", "workspace-b")) + .create_worker_scoped( + &scope("workspace-b", "server-b"), + scoped_task_request("b", "workspace-b"), + ) .unwrap(); assert_eq!(workspace_a.workspace_id.as_deref(), Some("workspace-a")); assert_eq!(workspace_b.workspace_id.as_deref(), Some("workspace-b")); assert_eq!( runtime - .list_workers_scoped("workspace-a") + .list_workers_scoped(&scope("workspace-a", "server-a")) .unwrap() .into_iter() .map(|worker| worker.worker_ref) @@ -2264,7 +2375,7 @@ mod tests { ); assert_eq!( runtime - .list_workers_scoped("workspace-b") + .list_workers_scoped(&scope("workspace-b", "server-b")) .unwrap() .into_iter() .map(|worker| worker.worker_ref) @@ -2273,13 +2384,13 @@ mod tests { ); let detail_error = runtime - .worker_detail_scoped("workspace-a", &workspace_b.worker_ref) + .worker_detail_scoped(&scope("workspace-a", "server-a"), &workspace_b.worker_ref) .unwrap_err(); assert!(matches!(detail_error, RuntimeError::WorkerNotFound { .. })); let input_error = runtime .send_input_scoped( - "workspace-a", + &scope("workspace-a", "server-a"), &workspace_b.worker_ref, WorkerInput::user("cross workspace"), ) @@ -2287,7 +2398,11 @@ mod tests { assert!(matches!(input_error, RuntimeError::WorkerNotFound { .. })); let protocol_error = runtime - .send_protocol_method_scoped("workspace-a", &workspace_b.worker_ref, Method::Shutdown) + .send_protocol_method_scoped( + &scope("workspace-a", "server-a"), + &workspace_b.worker_ref, + Method::Shutdown, + ) .unwrap_err(); assert!(matches!( protocol_error, @@ -2302,12 +2417,61 @@ mod tests { ); } + #[test] + fn workspace_owner_binding_rejects_other_backend_and_forgets_after_last_worker_delete() { + let runtime = runtime_with_backend(); + let server_a = scope("workspace-a", "server-a"); + let server_b = scope("workspace-a", "server-b"); + let first = runtime + .create_worker_scoped(&server_a, scoped_task_request("first", "workspace-a")) + .unwrap(); + let second = runtime + .create_worker_scoped(&server_a, scoped_task_request("second", "workspace-a")) + .unwrap(); + + let create_error = runtime + .create_worker_scoped(&server_b, scoped_task_request("stolen", "workspace-a")) + .unwrap_err(); + assert!(matches!( + create_error, + RuntimeError::WorkspaceOwnerMismatch { .. } + )); + let read_error = runtime + .worker_detail_scoped(&server_b, &first.worker_ref) + .unwrap_err(); + assert!(matches!( + read_error, + RuntimeError::WorkspaceOwnerMismatch { .. } + )); + + runtime.stop_worker(&first.worker_ref, None).unwrap(); + runtime.stop_worker(&second.worker_ref, None).unwrap(); + runtime + .delete_worker_scoped(&server_a, &first.worker_ref) + .unwrap(); + let still_owned_error = runtime + .create_worker_scoped(&server_b, scoped_task_request("still owned", "workspace-a")) + .unwrap_err(); + assert!(matches!( + still_owned_error, + RuntimeError::WorkspaceOwnerMismatch { .. } + )); + + runtime + .delete_worker_scoped(&server_a, &second.worker_ref) + .unwrap(); + let rebound = runtime + .create_worker_scoped(&server_b, scoped_task_request("rebound", "workspace-a")) + .unwrap(); + assert_eq!(rebound.workspace_id.as_deref(), Some("workspace-a")); + } + #[test] fn scoped_create_rejects_request_workspace_mismatch() { let runtime = runtime_with_backend(); let error = runtime .create_worker_scoped( - "workspace-a", + &scope("workspace-a", "server-a"), scoped_task_request("mismatch", "workspace-b"), ) .unwrap_err(); @@ -2323,12 +2487,12 @@ mod tests { assert!(legacy.workspace_id.is_none()); assert!( runtime - .list_workers_scoped("workspace-a") + .list_workers_scoped(&scope("workspace-a", "server-a")) .unwrap() .is_empty() ); let detail_error = runtime - .worker_detail_scoped("workspace-a", &legacy.worker_ref) + .worker_detail_scoped(&scope("workspace-a", "server-a"), &legacy.worker_ref) .unwrap_err(); assert!(matches!(detail_error, RuntimeError::WorkerNotFound { .. })); } @@ -2337,15 +2501,23 @@ mod tests { fn stopped_worker_scoped_list_uses_workspace_boundary() { let runtime = runtime_with_backend(); let workspace_a = runtime - .create_worker_scoped("workspace-a", scoped_task_request("a", "workspace-a")) + .create_worker_scoped( + &scope("workspace-a", "server-a"), + scoped_task_request("a", "workspace-a"), + ) .unwrap(); let workspace_b = runtime - .create_worker_scoped("workspace-b", scoped_task_request("b", "workspace-b")) + .create_worker_scoped( + &scope("workspace-b", "server-b"), + scoped_task_request("b", "workspace-b"), + ) .unwrap(); runtime.stop_worker(&workspace_a.worker_ref, None).unwrap(); runtime.stop_worker(&workspace_b.worker_ref, None).unwrap(); - let stopped = runtime.list_stopped_workers_scoped("workspace-a").unwrap(); + let stopped = runtime + .list_stopped_workers_scoped(&scope("workspace-a", "server-a")) + .unwrap(); assert_eq!(stopped.len(), 1); assert_eq!(stopped[0].worker_ref, workspace_a.worker_ref); } @@ -3012,7 +3184,7 @@ mod tests { runtime.store_config_bundle(test_bundle()).unwrap(); let scoped = runtime .create_worker_scoped( - "workspace-a", + &scope("workspace-a", "server-a"), scoped_task_request("persist workspace", "workspace-a"), ) .unwrap(); @@ -3037,7 +3209,7 @@ mod tests { assert_eq!(restored_scoped.workspace_id.as_deref(), Some("workspace-a")); assert_eq!( restored - .list_workers_scoped("workspace-a") + .list_workers_scoped(&scope("workspace-a", "server-a")) .unwrap() .into_iter() .map(|worker| worker.worker_ref) @@ -3045,16 +3217,29 @@ mod tests { vec![scoped.worker_ref.clone()] ); let legacy_error = restored - .worker_detail_scoped("workspace-a", &legacy.worker_ref) + .worker_detail_scoped(&scope("workspace-a", "server-a"), &legacy.worker_ref) .unwrap_err(); assert!(matches!(legacy_error, RuntimeError::WorkerNotFound { .. })); let recovered_legacy = restored - .worker_detail_scoped("workspace-b", &recoverable_legacy.worker_ref) + .worker_detail_scoped( + &scope("workspace-b", "server-b"), + &recoverable_legacy.worker_ref, + ) .unwrap(); assert_eq!( recovered_legacy.workspace_id.as_deref(), Some("workspace-b") ); + let stolen_legacy_error = restored + .worker_detail_scoped( + &scope("workspace-b", "server-c"), + &recoverable_legacy.worker_ref, + ) + .unwrap_err(); + assert!(matches!( + stolen_legacy_error, + RuntimeError::WorkspaceOwnerMismatch { .. } + )); let _ = std::fs::remove_dir_all(root); } diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 9bc479cc..aec8cdc5 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -3349,6 +3349,7 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti workdir_diagnostic.message.clone(), ), EmbeddedRuntimeError::InvalidRequest(_) + | EmbeddedRuntimeError::WorkspaceOwnerMismatch { .. } | EmbeddedRuntimeError::ConfigBundleMissing { .. } | EmbeddedRuntimeError::ConfigBundleDigestMismatch { .. } | EmbeddedRuntimeError::InvalidProfileSelector { .. }