diff --git a/crates/buzz-cli/src/commands/sandbox.rs b/crates/buzz-cli/src/commands/sandbox.rs index 5dbc233c503..89ca3b36467 100644 --- a/crates/buzz-cli/src/commands/sandbox.rs +++ b/crates/buzz-cli/src/commands/sandbox.rs @@ -8,7 +8,7 @@ use clap::Subcommand; use nostr::{Event, EventId, PublicKey, Timestamp}; use serde::Deserialize; use serde_json::{json, Value}; -use tokio::time::{sleep, timeout, Instant}; +use tokio::time::{sleep, timeout}; use uuid::Uuid; use crate::{client::BuzzClient, error::CliError}; diff --git a/crates/buzz-cli/src/commands/sandbox/terminal.rs b/crates/buzz-cli/src/commands/sandbox/terminal.rs index ba68391c901..3b63998d947 100644 --- a/crates/buzz-cli/src/commands/sandbox/terminal.rs +++ b/crates/buzz-cli/src/commands/sandbox/terminal.rs @@ -7,6 +7,7 @@ use std::os::unix::fs::OpenOptionsExt; use buzz_sdk::sandbox_attach::{Flow, Frame, Packet, Rate}; use buzz_ws_client::{NostrWsConnection, RelayMessage, WsClientError}; use tokio::io::unix::AsyncFd; +use tokio::time::Instant; use super::*; diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index 99b4aa68032..91d198ba8e1 100644 --- a/crates/buzz-core/src/lib.rs +++ b/crates/buzz-core/src/lib.rs @@ -86,3 +86,5 @@ pub mod test_helpers { pub mod artifact; /// NIP-AR sync filter parsing and limits. pub mod artifact_sync; +/// Canonical single-use sandbox attachment acceptance coordinates. +pub mod sandbox_attach; diff --git a/crates/buzz-core/src/sandbox_attach.rs b/crates/buzz-core/src/sandbox_attach.rs new file mode 100644 index 00000000000..4159c46b6f1 --- /dev/null +++ b/crates/buzz-core/src/sandbox_attach.rs @@ -0,0 +1,64 @@ +//! Coordinates whose first accepted event permanently consumes an attachment. + +use nostr::Event; +use uuid::Uuid; + +/// Parse a canonical terminal or snapshot acceptance coordinate. Requests, +/// approvals and end records retain ordinary replacement/deletion semantics. +pub fn parse_acceptance_coordinate(value: &str) -> Option<(&str, Uuid)> { + let mut parts = value.split(':'); + let mode = parts.next()?; + if !matches!(mode, "terminal" | "snapshot") { + return None; + } + let text = parts.next()?; + let session = Uuid::parse_str(text).ok()?; + if session.to_string() != text || parts.next()? != "accepted" || parts.next().is_some() { + return None; + } + Some((mode, session)) +} + +/// Identify an unambiguous kind30624 acceptance coordinate. This classifies a +/// stored envelope; the relay separately verifies its signed audit metadata. +pub fn acceptance_coordinate(event: &Event) -> Option<(&str, Uuid)> { + if u32::from(event.kind.as_u16()) != crate::kind::KIND_SANDBOX_ATTACH { + return None; + } + let mut tags = event + .tags + .iter() + .map(|tag| tag.as_slice()) + .filter(|tag| tag.first().is_some_and(|name| name == "d")); + let [_, value] = tags.next()? else { + return None; + }; + if tags.next().is_some() { + return None; + } + parse_acceptance_coordinate(value) +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn only_canonical_acceptance_coordinates_are_burns() { + let session: Uuid = "abcdefab-1234-4321-8765-abcdefabcdef".parse().unwrap(); + for mode in ["terminal", "snapshot"] { + assert_eq!( + parse_acceptance_coordinate(&format!("{mode}:{session}:accepted")), + Some((mode, session)) + ); + for value in [ + format!("{mode}:{session}:grant"), + format!("{mode}:{session}:request"), + format!("{mode}:{session}:ended"), + format!("{mode}:{session}:accepted:extra"), + format!("{mode}:{}:accepted", session.to_string().to_uppercase()), + ] { + assert!(parse_acceptance_coordinate(&value).is_none(), "{value}"); + } + } + } +} diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index ffc56214bd0..c4844ed808b 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -31,6 +31,9 @@ pub enum ParameterizedReplaceStatus { pub enum ParameterizedReplacePrecondition<'a> { /// Apply normal NIP-33 ordering without a revision precondition. Unconditional, + /// Insert only when this coordinate has never been accepted, including + /// after deletion of its event body. A watermark durably retains the burn. + ExpectNoHead, /// Require the live head to match this validated event ID. ExpectedRevision(&'a [u8]), /// Accept only an exact live-head replay and perform no mutation otherwise. @@ -111,6 +114,22 @@ async fn replace_parameterized_event_in_transaction_impl( ) -> Result { let community_id = tx.community(); let kind_i32 = buzz_core::kind::event_kind_i32(event); + let acceptance = buzz_core::sandbox_attach::acceptance_coordinate(event); + if acceptance.is_some_and(|(mode, session)| d_tag != format!("{mode}:{session}:accepted")) { + return Err(DbError::InvalidData( + "attachment acceptance coordinate mismatch".into(), + )); + } + let acceptance = acceptance.is_some(); + // This kind's acceptance cannot opt out through another caller or an + // omitted client tag. Requests and approvals use ordinary NIP-33 ordering. + let precondition = if acceptance { + ParameterizedReplacePrecondition::ExpectNoHead + } else { + precondition + }; + let retain_watermark = + acceptance || precondition == ParameterizedReplacePrecondition::ExpectNoHead; let pubkey_bytes = event.pubkey.to_bytes(); let created_at_secs = event.created_at.as_secs() as i64; let created_at = DateTime::from_timestamp(created_at_secs, 0) @@ -177,7 +196,7 @@ async fn replace_parameterized_event_in_transaction_impl( .bind(d_tag) .fetch_optional(tx.conn()) .await?; - let watermark: Option<(DateTime, Vec)> = if is_nip_rs { + let watermark: Option<(DateTime, Vec)> = if is_nip_rs || retain_watermark { sqlx::query_as( "SELECT created_at, event_id FROM parameterized_event_watermarks \ WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4", @@ -217,6 +236,17 @@ async fn replace_parameterized_event_in_transaction_impl( )); } + if precondition == ParameterizedReplacePrecondition::ExpectNoHead + && (existing.is_some() || watermark.is_some()) + { + return Ok(ParameterizedReplaceResult::new( + event, + received_at, + channel_id, + ParameterizedReplaceStatus::RevisionMismatch, + )); + } + if let ParameterizedReplacePrecondition::ExpectedRevision(expected_revision) = precondition { let status = match existing.as_ref() { None => Some(ParameterizedReplaceStatus::RevisionMissing), @@ -330,7 +360,7 @@ async fn replace_parameterized_event_in_transaction_impl( return Ok(false); } - if is_nip_rs { + if is_nip_rs || retain_watermark { sqlx::query( "INSERT INTO parameterized_event_watermarks \ (community_id, kind, pubkey, d_tag, created_at, event_id) \ diff --git a/crates/buzz-relay/src/handlers/attach_claim.rs b/crates/buzz-relay/src/handlers/attach_claim.rs new file mode 100644 index 00000000000..8552d98a1b8 --- /dev/null +++ b/crates/buzz-relay/src/handlers/attach_claim.rs @@ -0,0 +1,184 @@ +//! Agent-authored first acceptance of an owner-approved attachment. + +use buzz_core::tenant::TenantContext; +use buzz_sdk::sandbox_attach::Scope; +use nostr::{Event, PublicKey}; +use serde::Deserialize; +use uuid::Uuid; + +use super::ingest::IngestError; +use crate::state::AppState; + +#[derive(Deserialize)] +#[serde(tag = "mode", rename_all = "lowercase", deny_unknown_fields)] +enum Audit { + Terminal { + version: u8, + session: Uuid, + status: String, + grant: Event, + bytes_in: u64, + bytes_out: u64, + }, + Snapshot { + version: u8, + session: Uuid, + status: String, + grant: Event, + bytes: u64, + reason: Option, + }, +} + +fn one<'a>(event: &'a Event, name: &str) -> Result<&'a str, String> { + let mut tags = event + .tags + .iter() + .map(|tag| tag.as_slice()) + .filter(|tag| tag.first().is_some_and(|key| key == name)); + let Some([_, value]) = tags.next() else { + return Err("missing or malformed acceptance tag".into()); + }; + if tags.next().is_some() { + return Err("ambiguous acceptance tag".into()); + } + Ok(value) +} + +fn metadata(event: &Event) -> Result, String> { + if u32::from(event.kind.as_u16()) != buzz_core::kind::KIND_SANDBOX_ATTACH { + return Ok(None); + } + let candidate = event.tags.iter().any(|tag| { + tag.as_slice().first().is_some_and(|key| key == "d") + && tag + .as_slice() + .get(1) + .is_some_and(|value| value.ends_with(":accepted")) + }) || serde_json::from_str::(&event.content) + .ok() + .is_some_and(|value| value["status"] == "accepted"); + if !candidate { + return Ok(None); + } + let invalid = || "invalid attachment acceptance metadata".to_owned(); + let (mode, session) = + buzz_core::sandbox_attach::acceptance_coordinate(event).ok_or_else(invalid)?; + if event.content.len() > 8192 + || event.tags.iter().any(|tag| { + tag.as_slice() + .first() + .is_some_and(|key| key == "expiration") + }) + { + return Err("attachment acceptance must be bounded and have no expiration".into()); + } + let audit: Audit = serde_json::from_str(&event.content).map_err(|_| invalid())?; + let (version, body_mode, body_session, status, grant, empty) = match audit { + Audit::Terminal { + version, + session, + status, + grant, + bytes_in, + bytes_out, + } => ( + version, + "terminal", + session, + status, + grant, + bytes_in == 0 && bytes_out == 0, + ), + Audit::Snapshot { + version, + session, + status, + grant, + bytes, + reason, + } => ( + version, + "snapshot", + session, + status, + grant, + bytes == 0 && reason.is_none(), + ), + }; + if version != 1 + || body_mode != mode + || body_session != session + || status != "accepted" + || !empty + { + return Err(invalid()); + } + let nonce = one(event, "buzz:attach-claim")?; + if Uuid::parse_str(nonce).map_err(|_| invalid())?.to_string() != nonce { + return Err(invalid()); + } + let body: serde_json::Value = serde_json::from_str(&grant.content).map_err(|_| invalid())?; + let scope: Scope = serde_json::from_value(body["scope"].clone()).map_err(|_| invalid())?; + let (approved_session, requester) = if mode == "terminal" { + let approved = + buzz_sdk::sandbox_attach::Grant::parse(&grant, scope, event.created_at.as_secs()) + .map_err(|_| invalid())?; + (approved.session(), approved.requester()) + } else { + let approved = buzz_sdk::sandbox_attach::snapshot::Grant::parse( + &grant, + scope, + event.created_at.as_secs(), + ) + .map_err(|_| invalid())?; + (approved.session(), approved.requester()) + }; + if approved_session != session + || event.pubkey != scope.agent + || grant.pubkey == scope.agent + || one(&grant, "h")? != scope.channel.to_string() + || one(&grant, "e")? != scope.thread.to_hex() + || one(&grant, "p")? != scope.agent.to_hex() + || one(event, "h")? != scope.channel.to_string() + || one(event, "e")? != scope.thread.to_hex() + || one(event, "p")? != requester.to_hex() + { + return Err(invalid()); + } + Ok(Some(grant.pubkey)) +} + +/// Ordinary requests/grants remain mutable and revocable. An acceptance is +/// signed only by the recorded agent and retains the owner's signed approval. +pub(super) async fn validate( + tenant: &TenantContext, + event: &Event, + state: &AppState, +) -> Result<(), IngestError> { + let Some(owner) = + metadata(event).map_err(|error| IngestError::Rejected(format!("invalid: {error}")))? + else { + return Ok(()); + }; + let owned = state + .db + .is_agent_owner( + tenant.community(), + event.pubkey.as_bytes(), + owner.as_bytes(), + ) + .await + .map_err(|error| { + IngestError::Internal(format!("attachment owner lookup failed: {}", error.kind())) + })?; + if !owned { + return Err(IngestError::Rejected( + "restricted: only the approved agent may claim attachment acceptance".into(), + )); + } + Ok(()) +} + +#[cfg(test)] +mod postgres_tests; diff --git a/crates/buzz-relay/src/handlers/attach_claim/postgres_tests.rs b/crates/buzz-relay/src/handlers/attach_claim/postgres_tests.rs new file mode 100644 index 00000000000..72d492af691 --- /dev/null +++ b/crates/buzz-relay/src/handlers/attach_claim/postgres_tests.rs @@ -0,0 +1,324 @@ +use std::sync::Arc; + +use buzz_core::{channel::MemberRole, TenantContext}; +use buzz_db::EventQuery; +use buzz_sdk::sandbox_attach::{self as attach, snapshot, Scope}; +use nostr::{Event, EventBuilder, EventId, Keys, Kind, Tag, Timestamp}; +use uuid::Uuid; + +use crate::handlers::group_state_postgres_tests::{community, create_channel, http_auth, state}; +use crate::handlers::ingest::ingest_event; +use crate::state::AppState; + +struct Fixture { + state: Arc, + tenant: TenantContext, + owner: Keys, + agent: Keys, + scope: Scope, +} +impl Fixture { + async fn new() -> Self { + let state = state().await; + let tenant = community(&state, "attach-claims").await; + let (owner, agent) = (Keys::generate(), Keys::generate()); + for key in [&owner, &agent] { + state + .db + .ensure_user(tenant.community(), key.public_key().as_bytes()) + .await + .unwrap(); + } + state + .db + .set_agent_owner( + tenant.community(), + agent.public_key().as_bytes(), + owner.public_key().as_bytes(), + ) + .await + .unwrap(); + let channel = create_channel(&state, &tenant, &owner, &[]).await; + state + .db + .add_member( + tenant.community(), + channel, + agent.public_key().as_bytes(), + MemberRole::Member, + Some(owner.public_key().as_bytes()), + ) + .await + .unwrap(); + let scope = Scope { + channel, + thread: EventId::all_zeros(), + agent: agent.public_key(), + }; + Self { + state, + tenant, + owner, + agent, + scope, + } + } + + fn events(&self, mode: &str) -> (Event, Event, Event, Event) { + let now = Timestamp::now().as_secs(); + let session = Uuid::new_v4(); + if mode == "terminal" { + let request = attach::request(&self.owner, self.scope, session, now + 600).unwrap(); + let pending = attach::Pending::parse(&request, now).unwrap(); + let approval = attach::approve(&self.owner, &pending, now + 600).unwrap(); + let grant = attach::Grant::parse(&approval, self.scope, now).unwrap(); + ( + request, + approval, + attach::audit(&self.agent, &grant, "accepted", 0, 0).unwrap(), + attach::audit(&self.agent, &grant, "accepted", 0, 0).unwrap(), + ) + } else { + let request = snapshot::request(&self.owner, self.scope, session, now + 600).unwrap(); + let pending = snapshot::Pending::parse(&request, now).unwrap(); + let approval = snapshot::approve(&self.owner, &pending, now + 600).unwrap(); + let grant = snapshot::Grant::parse(&approval, self.scope, now).unwrap(); + ( + request, + approval, + snapshot::audit(&self.agent, &grant, "accepted", 0, None).unwrap(), + snapshot::audit(&self.agent, &grant, "accepted", 0, None).unwrap(), + ) + } + } + + async fn publish(&self, event: Event, signer: &Keys) -> super::super::ingest::IngestResult { + ingest_event(&self.state, &self.tenant, event, http_auth(signer)) + .await + .unwrap() + } + + async fn head(&self, event: &Event) -> Vec { + self.state + .db + .query_events(&EventQuery { + kinds: Some(vec![30624]), + authors: Some(vec![event.pubkey.to_bytes().to_vec()]), + d_tag: Some(d(event)), + limit: Some(1), + ..EventQuery::for_community(self.tenant.community()) + }) + .await + .unwrap() + } +} + +fn d(event: &Event) -> String { + event + .tags + .iter() + .find(|tag| tag.as_slice()[0] == "d") + .unwrap() + .as_slice()[1] + .clone() +} + +fn rebuild(event: &Event, keys: &Keys, tags: Vec, time: u64) -> Event { + EventBuilder::new(event.kind, event.content.clone()) + .tags(tags) + .allow_self_tagging() + .custom_created_at(Timestamp::from(time)) + .sign_with_keys(keys) + .unwrap() +} + +/// This drives the production ingest transaction on two concurrent writers. +/// A replacement/deletion/metadata purge cannot reopen either capability; the +/// per-write first-claim marker is never emitted for an exact replay or loser. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn attachment_acceptance_is_atomic_durable_and_narrowly_immutable() { + let f = Fixture::new().await; + for mode in ["terminal", "snapshot"] { + let (request, approval, first, second) = f.events(mode); + f.publish(request.clone(), &f.owner).await; + f.publish(approval.clone(), &f.owner).await; + assert_ne!( + first.id, second.id, + "same-second attempts have separate claim IDs" + ); + let (a, b) = tokio::join!( + ingest_event(&f.state, &f.tenant, first.clone(), http_auth(&f.agent)), + ingest_event(&f.state, &f.tenant, second.clone(), http_auth(&f.agent)), + ); + let (a, b) = (a.unwrap(), b.unwrap()); + assert_eq!( + [&a, &b] + .into_iter() + .filter(|reply| reply.message == attach::CLAIM_OK_MESSAGE) + .count(), + 1 + ); + let winner = if a.message == attach::CLAIM_OK_MESSAGE { + first + } else { + second + }; + assert_eq!(f.head(&winner).await[0].event.id, winner.id); + let replay = f.publish(winner.clone(), &f.agent).await; + assert!(replay.message.starts_with("duplicate")); + assert_ne!(replay.message, attach::CLAIM_OK_MESSAGE); + + for (kind, signer, tags) in [ + (5, &f.agent, vec![Tag::event(winner.id)]), + ( + 5, + &f.agent, + vec![Tag::parse([ + "a", + &format!("30624:{}:{}", f.agent.public_key(), d(&winner)), + ]) + .unwrap()], + ), + ( + 9005, + &f.owner, + vec![ + Tag::parse(["h", &f.scope.channel.to_string()]).unwrap(), + Tag::event(winner.id), + ], + ), + ] { + let deletion = EventBuilder::new(Kind::Custom(kind), "") + .tags(tags) + .sign_with_keys(signer) + .unwrap(); + f.publish(deletion, signer).await; + assert_eq!( + f.head(&winner).await[0].event.id, + winner.id, + "accepted burn survived kind{kind}" + ); + } + // Request and owner approval still revoke through ordinary NIP-09. + for target in [request.id, approval.id] { + let revoke = EventBuilder::new(Kind::Custom(5), "") + .tag(Tag::event(target)) + .sign_with_keys(&f.owner) + .unwrap(); + f.publish(revoke, &f.owner).await; + } + assert!(f.head(&request).await.is_empty()); + assert!(f.head(&approval).await.is_empty()); + + // Trusted operator retention may remove the audit body. The separate + // first-write watermark still denies a never-submitted, newer claim. + sqlx::query("DELETE FROM events WHERE community_id=$1 AND id=$2") + .bind(f.tenant.community().as_uuid()) + .bind(winner.id.as_bytes().as_slice()) + .execute(f.state.db.pool()) + .await + .unwrap(); + assert!(f.head(&winner).await.is_empty()); + let later = rebuild( + &winner, + &f.agent, + winner.tags.clone().to_vec(), + winner.created_at.as_secs() + 1, + ); + let blocked = f.publish(later, &f.agent).await; + assert_ne!(blocked.message, attach::CLAIM_OK_MESSAGE); + assert!( + f.head(&winner).await.is_empty(), + "metadata purge reopened consumed approval" + ); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn malformed_or_unowned_acceptance_never_gets_a_claim_receipt() { + let f = Fixture::new().await; + for mode in ["terminal", "snapshot"] { + let (_, _, audit, _) = f.events(mode); + let mut expiry = audit.tags.clone().to_vec(); + expiry.push(Tag::expiration( + Timestamp::now() + std::time::Duration::from_secs(10), + )); + let mut ambiguous = audit.tags.clone().to_vec(); + ambiguous.push(Tag::parse(["d", "unrelated"]).unwrap()); + let mut missing_nonce = audit.tags.clone().to_vec(); + missing_nonce.retain(|tag| tag.as_slice()[0] != "buzz:attach-claim"); + let mut wrong_thread = audit.tags.clone().to_vec(); + for tag in &mut wrong_thread { + if tag.as_slice()[0] == "e" { + *tag = Tag::parse(["e", &"22".repeat(32)]).unwrap(); + } + } + for event in [ + rebuild(&audit, &f.agent, wrong_thread, audit.created_at.as_secs()), + rebuild(&audit, &f.agent, expiry, audit.created_at.as_secs()), + rebuild(&audit, &f.agent, ambiguous, audit.created_at.as_secs()), + rebuild(&audit, &f.agent, missing_nonce, audit.created_at.as_secs()), + rebuild( + &audit, + &f.owner, + audit.tags.clone().to_vec(), + audit.created_at.as_secs(), + ), + ] { + assert!(ingest_event( + &f.state, + &f.tenant, + event.clone(), + http_auth(if event.pubkey == f.owner.public_key() { + &f.owner + } else { + &f.agent + }) + ) + .await + .is_err()); + assert!(f.head(&audit).await.is_empty()); + } + // An ordinary posting member cannot claim to be an agent merely by + // obtaining an owner-signed envelope that names its key. + let outsider = Keys::generate(); + f.state + .db + .ensure_user(f.tenant.community(), outsider.public_key().as_bytes()) + .await + .unwrap(); + f.state + .db + .add_member( + f.tenant.community(), + f.scope.channel, + outsider.public_key().as_bytes(), + MemberRole::Member, + Some(f.owner.public_key().as_bytes()), + ) + .await + .unwrap(); + let fake = Fixture { + state: f.state.clone(), + tenant: f.tenant.clone(), + owner: f.owner.clone(), + scope: Scope { + agent: outsider.public_key(), + ..f.scope + }, + agent: outsider, + }; + let (_, _, audit, _) = fake.events(mode); + assert!(ingest_event( + &fake.state, + &fake.tenant, + audit.clone(), + http_auth(&fake.agent) + ) + .await + .is_err()); + assert!(fake.head(&audit).await.is_empty()); + } +} diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index a3af129d3c1..d8452f57303 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -1884,6 +1884,18 @@ mod tests { (community_b, agent_bytes.clone(), owner_bytes.clone()), false, ); + // A failed owner check now tries gateway authority too. Keep this + // no-database test on the production cached-denial path in B. + state.observer_gateway_cache.insert( + (community_b, agent_bytes.clone(), owner_bytes.clone()), + super::GatewayVerdict { + authorized: false, + generation: state + .gateway_generation + .load(std::sync::atomic::Ordering::Acquire), + valid_until: u64::MAX, + }, + ); let encrypted = encrypt_observer_payload( &agent, diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 57f98bdb7eb..2af94c15277 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -2830,6 +2830,7 @@ async fn ingest_event_inner( // Consumers verify every head themselves; ingest only refuses heads that // could never be live or revoke, so a coordinate always holds a parseable // head. + super::attach_claim::validate(tenant, &event, state).await?; if kind_u32 == buzz_core::kind::KIND_AGENT_GATEWAY_DELEGATION { let (delegation, _) = buzz_sdk::gateway::Delegation::parse(&event) .map_err(|_| IngestError::Rejected("invalid: malformed gateway delegation".into()))?; @@ -3777,6 +3778,10 @@ async fn ingest_event_inner( }); } + if buzz_core::sandbox_attach::acceptance_coordinate(&event).is_some() { + message = buzz_sdk::sandbox_attach::CLAIM_OK_MESSAGE.into(); + } + // Fence observer gateway verdicts computed before this committed head. if kind_u32 == buzz_core::kind::KIND_AGENT_GATEWAY_DELEGATION { state diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 6e5c45cbf91..ba7a04ad2cb 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -7,6 +7,8 @@ mod agent_grant; pub mod artifact_sync; #[cfg(test)] mod artifact_sync_postgres_tests; +/// Atomic owner-approved terminal/snapshot acceptance. +mod attach_claim; pub mod auth; /// Pure NIP-29 channel membership-authority decisions (kinds 9000/9001/9022). pub mod channel_authz; diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 36417de40b0..2e04dbcc557 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -2660,6 +2660,10 @@ async fn plan_group_deletion_target( } _ => {} // Same channel — OK } + if buzz_core::sandbox_attach::acceptance_coordinate(&target.event).is_some() { + tracing::debug!("NIP-29 deletion ignored for attachment acceptance"); + return Ok(Vec::new()); + } Ok(vec![ PlannedDeletionTarget::load(tenant, state, target_id, target).await?, ]) @@ -3356,6 +3360,11 @@ async fn handle_a_tag_deletion( buzz_core::kind::KIND_AGENT_GATEWAY_DELEGATION => { tracing::debug!(d_tag, "NIP-09 deletion ignored for gateway delegation"); } + buzz_core::kind::KIND_SANDBOX_ATTACH + if buzz_core::sandbox_attach::parse_acceptance_coordinate(d_tag).is_some() => + { + tracing::debug!("NIP-09 deletion ignored for attachment acceptance"); + } buzz_core::kind::KIND_WORKFLOW_DEF => { // Workflow deletion belongs to the atomic persistence path in ingest. return Err(anyhow::anyhow!( @@ -3560,6 +3569,10 @@ async fn plan_standard_deletion_targets( ); continue; } + if buzz_core::sandbox_attach::acceptance_coordinate(&target.event).is_some() { + tracing::debug!("NIP-09 deletion ignored for attachment acceptance"); + continue; + } planned.push(PlannedDeletionTarget::load(tenant, state, target_id, target).await?); } Ok(planned) diff --git a/crates/buzz-sdk/src/sandbox_attach.rs b/crates/buzz-sdk/src/sandbox_attach.rs index b8447b05354..094230c25cd 100644 --- a/crates/buzz-sdk/src/sandbox_attach.rs +++ b/crates/buzz-sdk/src/sandbox_attach.rs @@ -33,6 +33,9 @@ pub const RETRY_MS: u64 = 1000; pub const MAX_RETRIES: u8 = 5; /// Largest frame rate, including retries and acknowledgements, per direction. pub const MAX_FRAMES_PER_SECOND: u16 = 64; +/// Only a relay transaction that inserted the first durable acceptance emits +/// this receipt. Generic success or duplicate receipts never authorize Open. +pub const CLAIM_OK_MESSAGE: &str = "sandbox-attach: first-claim-v1"; /// Frame admission budget. Apply before cryptography; backpressure sends and drop excess input. #[derive(Default)] @@ -375,6 +378,9 @@ pub fn audit( "d".into(), format!("terminal:{}:{status}", grant.session), ]); + if status == "accepted" { + values.push(vec!["buzz:attach-claim".into(), Uuid::new_v4().to_string()]); + } EventBuilder::new(Kind::from(kind::KIND_SANDBOX_ATTACH as u16), content) .tags(tags(values)?) .sign_with_keys(keys) diff --git a/crates/buzz-sdk/src/sandbox_attach/tests.rs b/crates/buzz-sdk/src/sandbox_attach/tests.rs index 175ad2a4f58..61e71c6be30 100644 --- a/crates/buzz-sdk/src/sandbox_attach/tests.rs +++ b/crates/buzz-sdk/src/sandbox_attach/tests.rs @@ -290,3 +290,24 @@ fn terminal_frame_rate_is_bounded_including_acknowledgements() { assert!(!rate.allow(999)); assert!(rate.allow(1000)); } + +#[test] +fn accepted_audits_have_unique_claims_without_expiration() { + let (owner, agent) = (Keys::generate(), Keys::generate()); + let scope = scope(&agent); + let now = Timestamp::now().as_secs(); + let pending = Pending::parse( + &request(&owner, scope, Uuid::new_v4(), now + 60).unwrap(), + now, + ) + .unwrap(); + let grant = Grant::parse(&approve(&owner, &pending, now + 60).unwrap(), scope, now).unwrap(); + let first = audit(&agent, &grant, "accepted", 0, 0).unwrap(); + let second = audit(&agent, &grant, "accepted", 0, 0).unwrap(); + let a = tag(&first, "buzz:attach-claim").unwrap(); + let b = tag(&second, "buzz:attach-claim").unwrap(); + assert_ne!(a, b); + assert_eq!(Uuid::parse_str(a).unwrap().to_string(), a); + assert_ne!(first.id, second.id); + assert!(tag(&first, "expiration").is_none()); +} diff --git a/docs/nips/NIP-SA.md b/docs/nips/NIP-SA.md index 5e6c19bee9b..e9c655256c9 100644 --- a/docs/nips/NIP-SA.md +++ b/docs/nips/NIP-SA.md @@ -17,6 +17,7 @@ Owner, agent, and requester must remain posting members in a strongly fetched re A gateway requester must remain a live delegate throughout creation and retrieval. Consumers fail closed on failed reads, expiry, deletion, replacement, or lost membership. The same atomic acceptance burn as terminal attachment makes each approval single-use. +Consumers require the relay's [atomic first-claim receipt](../sandbox-attachment-claims.md) before starting work. Only metadata is stored in plaintext: the approval, acceptance/end status, byte count, and optional failure category. Failure categories are `scan`, `budget`, `archive`, `upload`, or `authority`. diff --git a/docs/nips/NIP-TA.md b/docs/nips/NIP-TA.md index 73670014bfa..682849dca92 100644 --- a/docs/nips/NIP-TA.md +++ b/docs/nips/NIP-TA.md @@ -29,6 +29,7 @@ Missing, replaced, deleted, expired, or malformed records and read failures deny Consumers repeat those checks during attachment and close when authorization ends. The owner revokes an individual approval through NIP-09 deletion; the requester may delete its pending request. Durable acceptance makes each approval single-use; a new PTY requires a new request and owner approval. +Consumers require the relay's [atomic first-claim receipt](../sandbox-attachment-claims.md) before starting work. The acceptance and end records contain the signed metadata-only owner approval and byte counters. No keystrokes or file bytes belong in stored events. diff --git a/docs/sandbox-attachment-claims.md b/docs/sandbox-attachment-claims.md new file mode 100644 index 00000000000..f6849faf0b1 --- /dev/null +++ b/docs/sandbox-attachment-claims.md @@ -0,0 +1,50 @@ +# Single-use sandbox attachment approvals + +Terminal and snapshot approvals require a durable first acceptance before work starts. +The relay serializes acceptance at the existing addressable-event coordinate lock. +The acceptance coordinate is `terminal::accepted` or `snapshot::accepted`. +The UUID must use its canonical lowercase representation. + +The agent signs the acceptance and embeds the owner's signed, scoped approval. +The relay verifies the typed approval, immutable scope, agent ownership, and unique claim nonce. +The nonce is a canonical UUID in `buzz:attach-claim`. +Each attempt creates one event and retains that exact event for retries. +Acceptance byte counters start at zero, and acceptance events cannot expire. + +| Relay receipt | Consumer action | +| --- | --- | +| Accepted, matching event ID, message `sandbox-attach: first-claim-v1` | This attempt may start its approved operation. | +| Duplicate, ordinary success, rejection, timeout, or mismatched event ID | Refuse to start; request a new owner approval. | + +Only the transaction that inserts the first acceptance receives the first-claim message. +Exact event retries never receive that message. +An older relay's generic success therefore cannot authorize a new consumer. +Capability discovery alone cannot establish that a particular write was atomic. +After an ambiguous receipt, consumers must fail closed even when a later query finds their event. + +The existing `parameterized_event_watermarks` table retains the consumed coordinate in the same transaction. +Deleting an audit body does not reopen its coordinate. +NIP-09 event or address deletion and NIP-29 message deletion preserve acceptance records. +Pending requests and owner approvals retain ordinary deletion and replacement behavior. +Consumers must still check their current approval, pending request, membership, and delegation before and during work. + +Acceptance and end events contain signed approval metadata and byte counters. +They must never contain keystrokes, terminal output, or snapshot file bytes. +Operators may remove event bodies through trusted retention procedures; the consumed-coordinate watermark must remain. +Community deletion purges the existing tenant-scoped watermark table through the normal deletion path. +Trusted operator removal of that watermark resets this replay boundary and requires retiring the affected approval. +This protocol does not prevent an operator with direct database access from resetting authority state. + +Run the production-ingest regression against a disposable PostgreSQL database: + +```sh +DATABASE_URL=postgres://buzz:buzz_dev@127.0.0.1:55432/buzz \ + cargo test --locked -p buzz-relay --lib \ + handlers::attach_claim::postgres_tests -- --ignored +``` + +Both tests must pass. +They exercise concurrent claims, duplicate receipts, deletion, metadata retention, malformed envelopes, and unowned signers for both modes. +Runtime terminal and snapshot consumers are separate layers; this relay layer does not start either operation. + +See [NIP-TA](nips/NIP-TA.md) and [NIP-SA](nips/NIP-SA.md) for their distinct approvals.