From 5dc8e4750d4424f55cd622c4a00aefae2d8edf60 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Mon, 24 Aug 2026 23:21:40 -0400 Subject: [PATCH] fix(workflows): execute immutable signed revisions Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-cli/src/commands/workflows.rs | 17 +-- crates/buzz-db/src/channel.rs | 26 +++- crates/buzz-relay/src/api/bridge.rs | 55 +++---- crates/buzz-relay/src/api/workflows.rs | 66 ++++++++- .../src/handlers/command_executor.rs | 138 ++++++++---------- crates/buzz-relay/src/router.rs | 4 + crates/buzz-workflow/src/lib.rs | 101 +++++++++++-- desktop/src-tauri/src/commands/workflows.rs | 13 +- 8 files changed, 268 insertions(+), 152 deletions(-) diff --git a/crates/buzz-cli/src/commands/workflows.rs b/crates/buzz-cli/src/commands/workflows.rs index e87edd114b..ac3f876dc2 100644 --- a/crates/buzz-cli/src/commands/workflows.rs +++ b/crates/buzz-cli/src/commands/workflows.rs @@ -166,16 +166,13 @@ async fn current_workflow_revision( client: &BuzzClient, workflow_id: &str, ) -> Result { - let filter = serde_json::json!({ - "kinds": [30620], - "#d": [workflow_id], - "limit": 1 - }); - let response = client.query(&filter).await?; - let events: Vec = serde_json::from_str(&response).unwrap_or_default(); - events - .first() - .and_then(|event| event.get("id")) + 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"))) diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 109a9367d7..5dbeb7b0ea 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -473,7 +473,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, @@ -1866,6 +1866,30 @@ pub async fn get_member_role( Ok(row.map(|r| r.try_get("role")).transpose()?) } +/// Get an active member role using the caller's transaction. +/// +/// Callers that authorize a workflow run must first acquire +/// [`acquire_channel_membership_lock`] and hold the transaction through run +/// commit, serializing authorization with membership revocation. +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()?) +} + /// Archive ephemeral channels whose TTL deadline has passed. /// /// Returns the `(community_id, host, channel_id)` list that was archived. Idempotent — the diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index bbfcd8ecfe..998ce0eb3b 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2102,50 +2102,35 @@ pub async fn workflow_webhook( let run_id = state .db - .create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref()) + .create_workflow_run( + community_id, + id, + workflow.definition_event_id.as_deref(), + None, + trigger_ctx_json.as_ref(), + ) .await .map_err(|e| super::internal_error(&format!("db error: {e}")))?; // Spawn workflow execution asynchronously. 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!("webhook: failed to parse definition: {e}"); - if let Err(db_err) = db - .update_workflow_run( - community_id, - run_id, - buzz_db::workflow::RunStatus::Failed, - 0, - &serde_json::json!([]), - Some(buzz_db::workflow::WorkflowRunFailure { - code: "invalid_definition", - message: &format!("definition parse error: {e}"), - }), - ) - .await - { - tracing::error!("webhook: 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; diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index a3d5a6c729..70d4bfcbd8 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) @@ -97,15 +98,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>, @@ -129,8 +174,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( @@ -169,7 +221,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 60d5bb2e33..0db5e647ad 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -1059,6 +1059,13 @@ 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 @@ -1079,6 +1086,22 @@ async fn handle_workflow_trigger( "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 { @@ -1124,44 +1147,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; @@ -1473,15 +1475,38 @@ async fn resume_workflow_after_approval( workflow_id: Uuid, resume_index: usize, ) { - let run = match db.get_workflow_run(community_id, run_id).await { - Ok(r) => r, + let (run, def) = match engine.load_run_definition(community_id, run_id).await { + Ok(bound) => bound, Err(e) => { - tracing::error!("resume_workflow: failed to fetch run {run_id}: {e}"); + tracing::error!( + "resume_workflow: failed to load run-bound definition for {run_id}: {e}" + ); + if let Ok(run) = db.get_workflow_run(community_id, run_id).await { + let _ = db + .update_workflow_run( + community_id, + run_id, + RunStatus::Failed, + run.current_step, + &run.execution_trace, + Some(buzz_db::workflow::WorkflowRunFailure { + code: "invalid_definition_revision", + message: &e.to_string(), + }), + ) + .await; + } return; } }; - // Guard: only resume runs that are actually waiting for approval + if run.workflow_id != workflow_id { + tracing::error!( + "resume_workflow: approval workflow {workflow_id} does not match run {}", + run.workflow_id + ); + return; + } if run.status != RunStatus::WaitingApproval { tracing::warn!( "resume_workflow: run {run_id} has status '{}', expected 'waiting_approval'", @@ -1490,40 +1515,6 @@ async fn resume_workflow_after_approval( return; } - let workflow = match db.get_workflow(community_id, workflow_id).await { - Ok(w) => w, - Err(e) => { - tracing::error!("resume_workflow: failed to fetch workflow {workflow_id}: {e}"); - return; - } - }; - - let def: buzz_workflow::WorkflowDef = match serde_json::from_value(workflow.definition.clone()) - { - Ok(d) => d, - Err(e) => { - tracing::error!("resume_workflow: failed to parse workflow definition: {e}"); - if let Err(db_err) = db - .update_workflow_run( - community_id, - run_id, - RunStatus::Failed, - run.current_step, - &run.execution_trace, - Some(buzz_db::workflow::WorkflowRunFailure { - code: "invalid_definition", - message: &format!("definition parse error: {e}"), - }), - ) - .await - { - tracing::error!("resume_workflow: failed to mark run as failed: {db_err}"); - } - return; - } - }; - - // Reconstruct step_outputs from execution trace for template resolution let mut initial_outputs: std::collections::HashMap = std::collections::HashMap::new(); if let Some(trace_arr) = run.execution_trace.as_array() { @@ -1537,14 +1528,11 @@ async fn resume_workflow_after_approval( } } - // Restore trigger context for {{trigger.*}} templates let trigger_ctx: TriggerContext = run .trigger_context .as_ref() .and_then(|v| serde_json::from_value(v.clone()).ok()) .unwrap_or_default(); - - // Execute remaining steps let existing_trace = run.execution_trace.as_array().cloned(); let result = buzz_workflow::executor::execute_from_step( &engine, diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 1dce66e91e..794f155978 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -72,6 +72,10 @@ pub fn build_router(state: Arc) -> Router { .route("/events", post(api::bridge::submit_event)) .route("/query", post(api::bridge::query_events)) .route("/count", post(api::bridge::count_events)) + .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-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 6ab072a5bd..d97dab8698 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(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 @@ -407,6 +465,7 @@ impl WorkflowEngine { .create_workflow_run( community_id, workflow.id, + workflow.definition_event_id.as_deref(), Some(&trigger_event_id_bytes), Some(&trigger_ctx_json), ) @@ -426,13 +485,22 @@ impl WorkflowEngine { ); let engine = Arc::clone(self); - let def_clone = def.clone(); let ctx_clone = trigger_ctx.clone(); tokio::spawn(async move { - let result = - executor::execute_run(&engine, community_id, run_id, &def_clone, &ctx_clone) - .await; + let result = match engine.load_run_definition(community_id, run_id).await { + Ok((_, definition)) => { + executor::execute_run( + &engine, + community_id, + run_id, + &definition, + &ctx_clone, + ) + .await + } + Err(error) => Err((error, PartialProgress::default())), + }; engine .finalize_run(community_id, run_id, result, None) .await; @@ -669,6 +737,7 @@ impl WorkflowEngine { .create_workflow_run( community_id, workflow.id, + workflow.definition_event_id.as_deref(), None, // no trigger event for cron trigger_ctx_json.as_ref(), ) @@ -721,17 +790,21 @@ impl WorkflowEngine { ); let engine = Arc::clone(self); - let def_clone = def.clone(); let ctx_clone = trigger_ctx.clone(); tokio::spawn(async move { - let result = executor::execute_run( - &engine, - community_id, - run_id, - &def_clone, - &ctx_clone, - ) - .await; + let result = match engine.load_run_definition(community_id, run_id).await { + Ok((_, definition)) => { + executor::execute_run( + &engine, + community_id, + run_id, + &definition, + &ctx_clone, + ) + .await + } + Err(error) => Err((error, PartialProgress::default())), + }; engine .finalize_run(community_id, run_id, result, None) .await; diff --git a/desktop/src-tauri/src/commands/workflows.rs b/desktop/src-tauri/src/commands/workflows.rs index beac00e675..f2f99d76f4 100644 --- a/desktop/src-tauri/src/commands/workflows.rs +++ b/desktop/src-tauri/src/commands/workflows.rs @@ -326,16 +326,9 @@ pub async fn trigger_workflow( // 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 prior = query_relay( - &state, - &[serde_json::json!({ - "kinds": [30620], - "#d": [workflow_id.clone()], - "limit": 1 - })], - ) - .await?; - let definition_event_id = current_workflow_revision(&prior)?; + 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)