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.
This commit is contained in:
@@ -375,12 +375,14 @@ impl WorkdirTransportError {
|
|||||||
use WorkdirTransportErrorCode as Code;
|
use WorkdirTransportErrorCode as Code;
|
||||||
let error = match self.code {
|
let error = match self.code {
|
||||||
Code::NotFound => WorkdirError::NotFound("<remote>".into()),
|
Code::NotFound => WorkdirError::NotFound("<remote>".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(
|
Code::OutcomeUnknown => WorkdirError::OutcomeUnknown(
|
||||||
"Workdir operation may have effects; inspect before retrying".into(),
|
"Workdir operation may have effects; inspect before retrying".into(),
|
||||||
),
|
),
|
||||||
Code::Unsupported => WorkdirError::UnsupportedOperation(self.message),
|
Code::Unsupported => WorkdirError::UnsupportedOperation("Workdir operation is not supported".into()),
|
||||||
Code::InvalidRequest => WorkdirError::InvalidArgument(self.message),
|
Code::InvalidRequest => WorkdirError::InvalidArgument("Workdir operation request is invalid".into()),
|
||||||
Code::Denied => WorkdirError::Denied(crate::WorkdirDenial {
|
Code::Denied => WorkdirError::Denied(crate::WorkdirDenial {
|
||||||
reason: self.denial_reason,
|
reason: self.denial_reason,
|
||||||
// Provider-authored text is not trusted diagnostic data.
|
// Provider-authored text is not trusted diagnostic data.
|
||||||
@@ -409,12 +411,12 @@ impl WorkdirTransportError {
|
|||||||
target: "<remote-target>".into(),
|
target: "<remote-target>".into(),
|
||||||
},
|
},
|
||||||
Code::UnknownCommand => WorkdirError::UnknownCommand("<remote>".to_string()),
|
Code::UnknownCommand => WorkdirError::UnknownCommand("<remote>".to_string()),
|
||||||
Code::Unavailable => WorkdirError::Unavailable(self.message),
|
Code::Unavailable => WorkdirError::Unavailable("Workdir session is unavailable".into()),
|
||||||
Code::Io => WorkdirError::Io {
|
Code::Io => WorkdirError::Io {
|
||||||
path: "<remote>".into(),
|
path: "<remote>".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,
|
Code::Internal => WorkdirError::OperationFailed,
|
||||||
};
|
};
|
||||||
let error = if let Some(reason) = self.failure_reason {
|
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() {
|
if opened.workdir_id.as_str() != workdir_id.as_str() {
|
||||||
return Err(WorkdirError::Unavailable(
|
return Err(WorkdirError::Unavailable(
|
||||||
"Runtime opened a session for a different Workdir".to_string(),
|
"Runtime opened a session for a different Workdir".to_string(),
|
||||||
));
|
)
|
||||||
|
.with_failure_reason(crate::WorkdirFailureReason::SessionIdentityMismatch));
|
||||||
}
|
}
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
client,
|
client,
|
||||||
@@ -608,6 +611,7 @@ mod client {
|
|||||||
WorkdirError::Unavailable(format!(
|
WorkdirError::Unavailable(format!(
|
||||||
"Runtime returned a mismatched Workdir operation result; expected {expected}"
|
"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(
|
Ok(_) if mutation => Err(WorkdirError::OutcomeUnknown(
|
||||||
"mismatched response after checkout mutation".into(),
|
"mismatched response after checkout mutation".into(),
|
||||||
)),
|
)
|
||||||
|
.with_failure_reason(crate::WorkdirFailureReason::OperationResultMismatch)),
|
||||||
Ok(_) => Err(Self::mismatch("checkout_execute")),
|
Ok(_) => Err(Self::mismatch("checkout_execute")),
|
||||||
Err(error)
|
Err(error)
|
||||||
if mutation
|
if mutation
|
||||||
|
|||||||
@@ -414,6 +414,24 @@ pub enum WorkdirFailureReason {
|
|||||||
HttpTransfer,
|
HttpTransfer,
|
||||||
HttpResponseInvalid,
|
HttpResponseInvalid,
|
||||||
HttpResponseTooLarge,
|
HttpResponseTooLarge,
|
||||||
|
OperationResultMismatch,
|
||||||
|
SessionIdentityMismatch,
|
||||||
|
SessionOpenTaskFailed,
|
||||||
|
SessionOpenFailed,
|
||||||
|
AuthorizationMissing,
|
||||||
|
AuthorizationStale,
|
||||||
|
AuthorizationUnavailable,
|
||||||
|
SigningIdentityUnavailable,
|
||||||
|
AuthorizationSignFailed,
|
||||||
|
RuntimeStopped,
|
||||||
|
RuntimeStateUnavailable,
|
||||||
|
RuntimeBackendUnavailable,
|
||||||
|
RuntimeOwnerMismatch,
|
||||||
|
RuntimeInvalidRequest,
|
||||||
|
WorkingDirectoryNotFound,
|
||||||
|
WorkingDirectoryNotActive,
|
||||||
|
WorkingDirectoryRecordInvalid,
|
||||||
|
WorkingDirectoryAccessUnavailable,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl WorkdirFailureReason {
|
impl WorkdirFailureReason {
|
||||||
@@ -427,6 +445,83 @@ impl WorkdirFailureReason {
|
|||||||
Self::HttpTransfer => "http_transfer",
|
Self::HttpTransfer => "http_transfer",
|
||||||
Self::HttpResponseInvalid => "http_response_invalid",
|
Self::HttpResponseInvalid => "http_response_invalid",
|
||||||
Self::HttpResponseTooLarge => "http_response_too_large",
|
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",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -83,3 +83,34 @@ fn provider_cannot_inject_free_text_failure_reason() {
|
|||||||
.is_err()
|
.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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -2187,7 +2187,25 @@ impl RuntimeHttpWorkdirError {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn runtime(error: RuntimeError) -> Self {
|
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)
|
RuntimeError::WorkingDirectory(diagnostic)
|
||||||
if diagnostic.code == "working_directory_not_found" =>
|
if diagnostic.code == "working_directory_not_found" =>
|
||||||
{
|
{
|
||||||
@@ -2200,7 +2218,9 @@ impl RuntimeHttpWorkdirError {
|
|||||||
WorkdirTransportErrorCode::Unavailable,
|
WorkdirTransportErrorCode::Unavailable,
|
||||||
"Workdir session is 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(),
|
status = self.status.as_u16(),
|
||||||
code = self.payload.code.as_str(),
|
code = self.payload.code.as_str(),
|
||||||
failure_reason = reason.as_str(),
|
failure_reason = reason.as_str(),
|
||||||
|
failure_stage = reason.stage().as_str(),
|
||||||
"Workdir provider operation failed"
|
"Workdir provider operation failed"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
@@ -2434,6 +2455,7 @@ pub enum RuntimeHttpServerError {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
mod command_transport_tests;
|
mod command_transport_tests;
|
||||||
|
mod runtime_workdir_failure_tests;
|
||||||
#[cfg(feature = "fs-store")]
|
#[cfg(feature = "fs-store")]
|
||||||
mod session_material_tests;
|
mod session_material_tests;
|
||||||
|
|
||||||
|
|||||||
@@ -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<Mutex<Vec<u8>>>);
|
||||||
|
|
||||||
|
impl Write for LogCapture {
|
||||||
|
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
|
||||||
|
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::<serde_json::Value>(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<u8>,
|
||||||
|
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<Self>) {
|
||||||
|
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<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
|
||||||
|
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<WorkdirSessionHandle, WorkingDirectoryDiagnostic> {
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -66,13 +66,32 @@ impl DeferredWorkdirSession {
|
|||||||
let open = self.open.clone();
|
let open = self.open.clone();
|
||||||
let session = tokio::task::spawn_blocking(move || open())
|
let session = tokio::task::spawn_blocking(move || open())
|
||||||
.await
|
.await
|
||||||
.map_err(|_| WorkdirError::Unavailable("Workdir session opener failed".into()))?
|
.map_err(|_| {
|
||||||
.map_err(WorkdirError::Unavailable)?;
|
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() {
|
if session.workdir().id() != self.workdir.id() {
|
||||||
session.close().await?;
|
if let Err(error) = session.close().await {
|
||||||
return Err(WorkdirError::Unavailable(
|
// Retain a failed-close owner for explicit close; never use it.
|
||||||
"Workdir session identity mismatch".into(),
|
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() {
|
if let Some(mut source) = session.subscribe_command_events() {
|
||||||
let events = self.events.clone();
|
let events = self.events.clone();
|
||||||
@@ -224,10 +243,16 @@ mod tests {
|
|||||||
},
|
},
|
||||||
);
|
);
|
||||||
assert_eq!(attempts.load(Ordering::SeqCst), 0);
|
assert_eq!(attempts.load(Ordering::SeqCst), 0);
|
||||||
|
let failed = session.read(read_request()).await.unwrap_err();
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
session.read(read_request()).await,
|
failed.classification_source(),
|
||||||
Err(WorkdirError::Unavailable(_))
|
WorkdirError::Unavailable(_)
|
||||||
));
|
));
|
||||||
|
assert_eq!(
|
||||||
|
failed.failure_reason(),
|
||||||
|
Some(WorkdirFailureReason::SessionOpenFailed)
|
||||||
|
);
|
||||||
|
assert_eq!(attempts.load(Ordering::SeqCst), 1);
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
session.read(read_request()).await.unwrap().bytes,
|
session.read(read_request()).await.unwrap().bytes,
|
||||||
b"restored"
|
b"restored"
|
||||||
@@ -253,4 +278,126 @@ mod tests {
|
|||||||
);
|
);
|
||||||
session.close().await.unwrap();
|
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<AtomicUsize>,
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3995,12 +3995,27 @@ impl WorkdirHttpAuthorization for RemoteWorkdirAuthorization {
|
|||||||
return authorizer
|
return authorizer
|
||||||
.issue_async(method, path_and_query, operation, None, body)
|
.issue_async(method, path_and_query, operation, None, body)
|
||||||
.await
|
.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(|| {
|
authorization.fallback_bearer_token.ok_or_else(|| {
|
||||||
WorkdirError::Unavailable(
|
WorkdirError::Unavailable(
|
||||||
"remote Runtime does not have bearer authorization configured".to_string(),
|
"remote Runtime does not have bearer authorization configured".to_string(),
|
||||||
)
|
)
|
||||||
|
.with_failure_reason(workdir::WorkdirFailureReason::AuthorizationMissing)
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -315,3 +315,71 @@ async fn capability_claims_must_match_current_workspace_identity_before_signing(
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
verify(&fixture.verifier(), &token).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"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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()),
|
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()),
|
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_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,
|
status = event.status,
|
||||||
kind = event.kind.unwrap_or("unknown"),
|
kind = event.kind.unwrap_or("unknown"),
|
||||||
message = event.message.unwrap_or(""),
|
message = event.message.unwrap_or(""),
|
||||||
@@ -36942,8 +36943,9 @@ impl ApiErrorLog {
|
|||||||
log.denial_reason = error.denial_reason;
|
log.denial_reason = error.denial_reason;
|
||||||
log.failure_reason = error.failure_reason;
|
log.failure_reason = error.failure_reason;
|
||||||
if log.workdir_log.is_some() {
|
if log.workdir_log.is_some() {
|
||||||
// Existing public API details are unchanged; do not copy request-derived
|
// Public classification and arbitrary diagnostic text are not safe
|
||||||
// alias/handle text or session-open errors into operational diagnostics.
|
// 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());
|
log.kind = format!("workdir_session_operation_api_{}", status.as_u16());
|
||||||
// This can also fail after a command was dispatched. Do not claim
|
// This can also fail after a command was dispatched. Do not claim
|
||||||
// a pre-dispatch refusal or absence of effects in the safe message.
|
// a pre-dispatch refusal or absence of effects in the safe message.
|
||||||
|
|||||||
@@ -605,6 +605,10 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() {
|
|||||||
workdir::WorkdirFailureReason::SessionClosed,
|
workdir::WorkdirFailureReason::SessionClosed,
|
||||||
workdir::WorkdirFailureReason::CommandTerminalMissing,
|
workdir::WorkdirFailureReason::CommandTerminalMissing,
|
||||||
workdir::WorkdirFailureReason::HttpTimeout,
|
workdir::WorkdirFailureReason::HttpTimeout,
|
||||||
|
workdir::WorkdirFailureReason::OperationResultMismatch,
|
||||||
|
workdir::WorkdirFailureReason::SessionOpenTaskFailed,
|
||||||
|
workdir::WorkdirFailureReason::AuthorizationStale,
|
||||||
|
workdir::WorkdirFailureReason::RuntimeStopped,
|
||||||
] {
|
] {
|
||||||
let request = runtime_source_request(
|
let request = runtime_source_request(
|
||||||
&identity,
|
&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.len(), 1, "{events:?}");
|
||||||
assert_eq!(events[0]["failure_reason"], reason.as_str());
|
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]["operation"], "command_output");
|
||||||
assert_eq!(events[0]["workdir_stage"], "provider_dispatch");
|
assert_eq!(events[0]["workdir_stage"], "provider_dispatch");
|
||||||
let log = serde_json::to_string(&events).unwrap();
|
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_eq!(status, StatusCode::BAD_REQUEST);
|
||||||
assert!(events.is_empty());
|
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"], "[]");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user