use crate::config::{Config, NostrEventTransport};
#[cfg(feature = "experimental-decentralized-pubsub")]
use crate::nostr_relay::NostrRelay;
use crate::storage::{HashtreeStore, StorageRouter};
use anyhow::{Context, Result};
use hashtree_core::BlobRoute;
use hashtree_fips_transport::{
bind_fips_endpoint, bind_fips_endpoint_at_local_rendezvous, set_fips_peer_configs,
BoundFipsEndpoint, FipsEndpoint, FipsEndpointOptions, FipsPeerConfig, NostrRelayAdapter,
PeerIdentity, TcpBlobTransport, TcpBlobTransportConfig, DEFAULT_FIPS_DISCOVERY_SCOPE,
};
use hashtree_network::{
blob_resolver, BlobResolver, MeshReadSource, MeshRoutingConfig, NamedBlobRoute,
};
use nostr::nips::nip19::ToBech32;
use nostr::PublicKey;
#[cfg(feature = "experimental-decentralized-pubsub")]
use nostr_pubsub::{EventBus, EventSource, Filter, VerifiedEvent};
use nostr_pubsub_fips::{FipsPubsubClient, FipsPubsubClientOptions};
use std::collections::HashSet;
use std::sync::{Arc, Mutex};
use std::time::Duration;
#[cfg(feature = "experimental-decentralized-pubsub")]
use tokio::sync::mpsc;
#[cfg(feature = "experimental-decentralized-pubsub")]
use tokio::task::JoinHandle;
pub type DaemonBlobResolver = BlobResolver<StorageRouter>;
type DaemonBlobTransport = TcpBlobTransport<StorageRouter>;
const DAEMON_SAME_HOST_PROVIDER_PRIORITY: i16 = 100;
const BLOB_RESOLVER_REPLY_MARGIN: Duration = Duration::from_secs(1);
pub struct DaemonFipsHandle {
pub endpoint: Arc<FipsEndpoint>,
pub endpoint_npub: String,
pub discovery_scope: String,
pub nostr_provider: Option<Arc<dyn nostr_pubsub::PubsubProvider>>,
pub pubsub_client: Option<Arc<FipsPubsubClient>>,
pub blob_resolver: Arc<DaemonBlobResolver>,
blob_transport: Mutex<Option<Arc<DaemonBlobTransport>>>,
relay_adapter: Mutex<Option<NostrRelayAdapter>>,
}
impl DaemonFipsHandle {
pub async fn shutdown(&self) {
if let Ok(mut transport) = self.blob_transport.lock() {
transport.take();
}
let relay_adapter = self
.relay_adapter
.lock()
.ok()
.and_then(|mut adapter| adapter.take());
if let Some(relay_adapter) = relay_adapter {
relay_adapter.stop().await;
}
if let Err(err) = self.endpoint.shutdown().await {
tracing::warn!("failed to stop embedded FIPS endpoint: {err}");
}
}
}
#[cfg(feature = "experimental-decentralized-pubsub")]
pub struct DaemonNostrPubsubHandle {
_client: Arc<FipsPubsubClient>,
relay: Arc<NostrRelay>,
ingest_task: JoinHandle<()>,
outbound_task: JoinHandle<()>,
}
#[cfg(feature = "experimental-decentralized-pubsub")]
impl DaemonNostrPubsubHandle {
pub fn shutdown(&self) {
self.relay.set_decentralized_pubsub_sender(None);
self.ingest_task.abort();
self.outbound_task.abort();
}
}
pub async fn start_daemon_fips_transport(
config: &Config,
keys: &nostr::Keys,
store: Arc<HashtreeStore>,
peer_ids: Vec<String>,
) -> Result<Option<DaemonFipsHandle>> {
if !config.server.enable_fips || !config.server.mode.hash_get_enabled() {
return Ok(None);
}
let active_relays = config.nostr.active_relays();
let relays = config.server.resolved_fips_relays(&active_relays);
let relay_carrier_relays = relays.clone();
let discovery_scope = normalized_discovery_scope(&config.server.fips_discovery_scope);
let peer_configs = daemon_fips_peer_configs(config, peer_ids);
let identity_nsec = keys
.secret_key()
.to_bech32()
.context("Failed to encode daemon identity for FIPS endpoint")?;
let mut options = FipsEndpointOptions::new(identity_nsec);
options.discovery_scope = discovery_scope.clone();
options.relays = relays;
options.enable_udp = config.server.enable_fips_udp;
options.enable_webrtc = config.server.enable_fips_webrtc;
options.enable_local_rendezvous = true;
options.enable_lan_discovery = config.server.enable_fips_lan_discovery;
options.ethernet_interfaces = config.server.fips_ethernet_interfaces.clone();
options.udp_bind_addr = config.server.fips_udp_bind_addr.clone();
options.udp_public = config.server.fips_udp_public;
options.udp_external_addr = config.server.fips_udp_external_addr.clone();
options.webrtc_auto_connect = options.enable_webrtc && !peer_configs.is_empty();
options.webrtc_max_connections = hashtree_fips_transport::DEFAULT_FIPS_WEBRTC_MAX_CONNECTIONS;
options.open_discovery_max_pending = 0;
options.packet_channel_capacity = 1024;
let endpoint = if let Some(raw) = config.server.fips_local_rendezvous_addr.as_deref() {
let rendezvous_addr = raw
.parse()
.context("server.fips_local_rendezvous_addr must be an IPv4 socket address")?;
bind_fips_endpoint_at_local_rendezvous(options, rendezvous_addr).await
} else {
bind_fips_endpoint(options).await
}
.context("Failed to start FIPS endpoint")?;
let relay_adapter =
NostrRelayAdapter::start(endpoint.native_endpoint.clone(), &relay_carrier_relays)
.await
.map_err(anyhow::Error::msg)
.context("Failed to start FIPS Nostr relay carrier")?;
let request_timeout = Duration::from_millis(config.server.fips_request_timeout_ms.max(1));
let pubsub_client = if daemon_fips_pubsub_required(config) {
let options = FipsPubsubClientOptions {
query_timeout: request_timeout,
max_frame_bytes: config
.nostr
.decentralized_pubsub_max_event_bytes
.min(nostr_pubsub_fips::FIPS_NOSTR_PUBSUB_MAX_DATAGRAM_BYTES),
..Default::default()
};
Some(Arc::new(
FipsPubsubClient::start(endpoint.native_endpoint.clone(), options)
.await
.context("Failed to start FIPS Nostr pubsub provider")?,
))
} else {
None
};
let nostr_provider =
(config.nostr.event_transport == NostrEventTransport::FipsLocalOnly).then(|| {
pubsub_client
.as_ref()
.expect("FIPS local-only transport starts the shared pubsub client")
.clone() as Arc<dyn nostr_pubsub::PubsubProvider>
});
if !peer_configs.is_empty() {
set_fips_peer_configs(endpoint.native_endpoint.as_ref(), peer_configs.clone())
.await
.context("Failed to configure FIPS peers")?;
}
let (blob_resolver, blob_transport) =
bind_daemon_blob_resolver(&endpoint, store.store_arc(), &peer_configs, request_timeout)
.await?;
Ok(Some(DaemonFipsHandle {
endpoint: endpoint.native_endpoint,
endpoint_npub: endpoint.local_peer_id,
discovery_scope: endpoint.discovery_scope,
nostr_provider,
pubsub_client,
blob_resolver,
blob_transport: Mutex::new(Some(blob_transport)),
relay_adapter: Mutex::new(relay_adapter),
}))
}
fn daemon_fips_pubsub_required(config: &Config) -> bool {
config.nostr.event_transport == NostrEventTransport::FipsLocalOnly
|| config.nostr.decentralized_pubsub_enabled()
}
async fn bind_daemon_blob_resolver(
endpoint: &BoundFipsEndpoint,
store: Arc<StorageRouter>,
peers: &[FipsPeerConfig],
request_timeout: Duration,
) -> Result<(Arc<DaemonBlobResolver>, Arc<DaemonBlobTransport>)> {
let resolver = Arc::new(blob_resolver(
store.clone(),
endpoint.local_peer_id.clone(),
request_timeout,
MeshRoutingConfig::default(),
));
let transport = Arc::new(
TcpBlobTransport::bind_advertised_route_with_config(
endpoint.native_endpoint.clone(),
store,
resolver.clone(),
blob_transport_config(request_timeout),
DAEMON_SAME_HOST_PROVIDER_PRIORITY,
)
.await
.context("Failed to advertise htree's blob resolver")?,
);
let routes: Vec<Arc<dyn MeshReadSource>> = peers
.iter()
.filter_map(|peer| {
let identity = PeerIdentity::from_npub(&peer.npub).ok()?;
let route: Arc<dyn BlobRoute> = Arc::new(transport.weak_route_to(identity));
Some(
Arc::new(NamedBlobRoute::mesh_peer(peer.npub.clone(), route))
as Arc<dyn MeshReadSource>,
)
})
.collect();
resolver.set_read_sources(routes).await;
Ok((resolver, transport))
}
fn blob_transport_config(resolver_timeout: Duration) -> TcpBlobTransportConfig {
TcpBlobTransportConfig {
idle_timeout: resolver_timeout.saturating_add(BLOB_RESOLVER_REPLY_MARGIN),
}
}
pub async fn start_daemon_nostr_provider(
config: &Config,
fips_handle: Option<&DaemonFipsHandle>,
) -> Result<Option<Arc<dyn nostr_pubsub::PubsubProvider>>> {
match config.nostr.event_transport {
NostrEventTransport::Relay => {
let relays = config.nostr.active_relays();
if relays.is_empty() {
return Ok(None);
}
let event_bus = nostr_pubsub_relay::RelayEventBus::new(
relays,
Duration::from_millis(config.server.fips_request_timeout_ms.max(1)),
)
.await
.context("Failed to start Nostr relay event provider")?;
Ok(Some(Arc::new(event_bus)))
}
NostrEventTransport::FipsLocalOnly => fips_handle
.and_then(|handle| handle.nostr_provider.clone())
.map(Some)
.ok_or_else(|| {
anyhow::anyhow!(
"nostr.event_transport=fips-local-only requires the local FIPS Nostr pubsub provider"
)
}),
}
}
#[cfg(feature = "experimental-decentralized-pubsub")]
pub async fn start_daemon_nostr_pubsub(
config: &Config,
fips_handle: Option<&DaemonFipsHandle>,
relay: Option<Arc<NostrRelay>>,
) -> Result<Option<Arc<DaemonNostrPubsubHandle>>> {
if !daemon_decentralized_pubsub_ready(config, fips_handle.is_some(), relay.is_some()) {
return Ok(None);
}
let fips_handle = fips_handle.expect("checked by daemon_decentralized_pubsub_ready");
let relay = relay.expect("checked by daemon_decentralized_pubsub_ready");
let client = fips_handle.pubsub_client.clone().ok_or_else(|| {
anyhow::anyhow!("decentralized Nostr pubsub requires the shared FIPS pubsub client")
})?;
let (outbound_tx, outbound_rx) = mpsc::unbounded_channel();
relay.set_decentralized_pubsub_sender(Some(outbound_tx));
let ingest_task = spawn_daemon_nostr_pubsub_ingest(
Arc::clone(&client),
Arc::clone(&fips_handle.endpoint),
Arc::clone(&relay),
);
let outbound_task = spawn_daemon_nostr_pubsub_outbound(Arc::clone(&client), outbound_rx);
Ok(Some(Arc::new(DaemonNostrPubsubHandle {
_client: client,
relay,
ingest_task,
outbound_task,
})))
}
#[cfg(feature = "experimental-decentralized-pubsub")]
fn daemon_decentralized_pubsub_ready(config: &Config, has_fips: bool, has_relay: bool) -> bool {
config.nostr.decentralized_pubsub_enabled() && has_fips && has_relay
}
#[cfg(feature = "experimental-decentralized-pubsub")]
fn spawn_daemon_nostr_pubsub_ingest(
client: Arc<FipsPubsubClient>,
endpoint: Arc<FipsEndpoint>,
relay: Arc<NostrRelay>,
) -> JoinHandle<()> {
tokio::spawn(async move {
loop {
let subscribed_peers = connected_fips_peer_ids(&endpoint).await;
if subscribed_peers.is_empty() {
tokio::time::sleep(Duration::from_millis(250)).await;
continue;
}
let mut subscription = match client.subscribe(vec![Filter::new()]).await {
Ok(subscription) => subscription,
Err(err) => {
tracing::debug!("waiting to subscribe to FIPS Nostr peers: {err}");
tokio::time::sleep(Duration::from_millis(250)).await;
continue;
}
};
loop {
tokio::select! {
delivery = subscription.recv() => {
let Some(delivery) = delivery else {
break;
};
let source = delivery.source.id.as_str().to_string();
let event = delivery.event.into_event();
let event_id = event.id.to_hex();
match relay.ingest_peer_event_silent(event).await {
Ok(true) => tracing::debug!(
source,
event_id,
"ingested decentralized Nostr pubsub event"
),
Ok(false) => {}
Err(err) => tracing::warn!(
source,
event_id,
"nostr decentralized pubsub ingest failed: {err:#}"
),
}
}
() = tokio::time::sleep(Duration::from_millis(500)) => {
if connected_fips_peer_ids(&endpoint).await != subscribed_peers {
break;
}
}
}
}
}
})
}
#[cfg(feature = "experimental-decentralized-pubsub")]
async fn connected_fips_peer_ids(endpoint: &FipsEndpoint) -> Vec<String> {
let mut peers = endpoint
.peers()
.await
.unwrap_or_default()
.into_iter()
.filter(|peer| peer.connected)
.map(|peer| peer.npub)
.collect::<Vec<_>>();
peers.sort_unstable();
peers.dedup();
peers
}
#[cfg(feature = "experimental-decentralized-pubsub")]
fn spawn_daemon_nostr_pubsub_outbound(
client: Arc<FipsPubsubClient>,
mut outbound_rx: mpsc::UnboundedReceiver<nostr::Event>,
) -> JoinHandle<()> {
tokio::spawn(async move {
while let Some(event) = outbound_rx.recv().await {
let event_id = event.id.to_hex();
let verified = match VerifiedEvent::try_from(event) {
Ok(event) => event,
Err(err) => {
tracing::warn!(event_id, "invalid outbound Nostr event: {err}");
continue;
}
};
match client
.publish(verified, EventSource::local_index("htree-relay"))
.await
{
Ok(report) => tracing::debug!(
event_id,
accepted = report.accepted,
"published Nostr event over decentralized pubsub"
),
Err(err) => {
tracing::warn!(event_id, "nostr decentralized pubsub publish failed: {err}")
}
}
}
})
}
pub fn fips_peer_ids_from_pubkeys(pubkeys: Vec<[u8; 32]>) -> Vec<String> {
pubkeys
.into_iter()
.filter_map(|pubkey| PublicKey::from_slice(&pubkey).ok())
.filter_map(|pubkey| pubkey.to_bech32().ok())
.collect()
}
pub fn daemon_fips_peer_configs(
config: &Config,
discovered_peer_ids: Vec<String>,
) -> Vec<FipsPeerConfig> {
let mut seen = HashSet::new();
let mut peers = Vec::new();
for peer in &config.server.fips_peers {
let npub = peer.npub.trim().to_string();
if npub.is_empty() || !seen.insert(npub.clone()) {
continue;
}
let udp_addresses = peer
.udp_addresses
.iter()
.map(|addr| addr.trim().to_string())
.filter(|addr| !addr.is_empty())
.collect();
peers.push(FipsPeerConfig {
npub,
udp_addresses,
});
}
for peer_id in discovered_peer_ids {
let npub = peer_id.trim().to_string();
if npub.is_empty() || !seen.insert(npub.clone()) {
continue;
}
peers.push(FipsPeerConfig::new(npub));
}
peers
}
fn normalized_discovery_scope(scope: &str) -> String {
let scope = scope.trim();
if scope.is_empty() {
DEFAULT_FIPS_DISCOVERY_SCOPE.to_string()
} else {
scope.to_string()
}
}
#[cfg(test)]
mod tests {
use super::*;
use hashtree_core::Store;
#[cfg(feature = "experimental-decentralized-pubsub")]
use nostr::{EventBuilder, Kind, Tag};
#[cfg(feature = "experimental-decentralized-pubsub")]
use nostr_pubsub::{PubsubProviderMode, QueryOptions};
use sha2::{Digest, Sha256};
use tokio::time::timeout;
#[test]
fn fips_peer_ids_from_pubkeys_encodes_npbus() {
let keys = nostr::Keys::generate();
let expected = keys.public_key().to_bech32().unwrap();
assert_eq!(
fips_peer_ids_from_pubkeys(vec![keys.public_key().to_bytes()]),
vec![expected]
);
}
#[test]
fn empty_discovery_scope_uses_hashtree_default() {
assert_eq!(
normalized_discovery_scope(" "),
DEFAULT_FIPS_DISCOVERY_SCOPE.to_string()
);
}
#[tokio::test]
async fn fips_local_only_event_transport_requires_local_provider() {
let mut config = Config::default();
config.nostr.event_transport = NostrEventTransport::FipsLocalOnly;
let error = start_daemon_nostr_provider(&config, None)
.await
.err()
.expect("missing local FIPS provider must fail");
assert!(error.to_string().contains("fips-local-only"));
assert!(error.to_string().contains("requires"));
}
#[cfg(feature = "experimental-decentralized-pubsub")]
#[tokio::test]
async fn one_native_endpoint_serves_roots_and_trusted_decentralized_events() {
let scope = format!("htree-native-pubsub-{}", uuid::Uuid::new_v4());
let daemon_addr = reserve_udp_addr();
let rendezvous_addr = reserve_udp_addr();
let (remote_endpoint, remote_addr) = udp_endpoint(&scope).await;
let daemon_keys = nostr::Keys::generate();
let trusted_keys = nostr::Keys::generate();
let untrusted_keys = nostr::Keys::generate();
let outbound_keys = nostr::Keys::generate();
let temp = tempfile::tempdir().unwrap();
let store = Arc::new(HashtreeStore::new(temp.path().join("blobs")).unwrap());
let relay = trusted_test_relay(
temp.path(),
&[trusted_keys.public_key(), outbound_keys.public_key()],
)
.await;
let mut config = Config::default();
config.server.fips_discovery_scope = scope;
config.server.fips_relays = Some(Vec::new());
config.server.fips_udp_bind_addr = Some(daemon_addr.clone());
config.server.fips_local_rendezvous_addr = Some(rendezvous_addr);
config.server.enable_fips_udp = true;
config.server.enable_fips_webrtc = false;
config.server.fips_request_timeout_ms = 2_000;
config.server.fips_peers = vec![crate::config::ConfiguredFipsPeer {
npub: remote_endpoint.local_peer_id.clone(),
udp_addresses: vec![remote_addr],
}];
config.nostr.enabled = true;
config.nostr.event_transport = NostrEventTransport::FipsLocalOnly;
config.nostr.decentralized_pubsub = true;
let daemon = start_daemon_fips_transport(&config, &daemon_keys, store, Vec::new())
.await
.unwrap()
.expect("daemon FIPS endpoint");
let provider = start_daemon_nostr_provider(&config, Some(&daemon))
.await
.unwrap()
.expect("FIPS root provider");
assert_eq!(provider.mode(), PubsubProviderMode::LocalOnly);
let decentralized = start_daemon_nostr_pubsub(&config, Some(&daemon), Some(relay.clone()))
.await
.unwrap()
.expect("decentralized pubsub");
set_fips_peer_configs(
remote_endpoint.native_endpoint.as_ref(),
vec![FipsPeerConfig {
npub: daemon.endpoint_npub.clone(),
udp_addresses: vec![daemon_addr],
}],
)
.await
.unwrap();
let remote_client = Arc::new(
FipsPubsubClient::start(
remote_endpoint.native_endpoint.clone(),
FipsPubsubClientOptions {
query_timeout: Duration::from_secs(2),
..Default::default()
},
)
.await
.unwrap(),
);
wait_for_native_peer(&daemon.endpoint, &remote_endpoint.local_peer_id).await;
wait_for_peer(&remote_endpoint, &daemon.endpoint_npub).await;
timeout(Duration::from_secs(5), async {
while remote_client.peer_subscription_count().unwrap() == 0 {
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.expect("daemon did not subscribe to the authenticated peer");
let published_root = root_event(&outbound_keys, "published-root", "aa");
let mut published_root_rx = remote_client
.subscribe(vec![Filter::new().id(published_root.id)])
.await
.unwrap();
provider
.publish(
VerifiedEvent::try_from(published_root.clone()).unwrap(),
EventSource::local_index("root-publisher"),
)
.await
.unwrap();
let delivered_root = timeout(Duration::from_secs(5), published_root_rx.recv())
.await
.unwrap()
.expect("published root delivery");
assert_eq!(delivered_root.event.as_event().id, published_root.id);
let resolved_root = root_event(&trusted_keys, "resolved-root", "bb");
remote_client
.publish(
VerifiedEvent::try_from(resolved_root.clone()).unwrap(),
EventSource::local_index("remote-root"),
)
.await
.unwrap();
let resolution = provider
.query(
vec![Filter::new().id(resolved_root.id)],
QueryOptions { limit: Some(1) },
)
.await
.unwrap();
assert_eq!(resolution.events.len(), 1);
assert_eq!(resolution.events[0].event.as_event().id, resolved_root.id);
let rejected = EventBuilder::new(Kind::TextNote, "untrusted")
.sign_with_keys(&untrusted_keys)
.unwrap();
remote_client
.publish(
VerifiedEvent::try_from(rejected.clone()).unwrap(),
EventSource::local_index("remote-untrusted"),
)
.await
.unwrap();
let accepted = EventBuilder::new(Kind::TextNote, "trusted")
.sign_with_keys(&trusted_keys)
.unwrap();
remote_client
.publish(
VerifiedEvent::try_from(accepted.clone()).unwrap(),
EventSource::local_index("remote-trusted"),
)
.await
.unwrap();
wait_for_relay_event(&relay, accepted.id).await;
assert!(relay
.query_events(&Filter::new().id(rejected.id), 1)
.await
.is_empty());
let outbound = EventBuilder::new(Kind::TextNote, "relay outbound")
.sign_with_keys(&outbound_keys)
.unwrap();
let mut outbound_rx = remote_client
.subscribe(vec![Filter::new().id(outbound.id)])
.await
.unwrap();
let (client_tx, _client_rx) = mpsc::unbounded_channel();
relay.register_client(1, client_tx, None).await;
relay
.handle_client_message(1, nostr::ClientMessage::event(outbound.clone()))
.await;
let delivered = timeout(Duration::from_secs(5), outbound_rx.recv())
.await
.unwrap()
.expect("outbound relay publication");
assert_eq!(delivered.event.as_event().id, outbound.id);
decentralized.shutdown();
daemon.shutdown().await;
remote_endpoint.native_endpoint.shutdown().await.unwrap();
}
#[tokio::test]
async fn daemon_blob_resolver_advertises_and_serves_storage_router() {
let provider_endpoint = local_only_endpoint("htree-provider-test").await;
let observer_endpoint = local_only_endpoint("iris-drive-observer-test").await;
let temp = tempfile::tempdir().unwrap();
let store = Arc::new(HashtreeStore::new(temp.path()).unwrap());
let data = b"served directly from htree StorageRouter".to_vec();
let hash = Sha256::digest(&data).into();
store.store_arc().put(hash, data.clone()).await.unwrap();
let (blob_resolver, provider) = bind_daemon_blob_resolver(
&provider_endpoint,
store.store_arc(),
&[],
Duration::from_secs(1),
)
.await
.expect("bind htree provider");
let handle = DaemonFipsHandle {
endpoint: provider_endpoint.native_endpoint.clone(),
endpoint_npub: provider_endpoint.local_peer_id.clone(),
discovery_scope: provider_endpoint.discovery_scope.clone(),
nostr_provider: None,
pubsub_client: None,
blob_resolver,
blob_transport: Mutex::new(Some(provider)),
relay_adapter: Mutex::new(None),
};
timeout(Duration::from_secs(10), async {
loop {
if provider_advertised(&observer_endpoint, &provider_endpoint.local_peer_id) {
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.expect("htree provider capability did not converge");
let requester = hashtree_fips_transport::SameHostBlobStore::bind(
observer_endpoint.native_endpoint.clone(),
Arc::new(hashtree_core::MemoryStore::new()),
None,
hashtree_fips_transport::SameHostBlobStoreConfig::default(),
)
.await
.unwrap();
assert_eq!(requester.get(&hash).await.unwrap(), Some(data));
drop(requester);
handle.shutdown().await;
timeout(Duration::from_secs(10), async {
while provider_advertised(&observer_endpoint, &provider_endpoint.local_peer_id) {
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.expect("htree provider capability survived daemon shutdown");
drop(handle);
observer_endpoint.native_endpoint.shutdown().await.unwrap();
}
#[tokio::test]
async fn daemon_blob_resolvers_honor_multi_hop_htl_and_cache() {
let (source_endpoint, source_addr) = udp_endpoint("htree-source-test").await;
let (middle_endpoint, middle_addr) = udp_endpoint("htree-middle-test").await;
let (requester_endpoint, requester_addr) = udp_endpoint("htree-requester-test").await;
let source_dir = tempfile::tempdir().unwrap();
let middle_dir = tempfile::tempdir().unwrap();
let requester_dir = tempfile::tempdir().unwrap();
let source_store = Arc::new(HashtreeStore::new(source_dir.path()).unwrap());
let middle_store = Arc::new(HashtreeStore::new(middle_dir.path()).unwrap());
let requester_store = Arc::new(HashtreeStore::new(requester_dir.path()).unwrap());
let data = b"canonical multi-hop blob route".to_vec();
let hash = Sha256::digest(&data).into();
source_store
.store_arc()
.put(hash, data.clone())
.await
.unwrap();
set_fips_peer_configs(
source_endpoint.native_endpoint.as_ref(),
vec![FipsPeerConfig {
npub: middle_endpoint.local_peer_id.clone(),
udp_addresses: vec![middle_addr.clone()],
}],
)
.await
.unwrap();
set_fips_peer_configs(
middle_endpoint.native_endpoint.as_ref(),
vec![
FipsPeerConfig {
npub: source_endpoint.local_peer_id.clone(),
udp_addresses: vec![source_addr.clone()],
},
FipsPeerConfig {
npub: requester_endpoint.local_peer_id.clone(),
udp_addresses: vec![requester_addr],
},
],
)
.await
.unwrap();
set_fips_peer_configs(
requester_endpoint.native_endpoint.as_ref(),
vec![FipsPeerConfig {
npub: middle_endpoint.local_peer_id.clone(),
udp_addresses: vec![middle_addr.clone()],
}],
)
.await
.unwrap();
let (_source_resolver, source_transport) = bind_daemon_blob_resolver(
&source_endpoint,
source_store.store_arc(),
&[],
Duration::from_secs(2),
)
.await
.unwrap();
let (middle_resolver, middle_transport) = bind_daemon_blob_resolver(
&middle_endpoint,
middle_store.store_arc(),
&[FipsPeerConfig {
npub: source_endpoint.local_peer_id.clone(),
udp_addresses: vec![source_addr],
}],
Duration::from_secs(2),
)
.await
.unwrap();
let (requester_resolver, requester_transport) = bind_daemon_blob_resolver(
&requester_endpoint,
requester_store.store_arc(),
&[FipsPeerConfig {
npub: middle_endpoint.local_peer_id.clone(),
udp_addresses: vec![middle_addr],
}],
Duration::from_secs(2),
)
.await
.unwrap();
wait_for_peer(&requester_endpoint, &middle_endpoint.local_peer_id).await;
wait_for_peer(&middle_endpoint, &source_endpoint.local_peer_id).await;
assert_eq!(
requester_resolver
.route(hashtree_core::BlobRequest { hash, htl: 1 })
.await
.expect("exhausted search"),
hashtree_core::BlobReply::NoResult,
);
assert_eq!(
requester_resolver
.route(hashtree_core::BlobRequest { hash, htl: 2 })
.await
.expect("two-hop search"),
hashtree_core::BlobReply::Data(data.clone()),
);
assert_eq!(
middle_store.store_arc().get(&hash).await.unwrap(),
Some(data.clone())
);
assert_eq!(
requester_store.store_arc().get(&hash).await.unwrap(),
Some(data)
);
drop(requester_transport);
drop(middle_transport);
drop(source_transport);
requester_endpoint.native_endpoint.shutdown().await.unwrap();
middle_endpoint.native_endpoint.shutdown().await.unwrap();
source_endpoint.native_endpoint.shutdown().await.unwrap();
drop(middle_resolver);
}
fn provider_advertised(endpoint: &BoundFipsEndpoint, npub: &str) -> bool {
endpoint
.native_endpoint
.local_instance_advertisements()
.unwrap()
.iter()
.any(|advert| {
advert.npub == npub
&& advert
.capability(hashtree_fips_transport::TCP_BLOB_CAPABILITY)
.and_then(|capability| capability.fsp_port)
== Some(hashtree_fips_transport::TCP_BLOB_SERVICE_PORT)
})
}
async fn local_only_endpoint(scope: &str) -> hashtree_fips_transport::BoundFipsEndpoint {
let keys = nostr::Keys::generate();
let mut options = FipsEndpointOptions::new(keys.secret_key().to_bech32().unwrap());
options.discovery_scope = scope.to_string();
options.enable_udp = false;
options.enable_webrtc = false;
options.enable_local_rendezvous = true;
options.enable_lan_discovery = false;
options.share_local_candidates = false;
bind_fips_endpoint(options).await.unwrap()
}
async fn udp_endpoint(scope: &str) -> (BoundFipsEndpoint, String) {
let addr = reserve_udp_addr();
let keys = nostr::Keys::generate();
let mut options = FipsEndpointOptions::new(keys.secret_key().to_bech32().unwrap());
options.discovery_scope = scope.to_string();
options.enable_udp = true;
options.udp_bind_addr = Some(addr.clone());
options.enable_webrtc = false;
options.enable_local_rendezvous = false;
options.enable_lan_discovery = false;
options.share_local_candidates = false;
let endpoint = bind_fips_endpoint(options).await.unwrap();
(endpoint, addr)
}
fn reserve_udp_addr() -> String {
let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
socket.local_addr().unwrap().to_string()
}
async fn wait_for_peer(endpoint: &BoundFipsEndpoint, npub: &str) {
wait_for_native_peer(endpoint.native_endpoint.as_ref(), npub).await;
}
async fn wait_for_native_peer(endpoint: &FipsEndpoint, npub: &str) {
let result = timeout(Duration::from_secs(10), async {
loop {
if endpoint
.peers()
.await
.unwrap()
.iter()
.any(|peer| peer.npub == npub && peer.connected)
{
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await;
if result.is_err() {
panic!(
"FIPS peer {npub} did not connect; peers={:?}",
endpoint.peers().await
);
}
}
#[cfg(feature = "experimental-decentralized-pubsub")]
async fn trusted_test_relay(
root: &std::path::Path,
allowed: &[nostr::PublicKey],
) -> Arc<NostrRelay> {
let graph_dir = root.join("graph");
let data_dir = root.join("nostr");
std::fs::create_dir_all(&data_dir).unwrap();
let graph = {
let _guard = crate::socialgraph::test_lock().await;
crate::socialgraph::open_test_social_graph_store_with_mapsize(
&graph_dir,
Some(128 * 1024 * 1024),
)
.unwrap()
};
let backend: Arc<dyn crate::socialgraph::SocialGraphBackend> = graph;
let allowed = allowed
.iter()
.map(|pubkey| pubkey.to_hex())
.collect::<HashSet<_>>();
let access = Arc::new(crate::socialgraph::SocialGraphAccessControl::new(
Arc::clone(&backend),
0,
allowed.clone(),
));
Arc::new(
NostrRelay::new(
backend,
data_dir,
allowed,
Some(access),
crate::nostr_relay::NostrRelayConfig {
spambox_db_max_bytes: 0,
..Default::default()
},
)
.unwrap(),
)
}
#[cfg(feature = "experimental-decentralized-pubsub")]
fn root_event(keys: &nostr::Keys, name: &str, hash: &str) -> nostr::Event {
EventBuilder::new(Kind::Custom(30_078), hash)
.tags([
Tag::parse(["d", name]).unwrap(),
Tag::parse(["l", "hashtree"]).unwrap(),
Tag::parse(["hash", hash]).unwrap(),
])
.sign_with_keys(keys)
.unwrap()
}
#[cfg(feature = "experimental-decentralized-pubsub")]
async fn wait_for_relay_event(relay: &NostrRelay, id: nostr::EventId) {
timeout(Duration::from_secs(5), async {
loop {
if relay
.query_events(&Filter::new().id(id), 1)
.await
.iter()
.any(|event| event.id == id)
{
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.expect("trusted decentralized event was not indexed");
}
#[test]
fn daemon_fips_peer_configs_prefer_configured_peers() {
let mut config = Config::default();
config.server.fips_peers = vec![
crate::config::ConfiguredFipsPeer {
npub: " origin ".to_string(),
udp_addresses: vec![" udp:192.0.2.10:2121 ".to_string(), " ".to_string()],
},
crate::config::ConfiguredFipsPeer {
npub: "origin".to_string(),
udp_addresses: vec!["udp:ignored:2121".to_string()],
},
];
let peers = daemon_fips_peer_configs(
&config,
vec![
"origin".to_string(),
" followed ".to_string(),
" ".to_string(),
],
);
assert_eq!(
peers,
vec![
FipsPeerConfig {
npub: "origin".to_string(),
udp_addresses: vec!["udp:192.0.2.10:2121".to_string()],
},
FipsPeerConfig::new("followed"),
]
);
}
#[test]
fn blob_carrier_deadline_outlives_the_complete_resolver_budget() {
for resolver_timeout in [Duration::from_millis(1), Duration::from_secs(20)] {
assert!(blob_transport_config(resolver_timeout).idle_timeout > resolver_timeout);
}
}
#[cfg(feature = "experimental-decentralized-pubsub")]
#[test]
fn daemon_decentralized_pubsub_requires_config_fips_and_relay() {
let mut config = Config::default();
assert!(!daemon_decentralized_pubsub_ready(&config, true, true));
config.nostr.decentralized_pubsub = true;
assert!(daemon_decentralized_pubsub_ready(&config, true, true));
assert!(!daemon_decentralized_pubsub_ready(&config, false, true));
assert!(!daemon_decentralized_pubsub_ready(&config, true, false));
config.nostr.enabled = false;
assert!(!daemon_decentralized_pubsub_ready(&config, true, true));
}
}