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)]