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

Add authorized Unix pipe forwarding

Browse source
This commit is contained in:
Eric Wendland 2026-05-21 01:15:51 +02:00
commit 3d0da22eae
8 changed files with 491 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

39
crates/geth-cli/src/lib.rs
View file

@ -407,6 +407,16 @@ pub enum PipeCommand {
#[arg(long)]
bearer_secret: Option<String>,
},
ForwardUnix {
#[arg(long)]
listen: PathBuf,
#[arg(long)]
node: String,
#[arg(long)]
target: PathBuf,
#[arg(long)]
bearer_secret: Option<String>,
},
Send {
target: String,
message: Option<String>,
@ -635,6 +645,24 @@ pub async fn run() -> Result<()> {
.await
.context("run TCP forward through geth daemon")?;
}
Command::Pipe {
command:
PipeCommand::ForwardUnix {
listen,
node,
target,
bearer_secret,
},
} if !cli.json && !cli.jsonl => {
println!(
"forwarding unix {} -> {node}:{}",
listen.display(),
target.display()
);
geth_node::run_unix_forward(&paths, listen, node, target, bearer_secret)
.await
.context("run Unix socket forward through geth daemon")?;
}
command => {
let request = request_for_command(command)?;
let response = geth_node::send_control(&paths, request)
@ -926,6 +954,17 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
target_addr: target,
bearer_secret,
},
PipeCommand::ForwardUnix {
listen,
node,
target,
bearer_secret,
} => ControlRequest::PipeUnixForward {
listen_path: listen,
node,
target_path: target,
bearer_secret,
},
PipeCommand::Send {
target,
message,

40
crates/geth-control/src/lib.rs
View file

@ -313,6 +313,17 @@ pub enum ControlRequest {
target_addr: String,
bearer_secret: Option<String>,
},
PipeUnixForward {
node: String,
listen_path: PathBuf,
target_path: PathBuf,
bearer_secret: Option<String>,
},
PipeUnixStream {
node: String,
target_path: PathBuf,
bearer_secret: Option<String>,
},
PipeSend {
target: String,
data_base64: String,
@ -1050,6 +1061,12 @@ pub enum PipeWireRequest {
nonce: String,
bearer_proof: Option<BearerProof>,
},
UnixConnect {
peer_card: PeerCard,
target_path: PathBuf,
nonce: String,
bearer_proof: Option<BearerProof>,
},
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@ -2118,6 +2135,29 @@ mod tests {
response
);
let request = PipeWireRequest::UnixConnect {
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(),
},
},
target_path: PathBuf::from("/tmp/geth-test.sock"),
nonce: "nonce".to_owned(),
bearer_proof: None,
};
assert_eq!(
decode_pipe_wire_request(&encode_pipe_wire_request(&request).expect("encode"))
.expect("decode"),
request
);
let response = PeerControlResponse::SshProxyConnected {
node_id: "node:peer".to_owned(),
agent_id: "agent:peer".to_owned(),

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

@ -170,6 +170,13 @@ struct PipeTcpConnectWire {
bearer_proof: Option<BearerProof>,
}
struct PipeUnixConnectWire {
peer_card: PeerCard,
target_path: PathBuf,
nonce: String,
bearer_proof: Option<BearerProof>,
}
pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
paths.ensure_base_dirs()?;
if !paths.config_file().exists() {
@ -383,6 +390,87 @@ async fn stream_pipe_tcp(
}
}
pub async fn run_unix_forward(
paths: &GethPaths,
listen_path: PathBuf,
peer_node: String,
target_path: PathBuf,
bearer_secret: Option<String>,
) -> Result<(), NodeError> {
let listen_path = geth_pipe::validate_unix_forward_path(&listen_path)?;
geth_pipe::validate_unix_forward_path(&target_path)?;
let listener = UnixListener::bind(&listen_path)?;
tracing::info!(
listen = %listen_path.display(),
peer = %peer_node,
target = %target_path.display(),
"unix socket forward listening"
);
loop {
let (client, _) = listener.accept().await?;
let paths = paths.clone();
let peer_node = peer_node.clone();
let target_path = target_path.clone();
let bearer_secret = bearer_secret.clone();
tokio::spawn(async move {
if let Err(error) =
stream_pipe_unix(&paths, peer_node, target_path, bearer_secret, client).await
{
tracing::warn!(%error, "unix socket forward connection failed");
}
});
}
}
async fn stream_pipe_unix(
paths: &GethPaths,
peer_node: String,
target_path: PathBuf,
bearer_secret: Option<String>,
client: UnixStream,
) -> Result<(), NodeError> {
let mut stream = UnixStream::connect(paths.socket_path()).await?;
stream
.write_all(
geth_control::encode_request(&ControlRequest::PipeUnixStream {
node: peer_node,
target_path,
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 Unix stream response: {other:?}"
))),
}
}
pub async fn handle_request_async(
node: &LocalNode,
request: ControlRequest,
@ -451,12 +539,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::PipeTcpForward { .. }
| ControlRequest::PipeTcpStream { .. }
| ControlRequest::PipeUnixForward { .. }
| ControlRequest::PipeUnixStream { .. } => Err(NodeError::IrohPeer(
"pipe forwarding is a streaming control command; run without --json/--jsonl".to_owned(),
)),
ControlRequest::SshProxyConnect {
node: peer_node,
bearer_secret,
@ -493,6 +581,16 @@ async fn handle_stream(node: LocalNode, stream: UnixStream) -> Result<(), NodeEr
return handle_local_pipe_tcp_stream(node, &peer_node, target_addr, bearer_secret, stream)
.await;
}
if let ControlRequest::PipeUnixStream {
node: peer_node,
target_path,
bearer_secret,
} = request
{
let stream = reader.into_inner();
return handle_local_pipe_unix_stream(node, &peer_node, target_path, bearer_secret, stream)
.await;
}
let response = match handle_request_async(&node, request).await {
Ok(response) => response,
Err(error) => ControlResponse::Error {
@ -1816,6 +1914,129 @@ async fn handle_local_pipe_tcp_stream(
Ok(())
}
async fn handle_local_pipe_unix_stream(
node: LocalNode,
peer_node: &str,
target_path: PathBuf,
bearer_secret: Option<String>,
local_stream: UnixStream,
) -> Result<(), NodeError> {
let target_path = geth_pipe::validate_unix_forward_path(&target_path)?;
let target_display = target_path.display().to_string();
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-unix-stream\0{}\0{}",
node.node_id,
peer_node,
target_display,
geth_store::now_ms()
)
.as_bytes(),
);
let resource = format!("resource:pipe-unix:{target_display}");
let request = PipeWireRequest::UnixConnect {
peer_card: self_card,
target_path: target_path.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_display.clone(),
connected_at: UnixMillis(0),
local_listener_found: false,
note: "pipe Unix 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 Unix 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,
@ -3805,6 +4026,26 @@ async fn handle_pipe_wire_connection(
)
.await
}
PipeWireRequest::UnixConnect {
peer_card,
target_path,
nonce,
bearer_proof,
} => {
handle_pipe_unix_wire_connection(
node,
&remote_endpoint_id,
PipeUnixConnectWire {
peer_card,
target_path,
nonce,
bearer_proof,
},
send,
reader.into_inner(),
)
.await
}
}
}
@ -3982,6 +4223,120 @@ async fn handle_pipe_tcp_wire_connection(
Ok(())
}
async fn handle_pipe_unix_wire_connection(
node: LocalNode,
remote_endpoint_id: &str,
request: PipeUnixConnectWire,
mut send: iroh::endpoint::SendStream,
recv: iroh::endpoint::RecvStream,
) -> Result<(), NodeError> {
let PipeUnixConnectWire {
peer_card,
target_path,
nonce,
bearer_proof,
} = request;
let target_path = geth_pipe::validate_unix_forward_path(&target_path)?;
let target_display = target_path.display().to_string();
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-unix:{target_display}");
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 Unix forward denied before opening the remote Unix socket".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 unix = match UnixStream::connect(&target_path).await {
Ok(unix) => unix,
Err(error) => {
let response = PipeWireResponse::Error {
message: format!(
"authorized pipe Unix forward could not connect to {}: {error}",
target_path.display()
),
};
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_display.clone(),
connected_at: UnixMillis(geth_store::now_ms()),
local_listener_found: true,
note: "authorized Unix socket byte stream over the dedicated Iroh pipe ALPN".to_owned(),
}),
allowed: true,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "pipe Unix forward authenticated endpoint/card binding and required pipe.forward before connecting to the Unix socket 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 unix_read, mut unix_write) = unix.into_split();
let mut recv = recv;
let inbound = async {
tokio::io::copy(&mut recv, &mut unix_write)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
unix_write.shutdown().await.map_err(NodeError::from)
};
let outbound = async {
tokio::io::copy(&mut unix_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> {
@ -5317,9 +5672,10 @@ pub fn handle_request(
})
}
ControlRequest::PipeSend { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::PipeTcpForward { .. } | ControlRequest::PipeTcpStream { .. } => {
Err(NodeError::IrohEndpointUnavailable)
}
ControlRequest::PipeTcpForward { .. }
| ControlRequest::PipeTcpStream { .. }
| ControlRequest::PipeUnixForward { .. }
| ControlRequest::PipeUnixStream { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::PipeRecv { name, peek } => {
geth_pipe::validate_pipe_name(&name)?;
let messages = pipe_messages(node, &name, !peek)?;

23
crates/geth-pipe/src/lib.rs
View file

@ -1,6 +1,7 @@
use geth_types::{PipeId, ResourceId, UnixMillis};
use serde::{Deserialize, Serialize};
use std::net::SocketAddr;
use std::path::{Component, Path, PathBuf};
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PipeResource {
@ -42,6 +43,8 @@ pub enum PipeError {
InvalidTcpAddress(String),
#[error("TCP forwarding addresses must be loopback addresses: {0}")]
NonLoopbackTcpAddress(String),
#[error("invalid Unix socket path: {0}")]
InvalidUnixSocketPath(String),
}
pub fn validate_pipe_name(name: &str) -> Result<(), PipeError> {
@ -75,6 +78,19 @@ pub fn validate_tcp_forward_target_addr(addr: &str) -> Result<SocketAddr, PipeEr
Ok(addr)
}
pub fn validate_unix_forward_path(path: &Path) -> Result<PathBuf, PipeError> {
if path.as_os_str().is_empty() || !path.is_absolute() {
return Err(PipeError::InvalidUnixSocketPath(path.display().to_string()));
}
if path
.components()
.any(|component| matches!(component, Component::ParentDir))
{
return Err(PipeError::InvalidUnixSocketPath(path.display().to_string()));
}
Ok(path.to_path_buf())
}
#[must_use]
pub fn pipe_roadmap() -> &'static str {
"future pipes are authorized Iroh bidirectional streams for stdin/stdout and forwarding"
@ -107,4 +123,11 @@ mod tests {
assert!(validate_tcp_forward_target_addr("0.0.0.0:22").is_err());
assert!(validate_tcp_forward_target_addr("192.0.2.10:22").is_err());
}
#[test]
fn unix_forward_paths_must_be_absolute_without_parent_components() {
assert!(validate_unix_forward_path(Path::new("/tmp/geth.sock")).is_ok());
assert!(validate_unix_forward_path(Path::new("relative.sock")).is_err());
assert!(validate_unix_forward_path(Path::new("/tmp/../geth.sock")).is_err());
}
}