From f4dc6ddf3d02a3281392471c2d033a057bc9ceae Mon Sep 17 00:00:00 2001 From: David Grochowski Date: Fri, 6 Nov 2026 21:26:29 -0500 Subject: [PATCH] fix(relay): apply ingest's content checks to workflow SendMessage Problem: ingest refuses event content over 256 KB and content or tags containing a NUL byte (Postgres `text` cannot store `\0`). The workflow sink's SendMessage builds and inserts a relay-signed kind:9 directly, so it skipped both checks. Its text is rendered from templates that can include a `call_webhook` response body (up to 1 MiB, decoded with `String::from_utf8_lossy`, which keeps `\0`). How it fails: a webhook body between 256 KB and 1 MiB is stored as a message ingest would have refused; a body with a NUL byte fails at insert and the run step fails with `ActionSinkError::Database` instead of naming the bad input. Change: the size limit and the content NUL check become shared `pub(crate)` primitives in ingest (`MAX_EVENT_CONTENT_BYTES`, `check_content_size`, `check_content_nul`, with a `StoredContentError` whose Display omits the `invalid:` prefix). `ingest_event_inner` keeps its ordering: the size check first, the NUL check (content and tags) after auth, scope and the write restriction, before command routing. Its rejection messages are unchanged. The sink runs both checks on the rendered text before any database work and maps a failure to `ActionSinkError::InvalidInput`. Tags in the sink are built from parsed or stored values (UUID, pubkeys, resolved event ids), so the text is the only value that needs checking there. Why safe: ingest behaviour and messages are unchanged; the sink only refuses text that ingest would refuse or that could not be stored. Tests: `send_message_refuses_unstorable_text_before_touching_the_database` drives `RelayActionSink::send_message` against an unreachable database. Before the fix NUL text returned `Database("... pool timed out ...")`; after it NUL and over-limit text return `InvalidInput`, and text at the limit still reaches the database. Signed-off-by: David Grochowski --- crates/buzz-relay/src/handlers/ingest.rs | 60 ++++++++++++++++------ crates/buzz-relay/src/workflow_sink.rs | 63 ++++++++++++++++++++++++ 2 files changed, 107 insertions(+), 16 deletions(-) diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index e4b86e16828..f8ce2fcb448 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -2523,27 +2523,62 @@ pub async fn ingest_event( /// client ceiling (60 s, CANVAS_MAX_FUTURE_SKEW_SECS) < canvas relay bound (300 s) < relay general bound (900 s) const CANVAS_MAX_INGEST_FUTURE_SECS: i64 = 300; +/// Largest event `content`, in bytes, the relay will store. +pub(crate) const MAX_EVENT_CONTENT_BYTES: usize = 256 * 1024; // 256 KB + +/// Why event content cannot be stored. `Display` omits the `invalid:` prefix +/// so each write path can map it into its own error type. +#[derive(Debug, PartialEq, Eq, thiserror::Error)] +pub(crate) enum StoredContentError { + /// Content is longer than [`MAX_EVENT_CONTENT_BYTES`]. + #[error("content exceeds maximum size of {MAX_EVENT_CONTENT_BYTES} bytes (got {0})")] + TooLarge(usize), + /// Content contains `\0`, which Postgres `text` cannot store. + #[error("content contains a NUL byte")] + NulInContent, + /// A tag value contains `\0`, which Postgres `text` cannot store. + #[error("tag contains a NUL byte")] + NulInTag, +} + +/// Refuse content longer than [`MAX_EVENT_CONTENT_BYTES`]. Pure and cheap, so +/// callers run it before any database read or signature check. +pub(crate) fn check_content_size(content: &str) -> Result<(), StoredContentError> { + if content.len() > MAX_EVENT_CONTENT_BYTES { + return Err(StoredContentError::TooLarge(content.len())); + } + Ok(()) +} + +/// Refuse content containing a NUL byte. Postgres `text` cannot store `\0`, +/// so such content would otherwise fail at insert with an encoding error. +pub(crate) fn check_content_nul(content: &str) -> Result<(), StoredContentError> { + if content.contains('\0') { + return Err(StoredContentError::NulInContent); + } + Ok(()) +} + /// Reject events whose content or any tag value contains a NUL byte. /// Postgres `text` cannot store `\0`, so such an event would otherwise fail /// at insert with an encoding error and surface as an internal error. fn reject_nul_bytes(event: &Event) -> Result<(), IngestError> { - if event.content.contains('\0') { - return Err(IngestError::Rejected( - "invalid: content contains a NUL byte".into(), - )); - } + check_content_nul(&event.content).map_err(stored_content_rejection)?; if event .tags .iter() .any(|tag| tag.as_slice().iter().any(|value| value.contains('\0'))) { - return Err(IngestError::Rejected( - "invalid: tag contains a NUL byte".into(), - )); + return Err(stored_content_rejection(StoredContentError::NulInTag)); } Ok(()) } +/// Map a stored-content refusal to the client-facing `invalid:` rejection. +fn stored_content_rejection(error: StoredContentError) -> IngestError { + IngestError::Rejected(format!("invalid: {error}")) +} + /// Returns `Ok(())` if the canvas event timestamp is within the allowed future /// window, or `Err` with a rejection message otherwise. /// @@ -2605,14 +2640,7 @@ async fn ingest_event_inner( // The size check is a pure, cheap reject: it needs neither the DB nor // signature verification, so oversized events never cost a writer-pool // read or a blocking-pool Schnorr check. - const MAX_EVENT_CONTENT_BYTES: usize = 256 * 1024; // 256 KB - if event.content.len() > MAX_EVENT_CONTENT_BYTES { - return Err(IngestError::Rejected(format!( - "invalid: content exceeds maximum size of {} bytes (got {})", - MAX_EVENT_CONTENT_BYTES, - event.content.len() - ))); - } + check_content_size(&event.content).map_err(stored_content_rejection)?; // Durable community write fence: persistent ingest is a DB write the // deletion engine cannot exclude via serving-write leases (those cover diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 905d1c7b080..d8d0cf8ebd4 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -222,6 +222,16 @@ impl ActionSink for RelayActionSink { .upgrade() .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + // The rendered text is the only caller-supplied value stored + // unparsed (it can carry webhook output), so it gets ingest's + // stored-content checks before any database work; every tag below + // is built from parsed or stored values. Without them an oversized + // body is stored past ingest's limit and a NUL byte fails the + // insert as a database error. + crate::handlers::ingest::check_content_size(&text) + .and_then(|()| crate::handlers::ingest::check_content_nul(&text)) + .map_err(|e| ActionSinkError::InvalidInput(e.to_string()))?; + // The run carries its owning community (`community_id`); the // relay-signed kind:9 message belongs to *that* community, never the // deployment default. Re-deriving the tenant from `config.relay_url` @@ -773,6 +783,59 @@ mod tests { assert_eq!(tags.len(), 1); assert_eq!(tags[0].as_slice(), ["p", owner.as_str(), "", "owner"]); } + + /// Rendered workflow text can carry webhook output (up to 1 MiB, NUL + /// bytes kept by lossy UTF-8 decoding). The sink must refuse what ingest + /// refuses — oversized or NUL-bearing content — as invalid input before + /// it reads or writes the database. The database here is unreachable, so + /// any text that gets past the checks fails as `Database` instead. + /// Mutation: drop either check in `send_message` and its case fails. + #[tokio::test] + async fn send_message_refuses_unstorable_text_before_touching_the_database() { + let state = crate::state::tests::test_state_with_unreachable_database_url( + "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", // sadscan:disable np.postgres.1 + ) + .await; + let sink = RelayActionSink::new(&state); + let community = CommunityId::from_uuid(Uuid::new_v4()); + let channel = Uuid::new_v4().to_string(); + let owner = pk('1'); + let max = crate::handlers::ingest::MAX_EVENT_CONTENT_BYTES; + + let send = |text: String| { + let sink = &sink; + let channel = channel.clone(); + let owner = owner.clone(); + async move { + sink.send_message(community, &channel, &text, "{{text}}", &owner, None) + .await + } + }; + + for (text, expected) in [ + ( + "a\0b".to_string(), + "content contains a NUL byte".to_string(), + ), + ( + "x".repeat(max + 1), + format!("content exceeds maximum size of {max} bytes"), + ), + ] { + match send(text).await { + Err(ActionSinkError::InvalidInput(msg)) => { + assert!(msg.contains(&expected), "{expected}: {msg}"); + } + other => panic!("{expected}: must be refused as invalid input, got {other:?}"), + } + } + + // Text at the limit passes the checks and reaches the database. + match send("x".repeat(max)).await { + Err(ActionSinkError::Database(_)) => {} + other => panic!("text at the limit must reach the database, got {other:?}"), + } + } } #[cfg(test)]