From f55868e7b833db9a107c4b7fa42ca47cf3df0223 Mon Sep 17 00:00:00 2001 From: David Grochowski Date: Sat, 7 Nov 2026 10:56:10 -0500 Subject: [PATCH] fix(workflow): replace NUL in webhook output and trigger fields PostgreSQL jsonb rejects \u0000 ("unsupported Unicode escape sequence"). call_webhook decoded the response body with String::from_utf8_lossy, which keeps NUL, and the body lands in the run's execution_trace. The final status update then failed ("Failed to update run to Failed: ...") and the run stayed `running` forever. Webhook-trigger and command payload fields parsed from JSON ("\u0000" escapes) hit the same wall in workflow_runs.trigger_context, failing run creation with a 500. Replace U+0000 with U+FFFD at decode, the same substitution from_utf8_lossy already makes for invalid UTF-8. Refusing the step instead would fail a workflow over bytes a remote server chose to send, and the body is already a lossy, best-effort rendering; keeping the run going with a visible replacement character is consistent with that. Trigger fields from the webhook endpoint and the workflow command handler shared a duplicated flatten loop; both now go through TriggerContext::extend_webhook_fields, which applies the same replacement to keys and values. Signed-off-by: David Grochowski --- crates/buzz-relay/src/api/bridge.rs | 10 +-- .../bridge/workflow_webhook_postgres_tests.rs | 69 +++++++++++++- .../src/handlers/command_executor.rs | 8 +- crates/buzz-workflow/src/executor.rs | 90 +++++++++++++++++-- crates/buzz-workflow/src/lib.rs | 76 ++++++++++++++++ 5 files changed, 231 insertions(+), 22 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 0b6c481a887..cb0931c767c 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2949,14 +2949,8 @@ pub async fn workflow_webhook( .unwrap_or_default(), ..Default::default() }; - if let Some(Value::Object(ref map)) = body_json { - for (k, v) in map { - let val_str = match v { - Value::String(s) => s.clone(), - other => other.to_string(), - }; - trigger_ctx.webhook_fields.insert(k.clone(), val_str); - } + if let Some(Value::Object(map)) = body_json { + trigger_ctx.extend_webhook_fields(map); } let trigger_ctx_json = serde_json::to_value(&trigger_ctx).ok(); diff --git a/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs b/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs index ad22536291e..20a7c00a172 100644 --- a/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs +++ b/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs @@ -133,6 +133,16 @@ async fn post_webhook( workflow_id: Uuid, host: &str, secret: &str, +) -> (StatusCode, Value) { + post_webhook_with_body(state, workflow_id, host, secret, Bytes::new()).await +} + +async fn post_webhook_with_body( + state: &Arc, + workflow_id: Uuid, + host: &str, + secret: &str, + body: Bytes, ) -> (StatusCode, Value) { let mut headers = HeaderMap::new(); headers.insert( @@ -149,7 +159,7 @@ async fn post_webhook( Path(workflow_id.to_string()), Query(WebhookQuery { secret: None }), headers, - Bytes::new(), + body, ) .await { @@ -355,3 +365,60 @@ async fn webhook_admission_enforces_owner_ban_tenant_scope_and_lookup_failure() .await .expect("drop restriction lookup failure schema"); } + +/// PostgreSQL `jsonb` rejects `\u0000`; a webhook body field holding an +/// escaped NUL must still create the run instead of failing the insert. +#[tokio::test] +#[ignore = "requires PostgreSQL and Redis"] +async fn webhook_body_field_with_nul_creates_run() { + let pool = PgPool::connect(&database_url()) + .await + .expect("connect to isolated PostgreSQL test database"); + let db = buzz_db::Db::from_pool(pool.clone()); + + let community_id = Uuid::new_v4(); + let host = format!("webhook-nul-{}.test", Uuid::new_v4().simple()); + let owner_pubkey = [0x44; 32]; + let workflow_id = Uuid::new_v4(); + let secret = format!("webhook-secret-{}", Uuid::new_v4().simple()); + seed_community(&pool, community_id, &host).await; + seed_webhook_workflow( + &pool, + WebhookWorkflowSeed { + community_id, + workflow_id, + channel_id: Uuid::new_v4(), + owner_pubkey: &owner_pubkey, + owner_role: "owner", + name: "nul-field-workflow", + secret: &secret, + steps: json!([{"id": "hold", "action": "delay", "duration": "1s"}]), + }, + ) + .await; + + let (state, audit_shutdown) = app_state(db, &pool).await; + let (status, body) = post_webhook_with_body( + &state, + workflow_id, + &host, + &secret, + Bytes::from_static(br#"{"note": "a\u0000b"}"#), + ) + .await; + assert_eq!(status, StatusCode::ACCEPTED, "response: {body}"); + + let trigger_context: Value = sqlx::query_scalar( + "SELECT trigger_context FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2", + ) + .bind(community_id) + .bind(workflow_id) + .fetch_one(&pool) + .await + .expect("load run trigger context"); + assert_eq!(trigger_context["webhook_fields"]["note"], "a\u{FFFD}b"); + + audit_shutdown + .drain(std::time::Duration::from_secs(1)) + .await; +} diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 73d42985178..bd4f1e35a14 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -914,13 +914,7 @@ async fn handle_workflow_trigger( }; if !event.content.is_empty() { if let Ok(serde_json::Value::Object(map)) = serde_json::from_str(&event.content) { - for (k, v) in map { - let val_str = match v { - serde_json::Value::String(s) => s, - other => other.to_string(), - }; - trigger_ctx.webhook_fields.insert(k, val_str); - } + trigger_ctx.extend_webhook_fields(map); } } let trigger_ctx_json = serde_json::to_value(&trigger_ctx).ok(); diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 0fe33033bb4..07a32b57df9 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -61,6 +61,44 @@ impl TriggerContext { other => self.webhook_fields.get(other).map(|s| s.as_str()), } } + + /// Flatten a JSON object (webhook body, command payload) into + /// `webhook_fields`: strings verbatim, other values as compact JSON. + /// + /// U+0000 in keys and values becomes U+FFFD: the context is persisted as + /// `workflow_runs.trigger_context`, and PostgreSQL `jsonb` rejects `\u0000`. + pub fn extend_webhook_fields(&mut self, map: serde_json::Map) { + for (key, value) in map { + let value = match value { + JsonValue::String(s) => s, + other => other.to_string(), + }; + self.webhook_fields.insert(storable(key), storable(value)); + } + } +} + +/// Replace U+0000 with U+FFFD so an untrusted string can be persisted in run +/// state: PostgreSQL `jsonb` rejects `\u0000` ("unsupported Unicode escape +/// sequence"), which would fail the run's status update and strand it. +/// U+FFFD matches how `String::from_utf8_lossy` already treats invalid UTF-8. +fn storable(s: String) -> String { + if s.contains('\0') { + s.replace('\0', "\u{FFFD}") + } else { + s + } +} + +/// Step output of `call_webhook`: the status and the body decoded lossily +/// into a [`storable`] string, because the output lands in the run's +/// `execution_trace`. +#[cfg(any(feature = "reqwest", test))] +pub(crate) fn webhook_output(status: u16, body: &[u8]) -> JsonValue { + serde_json::json!({ + "status": status, + "body": storable(String::from_utf8_lossy(body).into_owned()), + }) } /// Resolve `{{trigger.X}}` and `{{steps.ID.output.X}}` placeholders in a string. @@ -953,12 +991,7 @@ async fn call_webhook_impl( } } - let body_text = String::from_utf8_lossy(&body_bytes).into_owned(); - - Ok(serde_json::json!({ - "status": status, - "body": body_text, - })) + Ok(webhook_output(status, &body_bytes)) } /// Returns a shared `reqwest::Client` reused across all workflow HTTP calls. @@ -1983,4 +2016,49 @@ mod tests { .expect("override should be accepted"); assert_eq!(resolved, override_channel_id.to_string()); } + + /// True if any string in `value` holds U+0000, which PostgreSQL `jsonb` + /// rejects as `\u0000`. + fn contains_nul(value: &JsonValue) -> bool { + match value { + JsonValue::String(s) => s.contains('\0'), + JsonValue::Array(items) => items.iter().any(contains_nul), + JsonValue::Object(map) => map.iter().any(|(k, v)| k.contains('\0') || contains_nul(v)), + _ => false, + } + } + + #[test] + fn webhook_output_replaces_nul_like_invalid_utf8() { + let trace = json!([{ + "step_id": "hook", + "status": "completed", + "output": webhook_output(200, b"ok\0tail\xff"), + }]); + assert!( + !contains_nul(&trace), + "trace is not jsonb-storable: {trace}" + ); + assert_eq!(trace[0]["output"]["body"], "ok\u{FFFD}tail\u{FFFD}"); + assert_eq!(trace[0]["output"]["status"], 200); + } + + #[test] + fn extend_webhook_fields_replaces_nul_in_keys_and_values() { + let body: JsonValue = serde_json::from_str( + r#"{"note": "a\u0000b", "k\u0000": 1, "nested": {"x": "\u0000"}}"#, + ) + .unwrap(); + let mut ctx = TriggerContext::default(); + ctx.extend_webhook_fields(body.as_object().cloned().unwrap()); + let stored = serde_json::to_value(&ctx).unwrap(); + assert!( + !contains_nul(&stored), + "context is not jsonb-storable: {stored}" + ); + assert_eq!(ctx.get_field("note"), Some("a\u{FFFD}b")); + assert_eq!(ctx.get_field("k\u{FFFD}"), Some("1")); + // Non-string values are stored as JSON text, where NUL is escaped. + assert_eq!(ctx.get_field("nested"), Some(r#"{"x":"\u0000"}"#)); + } } diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 9dabc5732a7..a2c9c783511 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -2618,4 +2618,80 @@ steps: run.error_message ); } + + /// PostgreSQL `jsonb` rejects `\u0000`. A `call_webhook` body carrying + /// U+0000 must still persist in the trace, or the final status update + /// fails and the run stays `running`. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn webhook_output_with_nul_completes_run() { + let db = setup_db().await; + let creator = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let member = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let (community, channel_id) = setup_channel(&db, &creator, &member).await; + let def = serde_json::json!({ + "name": "nul-webhook", + "trigger": {"on": "webhook"}, + "steps": [{"id": "hook", "action": "call_webhook", "url": "https://example.com/x"}], + "enabled": true, + }) + .to_string(); + let workflow_id = db + .create_workflow( + community, + Some(channel_id), + &creator, + "nul-webhook", + &def, + &[7u8; 32], + ) + .await + .expect("create workflow"); + + let run_id = db + .create_workflow_run(community, workflow_id, None, None) + .await + .expect("create run"); + db.update_workflow_run( + community, + run_id, + RunStatus::Running, + 0, + &serde_json::json!([]), + None, + ) + .await + .expect("mark run running"); + + let engine = WorkflowEngine::new(db.clone(), WorkflowConfig::default()); + let trace = vec![serde_json::json!({ + "step_id": "hook", + "status": "completed", + "output": executor::webhook_output(200, b"ok\0tail"), + })]; + engine + .finalize_run( + community, + run_id, + Ok(ExecutionResult { + approval_token: None, + step_index: 1, + step_outputs: HashMap::new(), + trace, + }), + None, + ) + .await; + + let run = db + .get_workflow_run(community, run_id) + .await + .expect("get run"); + assert_eq!( + run.status, + RunStatus::Completed, + "a NUL in the webhook body must not strand the run" + ); + assert_eq!(run.execution_trace[0]["output"]["body"], "ok\u{FFFD}tail"); + } }