Add authorized remote pipe listen

crates/geth-cli/src/lib.rs

@@ -365,6 +365,10 @@ pub enum PubsubCommand {
 pub enum PipeCommand {
     Listen {
         name: String,
+        #[arg(long)]
+        node: Option<String>,
+        #[arg(long)]
+        bearer_secret: Option<String>,
     },
     Connect {
         target: String,
@@ -801,7 +805,15 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
             },
         },
         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,

crates/geth-control/src/lib.rs

@@ -280,6 +280,8 @@ pub enum ControlRequest {
     },
     PipeListen {
         name: String,
+        node: Option<String>,
+        bearer_secret: Option<String>,
     },
     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<PipeListener>,
+        allowed: bool,
+        reason: String,
+        note: String,
+    },
     PipeConnected {
         connection: PipeConnection,
     },
@@ -716,6 +727,12 @@ pub enum PeerControlRequest {
         nonce: String,
         bearer_proof: Option<BearerProof>,
     },
+    PipeListen {
+        peer_card: PeerCard,
+        name: String,
+        nonce: String,
+        bearer_proof: Option<BearerProof>,
+    },
     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<PipeListener>,
+        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(),

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<ControlResponse,
         | PeerControlResponse::PubsubPublished { .. }
         | PeerControlResponse::PubsubSubscribed { .. }
         | PeerControlResponse::PipeConnected { .. }
+        | PeerControlResponse::PipeListening { .. }
         | PeerControlResponse::SshProxyConnected { .. }
         | PeerControlResponse::DocumentSynced { .. }
         | PeerControlResponse::DbSynced { .. } => 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<String>,
+) -> Result<ControlResponse, NodeError> {
+    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<PipeListener, NodeError> {
+    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 {

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()