syncweb_core/folder/
catalog.rs

1use std::{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}
159
160impl PackageCatalog {
161    #[must_use]
162    pub fn new(gossip: &GossipService, endpoint: &Endpoint) -> Self {
163        Self {
164            gossip: gossip.clone(),
165            signing_key: SigningKey::from_bytes(&endpoint.secret_key().to_bytes()),
166        }
167    }
168
169    /// Subscribe to the public package catalog.
170    ///
171    /// The caller must retain the returned topic while receiving
172    /// announcements.
173    ///
174    /// # Errors
175    ///
176    /// Returns an error if the gossip subscription cannot be created.
177    pub async fn subscribe(&self, bootstrap: Vec<PublicKey>) -> Result<GossipTopic> {
178        self.gossip.subscribe(catalog_topic(), bootstrap).await
179    }
180
181    /// Subscribe to the public package catalog and wait for a bootstrap
182    /// connection.
183    ///
184    /// # Errors
185    ///
186    /// Returns an error if the gossip subscription cannot join its bootstrap
187    /// peers.
188    pub async fn subscribe_and_join(&self, bootstrap: Vec<PublicKey>) -> Result<GossipTopic> {
189        self.gossip.subscribe_and_join(catalog_topic(), bootstrap).await
190    }
191
192    /// Broadcast an announcement to the public catalog.
193    ///
194    /// The announcement is Ed25519-signed so receivers can cryptographically
195    /// verify its origin.
196    ///
197    /// # Errors
198    ///
199    /// Returns an error if the announcement is invalid or cannot be sent.
200    pub async fn announce(&self, sender: &GossipSender, announcement: &PackageAnnouncement) -> Result<()> {
201        let content = announcement.to_bytes()?;
202        let wire = SignedMessage::sign(&self.signing_key, &content);
203        self.gossip.publish(sender, wire).await
204    }
205
206    /// Subscribe to a specific editorial channel.
207    ///
208    /// # Errors
209    ///
210    /// Returns an error if the gossip subscription cannot be created.
211    pub async fn subscribe_channel(&self, channel: &Channel, bootstrap: Vec<PublicKey>) -> Result<GossipTopic> {
212        self.gossip.subscribe(channel_topic(channel), bootstrap).await
213    }
214
215    /// Subscribe to a channel and block until at least one bootstrap peer is
216    /// connected.
217    ///
218    /// # Errors
219    ///
220    /// Returns an error if the gossip subscription cannot join its bootstrap
221    /// peers.
222    pub async fn subscribe_channel_and_join(
223        &self,
224        channel: &Channel,
225        bootstrap: Vec<PublicKey>,
226    ) -> Result<GossipTopic> {
227        self.gossip.subscribe_and_join(channel_topic(channel), bootstrap).await
228    }
229
230    /// Collect matching announcements until the timeout expires.
231    ///
232    /// Each announcement is verified: the Ed25519 signature is checked and the
233    /// claimed `publisher` field must match the signing key.
234    /// A timeout is a normal end condition because gossip has no finite
235    /// response boundary.
236    ///
237    /// # Errors
238    ///
239    /// Returns an error if a received announcement fails signature verification.
240    pub async fn search(
241        &self,
242        topic: &mut GossipTopic,
243        query: Option<&str>,
244        timeout: Duration,
245    ) -> Result<Vec<PackageAnnouncement>> {
246        let mut announcements = Vec::new();
247        let receive = async {
248            while let Some(event) = topic.next().await {
249                if let Event::Received(message) =
250                    event.map_err(|error| SyncwebError::operation("package catalog event failed", error))?
251                {
252                    // Verify the Ed25519 signature on the gossip message
253                    let (verifying_key, content) = SignedMessage::verify(&message.content).map_err(|error| {
254                        SyncwebError::operation("package catalog signature verification failed", error)
255                    })?;
256
257                    let announcement = PackageAnnouncement::from_bytes(content)?;
258
259                    // Verify that the claimed publisher matches the signing key
260                    let claimed_publisher: PublicKey = announcement
261                        .publisher
262                        .parse()
263                        .map_err(|error| SyncwebError::InvalidIdentity(format!("invalid publisher key: {error}")))?;
264                    let actual_publisher = PublicKey::from_bytes(&verifying_key.to_bytes()).map_err(|error| {
265                        SyncwebError::InvalidIdentity(format!("derived NodeId from verifying key is invalid: {error}"))
266                    })?;
267                    if claimed_publisher != actual_publisher {
268                        return Err(SyncwebError::operation(
269                            "package catalog publisher mismatch",
270                            format!("claimed {claimed_publisher}, signed by {actual_publisher}"),
271                        ));
272                    }
273
274                    if announcement.matches(query)
275                        && !announcements.iter().any(|item: &PackageAnnouncement| {
276                            item.collection_id == announcement.collection_id
277                                && item.version == announcement.version
278                                && item.manifest == announcement.manifest
279                        })
280                    {
281                        announcements.push(announcement);
282                    }
283                }
284            }
285            Ok::<(), SyncwebError>(())
286        };
287        if let Ok(result) = tokio::time::timeout(timeout, receive).await {
288            result?;
289        }
290        Ok(announcements)
291    }
292
293    /// Add the endpoint address for a discovered ticket to a memory lookup.
294    ///
295    /// This helper is useful to callers that construct a node with a shared
296    /// [`MemoryLookup`] and want to fetch an announcement immediately.
297    ///
298    /// # Errors
299    ///
300    /// Returns an error if the announcement's manifest ticket is invalid.
301    pub fn register_ticket_endpoint(lookup: &MemoryLookup, announcement: &PackageAnnouncement) -> Result<()> {
302        let ticket = announcement.ticket()?;
303        lookup.add_endpoint_info(ticket.addr().clone());
304        Ok(())
305    }
306}
syncweb_core/folder/catalog.rs

syncweb_core/folder/
catalog.rs
