1use crate::config::{Config, NostrEventTransport};
7#[cfg(feature = "experimental-decentralized-pubsub")]
8use crate::nostr_relay::NostrRelay;
9use crate::storage::{HashtreeStore, StorageRouter};
10use anyhow::{Context, Result};
11use hashtree_core::BlobRoute;
12use hashtree_fips_transport::{
13 bind_fips_endpoint, BoundFipsEndpoint, FipsEndpointOptions, FipsPeerConfig,
14 HashtreeFipsTransport, PeerIdentity, TcpBlobTransport, TcpBlobTransportConfig,
15 DEFAULT_FIPS_DISCOVERY_SCOPE,
16};
17use hashtree_network::{
18 blob_resolver, BlobResolver, MeshReadSource, MeshRoutingConfig, NamedBlobRoute,
19};
20use nostr::nips::nip19::ToBech32;
21use nostr::PublicKey;
22use std::collections::HashSet;
23use std::sync::{Arc, Mutex};
24use std::time::Duration;
25#[cfg(feature = "experimental-decentralized-pubsub")]
26use tokio::sync::mpsc;
27use tokio::task::JoinHandle;
28
29pub type DaemonFipsTransport = HashtreeFipsTransport<StorageRouter>;
30pub type DaemonBlobResolver = BlobResolver<StorageRouter>;
31type DaemonBlobTransport = TcpBlobTransport<StorageRouter>;
32
33const DAEMON_SAME_HOST_PROVIDER_PRIORITY: i16 = 100;
34const BLOB_RESOLVER_REPLY_MARGIN: Duration = Duration::from_secs(1);
35
36pub struct DaemonFipsHandle {
37 pub transport: Arc<DaemonFipsTransport>,
38 pub endpoint_npub: String,
39 pub discovery_scope: String,
40 pub nostr_provider: Option<Arc<dyn nostr_pubsub::PubsubProvider>>,
41 pub blob_resolver: Arc<DaemonBlobResolver>,
42 blob_transport: Mutex<Option<Arc<DaemonBlobTransport>>>,
43 receiver_task: JoinHandle<()>,
44}
45
46impl DaemonFipsHandle {
47 pub fn shutdown(&self) {
48 self.receiver_task.abort();
49 if let Ok(mut transport) = self.blob_transport.lock() {
50 transport.take();
51 }
52 }
53}
54
55#[cfg(feature = "experimental-decentralized-pubsub")]
56pub struct DaemonNostrPubsubHandle {
57 mesh: Arc<hashtree_fips_transport::FipsMeshPubsub<StorageRouter>>,
58 relay: Arc<NostrRelay>,
59 ingest_task: JoinHandle<()>,
60 outbound_task: JoinHandle<()>,
61}
62
63#[cfg(feature = "experimental-decentralized-pubsub")]
64impl DaemonNostrPubsubHandle {
65 pub fn shutdown(&self) {
66 self.relay.set_decentralized_pubsub_sender(None);
67 self.ingest_task.abort();
68 self.outbound_task.abort();
69 self.mesh.shutdown();
70 }
71}
72
73pub async fn start_daemon_fips_transport(
74 config: &Config,
75 keys: &nostr::Keys,
76 store: Arc<HashtreeStore>,
77 peer_ids: Vec<String>,
78) -> Result<Option<DaemonFipsHandle>> {
79 if !config.server.enable_fips || !config.server.mode.hash_get_enabled() {
80 return Ok(None);
81 }
82
83 let active_relays = config.nostr.active_relays();
84 let relays = config.server.resolved_fips_relays(&active_relays);
85 let discovery_scope = normalized_discovery_scope(&config.server.fips_discovery_scope);
86 let peer_configs = daemon_fips_peer_configs(config, peer_ids);
87 let identity_nsec = keys
88 .secret_key()
89 .to_bech32()
90 .context("Failed to encode daemon identity for FIPS endpoint")?;
91 let mut options = FipsEndpointOptions::new(identity_nsec);
92 options.discovery_scope = discovery_scope.clone();
93 options.relays = relays;
94 options.enable_udp = config.server.enable_fips_udp;
95 options.enable_webrtc = config.server.enable_fips_webrtc;
96 options.enable_local_rendezvous = true;
97 options.ethernet_interfaces = config.server.fips_ethernet_interfaces.clone();
98 options.udp_bind_addr = config.server.fips_udp_bind_addr.clone();
99 options.udp_public = config.server.fips_udp_public;
100 options.udp_external_addr = config.server.fips_udp_external_addr.clone();
101 options.webrtc_auto_connect = options.enable_webrtc && !peer_configs.is_empty();
105 options.webrtc_max_connections = hashtree_fips_transport::DEFAULT_FIPS_WEBRTC_MAX_CONNECTIONS;
106 options.open_discovery_max_pending = 0;
107 options.packet_channel_capacity = 1024;
108 let endpoint = bind_fips_endpoint(options)
109 .await
110 .context("Failed to start FIPS endpoint")?;
111
112 let request_timeout = Duration::from_millis(config.server.fips_request_timeout_ms.max(1));
113 let nostr_provider: Option<Arc<dyn nostr_pubsub::PubsubProvider>> =
114 if config.nostr.event_transport == NostrEventTransport::FipsLocalOnly {
115 let options = nostr_pubsub_fips::FipsPubsubClientOptions {
116 query_timeout: request_timeout,
117 ..Default::default()
118 };
119 let client = nostr_pubsub_fips::FipsPubsubClient::start(
120 endpoint.native_endpoint.clone(),
121 options,
122 )
123 .await
124 .context("Failed to start local-only FIPS Nostr pubsub provider")?;
125 Some(Arc::new(client))
126 } else {
127 None
128 };
129 let transport = Arc::new(
130 HashtreeFipsTransport::new(endpoint.endpoint.clone(), store.store_arc())
131 .with_request_timeout(request_timeout)
132 .with_cache_responses(false),
133 );
134 let (blob_resolver, blob_transport) =
135 bind_daemon_blob_resolver(&endpoint, store.store_arc(), &peer_configs, request_timeout)
136 .await?;
137 if !peer_configs.is_empty() {
138 transport.set_peer_configs(peer_configs).await;
139 }
140 let receiver_task = transport.start();
141
142 Ok(Some(DaemonFipsHandle {
143 transport,
144 endpoint_npub: endpoint.local_peer_id,
145 discovery_scope: endpoint.discovery_scope,
146 nostr_provider,
147 blob_resolver,
148 blob_transport: Mutex::new(Some(blob_transport)),
149 receiver_task,
150 }))
151}
152
153async fn bind_daemon_blob_resolver(
154 endpoint: &BoundFipsEndpoint,
155 store: Arc<StorageRouter>,
156 peers: &[FipsPeerConfig],
157 request_timeout: Duration,
158) -> Result<(Arc<DaemonBlobResolver>, Arc<DaemonBlobTransport>)> {
159 let resolver = Arc::new(blob_resolver(
160 store.clone(),
161 endpoint.local_peer_id.clone(),
162 request_timeout,
163 MeshRoutingConfig::default(),
164 ));
165 let transport = Arc::new(
166 TcpBlobTransport::bind_advertised_route_with_config(
167 endpoint.native_endpoint.clone(),
168 store,
169 resolver.clone(),
170 blob_transport_config(request_timeout),
171 DAEMON_SAME_HOST_PROVIDER_PRIORITY,
172 )
173 .await
174 .context("Failed to advertise htree's blob resolver")?,
175 );
176 let routes: Vec<Arc<dyn MeshReadSource>> = peers
177 .iter()
178 .filter_map(|peer| {
179 let identity = PeerIdentity::from_npub(&peer.npub).ok()?;
180 let route: Arc<dyn BlobRoute> = Arc::new(transport.weak_route_to(identity));
181 Some(
182 Arc::new(NamedBlobRoute::mesh_peer(peer.npub.clone(), route))
183 as Arc<dyn MeshReadSource>,
184 )
185 })
186 .collect();
187 resolver.set_read_sources(routes).await;
188 Ok((resolver, transport))
189}
190
191fn blob_transport_config(resolver_timeout: Duration) -> TcpBlobTransportConfig {
192 TcpBlobTransportConfig {
193 idle_timeout: resolver_timeout.saturating_add(BLOB_RESOLVER_REPLY_MARGIN),
194 }
195}
196
197pub async fn start_daemon_nostr_provider(
198 config: &Config,
199 fips_handle: Option<&DaemonFipsHandle>,
200) -> Result<Option<Arc<dyn nostr_pubsub::PubsubProvider>>> {
201 match config.nostr.event_transport {
202 NostrEventTransport::Relay => {
203 let relays = config.nostr.active_relays();
204 if relays.is_empty() {
205 return Ok(None);
206 }
207 let event_bus = nostr_pubsub_relay::RelayEventBus::new(
208 relays,
209 Duration::from_millis(config.server.fips_request_timeout_ms.max(1)),
210 )
211 .await
212 .context("Failed to start Nostr relay event provider")?;
213 Ok(Some(Arc::new(event_bus)))
214 }
215 NostrEventTransport::FipsLocalOnly => fips_handle
216 .and_then(|handle| handle.nostr_provider.clone())
217 .map(Some)
218 .ok_or_else(|| {
219 anyhow::anyhow!(
220 "nostr.event_transport=fips-local-only requires the local FIPS Nostr pubsub provider"
221 )
222 }),
223 }
224}
225
226#[cfg(feature = "experimental-decentralized-pubsub")]
227pub async fn start_daemon_nostr_pubsub(
228 config: &Config,
229 fips_handle: Option<&DaemonFipsHandle>,
230 store: Arc<HashtreeStore>,
231 relay: Option<Arc<NostrRelay>>,
232) -> Result<Option<Arc<DaemonNostrPubsubHandle>>> {
233 if !daemon_decentralized_pubsub_ready(config, fips_handle.is_some(), relay.is_some()) {
234 return Ok(None);
235 }
236
237 let fips_handle = fips_handle.expect("checked by daemon_decentralized_pubsub_ready");
238 let relay = relay.expect("checked by daemon_decentralized_pubsub_ready");
239 let request_timeout = Duration::from_millis(config.server.fips_request_timeout_ms.max(1));
240 let mesh = Arc::new(
241 fips_handle
242 .transport
243 .start_mesh_pubsub_with_options(
244 store.store_arc(),
245 fips_handle.endpoint_npub.clone(),
246 request_timeout,
247 hashtree_fips_transport::FipsMeshPubsubOptions {
248 forwarding: config.nostr.decentralized_pubsub_forwarding,
249 fanout: config.nostr.decentralized_pubsub_fanout,
250 max_hops: config.nostr.decentralized_pubsub_max_hops,
251 },
252 )
253 .await
254 .context("Failed to start decentralized Nostr pubsub over FIPS")?,
255 );
256 let _ = mesh
257 .subscribe_pubsub(crate::nostr_pubsub::NOSTR_EVENT_PUBSUB_STREAM)
258 .await;
259
260 let (outbound_tx, outbound_rx) = mpsc::unbounded_channel();
261 relay.set_decentralized_pubsub_sender(Some(outbound_tx));
262 let max_event_bytes = config.nostr.decentralized_pubsub_max_event_bytes;
263 let ingest_task =
264 spawn_daemon_nostr_pubsub_ingest(Arc::clone(&mesh), Arc::clone(&relay), max_event_bytes);
265 let outbound_task =
266 spawn_daemon_nostr_pubsub_outbound(Arc::clone(&mesh), outbound_rx, max_event_bytes);
267
268 Ok(Some(Arc::new(DaemonNostrPubsubHandle {
269 mesh,
270 relay,
271 ingest_task,
272 outbound_task,
273 })))
274}
275
276#[cfg(feature = "experimental-decentralized-pubsub")]
277fn daemon_decentralized_pubsub_ready(config: &Config, has_fips: bool, has_relay: bool) -> bool {
278 config.nostr.decentralized_pubsub_enabled() && has_fips && has_relay
279}
280
281#[cfg(feature = "experimental-decentralized-pubsub")]
282fn spawn_daemon_nostr_pubsub_ingest(
283 mesh: Arc<hashtree_fips_transport::FipsMeshPubsub<StorageRouter>>,
284 relay: Arc<NostrRelay>,
285 max_event_bytes: usize,
286) -> JoinHandle<()> {
287 tokio::spawn(async move {
288 loop {
289 let delivery = mesh.recv_pubsub_event().await;
290 let stream_id = delivery.stream_id.clone();
291 let origin_peer_id = delivery.origin_peer_id.clone();
292 match crate::nostr_pubsub::ingest_nostr_pubsub_payload_with_limit(
293 &relay,
294 &delivery.stream_id,
295 &delivery.payload,
296 max_event_bytes,
297 )
298 .await
299 {
300 Ok(Some(event)) => {
301 tracing::debug!(
302 stream_id,
303 origin_peer_id,
304 event_id = %event.id.to_hex(),
305 "ingested decentralized Nostr pubsub event"
306 );
307 }
308 Ok(None) => {}
309 Err(err) => {
310 tracing::warn!(
311 stream_id,
312 origin_peer_id,
313 "nostr decentralized pubsub ingest failed: {err:#}"
314 );
315 }
316 }
317 }
318 })
319}
320
321#[cfg(feature = "experimental-decentralized-pubsub")]
322fn spawn_daemon_nostr_pubsub_outbound(
323 mesh: Arc<hashtree_fips_transport::FipsMeshPubsub<StorageRouter>>,
324 mut outbound_rx: mpsc::UnboundedReceiver<nostr::Event>,
325 max_event_bytes: usize,
326) -> JoinHandle<()> {
327 tokio::spawn(async move {
328 let mut seq = 1u64;
329 while let Some(event) = outbound_rx.recv().await {
330 let event_id = event.id.to_hex();
331 match crate::nostr_pubsub::publish_fips_nostr_event_with_limit(
332 mesh.as_ref(),
333 seq,
334 &event,
335 max_event_bytes,
336 )
337 .await
338 {
339 Ok(stats) => {
340 tracing::debug!(
341 event_id,
342 seq,
343 selected_peers = stats.selected_peers,
344 sent_peers = stats.sent_peers,
345 sent_bytes = stats.sent_bytes,
346 "published Nostr event over decentralized pubsub"
347 );
348 }
349 Err(err) => {
350 tracing::warn!(
351 event_id,
352 seq,
353 "nostr decentralized pubsub publish failed: {err:#}"
354 );
355 }
356 }
357 seq = seq.wrapping_add(1);
358 if seq == 0 {
359 seq = 1;
360 }
361 }
362 })
363}
364
365pub fn fips_peer_ids_from_pubkeys(pubkeys: Vec<[u8; 32]>) -> Vec<String> {
366 pubkeys
367 .into_iter()
368 .filter_map(|pubkey| PublicKey::from_slice(&pubkey).ok())
369 .filter_map(|pubkey| pubkey.to_bech32().ok())
370 .collect()
371}
372
373pub fn daemon_fips_peer_configs(
374 config: &Config,
375 discovered_peer_ids: Vec<String>,
376) -> Vec<FipsPeerConfig> {
377 let mut seen = HashSet::new();
378 let mut peers = Vec::new();
379
380 for peer in &config.server.fips_peers {
381 let npub = peer.npub.trim().to_string();
382 if npub.is_empty() || !seen.insert(npub.clone()) {
383 continue;
384 }
385 let udp_addresses = peer
386 .udp_addresses
387 .iter()
388 .map(|addr| addr.trim().to_string())
389 .filter(|addr| !addr.is_empty())
390 .collect();
391 peers.push(FipsPeerConfig {
392 npub,
393 udp_addresses,
394 });
395 }
396
397 for peer_id in discovered_peer_ids {
398 let npub = peer_id.trim().to_string();
399 if npub.is_empty() || !seen.insert(npub.clone()) {
400 continue;
401 }
402 peers.push(FipsPeerConfig::new(npub));
403 }
404
405 peers
406}
407
408fn normalized_discovery_scope(scope: &str) -> String {
409 let scope = scope.trim();
410 if scope.is_empty() {
411 DEFAULT_FIPS_DISCOVERY_SCOPE.to_string()
412 } else {
413 scope.to_string()
414 }
415}
416
417#[cfg(test)]
418mod tests {
419 use super::*;
420 use hashtree_core::Store;
421 use sha2::{Digest, Sha256};
422 use tokio::time::timeout;
423
424 #[test]
425 fn fips_peer_ids_from_pubkeys_encodes_npbus() {
426 let keys = nostr::Keys::generate();
427 let expected = keys.public_key().to_bech32().unwrap();
428
429 assert_eq!(
430 fips_peer_ids_from_pubkeys(vec![keys.public_key().to_bytes()]),
431 vec![expected]
432 );
433 }
434
435 #[test]
436 fn empty_discovery_scope_uses_hashtree_default() {
437 assert_eq!(
438 normalized_discovery_scope(" "),
439 DEFAULT_FIPS_DISCOVERY_SCOPE.to_string()
440 );
441 }
442
443 #[tokio::test]
444 async fn fips_local_only_event_transport_requires_local_provider() {
445 let mut config = Config::default();
446 config.nostr.event_transport = NostrEventTransport::FipsLocalOnly;
447
448 let error = start_daemon_nostr_provider(&config, None)
449 .await
450 .err()
451 .expect("missing local FIPS provider must fail");
452
453 assert!(error.to_string().contains("fips-local-only"));
454 assert!(error.to_string().contains("requires"));
455 }
456
457 #[tokio::test]
458 async fn daemon_blob_resolver_advertises_and_serves_storage_router() {
459 let provider_endpoint = local_only_endpoint("htree-provider-test").await;
460 let observer_endpoint = local_only_endpoint("iris-drive-observer-test").await;
461 let temp = tempfile::tempdir().unwrap();
462 let store = Arc::new(HashtreeStore::new(temp.path()).unwrap());
463 let data = b"served directly from htree StorageRouter".to_vec();
464 let hash = Sha256::digest(&data).into();
465 store.store_arc().put(hash, data.clone()).await.unwrap();
466 let (blob_resolver, provider) = bind_daemon_blob_resolver(
467 &provider_endpoint,
468 store.store_arc(),
469 &[],
470 Duration::from_secs(1),
471 )
472 .await
473 .expect("bind htree provider");
474 let transport = Arc::new(HashtreeFipsTransport::new(
475 provider_endpoint.endpoint.clone(),
476 store.store_arc(),
477 ));
478 let receiver_task = transport.start();
479 let handle = DaemonFipsHandle {
480 transport,
481 endpoint_npub: provider_endpoint.local_peer_id.clone(),
482 discovery_scope: provider_endpoint.discovery_scope.clone(),
483 nostr_provider: None,
484 blob_resolver,
485 blob_transport: Mutex::new(Some(provider)),
486 receiver_task,
487 };
488
489 timeout(Duration::from_secs(10), async {
490 loop {
491 if provider_advertised(&observer_endpoint, &provider_endpoint.local_peer_id) {
492 break;
493 }
494 tokio::time::sleep(Duration::from_millis(20)).await;
495 }
496 })
497 .await
498 .expect("htree provider capability did not converge");
499
500 let requester = hashtree_fips_transport::SameHostBlobStore::bind(
501 observer_endpoint.native_endpoint.clone(),
502 Arc::new(hashtree_core::MemoryStore::new()),
503 None,
504 hashtree_fips_transport::SameHostBlobStoreConfig::default(),
505 )
506 .await
507 .unwrap();
508 assert_eq!(requester.get(&hash).await.unwrap(), Some(data));
509
510 drop(requester);
511 handle.shutdown();
512 timeout(Duration::from_secs(10), async {
513 while provider_advertised(&observer_endpoint, &provider_endpoint.local_peer_id) {
514 tokio::time::sleep(Duration::from_millis(20)).await;
515 }
516 })
517 .await
518 .expect("htree provider capability survived daemon shutdown");
519 drop(handle);
520 observer_endpoint.native_endpoint.shutdown().await.unwrap();
521 provider_endpoint.native_endpoint.shutdown().await.unwrap();
522 }
523
524 #[tokio::test]
525 async fn daemon_blob_resolvers_honor_multi_hop_htl_and_cache() {
526 let (source_endpoint, source_addr) = udp_endpoint("htree-source-test").await;
527 let (middle_endpoint, middle_addr) = udp_endpoint("htree-middle-test").await;
528 let (requester_endpoint, requester_addr) = udp_endpoint("htree-requester-test").await;
529 let source_dir = tempfile::tempdir().unwrap();
530 let middle_dir = tempfile::tempdir().unwrap();
531 let requester_dir = tempfile::tempdir().unwrap();
532 let source_store = Arc::new(HashtreeStore::new(source_dir.path()).unwrap());
533 let middle_store = Arc::new(HashtreeStore::new(middle_dir.path()).unwrap());
534 let requester_store = Arc::new(HashtreeStore::new(requester_dir.path()).unwrap());
535 let data = b"canonical multi-hop blob route".to_vec();
536 let hash = Sha256::digest(&data).into();
537 source_store
538 .store_arc()
539 .put(hash, data.clone())
540 .await
541 .unwrap();
542
543 source_endpoint
544 .endpoint
545 .set_peer_configs(vec![FipsPeerConfig {
546 npub: middle_endpoint.local_peer_id.clone(),
547 udp_addresses: vec![middle_addr.clone()],
548 }])
549 .await
550 .unwrap();
551 middle_endpoint
552 .endpoint
553 .set_peer_configs(vec![
554 FipsPeerConfig {
555 npub: source_endpoint.local_peer_id.clone(),
556 udp_addresses: vec![source_addr.clone()],
557 },
558 FipsPeerConfig {
559 npub: requester_endpoint.local_peer_id.clone(),
560 udp_addresses: vec![requester_addr],
561 },
562 ])
563 .await
564 .unwrap();
565 requester_endpoint
566 .endpoint
567 .set_peer_configs(vec![FipsPeerConfig {
568 npub: middle_endpoint.local_peer_id.clone(),
569 udp_addresses: vec![middle_addr.clone()],
570 }])
571 .await
572 .unwrap();
573
574 let (_source_resolver, source_transport) = bind_daemon_blob_resolver(
575 &source_endpoint,
576 source_store.store_arc(),
577 &[],
578 Duration::from_secs(2),
579 )
580 .await
581 .unwrap();
582 let (middle_resolver, middle_transport) = bind_daemon_blob_resolver(
583 &middle_endpoint,
584 middle_store.store_arc(),
585 &[FipsPeerConfig {
586 npub: source_endpoint.local_peer_id.clone(),
587 udp_addresses: vec![source_addr],
588 }],
589 Duration::from_secs(2),
590 )
591 .await
592 .unwrap();
593 let (requester_resolver, requester_transport) = bind_daemon_blob_resolver(
594 &requester_endpoint,
595 requester_store.store_arc(),
596 &[FipsPeerConfig {
597 npub: middle_endpoint.local_peer_id.clone(),
598 udp_addresses: vec![middle_addr],
599 }],
600 Duration::from_secs(2),
601 )
602 .await
603 .unwrap();
604
605 wait_for_peer(&requester_endpoint, &middle_endpoint.local_peer_id).await;
606 wait_for_peer(&middle_endpoint, &source_endpoint.local_peer_id).await;
607
608 assert_eq!(
609 requester_resolver
610 .route(hashtree_core::BlobRequest { hash, htl: 1 })
611 .await
612 .expect("exhausted search"),
613 hashtree_core::BlobReply::NoResult,
614 );
615 assert_eq!(
616 requester_resolver
617 .route(hashtree_core::BlobRequest { hash, htl: 2 })
618 .await
619 .expect("two-hop search"),
620 hashtree_core::BlobReply::Data(data.clone()),
621 );
622 assert_eq!(
623 middle_store.store_arc().get(&hash).await.unwrap(),
624 Some(data.clone())
625 );
626 assert_eq!(
627 requester_store.store_arc().get(&hash).await.unwrap(),
628 Some(data)
629 );
630
631 drop(requester_transport);
632 drop(middle_transport);
633 drop(source_transport);
634 requester_endpoint.native_endpoint.shutdown().await.unwrap();
635 middle_endpoint.native_endpoint.shutdown().await.unwrap();
636 source_endpoint.native_endpoint.shutdown().await.unwrap();
637 drop(middle_resolver);
638 }
639
640 fn provider_advertised(endpoint: &BoundFipsEndpoint, npub: &str) -> bool {
641 endpoint
642 .native_endpoint
643 .local_instance_advertisements()
644 .unwrap()
645 .iter()
646 .any(|advert| {
647 advert.npub == npub
648 && advert
649 .capability(hashtree_fips_transport::TCP_BLOB_CAPABILITY)
650 .and_then(|capability| capability.fsp_port)
651 == Some(hashtree_fips_transport::TCP_BLOB_SERVICE_PORT)
652 })
653 }
654
655 async fn local_only_endpoint(scope: &str) -> hashtree_fips_transport::BoundFipsEndpoint {
656 let keys = nostr::Keys::generate();
657 let mut options = FipsEndpointOptions::new(keys.secret_key().to_bech32().unwrap());
658 options.discovery_scope = scope.to_string();
659 options.enable_udp = false;
660 options.enable_webrtc = false;
661 options.enable_local_rendezvous = true;
662 options.enable_lan_discovery = false;
663 options.share_local_candidates = false;
664 bind_fips_endpoint(options).await.unwrap()
665 }
666
667 async fn udp_endpoint(scope: &str) -> (BoundFipsEndpoint, String) {
668 let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
669 let addr = socket.local_addr().unwrap();
670 drop(socket);
671 let keys = nostr::Keys::generate();
672 let mut options = FipsEndpointOptions::new(keys.secret_key().to_bech32().unwrap());
673 options.discovery_scope = scope.to_string();
674 options.enable_udp = true;
675 options.udp_bind_addr = Some(addr.to_string());
676 options.enable_webrtc = false;
677 options.enable_local_rendezvous = false;
678 options.enable_lan_discovery = false;
679 options.share_local_candidates = false;
680 let endpoint = bind_fips_endpoint(options).await.unwrap();
681 (endpoint, addr.to_string())
682 }
683
684 async fn wait_for_peer(endpoint: &BoundFipsEndpoint, npub: &str) {
685 let result = timeout(Duration::from_secs(10), async {
686 loop {
687 if endpoint
688 .native_endpoint
689 .peers()
690 .await
691 .unwrap()
692 .iter()
693 .any(|peer| peer.npub == npub && peer.connected)
694 {
695 break;
696 }
697 tokio::time::sleep(Duration::from_millis(20)).await;
698 }
699 })
700 .await;
701 if result.is_err() {
702 panic!(
703 "FIPS peer {npub} did not connect; peers={:?}",
704 endpoint.native_endpoint.peers().await
705 );
706 }
707 }
708
709 #[test]
710 fn daemon_fips_peer_configs_prefer_configured_peers() {
711 let mut config = Config::default();
712 config.server.fips_peers = vec![
713 crate::config::ConfiguredFipsPeer {
714 npub: " origin ".to_string(),
715 udp_addresses: vec![" udp:192.0.2.10:2121 ".to_string(), " ".to_string()],
716 },
717 crate::config::ConfiguredFipsPeer {
718 npub: "origin".to_string(),
719 udp_addresses: vec!["udp:ignored:2121".to_string()],
720 },
721 ];
722
723 let peers = daemon_fips_peer_configs(
724 &config,
725 vec![
726 "origin".to_string(),
727 " followed ".to_string(),
728 " ".to_string(),
729 ],
730 );
731
732 assert_eq!(
733 peers,
734 vec![
735 FipsPeerConfig {
736 npub: "origin".to_string(),
737 udp_addresses: vec!["udp:192.0.2.10:2121".to_string()],
738 },
739 FipsPeerConfig::new("followed"),
740 ]
741 );
742 }
743
744 #[test]
745 fn blob_carrier_deadline_outlives_the_complete_resolver_budget() {
746 for resolver_timeout in [Duration::from_millis(1), Duration::from_secs(20)] {
747 assert!(blob_transport_config(resolver_timeout).idle_timeout > resolver_timeout);
748 }
749 }
750
751 #[cfg(feature = "experimental-decentralized-pubsub")]
752 #[test]
753 fn daemon_decentralized_pubsub_requires_config_fips_and_relay() {
754 let mut config = Config::default();
755 assert!(!daemon_decentralized_pubsub_ready(&config, true, true));
756
757 config.nostr.decentralized_pubsub = true;
758 assert!(daemon_decentralized_pubsub_ready(&config, true, true));
759 assert!(!daemon_decentralized_pubsub_ready(&config, false, true));
760 assert!(!daemon_decentralized_pubsub_ready(&config, true, false));
761
762 config.nostr.enabled = false;
763 assert!(!daemon_decentralized_pubsub_ready(&config, true, true));
764 }
765}