diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 3b448d409..951a21566 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -125,6 +125,8 @@ pub struct WorkdirOperationApiError { pub message: String, #[serde(default, skip_serializing_if = "Option::is_none")] pub diagnostics: Option>, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub audit_failures: Vec, #[serde(skip, default = "default_repository_error_status")] status: u16, /// Server-local metadata retained by generated error response extensions, never the wire body. @@ -156,6 +158,7 @@ impl WorkdirOperationApiError { code: None, message: error.message, diagnostics: Some(error.diagnostics), + audit_failures: Vec::new(), status: error.status, workdir_log: None, denial_reason: None, @@ -170,6 +173,7 @@ impl WorkdirOperationApiError { code: Some(error.code), message: error.message, diagnostics: None, + audit_failures: Vec::new(), status, workdir_log: None, denial_reason: error.denial_reason, @@ -7522,7 +7526,9 @@ pub struct CurrentWorkerWorkdirOperationRequest { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] #[serde(transparent)] -pub struct CurrentWorkerWorkdirOperationResponse(pub workdir::http::WorkdirSessionOperationResult); +pub struct CurrentWorkerWorkdirOperationResponse( + pub workdir::http::WorkdirSessionOperationResponse, +); #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] diff --git a/crates/tools/src/file_target.rs b/crates/tools/src/file_target.rs index c3bfb535e..af307e0cc 100644 --- a/crates/tools/src/file_target.rs +++ b/crates/tools/src/file_target.rs @@ -167,6 +167,7 @@ pub(crate) fn checked_error(error: workdir::WorkdirError) -> ToolError { | WorkdirError::IsDirectory { .. } | WorkdirError::InvalidGlob { .. } | WorkdirError::InvalidRegex { .. } => "checkout_invalid", + WorkdirError::NotDispatched => "checkout_not_dispatched", WorkdirError::Unavailable { .. } | WorkdirError::SessionClosed => "checkout_unavailable", // IO/transport failures may arrive after effects have begun. Do not infer // no-effects from their text or strip explicit OutcomeUnknown evidence. diff --git a/crates/workdir/src/http.rs b/crates/workdir/src/http.rs index f99f804c5..805aea2de 100644 --- a/crates/workdir/src/http.rs +++ b/crates/workdir/src/http.rs @@ -119,6 +119,51 @@ pub enum WorkdirSessionOperationResult { CommandCancel, } +/// Audit persistence is independent of the provider result and never authorizes a resend. +/// Only server-generated correlation and closed classifications cross this boundary. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct WorkdirAuditFailure { + pub operation_id: String, + pub stage: WorkdirAuditStage, + pub operation_outcome: WorkdirAuditOperationOutcome, + pub reason: WorkdirAuditFailureReason, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum WorkdirAuditStage { + BeforeDispatch, + DispatchRejected, + ResponseReceived, + ResponseUnknown, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum WorkdirAuditOperationOutcome { + NotDispatched, + Succeeded, + Failed { code: WorkdirTransportErrorCode }, + Unknown, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum WorkdirAuditFailureReason { + AuditStoreUnavailable, +} + +/// A successful operation can still have failed audit persistence. Consumers must +/// surface that degradation separately, without discarding the result or resending. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +pub struct WorkdirSessionOperationResponse { + #[serde(flatten)] + pub result: WorkdirSessionOperationResult, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub audit_failures: Vec, +} + /// Dispatch one typed operation against any provider-neutral Workdir session. /// /// Runtime HTTP handlers, outbound External providers, and Backend brokers can @@ -208,6 +253,7 @@ pub enum WorkdirTransportErrorCode { NotFound, Conflict, OutcomeUnknown, + NotDispatched, Unsupported, InvalidRequest, Denied, @@ -231,6 +277,7 @@ impl WorkdirTransportErrorCode { Self::NotFound => "not_found", Self::Conflict => "conflict", Self::OutcomeUnknown => "outcome_unknown", + Self::NotDispatched => "not_dispatched", Self::Unsupported => "unsupported", Self::InvalidRequest => "invalid_request", Self::Denied => "denied", @@ -261,7 +308,7 @@ impl WorkdirTransportErrorCode { | Self::SymlinkTargetIsDirectory | Self::IsDirectory | Self::SymlinkDirectoryNotTraversed => 400, - Self::Unavailable => 503, + Self::Unavailable | Self::NotDispatched => 503, Self::Io | Self::Internal | Self::OutcomeUnknown => 500, Self::Transport => 502, } @@ -298,6 +345,9 @@ impl WorkdirTransportError { Code::OutcomeUnknown, "Workdir operation may have effects; inspect before retrying", ), + WorkdirError::NotDispatched => { + (Code::NotDispatched, "Workdir operation was not dispatched") + } WorkdirError::NotFound(_) => (Code::NotFound, "Workdir path was not found"), WorkdirError::Conflict(_) => ( Code::Conflict, @@ -410,6 +460,7 @@ impl WorkdirTransportError { }, Code::UnknownCommand => WorkdirError::UnknownCommand("".to_string()), Code::Unavailable => WorkdirError::Unavailable(self.message), + Code::NotDispatched => WorkdirError::NotDispatched, Code::Io => WorkdirError::Io { path: "".into(), source: std::io::Error::other(self.message), @@ -1210,6 +1261,37 @@ pub use client::{ mod tests { use super::*; + #[test] + fn operation_response_keeps_success_and_independent_audit_failure() { + let original = + WorkdirSessionOperationResult::CommandStart(crate::CommandHandle("command-a".into())); + let response = WorkdirSessionOperationResponse { + result: original.clone(), + audit_failures: vec![WorkdirAuditFailure { + operation_id: "op-00000000-0000-7000-8000-000000000001".into(), + stage: WorkdirAuditStage::ResponseReceived, + operation_outcome: WorkdirAuditOperationOutcome::Succeeded, + reason: WorkdirAuditFailureReason::AuditStoreUnavailable, + }], + }; + let json = serde_json::to_value(&response).unwrap(); + assert_eq!(json["operation"], "command_start"); + assert_eq!(json["result"], "command-a"); + assert_eq!( + json["audit_failures"][0]["operation_outcome"]["kind"], + "succeeded" + ); + assert_eq!( + serde_json::from_value::(json).unwrap(), + response + ); + let normal = serde_json::to_value(&original).unwrap(); + let decoded: WorkdirSessionOperationResponse = + serde_json::from_value(normal.clone()).unwrap(); + assert!(decoded.audit_failures.is_empty()); + assert_eq!(serde_json::to_value(decoded).unwrap(), normal); + } + #[test] fn transport_error_round_trip_keeps_public_classification() { for (code, expected_status, expected_error) in [ @@ -1229,6 +1311,11 @@ mod tests { 409, "The target file's content or existence changed since it was last observed", ), + ( + WorkdirTransportErrorCode::NotDispatched, + 503, + "not dispatched", + ), (WorkdirTransportErrorCode::Unsupported, 400, "unsupported"), (WorkdirTransportErrorCode::Denied, 403, "denied"), ( diff --git a/crates/workdir/src/lib.rs b/crates/workdir/src/lib.rs index 1a515ac16..c208253f0 100644 --- a/crates/workdir/src/lib.rs +++ b/crates/workdir/src/lib.rs @@ -409,6 +409,7 @@ pub enum WorkdirFailureReason { SessionClosed, ProviderUnavailable, CommandTerminalMissing, + AuditStoreUnavailable, HttpTimeout, HttpConnect, HttpTransfer, @@ -422,6 +423,7 @@ impl WorkdirFailureReason { Self::SessionClosed => "session_closed", Self::ProviderUnavailable => "provider_unavailable", Self::CommandTerminalMissing => "command_terminal_missing", + Self::AuditStoreUnavailable => "audit_store_unavailable", Self::HttpTimeout => "http_timeout", Self::HttpConnect => "http_connect", Self::HttpTransfer => "http_transfer", @@ -466,6 +468,10 @@ pub enum WorkdirError { #[error("Workdir session is unavailable: {0}")] Unavailable(String), + /// The broker has proof that this operation never reached the provider. + #[error("Workdir operation was not dispatched")] + NotDispatched, + #[error("Workdir operation failed")] OperationFailed, diff --git a/crates/worker/src/feature/builtin/manage_workdir.rs b/crates/worker/src/feature/builtin/manage_workdir.rs index 12cfdb65d..d8f2302f2 100644 --- a/crates/worker/src/feature/builtin/manage_workdir.rs +++ b/crates/worker/src/feature/builtin/manage_workdir.rs @@ -291,6 +291,11 @@ impl WorkspaceAttachedWorkdirSession { .execute(request) .map_err(workspace_workdir_error)?; if !response.is_success() { + if let Ok(error) = + serde_json::from_str::(&response.body) + { + notify_workdir_audit_failures(&error.audit_failures); + } return Err( serde_json::from_str::(&response.body) .map(workdir::http::WorkdirTransportError::into_workdir_error) @@ -302,11 +307,14 @@ impl WorkspaceAttachedWorkdirSession { }), ); } - serde_json::from_str(&response.body).map_err(|error| { - WorkdirError::Transport(format!( - "failed to decode Workspace Workdir operation result: {error}" - )) - }) + let response: workdir::http::WorkdirSessionOperationResponse = + serde_json::from_str(&response.body).map_err(|_| { + WorkdirError::Transport( + "failed to decode Workspace Workdir operation result".into(), + ) + })?; + notify_workdir_audit_failures(&response.audit_failures); + Ok(response.result) } fn mismatch(expected: &str) -> WorkdirError { @@ -316,6 +324,27 @@ impl WorkspaceAttachedWorkdirSession { } } +// Notification is separate from the operation return: callers retain successful command +// handles, checkout observations and original failures, and must not resend for audit loss. +fn notify_workdir_audit_failures(failures: &[workdir::http::WorkdirAuditFailure]) { + use sha2::Digest; + for failure in failures { + // Response strings are untrusted. Never copy them into the receiving process log. + let operation_id_hash = sha2::Sha256::digest(failure.operation_id.as_bytes()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + tracing::error!( + event = "external_workdir_audit_failed", + operation_id_hash, + audit_stage = ?failure.stage, + audit_reason = ?failure.reason, + operation_outcome = ?failure.operation_outcome, + "External Workdir audit persistence failed; operation result is independent; do not resend" + ); + } +} + fn workspace_workdir_error(error: WorkspaceClientError) -> WorkdirError { match error { WorkspaceClientError::Unavailable(message) => WorkdirError::Unavailable(message), @@ -1653,6 +1682,67 @@ mod tests { } } + #[tokio::test] + async fn attached_session_keeps_operation_results_when_audit_persistence_failed() { + let failure = json!({ + "operation_id":"op-safe-id", "stage":"response_received", + "reason":"audit_store_unavailable", "operation_outcome":{"kind":"succeeded"} + }); + let mut responses = vec![response(json!({ + "operation":"command_start", "result":"retained-command", + "audit_failures":[failure] + }))]; + for (code, status, outcome) in [ + ("denied", 403, json!({"kind":"failed","code":"denied"})), + ("outcome_unknown", 500, json!({"kind":"unknown"})), + ("not_dispatched", 503, json!({"kind":"not_dispatched"})), + ] { + let mut failure = failure.clone(); + failure["operation_outcome"] = outcome; + responses.push(WorkspaceResponse { status, body: json!({ + "code":code, "message":"safe provider classification", "audit_failures":[failure] + }).to_string() }); + } + let client = Arc::new(RecordingWorkspaceClient::new(responses)); + let session = WorkspaceAttachedWorkdirSession::handle(client.clone(), "checkout"); + let handle = session + .start_command(CommandRequest { + command: "true".into(), + timeout_secs: 1, + transport_byte_budget: workdir::MAX_COMMAND_TRANSPORT_BYTES, + cwd: workdir::WorkdirPath::root(), + tool_call_id: None, + }) + .await + .unwrap(); + assert_eq!(handle.0, "retained-command"); + let request = workdir::CheckoutRequest { + target: workdir::WorkdirPath::new("file.txt").unwrap(), + validator: vec![1], + operation: workdir::CheckoutOperation::Write { + content: b"new".to_vec(), + expected_hash: [0; 32], + }, + }; + assert!(matches!( + session.checkout_execute(request.clone()).await, + Err(WorkdirError::Denied(_)) + )); + assert!(matches!( + session.checkout_execute(request.clone()).await, + Err(WorkdirError::OutcomeUnknown(_)) + )); + assert!(matches!( + session.checkout_execute(request).await, + Err(WorkdirError::NotDispatched) + )); + assert_eq!( + client.requests().len(), + 4, + "audit loss must not trigger a resend" + ); + } + #[tokio::test] async fn attached_session_does_not_send_worker_local_bash_spill_path() { let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({ diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 76d394da4..69a1b035d 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -26,6 +26,7 @@ use subjektiv::api::*; mod backend_job_profile; +mod external_operation_audit; mod subject_authorization; mod subjektiv_job; mod subjektiv_journal; @@ -933,6 +934,10 @@ struct ExternalProviderAuditContext { /// output confirmation from the old connection; unconfirmed cleanup is not reported as success. /// Only confirmed provider shutdown can replace per-command confirmation. Reconnection does /// not prove that missing commands have stopped. +/// +/// Required request auditing fails closed before dispatch. Completion audit failure cannot +/// undo a provider result: it is emitted independently and retained in the current API +/// response, including successful command handles. Audit loss never triggers a resend. struct ExternalProviderWorkdirSession { workdir: workdir::Workdir, connection: Arc, @@ -991,18 +996,22 @@ impl ExternalProviderWorkdirSession { async fn audit_operation( &self, operation_id: &ExternalWorkdirOperationId, - operation_kind: &str, + operation_kind: &'static str, outcome: &str, - ) -> std::result::Result<(), workdir::WorkdirError> { + stage: workdir::http::WorkdirAuditStage, + operation_outcome: workdir::http::WorkdirAuditOperationOutcome, + ) -> std::result::Result<(), workdir::http::WorkdirAuditFailure> { let Some(audit) = &self.audit else { return Ok(()); }; let audit = audit.clone(); + let operation_id_for_notification = operation_id.as_str().to_owned(); + let operation_kind_for_notification = operation_kind; let operation_id = operation_id.clone(); let operation_kind = operation_kind.to_owned(); let outcome = outcome.to_owned(); let connection = self.connection.clone(); - audit + let result = audit .store .db_executor() .run(move || { @@ -1024,12 +1033,17 @@ impl ExternalProviderWorkdirSession { &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), ) }) - .await - .map_err(|error| { - workdir::WorkdirError::Unavailable(format!( - "External Workdir operation audit failed: {error}" - )) - }) + .await; + result.map_err(|_| { + let failure = workdir::http::WorkdirAuditFailure { + operation_id: operation_id_for_notification, + stage, + operation_outcome, + reason: workdir::http::WorkdirAuditFailureReason::AuditStoreUnavailable, + }; + external_operation_audit::notify(failure.clone(), operation_kind_for_notification); + failure + }) } async fn operate( @@ -1079,8 +1093,19 @@ impl ExternalProviderWorkdirSession { } let operation_id = ExternalWorkdirOperationId::new(format!("op-{}", Uuid::now_v7())) .map_err(workdir::WorkdirError::InvalidArgument)?; - self.audit_operation(&operation_id, operation_kind, "requested") - .await?; + use workdir::http::{WorkdirAuditOperationOutcome as Outcome, WorkdirAuditStage as Stage}; + self.audit_operation( + &operation_id, + operation_kind, + "requested", + Stage::BeforeDispatch, + Outcome::NotDispatched, + ) + .await + .map_err(|_| { + workdir::WorkdirError::NotDispatched + .with_failure_reason(workdir::WorkdirFailureReason::AuditStoreUnavailable) + })?; let (response, receive) = tokio::sync::oneshot::channel(); let mut cancel_guard = ExternalOperationCancelGuard { sender: self.connection.sender.clone(), @@ -1097,39 +1122,72 @@ impl ExternalProviderWorkdirSession { .await .is_err() { - self.audit_operation(&operation_id, operation_kind, "provider_unavailable") - .await?; - return Err(workdir::WorkdirError::Unavailable( - "External Workdir provider disconnected".to_string(), - )); + cancel_guard.disarm(); + // A rejected channel send proves there was no provider dispatch. + let _audit_result = self + .audit_operation( + &operation_id, + operation_kind, + "not_dispatched", + Stage::DispatchRejected, + Outcome::NotDispatched, + ) + .await; + return Err(workdir::WorkdirError::NotDispatched + .with_failure_reason(workdir::WorkdirFailureReason::ProviderUnavailable)); } - let result = match tokio::time::timeout(operation_timeout, receive).await { + let (result, stage) = match tokio::time::timeout(operation_timeout, receive).await { Ok(result) => { cancel_guard.disarm(); match result { - Ok(result) => result.map_err(WorkdirTransportError::into_workdir_error), - Err(_) => Err(workdir::WorkdirError::Unavailable( - "External Workdir provider disconnected".to_string(), - )), + Ok(result) => ( + result.map_err(WorkdirTransportError::into_workdir_error), + Stage::ResponseReceived, + ), + Err(_) => ( + Err(workdir::WorkdirError::OutcomeUnknown( + "External Workdir provider response lost; inspect before retrying" + .to_string(), + )), + Stage::ResponseUnknown, + ), } } Err(_) => { cancel_guard.cancel(); - Err(workdir::WorkdirError::Unavailable( - "External Workdir provider operation timed out".to_string(), - )) + ( + Err(workdir::WorkdirError::OutcomeUnknown( + "External Workdir provider response timed out; inspect before retrying" + .to_string(), + )), + Stage::ResponseUnknown, + ) } }; - self.audit_operation( - &operation_id, - operation_kind, - if result.is_ok() { - "completed" - } else { - "failed" + let outcome = match &result { + Ok(_) => Outcome::Succeeded, + Err(error) + if matches!( + error.classification_source(), + workdir::WorkdirError::OutcomeUnknown(_) + ) => + { + Outcome::Unknown + } + Err(error) => Outcome::Failed { + code: WorkdirTransportError::from_workdir_error(error).code, }, - ) - .await?; + }; + let audit_outcome = match outcome { + Outcome::Succeeded => "completed", + Outcome::Failed { .. } => "failed", + Outcome::Unknown => "outcome_unknown", + Outcome::NotDispatched => unreachable!("provider result follows dispatch"), + }; + // Failure is notified independently, never substituted for the provider result. + let _audit_result = self + .audit_operation(&operation_id, operation_kind, audit_outcome, stage, outcome) + .await; result } @@ -13169,26 +13227,38 @@ impl server_api::ServerApi for ServerApiContractService { ), ..Default::default() }; - let Json(response) = scoped_execute_current_worker_workdir_operation( - State( - self.workspace_api() - .map_err(server_api::WorkdirOperationApiError::api)? - .clone(), - ), - AxumPath(ScopedWorkspacePath { workspace_id }), - current_worker_contract_headers(&context).map_err(|error| { - let mut error = server_api::WorkdirOperationApiError::api(error); - error.workdir_log = Some(fallback_log); - error - })?, - Json(WorkspaceWorkdirSessionOperationRequest { - target_workdir: request.target_workdir, - operation: request.operation, - }), - ) - .await - .map_err(InternalWorkdirOperationError::into_contract_error)?; - Ok(server_api::CurrentWorkerWorkdirOperationResponse(response)) + let (result, audit_failures) = + external_operation_audit::capture(scoped_execute_current_worker_workdir_operation( + State( + self.workspace_api() + .map_err(server_api::WorkdirOperationApiError::api)? + .clone(), + ), + AxumPath(ScopedWorkspacePath { workspace_id }), + current_worker_contract_headers(&context).map_err(|error| { + let mut error = server_api::WorkdirOperationApiError::api(error); + error.workdir_log = Some(fallback_log); + error + })?, + Json(WorkspaceWorkdirSessionOperationRequest { + target_workdir: request.target_workdir, + operation: request.operation, + }), + )) + .await; + match result { + Ok(Json(result)) => Ok(server_api::CurrentWorkerWorkdirOperationResponse( + workdir::http::WorkdirSessionOperationResponse { + result, + audit_failures, + }, + )), + Err(error) => { + let mut error = error.into_contract_error(); + error.audit_failures = audit_failures; + Err(error) + } + } } async fn external_workdir_grant_create( diff --git a/crates/workspace-server/src/server/external_operation_audit.rs b/crates/workspace-server/src/server/external_operation_audit.rs new file mode 100644 index 000000000..1b36808d9 --- /dev/null +++ b/crates/workspace-server/src/server/external_operation_audit.rs @@ -0,0 +1,36 @@ +//! Per-request audit notifications, independent of provider result classification. +//! +//! The scope follows the operation future, not a session-wide last-result slot: concurrent +//! requests cannot consume one another's diagnostics. Every failure is also emitted immediately +//! for direct session callers and requests cancelled before a response. This is notification, +//! not a replacement durable audit ledger or crash-recovery/effect-deduplication guarantee. + +use std::{cell::RefCell, future::Future}; +use workdir::http::WorkdirAuditFailure; + +tokio::task_local! { + static FAILURES: RefCell>; +} + +pub(super) async fn capture(future: impl Future) -> (T, Vec) { + FAILURES + .scope(RefCell::new(Vec::new()), async { + let result = future.await; + let failures = FAILURES.with(|failures| failures.take()); + (result, failures) + }) + .await +} + +pub(super) fn notify(failure: WorkdirAuditFailure, operation: &'static str) { + tracing::error!( + event = "external_workdir_audit_failed", + operation, + operation_id = %failure.operation_id, + audit_stage = ?failure.stage, + audit_reason = ?failure.reason, + operation_outcome = ?failure.operation_outcome, + "External Workdir audit persistence failed; operation result is independent; do not resend" + ); + let _ = FAILURES.try_with(|failures| failures.borrow_mut().push(failure)); +} diff --git a/crates/workspace-server/src/server/tests/workspace_admission_tests.rs b/crates/workspace-server/src/server/tests/workspace_admission_tests.rs index 877ce5dff..facd8fe80 100644 --- a/crates/workspace-server/src/server/tests/workspace_admission_tests.rs +++ b/crates/workspace-server/src/server/tests/workspace_admission_tests.rs @@ -1,3 +1,5 @@ +mod audit_tests; + use super::*; use sha2::Digest; use std::sync::atomic::AtomicI64; @@ -459,7 +461,7 @@ async fn provider_error_and_disconnect_release_dispatch_admission() { assert_json( result, if disconnect { - StatusCode::SERVICE_UNAVAILABLE + StatusCode::INTERNAL_SERVER_ERROR } else { StatusCode::FORBIDDEN }, @@ -575,8 +577,8 @@ async fn provider_timeout_releases_admission_without_wall_clock_wait() { let (long, _reply) = f.start_read("1", &mut a).await; tokio::time::pause(); tokio::time::advance(std::time::Duration::from_secs(31)).await; - let body = assert_json(long.await.unwrap(), StatusCode::SERVICE_UNAVAILABLE).await; - assert_eq!(body["code"], "unavailable"); + let body = assert_json(long.await.unwrap(), StatusCode::INTERNAL_SERVER_ERROR).await; + assert_eq!(body["code"], "outcome_unknown"); tokio::time::resume(); let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await; drop( diff --git a/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs b/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs new file mode 100644 index 000000000..212198b6f --- /dev/null +++ b/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs @@ -0,0 +1,546 @@ +//! Integration coverage for signed loopback HTTP -> production admission/auth -> external +//! Workdir session -> real SQLite audit persistence. Only the external provider boundary +//! is substituted with a controlled command receiver. Temporary DBs, listeners and bounded +//! requests are isolated per case; these run in the normal workspace-server library suite. + +use super::*; +use std::io::Write; +use tracing::instrument::WithSubscriber; + +const AUDIT_SECRET: &str = "audit-store-password-secret"; +const WRITE_SECRET: &[u8] = b"write-content-secret"; +const PATH_SECRET: &str = "private-request-path-secret.txt"; +const PROVIDER_SECRET: &str = "/host/private/provider-secret"; + +type ProviderReply = + oneshot::Sender>; + +#[derive(Clone, Default)] +struct CapturedAuditLog(Arc>>); + +impl Write for CapturedAuditLog { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +struct PendingHttpWrite { + task: tokio::task::JoinHandle<(StatusCode, Value)>, + proof: String, +} + +impl Drop for PendingHttpWrite { + fn drop(&mut self) { + self.task.abort(); + } +} + +struct ProviderProbe { + receiver: mpsc::Receiver, + dispatches: usize, +} + +impl ProviderProbe { + async fn receive_write(&mut self, pending: &mut PendingHttpWrite) -> (String, ProviderReply) { + let command = tokio::time::timeout(BOUND, async { + tokio::select! { + command = self.receiver.recv() => command.expect("provider channel closed"), + response = &mut pending.task => { + panic!("HTTP completed before write dispatch: {:?}", response.unwrap()); + } + } + }) + .await + .expect("signed HTTP write never reached provider"); + let ExternalProviderCommand::Operation { + operation_id, + operation, + response, + } = command + else { + panic!("expected write dispatch, received cancellation") + }; + self.dispatches += 1; + let WorkdirSessionOperation::Write(request) = operation else { + panic!("expected mutation, got {operation:?}") + }; + assert_eq!( + request.path, + workdir::WorkdirPath::new(PATH_SECRET).unwrap() + ); + assert_eq!(request.content, WRITE_SECRET); + assert_eq!(request.expected_hash, None); + (operation_id.as_str().to_owned(), response) + } + + fn assert_dispatches(&mut self, expected: usize) { + while let Ok(command) = self.receiver.try_recv() { + if matches!(command, ExternalProviderCommand::Operation { .. }) { + self.dispatches += 1; + } + } + assert_eq!( + self.dispatches, expected, + "unexpected mutation dispatch/retry" + ); + } +} + +impl Fixture { + fn attach_audited_writer(&self) -> ProviderProbe { + let receiver = self.attach("1"); + // Setup changes only: real signed requests still traverse production authorization. + // All three capability sources must agree before the session is first resolved. + let connection = rusqlite::Connection::open(&self.api.config.database_path).unwrap(); + assert_eq!( + connection + .execute( + "UPDATE external_workdir_grants SET permissions='read_write' \ + WHERE workspace_id=?1 AND grant_id='external-1'", + [TEST_WORKSPACE_ID], + ) + .unwrap(), + 1 + ); + assert_eq!( + connection + .execute( + "UPDATE worker_workdir_links SET capabilities='read_write' \ + WHERE workspace_id=?1 AND runtime_id='workdirless-runtime' \ + AND worker_id='1' AND alias='attachment' AND unlinked_at IS NULL", + [TEST_WORKSPACE_ID], + ) + .unwrap(), + 1 + ); + let mut registry = self.api.external_workdir_providers.lock().unwrap(); + let provider = registry.connections.get_mut("external-1").unwrap(); + Arc::get_mut(provider).unwrap().capabilities = + workdir::WorkdirSessionCapabilities::READ_WRITE; + ProviderProbe { + receiver, + dispatches: 0, + } + } +} + +struct AuditHttpFixture { + fixture: Fixture, + logs: CapturedAuditLog, + client: reqwest::Client, + base_url: String, + server: tokio::task::JoinHandle<()>, +} + +impl Drop for AuditHttpFixture { + fn drop(&mut self) { + self.server.abort(); + } +} + +impl AuditHttpFixture { + async fn new() -> (Self, ProviderProbe) { + let fixture = Fixture::new().await; + let provider = fixture.attach_audited_writer(); + let logs = CapturedAuditLog::default(); + let writer = logs.clone(); + let subscriber = tracing_subscriber::fmt() + .without_time() + .with_ansi(false) + .json() + .flatten_event(true) + .with_writer(move || writer.clone()) + .finish(); + let dispatch = tracing::Dispatch::new(subscriber); + // Axum spawns connection tasks: scope the subscriber around each request rather + // than relying on the listener task's thread-local subscriber being inherited. + let app = fixture.app.clone().layer(axum::middleware::from_fn( + move |request: Request, next: axum::middleware::Next| { + let dispatch = dispatch.clone(); + async move { next.run(request).with_subscriber(dispatch).await } + }, + )); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let client = reqwest::Client::builder() + .no_proxy() + .timeout(BOUND) + .build() + .unwrap(); + ( + Self { + fixture, + logs, + client, + base_url, + server, + }, + provider, + ) + } + + fn fail_audit(&self, before_dispatch: bool) { + let filter = if before_dispatch { "=" } else { "<>" }; + rusqlite::Connection::open(&self.fixture.api.config.database_path) + .unwrap() + .execute_batch(&format!( + "CREATE TRIGGER fail_operation_audit BEFORE INSERT ON audit_events \ + WHEN NEW.action='external_workdir_operation' AND NEW.outcome {filter} 'requested' \ + BEGIN SELECT RAISE(FAIL, '{AUDIT_SECRET}'); END;" + )) + .unwrap(); + } + + fn start_write(&self) -> PendingHttpWrite { + let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-session/operations"); + let body = serde_json::to_vec(&WorkspaceWorkdirSessionOperationRequest { + target_workdir: "attachment".into(), + operation: WorkdirSessionOperation::Write(workdir::WriteRequest { + path: workdir::WorkdirPath::new(PATH_SECRET).unwrap(), + content: WRITE_SECRET.to_vec(), + expected_hash: None, + }), + }) + .unwrap(); + let request = self.fixture.request( + Some("1"), + &path, + body.clone(), + START, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + let proof = request.headers()[worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER] + .to_str() + .unwrap() + .to_owned(); + let url = format!("{}{path}", self.base_url); + let client = self.client.clone(); + let headers = request.headers().clone(); + let task = tokio::spawn(async move { + let response = client + .post(url) + .headers(headers) + .body(body) + .send() + .await + .unwrap(); + let status = response.status(); + assert_eq!(response.headers()[CONTENT_TYPE], "application/json"); + let bytes = response.bytes().await.unwrap(); + (status, serde_json::from_slice(&bytes).unwrap()) + }); + PendingHttpWrite { task, proof } + } + + async fn finish( + &self, + mut pending: PendingHttpWrite, + expected_status: StatusCode, + ) -> (Value, Vec) { + let (status, body) = tokio::time::timeout(BOUND, &mut pending.task) + .await + .expect("HTTP did not finish; possible automatic retry") + .unwrap(); + assert_eq!(status, expected_status, "{body}"); + let text = String::from_utf8(self.logs.0.lock().unwrap().clone()).unwrap(); + let encoded_content = serde_json::to_string(WRITE_SECRET).unwrap(); + for output in [body.to_string(), text.clone()] { + for secret in [ + AUDIT_SECRET, + std::str::from_utf8(WRITE_SECRET).unwrap(), + encoded_content.as_str(), + PATH_SECRET, + PROVIDER_SECRET, + pending.proof.as_str(), + self.fixture._root.path().to_str().unwrap(), + ] { + assert!(!output.contains(secret), "sensitive data escaped: {output}"); + } + } + let events = text + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + (body, events) + } + + fn assert_audits(&self, expected: &[(&str, &str)]) { + let connection = + rusqlite::Connection::open(&self.fixture.api.config.database_path).unwrap(); + let mut statement = connection + .prepare( + "SELECT request_id, outcome, summary FROM audit_events \ + WHERE workspace_id=?1 AND action='external_workdir_operation' \ + AND target_id='external-1' ORDER BY rowid", + ) + .unwrap(); + let rows: Vec<(String, String, String)> = statement + .query_map([TEST_WORKSPACE_ID], |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?)) + }) + .unwrap() + .collect::>() + .unwrap(); + let actual: Vec<(&str, &str)> = rows + .iter() + .map(|(operation_id, outcome, _)| (operation_id.as_str(), outcome.as_str())) + .collect(); + assert_eq!(actual, expected, "persisted audits: {rows:?}"); + for (_, _, summary) in rows { + assert!(summary.contains("operation_kind=write"), "{summary}"); + for secret in [ + AUDIT_SECRET, + PATH_SECRET, + PROVIDER_SECRET, + "write-content-secret", + ] { + assert!( + !summary.contains(secret), + "sensitive audit summary: {summary}" + ); + } + } + } + + fn apply_provider_write(&self) { + std::fs::write(self.effect_path(), WRITE_SECRET).unwrap(); + } + + fn effect_path(&self) -> std::path::PathBuf { + self.fixture._root.path().join("provider-effect") + } +} + +fn write_result() -> WorkdirSessionOperationResult { + WorkdirSessionOperationResult::Write(workdir::WriteResult { + bytes_written: WRITE_SECRET.len(), + created: true, + }) +} + +fn assert_audit_failure( + body: &Value, + events: &[Value], + stage: &str, + debug_stage: &str, + outcome: Value, + debug_outcome: &str, +) -> String { + let failures = body["audit_failures"] + .as_array() + .expect("independent audit receipt"); + assert_eq!(failures.len(), 1, "{body}"); + let operation_id = failures[0]["operation_id"].as_str().unwrap(); + assert!(operation_id.starts_with("op-"), "{body}"); + assert_eq!( + failures[0], + json!({ + "operation_id": operation_id, + "stage": stage, + "reason": "audit_store_unavailable", + "operation_outcome": outcome, + }) + ); + let notifications: Vec<&Value> = events + .iter() + .filter(|event| event["event"] == "external_workdir_audit_failed") + .collect(); + assert_eq!(notifications.len(), 1, "{events:?}"); + let event = notifications[0]; + assert_eq!(event["operation_id"], operation_id); + assert_eq!(event["operation"], "write"); + assert_eq!(event["audit_stage"], debug_stage); + assert_eq!(event["audit_reason"], "AuditStoreUnavailable"); + assert_eq!(event["operation_outcome"], debug_outcome); + operation_id.to_owned() +} + +fn assert_failure_reason(events: &[Value], reason: &str) { + let errors: Vec<&Value> = events + .iter() + .filter(|event| event["event"] == "api_error") + .collect(); + assert_eq!(errors.len(), 1, "{events:?}"); + assert_eq!(errors[0]["failure_reason"], reason, "{events:?}"); +} + +#[tokio::test] +async fn successful_write_after_audit_failure_preserves_result_and_does_not_repeat_mutation() { + let (h, mut provider) = AuditHttpFixture::new().await; + h.fail_audit(false); + let mut pending = h.start_write(); + let (dispatched_id, reply) = provider.receive_write(&mut pending).await; + h.apply_provider_write(); + reply.send(Ok(write_result())).unwrap(); + let (mut body, events) = h.finish(pending, StatusCode::OK).await; + let receipt_id = assert_audit_failure( + &body, + &events, + "response_received", + "ResponseReceived", + json!({"kind":"succeeded"}), + "Succeeded", + ); + assert_eq!(receipt_id, dispatched_id); + body.as_object_mut().unwrap().remove("audit_failures"); + assert_eq!(body, serde_json::to_value(write_result()).unwrap()); + assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET); + provider.assert_dispatches(1); + h.assert_audits(&[(&dispatched_id, "requested")]); +} + +#[tokio::test] +async fn denied_write_after_audit_failure_preserves_provider_error_and_separate_receipt() { + let (h, mut provider) = AuditHttpFixture::new().await; + h.fail_audit(false); + let mut pending = h.start_write(); + let (dispatched_id, reply) = provider.receive_write(&mut pending).await; + let mut original = WorkdirTransportError::from_workdir_error(&workdir::WorkdirError::Denied( + PROVIDER_SECRET.into(), + )); + let expected_message = original.message.clone(); + original.message = PROVIDER_SECRET.into(); + reply.send(Err(original)).unwrap(); + let (body, events) = h.finish(pending, StatusCode::FORBIDDEN).await; + assert_eq!(body["code"], "denied"); + assert_eq!(body["message"], expected_message); + let receipt_id = assert_audit_failure( + &body, + &events, + "response_received", + "ResponseReceived", + json!({"kind":"failed", "code":"denied"}), + "Failed { code: Denied }", + ); + assert_eq!(receipt_id, dispatched_id); + assert!(!h.effect_path().exists()); + provider.assert_dispatches(1); + h.assert_audits(&[(&dispatched_id, "requested")]); +} + +#[tokio::test] +async fn lost_write_response_after_audit_failure_stays_unknown_without_repeating_effect() { + let (h, mut provider) = AuditHttpFixture::new().await; + h.fail_audit(false); + let mut pending = h.start_write(); + let (dispatched_id, reply) = provider.receive_write(&mut pending).await; + h.apply_provider_write(); + // The provider applied the effect, but the server cannot infer that from response loss. + drop(reply); + let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await; + assert_eq!(body["code"], "outcome_unknown"); + let receipt_id = assert_audit_failure( + &body, + &events, + "response_unknown", + "ResponseUnknown", + json!({"kind":"unknown"}), + "Unknown", + ); + assert_eq!(receipt_id, dispatched_id); + assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET); + provider.assert_dispatches(1); + h.assert_audits(&[(&dispatched_id, "requested")]); +} + +#[tokio::test] +async fn before_write_audit_failure_is_not_dispatched_and_has_no_provider_effect() { + let (h, mut provider) = AuditHttpFixture::new().await; + h.fail_audit(true); + let (body, events) = h + .finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE) + .await; + assert_eq!(body["code"], "not_dispatched"); + assert_audit_failure( + &body, + &events, + "before_dispatch", + "BeforeDispatch", + json!({"kind":"not_dispatched"}), + "NotDispatched", + ); + assert_failure_reason(&events, "audit_store_unavailable"); + assert!( + body.get("failure_reason").is_none(), + "internal reason leaked: {body}" + ); + assert!(!h.effect_path().exists()); + provider.assert_dispatches(0); + h.assert_audits(&[]); +} + +#[tokio::test] +async fn rejected_write_send_with_audit_failure_is_not_dispatched_not_unknown() { + let (h, mut provider) = AuditHttpFixture::new().await; + h.fail_audit(false); + provider.receiver.close(); + let (body, events) = h + .finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE) + .await; + assert_eq!(body["code"], "not_dispatched"); + let receipt_id = assert_audit_failure( + &body, + &events, + "dispatch_rejected", + "DispatchRejected", + json!({"kind":"not_dispatched"}), + "NotDispatched", + ); + assert_failure_reason(&events, "provider_unavailable"); + assert!(!h.effect_path().exists()); + provider.assert_dispatches(0); + h.assert_audits(&[(&receipt_id, "requested")]); +} + +#[tokio::test] +async fn successful_write_persists_requested_and_completed_with_same_operation_id() { + let (h, mut provider) = AuditHttpFixture::new().await; + let mut pending = h.start_write(); + let (operation_id, reply) = provider.receive_write(&mut pending).await; + h.apply_provider_write(); + reply.send(Ok(write_result())).unwrap(); + let (body, events) = h.finish(pending, StatusCode::OK).await; + assert_eq!(body, serde_json::to_value(write_result()).unwrap()); + assert!( + events + .iter() + .all(|event| event["event"] != "external_workdir_audit_failed") + ); + assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET); + provider.assert_dispatches(1); + h.assert_audits(&[(&operation_id, "requested"), (&operation_id, "completed")]); +} + +#[tokio::test] +async fn provider_reported_unknown_preserves_received_response_and_unknown_effects() { + let (h, mut provider) = AuditHttpFixture::new().await; + h.fail_audit(false); + let mut pending = h.start_write(); + let (dispatched_id, reply) = provider.receive_write(&mut pending).await; + reply + .send(Err(WorkdirTransportError::from_workdir_error( + &workdir::WorkdirError::OutcomeUnknown(PROVIDER_SECRET.into()), + ))) + .unwrap(); + let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await; + assert_eq!(body["code"], "outcome_unknown"); + let receipt_id = assert_audit_failure( + &body, + &events, + "response_received", + "ResponseReceived", + json!({"kind":"unknown"}), + "Unknown", + ); + assert_eq!(receipt_id, dispatched_id); + provider.assert_dispatches(1); + h.assert_audits(&[(&dispatched_id, "requested")]); +}