diff --git a/apps/buzz-gpui/Cargo.lock b/apps/buzz-gpui/Cargo.lock index 0ac2cb5e87f..93d22d536ea 100644 --- a/apps/buzz-gpui/Cargo.lock +++ b/apps/buzz-gpui/Cargo.lock @@ -1238,12 +1238,14 @@ dependencies = [ "base64 0.22.1", "buzz-desktop", "buzz-identity", + "buzz-sdk", "chrono", "crc32fast", "futures", "image", "markdown", "moxcms", + "nostr 0.44.8", "nostr 0.45.5", "percent-encoding", "regex", diff --git a/apps/buzz-gpui/crates/identity/src/lib.rs b/apps/buzz-gpui/crates/identity/src/lib.rs index ee411099dfc..acd76569b3b 100644 --- a/apps/buzz-gpui/crates/identity/src/lib.rs +++ b/apps/buzz-gpui/crates/identity/src/lib.rs @@ -211,8 +211,8 @@ pub const PLUGIN_RELEASE: u16 = 1064; /// Kinds the app writes as ordinary content. const CONTENT_KINDS: &[u16] = &[ 0, 5, 7, 9, 1984, 9000, 9001, 9002, 9007, 9008, 9021, 9022, 9030, 9031, 9032, 9035, 9036, - 28936, 30030, 30175, 30177, 30178, 30315, 30620, 40003, 40100, 41010, 41012, 42000, 45010, - 46020, + 28936, 30030, 30175, 30177, 30178, 30180, 30315, 30620, 40003, 40100, 41010, 41012, 42000, + 45010, 46020, ]; impl Identity { @@ -266,6 +266,9 @@ impl Identity { content: String, ) -> Result { check_kind(purpose, kind)?; + if kind == 30180 { + check_channel_agent_grant(&self.pubkey_hex(), &tags, &content)?; + } if kind == AGENT_PRESENCE { check_agent_presence(&tags, &content)?; } @@ -290,6 +293,14 @@ impl Identity { /// Signs an event built by [`Self::template`], keeping its ID. pub fn sign(&self, purpose: Purpose, unsigned: UnsignedEvent) -> Result { check_kind(purpose, unsigned.kind.as_u16())?; + if unsigned.kind.as_u16() == 30180 { + let tags = unsigned + .tags + .iter() + .map(|tag| tag.as_slice().to_vec()) + .collect::>(); + check_channel_agent_grant(&self.pubkey_hex(), &tags, &unsigned.content)?; + } if unsigned.kind.as_u16() == AGENT_PRESENCE { let tags: Vec> = unsigned .tags @@ -654,6 +665,44 @@ fn check_kind(purpose: Purpose, kind: u16) -> Result<(), IdentityError> { } } +/// Grant signing never accepts credentials, an ambiguous scope, or a self grant. +fn check_channel_agent_grant( + viewer: &str, + tags: &[Vec], + content: &str, +) -> Result<(), IdentityError> { + let value: serde_json::Value = + serde_json::from_str(content).map_err(|_| IdentityError::NotAllowed)?; + let object = value.as_object().ok_or(IdentityError::NotAllowed)?; + let agent = value["agent"].as_str().ok_or(IdentityError::NotAllowed)?; + let channel = value["channel"] + .as_str() + .and_then(|id| uuid::Uuid::parse_str(id).ok()) + .ok_or(IdentityError::NotAllowed)?; + if object.len() != 4 + || value["version"] != 1 + || value["enabled"].as_bool().is_none() + || !PublicKey::from_hex(agent).is_ok_and(|key| key.to_hex() == agent) + || agent == viewer + { + return Err(IdentityError::NotAllowed); + } + for (name, expected) in [ + ("d", format!("{agent}:{channel}")), + ("h", channel.to_string()), + ("p", agent.to_owned()), + ] { + let matches = tags + .iter() + .filter(|tag| tag.first().map(String::as_str) == Some(name)) + .collect::>(); + if matches.len() != 1 || matches[0].as_slice() != [name, &expected] { + return Err(IdentityError::NotAllowed); + } + } + Ok(()) +} + /// Owner presence for the viewer's agents. pub const AGENT_PRESENCE: u16 = 20003; diff --git a/apps/buzz-gpui/crates/relay/Cargo.toml b/apps/buzz-gpui/crates/relay/Cargo.toml index 02712267235..91e8dfe0797 100644 --- a/apps/buzz-gpui/crates/relay/Cargo.toml +++ b/apps/buzz-gpui/crates/relay/Cargo.toml @@ -37,6 +37,8 @@ url = "2" uuid = { version = "1", features = ["v4"] } [dev-dependencies] +buzz-sdk = { path = "../../../../crates/buzz-sdk" } +nostr-wire = { package = "nostr", version = "0.44" } crc32fast = "1" tempfile = "3" # Test builds also trust `SSL_CERT_FILE`, for the test relay's own CA. diff --git a/apps/buzz-gpui/crates/relay/src/agent_sharing.rs b/apps/buzz-gpui/crates/relay/src/agent_sharing.rs new file mode 100644 index 00000000000..f8d09e3c79b --- /dev/null +++ b/apps/buzz-gpui/crates/relay/src/agent_sharing.rs @@ -0,0 +1,159 @@ +//! Owner opt-in for shared channel agents (kind 30180). + +#[cfg(test)] +mod tests; + +use crate::{Verified, event::kinds, messages::Draft, profiles}; +use nostr::prelude::EventId; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +/// One agent the viewer may share from channel settings. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ChannelAgent { + /// Agent identity whose owner attestation was verified in this read. + pub pubkey: String, + /// The agent's current profile name. + pub name: String, + /// The confirmed owner permission, rather than an optimistic local value. + pub enabled: bool, + /// Why sharing cannot be enabled; does not block revocation. + pub enable_reason: Option, + /// A journaled operation that still needs delivery or confirmation. + pub pending: Option, +} + +/// Durable delivery state retained when the settings panel closes. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PendingGrant { + /// The exact journaled event, reused on retry. + pub id: EventId, + /// The requested permission value. + pub enabled: bool, + /// Delivery failure or uncertainty retained by the outbox. + pub error: Option, + /// True only for proven non-delivery, which permits discarding this event. + pub failed: bool, +} + +/// A verified, fresh channel settings read. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct ChannelAgents { + /// Verified owned remote agents and owned agents with existing grants. + pub agents: Vec, +} + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub(crate) struct Grant { + pub version: u32, + pub agent: String, + pub channel: Uuid, + pub enabled: bool, +} + +impl Grant { + pub(crate) fn parse(event: &Verified) -> Result { + let grant: Self = + serde_json::from_str(&event.content).map_err(|_| "Invalid sharing permission")?; + if event.kind_u16() != kinds::AGENT_CHANNEL_GRANT + || grant.version != 1 + || !crate::event::is_hex64(&grant.agent) + || grant.agent == event.pubkey_hex() + { + return Err("Invalid sharing permission".into()); + } + for (name, expected) in [ + ("d", format!("{}:{}", grant.agent, grant.channel)), + ("h", grant.channel.to_string()), + ("p", grant.agent.clone()), + ] { + let tags = event.tags_named(name).collect::>(); + if tags.len() != 1 || tags[0] != [name, &expected] { + return Err("Ambiguous sharing permission".into()); + } + } + Ok(grant) + } + + pub(crate) fn draft(agent: &str, channel: &str, enabled: bool) -> Result { + let channel = Uuid::parse_str(channel).map_err(|_| "Invalid channel")?; + if !crate::event::is_hex64(agent) { + return Err("Invalid agent".into()); + } + let grant = Self { + version: 1, + agent: agent.into(), + channel, + enabled, + }; + Ok(Draft { + kind: kinds::AGENT_CHANNEL_GRANT, + tags: vec![ + vec!["d".into(), format!("{agent}:{channel}")], + vec!["h".into(), channel.to_string()], + vec!["p".into(), agent.into()], + ], + content: serde_json::to_string(&grant).map_err(|error| error.to_string())?, + }) + } +} + +/// Match the relay's current NIP-OA authentication-time ownership check. +pub(crate) fn owner(profile: &Verified, now: u64) -> Option { + if profile.kind_u16() != kinds::PROFILE { + return None; + } + let tags = profile + .tags_named("auth") + .map(<[String]>::to_vec) + .collect::>(); + profiles::attested_owner_of(&profile.pubkey_hex(), 22242, now, &tags) +} + +pub(crate) fn recovery_key(channel: &str, agent: &str) -> String { + format!("agent-share:{channel}:{agent}") +} + +/// Current runtime selection, never a filtered query that can expose an old head. +pub(crate) fn remote_listing(event: &Verified, viewer: &str) -> Option { + if event.kind_u16() != kinds::MANAGED_AGENT || event.pubkey_hex() != viewer { + return None; + } + let ids = event.tags_named("d").collect::>(); + let runtimes = event.tags_named("runtime").collect::>(); + if ids.len() != 1 + || ids[0].len() != 2 + || !crate::event::is_hex64(&ids[0][1]) + || runtimes.len() != 1 + || runtimes[0] != ["runtime", "voltage.remote-workspace/sandbox"] + { + return None; + } + Some(ids[0][1].clone()) +} + +pub(crate) fn remote_agents( + events: &[Verified], + viewer: &str, +) -> std::collections::BTreeSet { + let mut heads: std::collections::BTreeMap<&str, &Verified> = std::collections::BTreeMap::new(); + for event in events + .iter() + .filter(|e| e.kind_u16() == kinds::MANAGED_AGENT && e.pubkey_hex() == viewer) + { + let Some(id) = event.tag("d") else { + continue; + }; + if heads + .get(id) + .is_none_or(|old| crate::event::is_newer(event, old)) + { + heads.insert(id, event); + } + } + heads + .values() + .filter_map(|event| remote_listing(event, viewer)) + .collect() +} diff --git a/apps/buzz-gpui/crates/relay/src/agent_sharing/tests.rs b/apps/buzz-gpui/crates/relay/src/agent_sharing/tests.rs new file mode 100644 index 00000000000..49c10a6217e --- /dev/null +++ b/apps/buzz-gpui/crates/relay/src/agent_sharing/tests.rs @@ -0,0 +1,162 @@ +use super::*; +use buzz_identity::{Identity, Purpose}; +use nostr::prelude::{Event, EventBuilder, FinalizeEvent, Keys, Kind, Tag, Timestamp}; +use nostr_wire::JsonUtil as _; + +const CHANNEL: &str = "00000000-0000-4000-8000-000000000001"; + +fn signed(identity: &Identity, draft: Draft, at: u64) -> Event { + identity + .sign( + Purpose::Content, + identity + .template(Purpose::Content, draft.kind, at, draft.tags, draft.content) + .unwrap(), + ) + .unwrap() +} + +fn verified(event: Event) -> Verified { + crate::Verifier::default().verify(event).unwrap() +} + +#[test] +fn native_owner_grants_match_the_deployed_sdk_and_revoke_the_same_coordinate() { + let identity = Identity::ephemeral(); + let agent = Identity::ephemeral().pubkey_hex(); + for enabled in [true, false] { + let event = signed( + &identity, + Grant::draft(&agent, CHANNEL, enabled).unwrap(), + 100, + ); + let native = Grant::parse(&verified(event.clone())).unwrap(); + let wire = nostr_wire::Event::from_json(serde_json::to_string(&event).unwrap()).unwrap(); + let server = buzz_sdk::sharing::Grant::parse(&wire).unwrap(); + assert_eq!(native.enabled, enabled); + assert_eq!(server.enabled, enabled); + assert_eq!(server.agent.to_hex(), agent); + assert_eq!(server.channel.to_string(), CHANNEL); + assert_eq!(event.pubkey, identity.public_key()); + assert!(event.verify().is_ok()); + } +} + +#[test] +fn secure_signer_refuses_ambiguous_secret_bearing_or_self_grants() { + let identity = Identity::ephemeral(); + let agent = Identity::ephemeral().pubkey_hex(); + let valid = || Grant::draft(&agent, CHANNEL, true).unwrap(); + let mut cases = Vec::new(); + for name in ["d", "h", "p"] { + let mut draft = valid(); + draft.tags.push(vec![name.into(), "wrong".into()]); + cases.push(draft); + } + let mut draft = valid(); + let mut value: serde_json::Value = serde_json::from_str(&draft.content).unwrap(); + value["private_key"] = "not-a-permission".into(); + draft.content = value.to_string(); + cases.push(draft); + cases.push(Grant::draft(&identity.pubkey_hex(), CHANNEL, true).unwrap()); + for draft in cases { + assert!( + identity + .template( + Purpose::Content, + draft.kind, + 100, + draft.tags.clone(), + draft.content.clone() + ) + .is_err() + ); + // Callers can construct an unsigned event without using template(). + let unsigned = nostr::prelude::UnsignedEvent::new( + identity.public_key(), + Timestamp::from_secs(100), + Kind::from_u16(draft.kind), + draft.tags.into_iter().map(|tag| Tag::parse(tag).unwrap()), + draft.content, + ); + assert!(matches!( + identity.sign(Purpose::Content, unsigned), + Err(buzz_identity::IdentityError::NotAllowed) + )); + } +} + +#[test] +fn parser_checks_scope_even_for_validly_signed_events() { + let keys = Keys::generate(); + let agent = Identity::ephemeral().pubkey_hex(); + let draft = Grant::draft(&agent, CHANNEL, true).unwrap(); + let tags = draft + .tags + .into_iter() + .map(|t| Tag::parse(t).unwrap()) + .chain([Tag::custom("h", ["wrong"])]); + let event = EventBuilder::new(Kind::from_u16(kinds::AGENT_CHANNEL_GRANT), draft.content) + .tags(tags) + .finalize(&keys) + .unwrap(); + assert!(Grant::parse(&verified(event)).is_err()); +} + +#[test] +fn current_authentication_conditions_determine_ownership() { + let human = Identity::ephemeral(); + let keys = Keys::generate(); + let wire_human = nostr_wire::Keys::parse(human.nsec().as_str()).unwrap(); + let wire_agent = nostr_wire::PublicKey::from_hex(&keys.public_key().to_hex()).unwrap(); + for (conditions, allowed) in [ + ("", true), + ("kind=22242&created_at<101", true), + ("created_at<99", false), + ("kind=9", false), + ] { + let auth = + buzz_sdk::nip_oa::compute_auth_tag(&wire_human, &wire_agent, conditions).unwrap(); + let tag: Vec = serde_json::from_str(&auth).unwrap(); + let event = EventBuilder::new(Kind::Metadata, "{}") + .tags([Tag::parse(tag).unwrap()]) + .custom_created_at(Timestamp::from_secs(1)) + .finalize(&keys) + .unwrap(); + assert_eq!( + owner(&verified(event), 100).is_some(), + allowed, + "{conditions}" + ); + } +} + +#[test] +fn a_new_local_listing_disables_an_older_remote_listing() { + let human = Identity::ephemeral(); + let agent = Identity::ephemeral().pubkey_hex(); + let listing = |runtime: &str, at| { + verified(signed( + &human, + Draft { + kind: kinds::MANAGED_AGENT, + content: "{}".into(), + tags: vec![ + vec!["d".into(), agent.clone()], + vec!["runtime".into(), runtime.into()], + ], + }, + at, + )) + }; + let old = listing("voltage.remote-workspace/sandbox", 1); + let new = listing("buzz.codex/codex", 2); + assert!(remote_agents(&[new, old], &human.pubkey_hex()).is_empty()); + assert_eq!( + remote_agents( + &[listing("voltage.remote-workspace/sandbox", 3)], + &human.pubkey_hex() + ), + [agent].into() + ); +} diff --git a/apps/buzz-gpui/crates/relay/src/event.rs b/apps/buzz-gpui/crates/relay/src/event.rs index 09d0edaf7c7..2027acbe577 100644 --- a/apps/buzz-gpui/crates/relay/src/event.rs +++ b/apps/buzz-gpui/crates/relay/src/event.rs @@ -34,6 +34,8 @@ pub mod kinds { pub const CUSTOM_EMOJI: u16 = 30030; pub const APP_DATA: u16 = 30078; pub const MANAGED_AGENT: u16 = 30177; + /// Owner-signed permission to invoke one agent in one channel. + pub const AGENT_CHANNEL_GRANT: u16 = 30180; pub const REMINDERS: u16 = 30300; pub const USER_STATUS: u16 = 30315; pub const HIDDEN_DMS: u16 = 30622; diff --git a/apps/buzz-gpui/crates/relay/src/lib.rs b/apps/buzz-gpui/crates/relay/src/lib.rs index 99a0142fcb6..609eb087c8a 100644 --- a/apps/buzz-gpui/crates/relay/src/lib.rs +++ b/apps/buzz-gpui/crates/relay/src/lib.rs @@ -1,6 +1,7 @@ //! Client for Buzz relays. pub mod account; +pub mod agent_sharing; pub mod archive; pub mod audience; pub mod budget; diff --git a/apps/buzz-gpui/crates/relay/src/outbox.rs b/apps/buzz-gpui/crates/relay/src/outbox.rs index c112f68169b..6c1ca9efbd1 100644 --- a/apps/buzz-gpui/crates/relay/src/outbox.rs +++ b/apps/buzz-gpui/crates/relay/src/outbox.rs @@ -177,7 +177,8 @@ impl Operation { /// Confirmed, without caller recovery: moves to retained history. fn complete(&self) -> bool { - (self.kind() == kinds::PUT_USER && self.acknowledged) || self.delivery == Delivery::Seen + (matches!(self.kind(), kinds::PUT_USER | kinds::AGENT_CHANNEL_GRANT) && self.acknowledged) + || self.delivery == Delivery::Seen } } @@ -776,6 +777,25 @@ impl Outbox { /// Dates and identifies the draft, shows it as `sending`, and queues /// its delivery. Returns the final event ID. pub fn send(&self, draft: Draft, options: SendOptions) -> Result { + self.send_with_time(draft, options, 0) + } + + /// Keeps a replaceable permission newer than its confirmed head. + pub(crate) fn send_after( + &self, + draft: Draft, + options: SendOptions, + head: u64, + ) -> Result { + self.send_with_time(draft, options, head.saturating_add(1)) + } + + fn send_with_time( + &self, + draft: Draft, + options: SendOptions, + minimum_time: u64, + ) -> Result { if options.gate.as_ref().is_some_and(|gate| !gate()) { return Err(OutboxError::Cancelled); } @@ -831,7 +851,13 @@ impl Outbox { Some(channel) => inner.options.message_clock.next(channel, now_ms), None => now_ms, }; - let mut created_at = ms / 1000; + let mut created_at = (ms / 1000).max(minimum_time); + if created_at > ms / 1000 + MAX_EDIT_LEAD_SECS { + return Err(OutboxError::Refused( + "Changes are arriving too quickly or your clock changed. Wait a moment and retry." + .into(), + )); + } if draft.kind == kinds::EDIT { // Edits have second precision on every client. Keep this // device's successive edits of one target strictly increasing @@ -936,7 +962,8 @@ impl Outbox { return Ok(()); }; if state.closed - || (op.kind() == kinds::PUT_USER && op.acknowledged) + || (matches!(op.kind(), kinds::PUT_USER | kinds::AGENT_CHANNEL_GRANT) + && op.acknowledged) || is_workflow_operation(&op.event) || state.attempts.contains_key(id) || state.dismissing.contains_key(id) @@ -1026,6 +1053,30 @@ impl Outbox { self.inner.persist(Action::Acknowledge(*id)).await } + /// A newer signed head makes a pending grant permanently obsolete, even + /// when its HTTP acknowledgement was lost. Do not claim the old ID was seen. + pub(crate) async fn supersede_agent_grant( + &self, + id: &EventId, + head: &Verified, + ) -> Result<(), OutboxError> { + let Some(op) = self.get(id) else { + return Ok(()); + }; + if op.kind() != kinds::AGENT_CHANNEL_GRANT + || head.kind_u16() != op.kind() + || head.pubkey != op.event.pubkey + || head.tag("d") != op.tag("d") + || !(head.created_at_secs() > op.created_at() + || (head.created_at_secs() == op.created_at() && head.id < op.id())) + { + return Err(OutboxError::Refused( + "Sharing has not been superseded".into(), + )); + } + self.inner.persist(Action::Acknowledge(*id)).await + } + /// Attaches caller recovery to an existing operation, pending or /// confirmed, before the caller resumes it. pub async fn recover(&self, id: &EventId, recovery: Recovery) -> Result<(), OutboxError> { @@ -1212,7 +1263,8 @@ impl Inner { let mut cleared = (*current).clone(); cleared.recovery = None; cleared.acknowledged = true; - if (cleared.delivery == Delivery::Seen || cleared.kind() == kinds::PUT_USER) + if (cleared.delivery == Delivery::Seen + || matches!(cleared.kind(), kinds::PUT_USER | kinds::AGENT_CHANNEL_GRANT)) && !state.attempts.contains_key(&id) { state.remove_pending(&id); diff --git a/apps/buzz-gpui/crates/relay/src/outbox/agent_sharing_tests.rs b/apps/buzz-gpui/crates/relay/src/outbox/agent_sharing_tests.rs new file mode 100644 index 00000000000..b452ea3d62b --- /dev/null +++ b/apps/buzz-gpui/crates/relay/src/outbox/agent_sharing_tests.rs @@ -0,0 +1,136 @@ +use super::*; +use crate::agent_sharing::{Grant, recovery_key}; +use nostr_wire::JsonUtil as _; + +const CHANNEL: &str = "00000000-0000-4000-8000-000000000001"; + +fn options(agent: &str) -> SendOptions { + SendOptions { + recovery: Some(Recovery { + key: recovery_key(CHANNEL, agent), + value: "true".into(), + }), + ..SendOptions::default() + } +} + +#[tokio::test(start_paused = true)] +async fn agent_grant_clock_is_strictly_newer_and_native_bytes_match_server() { + let h = Harness::new(Mode::Accept); + h.outbox.ready().await.unwrap(); + let agent = Identity::ephemeral().pubkey_hex(); + let id = h + .outbox + .send_after( + Grant::draft(&agent, CHANNEL, false).unwrap(), + SendOptions::default(), + NOW, + ) + .unwrap(); + flush().await; + let event = h + .relay + .posted() + .into_iter() + .find(|event| event.id == id) + .unwrap(); + assert_eq!(event.created_at.as_secs(), NOW + 1); + let wire = nostr_wire::Event::from_json(serde_json::to_string(&event).unwrap()).unwrap(); + assert!(!buzz_sdk::sharing::Grant::parse(&wire).unwrap().enabled); + assert!( + h.outbox + .send_after( + Grant::draft(&agent, CHANNEL, true).unwrap(), + SendOptions::default(), + NOW + 60 + ) + .is_err() + ); +} + +#[tokio::test(start_paused = true)] +async fn agent_grant_lost_ack_survives_reopen_and_retries_the_same_event() { + let h = Harness::new(Mode::Lose); + h.outbox.ready().await.unwrap(); + let agent = Identity::ephemeral().pubkey_hex(); + let id = h + .outbox + .send_after( + Grant::draft(&agent, CHANNEL, true).unwrap(), + options(&agent), + 0, + ) + .unwrap(); + flush().await; + assert_eq!(h.delivery(&id), Delivery::Unknown); + let next = h.reopen(Mode::Accept); + next.outbox.ready().await.unwrap(); + flush().await; + assert!(next.relay.posted().is_empty()); + assert!( + next.outbox + .send_after( + Grant::draft(&agent, CHANNEL, false).unwrap(), + options(&agent), + NOW + ) + .is_err() + ); + next.outbox.retry(&id, None).unwrap(); + flush().await; + let sent = next.relay.posted(); + assert_eq!(sent.len(), 1); + assert_eq!(sent[0].id, id); + next.outbox.observe(&[verified(&sent[0])]); + next.outbox.acknowledge(&id).await.unwrap(); + assert!(next.op(&id).recovery.is_none()); +} + +#[tokio::test(start_paused = true)] +async fn newer_permission_releases_old_intent_without_claiming_it_was_seen() { + let h = Harness::new(Mode::Lose); + h.outbox.ready().await.unwrap(); + let agent = Identity::ephemeral().pubkey_hex(); + let id = h + .outbox + .send_after( + Grant::draft(&agent, CHANNEL, true).unwrap(), + options(&agent), + 0, + ) + .unwrap(); + flush().await; + let draft = Grant::draft(&agent, CHANNEL, false).unwrap(); + let unsigned = h + .identity + .template( + Purpose::Content, + draft.kind, + NOW + 1, + draft.tags, + draft.content, + ) + .unwrap(); + let head = verified(&h.identity.sign(Purpose::Content, unsigned).unwrap()); + let wrong = Identity::ephemeral(); + let draft = Grant::draft(&agent, CHANNEL, false).unwrap(); + let unsigned = wrong + .template( + Purpose::Content, + draft.kind, + NOW + 2, + draft.tags, + draft.content, + ) + .unwrap(); + let forged = verified(&wrong.sign(Purpose::Content, unsigned).unwrap()); + assert!(h.outbox.supersede_agent_grant(&id, &forged).await.is_err()); + h.outbox.supersede_agent_grant(&id, &head).await.unwrap(); + assert!(h.op(&id).acknowledged); + assert_eq!(h.delivery(&id), Delivery::Unknown); + assert!(h.op(&id).recovery.is_none()); + let count = h.relay.posted().len(); + h.outbox.retry(&id, None).unwrap(); + flush().await; + assert_eq!(h.relay.posted().len(), count); +} diff --git a/apps/buzz-gpui/crates/relay/src/outbox/tests.rs b/apps/buzz-gpui/crates/relay/src/outbox/tests.rs index 9c8519df0e9..10bc5afb6b9 100644 --- a/apps/buzz-gpui/crates/relay/src/outbox/tests.rs +++ b/apps/buzz-gpui/crates/relay/src/outbox/tests.rs @@ -3,6 +3,8 @@ //! Everything runs on paused tokio time. use super::*; +#[path = "agent_sharing_tests.rs"] +mod agent_sharing_tests; use crate::event::Verifier; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; diff --git a/apps/buzz-gpui/crates/relay/src/reader.rs b/apps/buzz-gpui/crates/relay/src/reader.rs index 3e2b5763589..4467d75f1ab 100644 --- a/apps/buzz-gpui/crates/relay/src/reader.rs +++ b/apps/buzz-gpui/crates/relay/src/reader.rs @@ -797,7 +797,7 @@ mod scheduling_tests { assert_eq!(fake.peak.load(SeqCst), 1); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn deadline_counts_queue_time() { let fake = Fake::new(false); let reader = spawn_reader( diff --git a/apps/buzz-gpui/crates/relay/src/session/agent_sharing.rs b/apps/buzz-gpui/crates/relay/src/session/agent_sharing.rs new file mode 100644 index 00000000000..8b0ab4dcfa3 --- /dev/null +++ b/apps/buzz-gpui/crates/relay/src/session/agent_sharing.rs @@ -0,0 +1,391 @@ +//! Sharing uses the session's verified reads and durable outbox, never a +//! private key export or a local preference presented as relay permission. + +#[cfg(test)] +#[path = "agent_sharing_tests.rs"] +mod tests; + +use super::Session; +use super::actions::{Context, Ops, read}; +use super::util::now; +use crate::agent_sharing::{ + ChannelAgent, ChannelAgents, Grant, PendingGrant, owner, recovery_key, remote_agents, +}; +use crate::event::{Verified, is_newer, kinds}; +use crate::lifecycle::{self, CONFIRM_DELAYS, ChannelState, StateType}; +use crate::messages::Draft; +use crate::outbox::{Delivery, Operation, Recovery, SendOptions}; +use crate::socket::filter; +use nostr::prelude::EventId; +use serde_json::json; +use std::collections::BTreeMap; + +const LIMIT: usize = 500; +const PENDING: &str = + "The sharing change is saved but not confirmed. Refresh its status or retry the saved change."; + +impl Session { + /// Fresh, relay-authoritative sharing for the viewer's remote agents. + pub async fn channel_agent_sharing(&self, channel: &str) -> Result { + let (ops, channel) = (self.ops(), channel.to_owned()); + self.run(async move { + let ctx = ops.context().await?; + let result = load(&ctx, &channel).await?; + check_epoch(&ops, ctx.epoch).await?; + Ok(result) + }) + .await + } + + /// One owner action publishes one signed, journaled permission event. + pub async fn set_channel_agent_sharing( + &self, + channel: &str, + agent: &str, + enabled: bool, + ) -> Result<(), String> { + let (ops, channel, agent) = (self.ops(), channel.to_owned(), agent.to_owned()); + self.run(async move { + let ctx = ops.context().await?; + let Some(change) = prepare(&ctx, &channel, &agent, enabled).await? else { + return Ok(()); + }; + check_epoch(&ops, ctx.epoch).await?; + let id = ctx + .outbox + .send_after(change.draft, change.options, change.after) + .map_err(|e| e.to_string())?; + confirm(&ctx, &channel, &agent, id).await + }) + .await + } + + /// Retry the exact saved event, after rechecking its current authority. + pub async fn retry_channel_agent_sharing(&self, id: EventId) -> Result<(), String> { + let ops = self.ops(); + self.run(async move { + let ctx = ops.context().await?; + let op = ctx + .outbox + .get(&id) + .ok_or("The sharing change is no longer pending")?; + let grant: Grant = serde_json::from_str(&op.event.content) + .map_err(|_| "Invalid saved sharing change")?; + if op.kind() != kinds::AGENT_CHANNEL_GRANT || op.event.pubkey.to_hex() != ctx.viewer { + return Err("Invalid saved sharing change".into()); + } + let channel = grant.channel.to_string(); + preflight(&ctx, &channel, &grant.agent, grant.enabled).await?; + check_epoch(&ops, ctx.epoch).await?; + ctx.outbox.retry(&id, None).map_err(|e| e.to_string())?; + confirm(&ctx, &channel, &grant.agent, id).await + }) + .await + } + + /// Discard only a sharing event proven not to have been delivered. + pub async fn discard_failed_agent_sharing(&self, id: EventId) -> Result<(), String> { + let ops = self.ops(); + self.run(async move { + let ctx = ops.context().await?; + if let Some(op) = ctx.outbox.get(&id) { + if op.kind() != kinds::AGENT_CHANNEL_GRANT || op.delivery != Delivery::Failed { + return Err(PENDING.into()); + } + ctx.outbox.dismiss(&id).await.map_err(|e| e.to_string())?; + } + Ok(()) + }) + .await + } +} + +struct Change { + draft: Draft, + options: SendOptions, + after: u64, +} + +fn pending_for(op: &Operation, channel: &str, agent: &str) -> bool { + op.kind() == kinds::AGENT_CHANNEL_GRANT + && op.tag("h") == Some(channel) + && op.tag("p") == Some(agent) + && op.recovery.is_some() +} + +async fn prepare( + ctx: &Context, + channel: &str, + agent: &str, + enabled: bool, +) -> Result, String> { + ctx.outbox.ready().await.map_err(|e| e.to_string())?; + preflight(ctx, channel, agent, enabled).await?; + let heads = grants(ctx, channel).await?; + settle(ctx, &heads).await?; + let head = heads.get(agent); + let pending = ctx.outbox.pending(); + let pending = pending + .iter() + .filter(|op| pending_for(op, channel, agent)) + .collect::>(); + if !pending.is_empty() { + // An uncertain enable may be cancelled, but never replaced with a new + // enable. A pending revoke must be reconciled or retried byte-for-byte. + if enabled || pending.iter().any(|op| !pending_grant(op).enabled) { + return Err(PENDING.into()); + } + } else if head + .map(Grant::parse) + .transpose()? + .is_some_and(|g| g.enabled == enabled) + { + return Ok(None); + } + let after = pending + .iter() + .map(|op| op.created_at()) + .chain(head.map(Verified::created_at_secs)) + .max() + .unwrap_or(0); + let key = recovery_key(channel, agent); + Ok(Some(Change { + draft: Grant::draft(agent, channel, enabled)?, + options: SendOptions { + // Keep the old journal until a strictly newer signed revoke is + // observed. The separate key permits this one cancellation write. + recovery: Some(Recovery { + key: if enabled { + key + } else { + format!("{key}:revoke") + }, + value: enabled.to_string(), + }), + ..SendOptions::default() + }, + after, + })) +} + +async fn check_epoch(ops: &Ops, epoch: u64) -> Result<(), String> { + if ops.with(move |actor| actor.epoch == epoch).await? { + Ok(()) + } else { + Err("Community access changed. Refresh sharing settings.".into()) + } +} + +async fn state(ctx: &Context, channel: &str) -> Result { + lifecycle::channel_id(channel).map_err(|e| e.to_string())?; + ChannelState::parse( + &read(ctx, lifecycle::state_filters(&ctx.relay, channel)).await?, + channel, + &ctx.relay, + ) + .map_err(|e| e.to_string()) +} + +async fn listings(ctx: &Context) -> Result, String> { + let events=read(ctx,vec![filter(json!({"kinds":[kinds::MANAGED_AGENT],"authors":[ctx.viewer],"consistency":"strong","limit":LIMIT}))]).await?; + if events.len() >= LIMIT { + return Err("Too many agent records to verify sharing. Contact support.".into()); + } + Ok(events) +} + +async fn grants(ctx: &Context, channel: &str) -> Result, String> { + let events=read(ctx,vec![filter(json!({"kinds":[kinds::AGENT_CHANNEL_GRANT],"authors":[ctx.viewer],"#h":[channel],"consistency":"strong","limit":LIMIT}))]).await?; + if events.len() >= LIMIT { + return Err("Too many sharing records to verify safely.".into()); + } + let mut heads: BTreeMap = BTreeMap::new(); + for event in events { + if event.pubkey_hex() != ctx.viewer { + return Err("Unexpected sharing author".into()); + } + let grant = Grant::parse(&event)?; + if grant.channel.to_string() != channel { + return Err("Unexpected sharing channel".into()); + } + if heads + .get(&grant.agent) + .is_none_or(|old| is_newer(&event, old)) + { + heads.insert(grant.agent, event); + } + } + Ok(heads) +} + +async fn profile(ctx: &Context, agent: &str) -> Result { + let events = read( + ctx, + vec![filter( + json!({"kinds":[kinds::PROFILE],"authors":[agent],"consistency":"strong","limit":1}), + )], + ) + .await?; + let event = crate::account::newest_profile(&events, agent) + .ok_or("The agent's current owner could not be verified")?; + if owner(event, now()).as_deref() != Some(&ctx.viewer) { + return Err("Only this agent's current owner can change sharing".into()); + } + Ok(event.clone()) +} + +async fn preflight(ctx: &Context, channel: &str, agent: &str, enabled: bool) -> Result<(), String> { + Grant::draft(agent, channel, enabled)?; + profile(ctx, agent).await?; + if enabled { + let current = state(ctx, channel).await?; + if current.channel_type == StateType::Dm + || current.archived + || !current.roles.contains_key(&ctx.viewer) + || !current.roles.contains_key(agent) + { + return Err( + "Sharing needs an active channel with both you and the agent as members.".into(), + ); + } + if !remote_agents(&listings(ctx).await?, &ctx.viewer).contains(agent) { + return Err("Move this agent to Remote workspace before sharing it.".into()); + } + } + Ok(()) +} + +async fn settle(ctx: &Context, heads: &BTreeMap) -> Result<(), String> { + ctx.outbox.ready().await.map_err(|e| e.to_string())?; + for op in ctx + .outbox + .pending() + .iter() + .filter(|op| op.kind() == kinds::AGENT_CHANNEL_GRANT && op.recovery.is_some()) + { + let Some(agent) = op.tag("p") else { + continue; + }; + if let Some(head) = heads.get(agent) { + if head.id == op.id() { + ctx.outbox.observe(std::slice::from_ref(head)); + ctx.outbox + .acknowledge(&op.id()) + .await + .map_err(|e| e.to_string())?; + } else if head.tag("d") == op.tag("d") + && (head.created_at_secs() > op.created_at() + || (head.created_at_secs() == op.created_at() && head.id < op.id())) + { + ctx.outbox + .supersede_agent_grant(&op.id(), head) + .await + .map_err(|e| e.to_string())?; + } + } + } + Ok(()) +} + +async fn load(ctx: &Context, channel: &str) -> Result { + let current = state(ctx, channel).await?; + if current.channel_type == StateType::Dm { + return Err("Direct messages do not share agents.".into()); + } + let heads = grants(ctx, channel).await?; + settle(ctx, &heads).await?; + let remote = remote_agents(&listings(ctx).await?, &ctx.viewer); + let pending = ctx.outbox.pending(); + let mut keys = remote + .iter() + .chain(heads.keys()) + .cloned() + .collect::>(); + // A runtime change must not hide the only way to reconcile a saved change. + keys.extend(pending.iter().filter_map(|op| { + (op.kind() == kinds::AGENT_CHANNEL_GRANT && op.tag("h") == Some(channel)) + .then(|| op.tag("p").map(str::to_owned)) + .flatten() + })); + if keys.len() > 128 { + return Err("Too many agents to verify in one settings panel.".into()); + } + let profiles = if keys.is_empty() { + Vec::new() + } else { + read(ctx, vec![filter(json!({"kinds":[kinds::PROFILE],"authors":keys,"consistency":"strong","limit":LIMIT}))]).await? + }; + let mut agents = Vec::new(); + for agent in keys { + let Some(event) = crate::account::newest_profile(&profiles, &agent) else { + continue; + }; + if owner(event, now()).as_deref() != Some(&ctx.viewer) { + continue; + } + let shown = crate::profiles::Profile::from_event(event); + let enabled = heads + .get(&agent) + .map(Grant::parse) + .transpose()? + .is_some_and(|g| g.enabled); + let enable_reason = if current.archived { + Some("This channel is archived.".into()) + } else if !current.roles.contains_key(&ctx.viewer) || !current.roles.contains_key(&agent) { + Some("Add yourself and this agent in Members before sharing.".into()) + } else if !remote.contains(&agent) { + Some("Move this agent to Remote workspace before sharing.".into()) + } else { + None + }; + let op = pending + .iter() + .filter(|op| pending_for(op, channel, &agent)) + .max_by_key(|op| op.created_at()); + agents.push(ChannelAgent { + pubkey: agent, + name: shown.name, + enabled, + enable_reason, + pending: op.map(|op| pending_grant(op)), + }); + } + agents.sort_by(|a, b| a.name.cmp(&b.name).then(a.pubkey.cmp(&b.pubkey))); + Ok(ChannelAgents { agents }) +} + +fn pending_grant(op: &Operation) -> PendingGrant { + PendingGrant { + id: op.id(), + enabled: serde_json::from_str::(&op.event.content).is_ok_and(|g| g.enabled), + error: op.error.clone(), + failed: op.delivery == Delivery::Failed, + } +} + +async fn confirm(ctx: &Context, channel: &str, agent: &str, id: EventId) -> Result<(), String> { + let mut changes = ctx.outbox.changes(); + loop { + match ctx.outbox.get(&id).map(|op| op.delivery) { + Some(Delivery::Accepted | Delivery::Seen) => break, + Some(Delivery::Failed) => { + return Err(ctx + .outbox + .get(&id) + .and_then(|op| op.error.clone()) + .unwrap_or_else(|| "Sharing was refused.".into())); + } + Some(Delivery::Unknown) | None => return Err(PENDING.into()), + Some(Delivery::Sending) => changes.changed().await.map_err(|_| PENDING.to_owned())?, + } + } + for delay in CONFIRM_DELAYS { + tokio::time::sleep(delay).await; + let heads = grants(ctx, channel).await?; + settle(ctx, &heads).await?; + if heads.get(agent).is_some_and(|head| head.id == id) { + return Ok(()); + } + } + Err(PENDING.into()) +} diff --git a/apps/buzz-gpui/crates/relay/src/session/agent_sharing_tests.rs b/apps/buzz-gpui/crates/relay/src/session/agent_sharing_tests.rs new file mode 100644 index 00000000000..55cb1ac2152 --- /dev/null +++ b/apps/buzz-gpui/crates/relay/src/session/agent_sharing_tests.rs @@ -0,0 +1,423 @@ +use super::*; +use crate::outbox::{MemoryJournal, Outbox, OutboxOptions, Publisher}; +use crate::reader::{Priority, Reader, ReaderOptions, Source}; +use crate::{RelayHttp, RelayOrigin, Verifier}; +use buzz_identity::{Identity, Purpose}; +use futures::future::BoxFuture; +use nostr::prelude::{Event, EventBuilder, FinalizeEvent, Keys, Kind, Tag, Timestamp}; +use serde_json::Value; +use std::sync::{Arc, Mutex}; + +const CHANNEL: &str = "00000000-0000-4000-8000-000000000001"; + +#[derive(Default)] +struct SourceState { + events: Vec, + fail: bool, + fail_publish: bool, + posts: usize, +} + +#[derive(Clone, Default)] +struct Fake(Arc>); + +impl Source for Fake { + fn query( + &self, + filters: Vec, + _: Priority, + ) -> BoxFuture<'static, Result, crate::HttpError>> { + let state = self.0.lock().unwrap(); + let result = if state.fail { + Err(crate::HttpError::Transport("offline".into())) + } else { + let found = state + .events + .iter() + .filter(|event| { + filters.iter().any(|filter| { + assert_eq!( + filter.get("consistency").and_then(Value::as_str), + Some("strong") + ); + let kinds = filter["kinds"].as_array().unwrap(); + let authors = filter["authors"].as_array().unwrap(); + kinds + .iter() + .any(|k| k.as_u64() == Some(event.kind.as_u16() as u64)) + && authors + .iter() + .any(|a| a.as_str() == Some(event.pubkey.to_hex().as_str())) + && ["h", "d"].iter().all(|name| { + filter.get(&format!("#{name}")).is_none_or(|values| { + values.as_array().unwrap().iter().any(|v| { + event.tags.iter().any(|tag| { + tag.as_slice().first().map(String::as_str) + == Some(*name) + && tag.as_slice().get(1).map(String::as_str) + == v.as_str() + }) + }) + }) + }) + }) + }) + .map(|event| serde_json::to_value(event).unwrap()) + .collect(); + Ok(found) + }; + Box::pin(async move { result }) + } +} + +impl Publisher for Fake { + fn post_event(&self, event: &Event) -> BoxFuture<'static, Result, crate::HttpError>> { + let id = event.id.to_hex(); + let mut state = self.0.lock().unwrap(); + state.posts += 1; + let result = if state.fail_publish { + Err(crate::HttpError::Transport("delivery uncertain".into())) + } else { + state.events.push(event.clone()); + Ok(serde_json::to_vec(&json!({"event_id":id,"accepted":true,"message":""})).unwrap()) + }; + Box::pin(async move { result }) + } + fn find_event(&self, _: &Event) -> BoxFuture<'static, Result> { + Box::pin(async { Ok(false) }) + } + fn read_echo(&self, _: EventId, _: Priority) -> BoxFuture<'static, Vec> { + Box::pin(async { Vec::new() }) + } +} + +struct Fixture { + ctx: Context, + source: Fake, + agent: Keys, + relay: Keys, +} + +impl Fixture { + fn new() -> Self { + let identity = Arc::new(Identity::ephemeral()); + let source = Fake::default(); + let relay = Keys::generate(); + let agent = Keys::generate(); + let origin = RelayOrigin::parse("http://127.0.0.1:9").unwrap(); + let reader = Reader::spawn( + &tokio::runtime::Handle::current(), + Arc::new(source.clone()), + Arc::new(Verifier::default()), + ReaderOptions::default(), + ); + let ctx = Context { + reader, + http: RelayHttp::new(origin, identity.clone()), + identity: identity.clone(), + viewer: identity.pubkey_hex(), + relay: relay.public_key().to_hex(), + epoch: 1, + outbox: Outbox::new( + identity, + Arc::new(source.clone()), + Arc::new(MemoryJournal::new()), + OutboxOptions::default(), + ), + }; + let fixture = Self { + ctx, + source, + agent, + relay, + }; + fixture.channel(false, true); + fixture.attest(&fixture.ctx.identity); + fixture.listing("voltage.remote-workspace/sandbox"); + fixture + } + + fn push(&self, event: Event) { + self.source.0.lock().unwrap().events.push(event); + } + + fn remove_kind(&self, kind: u16) { + self.source + .0 + .lock() + .unwrap() + .events + .retain(|e| e.kind.as_u16() != kind); + } + + fn event(&self, keys: &Keys, kind: u16, tags: Vec>, content: String) -> Event { + EventBuilder::new(Kind::from_u16(kind), content) + .tags(tags.into_iter().map(|t| Tag::parse(t).unwrap())) + .custom_created_at(Timestamp::from_secs(now())) + .finalize(keys) + .unwrap() + } + + fn channel(&self, archived: bool, member: bool) { + for kind in [ + kinds::CHANNEL_METADATA, + kinds::CHANNEL_ADMINS, + kinds::CHANNEL_MEMBERS, + ] { + self.remove_kind(kind); + } + let mut meta = vec![ + vec!["d".into(), CHANNEL.into()], + vec!["t".into(), "stream".into()], + vec!["name".into(), "engineering".into()], + vec!["public".into()], + ]; + if archived { + meta.push(vec!["archived".into(), "true".into()]); + } + self.push(self.event(&self.relay, kinds::CHANNEL_METADATA, meta, String::new())); + self.push(self.event( + &self.relay, + kinds::CHANNEL_ADMINS, + vec![ + vec!["d".into(), CHANNEL.into()], + vec!["p".into(), self.ctx.viewer.clone(), "owner".into()], + ], + String::new(), + )); + let mut roster = vec![ + vec!["d".into(), CHANNEL.into()], + vec![ + "p".into(), + self.ctx.viewer.clone(), + String::new(), + "owner".into(), + ], + ]; + if member { + roster.push(vec![ + "p".into(), + self.agent.public_key().to_hex(), + String::new(), + "bot".into(), + ]); + } + self.push(self.event(&self.relay, kinds::CHANNEL_MEMBERS, roster, String::new())); + } + + fn attest(&self, owner: &Identity) { + self.remove_kind(kinds::PROFILE); + let auth: Vec = serde_json::from_str( + &owner + .attest_agent(&self.agent.public_key().to_hex()) + .unwrap(), + ) + .unwrap(); + self.push(self.event( + &self.agent, + kinds::PROFILE, + vec![auth], + json!({"name":"Brokkr"}).to_string(), + )); + } + + fn listing(&self, runtime: &str) { + self.remove_kind(kinds::MANAGED_AGENT); + let tags = vec![ + vec!["d".into(), self.agent.public_key().to_hex()], + vec!["runtime".into(), runtime.into()], + ]; + let unsigned = self + .ctx + .identity + .template( + Purpose::Content, + kinds::MANAGED_AGENT, + now(), + tags, + "{}".into(), + ) + .unwrap(); + self.push(self.ctx.identity.sign(Purpose::Content, unsigned).unwrap()); + } + + fn grant(&self, enabled: bool) { + let draft = Grant::draft(&self.agent.public_key().to_hex(), CHANNEL, enabled).unwrap(); + let unsigned = self + .ctx + .identity + .template( + Purpose::Content, + draft.kind, + now(), + draft.tags, + draft.content, + ) + .unwrap(); + self.push(self.ctx.identity.sign(Purpose::Content, unsigned).unwrap()); + } +} + +#[tokio::test] +async fn settings_read_uses_signed_membership_ownership_listing_and_permission() { + let f = Fixture::new(); + let off = load(&f.ctx, CHANNEL).await.unwrap(); + assert_eq!(off.agents.len(), 1); + assert!(!off.agents[0].enabled); + assert!(off.agents[0].enable_reason.is_none()); + f.grant(true); + assert!(load(&f.ctx, CHANNEL).await.unwrap().agents[0].enabled); + f.attest(&Identity::ephemeral()); + assert!(load(&f.ctx, CHANNEL).await.unwrap().agents.is_empty()); + assert!( + preflight(&f.ctx, CHANNEL, &f.agent.public_key().to_hex(), false) + .await + .is_err() + ); +} + +#[tokio::test] +async fn archive_removed_agent_and_runtime_change_block_enable_but_preserve_revoke() { + let f = Fixture::new(); + let agent = f.agent.public_key().to_hex(); + assert!(preflight(&f.ctx, CHANNEL, &agent, true).await.is_ok()); + f.grant(true); + f.channel(true, true); + let state = load(&f.ctx, CHANNEL).await.unwrap(); + assert!(state.agents[0].enabled && state.agents[0].enable_reason.is_some()); + assert!(preflight(&f.ctx, CHANNEL, &agent, true).await.is_err()); + assert!(preflight(&f.ctx, CHANNEL, &agent, false).await.is_ok()); + f.channel(false, false); + assert!(preflight(&f.ctx, CHANNEL, &agent, true).await.is_err()); + assert!(preflight(&f.ctx, CHANNEL, &agent, false).await.is_ok()); + f.channel(false, true); + f.listing("buzz.codex/codex"); + assert!(preflight(&f.ctx, CHANNEL, &agent, true).await.is_err()); + assert!(preflight(&f.ctx, CHANNEL, &agent, false).await.is_ok()); +} + +#[tokio::test] +async fn failed_reads_never_turn_existing_permission_into_an_empty_success() { + let f = Fixture::new(); + f.grant(true); + assert!(load(&f.ctx, CHANNEL).await.unwrap().agents[0].enabled); + f.source.0.lock().unwrap().fail = true; + assert!(load(&f.ctx, CHANNEL).await.is_err()); +} + +#[tokio::test] +async fn changing_runtime_does_not_hide_a_saved_sharing_change() { + let f = Fixture::new(); + f.ctx.outbox.ready().await.unwrap(); + let agent = f.agent.public_key().to_hex(); + let id = f + .ctx + .outbox + .send_after( + Grant::draft(&agent, CHANNEL, true).unwrap(), + SendOptions { + recovery: Some(Recovery { + key: recovery_key(CHANNEL, &agent), + value: "true".into(), + }), + ..SendOptions::default() + }, + 0, + ) + .unwrap(); + let mut changes = f.ctx.outbox.changes(); + tokio::time::timeout(std::time::Duration::from_secs(3), async { + while f.ctx.outbox.get(&id).unwrap().delivery == Delivery::Sending { + changes.changed().await.unwrap(); + } + }) + .await + .unwrap(); + f.remove_kind(kinds::AGENT_CHANNEL_GRANT); + f.listing("buzz.codex/codex"); + let state = load(&f.ctx, CHANNEL).await.unwrap(); + assert_eq!(state.agents.len(), 1); + assert_eq!(state.agents[0].pending.as_ref().unwrap().id, id); + assert!(state.agents[0].enable_reason.is_some()); +} + +async fn delivery(f: &Fixture, id: EventId) -> Delivery { + let mut changes = f.ctx.outbox.changes(); + tokio::time::timeout(std::time::Duration::from_secs(3), async { + loop { + let delivery = f.ctx.outbox.get(&id).unwrap().delivery; + if delivery != Delivery::Sending { + return delivery; + } + changes.changed().await.unwrap(); + } + }) + .await + .unwrap() +} + +#[tokio::test] +async fn explicit_revoke_cancels_an_uncertain_enable_after_authority_changes() { + for case in ["archive", "removed", "local"] { + let f = Fixture::new(); + let agent = f.agent.public_key().to_hex(); + if case != "removed" { + f.grant(false); + } + f.source.0.lock().unwrap().fail_publish = true; + let change = prepare(&f.ctx, CHANNEL, &agent, true) + .await + .unwrap() + .unwrap(); + let enable = f + .ctx + .outbox + .send_after(change.draft, change.options, change.after) + .unwrap(); + assert_eq!(delivery(&f, enable).await, Delivery::Unknown); + match case { + "archive" => f.channel(true, true), + "removed" => f.channel(false, false), + _ => f.listing("buzz.codex/codex"), + } + assert!(preflight(&f.ctx, CHANNEL, &agent, true).await.is_err()); + assert!(prepare(&f.ctx, CHANNEL, &agent, true).await.is_err()); + assert_eq!(f.source.0.lock().unwrap().posts, 1); + let change = prepare(&f.ctx, CHANNEL, &agent, false) + .await + .unwrap() + .unwrap(); + let revoke = f + .ctx + .outbox + .send_after(change.draft, change.options, change.after) + .unwrap(); + assert_eq!(delivery(&f, revoke).await, Delivery::Unknown); + assert!( + f.ctx.outbox.get(&revoke).unwrap().created_at() + > f.ctx.outbox.get(&enable).unwrap().created_at() + ); + assert!(f.ctx.outbox.get(&enable).unwrap().recovery.is_some()); + let state = load(&f.ctx, CHANNEL).await.unwrap(); + let shown = state.agents[0].pending.as_ref().unwrap(); + assert_eq!(shown.id, revoke); + assert!(!shown.enabled); + assert!(prepare(&f.ctx, CHANNEL, &agent, false).await.is_err()); + f.source.0.lock().unwrap().fail_publish = false; + f.ctx.outbox.retry(&revoke, None).unwrap(); + assert_eq!(delivery(&f, revoke).await, Delivery::Accepted); + confirm(&f.ctx, CHANNEL, &agent, revoke).await.unwrap(); + assert!(!load(&f.ctx, CHANNEL).await.unwrap().agents[0].enabled); + assert!(f.ctx.outbox.get(&enable).unwrap().acknowledged); + assert!( + !f.ctx + .outbox + .pending() + .iter() + .any(|op| pending_for(op, CHANNEL, &agent)) + ); + let posts = f.source.0.lock().unwrap().posts; + f.ctx.outbox.retry(&enable, None).unwrap(); + assert_eq!(f.source.0.lock().unwrap().posts, posts); + } +} diff --git a/apps/buzz-gpui/crates/relay/src/session/mod.rs b/apps/buzz-gpui/crates/relay/src/session/mod.rs index d5d3fe1bfa3..cfd4297e188 100644 --- a/apps/buzz-gpui/crates/relay/src/session/mod.rs +++ b/apps/buzz-gpui/crates/relay/src/session/mod.rs @@ -17,6 +17,7 @@ mod account; mod actions; mod activity; mod admins; +mod agent_sharing; mod canvas; mod channels; mod control; diff --git a/apps/buzz-gpui/crates/relay/tests/channel_agent_sharing.rs b/apps/buzz-gpui/crates/relay/tests/channel_agent_sharing.rs new file mode 100644 index 00000000000..4bc67733503 --- /dev/null +++ b/apps/buzz-gpui/crates/relay/tests/channel_agent_sharing.rs @@ -0,0 +1,231 @@ +//! Native settings operations against an isolated relay with the same grant +//! protocol as deployment. Uses the local sharing fixture's synthetic owners. + +use buzz_identity::{Identity, Purpose}; +use buzz_relay_client::lifecycle::{Action, NewChannel}; +use buzz_relay_client::{RelayHttp, RelayOrigin, RosterStatus, Session, SessionStatus, Store}; +use nostr::nips::nip19::ToBech32; +use nostr::prelude::{Event, Keys}; +use serde_json::json; +use std::sync::Arc; +use std::time::Duration; + +fn identity(byte: &str) -> Arc { + let keys = Keys::parse(&byte.repeat(32)).unwrap(); + Arc::new(Identity::from_nsec(&keys.secret_key().to_bech32().unwrap()).unwrap()) +} + +async fn ready(origin: RelayOrigin, identity: Arc) -> Arc { + let session = Arc::new(Session::open( + &tokio::runtime::Handle::current(), + origin, + identity, + Store::in_memory().unwrap(), + )); + let mut snapshot = session.snapshot(); + tokio::time::timeout(Duration::from_secs(15), async { + loop { + match &snapshot.borrow().status { + SessionStatus::Ready + if snapshot.borrow().live + && matches!(snapshot.borrow().roster, RosterStatus::Ready) => + { + return; + } + SessionStatus::Failed(error) => panic!("Native session failed: {error}"), + SessionStatus::Connecting | SessionStatus::Ready => {} + } + snapshot.changed().await.unwrap(); + } + }) + .await + .unwrap(); + session +} + +async fn publish(http: &RelayHttp, event: Event) { + let response = http + .post("/events", serde_json::to_string(&event).unwrap()) + .await + .unwrap(); + let receipt: serde_json::Value = serde_json::from_slice(&response).unwrap(); + assert_eq!(receipt["accepted"], true, "{receipt}"); +} + +#[tokio::test] +#[ignore = "requires the isolated sharing relay and its synthetic membership fixture"] +async fn channel_settings_share_revoke_reopen_and_archived_channel() { + let relay = + std::env::var("BUZZ_CHANNEL_SHARING_RELAY").expect("set loopback sharing relay URL"); + let url = url::Url::parse(&relay).unwrap(); + assert_eq!(url.host_str(), Some("127.0.0.1")); + assert_eq!(url.scheme(), "http"); + let origin = RelayOrigin::parse(&relay).unwrap(); + let owner = identity("33"); + let agent = identity("22").pubkey_hex(); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let tags = vec![ + vec!["d".into(), agent.clone()], + vec!["runtime".into(), "voltage.remote-workspace/sandbox".into()], + ]; + let record = owner + .template( + Purpose::Content, + 30177, + now, + tags, + json!({"name":"Brokkr","parallelism":1,"respond_to":"owner-only"}).to_string(), + ) + .unwrap(); + publish( + &RelayHttp::new(origin.clone(), owner.clone()), + owner.sign(Purpose::Content, record).unwrap(), + ) + .await; + let session = ready(origin.clone(), owner.clone()).await; + let channel = session + .create_channel(NewChannel { + name: format!( + "sharing-settings-{}", + &uuid::Uuid::new_v4().to_string()[..8] + ), + description: "Native channel sharing settings acceptance".into(), + private: false, + ttl: None, + labels: Vec::new(), + }) + .await + .unwrap(); + // Channel creation confirms the relay state before the sidebar scan has + // necessarily adopted it. The native Members action needs that projection. + session.refresh_roster(); + let mut channels = session.snapshot(); + tokio::time::timeout(Duration::from_secs(15), async { + loop { + if channels + .borrow() + .channels + .iter() + .any(|row| row.id == channel && row.members.contains(&owner.pubkey_hex())) + { + break; + } + channels.changed().await.unwrap(); + } + }) + .await + .unwrap(); + session.add_agent_member(&channel, &agent).await.unwrap(); + let snapshot = session.channel_agent_sharing(&channel).await.unwrap(); + assert!( + snapshot + .agents + .iter() + .any(|row| row.pubkey == agent && !row.enabled && row.enable_reason.is_none()) + ); + session + .set_channel_agent_sharing(&channel, &agent, true) + .await + .unwrap(); + assert!( + session + .channel_agent_sharing(&channel) + .await + .unwrap() + .agents + .iter() + .find(|row| row.pubkey == agent) + .unwrap() + .enabled + ); + // An immediate second action must beat the first event's second-level time. + session + .set_channel_agent_sharing(&channel, &agent, false) + .await + .unwrap(); + assert!( + !session + .channel_agent_sharing(&channel) + .await + .unwrap() + .agents + .iter() + .find(|row| row.pubkey == agent) + .unwrap() + .enabled + ); + let reopened = ready(origin.clone(), owner.clone()).await; + assert!( + !reopened + .channel_agent_sharing(&channel) + .await + .unwrap() + .agents + .iter() + .find(|row| row.pubkey == agent) + .unwrap() + .enabled + ); + let other = identity("44"); + session + .add_member(&channel, &other.pubkey_hex()) + .await + .unwrap(); + let outsider = ready(origin, other).await; + assert!( + outsider + .set_channel_agent_sharing(&channel, &agent, true) + .await + .is_err() + ); + session + .set_channel_agent_sharing(&channel, &agent, true) + .await + .unwrap(); + session + .channel_action(Action::Archive, &channel) + .await + .unwrap(); + let archived = session.channel_agent_sharing(&channel).await.unwrap(); + let row = archived + .agents + .iter() + .find(|row| row.pubkey == agent) + .unwrap(); + assert!(row.enabled && row.enable_reason.is_some()); + session + .set_channel_agent_sharing(&channel, &agent, false) + .await + .unwrap(); + assert!( + !session + .channel_agent_sharing(&channel) + .await + .unwrap() + .agents + .iter() + .find(|row| row.pubkey == agent) + .unwrap() + .enabled + ); + session + .channel_action(Action::Unarchive, &channel) + .await + .unwrap(); + if let Ok(path) = std::env::var("BUZZ_CHANNEL_SHARING_UI_FIXTURE") { + assert!(path.starts_with("/tmp/")); + let value = json!({"relay":relay,"channel":channel,"owner":owner.pubkey_hex(),"owner_nsec":owner.nsec().as_str(),"agent":agent}); + std::fs::write(&path, value.to_string()).unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).unwrap(); + } + } + println!( + "Native sharing settings passed: owner grant, immediate revoke, reopened state, non-owner denial, archived revoke." + ); +}