diff --git a/Cargo.lock b/Cargo.lock index 4562675..59178c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1621,6 +1621,7 @@ dependencies = [ "geth-types", "hex", "iroh", + "iroh-blobs", "rusqlite", "serde", "serde_json", diff --git a/README.md b/README.md index 3b0eb06..8485a48 100644 --- a/README.md +++ b/README.md @@ -192,18 +192,17 @@ imported peer card and exchange a signed candidate-only peer-card ping. `geth peer auth-check ` sends a protected Iroh control request: the remote daemon verifies that the caller's signed peer card binds the actual Iroh EndpointID before reducing resource-local auth ops. -`geth cas fetch ` uses the same protected Iroh control path to -request a blob from a peer. The remote daemon only returns bytes when the caller -has `cas.fetch` on `resource:cas:local`, and the caller verifies that the bytes -hash to the requested BLAKE3 CAS hash before storing them locally. Successful -fetches record the serving peer as a local provider, visible with -`geth cas providers `. This is the bootstrap transfer path; future work -will move provider/fetch behavior to `iroh-blobs`. `geth-iroh` is now pinned to -`iroh 0.95.1` and compiles the native backend libraries `iroh-blobs 0.97.0`, -`iroh-docs 0.95.0`, and `iroh-gossip 0.95.0` against the same daemon-owned -endpoint generation. `geth status` reports those backends as ready to wire; the -module implementations still use the explicit bootstrap control path until the -module-specific migrations replace it. +`geth cas fetch ` uses the same protected Iroh control path as +an authorization preflight. The remote daemon verifies the caller's signed peer +card against the observed Iroh EndpointID and requires `cas.fetch` on +`resource:cas:local`. After that preflight succeeds, the requester fetches the +blob payload over native `iroh-blobs` (`/iroh-bytes/4`) on the same daemon-owned +Iroh endpoint, verifies the BLAKE3 hash, stores it in local CAS, and records the +serving peer as a provider visible with `geth cas providers `. +`geth-iroh` is pinned to `iroh 0.95.1` and compiles the native backend +libraries `iroh-blobs 0.97.0`, `iroh-docs 0.95.0`, and `iroh-gossip 0.95.0` +against the same daemon-owned endpoint generation. KV and pubsub still use their +documented bootstrap equivalents until their native protocol migrations land. Remote resource commands that accept `--bearer-secret` can also authorize with a resource-scoped bearer proof generated from the private bearer token returned at creation time. The persisted auth log stores a public bearer id and token diff --git a/crates/geth-iroh/src/lib.rs b/crates/geth-iroh/src/lib.rs index 1f33bf3..9b4f953 100644 --- a/crates/geth-iroh/src/lib.rs +++ b/crates/geth-iroh/src/lib.rs @@ -44,7 +44,7 @@ impl GethIrohConfig { local_discovery: true, bind_ipv4: None, bind_ipv6: None, - alpns: default_protocol_router().alpns(), + alpns: default_endpoint_alpns(), } } } @@ -161,6 +161,17 @@ pub fn default_protocol_descriptors() -> Vec { ] } +#[must_use] +pub fn default_endpoint_alpns() -> Vec> { + let mut alpns = default_protocol_router().alpns(); + alpns.push(iroh_blobs::ALPN.to_vec()); + alpns.push(iroh_docs::ALPN.to_vec()); + alpns.push(iroh_gossip::ALPN.to_vec()); + alpns.sort(); + alpns.dedup(); + alpns +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct NativeIrohLibrary { pub module: String, @@ -447,6 +458,15 @@ mod tests { })); } + #[test] + fn default_endpoint_alpns_include_native_libraries() { + let alpns = default_endpoint_alpns(); + assert!(alpns.contains(&ALPN_CONTROL.to_vec())); + assert!(alpns.contains(&iroh_blobs::ALPN.to_vec())); + assert!(alpns.contains(&iroh_docs::ALPN.to_vec())); + assert!(alpns.contains(&iroh_gossip::ALPN.to_vec())); + } + #[test] fn default_router_registers_all_protocols() { let router = default_protocol_router(); diff --git a/crates/geth-node/Cargo.toml b/crates/geth-node/Cargo.toml index 6f23fd6..b4ed8aa 100644 --- a/crates/geth-node/Cargo.toml +++ b/crates/geth-node/Cargo.toml @@ -34,6 +34,7 @@ geth-store = { path = "../geth-store" } geth-types = { path = "../geth-types" } hex.workspace = true iroh.workspace = true +iroh-blobs.workspace = true swarm-discovery.workspace = true [dev-dependencies] diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 0ef4c3b..d251b15 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -51,6 +51,7 @@ use geth_types::{ AuthOpId, BlobHash, Capability, DeviceId, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind, ResourceName, SshCertId, SshCertRequestId, UnixMillis, UserId, }; +use iroh::protocol::ProtocolHandler; use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::path::{Component, Path, PathBuf}; use std::sync::{Arc, Mutex}; @@ -157,6 +158,7 @@ pub struct LocalNode { pub node_id: String, pub iroh_status: EndpointStatus, iroh_endpoint: Arc>>, + iroh_blob_store: Arc>>, runtime: Arc, } @@ -265,6 +267,7 @@ pub fn init_node(paths: &GethPaths) -> Result { node_id, iroh_status: EndpointStatus::scaffolded(), iroh_endpoint: Arc::new(Mutex::new(None)), + iroh_blob_store: Arc::new(Mutex::new(None)), runtime: Arc::new(NodeRuntime { pubsub: Mutex::new(PubsubRuntime::default()), pipes: Mutex::new(PipeRuntime::default()), @@ -748,7 +751,19 @@ pub async fn handle_request_async( name, bearer_secret, } => document_sync_from_peer(node, &peer_node, &name, bearer_secret).await, - other => handle_request(node, other), + other => { + let response = handle_request(node, other)?; + match &response { + ControlResponse::CasAdded { hash, .. } => { + mirror_local_blob_to_iroh_blobs(node, hash).await?; + } + ControlResponse::CasPrivateAdded { encrypted_hash, .. } => { + mirror_local_blob_to_iroh_blobs(node, encrypted_hash).await?; + } + _ => {} + } + Ok(response) + } } } @@ -1275,7 +1290,7 @@ async fn cas_fetch_from_peer( let conn = endpoint .endpoint() - .connect(node_addr, geth_iroh::ALPN_CONTROL) + .connect(node_addr.clone(), geth_iroh::ALPN_CONTROL) .await .map_err(|error| NodeError::IrohPeer(error.to_string()))?; let (mut send, mut recv) = conn @@ -1300,7 +1315,7 @@ async fn cas_fetch_from_peer( endpoint_id, hash: response_hash, size_bytes, - content_base64, + content_base64: _, allowed, reason, nonce: response_nonce, @@ -1319,27 +1334,19 @@ async fn cas_fetch_from_peer( note, }); } - let content = content_base64 - .ok_or_else(|| NodeError::IrohPeer("peer omitted CAS content".to_owned())) - .and_then(|content| { - base64::engine::general_purpose::STANDARD - .decode(content) - .map_err(|error| NodeError::IrohPeer(error.to_string())) - })?; - if content.len() as u64 != size_bytes { - return Err(NodeError::IrohPeer(format!( - "peer announced {size_bytes} CAS bytes but returned {} bytes", - content.len() - ))); - } - let cas = LocalCas::new(node.paths.cas_dir()); - let info = cas.add_bytes(&content)?; + let info = fetch_blob_via_iroh_blobs(node, node_addr, response_hash.clone()).await?; if info.hash != response_hash { return Err(NodeError::IrohPeer(format!( "peer returned content hash {} for requested {}", info.hash, response_hash ))); } + if info.size_bytes != size_bytes { + return Err(NodeError::IrohPeer(format!( + "peer authorized {size_bytes} CAS bytes but iroh-blobs fetched {} bytes", + info.size_bytes + ))); + } store.record_cas_object( info.hash.as_str(), info.size_bytes, @@ -1384,6 +1391,105 @@ async fn cas_fetch_from_peer( } } +async fn fetch_blob_via_iroh_blobs( + node: &LocalNode, + node_addr: iroh::EndpointAddr, + hash: BlobHash, +) -> Result { + let endpoint = node + .iroh_endpoint + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)? + .clone() + .ok_or(NodeError::IrohEndpointUnavailable)?; + let blob_store = iroh_blob_store(node)?.ok_or_else(|| { + NodeError::IrohPeer( + "iroh-blobs store is unavailable; restart the daemon with native Iroh enabled" + .to_owned(), + ) + })?; + let iroh_hash = iroh_blob_hash(&hash)?; + let conn = endpoint + .endpoint() + .connect(node_addr, iroh_blobs::ALPN) + .await + .map_err(|error| NodeError::IrohPeer(format!("iroh-blobs connect failed: {error}")))?; + blob_store + .remote() + .fetch(conn, iroh_blobs::HashAndFormat::raw(iroh_hash)) + .await + .map_err(|error| NodeError::IrohPeer(format!("iroh-blobs fetch failed: {error}")))?; + let bytes = blob_store + .get_bytes(iroh_hash) + .await + .map_err(|error| NodeError::IrohPeer(format!("iroh-blobs export failed: {error}")))?; + let cas = LocalCas::new(node.paths.cas_dir()); + let info = cas.add_bytes(bytes.as_ref())?; + mirror_local_blob_to_iroh_blobs(node, &info.hash).await?; + Ok(info) +} + +fn iroh_blob_store(node: &LocalNode) -> Result, NodeError> { + node.iroh_blob_store + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned) + .map(|store| store.clone()) +} + +fn iroh_blob_hash(hash: &BlobHash) -> Result { + let bytes = hex::decode(hash.as_str()).map_err(|error| { + NodeError::IrohPeer(format!("invalid CAS hash for iroh-blobs: {error}")) + })?; + let bytes: [u8; 32] = bytes + .try_into() + .map_err(|_| NodeError::IrohPeer("invalid CAS hash length for iroh-blobs".to_owned()))?; + Ok(iroh_blobs::Hash::from_bytes(bytes)) +} + +async fn mirror_local_blob_to_iroh_blobs( + node: &LocalNode, + hash: &BlobHash, +) -> Result<(), NodeError> { + let Some(blob_store) = iroh_blob_store(node)? else { + return Ok(()); + }; + let content = LocalCas::new(node.paths.cas_dir()).read_bytes(hash)?; + let tag = blob_store + .add_slice(&content) + .await + .map_err(|error| NodeError::IrohPeer(format!("iroh-blobs import failed: {error}")))?; + if tag.hash.to_string() != hash.as_str() { + return Err(NodeError::IrohPeer(format!( + "iroh-blobs imported hash {} but local CAS hash is {}", + tag.hash, hash + ))); + } + Ok(()) +} + +async fn mirror_local_cas_to_iroh_blobs( + paths: &GethPaths, + blob_store: &iroh_blobs::store::fs::FsStore, +) -> Result { + let cas = LocalCas::new(paths.cas_dir()); + let mut count = 0; + for blob in cas.list()? { + let content = cas.read_bytes(&blob.hash)?; + let tag = blob_store + .add_slice(&content) + .await + .map_err(|error| NodeError::IrohPeer(format!("iroh-blobs import failed: {error}")))?; + if tag.hash.to_string() != blob.hash.as_str() { + return Err(NodeError::IrohPeer(format!( + "iroh-blobs imported hash {} but local CAS hash is {}", + tag.hash, blob.hash + ))); + } + count += 1; + } + Ok(count) +} + async fn cas_root_sync_from_peer( node: &LocalNode, peer_node: &str, @@ -4442,6 +4548,16 @@ async fn handle_iroh_control_connection( .map_err(|error| NodeError::IrohPeer(error.to_string()))?; let remote_endpoint_id = conn.remote_id().to_string(); let alpn = display_alpn(conn.alpn()); + if alpn == display_alpn(iroh_blobs::ALPN) { + let blob_store = iroh_blob_store(&node)?.ok_or_else(|| { + NodeError::IrohPeer("iroh-blobs ALPN accepted but blob store is unavailable".to_owned()) + })?; + let protocol = iroh_blobs::BlobsProtocol::new(&blob_store, None); + return protocol + .accept(conn) + .await + .map_err(|error| NodeError::IrohPeer(format!("iroh-blobs accept failed: {error}"))); + } let (mut send, mut recv) = conn .accept_bi() .await @@ -9557,11 +9673,26 @@ async fn start_daemon_iroh_endpoint( let store = Store::open(&node.paths.metadata_db())?; store.upsert_node_endpoint(endpoint_id, &node.node_id, &node.agent_id, "iroh")?; } + let blob_store = + iroh_blobs::store::fs::FsStore::load(node.paths.cas_dir().join("iroh-blobs")) + .await + .map_err(|error| { + NodeError::IrohPeer(format!("failed to open iroh-blobs store: {error}")) + })?; + let mirrored = mirror_local_cas_to_iroh_blobs(&node.paths, &blob_store).await?; + tracing::info!( + mirrored_blobs = mirrored, + "native iroh-blobs store is ready" + ); node.iroh_status = status; *node .iroh_endpoint .lock() .map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(endpoint.clone()); + *node + .iroh_blob_store + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(blob_store); Ok(Some(endpoint)) } Err(error) => { diff --git a/docs/architecture.md b/docs/architecture.md index 823500f..b5db68e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -31,17 +31,21 @@ uses Iroh's default relay policy; local-only/offline development can set The native module-backend crates for the intended CAS, KV, and pubsub replacements now compile against the same endpoint generation: `iroh-blobs 0.97.0`, `iroh-docs 0.95.0`, and `iroh-gossip 0.95.0`. `geth-iroh` -exposes their native ALPNs so module migrations can register handlers without -creating a second daemon endpoint. `geth status` reports these backends as -ready to wire. The module implementations still use explicit bootstrap -control-ALPN paths until each module is migrated to its native protocol. +exposes their native ALPNs without creating a second daemon endpoint. CAS now +registers an `iroh-blobs` provider handler on `/iroh-bytes/4`; local CAS writes +are mirrored into the native blob store, and remote `geth cas fetch` performs a +geth control-ALPN authorization preflight before transferring payload bytes over +`iroh-blobs`. `geth status` reports these native backend libraries. KV and +pubsub still use explicit documented bootstrap equivalents until each module is +migrated to its native protocol. Module ALPNs are registered through `geth-iroh`'s protocol router scaffold. The router owns the default protocol descriptors, rejects duplicate ALPN registrations, and returns explicit unknown-ALPN errors. The current daemon -accept loop dispatches geth control, pipe, and SSH-proxy streams directly; the -next backend migrations should attach iroh-blobs, iroh-docs, and iroh-gossip -handlers to the same endpoint instead of creating parallel endpoints. +accept loop dispatches geth control, pipe, SSH-proxy, and native CAS blob +streams directly; the next backend migrations should attach iroh-docs and +iroh-gossip handlers to the same endpoint instead of creating parallel +endpoints. The target product should use Iroh relay support for practical internet connectivity and mDNS/LAN discovery for local networks. These are connectivity diff --git a/docs/roadmap.md b/docs/roadmap.md index 6ac5d62..253a765 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -458,9 +458,13 @@ authorization and durable-state boundaries clear. the CAS hash. - `[x]` `geth cas providers ` lists locally known providers. - `[x]` Tests cover local provider metadata storage. - - `[ ]` Replace the bootstrap control-ALPN transfer with `iroh-blobs` + - `[x]` Replace the bootstrap control-ALPN byte transfer with `iroh-blobs` provider/fetch behavior using the daemon-owned `iroh 0.95.1` endpoint and pinned `iroh-blobs 0.97.0`. + - `[x]` CAS fetch keeps the geth control-ALPN authorization preflight before + opening the native `iroh-blobs` payload transfer. + - `[x]` Local CAS adds and daemon startup mirror available blobs into the + native `iroh-blobs` store so peers can fetch them through `/iroh-bytes/4`. - `[x]` CAS pin and cache policy. Acceptance criteria: