eric/geth
Watch
1
0
Fork
You've already forked geth
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: bound remote iroh line reads

Browse source
This commit is contained in:
Eric Wendland 2026-07-05 18:22:07 +02:00
commit ccd40f0224
5 changed files with 91 additions and 68 deletions
Download patch file Download diff file Expand all files Collapse all files

67
crates/geth-node/src/lib.rs
View file

@ -76,7 +76,10 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream};
use tokio::sync::{mpsc, oneshot};
use tun_rs::{DeviceBuilder, Layer};
use wire::{finish_iroh_send, read_iroh_line};
use wire::{
PEER_CONTROL_LINE_MAX, STREAM_HANDSHAKE_LINE_MAX, WIRE_REQUEST_LINE_MAX, finish_iroh_send,
read_iroh_line,
};
#[derive(Debug, thiserror::Error)]
pub enum NodeError {
@ -1128,11 +1131,8 @@ async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result<ControlResponse,
NodeError::IrohPeer(format!("peer ping request delivery failed: {error}"))
})?;
drop(send);
let mut response = String::new();
let mut reader = BufReader::new(recv);
reader.read_line(&mut response).await.map_err(|error| {
NodeError::IrohPeer(format!("peer ping response line read failed: {error}"))
})?;
let mut recv = recv;
let response = read_iroh_line(&mut recv, PEER_CONTROL_LINE_MAX).await?;
match geth_control::decode_peer_response(&response)? {
PeerControlResponse::Pong {
node_id,
@ -1243,12 +1243,8 @@ async fn peer_auth_check(
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
drop(send);
let mut response = String::new();
let mut reader = BufReader::new(recv);
reader
.read_line(&mut response)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let mut recv = recv;
let response = read_iroh_line(&mut recv, PEER_CONTROL_LINE_MAX).await?;
match geth_control::decode_peer_response(&response)? {
PeerControlResponse::AuthChecked {
node_id,
@ -2792,7 +2788,7 @@ async fn handle_local_pipe_tcp_stream(
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let mut remote_recv = remote_recv;
let line = read_iroh_line(&mut remote_recv, 64 * 1024).await?;
let line = read_iroh_line(&mut remote_recv, STREAM_HANDSHAKE_LINE_MAX).await?;
let response = geth_control::decode_pipe_wire_response(&line)?;
let local_response = match response {
PipeWireResponse::Connected {
@ -2910,7 +2906,7 @@ async fn handle_local_pipe_unix_stream(
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let mut remote_recv = remote_recv;
let line = read_iroh_line(&mut remote_recv, 64 * 1024).await?;
let line = read_iroh_line(&mut remote_recv, STREAM_HANDSHAKE_LINE_MAX).await?;
let response = geth_control::decode_pipe_wire_response(&line)?;
let local_response = match response {
PipeWireResponse::Connected {
@ -3120,7 +3116,7 @@ async fn handle_local_ssh_proxy_stream(
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let mut remote_recv = remote_recv;
let line = read_iroh_line(&mut remote_recv, 64 * 1024).await?;
let line = read_iroh_line(&mut remote_recv, STREAM_HANDSHAKE_LINE_MAX).await?;
let response = geth_control::decode_peer_response(&line)?;
let local_response = match response {
PeerControlResponse::SshProxyConnected {
@ -4141,12 +4137,9 @@ async fn handle_iroh_control_connection(
if alpn == display_alpn(geth_iroh::ALPN_OVERLAY) {
return handle_overlay_wire_connection(node, remote_endpoint_id, send, recv).await;
}
let mut request_line = String::new();
let mut reader = BufReader::new(recv);
let request_bytes = reader
.read_line(&mut request_line)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let mut recv = recv;
let request_line = read_iroh_line(&mut recv, PEER_CONTROL_LINE_MAX).await?;
let request_bytes = request_line.len();
tracing::debug!(remote_endpoint_id = %log_remote_endpoint_id, alpn = %log_alpn, request_bytes, "read iroh peer-control request line");
let response = match geth_control::decode_peer_request(&request_line)? {
PeerControlRequest::Ping { peer_card, nonce } => {
@ -5179,14 +5172,9 @@ async fn handle_ssh_proxy_wire_connection(
node: LocalNode,
remote_endpoint_id: String,
mut send: iroh::endpoint::SendStream,
recv: iroh::endpoint::RecvStream,
mut recv: iroh::endpoint::RecvStream,
) -> Result<(), NodeError> {
let mut reader = BufReader::new(recv);
let mut line = String::new();
reader
.read_line(&mut line)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let line = read_iroh_line(&mut recv, STREAM_HANDSHAKE_LINE_MAX).await?;
let request = geth_control::decode_peer_request(&line)?;
let PeerControlRequest::SshProxyConnect {
peer_card,
@ -5291,7 +5279,6 @@ async fn handle_ssh_proxy_wire_connection(
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let mut recv = reader.into_inner();
let (mut tcp_read, mut tcp_write) = tcp.into_split();
let inbound = async {
tokio::io::copy(&mut recv, &mut tcp_write)
@ -5314,14 +5301,9 @@ async fn handle_pipe_wire_connection(
node: LocalNode,
remote_endpoint_id: String,
mut send: iroh::endpoint::SendStream,
recv: iroh::endpoint::RecvStream,
mut recv: iroh::endpoint::RecvStream,
) -> Result<(), NodeError> {
let mut reader = BufReader::new(recv);
let mut line = String::new();
reader
.read_line(&mut line)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let line = read_iroh_line(&mut recv, WIRE_REQUEST_LINE_MAX).await?;
match geth_control::decode_pipe_wire_request(&line)? {
PipeWireRequest::Send {
peer_card,
@ -5361,7 +5343,7 @@ async fn handle_pipe_wire_connection(
bearer_proof,
},
send,
reader.into_inner(),
recv,
)
.await
}
@ -5381,7 +5363,7 @@ async fn handle_pipe_wire_connection(
bearer_proof,
},
send,
reader.into_inner(),
recv,
)
.await
}
@ -5392,14 +5374,9 @@ async fn handle_overlay_wire_connection(
node: LocalNode,
remote_endpoint_id: String,
mut send: iroh::endpoint::SendStream,
recv: iroh::endpoint::RecvStream,
mut recv: iroh::endpoint::RecvStream,
) -> Result<(), NodeError> {
let mut reader = BufReader::new(recv);
let mut line = String::new();
reader
.read_line(&mut line)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let line = read_iroh_line(&mut recv, WIRE_REQUEST_LINE_MAX).await?;
let response = match geth_control::decode_overlay_wire_request(&line)? {
OverlayWireRequest::Packet {
peer_card,

16
crates/geth-node/src/peer_client.rs
View file

@ -1,6 +1,6 @@
//! Outbound Iroh peer request helpers.
use crate::wire::{finish_iroh_send, read_iroh_line};
use crate::wire::{PEER_CONTROL_LINE_MAX, WIRE_REQUEST_LINE_MAX, finish_iroh_send, read_iroh_line};
use crate::{LocalNode, NodeError};
use geth_control::{
OverlayWireRequest, OverlayWireResponse, PeerControlRequest, PeerControlResponse,
@ -8,7 +8,6 @@ use geth_control::{
};
use geth_discovery::{DiscoverySource, PeerCard};
use geth_store::Store;
use tokio::io::{AsyncBufReadExt, BufReader};
pub(crate) async fn request_peer_control(
node: &LocalNode,
@ -43,7 +42,7 @@ pub(crate) async fn request_peer_control(
.connect(node_addr, geth_iroh::ALPN_CONTROL)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let (mut send, recv) = conn
let (mut send, mut recv) = conn
.open_bi()
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
@ -56,12 +55,7 @@ pub(crate) async fn request_peer_control(
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
drop(send);
let mut response_line = String::new();
let mut reader = BufReader::new(recv);
reader
.read_line(&mut response_line)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let response_line = read_iroh_line(&mut recv, PEER_CONTROL_LINE_MAX).await?;
let response = geth_control::decode_peer_response(&response_line)?;
match &response {
PeerControlResponse::SshCertSynced {
@ -176,7 +170,7 @@ pub(crate) async fn request_pipe_wire(
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
finish_iroh_send(&mut send).await?;
drop(send);
let text = read_iroh_line(&mut recv, 16 * 1024 * 1024).await?;
let text = read_iroh_line(&mut recv, WIRE_REQUEST_LINE_MAX).await?;
let response = geth_control::decode_pipe_wire_response(&text)?;
match &response {
PipeWireResponse::Sent {
@ -232,7 +226,7 @@ pub(crate) async fn request_overlay_wire(
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
finish_iroh_send(&mut send).await?;
drop(send);
let text = read_iroh_line(&mut recv, 16 * 1024 * 1024).await?;
let text = read_iroh_line(&mut recv, WIRE_REQUEST_LINE_MAX).await?;
let response = geth_control::decode_overlay_wire_response(&text)?;
match &response {
OverlayWireResponse::PacketAccepted {

60
crates/geth-node/src/wire.rs
View file

@ -1,19 +1,34 @@
//! Shared Iroh stream helpers.
use crate::NodeError;
use std::time::Duration;
pub(crate) const PEER_CONTROL_LINE_MAX: usize = 16 * 1024 * 1024;
pub(crate) const WIRE_REQUEST_LINE_MAX: usize = 16 * 1024 * 1024;
pub(crate) const STREAM_HANDSHAKE_LINE_MAX: usize = 64 * 1024;
pub(crate) const IROH_LINE_READ_TIMEOUT: Duration = Duration::from_secs(30);
pub(crate) async fn read_iroh_line(
recv: &mut iroh::endpoint::RecvStream,
max_len: usize,
) -> Result<String, NodeError> {
tokio::time::timeout(IROH_LINE_READ_TIMEOUT, read_iroh_line_inner(recv, max_len))
.await
.map_err(|_| {
NodeError::IrohPeer(format!(
"iroh line read timed out after {} seconds",
IROH_LINE_READ_TIMEOUT.as_secs()
))
})?
}
async fn read_iroh_line_inner(
recv: &mut iroh::endpoint::RecvStream,
max_len: usize,
) -> Result<String, NodeError> {
let mut bytes = Vec::new();
let mut byte = [0_u8; 1];
loop {
if bytes.len() >= max_len {
return Err(NodeError::IrohPeer(format!(
"iroh response line exceeded {max_len} bytes"
)));
}
let Some(n) = recv
.read(&mut byte)
.await
@ -29,14 +44,23 @@ pub(crate) async fn read_iroh_line(
if n == 0 {
continue;
}
bytes.push(byte[0]);
if byte[0] == b'\n' {
if push_line_byte(&mut bytes, byte[0], max_len)? {
break;
}
}
String::from_utf8(bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))
}
fn push_line_byte(bytes: &mut Vec<u8>, byte: u8, max_len: usize) -> Result<bool, NodeError> {
if bytes.len() >= max_len {
return Err(NodeError::IrohPeer(format!(
"iroh line exceeded {max_len} bytes"
)));
}
bytes.push(byte);
Ok(byte == b'\n')
}
pub(crate) async fn finish_iroh_send(
send: &mut iroh::endpoint::SendStream,
) -> Result<(), NodeError> {
@ -47,3 +71,25 @@ pub(crate) async fn finish_iroh_send(
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn bounded_line_rejects_byte_past_limit_without_appending() {
let mut bytes = b"abc".to_vec();
let error = push_line_byte(&mut bytes, b'd', 3).expect_err("line should exceed limit");
assert_eq!(bytes, b"abc");
assert!(error.to_string().contains("iroh line exceeded 3 bytes"));
}
#[test]
fn bounded_line_accepts_newline_within_limit() {
let mut bytes = b"abc".to_vec();
assert!(push_line_byte(&mut bytes, b'\n', 4).expect("push newline"));
assert_eq!(bytes, b"abc\n");
}
}