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

fix: prevent subject startup deadlocks and diagnose auth rejection

Browse Source
This commit is contained in:
Hare
2026-10-07 17:56:48 +09:00
parent 5c95859d39
commit cfacf04260
3 changed files with 496 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/server.rs
+122 -37
View File
@@ -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<ServerConfig>,
@@ -2145,19 +2176,19 @@ pub struct WorkspaceServerApi {
signing_materials: Arc<dyn WorkspaceSigningMaterialStore>,
routers: Arc<AsyncMutex<HashMap<String, Router>>>,
apis: Arc<AsyncMutex<HashMap<String, WorkspaceApi>>>,
mutation_locks: Arc<AsyncMutex<HashMap<String, Arc<AsyncMutex<()>>>>>,
mutation_locks: Arc<AsyncMutex<HashMap<String, Arc<WorkspaceMutationGate>>>>,
running_deletions: Arc<AsyncMutex<HashSet<String>>>,
hook_handles: Arc<AsyncMutex<HashMap<String, tokio::task::AbortHandle>>>,
}
async fn workspace_mutation_lock(
locks: &Arc<AsyncMutex<HashMap<String, Arc<AsyncMutex<()>>>>>,
locks: &Arc<AsyncMutex<HashMap<String, Arc<WorkspaceMutationGate>>>>,
workspace_id: &str,
) -> Arc<AsyncMutex<()>> {
) -> Arc<WorkspaceMutationGate> {
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<AsyncMutex<()>> {
async fn mutation_lock(&self, workspace_id: &str) -> Arc<WorkspaceMutationGate> {
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;
crates/workspace-server/src/server/tests/auth_logging_tests.rs
+173
View File
@@ -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<std::sync::Mutex<Vec<u8>>>);
impl Write for CapturedAuthLog {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
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<Value> = text
.lines()
.map(|line| serde_json::from_str::<Value>(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");
}
}
}
}
crates/workspace-server/src/server/tests/subject_spawn_tests.rs
+201
View File
@@ -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()
);
}
}