From 3382899628588788ca65ce69e5b05069e22dc92e Mon Sep 17 00:00:00 2001 From: Eric Wendland Date: Tue, 19 May 2026 19:21:42 +0200 Subject: [PATCH] Add authorized remote pipe listen --- AGENTS.md | 6 +- README.md | 7 +- crates/geth-cli/src/lib.rs | 37 +++++- crates/geth-control/src/lib.rs | 72 ++++++++++++ crates/geth-node/src/lib.rs | 204 ++++++++++++++++++++++++++++++--- crates/geth/tests/bootstrap.rs | 4 + docs/architecture.md | 6 +- docs/roadmap.md | 7 +- 8 files changed, 322 insertions(+), 21 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index d06808c..3ab0fa3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -169,8 +169,10 @@ Roadmap items should be actionable and checkable: roadmap work. - Pipe listen/connect supports a daemon-lifetime registry. Remote `geth pipe connect --node ` uses the protected Iroh control - ALPN and requires `pipe.connect` on `resource:pipe:`. Iroh byte streams, - TCP/Unix forwarding, and remote listener creation are still roadmap work. + 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. - `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/README.md b/README.md index 8594611..f549333 100644 --- a/README.md +++ b/README.md @@ -139,7 +139,8 @@ 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 ` and +- pipe registry/connect commands: + `geth pipe listen [--node ] [--bearer-secret ]` and `geth pipe connect [--node ] [--bearer-secret ]` `geth peer export/import/list` is for untrusted peer-card exchange. Peer cards @@ -185,6 +186,10 @@ 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. +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. `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/src/lib.rs b/crates/geth-cli/src/lib.rs index 9c4b52e..f0226dd 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -365,6 +365,10 @@ pub enum PubsubCommand { pub enum PipeCommand { Listen { name: String, + #[arg(long)] + node: Option, + #[arg(long)] + bearer_secret: Option, }, Connect { target: String, @@ -801,7 +805,15 @@ fn request_for_command(command: Command) -> Result { }, }, Command::Pipe { command } => match command { - PipeCommand::Listen { name } => ControlRequest::PipeListen { name }, + PipeCommand::Listen { + name, + node, + bearer_secret, + } => ControlRequest::PipeListen { + name, + node, + bearer_secret, + }, PipeCommand::Connect { target, node, @@ -1793,6 +1805,29 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { println!("connected_at_ms: {}", connection.connected_at.0); println!("note: {}", connection.note); } + ControlResponse::PipeRemoteListening { + peer_node_id, + peer_agent_id, + endpoint_id, + listener, + allowed, + reason, + note, + } => { + if let Some(listener) = listener { + println!("listening pipe: {}", listener.name); + println!("peer: {peer_node_id}"); + println!("id: {}", listener.id); + println!("listened_at_ms: {}", listener.listened_at.0); + } else { + println!("pipe listen denied by {peer_node_id}"); + } + println!("agent: {peer_agent_id}"); + println!("endpoint: {endpoint_id}"); + println!("allowed: {allowed}"); + println!("reason: {reason}"); + println!("note: {note}"); + } ControlResponse::PipeRemoteConnected { peer_node_id, peer_agent_id, diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index cf6c734..4869885 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -280,6 +280,8 @@ pub enum ControlRequest { }, PipeListen { name: String, + node: Option, + bearer_secret: Option, }, PipeConnect { target: String, @@ -579,6 +581,15 @@ pub enum ControlResponse { PipeListening { listener: PipeListener, }, + PipeRemoteListening { + peer_node_id: String, + peer_agent_id: String, + endpoint_id: String, + listener: Option, + allowed: bool, + reason: String, + note: String, + }, PipeConnected { connection: PipeConnection, }, @@ -716,6 +727,12 @@ pub enum PeerControlRequest { nonce: String, bearer_proof: Option, }, + PipeListen { + peer_card: PeerCard, + name: String, + nonce: String, + bearer_proof: Option, + }, SshProxyConnect { peer_card: PeerCard, nonce: String, @@ -864,6 +881,18 @@ pub enum PeerControlResponse { nonce: String, note: String, }, + PipeListening { + node_id: String, + agent_id: String, + endpoint_id: String, + remote_endpoint_id: String, + listener: Option, + allowed: bool, + reason: String, + evaluated_ops: usize, + nonce: String, + note: String, + }, SshProxyConnected { node_id: String, agent_id: String, @@ -1163,6 +1192,8 @@ mod tests { let request = ControlRequest::PipeListen { name: "inbox".to_owned(), + node: Some("node:peer".to_owned()), + bearer_secret: Some("bearer:test".to_owned()), }; assert_eq!( decode_request(&encode_request(&request).expect("encode")).expect("decode"), @@ -1256,6 +1287,25 @@ mod tests { response ); + let response = ControlResponse::PipeRemoteListening { + peer_node_id: "node:peer".to_owned(), + peer_agent_id: "agent:peer".to_owned(), + endpoint_id: "endpoint:peer".to_owned(), + listener: Some(PipeListener { + id: "pipe:inbox".into(), + name: "inbox".to_owned(), + listened_at: geth_types::UnixMillis(1), + note: "remote".to_owned(), + }), + allowed: true, + reason: "direct grant".to_owned(), + note: "pipe listen".to_owned(), + }; + assert_eq!( + decode_response(&encode_response(&response).expect("encode")).expect("decode"), + response + ); + let request = ControlRequest::SshProxyConnect { node: "node:peer".to_owned(), bearer_secret: None, @@ -1662,6 +1712,28 @@ mod tests { request ); + let request = PeerControlRequest::PipeListen { + 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(), + }, + }, + name: "inbox".to_owned(), + nonce: "nonce".to_owned(), + bearer_proof: None, + }; + assert_eq!( + decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"), + request + ); + 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 c4fcbc1..3de9b71 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -287,6 +287,11 @@ pub async fn handle_request_async( node: Some(peer_node), bearer_secret, } => pubsub_subscribe_from_peer(node, &peer_node, topic, bearer_secret).await, + ControlRequest::PipeListen { + name, + node: Some(peer_node), + bearer_secret, + } => pipe_listen_on_peer(node, &peer_node, name, bearer_secret).await, ControlRequest::PipeConnect { target, node: Some(peer_node), @@ -582,6 +587,7 @@ async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result Err(NodeError::IrohPeer( @@ -697,6 +703,7 @@ async fn peer_auth_check( | PeerControlResponse::PubsubPublished { .. } | PeerControlResponse::PubsubSubscribed { .. } | PeerControlResponse::PipeConnected { .. } + | PeerControlResponse::PipeListening { .. } | PeerControlResponse::SshProxyConnected { .. } | PeerControlResponse::DocumentSynced { .. } | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer( @@ -844,6 +851,7 @@ async fn cas_fetch_from_peer( | PeerControlResponse::PubsubPublished { .. } | PeerControlResponse::PubsubSubscribed { .. } | PeerControlResponse::PipeConnected { .. } + | PeerControlResponse::PipeListening { .. } | PeerControlResponse::SshProxyConnected { .. } | PeerControlResponse::DocumentSynced { .. } | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer( @@ -1257,6 +1265,53 @@ async fn pipe_connect_to_peer( } } +async fn pipe_listen_on_peer( + node: &LocalNode, + peer_node: &str, + name: String, + bearer_secret: Option, +) -> Result { + geth_pipe::validate_pipe_name(&name)?; + let response = request_peer_control(node, peer_node, "pipe-listen", |peer_card, nonce| { + PeerControlRequest::PipeListen { + peer_card, + name: name.clone(), + nonce: nonce.clone(), + bearer_proof: bearer_proof( + bearer_secret, + &format!("resource:pipe:{name}"), + "pipe.listen", + &nonce, + ), + } + }) + .await?; + match response { + PeerControlResponse::PipeListening { + node_id, + agent_id, + endpoint_id, + listener, + allowed, + reason, + note, + .. + } => Ok(ControlResponse::PipeRemoteListening { + peer_node_id: node_id, + peer_agent_id: agent_id, + endpoint_id, + listener, + allowed, + reason, + note, + }), + PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)), + _ => Err(NodeError::IrohPeer( + "peer returned wrong response type to pipe listen".to_owned(), + )), + } +} + async fn ssh_proxy_connect_to_peer( node: &LocalNode, peer_node: &str, @@ -1862,6 +1917,10 @@ async fn request_peer_control( nonce: response_nonce, .. } + | PeerControlResponse::PipeListening { + nonce: response_nonce, + .. + } | PeerControlResponse::SshProxyConnected { nonce: response_nonce, .. @@ -2513,6 +2572,54 @@ async fn handle_iroh_control_connection( note: "pipe connect authenticated endpoint/card binding and required pipe.connect on the remote pipe resource; byte streams are not implemented yet".to_owned(), } } + PeerControlRequest::PipeListen { + peer_card, + name, + nonce, + bearer_proof, + } => { + geth_pipe::validate_pipe_name(&name)?; + 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:{name}"); + let capability = "pipe.listen".to_owned(); + let explanation = explain_peer_or_bearer( + &store, + peer_card.node_id.as_str(), + &resource, + &capability, + &nonce, + bearer_proof.as_ref(), + )?; + let listener = if explanation.allowed { + Some(record_pipe_listener(&node, name)?) + } else { + None + }; + PeerControlResponse::PipeListening { + 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, + listener, + allowed: explanation.allowed, + 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(), + } + } PeerControlRequest::SshProxyConnect { peer_card, nonce, @@ -3981,22 +4088,14 @@ pub fn handle_request( }) } ControlRequest::PubsubSub { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable), - ControlRequest::PipeListen { name } => { + ControlRequest::PipeListen { + name, node: None, .. + } => { geth_pipe::validate_pipe_name(&name)?; - let listener = PipeListener { - id: format!("pipe:{name}").into(), - name: name.clone(), - listened_at: UnixMillis(geth_store::now_ms()), - note: geth_pipe::local_pipe_runtime_note().to_owned(), - }; - let mut runtime = node - .runtime - .pipes - .lock() - .map_err(|_| NodeError::RuntimeLockPoisoned)?; - runtime.listeners.insert(name, listener.clone()); + let listener = record_pipe_listener(node, name)?; Ok(ControlResponse::PipeListening { listener }) } + ControlRequest::PipeListen { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable), ControlRequest::PipeConnect { target, node: None, .. } => { @@ -4144,6 +4243,22 @@ fn pubsub_messages_for_topic( .collect()) } +fn record_pipe_listener(node: &LocalNode, name: String) -> Result { + let listener = PipeListener { + id: format!("pipe:{name}").into(), + name: name.clone(), + listened_at: UnixMillis(geth_store::now_ms()), + note: geth_pipe::local_pipe_runtime_note().to_owned(), + }; + let mut runtime = node + .runtime + .pipes + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)?; + runtime.listeners.insert(name, listener.clone()); + Ok(listener) +} + fn record_pipe_connection( node: &LocalNode, target: String, @@ -5041,6 +5156,8 @@ mod tests { &right, ControlRequest::PipeListen { name: "inbox".to_owned(), + node: None, + bearer_secret: None, }, ) .expect("right pipe listen"); @@ -5255,6 +5372,30 @@ mod tests { other => panic!("unexpected denied pipe connect response: {other:?}"), } + let denied_pipe_listen = handle_request_async( + &left, + ControlRequest::PipeListen { + name: "remote-inbox".to_owned(), + node: Some(right_card.node_id.to_string()), + bearer_secret: None, + }, + ) + .await + .expect("denied remote pipe listen"); + match denied_pipe_listen { + ControlResponse::PipeRemoteListening { + allowed, + reason, + listener, + .. + } => { + assert!(!allowed); + assert!(listener.is_none()); + assert!(reason.contains("no active direct or group grant")); + } + other => panic!("unexpected denied pipe listen response: {other:?}"), + } + let denied_ssh_proxy = handle_request_async( &left, ControlRequest::SshProxyConnect { @@ -5420,6 +5561,16 @@ mod tests { }, ) .expect("grant left pipe connect"); + handle_request( + &right, + ControlRequest::AuthGrant { + subject: left.node_id.clone(), + resource: "resource:pipe:remote-inbox".to_owned(), + capability: "pipe.listen".to_owned(), + grant_id: Some("grant:left-pipe-listen".to_owned()), + }, + ) + .expect("grant left pipe listen"); handle_request( &right, ControlRequest::AuthGrant { @@ -5652,6 +5803,33 @@ mod tests { other => panic!("unexpected allowed pipe connect response: {other:?}"), } + let remote_pipe_listen = handle_request_async( + &left, + ControlRequest::PipeListen { + name: "remote-inbox".to_owned(), + node: Some(right_card.node_id.to_string()), + bearer_secret: None, + }, + ) + .await + .expect("allowed remote pipe listen"); + match remote_pipe_listen { + ControlResponse::PipeRemoteListening { + allowed, + listener, + reason, + note, + .. + } => { + assert!(allowed); + let listener = listener.expect("remote listener metadata"); + assert_eq!(listener.name, "remote-inbox"); + assert!(reason.contains("direct grant")); + assert!(note.contains("pipe.listen")); + } + other => panic!("unexpected allowed pipe listen response: {other:?}"), + } + let ssh_proxy = handle_request_async( &left, ControlRequest::SshProxyConnect { diff --git a/crates/geth/tests/bootstrap.rs b/crates/geth/tests/bootstrap.rs index 68e09ad..0ed0e94 100644 --- a/crates/geth/tests/bootstrap.rs +++ b/crates/geth/tests/bootstrap.rs @@ -1444,6 +1444,8 @@ fn pipe_listen_connect_uses_local_runtime_registry() { &node, geth_control::ControlRequest::PipeListen { name: "inbox".to_owned(), + node: None, + bearer_secret: None, }, ) .expect("listen pipe"); @@ -1500,6 +1502,8 @@ fn pipe_listen_connect_uses_local_runtime_registry() { &node, geth_control::ControlRequest::PipeListen { name: "../bad".to_owned(), + node: None, + bearer_secret: None, }, ) .is_err() diff --git a/docs/architecture.md b/docs/architecture.md index 1e0c37e..0f6d3e2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -198,8 +198,10 @@ 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. -This is still a control-plane scaffold for names and connection attempts only; -it does not carry bytes or forward sockets yet. +`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-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 4346a50..a51ba5b 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -349,8 +349,11 @@ Goal: add authorized stream-oriented management workflows over Iroh. imported peer over Iroh at the control-plane level. - `[x]` Remote pipe connect requires `pipe.connect` on `resource:pipe:`. - - `[ ]` Remote pipe listen requires `pipe.listen` when remote listener - creation is added. + - `[x]` `geth pipe listen --node ` can register an authorized + daemon-lifetime listener on an imported peer. + - `[x]` Remote pipe listen requires `pipe.listen` on + `resource:pipe:`. + - `[x]` Tests cover denied and allowed remote listener registration. - `[ ]` Pipe connect carries bidirectional byte streams over Iroh. - `[ ]` Streams close cleanly and propagate errors.