jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-30 05:51:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add an opt-in single-channel local IFC broker

Browse Source 
Signed-off-by: Jordan Mecom <jm@squareup.com>
This commit is contained in:
Jordan Mecom
2026-09-11 12:38:00 -07:00
parent fe75084a9e
commit bebbc5cd47
9 changed files with 1359 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -983,6 +983,7 @@ dependencies = [
"axum",
"base64 0.22.1",
"buzz-core",
"buzz-ifc",
"buzz-persona",
"buzz-sdk",
"buzz-ws-client",

4
crates/buzz-cli/Cargo.toml
View File

@@ -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"

7
crates/buzz-cli/README.md
View File

@@ -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.

88
crates/buzz-cli/src/broker/README.md Normal file
View File

@@ -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 <community-uuid> \
--channel <channel-uuid> \
--relay-key <relay-signing-public-key> \
--requester <requesting-person-public-key>
```
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 <socket-path> --limit 20
buzz broker reply --socket <socket-path> --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
```

285
crates/buzz-cli/src/broker/mod.rs Normal file
View File

@@ -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<Self, CliError> {
let domain = relay.domain(&scope).await?;
Ok(Self {
relay,
scope,
session: IfcSession::enter(domain),
})
}
async fn check_scope(&mut self) -> Result<ExecutionDomain, CliError> {
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(&current))
.map_err(|_| denied("resource is outside the session"))?;
Ok(current)
}
async fn handle(&mut self, request: Request) -> Result<Value, CliError> {
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<Tag>,
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::<Request>(&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::<Response>(&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<T: DeserializeOwned>(stream: &mut UnixStream, max: usize) -> Result<T, CliError> {
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<T: Serialize>(
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<String>) -> CliError {
CliError::Other(message.into())
}
fn io_error(error: std::io::Error) -> CliError {
denied(format!("local broker I/O failed: {error}"))
}

260
crates/buzz-cli/src/broker/relay.rs Normal file
View File

@@ -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<Tag>,
}
impl Relay {
pub(super) fn new(url: String, keys: Keys, auth: Option<Tag>) -> Result<Self, CliError> {
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::<std::net::IpAddr>()
.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<u8>) -> Result<Value, CliError> {
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<Vec<Event>, 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<ExecutionDomain, CliError> {
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<Vec<Event>, 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<Event, CliError> {
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<Value, CliError> {
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])
}

685
crates/buzz-cli/src/broker/tests.rs Normal file
View File

@@ -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<Vec<String>>, 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<Event> {
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<Event>,
messages: Vec<Event>,
sent: Vec<Event>,
filters: Vec<Value>,
// Change membership while a read is in flight, after its first check.
policy_after_read: Option<Vec<Event>>,
reject_publish: bool,
}
struct Fixture {
url: String,
state: Arc<Mutex<RelayState>>,
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<Arc<Mutex<RelayState>>>,
headers: HeaderMap,
body: Bytes,
) -> Json<Value> {
authenticate(&headers, &body);
let filters: Vec<Value> = 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<Arc<Mutex<RelayState>>>,
headers: HeaderMap,
body: Bytes,
) -> Json<Value> {
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![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::<Response>(&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>(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::<Response>(&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::<Value>(&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"
);
}

2
crates/buzz-cli/src/client.rs
View File

@@ -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,

29
crates/buzz-cli/src/lib.rs
View File

@@ -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\": \"<category>\", \"message\": \"<detail>\"}"
@@ -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![