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

Add authorized TCP pipe forwarding

Browse source
This commit is contained in:
Eric Wendland 2026-05-21 01:12:01 +02:00
commit 6bc2666993
8 changed files with 645 additions and 81 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -48,7 +48,7 @@ use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpStream, UnixListener, UnixStream};
use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream};
#[derive(Debug, thiserror::Error)]
pub enum NodeError {
@ -163,6 +163,13 @@ struct LiveSyncCursor {
cursor_ms: i64,
}
struct PipeTcpConnectWire {
peer_card: PeerCard,
target_addr: String,
nonce: String,
bearer_proof: Option<BearerProof>,
}
pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
paths.ensure_base_dirs()?;
if !paths.config_file().exists() {
@ -295,6 +302,87 @@ pub async fn stream_ssh_proxy(
}
}
pub async fn run_tcp_forward(
paths: &GethPaths,
listen_addr: String,
peer_node: String,
target_addr: String,
bearer_secret: Option<String>,
) -> Result<(), NodeError> {
let listen_addr = geth_pipe::validate_tcp_forward_listen_addr(&listen_addr)?;
geth_pipe::validate_tcp_forward_target_addr(&target_addr)?;
let listener = TcpListener::bind(listen_addr).await?;
tracing::info!(
listen = %listen_addr,
peer = %peer_node,
target = %target_addr,
"tcp forward listening"
);
loop {
let (client, client_addr) = listener.accept().await?;
let paths = paths.clone();
let peer_node = peer_node.clone();
let target_addr = target_addr.clone();
let bearer_secret = bearer_secret.clone();
tokio::spawn(async move {
if let Err(error) =
stream_pipe_tcp(&paths, peer_node, target_addr, bearer_secret, client).await
{
tracing::warn!(%error, %client_addr, "tcp forward connection failed");
}
});
}
}
async fn stream_pipe_tcp(
paths: &GethPaths,
peer_node: String,
target_addr: String,
bearer_secret: Option<String>,
client: TcpStream,
) -> Result<(), NodeError> {
let mut stream = UnixStream::connect(paths.socket_path()).await?;
stream
.write_all(
geth_control::encode_request(&ControlRequest::PipeTcpStream {
node: peer_node,
target_addr,
bearer_secret,
})?
.as_bytes(),
)
.await?;
let mut reader = BufReader::new(stream);
let mut line = String::new();
reader.read_line(&mut line).await?;
match geth_control::decode_response(&line)? {
ControlResponse::PipeRemoteConnected { allowed: true, .. } => {
let daemon_stream = reader.into_inner();
let (mut daemon_read, mut daemon_write) = daemon_stream.into_split();
let (mut client_read, mut client_write) = client.into_split();
let upload = async {
tokio::io::copy(&mut client_read, &mut daemon_write).await?;
daemon_write.shutdown().await
};
let download = async {
tokio::io::copy(&mut daemon_read, &mut client_write).await?;
client_write.shutdown().await
};
tokio::try_join!(upload, download)?;
Ok(())
}
ControlResponse::PipeRemoteConnected {
allowed: false,
reason,
..
} => Err(NodeError::Unauthorized(reason)),
ControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
other => Err(NodeError::IrohPeer(format!(
"daemon returned unexpected pipe TCP stream response: {other:?}"
))),
}
}
pub async fn handle_request_async(
node: &LocalNode,
request: ControlRequest,
@ -363,6 +451,12 @@ pub async fn handle_request_async(
node: Some(peer_node),
bearer_secret,
} => pipe_send_to_peer(node, &peer_node, target, data_base64, bearer_secret).await,
ControlRequest::PipeTcpForward { .. } | ControlRequest::PipeTcpStream { .. } => {
Err(NodeError::IrohPeer(
"TCP forwarding is a streaming control command; run without --json/--jsonl"
.to_owned(),
))
}
ControlRequest::SshProxyConnect {
node: peer_node,
bearer_secret,
@ -389,6 +483,16 @@ async fn handle_stream(node: LocalNode, stream: UnixStream) -> Result<(), NodeEr
let stream = reader.into_inner();
return handle_local_ssh_proxy_stream(node, &peer_node, bearer_secret, stream).await;
}
if let ControlRequest::PipeTcpStream {
node: peer_node,
target_addr,
bearer_secret,
} = request
{
let stream = reader.into_inner();
return handle_local_pipe_tcp_stream(node, &peer_node, target_addr, bearer_secret, stream)
.await;
}
let response = match handle_request_async(&node, request).await {
Ok(response) => response,
Err(error) => ControlResponse::Error {
@ -1584,9 +1688,134 @@ async fn pipe_send_to_peer(
note,
}),
PipeWireResponse::Error { message } => Err(NodeError::IrohPeer(message)),
PipeWireResponse::Connected { .. } => Err(NodeError::IrohPeer(
"peer returned TCP stream response to pipe send".to_owned(),
)),
}
}
async fn handle_local_pipe_tcp_stream(
node: LocalNode,
peer_node: &str,
target_addr: String,
bearer_secret: Option<String>,
local_stream: UnixStream,
) -> Result<(), NodeError> {
geth_pipe::validate_tcp_forward_target_addr(&target_addr)?;
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{}\0pipe-tcp-stream\0{}\0{}",
node.node_id,
peer_node,
target_addr,
geth_store::now_ms()
)
.as_bytes(),
);
let resource = format!("resource:pipe-tcp:{target_addr}");
let request = PipeWireRequest::TcpConnect {
peer_card: self_card,
target_addr: target_addr.clone(),
nonce: nonce.clone(),
bearer_proof: bearer_proof(bearer_secret, &resource, "pipe.forward", &nonce),
};
let conn = endpoint
.endpoint()
.connect(node_addr, geth_iroh::ALPN_PIPE)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let (mut remote_send, remote_recv) = conn
.open_bi()
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
remote_send
.write_all(geth_control::encode_pipe_wire_request(&request)?.as_bytes())
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let mut remote_reader = BufReader::new(remote_recv);
let mut line = String::new();
remote_reader
.read_line(&mut line)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let response = geth_control::decode_pipe_wire_response(&line)?;
let local_response = match response {
PipeWireResponse::Connected {
node_id,
agent_id,
endpoint_id,
connection,
allowed,
reason,
note,
nonce: response_nonce,
..
} if response_nonce == nonce => ControlResponse::PipeRemoteConnected {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
connection: connection.unwrap_or_else(|| PipeConnection {
target: target_addr.clone(),
connected_at: UnixMillis(0),
local_listener_found: false,
note: "pipe TCP forward was denied before connecting".to_owned(),
}),
allowed,
reason,
note,
},
PipeWireResponse::Error { message } => ControlResponse::Error { message },
_ => ControlResponse::Error {
message: "peer returned wrong response type to pipe TCP stream".to_owned(),
},
};
let mut local_stream = local_stream;
local_stream
.write_all(geth_control::encode_response(&local_response)?.as_bytes())
.await?;
let ControlResponse::PipeRemoteConnected { allowed: true, .. } = local_response else {
return Ok(());
};
let remote_recv = remote_reader.into_inner();
let (mut local_read, mut local_write) = local_stream.into_split();
let upload = async {
tokio::io::copy(&mut local_read, &mut remote_send)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
remote_send
.finish()
.map_err(|error| NodeError::IrohPeer(error.to_string()))
};
let mut remote_recv = remote_recv;
let download = async {
tokio::io::copy(&mut remote_recv, &mut local_write)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
local_write.shutdown().await.map_err(NodeError::from)
};
tokio::try_join!(upload, download)?;
Ok(())
}
async fn ssh_proxy_connect_to_peer(
node: &LocalNode,
peer_node: &str,
@ -2577,21 +2806,15 @@ async fn handle_iroh_control_connection(
if alpn == display_alpn(geth_iroh::ALPN_SSH_PROXY.to_vec()) {
return handle_ssh_proxy_wire_connection(node, remote_endpoint_id, send, recv).await;
}
if alpn == display_alpn(geth_iroh::ALPN_PIPE.to_vec()) {
return handle_pipe_wire_connection(node, remote_endpoint_id, send, recv).await;
}
let bytes = recv
.read_to_end(64 * 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()))?;
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()?;
@ -3526,12 +3749,19 @@ async fn handle_ssh_proxy_wire_connection(
Ok(())
}
fn handle_pipe_wire_request(
node: &LocalNode,
remote_endpoint_id: &str,
text: &str,
) -> Result<PipeWireResponse, NodeError> {
match geth_control::decode_pipe_wire_request(text)? {
async fn handle_pipe_wire_connection(
node: LocalNode,
remote_endpoint_id: String,
mut send: iroh::endpoint::SendStream,
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()))?;
match geth_control::decode_pipe_wire_request(&line)? {
PipeWireRequest::Send {
peer_card,
target,
@ -3539,61 +3769,219 @@ fn handle_pipe_wire_request(
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,
let response = handle_pipe_send_wire_request(
&node,
&remote_endpoint_id,
peer_card,
target,
data_base64,
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(),
})
bearer_proof,
)?;
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()))?;
Ok(())
}
PipeWireRequest::TcpConnect {
peer_card,
target_addr,
nonce,
bearer_proof,
} => {
handle_pipe_tcp_wire_connection(
node,
&remote_endpoint_id,
PipeTcpConnectWire {
peer_card,
target_addr,
nonce,
bearer_proof,
},
send,
reader.into_inner(),
)
.await
}
}
}
fn handle_pipe_send_wire_request(
node: &LocalNode,
remote_endpoint_id: &str,
peer_card: PeerCard,
target: String,
data_base64: String,
nonce: String,
bearer_proof: Option<BearerProof>,
) -> Result<PipeWireResponse, NodeError> {
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(),
})
}
async fn handle_pipe_tcp_wire_connection(
node: LocalNode,
remote_endpoint_id: &str,
request: PipeTcpConnectWire,
mut send: iroh::endpoint::SendStream,
recv: iroh::endpoint::RecvStream,
) -> Result<(), NodeError> {
let PipeTcpConnectWire {
peer_card,
target_addr,
nonce,
bearer_proof,
} = request;
let target = geth_pipe::validate_tcp_forward_target_addr(&target_addr)?;
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-tcp:{target_addr}");
let capability = "pipe.forward".to_owned();
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&resource,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
if !explanation.allowed {
let response = PipeWireResponse::Connected {
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(),
connection: None,
allowed: false,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "pipe TCP forward denied before opening the remote TCP target".to_owned(),
};
send.write_all(geth_control::encode_pipe_wire_response(&response)?.as_bytes())
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
return send
.finish()
.map_err(|error| NodeError::IrohPeer(error.to_string()));
}
let tcp = match TcpStream::connect(target).await {
Ok(tcp) => tcp,
Err(error) => {
let response = PipeWireResponse::Error {
message: format!(
"authorized pipe TCP forward could not connect to {target_addr}: {error}"
),
};
send.write_all(geth_control::encode_pipe_wire_response(&response)?.as_bytes())
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
return send
.finish()
.map_err(|error| NodeError::IrohPeer(error.to_string()));
}
};
let response = PipeWireResponse::Connected {
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(),
connection: Some(PipeConnection {
target: target_addr.clone(),
connected_at: UnixMillis(geth_store::now_ms()),
local_listener_found: true,
note: "authorized TCP byte stream over the dedicated Iroh pipe ALPN".to_owned(),
}),
allowed: true,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "pipe TCP forward authenticated endpoint/card binding and required pipe.forward before connecting to the loopback TCP target".to_owned(),
};
send.write_all(geth_control::encode_pipe_wire_response(&response)?.as_bytes())
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let (mut tcp_read, mut tcp_write) = tcp.into_split();
let mut recv = recv;
let inbound = async {
tokio::io::copy(&mut recv, &mut tcp_write)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
tcp_write.shutdown().await.map_err(NodeError::from)
};
let outbound = async {
tokio::io::copy(&mut tcp_read, &mut send)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
send.finish()
.map_err(|error| NodeError::IrohPeer(error.to_string()))
};
tokio::try_join!(inbound, outbound)?;
Ok(())
}
fn iroh_node_addr_from_candidate(
candidate: &EndpointCandidate,
) -> Result<iroh::NodeAddr, NodeError> {
@ -4929,6 +5317,9 @@ pub fn handle_request(
})
}
ControlRequest::PipeSend { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::PipeTcpForward { .. } | ControlRequest::PipeTcpStream { .. } => {
Err(NodeError::IrohEndpointUnavailable)
}
ControlRequest::PipeRecv { name, peek } => {
geth_pipe::validate_pipe_name(&name)?;
let messages = pipe_messages(node, &name, !peek)?;