From 9370cfad440119b196fe088aecee031f6698de42 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Tue, 10 Nov 2026 14:28:48 -0500 Subject: [PATCH 1/2] feat(agents): admit gateway-owned workflow prompts Signed-off-by: Tee8z (cherry picked from commit 2a73e9a6893a506d2055fa0f2cad66c59a601a4f) Signed-off-by: Tee8z --- crates/buzz-acp/src/gateway.rs | 103 ++++++++---- crates/buzz-acp/src/gateway/tests.rs | 7 +- crates/buzz-acp/src/gateway/tests/workflow.rs | 152 ++++++++++++++++++ crates/buzz-acp/src/lib.rs | 14 +- crates/buzz-acp/src/pool.rs | 6 +- crates/buzz-sandbox-reaper/src/manager.rs | 11 +- .../src/mentions/gateways.rs | 4 +- .../buzz-sandbox-reaper/src/mentions/tests.rs | 1 + .../src/mentions/tests/gateway.rs | 4 +- .../src/mentions/tests/relay.rs | 2 +- .../src/mentions/tests/workflow.rs | 109 +++++++++++++ .../src/mentions/worker.rs | 67 ++++++-- .../buzz-sandbox-reaper/src/mobile_launch.rs | 28 ++++ .../src/mobile_launch/tests.rs | 49 ++++++ docs/agent-gateway-key.md | 39 +++++ 15 files changed, 531 insertions(+), 65 deletions(-) create mode 100644 crates/buzz-acp/src/gateway/tests/workflow.rs create mode 100644 crates/buzz-sandbox-reaper/src/mentions/tests/workflow.rs diff --git a/crates/buzz-acp/src/gateway.rs b/crates/buzz-acp/src/gateway.rs index 2948b96c2b3..dfacf39b2e6 100644 --- a/crates/buzz-acp/src/gateway.rs +++ b/crates/buzz-acp/src/gateway.rs @@ -37,16 +37,15 @@ const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5); #[error("gateway delegation lookup failed: {0}")] pub(crate) struct LookupError(String); +struct Authority { + gateways: Vec, + relay: Option, +} + enum Cached { Empty, - Fresh { - at: Instant, - gateways: Vec, - }, - Failed { - at: Instant, - error: LookupError, - }, + Fresh { at: Instant, authority: Authority }, + Failed { at: Instant, error: LookupError }, } /// The owner's live gateways for one agent, from the relay, with a bounded @@ -82,6 +81,10 @@ impl GatewayAccess { /// The live gateways now, newest delegation first, at most /// [`buzz_sdk::gateway::MAX_LIVE_GATEWAYS`]. pub(crate) async fn live(&self) -> Result, LookupError> { + Ok(self.authority().await?.0) + } + + async fn authority(&self) -> Result<(Vec, Option), LookupError> { let mut cached = self.cached.lock().await; let reuse = match &*cached { Cached::Fresh { at, .. } => at.elapsed() < VERDICT_TTL, @@ -93,9 +96,9 @@ impl GatewayAccess { // cannot extend the authority cache beyond the documented TTL. let lookup_started = Instant::now(); *cached = match tokio::time::timeout(LOOKUP_TIMEOUT, self.lookup()).await { - Ok(Ok(gateways)) => Cached::Fresh { + Ok(Ok(authority)) => Cached::Fresh { at: lookup_started, - gateways, + authority, }, Ok(Err(error)) => Cached::Failed { at: Instant::now(), @@ -110,11 +113,15 @@ impl GatewayAccess { let now = unix_now(); match &*cached { // A cached delegation never outlives its own expiration. - Cached::Fresh { gateways, .. } => Ok(gateways - .iter() - .filter(|live| live.is_live_at(now)) - .map(|live| live.gateway()) - .collect()), + Cached::Fresh { authority, .. } => Ok(( + authority + .gateways + .iter() + .filter(|live| live.is_live_at(now)) + .map(|live| live.gateway()) + .collect(), + authority.relay, + )), Cached::Failed { error, .. } => Err(error.clone()), Cached::Empty => Err(LookupError("no lookup".into())), } @@ -128,7 +135,33 @@ impl GatewayAccess { Ok(self.live().await?.contains(key)) } - async fn lookup(&self) -> Result, LookupError> { + /// Admit a direct gateway prompt, or a workflow prompt whose gateway owner + /// is proven by the current relay's signature and explicit agent mention. + /// Workflow routing tags alone never confer gateway authority. + pub(crate) async fn prompt_author( + &self, + event: &Event, + ) -> Result, LookupError> { + let (gateways, relay) = self.authority().await?; + if gateways.contains(&event.pubkey) { + return Ok(Some(event.pubkey)); + } + let Some(relay) = relay else { + return Ok(None); + }; + let Some(owner) = buzz_sdk::workflow::owner(event, relay, self.agent) else { + return Ok(None); + }; + if !gateways.contains(&owner) + || !buzz_sdk::sharing::task::scope(event, self.agent, unix_now()) + .is_ok_and(|scope| scope.channel == self.channel) + { + return Ok(None); + } + Ok(Some(owner)) + } + + async fn lookup(&self) -> Result { let now = unix_now(); let heads = self .query(json!({ @@ -142,7 +175,10 @@ impl GatewayAccess { .map(|live| live.gateway().to_hex()) .collect(); if candidates.is_empty() { - return Ok(Vec::new()); + return Ok(Authority { + gateways: Vec::new(), + relay: None, + }); } let profiles = self .query(json!({ @@ -166,22 +202,31 @@ impl GatewayAccess { return Err(LookupError("ambiguous channel membership".into())); } let Some(roster) = rosters.first() else { - return Ok(Vec::new()); + return Ok(Authority { + gateways: Vec::new(), + relay: Some(relay), + }); }; let member = |key| buzz_sdk::gateway::channel_member(roster, relay, self.channel, key); if !member(self.owner) || !member(self.agent) { - return Ok(Vec::new()); + return Ok(Authority { + gateways: Vec::new(), + relay: Some(relay), + }); } - Ok(buzz_sdk::gateway::resolve( - self.owner, - &heads, - &profiles, - &[self.owner, self.agent], - now, - ) - .into_iter() - .filter(|live| member(live.gateway())) - .collect()) + Ok(Authority { + gateways: buzz_sdk::gateway::resolve( + self.owner, + &heads, + &profiles, + &[self.owner, self.agent], + now, + ) + .into_iter() + .filter(|live| member(live.gateway())) + .collect(), + relay: Some(relay), + }) } /// One `/query`, refusing any response that is not an array of events. diff --git a/crates/buzz-acp/src/gateway/tests.rs b/crates/buzz-acp/src/gateway/tests.rs index a0f88568a05..8386ad0972d 100644 --- a/crates/buzz-acp/src/gateway/tests.rs +++ b/crates/buzz-acp/src/gateway/tests.rs @@ -9,6 +9,8 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use super::*; +mod workflow; + /// Fake relay state shared with the serving task. pub(crate) struct FakeRelay { pub(crate) heads: Arc>>, @@ -259,7 +261,10 @@ async fn gateway_access_is_cached_bounded_and_fails_closed() { .unwrap(); *access.cached.lock().await = Cached::Fresh { at: tokio::time::Instant::now(), - gateways: buzz_sdk::gateway::live_heads(owner.public_key(), &[expired], now - 1), + authority: Authority { + gateways: buzz_sdk::gateway::live_heads(owner.public_key(), &[expired], now - 1), + relay: None, + }, }; assert!( !access.is_live(&gw).await.unwrap(), diff --git a/crates/buzz-acp/src/gateway/tests/workflow.rs b/crates/buzz-acp/src/gateway/tests/workflow.rs new file mode 100644 index 00000000000..418f27b03f5 --- /dev/null +++ b/crates/buzz-acp/src/gateway/tests/workflow.rs @@ -0,0 +1,152 @@ +use super::*; + +fn workflow(signer: &Keys, gateway: PublicKey, agent: PublicKey, channel: uuid::Uuid) -> Event { + EventBuilder::new(Kind::Custom(9), "@agent run the gateway workflow") + .tags([ + nostr::Tag::parse(["h", &channel.to_string()]).unwrap(), + nostr::Tag::public_key(agent), + nostr::Tag::parse(["buzz:workflow", "true"]).unwrap(), + nostr::Tag::parse(["buzz:workflow-owner", &gateway.to_hex()]).unwrap(), + nostr::Tag::parse(["buzz:workflow-mention", &agent.to_hex()]).unwrap(), + ]) + .sign_with_keys(signer) + .unwrap() +} + +/// Exercise both production admission boundaries. Removing provenance, scope, +/// current delegation or membership checks admits one of the negative cases. +#[tokio::test] +async fn gateway_workflows_require_provenance_scope_and_live_authority() { + let relay = fake_relay().await; + let (owner, agent, gateway) = (Keys::generate(), Keys::generate(), Keys::generate()); + let relay_key = Keys::parse(&"11".repeat(32)).unwrap(); + relay + .heads + .lock() + .unwrap() + .push(grant(&owner, gateway.public_key(), 10)); + relay.admit(&[owner.public_key(), agent.public_key(), gateway.public_key()]); + let access = Arc::new(GatewayAccess::new( + owner.public_key(), + agent.public_key(), + relay.rest.clone(), + channel(), + )); + let mut gate = crate::InboundAuthorGate::connect( + &relay.rest, + &agent.public_key().to_hex(), + "workflow test", + ) + .await; + gate.enable_gateways(Some(access.clone())); + let owner_cache = crate::OwnerCache::new(Some(owner.public_key().to_hex())); + let allowlist = std::collections::HashSet::new(); + let mut ctx = crate::pool::tests::make_prompt_context_no_owner(); + ctx.agent_owner_pubkey = Some(owner.public_key()); + ctx.gateways = Some(access.clone()); + let check = |event: Event, expected: bool| { + let (gate, cache, allowlist, ctx, rest) = + (&gate, &owner_cache, &allowlist, &ctx, &relay.rest); + let gateway = gateway.public_key(); + async move { + let decision = gate + .evaluate_for_test( + &event, + &crate::RespondTo::Owner, + allowlist, + false, + cache, + rest, + ) + .await; + assert_eq!(decision.allowed, expected, "ingress {}", event.id); + if expected { + assert_eq!(decision.effective_author, gateway.to_hex()); + } + let batch = crate::queue::FlushBatch { + channel_id: channel(), + scope: crate::scope::SessionScope::Thread { + channel_id: channel(), + root_event_id: event.id.to_hex(), + }, + events: vec![crate::queue::BatchEvent { + event, + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + edit: None, + }], + cancelled_events: vec![], + cancel_reason: None, + }; + assert_eq!( + crate::pool::dispatch_refusal(ctx, &batch).await.is_none(), + expected, + "dispatch" + ); + } + }; + let valid = workflow( + &relay_key, + gateway.public_key(), + agent.public_key(), + channel(), + ); + check(valid.clone(), true).await; + check( + workflow( + &Keys::generate(), + gateway.public_key(), + agent.public_key(), + channel(), + ), + false, + ) + .await; + check( + workflow( + &relay_key, + gateway.public_key(), + agent.public_key(), + uuid::Uuid::new_v4(), + ), + false, + ) + .await; + check( + workflow( + &relay_key, + gateway.public_key(), + Keys::generate().public_key(), + channel(), + ), + false, + ) + .await; + let mut tampered = valid.clone(); + tampered.content.push_str(" forged"); + check(tampered, false).await; + + relay.admit(&[owner.public_key(), agent.public_key()]); + skip(VERDICT_TTL).await; + check(valid.clone(), false).await; + relay.admit(&[owner.public_key(), agent.public_key(), gateway.public_key()]); + skip(VERDICT_TTL).await; + check(valid.clone(), true).await; + + let now = unix_now(); + *relay.heads.lock().unwrap() = + vec![ + buzz_sdk::gateway::Delegation::grant(gateway.public_key(), now, now - 10) + .unwrap() + .sign_with_keys(&owner) + .unwrap(), + ]; + skip(VERDICT_TTL).await; + check(valid.clone(), false).await; + *relay.heads.lock().unwrap() = vec![ + grant(&owner, gateway.public_key(), 10), + revoke(&owner, gateway.public_key(), 5), + ]; + skip(VERDICT_TTL).await; + check(valid, false).await; +} diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 702e14f78e9..06ed19732bd 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -658,7 +658,7 @@ mod inbound_author_gate { owner_cache: &OwnerCache, rest_client: &relay::RestClient, ) -> InboundAuthorGateDecision { - let effective_author = if matches!(respond_to, RespondTo::Owner) { + let mut effective_author = if matches!(respond_to, RespondTo::Owner) { event.pubkey.to_hex() } else { effective_prompt_author(event, self.relay_self.as_deref(), &self.agent_pubkey_hex) @@ -673,12 +673,16 @@ mod inbound_author_gate { ) .await; // Strict owner mode also admits a live owner-delegated gateway, in - // channels only. The raw signer is the effective author here, and - // the owner's DMs are never delegated. + // channels only. A relay workflow must cryptographically prove + // its gateway owner and signed agent/channel scope. if !allowed && !is_dm && matches!(respond_to, RespondTo::Owner) { if let Some(gateways) = &self.gateways { - allowed = match gateways.is_live(&event.pubkey).await { - Ok(live) => live, + allowed = match gateways.prompt_author(event).await { + Ok(Some(author)) => { + effective_author = author.to_hex(); + true + } + Ok(None) => false, Err(error) => { tracing::warn!( author = %event.pubkey, diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index d51399ae0fc..9d52784024f 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -4943,9 +4943,9 @@ pub(crate) async fn dispatch_refusal(ctx: &PromptContext, pending: &FlushBatch) } let mut gateway_error = None; if let Some(gateways) = &ctx.gateways { - match gateways.is_live(&item.event.pubkey).await { - Ok(true) => continue, - Ok(false) => {} + match gateways.prompt_author(&item.event).await { + Ok(Some(_)) => continue, + Ok(None) => {} Err(error) => gateway_error = Some(error.to_string()), } } diff --git a/crates/buzz-sandbox-reaper/src/manager.rs b/crates/buzz-sandbox-reaper/src/manager.rs index 2f5b334bb90..168e3ea8680 100644 --- a/crates/buzz-sandbox-reaper/src/manager.rs +++ b/crates/buzz-sandbox-reaper/src/manager.rs @@ -464,27 +464,26 @@ impl Manager { /// Reuse mobile start validation without manufacturing an owner control event. /// Start the agent for one mention. The trigger's signer is the owner, a - /// live owner-delegated gateway (`delegation`, freshly verified by the - /// caller for this trigger), or a shared-task requester. + /// live owner-delegated gateway, a relay-proven gateway workflow owner + /// (`delegation`, freshly verified for this trigger), or a shared requester. pub(crate) async fn start_mention( &self, owner: PublicKey, agent: PublicKey, relay: &str, event: &Event, - delegation: Option<&buzz_sdk::gateway::Live>, + delegation: Option<&crate::mobile_launch::GatewayRequest>, ) -> serde_json::Value { let Some(developer) = self.config.developer(&owner) else { return refusal(ErrorCode::UnauthorizedOwner, "owner is not configured"); }; let payload = serde_json::json!({"agent_pubkey": agent, "relay_url": relay, "trigger": event}); - let delegation = - delegation.filter(|live| live.owner() == owner && live.gateway() == event.pubkey); + let delegation = delegation.filter(|request| request.delegation.owner() == owner); let admitted = if event.pubkey == owner { Start::admit(payload, owner, Timestamp::now().as_secs()) } else if let Some(delegation) = delegation { - Start::delegated(payload, delegation, Timestamp::now().as_secs()) + Start::gateway(payload, delegation, Timestamp::now().as_secs()) } else { let request = crate::shared_tasks::Request { relay_url: relay.into(), diff --git a/crates/buzz-sandbox-reaper/src/mentions/gateways.rs b/crates/buzz-sandbox-reaper/src/mentions/gateways.rs index 8835d6ebd6a..4e8d4363cfa 100644 --- a/crates/buzz-sandbox-reaper/src/mentions/gateways.rs +++ b/crates/buzz-sandbox-reaper/src/mentions/gateways.rs @@ -64,10 +64,8 @@ pub(super) async fn member( identity: &Identity, channel: uuid::Uuid, gateway: nostr::PublicKey, + relay: nostr::PublicKey, ) -> Result { - let relay = crate::shared_tasks::relay_key(&identity.relay) - .await - .map_err(|_| Error::Relay)?; let rosters = fetch( connection, "mention-gateway-channel", diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests.rs b/crates/buzz-sandbox-reaper/src/mentions/tests.rs index 945edd1b3fa..ec6e6665ef3 100644 --- a/crates/buzz-sandbox-reaper/src/mentions/tests.rs +++ b/crates/buzz-sandbox-reaper/src/mentions/tests.rs @@ -6,6 +6,7 @@ mod live_relay; mod relay; mod revoked; mod shared_live; +mod workflow; use std::sync::Arc; use std::time::Duration; diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests/gateway.rs b/crates/buzz-sandbox-reaper/src/mentions/tests/gateway.rs index 5e27302d5e0..d91fad0af7e 100644 --- a/crates/buzz-sandbox-reaper/src/mentions/tests/gateway.rs +++ b/crates/buzz-sandbox-reaper/src/mentions/tests/gateway.rs @@ -10,7 +10,7 @@ fn signed_mention(signer: &Keys, content: &str) -> Event { .unwrap() } -fn delegation(enabled: bool, age: u64) -> Event { +pub(super) fn delegation(enabled: bool, age: u64) -> Event { let now = Timestamp::now().as_secs(); let gateway = keys("44").public_key(); let builder = if enabled { @@ -21,7 +21,7 @@ fn delegation(enabled: bool, age: u64) -> Event { builder.unwrap().sign_with_keys(&keys("33")).unwrap() } -fn roster(gateway: bool) -> Event { +pub(super) fn roster(gateway: bool) -> Event { let mut tags = vec![Tag::parse(["d", CHANNEL]).unwrap()]; for key in ["22", "33"].into_iter().chain(gateway.then_some("44")) { tags.push(Tag::parse(["p", &keys(key).public_key().to_hex(), "", "member"]).unwrap()); diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests/relay.rs b/crates/buzz-sandbox-reaper/src/mentions/tests/relay.rs index e758d161404..ca36ad45caa 100644 --- a/crates/buzz-sandbox-reaper/src/mentions/tests/relay.rs +++ b/crates/buzz-sandbox-reaper/src/mentions/tests/relay.rs @@ -193,7 +193,7 @@ async fn connection( assert!(author.is_some()); let id = input[1].as_str().unwrap().to_string(); let filter = input[2].clone(); - if matches!(id.as_str(), "mention-gateways" | "mention-gateway-profiles" | "mention-gateway-channel") { + if matches!(id.as_str(), "mention-gateways" | "mention-gateway-profiles" | "mention-gateway-channel" | "mention-channel" | "mention-listing") { assert_eq!(filter["consistency"], json!("strong")); } let allowed = |event: &Event| matches!(event.kind.as_u16(), 0 | 24210 | 30177 | 30181) || event.tags.iter().any(|tag| tag.as_slice() == ["h", CHANNEL] || tag.as_slice() == ["d", CHANNEL]); diff --git a/crates/buzz-sandbox-reaper/src/mentions/tests/workflow.rs b/crates/buzz-sandbox-reaper/src/mentions/tests/workflow.rs new file mode 100644 index 00000000000..951af23f873 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/mentions/tests/workflow.rs @@ -0,0 +1,109 @@ +use super::*; + +fn workflow(content: &str, mentioned: nostr::PublicKey) -> Event { + EventBuilder::new(Kind::Custom(9), content) + .tags([ + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::public_key(keys("22").public_key()), + Tag::parse(["buzz:workflow", "true"]).unwrap(), + Tag::parse(["buzz:workflow-owner", &keys("44").public_key().to_hex()]).unwrap(), + Tag::parse(["buzz:workflow-mention", &mentioned.to_hex()]).unwrap(), + ]) + .sign_with_keys(&keys("11")) + .unwrap() +} + +/// Drive the real listener, fresh delivery gate and manager launch boundary. +/// A relay signature alone does not authorize an implicit workflow mention; +/// a valid gateway-owned workflow launches, and later revocation still wins. +#[tokio::test] +async fn gateway_owned_workflow_launches_only_with_live_proven_authority() { + let relay = relay::Relay::start().await; + let cluster = sandbox_fake::Shared::default(); + register(&cluster, &relay.url).await; + relay + .events + .lock() + .unwrap() + .push(gateway::delegation(true, 10)); + relay + .events + .lock() + .unwrap() + .retain(|e| e.kind.as_u16() != 39002); + relay.events.lock().unwrap().push(gateway::roster(true)); + let stop = CancellationToken::new(); + let server = { + let manager = manager(&cluster); + let stop = stop.clone(); + let url = relay.url.clone(); + tokio::spawn(async move { + crate::serve::run_services(manager, &url, None, true, stop.cancelled()).await + }) + }; + relay.wait_subscription("mentions-history").await; + let mut workflow_sender = + NostrWsConnection::connect_authenticated(&relay.url, &keys("11"), None) + .await + .unwrap(); + let forged = workflow( + "@agent implicit p tag is insufficient", + keys("55").public_key(), + ); + assert!( + workflow_sender + .send_event(forged.clone()) + .await + .unwrap() + .accepted + ); + let valid = workflow("@agent explicit gateway workflow", keys("22").public_key()); + assert!( + workflow_sender + .send_event(valid.clone()) + .await + .unwrap() + .accepted + ); + wait_for(|| completed(&cluster, &valid)).await; + assert!( + !completed(&cluster, &forged), + "invalid provenance was journaled" + ); + assert_eq!(cluster.lock().unwrap().sandboxes.len(), 1); + + relay + .events + .lock() + .unwrap() + .push(gateway::delegation(false, 5)); + let revoked = workflow( + "@agent workflow after gateway revoked", + keys("22").public_key(), + ); + assert!( + workflow_sender + .send_event(revoked.clone()) + .await + .unwrap() + .accepted + ); + wait_for(|| { + relay.events.lock().unwrap().iter().any(|event| { + event.content.contains("not currently delegated") + && event + .tags + .iter() + .any(|tag| tag.as_slice().get(1) == Some(&revoked.id.to_hex())) + }) + }) + .await; + assert_eq!( + cluster.lock().unwrap().sandboxes.len(), + 1, + "revoked workflow launched" + ); + stop.cancel(); + tokio::time::timeout(BOUND, server).await.unwrap().unwrap(); + relay.stop().await; +} diff --git a/crates/buzz-sandbox-reaper/src/mentions/worker.rs b/crates/buzz-sandbox-reaper/src/mentions/worker.rs index 0b6a262eef7..49e4e0235c7 100644 --- a/crates/buzz-sandbox-reaper/src/mentions/worker.rs +++ b/crates/buzz-sandbox-reaper/src/mentions/worker.rs @@ -10,7 +10,12 @@ use super::journal::Journal; use super::registry::Identity; use super::{Error, IO_TIMEOUT}; use crate::manager::Manager; -use crate::mobile_launch::Start; +use crate::mobile_launch::{GatewayRequest, Start}; + +struct GatewayAuthors { + keys: Vec, + relay: Option, +} const PAGE: usize = 1000; const MAX_PAGES: usize = 16; @@ -20,9 +25,9 @@ const MAX_ATTEMPTS: u8 = 3; const MAX_EVENT_BYTES: usize = 60_000; enum LaunchPermission { - /// Deliverable; carries the trigger signer's fresh delegation when the - /// signer is a live owner-delegated gateway. - Allowed(Option), + /// Deliverable; carries fresh gateway authority for the direct signer or + /// the owner cryptographically proven by relay workflow provenance. + Allowed(Option), UnsupportedConversation, NoLongerRemote, GatewayNotMember, @@ -117,13 +122,27 @@ pub(super) async fn session( .iter() .map(buzz_sdk::gateway::Live::gateway) .collect(); + let relay = if gateways.is_empty() { + None + } else { + Some( + crate::shared_tasks::relay_key(&identity.relay) + .await + .map_err(|_| Error::Relay)?, + ) + }; + let gateways = GatewayAuthors { + keys: gateways, + relay, + }; let mut connected_at = Timestamp::now().as_secs(); let started = tokio::time::Instant::now(); let mut filter = json!({"kinds": [9, 40002], "#h":channels, "#p": [identity.keys.public_key()]}); if !manager.sharing_enabled() { let mut authors = vec![identity.owner]; - authors.extend(gateways.iter().copied()); + authors.extend(gateways.keys.iter().copied()); + authors.extend(gateways.relay); filter["authors"] = json!(authors); } let mut live = filter.clone(); @@ -263,7 +282,7 @@ fn eligible( journal: &Journal, event: &Event, shared: bool, - gateways: Option<&[nostr::PublicKey]>, + gateways: Option<&GatewayAuthors>, ) -> bool { if event.created_at.as_secs() < journal.state.activated || serde_json::to_vec(event).map_or(true, |v| v.len() > MAX_EVENT_BYTES) @@ -273,7 +292,15 @@ fn eligible( if event.pubkey == identity.owner { return conversation(identity, event).is_ok(); } - let delegated = gateways.is_none_or(|gateways| gateways.contains(&event.pubkey)); + let delegated = gateways.is_none_or(|gateways| { + gateways.keys.contains(&event.pubkey) + || gateways + .relay + .and_then(|relay| { + buzz_sdk::workflow::owner(event, relay, identity.keys.public_key()) + }) + .is_some_and(|owner| gateways.keys.contains(&owner)) + }); if delegated && gateway_thread(identity, event).is_ok() { return true; } @@ -536,21 +563,31 @@ async fn channel_launch_allowed( let delegation = if trigger.pubkey == identity.owner { None } else { - super::gateways::read(&mut connection, identity) + let relay = crate::shared_tasks::relay_key(&identity.relay).await.map_err(|_| Error::Relay)?; + let workflow_relay = (trigger.pubkey == relay).then_some(relay); + let requester = if workflow_relay.is_some() { + buzz_sdk::workflow::owner(trigger, relay, identity.keys.public_key()) + } else { + Some(trigger.pubkey) + }; + let delegation = super::gateways::read(&mut connection, identity) .await? .into_iter() - .find(|live| live.gateway() == trigger.pubkey) + .find(|live| Some(live.gateway()) == requester); + if let Some(live) = &delegation { + if !super::gateways::member(&mut connection, identity, + channel.parse().map_err(|_| Error::Registration)?, live.gateway(), relay).await? { + return Ok(LaunchPermission::GatewayNotMember); + } + } + delegation.map(|delegation| GatewayRequest { delegation, workflow_relay }) }; - if delegation.is_some() && !super::gateways::member(&mut connection, identity, - channel.parse().map_err(|_| Error::Registration)?, trigger.pubkey).await? { - return Ok(LaunchPermission::GatewayNotMember); - } send(&mut connection, json!(["REQ", "mention-trigger", {"kinds":[9,40002],"ids":[trigger.id],"#h":[channel],"limit":1}])).await?; - send(&mut connection, json!(["REQ", "mention-channel", {"kinds":[39000],"#d":[channel],"limit":1}])).await?; + send(&mut connection, json!(["REQ", "mention-channel", {"kinds":[39000],"#d":[channel],"limit":1,"consistency":"strong"}])).await?; // Query the current coordinate without a runtime filter: an old remote // record must not survive a newer local record or coordinate deletion. let agent = identity.keys.public_key().to_hex(); - send(&mut connection, json!(["REQ", "mention-listing", {"kinds":[30177],"authors":[identity.owner],"#d":[agent],"limit":1}])).await?; + send(&mut connection, json!(["REQ", "mention-listing", {"kinds":[30177],"authors":[identity.owner],"#d":[agent],"limit":1,"consistency":"strong"}])).await?; let (mut message_seen, mut metadata, mut trigger_done, mut channel_done, mut listing_done, mut remote) = (false, None, false, false, false, false); for _ in 0..16 { match connection.next_event(IO_TIMEOUT).await.map_err(|_| Error::Relay)? { diff --git a/crates/buzz-sandbox-reaper/src/mobile_launch.rs b/crates/buzz-sandbox-reaper/src/mobile_launch.rs index 90b033d0a74..55bd3f1a836 100644 --- a/crates/buzz-sandbox-reaper/src/mobile_launch.rs +++ b/crates/buzz-sandbox-reaper/src/mobile_launch.rs @@ -146,7 +146,35 @@ struct StartPayload { trigger: Event, } +/// Fresh gateway authority, optionally carrying the relay identity needed to +/// reverify workflow provenance. This context is never accepted from the wire. +pub(crate) struct GatewayRequest { + pub(crate) delegation: buzz_sdk::gateway::Live, + pub(crate) workflow_relay: Option, +} + impl Start { + /// Verify gateway authority again at the launch boundary. A workflow may + /// act only for its proven gateway owner, never for an arbitrary tagged key. + pub(crate) fn gateway( + payload: Value, + request: &GatewayRequest, + request_time: u64, + ) -> Result { + let Some(relay) = request.workflow_relay else { + return Self::delegated(payload, &request.delegation, request_time); + }; + let parsed: StartPayload = + serde_json::from_value(payload.clone()).map_err(|_| invalid())?; + if !request.delegation.is_live_at(request_time) + || parsed.agent_pubkey == request.delegation.gateway() + || buzz_sdk::workflow::owner(&parsed.trigger, relay, parsed.agent_pubkey) + != Some(request.delegation.gateway()) + { + return Err(invalid()); + } + Self::admit_signed(payload, request.delegation.owner(), relay, request_time) + } /// Construct a confined launch only after fresh sharing verification and /// atomic budget admission. This never manufactures an owner-signed event. pub(crate) fn shared( diff --git a/crates/buzz-sandbox-reaper/src/mobile_launch/tests.rs b/crates/buzz-sandbox-reaper/src/mobile_launch/tests.rs index 2affee457de..aedf1abe0f7 100644 --- a/crates/buzz-sandbox-reaper/src/mobile_launch/tests.rs +++ b/crates/buzz-sandbox-reaper/src/mobile_launch/tests.rs @@ -396,3 +396,52 @@ fn invalid_desktop_registration_has_a_deploy_specific_refusal_without_url_secret assert_eq!(error["code"], "invalid_launch_registration"); assert!(!error.to_string().contains("private")); } + +/// A workflow launch revalidates the current relay's proof at the final Start +/// boundary. Raw owner tags, a different target, expiration and signature +/// tampering cannot turn into an owner-scoped launch. +#[test] +fn gateway_workflow_start_revalidates_the_owner_proof() { + let gateway = Keys::parse(&"44".repeat(32)).unwrap(); + let relay = Keys::parse(&"11".repeat(32)).unwrap(); + let at = 1_800_000_000; + let head = buzz_sdk::gateway::Delegation::grant(gateway.public_key(), at + 60, at - 10) + .unwrap() + .sign_with_keys(&owner()) + .unwrap(); + let delegation = + buzz_sdk::gateway::live(&head, owner().public_key(), gateway.public_key(), at).unwrap(); + let request = super::GatewayRequest { + delegation, + workflow_relay: Some(relay.public_key()), + }; + let payload = |signer: &Keys, mentioned: nostr::PublicKey| { + let event = EventBuilder::new(Kind::Custom(9), "@agent workflow") + .tags([ + Tag::parse(["h", CHANNEL]).unwrap(), + Tag::public_key(agent().public_key()), + Tag::parse(["buzz:workflow", "true"]).unwrap(), + Tag::parse(["buzz:workflow-owner", &gateway.public_key().to_hex()]).unwrap(), + Tag::parse(["buzz:workflow-mention", &mentioned.to_hex()]).unwrap(), + ]) + .custom_created_at(Timestamp::from(at)) + .sign_with_keys(signer) + .unwrap(); + json!({"agent_pubkey":agent().public_key(),"relay_url":RELAY,"trigger":event}) + }; + let valid = payload(&relay, agent().public_key()); + let start = Start::gateway(valid.clone(), &request, at).unwrap(); + assert_eq!(start.owner, owner().public_key()); + assert_eq!(start.scope["channel_id"], CHANNEL); + assert!(Start::gateway( + payload(&Keys::generate(), agent().public_key()), + &request, + at + ) + .is_err()); + assert!(Start::gateway(payload(&relay, gateway.public_key()), &request, at).is_err()); + assert!(Start::gateway(valid.clone(), &request, at + 60).is_err()); + let mut tampered = valid; + tampered["trigger"]["content"] = json!("forged"); + assert!(Start::gateway(tampered, &request, at).is_err()); +} diff --git a/docs/agent-gateway-key.md b/docs/agent-gateway-key.md index ff86480f3f8..a4f1e766f11 100644 --- a/docs/agent-gateway-key.md +++ b/docs/agent-gateway-key.md @@ -85,6 +85,45 @@ daemon Secret, and the owner enrolls again to generate new credentials. Admit and grant the new public key. Remove the old key's relay and channel membership when its records no longer need to remain accessible to that account. +## Gateway-owned workflows + +A gateway can create and manage its own workflows using its ordinary channel +membership. With a live owner grant, those workflows can prompt that owner's +remote agents in channels where the owner, agent and gateway remain posting +members. The grant provides no authority over the owner's workflow definitions. + +Use the gateway key for the existing workflow commands: + +```sh +buzz workflows create --channel "$CHANNEL_ID" --yaml ./workflow.yaml +buzz workflows trigger --workflow "$WORKFLOW_ID" +``` + +Name the target agent explicitly in the stored workflow step text. For example, +replace `@agent` below with the agent's channel mention, then save `workflow.yaml`: + +```yaml +name: Gateway agent check +trigger: + on: webhook +steps: + - id: ask_agent + action: send_message + text: "@agent reply with gateway workflow confirmed" +``` + +Use the workflow ID returned by `create` for the explicit `trigger` command. +A mention introduced only by substituted trigger text does not carry authority. The relay signs the emitted prompt and +records the workflow owner and explicitly authored agent target. The launcher +and harness verify that signature and provenance before applying the gateway's +current grant and channel membership. The signed prompt must address the same +agent and channel; a custom owner tag by itself grants nothing. + +Queued workflow prompts are checked again before model dispatch. Expiration, +revocation and channel removal have the same bounds as direct gateway prompts. +Workflows cannot turn a gateway grant into owner administration, model changes +or access to owner DMs. + ## Revocation and failure behavior Revocation uses a newer disabled replacement, without expiration. The relay From d0149cdfc79f65e9ec3f75a4d7ae331e74d6fcc0 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Tue, 10 Nov 2026 14:45:50 -0500 Subject: [PATCH 2/2] docs(agents): pass workflow YAML through stdin Signed-off-by: Tee8z --- docs/agent-gateway-key.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/agent-gateway-key.md b/docs/agent-gateway-key.md index a4f1e766f11..5bef49ec7ed 100644 --- a/docs/agent-gateway-key.md +++ b/docs/agent-gateway-key.md @@ -95,7 +95,7 @@ members. The grant provides no authority over the owner's workflow definitions. Use the gateway key for the existing workflow commands: ```sh -buzz workflows create --channel "$CHANNEL_ID" --yaml ./workflow.yaml +buzz workflows create --channel "$CHANNEL_ID" --yaml - < workflow.yaml buzz workflows trigger --workflow "$WORKFLOW_ID" ```