diff --git a/crates/buzz-cli/src/commands/workflows.rs b/crates/buzz-cli/src/commands/workflows.rs index 0028dfc766..ac3f876dc2 100644 --- a/crates/buzz-cli/src/commands/workflows.rs +++ b/crates/buzz-cli/src/commands/workflows.rs @@ -162,6 +162,22 @@ pub async fn cmd_delete_workflow(client: &BuzzClient, workflow_id: &str) -> Resu Ok(()) } +async fn current_workflow_revision( + client: &BuzzClient, + workflow_id: &str, +) -> Result { + let response = client + .get_authed(&format!("/workflows/{workflow_id}/revision")) + .await?; + let event: serde_json::Value = serde_json::from_str(&response) + .map_err(|e| CliError::Other(format!("invalid workflow revision response: {e}")))?; + event + .get("id") + .and_then(|id| id.as_str()) + .map(str::to_owned) + .ok_or_else(|| CliError::NotFound(format!("workflow {workflow_id} not found"))) +} + /// Trigger a workflow — sign and submit a kind:46020 event. /// /// When `inputs` is provided, it is parsed as a JSON object and used as the @@ -172,6 +188,7 @@ pub async fn cmd_trigger_workflow( inputs: Option<&str>, ) -> Result<(), CliError> { let wf_uuid = parse_uuid(workflow_id)?; + let revision = current_workflow_revision(client, workflow_id).await?; if let Some(raw) = inputs { // Parse and validate it is a JSON object, then build the event manually @@ -183,8 +200,12 @@ pub async fn cmd_trigger_workflow( } let content = serde_json::to_string(&parsed).unwrap_or_default(); use nostr::{EventBuilder, Kind, Tag}; - let tags = vec![Tag::parse(["d", &wf_uuid.to_string()]) - .map_err(|e| CliError::Other(format!("tag error: {e}")))?]; + let tags = vec![ + Tag::parse(["d", &wf_uuid.to_string()]) + .map_err(|e| CliError::Other(format!("tag error: {e}")))?, + Tag::parse(["e", revision.as_str()]) + .map_err(|e| CliError::Other(format!("tag error: {e}")))?, + ]; let builder = EventBuilder::new( Kind::Custom(buzz_sdk::kind::KIND_WORKFLOW_TRIGGER as u16), &content, @@ -194,7 +215,7 @@ pub async fn cmd_trigger_workflow( let resp = client.submit_event(event).await?; println!("{}", normalize_write_response(&resp)); } else { - let builder = buzz_sdk::build_workflow_trigger(wf_uuid).map_err(sdk_err)?; + let builder = buzz_sdk::build_workflow_trigger(wf_uuid, &revision).map_err(sdk_err)?; let event = client.sign_event(builder)?; let resp = client.submit_event(event).await?; println!("{}", normalize_write_response(&resp)); diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 5e89b1101b..12065c3a06 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -21,8 +21,8 @@ pub use buzz_core::channel::{ChannelType, ChannelVisibility, MemberRole}; // Keep the established channel module paths compatible while membership SQL // and invariants live in their dedicated store module. pub use crate::channel_members::{ - add_member, get_accessible_channel_ids, get_accessible_channels, get_bot_members, - get_member_count, get_member_counts_bulk, get_member_role, get_members, get_members_bulk, + acquire_channel_membership_lock, add_member, get_accessible_channel_ids, get_accessible_channels, get_bot_members, + get_member_count, get_member_counts_bulk, get_member_role, get_member_role_in_transaction, get_members, get_members_bulk, get_users_bulk, is_member, list_large_channel_rosters_needing_reconciliation, lock_member_snapshot, membership_pairs, remove_member, verify_channel_roster_fence_behavior, verify_channel_roster_fence_catalog, AccessibleChannel, BotChannelEntry, BotMemberRecord, diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 8280ca01f8..8443f2872f 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -178,7 +178,7 @@ pub async fn verify_channel_roster_fence_behavior(pool: &sqlx::PgPool) -> Result /// Take the per-channel membership lock. MUST be the first statement in the /// transaction that then reads roles/owner counts and writes membership, so the /// whole check-then-write sequence is atomic against a concurrent one. -async fn acquire_channel_membership_lock( +pub async fn acquire_channel_membership_lock( tx: &mut Transaction<'_, Postgres>, community_id: CommunityId, channel_id: Uuid, @@ -3210,3 +3210,27 @@ mod postgres_tests { drop_scratch_db(&admin, pool, &scratch_name).await; } } + +/// Get an active member role using the caller's transaction. +/// +/// Authorization callers must acquire [`acquire_channel_membership_lock`] first +/// and hold the transaction through the dependent mutation. +pub async fn get_member_role_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + channel_id: Uuid, + pubkey: &[u8], +) -> Result> { + let row = sqlx::query( + "SELECT cm.role::text AS role FROM channel_members cm \ + JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \ + WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(pubkey) + .fetch_optional(&mut **tx) + .await?; + Ok(row.map(|r| r.try_get("role")).transpose()?) +} + diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index cb45809ead..4101c50661 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1164,6 +1164,26 @@ pub(crate) async fn get_event_by_id_with_operation( } } +/// Fetch a single live event by ID on the caller's transaction. +pub async fn get_event_by_id_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + id_bytes: &[u8], +) -> Result> { + let row = sqlx::query( + "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \ + FROM events WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(id_bytes) + .fetch_optional(&mut **tx) + .await?; + match row { + Some(r) => row_to_stored_event(r), + None => Ok(None), + } +} + /// Fetches the latest global (non-channel, `channel_id IS NULL`) replaceable event /// for a (kind, pubkey) pair. /// @@ -1528,6 +1548,18 @@ pub async fn insert_event_with_thread_metadata( } impl Db { + /// Fetch a live event by ID using the caller's transaction. + #[datastore_span(name = "get_event_by_id_in_transaction", system = "postgresql")] + pub async fn get_event_by_id_in_transaction( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + id_bytes: &[u8], + ) -> Result> { + crate::event::get_event_by_id_in_transaction(tx, community_id, id_bytes).await + } + + /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate. #[datastore_span(name = "insert_event", system = "postgresql")] pub async fn insert_event( diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 623d979e10..45c42a7d29 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -409,6 +409,33 @@ pub async fn get_workflow( row_to_workflow_record(row) } +/// Fetch and share-lock one workflow on an existing transaction. +/// +/// Definition replacement updates this row, so holding this lock through run +/// creation keeps revision validation and the dependent run atomic. +pub async fn get_workflow_for_share_in_transaction( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + id: Uuid, +) -> Result { + let row = sqlx::query( + r#" + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, + status::text AS status, enabled, created_at, updated_at + FROM workflows + WHERE community_id = $1 AND id = $2 + FOR SHARE + "#, + ) + .bind(community_id.as_uuid()) + .bind(id) + .fetch_optional(&mut **tx) + .await? + .ok_or_else(|| DbError::NotFound(format!("workflow {id}")))?; + + row_to_workflow_record(row) +} + /// List workflows for a channel, ordered newest first. /// /// `limit` is capped at [`LIST_MAX_LIMIT`]. Pass `None` to use [`LIST_DEFAULT_LIMIT`]. @@ -817,6 +844,34 @@ pub async fn delete_workflow_for_owner( // -- Workflow Run CRUD -------------------------------------------------------- +/// Insert a new exact-revision workflow run on the caller's transaction. +pub async fn create_workflow_run_in_transaction( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + workflow_id: Uuid, + definition_event_id: &[u8], + trigger_event_id: Option<&[u8]>, + trigger_context: Option<&serde_json::Value>, +) -> Result { + let id = Uuid::new_v4(); + sqlx::query( + r#" + INSERT INTO workflow_runs + (community_id, id, workflow_id, definition_event_id, status, trigger_event_id, current_step, execution_trace, trigger_context) + VALUES ($1, $2, $3, $4, 'pending', $5, 0, '[]', $6) + "#, + ) + .bind(community_id.as_uuid()) + .bind(id) + .bind(workflow_id) + .bind(definition_event_id) + .bind(trigger_event_id) + .bind(trigger_context) + .execute(&mut **tx) + .await?; + Ok(id) +} + /// Insert a new workflow run. Returns the new run's UUID. /// /// `trigger_context` is the serialized `TriggerContext` for this run. It is stored @@ -1297,6 +1352,40 @@ pub async fn find_by_owner_and_name( // -- Run and approval Db API -------------------------------------------------- impl Db { + /// Fetch and share-lock one workflow on an existing transaction. + #[datastore_span(name = "get_workflow_for_share_in_transaction", system = "postgresql")] + pub async fn get_workflow_for_share_in_transaction( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + id: Uuid, + ) -> Result { + crate::workflow::get_workflow_for_share_in_transaction(tx, community_id, id).await + } + + /// Create an exact-revision workflow run on an existing transaction. + #[datastore_span(name = "create_workflow_run_in_transaction", system = "postgresql")] + pub async fn create_workflow_run_in_transaction( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + workflow_id: Uuid, + definition_event_id: &[u8], + trigger_event_id: Option<&[u8]>, + trigger_context: Option<&serde_json::Value>, + ) -> Result { + crate::workflow::create_workflow_run_in_transaction( + tx, + community_id, + workflow_id, + definition_event_id, + trigger_event_id, + trigger_context, + ) + .await + } + + /// Create a new workflow run. #[datastore_span(name = "create_workflow_run", system = "postgresql")] pub async fn create_workflow_run( diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index c7fa09bebd..068b6cbf3b 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -46,6 +46,7 @@ async fn authorize_workflow_read( path: &str, raw_query: Option<&str>, workflow_id: Uuid, + allow_immutable_owner: bool, ) -> Result)> { let raw_host = headers .get(axum::http::header::HOST) @@ -99,15 +100,59 @@ async fn authorize_workflow_read( .await .map_err(|error| internal_error(&format!("workflow channel access lookup: {error}")))?; if !accessible.contains(&channel_id) { - return Err(api_error( - StatusCode::FORBIDDEN, - "workflow is not accessible", - )); + let controls = allow_immutable_owner + && (workflow.owner_pubkey == pubkey_bytes + || state + .db + .is_agent_owner(tenant.community(), &workflow.owner_pubkey, &pubkey_bytes) + .await + .map_err(|error| internal_error(&format!("workflow owner lookup: {error}")))?); + if !controls { + return Err(api_error( + StatusCode::FORBIDDEN, + "workflow is not accessible", + )); + } } Ok(tenant) } +/// `GET /workflows/{workflow_id}/revision` — current signed revision for an +/// authorized channel reader or the managed agent's immutable human owner. +/// +/// This narrow endpoint does not grant channel visibility; it returns only the +/// exact owner-signed definition event needed to construct a revision-bound +/// manual trigger. +pub async fn workflow_revision( + State(state): State>, + Path(workflow_id): Path, + headers: HeaderMap, +) -> Result, (StatusCode, Json)> { + let path = format!("/workflows/{workflow_id}/revision"); + let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id, true).await?; + let workflow = state + .db + .get_workflow(tenant.community(), workflow_id) + .await + .map_err(|_| api_error(StatusCode::NOT_FOUND, "workflow not found"))?; + let revision = workflow.definition_event_id.as_deref().ok_or_else(|| { + api_error( + StatusCode::CONFLICT, + "owner-signed workflow revision is unavailable", + ) + })?; + let event = state + .db + .get_event_by_id(tenant.community(), revision) + .await + .map_err(|error| internal_error(&format!("get workflow revision: {error}")))? + .ok_or_else(|| api_error(StatusCode::CONFLICT, "workflow revision is unavailable"))?; + Ok(Json(serde_json::to_value(event.event).map_err( + |error| internal_error(&format!("serialize workflow revision: {error}")), + )?)) +} + /// `GET /workflows/{workflow_id}/runs` — one authorized, keyset-paginated page. pub async fn workflow_runs( State(state): State>, @@ -131,8 +176,15 @@ pub async fn workflow_runs( } let path = format!("/workflows/{workflow_id}/runs"); - let tenant = - authorize_workflow_read(&state, &headers, &path, raw_query.as_deref(), workflow_id).await?; + let tenant = authorize_workflow_read( + &state, + &headers, + &path, + raw_query.as_deref(), + workflow_id, + false, + ) + .await?; let mut rows = state .db .list_workflow_runs_page( @@ -171,7 +223,7 @@ pub async fn run_approvals( headers: HeaderMap, ) -> Result, (StatusCode, Json)> { let path = format!("/workflows/{workflow_id}/runs/{run_id}/approvals"); - let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id).await?; + let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id, false).await?; let run = state .db diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 48675e9019..1acd4377b6 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -92,11 +92,11 @@ enum PersistResult { /// If the event is a duplicate (ON CONFLICT DO NOTHING), the transaction is /// rolled back and `PersistResult::Duplicate` is returned — no mutations needed. /// -/// Most domain mutations still execute on the connection pool rather than this -/// transaction, so their handlers rely on idempotency if event commit fails. -/// Workflow definition ingest is the exception: it materializes the workflow -/// revision on this same transaction so the signed event and revision pointer -/// commit or roll back together. +/// NOTE: Most domain mutations still execute on the connection pool rather +/// than in this transaction. Workflow-definition ingestion is the exception: +/// its materialized workflow row and exact signed revision are written through +/// this transaction so the event and revision binding commit atomically. +/// Other operations remain idempotent but not strictly atomic. #[datastore_span(name = "persist_command_event", system = "postgresql")] async fn persist_command_event( db: &buzz_db::Db, @@ -736,7 +736,8 @@ async fn handle_workflow_def( .map_err(|e| IngestError::Internal(format!("error: json serialize: {e}")))?; let hash = compute_definition_hash(&definition_json_final); - // Persist the signed definition and materialized revision atomically. + // Persist the command event — returns the transaction that will also own + // the materialized workflow revision update. let mut tx = match persist_command_event(&state.db, tenant, event, Some(channel_id)).await? { PersistResult::Duplicate => { return Ok(IngestResult { @@ -787,11 +788,14 @@ async fn handle_workflow_def( other => IngestError::Internal(format!("error: db upsert_workflow: {other}")), })?; - // Commit before cache invalidation so a concurrent refill can observe the new revision. + // Commit the event transaction after the idempotent workflow upsert succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; + // Invalidate only after commit. Invalidating while the new row is still + // invisible lets a concurrent trigger refill the cache with the old + // definition and retain it until TTL expiry. state .workflow_engine .invalidate_channel_workflows(community_id, channel_id); @@ -811,6 +815,101 @@ async fn handle_workflow_def( }) } +async fn caller_controls_workflow( + state: &Arc, + community_id: CommunityId, + workflow_owner: &[u8], + caller: &[u8], +) -> Result { + if workflow_owner == caller { + return Ok(true); + } + + state + .db + .is_agent_owner(community_id, workflow_owner, caller) + .await + .map_err(|e| IngestError::Internal(format!("error: workflow owner check: {e}"))) +} + +fn exact_tag_value<'a>(event: &'a Event, name: &str) -> Option<&'a str> { + let mut values = event.tags.iter().filter_map(|tag| { + (tag.kind().to_string() == name) + .then(|| tag.content()) + .flatten() + }); + let value = values.next()?; + values.next().is_none().then_some(value) +} + +async fn verify_workflow_revision( + state: &Arc, + mut tx: Option<&mut sqlx::Transaction<'_, sqlx::Postgres>>, + community_id: CommunityId, + workflow: &buzz_db::workflow::WorkflowRecord, + requested_revision: &[u8], +) -> Result<(), IngestError> { + let Some(persisted_revision) = workflow.definition_event_id.as_deref() else { + return Err(IngestError::Rejected( + "invalid: owner-signed workflow revision is unavailable".into(), + )); + }; + if persisted_revision != requested_revision { + return Err(IngestError::Rejected( + "conflict: workflow revision does not match current definition".into(), + )); + } + + let stored = match tx.as_mut() { + Some(tx) => { + state + .db + .get_event_by_id_in_transaction(tx, community_id, persisted_revision) + .await + } + None => { + state + .db + .get_event_by_id(community_id, persisted_revision) + .await + } + } + .map_err(|e| IngestError::Internal(format!("error: workflow revision lookup: {e}")))? + .ok_or_else(|| IngestError::Rejected("invalid: signed workflow revision not found".into()))?; + let definition_event = &stored.event; + let workflow_id = workflow.id.to_string(); + let workflow_channel_id = workflow.channel_id.map(|id| id.to_string()); + if definition_event.id.as_bytes() != persisted_revision + || !definition_event.verify_id() + || !definition_event.verify_signature() + || definition_event.kind.as_u16() as u32 != KIND_WORKFLOW_DEF + || definition_event.pubkey.to_bytes().as_slice() != workflow.owner_pubkey + || exact_tag_value(definition_event, "d") != Some(workflow_id.as_str()) + || workflow_channel_id.is_none() + || exact_tag_value(definition_event, "h") != workflow_channel_id.as_deref() + || stored.channel_id != workflow.channel_id + { + return Err(IngestError::Rejected( + "invalid: signed workflow revision binding mismatch".into(), + )); + } + + let (_, signed_json) = buzz_workflow::WorkflowEngine::parse_yaml(&definition_event.content) + .map_err(|_| { + IngestError::Rejected("invalid: signed workflow revision is malformed".into()) + })?; + let signed_definition: serde_json::Value = + serde_json::from_str(&signed_json).map_err(|_| { + IngestError::Rejected("invalid: signed workflow revision is malformed".into()) + })?; + if signed_definition != webhook_secret::strip_secret(&workflow.definition) { + return Err(IngestError::Rejected( + "invalid: signed workflow revision differs from materialized definition".into(), + )); + } + Ok(()) +} + async fn handle_workflow_trigger( tenant: &TenantContext, state: &Arc, @@ -819,13 +918,21 @@ async fn handle_workflow_trigger( ) -> Result { let self_bytes = auth.pubkey().to_bytes().to_vec(); - // 1. Extract workflow reference from `d` tag or `e` tag - let workflow_id_str = extract_d_tag(event) - .or_else(|| extract_e_tag(event)) - .ok_or_else(|| { - IngestError::Rejected("invalid: missing workflow reference (d or e tag)".into()) - })?; - let workflow_id = Uuid::parse_str(&workflow_id_str) + // 1. Bind the command to both the workflow UUID and one exact signed revision. + let workflow_id_str = exact_tag_value(event, "d").ok_or_else(|| { + IngestError::Rejected("invalid: expected exactly one workflow d tag".into()) + })?; + let revision_hex = exact_tag_value(event, "e").ok_or_else(|| { + IngestError::Rejected("invalid: expected exactly one workflow revision e tag".into()) + })?; + let requested_revision = hex::decode(revision_hex) + .map_err(|_| IngestError::Rejected("invalid: bad workflow revision event id".into()))?; + if requested_revision.len() != 32 { + return Err(IngestError::Rejected( + "invalid: bad workflow revision event id".into(), + )); + } + let workflow_id = Uuid::parse_str(workflow_id_str) .map_err(|_| IngestError::Rejected("invalid: bad workflow_id format".into()))?; // 2. Validate workflow exists — scoped to the caller's community. The same @@ -839,14 +946,20 @@ async fn handle_workflow_trigger( .await .map_err(|_| IngestError::Rejected("invalid: workflow not found".into()))?; - // 3. Manual triggers execute with the workflow owner's authority, so only - // the owner may start them. Channel membership alone is insufficient: a - // member could otherwise invoke another user's webhook or message actions. - if workflow.owner_pubkey != self_bytes { + // 3. Manual triggers execute with the workflow owner's authority. Permit + // that principal and, for a managed agent, its immutable human owner. + // Channel membership alone remains insufficient. + if !caller_controls_workflow(state, community_id, &workflow.owner_pubkey, &self_bytes).await? { return Err(IngestError::Rejected( "forbidden: not authorized to trigger this workflow".into(), )); } + // Managed-agent ownership is immutable. Carry the authorized workflow + // principal across the transaction boundary so no pool-backed ownership + // lookup is attempted while the command transaction holds its connection. + let authorized_workflow_owner = workflow.owner_pubkey.clone(); + + verify_workflow_revision(state, None, community_id, &workflow, &requested_revision).await?; // SEC-006: manual triggers must honor the workflow's lifecycle state and // recheck the owner's *current* channel authority before creating a run. @@ -876,7 +989,7 @@ async fn handle_workflow_trigger( // Persist the command event under the workflow channel even though the // trigger event itself only carries the workflow UUID. Storing channel // triggers as global events leaks workflow IDs to unrelated relay members. - let tx = match persist_command_event(&state.db, tenant, event, workflow.channel_id).await? { + let mut tx = match persist_command_event(&state.db, tenant, event, workflow.channel_id).await? { PersistResult::Duplicate => { return Ok(IngestResult { event_id: event.id.to_hex(), @@ -887,6 +1000,57 @@ async fn handle_workflow_trigger( PersistResult::Inserted(tx) => tx, }; + // Serialize the final authority check and run commit with channel + // membership writers. If revocation commits first we observe no role; if + // this lock wins, revocation cannot commit until this run is durable. + buzz_db::channel::acquire_channel_membership_lock(&mut tx, community_id, wf_channel_id) + .await + .map_err(|e| IngestError::Internal(format!("error: membership lock: {e}")))?; + + // Re-read the workflow under a row lock on the same transaction that will + // commit the trigger event and run. Definition replacement updates this row, + // so it cannot commit between this exact-revision check and our commit. A + // replacement that won first is observed here and rejected as stale. + let workflow = state + .db + .get_workflow_for_share_in_transaction(&mut tx, community_id, workflow_id) + .await + .map_err(|_| IngestError::Rejected("invalid: workflow not found".into()))?; + if workflow.owner_pubkey != authorized_workflow_owner { + return Err(IngestError::Rejected( + "conflict: workflow owner changed while trigger was being processed".into(), + )); + } + verify_workflow_revision( + state, + Some(&mut tx), + community_id, + &workflow, + &requested_revision, + ) + .await?; + if !workflow.enabled || workflow.status != buzz_db::workflow::WorkflowStatus::Active { + return Err(IngestError::Rejected( + "forbidden: workflow is disabled or inactive".into(), + )); + } + let role = buzz_db::channel::get_member_role_in_transaction( + &mut tx, + community_id, + wf_channel_id, + &workflow.owner_pubkey, + ) + .await + .map_err(|e| IngestError::Internal(format!("error: owner authority lookup: {e}")))?; + if !matches!( + (role.as_deref(), def.requires_elevated_authority()), + (Some(_), false) | (Some("owner" | "admin"), true) + ) { + return Err(IngestError::Rejected( + "forbidden: not authorized to trigger this workflow".into(), + )); + } + // 4. Execute: create workflow run let mut trigger_ctx = TriggerContext { channel_id: workflow @@ -912,10 +1076,11 @@ async fn handle_workflow_trigger( let event_id_bytes = event.id.as_bytes().to_vec(); let run_id = state .db - .create_workflow_run( + .create_workflow_run_in_transaction( + &mut tx, community_id, workflow_id, - workflow.definition_event_id.as_deref(), + &requested_revision, Some(&event_id_bytes), trigger_ctx_json.as_ref(), ) @@ -929,44 +1094,23 @@ async fn handle_workflow_trigger( // 5. Spawn workflow execution let engine = Arc::clone(&state.workflow_engine); - let db = state.db.clone(); - let def_value = workflow.definition.clone(); let trigger_ctx_clone = trigger_ctx.clone(); tokio::spawn(async move { - let def: buzz_workflow::WorkflowDef = match serde_json::from_value(def_value) { - Ok(d) => d, - Err(e) => { - tracing::error!("workflow_trigger: failed to parse definition: {e}"); - if let Err(db_err) = db - .update_workflow_run( - community_id, - run_id, - RunStatus::Failed, - 0, - &serde_json::json!([]), - Some(buzz_db::workflow::WorkflowRunFailure { - code: "invalid_definition", - message: &format!("definition parse error: {e}"), - }), - ) - .await - { - tracing::error!("workflow_trigger: failed to mark run as failed: {db_err}"); - } - return; + let result = match engine.load_run_definition(community_id, run_id).await { + Ok((_, definition)) => { + buzz_workflow::executor::execute_from_step( + &engine, + community_id, + run_id, + &definition, + &trigger_ctx_clone, + 0, + None, + ) + .await } + Err(error) => Err((error, buzz_workflow::error::PartialProgress::default())), }; - - let result = buzz_workflow::executor::execute_from_step( - &engine, - community_id, - run_id, - &def, - &trigger_ctx_clone, - 0, - None, - ) - .await; engine .finalize_run(community_id, run_id, result, None) .await; @@ -1375,7 +1519,10 @@ mod postgres_tests { let url = std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); // sadscan:disable np.postgres.1 -- local test-only credentials - let pool = sqlx::PgPool::connect(&url) + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(std::time::Duration::from_secs(1)) + .connect(&url) .await .expect("connect workflow persistence test database"); let db = buzz_db::Db::from_pool(pool); @@ -1393,6 +1540,180 @@ mod postgres_tests { (db, TenantContext::resolved(community, host)) } + async fn manual_trigger_test_context() -> (Arc, TenantContext, Keys, Keys, Uuid, Event) + { + use buzz_core::channel::{ChannelType, ChannelVisibility}; + + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); + let setup_pool = sqlx::PgPool::connect(&url) + .await + .expect("connect workflow trigger setup database"); + let setup_db = buzz_db::Db::from_pool(setup_pool.clone()); + setup_db + .migrate() + .await + .expect("migrate workflow trigger test database"); + + let host = format!("workflow-trigger-{}.example", Uuid::new_v4().simple()); + let community = setup_db + .ensure_configured_community(&host) + .await + .expect("create workflow trigger test community") + .id; + let tenant = TenantContext::resolved(community, host.clone()); + let human = Keys::generate(); + let agent = Keys::generate(); + let human_bytes = human.public_key().to_bytes(); + let agent_bytes = agent.public_key().to_bytes(); + setup_db + .ensure_user(community, &human_bytes) + .await + .expect("ensure human owner"); + setup_db + .ensure_user(community, &agent_bytes) + .await + .expect("ensure managed agent"); + assert!(setup_db + .set_agent_owner(community, &agent_bytes, &human_bytes) + .await + .expect("set immutable agent owner")); + let channel = setup_db + .create_channel( + community, + "manual-trigger-pool", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &agent_bytes, + None, + ) + .await + .expect("create workflow channel"); + let workflow_id = Uuid::new_v4(); + let definition = EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + concat!( + "name: manual-trigger-pool\n", + "trigger:\n on: message_posted\n", + "steps:\n - id: send\n action: send_message\n text: done\n", + ), + ) + .tags(vec![ + Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"), + Tag::parse(["h", channel.id.to_string().as_str()]).expect("h tag"), + ]) + .sign_with_keys(&agent) + .expect("sign workflow definition"); + let (_, definition_json) = buzz_workflow::WorkflowEngine::parse_yaml(&definition.content) + .expect("parse signed workflow definition"); + let definition_hash = compute_definition_hash(&definition_json); + let mut tx = setup_db + .begin_transaction() + .await + .expect("begin workflow seed"); + buzz_db::event::insert_event_in_transaction( + &mut tx, + community, + &definition, + Some(channel.id), + ) + .await + .expect("persist signed workflow definition"); + setup_db + .upsert_workflow( + &mut tx, + community, + workflow_id, + Some(channel.id), + &agent_bytes, + "manual-trigger-pool", + &definition_json, + &definition_hash, + definition.id.as_bytes(), + ) + .await + .expect("materialize signed workflow"); + tx.commit().await.expect("commit signed workflow"); + + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(std::time::Duration::from_secs(1)) + .connect(&url) + .await + .expect("connect one-connection workflow trigger pool"); + let db = buzz_db::Db::from_pool(pool.clone()); + let mut config = crate::config::Config::from_env().expect("config from env"); + config.database_url = url; + config.redis_url = "redis://127.0.0.1:1".to_string(); + config.relay_url = format!("wss://{host}"); + config.require_relay_membership = false; + let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool config"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) + .await + .expect("pubsub manager"), + ); + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage"); + let (state, _audit_shutdown) = AppState::new( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + Keys::generate(), + media_storage, + ); + setup_pool.close().await; + ( + Arc::new(state), + tenant, + human, + agent, + workflow_id, + definition, + ) + } + + fn workflow_trigger_event_for_revision( + keys: &Keys, + workflow_id: Uuid, + revision: &str, + ) -> Event { + EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "") + .tags(vec![ + Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"), + Tag::parse(["e", revision]).expect("revision tag"), + ]) + .sign_with_keys(keys) + .expect("sign workflow trigger") + } + + fn workflow_trigger_event(keys: &Keys, workflow_id: Uuid, revision: &Event) -> Event { + workflow_trigger_event_for_revision(keys, workflow_id, &revision.id.to_hex()) + } + + fn http_auth(keys: &Keys) -> IngestAuth { + IngestAuth::Http { + pubkey: keys.public_key(), + scopes: vec![buzz_auth::Scope::MessagesWrite], + auth_method: super::super::ingest::HttpAuthMethod::Nip98, + } + } + fn workflow_event( keys: &Keys, workflow_id: Uuid, @@ -1428,6 +1749,134 @@ mod postgres_tests { } } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn human_owner_manual_trigger_completes_with_one_connection() { + let (state, tenant, human, agent, workflow_id, revision) = + manual_trigger_test_context().await; + + let trigger = workflow_trigger_event(&human, workflow_id, &revision); + let result = tokio::time::timeout( + std::time::Duration::from_secs(3), + handle_workflow_trigger(&tenant, &state, &trigger, &http_auth(&human)), + ) + .await + .expect("human-owner trigger must not wait for a second pool connection") + .expect("human-owner trigger must succeed"); + let run_id = Uuid::parse_str( + serde_json::from_str::( + result + .message + .strip_prefix("response:") + .expect("workflow trigger response prefix"), + ) + .expect("workflow trigger response JSON")["run_id"] + .as_str() + .expect("workflow trigger run id"), + ) + .expect("workflow trigger run UUID"); + let (_, loaded_definition) = state + .workflow_engine + .load_run_definition(tenant.community(), run_id) + .await + .expect("manual execution must load its exact signed revision"); + assert_eq!(loaded_definition.name, "manual-trigger-pool"); + + let agent_trigger = workflow_trigger_event(&agent, workflow_id, &revision); + let agent_result = + handle_workflow_trigger(&tenant, &state, &agent_trigger, &http_auth(&agent)) + .await + .expect("workflow principal must be able to trigger its own workflow"); + assert!(agent_result.message.contains("\"run_id\"")); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn manual_trigger_rejects_non_owner_and_stale_or_missing_revision() { + let (state, tenant, human, _agent, workflow_id, revision) = + manual_trigger_test_context().await; + let stranger = Keys::generate(); + let unauthorized = workflow_trigger_event(&stranger, workflow_id, &revision); + let unauthorized_error = + match handle_workflow_trigger(&tenant, &state, &unauthorized, &http_auth(&stranger)) + .await + { + Err(error) => error, + Ok(_) => panic!("channel membership must not grant manual trigger authority"), + }; + assert!(matches!( + unauthorized_error, + IngestError::Rejected(ref message) + if message == "forbidden: not authorized to trigger this workflow" + )); + + let stale = workflow_trigger_event_for_revision(&human, workflow_id, &"42".repeat(32)); + let stale_error = + match handle_workflow_trigger(&tenant, &state, &stale, &http_auth(&human)).await { + Err(error) => error, + Ok(_) => panic!("a stale signed revision must be rejected"), + }; + assert!(matches!( + stale_error, + IngestError::Rejected(ref message) + if message == "conflict: workflow revision does not match current definition" + )); + + let missing = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "") + .tag(Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag")) + .sign_with_keys(&human) + .expect("sign revision-less trigger"); + let missing_error = + match handle_workflow_trigger(&tenant, &state, &missing, &http_auth(&human)).await { + Err(error) => error, + Ok(_) => panic!("a revision-less trigger must fail closed"), + }; + assert!(matches!( + missing_error, + IngestError::Rejected(ref message) + if message == "invalid: expected exactly one workflow revision e tag" + )); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn concurrent_human_owner_manual_triggers_do_not_starve_one_connection_pool() { + let (state, tenant, human, _agent, workflow_id, revision) = + manual_trigger_test_context().await; + let triggers = (0..8) + .map(|_| workflow_trigger_event(&human, workflow_id, &revision)) + .collect::>(); + + let results = tokio::time::timeout(std::time::Duration::from_secs(8), async { + let mut tasks = tokio::task::JoinSet::new(); + for trigger in triggers { + let state = Arc::clone(&state); + let tenant = tenant.clone(); + let auth = http_auth(&human); + tasks.spawn(async move { + handle_workflow_trigger(&tenant, &state, &trigger, &auth).await + }); + } + let mut results = Vec::new(); + while let Some(result) = tasks.join_next().await { + results.push(result.expect("trigger task must not panic")); + } + results + }) + .await + .expect("concurrent triggers must drain rather than pool-starve"); + + assert_eq!(results.len(), 8); + for result in results { + assert!( + result + .expect("concurrent human-owner trigger must succeed") + .accepted, + "manual trigger should be accepted" + ); + } + } + #[test] fn workflow_revision_parser_accepts_create_and_valid_update() { let revision = [0x42; 32]; @@ -1476,6 +1925,10 @@ mod postgres_tests { let workflow_id = Uuid::new_v4(); let created_at = Timestamp::now().as_secs(); let create = workflow_event(&keys, workflow_id, created_at, None, "create"); + let channel_id = Uuid::parse_str( + exact_tag_value(&create, "h").expect("workflow definition channel tag"), + ) + .expect("workflow definition channel UUID"); let missing_revision = hex::encode([0x24; 32]); let missing_revision_update = workflow_event( @@ -1496,15 +1949,22 @@ mod postgres_tests { if message == "conflict: workflow revision does not exist" )); - let PersistResult::Inserted(tx) = persist_command_event(&db, &tenant, &create, None) - .await - .expect("persist create") + let PersistResult::Inserted(tx) = + persist_command_event(&db, &tenant, &create, Some(channel_id)) + .await + .expect("persist create") else { panic!("first create must insert"); }; tx.commit().await.expect("commit create"); + let stored_create = db + .get_event_by_id(tenant.community(), create.id.as_bytes()) + .await + .expect("load persisted workflow definition") + .expect("persisted workflow definition"); + assert_eq!(stored_create.channel_id, Some(channel_id)); assert!(matches!( - persist_command_event(&db, &tenant, &create, None) + persist_command_event(&db, &tenant, &create, Some(channel_id)) .await .expect("replay create"), PersistResult::Duplicate diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 61aedf70be..96cff655f3 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -76,6 +76,10 @@ pub fn build_router(state: Arc) -> Router { // Relay-owned third-party GIF metadata proxy (NIP-98 auth). .route(api::gifs::SEARCH_PATH, post(api::gifs::search)) .route(api::gifs::SHARE_PATH, post(api::gifs::share)) + .route( + "/workflows/{workflow_id}/revision", + get(api::workflows::workflow_revision), + ) .route( "/workflows/{workflow_id}/runs", get(api::workflows::workflow_runs), diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index f43887b65b..4d7599cc84 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -1658,9 +1658,16 @@ pub fn build_workflow_delete( build_delete_addressable(KIND_WORKFLOW_DEF, author_pubkey, &workflow_id.to_string()) } -/// Build a workflow trigger event (kind 46020). -pub fn build_workflow_trigger(workflow_id: Uuid) -> Result { - let tags = vec![tag(&["d", &workflow_id.to_string()])?]; +/// Build a workflow trigger event (kind 46020) bound to an exact signed revision. +pub fn build_workflow_trigger( + workflow_id: Uuid, + definition_event_id: &str, +) -> Result { + let revision = check_hex_exact(definition_event_id, 64, "definition_event_id")?; + let tags = vec![ + tag(&["d", &workflow_id.to_string()])?, + tag(&["e", &revision])?, + ]; Ok(EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "").tags(tags)) } @@ -4113,9 +4120,11 @@ mod tests { #[test] fn workflow_trigger_happy_path() { let wid = uuid(); - let ev = sign(build_workflow_trigger(wid).unwrap()); + let ev = sign(build_workflow_trigger(wid, &"ab".repeat(32)).unwrap()); assert_eq!(ev.kind.as_u16(), 46020); assert!(has_tag(&ev, "d", &wid.to_string())); + assert!(has_tag(&ev, "e", &"ab".repeat(32))); + assert!(build_workflow_trigger(wid, "not-an-event-id").is_err()); } #[test] diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 390b9b14ac..fb44b6d568 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -44,7 +44,9 @@ use std::collections::HashMap; use std::sync::Arc; use std::sync::OnceLock; -use buzz_core::kind::{event_kind_u32, is_workflow_execution_kind, KIND_REACTION}; +use buzz_core::kind::{ + event_kind_u32, is_workflow_execution_kind, KIND_REACTION, KIND_WORKFLOW_DEF, +}; use buzz_core::tenant::CommunityId; use buzz_db::workflow::RunStatus; use buzz_db::Db; @@ -122,6 +124,62 @@ impl WorkflowEngine { } } + /// Load and verify the exact owner-signed definition bound to a run. + /// + /// The mutable `workflows` row supplies only immutable identity/channel + /// binding. Definition content always comes from the run's signed event; + /// legacy runs without a revision fail closed. + pub async fn load_run_definition( + &self, + community_id: CommunityId, + run_id: Uuid, + ) -> Result<(buzz_db::workflow::WorkflowRunRecord, WorkflowDef), WorkflowError> { + let run = self.db.get_workflow_run(community_id, run_id).await?; + let revision = run.definition_event_id.as_deref().ok_or_else(|| { + WorkflowError::InvalidDefinition( + "workflow run has no owner-signed definition revision".into(), + ) + })?; + let workflow = self.db.get_workflow(community_id, run.workflow_id).await?; + let stored = self + .db + .get_event_by_id_including_deleted(community_id, revision) + .await? + .ok_or_else(|| { + WorkflowError::InvalidDefinition( + "workflow run definition event is unavailable".into(), + ) + })?; + let event = &stored.event; + let workflow_id = run.workflow_id.to_string(); + let channel_id = workflow.channel_id.map(|id| id.to_string()); + let exact_tag = |name: &str| { + let mut values = event.tags.iter().filter_map(|tag| { + (tag.kind().to_string() == name) + .then(|| tag.content()) + .flatten() + }); + let value = values.next(); + value.filter(|_| values.next().is_none()) + }; + if event.id.as_bytes() != revision + || !event.verify_id() + || !event.verify_signature() + || event_kind_u32(event) != KIND_WORKFLOW_DEF + || event.pubkey.to_bytes().as_slice() != workflow.owner_pubkey + || exact_tag("d") != Some(workflow_id.as_str()) + || channel_id.is_none() + || exact_tag("h") != channel_id.as_deref() + || stored.channel_id != workflow.channel_id + { + return Err(WorkflowError::InvalidDefinition( + "workflow run definition event binding mismatch".into(), + )); + } + let (definition, _) = Self::parse_yaml(&event.content)?; + Ok((run, definition)) + } + /// Drop the cached enabled-workflow list for a channel. /// /// Must be called after any write to a workflow's trigger eligibility or diff --git a/desktop/src-tauri/src/commands/workflows.rs b/desktop/src-tauri/src/commands/workflows.rs index c4e5d38c8b..31aa948b93 100644 --- a/desktop/src-tauri/src/commands/workflows.rs +++ b/desktop/src-tauri/src/commands/workflows.rs @@ -323,7 +323,13 @@ pub async fn trigger_workflow( workflow_id: String, state: State<'_, AppState>, ) -> Result { - let builder = events::build_workflow_trigger(&workflow_id)?; + // Resolve the current signed definition at trigger time. The relay binds + // authorization and execution to this exact revision and rejects a stale + // result if an update races this command. + let revision: nostr::Event = + get_relay_json(&state, &format!("/workflows/{workflow_id}/revision")).await?; + let definition_event_id = revision.id.to_hex(); + let builder = events::build_workflow_trigger(&workflow_id, &definition_event_id)?; let result = submit_event(builder, &state).await?; trigger_wire_from_message(workflow_id, &result.message) } diff --git a/desktop/src-tauri/src/events/workflows.rs b/desktop/src-tauri/src/events/workflows.rs index 8615f73851..b73f8fca4f 100644 --- a/desktop/src-tauri/src/events/workflows.rs +++ b/desktop/src-tauri/src/events/workflows.rs @@ -31,9 +31,16 @@ pub fn build_workflow_delete( Ok(EventBuilder::new(Kind::Custom(5), "").tags(tags)) } -/// Kind 46020 — trigger a workflow run by id. -pub fn build_workflow_trigger(workflow_id: &str) -> Result { - let tags = vec![tag(vec!["d", workflow_id])?]; +/// Kind 46020 — trigger a workflow run by id, bound to one exact definition revision. +pub fn build_workflow_trigger( + workflow_id: &str, + definition_event_id: &str, +) -> Result { + EventId::from_hex(definition_event_id).map_err(|_| "invalid workflow revision".to_string())?; + let tags = vec![ + tag(vec!["d", workflow_id])?, + tag(vec!["e", definition_event_id])?, + ]; Ok(EventBuilder::new(Kind::Custom(46020), "").tags(tags)) }