From cfacf0426056f91fdbf3fcf0efbe8fa790b8eea1 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 7 Oct 2026 17:56:48 +0900 Subject: [PATCH] fix: prevent subject startup deadlocks and diagnose auth rejection --- crates/workspace-server/src/server.rs | 159 ++++++++++---- .../src/server/tests/auth_logging_tests.rs | 173 +++++++++++++++ .../src/server/tests/subject_spawn_tests.rs | 201 ++++++++++++++++++ 3 files changed, 496 insertions(+), 37 deletions(-) create mode 100644 crates/workspace-server/src/server/tests/auth_logging_tests.rs create mode 100644 crates/workspace-server/src/server/tests/subject_spawn_tests.rs diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 84fb046e1..a610b777c 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2137,6 +2137,37 @@ impl WorkerRemovalService { } } +/// Session attribution is a synchronous callback from Worker create/restore. +/// It must not wait for the request that is waiting for that callback. Deletion +/// acquires both lanes, in request-first order, so it cannot fence out a callback +/// needed by an in-flight create/restore while waiting for that request to finish. +#[derive(Default)] +struct WorkspaceMutationGate { + requests: AsyncMutex<()>, + session_attributions: AsyncMutex<()>, +} + +impl WorkspaceMutationGate { + fn request_lock(&self, method: &Method, path: &str, workspace_id: &str) -> &AsyncMutex<()> { + if *method == Method::POST && path == format!("/api/w/{workspace_id}/subjektiv/sessions") { + &self.session_attributions + } else { + &self.requests + } + } + + async fn lock_deletion( + &self, + ) -> ( + tokio::sync::MutexGuard<'_, ()>, + tokio::sync::MutexGuard<'_, ()>, + ) { + let requests = self.requests.lock().await; + let attributions = self.session_attributions.lock().await; + (requests, attributions) + } +} + #[derive(Clone)] pub struct WorkspaceServerApi { template: Arc, @@ -2145,19 +2176,19 @@ pub struct WorkspaceServerApi { signing_materials: Arc, routers: Arc>>, apis: Arc>>, - mutation_locks: Arc>>>>, + mutation_locks: Arc>>>, running_deletions: Arc>>, hook_handles: Arc>>, } async fn workspace_mutation_lock( - locks: &Arc>>>>, + locks: &Arc>>>, workspace_id: &str, -) -> Arc> { +) -> Arc { let mut locks = locks.lock().await; locks .entry(workspace_id.to_string()) - .or_insert_with(|| Arc::new(AsyncMutex::new(()))) + .or_insert_with(|| Arc::new(WorkspaceMutationGate::default())) .clone() } @@ -2180,7 +2211,7 @@ impl WorkspaceServerApi { } } - async fn mutation_lock(&self, workspace_id: &str) -> Arc> { + async fn mutation_lock(&self, workspace_id: &str) -> Arc { workspace_mutation_lock(&self.mutation_locks, workspace_id).await } @@ -2557,6 +2588,49 @@ fn repository_api_rejection(path: &str, status: StatusCode, message: &str) -> Re status.into_response() } +// Log here, not only in the inner Workspace router: the catalog dispatcher can +// reject authentication before that router (and its API error logger) is entered. +fn runtime_request_proof_rejection( + method: &str, + path: &str, + workspace_id: &str, + error: &crate::worker_source::WorkerMutationSourceProofError, +) -> Response { + use crate::worker_source::WorkerMutationSourceProofError; + + // Do not format the error itself: Authority and MissingPermission can carry + // internal or request-derived strings. Only emit allowlisted reason codes. + let reason = match error { + WorkerMutationSourceProofError::Missing => "missing", + WorkerMutationSourceProofError::Invalid => "invalid", + WorkerMutationSourceProofError::WrongAudience => "wrong_audience", + WorkerMutationSourceProofError::WrongWorkspace => "wrong_workspace", + WorkerMutationSourceProofError::WrongActor => "wrong_actor", + WorkerMutationSourceProofError::MissingPermission(_) => "missing_permission", + WorkerMutationSourceProofError::Expired => "expired", + WorkerMutationSourceProofError::RevokedRuntimeTrust => "runtime_trust_missing_or_revoked", + WorkerMutationSourceProofError::Replay => "replay", + WorkerMutationSourceProofError::WorkerCatalogMembership => "worker_catalog_membership", + WorkerMutationSourceProofError::Authority(_) => "authority_unavailable", + }; + let path = path.split('?').next().unwrap_or(path); + tracing::warn!( + target: "yoi::auth", + event = "runtime_request_proof_rejected", + method, + path, + workspace_id, + status = 401, + reason, + "Runtime request proof rejected", + ); + repository_api_rejection( + path, + StatusCode::UNAUTHORIZED, + "invalid runtime request proof", + ) +} + /// Authored configuration editors are user-facing authority, not a generic /// Runtime workspace.request capability. Worker configuration access is only /// through the grant- and connection-bound WIP adapter. Keep runtime projections @@ -2627,12 +2701,8 @@ async fn authorize_scoped_workspace_request( &digest, ) .await - .map_err(|_| { - repository_api_rejection( - &request_path, - StatusCode::UNAUTHORIZED, - "invalid runtime request proof", - ) + .map_err(|error| { + runtime_request_proof_rejection(&method, &request_path, workspace_id, &error) })?; if is_workspace_config_editor_path(&request_path) { return Err(repository_api_rejection( @@ -2756,7 +2826,7 @@ async fn authorize_workspace_api_request( } else { worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION }; - let Ok(source) = crate::worker_source::verify_runtime_request_source_proof( + let source = match crate::worker_source::verify_runtime_request_source_proof( &api, &proof, &workspace_id, @@ -2766,12 +2836,16 @@ async fn authorize_workspace_api_request( &digest, ) .await - else { - return repository_api_rejection( - &request_path, - StatusCode::UNAUTHORIZED, - "invalid runtime request proof", - ); + { + Ok(source) => source, + Err(error) => { + return runtime_request_proof_rejection( + &method, + &request_path, + &workspace_id, + &error, + ); + } }; if is_workspace_config_editor_path(&request_path) { return repository_api_rejection( @@ -2911,10 +2985,19 @@ async fn dispatch_workspace_request( ) -> Response { let path = request.uri().path().to_owned(); let workspace_id = scoped_workspace_id(&path); - let _mutation_guard = if let Some(workspace_id) = workspace_id + let mutation_gate = if let Some(workspace_id) = workspace_id && workspace_request_requires_mutation_lock(request.method(), &path, workspace_id) { - Some(api.mutation_lock(workspace_id).await.lock_owned().await) + Some((api.mutation_lock(workspace_id).await, workspace_id)) + } else { + None + }; + let _mutation_guard = if let Some((gate, workspace_id)) = mutation_gate.as_ref() { + Some( + gate.request_lock(request.method(), &path, workspace_id) + .lock() + .await, + ) } else { None }; @@ -7145,7 +7228,7 @@ impl server_api::ServerApi for ServerApiContractService { ) })?; let mutation_lock = api.mutation_lock(&workspace_id).await; - let _mutation_guard = mutation_lock.lock().await; + let _mutation_guard = mutation_lock.lock_deletion().await; let existing = api .store .workspace_deletion_operation(&actor.account_id, &request.operation_id) @@ -29570,6 +29653,20 @@ fn compensate_failed_worker_spawn( &worker.worker, context, )); + if let Ok(worker_id) = parse_runtime_worker_id_for_registry(&worker.worker.worker_id) + && let Err(error) = api + .config_store + .release_removed_worker_resource_key(&api.config.workspace_id, worker_id) + { + diagnostics.push(spawn_compensation_diagnostic( + "worker_spawn_compensation_resource_key_release_failed", + format!( + "Failed to release removed Workspace Worker resource key {}: {}", + worker.worker.worker_id, + sanitize_backend_error(&error.to_string()) + ), + )); + } diagnostics } @@ -29637,20 +29734,6 @@ fn finalize_spawn_compensation_after_worker_delete( )), } } - if let Ok(worker_id) = parse_runtime_worker_id_for_registry(&worker_ref.worker_id) - && let Err(error) = api - .config_store - .release_removed_worker_resource_key(&api.config.workspace_id, worker_id) - { - diagnostics.push(spawn_compensation_diagnostic( - "worker_spawn_compensation_resource_key_release_failed", - format!( - "Failed to release removed Workspace Worker resource key {}: {}", - worker_ref.worker_id, - sanitize_backend_error(&error.to_string()) - ), - )); - } diagnostics } @@ -33465,6 +33548,8 @@ impl IntoResponse for ApiError { #[cfg(test)] mod tests { + mod auth_logging_tests; + mod subject_spawn_tests; mod subjektiv_jobs_tests; mod worker_operations_tests; include!("server_workspace_config_tests.rs"); @@ -36226,10 +36311,10 @@ mod tests { let deletion = workspace_mutation_lock(&locks, "workspace-a").await; assert!(Arc::ptr_eq(&active_mutation, &deletion)); - let active_guard = active_mutation.lock_owned().await; + let active_guard = active_mutation.requests.lock().await; let (acquired_tx, mut acquired_rx) = tokio::sync::oneshot::channel(); let waiter = tokio::spawn(async move { - let _deletion_guard = deletion.lock_owned().await; + let _deletion_guard = deletion.lock_deletion().await; let _ = acquired_tx.send(()); }); tokio::task::yield_now().await; diff --git a/crates/workspace-server/src/server/tests/auth_logging_tests.rs b/crates/workspace-server/src/server/tests/auth_logging_tests.rs new file mode 100644 index 000000000..24a61fc03 --- /dev/null +++ b/crates/workspace-server/src/server/tests/auth_logging_tests.rs @@ -0,0 +1,173 @@ +use super::*; +use std::io::Write; +use tracing::instrument::WithSubscriber; +use worker_runtime::auth::{ + RUNTIME_REQUEST_SOURCE_PROOF_HEADER, RuntimeRequestSourceSigner, WORKSPACE_REQUEST_PERMISSION, +}; + +#[derive(Clone, Default)] +struct CapturedAuthLog(Arc>>); + +impl Write for CapturedAuthLog { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[test] +fn runtime_proof_diagnostics_redact_error_payloads() { + use crate::worker_source::WorkerMutationSourceProofError; + + for (error, reason) in [ + ( + WorkerMutationSourceProofError::Authority("private-database-path".to_string()), + "authority_unavailable", + ), + ( + WorkerMutationSourceProofError::MissingPermission("private-permission".to_string()), + "missing_permission", + ), + ] { + let logs = CapturedAuthLog::default(); + let writer = logs.clone(); + let subscriber = tracing_subscriber::fmt() + .without_time() + .with_ansi(false) + .json() + .flatten_event(true) + .with_writer(move || writer.clone()) + .finish(); + tracing::subscriber::with_default(subscriber, || { + let response = runtime_request_proof_rejection( + "POST", + "/api/w/workspace/workers/self/workdir-session/operations?secret=query-secret", + "workspace", + &error, + ); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + }); + let text = String::from_utf8(logs.0.lock().unwrap().clone()).unwrap(); + let event: Value = serde_json::from_str(text.trim()).unwrap(); + assert_eq!(event["reason"], reason); + assert!(!text.contains("private-"), "auth log leaked error details"); + assert!( + !text.contains("query-secret"), + "auth log leaked query values" + ); + } +} + +#[tokio::test] +async fn rejected_runtime_proofs_log_reason_without_request_secrets_on_both_routers() { + let workspace = tempfile::tempdir().unwrap(); + let mut api = test_api(workspace.path()).await; + let identity = RuntimeIdentityMaterial::generate("runtime-test").unwrap(); + configure_runtime_request_auth(&mut api, &identity, "runtime-test"); + let signer = RuntimeRequestSourceSigner::from_identity(&identity); + let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-session/operations"); + let target = format!("{path}?access_token=query-secret"); + let body = "body-secret"; + let apps = [ + build_router(api.clone()), + workspace_server_router(WorkspaceServerApi::new( + api.config.clone(), + api.store.clone(), + )), + ]; + + for app in apps { + for reason in ["expired", "replay", "invalid"] { + let now = i64::try_from(worker_runtime::auth::unix_now_seconds()).unwrap(); + let proof = if reason == "invalid" { + "invalid-proof-secret".to_string() + } else { + signer + .issue( + "server-test", + TEST_WORKSPACE_ID, + None, + WORKSPACE_REQUEST_PERMISSION, + "POST", + &target, + body.as_bytes(), + if reason == "expired" { now - 120 } else { now }, + 30, + ) + .unwrap() + }; + if reason == "replay" { + crate::worker_source::verify_runtime_request_source_proof( + &api, + &proof, + TEST_WORKSPACE_ID, + WORKSPACE_REQUEST_PERMISSION, + "POST", + &target, + &worker_runtime::auth::request_body_digest(body.as_bytes()), + ) + .await + .unwrap(); + } + let logs = CapturedAuthLog::default(); + let writer = logs.clone(); + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::WARN) + .without_time() + .with_ansi(false) + .json() + .flatten_event(true) + .with_writer(move || writer.clone()) + .finish(); + let response = app + .clone() + .oneshot( + Request::builder() + .method(Method::POST) + .uri(&target) + .header(RUNTIME_REQUEST_SOURCE_PROOF_HEADER, &proof) + .header(axum::http::header::AUTHORIZATION, "Bearer bearer-secret") + .header(CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .unwrap(), + ) + .with_subscriber(subscriber) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + let response_body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert!(response_body.is_empty(), "{response_body:?}"); + + let text = String::from_utf8(logs.0.lock().unwrap().clone()).unwrap(); + let events: Vec = text + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .filter(|event| event["event"] == "runtime_request_proof_rejected") + .collect(); + assert_eq!( + events.len(), + 1, + "missing or duplicated auth diagnostic: {text}" + ); + let event = &events[0]; + assert_eq!(event["reason"], reason); + assert_eq!(event["status"], 401); + assert_eq!(event["method"], "POST"); + assert_eq!(event["path"], path); + assert_eq!(event["workspace_id"], TEST_WORKSPACE_ID); + for secret in [ + &proof, + "access_token", + "query-secret", + body, + "bearer-secret", + ] { + assert!(!text.contains(secret), "auth log leaked request data"); + } + } + } +} diff --git a/crates/workspace-server/src/server/tests/subject_spawn_tests.rs b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs new file mode 100644 index 000000000..44d7d7811 --- /dev/null +++ b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs @@ -0,0 +1,201 @@ +use super::*; + +#[tokio::test] +async fn subject_session_callback_completes_while_create_and_deletion_are_waiting() { + let workspace = tempfile::tempdir().unwrap(); + let mut api = test_api(workspace.path()).await; + let identity = RuntimeIdentityMaterial::generate("runtime-test").unwrap(); + configure_runtime_request_auth(&mut api, &identity, "runtime-test"); + let subjektiv = open_subjektiv_store(&api).unwrap(); + let subject = subjektiv + .create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap()) + .unwrap(); + let singleton_key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap(); + let memory = api + .config_store + .get_workspace_memory_settings(TEST_WORKSPACE_ID) + .unwrap(); + let reserved = api + .config_store + .reserve_worker_create( + TEST_WORKSPACE_ID, + "runtime-test", + "subject-start", + &"f".repeat(64), + Some(&singleton_key), + &memory, + ) + .unwrap(); + let worker_id = reserved.worker_id.to_string(); + let server = WorkspaceServerApi::new(api.config.clone(), api.store.clone()); + // Reuse the fixture API, but exercise the real outer dispatcher and both auth layers. + server + .apis + .lock() + .await + .insert(TEST_WORKSPACE_ID.to_string(), api.clone()); + server.routers.lock().await.insert( + TEST_WORKSPACE_ID.to_string(), + build_inner_router(api.clone()), + ); + let gate = server.mutation_lock(TEST_WORKSPACE_ID).await; + let create_path = format!("/api/w/{TEST_WORKSPACE_ID}/workers"); + let create = gate + .request_lock(&Method::POST, &create_path, TEST_WORKSPACE_ID) + .lock() + .await; + let mut deletion = Box::pin(gate.lock_deletion()); + assert!(futures::poll!(&mut deletion).is_pending()); + let app = workspace_server_router(server); + let path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/sessions"); + let session_id = session_store::new_session_id().to_string(); + let body = serde_json::to_vec(&json!({ "session_id": session_id, "create_if_missing": true })) + .unwrap(); + let response = tokio::time::timeout( + std::time::Duration::from_secs(2), + app.clone().oneshot(runtime_source_request( + &identity, + Some(&worker_id), + "POST", + &path, + body.clone(), + )), + ) + .await + .expect("Session attribution must not wait for its own Worker creation") + .unwrap(); + let status = response.status(); + let response_body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!(status, StatusCode::OK, "{response_body:?}"); + let response: Value = serde_json::from_slice(&response_body).unwrap(); + assert_eq!(response["subject_id"], subject.id); + assert_eq!(response["session_id"], session_id); + let stored = subjektiv.session_attribution(&session_id).unwrap().unwrap(); + assert_eq!(stored.worker_id, worker_id); + assert_eq!(stored.subject_id, subject.id); + + // The callback lane does not bypass authentication. + let response = tokio::time::timeout( + std::time::Duration::from_secs(2), + app.clone().oneshot( + Request::builder() + .method("POST") + .uri(&path) + .header(CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .unwrap(), + ), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert!(futures::poll!(&mut deletion).is_pending()); + drop(create); + let _deletion = tokio::time::timeout(std::time::Duration::from_secs(2), deletion) + .await + .unwrap(); + assert!(gate.session_attributions.try_lock().is_err()); +} + +#[tokio::test] +async fn workspace_deletion_waits_for_session_attribution_and_fences_both_lanes() { + let gate = WorkspaceMutationGate::default(); + let attribution = gate.session_attributions.lock().await; + let mut deletion = Box::pin(gate.lock_deletion()); + assert!(futures::poll!(&mut deletion).is_pending()); + assert!(gate.requests.try_lock().is_err()); + drop(attribution); + let deletion = deletion.await; + assert!(gate.requests.try_lock().is_err()); + assert!(gate.session_attributions.try_lock().is_err()); + drop(deletion); + assert!(gate.requests.try_lock().is_ok()); + assert!(gate.session_attributions.try_lock().is_ok()); +} + +#[tokio::test] +async fn only_exact_session_attribution_posts_can_reenter_worker_creation() { + let gate = WorkspaceMutationGate::default(); + let _create = gate.requests.lock().await; + for (method, path) in [ + (Method::POST, "/api/w/workspace-a/subjektiv/sessions"), + (Method::POST, "/api/w/workspace-a/subjektiv/memory"), + (Method::POST, "/api/w/workspace-b/subjektiv/sessions"), + (Method::POST, "/api/w/workspace-a/subjektiv/sessions/extra"), + (Method::DELETE, "/api/w/workspace-a/subjektiv/sessions"), + (Method::POST, "/api/w/workspace-a/workers"), + ] { + let available = gate + .request_lock(&method, path, "workspace-a") + .try_lock() + .is_ok(); + assert_eq!( + available, + method == Method::POST && path == "/api/w/workspace-a/subjektiv/sessions", + "{method} {path}" + ); + } +} + +#[tokio::test] +async fn rejected_create_releases_reservation_and_key_without_cleanup_errors() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + let memory = api + .config_store + .get_workspace_memory_settings(TEST_WORKSPACE_ID) + .unwrap(); + let reserved = api + .config_store + .reserve_worker_create( + TEST_WORKSPACE_ID, + EMBEDDED_RUNTIME_ID, + "rejected-create", + &"f".repeat(64), + Some("subjektiv:rejected-create"), + &memory, + ) + .unwrap(); + let worker = RuntimeWorkerRef::new(EMBEDDED_RUNTIME_ID, reserved.worker_id.to_string()); + assert!( + api.store + .worker_resource_key(TEST_WORKSPACE_ID, &worker) + .unwrap() + .is_some() + ); + let context = WorkerSpawnCompensationContext { + assignment: None, + spawned_workdir_ids: &[], + }; + for _ in 0..2 { + let diagnostics = compensate_failed_workspace_worker_create( + &api, + EMBEDDED_RUNTIME_ID, + reserved.worker_id, + &reserved.create_fingerprint, + "runtime_spawn_rejected", + None, + &context, + &[], + ); + assert!(diagnostics.is_empty(), "{diagnostics:?}"); + assert!( + !api.store + .has_active_worker_create_reservation(TEST_WORKSPACE_ID, &worker) + .unwrap() + ); + assert!( + api.store + .worker_resource_key(TEST_WORKSPACE_ID, &worker) + .unwrap() + .is_none() + ); + assert!( + api.store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, "subjektiv:rejected-create") + .unwrap() + .is_none() + ); + } +}