diff --git a/Cargo.lock b/Cargo.lock index 2ab0ef8de5e..90f0eab62cb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -983,6 +983,7 @@ dependencies = [ "axum", "base64 0.22.1", "buzz-core", + "buzz-ifc", "buzz-persona", "buzz-sdk", "buzz-ws-client", diff --git a/crates/buzz-cli/Cargo.toml b/crates/buzz-cli/Cargo.toml index 59d1bb2cee6..d8bbb80203a 100644 --- a/crates/buzz-cli/Cargo.toml +++ b/crates/buzz-cli/Cargo.toml @@ -86,6 +86,10 @@ rustls = { version = "0.23", default-features = false, features = ["ring", "std" # Random number generation — full jitter for exponential backoff in with_retry rand = { workspace = true } +[target.'cfg(unix)'.dependencies] +buzz-ifc = { workspace = true } +tempfile = "3" + [dev-dependencies] # Scratch files for channel-templates.json fixtures in tests tempfile = "3" diff --git a/crates/buzz-cli/README.md b/crates/buzz-cli/README.md index 29ec19a04c2..d2607684f40 100644 --- a/crates/buzz-cli/README.md +++ b/crates/buzz-cli/README.md @@ -20,6 +20,13 @@ export BUZZ_PRIVATE_KEY="nsec1..." buzz channels list ``` +## Local broker prototype + +On Unix, `buzz broker serve` holds the key for one fixed channel; +`buzz broker read` and `buzz broker reply` use its socket without a key. +This is opt-in and does not isolate the broker from other same-user processes. +See [setup and limits](src/broker/README.md). + ## Usage All output is JSON on stdout. Errors are JSON on stderr. Exit codes: 0=ok, 1=user error, 2=network, 3=auth, 4=other, 5=write conflict. diff --git a/crates/buzz-cli/src/broker/README.md b/crates/buzz-cli/src/broker/README.md new file mode 100644 index 00000000000..6def075325b --- /dev/null +++ b/crates/buzz-cli/src/broker/README.md @@ -0,0 +1,88 @@ +# Local broker prototype + +One process holds the Buzz key and one `IfcSession`. A keyless client can read +recent messages from one channel or post plain text there. The operator fixes +the relay, channel, community, and requester at startup. Requests cannot select +another destination, sign an arbitrary event, or change the session. + +This is an opt-in CLI prototype, not a hardened local security boundary. It does +not change how Desktop or `buzz-acp` launches agents; those paths still supply +credentials as before. It does not implement the SDK's full HTTP broker contract. + +## Run it + +Build with `cargo build -p buzz-cli`. In an operator terminal that already has +the agent's `BUZZ_PRIVATE_KEY` (and `BUZZ_AUTH_TAG`, if needed), start: + +```sh +buzz --relay https://your-community.example broker serve \ + --community \ + --channel \ + --relay-key \ + --requester +``` + +Get the community UUID and relay signing key from a trusted operator, not from +the agent. Both the agent and requester must be current channel members. Only +public or private stream channels are supported; DMs, forums, thread routing, +mentions, media, and arbitrary event kinds are deliberately absent. + +The foreground process prints `{"socket":"...","channel":"..."}`. Give only +that socket path to a **fresh** agent with no Buzz credential environment: + +```sh +buzz broker read --socket --limit 20 +buzz broker reply --socket --content 'Plain text answer' +``` + +`reply` posts a top-level message in the fixed channel. It does not resolve +mentions or expand text into additional operations. The read limit is 1–100; +reply text is at most 16 KiB. The client does not connect directly to the relay +or fall back to a private key when the socket is unavailable. + +Stop the broker with Ctrl-C. Each startup creates a new socket in a private +temporary directory. Do not reconnect an agent's old history or files to a new +broker: it would discard the old IFC state. This prototype does not manage that +agent lifetime for you. + +## Checks and limits + +The broker verifies relay signatures on metadata and the complete membership +snapshot before creating the domain, before each operation, and again before +delivering a read. It validates each returned message's signature, kind, and +channel. A changed snapshot or failed policy check permanently blocks new +publications from that session, including after a policy rollback. Even a +metadata edit or temporary policy-query failure requires a fresh agent and +broker before writing again; there is no recovery API that clears IFC state. + +The broker builds the outgoing event itself, passes its serialized bytes to +`IfcSession::publish`, and sends exactly those bytes. It never automatically +retries a publication. If the connection fails, the message may already have +been accepted; inspect the channel before deciding to send again. + +Membership checks and publication are **not atomic**. Membership may change +after the final check but before the relay accepts a message. The relay remains +trusted for current access checks, and a future relay-side policy-version +precondition is needed to close that race. This is not a claim of complete IFC +enforcement under concurrent membership changes. + +The socket's directory is mode 0700 and the socket is 0600. That excludes other +OS users, not another process running as the same user. Such a process may read +the broker's environment or memory, replace its executable, or use its socket. +There is no signed helper, Hardened Runtime enforcement, or protected Keychain +storage here. Do not treat this prototype as protection against a hostile local +agent. It also cannot label the agent's filesystem, model history, other tools, +or network traffic: start with fresh state and do not feed it unrelated private +data. End-to-end confidentiality requires those boundaries too. + +Tests live beside the code, with each invariant explained inline. Run +`cargo test -p buzz-cli broker` for the socket/HTTP tests against a signed mock +relay; they do not require a running Buzz deployment. + +To also exercise separate broker/client processes and shutdown on macOS/Unix: + +```sh +cargo build -p buzz-cli +BUZZ_TEST_BROKER_BIN="$PWD/target/debug/buzz" \ + cargo test -p buzz-cli broker_executable_smoke -- --ignored +``` diff --git a/crates/buzz-cli/src/broker/mod.rs b/crates/buzz-cli/src/broker/mod.rs new file mode 100644 index 00000000000..b9eb82eb5c5 --- /dev/null +++ b/crates/buzz-cli/src/broker/mod.rs @@ -0,0 +1,285 @@ +//! A deliberately small local broker: one signer, one channel, one IFC session. +//! +//! The operator chooses the scope before the socket exists. A socket caller +//! cannot choose a relay, channel, audience, event kind, or signing payload. +//! This is not a same-user security boundary: the executable, environment, +//! socket, and agent's other inputs are not isolated by the OS. See README.md. + +mod relay; +#[cfg(test)] +mod tests; + +use std::{os::unix::fs::PermissionsExt, path::PathBuf, time::Duration}; + +use buzz_ifc::{ExecutionDomain, IfcSession, ResourceLabel}; +use nostr::{Keys, PublicKey, Tag}; +use serde::{de::DeserializeOwned, Deserialize, Serialize}; +use serde_json::{json, Value}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{UnixListener, UnixStream}, + time::timeout, +}; +use uuid::Uuid; + +use crate::error::CliError; +use relay::Relay; + +const READ: &str = "channel.read"; +const REPLY: &str = "message.post"; +const REQUEST_BYTES: usize = 32 * 1024; +const RESPONSE_BYTES: usize = 2 * 1024 * 1024; +const DEADLINE: Duration = Duration::from_secs(30); + +#[derive(clap::Subcommand)] +pub(crate) enum Command { + /// Hold the key and serve one fixed channel; prints the socket path as JSON + Serve(Scope), + /// Read recent messages without a private key + Read { + #[arg(long)] + socket: PathBuf, + #[arg(long, default_value_t = 20, value_parser = clap::value_parser!(u32).range(1..=100))] + limit: u32, + }, + /// Post plain text to the broker's channel (no thread or mention expansion) + Reply { + #[arg(long)] + socket: PathBuf, + #[arg(long)] + content: String, + }, +} + +/// Trusted startup configuration, never deserialized from a socket request. +#[derive(Clone, clap::Args)] +pub(crate) struct Scope { + /// Community UUID belonging to the configured relay, supplied by its operator + #[arg(long)] + community: Uuid, + /// The only channel this broker may read or write + #[arg(long)] + channel: Uuid, + /// Pinned relay signing key for channel metadata and membership + #[arg(long)] + relay_key: PublicKey, + /// Person authorizing this work; must be a current channel member + #[arg(long)] + requester: PublicKey, +} + +#[derive(Serialize, Deserialize)] +#[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)] +enum Request { + Read { limit: u32 }, + Reply { content: String }, +} + +#[derive(Serialize, Deserialize)] +#[serde(tag = "status", rename_all = "snake_case", deny_unknown_fields)] +enum Response { + Ok { result: Value }, + Error { message: String }, +} + +struct Broker { + relay: Relay, + scope: Scope, + session: IfcSession, +} + +impl Broker { + async fn open(relay: Relay, scope: Scope) -> Result { + let domain = relay.domain(&scope).await?; + Ok(Self { + relay, + scope, + session: IfcSession::enter(domain), + }) + } + + async fn check_scope(&mut self) -> Result { + let current = match self.relay.domain(&self.scope).await { + Ok(domain) => domain, + Err(error) => { + self.session.mark_unknown_input(); + return Err(error); + } + }; + if current.key() != self.session.domain_key() { + // Never replace the session: the agent may still hold data from + // the old audience. Even a subsequent policy rollback cannot + // make this instance safe to publish again. + self.session.mark_unknown_input(); + return Err(denied( + "channel policy changed; start a fresh agent and broker", + )); + } + self.session + .read(&ResourceLabel::from_domain(¤t)) + .map_err(|_| denied("resource is outside the session"))?; + Ok(current) + } + + async fn handle(&mut self, request: Request) -> Result { + match &request { + Request::Read { limit } if !(1..=100).contains(limit) => { + return Err(denied("read limit must be between 1 and 100")) + } + Request::Reply { content } + if content.trim().is_empty() || content.len() > 16 * 1024 => + { + return Err(denied("reply must contain 1 to 16384 bytes of text")) + } + _ => {} + } + let domain = self.check_scope().await?; + match request { + Request::Read { limit } => { + self.session + .call(READ) + .map_err(|_| denied("read is not allowed"))?; + let events = self.relay.read(self.scope.channel, limit).await?; + // Do not deliver a fetch that overlapped an observed policy + // change. A fresh query is not an atomic relay transaction; + // the remaining race is documented alongside this prototype. + self.check_scope().await?; + Ok(json!(events)) + } + Request::Reply { content } => { + let event = self.relay.message(self.scope.channel, &content)?; + let bytes = + serde_json::to_vec(&event).map_err(|_| denied("cannot encode reply"))?; + let authorization = self + .session + .publish(REPLY, domain.audience(), bytes) + .map_err(|_| denied("session cannot publish"))?; + self.relay.publish(authorization, event.id).await + } + } + } +} + +pub(crate) async fn serve( + url: String, + keys: Keys, + auth: Option, + scope: Scope, +) -> Result<(), CliError> { + eprintln!("WARNING: local broker prototype; no OS isolation from same-user processes."); + let relay = Relay::new(url, keys, auth)?; + let mut broker = Broker::open(relay, scope).await?; + let (_directory, listener, socket) = bind_socket()?; + println!( + "{}", + json!({"socket": socket, "channel": broker.scope.channel}) + ); + loop { + tokio::select! { + signal = tokio::signal::ctrl_c() => { + signal.map_err(io_error)?; + return Ok(()); + } + incoming = listener.accept() => { + let (stream, _) = incoming.map_err(io_error)?; + // One request at a time keeps IFC state changes serialized. + // A silent client cannot hold the broker forever. + if let Err(error) = connection(&mut broker, stream).await { + eprintln!("local broker connection failed: {error}"); + } + } + } + } +} + +fn bind_socket() -> Result<(tempfile::TempDir, UnixListener, PathBuf), CliError> { + // tempfile creates a new 0700 directory. Never unlink a caller-selected + // socket or reuse its name for a new IFC session after a restart. + let directory = tempfile::Builder::new() + .prefix("buzz-broker-") + .permissions(std::fs::Permissions::from_mode(0o700)) + .tempdir() + .map_err(io_error)?; + let socket = directory.path().join("broker.sock"); + let listener = UnixListener::bind(&socket).map_err(io_error)?; + std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o600)).map_err(io_error)?; + Ok((directory, listener, socket)) +} + +async fn connection(broker: &mut Broker, mut stream: UnixStream) -> Result<(), CliError> { + timeout(DEADLINE, async { + let result = match receive::(&mut stream, REQUEST_BYTES).await { + Ok(request) => broker.handle(request).await, + Err(error) => Err(error), + }; + let response = match result { + Ok(result) => Response::Ok { result }, + Err(error) => Response::Error { + message: error.to_string(), + }, + }; + send(&mut stream, &response, RESPONSE_BYTES).await + }) + .await + .map_err(|_| denied("broker request timed out; publication outcome may be unknown"))? +} + +pub(crate) async fn call(command: &Command) -> Result<(), CliError> { + let (socket, request) = match command { + Command::Read { socket, limit } => (socket, Request::Read { limit: *limit }), + Command::Reply { socket, content } => ( + socket, + Request::Reply { + content: content.clone(), + }, + ), + Command::Serve(_) => return Err(denied("serve requires the operator's credentials")), + }; + let result = timeout(DEADLINE + Duration::from_secs(5), async { + let mut stream = UnixStream::connect(socket).await.map_err(io_error)?; + send(&mut stream, &request, REQUEST_BYTES).await?; + match receive::(&mut stream, RESPONSE_BYTES).await? { + Response::Ok { result } => Ok(result), + Response::Error { message } => Err(denied(message)), + } + }) + .await + .map_err(|_| denied("broker did not respond; publication outcome may be unknown"))??; + println!("{result}"); + Ok(()) +} + +// Length-prefixed JSON bounds allocation before decoding. No agent-controlled +// string reaches a shell, a URL, or a relay filter. +async fn receive(stream: &mut UnixStream, max: usize) -> Result { + let len = stream.read_u32().await.map_err(io_error)? as usize; + if len > max { + return Err(denied("broker frame is too large")); + } + let mut bytes = vec![0; len]; + stream.read_exact(&mut bytes).await.map_err(io_error)?; + serde_json::from_slice(&bytes).map_err(|_| denied("invalid broker request or response")) +} + +async fn send( + stream: &mut UnixStream, + value: &T, + max: usize, +) -> Result<(), CliError> { + let bytes = serde_json::to_vec(value).map_err(|_| denied("cannot encode broker frame"))?; + if bytes.len() > max { + return Err(denied("broker frame is too large")); + } + stream + .write_u32(bytes.len() as u32) + .await + .map_err(io_error)?; + stream.write_all(&bytes).await.map_err(io_error) +} + +fn denied(message: impl Into) -> CliError { + CliError::Other(message.into()) +} +fn io_error(error: std::io::Error) -> CliError { + denied(format!("local broker I/O failed: {error}")) +} diff --git a/crates/buzz-cli/src/broker/relay.rs b/crates/buzz-cli/src/broker/relay.rs new file mode 100644 index 00000000000..1de48da4d01 --- /dev/null +++ b/crates/buzz-cli/src/broker/relay.rs @@ -0,0 +1,260 @@ +use std::{collections::BTreeSet, time::Duration}; + +use buzz_core::kind::{ + KIND_NIP29_GROUP_MEMBERS, KIND_NIP29_GROUP_METADATA, KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_V2, +}; +use buzz_ifc::{ + derive_execution_domain, AuthorizedPublication, CapabilityPolicy, CapabilitySet, CommunityId, + ConversationKind, DomainFacts, ExecutionDomain, MembershipEpoch, OperationEffect, Principal, +}; +use nostr::{Event, EventId, Keys, Tag}; +use serde_json::{json, Value}; +use uuid::Uuid; + +use super::{denied, Scope, READ, REPLY}; +use crate::{client::sign_nip98, error::CliError}; + +const RELAY_BYTES: usize = 1024 * 1024; + +pub(super) struct Relay { + http: reqwest::Client, + url: String, + keys: Keys, + auth: Option, +} + +impl Relay { + pub(super) fn new(url: String, keys: Keys, auth: Option) -> Result { + let parsed = url::Url::parse(&url).map_err(|_| denied("invalid relay URL"))?; + let loopback = parsed.host_str().is_some_and(|host| { + host == "localhost" + || host + .trim_matches(['[', ']']) + .parse::() + .is_ok_and(|ip| ip.is_loopback()) + }); + if !(parsed.scheme() == "https" || parsed.scheme() == "http" && loopback) + || !parsed.username().is_empty() + || parsed.password().is_some() + || parsed.query().is_some() + || parsed.fragment().is_some() + || parsed.path() != "/" + { + return Err(denied( + "relay must be an HTTPS origin (HTTP is allowed only on loopback)", + )); + } + let http = reqwest::Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .timeout(Duration::from_secs(5)) + .build() + .map_err(|_| denied("cannot create relay client"))?; + Ok(Self { + http, + url: url.trim_end_matches('/').to_owned(), + keys, + auth, + }) + } + + async fn post(&self, path: &str, bytes: Vec) -> Result { + let url = format!("{}{path}", self.url); + let auth = sign_nip98(&self.keys, "POST", &url, Some(&bytes))?; + let mut request = self + .http + .post(url) + .header("Authorization", auth) + .header("Content-Type", "application/json") + .body(bytes); + if let Some(tag) = &self.auth { + request = request.header( + "x-auth-tag", + serde_json::to_string(tag.as_slice()) + .map_err(|_| denied("cannot encode owner attestation"))?, + ); + } + let mut response = request + .send() + .await + .map_err(|_| denied("relay request failed; publication outcome may be unknown"))?; + if !response.status().is_success() { + // Never forward an unscoped relay error body to the agent. + return Err(denied(format!( + "relay rejected request (HTTP {})", + response.status().as_u16() + ))); + } + let mut body = Vec::new(); + while let Some(chunk) = response + .chunk() + .await + .map_err(|_| denied("incomplete relay response; publication outcome may be unknown"))? + { + if chunk.len() > RELAY_BYTES.saturating_sub(body.len()) { + return Err(denied("relay response is too large")); + } + body.extend_from_slice(&chunk); + } + serde_json::from_slice(&body).map_err(|_| denied("invalid relay response")) + } + + async fn query(&self, filter: Value) -> Result, CliError> { + let body = serde_json::to_vec(&[filter]).map_err(|_| denied("cannot encode query"))?; + serde_json::from_value(self.post("/query", body).await?) + .map_err(|_| denied("invalid relay events")) + } + + pub(super) async fn domain(&self, scope: &Scope) -> Result { + let events = self + .query(json!({ + "kinds": [KIND_NIP29_GROUP_METADATA, KIND_NIP29_GROUP_MEMBERS], + "authors": [scope.relay_key.to_hex()], "#d": [scope.channel.to_string()], "limit": 2 + })) + .await?; + let mut metadata = None; + let mut membership = None; + for event in &events { + if event.pubkey != scope.relay_key + || event.verify().is_err() + || !exact_tag(event, "d", &scope.channel.to_string()) + { + return Err(denied("untrusted channel state")); + } + let slot = match u32::from(event.kind.as_u16()) { + KIND_NIP29_GROUP_METADATA => &mut metadata, + KIND_NIP29_GROUP_MEMBERS => &mut membership, + _ => return Err(denied("unexpected channel state")), + }; + if slot.replace(event).is_some() { + return Err(denied("ambiguous channel state")); + } + } + let (Some(metadata), Some(membership)) = (metadata, membership) else { + return Err(denied("channel metadata or membership is missing")); + }; + let public = flag(metadata, "public"); + if public == flag(metadata, "private") + || !exact_tag(metadata, "t", "stream") + || metadata + .tags + .iter() + .any(|tag| tag.as_slice().first().is_some_and(|s| s == "archived")) + { + return Err(denied( + "broker requires an active public or private stream channel", + )); + } + let mut members = BTreeSet::new(); + for tag in membership + .tags + .iter() + .filter(|tag| tag.as_slice().first().is_some_and(|s| s == "p")) + { + let key = tag + .as_slice() + .get(1) + .ok_or_else(|| denied("invalid channel member"))?; + members.insert(Principal::from_hex(key).map_err(|_| denied("invalid channel member"))?); + } + let agent = Principal::from_public_key(&self.keys.public_key()) + .map_err(|_| denied("invalid agent key"))?; + let requester = Principal::from_public_key(&scope.requester) + .map_err(|_| denied("invalid requester key"))?; + if !members.contains(&agent) || !members.contains(&requester) { + return Err(denied( + "agent and requester must be current channel members", + )); + } + let capabilities = CapabilitySet::from_operations([ + (READ, OperationEffect::NonEgressing), + (REPLY, OperationEffect::Publication), + ]); + derive_execution_domain( + DomainFacts { + community: CommunityId::from_uuid(scope.community), + channel_id: scope.channel, + kind: if public { + ConversationKind::Public + } else { + ConversationKind::Restricted + }, + epoch: MembershipEpoch::new(format!("{}:{}", metadata.id, membership.id)), + members, + executing_agent: agent, + requesters: BTreeSet::from([requester]), + system_principal: None, + owner: None, + }, + &CapabilityPolicy::new(capabilities.clone(), capabilities), + ) + .map_err(|_| denied("invalid execution domain")) + } + + pub(super) async fn read(&self, channel: Uuid, limit: u32) -> Result, CliError> { + let events = self + .query(json!({ + "kinds": [KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2], + "#h": [channel.to_string()], "limit": limit + })) + .await?; + if events.len() > limit as usize + || events.iter().any(|event| { + !matches!( + u32::from(event.kind.as_u16()), + KIND_STREAM_MESSAGE | KIND_STREAM_MESSAGE_V2 + ) || !exact_tag(event, "h", &channel.to_string()) + || event.verify().is_err() + }) + { + return Err(denied("relay returned invalid or out-of-channel messages")); + } + Ok(events) + } + + pub(super) fn message(&self, channel: Uuid, content: &str) -> Result { + let mut builder = + buzz_sdk::builders::build_message(channel, content, None, &[], false, &[], &[]) + .map_err(|_| denied("invalid message"))?; + if let Some(auth) = &self.auth { + builder = builder.tags([auth.clone()]); + } + builder + .sign_with_keys(&self.keys) + .map_err(|_| denied("cannot sign reply")) + } + + pub(super) async fn publish( + &self, + authorization: AuthorizedPublication, + expected: EventId, + ) -> Result { + let (_, _, bytes) = authorization.into_parts(); + // Send the exact serialized event checked by IFC. No retries: a lost + // acknowledgement must not silently turn into a second message. + let response = self.post("/events", bytes).await?; + if response.get("accepted").and_then(Value::as_bool) != Some(true) + || response.get("event_id").and_then(Value::as_str) != Some(expected.to_hex().as_str()) + { + return Err(denied("relay did not acknowledge the exact reply")); + } + Ok(json!({"event_id": expected, "accepted": true, "message": ""})) + } +} + +// Reject duplicate routing tags even when they agree. Different consumers +// must not be able to disagree about which channel a signed event names. +fn exact_tag(event: &Event, name: &str, value: &str) -> bool { + let mut tags = event + .tags + .iter() + .filter(|tag| tag.as_slice().first().is_some_and(|s| s == name)); + tags.next() + .is_some_and(|tag| tag.as_slice() == [name, value]) + && tags.next().is_none() +} + +fn flag(event: &Event, name: &str) -> bool { + event.tags.iter().any(|tag| tag.as_slice() == [name]) +} diff --git a/crates/buzz-cli/src/broker/tests.rs b/crates/buzz-cli/src/broker/tests.rs new file mode 100644 index 00000000000..46e4aab536b --- /dev/null +++ b/crates/buzz-cli/src/broker/tests.rs @@ -0,0 +1,685 @@ +use std::sync::{Arc, Mutex}; + +use axum::{body::Bytes, extract::State, http::HeaderMap, routing::post, Json, Router}; +use base64::Engine; +use nostr::{Event, EventBuilder, Kind, Timestamp}; +use sha2::{Digest, Sha256}; + +use super::*; + +fn keys(n: u8) -> Keys { + let _ = rustls::crypto::ring::default_provider().install_default(); + Keys::parse(&format!("{n:064x}")).expect("fixture key") +} + +fn scope() -> Scope { + Scope { + community: Uuid::from_u128(1), + channel: Uuid::from_u128(2), + relay_key: keys(1).public_key(), + requester: keys(3).public_key(), + } +} + +fn event(signer: &Keys, kind: u16, tags: Vec>, content: &str) -> Event { + EventBuilder::new(Kind::Custom(kind), content) + .tags( + tags.into_iter() + .map(|tag| Tag::parse(tag).expect("fixture tag")), + ) + .custom_created_at(Timestamp::from_secs(100)) + .sign_with_keys(signer) + .expect("fixture signature") +} + +fn policy(public: bool, members: &[u8]) -> Vec { + let id = scope().channel.to_string(); + vec![ + event( + &keys(1), + 39000, + vec![ + vec!["d".into(), id.clone()], + vec![if public { "public" } else { "private" }.into()], + // emit_group_discovery_events uses `t`, not `type` or + // the creation command's `channel_type` tag. + vec!["t".into(), "stream".into()], + ], + "", + ), + event( + &keys(1), + 39002, + std::iter::once(vec!["d".into(), id]) + .chain( + members + .iter() + .map(|n| vec!["p".into(), keys(*n).public_key().to_hex()]), + ) + .collect(), + "", + ), + ] +} + +fn message(channel: Uuid) -> Event { + event( + &keys(3), + 9, + vec![vec!["h".into(), channel.to_string()]], + "hello", + ) +} + +struct RelayState { + policy: Vec, + messages: Vec, + sent: Vec, + filters: Vec, + // Change membership while a read is in flight, after its first check. + policy_after_read: Option>, + reject_publish: bool, +} + +struct Fixture { + url: String, + state: Arc>, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.task.abort(); + } +} + +// Check real NIP-98 signatures and body hashes in the mock relay. These tests +// exercise HTTP signing, JSON framing, event verification, and IFC together. +fn authenticate(headers: &HeaderMap, body: &[u8]) { + let auth = headers["authorization"] + .to_str() + .expect("auth string") + .strip_prefix("Nostr ") + .expect("Nostr auth"); + let event: Event = serde_json::from_slice( + &base64::engine::general_purpose::STANDARD + .decode(auth) + .expect("base64"), + ) + .expect("auth event"); + event.verify().expect("valid signature"); + assert_eq!(event.pubkey, keys(2).public_key()); + assert!(event + .tags + .iter() + .any(|tag| tag.as_slice() == ["payload", &hex::encode(Sha256::digest(body))])); +} + +async fn query( + State(state): State>>, + headers: HeaderMap, + body: Bytes, +) -> Json { + authenticate(&headers, &body); + let filters: Vec = serde_json::from_slice(&body).expect("filters"); + assert_eq!(filters.len(), 1); + let mut state = state.lock().expect("state"); + let filter = &filters[0]; + state.filters.push(filter.clone()); + let events = if filter["kinds"][0] == 39000 { + state.policy.clone() + } else { + assert_eq!(filter["#h"], json!([scope().channel.to_string()])); + assert_eq!(filter["kinds"], json!([9, 40002])); + if let Some(policy) = state.policy_after_read.take() { + state.policy = policy; + } + state.messages.clone() + }; + Json(json!(events)) +} + +async fn publish( + State(state): State>>, + headers: HeaderMap, + body: Bytes, +) -> Json { + authenticate(&headers, &body); + let event: Event = serde_json::from_slice(&body).expect("published event"); + event.verify().expect("valid signature"); + assert_eq!(event.pubkey, keys(2).public_key()); + assert_eq!(event.kind, Kind::Custom(9)); + assert_eq!( + event + .tags + .iter() + .map(|t| t.as_slice().to_vec()) + .collect::>(), + vec![vec!["h".to_owned(), scope().channel.to_string()]] + ); + let mut state = state.lock().expect("state"); + state.sent.push(event.clone()); + Json( + json!({"event_id": event.id, "accepted": !state.reject_publish, "message": "not forwarded"}), + ) +} + +impl Fixture { + async fn new(public: bool) -> Self { + let state = Arc::new(Mutex::new(RelayState { + policy: policy(public, &[2, 3]), + messages: vec![message(scope().channel)], + sent: vec![], + filters: vec![], + policy_after_read: None, + reject_publish: false, + })); + let router = Router::new() + .route("/query", post(query)) + .route("/events", post(publish)) + .with_state(state.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("listener"); + let url = format!("http://{}", listener.local_addr().expect("address")); + let task = tokio::spawn(async move { + axum::serve(listener, router).await.expect("relay"); + }); + Self { url, state, task } + } + + async fn broker(&self) -> Broker { + Broker::open( + Relay::new(self.url.clone(), keys(2), None).expect("client"), + scope(), + ) + .await + .expect("broker") + } +} + +// Pin the production socket path, not just a test-only policy predicate. Only +// the broker fixture has a key; its caller gets signed, scoped messages and +// a receipt, never a signer or arbitrary relay transport. +#[tokio::test] +async fn keyless_socket_client_reads_and_replies_in_public_and_private_channels() { + for public in [true, false] { + let fixture = Fixture::new(public).await; + let mut broker = fixture.broker().await; + for request in [ + Request::Read { limit: 20 }, + Request::Reply { + content: "answer".into(), + }, + ] { + let (mut client, server) = UnixStream::pair().expect("socket pair"); + let (served, received) = tokio::join!(connection(&mut broker, server), async { + send(&mut client, &request, REQUEST_BYTES) + .await + .expect("request"); + receive::(&mut client, RESPONSE_BYTES) + .await + .expect("response") + }); + served.expect("served"); + let Response::Ok { result } = received else { + panic!("expected success") + }; + match request { + Request::Read { .. } => assert_eq!(result[0]["content"], "hello"), + Request::Reply { .. } => assert_eq!(result["accepted"], true), + } + assert!(!result + .to_string() + .contains(&keys(2).secret_key().to_secret_hex())); + } + assert_eq!( + fixture.state.lock().expect("state").sent[0].content, + "answer" + ); + } +} + +// Removing any signature, channel, or kind check must admit one of these +// responses. Rejection happens before a single event is returned to the agent. +#[tokio::test] +async fn reads_reject_forged_cross_channel_duplicate_tag_and_wrong_kind_events() { + let fixture = Fixture::new(false).await; + let mut broker = fixture.broker().await; + let mut forged = message(scope().channel); + forged.content = "unsigned edit".into(); + let duplicate = event( + &keys(3), + 9, + vec![ + vec!["h".into(), scope().channel.to_string()], + vec!["h".into(), Uuid::from_u128(99).to_string()], + ], + "secret", + ); + let wrong_kind = event( + &keys(3), + 1, + vec![vec!["h".into(), scope().channel.to_string()]], + "secret", + ); + for message in [forged, message(Uuid::from_u128(99)), duplicate, wrong_kind] { + fixture.state.lock().expect("state").messages = vec![message]; + assert!(broker.handle(Request::Read { limit: 20 }).await.is_err()); + } +} + +// A valid signature from another identity is not relay authority. Missing or +// duplicate snapshots must not default to a public audience. +#[tokio::test] +async fn startup_rejects_untrusted_or_incomplete_channel_policy() { + let fixture = Fixture::new(false).await; + let original = policy(false, &[2, 3]); + let mut forged = original.clone(); + forged[0].content = "tampered".into(); + let mut wrong_author = original.clone(); + wrong_author[0] = event( + &keys(4), + 39000, + original[0] + .tags + .iter() + .map(|t| t.as_slice().to_vec()) + .collect(), + "", + ); + let mut duplicate = original.clone(); + duplicate.push(original[0].clone()); + for policy in [ + vec![], + vec![original[0].clone()], + forged, + wrong_author, + duplicate, + policy(false, &[3]), + policy(false, &[2, 4]), + ] { + fixture.state.lock().expect("state").policy = policy; + assert!(Broker::open( + Relay::new(fixture.url.clone(), keys(2), None).expect("client"), + scope() + ) + .await + .is_err()); + } +} + +// Observing a membership or visibility change poisons the retained session. +// Restoring the old snapshot must not reset it and authorize another reply. +#[tokio::test] +async fn policy_change_permanently_blocks_publication_even_after_rollback() { + for changed in [ + policy(false, &[2, 3, 4]), + policy(true, &[2, 3]), + policy(false, &[3]), + ] { + let fixture = Fixture::new(false).await; + let mut broker = fixture.broker().await; + fixture.state.lock().expect("state").policy = changed; + assert!(broker + .handle(Request::Reply { + content: "secret".into() + }) + .await + .is_err()); + fixture.state.lock().expect("state").policy = policy(false, &[2, 3]); + assert!(broker + .handle(Request::Reply { + content: "still secret".into() + }) + .await + .is_err()); + assert!(fixture.state.lock().expect("state").sent.is_empty()); + } +} + +// The check after fetching is required: checking only before a read delivers +// data even when a membership change was observable before delivery. +#[tokio::test] +async fn membership_change_during_read_prevents_delivery() { + let fixture = Fixture::new(false).await; + let mut broker = fixture.broker().await; + fixture.state.lock().expect("state").policy_after_read = Some(policy(false, &[2, 3, 4])); + assert!(broker.handle(Request::Read { limit: 20 }).await.is_err()); +} + +// Unknown JSON fields are rejected, not ignored. Otherwise an agent could +// believe a destination, signer, or raw event it supplied had been honored. +#[test] +fn requests_cannot_expand_the_broker_scope() { + for request in [ + json!({"op":"sign", "bytes":"secret"}), + json!({"op":"read", "limit":20, "channel":"elsewhere"}), + json!({"op":"reply", "content":"text", "relay":"https://attacker.test"}), + json!({"op":"reply", "content":"text", "tags":[["h","elsewhere"]]}), + json!({"op":"reply", "content":"text", "pubkey":"another identity"}), + ] { + assert!(serde_json::from_value::(request).is_err()); + } +} + +// Bounds are applied before relay work and before allocating an IPC frame. +#[tokio::test] +async fn malformed_or_oversized_requests_never_reach_the_relay() { + let fixture = Fixture::new(true).await; + let mut broker = fixture.broker().await; + fixture.state.lock().expect("state").filters.clear(); + for request in [ + Request::Read { limit: 0 }, + Request::Read { limit: 101 }, + Request::Reply { + content: " ".into(), + }, + Request::Reply { + content: "x".repeat(16385), + }, + ] { + assert!(broker.handle(request).await.is_err()); + } + let (mut client, server) = UnixStream::pair().expect("sockets"); + let (served, response) = tokio::join!(connection(&mut broker, server), async { + client.write_u32(u32::MAX).await.expect("length"); + receive::(&mut client, RESPONSE_BYTES) + .await + .expect("response") + }); + served.expect("error response sent"); + assert!(matches!(response, Response::Error { .. })); + assert!(fixture.state.lock().expect("state").filters.is_empty()); +} + +// A relay rejection is an error, never an accepted receipt, and is not retried. +#[tokio::test] +async fn failed_publication_is_not_reported_as_success_or_retried() { + let fixture = Fixture::new(true).await; + let mut broker = fixture.broker().await; + fixture.state.lock().expect("state").reject_publish = true; + assert!(broker + .handle(Request::Reply { + content: "hello".into() + }) + .await + .is_err()); + assert_eq!(fixture.state.lock().expect("state").sent.len(), 1); +} + +// An idle local peer cannot monopolize the single-session broker indefinitely. +#[tokio::test(start_paused = true)] +async fn silent_socket_client_has_a_deadline() { + // No HTTP needed: the deadline expires before any request is decoded. + let relay = Relay::new("http://127.0.0.1:1".into(), keys(2), None).expect("relay"); + let capabilities = buzz_ifc::CapabilitySet::default(); + let domain = buzz_ifc::derive_execution_domain( + buzz_ifc::DomainFacts { + community: buzz_ifc::CommunityId::from_uuid(scope().community), + channel_id: scope().channel, + kind: buzz_ifc::ConversationKind::Public, + epoch: buzz_ifc::MembershipEpoch::new("test"), + members: Default::default(), + executing_agent: buzz_ifc::Principal::from_public_key(&keys(2).public_key()) + .expect("principal"), + requesters: [ + buzz_ifc::Principal::from_public_key(&keys(3).public_key()).expect("principal") + ] + .into(), + system_principal: None, + owner: None, + }, + &buzz_ifc::CapabilityPolicy::new(capabilities.clone(), capabilities), + ) + .expect("domain"); + let mut broker = Broker { + relay, + scope: scope(), + session: IfcSession::enter(domain), + }; + let (_client, server) = UnixStream::pair().expect("sockets"); + assert!(connection(&mut broker, server).await.is_err()); +} + +#[test] +fn relay_url_rejects_insecure_or_ambiguous_destinations() { + for url in [ + "http://example.com", + "https://user:pass@example.com", + "https://example.com/path", + "https://example.com/?relay=other", + "https://example.com/#fragment", + "file:///tmp/key", + ] { + assert!(Relay::new(url.into(), keys(2), None).is_err(), "{url}"); + } +} + +// Exercise the CLI dispatch before key parsing, using the real listener and +// handler. Removing the early keyless branch makes both calls fail. +#[tokio::test] +async fn cli_clients_need_neither_a_key_nor_a_relay_and_never_fall_back() { + let fixture = Fixture::new(false).await; + let mut broker = fixture.broker().await; + let (_directory, listener, socket) = bind_socket().expect("socket"); + for command in [ + Command::Read { + socket: socket.clone(), + limit: 20, + }, + Command::Reply { + socket: socket.clone(), + content: "from CLI".into(), + }, + ] { + let cli = crate::Cli { + relay: "not a relay URL".into(), + private_key: None, + auth_tag: None, + format: crate::OutputFormat::Json, + command: crate::Cmd::Broker(command), + }; + let (served, result) = tokio::join!( + async { + let (stream, _) = listener.accept().await.expect("accept"); + connection(&mut broker, stream).await + }, + crate::run(cli) + ); + served.expect("serve client"); + result.expect("keyless CLI call"); + } + drop(listener); + assert!(call(&Command::Reply { + socket, + content: "must not go direct".into() + }) + .await + .is_err()); + assert_eq!(fixture.state.lock().expect("state").sent.len(), 1); +} + +// Permissions belong to the actual listener setup. A temporary-directory +// default of 0755 would expose the channel socket to other OS users. +#[tokio::test] +async fn socket_is_private_and_a_new_session_never_reuses_its_name() { + let (directory, listener, path) = bind_socket().expect("socket"); + assert_eq!( + std::fs::metadata(directory.path()) + .expect("directory") + .permissions() + .mode() + & 0o777, + 0o700 + ); + assert_eq!( + std::fs::metadata(&path) + .expect("socket") + .permissions() + .mode() + & 0o777, + 0o600 + ); + let (_another_directory, _another_listener, another_path) = + bind_socket().expect("second socket"); + assert_ne!(path, another_path); + drop(listener); + drop(directory); + assert!(!path.exists()); +} + +// A disconnected caller may have lost an acknowledgement. The broker must +// not repeat the signed publication when writing its response fails. +#[tokio::test] +async fn disconnected_client_does_not_repeat_publication() { + let fixture = Fixture::new(true).await; + let mut broker = fixture.broker().await; + let (mut client, server) = UnixStream::pair().expect("sockets"); + send( + &mut client, + &Request::Reply { + content: "once".into(), + }, + REQUEST_BYTES, + ) + .await + .expect("request"); + drop(client); + assert!(connection(&mut broker, server).await.is_err()); + assert_eq!(fixture.state.lock().expect("state").sent.len(), 1); +} + +// A malicious or broken relay must not turn its response body into unbounded +// allocation, leak an error body, or redirect a signed request to another host. +#[tokio::test] +async fn relay_responses_are_bounded_and_redirects_and_error_bodies_are_rejected() { + use axum::http::StatusCode; + for (status, body) in [ + (StatusCode::OK, "x".repeat(1024 * 1024 + 1)), + (StatusCode::FORBIDDEN, "private relay diagnostic".into()), + ( + StatusCode::TEMPORARY_REDIRECT, + "private redirect body".into(), + ), + ] { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("listener"); + let url = format!("http://{}", listener.local_addr().expect("address")); + let app = Router::new().route( + "/query", + post(move || { + let body = body.clone(); + async move { (status, [("location", "https://invalid.example/")], body) } + }), + ); + let task = tokio::spawn(async move { + axum::serve(listener, app).await.expect("relay"); + }); + let result = Relay::new(url, keys(2), None) + .expect("client") + .domain(&scope()) + .await; + task.abort(); + let error = result.expect_err("bad relay response").to_string(); + assert!( + !error.contains("private"), + "unscoped relay body must not escape" + ); + if status.is_redirection() { + assert!(error.contains("307")); + } + if status.is_success() { + assert!(error.contains("too large")); + } + } +} + +// Run the actual executable on macOS/Unix, including startup, JSON readiness, +// keyless child environments, and Ctrl-C cleanup. The relay still uses only +// fixture keys; no user's Buzz account or deployed channel is touched. +#[tokio::test] +#[ignore = "build buzz and set BUZZ_TEST_BROKER_BIN to its absolute path"] +async fn broker_executable_smoke() { + use tokio::io::{AsyncBufReadExt, BufReader}; + use tokio::process::Command as Process; + + let binary = std::env::var("BUZZ_TEST_BROKER_BIN").expect("built buzz path"); + let fixture = Fixture::new(false).await; + let mut child = Process::new(&binary) + .env_clear() + .env("BUZZ_PRIVATE_KEY", keys(2).secret_key().to_secret_hex()) + .args([ + "--relay", + &fixture.url, + "broker", + "serve", + "--community", + &scope().community.to_string(), + "--channel", + &scope().channel.to_string(), + "--relay-key", + &scope().relay_key.to_hex(), + "--requester", + &scope().requester.to_hex(), + ]) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::inherit()) + .kill_on_drop(true) + .spawn() + .expect("broker process"); + let mut stdout = BufReader::new(child.stdout.take().expect("stdout").take(4096)); + let mut line = String::new(); + timeout(Duration::from_secs(10), stdout.read_line(&mut line)) + .await + .expect("startup deadline") + .expect("readiness"); + let ready: Value = serde_json::from_str(&line).expect("JSON readiness"); + let socket = ready["socket"].as_str().expect("socket path"); + for args in [ + vec!["broker", "read", "--socket", socket], + vec![ + "broker", + "reply", + "--socket", + socket, + "--content", + "process smoke", + ], + ] { + let output = timeout( + Duration::from_secs(10), + Process::new(&binary) + .env_clear() + .args(args) + .kill_on_drop(true) + .output(), + ) + .await + .expect("client deadline") + .expect("client process"); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + serde_json::from_slice::(&output.stdout).expect("JSON result"); + } + assert_eq!(fixture.state.lock().expect("state").sent.len(), 1); + let stopped = Process::new("/bin/kill") + .args(["-INT", &child.id().expect("pid").to_string()]) + .status() + .await + .expect("interrupt broker"); + assert!(stopped.success()); + assert!(timeout(Duration::from_secs(5), child.wait()) + .await + .expect("shutdown deadline") + .expect("wait") + .success()); + assert!( + !std::path::Path::new(socket).exists(), + "normal shutdown removes the socket" + ); +} diff --git a/crates/buzz-cli/src/client.rs b/crates/buzz-cli/src/client.rs index 6ade19f1cad..00c50d0845c 100644 --- a/crates/buzz-cli/src/client.rs +++ b/crates/buzz-cli/src/client.rs @@ -81,7 +81,7 @@ const MAX_VIDEO_BYTES: u64 = 500 * 1024 * 1024; /// - `u` tag: the full request URL /// - `method` tag: HTTP method (GET, POST, PUT, DELETE) /// - `payload` tag: SHA-256 hex of the request body (if present) -fn sign_nip98( +pub(crate) fn sign_nip98( keys: &Keys, method: &str, url: &str, diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index b38486417a8..132be48f7b5 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1,4 +1,6 @@ pub mod agent_management; +#[cfg(unix)] +mod broker; mod client; mod commands; mod error; @@ -105,10 +107,11 @@ Buzz CLI — interact with a Buzz relay Configuration (flags override env vars): BUZZ_RELAY_URL Relay base URL [default: http://localhost:3000] - BUZZ_PRIVATE_KEY Nostr private key (hex or nsec) [required] + BUZZ_PRIVATE_KEY Nostr private key (hex or nsec) [relay operations / broker serve] BUZZ_AUTH_TAG NIP-OA auth tag JSON [optional] The 'pack' subcommand runs locally and does not require a relay connection. +On Unix, 'broker read' and 'broker reply' use only a local socket and need no key. Exit codes: 0=ok 1=bad input 2=relay/network error 3=auth error 4=other 5=write conflict Errors are JSON on stderr: {\"error\": \"\", \"message\": \"\"}" @@ -210,6 +213,10 @@ pub enum OutputFormat { #[derive(Subcommand)] enum Cmd { + /// Opt-in, single-channel local broker (no OS isolation yet) + #[cfg(unix)] + #[command(subcommand)] + Broker(broker::Command), /// Draft owner-reviewed agent creation and updates #[command(subcommand)] Agents(AgentsCmd), @@ -2124,6 +2131,15 @@ fn normalize_auth_tag_input(input: &str) -> String { async fn run(cli: Cli) -> Result<(), CliError> { let relay_url = client::normalize_relay_url(&cli.relay); + // These clients talk only to the local socket. Never construct a signer + // or fall back to the relay when the broker is unavailable. + #[cfg(unix)] + if let Cmd::Broker(ref command) = cli.command { + if !matches!(command, broker::Command::Serve(_)) { + return broker::call(command).await; + } + } + // Pack commands are local-only — no relay connection needed. if let Cmd::Pack(ref sub) = cli.command { return match sub { @@ -2167,9 +2183,16 @@ async fn run(cli: Cli) -> Result<(), CliError> { _ => (None, None), }; + #[cfg(unix)] + if let Cmd::Broker(broker::Command::Serve(scope)) = cli.command { + return broker::serve(relay_url, keys, auth_tag, scope).await; + } + let client = BuzzClient::new(relay_url, keys, auth_tag, auth_tag_json)?; match cli.command { + #[cfg(unix)] + Cmd::Broker(_) => Err(CliError::Other("broker command was not dispatched".into())), Cmd::Agents(sub) => commands::agents::dispatch(sub, &client).await, Cmd::Messages(sub) => commands::messages::dispatch(sub, &client, &cli.format).await, Cmd::Channels(sub) => commands::channels::dispatch(sub, &client, &cli.format).await, @@ -2321,6 +2344,8 @@ mod tests { fn command_inventory_is_stable() { let expected_groups: Vec<&str> = vec![ "agents", + #[cfg(unix)] + "broker", "canvas", "channels", "dms", @@ -2384,6 +2409,8 @@ mod tests { } let cmd = Cli::command(); + #[cfg(unix)] + assert_eq!(names(&cmd, "broker"), vec!["read", "reply", "serve"]); assert_eq!( names(&cmd, "agents"), vec![