bl
4033c0ffab
Three capabilities the iroh-blobs 0.103 line makes possible:
- Push (new API surface, protocol v2): `Push { hash, node_id }` hands
fully-present content to a trusted peer, unprompted. Consent is
mutual — the sender pushes only to peers it trusts, and the receiver
(which now accepts connections unconditionally and gates per request)
admits pushes only from peers *it* trusts, deferring with RateLimited
while metered. An accepted push is pinned by the receiver (default
policy, format inferred from the request ranges) once its transfer
completes, and surfaces as a PushReceived event. Because QUIC writes
are fire-and-forget, the sender confirms delivery by observing the
receiver's bitfields (root + last hashseq child) before replying —
Pushed { bytes } means verified received, not merely sent. New
varde-ctl `push` command.
- Truthful partial presence: Status/List report bytes actually present
and verified for partial blobs, from the store's bitfields via
observe, instead of the old "0 until complete".
- Split downloads: multi-provider fetches stripe one request across
providers (SplitStrategy::Split) instead of trying them serially.
Trusted peers may observe bitfields even when serving is disabled or
metered — bitfields are metadata, and push confirmation rides on them.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
417 lines
16 KiB
Rust
417 lines
16 KiB
Rust
//! Milestone 4: trust gating, rate limiting, event streaming, and
|
|
//! mdns-based auto-sync between trusted daemons.
|
|
|
|
mod support;
|
|
|
|
use std::path::Path;
|
|
use std::str::FromStr;
|
|
use std::time::{Duration, Instant};
|
|
|
|
use varde_proto::{MaterializeMode, PinPolicy, Request, ResponseData};
|
|
|
|
fn add_file(client: &mut support::Client, path: &Path) -> String {
|
|
let reply = client.request(&Request::Add {
|
|
path: path.display().to_string(),
|
|
recursive: false,
|
|
});
|
|
assert!(reply.ok, "add failed: {:?}", reply.error);
|
|
match reply.data {
|
|
Some(ResponseData::Added { hash, .. }) => hash,
|
|
other => panic!("unexpected add reply: {other:?}"),
|
|
}
|
|
}
|
|
|
|
fn pin(client: &mut support::Client, hash: &str, open_lan: bool) {
|
|
let reply = client.request(&Request::Pin {
|
|
hash: hash.to_string(),
|
|
policy: PinPolicy { open_lan },
|
|
});
|
|
assert!(reply.ok, "pin failed: {:?}", reply.error);
|
|
}
|
|
|
|
fn export_ticket(client: &mut support::Client, hash: &str) -> String {
|
|
let reply = client.request(&Request::TicketExport {
|
|
hash: hash.to_string(),
|
|
});
|
|
assert!(reply.ok, "ticket export failed: {:?}", reply.error);
|
|
match reply.data {
|
|
Some(ResponseData::Ticket { ticket }) => ticket,
|
|
other => panic!("unexpected ticket reply: {other:?}"),
|
|
}
|
|
}
|
|
|
|
fn import_ticket(client: &mut support::Client, ticket: &str) {
|
|
let reply = client.request(&Request::TicketImport {
|
|
ticket: ticket.to_string(),
|
|
pin_policy: PinPolicy::default(),
|
|
});
|
|
assert!(reply.ok, "ticket import failed: {:?}", reply.error);
|
|
}
|
|
|
|
fn node_id(client: &mut support::Client) -> String {
|
|
let reply = client.request(&Request::Status { hash: None });
|
|
match reply.data {
|
|
Some(ResponseData::Status(s)) => s.node_id.expect("daemon has a node id"),
|
|
other => panic!("unexpected status reply: {other:?}"),
|
|
}
|
|
}
|
|
|
|
fn is_complete(socket: &Path, hash: &str) -> bool {
|
|
let mut client = support::Client::connect(socket);
|
|
let reply = client.request(&Request::Status {
|
|
hash: Some(hash.to_string()),
|
|
});
|
|
matches!(
|
|
reply.data,
|
|
Some(ResponseData::HashStatus(s)) if s.complete
|
|
)
|
|
}
|
|
|
|
fn wait_complete(socket: &Path, hash: &str, timeout: Duration) -> bool {
|
|
let deadline = Instant::now() + timeout;
|
|
while Instant::now() < deadline {
|
|
if is_complete(socket, hash) {
|
|
return true;
|
|
}
|
|
std::thread::sleep(Duration::from_millis(100));
|
|
}
|
|
false
|
|
}
|
|
|
|
/// Re-target a ticket at another hash on the same provider: what an
|
|
/// attacker who learned a hash out of band would hand to their daemon.
|
|
fn forge_ticket(real_ticket: &str, target_hash: &str) -> String {
|
|
let ticket = iroh_blobs::ticket::BlobTicket::from_str(real_ticket).unwrap();
|
|
let hash = iroh_blobs::Hash::from_str(target_hash).unwrap();
|
|
iroh_blobs::ticket::BlobTicket::new(ticket.addr().clone(), hash, iroh_blobs::BlobFormat::Raw)
|
|
.to_string()
|
|
}
|
|
|
|
#[test]
|
|
fn trust_gates_serving_per_hash() {
|
|
let provider = support::spawn_daemon();
|
|
let fetcher_dir = tempfile::tempdir().unwrap();
|
|
let fetcher = support::spawn_daemon_at(fetcher_dir.path());
|
|
let mut provider_client = support::Client::connect(&provider.socket);
|
|
let mut fetcher_client = support::Client::connect(&fetcher.socket);
|
|
let dir = tempfile::tempdir().unwrap();
|
|
|
|
// `shared` is deliberately published via ticket; `private` is pinned
|
|
// but never shared.
|
|
std::fs::write(dir.path().join("shared.bin"), vec![1u8; 300_000]).unwrap();
|
|
std::fs::write(dir.path().join("private.bin"), vec![2u8; 300_000]).unwrap();
|
|
let shared = add_file(&mut provider_client, &dir.path().join("shared.bin"));
|
|
let private = add_file(&mut provider_client, &dir.path().join("private.bin"));
|
|
pin(&mut provider_client, &shared, false);
|
|
pin(&mut provider_client, &private, false);
|
|
|
|
let ticket = export_ticket(&mut provider_client, &shared);
|
|
|
|
// The exported content transfers to the untrusted fetcher.
|
|
import_ticket(&mut fetcher_client, &ticket);
|
|
assert!(
|
|
wait_complete(&fetcher.socket, &shared, Duration::from_secs(60)),
|
|
"exported content must be fetchable without trust"
|
|
);
|
|
|
|
// A forged ticket for the never-shared hash must not: the provider
|
|
// answers "not found" to untrusted peers.
|
|
let forged = forge_ticket(&ticket, &private);
|
|
import_ticket(&mut fetcher_client, &forged);
|
|
std::thread::sleep(Duration::from_secs(5));
|
|
assert!(
|
|
!is_complete(&fetcher.socket, &private),
|
|
"unshared content leaked to an untrusted peer"
|
|
);
|
|
|
|
// Once the provider trusts the fetcher, the same request succeeds.
|
|
// Trust is evaluated when a connection is accepted, so restart the
|
|
// fetcher (same store, same identity) to shed the connection that
|
|
// was established while untrusted.
|
|
let fetcher_id = node_id(&mut fetcher_client);
|
|
let reply = provider_client.request(&Request::PeerTrust {
|
|
node_id: fetcher_id,
|
|
});
|
|
assert!(reply.ok, "peer trust failed: {:?}", reply.error);
|
|
drop(fetcher);
|
|
let fetcher = support::spawn_daemon_at(fetcher_dir.path());
|
|
let mut fetcher_client = support::Client::connect(&fetcher.socket);
|
|
import_ticket(&mut fetcher_client, &forged);
|
|
assert!(
|
|
wait_complete(&fetcher.socket, &private, Duration::from_secs(60)),
|
|
"trusted peer should be served everything"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn upload_rate_limit_is_enforced() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
// 256 KiB/s upload cap on the provider.
|
|
let provider = support::spawn_daemon_with_env(dir.path(), &[("VARDE_MAX_UPLOAD", "262144")]);
|
|
let fetcher = support::spawn_daemon();
|
|
let mut provider_client = support::Client::connect(&provider.socket);
|
|
let mut fetcher_client = support::Client::connect(&fetcher.socket);
|
|
|
|
let payload = vec![7u8; 1_000_000];
|
|
let src = dir.path().join("capped.bin");
|
|
std::fs::write(&src, &payload).unwrap();
|
|
let hash = add_file(&mut provider_client, &src);
|
|
pin(&mut provider_client, &hash, false);
|
|
let ticket = export_ticket(&mut provider_client, &hash);
|
|
|
|
let start = Instant::now();
|
|
import_ticket(&mut fetcher_client, &ticket);
|
|
assert!(
|
|
wait_complete(&fetcher.socket, &hash, Duration::from_secs(120)),
|
|
"capped fetch never completed"
|
|
);
|
|
let elapsed = start.elapsed();
|
|
// 1 MB at 256 KiB/s is ~4 s minus one 256 KiB burst: ~3 s. Anything
|
|
// under 1.5 s means the limiter did nothing (uncapped localhost
|
|
// moves this in well under 200 ms).
|
|
assert!(
|
|
elapsed >= Duration::from_millis(1500),
|
|
"1 MB fetch finished in {elapsed:?} despite 256 KiB/s upload cap"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn subscribe_streams_progress_and_completion() {
|
|
let provider = support::spawn_daemon();
|
|
let fetcher = support::spawn_daemon();
|
|
let mut provider_client = support::Client::connect(&provider.socket);
|
|
let mut fetcher_client = support::Client::connect(&fetcher.socket);
|
|
let dir = tempfile::tempdir().unwrap();
|
|
|
|
let src = dir.path().join("watched.bin");
|
|
std::fs::write(&src, vec![5u8; 2_000_000]).unwrap();
|
|
let hash = add_file(&mut provider_client, &src);
|
|
pin(&mut provider_client, &hash, false);
|
|
let ticket = export_ticket(&mut provider_client, &hash);
|
|
|
|
// Subscribe on the fetcher before starting the transfer.
|
|
let mut events = support::Client::connect(&fetcher.socket);
|
|
let ack = events.request(&Request::Subscribe {});
|
|
assert!(ack.ok);
|
|
events.set_read_timeout(Duration::from_secs(60));
|
|
|
|
import_ticket(&mut fetcher_client, &ticket);
|
|
|
|
let mut saw_progress = false;
|
|
let mut saw_complete = false;
|
|
while let Some(line) = events.read_event_line() {
|
|
let event: serde_json::Value = serde_json::from_str(&line).expect("event is JSON");
|
|
match event["event"].as_str() {
|
|
Some("transfer_progress") => {
|
|
assert_eq!(event["hash"].as_str(), Some(hash.as_str()));
|
|
assert_eq!(event["direction"].as_str(), Some("download"));
|
|
saw_progress = true;
|
|
}
|
|
Some("pin_complete") => {
|
|
assert_eq!(event["hash"].as_str(), Some(hash.as_str()));
|
|
saw_complete = true;
|
|
break;
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
assert!(saw_progress, "no transfer_progress event seen");
|
|
assert!(saw_complete, "no pin_complete event seen");
|
|
}
|
|
|
|
/// Two trusted daemons with LAN discovery on: a pin that exists on one
|
|
/// side auto-syncs to the other with no ticket exchange. Requires
|
|
/// working multicast; skips (with a note) where mdns can't see the
|
|
/// sibling process.
|
|
#[test]
|
|
fn overlapping_pins_auto_sync_via_lan_discovery() {
|
|
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);
|
|
|
|
// Mutual trust: A serves B, B auto-fetches from A.
|
|
let id_a = node_id(&mut client_a);
|
|
let id_b = node_id(&mut client_b);
|
|
assert!(client_a.request(&Request::PeerTrust { node_id: id_b }).ok);
|
|
assert!(client_b.request(&Request::PeerTrust { node_id: id_a }).ok);
|
|
|
|
// Content pinned on A only.
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let src = dir.path().join("synced.bin");
|
|
let payload = vec![9u8; 500_000];
|
|
std::fs::write(&src, &payload).unwrap();
|
|
let hash = add_file(&mut client_a, &src);
|
|
pin(&mut client_a, &hash, false);
|
|
|
|
// B pins the same hash without having the bytes.
|
|
pin(&mut client_b, &hash, false);
|
|
|
|
// Wait for B to see A via discovery at all.
|
|
let deadline = Instant::now() + Duration::from_secs(30);
|
|
let mut discovered = false;
|
|
while Instant::now() < deadline {
|
|
let reply = client_b.request(&Request::PeerList {});
|
|
if let Some(ResponseData::Peers { peers }) = reply.data {
|
|
if peers.iter().any(|p| p.connected) {
|
|
discovered = true;
|
|
break;
|
|
}
|
|
}
|
|
std::thread::sleep(Duration::from_millis(250));
|
|
}
|
|
if !discovered {
|
|
eprintln!("SKIP: mdns discovery saw no peers (multicast unavailable in this environment)");
|
|
return;
|
|
}
|
|
|
|
assert!(
|
|
wait_complete(&b.socket, &hash, Duration::from_secs(60)),
|
|
"pinned content did not auto-sync from trusted LAN peer"
|
|
);
|
|
let dest = dir.path().join("synced-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);
|
|
}
|
|
|
|
/// 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));
|
|
}
|