From 7394ec6784727e44d9f4b10776631627abda3b1c Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 06:13:07 +0900 Subject: [PATCH] fix: retain safe Workdir open authorization and response failure causes Keep bounded reasons and derived failure stages separate from public classification. Reject provider-authored raw error text on transport decoding. Preserve identity mismatch when cleanup fails and retain its handle for explicit close without opening again. Add signed Runtime HTTP, Registry/API logging, remote mismatch, authorization no-dispatch and deferred fault coverage. Focused tests and Workdir suite pass; full-source validation follows T773 audit contract merge. --- crates/workdir/src/http.rs | 21 +- crates/workdir/src/lib.rs | 95 ++++ crates/workdir/tests/failure_diagnostics.rs | 31 + crates/workdir/tests/remote_denial_reasons.rs | 103 ++++ crates/worker-runtime/src/http_server.rs | 26 +- .../tests/runtime_workdir_failure_tests.rs | 528 ++++++++++++++++++ .../src/worker_backend/deferred_workdir.rs | 163 +++++- crates/workspace-server/src/hosts.rs | 17 +- .../src/hosts/tests/workspace_auth_tests.rs | 68 +++ crates/workspace-server/src/server.rs | 6 +- .../src/server/tests/api_logging_tests.rs | 56 ++ 11 files changed, 1093 insertions(+), 21 deletions(-) create mode 100644 crates/worker-runtime/src/http_server/tests/runtime_workdir_failure_tests.rs diff --git a/crates/workdir/src/http.rs b/crates/workdir/src/http.rs index f99f804c5..889a7eb93 100644 --- a/crates/workdir/src/http.rs +++ b/crates/workdir/src/http.rs @@ -375,12 +375,14 @@ impl WorkdirTransportError { use WorkdirTransportErrorCode as Code; let error = match self.code { Code::NotFound => WorkdirError::NotFound("".into()), - Code::Conflict => WorkdirError::Conflict(self.message), + Code::Conflict => WorkdirError::Conflict( + "The target file's content or existence changed since it was last observed; read the file again before retrying".into(), + ), Code::OutcomeUnknown => WorkdirError::OutcomeUnknown( "Workdir operation may have effects; inspect before retrying".into(), ), - Code::Unsupported => WorkdirError::UnsupportedOperation(self.message), - Code::InvalidRequest => WorkdirError::InvalidArgument(self.message), + Code::Unsupported => WorkdirError::UnsupportedOperation("Workdir operation is not supported".into()), + Code::InvalidRequest => WorkdirError::InvalidArgument("Workdir operation request is invalid".into()), Code::Denied => WorkdirError::Denied(crate::WorkdirDenial { reason: self.denial_reason, // Provider-authored text is not trusted diagnostic data. @@ -409,12 +411,12 @@ impl WorkdirTransportError { target: "".into(), }, Code::UnknownCommand => WorkdirError::UnknownCommand("".to_string()), - Code::Unavailable => WorkdirError::Unavailable(self.message), + Code::Unavailable => WorkdirError::Unavailable("Workdir session is unavailable".into()), Code::Io => WorkdirError::Io { path: "".into(), - source: std::io::Error::other(self.message), + source: std::io::Error::other("Workdir I/O operation failed"), }, - Code::Transport => WorkdirError::Transport(self.message), + Code::Transport => WorkdirError::Transport("Workdir transport failed".into()), Code::Internal => WorkdirError::OperationFailed, }; let error = if let Some(reason) = self.failure_reason { @@ -549,7 +551,8 @@ mod client { if opened.workdir_id.as_str() != workdir_id.as_str() { return Err(WorkdirError::Unavailable( "Runtime opened a session for a different Workdir".to_string(), - )); + ) + .with_failure_reason(crate::WorkdirFailureReason::SessionIdentityMismatch)); } Ok(Self { client, @@ -608,6 +611,7 @@ mod client { WorkdirError::Unavailable(format!( "Runtime returned a mismatched Workdir operation result; expected {expected}" )) + .with_failure_reason(crate::WorkdirFailureReason::OperationResultMismatch) } } @@ -770,7 +774,8 @@ mod client { } Ok(_) if mutation => Err(WorkdirError::OutcomeUnknown( "mismatched response after checkout mutation".into(), - )), + ) + .with_failure_reason(crate::WorkdirFailureReason::OperationResultMismatch)), Ok(_) => Err(Self::mismatch("checkout_execute")), Err(error) if mutation diff --git a/crates/workdir/src/lib.rs b/crates/workdir/src/lib.rs index 1a515ac16..10ba3bc6c 100644 --- a/crates/workdir/src/lib.rs +++ b/crates/workdir/src/lib.rs @@ -414,6 +414,24 @@ pub enum WorkdirFailureReason { HttpTransfer, HttpResponseInvalid, HttpResponseTooLarge, + OperationResultMismatch, + SessionIdentityMismatch, + SessionOpenTaskFailed, + SessionOpenFailed, + AuthorizationMissing, + AuthorizationStale, + AuthorizationUnavailable, + SigningIdentityUnavailable, + AuthorizationSignFailed, + RuntimeStopped, + RuntimeStateUnavailable, + RuntimeBackendUnavailable, + RuntimeOwnerMismatch, + RuntimeInvalidRequest, + WorkingDirectoryNotFound, + WorkingDirectoryNotActive, + WorkingDirectoryRecordInvalid, + WorkingDirectoryAccessUnavailable, } impl WorkdirFailureReason { @@ -427,6 +445,83 @@ impl WorkdirFailureReason { Self::HttpTransfer => "http_transfer", Self::HttpResponseInvalid => "http_response_invalid", Self::HttpResponseTooLarge => "http_response_too_large", + Self::OperationResultMismatch => "operation_result_mismatch", + Self::SessionIdentityMismatch => "session_identity_mismatch", + Self::SessionOpenTaskFailed => "session_open_task_failed", + Self::SessionOpenFailed => "session_open_failed", + Self::AuthorizationMissing => "authorization_missing", + Self::AuthorizationStale => "authorization_stale", + Self::AuthorizationUnavailable => "authorization_unavailable", + Self::SigningIdentityUnavailable => "signing_identity_unavailable", + Self::AuthorizationSignFailed => "authorization_sign_failed", + Self::RuntimeStopped => "runtime_stopped", + Self::RuntimeStateUnavailable => "runtime_state_unavailable", + Self::RuntimeBackendUnavailable => "runtime_backend_unavailable", + Self::RuntimeOwnerMismatch => "runtime_owner_mismatch", + Self::RuntimeInvalidRequest => "runtime_invalid_request", + Self::WorkingDirectoryNotFound => "working_directory_not_found", + Self::WorkingDirectoryNotActive => "working_directory_not_active", + Self::WorkingDirectoryRecordInvalid => "working_directory_record_invalid", + Self::WorkingDirectoryAccessUnavailable => "working_directory_access_unavailable", + } + } + + /// The observed failing boundary, not a dispatch or effects assertion. + pub const fn stage(self) -> WorkdirFailureStage { + use WorkdirFailureStage as Stage; + match self { + Self::SessionClosed => Stage::Session, + Self::ProviderUnavailable => Stage::Provider, + Self::CommandTerminalMissing => Stage::Command, + Self::HttpTimeout | Self::HttpConnect | Self::HttpTransfer => Stage::Http, + Self::HttpResponseInvalid + | Self::HttpResponseTooLarge + | Self::OperationResultMismatch => Stage::ResponseValidation, + Self::SessionIdentityMismatch + | Self::SessionOpenTaskFailed + | Self::SessionOpenFailed + | Self::WorkingDirectoryNotFound + | Self::WorkingDirectoryNotActive + | Self::WorkingDirectoryRecordInvalid + | Self::WorkingDirectoryAccessUnavailable => Stage::SessionOpen, + Self::AuthorizationMissing + | Self::AuthorizationStale + | Self::AuthorizationUnavailable + | Self::SigningIdentityUnavailable + | Self::AuthorizationSignFailed => Stage::AuthorizationPreparation, + Self::RuntimeStopped + | Self::RuntimeStateUnavailable + | Self::RuntimeBackendUnavailable + | Self::RuntimeOwnerMismatch + | Self::RuntimeInvalidRequest => Stage::Runtime, + } + } +} + +/// Internal operational boundary derived only from a bounded reason. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum WorkdirFailureStage { + Session, + Provider, + Command, + Http, + ResponseValidation, + SessionOpen, + AuthorizationPreparation, + Runtime, +} + +impl WorkdirFailureStage { + pub const fn as_str(self) -> &'static str { + match self { + Self::Session => "session", + Self::Provider => "provider", + Self::Command => "command", + Self::Http => "http", + Self::ResponseValidation => "response_validation", + Self::SessionOpen => "session_open", + Self::AuthorizationPreparation => "authorization_preparation", + Self::Runtime => "runtime", } } } diff --git a/crates/workdir/tests/failure_diagnostics.rs b/crates/workdir/tests/failure_diagnostics.rs index a94f4fe9f..b39336f04 100644 --- a/crates/workdir/tests/failure_diagnostics.rs +++ b/crates/workdir/tests/failure_diagnostics.rs @@ -83,3 +83,34 @@ fn provider_cannot_inject_free_text_failure_reason() { .is_err() ); } + +#[test] +fn provider_authored_operational_messages_never_enter_error_display_or_debug() { + for code in [ + Code::Conflict, + Code::OutcomeUnknown, + Code::Unsupported, + Code::InvalidRequest, + Code::Unavailable, + Code::Io, + Code::Transport, + Code::Internal, + ] { + let wire = serde_json::json!({ + "code": code, "message":"/host/private command-secret body-secret proof-secret key-secret Authorization: Bearer bearer-secret", + "failure_reason": "session_open_task_failed" + }); + let decoded: WorkdirTransportError = serde_json::from_value(wire).unwrap(); + let error = decoded.into_workdir_error(); + for text in [ + format!("{error}"), + format!("{error:?}"), + serde_json::to_string(&WorkdirTransportError::from_workdir_error(&error)).unwrap(), + ] { + assert!(!text.contains("secret"), "{text}"); + assert!(!text.contains("/host/private"), "{text}"); + } + assert_eq!(error.failure_reason(), Some(Reason::SessionOpenTaskFailed)); + assert_eq!(WorkdirTransportError::from_workdir_error(&error).code, code); + } +} diff --git a/crates/workdir/tests/remote_denial_reasons.rs b/crates/workdir/tests/remote_denial_reasons.rs index f14acc816..6769fb136 100644 --- a/crates/workdir/tests/remote_denial_reasons.rs +++ b/crates/workdir/tests/remote_denial_reasons.rs @@ -216,3 +216,106 @@ async fn remote_checkout_diagnostics_preserve_mutation_outcome_and_read_failures } } } + +#[tokio::test] +async fn mismatched_http_results_keep_cause_and_mutation_uncertainty() { + use workdir::{ + CheckoutOperation, CheckoutRequest, CommandHandle, CommandOutputRequest, + WorkdirFailureReason as Failure, + }; + for case in ["variant", "bytes", "mutation", "identity"] { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = thread::spawn(move || { + reply( + &listener, + "200 OK", + if case == "identity" { + r#"{"session_id":"s","workdir_id":"wrong-secret","capabilities":{"bits":63}}"# + } else { + r#"{"session_id":"s","workdir_id":"w","capabilities":{"bits":63}}"# + }, + ); + if case != "identity" { + let result = if case == "bytes" { + serde_json::json!({"operation":"command_output","result":{ + "chunks": [], "next_cursor": 123, "terminal": null + }}) + } else { + serde_json::to_value( + workdir::http::WorkdirSessionOperationResult::CommandCancel, + ) + .unwrap() + }; + reply(&listener, "200 OK", &result.to_string()); + } + }); + let opened = RemoteWorkdirSession::open( + reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(5)) + .build() + .unwrap(), + format!("http://{address}").parse().unwrap(), + "request-secret", + Workdir::new("w").id().clone(), + OpenWorkdirSessionRequest::default(), + ) + .await; + let error = if case == "identity" { + opened.unwrap_err() + } else { + let remote = opened.unwrap(); + match case { + "variant" => remote + .stat(StatRequest { + path: WorkdirPath::new("file-secret").unwrap(), + }) + .await + .unwrap_err(), + "bytes" => remote + .command_output(CommandOutputRequest { + handle: CommandHandle("command-secret".into()), + cursor: 0, + limit: 16, + wait: false, + }) + .await + .unwrap_err(), + "mutation" => remote + .checkout_execute(CheckoutRequest { + target: WorkdirPath::new("file-secret").unwrap(), + validator: vec![1], + operation: CheckoutOperation::Write { + content: b"body-secret".to_vec(), + expected_hash: [0; 32], + }, + }) + .await + .unwrap_err(), + _ => unreachable!(), + } + }; + server.join().unwrap(); + let expected = if case == "identity" { + Failure::SessionIdentityMismatch + } else { + Failure::OperationResultMismatch + }; + assert_eq!(error.failure_reason(), Some(expected), "{case}: {error:?}"); + let transport = WorkdirTransportError::from_workdir_error(&error); + assert_eq!( + transport.code, + if case == "mutation" { + Code::OutcomeUnknown + } else { + Code::Unavailable + } + ); + let wire = serde_json::to_string(&transport).unwrap(); + assert!(!wire.contains("secret"), "{wire}"); + assert!(!format!("{error:?}").contains("secret")); + let forwarded = transport.into_workdir_error(); + assert_eq!(forwarded.failure_reason(), Some(expected)); + } +} diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index aefc66973..1a6aeefe9 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2187,7 +2187,25 @@ impl RuntimeHttpWorkdirError { } fn runtime(error: RuntimeError) -> Self { - match error { + use workdir::WorkdirFailureReason as Reason; + let reason = match &error { + RuntimeError::WorkingDirectory(diagnostic) => match diagnostic.code.as_str() { + "working_directory_not_found" => Reason::WorkingDirectoryNotFound, + "working_directory_not_active" => Reason::WorkingDirectoryNotActive, + "working_directory_record_invalid" => Reason::WorkingDirectoryRecordInvalid, + "working_directory_repository_access_unavailable" => { + Reason::WorkingDirectoryAccessUnavailable + } + _ => Reason::SessionOpenFailed, + }, + RuntimeError::RuntimeStopped => Reason::RuntimeStopped, + RuntimeError::StatePoisoned => Reason::RuntimeStateUnavailable, + RuntimeError::ExecutionBackendUnavailable { .. } => Reason::RuntimeBackendUnavailable, + RuntimeError::WorkspaceOwnerMismatch { .. } => Reason::RuntimeOwnerMismatch, + RuntimeError::InvalidRequest(_) => Reason::RuntimeInvalidRequest, + _ => Reason::SessionOpenFailed, + }; + let mut response = match error { RuntimeError::WorkingDirectory(diagnostic) if diagnostic.code == "working_directory_not_found" => { @@ -2200,7 +2218,9 @@ impl RuntimeHttpWorkdirError { WorkdirTransportErrorCode::Unavailable, "Workdir session is unavailable", ), - } + }; + response.payload.failure_reason = Some(reason); + response } } @@ -2222,6 +2242,7 @@ impl IntoResponse for RuntimeHttpWorkdirError { status = self.status.as_u16(), code = self.payload.code.as_str(), failure_reason = reason.as_str(), + failure_stage = reason.stage().as_str(), "Workdir provider operation failed" ); } @@ -2434,6 +2455,7 @@ pub enum RuntimeHttpServerError { #[cfg(test)] mod tests { mod command_transport_tests; + mod runtime_workdir_failure_tests; #[cfg(feature = "fs-store")] mod session_material_tests; diff --git a/crates/worker-runtime/src/http_server/tests/runtime_workdir_failure_tests.rs b/crates/worker-runtime/src/http_server/tests/runtime_workdir_failure_tests.rs new file mode 100644 index 000000000..55873420b --- /dev/null +++ b/crates/worker-runtime/src/http_server/tests/runtime_workdir_failure_tests.rs @@ -0,0 +1,528 @@ +//! Session-open integration through the production signed router, Runtime owner/binding +//! checks, HTTP DTO and tracing boundary. Only the execution backend is replaced +//! for deterministic lookup/open faults and inert owner-binding Worker fixtures; +//! no handler, middleware or grant is replaced. Runs in the normal http-server lib +//! suite with isolated loopback sockets. + +use super::*; +use crate::working_directory::WorkingDirectoryDiagnostic; +use std::io::{self, Write}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +const OPEN_PATH: &str = "/v1/working-directories/private-workdir-marker/sessions"; +const PRIVATE_DIAGNOSTIC: &str = concat!( + "raw-runtime-error-marker /private/host-path-marker ", + "command=private-command-marker contents=private-file-marker ", + "proof=private-proof-marker Authorization: Bearer private-header-marker ", + "working_directory_not_found authorization_stale runtime_stopped", +); +const PRIVATE_MARKERS: &[&str] = &[ + "raw-runtime-error-marker", + "/private/host-path-marker", + "private-command-marker", + "private-file-marker", + "private-proof-marker", + "private-header-marker", + "private-workdir-marker", + "private-owner-marker", +]; + +#[derive(Clone, Default)] +struct LogCapture(Arc>>); + +impl Write for LogCapture { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl LogCapture { + fn subscriber(&self) -> impl tracing::Subscriber + Send + Sync + 'static { + let writer = self.clone(); + tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_max_level(tracing::Level::WARN) + .with_writer(move || writer.clone()) + .finish() + } + + fn take(&self) -> String { + String::from_utf8(std::mem::take(&mut *self.0.lock().unwrap())).unwrap() + } +} + +fn assert_private_absent(text: &str) { + for marker in PRIVATE_MARKERS { + assert!( + !text.contains(marker), + "private input escaped error boundary" + ); + } +} + +fn assert_failure_log(log: &str, status: StatusCode, code: &str, reason: &str, stage: &str) { + let events: Vec<_> = log + .lines() + .filter(|line| line.contains("workdir_provider_failure")) + .collect(); + assert_eq!(events.len(), 1, "one diagnostic per failed request: {log}"); + let event = events[0]; + for field in [ + "WARN".to_string(), + "worker_runtime::workdir".to_string(), + format!("status={}", status.as_u16()), + format!("code=\"{code}\""), + format!("failure_reason=\"{reason}\""), + format!("failure_stage=\"{stage}\""), + ] { + assert!(event.contains(&field), "missing {field}: {event}"); + } + assert_private_absent(log); +} + +fn assert_failure_body(body: &str, code: &str, message: &str, reason: &str) { + assert_private_absent(body); + // Exact safe transport vocabulary; raw diagnostics and stage are not wire fields. + assert_eq!( + serde_json::from_str::(body).unwrap(), + serde_json::json!({"code": code, "message": message, "failure_reason": reason}), + ); +} + +struct Harness { + server: tokio::task::JoinHandle<()>, + client: reqwest::Client, + base: String, + identity: RuntimeIdentityMaterial, + trust: WorkspaceIssuerTrustRecord, +} + +impl Drop for Harness { + fn drop(&mut self) { + self.server.abort(); + } +} + +impl Harness { + async fn new(runtime: Runtime) -> Self { + let identity = RuntimeIdentityMaterial::generate("runtime-failure-test-key").unwrap(); + let trust = workspace_trust(&identity); + let app = runtime_http_router_with_workspace_auth( + runtime, + None, + workspace_test_auth(vec![trust.clone()]), + ); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + Self { + server, + client: reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(5)) + .build() + .unwrap(), + base, + identity, + trust, + } + } + + async fn send( + &self, + claims: &WorkspaceCapabilityClaims, + body: Vec, + workspace: &str, + ) -> (StatusCode, String, String) { + let token = issue_workspace_capability_token(&self.identity.signing_key().unwrap(), claims) + .unwrap(); + let response = self + .client + .post(format!("{}{OPEN_PATH}", self.base)) + .bearer_auth(&token) + .header(header::CONTENT_TYPE, "application/json") + .header(RUNTIME_WORKSPACE_SCOPE_HEADER, workspace) + .header("x-private-test-header", "private-header-marker") + .body(body) + .send() + .await + .unwrap(); + let status = StatusCode::from_u16(response.status().as_u16()).unwrap(); + let headers = format!("{:?}", response.headers()); + let body = response.text().await.unwrap(); + assert_private_absent(&headers); + for output in [&body, &headers] { + assert!(!output.contains(&token), "signed bearer token leaked"); + assert!( + !output.contains(&self.identity.private_key), + "signing key leaked" + ); + } + (status, body, token) + } + + async fn open(&self) -> (StatusCode, String, String) { + let body = serde_json::to_vec(&OpenWorkdirSessionRequest::default()).unwrap(); + let claims = signed_claims(&self.trust, "POST", OPEN_PATH, &body); + self.send(&claims, body, "workspace-a").await + } +} + +#[derive(Clone, Copy)] +enum FailureAt { + Lookup, + Open, +} + +struct FailingBackend { + at: FailureAt, + code: &'static str, + lookups: AtomicUsize, + opens: AtomicUsize, +} + +impl FailingBackend { + fn runtime(at: FailureAt, code: &'static str) -> (Runtime, Arc) { + let backend = Arc::new(Self { + at, + code, + lookups: AtomicUsize::new(0), + opens: AtomicUsize::new(0), + }); + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), backend.clone()).unwrap(); + (runtime, backend) + } + + fn failure(&self) -> WorkingDirectoryDiagnostic { + WorkingDirectoryDiagnostic::rejected(self.code, PRIVATE_DIAGNOSTIC) + } +} + +impl WorkerExecutionBackend for FailingBackend { + fn backend_id(&self) -> &str { + "runtime-failure-test" + } + + fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { + AcceptingBackend.spawn_worker(request) + } + + fn dispatch_input(&self, worker: &WorkerRef, input: WorkerInput) -> WorkerExecutionResult { + AcceptingBackend.dispatch_input(worker, input) + } + + fn stop_worker(&self, worker: &WorkerRef) -> WorkerExecutionResult { + AcceptingBackend.stop_worker(worker) + } + + fn working_directory( + &self, + id: &str, + ) -> Result { + self.lookups.fetch_add(1, Ordering::SeqCst); + match self.at { + FailureAt::Lookup => Err(self.failure()), + FailureAt::Open => Ok(WorkingDirectoryStatus { + summary: crate::catalog::WorkingDirectorySummary { + working_directory_id: id.into(), + display_name: None, + repository_id: "repository-test".into(), + creation_selector: None, + creation_ref: None, + creation_tree: None, + current_selector: None, + current_ref: None, + current_tree: None, + observed_at_epoch_seconds: None, + materializer_kind: crate::catalog::MaterializerKind::RuntimeGitClone, + cleanup_target: None, + status: crate::catalog::WorkingDirectoryStatusKind::Active, + cleanliness: None, + occupied_by: None, + }, + }), + } + } + + fn open_workdir_session( + &self, + _: &str, + ) -> Result { + self.opens.fetch_add(1, Ordering::SeqCst); + Err(self.failure()) + } +} + +// Current-thread execution keeps the server tasks inside the thread-local tracing +// capture across awaits, without replacing the process-global subscriber. +#[tokio::test(flavor = "current_thread")] +async fn signed_http_session_open_preserves_runtime_availability_and_owner_failures() { + let capture = LogCapture::default(); + let _guard = tracing::subscriber::set_default(capture.subscriber()); + tracing::callsite::rebuild_interest_cache(); + tokio::time::timeout(Duration::from_secs(15), async { + let stopped = Runtime::new_memory(); + stopped.stop_runtime().unwrap(); + let (owned, backend) = FailingBackend::runtime(FailureAt::Lookup, "must_not_reach_backend"); + let mut request = task_request("bind owner"); + request.config_bundle = None; + request.memory_settings.as_mut().unwrap().workspace_id = "workspace-a".into(); + owned + .create_worker_scoped( + &RuntimeWorkspaceScope::new("workspace-a", "private-owner-marker"), + request, + ) + .unwrap(); + for (runtime, status, code, message, reason) in [ + ( + stopped, + StatusCode::SERVICE_UNAVAILABLE, + "unavailable", + "Workdir session is unavailable", + "runtime_stopped", + ), + ( + Runtime::new_memory(), + StatusCode::SERVICE_UNAVAILABLE, + "unavailable", + "Workdir session is unavailable", + "runtime_backend_unavailable", + ), + ( + owned, + StatusCode::NOT_FOUND, + "not_found", + "Workdir session was not found", + "runtime_owner_mismatch", + ), + ] { + let harness = Harness::new(runtime).await; + capture.take(); + let (actual, body, token) = harness.open().await; + assert_eq!(actual, status); + assert_failure_body(&body, code, message, reason); + let log = capture.take(); + assert_failure_log(&log, status, code, reason, "runtime"); + assert!(!log.contains(&token)); + assert!(!log.contains(&harness.identity.private_key)); + } + assert_eq!(backend.lookups.load(Ordering::SeqCst), 0); + assert_eq!(backend.opens.load(Ordering::SeqCst), 0); + }) + .await + .expect("bounded Runtime failure HTTP requests"); +} + +#[tokio::test(flavor = "current_thread")] +async fn signed_http_session_open_allowlists_diagnostic_codes_without_guessing_unknown_causes() { + let capture = LogCapture::default(); + let _guard = tracing::subscriber::set_default(capture.subscriber()); + tracing::callsite::rebuild_interest_cache(); + tokio::time::timeout(Duration::from_secs(20), async { + for (diagnostic, reason) in [ + ("working_directory_not_found", "working_directory_not_found"), + ( + "working_directory_not_active", + "working_directory_not_active", + ), + ( + "working_directory_record_invalid", + "working_directory_record_invalid", + ), + ( + "working_directory_repository_access_unavailable", + "working_directory_access_unavailable", + ), + // Plausible provider text/code is not evidence of a typed cause. + ("authorization_stale", "session_open_failed"), + ( + "working_directory_not_found_private-proof-marker", + "session_open_failed", + ), + ] { + for at in [FailureAt::Lookup, FailureAt::Open] { + let (runtime, backend) = FailingBackend::runtime(at, diagnostic); + let harness = Harness::new(runtime).await; + capture.take(); + let (status, body, token) = harness.open().await; + let (expected, code, message) = if diagnostic == "working_directory_not_found" { + ( + StatusCode::NOT_FOUND, + "not_found", + "Workdir session was not found", + ) + } else { + ( + StatusCode::SERVICE_UNAVAILABLE, + "unavailable", + "Workdir session is unavailable", + ) + }; + assert_eq!(status, expected, "{diagnostic}"); + assert_failure_body(&body, code, message, reason); + let log = capture.take(); + assert_failure_log(&log, status, code, reason, "session_open"); + assert!(!log.contains(&token)); + assert!(!log.contains(&harness.identity.private_key)); + assert_eq!(backend.lookups.load(Ordering::SeqCst), 1, "no hidden retry"); + assert_eq!( + backend.opens.load(Ordering::SeqCst), + usize::from(matches!(at, FailureAt::Open)), + ); + } + } + }) + .await + .expect("bounded diagnostic fault matrix"); +} + +#[tokio::test(flavor = "current_thread")] +async fn signed_http_session_open_rejects_bad_capability_scope_and_owner_binding_before_backend() { + let capture = LogCapture::default(); + let _guard = tracing::subscriber::set_default(capture.subscriber()); + tracing::callsite::rebuild_interest_cache(); + tokio::time::timeout(Duration::from_secs(15), async { + let (runtime, backend) = + FailingBackend::runtime(FailureAt::Open, "working_directory_record_invalid"); + let mut request = task_request("foreign workspace binding"); + request.config_bundle = None; + request.memory_settings.as_mut().unwrap().workspace_id = "other-workspace".into(); + let foreign_worker = runtime + .create_worker_scoped( + &RuntimeWorkspaceScope::new("other-workspace", "https://backend.test"), + request, + ) + .unwrap() + .worker_ref; + let harness = Harness::new(runtime).await; + for case in ["permission", "body", "workspace", "owner_binding"] { + let body = if case == "owner_binding" { + serde_json::to_vec(&OpenWorkdirSessionRequest { + owner_worker_id: Some(foreign_worker.worker_id.to_string()), + }) + .unwrap() + } else { + b"{}".to_vec() + }; + let mut claims = signed_claims(&harness.trust, "POST", OPEN_PATH, &body); + if case == "permission" { + claims.operation = "workers:read".into(); + } + if case == "workspace" { + claims.issuer_workspace_id = "other-workspace".into(); + } + let sent = if case == "body" { + b"{ }".to_vec() + } else { + body + }; + capture.take(); + let (status, body, token) = harness.send(&claims, sent, "workspace-a").await; + assert_eq!( + status, + if case == "owner_binding" { + StatusCode::NOT_FOUND + } else { + StatusCode::UNAUTHORIZED + }, + "{case}", + ); + assert_private_absent(&body); + let log = capture.take(); + assert_private_absent(&log); + assert!(!log.contains(&token)); + assert!(!log.contains(&harness.identity.private_key)); + if case == "owner_binding" { + assert_failure_body( + &body, + "not_found", + "Workdir session was not found", + "working_directory_not_found", + ); + assert_failure_log( + &log, + status, + "not_found", + "working_directory_not_found", + "session_open", + ); + } else { + let wire: serde_json::Value = serde_json::from_str(&body).unwrap(); + assert_eq!(wire["error"]["code"], "unauthorized"); + assert!( + !log.contains("workdir_provider_failure"), + "auth rejection is not a provider fault", + ); + } + } + assert_eq!(backend.lookups.load(Ordering::SeqCst), 0); + assert_eq!(backend.opens.load(Ordering::SeqCst), 0); + }) + .await + .expect("bounded authorization rejection requests"); +} + +#[tokio::test(flavor = "current_thread")] +async fn runtime_error_response_keeps_raw_details_private_and_unclassified_causes_generic() { + let capture = LogCapture::default(); + let _guard = tracing::subscriber::set_default(capture.subscriber()); + tracing::callsite::rebuild_interest_cache(); + for (error, status, code, message, reason, stage) in [ + ( + RuntimeError::ExecutionBackendUnavailable { + message: PRIVATE_DIAGNOSTIC.into(), + }, + StatusCode::SERVICE_UNAVAILABLE, + "unavailable", + "Workdir session is unavailable", + "runtime_backend_unavailable", + "runtime", + ), + ( + RuntimeError::StatePoisoned, + StatusCode::SERVICE_UNAVAILABLE, + "unavailable", + "Workdir session is unavailable", + "runtime_state_unavailable", + "runtime", + ), + ( + RuntimeError::InvalidRequest(PRIVATE_DIAGNOSTIC.into()), + StatusCode::BAD_REQUEST, + "invalid_request", + "Workdir operation request is invalid", + "runtime_invalid_request", + "runtime", + ), + ( + RuntimeError::StoreCorrupt { + operation: "session_open", + path: PathBuf::from("/private/host-path-marker"), + message: PRIVATE_DIAGNOSTIC.into(), + }, + StatusCode::SERVICE_UNAVAILABLE, + "unavailable", + "Workdir session is unavailable", + "session_open_failed", + "session_open", + ), + ] { + capture.take(); + let response = RuntimeHttpWorkdirError::runtime(error).into_response(); + assert_eq!(response.status(), status); + let bytes = to_bytes(response.into_body(), 16384).await.unwrap(); + let body = std::str::from_utf8(&bytes).unwrap(); + assert_failure_body(body, code, message, reason); + assert_failure_log(&capture.take(), status, code, reason, stage); + } +} diff --git a/crates/worker-runtime/src/worker_backend/deferred_workdir.rs b/crates/worker-runtime/src/worker_backend/deferred_workdir.rs index eebedde57..9bcb3962d 100644 --- a/crates/worker-runtime/src/worker_backend/deferred_workdir.rs +++ b/crates/worker-runtime/src/worker_backend/deferred_workdir.rs @@ -66,13 +66,32 @@ impl DeferredWorkdirSession { let open = self.open.clone(); let session = tokio::task::spawn_blocking(move || open()) .await - .map_err(|_| WorkdirError::Unavailable("Workdir session opener failed".into()))? - .map_err(WorkdirError::Unavailable)?; + .map_err(|_| { + WorkdirError::Unavailable("Workdir session opener failed".into()) + .with_failure_reason(WorkdirFailureReason::SessionOpenTaskFailed) + })? + .map_err(|_| { + // The opener's string contract cannot prove a more specific cause. + WorkdirError::Unavailable("Workdir session could not be opened".into()) + .with_failure_reason(WorkdirFailureReason::SessionOpenFailed) + })?; if session.workdir().id() != self.workdir.id() { - session.close().await?; - return Err(WorkdirError::Unavailable( - "Workdir session identity mismatch".into(), - )); + if let Err(error) = session.close().await { + // Retain a failed-close owner for explicit close; never use it. + state.closed = true; + state.session = Some(session); + tracing::warn!( + target: "worker_runtime::workdir", + event = "workdir_session_identity_cleanup_failed", + failure_reason = error.failure_reason().map(WorkdirFailureReason::as_str), + denial_reason = error.denial_reason().map(WorkdirDenialReason::as_str), + "Mismatched Workdir session cleanup failed" + ); + } + return Err( + WorkdirError::Unavailable("Workdir session identity mismatch".into()) + .with_failure_reason(WorkdirFailureReason::SessionIdentityMismatch), + ); } if let Some(mut source) = session.subscribe_command_events() { let events = self.events.clone(); @@ -224,10 +243,16 @@ mod tests { }, ); assert_eq!(attempts.load(Ordering::SeqCst), 0); + let failed = session.read(read_request()).await.unwrap_err(); assert!(matches!( - session.read(read_request()).await, - Err(WorkdirError::Unavailable(_)) + failed.classification_source(), + WorkdirError::Unavailable(_) )); + assert_eq!( + failed.failure_reason(), + Some(WorkdirFailureReason::SessionOpenFailed) + ); + assert_eq!(attempts.load(Ordering::SeqCst), 1); assert_eq!( session.read(read_request()).await.unwrap().bytes, b"restored" @@ -253,4 +278,126 @@ mod tests { ); session.close().await.unwrap(); } + + #[tokio::test] + async fn opener_task_and_open_errors_keep_distinct_safe_causes() { + for panic in [false, true] { + let attempts = Arc::new(AtomicUsize::new(0)); + let observed = attempts.clone(); + let session = DeferredWorkdirSession::handle( + "saved", + WorkdirSessionCapabilities::READ_ONLY, + move || { + observed.fetch_add(1, Ordering::SeqCst); + if panic { + panic!( + "/host/private bearer-secret command-secret proof-secret key-secret" + ); + } + Err("/host/private bearer-secret command-secret proof-secret key-secret".into()) + }, + ); + let error = session.read(read_request()).await.unwrap_err(); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + let reason = if panic { + WorkdirFailureReason::SessionOpenTaskFailed + } else { + WorkdirFailureReason::SessionOpenFailed + }; + assert_eq!(error.failure_reason(), Some(reason)); + assert_eq!(reason.stage().as_str(), "session_open"); + let transport = workdir::http::WorkdirTransportError::from_workdir_error(&error); + assert_eq!(transport.code.http_status(), 503); + assert!( + !serde_json::to_string(&transport) + .unwrap() + .contains("secret") + ); + assert!(!format!("{error:?}").contains("secret")); + session.close().await.unwrap(); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + } + } + + #[derive(Debug)] + struct MismatchedSession { + workdir: Workdir, + closes: Arc, + fail_close: bool, + } + + macro_rules! mismatched_session { + ($( $method:ident($request:ident: $request_type:ty) -> $result:ty; )*) => { + #[async_trait] + impl WorkdirSession for MismatchedSession { + fn workdir(&self) -> &Workdir { &self.workdir } + fn capabilities(&self) -> WorkdirSessionCapabilities { WorkdirSessionCapabilities::READ_ONLY } + $(async fn $method(&self, _: $request_type) -> Result<$result, WorkdirError> { + panic!("mismatched session must never be used"); + })* + async fn close(&self) -> Result<(), WorkdirError> { + let attempt = self.closes.fetch_add(1, Ordering::SeqCst); + if self.fail_close && attempt == 0 { + Err(WorkdirError::Transport("/host/private key-secret".into()) + .with_failure_reason(WorkdirFailureReason::HttpTransfer)) + } else { Ok(()) } + } + } + }; + } + mismatched_session! { + stat(request: StatRequest) -> StatResult; + read(request: ReadRequest) -> ReadResult; + write(request: WriteRequest) -> WriteResult; + edit(request: EditRequest) -> EditResult; + list(request: ListRequest) -> ListResult; + glob(request: GlobRequest) -> GlobResult; + grep(request: GrepRequest) -> GrepResult; + start_command(request: CommandRequest) -> CommandHandle; + command_status(handle: CommandHandle) -> CommandStatus; + command_output(request: CommandOutputRequest) -> CommandOutput; + cancel_command(handle: CommandHandle) -> (); + } + + #[tokio::test] + async fn identity_mismatch_preserves_primary_cause_and_failed_close_owner() { + for fail_close in [false, true] { + let closes = Arc::new(AtomicUsize::new(0)); + let raw: WorkdirSessionHandle = Arc::new(MismatchedSession { + workdir: Workdir::new("wrong-secret"), + closes: closes.clone(), + fail_close, + }); + let attempts = Arc::new(AtomicUsize::new(0)); + let observed = attempts.clone(); + let session = DeferredWorkdirSession::handle( + "saved", + WorkdirSessionCapabilities::READ_ONLY, + move || { + observed.fetch_add(1, Ordering::SeqCst); + Ok(raw.clone()) + }, + ); + let error = session.read(read_request()).await.unwrap_err(); + assert_eq!( + error.failure_reason(), + Some(WorkdirFailureReason::SessionIdentityMismatch) + ); + assert!(!format!("{error:?}").contains("secret")); + assert_eq!(closes.load(Ordering::SeqCst), 1); + if fail_close { + assert!(matches!( + session.read(read_request()).await, + Err(WorkdirError::SessionClosed) + )); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + } + session.close().await.unwrap(); + assert_eq!( + closes.load(Ordering::SeqCst), + if fail_close { 2 } else { 1 } + ); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + } + } } diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 4027b221a..7fbb8bf25 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -3995,12 +3995,27 @@ impl WorkdirHttpAuthorization for RemoteWorkdirAuthorization { return authorizer .issue_async(method, path_and_query, operation, None, body) .await - .map_err(|error| WorkdirError::Unavailable(error.message)); + .map_err(|error| { + use workdir::WorkdirFailureReason as Reason; + let reason = match error.code.as_str() { + "workspace_runtime_authorization_stale" => Reason::AuthorizationStale, + "workspace_signing_identity_unavailable" => { + Reason::SigningIdentityUnavailable + } + "workspace_runtime_authorization_sign_failed" => { + Reason::AuthorizationSignFailed + } + _ => Reason::AuthorizationUnavailable, + }; + WorkdirError::Unavailable("Workdir authorization is unavailable".into()) + .with_failure_reason(reason) + }); } authorization.fallback_bearer_token.ok_or_else(|| { WorkdirError::Unavailable( "remote Runtime does not have bearer authorization configured".to_string(), ) + .with_failure_reason(workdir::WorkdirFailureReason::AuthorizationMissing) }) }) } diff --git a/crates/workspace-server/src/hosts/tests/workspace_auth_tests.rs b/crates/workspace-server/src/hosts/tests/workspace_auth_tests.rs index 29c0cb52c..f6b8f1c73 100644 --- a/crates/workspace-server/src/hosts/tests/workspace_auth_tests.rs +++ b/crates/workspace-server/src/hosts/tests/workspace_auth_tests.rs @@ -315,3 +315,71 @@ async fn capability_claims_must_match_current_workspace_identity_before_signing( .unwrap(); verify(&fixture.verifier(), &token).unwrap(); } + +#[tokio::test] +async fn remote_workdir_authorization_failures_keep_safe_reason_without_dispatch_or_fallback() { + use workdir::WorkdirFailureReason as Reason; + use workdir::http::{OpenWorkdirSessionRequest, RemoteWorkdirSession, WorkdirTransportError}; + for case in ["stale", "missing_material", "missing_authorization"] { + let fixture = Fixture::new().await; + if case == "stale" { + fixture + .store + .revoke_workspace_runtime_binding(WORKSPACE, RUNTIME, "revoked") + .unwrap(); + } else if case == "missing_material" { + fixture + .materials + .delete(&fixture.identity.private_material_ref) + .unwrap(); + } + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let authorization = RemoteWorkdirAuthorization { + workspace_authorization: (case != "missing_authorization") + .then(|| fixture.authorizer.clone()), + fallback_bearer_token: (case != "missing_authorization") + .then(|| "bearer-secret".into()), + }; + let error = RemoteWorkdirSession::open_with_authorization( + reqwest::Client::builder().no_proxy().build().unwrap(), + format!("http://{}", listener.local_addr().unwrap()) + .parse() + .unwrap(), + Arc::new(authorization), + workdir::Workdir::new("workdir-secret").id().clone(), + OpenWorkdirSessionRequest::default(), + ) + .await + .unwrap_err(); + let reason = match case { + "stale" => Reason::AuthorizationStale, + "missing_material" => Reason::AuthorizationUnavailable, + _ => Reason::AuthorizationMissing, + }; + assert_eq!(error.failure_reason(), Some(reason)); + assert_eq!(reason.stage().as_str(), "authorization_preparation"); + let transport = WorkdirTransportError::from_workdir_error(&error); + assert_eq!(transport.code.http_status(), 503); + assert_eq!(transport.message, "Workdir session is unavailable"); + assert!( + !serde_json::to_string(&transport) + .unwrap() + .contains("secret") + ); + let debug = format!("{error:?}"); + for secret in [ + "secret", + "workspace-signing/", + "private material", + "Authorization:", + "Bearer ", + ] { + assert!(!debug.contains(secret), "{case}: {debug}"); + } + assert!( + matches!(listener.accept(), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock), + "authorization failure sent HTTP or fell back" + ); + } +} diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index a1b36997f..33bed6818 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -13506,6 +13506,7 @@ async fn log_failed_api_response(request: Request, next: Next) -> Response { request_token_id_hash = context.and(request_token_id_hash.as_deref()), denial_reason = error.and_then(|error| error.denial_reason).map(|reason| reason.as_str()), failure_reason = error.and_then(|error| error.failure_reason).map(|reason| reason.as_str()), + failure_stage = error.and_then(|error| error.failure_reason).map(|reason| reason.stage().as_str()), status = event.status, kind = event.kind.unwrap_or("unknown"), message = event.message.unwrap_or(""), @@ -36942,8 +36943,9 @@ impl ApiErrorLog { log.denial_reason = error.denial_reason; log.failure_reason = error.failure_reason; if log.workdir_log.is_some() { - // Existing public API details are unchanged; do not copy request-derived - // alias/handle text or session-open errors into operational diagnostics. + // Public classification and arbitrary diagnostic text are not safe + // operational causes. Keep only the separately typed reason/stage; + // never infer a reason by parsing request- or provider-authored text. log.kind = format!("workdir_session_operation_api_{}", status.as_u16()); // This can also fail after a command was dispatched. Do not claim // a pre-dispatch refusal or absence of effects in the safe message. diff --git a/crates/workspace-server/src/server/tests/api_logging_tests.rs b/crates/workspace-server/src/server/tests/api_logging_tests.rs index fc883168b..604296ac5 100644 --- a/crates/workspace-server/src/server/tests/api_logging_tests.rs +++ b/crates/workspace-server/src/server/tests/api_logging_tests.rs @@ -605,6 +605,10 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() { workdir::WorkdirFailureReason::SessionClosed, workdir::WorkdirFailureReason::CommandTerminalMissing, workdir::WorkdirFailureReason::HttpTimeout, + workdir::WorkdirFailureReason::OperationResultMismatch, + workdir::WorkdirFailureReason::SessionOpenTaskFailed, + workdir::WorkdirFailureReason::AuthorizationStale, + workdir::WorkdirFailureReason::RuntimeStopped, ] { let request = runtime_source_request( &identity, @@ -653,6 +657,7 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() { ); assert_eq!(events.len(), 1, "{events:?}"); assert_eq!(events[0]["failure_reason"], reason.as_str()); + assert_eq!(events[0]["failure_stage"], reason.stage().as_str()); assert_eq!(events[0]["operation"], "command_output"); assert_eq!(events[0]["workdir_stage"], "provider_dispatch"); let log = serde_json::to_string(&events).unwrap(); @@ -1249,3 +1254,54 @@ async fn non_api_workspace_paths_are_not_logged_as_ticket_api_errors() { assert_eq!(status, StatusCode::BAD_REQUEST); assert!(events.is_empty()); } + +#[tokio::test] +async fn session_open_failure_causes_survive_registry_api_conversion_without_raw_diagnostics() { + use workdir::WorkdirFailureReason as Reason; + for reason in [ + Reason::SessionOpenFailed, + Reason::SessionOpenTaskFailed, + Reason::SessionIdentityMismatch, + Reason::AuthorizationStale, + Reason::AuthorizationUnavailable, + Reason::RuntimeStopped, + Reason::WorkingDirectoryRecordInvalid, + Reason::OperationResultMismatch, + ] { + let mut fixture = manual_worker_assignment_fixture().await; + fixture.runtime.workdir_session_factory = Some(Arc::new(move |_| { + Err(workdir::WorkdirError::Unavailable( + "/host/private body-secret proof-secret key-secret credential-secret".into(), + ) + .with_failure_reason(reason)) + })); + fixture + .api + .runtime + .register_or_replace(fixture.runtime.clone()); + let identity = + worker_runtime::auth::RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id) + .unwrap(); + configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id); + let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-session/operations"); + let request = runtime_source_request(&identity, Some(&fixture.worker.worker_id), "POST", &path, serde_json::to_vec(&json!({ + "target_workdir": "checkout", + "operation": {"operation": "read", "request": {"path":"README.md", "offset":0, "limit":10, "max_bytes":1024}} + })).unwrap()); + let (status, body, events) = + request_with_logs(build_router(fixture.api.clone()), request).await; + assert_eq!(status, StatusCode::BAD_GATEWAY, "{body}"); + assert_eq!( + body["message"], + "workdir_session_open_failed: Workdir session is unavailable" + ); + assert!(body.get("failure_reason").is_none(), "{body}"); + assert!(!body.to_string().contains("secret")); + assert_eq!(events.len(), 1, "{events:?}"); + assert_eq!(events[0]["workdir_stage"], "session_open"); + assert_eq!(events[0]["failure_reason"], reason.as_str()); + assert_eq!(events[0]["failure_stage"], reason.stage().as_str()); + assert_eq!(events[0]["message"], "Workdir operation failed"); + assert_eq!(events[0]["diagnostics"], "[]"); + } +}