Milestone 6: DiscoveryProvider seam, signed announcements, redoal sketch

The discovery module defines the seam: DiscoveryProvider (subscribe/
announce over 32-byte TopicKeys) and a signed Announcement carrying
root hash, ed25519 author, metadata and provider addresses. Signatures
cover a deterministic postcard encoding including the topic (no cross-
channel replay) and the provider identities (addresses stay refreshable
hints). LanDiscovery conforms to the trait — mdns sightings become
locally-authored announcements, one per pinned root — and the daemon's
auto-sync now runs entirely through it: verify, index unconditionally,
fetch only for trusted authors from allowlisted providers on already-
pinned incomplete roots. The milestone-4 real-mdns sync test passes
unchanged through the new path. Verification unit tests cover round
trip, tampered root, wrong topic, forged author, mismatched signing
key, and serde survival. docs/redoal-integration.md sketches the
gesture-topic gossip provider against this contract.

Also: fix a flaky hang in the socket-activation test (dup2(3,3) leaves
CLOEXEC set when the listener already sits on fd 3; parent's listener
copy masked daemon death), and give the test client a read-timeout hang
guard. Add a top-level README.

Dependencies: ed25519-dalek (Signature type; same implementation iroh
keys use), postcard (deterministic signed encoding, iroh's canonical
compact codec).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

varde-daemon/Cargo.toml

@@ -24,6 +24,10 @@ futures-lite = "2"
 bytes = "1"
 # Feature-gated D-Bus client for NetworkManager metered status.
 zbus = { version = "5", optional = true, default-features = false, features = ["tokio"] }
+# Announcement signatures: same ed25519 implementation iroh keys use,
+# postcard is the deterministic encoding signed over.
+ed25519-dalek = "2"
+postcard = { version = "1", features = ["use-std"] }
 iroh-io = { workspace = true }
 reflink-copy = { workspace = true }
 anyhow = { workspace = true }

varde-daemon/src/daemon.rs

@@ -15,6 +15,7 @@ use varde_proto::{
 };
 
 use crate::config::Config;
+use crate::discovery::{DiscoveryProvider, LanDiscovery, SignedAnnouncement, TopicKey, LAN_TOPIC};
 use crate::meta::{Meta, StoredFormat};
 use crate::store::{BlobStore, OpError};
 use crate::transfer::Transfer;
@@ -39,6 +40,9 @@ pub struct Daemon {
     open_set: Arc<RwLock<BTreeSet<Hash>>>,
     /// LAN peers seen via discovery: node id -> last sighting.
     lan_peers: Mutex<BTreeMap<String, Instant>>,
+    /// Everything announced on subscribed discovery topics, newest per
+    /// root. Indexing is unconditional; *acting* requires trust.
+    announcements: Mutex<BTreeMap<Hash, SignedAnnouncement>>,
 }
 
 impl Daemon {
@@ -78,6 +82,7 @@ impl Daemon {
             events,
             open_set,
             lan_peers: Mutex::new(BTreeMap::new()),
+            announcements: Mutex::new(BTreeMap::new()),
         });
         daemon.recompute_open_set().await;
 
@@ -99,14 +104,40 @@ impl Daemon {
             }
         });
 
-        // Discovery loop: track LAN peer sightings, emit join events for
-        // trusted peers and sync any incomplete pins from them.
+        // 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.
         if let Some(mut discovered) = daemon.transfer.take_discovery() {
+            let (lan_tx, lan_rx) = tokio::sync::mpsc::channel(64);
+            let pins_snapshot: Arc<dyn Fn() -> Vec<Hash> + Send + Sync> = {
+                let meta = daemon.meta.clone();
+                Arc::new(move || {
+                    meta.pins()
+                        .into_iter()
+                        .filter_map(|(hex, _)| Hash::from_str(&hex).ok())
+                        .collect()
+                })
+            };
+            let lan = LanDiscovery::new(daemon.transfer.secret_key(), lan_rx, pins_snapshot);
+            let mut announcements = lan.subscribe(LAN_TOPIC);
+
             let weak = Arc::downgrade(&daemon);
             tokio::spawn(async move {
                 while let Some(node_id) = discovered.recv().await {
                     let Some(daemon) = weak.upgrade() else { break };
-                    daemon.on_peer_seen(node_id).await;
+                    daemon.on_peer_seen(node_id);
+                    if lan_tx.send(node_id).await.is_err() {
+                        break;
+                    }
+                }
+            });
+
+            let weak = Arc::downgrade(&daemon);
+            tokio::spawn(async move {
+                use n0_future::StreamExt;
+                while let Some(signed) = announcements.next().await {
+                    let Some(daemon) = weak.upgrade() else { break };
+                    daemon.on_announcement(LAN_TOPIC, signed).await;
                 }
             });
         }
@@ -161,8 +192,9 @@ impl Daemon {
         *self.open_set.write().expect("open set lock") = open;
     }
 
-    /// Handle a discovery sighting of `node_id`.
-    async fn on_peer_seen(&self, node_id: iroh::NodeId) {
+    /// Track presence for a discovery sighting of `node_id`. Content
+    /// sync happens via the announcement path, not here.
+    fn on_peer_seen(&self, node_id: iroh::NodeId) {
         let id = node_id.to_string();
         let rejoined = {
             let mut peers = self.lan_peers.lock().expect("lan peers lock");
@@ -174,14 +206,64 @@ impl Daemon {
             peers.insert(id.clone(), now);
             fresh
         };
-        if !self.meta.is_trusted(&id) {
-            return;
-        }
-        if rejoined {
+        if rejoined && self.meta.is_trusted(&id) {
             info!(peer = %id, "trusted peer present on LAN");
             let _ = self.events.send(Event::PeerJoined { node_id: id });
         }
-        self.sync_pins_from(node_id).await;
+    }
+
+    /// Handle one announcement from a discovery provider: index it
+    /// unconditionally, act on it only within the trust policy.
+    async fn on_announcement(&self, topic: TopicKey, signed: SignedAnnouncement) {
+        if let Err(e) = signed.verify(topic) {
+            warn!(error = %e, "dropping announcement with bad signature");
+            return;
+        }
+        let ann = &signed.announcement;
+        let root = ann.root;
+        self.announcements
+            .lock()
+            .expect("announcements lock")
+            .insert(root, signed.clone());
+        let _ = self.events.send(Event::Announcement {
+            root: root.to_hex().to_string(),
+            author: ann.author.to_string(),
+            meta: ann.meta.clone(),
+        });
+
+        // Consent tier 1: only act on announcements from trusted authors
+        // (our own key vouches for locally observed LAN sightings).
+        let author_trusted = ann.author == self.transfer.secret_key().public()
+            || self.meta.is_trusted(&ann.author.to_string());
+        if !author_trusted {
+            return;
+        }
+        // Consent tier 2: only content we already pinned, and only from
+        // providers on the peer allowlist.
+        let hex = root.to_hex().to_string();
+        if !self.meta.is_pinned(&hex) {
+            return;
+        }
+        if let Ok((_, _, true)) = self.store.presence(root).await {
+            return; // already complete
+        }
+        let providers: Vec<iroh::NodeAddr> = ann
+            .providers
+            .iter()
+            .filter(|p| self.meta.is_trusted(&p.node_id.to_string()))
+            .cloned()
+            .collect();
+        if providers.is_empty() {
+            return;
+        }
+        debug!(root = %hex, providers = providers.len(), "fetching announced content");
+        self.transfer.spawn_fetch(
+            HashAndFormat {
+                hash: root,
+                format: blob_format(self.meta.format_of(&hex)),
+            },
+            providers,
+        );
     }
 
     /// Queue fetches for every incomplete pin from a trusted LAN peer.

varde-daemon/src/discovery.rs

@@ -0,0 +1,261 @@
+//! The content-discovery seam.
+//!
+//! A [`DiscoveryProvider`] is a source of signed content announcements
+//! the daemon may subscribe to. v1 ships [`LanDiscovery`] (backed by
+//! mDNS peer sightings) and the implicit "provider" that is ticket
+//! import. A future provider maps gesture-derived gossip TopicIds
+//! (redoal) to announcement streams over iroh-gossip — see
+//! `docs/redoal-integration.md`.
+//!
+//! Consent tiering: the daemon indexes every announcement on subscribed
+//! topics, but only auto-fetches content announced by trusted author
+//! keys, and only from providers on the peer allowlist. Verification of
+//! the fetched bytes is BLAKE3 as always — trust gates participation,
+//! not integrity.
+
+use std::collections::BTreeMap;
+use std::pin::Pin;
+use std::sync::Arc;
+
+use anyhow::Result;
+use iroh::{NodeAddr, NodeId, PublicKey, SecretKey};
+use iroh_blobs::Hash;
+use n0_future::Stream;
+use serde::{Deserialize, Serialize};
+use tokio::sync::mpsc;
+
+/// A stream of announcements, boxed for trait-object use.
+pub type BoxStream<T> = Pin<Box<dyn Stream<Item = T> + Send + 'static>>;
+
+/// Identifies an announcement channel. For the LAN provider there is a
+/// single well-known topic; gossip providers derive topics from shared
+/// secrets (e.g. redoal gesture keys).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+pub struct TopicKey(pub [u8; 32]);
+
+/// The well-known topic for local-network presence announcements.
+pub const LAN_TOPIC: TopicKey = TopicKey(*b"varde/v1/lan-presence\0\0\0\0\0\0\0\0\0\0\0");
+
+/// A content announcement: "content `root` is available from
+/// `providers`", vouched for by `author`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Announcement {
+    /// The announced content (HashSeq root or single blob hash).
+    pub root: Hash,
+    /// ed25519 key of whoever signs this announcement.
+    pub author: PublicKey,
+    /// Free-form metadata (name hints, sizes, provenance...).
+    pub meta: BTreeMap<String, String>,
+    /// Peers believed to hold the content.
+    pub providers: Vec<NodeAddr>,
+}
+
+/// An [`Announcement`] plus the author's signature over it and its topic.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct SignedAnnouncement {
+    pub announcement: Announcement,
+    /// 64-byte ed25519 signature (serde can't derive for [u8; 64]).
+    signature_bytes: Vec<u8>,
+}
+
+impl SignedAnnouncement {
+    /// Sign `announcement` for `topic`. The signature covers the topic
+    /// too, so an announcement can't be replayed onto another channel.
+    /// `secret` must be the key matching `announcement.author`.
+    pub fn sign(secret: &SecretKey, topic: TopicKey, announcement: Announcement) -> Result<Self> {
+        anyhow::ensure!(
+            secret.public() == announcement.author,
+            "signing key does not match announcement author"
+        );
+        let payload = signing_payload(topic, &announcement)?;
+        let signature = secret.sign(&payload);
+        Ok(SignedAnnouncement {
+            announcement,
+            signature_bytes: signature.to_bytes().to_vec(),
+        })
+    }
+
+    /// Verify the signature against the embedded author key and `topic`.
+    pub fn verify(&self, topic: TopicKey) -> Result<()> {
+        let payload = signing_payload(topic, &self.announcement)?;
+        let bytes: &[u8; 64] = self
+            .signature_bytes
+            .as_slice()
+            .try_into()
+            .map_err(|_| anyhow::anyhow!("signature must be 64 bytes"))?;
+        let signature = ed25519_dalek::Signature::from_bytes(bytes);
+        self.announcement
+            .author
+            .verify(&payload, &signature)
+            .map_err(|e| anyhow::anyhow!("announcement signature invalid: {e}"))
+    }
+}
+
+/// Deterministic bytes covered by the signature: postcard of
+/// (topic, root, author, meta, providers' node ids). Provider *addresses*
+/// are excluded — they are hints that relays may legitimately refresh —
+/// but the provider identities are covered.
+fn signing_payload(topic: TopicKey, announcement: &Announcement) -> Result<Vec<u8>> {
+    let provider_ids: Vec<NodeId> = announcement.providers.iter().map(|p| p.node_id).collect();
+    let payload = (
+        topic,
+        announcement.root,
+        announcement.author,
+        &announcement.meta,
+        provider_ids,
+    );
+    Ok(postcard::to_stdvec(&payload)?)
+}
+
+/// A source of content announcements the daemon may subscribe to.
+pub trait DiscoveryProvider: Send + Sync {
+    /// Announcements arriving on `topic`. Invalid signatures are the
+    /// subscriber's problem to reject (call [`SignedAnnouncement::verify`]).
+    fn subscribe(&self, topic: TopicKey) -> BoxStream<SignedAnnouncement>;
+
+    /// Publish an announcement on `topic`.
+    fn announce(&self, topic: TopicKey, ann: SignedAnnouncement) -> Result<()>;
+}
+
+/// The v1 LAN provider: turns mDNS peer sightings into announcements.
+///
+/// When a peer appears on the LAN, we synthesize one announcement per
+/// locally pinned root, authored (and signed) by *this* daemon: "I saw
+/// peer N; N may hold root R". The daemon's own trust policy then
+/// decides whether to dial N. Outbound `announce` is a no-op: presence
+/// on the LAN is already advertised by mDNS itself, and varde never
+/// broadcasts its content list.
+pub struct LanDiscovery {
+    secret: SecretKey,
+    sightings: std::sync::Mutex<Option<mpsc::Receiver<NodeId>>>,
+    /// Snapshot of locally pinned roots, queried per sighting.
+    pins: Arc<dyn Fn() -> Vec<Hash> + Send + Sync>,
+}
+
+impl LanDiscovery {
+    pub fn new(
+        secret: SecretKey,
+        sightings: mpsc::Receiver<NodeId>,
+        pins: Arc<dyn Fn() -> Vec<Hash> + Send + Sync>,
+    ) -> LanDiscovery {
+        LanDiscovery {
+            secret,
+            sightings: std::sync::Mutex::new(Some(sightings)),
+            pins,
+        }
+    }
+}
+
+impl DiscoveryProvider for LanDiscovery {
+    fn subscribe(&self, topic: TopicKey) -> BoxStream<SignedAnnouncement> {
+        if topic != LAN_TOPIC {
+            // The LAN has exactly one channel; other topics are empty.
+            return Box::pin(n0_future::stream::empty());
+        }
+        let Some(mut sightings) = self.sightings.lock().expect("sightings lock").take() else {
+            return Box::pin(n0_future::stream::empty());
+        };
+        let secret = self.secret.clone();
+        let pins = self.pins.clone();
+        let (tx, rx) = mpsc::channel(64);
+        tokio::spawn(async move {
+            while let Some(node_id) = sightings.recv().await {
+                for root in pins() {
+                    let announcement = Announcement {
+                        root,
+                        author: secret.public(),
+                        meta: BTreeMap::from([("source".to_string(), "mdns".to_string())]),
+                        providers: vec![NodeAddr::new(node_id)],
+                    };
+                    match SignedAnnouncement::sign(&secret, LAN_TOPIC, announcement) {
+                        Ok(signed) => {
+                            if tx.send(signed).await.is_err() {
+                                return;
+                            }
+                        }
+                        Err(e) => tracing::warn!(error = %e, "signing lan announcement"),
+                    }
+                }
+            }
+        });
+        Box::pin(tokio_stream_from(rx))
+    }
+
+    fn announce(&self, _topic: TopicKey, _ann: SignedAnnouncement) -> Result<()> {
+        // Presence is already advertised by mDNS; varde does not
+        // broadcast content lists on the LAN.
+        Ok(())
+    }
+}
+
+fn tokio_stream_from<T: Send + 'static>(mut rx: mpsc::Receiver<T>) -> impl Stream<Item = T> {
+    n0_future::stream::poll_fn(move |cx| rx.poll_recv(cx))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn keypair() -> SecretKey {
+        SecretKey::generate(rand::rngs::OsRng)
+    }
+
+    fn sample(author: PublicKey) -> Announcement {
+        Announcement {
+            root: Hash::new(b"content"),
+            author,
+            meta: BTreeMap::from([("name".to_string(), "demo".to_string())]),
+            providers: vec![NodeAddr::new(keypair().public())],
+        }
+    }
+
+    #[test]
+    fn sign_verify_round_trip() {
+        let secret = keypair();
+        let signed = SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
+        signed.verify(LAN_TOPIC).expect("valid signature verifies");
+    }
+
+    #[test]
+    fn tampered_root_fails() {
+        let secret = keypair();
+        let mut signed =
+            SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
+        signed.announcement.root = Hash::new(b"evil");
+        assert!(signed.verify(LAN_TOPIC).is_err());
+    }
+
+    #[test]
+    fn wrong_topic_fails() {
+        let secret = keypair();
+        let signed = SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
+        assert!(signed.verify(TopicKey([9u8; 32])).is_err());
+    }
+
+    #[test]
+    fn forged_author_fails() {
+        let secret = keypair();
+        let mut signed =
+            SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
+        // Claim a different author without their key.
+        signed.announcement.author = keypair().public();
+        assert!(signed.verify(LAN_TOPIC).is_err());
+    }
+
+    #[test]
+    fn signing_with_mismatched_key_is_rejected() {
+        let secret = keypair();
+        let other = keypair();
+        assert!(SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(other.public())).is_err());
+    }
+
+    #[test]
+    fn survives_serde_round_trip() {
+        let secret = keypair();
+        let signed = SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
+        let json = serde_json::to_string(&signed).unwrap();
+        let back: SignedAnnouncement = serde_json::from_str(&json).unwrap();
+        back.verify(LAN_TOPIC).expect("signature survives serde");
+        assert_eq!(back, signed);
+    }
+}

