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

runtime: scope worker access by workspace

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-30 00:25:46 +09:00
parent 0ff8a95da3
commit 3dda06cbe3
No known key found for this signature in database
5 changed files with 754 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

4
crates/worker-runtime/src/catalog.rs
View File

@ -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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory: Option<WorkingDirectoryStatus>,
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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory: Option<WorkingDirectoryStatus>,
pub profile: ProfileSelector,
#[serde(default, skip_serializing_if = "Option::is_none")]

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

@ -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<String>,
pub(crate) working_directory: Option<WorkingDirectoryStatus>,
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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
working_directory: Option<WorkingDirectoryStatus>,
/// 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)

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

@ -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<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
query: Result<Query<RuntimeHttpWorkersQuery>, QueryRejection>,
) -> RestResult<RuntimeHttpWorkersResponse> {
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<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
) -> RestResult<RuntimeHttpWorkerResponse> {
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<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
) -> RestResult<RuntimeHttpWorkerDeleteResponse> {
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<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
body: Result<Json<CreateWorkerRequest>, JsonRejection>,
) -> RestResult<RuntimeHttpWorkerResponse> {
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<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
) -> RestResult<RuntimeHttpWorkerResponse> {
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<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
Query(query): Query<RuntimeWorkerEventsWsQuery>,
ws: WebSocketUpgrade,
) -> Result<Response, RuntimeHttpRestError> {
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<String>,
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<String>) -> protocol::Event {
async fn send_worker_input(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
body: Result<Json<WorkerInput>, JsonRejection>,
) -> RestResult<RuntimeHttpWorkerInputResponse> {
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<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
body: Result<Json<RuntimeHttpWorkerCompletionsRequest>, JsonRejection>,
) -> RestResult<RuntimeHttpWorkerCompletionsResponse> {
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<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
body: Bytes,
) -> RestResult<RuntimeHttpWorkerLifecycleResponse> {
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<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
body: Bytes,
) -> RestResult<RuntimeHttpWorkerLifecycleResponse> {
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<RuntimeAuthContext>>,
) -> Result<Option<&'a str>, 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<const N: usize>(
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<str>,
token: &str,
body: impl Into<Body>,
) -> Request<Body> {
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());

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

@ -320,11 +320,29 @@ impl Runtime {
pub fn create_worker(
&self,
request: CreateWorkerRequest,
) -> Result<WorkerDetail, RuntimeError> {
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<WorkerDetail, RuntimeError> {
self.create_worker_with_workspace(request, Some(workspace_id))
}
fn create_worker_with_workspace(
&self,
request: CreateWorkerRequest,
workspace_id: Option<&str>,
) -> 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)?;
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<Vec<WorkerSummary>, 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<Vec<WorkerSummary>, 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<WorkerDetail, RuntimeError> {
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<WorkerDetail, RuntimeError> {
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<WorkerDetail, RuntimeError> {
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<WorkerInteractionAck, RuntimeError> {
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<Vec<protocol::CompletionEntry>, 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<Vec<Event>, 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<String>,
) -> Result<WorkerLifecycleAck, RuntimeError> {
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<String>,
) -> Result<WorkerLifecycleAck, RuntimeError> {
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<WorkerDeleteResult, RuntimeError> {
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<String>,
request: CreateWorkerRequest,
working_directory: Option<CatalogWorkingDirectoryStatus>,
execution_handle: Option<WorkerExecutionHandle>,
@ -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<_>>(),
vec![workspace_a.worker_ref.clone()]
);
assert_eq!(
runtime
.list_workers_scoped("workspace-b")
.unwrap()
.into_iter()
.map(|worker| worker.worker_ref)
.collect::<Vec<_>>(),
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<_>>(),
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() {