From c9918251270d08d373dc2e727aaa983ca8c3d0a9 Mon Sep 17 00:00:00 2001 From: Eric Wendland Date: Sun, 17 May 2026 20:17:26 +0200 Subject: [PATCH] Add local pipe registry commands --- AGENTS.md | 3 ++ Cargo.lock | 3 ++ README.md | 6 ++- crates/geth-cli/src/lib.rs | 18 +++++++-- crates/geth-control/Cargo.toml | 1 + crates/geth-control/src/lib.rs | 21 +++++++++++ crates/geth-node/Cargo.toml | 1 + crates/geth-node/src/lib.rs | 50 ++++++++++++++++++++++++- crates/geth-pipe/Cargo.toml | 1 + crates/geth-pipe/src/lib.rs | 55 ++++++++++++++++++++++++++- crates/geth/tests/bootstrap.rs | 68 ++++++++++++++++++++++++++++++++++ docs/architecture.md | 8 +++- docs/roadmap.md | 12 ++++-- 13 files changed, 234 insertions(+), 13 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index c3fff78..bd467d5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -131,6 +131,9 @@ Roadmap items should be actionable and checkable: - Pubsub supports local daemon-lifetime publish/subscribe snapshots through a bounded in-memory ring buffer. Iroh-gossip replication, private topics, and pubsub capability enforcement are still roadmap work. +- Pipe listen/connect supports a local daemon-lifetime registry only. Iroh byte + streams, TCP/Unix forwarding, and pipe capability enforcement are still + roadmap work. - Resource secret epoch metadata can be created, rotated, and listed locally. Bearer access metadata can be created/listed/revoked as resource-scoped auth ops and must not allow trust graph mutation capabilities. Payload encryption, diff --git a/Cargo.lock b/Cargo.lock index 1cb2265..b59bce4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1120,6 +1120,7 @@ dependencies = [ "geth-document", "geth-keychain", "geth-kv", + "geth-pipe", "geth-pubsub", "geth-resource", "geth-secrets", @@ -1223,6 +1224,7 @@ dependencies = [ "geth-iroh", "geth-keychain", "geth-kv", + "geth-pipe", "geth-pubsub", "geth-resource", "geth-secrets", @@ -1241,6 +1243,7 @@ version = "0.1.0" dependencies = [ "geth-types", "serde", + "thiserror 2.0.18", ] [[package]] diff --git a/README.md b/README.md index 50df0d2..6d89a56 100644 --- a/README.md +++ b/README.md @@ -101,8 +101,9 @@ The bootstrap implementation provides: - `geth ssh revocation add ` - `geth ssh revocation list` - `geth ssh revocation export --out [--format jsonl|openssh-krl-spec]` +- local pipe registry commands: `geth pipe listen/connect` -Other command groups exist as explicit stubs: `pipe` and `ssh`. +Other command groups exist as explicit stubs: `ssh`. ## Resource Modules @@ -110,7 +111,8 @@ Everything meaningful is modeled as a resource. Planned resource kinds are: - `db`: SQLite/cr-sqlite synchronization - `kv`: Iroh Documents backed key-value stores -- `pipe`: dumbpipe-like byte streams over Iroh +- `pipe`: dumbpipe-like byte streams over Iroh; the bootstrap has a local + daemon registry only - `document`: Automerge documents over Iroh streams - `pubsub`: lossy notifications, not authoritative storage; the bootstrap keeps only an in-memory daemon-lifetime ring buffer diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index 7c7951d..78e0afb 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -462,9 +462,9 @@ fn request_for_command(command: Command) -> Result { PubsubCommand::Pub { topic, message } => ControlRequest::PubsubPub { topic, message }, PubsubCommand::Sub { topic } => ControlRequest::PubsubSub { topic }, }, - Command::Pipe { command } => ControlRequest::ModuleStub { - module: "pipe".to_owned(), - command: format!("{command:?}"), + Command::Pipe { command } => match command { + PipeCommand::Listen { name } => ControlRequest::PipeListen { name }, + PipeCommand::Connect { target } => ControlRequest::PipeConnect { target }, }, Command::Db { command } => match command { DbCommand::Add { name, path } => ControlRequest::DbAdd { name, path }, @@ -947,6 +947,18 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { } println!("note: {note}"); } + ControlResponse::PipeListening { listener } => { + println!("listening pipe: {}", listener.name); + println!("id: {}", listener.id); + println!("listened_at_ms: {}", listener.listened_at.0); + println!("note: {}", listener.note); + } + ControlResponse::PipeConnected { connection } => { + println!("pipe target: {}", connection.target); + println!("local_listener_found: {}", connection.local_listener_found); + println!("connected_at_ms: {}", connection.connected_at.0); + println!("note: {}", connection.note); + } ControlResponse::NotImplemented { module, command } => { println!("{module} {command}: not implemented yet"); } diff --git a/crates/geth-control/Cargo.toml b/crates/geth-control/Cargo.toml index 45d163a..a711644 100644 --- a/crates/geth-control/Cargo.toml +++ b/crates/geth-control/Cargo.toml @@ -14,6 +14,7 @@ geth-db = { path = "../geth-db" } geth-document = { path = "../geth-document" } geth-keychain = { path = "../geth-keychain" } geth-kv = { path = "../geth-kv" } +geth-pipe = { path = "../geth-pipe" } geth-pubsub = { path = "../geth-pubsub" } geth-resource = { path = "../geth-resource" } geth-secrets = { path = "../geth-secrets" } diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index 9dfe8d6..f60d094 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -3,6 +3,7 @@ use geth_db::DbResource; use geth_document::{DocumentResource, DocumentState}; use geth_keychain::KeychainOp; use geth_kv::{KvEntry, KvResource}; +use geth_pipe::{PipeConnection, PipeListener}; use geth_pubsub::PubsubMessage; use geth_resource::ResourceDescriptor; use geth_secrets::{BearerAccess, ResourceMasterSecret}; @@ -152,6 +153,12 @@ pub enum ControlRequest { PubsubSub { topic: String, }, + PipeListen { + name: String, + }, + PipeConnect { + target: String, + }, ModuleStub { module: String, command: String, @@ -284,6 +291,12 @@ pub enum ControlResponse { messages: Vec, note: String, }, + PipeListening { + listener: PipeListener, + }, + PipeConnected { + connection: PipeConnection, + }, NotImplemented { module: String, command: String, @@ -406,5 +419,13 @@ mod tests { decode_request(&encode_request(&request).expect("encode")).expect("decode"), request ); + + let request = ControlRequest::PipeListen { + name: "inbox".to_owned(), + }; + assert_eq!( + decode_request(&encode_request(&request).expect("encode")).expect("decode"), + request + ); } } diff --git a/crates/geth-node/Cargo.toml b/crates/geth-node/Cargo.toml index c18aa53..e178f6a 100644 --- a/crates/geth-node/Cargo.toml +++ b/crates/geth-node/Cargo.toml @@ -20,6 +20,7 @@ geth-document = { path = "../geth-document" } geth-iroh = { path = "../geth-iroh" } geth-keychain = { path = "../geth-keychain" } geth-kv = { path = "../geth-kv" } +geth-pipe = { path = "../geth-pipe" } geth-pubsub = { path = "../geth-pubsub" } geth-resource = { path = "../geth-resource" } geth-secrets = { path = "../geth-secrets" } diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index f4e8a11..8fbcae7 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -13,6 +13,7 @@ use geth_document::{DocumentResource, DocumentState}; use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode}; use geth_keychain::{KeychainOp, KeychainOpKind}; use geth_kv::{KvEntry, KvResource}; +use geth_pipe::{PipeConnection, PipeListener}; use geth_pubsub::PubsubMessage; use geth_resource::ResourceDescriptor; use geth_secrets::{BearerAccess, ResourceMasterSecret}; @@ -30,7 +31,7 @@ use geth_types::{ AuthOpId, Capability, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind, ResourceName, SshCertId, SshCertRequestId, UnixMillis, }; -use std::collections::VecDeque; +use std::collections::{BTreeMap, VecDeque}; use std::path::Path; use std::sync::{Arc, Mutex}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; @@ -80,6 +81,8 @@ pub enum NodeError { Secrets(#[from] geth_secrets::SecretsError), #[error("pubsub error: {0}")] Pubsub(#[from] geth_pubsub::PubsubError), + #[error("pipe error: {0}")] + Pipe(#[from] geth_pipe::PipeError), #[error("runtime state lock poisoned")] RuntimeLockPoisoned, #[error("invalid ssh certificate kind: {0}")] @@ -110,6 +113,7 @@ pub struct LocalNode { #[derive(Debug)] struct NodeRuntime { pubsub: Mutex, + pipes: Mutex, } #[derive(Debug, Default)] @@ -117,7 +121,14 @@ struct PubsubRuntime { messages: VecDeque, } +#[derive(Debug, Default)] +struct PipeRuntime { + listeners: BTreeMap, + connections: VecDeque, +} + const PUBSUB_RING_LIMIT: usize = 256; +const PIPE_CONNECTION_RING_LIMIT: usize = 256; pub fn init_node(paths: &GethPaths) -> Result { paths.ensure_base_dirs()?; @@ -143,6 +154,7 @@ pub fn init_node(paths: &GethPaths) -> Result { iroh_status: EndpointStatus::scaffolded(), runtime: Arc::new(NodeRuntime { pubsub: Mutex::new(PubsubRuntime::default()), + pipes: Mutex::new(PipeRuntime::default()), }), }) } @@ -890,6 +902,42 @@ pub fn handle_request( note: geth_pubsub::pubsub_storage_warning().to_owned(), }) } + ControlRequest::PipeListen { name } => { + geth_pipe::validate_pipe_name(&name)?; + let listener = PipeListener { + id: format!("pipe:{name}").into(), + name: name.clone(), + listened_at: UnixMillis(geth_store::now_ms()), + note: geth_pipe::local_pipe_runtime_note().to_owned(), + }; + let mut runtime = node + .runtime + .pipes + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)?; + runtime.listeners.insert(name, listener.clone()); + Ok(ControlResponse::PipeListening { listener }) + } + ControlRequest::PipeConnect { target } => { + geth_pipe::validate_pipe_name(&target)?; + let mut runtime = node + .runtime + .pipes + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)?; + let local_listener_found = runtime.listeners.contains_key(&target); + let connection = PipeConnection { + target, + connected_at: UnixMillis(geth_store::now_ms()), + local_listener_found, + note: geth_pipe::local_pipe_runtime_note().to_owned(), + }; + runtime.connections.push_back(connection.clone()); + while runtime.connections.len() > PIPE_CONNECTION_RING_LIMIT { + runtime.connections.pop_front(); + } + Ok(ControlResponse::PipeConnected { connection }) + } ControlRequest::ModuleStub { module, command } => { Ok(ControlResponse::NotImplemented { module, command }) } diff --git a/crates/geth-pipe/Cargo.toml b/crates/geth-pipe/Cargo.toml index 9bcefed..d70fd34 100644 --- a/crates/geth-pipe/Cargo.toml +++ b/crates/geth-pipe/Cargo.toml @@ -7,4 +7,5 @@ license.workspace = true [dependencies] serde.workspace = true +thiserror.workspace = true geth-types = { path = "../geth-types" } diff --git a/crates/geth-pipe/src/lib.rs b/crates/geth-pipe/src/lib.rs index 6f7a60c..b18a2d1 100644 --- a/crates/geth-pipe/src/lib.rs +++ b/crates/geth-pipe/src/lib.rs @@ -1,4 +1,4 @@ -use geth_types::{PipeId, ResourceId}; +use geth_types::{PipeId, ResourceId, UnixMillis}; use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -8,7 +8,60 @@ pub struct PipeResource { pub name: String, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PipeListener { + pub id: PipeId, + pub name: String, + pub listened_at: UnixMillis, + pub note: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PipeConnection { + pub target: String, + pub connected_at: UnixMillis, + pub local_listener_found: bool, + pub note: String, +} + +#[derive(Debug, thiserror::Error)] +pub enum PipeError { + #[error("invalid pipe name or target: {0}")] + InvalidName(String), +} + +pub fn validate_pipe_name(name: &str) -> Result<(), PipeError> { + if name.is_empty() + || !name + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':')) + { + return Err(PipeError::InvalidName(name.to_owned())); + } + Ok(()) +} + #[must_use] pub fn pipe_roadmap() -> &'static str { "future pipes are authorized Iroh bidirectional streams for stdin/stdout and forwarding" } + +#[must_use] +pub fn local_pipe_runtime_note() -> &'static str { + "local daemon registry only; byte streams and Iroh transport are not implemented yet" +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pipe_name_validation_rejects_paths_and_empty_names() { + assert!(validate_pipe_name("inbox").is_ok()); + assert!(validate_pipe_name("node:laptop").is_ok()); + assert!(validate_pipe_name("").is_err()); + assert!(validate_pipe_name("../inbox").is_err()); + assert!(validate_pipe_name("inbox/main").is_err()); + assert!(validate_pipe_name("inbox main").is_err()); + } +} diff --git a/crates/geth/tests/bootstrap.rs b/crates/geth/tests/bootstrap.rs index 73465a4..e85d5c9 100644 --- a/crates/geth/tests/bootstrap.rs +++ b/crates/geth/tests/bootstrap.rs @@ -873,6 +873,74 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() { ); } +#[test] +fn pipe_listen_connect_uses_local_runtime_registry() { + let home = tempfile::tempdir().expect("tempdir"); + let paths = geth_config::GethPaths::from_home(home.path()); + let node = geth_node::init_node(&paths).expect("init node"); + + let response = geth_node::handle_request( + &node, + geth_control::ControlRequest::PipeListen { + name: "inbox".to_owned(), + }, + ) + .expect("listen pipe"); + match response { + geth_control::ControlResponse::PipeListening { listener } => { + assert_eq!(listener.id.to_string(), "pipe:inbox"); + assert_eq!(listener.name, "inbox"); + assert!(listener.note.contains("local daemon registry")); + } + other => panic!("unexpected response: {other:?}"), + } + + let response = geth_node::handle_request( + &node, + geth_control::ControlRequest::PipeConnect { + target: "inbox".to_owned(), + }, + ) + .expect("connect pipe"); + match response { + geth_control::ControlResponse::PipeConnected { connection } => { + assert_eq!(connection.target, "inbox"); + assert!(connection.local_listener_found); + assert!( + connection + .note + .contains("Iroh transport are not implemented") + ); + } + other => panic!("unexpected response: {other:?}"), + } + + let reopened = geth_node::open_node(&paths).expect("reopen node"); + let response = geth_node::handle_request( + &reopened, + geth_control::ControlRequest::PipeConnect { + target: "inbox".to_owned(), + }, + ) + .expect("connect after reopen"); + match response { + geth_control::ControlResponse::PipeConnected { connection } => { + assert!(!connection.local_listener_found); + } + other => panic!("unexpected response: {other:?}"), + } + + assert!( + geth_node::handle_request( + &node, + geth_control::ControlRequest::PipeListen { + name: "../bad".to_owned(), + }, + ) + .is_err() + ); +} + #[test] fn ssh_cert_request_approval_and_revocation_export_use_local_state() { let home = tempfile::tempdir().expect("tempdir"); diff --git a/docs/architecture.md b/docs/architecture.md index b8a7840..b2e86ec 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -114,8 +114,12 @@ are lost when the daemon stops. This is deliberate: pubsub is a lossy wakeup and presence channel, not authoritative storage. Iroh-gossip replication, private topics, and capability enforcement are future work. -`geth-pipe` and `geth-ssh-proxy` currently define types, command shape, and -roadmap stubs. +`geth-pipe` currently supports `pipe listen/connect` against a local +daemon-lifetime registry. This is a control-plane scaffold for names and +connection attempts only; it does not carry bytes, forward sockets, or use Iroh +streams yet. + +`geth-ssh-proxy` currently defines types, command shape, and roadmap stubs. `geth-ssh-identity` defines SSH trust namespaces plus certificate request, approval, certificate import, and revocation-list data models. The bootstrap diff --git a/docs/roadmap.md b/docs/roadmap.md index 9496ab5..aca8a30 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -251,11 +251,15 @@ authorization and durable-state boundaries clear. Goal: add authorized stream-oriented management workflows over Iroh. -- `[ ]` Dumbpipe-style Iroh streams. +- `[~]` Dumbpipe-style Iroh streams. Acceptance criteria: - - `geth pipe listen/connect` can connect two local test nodes. - - Pipe access requires `pipe.listen` or `pipe.connect`. - - Streams close cleanly and propagate errors. + - `[x]` `geth pipe listen/connect` works against a local daemon-lifetime + registry. + - `[x]` Tests cover local listener registration, local connect matching, and + daemon restart behavior. + - `[ ]` `geth pipe listen/connect` can connect two local test nodes over Iroh. + - `[ ]` Pipe access requires `pipe.listen` or `pipe.connect`. + - `[ ]` Streams close cleanly and propagate errors. - `[ ]` TCP forwarding. Acceptance criteria: