diff --git a/README.md b/README.md index 65c846e..186361b 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,13 @@ $ varde-ctl status # shows this daemon's node id $ varde-ctl peer trust # on both ends ``` +Trusted peers can also hand content to each other directly, no ticket +round-trip — the receiver pins what it accepted: + +```console +$ varde-ctl push 5b1c…e0 +``` + ## Building and testing ```console diff --git a/SPECS.md b/SPECS.md index 3b5fb94..bad031a 100644 --- a/SPECS.md +++ b/SPECS.md @@ -53,8 +53,9 @@ Requests (define these as enums in `varde-proto`): | `Status { hash? }` | Global or per-hash: have/missing bytes, peers, transfer rates. | | `List {}` | All pins with policies and completeness. | | `TicketExport { hash }` / `TicketImport { ticket, pin_policy }` | iroh blob tickets — the v1 out-of-band sharing mechanism. | -| `Gc {}` | Drop unpinned blobs. | -| `Subscribe {}` | Switch connection to event stream (transfer progress, peer joined, pin complete). | +| `Push { hash, node_id }` | Hand fully-present content to a trusted peer, unprompted. Mutual consent: sender pushes only to peers it trusts, receiver accepts only from peers it trusts, and pins what it accepted. Delivery is verified via the peer's bitfields before the reply. | +| `Gc {}` | Answers unimplemented since iroh-blobs 0.103: gc runs continuously inside the store (`gc_interval_secs`, default 300), protecting pins via a roots snapshot. | +| `Subscribe {}` | Switch connection to event stream (transfer progress, peer joined, pin complete, push received). | Every response carries `{"ok": bool, ...}`. Errors are structured (`code`, `message`), never bare strings. diff --git a/varde-ctl/src/main.rs b/varde-ctl/src/main.rs index 4d55e90..a58a77d 100644 --- a/varde-ctl/src/main.rs +++ b/varde-ctl/src/main.rs @@ -73,6 +73,14 @@ enum Command { /// Peer trust operations #[command(subcommand)] Peer(PeerCommand), + /// Push fully-present content to a trusted peer + Push { + /// BLAKE3 hash (hex) + hash: String, + /// The receiving peer's iroh NodeId (z-base-32); must be trusted + /// on both sides + node_id: String, + }, /// Drop all unpinned blobs Gc, /// Stream daemon events to stdout (one JSON object per line) @@ -180,6 +188,7 @@ fn to_request(command: Command) -> Result { Command::Peer(PeerCommand::Trust { node_id }) => Request::PeerTrust { node_id }, Command::Peer(PeerCommand::Untrust { node_id }) => Request::PeerUntrust { node_id }, Command::Peer(PeerCommand::List) => Request::PeerList {}, + Command::Push { hash, node_id } => Request::Push { hash, node_id }, Command::Gc => Request::Gc {}, Command::Subscribe => Request::Subscribe {}, }) @@ -291,5 +300,8 @@ fn print_human(data: Option<&ResponseData>) { ResponseData::GcDone { blobs_removed } => { println!("gc: removed {blobs_removed} blobs"); } + ResponseData::Pushed { hash, bytes } => { + println!("pushed {hash} ({bytes} bytes)"); + } } } diff --git a/varde-daemon/src/daemon.rs b/varde-daemon/src/daemon.rs index 33ddac9..e05facf 100644 --- a/varde-daemon/src/daemon.rs +++ b/varde-daemon/src/daemon.rs @@ -125,6 +125,29 @@ impl Daemon { } }); + // Inbound pushes: content a trusted peer handed us, fully + // verified. Pin it (default policy) so it survives gc, and tell + // subscribers where it came from. + if let Some(mut pushes) = daemon.transfer.take_pushes() { + let weak = Arc::downgrade(&daemon); + tokio::spawn(async move { + while let Some(push) = pushes.recv().await { + let Some(daemon) = weak.upgrade() else { break }; + let hex = push.hash.to_hex().to_string(); + let from = push.from.clone().unwrap_or_default(); + info!(hash = %hex, from = %from, "push received, pinning"); + if let Err(e) = daemon.meta.record_format(&hex, push.format) { + warn!(hash = %hex, error = %e, "recording pushed format"); + } + if let Err(e) = daemon.meta.pin(&hex, varde_proto::PinPolicy::default()) { + warn!(hash = %hex, error = %e, "pinning pushed content"); + } + daemon.recompute_open_set().await; + let _ = daemon.events.send(Event::PushReceived { hash: hex, from }); + } + }); + } + // Discovery: raw mdns sightings feed presence tracking, and are // re-emitted as signed announcements through the LanDiscovery // provider — the same seam a future gossip provider plugs into. @@ -513,6 +536,35 @@ impl Daemon { Request::PeerList {} => Ok(ResponseData::Peers { peers: self.present_trusted_peers(), }), + Request::Push { hash, node_id } => { + let parsed = parse_hash(&hash)?; + let peer: iroh::EndpointId = node_id + .parse() + .map_err(|e| OpError::InvalidArgument(format!("invalid node id: {e}")))?; + // Consent, sending side: we only push to peers we trust. + // (The receiver independently requires that it trusts us.) + if !self.meta.is_trusted(&peer.to_string()) { + return Err(OpError::InvalidArgument(format!( + "{peer} is not a trusted peer; trust it before pushing" + ))); + } + let (_, _, complete) = self.store.presence(parsed).await?; + if !complete { + return Err(OpError::NotFound(format!( + "{} is not fully present; cannot push", + parsed.to_hex() + ))); + } + let content = HashAndFormat { + hash: parsed, + format: blob_format(self.meta.format_of(&parsed.to_hex())), + }; + let bytes = self.transfer.push_to(peer, content).await?; + Ok(ResponseData::Pushed { + hash: parsed.to_hex().to_string(), + bytes, + }) + } Request::Subscribe {} => Ok(ResponseData::Done {}), } } diff --git a/varde-daemon/src/shaped.rs b/varde-daemon/src/shaped.rs index 95b8371..0a9037f 100644 --- a/varde-daemon/src/shaped.rs +++ b/varde-daemon/src/shaped.rs @@ -24,7 +24,7 @@ use tokio::sync::broadcast; use tracing::debug; use varde_proto::{Direction, Event}; -use crate::meta::Meta; +use crate::meta::{Meta, StoredFormat}; use crate::metered::MeteredState; /// An async token bucket. Rate 0 means unlimited. @@ -86,6 +86,15 @@ impl TokenBucket { /// Predicate deciding whether a hash may be served to untrusted peers. pub type ServeFilter = Arc bool + Send + Sync>; +/// A completed inbound push, reported to the daemon for pinning. +#[derive(Debug)] +pub struct ReceivedPush { + pub hash: Hash, + pub format: StoredFormat, + /// NodeId of the pushing peer, when the connection exposed one. + pub from: Option, +} + /// Provider gate configuration. pub struct Gate { pub meta: Arc, @@ -94,6 +103,15 @@ pub struct Gate { pub serving_enabled: bool, pub metered: MeteredState, pub events: broadcast::Sender, + /// Completed pushes flow to the daemon here. + pub pushes: tokio::sync::mpsc::Sender, +} + +/// What the gate knows about one live connection. +#[derive(Debug, Clone)] +struct ConnInfo { + trusted: bool, + endpoint_id: Option, } /// Spawn the provider event handler and return the [`EventSender`] to @@ -103,18 +121,22 @@ pub fn provider_events(gate: Gate) -> EventSender { connected: ConnectMode::Intercept, get: RequestMode::InterceptLog, get_many: RequestMode::InterceptLog, - push: RequestMode::Disabled, + // Push writes to the local store: intercepted and allowed for + // trusted peers only, with completion tracked for pinning. + push: RequestMode::InterceptLog, observe: ObserveMode::Intercept, throttle: ThrottleMode::Intercept, }; let (sender, mut rx) = EventSender::channel(32, mask); tokio::spawn(async move { // Trust decided once per connection, consulted per request. - let mut trusted_conns: HashMap = HashMap::new(); + let mut conns: HashMap = HashMap::new(); while let Some(msg) = rx.recv().await { match msg { ProviderMessage::ClientConnected(m) => { - let allowed = gate.serving_enabled && !gate.metered.is_metered(); + // Connections are always accepted; serving and + // metered state gate individual requests, so a + // non-serving daemon can still receive pushes. let trusted = m .inner .endpoint_id @@ -124,22 +146,21 @@ pub fn provider_events(gate: Gate) -> EventSender { connection_id, endpoint_id, } = m.inner; - debug!(?endpoint_id, trusted, allowed, "incoming blobs connection"); - if allowed { - trusted_conns.insert(connection_id, trusted); - } - let verdict = if allowed { - Ok(()) - } else { - Err(AbortReason::Permission) - }; - m.tx.send(verdict).await.ok(); + debug!(?endpoint_id, trusted, "incoming blobs connection"); + conns.insert( + connection_id, + ConnInfo { + trusted, + endpoint_id: endpoint_id.map(|id| id.to_string()), + }, + ); + m.tx.send(Ok(())).await.ok(); } ProviderMessage::ConnectionClosed(m) => { - trusted_conns.remove(&m.inner.connection_id); + conns.remove(&m.inner.connection_id); } ProviderMessage::GetRequestReceived(m) => { - let allowed = allow(&gate, &trusted_conns, m.inner.connection_id, [ + let allowed = allow(&gate, &conns, m.inner.connection_id, [ m.inner.request.hash, ]); let verdict = if allowed { @@ -155,7 +176,7 @@ pub fn provider_events(gate: Gate) -> EventSender { ProviderMessage::GetManyRequestReceived(m) => { let allowed = allow( &gate, - &trusted_conns, + &conns, m.inner.connection_id, m.inner.request.hashes.iter().copied(), ); @@ -171,9 +192,17 @@ pub fn provider_events(gate: Gate) -> EventSender { } ProviderMessage::ObserveRequestReceived(m) => { // Presence is information too: gate it like data. - let allowed = allow(&gate, &trusted_conns, m.inner.connection_id, [ - m.inner.request.hash, - ]); + // Bitfields are metadata, not payload, so trusted + // peers may observe even while serving is off or the + // connection is metered (push confirmation needs it). + let trusted = conns + .get(&m.inner.connection_id) + .map(|i| i.trusted) + .unwrap_or(false); + let allowed = trusted + || allow(&gate, &conns, m.inner.connection_id, [ + m.inner.request.hash, + ]); let verdict = if allowed { Ok(()) } else { @@ -181,6 +210,34 @@ pub fn provider_events(gate: Gate) -> EventSender { }; m.tx.send(verdict).await.ok(); } + ProviderMessage::PushRequestReceived(m) => { + // Inbound writes: trusted peers only, deferred while + // metered (RateLimited invites a retry later). + let info = conns.get(&m.inner.connection_id); + let trusted = info.map(|i| i.trusted).unwrap_or(false); + debug!(hash = %m.inner.request.hash.to_hex(), trusted, "push request received"); + let verdict = if !trusted { + Err(AbortReason::Permission) + } else if gate.metered.is_metered() { + Err(AbortReason::RateLimited) + } else { + Ok(()) + }; + let allowed = verdict.is_ok(); + m.tx.send(verdict).await.ok(); + if allowed { + let received = ReceivedPush { + hash: m.inner.request.hash, + format: if m.inner.request.ranges.is_blob() { + StoredFormat::Raw + } else { + StoredFormat::HashSeq + }, + from: info.and_then(|i| i.endpoint_id.clone()), + }; + watch_push(m.rx, received, gate.pushes.clone()); + } + } ProviderMessage::Throttle(m) => { // Pay for the chunk from the shared upload bucket off // the handler loop, so a deep debt doesn't stall @@ -191,8 +248,8 @@ pub fn provider_events(gate: Gate) -> EventSender { m.tx.send(Ok(())).await.ok(); }); } - // Push is disabled in the mask; notify variants are not - // subscribed. Ignore anything else. + // Notify variants are not subscribed. Ignore anything + // else. _ => {} } } @@ -200,20 +257,51 @@ pub fn provider_events(gate: Gate) -> EventSender { sender } -/// A request is allowed if the connection is trusted, or every requested +/// A get/observe request is allowed if serving is on, the connection is +/// not metered, and either the connection is trusted or every requested /// hash is openly served. fn allow( gate: &Gate, - trusted_conns: &HashMap, + conns: &HashMap, connection_id: u64, hashes: impl IntoIterator, ) -> bool { - if trusted_conns.get(&connection_id).copied().unwrap_or(false) { + if !gate.serving_enabled || gate.metered.is_metered() { + return false; + } + if conns + .get(&connection_id) + .map(|i| i.trusted) + .unwrap_or(false) + { return true; } hashes.into_iter().all(|hash| (gate.open_filter)(&hash)) } +/// Watch a push request's update stream; report it to the daemon only if +/// the transfer runs to completion (failed pushes stay unpinned and are +/// swept by gc). +fn watch_push( + mut rx: irpc::channel::mpsc::Receiver, + received: ReceivedPush, + pushes: tokio::sync::mpsc::Sender, +) { + use iroh_blobs::provider::events::RequestUpdate; + tokio::spawn(async move { + while let Ok(Some(update)) = rx.recv().await { + match update { + RequestUpdate::Completed(_) => { + pushes.send(received).await.ok(); + return; + } + RequestUpdate::Aborted(_) => return, + _ => {} + } + } + }); +} + /// Forward per-request transfer updates into the daemon event stream. fn forward_progress( mut rx: irpc::channel::mpsc::Receiver, diff --git a/varde-daemon/src/store.rs b/varde-daemon/src/store.rs index f965fa2..3229d10 100644 --- a/varde-daemon/src/store.rs +++ b/varde-daemon/src/store.rs @@ -330,9 +330,13 @@ impl BlobStore { pub async fn presence(&self, hash: Hash) -> Result<(u64, Option, bool), OpError> { match self.store.blobs().status(hash).await.map_err(internal)? { BlobStatus::NotFound => Ok((0, None, false)), - // Valid-range accounting for partials arrives with the - // transfer milestone; absence of data is the safe report. - BlobStatus::Partial { size } => Ok((0, size, false)), + BlobStatus::Partial { size } => { + // Chunk-accurate accounting from the store's bitfield: + // report the bytes actually present and verified. + let bitfield = self.store.blobs().observe(hash).await.map_err(internal)?; + let total = bitfield.validated_size().or(size); + Ok((bitfield.total_bytes().min(total.unwrap_or(u64::MAX)), total, false)) + } BlobStatus::Complete { size } => Ok((size, Some(size), true)), } } diff --git a/varde-daemon/src/transfer.rs b/varde-daemon/src/transfer.rs index fc8bd00..78b7494 100644 --- a/varde-daemon/src/transfer.rs +++ b/varde-daemon/src/transfer.rs @@ -13,6 +13,7 @@ use iroh::address_lookup::memory::MemoryLookup; use iroh::protocol::Router; use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey}; use iroh_blobs::api::downloader::{DownloadOptions, Downloader, Shuffled, SplitStrategy}; +use iroh_blobs::protocol::GetRequest; use iroh_blobs::{BlobFormat, BlobsProtocol, Hash, HashAndFormat}; use iroh_mdns_address_lookup::{DiscoveryEvent, MdnsAddressLookup}; use n0_future::StreamExt; @@ -23,7 +24,7 @@ use varde_proto::{Direction, Event}; use crate::config::Config; use crate::meta::Meta; use crate::metered::MeteredState; -use crate::shaped::{Gate, ServeFilter, TokenBucket}; +use crate::shaped::{Gate, ReceivedPush, ServeFilter, TokenBucket}; use crate::store::BlobStore; /// The network side of the daemon: one iroh endpoint serving the blob @@ -33,6 +34,7 @@ use crate::store::BlobStore; pub struct Transfer { endpoint: Endpoint, router: Router, + store: iroh_blobs::api::Store, downloader: Downloader, /// Address book for provider addresses learned out-of-band (ticket /// imports): the downloader dials by endpoint id and resolves @@ -41,6 +43,7 @@ pub struct Transfer { events: broadcast::Sender, metered: MeteredState, discovery_rx: std::sync::Mutex>>, + pushes_rx: std::sync::Mutex>>, } impl Transfer { @@ -108,6 +111,7 @@ impl Transfer { // Best-effort low-priority marking of our UDP traffic. crate::dscp::mark_endpoint_sockets(&endpoint.bound_sockets()); + let (pushes_tx, pushes_rx) = mpsc::channel(16); let provider_events = crate::shaped::provider_events(Gate { meta, open_filter, @@ -115,6 +119,7 @@ impl Transfer { serving_enabled: config.max_upload_bytes_per_sec > 0, metered: metered.clone(), events: events.clone(), + pushes: pushes_tx, }); let provider = BlobsProtocol::new(store.api(), Some(provider_events)); @@ -126,11 +131,13 @@ impl Transfer { Ok(Transfer { endpoint, router, + store: store.api().clone(), downloader, known_addrs, events, metered, discovery_rx: std::sync::Mutex::new(Some(discovery_rx)), + pushes_rx: std::sync::Mutex::new(Some(pushes_rx)), }) } @@ -151,6 +158,108 @@ impl Transfer { self.discovery_rx.lock().expect("discovery lock").take() } + /// Take the stream of completed inbound pushes. Callable once. + pub fn take_pushes(&self) -> Option> { + self.pushes_rx.lock().expect("pushes lock").take() + } + + /// Push fully-present `content` to `peer`, unprompted. The peer + /// accepts only if it trusts us. Returns payload bytes written. + pub async fn push_to(&self, peer: EndpointId, content: HashAndFormat) -> Result { + anyhow::ensure!( + !self.metered.is_metered(), + "metered connection: refusing to push" + ); + let conn = self + .endpoint + .connect(peer, iroh_blobs::ALPN) + .await + .with_context(|| format!("connecting to {peer}"))?; + let request = match content.format { + BlobFormat::Raw => { + iroh_blobs::protocol::PushRequest::from(GetRequest::blob(content.hash)) + } + BlobFormat::HashSeq => { + iroh_blobs::protocol::PushRequest::from(GetRequest::all(content.hash)) + } + }; + // Upstream returns zeroed Stats for pushes; the byte count rides + // the progress stream instead. + use iroh_blobs::api::remote::PushProgressItem; + let mut stream = self + .store + .remote() + .execute_push(conn.clone(), request) + .stream(); + let mut bytes = 0u64; + let mut done = false; + while let Some(item) = stream.next().await { + match item { + PushProgressItem::Progress(sent) => { + bytes = sent; + let _ = self.events.send(Event::TransferProgress { + hash: content.hash.to_hex().to_string(), + direction: Direction::Upload, + bytes_done: sent, + bytes_total: None, + }); + } + PushProgressItem::Done(_) => { + done = true; + break; + } + PushProgressItem::Error(e) => { + return Err(anyhow::anyhow!(e)).context("pushing content"); + } + } + } + anyhow::ensure!(done, "push stream ended without a result"); + + // Writing is fire-and-forget at the QUIC level: closing the + // connection now could discard data the peer has not read yet, + // and says nothing about acceptance. Delivery is confirmed by + // observing the peer's bitfields until they report complete — + // the peer imports the root first, then children in order, so + // root + last child covers the whole request. + self.wait_remote_complete(&conn, content.hash).await?; + if content.format == BlobFormat::HashSeq { + let seq_bytes = self + .store + .blobs() + .get_bytes(content.hash) + .await + .context("reading pushed hashseq")?; + let seq = iroh_blobs::hashseq::HashSeq::try_from(seq_bytes) + .map_err(|e| anyhow::anyhow!("invalid hashseq: {e}"))?; + if let Some(last) = seq.iter().last() { + self.wait_remote_complete(&conn, last).await?; + } + } + info!(hash = %content.hash.to_hex(), peer = %peer, bytes, "push complete, verified by peer"); + Ok(bytes) + } + + /// Watch `hash` on the remote end of `conn` until its bitfield + /// reports the blob complete. + async fn wait_remote_complete( + &self, + conn: &iroh::endpoint::Connection, + hash: Hash, + ) -> Result<()> { + let observe = self.store.remote().observe( + conn.clone(), + iroh_blobs::protocol::ObserveRequest::new(hash), + ); + let mut observe = std::pin::pin!(observe); + while let Some(bitfield) = observe.next().await { + let bitfield = bitfield.context("observing pushed content on the peer")?; + if bitfield.is_complete() { + return Ok(()); + } + } + anyhow::bail!("peer stopped reporting before the pushed content completed"); + } + /// Produce a ticket for out-of-band sharing of `content`. pub async fn export_ticket(&self, content: HashAndFormat) -> Result { let addr = self.endpoint.addr(); @@ -181,13 +290,20 @@ impl Transfer { } } + // With several providers one fetch stripes across all of them; + // a lone provider gets the whole request. + let strategy = if ids.len() > 1 { + SplitStrategy::Split + } else { + SplitStrategy::None + }; let downloader = self.downloader.clone(); let events = self.events.clone(); tokio::spawn(async move { let progress = downloader.download_with_opts(DownloadOptions::new( content, Shuffled::new(ids), - SplitStrategy::None, + strategy, )); let mut stream = match progress.stream().await { Ok(stream) => stream, diff --git a/varde-daemon/tests/lan_trust.rs b/varde-daemon/tests/lan_trust.rs index 4a30fcf..51eff2e 100644 --- a/varde-daemon/tests/lan_trust.rs +++ b/varde-daemon/tests/lan_trust.rs @@ -280,3 +280,137 @@ fn overlapping_pins_auto_sync_via_lan_discovery() { assert!(reply.ok, "materialize failed: {:?}", reply.error); assert_eq!(std::fs::read(&dest).unwrap(), payload); } + +/// Wait until `client` sees at least one connected trusted peer; false +/// if the deadline passes (multicast unavailable). +fn wait_peer_connected(client: &mut support::Client, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let reply = client.request(&Request::PeerList {}); + if let Some(ResponseData::Peers { peers }) = reply.data { + if peers.iter().any(|p| p.connected) { + return true; + } + } + std::thread::sleep(Duration::from_millis(250)); + } + false +} + +/// Push: A hands pinned content to B unprompted. Requires mutual trust +/// and mdns (skips without multicast). The receiver pins what it +/// accepted, so the content survives gc and lists as a pin. +#[test] +fn push_hands_content_to_trusted_peer() { + let dir_a = tempfile::tempdir().unwrap(); + let dir_b = tempfile::tempdir().unwrap(); + let a = support::spawn_daemon_with_env(dir_a.path(), &[("VARDE_DISCOVERY", "true")]); + let b = support::spawn_daemon_with_env(dir_b.path(), &[("VARDE_DISCOVERY", "true")]); + let mut client_a = support::Client::connect(&a.socket); + let mut client_b = support::Client::connect(&b.socket); + + let id_a = node_id(&mut client_a); + let id_b = node_id(&mut client_b); + + let dir = tempfile::tempdir().unwrap(); + let src = dir.path().join("gift.bin"); + let payload = vec![3u8; 300_000]; + std::fs::write(&src, &payload).unwrap(); + let hash = add_file(&mut client_a, &src); + pin(&mut client_a, &hash, false); + + // Sender-side consent: pushing to a peer we don't trust is refused + // locally, before any connection is made. + let reply = client_a.request(&Request::Push { + hash: hash.clone(), + node_id: id_b.clone(), + }); + assert!(!reply.ok, "push to untrusted peer must be refused"); + + // Mutual trust, established before any blob connection exists. + assert!(client_a.request(&Request::PeerTrust { node_id: id_b }).ok); + assert!(client_b.request(&Request::PeerTrust { node_id: id_a }).ok); + + if !wait_peer_connected(&mut client_a, Duration::from_secs(30)) { + eprintln!("SKIP: mdns discovery saw no peers (multicast unavailable in this environment)"); + return; + } + + let reply = client_a.request(&Request::Push { + hash: hash.clone(), + node_id: node_id(&mut client_b), + }); + assert!(reply.ok, "push failed: {:?}", reply.error); + match reply.data { + Some(ResponseData::Pushed { bytes, .. }) => { + assert_eq!(bytes, payload.len() as u64, "push moved the payload") + } + other => panic!("unexpected push reply: {other:?}"), + } + + // The receiver has the bytes, and pinned them. + if !wait_complete(&b.socket, &hash, Duration::from_secs(30)) { + let reply = client_b.request(&Request::Status { + hash: Some(hash.clone()), + }); + panic!("pushed content not complete on receiver; status: {reply:?}"); + } + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let reply = client_b.request(&Request::List {}); + if let Some(ResponseData::Pins { pins }) = &reply.data { + if pins.iter().any(|p| p.hash == hash && p.complete) { + break; + } + } + assert!( + Instant::now() < deadline, + "receiver did not pin pushed content" + ); + std::thread::sleep(Duration::from_millis(250)); + } + let dest = dir.path().join("gift-out.bin"); + let reply = client_b.request(&Request::Materialize { + hash, + dest: dest.display().to_string(), + mode: MaterializeMode::Copy, + }); + assert!(reply.ok, "materialize failed: {:?}", reply.error); + assert_eq!(std::fs::read(&dest).unwrap(), payload); +} + +/// Consent is mutual: a receiver that does not trust the sender rejects +/// the push, even though the sender trusts the receiver. +#[test] +fn push_rejected_without_receiver_trust() { + let dir_a = tempfile::tempdir().unwrap(); + let dir_b = tempfile::tempdir().unwrap(); + let a = support::spawn_daemon_with_env(dir_a.path(), &[("VARDE_DISCOVERY", "true")]); + let b = support::spawn_daemon_with_env(dir_b.path(), &[("VARDE_DISCOVERY", "true")]); + let mut client_a = support::Client::connect(&a.socket); + let mut client_b = support::Client::connect(&b.socket); + + let id_b = node_id(&mut client_b); + + let dir = tempfile::tempdir().unwrap(); + let src = dir.path().join("unwanted.bin"); + std::fs::write(&src, vec![5u8; 100_000]).unwrap(); + let hash = add_file(&mut client_a, &src); + pin(&mut client_a, &hash, false); + + // One-way trust only: A trusts B, B has never heard of A. + assert!(client_a.request(&Request::PeerTrust { node_id: id_b }).ok); + if !wait_peer_connected(&mut client_a, Duration::from_secs(30)) { + eprintln!("SKIP: mdns discovery saw no peers (multicast unavailable in this environment)"); + return; + } + + let reply = client_a.request(&Request::Push { + hash: hash.clone(), + node_id: node_id(&mut client_b), + }); + assert!(!reply.ok, "receiver without trust must reject the push"); + + // And the receiver holds none of it. + assert!(!is_complete(&b.socket, &hash)); +} diff --git a/varde-proto/src/lib.rs b/varde-proto/src/lib.rs index 85a04a5..25c6573 100644 --- a/varde-proto/src/lib.rs +++ b/varde-proto/src/lib.rs @@ -18,7 +18,7 @@ use serde::{Deserialize, Serialize}; /// Version of the socket protocol described by this crate. /// /// Bumped on incompatible changes; reported in [`GlobalStatus`]. -pub const PROTOCOL_VERSION: u32 = 1; +pub const PROTOCOL_VERSION: u32 = 2; /// A request from a client to the daemon. One JSON object per line. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -96,6 +96,18 @@ pub enum Request { }, /// List trusted peers and their current connectivity. PeerList {}, + /// Push fully-present content to a trusted peer, unprompted. + /// + /// Consent is mutual: this daemon only pushes to peers it trusts, and + /// the receiver only accepts pushes from peers *it* trusts. On the + /// receiving side an accepted push is recorded as a pin (default + /// policy), so the content survives gc and shows up in `List`. + Push { + /// Hash (hex) of the fully-present blob or HashSeq root to push. + hash: String, + /// The receiving peer's iroh NodeId (z-base-32). + node_id: String, + }, /// Drop all blobs not reachable from a pin. Explicit, never automatic. Gc {}, /// Switch this connection to an event stream. The daemon acknowledges @@ -240,6 +252,13 @@ pub enum ResponseData { /// Number of blobs removed. blobs_removed: u64, }, + /// Content was pushed to a peer. Reply to [`Request::Push`]. + Pushed { + /// The pushed root hash (hex). + hash: String, + /// Payload bytes written to the peer. + bytes: u64, + }, } /// Global daemon status. @@ -334,6 +353,14 @@ pub enum Event { /// The pinned root hash (hex). hash: String, }, + /// A trusted peer pushed content to this daemon; it has been pinned + /// with the default policy. + PushReceived { + /// The pushed root hash (hex). + hash: String, + /// NodeId (z-base-32) of the pushing peer. + from: String, + }, /// A signed announcement was received on a subscribed discovery topic. Announcement { /// The announced HashSeq root (hex).