From 3dda06cbe3ab69a8ec26e042775a3d03e695f248 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 30 Jul 2026 00:25:46 +0900 Subject: [PATCH] runtime: scope worker access by workspace --- crates/worker-runtime/src/auth.rs | 5 + crates/worker-runtime/src/catalog.rs | 4 + crates/worker-runtime/src/fs_store.rs | 11 + crates/worker-runtime/src/http_server.rs | 418 +++++++++++++++++++---- crates/worker-runtime/src/runtime.rs | 375 +++++++++++++++++++- 5 files changed, 754 insertions(+), 59 deletions(-) diff --git a/crates/worker-runtime/src/auth.rs b/crates/worker-runtime/src/auth.rs index ba233d36..3c90b176 100644 --- a/crates/worker-runtime/src/auth.rs +++ b/crates/worker-runtime/src/auth.rs @@ -37,6 +37,8 @@ pub enum RuntimeAuthError { WrongAudience { expected: String, actual: String }, #[error("capability token is expired")] Expired, + #[error("capability token is missing workspace scope")] + MissingWorkspaceScope, #[error("capability token is missing required permission `{0}`")] MissingPermission(String), } @@ -187,6 +189,9 @@ pub fn verify_capability_token( if claims.exp < now_seconds { return Err(RuntimeAuthError::Expired); } + if claims.workspace_id.trim().is_empty() { + return Err(RuntimeAuthError::MissingWorkspaceScope); + } if let Some(required) = required_permission { if !claims .permissions diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index fa69b9da..3ba2a272 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -230,6 +230,8 @@ pub struct WorkerSummary { pub worker_id: WorkerId, pub status: WorkerStatus, #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspace_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub working_directory: Option, pub profile: ProfileSelector, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -247,6 +249,8 @@ pub struct WorkerDetail { pub worker_id: WorkerId, pub status: WorkerStatus, #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspace_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub working_directory: Option, pub profile: ProfileSelector, #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index adc036da..6ec27e70 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -302,6 +302,7 @@ pub(crate) struct PersistedWorkerRecord { pub(crate) worker_ref: WorkerRef, pub(crate) worker_id: WorkerId, pub(crate) request: CreateWorkerRequest, + pub(crate) workspace_id: Option, pub(crate) working_directory: Option, pub(crate) last_event_id: u64, } @@ -401,6 +402,8 @@ struct WorkerSnapshot { worker_id: WorkerId, request: CreateWorkerRequest, #[serde(default, skip_serializing_if = "Option::is_none")] + workspace_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] working_directory: Option, /// One-way migration input for schema-v1 snapshots. New snapshots never /// write the removed execution projection. @@ -422,6 +425,7 @@ impl WorkerSnapshot { worker_ref: worker.worker_ref.clone(), worker_id: worker.worker_id.clone(), request: worker.request.clone(), + workspace_id: worker.workspace_id.clone(), working_directory: worker.working_directory.clone(), legacy_execution: None, last_event_id: worker.last_event_id, @@ -453,10 +457,17 @@ impl WorkerSnapshot { } fn into_persisted(self) -> PersistedWorkerRecord { + let workspace_id = self.workspace_id.or_else(|| { + self.request + .workspace_api + .as_ref() + .map(|workspace_api| workspace_api.workspace_id.clone()) + }); PersistedWorkerRecord { worker_ref: self.worker_ref, worker_id: self.worker_id, request: self.request, + workspace_id, working_directory: self.working_directory.or_else(|| { self.legacy_execution .and_then(|execution| execution.working_directory) diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index a8d8a3d7..c8fa053e 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -8,7 +8,8 @@ use crate::Runtime; use crate::auth::{ - RuntimeAuthContext, RuntimeHttpAuthConfig, unix_now_seconds, verify_capability_token, + RuntimeAuthContext, RuntimeAuthError, RuntimeHttpAuthConfig, unix_now_seconds, + verify_capability_token, }; use crate::catalog::{ ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary, @@ -25,7 +26,7 @@ use axum::body::{Body, Bytes}; use axum::extract::rejection::{JsonRejection, QueryRejection}; #[cfg(feature = "ws-server")] use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; -use axum::extract::{Path, Query, State}; +use axum::extract::{Extension, Path, Query, State}; use axum::http::{Method, Request, StatusCode, header}; use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Response}; @@ -383,12 +384,20 @@ async fn check_config_bundle( async fn list_workers( State(state): State, + auth: Option>, query: Result, QueryRejection>, ) -> RestResult { let Query(query) = query.map_err(RuntimeHttpRestError::query_rejection)?; - let workers = match query.status { - Some(RuntimeHttpWorkerStatusFilter::Stopped) => state.runtime.list_stopped_workers(), - None => state.runtime.list_workers(), + 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) + } + (Some(RuntimeHttpWorkerStatusFilter::Stopped), None) => { + state.runtime.list_stopped_workers() + } + (None, Some(workspace_id)) => state.runtime.list_workers_scoped(workspace_id), + (None, None) => state.runtime.list_workers(), } .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkersResponse { workers })) @@ -448,67 +457,87 @@ async fn cleanup_working_directory( async fn get_worker( State(state): State, + auth: Option>, Path(worker_id): Path, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let worker = state - .runtime - .worker_detail(&worker_ref) - .map_err(RuntimeHttpRestError::runtime)?; + let worker = match auth_workspace_id(&state, auth.as_ref())? { + Some(workspace_id) => state + .runtime + .worker_detail_scoped(workspace_id, &worker_ref), + None => state.runtime.worker_detail(&worker_ref), + } + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkerResponse { worker })) } async fn delete_worker( State(state): State, + auth: Option>, Path(worker_id): Path, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let worker = state - .runtime - .delete_worker(&worker_ref) - .map_err(RuntimeHttpRestError::runtime)?; + let worker = match auth_workspace_id(&state, auth.as_ref())? { + Some(workspace_id) => state + .runtime + .delete_worker_scoped(workspace_id, &worker_ref), + None => state.runtime.delete_worker(&worker_ref), + } + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkerDeleteResponse { worker })) } async fn create_worker( State(state): State, + auth: Option>, body: Result, JsonRejection>, ) -> RestResult { let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?; - let worker = state - .runtime - .create_worker(request) - .map_err(RuntimeHttpRestError::runtime)?; + let worker = match auth_workspace_id(&state, auth.as_ref())? { + Some(workspace_id) => state.runtime.create_worker_scoped(workspace_id, request), + None => state.runtime.create_worker(request), + } + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkerResponse { worker })) } async fn restore_worker( State(state): State, + auth: Option>, Path(worker_id): Path, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let worker = state - .runtime - .restore_worker(&worker_ref) - .map_err(RuntimeHttpRestError::runtime)?; + let worker = match auth_workspace_id(&state, auth.as_ref())? { + Some(workspace_id) => state + .runtime + .restore_worker_scoped(workspace_id, &worker_ref), + None => state.runtime.restore_worker(&worker_ref), + } + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkerResponse { worker })) } #[cfg(feature = "ws-server")] async fn worker_protocol_ws( State(state): State, + auth: Option>, Path(worker_id): Path, Query(query): Query, ws: WebSocketUpgrade, ) -> Result { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - state - .runtime - .worker_detail(&worker_ref) - .map_err(RuntimeHttpRestError::runtime)?; + let workspace_id = auth_workspace_id(&state, auth.as_ref())?.map(ToOwned::to_owned); + match workspace_id.as_deref() { + Some(workspace_id) => state + .runtime + .worker_detail_scoped(workspace_id, &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, worker_ref, query, socket) + worker_protocol_ws_session(state.runtime, workspace_id, worker_ref, query, socket) }) .into_response()) } @@ -516,6 +545,7 @@ async fn worker_protocol_ws( #[cfg(feature = "ws-server")] async fn worker_protocol_ws_session( runtime: Runtime, + workspace_id: Option, worker_ref: WorkerRef, query: RuntimeWorkerEventsWsQuery, mut socket: WebSocket, @@ -599,20 +629,30 @@ async fn worker_protocol_ws_session( inbound = socket.next() => { match inbound { Some(Ok(WsMessage::Text(text))) => match decode_method(&text) { - Ok(method) => match runtime.send_protocol_method(&worker_ref, method) { - Ok(events) => { - for event in events { + Ok(method) => { + let result = match workspace_id.as_deref() { + Some(workspace_id) => runtime.send_protocol_method_scoped( + workspace_id, + &worker_ref, + method, + ), + None => runtime.send_protocol_method(&worker_ref, method), + }; + match result { + Ok(events) => { + for event in events { + if !send_protocol_event(&mut socket, &event).await { + return; + } + } + } + Err(error) => { + let event = protocol_error_event(error.to_string()); if !send_protocol_event(&mut socket, &event).await { return; } } } - Err(error) => { - let event = protocol_error_event(error.to_string()); - if !send_protocol_event(&mut socket, &event).await { - return; - } - } }, Err(error) => { let event = protocol_error_event(format!( @@ -688,29 +728,42 @@ fn protocol_error_event(message: impl Into) -> protocol::Event { async fn send_worker_input( State(state): State, + auth: Option>, Path(worker_id): Path, body: Result, JsonRejection>, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; let Json(input) = body.map_err(RuntimeHttpRestError::json_rejection)?; - let ack = state - .runtime - .send_input(&worker_ref, input) - .map_err(RuntimeHttpRestError::runtime)?; + let ack = match auth_workspace_id(&state, auth.as_ref())? { + Some(workspace_id) => state + .runtime + .send_input_scoped(workspace_id, &worker_ref, input), + None => state.runtime.send_input(&worker_ref, input), + } + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkerInputResponse { ack })) } async fn worker_completions( State(state): State, + auth: Option>, Path(worker_id): Path, body: Result, JsonRejection>, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?; - let entries = state - .runtime - .worker_completions(&worker_ref, request.kind, &request.prefix) - .map_err(RuntimeHttpRestError::runtime)?; + let entries = match auth_workspace_id(&state, auth.as_ref())? { + Some(workspace_id) => state.runtime.worker_completions_scoped( + workspace_id, + &worker_ref, + request.kind, + &request.prefix, + ), + None => state + .runtime + .worker_completions(&worker_ref, request.kind, &request.prefix), + } + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkerCompletionsResponse { kind: request.kind, prefix: request.prefix, @@ -720,29 +773,41 @@ async fn worker_completions( async fn stop_worker( State(state): State, + auth: Option>, Path(worker_id): Path, body: Bytes, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; let request = parse_optional_lifecycle_request(body)?; - let ack = state - .runtime - .stop_worker(&worker_ref, request.reason) - .map_err(RuntimeHttpRestError::runtime)?; + let ack = match auth_workspace_id(&state, auth.as_ref())? { + Some(workspace_id) => { + state + .runtime + .stop_worker_scoped(workspace_id, &worker_ref, request.reason) + } + None => state.runtime.stop_worker(&worker_ref, request.reason), + } + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkerLifecycleResponse { ack })) } async fn cancel_worker( State(state): State, + auth: Option>, Path(worker_id): Path, body: Bytes, ) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; let request = parse_optional_lifecycle_request(body)?; - let ack = state - .runtime - .cancel_worker(&worker_ref, request.reason) - .map_err(RuntimeHttpRestError::runtime)?; + let ack = match auth_workspace_id(&state, auth.as_ref())? { + Some(workspace_id) => { + state + .runtime + .cancel_worker_scoped(workspace_id, &worker_ref, request.reason) + } + None => state.runtime.cancel_worker(&worker_ref, request.reason), + } + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkerLifecycleResponse { ack })) } @@ -806,12 +871,7 @@ async fn require_runtime_auth( return next.run(request).await; } Err(error) => { - return RuntimeHttpRestError::new( - StatusCode::UNAUTHORIZED, - "unauthorized", - format!("invalid Runtime capability token: {error}"), - ) - .into_response(); + return runtime_auth_error_response(error).into_response(); } } } @@ -836,6 +896,51 @@ async fn require_runtime_auth( next.run(request).await } +fn runtime_auth_error_response(error: RuntimeAuthError) -> RuntimeHttpRestError { + match error { + RuntimeAuthError::MissingPermission(permission) => RuntimeHttpRestError::new( + StatusCode::FORBIDDEN, + "forbidden", + format!("Runtime capability token is missing required permission `{permission}`"), + ), + RuntimeAuthError::MissingWorkspaceScope => RuntimeHttpRestError::new( + StatusCode::FORBIDDEN, + "workspace_scope_required", + "Runtime capability token is missing workspace scope", + ), + other => RuntimeHttpRestError::new( + StatusCode::UNAUTHORIZED, + "unauthorized", + format!("invalid Runtime capability token: {other}"), + ), + } +} + +fn auth_workspace_id<'a>( + state: &RuntimeHttpState, + auth: Option<&'a Extension>, +) -> Result, RuntimeHttpRestError> { + let Some(Extension(context)) = auth else { + if state.auth.is_some() || state.local_token.is_some() { + return Err(RuntimeHttpRestError::new( + StatusCode::FORBIDDEN, + "workspace_scope_required", + "Runtime worker operation requires a workspace-scoped authorization context", + )); + } + return Ok(None); + }; + let workspace_id = context.workspace_id.trim(); + if workspace_id.is_empty() { + return Err(RuntimeHttpRestError::new( + StatusCode::FORBIDDEN, + "workspace_scope_required", + "Runtime worker operation requires a non-empty workspace scope", + )); + } + Ok(Some(workspace_id)) +} + fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static str> { if path == "/v1/runtime" { return None; @@ -991,7 +1096,11 @@ pub enum RuntimeHttpServerError { #[cfg(test)] mod tests { use super::*; - use crate::catalog::{ConfigBundleRef, ProfileSelector, WorkerStatus}; + use crate::auth::{ + CapabilityTokenSigner, RuntimeHttpAuthConfig, RuntimeIdentityMaterial, TrustedServerKey, + capability_claims, + }; + use crate::catalog::{ConfigBundleRef, ProfileSelector, WorkerStatus, WorkspaceApiRef}; use crate::config_bundle::{ ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigProfileDescriptor, }; @@ -1029,6 +1138,199 @@ mod tests { .with_computed_digest() } + fn scoped_task_request(objective: &str, workspace_id: &str) -> CreateWorkerRequest { + let mut request = task_request(objective); + request.workspace_api = Some(WorkspaceApiRef { + workspace_id: workspace_id.to_string(), + base_url: format!("https://workspace.example/{workspace_id}"), + }); + request + } + + fn auth_config_and_signer() -> (RuntimeHttpAuthConfig, CapabilityTokenSigner) { + let identity = RuntimeIdentityMaterial::generate("server-a").unwrap(); + let signer = CapabilityTokenSigner::new(identity.identity_id.clone(), identity.private_key); + let auth = RuntimeHttpAuthConfig { + runtime_id: "runtime-test".to_string(), + trusted_servers: vec![TrustedServerKey { + server_id: identity.identity_id, + public_key: identity.public_key, + display_name: None, + }], + }; + (auth, signer) + } + + fn token_for_workspace(signer: &CapabilityTokenSigner, workspace_id: &str) -> String { + token_for_workspace_with_permissions( + signer, + workspace_id, + [ + "workers:list", + "workers:create", + "workers:read", + "workers:input", + "workers:stop", + "workers:protocol", + "workers:delete", + ], + ) + } + + fn token_for_workspace_with_permissions( + signer: &CapabilityTokenSigner, + workspace_id: &str, + permissions: [&str; N], + ) -> String { + let claims = capability_claims( + signer.server_id(), + "runtime-test", + workspace_id, + permissions.into_iter().map(str::to_string).collect(), + 3600, + ) + .unwrap(); + signer.sign(&claims).unwrap() + } + + fn bearer_request( + method: Method, + uri: impl AsRef, + token: &str, + body: impl Into, + ) -> Request { + Request::builder() + .method(method) + .uri(uri.as_ref()) + .header(header::AUTHORIZATION, format!("Bearer {token}")) + .header(header::CONTENT_TYPE, "application/json") + .body(body.into()) + .unwrap() + } + + #[tokio::test] + async fn capability_workspace_scope_filters_list_and_hides_detail() { + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(AcceptingBackend)) + .unwrap(); + let (auth, signer) = auth_config_and_signer(); + let token_a = token_for_workspace(&signer, "workspace-a"); + let token_b = token_for_workspace(&signer, "workspace-b"); + let app = runtime_http_router_with_auth(runtime, None, Some(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 body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let worker_a: RuntimeHttpWorkerResponse = serde_json::from_slice(&body).unwrap(); + assert_eq!(worker_a.worker.workspace_id.as_deref(), Some("workspace-a")); + + let create_b = scoped_task_request("b", "workspace-b"); + let response = app + .clone() + .oneshot(bearer_request( + Method::POST, + "/v1/workers", + &token_b, + serde_json::to_vec(&create_b).unwrap(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let worker_b: RuntimeHttpWorkerResponse = serde_json::from_slice(&body).unwrap(); + assert_eq!(worker_b.worker.workspace_id.as_deref(), Some("workspace-b")); + + let response = app + .clone() + .oneshot(bearer_request( + Method::GET, + "/v1/workers", + &token_a, + Body::empty(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let workers: RuntimeHttpWorkersResponse = serde_json::from_slice(&body).unwrap(); + assert_eq!(workers.workers.len(), 1); + assert_eq!(workers.workers[0].worker_ref, worker_a.worker.worker_ref); + assert_eq!( + workers.workers[0].workspace_id.as_deref(), + Some("workspace-a") + ); + + let response = app + .oneshot(bearer_request( + Method::GET, + format!("/v1/workers/{}", worker_b.worker.worker_id), + &token_a, + Body::empty(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + } + + #[tokio::test] + async fn capability_token_without_workspace_scope_is_forbidden() { + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(AcceptingBackend)) + .unwrap(); + let (auth, signer) = auth_config_and_signer(); + let token = token_for_workspace_with_permissions(&signer, "", ["workers:list"]); + let app = runtime_http_router_with_auth(runtime, None, Some(auth)); + + let response = app + .oneshot(bearer_request( + Method::GET, + "/v1/workers", + &token, + Body::empty(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + } + + #[tokio::test] + async fn capability_token_without_worker_permission_is_forbidden() { + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(AcceptingBackend)) + .unwrap(); + let (auth, signer) = auth_config_and_signer(); + let token = token_for_workspace_with_permissions(&signer, "workspace-a", ["workers:list"]); + let app = runtime_http_router_with_auth(runtime, None, Some(auth)); + let create = scoped_task_request("a", "workspace-a"); + + let response = app + .oneshot(bearer_request( + Method::POST, + "/v1/workers", + &token, + serde_json::to_vec(&create).unwrap(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + } + fn task_request(_objective: &str) -> CreateWorkerRequest { let profile = ProfileSelector::Builtin("builtin:coder".to_string()); let bundle = test_bundle(profile.clone()); diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 4d19f342..8ea6695a 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -320,11 +320,29 @@ impl Runtime { pub fn create_worker( &self, request: CreateWorkerRequest, + ) -> Result { + self.create_worker_with_workspace(request, None) + } + + /// Create a Worker scoped to a workspace authorization context. + pub fn create_worker_scoped( + &self, + workspace_id: &str, + request: CreateWorkerRequest, + ) -> Result { + self.create_worker_with_workspace(request, Some(workspace_id)) + } + + fn create_worker_with_workspace( + &self, + request: CreateWorkerRequest, + workspace_id: Option<&str>, ) -> 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)?; state.validate_worker_config_boundary(&request)?; if let Some(working_directory_id) = requested_primary_workdir_id(&request) { if let Some(owner_worker_id) = @@ -354,6 +372,7 @@ impl Runtime { worker_ref: worker_ref.clone(), worker_id: worker_id.clone(), status: WorkerStatus::Stopped, + workspace_id: workspace_id.map(ToOwned::to_owned), request: request.clone(), working_directory: None, execution_handle: None, @@ -446,6 +465,52 @@ impl Runtime { .collect()) } + /// List Workers visible to a workspace-scoped Runtime authorization context. + pub fn list_workers_scoped( + &self, + workspace_id: &str, + ) -> Result, RuntimeError> { + let state = self.lock()?; + Ok(state + .workers + .values() + .filter(|worker| worker.belongs_to_workspace(workspace_id)) + .map(WorkerRecord::summary) + .collect()) + } + + /// List stopped Workers visible to a workspace-scoped Runtime authorization context. + pub fn list_stopped_workers_scoped( + &self, + workspace_id: &str, + ) -> Result, RuntimeError> { + let state = self.lock()?; + Ok(state + .workers + .values() + .filter(|worker| { + worker.status == WorkerStatus::Stopped && worker.belongs_to_workspace(workspace_id) + }) + .map(WorkerRecord::summary) + .collect()) + } + + /// Fetch Worker detail through a workspace-scoped Runtime authorization context. + pub fn worker_detail_scoped( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + ) -> Result { + let state = self.lock()?; + let worker = state.worker(worker_ref)?; + if !worker.belongs_to_workspace(workspace_id) { + return Err(RuntimeError::WorkerNotFound { + worker_id: worker_ref.worker_id, + }); + } + Ok(worker.detail()) + } + /// Fetch Worker detail. The supplied [`WorkerRef`] must match this Runtime. pub fn worker_detail(&self, worker_ref: &WorkerRef) -> Result { let state = self.lock()?; @@ -453,6 +518,32 @@ impl Runtime { Ok(worker.detail()) } + fn ensure_worker_in_workspace( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + ) -> Result<(), RuntimeError> { + let state = self.lock()?; + let worker = state.worker(worker_ref)?; + if worker.belongs_to_workspace(workspace_id) { + Ok(()) + } else { + Err(RuntimeError::WorkerNotFound { + worker_id: worker_ref.worker_id, + }) + } + } + + /// Attach a live execution through a workspace-scoped Runtime authorization context. + pub fn restore_worker_scoped( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + ) -> Result { + self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.restore_worker(worker_ref) + } + /// Attach a live execution to a persisted Worker definition. /// /// Current liveness is never read from disk. If a handle is already @@ -541,6 +632,17 @@ impl Runtime { self.restore_worker(worker_ref).map(|_| ()) } + /// Accept input into a Worker through a workspace-scoped Runtime authorization context. + pub fn send_input_scoped( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + input: WorkerInput, + ) -> Result { + self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.send_input(worker_ref, input) + } + /// Accept input into a Worker. pub fn send_input( &self, @@ -612,6 +714,18 @@ impl Runtime { }) } + /// Return live completion entries through a workspace-scoped Runtime authorization context. + pub fn worker_completions_scoped( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + kind: protocol::CompletionKind, + prefix: &str, + ) -> Result, RuntimeError> { + self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.worker_completions(worker_ref, kind, prefix) + } + /// Return live completion entries for the Worker composer. pub fn worker_completions( &self, @@ -634,6 +748,17 @@ impl Runtime { Ok(backend.worker_completions(&handle, kind, prefix)) } + /// Accept a protocol method through a workspace-scoped Runtime authorization context. + pub fn send_protocol_method_scoped( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + method: Method, + ) -> Result, RuntimeError> { + self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.send_protocol_method(worker_ref, method) + } + /// Accept a protocol method for a Worker through a Backend/runtime transport. /// /// Most methods are delivered to the execution backend unchanged. Methods with @@ -783,6 +908,17 @@ impl Runtime { }) } + /// Stop a Worker through a workspace-scoped Runtime authorization context. + pub fn stop_worker_scoped( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + reason: Option, + ) -> Result { + self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.stop_worker(worker_ref, reason) + } + /// Stop a Worker. Repeated stops are idempotent and return the last event id. pub fn stop_worker( &self, @@ -798,6 +934,17 @@ impl Runtime { ) } + /// Cancel a Worker through a workspace-scoped Runtime authorization context. + pub fn cancel_worker_scoped( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + reason: Option, + ) -> Result { + self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.cancel_worker(worker_ref, reason) + } + /// Cancel a Worker. Repeated cancels are idempotent and return the last event id. pub fn cancel_worker( &self, @@ -813,6 +960,16 @@ impl Runtime { ) } + /// Delete a non-running Worker through a workspace-scoped Runtime authorization context. + pub fn delete_worker_scoped( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + ) -> Result { + self.ensure_worker_in_workspace(workspace_id, worker_ref)?; + self.delete_worker(worker_ref) + } + /// Delete a non-running Worker from Runtime state and persisted Worker storage. pub fn delete_worker( &self, @@ -1298,6 +1455,7 @@ impl RuntimeState { worker_ref: worker.worker_ref, worker_id: worker.worker_id, status: WorkerStatus::Stopped, + workspace_id: worker.workspace_id, request: worker.request, working_directory: worker.working_directory, execution_handle: None, @@ -1661,6 +1819,7 @@ struct WorkerRecord { worker_ref: WorkerRef, worker_id: WorkerId, status: WorkerStatus, + workspace_id: Option, request: CreateWorkerRequest, working_directory: Option, execution_handle: Option, @@ -1668,11 +1827,16 @@ struct WorkerRecord { } impl WorkerRecord { + fn belongs_to_workspace(&self, workspace_id: &str) -> bool { + self.workspace_id.as_deref() == Some(workspace_id) + } + fn summary(&self) -> WorkerSummary { WorkerSummary { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id, status: self.status, + workspace_id: self.workspace_id.clone(), working_directory: self.working_directory.clone(), profile: self.request.profile.clone(), display_name: self.request.display_name.clone(), @@ -1687,6 +1851,7 @@ impl WorkerRecord { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id, status: self.status, + workspace_id: self.workspace_id.clone(), working_directory: self.working_directory.clone(), profile: self.request.profile.clone(), display_name: self.request.display_name.clone(), @@ -1702,6 +1867,7 @@ impl WorkerRecord { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id.clone(), request: self.request.clone(), + workspace_id: self.workspace_id.clone(), working_directory: self.working_directory.clone(), last_event_id: self.last_event_id, } @@ -1766,6 +1932,32 @@ fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), R Ok(()) } +fn validate_create_workspace_scope( + request: &CreateWorkerRequest, + workspace_id: Option<&str>, +) -> Result<(), RuntimeError> { + let Some(workspace_id) = workspace_id else { + return Ok(()); + }; + if workspace_id.trim().is_empty() { + return Err(RuntimeError::InvalidRequest( + "Runtime auth workspace_id must not be empty".to_string(), + )); + } + if let Some(request_workspace_id) = request + .workspace_api + .as_ref() + .map(|workspace_api| workspace_api.workspace_id.as_str()) + { + if request_workspace_id != workspace_id { + return Err(RuntimeError::InvalidRequest(format!( + "request workspace_id {request_workspace_id} does not match Runtime auth workspace_id {workspace_id}" + ))); + } + } + Ok(()) +} + fn validate_worker_input(input: &WorkerInput) -> Result<(), RuntimeError> { if !input.kind.is_empty_content_allowed() && input.content.trim().is_empty() { return Err(RuntimeError::InvalidRequest( @@ -1815,7 +2007,7 @@ fn input_protocol_event(input: &WorkerInput) -> protocol::Event { #[cfg(test)] mod tests { use super::*; - use crate::catalog::{ConfigBundleRef, ProfileSelector}; + use crate::catalog::{ConfigBundleRef, ProfileSelector, WorkspaceApiRef}; use crate::config_bundle::{ ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigDeclaration, ConfigDeclarationKind, ConfigProfileDescriptor, @@ -1863,6 +2055,15 @@ mod tests { } } + fn scoped_task_request(objective: &str, workspace_id: &str) -> CreateWorkerRequest { + let mut request = task_request(objective); + request.workspace_api = Some(WorkspaceApiRef { + workspace_id: workspace_id.to_string(), + base_url: format!("https://workspace.example/{workspace_id}"), + }); + request + } + fn test_bundle_for_profile(profile: ProfileSelector) -> ConfigBundle { ConfigBundle { metadata: ConfigBundleMetadata { @@ -2040,6 +2241,115 @@ mod tests { request } + #[test] + 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")) + .unwrap(); + let workspace_b = runtime + .create_worker_scoped("workspace-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") + .unwrap() + .into_iter() + .map(|worker| worker.worker_ref) + .collect::>(), + vec![workspace_a.worker_ref.clone()] + ); + assert_eq!( + runtime + .list_workers_scoped("workspace-b") + .unwrap() + .into_iter() + .map(|worker| worker.worker_ref) + .collect::>(), + vec![workspace_b.worker_ref.clone()] + ); + + let detail_error = runtime + .worker_detail_scoped("workspace-a", &workspace_b.worker_ref) + .unwrap_err(); + assert!(matches!(detail_error, RuntimeError::WorkerNotFound { .. })); + + let input_error = runtime + .send_input_scoped( + "workspace-a", + &workspace_b.worker_ref, + WorkerInput::user("cross workspace"), + ) + .unwrap_err(); + assert!(matches!(input_error, RuntimeError::WorkerNotFound { .. })); + + let protocol_error = runtime + .send_protocol_method_scoped("workspace-a", &workspace_b.worker_ref, Method::Shutdown) + .unwrap_err(); + assert!(matches!( + protocol_error, + RuntimeError::WorkerNotFound { .. } + )); + assert_eq!( + runtime + .worker_detail(&workspace_b.worker_ref) + .unwrap() + .status, + WorkerStatus::Idle + ); + } + + #[test] + fn scoped_create_rejects_request_workspace_mismatch() { + let runtime = runtime_with_backend(); + let error = runtime + .create_worker_scoped( + "workspace-a", + scoped_task_request("mismatch", "workspace-b"), + ) + .unwrap_err(); + assert!(matches!(error, RuntimeError::InvalidRequest(_))); + assert!(runtime.list_workers().unwrap().is_empty()); + } + + #[test] + fn unscoped_legacy_workers_are_hidden_from_scoped_access() { + let runtime = runtime_with_backend(); + let legacy = runtime.create_worker(task_request("legacy")).unwrap(); + + assert!(legacy.workspace_id.is_none()); + assert!( + runtime + .list_workers_scoped("workspace-a") + .unwrap() + .is_empty() + ); + let detail_error = runtime + .worker_detail_scoped("workspace-a", &legacy.worker_ref) + .unwrap_err(); + assert!(matches!(detail_error, RuntimeError::WorkerNotFound { .. })); + } + + #[test] + 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")) + .unwrap(); + let workspace_b = runtime + .create_worker_scoped("workspace-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(); + assert_eq!(stopped.len(), 1); + assert_eq!(stopped[0].worker_ref, workspace_a.worker_ref); + } + #[test] fn create_list_and_detail_preserve_runtime_local_worker_authority() { let runtime = runtime_with_backend(); @@ -2686,6 +2996,69 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[cfg(feature = "fs-store")] + #[test] + fn fs_store_restores_workspace_scope_and_hides_legacy_workers_from_scoped_access() { + let root = fs_store_root("workspace-scope"); + let runtime = Runtime::with_fs_store_and_execution_backend( + crate::fs_store::FsRuntimeStoreOptions { + root: root.clone(), + display_name: None, + limits: RuntimeLimits::default(), + }, + Arc::new(TestExecutionBackend::default()), + ) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let scoped = runtime + .create_worker_scoped( + "workspace-a", + scoped_task_request("persist workspace", "workspace-a"), + ) + .unwrap(); + let legacy = runtime + .create_worker(task_request("legacy persist")) + .unwrap(); + let recoverable_legacy = runtime + .create_worker(scoped_task_request( + "legacy recoverable persist", + "workspace-b", + )) + .unwrap(); + drop(runtime); + + let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { + root: root.clone(), + display_name: None, + limits: RuntimeLimits::default(), + }) + .unwrap(); + let restored_scoped = restored.worker_detail(&scoped.worker_ref).unwrap(); + assert_eq!(restored_scoped.workspace_id.as_deref(), Some("workspace-a")); + assert_eq!( + restored + .list_workers_scoped("workspace-a") + .unwrap() + .into_iter() + .map(|worker| worker.worker_ref) + .collect::>(), + vec![scoped.worker_ref.clone()] + ); + let legacy_error = restored + .worker_detail_scoped("workspace-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) + .unwrap(); + assert_eq!( + recovered_legacy.workspace_id.as_deref(), + Some("workspace-b") + ); + + let _ = std::fs::remove_dir_all(root); + } + #[cfg(feature = "fs-store")] #[test] fn fs_store_restores_active_worker_execution_handles() {