From a32cab645f753c6f2e8901ad3ece8b711731fed8 Mon Sep 17 00:00:00 2001 From: Eric Wendland Date: Sun, 5 Jul 2026 23:51:08 +0200 Subject: [PATCH] refactor: centralize peer control routing --- crates/geth-node/src/lib.rs | 175 +++++++++------------------ crates/geth-node/src/peer_control.rs | 151 +++++++++++++++++++++++ docs/architecture.md | 14 ++- docs/production-readiness-roadmap.md | 4 +- 4 files changed, 218 insertions(+), 126 deletions(-) create mode 100644 crates/geth-node/src/peer_control.rs diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 3a53166..d151a35 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -3,6 +3,7 @@ mod daemon; pub mod doctor; mod local_control; mod peer_client; +mod peer_control; mod resource_contracts; mod runtime; pub mod service; @@ -68,6 +69,10 @@ use iroh::protocol::ProtocolHandler; use iroh_docs::api::protocol::{AddrInfoOptions, ShareMode}; pub use local_control::handle_request_async; use peer_client::{request_overlay_wire, request_peer_control, request_pipe_wire}; +use peer_control::{ + IrohAlpnRoute, PeerControlCaller, authenticate_peer_control_caller, classify_iroh_alpn, + display_alpn, ensure_peer_card_matches_endpoint, iroh_node_addr_from_candidate, +}; use runtime::{ NodeRuntime, OverlayTunCounters, OverlayTunRuntime, PipeRuntime, PubsubGossipTopicRuntime, PubsubRuntime, @@ -223,12 +228,6 @@ struct PipeUnixConnectWire { bearer_proof: Option, } -struct PeerControlCaller { - peer_card: PeerCard, - remote_endpoint_id: String, - store: Store, -} - #[derive(Clone, Debug)] pub struct InitOwnerOptions { pub admin_key_path: Option, @@ -4064,41 +4063,48 @@ async fn handle_iroh_control_connection( .await .map_err(|error| NodeError::IrohPeer(error.to_string()))?; let remote_endpoint_id = conn.remote_id().to_string(); + let route = classify_iroh_alpn(conn.alpn()); let alpn = display_alpn(conn.alpn()); let log_remote_endpoint_id = remote_endpoint_id.clone(); let log_alpn = alpn.clone(); tracing::debug!(remote_endpoint_id = %log_remote_endpoint_id, alpn = %log_alpn, "iroh connection established"); - if alpn == display_alpn(iroh_blobs::ALPN) { - let blob_store = iroh_blob_store(&node)?.ok_or_else(|| { - NodeError::IrohPeer("iroh-blobs ALPN accepted but blob store is unavailable".to_owned()) - })?; - let protocol = iroh_blobs::BlobsProtocol::new(&blob_store, None); - return protocol - .accept(conn) - .await - .map_err(|error| NodeError::IrohPeer(format!("iroh-blobs accept failed: {error}"))); - } - if alpn == display_alpn(iroh_docs::ALPN) { - let docs = iroh_docs(&node)?.ok_or_else(|| { - NodeError::IrohPeer( - "iroh-docs ALPN accepted but docs runtime is unavailable".to_owned(), - ) - })?; - return docs - .accept(conn) - .await - .map_err(|error| NodeError::IrohPeer(format!("iroh-docs accept failed: {error}"))); - } - if alpn == display_alpn(iroh_gossip::ALPN) { - let gossip = iroh_gossip(&node)?.ok_or_else(|| { - NodeError::IrohPeer( - "iroh-gossip ALPN accepted but gossip runtime is unavailable".to_owned(), - ) - })?; - return gossip - .accept(conn) - .await - .map_err(|error| NodeError::IrohPeer(format!("iroh-gossip accept failed: {error}"))); + match route { + IrohAlpnRoute::NativeBlobs => { + let blob_store = iroh_blob_store(&node)?.ok_or_else(|| { + NodeError::IrohPeer( + "iroh-blobs ALPN accepted but blob store is unavailable".to_owned(), + ) + })?; + let protocol = iroh_blobs::BlobsProtocol::new(&blob_store, None); + return protocol.accept(conn).await.map_err(|error| { + NodeError::IrohPeer(format!("iroh-blobs accept failed: {error}")) + }); + } + IrohAlpnRoute::NativeDocs => { + let docs = iroh_docs(&node)?.ok_or_else(|| { + NodeError::IrohPeer( + "iroh-docs ALPN accepted but docs runtime is unavailable".to_owned(), + ) + })?; + return docs + .accept(conn) + .await + .map_err(|error| NodeError::IrohPeer(format!("iroh-docs accept failed: {error}"))); + } + IrohAlpnRoute::NativeGossip => { + let gossip = iroh_gossip(&node)?.ok_or_else(|| { + NodeError::IrohPeer( + "iroh-gossip ALPN accepted but gossip runtime is unavailable".to_owned(), + ) + })?; + return gossip.accept(conn).await.map_err(|error| { + NodeError::IrohPeer(format!("iroh-gossip accept failed: {error}")) + }); + } + IrohAlpnRoute::SshProxy + | IrohAlpnRoute::Pipe + | IrohAlpnRoute::Overlay + | IrohAlpnRoute::Control => {} } tracing::debug!(%remote_endpoint_id, %alpn, "waiting for iroh bidirectional stream"); let (mut send, recv) = conn @@ -4106,14 +4112,20 @@ async fn handle_iroh_control_connection( .await .map_err(|error| NodeError::IrohPeer(error.to_string()))?; tracing::debug!(%remote_endpoint_id, %alpn, "accepted iroh bidirectional stream"); - if alpn == display_alpn(geth_iroh::ALPN_SSH_PROXY) { - return handle_ssh_proxy_wire_connection(node, remote_endpoint_id, send, recv).await; - } - if alpn == display_alpn(geth_iroh::ALPN_PIPE) { - return handle_pipe_wire_connection(node, remote_endpoint_id, send, recv).await; - } - if alpn == display_alpn(geth_iroh::ALPN_OVERLAY) { - return handle_overlay_wire_connection(node, remote_endpoint_id, send, recv).await; + match route { + IrohAlpnRoute::SshProxy => { + return handle_ssh_proxy_wire_connection(node, remote_endpoint_id, send, recv).await; + } + IrohAlpnRoute::Pipe => { + return handle_pipe_wire_connection(node, remote_endpoint_id, send, recv).await; + } + IrohAlpnRoute::Overlay => { + return handle_overlay_wire_connection(node, remote_endpoint_id, send, recv).await; + } + IrohAlpnRoute::Control => {} + IrohAlpnRoute::NativeBlobs | IrohAlpnRoute::NativeDocs | IrohAlpnRoute::NativeGossip => { + unreachable!("native ALPN routes return before bidirectional stream handling") + } } let mut recv = recv; let request_line = read_iroh_line(&mut recv, PEER_CONTROL_LINE_MAX).await?; @@ -5470,75 +5482,6 @@ async fn handle_pipe_unix_wire_connection( Ok(()) } -fn iroh_node_addr_from_candidate( - candidate: &EndpointCandidate, -) -> Result { - let endpoint_id = candidate - .endpoint_id - .parse::() - .map_err(|error| NodeError::IrohPeer(error.to_string()))?; - let direct_addresses = candidate - .direct_addresses - .iter() - .map(|addr| { - addr.parse::() - .map_err(|error| NodeError::IrohPeer(error.to_string())) - }) - .collect::, _>>()?; - let mut node_addr = iroh::EndpointAddr::new(endpoint_id); - for address in direct_addresses { - node_addr = node_addr.with_ip_addr(address); - } - if let Some(relay_url) = &candidate.relay_url { - node_addr = node_addr.with_relay_url( - relay_url - .parse::() - .map_err(|error| NodeError::IrohPeer(error.to_string()))?, - ); - } - Ok(node_addr) -} - -fn ensure_peer_card_matches_endpoint(card: &PeerCard, endpoint_id: &str) -> Result<(), NodeError> { - if card - .endpoints - .iter() - .any(|candidate| candidate.endpoint_id == endpoint_id) - { - Ok(()) - } else { - Err(NodeError::IrohPeer(format!( - "signed peer card for {} does not bind Iroh endpoint {}", - card.node_id, endpoint_id - ))) - } -} - -fn authenticate_peer_control_caller( - node: &LocalNode, - peer_card: PeerCard, - remote_endpoint_id: &str, -) -> Result { - 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, - })?; - Ok(PeerControlCaller { - peer_card, - remote_endpoint_id: remote_endpoint_id.to_owned(), - store, - }) -} - fn peer_gossip_endpoint_id( node: &LocalNode, peer_node: &str, @@ -5581,10 +5524,6 @@ fn restricted_admin_shell_output(node: &LocalNode, command: &str) -> Result String { - String::from_utf8_lossy(alpn).into_owned() -} - pub fn handle_request( node: &LocalNode, request: ControlRequest, diff --git a/crates/geth-node/src/peer_control.rs b/crates/geth-node/src/peer_control.rs new file mode 100644 index 0000000..258f0af --- /dev/null +++ b/crates/geth-node/src/peer_control.rs @@ -0,0 +1,151 @@ +//! Protected peer-control routing and caller authentication helpers. + +use super::*; + +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub(crate) enum IrohAlpnRoute { + NativeBlobs, + NativeDocs, + NativeGossip, + SshProxy, + Pipe, + Overlay, + Control, +} + +pub(crate) struct PeerControlCaller { + pub(crate) peer_card: PeerCard, + pub(crate) remote_endpoint_id: String, + pub(crate) store: Store, +} + +pub(crate) fn classify_iroh_alpn(alpn: &[u8]) -> IrohAlpnRoute { + if alpn == iroh_blobs::ALPN { + IrohAlpnRoute::NativeBlobs + } else if alpn == iroh_docs::ALPN { + IrohAlpnRoute::NativeDocs + } else if alpn == iroh_gossip::ALPN { + IrohAlpnRoute::NativeGossip + } else if alpn == geth_iroh::ALPN_SSH_PROXY { + IrohAlpnRoute::SshProxy + } else if alpn == geth_iroh::ALPN_PIPE { + IrohAlpnRoute::Pipe + } else if alpn == geth_iroh::ALPN_OVERLAY { + IrohAlpnRoute::Overlay + } else { + IrohAlpnRoute::Control + } +} + +pub(crate) fn display_alpn(alpn: &[u8]) -> String { + String::from_utf8_lossy(alpn).into_owned() +} + +pub(crate) fn iroh_node_addr_from_candidate( + candidate: &EndpointCandidate, +) -> Result { + let endpoint_id = candidate + .endpoint_id + .parse::() + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + let direct_addresses = candidate + .direct_addresses + .iter() + .map(|addr| { + addr.parse::() + .map_err(|error| NodeError::IrohPeer(error.to_string())) + }) + .collect::, _>>()?; + let mut node_addr = iroh::EndpointAddr::new(endpoint_id); + for address in direct_addresses { + node_addr = node_addr.with_ip_addr(address); + } + if let Some(relay_url) = &candidate.relay_url { + node_addr = node_addr.with_relay_url( + relay_url + .parse::() + .map_err(|error| NodeError::IrohPeer(error.to_string()))?, + ); + } + Ok(node_addr) +} + +pub(crate) fn ensure_peer_card_matches_endpoint( + card: &PeerCard, + endpoint_id: &str, +) -> Result<(), NodeError> { + if card + .endpoints + .iter() + .any(|candidate| candidate.endpoint_id == endpoint_id) + { + Ok(()) + } else { + Err(NodeError::IrohPeer(format!( + "signed peer card for {} does not bind Iroh endpoint {}", + card.node_id, endpoint_id + ))) + } +} + +pub(crate) fn authenticate_peer_control_caller( + node: &LocalNode, + peer_card: PeerCard, + remote_endpoint_id: &str, +) -> Result { + 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, + })?; + Ok(PeerControlCaller { + peer_card, + remote_endpoint_id: remote_endpoint_id.to_owned(), + store, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn classifies_geth_and_native_alpns() { + assert_eq!( + classify_iroh_alpn(iroh_blobs::ALPN), + IrohAlpnRoute::NativeBlobs + ); + assert_eq!( + classify_iroh_alpn(iroh_docs::ALPN), + IrohAlpnRoute::NativeDocs + ); + assert_eq!( + classify_iroh_alpn(iroh_gossip::ALPN), + IrohAlpnRoute::NativeGossip + ); + assert_eq!( + classify_iroh_alpn(geth_iroh::ALPN_SSH_PROXY), + IrohAlpnRoute::SshProxy + ); + assert_eq!( + classify_iroh_alpn(geth_iroh::ALPN_PIPE), + IrohAlpnRoute::Pipe + ); + assert_eq!( + classify_iroh_alpn(geth_iroh::ALPN_OVERLAY), + IrohAlpnRoute::Overlay + ); + assert_eq!( + classify_iroh_alpn(geth_iroh::ALPN_CONTROL), + IrohAlpnRoute::Control + ); + } +} diff --git a/docs/architecture.md b/docs/architecture.md index 676b9ee..de2f02c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -11,12 +11,14 @@ Within `geth-node`, daemon lifecycle code is separated from feature handlers: signal handling, Iroh endpoint startup, the Iroh accept loop, and background live-sync task spawning. `local_control.rs` owns async local `ControlRequest` routing, safe trace-field classification, and named peer/resource/local handler -families before delegating to feature implementations. `resource_contracts.rs` -records the review boundary for each resource family: resource ID patterns, -capabilities, and mutation or host-access points. Runtime registries for -pubsub, pipes, and overlays live behind narrow mutex-protected structs in -`runtime.rs`. Protected peer-control ALPN dispatch remains a separate refactor -target. +families before delegating to feature implementations. `peer_control.rs` +centralizes inbound Iroh ALPN classification, peer endpoint-address parsing, +signed peer-card endpoint binding validation, and authenticated caller context +creation for protected peer-control, pipe, SSH-proxy, and overlay paths. +`resource_contracts.rs` records the review boundary for each resource family: +resource ID patterns, capabilities, and mutation or host-access points. Runtime +registries for pubsub, pipes, and overlays live behind narrow mutex-protected +structs in `runtime.rs`. The local metadata store is SQLite product state. `geth-store` tracks a numeric `schema_version` in the `meta` table and applies ordered migrations up to the diff --git a/docs/production-readiness-roadmap.md b/docs/production-readiness-roadmap.md index 6e482bd..daea695 100644 --- a/docs/production-readiness-roadmap.md +++ b/docs/production-readiness-roadmap.md @@ -52,13 +52,13 @@ behavior. - `[x]` Each command family has a small handler module or function group. - `[x]` Local-only behavior remains covered by existing integration tests. -- `[~]` Extract protected peer-control routing. +- `[x]` Extract protected peer-control routing. Acceptance criteria: - `[x]` Shared bounded Iroh line-read and send-finish helpers live outside the main feature handler module. - `[x]` Outbound peer-control, pipe-wire, and overlay-wire request helpers live outside the main feature handler module. - - `[~]` Iroh control ALPN handling, nonce checks, peer-card validation, and + - `[x]` Iroh control ALPN handling, nonce checks, peer-card validation, and endpoint-binding validation are centralized. - `[x]` Feature handlers receive authenticated caller context rather than repeating peer-card boilerplate.