fix: retain external workdir results independently of audit failures
This commit is contained in:
@@ -125,6 +125,8 @@ pub struct WorkdirOperationApiError {
|
|||||||
pub message: String,
|
pub message: String,
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||||
pub diagnostics: Option<Vec<Diagnostic>>,
|
pub diagnostics: Option<Vec<Diagnostic>>,
|
||||||
|
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||||
|
pub audit_failures: Vec<workdir::http::WorkdirAuditFailure>,
|
||||||
#[serde(skip, default = "default_repository_error_status")]
|
#[serde(skip, default = "default_repository_error_status")]
|
||||||
status: u16,
|
status: u16,
|
||||||
/// Server-local metadata retained by generated error response extensions, never the wire body.
|
/// Server-local metadata retained by generated error response extensions, never the wire body.
|
||||||
@@ -156,6 +158,7 @@ impl WorkdirOperationApiError {
|
|||||||
code: None,
|
code: None,
|
||||||
message: error.message,
|
message: error.message,
|
||||||
diagnostics: Some(error.diagnostics),
|
diagnostics: Some(error.diagnostics),
|
||||||
|
audit_failures: Vec::new(),
|
||||||
status: error.status,
|
status: error.status,
|
||||||
workdir_log: None,
|
workdir_log: None,
|
||||||
denial_reason: None,
|
denial_reason: None,
|
||||||
@@ -170,6 +173,7 @@ impl WorkdirOperationApiError {
|
|||||||
code: Some(error.code),
|
code: Some(error.code),
|
||||||
message: error.message,
|
message: error.message,
|
||||||
diagnostics: None,
|
diagnostics: None,
|
||||||
|
audit_failures: Vec::new(),
|
||||||
status,
|
status,
|
||||||
workdir_log: None,
|
workdir_log: None,
|
||||||
denial_reason: error.denial_reason,
|
denial_reason: error.denial_reason,
|
||||||
@@ -7522,7 +7526,9 @@ pub struct CurrentWorkerWorkdirOperationRequest {
|
|||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
|
||||||
#[serde(transparent)]
|
#[serde(transparent)]
|
||||||
pub struct CurrentWorkerWorkdirOperationResponse(pub workdir::http::WorkdirSessionOperationResult);
|
pub struct CurrentWorkerWorkdirOperationResponse(
|
||||||
|
pub workdir::http::WorkdirSessionOperationResponse,
|
||||||
|
);
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
|
||||||
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
|
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
|
||||||
|
|||||||
@@ -167,6 +167,7 @@ pub(crate) fn checked_error(error: workdir::WorkdirError) -> ToolError {
|
|||||||
| WorkdirError::IsDirectory { .. }
|
| WorkdirError::IsDirectory { .. }
|
||||||
| WorkdirError::InvalidGlob { .. }
|
| WorkdirError::InvalidGlob { .. }
|
||||||
| WorkdirError::InvalidRegex { .. } => "checkout_invalid",
|
| WorkdirError::InvalidRegex { .. } => "checkout_invalid",
|
||||||
|
WorkdirError::NotDispatched => "checkout_not_dispatched",
|
||||||
WorkdirError::Unavailable { .. } | WorkdirError::SessionClosed => "checkout_unavailable",
|
WorkdirError::Unavailable { .. } | WorkdirError::SessionClosed => "checkout_unavailable",
|
||||||
// IO/transport failures may arrive after effects have begun. Do not infer
|
// IO/transport failures may arrive after effects have begun. Do not infer
|
||||||
// no-effects from their text or strip explicit OutcomeUnknown evidence.
|
// no-effects from their text or strip explicit OutcomeUnknown evidence.
|
||||||
|
|||||||
@@ -119,6 +119,51 @@ pub enum WorkdirSessionOperationResult {
|
|||||||
CommandCancel,
|
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<WorkdirAuditFailure>,
|
||||||
|
}
|
||||||
|
|
||||||
/// Dispatch one typed operation against any provider-neutral Workdir session.
|
/// Dispatch one typed operation against any provider-neutral Workdir session.
|
||||||
///
|
///
|
||||||
/// Runtime HTTP handlers, outbound External providers, and Backend brokers can
|
/// Runtime HTTP handlers, outbound External providers, and Backend brokers can
|
||||||
@@ -208,6 +253,7 @@ pub enum WorkdirTransportErrorCode {
|
|||||||
NotFound,
|
NotFound,
|
||||||
Conflict,
|
Conflict,
|
||||||
OutcomeUnknown,
|
OutcomeUnknown,
|
||||||
|
NotDispatched,
|
||||||
Unsupported,
|
Unsupported,
|
||||||
InvalidRequest,
|
InvalidRequest,
|
||||||
Denied,
|
Denied,
|
||||||
@@ -231,6 +277,7 @@ impl WorkdirTransportErrorCode {
|
|||||||
Self::NotFound => "not_found",
|
Self::NotFound => "not_found",
|
||||||
Self::Conflict => "conflict",
|
Self::Conflict => "conflict",
|
||||||
Self::OutcomeUnknown => "outcome_unknown",
|
Self::OutcomeUnknown => "outcome_unknown",
|
||||||
|
Self::NotDispatched => "not_dispatched",
|
||||||
Self::Unsupported => "unsupported",
|
Self::Unsupported => "unsupported",
|
||||||
Self::InvalidRequest => "invalid_request",
|
Self::InvalidRequest => "invalid_request",
|
||||||
Self::Denied => "denied",
|
Self::Denied => "denied",
|
||||||
@@ -261,7 +308,7 @@ impl WorkdirTransportErrorCode {
|
|||||||
| Self::SymlinkTargetIsDirectory
|
| Self::SymlinkTargetIsDirectory
|
||||||
| Self::IsDirectory
|
| Self::IsDirectory
|
||||||
| Self::SymlinkDirectoryNotTraversed => 400,
|
| Self::SymlinkDirectoryNotTraversed => 400,
|
||||||
Self::Unavailable => 503,
|
Self::Unavailable | Self::NotDispatched => 503,
|
||||||
Self::Io | Self::Internal | Self::OutcomeUnknown => 500,
|
Self::Io | Self::Internal | Self::OutcomeUnknown => 500,
|
||||||
Self::Transport => 502,
|
Self::Transport => 502,
|
||||||
}
|
}
|
||||||
@@ -298,6 +345,9 @@ impl WorkdirTransportError {
|
|||||||
Code::OutcomeUnknown,
|
Code::OutcomeUnknown,
|
||||||
"Workdir operation may have effects; inspect before retrying",
|
"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::NotFound(_) => (Code::NotFound, "Workdir path was not found"),
|
||||||
WorkdirError::Conflict(_) => (
|
WorkdirError::Conflict(_) => (
|
||||||
Code::Conflict,
|
Code::Conflict,
|
||||||
@@ -410,6 +460,7 @@ impl WorkdirTransportError {
|
|||||||
},
|
},
|
||||||
Code::UnknownCommand => WorkdirError::UnknownCommand("<remote>".to_string()),
|
Code::UnknownCommand => WorkdirError::UnknownCommand("<remote>".to_string()),
|
||||||
Code::Unavailable => WorkdirError::Unavailable(self.message),
|
Code::Unavailable => WorkdirError::Unavailable(self.message),
|
||||||
|
Code::NotDispatched => WorkdirError::NotDispatched,
|
||||||
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(self.message),
|
||||||
@@ -1210,6 +1261,37 @@ pub use client::{
|
|||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
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::<WorkdirSessionOperationResponse>(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]
|
#[test]
|
||||||
fn transport_error_round_trip_keeps_public_classification() {
|
fn transport_error_round_trip_keeps_public_classification() {
|
||||||
for (code, expected_status, expected_error) in [
|
for (code, expected_status, expected_error) in [
|
||||||
@@ -1229,6 +1311,11 @@ mod tests {
|
|||||||
409,
|
409,
|
||||||
"The target file's content or existence changed since it was last observed",
|
"The target file's content or existence changed since it was last observed",
|
||||||
),
|
),
|
||||||
|
(
|
||||||
|
WorkdirTransportErrorCode::NotDispatched,
|
||||||
|
503,
|
||||||
|
"not dispatched",
|
||||||
|
),
|
||||||
(WorkdirTransportErrorCode::Unsupported, 400, "unsupported"),
|
(WorkdirTransportErrorCode::Unsupported, 400, "unsupported"),
|
||||||
(WorkdirTransportErrorCode::Denied, 403, "denied"),
|
(WorkdirTransportErrorCode::Denied, 403, "denied"),
|
||||||
(
|
(
|
||||||
|
|||||||
@@ -409,6 +409,7 @@ pub enum WorkdirFailureReason {
|
|||||||
SessionClosed,
|
SessionClosed,
|
||||||
ProviderUnavailable,
|
ProviderUnavailable,
|
||||||
CommandTerminalMissing,
|
CommandTerminalMissing,
|
||||||
|
AuditStoreUnavailable,
|
||||||
HttpTimeout,
|
HttpTimeout,
|
||||||
HttpConnect,
|
HttpConnect,
|
||||||
HttpTransfer,
|
HttpTransfer,
|
||||||
@@ -422,6 +423,7 @@ impl WorkdirFailureReason {
|
|||||||
Self::SessionClosed => "session_closed",
|
Self::SessionClosed => "session_closed",
|
||||||
Self::ProviderUnavailable => "provider_unavailable",
|
Self::ProviderUnavailable => "provider_unavailable",
|
||||||
Self::CommandTerminalMissing => "command_terminal_missing",
|
Self::CommandTerminalMissing => "command_terminal_missing",
|
||||||
|
Self::AuditStoreUnavailable => "audit_store_unavailable",
|
||||||
Self::HttpTimeout => "http_timeout",
|
Self::HttpTimeout => "http_timeout",
|
||||||
Self::HttpConnect => "http_connect",
|
Self::HttpConnect => "http_connect",
|
||||||
Self::HttpTransfer => "http_transfer",
|
Self::HttpTransfer => "http_transfer",
|
||||||
@@ -466,6 +468,10 @@ pub enum WorkdirError {
|
|||||||
#[error("Workdir session is unavailable: {0}")]
|
#[error("Workdir session is unavailable: {0}")]
|
||||||
Unavailable(String),
|
Unavailable(String),
|
||||||
|
|
||||||
|
/// The broker has proof that this operation never reached the provider.
|
||||||
|
#[error("Workdir operation was not dispatched")]
|
||||||
|
NotDispatched,
|
||||||
|
|
||||||
#[error("Workdir operation failed")]
|
#[error("Workdir operation failed")]
|
||||||
OperationFailed,
|
OperationFailed,
|
||||||
|
|
||||||
|
|||||||
@@ -291,6 +291,11 @@ impl WorkspaceAttachedWorkdirSession {
|
|||||||
.execute(request)
|
.execute(request)
|
||||||
.map_err(workspace_workdir_error)?;
|
.map_err(workspace_workdir_error)?;
|
||||||
if !response.is_success() {
|
if !response.is_success() {
|
||||||
|
if let Ok(error) =
|
||||||
|
serde_json::from_str::<server_api::WorkdirOperationApiError>(&response.body)
|
||||||
|
{
|
||||||
|
notify_workdir_audit_failures(&error.audit_failures);
|
||||||
|
}
|
||||||
return Err(
|
return Err(
|
||||||
serde_json::from_str::<workdir::http::WorkdirTransportError>(&response.body)
|
serde_json::from_str::<workdir::http::WorkdirTransportError>(&response.body)
|
||||||
.map(workdir::http::WorkdirTransportError::into_workdir_error)
|
.map(workdir::http::WorkdirTransportError::into_workdir_error)
|
||||||
@@ -302,11 +307,14 @@ impl WorkspaceAttachedWorkdirSession {
|
|||||||
}),
|
}),
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
serde_json::from_str(&response.body).map_err(|error| {
|
let response: workdir::http::WorkdirSessionOperationResponse =
|
||||||
WorkdirError::Transport(format!(
|
serde_json::from_str(&response.body).map_err(|_| {
|
||||||
"failed to decode Workspace Workdir operation result: {error}"
|
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 {
|
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::<String>();
|
||||||
|
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 {
|
fn workspace_workdir_error(error: WorkspaceClientError) -> WorkdirError {
|
||||||
match error {
|
match error {
|
||||||
WorkspaceClientError::Unavailable(message) => WorkdirError::Unavailable(message),
|
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]
|
#[tokio::test]
|
||||||
async fn attached_session_does_not_send_worker_local_bash_spill_path() {
|
async fn attached_session_does_not_send_worker_local_bash_spill_path() {
|
||||||
let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
|
let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
|
||||||
|
|||||||
@@ -26,6 +26,7 @@
|
|||||||
|
|
||||||
use subjektiv::api::*;
|
use subjektiv::api::*;
|
||||||
mod backend_job_profile;
|
mod backend_job_profile;
|
||||||
|
mod external_operation_audit;
|
||||||
mod subject_authorization;
|
mod subject_authorization;
|
||||||
mod subjektiv_job;
|
mod subjektiv_job;
|
||||||
mod subjektiv_journal;
|
mod subjektiv_journal;
|
||||||
@@ -933,6 +934,10 @@ struct ExternalProviderAuditContext {
|
|||||||
/// output confirmation from the old connection; unconfirmed cleanup is not reported as success.
|
/// output confirmation from the old connection; unconfirmed cleanup is not reported as success.
|
||||||
/// Only confirmed provider shutdown can replace per-command confirmation. Reconnection does
|
/// Only confirmed provider shutdown can replace per-command confirmation. Reconnection does
|
||||||
/// not prove that missing commands have stopped.
|
/// 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 {
|
struct ExternalProviderWorkdirSession {
|
||||||
workdir: workdir::Workdir,
|
workdir: workdir::Workdir,
|
||||||
connection: Arc<ExternalProviderConnection>,
|
connection: Arc<ExternalProviderConnection>,
|
||||||
@@ -991,18 +996,22 @@ impl ExternalProviderWorkdirSession {
|
|||||||
async fn audit_operation(
|
async fn audit_operation(
|
||||||
&self,
|
&self,
|
||||||
operation_id: &ExternalWorkdirOperationId,
|
operation_id: &ExternalWorkdirOperationId,
|
||||||
operation_kind: &str,
|
operation_kind: &'static str,
|
||||||
outcome: &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 {
|
let Some(audit) = &self.audit else {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
};
|
};
|
||||||
let audit = audit.clone();
|
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_id = operation_id.clone();
|
||||||
let operation_kind = operation_kind.to_owned();
|
let operation_kind = operation_kind.to_owned();
|
||||||
let outcome = outcome.to_owned();
|
let outcome = outcome.to_owned();
|
||||||
let connection = self.connection.clone();
|
let connection = self.connection.clone();
|
||||||
audit
|
let result = audit
|
||||||
.store
|
.store
|
||||||
.db_executor()
|
.db_executor()
|
||||||
.run(move || {
|
.run(move || {
|
||||||
@@ -1024,11 +1033,16 @@ impl ExternalProviderWorkdirSession {
|
|||||||
&Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
|
&Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
|
||||||
)
|
)
|
||||||
})
|
})
|
||||||
.await
|
.await;
|
||||||
.map_err(|error| {
|
result.map_err(|_| {
|
||||||
workdir::WorkdirError::Unavailable(format!(
|
let failure = workdir::http::WorkdirAuditFailure {
|
||||||
"External Workdir operation audit failed: {error}"
|
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
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1079,8 +1093,19 @@ impl ExternalProviderWorkdirSession {
|
|||||||
}
|
}
|
||||||
let operation_id = ExternalWorkdirOperationId::new(format!("op-{}", Uuid::now_v7()))
|
let operation_id = ExternalWorkdirOperationId::new(format!("op-{}", Uuid::now_v7()))
|
||||||
.map_err(workdir::WorkdirError::InvalidArgument)?;
|
.map_err(workdir::WorkdirError::InvalidArgument)?;
|
||||||
self.audit_operation(&operation_id, operation_kind, "requested")
|
use workdir::http::{WorkdirAuditOperationOutcome as Outcome, WorkdirAuditStage as Stage};
|
||||||
.await?;
|
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 (response, receive) = tokio::sync::oneshot::channel();
|
||||||
let mut cancel_guard = ExternalOperationCancelGuard {
|
let mut cancel_guard = ExternalOperationCancelGuard {
|
||||||
sender: self.connection.sender.clone(),
|
sender: self.connection.sender.clone(),
|
||||||
@@ -1097,39 +1122,72 @@ impl ExternalProviderWorkdirSession {
|
|||||||
.await
|
.await
|
||||||
.is_err()
|
.is_err()
|
||||||
{
|
{
|
||||||
self.audit_operation(&operation_id, operation_kind, "provider_unavailable")
|
cancel_guard.disarm();
|
||||||
.await?;
|
// A rejected channel send proves there was no provider dispatch.
|
||||||
return Err(workdir::WorkdirError::Unavailable(
|
let _audit_result = self
|
||||||
"External Workdir provider disconnected".to_string(),
|
.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) => {
|
Ok(result) => {
|
||||||
cancel_guard.disarm();
|
cancel_guard.disarm();
|
||||||
match result {
|
match result {
|
||||||
Ok(result) => result.map_err(WorkdirTransportError::into_workdir_error),
|
Ok(result) => (
|
||||||
Err(_) => Err(workdir::WorkdirError::Unavailable(
|
result.map_err(WorkdirTransportError::into_workdir_error),
|
||||||
"External Workdir provider disconnected".to_string(),
|
Stage::ResponseReceived,
|
||||||
|
),
|
||||||
|
Err(_) => (
|
||||||
|
Err(workdir::WorkdirError::OutcomeUnknown(
|
||||||
|
"External Workdir provider response lost; inspect before retrying"
|
||||||
|
.to_string(),
|
||||||
)),
|
)),
|
||||||
|
Stage::ResponseUnknown,
|
||||||
|
),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Err(_) => {
|
Err(_) => {
|
||||||
cancel_guard.cancel();
|
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(
|
let outcome = match &result {
|
||||||
&operation_id,
|
Ok(_) => Outcome::Succeeded,
|
||||||
operation_kind,
|
Err(error)
|
||||||
if result.is_ok() {
|
if matches!(
|
||||||
"completed"
|
error.classification_source(),
|
||||||
} else {
|
workdir::WorkdirError::OutcomeUnknown(_)
|
||||||
"failed"
|
) =>
|
||||||
|
{
|
||||||
|
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
|
result
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -13169,7 +13227,8 @@ impl server_api::ServerApi for ServerApiContractService {
|
|||||||
),
|
),
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
let Json(response) = scoped_execute_current_worker_workdir_operation(
|
let (result, audit_failures) =
|
||||||
|
external_operation_audit::capture(scoped_execute_current_worker_workdir_operation(
|
||||||
State(
|
State(
|
||||||
self.workspace_api()
|
self.workspace_api()
|
||||||
.map_err(server_api::WorkdirOperationApiError::api)?
|
.map_err(server_api::WorkdirOperationApiError::api)?
|
||||||
@@ -13185,10 +13244,21 @@ impl server_api::ServerApi for ServerApiContractService {
|
|||||||
target_workdir: request.target_workdir,
|
target_workdir: request.target_workdir,
|
||||||
operation: request.operation,
|
operation: request.operation,
|
||||||
}),
|
}),
|
||||||
)
|
))
|
||||||
.await
|
.await;
|
||||||
.map_err(InternalWorkdirOperationError::into_contract_error)?;
|
match result {
|
||||||
Ok(server_api::CurrentWorkerWorkdirOperationResponse(response))
|
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(
|
async fn external_workdir_grant_create(
|
||||||
|
|||||||
@@ -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<Vec<WorkdirAuditFailure>>;
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(super) async fn capture<T>(future: impl Future<Output = T>) -> (T, Vec<WorkdirAuditFailure>) {
|
||||||
|
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));
|
||||||
|
}
|
||||||
@@ -1,3 +1,5 @@
|
|||||||
|
mod audit_tests;
|
||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
use sha2::Digest;
|
use sha2::Digest;
|
||||||
use std::sync::atomic::AtomicI64;
|
use std::sync::atomic::AtomicI64;
|
||||||
@@ -459,7 +461,7 @@ async fn provider_error_and_disconnect_release_dispatch_admission() {
|
|||||||
assert_json(
|
assert_json(
|
||||||
result,
|
result,
|
||||||
if disconnect {
|
if disconnect {
|
||||||
StatusCode::SERVICE_UNAVAILABLE
|
StatusCode::INTERNAL_SERVER_ERROR
|
||||||
} else {
|
} else {
|
||||||
StatusCode::FORBIDDEN
|
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;
|
let (long, _reply) = f.start_read("1", &mut a).await;
|
||||||
tokio::time::pause();
|
tokio::time::pause();
|
||||||
tokio::time::advance(std::time::Duration::from_secs(31)).await;
|
tokio::time::advance(std::time::Duration::from_secs(31)).await;
|
||||||
let body = assert_json(long.await.unwrap(), StatusCode::SERVICE_UNAVAILABLE).await;
|
let body = assert_json(long.await.unwrap(), StatusCode::INTERNAL_SERVER_ERROR).await;
|
||||||
assert_eq!(body["code"], "unavailable");
|
assert_eq!(body["code"], "outcome_unknown");
|
||||||
tokio::time::resume();
|
tokio::time::resume();
|
||||||
let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await;
|
let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await;
|
||||||
drop(
|
drop(
|
||||||
|
|||||||
@@ -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<std::result::Result<WorkdirSessionOperationResult, WorkdirTransportError>>;
|
||||||
|
|
||||||
|
#[derive(Clone, Default)]
|
||||||
|
struct CapturedAuditLog(Arc<std::sync::Mutex<Vec<u8>>>);
|
||||||
|
|
||||||
|
impl Write for CapturedAuditLog {
|
||||||
|
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
|
||||||
|
self.0.lock().unwrap().extend_from_slice(bytes);
|
||||||
|
Ok(bytes.len())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn flush(&mut self) -> std::io::Result<()> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
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<ExternalProviderCommand>,
|
||||||
|
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<Body>, 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<Value>) {
|
||||||
|
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::<rusqlite::Result<_>>()
|
||||||
|
.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")]);
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user