diff --git a/AGENTS.md b/AGENTS.md index 1cdd8e2..cce6fa6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -183,8 +183,11 @@ Roadmap items should be actionable and checkable: forward-tcp --listen 127.0.0.1: --node --target 127.0.0.1:` opens authorized bidirectional byte streams over `/geth/pipe/1`; the remote daemon requires `pipe.forward` on - `resource:pipe-tcp:` before connecting to the loopback target. Unix - socket forwarding is still roadmap work. + `resource:pipe-tcp:` before connecting to the loopback target. `geth + pipe forward-unix --listen --node --target + ` uses the same Iroh path and requires `pipe.forward` on + `resource:pipe-unix:` before connecting to an absolute Unix socket + path. - `geth ssh proxy ` is a streaming OpenSSH ProxyCommand-style path. The CLI streams through the local daemon, the daemon uses `/geth/ssh-proxy/1` over Iroh, the remote daemon requires `ssh_proxy.connect` on diff --git a/README.md b/README.md index f425bc6..3ef4e2b 100644 --- a/README.md +++ b/README.md @@ -148,6 +148,7 @@ The bootstrap implementation provides: `geth pipe send [message|--in |--in -] [--node ] [--bearer-secret ]`, `geth pipe recv [--peek]`, and `geth pipe forward-tcp --listen 127.0.0.1: --node --target 127.0.0.1:` + or `geth pipe forward-unix --listen /tmp/local.sock --node --target /tmp/remote.sock` `geth peer export/import/list` is for untrusted peer-card exchange. Peer cards include the Iroh EndpointID plus currently known relay/direct addresses. @@ -216,6 +217,10 @@ stream to the peer. The remote daemon validates the signed endpoint/card binding and requires `pipe.forward` on `resource:pipe-tcp:` before connecting to the remote loopback TCP target. This is loopback-only in the prototype to avoid turning geth into an accidental open proxy. +`geth pipe forward-unix --listen --node --target +` uses the same `/geth/pipe/1` byte stream and requires +`pipe.forward` on `resource:pipe-unix:` before connecting to the remote +Unix socket. Unix socket paths must be absolute. `geth ssh proxy ` is usable as an OpenSSH `ProxyCommand`: the CLI opens a local daemon stream, the daemon opens the dedicated `/geth/ssh-proxy/1` Iroh ALPN, the remote daemon validates the caller's endpoint/card binding and diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index c778250..28746c5 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -407,6 +407,16 @@ pub enum PipeCommand { #[arg(long)] bearer_secret: Option, }, + ForwardUnix { + #[arg(long)] + listen: PathBuf, + #[arg(long)] + node: String, + #[arg(long)] + target: PathBuf, + #[arg(long)] + bearer_secret: Option, + }, Send { target: String, message: Option, @@ -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 { 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, diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index 5d59fa6..76fcc0e 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -313,6 +313,17 @@ pub enum ControlRequest { target_addr: String, bearer_secret: Option, }, + PipeUnixForward { + node: String, + listen_path: PathBuf, + target_path: PathBuf, + bearer_secret: Option, + }, + PipeUnixStream { + node: String, + target_path: PathBuf, + bearer_secret: Option, + }, PipeSend { target: String, data_base64: String, @@ -1050,6 +1061,12 @@ pub enum PipeWireRequest { nonce: String, bearer_proof: Option, }, + UnixConnect { + peer_card: PeerCard, + target_path: PathBuf, + nonce: String, + bearer_proof: Option, + }, } #[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(), diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 6214b8f..5a83a65 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -170,6 +170,13 @@ struct PipeTcpConnectWire { bearer_proof: Option, } +struct PipeUnixConnectWire { + peer_card: PeerCard, + target_path: PathBuf, + nonce: String, + bearer_proof: Option, +} + pub fn init_node(paths: &GethPaths) -> Result { 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, +) -> 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, + 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, + 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 { @@ -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)?; diff --git a/crates/geth-pipe/src/lib.rs b/crates/geth-pipe/src/lib.rs index 5e0a01c..015d2b8 100644 --- a/crates/geth-pipe/src/lib.rs +++ b/crates/geth-pipe/src/lib.rs @@ -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 Result { + 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()); + } } diff --git a/docs/architecture.md b/docs/architecture.md index c68a29e..6ee96eb 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -219,7 +219,10 @@ local loopback listener and opens one authorized `/geth/pipe/1` byte stream per accepted connection. The remote daemon validates endpoint/card binding and requires `pipe.forward` on `resource:pipe-tcp:` before connecting to the remote loopback TCP target. TCP forwarding is loopback-only in the prototype; -Unix socket forwarding is still future work. +`geth pipe forward-unix --listen --node --target +` uses the same authorized Iroh pipe stream and requires +`pipe.forward` on `resource:pipe-unix:` before connecting to an absolute +remote Unix socket path. `geth-ssh-proxy` defines proxy target and connection metadata. `geth ssh proxy ` is a streaming command intended for OpenSSH `ProxyCommand`: the CLI diff --git a/docs/roadmap.md b/docs/roadmap.md index a1d572d..c86d662 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -385,11 +385,17 @@ Goal: add authorized stream-oriented management workflows over Iroh. request/response serialization. - `[ ]` Tests cover a full two-node request/response forwarding exchange. -- `[ ]` Unix socket forwarding where supported. +- `[~]` Unix socket forwarding where supported. Acceptance criteria: - - Unix socket forwarding is available on Unix platforms. - - Unsupported platforms return clear errors. - - Tests skip or use cfg guards where sockets are unavailable. + - `[x]` Unix socket forwarding is available on Unix platforms through + `geth pipe forward-unix`. + - `[x]` Forwarding is resource-scoped with `pipe.forward` on + `resource:pipe-unix:`. + - `[x]` Unix socket paths must be absolute and reject parent-directory + components. + - `[x]` Tests cover Unix path validation and pipe wire request serialization. + - `[ ]` Unsupported platforms return clear errors. + - `[ ]` Tests cover a full two-node Unix socket forwarding exchange. - `[~]` SSH proxy over Iroh. Acceptance criteria: