syncweb_core/folder/
catalog.rs

1use std::{collections::HashSet, str::FromStr, time::Duration};
2
3use ed25519_dalek::SigningKey;
4use iroh::{Endpoint, PublicKey, address_lookup::memory::MemoryLookup};
5use iroh_blobs::{Hash, ticket::BlobTicket};
6use iroh_gossip::{
7    TopicId,
8    api::{Event, GossipSender, GossipTopic},
9};
10use n0_future::StreamExt;
11use semver::Version;
12use serde::{Deserialize, Serialize};
13use uuid::Uuid;
14
15use crate::{
16    editorial::Channel,
17    error::{Result, SyncwebError},
18    node::gossip_service::GossipService,
19    node::signed_message::SignedMessage,
20};
21
22const CATALOG_TOPIC_SEED: &[u8] = b"syncweb/public-package-catalog/v1";
23
24/// The public gossip topic used for package announcements.
25#[must_use]
26pub fn catalog_topic() -> TopicId {
27    TopicId::from_bytes(*blake3::hash(CATALOG_TOPIC_SEED).as_bytes())
28}
29
30/// Derive a gossip topic ID from an editorial channel name.
31///
32/// Each channel forms a fully independent gossip swarm so that consumers
33/// can subscribe only to the curated views they trust.
34#[must_use]
35pub fn channel_topic_id(name: impl AsRef<str>) -> TopicId {
36    let seed = format!("syncweb/editorial/{}/v1", name.as_ref());
37    TopicId::from_bytes(*blake3::hash(seed.as_bytes()).as_bytes())
38}
39
40/// Shorthand for [`channel_topic_id`] that accepts a [`Channel`].
41#[must_use]
42pub fn channel_topic(channel: &Channel) -> TopicId {
43    channel_topic_id(&channel.name)
44}
45
46/// A gossip announcement for a publicly fetchable collection manifest.
47#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
48#[non_exhaustive]
49pub struct PackageAnnouncement {
50    pub collection_id: Uuid,
51    pub name: String,
52    pub version: String,
53    pub sequence: u64,
54    pub manifest: Hash,
55    pub manifest_ticket: String,
56    pub publisher: String,
57}
58
59impl PackageAnnouncement {
60    /// # Errors
61    ///
62    /// Returns an error if the ticket is invalid or does not identify the
63    /// announced manifest.
64    pub fn new(
65        collection_id: Uuid,
66        name: impl Into<String>,
67        version: impl Into<String>,
68        sequence: u64,
69        manifest: Hash,
70        manifest_ticket: impl Into<String>,
71        publisher: PublicKey,
72    ) -> Result<Self> {
73        let announcement = Self {
74            collection_id,
75            name: name.into(),
76            version: version.into(),
77            sequence,
78            manifest,
79            manifest_ticket: manifest_ticket.into(),
80            publisher: publisher.to_string(),
81        };
82        announcement.validate()?;
83        Ok(announcement)
84    }
85
86    /// # Errors
87    ///
88    /// Returns an error if the announcement has invalid metadata or a ticket
89    /// that does not match the manifest hash.
90    pub fn validate(&self) -> Result<()> {
91        if self.name.trim().is_empty() {
92            return Err(SyncwebError::InvalidConfig(
93                "package announcement name cannot be empty".to_owned(),
94            ));
95        }
96        if self.sequence == 0 {
97            return Err(SyncwebError::InvalidConfig(
98                "package announcement sequence must be greater than zero".to_owned(),
99            ));
100        }
101        Version::parse(&self.version)
102            .map_err(|error| SyncwebError::InvalidConfig(format!("invalid package version: {error}")))?;
103        let ticket = BlobTicket::from_str(&self.manifest_ticket)
104            .map_err(|error| SyncwebError::InvalidTicket(error.to_string()))?;
105        if ticket.hash() != self.manifest {
106            return Err(SyncwebError::InvalidTicket(
107                "package manifest ticket does not match its manifest hash".to_owned(),
108            ));
109        }
110        self.publisher
111            .parse::<PublicKey>()
112            .map_err(|error| SyncwebError::InvalidIdentity(error.to_string()))?;
113        Ok(())
114    }
115
116    /// # Errors
117    ///
118    /// Returns an error if the announcement is invalid or cannot be encoded.
119    pub fn to_bytes(&self) -> Result<Vec<u8>> {
120        self.validate()?;
121        serde_json::to_vec(self)
122            .map_err(|error| SyncwebError::operation("failed to serialize package announcement", error))
123    }
124
125    /// # Errors
126    ///
127    /// Returns an error if the announcement cannot be decoded or validated.
128    pub fn from_bytes(bytes: impl AsRef<[u8]>) -> Result<Self> {
129        let announcement: Self = serde_json::from_slice(bytes.as_ref())
130            .map_err(|error| SyncwebError::operation("failed to deserialize package announcement", error))?;
131        announcement.validate()?;
132        Ok(announcement)
133    }
134
135    /// Parse the manifest blob ticket carried by this announcement.
136    ///
137    /// # Errors
138    ///
139    /// Returns an error if the ticket is invalid.
140    pub fn ticket(&self) -> Result<BlobTicket> {
141        BlobTicket::from_str(&self.manifest_ticket).map_err(|error| SyncwebError::InvalidTicket(error.to_string()))
142    }
143
144    fn matches(&self, query: Option<&str>) -> bool {
145        query.is_none_or(|needle| {
146            self.name.contains(needle)
147                || self.collection_id.to_string().contains(needle)
148                || self.version.contains(needle)
149        })
150    }
151}
152
153/// Gossip-backed discovery for publicly published package manifests.
154#[derive(Clone)]
155pub struct PackageCatalog {
156    gossip: GossipService,
157    signing_key: SigningKey,
158    /// Optional publisher allowlist. If non-empty, only announcements
159    /// from publishers in this set are returned by `search()`.
160    allowed_publishers: HashSet<PublicKey>,
161}
162
163impl PackageCatalog {
164    #[must_use]
165    pub fn new(gossip: &GossipService, endpoint: &Endpoint) -> Self {
166        Self {
167            gossip: gossip.clone(),
168            signing_key: SigningKey::from_bytes(&endpoint.secret_key().to_bytes()),
169            allowed_publishers: HashSet::new(),
170        }
171    }
172
173    /// Create a catalog with an explicit publisher allowlist.
174    #[must_use]
175    pub fn with_publishers(gossip: &GossipService, endpoint: &Endpoint, allowed_publishers: Vec<PublicKey>) -> Self {
176        Self {
177            gossip: gossip.clone(),
178            signing_key: SigningKey::from_bytes(&endpoint.secret_key().to_bytes()),
179            allowed_publishers: allowed_publishers.into_iter().collect(),
180        }
181    }
182
183    /// Subscribe to the public package catalog.
184    ///
185    /// The caller must retain the returned topic while receiving
186    /// announcements.
187    ///
188    /// # Errors
189    ///
190    /// Returns an error if the gossip subscription cannot be created.
191    pub async fn subscribe(&self, bootstrap: Vec<PublicKey>) -> Result<GossipTopic> {
192        self.gossip.subscribe(catalog_topic(), bootstrap).await
193    }
194
195    /// Subscribe to the public package catalog and wait for a bootstrap
196    /// connection.
197    ///
198    /// # Errors
199    ///
200    /// Returns an error if the gossip subscription cannot join its bootstrap
201    /// peers.
202    pub async fn subscribe_and_join(&self, bootstrap: Vec<PublicKey>) -> Result<GossipTopic> {
203        self.gossip.subscribe_and_join(catalog_topic(), bootstrap).await
204    }
205
206    /// Broadcast an announcement to the public catalog.
207    ///
208    /// The announcement is Ed25519-signed so receivers can cryptographically
209    /// verify its origin.
210    ///
211    /// # Errors
212    ///
213    /// Returns an error if the announcement is invalid or cannot be sent.
214    pub async fn announce(&self, sender: &GossipSender, announcement: &PackageAnnouncement) -> Result<()> {
215        let content = announcement.to_bytes()?;
216        let wire = SignedMessage::sign(&self.signing_key, &content);
217        self.gossip.publish(sender, wire).await
218    }
219
220    /// Subscribe to a specific editorial channel.
221    ///
222    /// # Errors
223    ///
224    /// Returns an error if the gossip subscription cannot be created.
225    pub async fn subscribe_channel(&self, channel: &Channel, bootstrap: Vec<PublicKey>) -> Result<GossipTopic> {
226        self.gossip.subscribe(channel_topic(channel), bootstrap).await
227    }
228
229    /// Subscribe to a channel and block until at least one bootstrap peer is
230    /// connected.
231    ///
232    /// # Errors
233    ///
234    /// Returns an error if the gossip subscription cannot join its bootstrap
235    /// peers.
236    pub async fn subscribe_channel_and_join(
237        &self,
238        channel: &Channel,
239        bootstrap: Vec<PublicKey>,
240    ) -> Result<GossipTopic> {
241        self.gossip.subscribe_and_join(channel_topic(channel), bootstrap).await
242    }
243
244    /// Collect matching announcements until the timeout expires.
245    ///
246    /// Each announcement is verified: the Ed25519 signature is checked and the
247    /// claimed `publisher` field must match the signing key.
248    /// A timeout is a normal end condition because gossip has no finite
249    /// response boundary.
250    ///
251    /// When `allowed_publishers` is non-empty, only announcements from
252    /// publishers in that set are returned.
253    ///
254    /// # Errors
255    ///
256    /// Returns an error if a received announcement fails signature verification.
257    pub async fn search(
258        &self,
259        topic: &mut GossipTopic,
260        query: Option<&str>,
261        timeout: Duration,
262    ) -> Result<Vec<PackageAnnouncement>> {
263        self.search_inner(topic, query, timeout, &self.allowed_publishers).await
264    }
265
266    /// Collect matching announcements on a channel topic, applying the
267    /// intersection of the catalog's global allowlist and the channel's
268    /// own allowlist.
269    ///
270    /// If both the catalog and the channel have empty allowlists, all
271    /// publishers are accepted.
272    ///
273    /// # Errors
274    ///
275    /// Returns an error if a received announcement fails signature verification.
276    pub async fn search_channel(
277        &self,
278        topic: &mut GossipTopic,
279        channel: &Channel,
280        query: Option<&str>,
281        timeout: Duration,
282    ) -> Result<Vec<PackageAnnouncement>> {
283        let effective = self.effective_allowlist(channel);
284        self.search_inner(topic, query, timeout, &effective).await
285    }
286
287    fn effective_allowlist(&self, channel: &Channel) -> HashSet<PublicKey> {
288        if self.allowed_publishers.is_empty() {
289            return parse_publishers(&channel.allowed_publishers);
290        }
291        if channel.allowed_publishers.is_empty() {
292            return self.allowed_publishers.clone();
293        }
294        let channel_set = parse_publishers(&channel.allowed_publishers);
295        self.allowed_publishers.intersection(&channel_set).copied().collect()
296    }
297
298    async fn search_inner(
299        &self,
300        topic: &mut GossipTopic,
301        query: Option<&str>,
302        timeout: Duration,
303        allowed: &HashSet<PublicKey>,
304    ) -> Result<Vec<PackageAnnouncement>> {
305        let mut announcements = Vec::new();
306        let receive = async {
307            while let Some(event) = topic.next().await {
308                if let Event::Received(message) =
309                    event.map_err(|error| SyncwebError::operation("package catalog event failed", error))?
310                {
311                    // Verify the Ed25519 signature on the gossip message
312                    let (verifying_key, content) = SignedMessage::verify(&message.content).map_err(|error| {
313                        SyncwebError::operation("package catalog signature verification failed", error)
314                    })?;
315
316                    let announcement = PackageAnnouncement::from_bytes(content)?;
317
318                    // Verify that the claimed publisher matches the signing key
319                    let claimed_publisher: PublicKey = announcement
320                        .publisher
321                        .parse()
322                        .map_err(|error| SyncwebError::InvalidIdentity(format!("invalid publisher key: {error}")))?;
323                    let actual_publisher = PublicKey::from_bytes(&verifying_key.to_bytes()).map_err(|error| {
324                        SyncwebError::InvalidIdentity(format!("derived NodeId from verifying key is invalid: {error}"))
325                    })?;
326                    if claimed_publisher != actual_publisher {
327                        return Err(SyncwebError::operation(
328                            "package catalog publisher mismatch",
329                            format!("claimed {claimed_publisher}, signed by {actual_publisher}"),
330                        ));
331                    }
332
333                    // Filter by publisher allowlist
334                    if !allowed.is_empty() && !allowed.contains(&actual_publisher) {
335                        tracing::debug!(
336                            "skipping announcement from {} (not in publisher allowlist)",
337                            actual_publisher
338                        );
339                        continue;
340                    }
341
342                    if announcement.matches(query)
343                        && !announcements.iter().any(|item: &PackageAnnouncement| {
344                            item.collection_id == announcement.collection_id
345                                && item.version == announcement.version
346                                && item.manifest == announcement.manifest
347                        })
348                    {
349                        announcements.push(announcement);
350                    }
351                }
352            }
353            Ok::<(), SyncwebError>(())
354        };
355        if let Ok(result) = tokio::time::timeout(timeout, receive).await {
356            result?;
357        }
358        Ok(announcements)
359    }
360
361    /// Add the endpoint address for a discovered ticket to a memory lookup.
362    ///
363    /// This helper is useful to callers that construct a node with a shared
364    /// [`MemoryLookup`] and want to fetch an announcement immediately.
365    ///
366    /// # Errors
367    ///
368    /// Returns an error if the announcement's manifest ticket is invalid.
369    pub fn register_ticket_endpoint(lookup: &MemoryLookup, announcement: &PackageAnnouncement) -> Result<()> {
370        let ticket = announcement.ticket()?;
371        lookup.add_endpoint_info(ticket.addr().clone());
372        Ok(())
373    }
374}
375
376fn parse_publishers(keys: &[String]) -> HashSet<PublicKey> {
377    keys.iter()
378        .filter_map(|key| {
379            let stripped = key.strip_prefix("ed25519:").unwrap_or(key);
380            stripped.parse::<PublicKey>().ok()
381        })
382        .collect()
383}
syncweb_core/folder/catalog.rs

syncweb_core/folder/
catalog.rs
