diff --git a/AGENTS.md b/AGENTS.md index 6d8ed42..08d95bc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -176,8 +176,11 @@ Roadmap items should be actionable and checkable: `geth pipe connect --node ` uses the protected Iroh control ALPN and requires `pipe.connect` on `resource:pipe:`. Remote `geth pipe listen --node ` requires `pipe.listen` on the same - resource before creating a daemon-lifetime listener on the peer. Iroh byte - streams and TCP/Unix forwarding are still roadmap work. + resource before creating a daemon-lifetime listener on the peer. `geth pipe + send --node ` carries a byte message over the + dedicated `/geth/pipe/1` ALPN when `pipe.connect` is authorized, and + `geth pipe recv ` drains local daemon-lifetime messages. Long-lived + bidirectional streams and TCP/Unix forwarding are still roadmap work. - `geth ssh proxy ` performs an authorized control-plane handshake over the protected Iroh control ALPN and requires `ssh_proxy.connect` on `resource:ssh-proxy:local` before returning proxy metadata. It does not carry diff --git a/Cargo.lock b/Cargo.lock index df67807..7a63392 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1087,11 +1087,13 @@ name = "geth-cli" version = "0.1.0" dependencies = [ "anyhow", + "base64", "clap", "geth-cas", "geth-config", "geth-control", "geth-node", + "geth-pipe", "serde_json", "tokio", ] diff --git a/README.md b/README.md index f51c4d1..6e518ec 100644 --- a/README.md +++ b/README.md @@ -142,9 +142,11 @@ The bootstrap implementation provides: - `geth ssh revocation import [--format jsonl|openssh-krl-spec] [--subject ]` - `geth ssh revocation sync [--bearer-secret ]` - SSH proxy authorization probe: `geth ssh proxy [--bearer-secret ]` -- pipe registry/connect commands: - `geth pipe listen [--node ] [--bearer-secret ]` and - `geth pipe connect [--node ] [--bearer-secret ]` +- pipe registry/message commands: + `geth pipe listen [--node ] [--bearer-secret ]`, + `geth pipe connect [--node ] [--bearer-secret ]`, + `geth pipe send [--node ] [--bearer-secret ]`, + and `geth pipe recv [--peek]` `geth peer export/import/list` is for untrusted peer-card exchange. Peer cards include the Iroh EndpointID plus currently known relay/direct addresses. @@ -198,12 +200,15 @@ requires `pubsub.subscribe` on `resource:pubsub:` before returning the peer's current daemon-lifetime snapshot for that topic. Remote pipe connect uses the same protected Iroh control path and requires `pipe.connect` on `resource:pipe:`. The current prototype records a remote -connection attempt and whether a listener exists; byte streaming and forwarding -are still future work. +connection attempt and whether a listener exists. `geth pipe send + --node ` uses the dedicated `/geth/pipe/1` Iroh ALPN to +write a byte message to an authorized peer listener, and `geth pipe recv ` +drains local daemon-lifetime messages. Remote pipe listen uses the same protected path: `geth pipe listen --node ` requires `pipe.listen` on `resource:pipe:` before registering a daemon-lifetime listener on the -peer. +peer. Long-lived stdin/stdout streaming and socket forwarding are still future +work. `geth ssh proxy ` also uses the protected Iroh control path. The remote peer validates the caller's endpoint/card binding and requires `ssh_proxy.connect` on `resource:ssh-proxy:local` before returning proxy diff --git a/crates/geth-cli/Cargo.toml b/crates/geth-cli/Cargo.toml index c2a0abf..39f1451 100644 --- a/crates/geth-cli/Cargo.toml +++ b/crates/geth-cli/Cargo.toml @@ -7,6 +7,7 @@ license.workspace = true [dependencies] anyhow.workspace = true +base64.workspace = true clap.workspace = true serde_json.workspace = true tokio.workspace = true @@ -14,3 +15,4 @@ geth-config = { path = "../geth-config" } geth-control = { path = "../geth-control" } geth-cas = { path = "../geth-cas" } geth-node = { path = "../geth-node" } +geth-pipe = { path = "../geth-pipe" } diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index 47cad12..ba41f70 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -1,4 +1,5 @@ use anyhow::{Context, Result, bail}; +use base64::Engine; use clap::{Args, Parser, Subcommand}; use geth_config::GethPaths; use geth_control::{ControlRequest, ControlResponse}; @@ -395,6 +396,19 @@ pub enum PipeCommand { #[arg(long)] bearer_secret: Option, }, + Send { + target: String, + message: String, + #[arg(long)] + node: Option, + #[arg(long)] + bearer_secret: Option, + }, + Recv { + name: String, + #[arg(long)] + peek: bool, + }, } #[derive(Debug, Subcommand)] @@ -863,6 +877,18 @@ fn request_for_command(command: Command) -> Result { node, bearer_secret, }, + PipeCommand::Send { + target, + message, + node, + bearer_secret, + } => ControlRequest::PipeSend { + target, + data_base64: base64::engine::general_purpose::STANDARD.encode(message.as_bytes()), + node, + bearer_secret, + }, + PipeCommand::Recv { name, peek } => ControlRequest::PipeRecv { name, peek }, }, Command::Db { command } => match command { DbCommand::Add { name, path } => ControlRequest::DbAdd { name, path }, @@ -1951,6 +1977,55 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { println!("reason: {reason}"); println!("note: {note}"); } + ControlResponse::PipeSent { + message, + listener_found, + note, + } => { + println!("listener_found: {listener_found}"); + if let Some(message) = message { + println!("pipe: {}", message.pipe); + println!("received_at_ms: {}", message.received_at.0); + print_pipe_message_data(&message)?; + } + println!("note: {note}"); + } + ControlResponse::PipeRemoteSent { + peer_node_id, + peer_agent_id, + endpoint_id, + message, + listener_found, + allowed, + reason, + note, + } => { + println!("peer: {peer_node_id}"); + println!("agent: {peer_agent_id}"); + println!("endpoint: {endpoint_id}"); + println!("allowed: {allowed}"); + println!("listener_found: {listener_found}"); + if let Some(message) = message { + println!("pipe: {}", message.pipe); + println!("received_at_ms: {}", message.received_at.0); + } + println!("reason: {reason}"); + println!("note: {note}"); + } + ControlResponse::PipeMessages { + name, + messages, + drained, + note, + } => { + println!("pipe: {name}"); + println!("messages: {}", messages.len()); + println!("drained: {drained}"); + for message in messages { + print_pipe_message_data(&message)?; + } + println!("note: {note}"); + } ControlResponse::SshProxyConnected { peer_node_id, peer_agent_id, @@ -2051,6 +2126,20 @@ fn print_file_conflict(conflict: &geth_cas::FileConflict) { } } +fn print_pipe_message_data(message: &geth_pipe::PipeMessage) -> Result<()> { + let bytes = base64::engine::general_purpose::STANDARD + .decode(&message.data_base64) + .context("decode pipe message")?; + match String::from_utf8(bytes) { + Ok(text) => println!("{text}"), + Err(error) => println!( + "base64:{}", + base64::engine::general_purpose::STANDARD.encode(error.into_bytes()) + ), + } + Ok(()) +} + fn shell_quote_command(command: &[String]) -> String { command .iter() diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index a765712..a793163 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -5,7 +5,7 @@ use geth_discovery::{DiscoveredPeer, PeerCard}; use geth_document::{DocumentResource, DocumentState}; use geth_keychain::{KeychainOp, KeychainOpSignature}; use geth_kv::{KvEntry, KvResource, KvSyncEntry}; -use geth_pipe::{PipeConnection, PipeListener}; +use geth_pipe::{PipeConnection, PipeListener, PipeMessage}; use geth_pubsub::PubsubMessage; use geth_resource::ResourceDescriptor; use geth_secrets::{BearerAccess, BearerChallenge, BearerProof, ResourceMasterSecret}; @@ -298,6 +298,16 @@ pub enum ControlRequest { node: Option, bearer_secret: Option, }, + PipeSend { + target: String, + data_base64: String, + node: Option, + bearer_secret: Option, + }, + PipeRecv { + name: String, + peek: bool, + }, ModuleStub { module: String, command: String, @@ -632,6 +642,27 @@ pub enum ControlResponse { reason: String, note: String, }, + PipeSent { + message: Option, + listener_found: bool, + note: String, + }, + PipeRemoteSent { + peer_node_id: String, + peer_agent_id: String, + endpoint_id: String, + message: Option, + listener_found: bool, + allowed: bool, + reason: String, + note: String, + }, + PipeMessages { + name: String, + messages: Vec, + drained: bool, + note: String, + }, SshProxyConnected { peer_node_id: String, peer_agent_id: String, @@ -988,6 +1019,39 @@ pub enum PeerControlResponse { }, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "kebab-case")] +pub enum PipeWireRequest { + Send { + peer_card: PeerCard, + target: String, + data_base64: String, + nonce: String, + bearer_proof: Option, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "kebab-case")] +pub enum PipeWireResponse { + Sent { + node_id: String, + agent_id: String, + endpoint_id: String, + remote_endpoint_id: String, + message: Box>, + listener_found: bool, + allowed: bool, + reason: String, + evaluated_ops: usize, + nonce: String, + note: String, + }, + Error { + message: String, + }, +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct SyncWatermark { pub stream: String, @@ -1040,6 +1104,26 @@ pub fn decode_peer_response(line: &str) -> Result Result { + let mut line = serde_json::to_string(request)?; + line.push('\n'); + Ok(line) +} + +pub fn decode_pipe_wire_request(line: &str) -> Result { + serde_json::from_str(line).map_err(ControlError::from) +} + +pub fn encode_pipe_wire_response(response: &PipeWireResponse) -> Result { + let mut line = serde_json::to_string(response)?; + line.push('\n'); + Ok(line) +} + +pub fn decode_pipe_wire_response(line: &str) -> Result { + serde_json::from_str(line).map_err(ControlError::from) +} + #[cfg(test)] mod tests { use super::*; @@ -1260,6 +1344,43 @@ mod tests { request ); + let request = ControlRequest::PipeSend { + target: "inbox".to_owned(), + data_base64: "aGVsbG8=".to_owned(), + node: Some("node:peer".to_owned()), + bearer_secret: None, + }; + assert_eq!( + decode_request(&encode_request(&request).expect("encode")).expect("decode"), + request + ); + + let request = ControlRequest::PipeRecv { + name: "inbox".to_owned(), + peek: true, + }; + assert_eq!( + decode_request(&encode_request(&request).expect("encode")).expect("decode"), + request + ); + + let response = ControlResponse::PipeMessages { + name: "inbox".to_owned(), + messages: vec![PipeMessage { + pipe: "inbox".to_owned(), + data_base64: "aGVsbG8=".to_owned(), + received_at: geth_types::UnixMillis(1), + source_node: Some("node:peer".to_owned()), + note: "pipe".to_owned(), + }], + drained: false, + note: "pipe messages".to_owned(), + }; + assert_eq!( + decode_response(&encode_response(&response).expect("encode")).expect("decode"), + response + ); + let request = ControlRequest::PubsubPub { topic: "presence/test".to_owned(), message: "online".to_owned(), @@ -1854,6 +1975,55 @@ mod tests { request ); + let request = PipeWireRequest::Send { + peer_card: PeerCard { + node_id: "node:caller".into(), + agent_id: "agent:caller".into(), + endpoints: Vec::new(), + issued_at: geth_types::UnixMillis(1), + signature: geth_discovery::SignatureMetadata { + namespace: "geth.peer-card.v1@geth.local".to_owned(), + signer: "agent:caller".to_owned(), + public_key: "key".to_owned(), + signature: "sig".to_owned(), + }, + }, + target: "inbox".to_owned(), + data_base64: "aGVsbG8=".to_owned(), + nonce: "nonce".to_owned(), + bearer_proof: None, + }; + assert_eq!( + decode_pipe_wire_request(&encode_pipe_wire_request(&request).expect("encode")) + .expect("decode"), + request + ); + + let response = PipeWireResponse::Sent { + node_id: "node:peer".to_owned(), + agent_id: "agent:peer".to_owned(), + endpoint_id: "endpoint:peer".to_owned(), + remote_endpoint_id: "endpoint:caller".to_owned(), + message: Box::new(Some(PipeMessage { + pipe: "inbox".to_owned(), + data_base64: "aGVsbG8=".to_owned(), + received_at: geth_types::UnixMillis(1), + source_node: Some("node:caller".to_owned()), + note: "pipe".to_owned(), + })), + listener_found: true, + allowed: true, + reason: "direct grant".to_owned(), + evaluated_ops: 1, + nonce: "nonce".to_owned(), + note: "pipe wire".to_owned(), + }; + assert_eq!( + decode_pipe_wire_response(&encode_pipe_wire_response(&response).expect("encode")) + .expect("decode"), + response + ); + let response = PeerControlResponse::SshProxyConnected { node_id: "node:peer".to_owned(), agent_id: "agent:peer".to_owned(), diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 7eae759..b4d2b82 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -9,7 +9,8 @@ use geth_cas::{ use geth_config::{GethConfig, GethPaths, RelayMode}; use geth_control::{ CasBlob, CasProvider, ControlRequest, ControlResponse, KeychainStatusResponse, NodeIdResponse, - PeerControlRequest, PeerControlResponse, StatusResponse, SyncWatermark, + PeerControlRequest, PeerControlResponse, PipeWireRequest, PipeWireResponse, StatusResponse, + SyncWatermark, }; use geth_crypto::AgentKey; use geth_db::DbResource; @@ -21,7 +22,7 @@ use geth_document::{DocumentResource, DocumentState}; use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode}; use geth_keychain::{KeychainOp, KeychainOpKind, KeychainOpSignature}; use geth_kv::{KvEntry, KvResource, KvSyncEntry}; -use geth_pipe::{PipeConnection, PipeListener}; +use geth_pipe::{PipeConnection, PipeListener, PipeMessage}; use geth_pubsub::PubsubMessage; use geth_resource::ResourceDescriptor; use geth_secrets::{BearerAccess, BearerChallenge, BearerProof, ResourceMasterSecret}; @@ -150,10 +151,12 @@ struct PubsubRuntime { struct PipeRuntime { listeners: BTreeMap, connections: VecDeque, + messages: VecDeque, } const PUBSUB_RING_LIMIT: usize = 256; const PIPE_CONNECTION_RING_LIMIT: usize = 256; +const PIPE_MESSAGE_RING_LIMIT: usize = 1024; #[derive(Debug, serde::Deserialize, serde::Serialize)] struct LiveSyncCursor { @@ -307,6 +310,12 @@ pub async fn handle_request_async( node: Some(peer_node), bearer_secret, } => pipe_connect_to_peer(node, &peer_node, target, bearer_secret).await, + ControlRequest::PipeSend { + target, + data_base64, + node: Some(peer_node), + bearer_secret, + } => pipe_send_to_peer(node, &peer_node, target, data_base64, bearer_secret).await, ControlRequest::SshProxyConnect { node: peer_node, bearer_secret, @@ -1472,6 +1481,57 @@ async fn pipe_listen_on_peer( } } +async fn pipe_send_to_peer( + node: &LocalNode, + peer_node: &str, + target: String, + data_base64: String, + bearer_secret: Option, +) -> Result { + geth_pipe::validate_pipe_name(&target)?; + base64::engine::general_purpose::STANDARD + .decode(&data_base64) + .map_err(|error| NodeError::IrohPeer(format!("invalid pipe base64 payload: {error}")))?; + let response = request_pipe_wire(node, peer_node, "pipe-send", |peer_card, nonce| { + PipeWireRequest::Send { + peer_card, + target: target.clone(), + data_base64: data_base64.clone(), + nonce: nonce.clone(), + bearer_proof: bearer_proof( + bearer_secret, + &format!("resource:pipe:{target}"), + "pipe.connect", + &nonce, + ), + } + }) + .await?; + match response { + PipeWireResponse::Sent { + node_id, + agent_id, + endpoint_id, + message, + listener_found, + allowed, + reason, + note, + .. + } => Ok(ControlResponse::PipeRemoteSent { + peer_node_id: node_id, + peer_agent_id: agent_id, + endpoint_id, + message: *message, + listener_found, + allowed, + reason, + note, + }), + PipeWireResponse::Error { message } => Err(NodeError::IrohPeer(message)), + } +} + async fn ssh_proxy_connect_to_peer( node: &LocalNode, peer_node: &str, @@ -2125,6 +2185,75 @@ async fn request_peer_control( } } +async fn request_pipe_wire( + node: &LocalNode, + peer_node: &str, + operation: &str, + build_request: impl FnOnce(PeerCard, String) -> PipeWireRequest, +) -> Result { + let store = Store::open(&node.paths.metadata_db())?; + let stored = store + .get_peer_card(peer_node)? + .ok_or_else(|| NodeError::PeerNotFound(peer_node.to_owned()))?; + let peer_card: PeerCard = serde_json::from_str(&stored.card_json)?; + peer_card.validate_candidate()?; + let candidate = peer_card + .endpoints + .first() + .ok_or(geth_discovery::DiscoveryError::MissingEndpoint)?; + ensure_peer_card_matches_endpoint(&peer_card, &candidate.endpoint_id)?; + let node_addr = iroh_node_addr_from_candidate(candidate)?; + let endpoint = node + .iroh_endpoint + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)? + .clone() + .ok_or(NodeError::IrohEndpointUnavailable)?; + let self_card = local_peer_card(node, DiscoverySource::PeerExchange, true).await?; + let nonce = geth_crypto::blake3_hex( + format!( + "{}\0{}\0{}\0{}", + node.node_id, + peer_node, + operation, + geth_store::now_ms() + ) + .as_bytes(), + ); + let request = build_request(self_card, nonce.clone()); + let conn = endpoint + .endpoint() + .connect(node_addr, geth_iroh::ALPN_PIPE) + .await + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + let (mut send, mut recv) = conn + .open_bi() + .await + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + send.write_all(geth_control::encode_pipe_wire_request(&request)?.as_bytes()) + .await + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + send.finish() + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + let bytes = recv + .read_to_end(16 * 1024 * 1024) + .await + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + let text = + std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?; + let response = geth_control::decode_pipe_wire_response(text)?; + match &response { + PipeWireResponse::Sent { + nonce: response_nonce, + .. + } if response_nonce == &nonce => Ok(response), + PipeWireResponse::Error { .. } => Ok(response), + _ => Err(NodeError::IrohPeer(format!( + "peer {operation} response did not match request" + ))), + } +} + fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) { let raw_endpoint = endpoint.endpoint(); tokio::spawn(async move { @@ -2279,6 +2408,15 @@ async fn handle_iroh_control_connection( .map_err(|error| NodeError::IrohPeer(error.to_string()))?; let text = std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?; + if alpn == display_alpn(geth_iroh::ALPN_PIPE.to_vec()) { + let response = handle_pipe_wire_request(&node, &remote_endpoint_id, text)?; + send.write_all(geth_control::encode_pipe_wire_response(&response)?.as_bytes()) + .await + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + send.finish() + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + return Ok(()); + } let response = match geth_control::decode_peer_request(text)? { PeerControlRequest::Ping { peer_card, nonce } => { peer_card.validate_candidate()?; @@ -2818,7 +2956,7 @@ async fn handle_iroh_control_connection( Some(record_pipe_connection( &node, target, - "remote pipe connect over protected Iroh control path; byte streams are not implemented yet".to_owned(), + "remote pipe connect over protected Iroh control path; byte messages use the dedicated Iroh pipe ALPN".to_owned(), )?) } else { None @@ -2833,7 +2971,7 @@ async fn handle_iroh_control_connection( reason: explanation.reason, evaluated_ops: explanation.evaluated_ops, nonce, - note: "pipe connect authenticated endpoint/card binding and required pipe.connect on the remote pipe resource; byte streams are not implemented yet".to_owned(), + note: "pipe connect authenticated endpoint/card binding and required pipe.connect on the remote pipe resource; byte messages use the dedicated Iroh pipe ALPN".to_owned(), } } PeerControlRequest::PipeListen { @@ -2881,7 +3019,7 @@ async fn handle_iroh_control_connection( reason: explanation.reason, evaluated_ops: explanation.evaluated_ops, nonce, - note: "pipe listen authenticated endpoint/card binding and required pipe.listen on the remote pipe resource; byte streams are not implemented yet".to_owned(), + note: "pipe listen authenticated endpoint/card binding and required pipe.listen on the remote pipe resource; listeners receive daemon-lifetime byte messages".to_owned(), } } PeerControlRequest::SshProxyConnect { @@ -3078,6 +3216,74 @@ async fn handle_iroh_control_connection( Ok(()) } +fn handle_pipe_wire_request( + node: &LocalNode, + remote_endpoint_id: &str, + text: &str, +) -> Result { + match geth_control::decode_pipe_wire_request(text)? { + PipeWireRequest::Send { + peer_card, + target, + data_base64, + nonce, + bearer_proof, + } => { + geth_pipe::validate_pipe_name(&target)?; + base64::engine::general_purpose::STANDARD + .decode(&data_base64) + .map_err(|error| NodeError::IrohPeer(format!("invalid pipe payload: {error}")))?; + peer_card.validate_candidate()?; + ensure_peer_card_matches_endpoint(&peer_card, remote_endpoint_id)?; + let discovered = DiscoveredPeer::candidate( + peer_card.clone(), + UnixMillis(geth_store::now_ms()), + DiscoverySource::PeerExchange, + )?; + let store = Store::open(&node.paths.metadata_db())?; + store.upsert_peer_card(&StoredPeerCard { + peer_id: peer_card.node_id.to_string(), + card_json: serde_json::to_string(&peer_card)?, + updated_at_ms: discovered.discovered_at.0, + })?; + let resource = format!("resource:pipe:{target}"); + let capability = "pipe.connect".to_owned(); + let explanation = explain_peer_or_bearer( + &store, + peer_card.node_id.as_str(), + &resource, + &capability, + &nonce, + bearer_proof.as_ref(), + )?; + let (message, listener_found) = if explanation.allowed { + record_pipe_message( + node, + target, + data_base64, + Some(peer_card.node_id.to_string()), + "remote pipe byte message over dedicated Iroh pipe ALPN".to_owned(), + )? + } else { + (None, false) + }; + Ok(PipeWireResponse::Sent { + node_id: node.node_id.clone(), + agent_id: node.agent_id.clone(), + endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(), + remote_endpoint_id: remote_endpoint_id.to_owned(), + message: Box::new(message), + listener_found, + allowed: explanation.allowed, + reason: explanation.reason, + evaluated_ops: explanation.evaluated_ops, + nonce, + note: "pipe send authenticated endpoint/card binding and required pipe.connect on the remote pipe resource; payload used the dedicated Iroh pipe ALPN".to_owned(), + }) + } + } +} + fn iroh_node_addr_from_candidate( candidate: &EndpointCandidate, ) -> Result { @@ -4389,6 +4595,40 @@ pub fn handle_request( ControlRequest::PipeConnect { node: Some(_), .. } => { Err(NodeError::IrohEndpointUnavailable) } + ControlRequest::PipeSend { + target, + data_base64, + node: None, + .. + } => { + geth_pipe::validate_pipe_name(&target)?; + base64::engine::general_purpose::STANDARD + .decode(&data_base64) + .map_err(|error| NodeError::IrohPeer(format!("invalid pipe payload: {error}")))?; + let (message, listener_found) = record_pipe_message( + node, + target, + data_base64, + Some(node.node_id.clone()), + "local pipe byte message".to_owned(), + )?; + Ok(ControlResponse::PipeSent { + message, + listener_found, + note: geth_pipe::local_pipe_runtime_note().to_owned(), + }) + } + ControlRequest::PipeSend { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable), + ControlRequest::PipeRecv { name, peek } => { + geth_pipe::validate_pipe_name(&name)?; + let messages = pipe_messages(node, &name, !peek)?; + Ok(ControlResponse::PipeMessages { + name, + messages, + drained: !peek, + note: geth_pipe::local_pipe_runtime_note().to_owned(), + }) + } ControlRequest::SshProxyConnect { .. } => Err(NodeError::IrohEndpointUnavailable), ControlRequest::ModuleStub { module, command } => { Ok(ControlResponse::NotImplemented { module, command }) @@ -4562,6 +4802,54 @@ fn record_pipe_connection( Ok(connection) } +fn record_pipe_message( + node: &LocalNode, + target: String, + data_base64: String, + source_node: Option, + note: String, +) -> Result<(Option, bool), NodeError> { + let mut runtime = node + .runtime + .pipes + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)?; + let listener_found = runtime.listeners.contains_key(&target); + if !listener_found { + return Ok((None, false)); + } + let message = PipeMessage { + pipe: target, + data_base64, + received_at: UnixMillis(geth_store::now_ms()), + source_node, + note, + }; + runtime.messages.push_back(message.clone()); + while runtime.messages.len() > PIPE_MESSAGE_RING_LIMIT { + runtime.messages.pop_front(); + } + Ok((Some(message), true)) +} + +fn pipe_messages(node: &LocalNode, name: &str, drain: bool) -> Result, NodeError> { + let mut runtime = node + .runtime + .pipes + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)?; + let messages = runtime + .messages + .iter() + .filter(|message| message.pipe == name) + .cloned() + .collect::>(); + if drain { + runtime.messages.retain(|message| message.pipe != name); + } + Ok(messages) +} + fn ensure_local_document(store: &Store, name: &str) -> Result { if let Some(document) = store.get_document_resource_by_name(name)? { return Ok(document); @@ -6554,11 +6842,64 @@ mod tests { assert!(allowed); assert!(connection.local_listener_found); assert!(reason.contains("direct grant")); - assert!(note.contains("byte streams are not implemented yet")); + assert!(note.contains("pipe.connect")); } other => panic!("unexpected allowed pipe connect response: {other:?}"), } + let remote_pipe_send = handle_request_async( + &left, + ControlRequest::PipeSend { + target: "inbox".to_owned(), + data_base64: "aGVsbG8gZnJvbSBsZWZ0".to_owned(), + node: Some(right_card.node_id.to_string()), + bearer_secret: None, + }, + ) + .await + .expect("allowed remote pipe send"); + match remote_pipe_send { + ControlResponse::PipeRemoteSent { + allowed, + listener_found, + message, + reason, + note, + .. + } => { + assert!(allowed); + assert!(listener_found); + let message = message.expect("remote pipe message"); + assert_eq!(message.pipe, "inbox"); + assert_eq!(message.data_base64, "aGVsbG8gZnJvbSBsZWZ0"); + assert!(reason.contains("direct grant")); + assert!(note.contains("Iroh pipe ALPN")); + } + other => panic!("unexpected allowed pipe send response: {other:?}"), + } + let right_messages = handle_request( + &right, + ControlRequest::PipeRecv { + name: "inbox".to_owned(), + peek: false, + }, + ) + .expect("right pipe recv after remote send"); + match right_messages { + ControlResponse::PipeMessages { + messages, drained, .. + } => { + assert!(drained); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].data_base64, "aGVsbG8gZnJvbSBsZWZ0"); + assert_eq!( + messages[0].source_node.as_deref(), + Some(left.node_id.as_str()) + ); + } + other => panic!("unexpected right pipe messages response: {other:?}"), + } + let remote_pipe_listen = handle_request_async( &left, ControlRequest::PipeListen { diff --git a/crates/geth-pipe/src/lib.rs b/crates/geth-pipe/src/lib.rs index b18a2d1..36e86d4 100644 --- a/crates/geth-pipe/src/lib.rs +++ b/crates/geth-pipe/src/lib.rs @@ -24,6 +24,15 @@ pub struct PipeConnection { pub note: String, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PipeMessage { + pub pipe: String, + pub data_base64: String, + pub received_at: UnixMillis, + pub source_node: Option, + pub note: String, +} + #[derive(Debug, thiserror::Error)] pub enum PipeError { #[error("invalid pipe name or target: {0}")] @@ -48,7 +57,7 @@ pub fn pipe_roadmap() -> &'static str { #[must_use] pub fn local_pipe_runtime_note() -> &'static str { - "local daemon registry only; byte streams and Iroh transport are not implemented yet" + "daemon-lifetime pipe runtime; byte messages are buffered locally and remote writes use the Iroh pipe ALPN" } #[cfg(test)] diff --git a/crates/geth/tests/bootstrap.rs b/crates/geth/tests/bootstrap.rs index 2be221a..660e33f 100644 --- a/crates/geth/tests/bootstrap.rs +++ b/crates/geth/tests/bootstrap.rs @@ -1543,7 +1543,7 @@ fn pipe_listen_connect_uses_local_runtime_registry() { geth_control::ControlResponse::PipeListening { listener } => { assert_eq!(listener.id.to_string(), "pipe:inbox"); assert_eq!(listener.name, "inbox"); - assert!(listener.note.contains("local daemon registry")); + assert!(listener.note.contains("daemon-lifetime pipe runtime")); } other => panic!("unexpected response: {other:?}"), } @@ -1561,11 +1561,63 @@ fn pipe_listen_connect_uses_local_runtime_registry() { geth_control::ControlResponse::PipeConnected { connection } => { assert_eq!(connection.target, "inbox"); assert!(connection.local_listener_found); - assert!( - connection - .note - .contains("Iroh transport are not implemented") - ); + assert!(connection.note.contains("pipe runtime")); + } + other => panic!("unexpected response: {other:?}"), + } + + let response = geth_node::handle_request( + &node, + geth_control::ControlRequest::PipeSend { + target: "inbox".to_owned(), + data_base64: "aGVsbG8gcGlwZQ==".to_owned(), + node: None, + bearer_secret: None, + }, + ) + .expect("send pipe message"); + match response { + geth_control::ControlResponse::PipeSent { + message, + listener_found, + .. + } => { + assert!(listener_found); + assert_eq!(message.expect("message").data_base64, "aGVsbG8gcGlwZQ=="); + } + other => panic!("unexpected response: {other:?}"), + } + + let response = geth_node::handle_request( + &node, + geth_control::ControlRequest::PipeRecv { + name: "inbox".to_owned(), + peek: false, + }, + ) + .expect("receive pipe messages"); + match response { + geth_control::ControlResponse::PipeMessages { + messages, drained, .. + } => { + assert!(drained); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].data_base64, "aGVsbG8gcGlwZQ=="); + } + other => panic!("unexpected response: {other:?}"), + } + + let response = geth_node::handle_request( + &node, + geth_control::ControlRequest::PipeRecv { + name: "inbox".to_owned(), + peek: false, + }, + ) + .expect("receive drained pipe messages"); + match response { + geth_control::ControlResponse::PipeMessages { messages, .. } => { + assert!(messages.is_empty()); } other => panic!("unexpected response: {other:?}"), } diff --git a/docs/architecture.md b/docs/architecture.md index 9d29190..09ce78d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -203,15 +203,18 @@ authorized peer's current snapshot for that topic over the same protected path when the caller has `pubsub.subscribe` on `resource:pubsub:`. Iroh-gossip replication and private topics are future work. -`geth-pipe` currently supports `pipe listen/connect` against a daemon-lifetime -registry. `geth pipe connect --node ` sends an authorized remote -connect request over the protected Iroh control ALPN. The remote daemon validates -endpoint/card binding and requires `pipe.connect` on `resource:pipe:` -before recording the connection attempt and reporting whether a listener exists. -`geth pipe listen --node ` can also ask a peer to register a -daemon-lifetime listener after checking `pipe.listen` on the same resource. This -is still a control-plane scaffold for names, listeners, and connection attempts -only; it does not carry bytes or forward sockets yet. +`geth-pipe` currently supports `pipe listen/connect/send/recv` against a +daemon-lifetime runtime. `geth pipe connect --node ` sends an +authorized remote connect request over the protected Iroh control ALPN. The +remote daemon validates endpoint/card binding and requires `pipe.connect` on +`resource:pipe:` before recording the connection attempt and reporting +whether a listener exists. `geth pipe send --node ` +uses the dedicated `/geth/pipe/1` ALPN to write a byte message to a peer +listener after the same endpoint/card and capability checks. `geth pipe recv +` drains local daemon-lifetime messages. `geth pipe listen --node +` can also ask a peer to register a daemon-lifetime listener after +checking `pipe.listen` on the same resource. Long-lived stdin/stdout streams and +TCP/Unix forwarding are still future work. `geth-ssh-proxy` currently defines proxy target and connection metadata. The daemon can authorize a remote proxy attempt over the protected Iroh control ALPN diff --git a/docs/roadmap.md b/docs/roadmap.md index b1196c8..ee51abc 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -364,6 +364,11 @@ Goal: add authorized stream-oriented management workflows over Iroh. - `[x]` Remote pipe listen requires `pipe.listen` on `resource:pipe:`. - `[x]` Tests cover denied and allowed remote listener registration. + - `[x]` `geth pipe send --node ` carries a byte + message over the dedicated `/geth/pipe/1` Iroh ALPN. + - `[x]` `geth pipe recv ` drains daemon-lifetime pipe messages. + - `[x]` Remote pipe send requires `pipe.connect` on + `resource:pipe:`. - `[ ]` Pipe connect carries bidirectional byte streams over Iroh. - `[ ]` Streams close cleanly and propagate errors.