diff --git a/docs/public/execution/run-configuration.mdx b/docs/public/execution/run-configuration.mdx index 1b51a9161..c8bc920a6 100644 --- a/docs/public/execution/run-configuration.mdx +++ b/docs/public/execution/run-configuration.mdx @@ -306,6 +306,30 @@ SERVICE_URL = "https://api.{{ env.REGION }}.example.com" Missing host variables produce a hard error pointing at the specific field and unresolved token. +Vault secrets use the same interpolation form with `{{ secrets.NAME }}`. Use +secret tokens for values that should be treated as sensitive even when they do +not look like credentials, such as environment names or deployment labels: + +```toml title="run.toml" +[environments.ci.env] +DEPLOY_ENV = "{{ secrets.DEPLOY_ENV }}" + +[[run.prepare.steps]] +command = ["./deploy", "--environment", "{{ secrets.DEPLOY_ENV }}"] +``` + +Declared secret values are registered when the run starts and are redacted by +exact match on structured run surfaces: events, `progress.jsonl`, and setup +failure errors. This redaction applies regardless of the value's shape, so a +low-entropy value such as `staging` is still replaced with `REDACTED` in those +surfaces. Fabro also keeps its content-based redaction pass for credential-shaped +values that were not declared with `{{ secrets.NAME }}`. + +The boundary is the structured data Fabro captures. Once a secret is passed into +a sandbox process environment, text the sandbox prints is covered where Fabro +captures it back into structured events or setup errors; this is not a blanket +guarantee for every live process stream or file the sandbox writes. + ### `[run.integrations.github.permissions]` Request a scoped GitHub App token for workflow stages that need `GITHUB_TOKEN` inside the sandbox. Values map directly to GitHub App permission names and access levels. diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index 9eba9a3d5..2fca9a250 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -178,12 +178,12 @@ pub(crate) async fn execute( fabro_run_tools, }; - let execution = async { + let execution = Box::pin(async { match mode { RunWorkerMode::Start => operations::start(&run_dir, services).await, RunWorkerMode::Resume => operations::resume(&run_dir, services).await, } - }; + }); if let Some(mut control_manager) = control_manager { tokio::select! { @@ -1457,7 +1457,10 @@ mod tests { ); let event = running_event(None); - sink.write_run_event(&event).await.unwrap(); + // This worker-stamp unit test emits no run-declared secret values. + sink.write_run_event(&event, &fabro_redact::SecretRedactor::default()) + .await + .unwrap(); let first = first.lock().await; let second = second.lock().await; diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index c246e396d..1a9b6632a 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -67,7 +67,7 @@ use fabro_llm::types::{ use fabro_mcp_store::McpServerStore; use fabro_model::catalog::LlmCatalogSettings; use fabro_model::{BilledTokenCounts, Catalog, ModelRef, ModelTestMode, ProviderId}; -use fabro_redact::redact_jsonl_line; +use fabro_redact::{redact_json_value, redact_jsonl_line}; use fabro_sandbox::daytona::{self, DaytonaSandbox}; use fabro_sandbox::details::sandbox_details; use fabro_sandbox::reconnect::reconnect_for_run; diff --git a/lib/crates/fabro-server/src/server/handler/events.rs b/lib/crates/fabro-server/src/server/handler/events.rs index b8508bd5c..bfebaeebf 100644 --- a/lib/crates/fabro-server/src/server/handler/events.rs +++ b/lib/crates/fabro-server/src/server/handler/events.rs @@ -11,8 +11,8 @@ use super::super::{ EventPayload, HashSet, IntoResponse, Json, KeepAlive, PaginatedEventList, PaginationMeta, Path, Query, RequireRunManagementTarget, RequireRunScoped, RequireRunStageScoped, RequiredUser, Response, Router, RunEvent, RunId, Sse, State, StatusCode, StreamExt, UnboundedReceiverStream, - broadcast, get, mpsc, parse_run_id_path, parse_stage_id_path, redact_jsonl_line, - reject_if_archived, update_live_run_from_event, + broadcast, get, mpsc, parse_run_id_path, parse_stage_id_path, redact_json_value, + redact_jsonl_line, reject_if_archived, update_live_run_from_event, }; pub(super) fn routes() -> Router> { @@ -175,10 +175,18 @@ async fn append_run_event( )) .into_response(); } - let payload = match EventPayload::new(value, &id) { + let redacted_value = redact_json_value(value); + let payload = match EventPayload::new(redacted_value, &id) { Ok(payload) => payload, Err(err) => return ApiError::bad_request(err.to_string()).into_response(), }; + let event = match RunEvent::from_ref(payload.as_value()) { + Ok(event) => event, + Err(err) => { + return ApiError::bad_request(format!("Invalid redacted run event: {err}")) + .into_response(); + } + }; match state.stores.runs.open_run(&id).await { Ok(run_store) => match run_store.append_event(&payload).await { @@ -352,16 +360,23 @@ fn event_properties(event: &RunEvent) -> serde_json::Map serde_json::Map { - build_redacted_event_payload(event, &event.run_id) - .ok() - .and_then(|payload| { - payload - .as_value() - .get("properties") - .and_then(serde_json::Value::as_object) - .cloned() - }) - .unwrap_or_else(|| event_properties(event)) + // Event detail is a read path. Stored events are trusted to have been + // redacted at worker/source and server ingest; no per-run registry is + // available here, so preserve the existing pattern-only projection. + build_redacted_event_payload( + event, + &event.run_id, + &fabro_redact::SecretRedactor::default(), + ) + .ok() + .and_then(|payload| { + payload + .as_value() + .get("properties") + .and_then(serde_json::Value::as_object) + .cloned() + }) + .unwrap_or_else(|| event_properties(event)) } fn truncate_content(value: String, max_content_length: usize) -> (String, bool) { diff --git a/lib/crates/fabro-server/src/server/handler/sessions.rs b/lib/crates/fabro-server/src/server/handler/sessions.rs index 3ac297f07..457200688 100644 --- a/lib/crates/fabro-server/src/server/handler/sessions.rs +++ b/lib/crates/fabro-server/src/server/handler/sessions.rs @@ -1346,7 +1346,7 @@ async fn append_run_session_event( actor: None, body, }; - let payload = EventPayload::new(event.to_value()?, &run_id)?; + let payload = EventPayload::new(fabro_redact::redact_json_value(event.to_value()?), &run_id)?; run_store.append_event_envelope(&payload).await } diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index 214bff225..28b2d9293 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -4824,7 +4824,12 @@ async fn append_scoped_stage_event( parallel_branch_id: None, }; let stored = fabro_workflow::event::to_run_event_at(&run_id, event, Utc::now(), Some(&scope)); - let payload = fabro_workflow::event::build_redacted_event_payload(&stored, &run_id).unwrap(); + let payload = fabro_workflow::event::build_redacted_event_payload( + &stored, + &run_id, + &fabro_redact::SecretRedactor::default(), + ) + .unwrap(); let run_store = state.stores.runs.open_run(&run_id).await.unwrap(); run_store.append_event(&payload).await.unwrap(); } @@ -9616,6 +9621,45 @@ async fn append_run_event_rejects_reserved_archive_event() { ); } +#[tokio::test] +async fn append_run_event_redacts_credential_shaped_values_before_storage() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = create_run(&app, MINIMAL_DOT) + .await + .parse::() + .unwrap(); + let secret = "sk-ant-api03-xK9mZ2vL8nQ5rT1wY4bC7dF0gH3jE6pA"; + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/events"))) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "id": "evt-run-notice-redaction", + "ts": "2026-04-19T12:00:00Z", + "run_id": run_id, + "event": "run.notice", + "properties": { + "level": "warn", + "code": "credential_detected", + "message": format!("token={secret}") + } + }) + .to_string(), + )) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + response_json!(response, StatusCode::OK).await; + + let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap(); + let serialized = serde_json::to_string(&run_store.list_events().await.unwrap()).unwrap(); + assert!(!serialized.contains(secret)); + assert!(serialized.contains("REDACTED")); +} + #[tokio::test] async fn get_checkpoint_returns_null_initially() { let state = test_app_state(); diff --git a/lib/crates/fabro-workflow/src/event/redaction.rs b/lib/crates/fabro-workflow/src/event/redaction.rs index 5ae44d111..1dca025d3 100644 --- a/lib/crates/fabro-workflow/src/event/redaction.rs +++ b/lib/crates/fabro-workflow/src/event/redaction.rs @@ -1,17 +1,26 @@ use ::fabro_types::{RunEvent, RunId}; use anyhow::{Context, Result}; -use fabro_redact::redact_json_value; +use fabro_redact::SecretRedactor; use fabro_store::EventPayload; use fabro_util::json::normalize_json_value; use serde_json::Value; -pub fn build_redacted_event_payload(event: &RunEvent, run_id: &RunId) -> Result { - let value = redacted_event_value(event)?; +pub fn build_redacted_event_payload( + event: &RunEvent, + run_id: &RunId, + redactor: &SecretRedactor, +) -> Result { + let value = redacted_event_value(event, redactor)?; EventPayload::new(value, run_id).map_err(anyhow::Error::from) } -pub fn redacted_event_json(event: &RunEvent) -> Result { - serde_json::to_string(&redacted_event_value(event)?).map_err(anyhow::Error::from) +pub fn redacted_event_json(event: &RunEvent, redactor: &SecretRedactor) -> Result { + serde_json::to_string(&redacted_event_value(event, redactor)?).map_err(anyhow::Error::from) +} + +pub(super) fn redacted_run_event(event: &RunEvent, redactor: &SecretRedactor) -> Result { + let value = redacted_event_value(event, redactor)?; + RunEvent::from_ref(&value).context("Failed to reparse redacted event payload") } fn normalized_event_value(event: &RunEvent) -> Result { @@ -19,8 +28,96 @@ fn normalized_event_value(event: &RunEvent) -> Result { Ok(normalize_json_value(value)) } -fn redacted_event_value(event: &RunEvent) -> Result { - Ok(redact_json_value(normalized_event_value(event)?)) +fn redacted_event_value(event: &RunEvent, redactor: &SecretRedactor) -> Result { + let mut value = fabro_redact::redact_json_value(normalized_event_value(event)?); + redact_registered_secrets_in_event_value(&mut value, redactor); + Ok(value) +} + +fn redact_registered_secrets_in_event_value(value: &mut Value, redactor: &SecretRedactor) { + if redactor.is_empty() { + return; + } + + if let Some(Value::String(node_label)) = value.get_mut("node_label") { + let redacted = redactor.redact_into(node_label); + if redacted != *node_label { + *node_label = redacted; + } + } + + if let Some(properties) = value.get_mut("properties") { + redact_registered_secrets_in_properties(properties, redactor); + } +} + +fn redact_registered_secrets_in_properties(value: &mut Value, redactor: &SecretRedactor) { + match value { + Value::Object(properties) => { + for (key, child) in properties { + // Registered values can be low-entropy words. Redacting them in + // every event field would corrupt structural values such as ids, + // enum strings, and event names. Until event-field redaction is + // derived from typed field metadata, add new free-form text + // properties here when they are introduced. + if is_free_form_text_property(key) { + *child = redactor.redact_json(std::mem::take(child)); + } else { + redact_registered_secrets_in_properties(child, redactor); + } + } + } + Value::Array(items) => { + for item in items { + redact_registered_secrets_in_properties(item, redactor); + } + } + _ => {} + } +} + +fn is_free_form_text_property(key: &str) -> bool { + matches!( + key, + // command/script I/O + "command" + | "script" + | "stdout" + | "stderr" + | "output" + | "input" + | "arguments" + | "exec_output_tail" + | "tool_input" + | "tool_output" + // agent/LLM text + | "prompt" + | "response" + | "answer" + | "question" + | "delta" + | "text" + | "message" + | "reason" + | "notes" + | "preview" + // errors + | "error" + | "error_message" + | "failure" + | "causes" + | "details" + | "description" + // diffs/config + | "diff" + | "final_patch" + | "workflow_config" + | "workflow_source" + // metadata text + | "goal" + | "subject" + | "title" + ) } pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result { @@ -31,6 +128,7 @@ pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result Result<()> { let stored = to_run_event(run_id, event); - let payload = build_redacted_event_payload(&stored, run_id)?; + // Direct RunDatabase appends are server/test lifecycle paths with no + // per-run declared-secret registry. Worker run output goes through + // RunEventSink/RunEventLogger with the run redactor. + let payload = build_redacted_event_payload(&stored, run_id, &SecretRedactor::default())?; run_store .append_event(&payload) .await @@ -27,9 +31,10 @@ pub async fn append_event_to_sink( sink: &RunEventSink, run_id: &RunId, event: &Event, + redactor: &SecretRedactor, ) -> Result<()> { let stored = to_run_event(run_id, event); - sink.write_run_event(&stored).await + sink.write_run_event(&stored, redactor).await } #[derive(Clone)] @@ -99,21 +104,25 @@ impl RunEventSink { } } - pub async fn write_run_event(&self, event: &RunEvent) -> Result<()> { + pub async fn write_run_event(&self, event: &RunEvent, redactor: &SecretRedactor) -> Result<()> { let mut pending = vec![(self, event.clone())]; while let Some((sink, event)) = pending.pop() { match sink { Self::Store(run_store) => { + let event = redacted_run_event(&event, redactor)?; run_store.append_run_event(&event).await?; } Self::JsonLines(writer) => { - let line = redacted_event_json(&event)?; + let line = redacted_event_json(&event, redactor)?; let mut writer = writer.lock().await; writer.write_all(line.as_bytes()).await?; writer.write_all(b"\n").await?; writer.flush().await?; } - Self::Callback(callback) => callback(event).await?, + Self::Callback(callback) => { + let event = redacted_run_event(&event, redactor)?; + callback(event).await?; + } Self::Map { transform, inner } => { pending.push((inner.as_ref(), transform(event))); } @@ -144,14 +153,15 @@ pub struct RunEventLogger { impl RunEventLogger { #[must_use] - pub fn new(sink: RunEventSink) -> Self { + pub fn new(sink: RunEventSink, redactor: SecretRedactor) -> Self { let (tx, mut rx) = mpsc::unbounded_channel(); + let logger_redactor = redactor; tokio::spawn(async move { while let Some(command) = rx.recv().await { match command { RunEventCommand::Event(event) => { - if let Err(err) = sink.write_run_event(&event).await { + if let Err(err) = sink.write_run_event(&event, &logger_redactor).await { tracing::warn!(error = %err, "Failed to write run event"); } } @@ -193,9 +203,9 @@ pub struct StoreProgressLogger { impl StoreProgressLogger { #[must_use] - pub fn new(run_store: impl Into) -> Self { + pub fn new(run_store: impl Into, redactor: SecretRedactor) -> Self { Self { - inner: RunEventLogger::new(RunEventSink::backend(run_store.into())), + inner: RunEventLogger::new(RunEventSink::backend(run_store.into()), redactor), } } @@ -261,7 +271,9 @@ mod tests { message: "notice".to_string(), exec_output_tail: None, }); - let payload = build_redacted_event_payload(&stored, &fixtures::RUN_7).unwrap(); + let payload = + build_redacted_event_payload(&stored, &fixtures::RUN_7, &SecretRedactor::default()) + .unwrap(); run_store.append_event(&payload).await.unwrap(); let events = run_store.list_events().await.unwrap(); @@ -282,8 +294,9 @@ mod tests { let (writer, reader) = tokio::io::duplex(4096); let sink = RunEventSink::json_lines(writer); let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested { actor: None }); + let redactor = SecretRedactor::default(); - sink.write_run_event(&event).await.unwrap(); + sink.write_run_event(&event, &redactor).await.unwrap(); let mut reader = BufReader::new(reader); let mut line = String::new(); @@ -323,8 +336,9 @@ mod tests { ]), ); let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested { actor: None }); + let redactor = SecretRedactor::default(); - sink.write_run_event(&event).await.unwrap(); + sink.write_run_event(&event, &redactor).await.unwrap(); let first = first.lock().await; let second = second.lock().await; @@ -340,7 +354,7 @@ mod tests { let (writer, reader) = tokio::io::duplex(4096); let sink = RunEventSink::json_lines(writer); - let logger = RunEventLogger::new(sink); + let logger = RunEventLogger::new(sink, SecretRedactor::default()); let emitter = Emitter::new(fixtures::RUN_8); logger.register(&emitter); @@ -354,4 +368,32 @@ mod tests { let payload = event_payload_from_redacted_json(line.trim_end(), &fixtures::RUN_8).unwrap(); assert_eq!(payload.as_value()["event"], "run.paused"); } + + #[tokio::test] + async fn run_event_sink_callback_receives_redacted_event() { + let received = Arc::new(AsyncMutex::new(Vec::new())); + let received_events = Arc::clone(&received); + let sink = RunEventSink::callback(move |event| { + let received_events = Arc::clone(&received_events); + async move { + received_events.lock().await.push(event); + Ok(()) + } + }); + let event = to_run_event(&fixtures::RUN_7, &Event::RunNotice { + level: RunNoticeLevel::Warn, + code: "example".to_string(), + message: "deploy to staging".to_string(), + exec_output_tail: None, + }); + let redactor = SecretRedactor::default(); + redactor.register("staging"); + + sink.write_run_event(&event, &redactor).await.unwrap(); + + let events = received.lock().await; + let text = serde_json::to_string(&events[0].to_value().unwrap()).unwrap(); + assert!(!text.contains("staging")); + assert!(text.contains("REDACTED")); + } } diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 16dbe0230..50dd32067 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -461,7 +461,10 @@ mod tests { .run .with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1))) .with_run_store(run_store.clone().into()); - let logger = crate::event::StoreProgressLogger::new(run_store.clone()); + let logger = crate::event::StoreProgressLogger::new( + run_store.clone(), + fabro_redact::SecretRedactor::default(), + ); logger.register(services.run.emitter.as_ref()); (services, run_store, logger) } diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 146324013..09186ffc3 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -328,7 +328,10 @@ mod tests { .run .with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1))) .with_run_store(run_store.clone().into()); - let logger = crate::event::StoreProgressLogger::new(run_store.clone()); + let logger = crate::event::StoreProgressLogger::new( + run_store.clone(), + fabro_redact::SecretRedactor::default(), + ); logger.register(services.run.emitter.as_ref()); (services, run_store, logger) } diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 22aa2aa9c..ffe92961c 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -783,7 +783,10 @@ mod tests { .run .with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1))) .with_run_store(run_store.clone().into()); - let logger = crate::event::StoreProgressLogger::new(run_store.clone()); + let logger = crate::event::StoreProgressLogger::new( + run_store.clone(), + fabro_redact::SecretRedactor::default(), + ); logger.register(services.run.emitter.as_ref()); let mut node = Node::new("par"); node.attrs.insert( @@ -836,7 +839,10 @@ mod tests { .run .with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1))) .with_run_store(run_store.clone().into()); - let logger = crate::event::StoreProgressLogger::new(run_store.clone()); + let logger = crate::event::StoreProgressLogger::new( + run_store.clone(), + fabro_redact::SecretRedactor::default(), + ); logger.register(services.run.emitter.as_ref()); let mut node = Node::new("par"); node.attrs.insert( diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 48cbeeb32..c56e38470 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -260,7 +260,10 @@ mod tests { .run .with_emitter(Arc::new(Emitter::new(fixtures::RUN_1))) .with_run_store(run_store.clone().into()); - let logger = crate::event::StoreProgressLogger::new(run_store.clone()); + let logger = crate::event::StoreProgressLogger::new( + run_store.clone(), + fabro_redact::SecretRedactor::default(), + ); logger.register(services.run.emitter.as_ref()); (services, run_store, logger) } diff --git a/lib/crates/fabro-workflow/src/operations/fork.rs b/lib/crates/fabro-workflow/src/operations/fork.rs index d2353bdfc..6e1996fc4 100644 --- a/lib/crates/fabro-workflow/src/operations/fork.rs +++ b/lib/crates/fabro-workflow/src/operations/fork.rs @@ -208,8 +208,15 @@ async fn replay_historical_projection_events( let mut event = envelope.event.clone(); event.id = format!("{new_run_id}-fork-{}", envelope.seq); event.run_id = new_run_id; - let payload = event::build_redacted_event_payload(&event, &new_run_id) - .map_err(|err| Error::engine(err.to_string()))?; + // Fork replay copies events that were already persisted through the + // source run's redaction boundary; no live run secret registry exists + // for the historical projection replay. + let payload = event::build_redacted_event_payload( + &event, + &new_run_id, + &fabro_redact::SecretRedactor::default(), + ) + .map_err(|err| Error::engine(err.to_string()))?; run_store .append_event(&payload) .await diff --git a/lib/crates/fabro-workflow/src/operations/resume.rs b/lib/crates/fabro-workflow/src/operations/resume.rs index beaad87a9..47dcd9490 100644 --- a/lib/crates/fabro-workflow/src/operations/resume.rs +++ b/lib/crates/fabro-workflow/src/operations/resume.rs @@ -1,5 +1,7 @@ use std::path::Path; +use fabro_redact::SecretRedactor; + use super::start::{StartServices, Started, execute_persisted_run}; use crate::error::Error; use crate::event::{Event, append_event_to_sink}; @@ -39,10 +41,14 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result>>, catalog: Arc, fabro_run_tools: Option, + secret_redactor: SecretRedactor, } struct ResolvedStartLlm { @@ -145,6 +147,9 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result Result Result = Err(error.clone()); let (_, _, run_status) = classify_engine_result(&engine_result); @@ -339,6 +365,7 @@ async fn persist_terminal_engine_failure( error, reason, crate::millis_u64(duration), + redactor, ) .await; } @@ -373,6 +400,7 @@ impl RunSession { let configured = configured_providers_for_start(services.vault.as_ref(), Arc::clone(&catalog)).await; let llm = resolve_start_llm(catalog.as_ref(), &configured, resolved)?; + let secret_redactor = SecretRedactor::default(); let vault_guard = match services.vault.as_ref() { Some(vault) => Some(vault.read().await), None => None, @@ -380,7 +408,9 @@ impl RunSession { // Token-only secrets lookup over the vault read guard, shared across // every run-boundary resolver. A missing or non-Token secret becomes // `None`, so resolution fails closed with a secret error. - let secret_lookup = |name: &str| vault_token_lookup(vault_guard.as_deref(), name); + let secret_lookup = |name: &str| { + registered_vault_token_lookup(vault_guard.as_deref(), &secret_redactor, name) + }; let mcp_servers = resolved .agent .mcps @@ -507,6 +537,7 @@ impl RunSession { vault: services.vault, catalog, fabro_run_tools: services.fabro_run_tools, + secret_redactor, }) } } @@ -566,6 +597,16 @@ fn vault_token_lookup(vault: Option<&Vault>, name: &str) -> Option { vault.and_then(|vault| fabro_auth::vault_get_token(vault, name).ok().flatten()) } +fn registered_vault_token_lookup( + vault: Option<&Vault>, + redactor: &SecretRedactor, + name: &str, +) -> Option { + let value = vault_token_lookup(vault, name)?; + redactor.register(value.clone()); + Some(value) +} + async fn load_accepted_run_definition( run_store: &RunStoreHandle, blob_id: fabro_types::RunBlobId, @@ -827,7 +868,8 @@ impl RunSession { }); } - let store_progress_logger = RunEventLogger::new(self.event_sink.clone()); + let store_progress_logger = + RunEventLogger::new(self.event_sink.clone(), self.secret_redactor.clone()); store_progress_logger.register(self.emitter.as_ref()); let init_options = InitOptions { @@ -854,8 +896,17 @@ impl RunSession { checkpoint, seed_context: self.seed_context, fabro_run_tools: self.fabro_run_tools, + secret_redactor: self.secret_redactor.clone(), + }; + let mut initialized = match Box::pin(pipeline::initialize(persisted, init_options)).await { + Ok(initialized) => initialized, + Err(err) => { + // Initialize can emit setup-failure events before returning; + // flush them before the terminal failure path appends run.failed. + store_progress_logger.flush().await; + return Err(err); + } }; - let mut initialized = Box::pin(pipeline::initialize(persisted, init_options)).await?; initialized.on_node = on_node; let sandbox_for_cleanup = Arc::clone(&initialized.engine.run.sandbox); @@ -969,6 +1020,9 @@ impl Drop for DetachedRunBootstrapGuard { let event_sink = self.event_sink.clone(); if let Ok(handle) = Handle::try_current() { handle.spawn(async move { + // Bootstrap drop failures happen before any run-boundary + // declared secret can be resolved into output. + let redactor = SecretRedactor::default(); emit_workflow_run_failed( run_id, &run_store, @@ -976,6 +1030,7 @@ impl Drop for DetachedRunBootstrapGuard { &Error::engine(reason.to_string()), reason, 0, + &redactor, ) .await; }); @@ -1005,11 +1060,12 @@ async fn run_store_reaches_terminal(run_store: &RunStoreHandle, timeout: Duratio } struct DetachedRunCompletionGuard { - event_sink: RunEventSink, - run_id: RunId, - run_store: RunStoreHandle, - cancel_token: CancellationToken, - active: bool, + event_sink: RunEventSink, + run_id: RunId, + run_store: RunStoreHandle, + cancel_token: CancellationToken, + secret_redactor: SecretRedactor, + active: bool, } impl DetachedRunCompletionGuard { @@ -1018,12 +1074,14 @@ impl DetachedRunCompletionGuard { run_store: RunStoreHandle, event_sink: RunEventSink, cancel_token: CancellationToken, + secret_redactor: SecretRedactor, ) -> Self { Self { event_sink, run_id, run_store, cancel_token, + secret_redactor, active: true, } } @@ -1058,6 +1116,7 @@ impl Drop for DetachedRunCompletionGuard { let event_sink = self.event_sink.clone(); let run_id = self.run_id; let run_store = self.run_store.clone(); + let secret_redactor = self.secret_redactor.clone(); if let Ok(handle) = Handle::try_current() { handle.spawn(async move { if run_store_reaches_terminal(&run_store, DETACHED_COMPLETION_GUARD_TERMINAL_GRACE) @@ -1072,14 +1131,20 @@ impl Drop for DetachedRunCompletionGuard { &Error::engine(message.to_string()), reason, 0, + &secret_redactor, ) .await; - let _ = append_event_to_sink(&event_sink, &run_id, &Event::RunNotice { - level: RunNoticeLevel::Error, - code: code.to_string(), - message: message.to_string(), - exec_output_tail: None, - }) + let _ = append_event_to_sink( + &event_sink, + &run_id, + &Event::RunNotice { + level: RunNoticeLevel::Error, + code: code.to_string(), + message: message.to_string(), + exec_output_tail: None, + }, + &secret_redactor, + ) .await; }); } @@ -1094,8 +1159,9 @@ async fn persist_detached_failure( phase: &'static str, reason: FailureReason, error: &Error, + redactor: &SecretRedactor, ) -> Result<(), Error> { - emit_workflow_run_failed(run_id, run_store, event_sink, error, reason, 0).await; + emit_workflow_run_failed(run_id, run_store, event_sink, error, reason, 0, redactor).await; let event = Event::RunNotice { level: RunNoticeLevel::Error, @@ -1103,7 +1169,7 @@ async fn persist_detached_failure( message: error.to_string(), exec_output_tail: None, }; - if let Err(err) = append_event_to_sink(event_sink, &run_id, &event).await { + if let Err(err) = append_event_to_sink(event_sink, &run_id, &event, redactor).await { tracing::warn!(error = %err, "Failed to append detached failure notice"); } @@ -1568,6 +1634,159 @@ reasoning = false ); } + #[tokio::test] + async fn run_session_registered_secret_redacts_pre_stage_stored_event() { + let temp = tempfile::tempdir().unwrap(); + let (storage_root, _run_dir) = storage_root_and_run_dir(&temp); + let mut settings = settings_from_run_layer(RunLayer { + execution: Some(RunExecutionLayer { + mode: Some(RunMode::DryRun), + ..RunExecutionLayer::default() + }), + ..RunLayer::default() + }); + settings.run.agent.mcps.insert( + "vaulted".to_string(), + ResolvedMcpEntry::Resolved(ResolvedMcpServerSettings { + name: "vaulted".to_string(), + transport: ResolvedMcpTransport::Stdio { + command: vec!["mcp-server".to_string()], + env: HashMap::from([( + "MCP_TOKEN".to_string(), + "{{ secrets.MCP_TOKEN }}".to_string(), + )]), + }, + ..ResolvedMcpServerSettings::default() + }), + ); + let (persisted, store) = + persisted_workflow_with_settings(MINIMAL_DOT, &storage_root, settings).await; + let emitter = Arc::new(Emitter::new(fixtures::RUN_1)); + let registry = Arc::new(test_registry()); + let vault = Arc::new(AsyncRwLock::new(token_vault("MCP_TOKEN", "staging"))); + + let session = RunSession::new(&persisted, StartServices { + vault: Some(vault), + ..test_start_services(&store, &storage_root, emitter, registry).await + }) + .await + .unwrap(); + + append_event_to_sink( + &session.event_sink, + &fixtures::RUN_1, + &Event::RunNotice { + level: fabro_types::RunNoticeLevel::Warn, + code: "mcp_ready".to_string(), + message: "MCP reported staging before stages".to_string(), + exec_output_tail: None, + }, + &session.secret_redactor, + ) + .await + .unwrap(); + + let events = session.run_store.list_events().await.unwrap(); + let serialized = serde_json::to_string(&events).unwrap(); + assert!(!serialized.contains("staging")); + assert!(serialized.contains("REDACTED")); + } + + #[tokio::test] + async fn start_redacts_declared_secret_from_setup_failure_events_and_error() { + let temp = tempfile::tempdir().unwrap(); + let (storage_root, run_dir) = storage_root_and_run_dir(&temp); + let mut settings = settings_from_run_layer(RunLayer { + execution: Some(RunExecutionLayer { + mode: Some(RunMode::DryRun), + ..RunExecutionLayer::default() + }), + ..RunLayer::default() + }); + settings.run.prepare = prepare_with_step(command_step( + &["sh", "-c", "printf %s \"$DEPLOY_ENV\" >&2; exit 7"], + HashMap::from([( + "DEPLOY_ENV".to_string(), + "{{ secrets.DEPLOY_ENV }}".to_string(), + )]), + )); + let (_persisted, store) = + persisted_workflow_with_settings(MINIMAL_DOT, &storage_root, settings).await; + let emitter = Arc::new(Emitter::new(fixtures::RUN_1)); + let registry = Arc::new(test_registry()); + let vault = Arc::new(AsyncRwLock::new(token_vault("DEPLOY_ENV", "staging"))); + + let Err(err) = start(&run_dir, StartServices { + vault: Some(vault), + ..test_start_services(&store, &storage_root, emitter, registry).await + }) + .await + else { + panic!("setup failure should fail the run"); + }; + + let error_text = err.to_string(); + assert!(!error_text.contains("staging")); + assert!(error_text.contains("REDACTED")); + + let run_store = store.open_run(&fixtures::RUN_1).await.unwrap(); + let events = run_store.list_events().await.unwrap(); + let serialized = serde_json::to_string(&events).unwrap(); + assert!( + events + .iter() + .any(|event| event.event.event_name() == "setup.failed") + ); + assert!(!serialized.contains("staging")); + assert!(serialized.contains("REDACTED")); + } + + #[tokio::test] + async fn run_session_secret_redactors_are_isolated_per_run() { + async fn session_with_secret(value: &str) -> RunSession { + let temp = tempfile::tempdir().unwrap(); + let (storage_root, _run_dir) = storage_root_and_run_dir(&temp); + let mut settings = settings_from_run_layer(RunLayer { + execution: Some(RunExecutionLayer { + mode: Some(RunMode::DryRun), + ..RunExecutionLayer::default() + }), + ..RunLayer::default() + }); + settings.run.environment.env.insert( + "DEPLOY_ENV".to_string(), + InterpString::parse("{{ secrets.DEPLOY_ENV }}"), + ); + let (persisted, store) = + persisted_workflow_with_settings(MINIMAL_DOT, &storage_root, settings).await; + let emitter = Arc::new(Emitter::new(fixtures::RUN_1)); + let registry = Arc::new(test_registry()); + let vault = Arc::new(AsyncRwLock::new(token_vault("DEPLOY_ENV", value))); + + RunSession::new(&persisted, StartServices { + vault: Some(vault), + ..test_start_services(&store, &storage_root, emitter, registry).await + }) + .await + .unwrap() + } + + let staging_session = session_with_secret("staging").await; + let production_session = session_with_secret("production").await; + + let staging_text = staging_session + .secret_redactor + .redact_into("staging production"); + let production_text = production_session + .secret_redactor + .redact_into("staging production"); + + assert!(!staging_text.contains("staging")); + assert!(staging_text.contains("production")); + assert!(production_text.contains("staging")); + assert!(!production_text.contains("production")); + } + #[tokio::test] async fn run_session_new_missing_secret_fails_startup() { let temp = tempfile::tempdir().unwrap(); @@ -1990,6 +2209,7 @@ reasoning = false &run_dir, &Error::engine("visit limit exceeded"), Duration::from_millis(9_999), + &SecretRedactor::default(), ) .await; @@ -2072,6 +2292,7 @@ reasoning = false run_store_handle, event_sink, CancellationToken::new(), + SecretRedactor::default(), ); } diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index e2b96ed37..793a444cc 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -15,6 +15,7 @@ use fabro_agent::Sandbox; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_hooks::HookSettings; use fabro_interview::AutoApproveInterviewer; +use fabro_redact::SecretRedactor; use fabro_sandbox::SandboxSpec; use fabro_store::Database; use fabro_types::settings::run::RunModelControls; @@ -251,7 +252,7 @@ async fn execute_test_run_with_options( let run_store = test_run_store(&run_id_value).await; seed_created_and_starting(&run_store, &run_options, &graph).await; let emitter = test_emitter_arc("test-run"); - let store_logger = StoreProgressLogger::new(run_store.clone()); + let store_logger = StoreProgressLogger::new(run_store.clone(), SecretRedactor::default()); store_logger.register(&emitter); let initialized = initialize( persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value), @@ -295,6 +296,7 @@ async fn execute_test_run_with_options( checkpoint: None, seed_context: None, fabro_run_tools: None, + secret_redactor: SecretRedactor::default(), }, ) .await @@ -358,6 +360,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { checkpoint: None, seed_context: None, fabro_run_tools: None, + secret_redactor: SecretRedactor::default(), }, ) .await @@ -429,6 +432,7 @@ async fn run_with_lifecycle( checkpoint: None, seed_context: None, fabro_run_tools: None, + secret_redactor: SecretRedactor::default(), }, ) .await?; diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index 6e90991dc..4d00dda40 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -1035,7 +1035,8 @@ mod tests { let inner_store = test_store().create_run(&test_run_id()).await.unwrap(); let run_store = inner_store; let emitter = Arc::new(Emitter::new(test_run_id())); - let store_logger = StoreProgressLogger::new(run_store.clone()); + let store_logger = + StoreProgressLogger::new(run_store.clone(), fabro_redact::SecretRedactor::default()); store_logger.register(&emitter); let sandbox: Arc = Arc::new(fabro_agent::LocalSandbox::new( std::env::current_dir().unwrap(), diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 54a6f1f45..636c054bd 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -559,10 +559,12 @@ pub async fn initialize( stderr: result.stderr.clone(), exec_output_tail, }); - return Err(Error::engine(format!( + let message = format!( "Setup command failed (exit code {}): {command}\n{}", exit_code, result.stderr, - ))); + ); + let message = fabro_redact::redact_string(&message); + return Err(Error::engine(options.secret_redactor.redact_into(&message))); } let exit_code = result.exit_code.unwrap_or(0); options.emitter.emit(&Event::SetupCommandCompleted { @@ -647,6 +649,7 @@ mod tests { use fabro_acp::test_support::fake_acp_agent_script; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_interview::AutoApproveInterviewer; + use fabro_redact::SecretRedactor; use fabro_sandbox::SandboxSpec; use fabro_store::Database; use fabro_types::settings::run::RunModelControls; @@ -798,6 +801,13 @@ mod tests { async fn initialize_with_setup_step( setup: crate::run_options::SetupCommand, + ) -> (crate::error::Result, Vec) { + initialize_with_setup_step_and_redactor(setup, SecretRedactor::default()).await + } + + async fn initialize_with_setup_step_and_redactor( + setup: crate::run_options::SetupCommand, + secret_redactor: SecretRedactor, ) -> (crate::error::Result, Vec) { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); @@ -812,18 +822,18 @@ mod tests { }); let result = initialize(persisted, InitOptions { - run_id: test_run_id(), - run_store: { + run_id: test_run_id(), + run_store: { let store = memory_store(); let inner = store.create_run(&test_run_id()).await.unwrap(); inner.into() }, - dry_run: false, - emitter: emitter.clone(), - sandbox: SandboxSpec::Local { + dry_run: false, + emitter: emitter.clone(), + sandbox: SandboxSpec::Local { working_directory: std::env::current_dir().unwrap(), }, - llm: LlmSpec { + llm: LlmSpec { model: "test-model".to_string(), provider_id: fabro_model::ProviderId::anthropic(), fallback_chain: Vec::new(), @@ -831,30 +841,31 @@ mod tests { model_controls: RunModelControls::default(), dry_run: true, }, - interviewer: Arc::new(AutoApproveInterviewer::engine()), - steering_hub: Arc::new(crate::steering_hub::SteeringHub::new(emitter.clone())), - catalog: test_catalog(), - lifecycle: crate::run_options::LifecycleOptions { + interviewer: Arc::new(AutoApproveInterviewer::engine()), + steering_hub: Arc::new(crate::steering_hub::SteeringHub::new(emitter.clone())), + catalog: test_catalog(), + lifecycle: crate::run_options::LifecycleOptions { setup_commands: vec![setup], setup_command_timeout_ms: 1_000, }, - run_options: test_settings(&run_dir), - workflow_path: None, - workflow_bundle: None, - hooks: fabro_hooks::HookSettings { hooks: vec![] }, - sandbox_env: SandboxEnvSpec { + run_options: test_settings(&run_dir), + workflow_path: None, + workflow_bundle: None, + hooks: fabro_hooks::HookSettings { hooks: vec![] }, + sandbox_env: SandboxEnvSpec { toml_env: HashMap::new(), github_permissions: None, origin_url: None, }, - vault: None, - git: None, - run_control: None, + vault: None, + git: None, + run_control: None, registry_override: None, - artifact_sink: None, - checkpoint: None, - seed_context: None, - fabro_run_tools: None, + artifact_sink: None, + checkpoint: None, + seed_context: None, + fabro_run_tools: None, + secret_redactor, }) .await; let events = seen.lock().unwrap().clone(); @@ -937,6 +948,7 @@ mod tests { checkpoint: None, seed_context: None, fabro_run_tools: None, + secret_redactor: SecretRedactor::default(), }) .await .unwrap(); @@ -1135,6 +1147,7 @@ mod tests { checkpoint: None, seed_context: None, fabro_run_tools: None, + secret_redactor: SecretRedactor::default(), }) .await .unwrap(); @@ -1183,7 +1196,7 @@ mod tests { let emitter = Arc::new(crate::event::Emitter::new(test_run_id())); let store = memory_store(); let run_store = store.create_run(&test_run_id()).await.unwrap(); - let store_logger = StoreProgressLogger::new(run_store.clone()); + let store_logger = StoreProgressLogger::new(run_store.clone(), SecretRedactor::default()); let seen = Arc::new(std::sync::Mutex::new(Vec::new())); emitter.on_event({ let seen = Arc::clone(&seen); @@ -1231,6 +1244,7 @@ mod tests { checkpoint: None, seed_context: None, fabro_run_tools: None, + secret_redactor: SecretRedactor::default(), }) .await .unwrap(); @@ -1317,6 +1331,40 @@ mod tests { } } + #[tokio::test] + async fn initialize_setup_failure_redacts_registered_secret_in_error_text() { + let redactor = SecretRedactor::default(); + redactor.register("staging"); + let setup = crate::run_options::SetupCommand { + command: "printf %s \"$DEPLOY_ENV\" >&2; exit 7".to_string(), + env: HashMap::from([("DEPLOY_ENV".to_string(), "staging".to_string())]), + }; + + let (result, _events) = initialize_with_setup_step_and_redactor(setup, redactor).await; + + let Err(err) = result else { + panic!("setup should fail"); + }; + let text = err.to_string(); + assert!(!text.contains("staging")); + assert!(text.contains("REDACTED")); + } + + #[tokio::test] + async fn initialize_setup_failure_keeps_content_redaction_with_empty_registry() { + let secret = "sk-ant-api03-xK9mZ2vL8nQ5rT1wY4bC7dF0gH3jE6pA"; + let command = format!("printf %s {secret} >&2; exit 7"); + + let (result, _events) = initialize_with_setup_command(&command).await; + + let Err(err) = result else { + panic!("setup should fail"); + }; + let text = err.to_string(); + assert!(!text.contains(secret)); + assert!(text.contains("REDACTED")); + } + #[tokio::test] async fn initialize_cancelled_setup_command_returns_cancelled() { let temp = tempfile::tempdir().unwrap(); @@ -1374,6 +1422,7 @@ mod tests { checkpoint: None, seed_context: None, fabro_run_tools: None, + secret_redactor: SecretRedactor::default(), }) .await; diff --git a/lib/crates/fabro-workflow/src/pipeline/types.rs b/lib/crates/fabro-workflow/src/pipeline/types.rs index 8449453a8..4f8e82fc6 100644 --- a/lib/crates/fabro-workflow/src/pipeline/types.rs +++ b/lib/crates/fabro-workflow/src/pipeline/types.rs @@ -6,6 +6,7 @@ use fabro_graphviz::graph::Graph; use fabro_interview::Interviewer; use fabro_mcp::config::McpServerSettings; use fabro_model::{Catalog, FallbackTarget, ProviderId}; +use fabro_redact::SecretRedactor; use fabro_sandbox::SandboxSpec; use fabro_template::TemplateContext; use fabro_types::settings::run::{PullRequestSettings, RunModelControls}; @@ -272,6 +273,7 @@ pub struct InitOptions { pub checkpoint: Option, pub seed_context: Option, pub fabro_run_tools: Option, + pub secret_redactor: SecretRedactor, } /// Output of the INITIALIZE phase. diff --git a/lib/crates/fabro-workflow/src/runtime_store.rs b/lib/crates/fabro-workflow/src/runtime_store.rs index 761a8c4fd..82bb8dba3 100644 --- a/lib/crates/fabro-workflow/src/runtime_store.rs +++ b/lib/crates/fabro-workflow/src/runtime_store.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use anyhow::Result; use async_trait::async_trait; use bytes::Bytes; +use fabro_redact::SecretRedactor; use fabro_store::{EventEnvelope, RunDatabase, RunProjection}; use fabro_types::{RunBlobId, RunEvent}; @@ -83,7 +84,11 @@ impl RunStoreBackend for LocalRunStoreBackend { } async fn append_run_event(&self, event: &RunEvent) -> Result<()> { - let payload = build_redacted_event_payload(event, &event.run_id)?; + // Local backends can be called from server/test paths that do not have a + // worker run registry. RunEventSink applies the per-run redactor before + // calling this backend during workflow execution. + let payload = + build_redacted_event_payload(event, &event.run_id, &SecretRedactor::default())?; self.run_store .append_event(&payload) .await diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index 1acf410a4..ba3a731d1 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -201,7 +201,8 @@ async fn initialized( .await .expect("failed to seed run.starting event in run store"); let emitter = bound_emitter(run_options.run_id, &emitter); - let store_logger = StoreProgressLogger::new(run_store.clone()); + let store_logger = + StoreProgressLogger::new(run_store.clone(), fabro_redact::SecretRedactor::default()); store_logger.register(emitter.as_ref()); let artifact_store = ArtifactStore::new( Arc::new(