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

runtime: bind workspaces to issuer

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-30 13:42:53 +09:00
parent ac6c8b275d
commit d94ef81b43
No known key found for this signature in database
6 changed files with 374 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
crates/worker-runtime/src/error.rs
View File

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

5
crates/worker-runtime/src/fs_store.rs
View File

@ -292,6 +292,7 @@ pub(crate) struct PersistedRuntimeState {
pub(crate) next_event_id: u64,
pub(crate) next_diagnostic_id: u64,
pub(crate) workers: BTreeMap<WorkerId, PersistedWorkerRecord>,
pub(crate) workspace_owners: BTreeMap<String, String>,
pub(crate) config_bundles: BTreeMap<String, ConfigBundle>,
pub(crate) events: Vec<RuntimeEvent>,
pub(crate) diagnostics: Vec<RuntimeDiagnostic>,
@ -319,6 +320,8 @@ struct RuntimeSnapshot {
next_diagnostic_id: u64,
#[serde(default)]
config_bundles: BTreeMap<String, ConfigBundle>,
#[serde(default)]
workspace_owners: BTreeMap<String, String>,
diagnostics: Vec<RuntimeDiagnostic>,
}
@ -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,
}

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

@ -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<Query<RuntimeHttpWorkersQuery>, QueryRejection>,
) -> RestResult<RuntimeHttpWorkersResponse> {
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<String>,
) -> RestResult<RuntimeHttpWorkerResponse> {
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<String>,
) -> RestResult<RuntimeHttpWorkerDeleteResponse> {
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<Json<CreateWorkerRequest>, JsonRejection>,
) -> RestResult<RuntimeHttpWorkerResponse> {
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<String>,
) -> RestResult<RuntimeHttpWorkerResponse> {
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<Response, RuntimeHttpRestError> {
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<String>,
scope: Option<RuntimeWorkspaceScope>,
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<RuntimeHttpWorkerInputResponse> {
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<RuntimeHttpWorkerCompletionsResponse> {
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<RuntimeHttpWorkerLifecycleResponse> {
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
let ack = match auth_workspace_scope(&state, auth.as_ref())? {
Some(scope) => state
.runtime
.stop_worker_scoped(workspace_id, &worker_ref, request.reason)
}
.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<RuntimeHttpWorkerLifecycleResponse> {
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
let ack = match auth_workspace_scope(&state, auth.as_ref())? {
Some(scope) => state
.runtime
.cancel_worker_scoped(workspace_id, &worker_ref, request.reason)
}
.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<RuntimeAuthContext>>,
) -> Result<Option<&'a str>, RuntimeHttpRestError> {
auth: Option<&Extension<RuntimeAuthContext>>,
) -> Result<Option<RuntimeWorkspaceScope>, 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();

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

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

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

@ -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<String>, server_id: impl Into<String>) -> 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<WorkerDetail, RuntimeError> {
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<WorkerDetail, RuntimeError> {
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<Vec<WorkerSummary>, 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<Vec<WorkerSummary>, 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<WorkerDetail, RuntimeError> {
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<WorkerDetail, RuntimeError> {
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<WorkerInteractionAck, RuntimeError> {
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<Vec<protocol::CompletionEntry>, 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<Vec<Event>, 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<String>,
) -> Result<WorkerLifecycleAck, RuntimeError> {
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<String>,
) -> Result<WorkerLifecycleAck, RuntimeError> {
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<WorkerDeleteResult, RuntimeError> {
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<WorkerId, WorkerRecord>,
workspace_owners: BTreeMap<String, String>,
config_bundles: BTreeMap<String, ConfigBundle>,
events: Vec<RuntimeEvent>,
diagnostics: Vec<RuntimeDiagnostic>,
@ -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<bool, RuntimeError> {
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<bool, RuntimeError> {
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);
}

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

@ -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 { .. }