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

Stream SSH proxy over Iroh

Browse source
This commit is contained in:
Eric Wendland 2026-05-21 01:03:38 +02:00
commit 87d8801d71
7 changed files with 366 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -600,6 +600,17 @@ pub async fn run() -> Result<()> {
run_service_command(&paths, command).context("manage geth user service")?;
print_service_report(report, cli.json || cli.jsonl)?;
}
Command::Ssh {
command:
SshCommand::Proxy {
node,
bearer_secret,
},
} if !cli.json && !cli.jsonl => {
geth_node::stream_ssh_proxy(&paths, node, bearer_secret)
.await
.context("stream SSH proxy through geth daemon")?;
}
command => {
let request = request_for_command(command)?;
let response = geth_node::send_control(&paths, request)

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

@ -223,6 +223,10 @@ pub enum ControlRequest {
node: String,
bearer_secret: Option<String>,
},
SshProxyStream {
node: String,
bearer_secret: Option<String>,
},
DbAdd {
name: String,
path: PathBuf,
@ -1486,6 +1490,15 @@ mod tests {
request
);
let request = ControlRequest::SshProxyStream {
node: "node:peer".to_owned(),
bearer_secret: Some("bearer:test".to_owned()),
};
assert_eq!(
decode_request(&encode_request(&request).expect("encode")).expect("decode"),
request
);
let response = ControlResponse::SshProxyConnected {
peer_node_id: "node:peer".to_owned(),
peer_agent_id: "agent:peer".to_owned(),

316
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::{UnixListener, UnixStream};
use tokio::net::{TcpStream, UnixListener, UnixStream};
#[derive(Debug, thiserror::Error)]
pub enum NodeError {
@ -248,6 +248,53 @@ pub async fn send_control(
Ok(geth_control::decode_response(&line)?)
}
pub async fn stream_ssh_proxy(
paths: &GethPaths,
peer_node: String,
bearer_secret: Option<String>,
) -> Result<(), NodeError> {
let mut stream = UnixStream::connect(paths.socket_path()).await?;
stream
.write_all(
geth_control::encode_request(&ControlRequest::SshProxyStream {
node: peer_node,
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::SshProxyConnected { allowed: true, .. } => {
let stream = reader.into_inner();
let (mut local_read, mut local_write) = stream.into_split();
let mut stdin = tokio::io::stdin();
let mut stdout = tokio::io::stdout();
let upload = async {
tokio::io::copy(&mut stdin, &mut local_write).await?;
local_write.shutdown().await
};
let download = async {
tokio::io::copy(&mut local_read, &mut stdout).await?;
stdout.flush().await
};
tokio::try_join!(upload, download)?;
Ok(())
}
ControlResponse::SshProxyConnected {
allowed: false,
reason,
..
} => Err(NodeError::Unauthorized(reason)),
ControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
other => Err(NodeError::IrohPeer(format!(
"daemon returned unexpected SSH proxy stream response: {other:?}"
))),
}
}
pub async fn handle_request_async(
node: &LocalNode,
request: ControlRequest,
@ -334,6 +381,14 @@ async fn handle_stream(node: LocalNode, stream: UnixStream) -> Result<(), NodeEr
let mut line = String::new();
reader.read_line(&mut line).await?;
let request = geth_control::decode_request(&line)?;
if let ControlRequest::SshProxyStream {
node: peer_node,
bearer_secret,
} = request
{
let stream = reader.into_inner();
return handle_local_ssh_proxy_stream(node, &peer_node, bearer_secret, stream).await;
}
let response = match handle_request_async(&node, request).await {
Ok(response) => response,
Err(error) => ControlResponse::Error {
@ -1577,6 +1632,123 @@ async fn ssh_proxy_connect_to_peer(
}
}
async fn handle_local_ssh_proxy_stream(
node: LocalNode,
peer_node: &str,
bearer_secret: Option<String>,
local_stream: UnixStream,
) -> Result<(), NodeError> {
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{}\0ssh-proxy-stream\0{}",
node.node_id,
peer_node,
geth_store::now_ms()
)
.as_bytes(),
);
let request = PeerControlRequest::SshProxyConnect {
peer_card: self_card,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
"resource:ssh-proxy:local",
"ssh_proxy.connect",
&nonce,
),
};
let conn = endpoint
.endpoint()
.connect(node_addr, geth_iroh::ALPN_SSH_PROXY)
.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_peer_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_peer_response(&line)?;
let local_response = match response {
PeerControlResponse::SshProxyConnected {
node_id,
agent_id,
endpoint_id,
connection,
allowed,
reason,
note,
nonce: response_nonce,
..
} if response_nonce == nonce => ControlResponse::SshProxyConnected {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
connection,
allowed,
reason,
note,
},
PeerControlResponse::Error { message } => ControlResponse::Error { message },
_ => ControlResponse::Error {
message: "peer returned wrong response type to SSH proxy stream".to_owned(),
},
};
let mut local_stream = local_stream;
local_stream
.write_all(geth_control::encode_response(&local_response)?.as_bytes())
.await?;
let ControlResponse::SshProxyConnected { 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 document_sync_from_peer(
node: &LocalNode,
peer_node: &str,
@ -2402,6 +2574,9 @@ async fn handle_iroh_control_connection(
.accept_bi()
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
if alpn == display_alpn(geth_iroh::ALPN_SSH_PROXY.to_vec()) {
return handle_ssh_proxy_wire_connection(node, remote_endpoint_id, send, recv).await;
}
let bytes = recv
.read_to_end(64 * 1024)
.await
@ -3216,6 +3391,141 @@ async fn handle_iroh_control_connection(
Ok(())
}
async fn handle_ssh_proxy_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()))?;
let request = geth_control::decode_peer_request(&line)?;
let PeerControlRequest::SshProxyConnect {
peer_card,
nonce,
bearer_proof,
} = request
else {
let response = PeerControlResponse::Error {
message: "SSH proxy stream expected ssh-proxy-connect handshake".to_owned(),
};
send.write_all(geth_control::encode_peer_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(());
};
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 = "resource:ssh-proxy:local".to_owned();
let capability = "ssh_proxy.connect".to_owned();
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&resource,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let connection = if explanation.allowed {
Some(SshProxyConnection {
target_node: node.node_id.clone().into(),
connected_at: UnixMillis(geth_store::now_ms()),
local_sshd_target: Some("127.0.0.1:22".to_owned()),
admin_shell_available: false,
note: "authorized SSH proxy byte stream over Iroh; SSH is not a geth transport"
.to_owned(),
})
} else {
None
};
if !explanation.allowed {
let response = PeerControlResponse::SshProxyConnected {
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,
connection,
allowed: false,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "SSH proxy denied before opening sshd; SSH is not a geth transport".to_owned(),
};
send.write_all(geth_control::encode_peer_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 tcp = match TcpStream::connect("127.0.0.1:22").await {
Ok(tcp) => tcp,
Err(error) => {
let response = PeerControlResponse::Error {
message: format!("authorized SSH proxy could not connect to 127.0.0.1:22: {error}"),
};
send.write_all(geth_control::encode_peer_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 = PeerControlResponse::SshProxyConnected {
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,
connection,
allowed: true,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "SSH proxy authenticated endpoint/card binding and required ssh_proxy.connect before connecting to local sshd; SSH is not a geth transport".to_owned(),
};
send.write_all(geth_control::encode_peer_response(&response)?.as_bytes())
.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)
.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 handle_pipe_wire_request(
node: &LocalNode,
remote_endpoint_id: &str,
@ -4629,7 +4939,9 @@ pub fn handle_request(
note: geth_pipe::local_pipe_runtime_note().to_owned(),
})
}
ControlRequest::SshProxyConnect { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::SshProxyConnect { .. } | ControlRequest::SshProxyStream { .. } => {
Err(NodeError::IrohEndpointUnavailable)
}
ControlRequest::ModuleStub { module, command } => {
Ok(ControlResponse::NotImplemented { module, command })
}