varde-daemon/src/lib.rs

@@ -5,6 +5,7 @@
 
 pub mod config;
 pub mod daemon;
+pub mod discovery;
 pub mod dscp;
 pub mod meta;
 pub mod metered;

varde-daemon/src/transfer.rs

@@ -190,6 +190,11 @@ impl Transfer {
         self.endpoint.node_id().to_string()
     }
 
+    /// The endpoint's identity key (also signs our announcements).
+    pub fn secret_key(&self) -> SecretKey {
+        self.endpoint.secret_key().clone()
+    }
+
     /// Take the stream of locally discovered node ids. Yields each
     /// sighting (mdns re-announces periodically); the daemon dedupes.
     /// Callable once; returns None afterwards or with discovery off.

varde-daemon/tests/citizenship.rs

@@ -13,8 +13,11 @@ use varde_proto::{PinPolicy, Request, ResponseData};
 
 extern "C" {
     fn dup2(oldfd: i32, newfd: i32) -> i32;
+    fn fcntl(fd: i32, cmd: i32, arg: i32) -> i32;
 }
 
+const F_SETFD: i32 = 2;
+
 /// Spawn the daemon with an activation socket on fd 3, the LISTEN_FDS
 /// protocol systemd uses.
 #[test]
@@ -38,17 +41,28 @@ fn systemd_socket_activation() {
         .env("VARDE_DISCOVERY", "false")
         .env("LISTEN_FDS", "1")
         .env_remove("LISTEN_PID");
-    // SAFETY: dup2 is async-signal-safe; we place the listener on fd 3
-    // in the child, which also clears CLOEXEC on the duplicate.
+    // SAFETY: dup2/fcntl are async-signal-safe; we place the listener on
+    // fd 3 in the child. dup2 clears CLOEXEC on the duplicate, but when
+    // the listener already *is* fd 3, dup2(3,3) is a no-op that leaves
+    // CLOEXEC set and exec would close the socket — clear it explicitly.
     unsafe {
         command.pre_exec(move || {
-            if dup2(fd, 3) < 0 {
+            let rc = if fd == 3 {
+                fcntl(3, F_SETFD, 0)
+            } else {
+                dup2(fd, 3)
+            };
+            if rc < 0 {
                 return Err(std::io::Error::last_os_error());
             }
             Ok(())
         });
     }
     let mut child = command.spawn().expect("spawning daemon");
+    // Drop our copy of the listener: if the daemon dies, connects must
+    // fail (and this test fail cleanly) rather than queue in the backlog
+    // of a socket nobody accepts on.
+    drop(listener);
 
     // The daemon must answer on the activation socket.
     let deadline = Instant::now() + Duration::from_secs(10);
@@ -56,6 +70,7 @@ fn systemd_socket_activation() {
     while Instant::now() < deadline {
         if std::os::unix::net::UnixStream::connect(&socket_path).is_ok() {
             let mut client = support::Client::connect(&socket_path);
+            client.set_read_timeout(Duration::from_secs(5));
             let reply = client.request(&Request::Status { hash: None });
             if reply.ok {
                 answered = true;

varde-daemon/tests/support/mod.rs

@@ -101,6 +101,11 @@ pub struct Client {
 impl Client {
     pub fn connect(socket: &Path) -> Client {
         let stream = UnixStream::connect(socket).expect("connecting to daemon");
+        // Hang guard: a daemon that accepts but never answers should fail
+        // the test, not wedge the whole suite.
+        stream
+            .set_read_timeout(Some(Duration::from_secs(120)))
+            .expect("setting read timeout");
         let reader = BufReader::new(stream.try_clone().expect("cloning stream"));
         Client {
             reader,