From 0ffaa6c741bd105df264e037c3064a3c2aeafd1e Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 3 Aug 2026 17:08:23 +0900 Subject: [PATCH] workspace: remove worker credential refresh flow --- crates/worker-runtime/src/catalog.rs | 6 - crates/worker-runtime/src/execution.rs | 21 - crates/worker-runtime/src/http_server.rs | 13 - crates/worker-runtime/src/runtime.rs | 73 +--- crates/worker-runtime/src/worker_backend.rs | 58 +-- crates/worker/src/controller.rs | 16 +- crates/worker/src/feature/builtin/memory.rs | 1 + .../worker/src/feature/builtin/objective.rs | 1 + .../src/feature/builtin/session_explore.rs | 1 + crates/worker/src/feature/builtin/ticket.rs | 9 +- crates/worker/src/skill.rs | 11 +- crates/worker/src/worker.rs | 218 +++------- crates/workspace-server/src/hosts.rs | 25 +- crates/workspace-server/src/lib.rs | 4 +- crates/workspace-server/src/server.rs | 377 +++--------------- crates/workspace-server/src/store.rs | 315 +-------------- ...tored-worker-workspace-credential-stale.md | 11 +- 17 files changed, 184 insertions(+), 976 deletions(-) diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 035e6b28..d1a513e8 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -178,8 +178,6 @@ pub struct WorkspaceApiRef { pub base_url: String, #[serde(default, skip_serializing_if = "Option::is_none")] pub runtime_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub access_token: Option, } impl std::fmt::Debug for WorkspaceApiRef { @@ -189,10 +187,6 @@ impl std::fmt::Debug for WorkspaceApiRef { .field("workspace_id", &self.workspace_id) .field("base_url", &self.base_url) .field("runtime_id", &self.runtime_id) - .field( - "access_token", - &self.access_token.as_ref().map(|_| "[redacted]"), - ) .finish() } } diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 7bcc7439..6f5144a8 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -31,7 +31,6 @@ pub enum WorkerExecutionOperation { Restore, Input, ProtocolMethod, - ReplaceWorkspaceAccessToken, Stop, Cancel, } @@ -332,17 +331,6 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { Vec::new() } - fn replace_workspace_access_token( - &self, - _handle: &WorkerExecutionHandle, - _access_token: String, - ) -> WorkerExecutionResult { - WorkerExecutionResult::unsupported( - WorkerExecutionOperation::ReplaceWorkspaceAccessToken, - "execution backend does not support replacing Workspace access tokens", - ) - } - fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult { WorkerExecutionResult::unsupported( WorkerExecutionOperation::Stop, @@ -455,15 +443,6 @@ impl WorkerExecutionBackendRef { self.backend.worker_completions(handle, kind, prefix) } - pub(crate) fn replace_workspace_access_token( - &self, - handle: &WorkerExecutionHandle, - access_token: String, - ) -> WorkerExecutionResult { - self.backend - .replace_workspace_access_token(handle, access_token) - } - pub(crate) fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult { self.backend.stop_worker(handle) } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 20391dea..7a9c9114 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -1466,7 +1466,6 @@ mod tests { workspace_id: workspace_id.to_string(), base_url: format!("https://workspace.example/{workspace_id}"), runtime_id: None, - access_token: None, }); request } @@ -1796,17 +1795,6 @@ mod tests { ) } - fn replace_workspace_access_token( - &self, - _handle: &WorkerExecutionHandle, - _access_token: String, - ) -> WorkerExecutionResult { - WorkerExecutionResult::accepted( - WorkerExecutionOperation::ReplaceWorkspaceAccessToken, - WorkerExecutionRunState::Idle, - ) - } - fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult { WorkerExecutionResult::accepted( WorkerExecutionOperation::Stop, @@ -1908,7 +1896,6 @@ mod tests { workspace_id: "local".to_string(), base_url: "http://127.0.0.1:8787".to_string(), runtime_id: None, - access_token: Some("workspace-access-token".to_string()), }, }, ) diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index bcc52385..0980e5af 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -720,8 +720,7 @@ impl Runtime { Ok(()) } - /// Replace the Workspace API binding persisted for a Worker and update the - /// live execution when one is connected. + /// Replace the Workspace API identity binding persisted for a Worker. pub fn replace_worker_workspace_api_scoped( &self, scope: &RuntimeWorkspaceScope, @@ -743,17 +742,7 @@ impl Runtime { worker_ref: &WorkerRef, workspace_api: WorkspaceApiRef, ) -> Result { - let access_token = workspace_api - .access_token - .as_ref() - .filter(|token| !token.trim().is_empty()) - .cloned() - .ok_or_else(|| { - RuntimeError::InvalidRequest( - "Workspace API replacement requires an access token".to_string(), - ) - })?; - let (previous_workspace_api, live_execution) = { + let previous_workspace_api = { let state = self.lock()?; let worker = state.worker(worker_ref)?; if let Some(existing) = worker.request.workspace_api.as_ref() @@ -769,14 +758,7 @@ impl Runtime { .to_string(), )); } - let live_execution = match ( - state.execution_backend.clone(), - worker.execution_handle.clone(), - ) { - (Some(backend), Some(handle)) => Some((backend, handle)), - _ => None, - }; - (worker.request.workspace_api.clone(), live_execution) + worker.request.workspace_api.clone() }; { @@ -788,22 +770,6 @@ impl Runtime { } } - if let Some((backend, handle)) = live_execution { - let result = backend.replace_workspace_access_token(&handle, access_token); - if !result.is_accepted() { - let mut state = self.lock()?; - state.worker_mut(worker_ref)?.request.workspace_api = previous_workspace_api; - state.persist_runtime_snapshot()?; - return Err(RuntimeError::WorkerExecutionRejected { - worker_id: worker_ref.worker_id.clone(), - operation: result.operation, - outcome: result.outcome, - message: result.message_or_default(), - result, - }); - } - } - let state = self.lock()?; Ok(state.worker(worker_ref)?.detail()) } @@ -1163,8 +1129,7 @@ impl Runtime { WorkerExecutionOperation::Spawn | WorkerExecutionOperation::Restore | WorkerExecutionOperation::Input - | WorkerExecutionOperation::ProtocolMethod - | WorkerExecutionOperation::ReplaceWorkspaceAccessToken => return Ok(()), + | WorkerExecutionOperation::ProtocolMethod => return Ok(()), }; if result.is_accepted() { return Ok(()); @@ -2475,7 +2440,6 @@ mod tests { workspace_id: workspace_id.to_string(), base_url: format!("https://workspace.example/{workspace_id}"), runtime_id: None, - access_token: None, }); request } @@ -2518,7 +2482,6 @@ mod tests { restore_result: Mutex>, restore_count: Mutex, contexts: Mutex>, - workspace_access_tokens: Mutex>, #[cfg(feature = "ws-server")] snapshots: Mutex>, } @@ -2607,21 +2570,6 @@ mod tests { }) } - fn replace_workspace_access_token( - &self, - handle: &WorkerExecutionHandle, - access_token: String, - ) -> WorkerExecutionResult { - self.workspace_access_tokens - .lock() - .unwrap() - .insert(handle.worker_ref().worker_id.clone(), access_token); - WorkerExecutionResult::accepted( - WorkerExecutionOperation::ReplaceWorkspaceAccessToken, - WorkerExecutionRunState::Idle, - ) - } - fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult { WorkerExecutionResult::accepted( WorkerExecutionOperation::Stop, @@ -2899,8 +2847,8 @@ mod tests { } #[test] - fn workspace_api_replacement_updates_live_execution_and_persisted_request() { - let (runtime, backend) = runtime_and_backend(); + fn workspace_api_replacement_updates_persisted_request() { + let (runtime, _backend) = runtime_and_backend(); let scope = scope("workspace-a", "server-a"); let worker = runtime .create_worker_scoped( @@ -2912,21 +2860,12 @@ mod tests { workspace_id: "workspace-a".to_string(), base_url: "https://workspace.example/workspace-a/".to_string(), runtime_id: Some("runtime-a".to_string()), - access_token: Some("replacement-token".to_string()), }; runtime .replace_worker_workspace_api_scoped(&scope, &worker.worker_ref, replacement.clone()) .unwrap(); - assert_eq!( - backend - .workspace_access_tokens - .lock() - .unwrap() - .get(&worker.worker_ref.worker_id), - Some(&"replacement-token".to_string()) - ); let state = runtime.lock().unwrap(); assert_eq!( state diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 5955cba7..17242635 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -261,17 +261,22 @@ enum RuntimeWorkspaceBackendRef { Http { workspace_id: String, base_url: String, - access_token: Option, + runtime_id: String, }, } impl RuntimeWorkspaceBackendRef { fn from_worker_request(request: &CreateWorkerRequest) -> Self { - if let Some(api) = request.workspace_api.as_ref() { + if let Some(api) = request.workspace_api.as_ref() + && let Some(runtime_id) = api + .runtime_id + .as_ref() + .filter(|runtime_id| !runtime_id.trim().is_empty()) + { return Self::Http { workspace_id: api.workspace_id.clone(), base_url: api.base_url.clone(), - access_token: api.access_token.clone(), + runtime_id: runtime_id.clone(), }; } Self::None @@ -283,17 +288,15 @@ impl RuntimeWorkspaceBackendRef { Self::Http { workspace_id, base_url, - access_token, + runtime_id, } => WorkerWorkspaceContext::with_client( WorkspaceId::new(workspace_id.clone()).ok(), - Arc::new( - RuntimeWorkspaceHttpClient::new( - workspace_id.clone(), - base_url.clone(), - worker_ref.worker_id.to_string(), - ) - .with_access_token(access_token.clone()), - ), + Arc::new(RuntimeWorkspaceHttpClient::new( + workspace_id.clone(), + base_url.clone(), + runtime_id.clone(), + worker_ref.worker_id.to_string(), + )), ), } } @@ -1160,34 +1163,6 @@ where result } - fn replace_workspace_access_token( - &self, - handle: &WorkerExecutionHandle, - access_token: String, - ) -> WorkerExecutionResult { - let (worker, _busy) = match self.get_execution(handle) { - Ok(execution) => execution, - Err(mut result) => { - result.operation = WorkerExecutionOperation::ReplaceWorkspaceAccessToken; - return result; - } - }; - worker - .replace_workspace_access_token(access_token) - .map(|_| { - WorkerExecutionResult::accepted( - WorkerExecutionOperation::ReplaceWorkspaceAccessToken, - WorkerExecutionRunState::Idle, - ) - }) - .unwrap_or_else(|error| { - WorkerExecutionResult::errored( - WorkerExecutionOperation::ReplaceWorkspaceAccessToken, - error.to_string(), - ) - }) - } - fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult { if handle.backend_id() != self.backend_id() { return WorkerExecutionResult::rejected( @@ -1775,8 +1750,7 @@ mod tests { request.workspace_api = Some(crate::catalog::WorkspaceApiRef { workspace_id: "ws-test".to_string(), base_url: "http://127.0.0.1:3999".to_string(), - runtime_id: None, - access_token: None, + runtime_id: Some("runtime-test".to_string()), }); let detail = runtime.create_worker(request).unwrap(); diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 75fe0bc7..b1530b02 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -26,10 +26,7 @@ use crate::shutdown_after_idle::{ use crate::spawn::comm_tools::{read_worker_output_tool, send_to_worker_tool, stop_worker_tool}; use crate::spawn::registry::SpawnedWorkerRegistry; use crate::spawn::tool::spawn_worker_tool; -use crate::worker::{ - SystemItemCommitter, Worker, WorkerError, WorkerRunResult, WorkspaceClient, - WorkspaceClientError, -}; +use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult}; use protocol::{ AlertLevel, AlertSource, ErrorCode, Event, Method, RewindTargetId, RunResult, Segment, TurnResult, WorkerStatus, @@ -43,7 +40,6 @@ use protocol::{ pub struct WorkerHandle { method_tx: mpsc::Sender, event_tx: broadcast::Sender, - workspace_client: Arc, pub shared_state: Arc, pub runtime_dir: Arc, pub alerter: Alerter, @@ -119,14 +115,6 @@ impl WorkerHandle { pub fn alert(&self, level: AlertLevel, source: AlertSource, message: String) { self.alerter.alert(level, source, message); } - - /// Replace the Runtime-issued Workspace access token used by this live Worker. - pub fn replace_workspace_access_token( - &self, - access_token: String, - ) -> Result<(), WorkspaceClientError> { - self.workspace_client.replace_access_token(access_token) - } } async fn set_controller_status( @@ -248,7 +236,6 @@ impl WorkerController { let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); let (method_tx, method_rx) = mpsc::channel::(32); let (event_tx, _) = broadcast::channel::(256); - let workspace_client = worker.workspace_client_handle(); let alerter = Alerter::new(event_tx.clone()); let in_flight = InFlightEvents::new(event_tx.clone()); worker.attach_in_flight_events(in_flight.clone()); @@ -367,7 +354,6 @@ impl WorkerController { let handle = WorkerHandle { method_tx, event_tx: event_tx.clone(), - workspace_client, shared_state: shared_state.clone(), runtime_dir: runtime_dir.clone(), alerter: alerter.clone(), diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index 97fc334a..8cc008e9 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -348,6 +348,7 @@ mod tests { Arc::new(crate::worker::RuntimeWorkspaceHttpClient::new( "workspace", "http://backend", + "test-runtime", "test-worker", )) } diff --git a/crates/worker/src/feature/builtin/objective.rs b/crates/worker/src/feature/builtin/objective.rs index 49f46515..0a0b3d45 100644 --- a/crates/worker/src/feature/builtin/objective.rs +++ b/crates/worker/src/feature/builtin/objective.rs @@ -627,6 +627,7 @@ mod tests { crate::worker::RuntimeWorkspaceHttpClient::new( "workspace", "http://backend", + "test-runtime", "test-worker", ), ))); diff --git a/crates/worker/src/feature/builtin/session_explore.rs b/crates/worker/src/feature/builtin/session_explore.rs index b71d48b7..93fb43b1 100644 --- a/crates/worker/src/feature/builtin/session_explore.rs +++ b/crates/worker/src/feature/builtin/session_explore.rs @@ -662,6 +662,7 @@ mod tests { Arc::new(crate::worker::RuntimeWorkspaceHttpClient::new( "test-workspace", format!("http://{addr}"), + "test-runtime", "test-worker", )), rx, diff --git a/crates/worker/src/feature/builtin/ticket.rs b/crates/worker/src/feature/builtin/ticket.rs index 2f9ca265..d017850b 100644 --- a/crates/worker/src/feature/builtin/ticket.rs +++ b/crates/worker/src/feature/builtin/ticket.rs @@ -1372,6 +1372,7 @@ provider = "github" crate::worker::RuntimeWorkspaceHttpClient::new( "workspace-a", "not-a-url", + "test-runtime", "test-worker", ), )); @@ -1407,6 +1408,7 @@ provider = "github" let client = Arc::new(crate::worker::RuntimeWorkspaceHttpClient::new( "workspace-a", format!("http://{address}"), + "test-runtime", "worker-a", )); let backend = WorkspaceHttpTicketBackend::new(client); @@ -1448,7 +1450,12 @@ provider = "github" }); let backend = WorkspaceHttpTicketBackend::new(Arc::new( - crate::worker::RuntimeWorkspaceHttpClient::new("workspace-a", base_url, "test-worker"), + crate::worker::RuntimeWorkspaceHttpClient::new( + "workspace-a", + base_url, + "test-runtime", + "test-worker", + ), )); let created = backend.create(NewTicket::new("HTTP ticket")).unwrap(); diff --git a/crates/worker/src/skill.rs b/crates/worker/src/skill.rs index 2b1a1691..cef7a564 100644 --- a/crates/worker/src/skill.rs +++ b/crates/worker/src/skill.rs @@ -193,11 +193,15 @@ mod tests { let mut request_line = String::new(); reader.read_line(&mut request_line).unwrap(); assert!(request_line.starts_with("GET /api/w/ws-1/skills HTTP/1.1")); + let mut runtime_header = None; let mut worker_header = None; let mut authorization = None; loop { let mut line = String::new(); reader.read_line(&mut line).unwrap(); + if let Some(value) = line.strip_prefix("x-yoi-runtime-id: ") { + runtime_header = Some(value.trim().to_string()); + } if let Some(value) = line.strip_prefix("x-yoi-worker-id: ") { worker_header = Some(value.trim().to_string()); } @@ -208,8 +212,9 @@ mod tests { break; } } + assert_eq!(runtime_header.as_deref(), Some("runtime-test")); assert_eq!(worker_header.as_deref(), Some("test-worker")); - assert_eq!(authorization.as_deref(), Some("Bearer test-credential")); + assert_eq!(authorization, None); let body = serde_json::json!({ "authority": "workspace-backend-skills-v0", "entries": [{ @@ -234,9 +239,9 @@ mod tests { let client = crate::worker::RuntimeWorkspaceHttpClient::new( "ws-1", format!("http://{addr}"), + "runtime-test", "test-worker", - ) - .with_access_token(Some("test-credential".to_string())); + ); let catalog = (&client as &dyn WorkspaceClient).list_skills().unwrap(); assert_eq!(catalog.entries[0].name, "triage-errors"); assert_eq!(catalog.entries[0].provenance.id, "workspace:triage-errors"); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index aa3753eb..de7fa371 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -216,13 +216,6 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync { fn is_available(&self) -> bool; fn execute(&self, request: WorkspaceRequest) -> Result; - - /// Replace the Runtime-issued Workspace access token for this live client. - fn replace_access_token(&self, _access_token: String) -> Result<(), WorkspaceClientError> { - Err(WorkspaceClientError::Unavailable( - "Workspace client does not support access token replacement".to_string(), - )) - } } /// HTTP forwarding client created by Runtime for one concrete Worker execution. @@ -233,8 +226,8 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync { pub struct RuntimeWorkspaceHttpClient { workspace_id: String, base_url: String, + runtime_id: String, worker_id: String, - access_token: Mutex>, } impl std::fmt::Debug for RuntimeWorkspaceHttpClient { @@ -243,15 +236,8 @@ impl std::fmt::Debug for RuntimeWorkspaceHttpClient { .debug_struct("RuntimeWorkspaceHttpClient") .field("workspace_id", &self.workspace_id) .field("base_url", &self.base_url) + .field("runtime_id", &self.runtime_id) .field("worker_id", &self.worker_id) - .field( - "access_token", - &self - .access_token - .lock() - .ok() - .and_then(|token| token.as_ref().map(|_| "[redacted]")), - ) .finish() } } @@ -260,20 +246,16 @@ impl RuntimeWorkspaceHttpClient { pub fn new( workspace_id: impl Into, base_url: impl Into, + runtime_id: impl Into, worker_id: impl Into, ) -> Self { Self { workspace_id: workspace_id.into(), base_url: base_url.into().trim_end_matches('/').to_string(), + runtime_id: runtime_id.into(), worker_id: worker_id.into(), - access_token: Mutex::new(None), } } - - pub fn with_access_token(self, access_token: Option) -> Self { - *self.access_token.lock().expect("new credential mutex") = access_token; - self - } } impl WorkspaceClient for RuntimeWorkspaceHttpClient { @@ -294,105 +276,26 @@ impl WorkspaceClient for RuntimeWorkspaceHttpClient { request: WorkspaceRequest, ) -> Result { let base_url = self.base_url.clone(); + let runtime_id = self.runtime_id.clone(); let worker_id = self.worker_id.clone(); - let access_token = self - .access_token - .lock() - .map_err(|_| { - WorkspaceClientError::Request("workspace credential lock poisoned".to_string()) - })? - .clone(); - let request_copy = request.clone(); - let result = if tokio::runtime::Handle::try_current().is_ok() { + if tokio::runtime::Handle::try_current().is_ok() { std::thread::spawn(move || { - execute_runtime_workspace_http_with_refresh( - &base_url, - &worker_id, - access_token, - request_copy, - ) + execute_runtime_workspace_http(&base_url, &runtime_id, &worker_id, request) }) .join() .map_err(|_| { WorkspaceClientError::Request("workspace request thread panicked".to_string()) })? } else { - execute_runtime_workspace_http_with_refresh( - &base_url, - &worker_id, - access_token, - request, - ) - }?; - if let Some(new_token) = result.1 { - *self.access_token.lock().map_err(|_| { - WorkspaceClientError::Request("workspace credential lock poisoned".to_string()) - })? = Some(new_token); + execute_runtime_workspace_http(&base_url, &runtime_id, &worker_id, request) } - Ok(result.0) } - - fn replace_access_token(&self, access_token: String) -> Result<(), WorkspaceClientError> { - *self.access_token.lock().map_err(|_| { - WorkspaceClientError::Request("workspace credential lock poisoned".to_string()) - })? = Some(access_token); - Ok(()) - } -} - -fn execute_runtime_workspace_http_with_refresh( - base_url: &str, - worker_id: &str, - access_token: Option, - request: WorkspaceRequest, -) -> Result<(WorkspaceResponse, Option), WorkspaceClientError> { - let response = execute_runtime_workspace_http( - base_url, - worker_id, - access_token.as_deref(), - request.clone(), - )?; - if response.status != 401 { - return Ok((response, None)); - } - let Some(expired_token) = access_token else { - return Ok((response, None)); - }; - let workspace_id = request - .path - .strip_prefix("/api/w/") - .and_then(|path| path.split('/').next()) - .ok_or_else(|| WorkspaceClientError::InvalidPath(request.path.clone()))?; - let refresh_url = format!("{base_url}/api/w/{workspace_id}/worker-credentials/refresh"); - let refresh = reqwest::blocking::Client::new() - .post(refresh_url) - .bearer_auth(expired_token) - .header("x-yoi-worker-id", worker_id) - .send() - .map_err(|error| WorkspaceClientError::Request(error.to_string()))?; - if !refresh.status().is_success() { - return Ok((response, None)); - } - let body: serde_json::Value = refresh - .json() - .map_err(|error| WorkspaceClientError::Request(error.to_string()))?; - let new_token = body - .get("access_token") - .and_then(serde_json::Value::as_str) - .ok_or_else(|| { - WorkspaceClientError::Request( - "Workspace credential refresh response omitted access_token".to_string(), - ) - })? - .to_string(); - let retried = execute_runtime_workspace_http(base_url, worker_id, Some(&new_token), request)?; - Ok((retried, Some(new_token))) } fn execute_runtime_workspace_http( base_url: &str, + runtime_id: &str, worker_id: &str, - access_token: Option<&str>, request: WorkspaceRequest, ) -> Result { if !request.path.starts_with('/') || request.path.starts_with("//") { @@ -409,10 +312,8 @@ fn execute_runtime_workspace_http( let client = reqwest::blocking::Client::new(); let mut request_builder = client .request(method, url) + .header("x-yoi-runtime-id", runtime_id) .header("x-yoi-worker-id", worker_id); - if let Some(access_token) = access_token { - request_builder = request_builder.bearer_auth(access_token); - } if let Some(body) = request.body { request_builder = request_builder .header(reqwest::header::CONTENT_TYPE, "application/json") @@ -6132,6 +6033,7 @@ mod build_summary_prompt_tests { Arc::new(RuntimeWorkspaceHttpClient::new( "test-memory", format!("http://{addr}"), + "test-runtime", "test-worker", )), ) @@ -6268,6 +6170,7 @@ mod build_summary_prompt_tests { Arc::new(RuntimeWorkspaceHttpClient::new( "ws-skill", format!("http://{addr}"), + "test-runtime", "test-worker", )), ), @@ -6311,88 +6214,57 @@ mod build_summary_prompt_tests { } #[test] - fn runtime_workspace_client_refreshes_expired_credential_and_retries() { + fn runtime_workspace_client_sends_runtime_worker_identity_without_bearer() { use std::io::{BufRead, BufReader, Write}; use std::net::TcpListener; let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let address = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { - for step in 0..3 { - let (mut stream, _) = listener.accept().unwrap(); - let mut reader = BufReader::new(stream.try_clone().unwrap()); - let mut first_line = String::new(); - reader.read_line(&mut first_line).unwrap(); - let mut authorization = String::new(); - loop { - let mut line = String::new(); - reader.read_line(&mut line).unwrap(); - if let Some(value) = line.strip_prefix("authorization: ") { - authorization = value.trim().to_string(); - } - if line == "\r\n" || line.is_empty() { - break; - } + let (mut stream, _) = listener.accept().unwrap(); + let mut reader = BufReader::new(stream.try_clone().unwrap()); + let mut first_line = String::new(); + reader.read_line(&mut first_line).unwrap(); + assert!(first_line.contains("/api/w/workspace-a/tickets/search")); + let mut runtime_id = String::new(); + let mut worker_id = String::new(); + let mut authorization = String::new(); + loop { + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + if let Some(value) = line.strip_prefix("x-yoi-runtime-id: ") { + runtime_id = value.trim().to_string(); } - match step { - 0 => { - assert_eq!(authorization, "Bearer expired-token"); - stream - .write_all(b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\n\r\n") - .unwrap(); - } - 1 => { - assert!(first_line.contains("/worker-credentials/refresh")); - assert_eq!(authorization, "Bearer expired-token"); - let body = - r#"{"access_token":"fresh-token","expires_at":"2099-01-01T00:00:00Z"}"#; - write!( - stream, - "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}", - body.len(), - body - ) - .unwrap(); - } - _ => { - assert_eq!(authorization, "Bearer fresh-token"); - stream - .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}") - .unwrap(); - } + if let Some(value) = line.strip_prefix("x-yoi-worker-id: ") { + worker_id = value.trim().to_string(); + } + if let Some(value) = line.strip_prefix("authorization: ") { + authorization = value.trim().to_string(); + } + if line == "\r\n" || line.is_empty() { + break; } } + assert_eq!(runtime_id, "runtime-a"); + assert_eq!(worker_id, "worker-a"); + assert!(authorization.is_empty()); + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}") + .unwrap(); }); let client = RuntimeWorkspaceHttpClient::new( - "workspace-refresh", + "workspace-a", format!("http://{address}"), - "worker-refresh", - ) - .with_access_token(Some("expired-token".to_string())); + "runtime-a", + "worker-a", + ); let response = client - .execute(WorkspaceRequest::get( - "/api/w/workspace-refresh/tickets/search", - )) + .execute(WorkspaceRequest::get("/api/w/workspace-a/tickets/search")) .unwrap(); assert_eq!(response.status, 200); server.join().unwrap(); } - #[test] - fn runtime_workspace_client_can_install_missing_access_token() { - let client = - RuntimeWorkspaceHttpClient::new("workspace-a", "https://workspace.example", "worker-a"); - - client - .replace_access_token("replacement-token".to_string()) - .unwrap(); - - assert_eq!( - client.access_token.lock().unwrap().as_deref(), - Some("replacement-token") - ); - } - fn minimal_manifest() -> WorkerManifest { let toml_str = r#" [worker] diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index a7c1951e..e0216136 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -427,25 +427,13 @@ pub struct ConfigBundleListResult { fn required_worker_workspace_api( request: &WorkerSpawnRequest, ) -> Result { - let workspace_api = request.resolved_workspace_api.clone().ok_or_else(|| { + request.resolved_workspace_api.clone().ok_or_else(|| { diagnostic( - "worker_workspace_credential_missing", + "worker_workspace_api_missing", DiagnosticSeverity::Error, - "Workspace-bound Worker spawn requires a resolved Workspace API credential", + "Workspace-bound Worker spawn requires a resolved Workspace API binding", ) - })?; - if workspace_api - .access_token - .as_deref() - .is_none_or(|token| token.trim().is_empty()) - { - return Err(diagnostic( - "worker_workspace_credential_missing", - DiagnosticSeverity::Error, - "Workspace-bound Worker spawn requires a non-empty Workspace API access token", - )); - } - Ok(workspace_api) + }) } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -3808,7 +3796,6 @@ mod tests { workspace_id: "workspace-test".to_string(), base_url: "http://127.0.0.1:8787".to_string(), runtime_id: Some("runtime-test".to_string()), - access_token: Some("workspace-access-token".to_string()), } } @@ -4339,7 +4326,7 @@ mod tests { } #[test] - fn embedded_runtime_rejects_tokenless_workspace_spawn() { + fn embedded_runtime_rejects_missing_workspace_api_binding() { let runtime = EmbeddedWorkerRuntime::new_memory_with_execution_backend( "local:test", Arc::new(AcceptingExecutionBackend::default()), @@ -4355,7 +4342,7 @@ mod tests { spawned .diagnostics .iter() - .any(|diagnostic| { diagnostic.code == "worker_workspace_credential_missing" }) + .any(|diagnostic| { diagnostic.code == "worker_workspace_api_missing" }) ); } diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index 8a3f4d40..ccd17b74 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -89,8 +89,8 @@ pub enum Error { WorkspaceIdMismatch, #[error("Ticket assignment conflict: {0}")] TicketAssignmentConflict(String), - #[error("Worker Workspace authentication failed: {0}")] - WorkerWorkspaceAuthentication(String), + #[error("Worker source identity is invalid: {0}")] + WorkerSourceIdentity(String), #[error("workspace identity error: {0}")] WorkspaceIdentity(String), #[error("store error: {0}")] diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 1d4be92a..2d028889 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -65,8 +65,7 @@ use crate::hosts::{ WorkerInputKind, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult, WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, - WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerWorkspaceApiResult, - WorkerWorkspaceSummary, + WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerWorkspaceSummary, }; use crate::identity::WorkspaceIdentity; use crate::memory_backend::execute_memory_backend_operation_with_authority; @@ -96,7 +95,7 @@ use crate::store::{ AccountRecord, ApiTokenRecord, AuthChallengeRecord, BrowserSessionRecord, ControlPlaneStore, DeviceLoginFlowRecord, PasskeyCredentialRecord, RepositoryRecord, TicketWorkerAssignmentRecord, UserRecord, WorkdirRegistryRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord, - WorkerWorkspaceCredentialRecord, WorkspaceRecord, + WorkspaceRecord, }; use crate::{Error, Result}; use worker_runtime::catalog::{ @@ -251,7 +250,6 @@ pub struct WorkspaceApi { observation_proxy: BackendObservationProxy, runtime_subscription_broker: RuntimeSubscriptionBroker, resource_broker: BackendResourceBroker, - credential_operation_lock: Arc>, } impl WorkspaceApi { @@ -351,7 +349,6 @@ impl WorkspaceApi { observation_proxy, runtime_subscription_broker, resource_broker, - credential_operation_lock: Arc::new(std::sync::Mutex::new(())), }) } @@ -363,49 +360,18 @@ impl WorkspaceApi { &self.runtime_subscription_broker } - fn mint_worker_workspace_credential( - &self, - runtime_id: &str, - worker_id: Option<&str>, - ) -> ApiResult<(WorkerWorkspaceCredentialRecord, WorkspaceApiRef)> { - let now = Utc::now(); - let token = mint_secret("wac"); - let credential = WorkerWorkspaceCredentialRecord { - credential_id: new_id("wac"), - token: token.clone(), + fn workspace_api_ref(&self, runtime_id: &str) -> WorkspaceApiRef { + WorkspaceApiRef { workspace_id: self.config.workspace_id.clone(), - runtime_id: runtime_id.to_string(), - worker_id: worker_id.map(ToOwned::to_owned), - created_at: now.to_rfc3339_opts(SecondsFormat::Secs, true), - expires_at: (now + chrono::Duration::hours(1)) - .to_rfc3339_opts(SecondsFormat::Secs, true), - revoked_at: None, - }; - self.store.upsert_worker_workspace_credential(&credential)?; - Ok(( - credential, - WorkspaceApiRef { - workspace_id: self.config.workspace_id.clone(), - base_url: self - .config - .backend_base_url - .clone() - .unwrap_or_else(|| "http://127.0.0.1:8787".to_string()) - .trim_end_matches('/') - .to_string(), - runtime_id: Some(runtime_id.to_string()), - access_token: Some(token), - }, - )) - } - - fn revoke_credential_record( - &self, - credential: &mut WorkerWorkspaceCredentialRecord, - ) -> ApiResult<()> { - credential.revoked_at = Some(Utc::now().to_rfc3339()); - self.store.upsert_worker_workspace_credential(credential)?; - Ok(()) + base_url: self + .config + .backend_base_url + .clone() + .unwrap_or_else(|| "http://127.0.0.1:8787".to_string()) + .trim_end_matches('/') + .to_string(), + runtime_id: Some(runtime_id.to_string()), + } } fn spawn_workspace_worker( @@ -413,18 +379,13 @@ impl WorkspaceApi { runtime_id: &str, mut request: WorkerSpawnRequest, ) -> ApiResult { - let (mut credential, workspace_api) = - self.mint_worker_workspace_credential(runtime_id, None)?; + let workspace_api = self.workspace_api_ref(runtime_id); request.resolved_workspace_api = Some(workspace_api.clone()); - let result = match self.runtime.spawn_worker(runtime_id, request) { - Ok(result) => result, - Err(error) => { - self.revoke_credential_record(&mut credential)?; - return Err(error.into_error().into()); - } - }; + let result = self + .runtime + .spawn_worker(runtime_id, request) + .map_err(|error| error.into_error())?; let Some(worker) = result.worker.as_ref() else { - self.revoke_credential_record(&mut credential)?; return Ok(result); }; let replacement = match self.runtime.replace_worker_workspace_api( @@ -435,13 +396,11 @@ impl WorkspaceApi { Ok(replacement) => replacement, Err(error) => { let _ = self.runtime.delete_worker(runtime_id, &worker.worker_id); - self.revoke_credential_record(&mut credential)?; return Err(error.into_error().into()); } }; if replacement.state != WorkerOperationState::Accepted { let _ = self.runtime.delete_worker(runtime_id, &worker.worker_id); - self.revoke_credential_record(&mut credential)?; return Err(Error::RuntimeOperationFailed { runtime_id: runtime_id.to_string(), code: "worker_workspace_api_replace_failed".to_string(), @@ -455,50 +414,6 @@ impl WorkspaceApi { } .into()); } - credential.worker_id = Some(worker.worker_id.clone()); - self.store.upsert_worker_workspace_credential(&credential)?; - self.store.revoke_worker_workspace_credentials_except( - &self.config.workspace_id, - runtime_id, - &worker.worker_id, - &credential.credential_id, - &Utc::now().to_rfc3339(), - )?; - Ok(result) - } - - fn rotate_worker_workspace_credential( - &self, - runtime_id: &str, - worker_id: &str, - ) -> ApiResult { - let _credential_guard = self.credential_operation_lock.lock().map_err(|_| { - Error::Config("Workspace credential operation lock poisoned".to_string()) - })?; - let (mut credential, workspace_api) = - self.mint_worker_workspace_credential(runtime_id, Some(worker_id))?; - let result = - match self - .runtime - .replace_worker_workspace_api(runtime_id, worker_id, workspace_api) - { - Ok(result) => result, - Err(error) => { - self.revoke_credential_record(&mut credential)?; - return Err(error.into_error().into()); - } - }; - if result.state != WorkerOperationState::Accepted { - self.revoke_credential_record(&mut credential)?; - return Ok(result); - } - self.store.revoke_worker_workspace_credentials_except( - &self.config.workspace_id, - runtime_id, - worker_id, - &credential.credential_id, - &Utc::now().to_rfc3339(), - )?; Ok(result) } @@ -507,12 +422,15 @@ impl WorkspaceApi { runtime_id: &str, worker_id: &str, ) -> ApiResult { - let rotation = self.rotate_worker_workspace_credential(runtime_id, worker_id)?; - if rotation.state != WorkerOperationState::Accepted { + let binding = self + .runtime + .replace_worker_workspace_api(runtime_id, worker_id, self.workspace_api_ref(runtime_id)) + .map_err(|error| error.into_error())?; + if binding.state != WorkerOperationState::Accepted { return Ok(WorkerRestoreResult { - state: rotation.state, - worker: rotation.worker, - diagnostics: rotation.diagnostics, + state: binding.state, + worker: binding.worker, + diagnostics: binding.diagnostics, }); } Ok(self @@ -656,10 +574,6 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/tickets", get(scoped_list_tickets).post(scoped_create_ticket_record), ) - .route( - "/api/w/{workspace_id}/worker-credentials/refresh", - post(scoped_refresh_worker_workspace_credential), - ) .route( "/api/w/{workspace_id}/memory", get(scoped_get_memory_document), @@ -2256,63 +2170,6 @@ async fn scoped_close_ticket( browser_ticket_detail(&api, &path.id) } -#[derive(Debug, Serialize)] -struct WorkerWorkspaceCredentialRefreshResponse { - access_token: String, - expires_at: String, -} - -async fn scoped_refresh_worker_workspace_credential( - State(api): State, - AxumPath(path): AxumPath, - headers: HeaderMap, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - let _credential_guard = api - .credential_operation_lock - .lock() - .map_err(|_| Error::Config("Workspace credential operation lock poisoned".to_string()))?; - let token = headers - .get(axum::http::header::AUTHORIZATION) - .and_then(|value| value.to_str().ok()) - .and_then(|value| value.strip_prefix("Bearer ")) - .filter(|value| !value.trim().is_empty()) - .ok_or_else(|| { - Error::WorkerWorkspaceAuthentication("missing expired credential".to_string()) - })?; - let worker_id = headers - .get("x-yoi-worker-id") - .and_then(|value| value.to_str().ok()) - .filter(|value| !value.trim().is_empty()) - .ok_or_else(|| { - Error::WorkerWorkspaceAuthentication("missing Runtime-bound Worker id".to_string()) - })?; - let new_token = mint_secret("wac"); - let expires_at = - (Utc::now() + chrono::Duration::hours(1)).to_rfc3339_opts(SecondsFormat::Secs, true); - let credential = api - .store - .refresh_worker_workspace_credential( - token, - &path.workspace_id, - worker_id, - &new_token, - &expires_at, - )? - .ok_or_else(|| { - Error::WorkerWorkspaceAuthentication("credential cannot be refreshed".to_string()) - })?; - api.runtime - .worker(&credential.runtime_id, worker_id) - .map_err(|_| { - Error::WorkerWorkspaceAuthentication("credential Worker no longer exists".to_string()) - })?; - Ok(Json(WorkerWorkspaceCredentialRefreshResponse { - access_token: new_token, - expires_at, - })) -} - async fn execute_worker_ticket_rest_operation( api: &WorkspaceApi, workspace_id: &str, @@ -2320,12 +2177,6 @@ async fn execute_worker_ticket_rest_operation( mut operation: TicketBackendOperation, ) -> ApiResult { validate_workspace_scope(api, workspace_id)?; - if headers.get(axum::http::header::AUTHORIZATION).is_none() { - return Err(Error::WorkerWorkspaceAuthentication( - "missing Runtime Workspace credential".to_string(), - ) - .into()); - } let config = ticket::config::TicketConfig::load_workspace(&api.config.workspace_root) .map_err(|error| Error::Config(format!("load Ticket workspace settings: {error}")))?; let mut backend = SqliteTicketBackend::new( @@ -3119,39 +2970,26 @@ fn build_ticket_notification_hook( fn authenticate_worker_mutation_source( api: &WorkspaceApi, - workspace_id: &str, + _workspace_id: &str, headers: &HeaderMap, ) -> Result { - let token = headers - .get(axum::http::header::AUTHORIZATION) + let runtime_id = headers + .get("x-yoi-runtime-id") .and_then(|value| value.to_str().ok()) - .and_then(|value| value.strip_prefix("Bearer ")) .filter(|value| !value.trim().is_empty()) - .ok_or_else(|| { - Error::WorkerWorkspaceAuthentication("missing Runtime Workspace credential".to_string()) - })?; + .ok_or_else(|| Error::WorkerSourceIdentity("missing Runtime id".to_string()))?; let worker_id = headers .get("x-yoi-worker-id") .and_then(|value| value.to_str().ok()) .filter(|value| !value.trim().is_empty()) .ok_or_else(|| { - Error::WorkerWorkspaceAuthentication("missing Runtime-bound Worker id".to_string()) - })?; - let credential = api - .store - .authenticate_worker_workspace_credential(token, workspace_id, worker_id)? - .ok_or_else(|| { - Error::WorkerWorkspaceAuthentication("invalid Runtime Workspace credential".to_string()) - })?; - api.runtime - .worker(&credential.runtime_id, worker_id) - .map_err(|_| { - Error::WorkerWorkspaceAuthentication( - "credential does not identify a current Runtime Worker".to_string(), - ) + Error::WorkerSourceIdentity("missing Runtime-bound Worker id".to_string()) })?; + api.runtime.worker(runtime_id, worker_id).map_err(|_| { + Error::WorkerSourceIdentity("Runtime-bound Worker identity does not exist".to_string()) + })?; Ok(WorkerMutationSource { - runtime_id: credential.runtime_id, + runtime_id: runtime_id.to_string(), worker_id: worker_id.to_string(), }) } @@ -4530,15 +4368,7 @@ fn cleanup_runtime_worker_for_execution( .runtime .delete_worker(runtime_id, candidate.runtime_worker_id.as_str()) { - Ok(result) if result.deleted && result.state == WorkerOperationState::Accepted => { - api.store.revoke_worker_workspace_credentials( - &api.config.workspace_id, - runtime_id, - candidate.runtime_worker_id.as_str(), - &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), - )?; - Ok(()) - } + Ok(result) if result.deleted && result.state == WorkerOperationState::Accepted => Ok(()), Ok(result) => Err(ApiError::with_diagnostics( Error::RuntimeOperationFailed { runtime_id: runtime_id.to_string(), @@ -8745,7 +8575,7 @@ impl IntoResponse for ApiError { fn into_response(self) -> Response { let status = match &self.error { Error::TicketAssignmentConflict(_) => StatusCode::CONFLICT, - Error::WorkerWorkspaceAuthentication(_) => StatusCode::UNAUTHORIZED, + Error::WorkerSourceIdentity(_) => StatusCode::BAD_REQUEST, Error::InvalidRuntimeIdentifier { .. } => StatusCode::BAD_REQUEST, Error::Ticket(ticket::TicketError::NotFound(_)) => StatusCode::NOT_FOUND, Error::Ticket( @@ -8880,7 +8710,6 @@ mod tests { workspace_id: TEST_WORKSPACE_ID.to_string(), base_url: "http://127.0.0.1:8787".to_string(), runtime_id: Some(runtime_id.to_string()), - access_token: Some("workspace-access-token".to_string()), } } @@ -9535,17 +9364,6 @@ mod tests { } } - fn replace_workspace_access_token( - &self, - _handle: &worker_runtime::execution::WorkerExecutionHandle, - _access_token: String, - ) -> worker_runtime::execution::WorkerExecutionResult { - worker_runtime::execution::WorkerExecutionResult::accepted( - worker_runtime::execution::WorkerExecutionOperation::ReplaceWorkspaceAccessToken, - worker_runtime::execution::WorkerExecutionRunState::Idle, - ) - } - fn dispatch_input( &self, handle: &worker_runtime::execution::WorkerExecutionHandle, @@ -9912,22 +9730,10 @@ mod tests { false, ) .unwrap(); - api.store - .upsert_worker_workspace_credential(&WorkerWorkspaceCredentialRecord { - credential_id: "source-credential".to_string(), - token: "source-secret".to_string(), - workspace_id: TEST_WORKSPACE_ID.to_string(), - runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), - worker_id: Some(source_worker.worker_id.clone()), - created_at: TEST_CREATED_AT.to_string(), - expires_at: "2099-01-01T00:00:00Z".to_string(), - revoked_at: None, - }) - .unwrap(); let mut headers = HeaderMap::new(); headers.insert( - axum::http::header::AUTHORIZATION, - axum::http::HeaderValue::from_static("Bearer source-secret"), + "x-yoi-runtime-id", + axum::http::HeaderValue::from_static(EMBEDDED_WORKER_RUNTIME_ID), ); headers.insert( "x-yoi-worker-id", @@ -9942,7 +9748,7 @@ mod tests { ticket_ref.id )) .header("content-type", "application/json") - .header("authorization", "Bearer source-secret") + .header("x-yoi-runtime-id", EMBEDDED_WORKER_RUNTIME_ID) .header("x-yoi-worker-id", &source_worker.worker_id) .body(Body::from( serde_json::to_vec(&NewTicketEvent::new( @@ -9995,7 +9801,7 @@ mod tests { "/api/w/{TEST_WORKSPACE_ID}/tickets/{}/record", ticket_ref.id )) - .header("authorization", "Bearer source-secret") + .header("x-yoi-runtime-id", EMBEDDED_WORKER_RUNTIME_ID) .header("x-yoi-worker-id", &source_worker.worker_id) .body(Body::empty()) .unwrap(), @@ -10099,7 +9905,7 @@ mod tests { Some("coder") ); - let unauthorized = execute_worker_ticket_test_operation( + let invalid_source = execute_worker_ticket_test_operation( State(api.clone()), AxumPath(ScopedWorkspacePath { workspace_id: TEST_WORKSPACE_ID.to_string(), @@ -10113,7 +9919,7 @@ mod tests { .await .unwrap_err() .into_response(); - assert_eq!(unauthorized.status(), StatusCode::UNAUTHORIZED); + assert_eq!(invalid_source.status(), StatusCode::BAD_REQUEST); } #[tokio::test] @@ -10183,26 +9989,14 @@ mod tests { }, ) .unwrap(); - api.store - .upsert_worker_workspace_credential(&WorkerWorkspaceCredentialRecord { - credential_id: "orchestrator-source-credential".to_string(), - token: "orchestrator-source-secret".to_string(), - workspace_id: TEST_WORKSPACE_ID.to_string(), - runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), - worker_id: Some(source.worker_id.clone()), - created_at: TEST_CREATED_AT.to_string(), - expires_at: "2099-01-01T00:00:00Z".to_string(), - revoked_at: None, - }) - .unwrap(); let backend = browser_ticket_backend(&api).unwrap(); let mut input = ticket::NewTicket::new("Queued notification"); input.workflow_state = Some(TicketWorkflowState::Queued); let ticket_ref = backend.create(input).unwrap(); let mut headers = HeaderMap::new(); headers.insert( - axum::http::header::AUTHORIZATION, - axum::http::HeaderValue::from_static("Bearer orchestrator-source-secret"), + "x-yoi-runtime-id", + axum::http::HeaderValue::from_static(EMBEDDED_WORKER_RUNTIME_ID), ); headers.insert( "x-yoi-worker-id", @@ -10234,66 +10028,6 @@ mod tests { ); } - #[tokio::test] - async fn workspace_credential_rotation_revokes_prior_bound_credential() { - let dir = tempfile::tempdir().unwrap(); - let api = test_api(dir.path()).await; - let result = api - .spawn_workspace_worker( - EMBEDDED_WORKER_RUNTIME_ID, - WorkerSpawnRequest { - requested_worker_name: Some("credential-boundary".to_string()), - intent: WorkerSpawnIntent::WorkspaceCoding, - acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { - expected_segments: 0, - }, - profile: ProfileSelector::Builtin("builtin:coder".to_string()), - ticket_assignment: None, - initial_input: None, - working_directory_request: None, - resolved_working_directory_request: None, - resolved_working_directory: None, - resolved_config_bundle: None, - resolved_workspace_api: Some(test_worker_workspace_api( - "embedded-worker-runtime", - )), - }, - ) - .unwrap(); - let worker = result.worker.unwrap(); - let old_token = "legacy-worker-token"; - api.store - .upsert_worker_workspace_credential(&WorkerWorkspaceCredentialRecord { - credential_id: new_id("wac"), - token: old_token.to_string(), - workspace_id: TEST_WORKSPACE_ID.to_string(), - runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), - worker_id: Some(worker.worker_id.clone()), - created_at: Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), - expires_at: (Utc::now() + chrono::Duration::hours(1)) - .to_rfc3339_opts(SecondsFormat::Secs, true), - revoked_at: None, - }) - .unwrap(); - - let rotation = api - .rotate_worker_workspace_credential(EMBEDDED_WORKER_RUNTIME_ID, &worker.worker_id) - .unwrap(); - - assert_eq!(rotation.state, WorkerOperationState::Accepted); - assert!( - api.store - .authenticate_worker_workspace_credential( - old_token, - TEST_WORKSPACE_ID, - &worker.worker_id, - ) - .unwrap() - .is_none(), - "repair must revoke the prior bound credential" - ); - } - #[tokio::test] async fn worker_spawn_and_restore_assignment_operations_are_idempotent() { let dir = tempfile::tempdir().unwrap(); @@ -11500,7 +11234,7 @@ mod tests { } #[tokio::test] - async fn ticket_rest_search_requires_worker_credential_and_rpc_route_is_removed() { + async fn ticket_rest_search_requires_worker_source_identity_and_rpc_route_is_removed() { let dir = tempfile::tempdir().unwrap(); let api = test_api(dir.path()).await; let app = build_router(api); @@ -11518,7 +11252,24 @@ mod tests { ) .await .unwrap(); - assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + + let response = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(format!( + "/api/w/{TEST_WORKSPACE_ID}/tickets/search?state=active" + )) + .header("x-yoi-runtime-id", "embedded-worker-runtime") + .header("x-yoi-worker-id", "missing-worker") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); let response = app .oneshot( diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index aaa9bc9b..4caee5fc 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -117,6 +117,11 @@ const MIGRATIONS: &[Migration] = &[ name: "reconcile Workdir revision and crash safe Worker lifecycle reservations", apply: strengthen_ticket_assignment_lifecycle_reservations, }, + Migration { + version: 21, + name: "remove per-Worker Workspace credentials", + apply: remove_worker_workspace_credentials, + }, ]; struct Migration { @@ -291,18 +296,6 @@ pub struct TicketWorkerAssignmentUpdate { pub previous: Option, } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub struct WorkerWorkspaceCredentialRecord { - pub credential_id: String, - pub token: String, - pub workspace_id: String, - pub runtime_id: String, - pub worker_id: Option, - pub created_at: String, - pub expires_at: String, - pub revoked_at: Option, -} - #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct TicketNotificationRecipient { pub runtime_id: String, @@ -632,39 +625,6 @@ pub trait ControlPlaneStore: Send + Sync { limit: usize, ) -> Result>; - fn upsert_worker_workspace_credential( - &self, - record: &WorkerWorkspaceCredentialRecord, - ) -> Result<()>; - fn authenticate_worker_workspace_credential( - &self, - token: &str, - workspace_id: &str, - worker_id: &str, - ) -> Result>; - fn refresh_worker_workspace_credential( - &self, - token: &str, - workspace_id: &str, - worker_id: &str, - new_token: &str, - new_expires_at: &str, - ) -> Result>; - fn revoke_worker_workspace_credentials_except( - &self, - workspace_id: &str, - runtime_id: &str, - worker_id: &str, - active_credential_id: &str, - revoked_at: &str, - ) -> Result<()>; - fn revoke_worker_workspace_credentials( - &self, - workspace_id: &str, - runtime_id: &str, - worker_id: &str, - revoked_at: &str, - ) -> Result<()>; fn enqueue_ticket_notification( &self, notification_id: &str, @@ -2223,171 +2183,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } - fn upsert_worker_workspace_credential( - &self, - record: &WorkerWorkspaceCredentialRecord, - ) -> Result<()> { - self.with_conn(|conn| { - conn.execute( - r#"INSERT INTO worker_workspace_credentials ( - credential_id, token, workspace_id, runtime_id, worker_id, created_at, - expires_at, revoked_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) - ON CONFLICT(credential_id) DO UPDATE SET - token = excluded.token, - workspace_id = excluded.workspace_id, - runtime_id = excluded.runtime_id, - worker_id = excluded.worker_id, - created_at = excluded.created_at, - expires_at = excluded.expires_at, - revoked_at = excluded.revoked_at"#, - params![ - record.credential_id, - record.token, - record.workspace_id, - record.runtime_id, - record.worker_id, - record.created_at, - record.expires_at, - record.revoked_at, - ], - )?; - Ok(()) - }) - } - - fn authenticate_worker_workspace_credential( - &self, - token: &str, - workspace_id: &str, - worker_id: &str, - ) -> Result> { - self.with_conn(|conn| { - let tx = conn.unchecked_transaction()?; - let record = tx - .query_row( - r#"SELECT credential_id, token, workspace_id, runtime_id, worker_id, created_at, - expires_at, revoked_at - FROM worker_workspace_credentials - WHERE token = ?1 AND workspace_id = ?2 - AND revoked_at IS NULL AND datetime(expires_at) > datetime('now')"#, - params![token, workspace_id], - |row| { - Ok(WorkerWorkspaceCredentialRecord { - credential_id: row.get(0)?, - token: row.get(1)?, - workspace_id: row.get(2)?, - runtime_id: row.get(3)?, - worker_id: row.get(4)?, - created_at: row.get(5)?, - expires_at: row.get(6)?, - revoked_at: row.get(7)?, - }) - }, - ) - .optional()?; - let Some(mut record) = record else { - tx.commit()?; - return Ok(None); - }; - if record.worker_id.as_deref().is_some_and(|bound| bound != worker_id) { - tx.commit()?; - return Ok(None); - } - if record.worker_id.is_none() { - tx.execute( - "UPDATE worker_workspace_credentials SET worker_id = ?1 WHERE credential_id = ?2 AND worker_id IS NULL", - params![worker_id, record.credential_id], - )?; - record.worker_id = Some(worker_id.to_string()); - } - tx.commit()?; - Ok(Some(record)) - }) - } - - fn refresh_worker_workspace_credential( - &self, - token: &str, - workspace_id: &str, - worker_id: &str, - new_token: &str, - new_expires_at: &str, - ) -> Result> { - self.with_conn(|conn| { - let tx = conn.unchecked_transaction()?; - let record = tx - .query_row( - r#"SELECT credential_id, runtime_id, created_at FROM worker_workspace_credentials - WHERE token = ?1 AND workspace_id = ?2 AND worker_id = ?3 AND revoked_at IS NULL"#, - params![token, workspace_id, worker_id], - |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?)), - ) - .optional()?; - let Some((credential_id, runtime_id, created_at)) = record else { - tx.commit()?; - return Ok(None); - }; - tx.execute( - "UPDATE worker_workspace_credentials SET token = ?1, expires_at = ?2 WHERE credential_id = ?3", - params![new_token, new_expires_at, credential_id], - )?; - tx.commit()?; - Ok(Some(WorkerWorkspaceCredentialRecord { - credential_id, - token: new_token.to_string(), - workspace_id: workspace_id.to_string(), - runtime_id, - worker_id: Some(worker_id.to_string()), - created_at, - expires_at: new_expires_at.to_string(), - revoked_at: None, - })) - }) - } - - fn revoke_worker_workspace_credentials_except( - &self, - workspace_id: &str, - runtime_id: &str, - worker_id: &str, - active_credential_id: &str, - revoked_at: &str, - ) -> Result<()> { - self.with_conn(|conn| { - conn.execute( - r#"UPDATE worker_workspace_credentials SET revoked_at = ?5 - WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 - AND credential_id <> ?4 AND revoked_at IS NULL"#, - params![ - workspace_id, - runtime_id, - worker_id, - active_credential_id, - revoked_at - ], - )?; - Ok(()) - }) - } - - fn revoke_worker_workspace_credentials( - &self, - workspace_id: &str, - runtime_id: &str, - worker_id: &str, - revoked_at: &str, - ) -> Result<()> { - self.with_conn(|conn| { - conn.execute( - r#"UPDATE worker_workspace_credentials SET revoked_at = ?4 - WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND revoked_at IS NULL"#, - params![workspace_id, runtime_id, worker_id, revoked_at], - )?; - Ok(()) - }) - } - fn enqueue_ticket_notification( &self, notification_id: &str, @@ -3543,6 +3338,11 @@ fn strengthen_ticket_assignment_lifecycle_reservations(conn: &Connection) -> Res Ok(()) } +fn remove_worker_workspace_credentials(conn: &Connection) -> Result<()> { + conn.execute_batch("DROP TABLE IF EXISTS worker_workspace_credentials;")?; + Ok(()) +} + fn create_objective_event_tables(conn: &Connection) -> Result<()> { conn.execute_batch( r#" @@ -4225,8 +4025,12 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 20); - + assert_eq!(store.schema_version().await.unwrap(), 21); + assert!( + !store + .with_conn(|conn| table_exists(conn, "worker_workspace_credentials")) + .unwrap() + ); let record = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -4238,7 +4042,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 20); + assert_eq!(reopened.schema_version().await.unwrap(), 21); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) @@ -4485,7 +4289,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); } #[tokio::test] - async fn worker_credential_binds_once_and_notification_outbox_is_durable() { + async fn notification_outbox_is_durable() { let dir = tempfile::tempdir().unwrap(); let db = dir.path().join("server.db"); let store = SqliteWorkspaceStore::open(&db).unwrap(); @@ -4500,81 +4304,6 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); }) .await .unwrap(); - store - .upsert_worker_workspace_credential(&WorkerWorkspaceCredentialRecord { - credential_id: "credential-1".to_string(), - token: "secret-token".to_string(), - workspace_id: "workspace-a".to_string(), - runtime_id: "runtime-1".to_string(), - worker_id: None, - created_at: "2026-07-31T00:00:01Z".to_string(), - expires_at: "2099-01-01T00:00:00Z".to_string(), - revoked_at: None, - }) - .unwrap(); - let bound = store - .authenticate_worker_workspace_credential("secret-token", "workspace-a", "worker-1") - .unwrap() - .unwrap(); - assert_eq!(bound.worker_id.as_deref(), Some("worker-1")); - let refreshed = store - .refresh_worker_workspace_credential( - "secret-token", - "workspace-a", - "worker-1", - "refreshed-token", - "2099-02-01T00:00:00Z", - ) - .unwrap() - .unwrap(); - assert_eq!(refreshed.token, "refreshed-token"); - assert!(store - .authenticate_worker_workspace_credential( - "secret-token", - "workspace-a", - "worker-1", - ) - .unwrap() - .is_none()); - assert!( - store - .authenticate_worker_workspace_credential( - "refreshed-token", - "workspace-a", - "worker-1", - ) - .unwrap() - .is_some() - ); - store - .revoke_worker_workspace_credentials( - "workspace-a", - "runtime-1", - "worker-1", - "2026-08-01T00:00:00Z", - ) - .unwrap(); - assert!( - store - .authenticate_worker_workspace_credential( - "refreshed-token", - "workspace-a", - "worker-1", - ) - .unwrap() - .is_none() - ); - assert!( - store - .authenticate_worker_workspace_credential( - "secret-token", - "workspace-a", - "worker-2", - ) - .unwrap() - .is_none() - ); - store .enqueue_ticket_notification( "notification-1", @@ -4850,7 +4579,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); .unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 20); + assert_eq!(store.schema_version().await.unwrap(), 21); store .with_conn(|conn| { @@ -5036,7 +4765,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn repository_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 20); + assert_eq!(store.schema_version().await.unwrap(), 21); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -5074,7 +4803,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn memory_authority_records_round_trip_and_close_staging() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 20); + assert_eq!(store.schema_version().await.unwrap(), 21); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -5252,7 +4981,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn account_and_login_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 20); + assert_eq!(store.schema_version().await.unwrap(), 21); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { account_id: "acct-user-alice".to_string(), diff --git a/docs/report/2026-07-30-restored-worker-workspace-credential-stale.md b/docs/report/2026-07-30-restored-worker-workspace-credential-stale.md index 13ef5845..ac77d381 100644 --- a/docs/report/2026-07-30-restored-worker-workspace-credential-stale.md +++ b/docs/report/2026-07-30-restored-worker-workspace-credential-stale.md @@ -16,13 +16,8 @@ The Server control-plane DB contained a current active `worker_workspace_credent - A Worker cannot report or ticket the restore regression through the intended typed authority. - The failure is easy to misattribute to the Browser multiplexer; in this incident Browser authentication/bootstrap was a separate issue. -## Suggested investigation +## Resolution -Trace the credential lifecycle across Backend restart and Runtime Worker restore: +The per-Worker bearer credential was removed rather than adding restore-time secret rotation and reinjection. Worker Workspace requests now carry the Runtime/Worker identity binding established by Runtime, and Server verifies that identity against the current Runtime catalog before applying Ticket role/assignment gates. -1. whether Backend rotates or recreates the credential record; -2. whether restored Worker execution receives the current plaintext credential rather than retaining an old environment/config bundle; -3. whether credential binding should remain stable across Backend restart or be explicitly refreshed before marking the Worker restored; -4. whether restore health should include a bounded Workspace API authentication probe. - -Do not treat a current DB credential row alone as proof that the restored Worker possesses it. +The removal also deletes token mint/rotate/revoke/refresh behavior, live Worker token replacement, and the control-plane credential table. Runtime/Server trust remains the security boundary; Worker role checks remain the accidental-misuse gate.