diff --git a/Cargo.lock b/Cargo.lock index 7a2332cf4ce..39adee8b384 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1659,10 +1659,12 @@ dependencies = [ "bytes", "chrono", "clap", + "crossterm", "diffy", "dirs", "hex", "infer", + "nix 0.31.3", "nostr 0.44.7", "rand 0.10.1", "reqwest 0.13.4", diff --git a/crates/buzz-cli/Cargo.toml b/crates/buzz-cli/Cargo.toml index 59d1bb2cee6..910ddaf0eae 100644 --- a/crates/buzz-cli/Cargo.toml +++ b/crates/buzz-cli/Cargo.toml @@ -86,6 +86,11 @@ 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] +# Safe local TTY handling; remote PTYs remain subordinate to buzz-acp. +crossterm = { version = "0.29", default-features = false } +nix = { version = "0.31", default-features = false, features = ["fs"] } + [dev-dependencies] # Scratch files for channel-templates.json fixtures in tests tempfile = "3" diff --git a/crates/buzz-cli/src/commands/mod.rs b/crates/buzz-cli/src/commands/mod.rs index 7c71016abd8..9a16b30ff8c 100644 --- a/crates/buzz-cli/src/commands/mod.rs +++ b/crates/buzz-cli/src/commands/mod.rs @@ -20,6 +20,7 @@ pub mod projects; pub mod reactions; mod repo_default_branch; pub mod repos; +pub mod sandbox; pub mod social; pub mod upload; pub mod users; diff --git a/crates/buzz-cli/src/commands/sandbox.rs b/crates/buzz-cli/src/commands/sandbox.rs new file mode 100644 index 00000000000..89ca3b36467 --- /dev/null +++ b/crates/buzz-cli/src/commands/sandbox.rs @@ -0,0 +1,464 @@ +//! Explicit, owner-approved terminal access through the community relay. + +use std::io::{BufRead, IsTerminal, Read, Write}; +use std::time::Duration; + +use buzz_sdk::sandbox_attach::{self as wire, Authority, Grant, Pending, Scope}; +use clap::Subcommand; +use nostr::{Event, EventId, PublicKey, Timestamp}; +use serde::Deserialize; +use serde_json::{json, Value}; +use tokio::time::{sleep, timeout}; +use uuid::Uuid; + +use crate::{client::BuzzClient, error::CliError}; + +#[cfg(unix)] +mod terminal; +#[cfg(test)] +mod tests; + +/// Owner approvals and interactive sandbox terminals. +#[derive(Subcommand)] +pub enum SandboxCmd { + /// Attach a local TTY; omitted targets are selected from BUZZ_PASEO_AGENTS + Attach { + /// Channel UUID containing the sandbox thread + #[arg(long, requires = "thread")] + channel: Option, + /// Thread root or a message inside the thread + #[arg(long, requires = "channel")] + thread: Option, + /// Agent public key (hex or npub); otherwise infer from the target + #[arg(long)] + agent: Option, + /// Human owner; hosted gateways must set this explicitly + #[arg(long, env = "BUZZ_PASEO_OWNER")] + owner: Option, + /// Session lifetime in minutes, maximum 60 + #[arg(long, default_value_t = 15, value_parser = clap::value_parser!(u64).range(1..=60))] + minutes: u64, + }, + /// Approve one pending request with the sandbox owner's primary key + Authorize { + /// Pending terminal request event ID + #[arg(long)] + request: EventId, + /// Approval lifetime in minutes, capped by the pending request + #[arg(long, default_value_t = 15, value_parser = clap::value_parser!(u64).range(1..=60))] + minutes: u64, + }, + /// Revoke one owner-signed approval by NIP-09 deletion + Revoke { + /// Owner approval event ID + #[arg(long)] + grant: EventId, + }, +} + +fn failure(message: impl ToString) -> CliError { + CliError::Other(message.to_string()) +} +fn now() -> u64 { + Timestamp::now().as_secs() +} +fn key(value: &str) -> Result { + PublicKey::parse(value).map_err(|_| CliError::Usage("invalid public key".into())) +} +fn tag<'a>(event: &'a Event, name: &str) -> Option<&'a str> { + let mut tags = event + .tags + .iter() + .filter(|tag| tag.as_slice().first().is_some_and(|k| k == name)); + let value = tags.next()?.as_slice().get(1)?.as_str(); + tags.next().is_none().then_some(value) +} + +fn websocket_url(relay: &str) -> Result { + let mut url = url::Url::parse(relay).map_err(failure)?; + if url.scheme() != "https" + || !url.username().is_empty() + || url.password().is_some() + || url.query().is_some() + || url.fragment().is_some() + { + return Err(CliError::Usage( + "sandbox access requires an HTTPS/WSS relay without URL credentials, query or fragment" + .into(), + )); + } + url.set_scheme("wss") + .map_err(|_| failure("invalid relay URL"))?; + Ok(url.to_string()) +} + +async fn query(client: &BuzzClient, mut filter: Value) -> Result, CliError> { + filter["consistency"] = json!("strong"); + let result = timeout(Duration::from_secs(5), client.query(&filter)) + .await + .map_err(|_| failure("terminal authority query timed out"))??; + if result.len() > 2 * 1024 * 1024 { + return Err(failure("terminal query exceeded response budget")); + } + let events: Vec = serde_json::from_str(&result).map_err(failure)?; + let limit = filter["limit"].as_u64().unwrap_or(1) as usize; + if events.len() > limit || events.iter().any(|event| event.verify().is_err()) { + return Err(failure("invalid terminal query response")); + } + Ok(events) +} +async fn one(client: &BuzzClient, filter: Value) -> Result { + let mut events = query(client, filter).await?; + if events.len() != 1 { + return Err(failure("terminal authority is absent or ambiguous")); + } + Ok(events.remove(0)) +} +async fn publish(client: &BuzzClient, event: Event) -> Result<(), CliError> { + let response = timeout(Duration::from_secs(5), client.submit_event(event)) + .await + .map_err(|_| failure("terminal publish timed out; acceptance is unknown"))??; + super::parse_write_response(&response, "terminal metadata changed concurrently")?; + Ok(()) +} +async fn relay_key(client: &BuzzClient) -> Result { + let raw = timeout(Duration::from_secs(5), client.get_public("/info")) + .await + .map_err(|_| failure("relay metadata timed out"))??; + let info: Value = serde_json::from_str(&raw).map_err(failure)?; + key(info["pubkey"] + .as_str() + .ok_or_else(|| failure("relay metadata has no signing key"))?) +} +async fn current_pending(client: &BuzzClient, pending: &Pending) -> Result<(), CliError> { + let head = one(client, json!({"kinds":[30624],"authors":[pending.requester()],"#d":[format!("terminal:{}:request",pending.session())],"limit":1})).await?; + if head.id != pending.id() || Pending::parse(&head, now()).is_err() { + return Err(failure("terminal request was replaced, revoked or expired")); + } + Ok(()) +} +async fn authorized( + client: &BuzzClient, + owner: PublicKey, + relay: PublicKey, + pending: &Pending, + grant: &Grant, +) -> Result<(), CliError> { + timeout(Duration::from_secs(5), async { + let (head, roster, _, delegations, profiles) = tokio::try_join!( + one(client,json!({"kinds":[30624],"authors":[owner],"#d":[format!("terminal:{}:grant",grant.session())],"limit":1})), + one(client,json!({"kinds":[39002],"authors":[relay],"#d":[grant.scope().channel],"limit":1})), + current_pending(client,pending), + query(client,json!({"kinds":[30181],"authors":[owner],"limit":32})), + query(client,json!({"kinds":[0],"authors":[grant.requester()],"limit":1})), + )?; + if head.id != grant.id() || !grant.authorized(pending,Authority{owner,relay,roster:&roster,delegations:&delegations,profiles:&profiles},now()) { + return Err(CliError::Auth("terminal approval, membership or gateway authority ended".into())); + } + Ok(()) + }).await.map_err(|_| failure("terminal authorization refresh timed out"))? +} + +/// Execute a terminal command. Hosted gateways cannot sign owner approvals. +pub async fn dispatch(command: SandboxCmd, client: &BuzzClient) -> Result<(), CliError> { + let ws_url = websocket_url(client.relay_url())?; + if client.auth_tag_owner_hex().is_some() { + return Err(CliError::Auth( + "agent attestations cannot authorize sandbox attachment".into(), + )); + } + match command { + SandboxCmd::Authorize { request, minutes } => { + let event = one(client, json!({"ids":[request],"kinds":[30624],"limit":1})).await?; + if event.id != request { + return Err(failure("relay returned another request")); + } + let pending = Pending::parse(&event, now()).map_err(failure)?; + let owner = client.keys().public_key(); + let expiry = pending.expires_at().min(now().saturating_add(minutes * 60)); + let approval = wire::approve(client.keys(), &pending, expiry).map_err(failure)?; + let grant = Grant::parse(&approval, pending.scope(), now()).map_err(failure)?; + let relay = relay_key(client).await?; + // Check current authority before publication, using the locally signed approval. + let (roster, delegations, profiles, _) = tokio::try_join!( + one( + client, + json!({"kinds":[39002],"authors":[relay],"#d":[pending.scope().channel],"limit":1}) + ), + query( + client, + json!({"kinds":[30181],"authors":[owner],"limit":32}) + ), + query( + client, + json!({"kinds":[0],"authors":[pending.requester()],"limit":1}) + ), + current_pending(client, &pending), + )?; + if !grant.authorized( + &pending, + Authority { + owner, + relay, + roster: &roster, + delegations: &delegations, + profiles: &profiles, + }, + now(), + ) { + return Err(CliError::Auth( + "requester is not an admitted owner/live gateway".into(), + )); + } + publish(client, approval.clone()).await?; + println!( + "{}", + json!({"event_id":approval.id,"accepted":true,"expires_at":expiry}) + ); + Ok(()) + } + SandboxCmd::Revoke { grant } => { + let event = one(client, json!({"ids":[grant],"kinds":[30624],"limit":1})).await?; + if event.id != grant + || event.pubkey != client.keys().public_key() + || !tag(&event, "d").is_some_and(|d| d.ends_with(":grant")) + { + return Err(CliError::Auth( + "only this approval's owner can revoke it".into(), + )); + } + let channel = tag(&event, "h").ok_or_else(|| failure("approval has no channel"))?; + let builder = + nostr::EventBuilder::new(nostr::Kind::EventDeletion, "terminal approval revoked") + .tags([ + nostr::Tag::parse(["e", &grant.to_hex()]).map_err(failure)?, + nostr::Tag::parse(["h", channel]).map_err(failure)?, + ]); + let deletion = client.sign_event(builder)?; + publish(client, deletion.clone()).await?; + println!("{}", json!({"event_id":deletion.id,"accepted":true})); + Ok(()) + } + SandboxCmd::Attach { + channel, + thread, + agent, + owner, + minutes, + } => { + if !std::io::stdin().is_terminal() || !std::io::stdout().is_terminal() { + return Err(CliError::Usage( + "sandbox attach requires a local TTY".into(), + )); + } + let owner = owner + .as_deref() + .map(key) + .transpose()? + .unwrap_or(client.keys().public_key()); + let scope = select_target(client, channel, thread, agent.as_deref()).await?; + let relay = relay_key(client).await?; + let event = wire::request(client.keys(), scope, Uuid::new_v4(), now() + minutes * 60) + .map_err(failure)?; + let pending = Pending::parse(&event, now()).map_err(failure)?; + publish(client, event).await?; + if owner == client.keys().public_key() { + publish( + client, + wire::approve(client.keys(), &pending, pending.expires_at()) + .map_err(failure)?, + ) + .await?; + } else { + eprintln!("Owner approval required for channel {} thread {} agent {} requester {}.\nOn the owner's machine, using the same relay, run:\nbuzz sandbox authorize --request {}",scope.channel,scope.thread,scope.agent,client.keys().public_key(),pending.id()); + } + let grant = loop { + if now() >= pending.expires_at() { + return Err(failure( + "terminal request expired while awaiting owner approval", + )); + } + current_pending(client, &pending).await?; + let approvals = query(client,json!({"kinds":[30624],"authors":[owner],"#d":[format!("terminal:{}:grant",pending.session())],"limit":1})).await?; + if let Some(event) = approvals.first() { + let grant = Grant::parse(event, scope, now()).map_err(failure)?; + authorized(client, owner, relay, &pending, &grant).await?; + break grant; + } + sleep(Duration::from_secs(2)).await; + }; + #[cfg(unix)] + return terminal::attach(client, &ws_url, owner, relay, &pending, &grant).await; + #[cfg(not(unix))] + { + let _ = (ws_url, owner, relay, pending, grant); + Err(CliError::Usage( + "sandbox terminals currently require Unix".into(), + )) + } + } + } +} + +#[derive(Deserialize)] +struct Mapping { + label: String, + agent: String, + channel: Uuid, +} +fn mappings(value: &str) -> Result, CliError> { + if value.len() > 65536 { + return Err(CliError::Usage("BUZZ_PASEO_AGENTS exceeds 64 KiB".into())); + } + let maps: Vec = serde_json::from_str(value).map_err(|_| { + CliError::Usage("BUZZ_PASEO_AGENTS must contain agent/channel mappings".into()) + })?; + if maps.len() > 64 { + return Err(CliError::Usage( + "at most 64 configured agents are supported".into(), + )); + } + for map in &maps { + key(&map.agent)?; + } + Ok(maps) +} +fn label(value: &str) -> String { + value + .chars() + .filter(|c| { + !c.is_control() && !matches!(*c,'\u{202a}'..='\u{202e}'|'\u{2066}'..='\u{2069}') + }) + .take(100) + .collect() +} +fn choose(labels: &[String]) -> Result { + if labels.is_empty() { + return Err(CliError::Usage( + "no targets found; start an agent thread first".into(), + )); + } + for (index, title) in labels.iter().enumerate() { + eprintln!("{}. {}", index + 1, label(title)); + } + eprint!("Choose 1-{}: ", labels.len()); + std::io::stderr().flush().map_err(failure)?; + let mut answer = String::new(); + std::io::stdin() + .lock() + .take(32) + .read_line(&mut answer) + .map_err(failure)?; + answer + .trim() + .parse::() + .ok() + .filter(|n| *n > 0 && *n <= labels.len()) + .map(|n| n - 1) + .ok_or_else(|| CliError::Usage("invalid target selection".into())) +} +async fn select_target( + client: &BuzzClient, + channel: Option, + thread: Option, + agent: Option<&str>, +) -> Result { + let maps = mappings(&std::env::var("BUZZ_PASEO_AGENTS").unwrap_or_else(|_| "[]".into()))?; + let (channel, thread, agent) = match (channel, thread) { + (Some(channel), Some(thread)) => (channel, thread, agent.map(key).transpose()?), + (None, None) => { + let index = choose( + &maps + .iter() + .map(|m| format!("{} ({})", m.label, m.channel)) + .collect::>(), + )?; + let map = &maps[index]; + let agent = key(&map.agent)?; + let roots = query( + client, + json!({"kinds":[9],"#h":[map.channel],"#p":[agent],"limit":64}), + ) + .await?; + let roots: Vec<_> = roots + .into_iter() + .filter(|event| { + tag(event, "h") == Some(map.channel.to_string().as_str()) + && buzz_core::nip10::parse_thread_markers(&event.tags) + .resolve() + .is_none() + }) + .collect(); + let selected = choose( + &roots + .iter() + .map(|event| format!("{} {}", event.id, label(&event.content))) + .collect::>(), + )?; + (map.channel, roots[selected].id, Some(agent)) + } + _ => { + return Err(CliError::Usage( + "provide both --channel and --thread".into(), + )) + } + }; + let event = one( + client, + json!({"kinds":[9],"ids":[thread],"#h":[channel],"limit":1}), + ) + .await?; + if event.id != thread || tag(&event, "h") != Some(channel.to_string().as_str()) { + return Err(failure("thread does not belong to channel")); + } + let root = buzz_core::nip10::parse_thread_markers(&event.tags) + .resolve() + .map(|(root, _)| EventId::from_hex(&root).map_err(failure)) + .transpose()? + .unwrap_or(event.id); + let root_event = if root == event.id { + event + } else { + one( + client, + json!({"kinds":[9],"ids":[root],"#h":[channel],"limit":1}), + ) + .await? + }; + if root_event.id != root + || tag(&root_event, "h") != Some(channel.to_string().as_str()) + || buzz_core::nip10::parse_thread_markers(&root_event.tags) + .resolve() + .is_some() + { + return Err(failure("invalid thread root")); + } + let mut configured: Vec<_> = maps + .iter() + .filter(|map| map.channel == channel) + .filter_map(|map| key(&map.agent).ok()) + .filter(|key| { + root_event.tags.iter().any(|tag| { + tag.as_slice().first().is_some_and(|name| name == "p") + && tag.as_slice().get(1) == Some(&key.to_hex()) + }) + }) + .collect(); + configured.sort(); + configured.dedup(); + let inferred = if configured.len() == 1 { + configured.first().copied() + } else if configured.is_empty() { + tag(&root_event, "p").and_then(|p| key(p).ok()) + } else { + None + }; + let agent = agent + .or(inferred) + .ok_or_else(|| CliError::Usage("ambiguous agent; provide --agent".into()))?; + Ok(Scope { + channel, + thread: root, + agent, + }) +} diff --git a/crates/buzz-cli/src/commands/sandbox/terminal.rs b/crates/buzz-cli/src/commands/sandbox/terminal.rs new file mode 100644 index 00000000000..3b63998d947 --- /dev/null +++ b/crates/buzz-cli/src/commands/sandbox/terminal.rs @@ -0,0 +1,251 @@ +//! Raw TTY forwarding with bounded queues and restoration on every return path. + +use std::fs::{File, OpenOptions}; +use std::io::{Read, Write}; +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::*; + +struct Tty(AsyncFd); +impl Tty { + fn open() -> Result { + let file = OpenOptions::new() + .read(true) + .write(true) + .custom_flags(nix::fcntl::OFlag::O_NONBLOCK.bits()) + .open("/dev/tty") + .map_err(failure)?; + Ok(Self(AsyncFd::new(file).map_err(failure)?)) + } + async fn read(&self, bytes: &mut [u8]) -> Result { + loop { + let mut guard = self.0.readable().await.map_err(failure)?; + match guard.try_io(|fd| { + let mut file = fd.get_ref(); + file.read(bytes) + }) { + Ok(result) => return result.map_err(failure), + Err(_) => continue, + } + } + } + async fn write(&self, mut bytes: &[u8]) -> Result<(), CliError> { + while !bytes.is_empty() { + let mut guard = self.0.writable().await.map_err(failure)?; + match guard.try_io(|fd| { + let mut file = fd.get_ref(); + file.write(bytes) + }) { + Ok(Ok(0)) => return Err(failure("terminal output closed")), + Ok(Ok(n)) => bytes = &bytes[n..], + Ok(Err(error)) => return Err(failure(error)), + Err(_) => {} + } + } + Ok(()) + } +} +struct Raw(bool); +impl Raw { + fn enter() -> Result { + crossterm::terminal::enable_raw_mode().map_err(failure)?; + Ok(Self(true)) + } + fn restore(mut self) -> Result<(), CliError> { + crossterm::terminal::disable_raw_mode().map_err(failure)?; + self.0 = false; + Ok(()) + } +} +impl Drop for Raw { + fn drop(&mut self) { + if self.0 { + let _ = crossterm::terminal::disable_raw_mode(); + } + } +} +fn dimensions() -> Result<(u16, u16), CliError> { + let (cols, rows) = crossterm::terminal::size().map_err(failure)?; + Ok((rows.clamp(1, 1000), cols.clamp(1, 1000))) +} +async fn send( + connection: &mut NostrWsConnection, + client: &BuzzClient, + grant: &Grant, + frame: &Frame, + rate: &mut Rate, + clock: Instant, +) -> Result<(), CliError> { + // Backpressure instead of allocating another queue when the rate is exhausted. + while !rate.allow(clock.elapsed().as_millis() as u64) { + sleep(Duration::from_millis(20)).await; + } + let event = + wire::encrypt(client.keys(), grant.scope().agent, grant.scope(), frame).map_err(failure)?; + timeout( + Duration::from_secs(1), + connection.send_raw(&json!(["EVENT", event])), + ) + .await + .map_err(|_| failure("terminal frame send timed out"))? + .map_err(failure) +} + +pub(super) async fn attach( + client: &BuzzClient, + url: &str, + owner: PublicKey, + relay: PublicKey, + pending: &Pending, + grant: &Grant, +) -> Result<(), CliError> { + let tty = Tty::open()?; + let mut connection = timeout( + Duration::from_secs(45), + NostrWsConnection::connect_authenticated(url, client.keys(), None), + ) + .await + .map_err(|_| failure("terminal relay connection timed out"))? + .map_err(failure)?; + let subscription = json!(["REQ","sandbox-terminal",{"kinds":[24203],"authors":[grant.scope().agent],"#p":[client.keys().public_key()],"#h":[grant.scope().channel],"#e":[grant.scope().thread],"since":now()}]); + timeout(Duration::from_secs(1), connection.send_raw(&subscription)) + .await + .map_err(|_| failure("terminal subscription timed out"))? + .map_err(failure)?; + authorized(client, owner, relay, pending, grant).await?; + let mut flow = Flow::new(grant.id()); + let clock = Instant::now(); + let mut outgoing = Rate::default(); + let mut incoming = Rate::default(); + let mut size = dimensions()?; + let open = flow + .send( + Packet::Open { + rows: size.0, + cols: size.1, + }, + 0, + ) + .map_err(failure)?; + send(&mut connection, client, grant, &open, &mut outgoing, clock).await?; + let mut raw = None; + let mut input = [0u8; wire::MAX_DATA_BYTES]; + let mut refresh = Instant::now(); + let mut activity = Instant::now(); + let mut heartbeat = Instant::now(); + let mut tick = tokio::time::interval(Duration::from_millis(100)); + let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .map_err(failure)?; + let mut hangup = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup()).map_err(failure)?; + let outcome = loop { + let step:Result=async{ + tokio::select!{ + _=terminate.recv()=>return Ok(true), + _=hangup.recv()=>return Ok(true), + _=tokio::signal::ctrl_c()=>return Ok(true), + result=tty.read(&mut input),if raw.is_some()&&flow.writable()=>{ + let count=result?; + let (bytes,detach)=split_input(&input[..count]); + if !bytes.is_empty(){ + let frame=flow.send(Packet::data(bytes).map_err(failure)?,clock.elapsed().as_millis() as u64).map_err(failure)?; + send(&mut connection,client,grant,&frame,&mut outgoing,clock).await?; + activity=Instant::now(); + } + if count==0||detach{return Ok(true);} + } + message=connection.next_event(Duration::from_secs(2))=>{ + match message{ + Ok(RelayMessage::Event{subscription_id,event})if subscription_id=="sandbox-terminal"=>{ + if !incoming.allow(clock.elapsed().as_millis() as u64){return Ok(false);} + let frame=wire::decrypt(client.keys(),grant.scope().agent,grant.scope(),grant.id(),&event,now()).map_err(failure)?; + let ack=frame.ack;let sequenced=frame.seq!=0; + let packet=flow.receive(frame).map_err(failure)?; + if raw.is_none()&&ack>=1{ + eprintln!("Connected. Ctrl-] detaches. Approval expires at {}.",grant.expires_at()); + raw=Some(Raw::enter()?); + } + match packet{ + Some(Packet::Data{data})=>{ + let bytes=Packet::Data{data}.bytes().map_err(failure)?; + timeout(Duration::from_secs(2),tty.write(&bytes)).await.map_err(|_|failure("local terminal stopped reading"))??; + if !bytes.is_empty(){activity=Instant::now();} + } + Some(Packet::Close)=>return Ok(true), + Some(Packet::Open{..}|Packet::Resize{..})=>return Err(failure("agent sent an invalid terminal control")), + _=>{}, + } + if sequenced{send(&mut connection,client,grant,&flow.acknowledgement(),&mut outgoing,clock).await?;} + } + Ok(RelayMessage::Ok(response))if !response.accepted=>return Err(failure(format!("relay rejected terminal frame: {}",response.message))), + Ok(RelayMessage::Closed{message,..})=>return Err(failure(message)), + Err(WsClientError::Timeout)=>{},Err(error)=>return Err(failure(error)),_=>{}, + } + } + _=tick.tick()=>{ + if now()>=grant.expires_at()||activity.elapsed()>=Duration::from_secs(300){return Ok(true);} + if refresh.elapsed()>=Duration::from_secs(5){ + authorized(client,owner,relay,pending,grant).await?;refresh=Instant::now(); + } + for frame in flow.retries(clock.elapsed().as_millis() as u64).map_err(failure)?{ + send(&mut connection,client,grant,&frame,&mut outgoing,clock).await?; + } + if raw.is_some()&&flow.writable(){ + let current=dimensions()?; + let packet=if current!=size{size=current;Some(Packet::Resize{rows:size.0,cols:size.1})} + else if heartbeat.elapsed()>=Duration::from_secs(5){heartbeat=Instant::now();Some(Packet::data(&[]).map_err(failure)?)}else{None}; + if let Some(packet)=packet{let frame=flow.send(packet,clock.elapsed().as_millis() as u64).map_err(failure)?;send(&mut connection,client,grant,&frame,&mut outgoing,clock).await?;} + } + } + } + Ok(false) + }.await; + match step { + Ok(false) => {} + Ok(true) => break Ok(()), + Err(error) => break Err(error), + } + }; + // Best-effort immediate close; the server independently bounds lifetime, authority and idle. + if flow.writable() { + if let Ok(frame) = flow.send(Packet::Close, clock.elapsed().as_millis() as u64) { + let _ = send(&mut connection, client, grant, &frame, &mut outgoing, clock).await; + } + } + let restored = raw.map(Raw::restore).transpose(); + eprintln!("\r\nSandbox terminal closed."); + match (outcome, restored) { + (Err(error), Err(restore)) => Err(failure(format!( + "{error}; TTY restoration failed: {restore}" + ))), + (Err(error), _) => Err(error), + (_, Err(error)) => Err(error), + _ => Ok(()), + } +} + +fn split_input(bytes: &[u8]) -> (&[u8], bool) { + match bytes.iter().position(|byte| *byte == 0x1d) { + Some(index) => (&bytes[..index], true), + None => (bytes, false), + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn raw_input_preserves_control_bytes_and_detach_is_local() { + let bytes = b"\x03\x1b[A\x00paste\r\n"; + assert_eq!(split_input(bytes), (bytes.as_slice(), false)); + assert_eq!( + split_input(b"before\x1dnever sent"), + (b"before".as_slice(), true) + ); + } +} diff --git a/crates/buzz-cli/src/commands/sandbox/tests.rs b/crates/buzz-cli/src/commands/sandbox/tests.rs new file mode 100644 index 00000000000..71581c01083 --- /dev/null +++ b/crates/buzz-cli/src/commands/sandbox/tests.rs @@ -0,0 +1,398 @@ +use super::*; +use axum::{extract::State, routing::post, Json, Router}; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use std::sync::{Arc, RwLock}; + +type Stored = Arc>>; +async fn serve(State(stored): State, Json(filters): Json>) -> Json> { + let filter = &filters[0]; + assert_eq!( + filter["consistency"], "strong", + "production authority reads must route to the writer" + ); + let events = stored.read().unwrap(); + Json( + events + .iter() + .filter(|event| { + filter["kinds"] + .as_array() + .unwrap() + .iter() + .any(|kind| kind.as_u64() == Some(event.kind.as_u16() as u64)) + && filter.get("authors").is_none_or(|authors| { + authors + .as_array() + .unwrap() + .iter() + .any(|author| author.as_str() == Some(event.pubkey.to_hex().as_str())) + }) + && filter.get("#d").is_none_or(|values| { + values + .as_array() + .unwrap() + .iter() + .any(|value| value.as_str() == tag(event, "d")) + }) + }) + .cloned() + .collect(), + ) +} +async fn fixture(events: Vec) -> (BuzzClient, Stored, tokio::task::JoinHandle<()>) { + let stored = Arc::new(RwLock::new(events)); + let app = Router::new() + .route("/query", post(serve)) + .with_state(stored.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + ( + BuzzClient::new(format!("http://{address}"), Keys::generate(), None, None).unwrap(), + stored, + server, + ) +} +#[tokio::test] +async fn current_authority_refresh_uses_strong_reads_and_detects_withdrawal_removal_and_revocation() +{ + let owner = Keys::generate(); + let agent = Keys::generate(); + let relay = Keys::generate(); + let scope = Scope { + channel: Uuid::new_v4(), + thread: EventId::all_zeros(), + agent: agent.public_key(), + }; + let request = wire::request(&owner, scope, Uuid::new_v4(), now() + 600).unwrap(); + let pending = Pending::parse(&request, now()).unwrap(); + let approval = wire::approve(&owner, &pending, now() + 599).unwrap(); + let grant = Grant::parse(&approval, scope, now()).unwrap(); + let roster = EventBuilder::new(Kind::from(39002), "") + .tags([ + Tag::parse(["d", &scope.channel.to_string()]).unwrap(), + Tag::parse(["p", &owner.public_key().to_hex(), "", "member"]).unwrap(), + Tag::parse(["p", &agent.public_key().to_hex(), "", "member"]).unwrap(), + ]) + .sign_with_keys(&relay) + .unwrap(); + let (client, stored, server) = + fixture(vec![request.clone(), approval.clone(), roster.clone()]).await; + authorized( + &client, + owner.public_key(), + relay.public_key(), + &pending, + &grant, + ) + .await + .unwrap(); + stored + .write() + .unwrap() + .retain(|event| event.id != request.id); + assert!( + authorized( + &client, + owner.public_key(), + relay.public_key(), + &pending, + &grant + ) + .await + .is_err(), + "request withdrawal must end the terminal" + ); + *stored.write().unwrap() = vec![request.clone(), roster.clone()]; + assert!( + authorized( + &client, + owner.public_key(), + relay.public_key(), + &pending, + &grant + ) + .await + .is_err(), + "deleted approval must end the terminal" + ); + let removed = EventBuilder::new(Kind::from(39002), "") + .tag(Tag::parse(["d", &scope.channel.to_string()]).unwrap()) + .sign_with_keys(&relay) + .unwrap(); + *stored.write().unwrap() = vec![request, approval, removed]; + assert!( + authorized( + &client, + owner.public_key(), + relay.public_key(), + &pending, + &grant + ) + .await + .is_err(), + "removal must end the terminal" + ); + *stored.write().unwrap() = vec![]; + assert!( + authorized( + &client, + owner.public_key(), + relay.public_key(), + &pending, + &grant + ) + .await + .is_err(), + "missing authority must fail closed" + ); + server.abort(); +} +#[tokio::test] +async fn plaintext_relays_are_rejected_by_real_dispatch_before_network() { + let client = + BuzzClient::new("http://127.0.0.1:9".into(), Keys::generate(), None, None).unwrap(); + let error = dispatch( + SandboxCmd::Authorize { + request: EventId::all_zeros(), + minutes: 15, + }, + &client, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("HTTPS/WSS")); + for relay in [ + "ws://localhost", + "https://user:secret@example.com", + "https://example.com/?token=x", + "https://example.com/#x", + ] { + assert!(websocket_url(relay).is_err()); + } + assert_eq!( + websocket_url("https://example.com").unwrap(), + "wss://example.com/" + ); +} +#[test] +fn configured_targets_and_untrusted_labels_have_budgets() { + let agent = Keys::generate().public_key(); + let channel = Uuid::new_v4(); + let entry = json!({"id":"agent","label":"Agent","agent":agent,"channel":channel}); + assert_eq!( + mappings(&json!([entry.clone()]).to_string()).unwrap().len(), + 1 + ); + assert!(mappings(&Value::Array(vec![entry; 65]).to_string()).is_err()); + assert!(mappings(&" ".repeat(65537)).is_err()); + assert_eq!(label("bad\x1b[2J\n\u{202e}name"), "bad[2Jname"); + assert_eq!(label(&"x".repeat(1000)).len(), 100); +} + +/// Ordinary cargo/CI runs create a controlling PTY for this production flow. +#[cfg(unix)] +#[tokio::test] +async fn real_tty_forwards_encrypted_raw_bytes_and_restores_after_close() { + use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; + use axum::routing::get; + use buzz_sdk::sandbox_attach::{Flow, Packet}; + if std::env::var_os("BUZZ_SANDBOX_TEST_PTY_CHILD").is_none() { + let driver = r#" +import fcntl, os, pty, select, struct, subprocess, sys, termios, time +master, slave = pty.openpty() +fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack('HHHH', 24, 80, 0, 0)) +original = termios.tcgetattr(slave) +def child(): + os.setsid() + fcntl.ioctl(slave, termios.TIOCSCTTY, 0) +process = subprocess.Popen([sys.argv[1], sys.argv[2], '--exact', '--nocapture'], + stdin=slave, stdout=slave, stderr=slave, preexec_fn=child, + env=dict(os.environ, BUZZ_SANDBOX_TEST_PTY_CHILD='1')) +output = bytearray() +sent = False +deadline = time.monotonic() + 25 +def consume(): + global sent + output.extend(os.read(master, 65536)) + if len(output) > 1048576: + raise RuntimeError('PTY test exceeded its output budget') + if b'REMOTE_OUTPUT_OK' in output and not sent: + os.write(master, b'\x03\x1b[Araw\r\n') + sent = True +try: + while process.poll() is None and time.monotonic() < deadline: + if select.select([master], [], [], .1)[0]: + consume() + process.wait(timeout=2) + while select.select([master], [], [], 0)[0]: + consume() + assert process.returncode == 0 and sent and b'REAL_PTY_SUCCESS' in output, output.decode(errors='replace') + assert original == termios.tcgetattr(slave), 'terminal attributes not restored' + print('Controlling PTY: encrypted control bytes and complete termios restoration passed.') +finally: + if process.poll() is None: + process.kill() + process.wait() + os.close(master) + os.close(slave) +"#; + let executable = std::env::current_exe().unwrap(); + let name = "commands::sandbox::tests::real_tty_forwards_encrypted_raw_bytes_and_restores_after_close"; + let result = std::process::Command::new("python3") + .args(["-c", driver]) + .arg(executable) + .arg(name) + .output() + .unwrap(); + assert!( + result.status.success(), + "PTY driver failed: {}{}", + String::from_utf8_lossy(&result.stdout), + String::from_utf8_lossy(&result.stderr) + ); + return; + } + assert!( + std::io::stdin().is_terminal(), + "PTY child must own a terminal" + ); + async fn receive(socket: &mut WebSocket) -> Value { + let frame = timeout(Duration::from_secs(10), socket.recv()) + .await + .unwrap() + .unwrap() + .unwrap(); + let Message::Text(text) = frame else { + panic!("expected signed relay frame") + }; + serde_json::from_str(text.as_str()).unwrap() + } + let owner = Keys::generate(); + let agent = Keys::generate(); + let relay = Keys::generate(); + let scope = Scope { + channel: Uuid::new_v4(), + thread: EventId::all_zeros(), + agent: agent.public_key(), + }; + let request = wire::request(&owner, scope, Uuid::new_v4(), now() + 600).unwrap(); + let pending = Pending::parse(&request, now()).unwrap(); + let approval = wire::approve(&owner, &pending, now() + 599).unwrap(); + let grant = Grant::parse(&approval, scope, now()).unwrap(); + let roster = EventBuilder::new(Kind::from(39002), "") + .tags([ + Tag::parse(["d", &scope.channel.to_string()]).unwrap(), + Tag::parse(["p", &owner.public_key().to_hex(), "", "member"]).unwrap(), + Tag::parse(["p", &agent.public_key().to_hex(), "", "member"]).unwrap(), + ]) + .sign_with_keys(&relay) + .unwrap(); + let (http, _, authority_server) = fixture(vec![request, approval, roster]).await; + let client = BuzzClient::new(http.relay_url().to_owned(), owner.clone(), None, None).unwrap(); + let grant_id = grant.id(); + let requester = owner.public_key(); + let (done_tx, done_rx) = tokio::sync::oneshot::channel(); + let done = Arc::new(std::sync::Mutex::new(Some(done_tx))); + let router = Router::new().route( + "/", + get(move |upgrade: WebSocketUpgrade| { + let agent = agent.clone(); + let done = done.clone(); + async move { + upgrade.on_upgrade(move |mut socket| async move { + socket + .send(Message::Text( + json!(["AUTH", "test-challenge"]).to_string().into(), + )) + .await + .unwrap(); + let auth = receive(&mut socket).await; + assert_eq!(auth[0], "AUTH"); + let event: Event = serde_json::from_value(auth[1].clone()).unwrap(); + event.verify().unwrap(); + assert_eq!(event.pubkey, requester); + socket + .send(Message::Text( + json!(["OK", event.id, true, ""]).to_string().into(), + )) + .await + .unwrap(); + assert_eq!(receive(&mut socket).await[0], "REQ"); + let event = receive(&mut socket).await; + let event: Event = serde_json::from_value(event[1].clone()).unwrap(); + let mut flow = Flow::new(grant_id); + let open = + wire::decrypt(&agent, requester, scope, grant_id, &event, now()).unwrap(); + assert!(matches!( + flow.receive(open).unwrap(), + Some(Packet::Open { .. }) + )); + let frame = flow + .send(Packet::data(b"REMOTE_OUTPUT_OK\r\n").unwrap(), 0) + .unwrap(); + let event = wire::encrypt(&agent, requester, scope, &frame).unwrap(); + socket + .send(Message::Text( + json!(["EVENT", "sandbox-terminal", event]) + .to_string() + .into(), + )) + .await + .unwrap(); + loop { + let value = receive(&mut socket).await; + let event: Event = serde_json::from_value(value[1].clone()).unwrap(); + let frame = + wire::decrypt(&agent, requester, scope, grant_id, &event, now()) + .unwrap(); + if let Some(packet) = flow.receive(frame).unwrap() { + assert_eq!(packet.bytes().unwrap(), b"\x03\x1b[Araw\r\n"); + break; + } + } + let close = flow.send(Packet::Close, 0).unwrap(); + let event = wire::encrypt(&agent, requester, scope, &close).unwrap(); + socket + .send(Message::Text( + json!(["EVENT", "sandbox-terminal", event]) + .to_string() + .into(), + )) + .await + .unwrap(); + done.lock().unwrap().take().unwrap().send(()).unwrap(); + sleep(Duration::from_millis(300)).await; + }) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/", listener.local_addr().unwrap()); + let websocket_server = + tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + timeout( + Duration::from_secs(15), + terminal::attach( + &client, + &url, + owner.public_key(), + relay.public_key(), + &pending, + &grant, + ), + ) + .await + .unwrap() + .unwrap(); + done_rx.await.unwrap(); + assert!( + !crossterm::terminal::is_raw_mode_enabled().unwrap(), + "raw mode must be restored after peer close" + ); + println!("REAL_PTY_SUCCESS"); + authority_server.abort(); + websocket_server.abort(); +} diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 77ddbc168f3..0aad53dccd8 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -266,6 +266,9 @@ enum Cmd { /// Delegate a separate key to drive your remote agents #[command(subcommand)] Gateway(GatewayCmd), + /// Attach to an explicitly owner-approved sandbox terminal + #[command(subcommand)] + Sandbox(commands::sandbox::SandboxCmd), /// Send, read, search, and manage messages #[command(subcommand)] Messages(MessagesCmd), @@ -2340,6 +2343,7 @@ async fn run(cli: Cli) -> Result<(), CliError> { match cli.command { Cmd::Agents(sub) => commands::agents::dispatch(sub, &client).await, Cmd::Gateway(sub) => commands::gateway::dispatch(sub, &client).await, + Cmd::Sandbox(sub) => commands::sandbox::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, Cmd::Canvas(sub) => commands::channels::dispatch_canvas(sub, &client).await, @@ -2550,6 +2554,7 @@ mod tests { "projects", "reactions", "repos", + "sandbox", "social", "upload", "users", diff --git a/docs/remote-agents.md b/docs/remote-agents.md index 18117d28989..db11201aec3 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -107,6 +107,12 @@ The defining constraint, stated as a design axiom: requires a channel to the substrate. The price is the staleness bounds in §Presence. + Owner-approved [terminal attachment](sandbox-attach-cli.md) also uses only + relay events: stored session metadata and ephemeral encrypted frames. It + expands relay protocol surface without a substrate exec or tunnel. Runtime + and CLI consumers must both be available; a gateway delegation alone does + not authorize a shell. + An agent's identity is a Nostr keypair. The **agent record** on `D` carries: `name`, `relay_url`, the nsec (keyring-hydrated), the NIP-OA `auth` tag attesting owner authorization, `agent_command`/`agent_args` (the ACP agent the diff --git a/docs/sandbox-attach-cli.md b/docs/sandbox-attach-cli.md new file mode 100644 index 00000000000..7dd99c52650 --- /dev/null +++ b/docs/sandbox-attach-cli.md @@ -0,0 +1,61 @@ +# Sandbox terminal CLI + +`buzz sandbox attach` forwards a local terminal through NIP-TA relay events. +It needs the SDK foundation and ACP terminal worker in the agent image. +This CLI PR alone does not enable a terminal in an existing sandbox; rebuild both CLI and agent images after the dependent changes land. +Use an HTTPS relay URL; the terminal connection uses WSS and normal certificate verification. + +Set `BUZZ_PRIVATE_KEY` and `BUZZ_RELAY_URL` on the attaching machine. +The owner, agent, and requester must already have posting membership in the channel. +A hosted gateway must also have a live owner delegation and set `BUZZ_PASEO_OWNER` to the human owner's public key. +That delegation permits requesting approval; an owner-signed approval is required for every terminal session. + +```sh +buzz sandbox attach --channel --thread --agent --owner +``` + +The CLI accepts a thread root or a reply, verifies its channel, and resolves its root. +Omit `--agent` only when configured mappings or the root's single recipient identify it unambiguously. +The default session lifetime is 15 minutes; `--minutes` accepts 1 through 60. + +For a hosted gateway, the CLI prints a pending request ID and waits without forwarding input. +On the owner's machine, using the same relay and the owner's primary signing key: + +```sh +buzz sandbox authorize --request --minutes 15 +``` + +The command checks current membership and gateway authority before signing the approval. +The worker independently verifies the immutable sandbox owner, agent, thread, and channel before accepting it. +An owner attaching with their primary key signs their own per-session approval as part of the attach action. +To end an approval early, run `buzz sandbox revoke --grant ` with its owner's key. + +```mermaid +sequenceDiagram + participant C as Attaching CLI + participant O as Owner CLI + participant R as Relay + participant A as Agent worker + C->>R: Signed pending request + O->>R: Owner approval naming request and requester + C->>R: Encrypted Open + R->>A: Open and current authority + A->>R: Durable acceptance, then Open acknowledgement + R->>C: Acknowledgement and encrypted output + C->>R: Encrypted input and resize + A->>R: Metadata-only end record +``` + +The local TTY enters raw mode only after the worker acknowledges Open. +Control keys and pasted bytes are forwarded unchanged; `Ctrl-]` detaches locally. +Resize is checked every 100 ms. Idle sessions close after five minutes; empty transport heartbeats do not reset idle time. +Both sides refresh current authority every five seconds with a five-second total query deadline, bounding revocation detection to ten seconds plus teardown. +The four-frame/64 KiB window, retry limits, and frame admission budget apply in both directions. +The CLI restores terminal settings after peer close, expiry, input/output failure, or authorization failure. + +For Paseo, configure a terminal profile with `command: "buzz"` and `args: ["sandbox", "attach"]`. +With no target flags, the command offers up to 64 `BUZZ_PASEO_AGENTS` mappings, then at most 64 recent addressed thread roots in the selected channel. +The variable uses the wrapper's existing `id`, `label`, `agent`, and `channel` JSON entries. +This menu does not launch an agent or create a sandbox; start a thread first. +Enabling the Paseo host-tools profile remains an operator choice after these consumers land and images are rebuilt. +No substrate connection, exec endpoint, broker, or tunnel is involved.