From 557c6a355b0dc7af88043179ddd0cbcdbbe69647 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 26 Aug 2026 12:04:20 -0400 Subject: [PATCH] feat(workflows): sign delivery verification receipts Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-core/src/kind.rs | 6 + crates/buzz-core/src/workflow_delivery.rs | 261 +++++++++++++++++- crates/buzz-db/src/lib.rs | 67 +++++ crates/buzz-db/src/migration.rs | 15 +- crates/buzz-db/src/workflow.rs | 98 +++++++ crates/buzz-relay/src/api/bridge.rs | 21 ++ crates/buzz-relay/src/api/workflows.rs | 80 +++++- .../0036_workflow_webhook_invocations.sql | 20 ++ schema/schema.sql | 19 ++ 9 files changed, 583 insertions(+), 4 deletions(-) create mode 100644 migrations/0036_workflow_webhook_invocations.sql diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 2a5145ba2a..35c5c2a906 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -440,6 +440,11 @@ pub const KIND_WINDOW_BOUNDS: u32 = 39006; /// Ephemeral, relay-authored hint that a durable workflow delivery is ready. pub const KIND_WORKFLOW_AGENT_WAKE: u32 = 24620; +/// Relay-signed proof binding a durable workflow delivery to its public inputs. +/// +/// Returned by the authenticated delivery API; not accepted as client-authored +/// state and not stored in the event log. +pub const KIND_WORKFLOW_DELIVERY_RECEIPT: u32 = 24621; /// Workflow definition (parameterized replaceable, d=workflow_uuid). pub const KIND_WORKFLOW_DEF: u32 = 30620; @@ -696,6 +701,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_THREAD_SUMMARY, KIND_WINDOW_BOUNDS, KIND_WORKFLOW_AGENT_WAKE, + KIND_WORKFLOW_DELIVERY_RECEIPT, KIND_PRESENCE_UPDATE, KIND_TYPING_INDICATOR, KIND_HUDDLE_REACTION, diff --git a/crates/buzz-core/src/workflow_delivery.rs b/crates/buzz-core/src/workflow_delivery.rs index 3339038e9d..8e061af8c0 100644 --- a/crates/buzz-core/src/workflow_delivery.rs +++ b/crates/buzz-core/src/workflow_delivery.rs @@ -6,11 +6,15 @@ use std::fmt; -use nostr::{Event, EventBuilder, EventId, Kind, PublicKey, Tag}; +use nostr::{Event, EventBuilder, EventId, Keys, Kind, PublicKey, Tag}; +use sha2::{Digest, Sha256}; use thiserror::Error; use uuid::Uuid; -use crate::{kind::KIND_WORKFLOW_AGENT_WAKE, tenant::CommunityId}; +use crate::{ + kind::{KIND_WORKFLOW_AGENT_WAKE, KIND_WORKFLOW_DELIVERY_RECEIPT}, + tenant::CommunityId, +}; /// The target class admitted by the durable workflow delivery protocol. pub const WORKFLOW_DELIVERY_TARGET: &str = "message-v1"; @@ -201,6 +205,135 @@ pub fn message_v1_targets(event: &Event) -> Result, WorkflowDeliv Ok(targets) } +/// Relay-signed proof that the relay produced one public workflow delivery. +/// +/// The receipt contains references and hashes only. Trigger context, webhook +/// fields, step outputs, execution traces, and secrets are deliberately absent. +/// Its timestamp is copied from the visible message, making signing +/// deterministic for one binding/message/key tuple. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowDeliveryReceipt { + delivery_id: WorkflowDeliveryId, + binding: WorkflowDeliveryBinding, + message_content_sha256: [u8; 32], +} + +impl WorkflowDeliveryReceipt { + /// Construct a receipt after validating the visible event against the binding. + pub fn new( + delivery_id: WorkflowDeliveryId, + binding: WorkflowDeliveryBinding, + message: &Event, + ) -> Result { + binding.validate_message_event(message)?; + Ok(Self { + delivery_id, + binding, + message_content_sha256: Sha256::digest(message.content.as_bytes()).into(), + }) + } + + /// Return the durable delivery identity. + pub const fn delivery_id(&self) -> WorkflowDeliveryId { + self.delivery_id + } + + /// Return the immutable delivery binding proved by this receipt. + pub const fn binding(&self) -> &WorkflowDeliveryBinding { + &self.binding + } + + /// Return the SHA-256 identity of the visible message content. + pub const fn message_content_sha256(&self) -> [u8; 32] { + self.message_content_sha256 + } + + /// Sign the canonical receipt with the relay identity. + pub fn sign(&self, relay_keys: &Keys, message: &Event) -> Result { + self.binding.validate_message_event(message)?; + if <[u8; 32]>::from(Sha256::digest(message.content.as_bytes())) + != self.message_content_sha256 + { + return Err(WorkflowDeliveryError::MessageContentHashMismatch); + } + EventBuilder::new(Kind::Custom(KIND_WORKFLOW_DELIVERY_RECEIPT as u16), "") + .tags(self.canonical_tags()?) + .custom_created_at(message.created_at) + .sign_with_keys(relay_keys) + .map_err(|error| WorkflowDeliveryError::ReceiptSigning(error.to_string())) + } + + /// Verify a receipt's relay signature, canonical shape, binding, and public message. + pub fn verify( + event: &Event, + relay_pubkey: PublicKey, + delivery_id: WorkflowDeliveryId, + binding: &WorkflowDeliveryBinding, + message: &Event, + ) -> Result { + if event.kind.as_u16() != KIND_WORKFLOW_DELIVERY_RECEIPT as u16 { + return Err(WorkflowDeliveryError::WrongReceiptKind(event.kind.as_u16())); + } + if event.pubkey != relay_pubkey || event.verify().is_err() { + return Err(WorkflowDeliveryError::InvalidReceiptSignature); + } + if !event.content.is_empty() { + return Err(WorkflowDeliveryError::ReceiptHasContent); + } + binding.validate_message_event(message)?; + if event.created_at != message.created_at { + return Err(WorkflowDeliveryError::NonCanonicalReceipt); + } + let expected = Self::new(delivery_id, binding.clone(), message)?; + let expected_tags = expected.canonical_tags()?; + if event.tags.as_slice() != expected_tags.as_slice() { + return Err(WorkflowDeliveryError::ReceiptBindingMismatch); + } + Ok(expected) + } + + fn canonical_tags(&self) -> Result, WorkflowDeliveryError> { + let mut tags = vec![ + parse_tag(["delivery", &self.delivery_id.to_string()])?, + parse_tag(["community", &self.binding.community_id().to_string()])?, + parse_tag(["workflow", &self.binding.workflow_id().to_string()])?, + parse_tag(["run", &self.binding.run_id().to_string()])?, + parse_tag(["step", self.binding.step_id()])?, + parse_tag(["p", &self.binding.target_pubkey().to_hex()])?, + parse_tag(["definition", &self.binding.definition_event_id().to_hex()])?, + parse_tag([ + "message", + &self.binding.message_event_id().to_hex(), + &sha256_hex(&self.message_content_sha256), + ])?, + ]; + tags.push(cause_tag(self.binding.cause())?); + Ok(tags) + } +} + +fn sha256_hex(digest: &[u8; 32]) -> String { + use std::fmt::Write as _; + + let mut encoded = String::with_capacity(64); + for byte in digest { + let _ = write!(encoded, "{byte:02x}"); + } + encoded +} + +fn cause_tag(cause: &WorkflowDeliveryCause) -> Result { + match cause { + WorkflowDeliveryCause::Event(event_id) => parse_tag(["cause", "event", &event_id.to_hex()]), + WorkflowDeliveryCause::Schedule { + scheduled_for_unix_seconds, + } => parse_tag(["cause", "schedule", &scheduled_for_unix_seconds.to_string()]), + WorkflowDeliveryCause::Webhook { invocation_id } => { + parse_tag(["cause", "webhook", &invocation_id.to_string()]) + } + } +} + /// An ephemeral identifier-only wake hint for a durable delivery. #[derive(Debug, Clone, PartialEq, Eq)] pub struct WorkflowDeliveryWake { @@ -300,6 +433,27 @@ pub enum WorkflowDeliveryError { /// A `message-v1` recipient appeared more than once. #[error("workflow delivery message has duplicate message-v1 target")] DuplicateMessageV1Target, + /// A receipt was not kind 24621. + #[error("workflow delivery receipt must be kind 24621, got {0}")] + WrongReceiptKind(u16), + /// A receipt was not signed by the expected relay or had an invalid signature. + #[error("workflow delivery receipt signature is invalid")] + InvalidReceiptSignature, + /// A receipt carried content rather than references only. + #[error("workflow delivery receipt content must be empty")] + ReceiptHasContent, + /// A receipt did not use the message timestamp. + #[error("workflow delivery receipt is not canonical")] + NonCanonicalReceipt, + /// A receipt's exact canonical tags disagreed with the expected binding. + #[error("workflow delivery receipt binding does not match")] + ReceiptBindingMismatch, + /// The visible message content did not match the receipt hash. + #[error("workflow delivery receipt message content hash does not match")] + MessageContentHashMismatch, + /// Nostr could not sign the receipt. + #[error("workflow delivery receipt signing failed: {0}")] + ReceiptSigning(String), /// A wake was not kind 24620. #[error("workflow delivery wake must be kind 24620, got {0}")] WrongWakeKind(u16), @@ -643,6 +797,109 @@ mod tests { ); } + #[test] + fn receipt_binds_every_identity_and_contains_no_private_inputs() { + let relay = Keys::generate(); + let original = binding(); + let message = EventBuilder::new(Kind::Custom(9), "rendered secret result") + .tags([original.message_v1_target_tag().unwrap()]) + .sign_with_keys(&relay) + .unwrap(); + let binding = WorkflowDeliveryBinding::new( + original.community_id(), + original.workflow_id(), + original.run_id(), + original.step_id(), + original.target_pubkey(), + original.definition_event_id(), + message.id, + original.cause().clone(), + ) + .unwrap(); + let delivery_id = WorkflowDeliveryId::from_uuid(Uuid::new_v4()); + let receipt = WorkflowDeliveryReceipt::new(delivery_id, binding.clone(), &message).unwrap(); + let signed = receipt.sign(&relay, &message).unwrap(); + + assert!(signed.content.is_empty()); + let wire = serde_json::to_string(&signed).unwrap(); + assert!(!wire.contains("rendered secret result")); + assert!(!wire.contains("trigger_context")); + assert!(!wire.contains("execution_trace")); + assert_eq!( + WorkflowDeliveryReceipt::verify( + &signed, + relay.public_key(), + delivery_id, + &binding, + &message, + ) + .unwrap(), + receipt + ); + + let other_target = Keys::generate().public_key(); + let tampered = WorkflowDeliveryBinding::new( + binding.community_id(), + binding.workflow_id(), + binding.run_id(), + binding.step_id(), + other_target, + binding.definition_event_id(), + binding.message_event_id(), + binding.cause().clone(), + ) + .unwrap(); + assert_eq!( + WorkflowDeliveryReceipt::verify( + &signed, + relay.public_key(), + delivery_id, + &tampered, + &message, + ), + Err(WorkflowDeliveryError::MissingMessageV1Target) + ); + assert_eq!( + WorkflowDeliveryReceipt::verify( + &signed, + Keys::generate().public_key(), + delivery_id, + &binding, + &message, + ), + Err(WorkflowDeliveryError::InvalidReceiptSignature) + ); + } + + #[test] + fn receipt_rejects_message_content_and_signature_tampering() { + let relay = Keys::generate(); + let original = binding(); + let message = EventBuilder::new(Kind::Custom(9), "visible") + .tags([original.message_v1_target_tag().unwrap()]) + .sign_with_keys(&relay) + .unwrap(); + let binding = WorkflowDeliveryBinding::new( + original.community_id(), + original.workflow_id(), + original.run_id(), + original.step_id(), + original.target_pubkey(), + original.definition_event_id(), + message.id, + original.cause().clone(), + ) + .unwrap(); + let id = WorkflowDeliveryId::from_uuid(Uuid::new_v4()); + let receipt = WorkflowDeliveryReceipt::new(id, binding.clone(), &message).unwrap(); + let mut signed = receipt.sign(&relay, &message).unwrap(); + signed.content = "private input".to_owned(); + assert_eq!( + WorkflowDeliveryReceipt::verify(&signed, relay.public_key(), id, &binding, &message), + Err(WorkflowDeliveryError::InvalidReceiptSignature) + ); + } + #[test] fn wake_round_trips_and_contains_no_authority_beyond_identifier_and_target() { let target = Keys::generate().public_key(); diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index d477a7ece2..e0a1a0279e 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3437,6 +3437,23 @@ impl Db { workflow::latest_scheduled_workflow_fire(&self.pool, community_id, workflow_id).await } + /// Read the run linked by one durable scheduled-fire claim. + #[datastore_span(name = "get_scheduled_workflow_fire_run", system = "postgresql")] + pub async fn get_scheduled_workflow_fire_run( + &self, + community_id: CommunityId, + workflow_id: Uuid, + scheduled_for: chrono::DateTime, + ) -> Result> { + workflow::get_scheduled_workflow_fire_run( + &self.pool, + community_id, + workflow_id, + scheduled_for, + ) + .await + } + /// Attach the workflow run id created from a won scheduled-fire claim. #[datastore_span(name = "attach_scheduled_workflow_run", system = "postgresql")] pub async fn attach_scheduled_workflow_run( @@ -3456,6 +3473,56 @@ impl Db { .await } + /// Persist a payload-free webhook invocation authority. + pub async fn create_workflow_webhook_invocation( + &self, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, + ) -> Result<()> { + workflow::create_workflow_webhook_invocation( + &self.pool, + community_id, + workflow_id, + invocation_id, + ) + .await + } + + /// Link a webhook invocation authority to the run it caused. + pub async fn attach_workflow_webhook_invocation_run( + &self, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, + workflow_run_id: Uuid, + ) -> Result { + workflow::attach_workflow_webhook_invocation_run( + &self.pool, + community_id, + workflow_id, + invocation_id, + workflow_run_id, + ) + .await + } + + /// Read the run linked by one opaque webhook invocation. + pub async fn get_workflow_webhook_invocation_run( + &self, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, + ) -> Result> { + workflow::get_workflow_webhook_invocation_run( + &self.pool, + community_id, + workflow_id, + invocation_id, + ) + .await + } + /// Delete old scheduled workflow fire claims before a retention cutoff. #[datastore_span(name = "prune_scheduled_workflow_fires_before", system = "postgresql")] pub async fn prune_scheduled_workflow_fires_before( diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index e5fd8ca7fb..41acbe980e 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -645,7 +645,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 35); + assert_eq!(migrations.len(), 36); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1106,6 +1106,16 @@ mod tests { desired_schema.contains("attach_community_write_fence('workflow_agent_deliveries')") ); + // Payload-free durable authority binds each opaque webhook invocation + // to its tenant, workflow, and eventual run. + assert_eq!(migrations[35].version, 36); + let webhook_invocations = migrations[35].sql.as_str(); + assert!(webhook_invocations.contains("CREATE TABLE workflow_webhook_invocations")); + assert!(webhook_invocations.contains("PRIMARY KEY (community_id, invocation_id)")); + assert!(webhook_invocations + .contains("attach_community_write_fence('workflow_webhook_invocations')")); + assert!(desired_schema.contains("CREATE TABLE workflow_webhook_invocations")); + // Fresh desired-state bootstrap must install the identical executable // fence as migration 0032. CI and isolated relay startup use schema.sql // without running migrations, so drift reopens rolling-deploy races. @@ -1584,6 +1594,9 @@ mod tests { // Migration 0035 attaches the fence to the durable managed-agent // delivery inbox after 0029; the desired-state schema carries it inline. expected_fences.insert("workflow_agent_deliveries".to_owned()); + // Migration 0036 likewise adds the payload-free webhook invocation + // authority after 0029. + expected_fences.insert("workflow_webhook_invocations".to_owned()); assert_eq!( expected_fences, schema.fence_attachments, "write-fence attachment targets differ after recovery policy" diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 32461830f8..27b959dc2a 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -599,6 +599,32 @@ pub async fn latest_scheduled_workflow_fire( row.try_get("scheduled_for").map_err(Into::into) } +/// Read the durable authority binding for one scheduled workflow fire. +/// +/// Returns the linked run only from the tenant/workflow/slot primary key; a +/// missing or not-yet-attached claim cannot authorize delivery. +pub async fn get_scheduled_workflow_fire_run( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + scheduled_for: DateTime, +) -> Result> { + sqlx::query_scalar( + r#" + SELECT workflow_run_id + FROM scheduled_workflow_fires + WHERE community_id = $1 AND workflow_id = $2 AND scheduled_for = $3 + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(scheduled_for) + .fetch_optional(pool) + .await + .map(|value| value.flatten()) + .map_err(Into::into) +} + /// Link a won scheduled-fire claim to the workflow run it created. /// /// This is for ops/audit forensics only; the claim row remains the dedupe @@ -632,6 +658,78 @@ pub async fn attach_scheduled_workflow_run( Ok(result.rows_affected() == 1) } +/// Persist a payload-free webhook invocation authority before run creation. +pub async fn create_workflow_webhook_invocation( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, +) -> Result<()> { + sqlx::query( + r#" + INSERT INTO workflow_webhook_invocations + (community_id, invocation_id, workflow_id) + SELECT community_id, $3, id + FROM workflows + WHERE community_id = $1 AND id = $2 + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(invocation_id) + .execute(pool) + .await?; + Ok(()) +} + +/// Link a durable webhook invocation to the run it caused. +pub async fn attach_workflow_webhook_invocation_run( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, + workflow_run_id: Uuid, +) -> Result { + let result = sqlx::query( + r#" + UPDATE workflow_webhook_invocations + SET workflow_run_id = $4 + WHERE community_id = $1 AND workflow_id = $2 AND invocation_id = $3 + AND workflow_run_id IS NULL + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(invocation_id) + .bind(workflow_run_id) + .execute(pool) + .await?; + Ok(result.rows_affected() == 1) +} + +/// Read a webhook invocation's durable tenant/workflow/run binding. +pub async fn get_workflow_webhook_invocation_run( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, +) -> Result> { + sqlx::query_scalar( + r#" + SELECT workflow_run_id + FROM workflow_webhook_invocations + WHERE community_id = $1 AND workflow_id = $2 AND invocation_id = $3 + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(invocation_id) + .fetch_optional(pool) + .await + .map(|value| value.flatten()) + .map_err(Into::into) +} + /// Delete old scheduled workflow fire claims for retention. /// /// Schedule claim rows are correctness metadata, but they grow with every fire. diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 50f6a74c10..d18c443303 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2111,6 +2111,16 @@ pub async fn workflow_webhook( .await .map_err(|_| not_found("workflow not found"))?; + // Mint durable opaque authority before run creation. The record contains + // no request body or secret; a NULL run link remains honest if insertion + // of the run later fails. + let webhook_invocation_id = uuid::Uuid::new_v4(); + state + .db + .create_workflow_webhook_invocation(community_id, id, webhook_invocation_id) + .await + .map_err(|e| super::internal_error(&format!("persist webhook invocation: {e}")))?; + let definition_event_id = workflow .definition_event_id .as_deref() @@ -2127,6 +2137,17 @@ pub async fn workflow_webhook( .await .map_err(|e| super::internal_error(&format!("db error: {e}")))?; + let attached = state + .db + .attach_workflow_webhook_invocation_run(community_id, id, webhook_invocation_id, run_id) + .await + .map_err(|e| super::internal_error(&format!("attach webhook invocation: {e}")))?; + if !attached { + return Err(super::internal_error( + "webhook invocation authority was not attachable", + )); + } + // Spawn workflow execution asynchronously. let engine = Arc::clone(&state.workflow_engine); let trigger_ctx_clone = trigger_ctx.clone(); diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index e50cc3f600..a4130c52d5 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -17,7 +17,9 @@ use serde_json::Value; use uuid::Uuid; use buzz_core::{ - workflow_delivery::{WorkflowDeliveryBinding, WorkflowDeliveryCause, WorkflowDeliveryId}, + workflow_delivery::{ + WorkflowDeliveryBinding, WorkflowDeliveryCause, WorkflowDeliveryId, WorkflowDeliveryReceipt, + }, TenantContext, }; @@ -700,6 +702,66 @@ fn terminal_matches( ) } +async fn validate_delivery_cause_authority( + state: &Arc, + tenant: &TenantContext, + binding: &WorkflowDeliveryBinding, +) -> Result<(), (StatusCode, Json)> { + let authoritative_run = match binding.cause() { + WorkflowDeliveryCause::Event(event_id) => { + let event = state + .db + .get_event_by_id(tenant.community(), event_id.as_bytes()) + .await + .map_err(|error| internal_error(&format!("delivery cause lookup: {error}")))? + .ok_or_else(|| { + api_error(StatusCode::CONFLICT, "delivery authority is unavailable") + })?; + if event.event.verify().is_err() { + return Err(api_error( + StatusCode::CONFLICT, + "delivery authority is unavailable", + )); + } + return Ok(()); + } + WorkflowDeliveryCause::Schedule { + scheduled_for_unix_seconds, + } => { + let scheduled_for = + chrono::DateTime::::from_timestamp(*scheduled_for_unix_seconds, 0) + .ok_or_else(|| { + api_error(StatusCode::CONFLICT, "delivery authority is unavailable") + })?; + state + .db + .get_scheduled_workflow_fire_run( + tenant.community(), + binding.workflow_id(), + scheduled_for, + ) + .await + .map_err(|error| internal_error(&format!("schedule authority lookup: {error}")))? + } + WorkflowDeliveryCause::Webhook { invocation_id } => state + .db + .get_workflow_webhook_invocation_run( + tenant.community(), + binding.workflow_id(), + *invocation_id, + ) + .await + .map_err(|error| internal_error(&format!("webhook authority lookup: {error}")))?, + }; + if authoritative_run != Some(binding.run_id()) { + return Err(api_error( + StatusCode::CONFLICT, + "delivery authority is unavailable", + )); + } + Ok(()) +} + async fn delivery_response( state: &Arc, tenant: &TenantContext, @@ -746,10 +808,16 @@ async fn delivery_response( .await .map_err(|error| internal_error(&format!("delivery message lookup: {error}")))? .ok_or_else(|| api_error(StatusCode::CONFLICT, "delivery binding is unavailable"))?; + validate_delivery_cause_authority(state, tenant, &delivery.binding).await?; + let receipt = + WorkflowDeliveryReceipt::new(delivery.id, delivery.binding.clone(), &message.event) + .and_then(|receipt| receipt.sign(&state.relay_keypair, &message.event)) + .map_err(|error| internal_error(&format!("sign delivery receipt: {error}")))?; Ok(Json(serde_json::json!({ "delivery": delivery_json(delivery, lease), "definition_event": definition.event, "message_event": message.event, + "receipt_event": receipt, }))) } @@ -1154,6 +1222,16 @@ mod tests { "private execution input" ); assert!(claimed.get("execution_trace").is_none()); + assert!(claimed.get("trigger_context").is_none()); + assert!(claimed.get("webhook_fields").is_none()); + let receipt: nostr::Event = + serde_json::from_value(claimed["receipt_event"].clone()).expect("signed receipt event"); + assert_eq!(receipt.pubkey, fixture.state.relay_keypair.public_key()); + assert!(receipt.verify().is_ok()); + assert!(receipt.content.is_empty()); + assert!(!claimed["receipt_event"] + .to_string() + .contains("private execution input")); let lease = serde_json::json!({ "lease_generation": 1, diff --git a/migrations/0036_workflow_webhook_invocations.sql b/migrations/0036_workflow_webhook_invocations.sql new file mode 100644 index 0000000000..650ca8c154 --- /dev/null +++ b/migrations/0036_workflow_webhook_invocations.sql @@ -0,0 +1,20 @@ +-- Durable, payload-free authority for webhook-triggered workflow runs. +-- The body and secret stay in transient trigger context; this table retains +-- only an opaque invocation identity and its tenant/workflow/run binding. +CREATE TABLE workflow_webhook_invocations ( + community_id UUID NOT NULL REFERENCES communities(id), + invocation_id UUID NOT NULL, + workflow_id UUID NOT NULL, + workflow_run_id UUID, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, invocation_id), + FOREIGN KEY (community_id, workflow_id) + REFERENCES workflows (community_id, id) ON DELETE CASCADE, + FOREIGN KEY (community_id, workflow_run_id) + REFERENCES workflow_runs (community_id, id) ON DELETE NO ACTION +); + +CREATE INDEX idx_workflow_webhook_invocations_created_at + ON workflow_webhook_invocations (created_at); + +SELECT attach_community_write_fence('workflow_webhook_invocations'); diff --git a/schema/schema.sql b/schema/schema.sql index ba897133cf..a91bcd9c43 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -536,6 +536,24 @@ CREATE INDEX idx_workflow_agent_deliveries_lease ON workflow_agent_deliveries (lease_until) WHERE status = 'claimed'; +-- ── Workflow webhook invocations ───────────────────────────────────────────── +-- Payload-free durable authority linking an opaque webhook call to its run. +CREATE TABLE workflow_webhook_invocations ( + community_id UUID NOT NULL REFERENCES communities(id), + invocation_id UUID NOT NULL, + workflow_id UUID NOT NULL, + workflow_run_id UUID, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, invocation_id), + FOREIGN KEY (community_id, workflow_id) + REFERENCES workflows (community_id, id) ON DELETE CASCADE, + FOREIGN KEY (community_id, workflow_run_id) + REFERENCES workflow_runs (community_id, id) ON DELETE NO ACTION +); + +CREATE INDEX idx_workflow_webhook_invocations_created_at + ON workflow_webhook_invocations (created_at); + -- ── API tokens ──────────────────────────────────────────────────────────────── -- Conformance: "API tokens and NIP-98 replay". token_hash uniqueness scoped to -- (community_id, token_hash); channel claims reference channels in same community. @@ -1810,6 +1828,7 @@ SELECT attach_community_write_fence('reactions'); SELECT attach_community_write_fence('relay_invites'); SELECT attach_community_write_fence('relay_members'); SELECT attach_community_write_fence('scheduled_workflow_fires'); +SELECT attach_community_write_fence('workflow_webhook_invocations'); SELECT attach_community_write_fence('subscriptions'); SELECT attach_community_write_fence('thread_metadata'); SELECT attach_community_write_fence('users');