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, StoreBlobRoute};
use hashtree_fips_transport::{
bind_fips_endpoint, bind_fips_endpoint_at_local_rendezvous, set_fips_peer_configs,
BoundFipsEndpoint, FipsBlobRoute, FipsEndpoint, FipsEndpointOptions, FipsPeerConfig,
NostrRelayAdapter, PeerIdentity, TcpBlobTransport, TcpBlobTransportConfig,
DEFAULT_FIPS_DISCOVERY_SCOPE,
};
use hashtree_network::{BlobRouteEntry, BlobRouter, BlobRouterConfig};
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 = BlobRouter;
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 = config.server.fips_open_discovery_max_pending;
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 mut routes = vec![BlobRouteEntry::new(
"configured-store",
Arc::new(StoreBlobRoute::new(store.clone())),
)];
let resolver = Arc::new(
BlobRouter::new(
routes.clone(),
Some(store.clone()),
BlobRouterConfig {
request_timeout,
..Default::default()
},
)
.map_err(anyhow::Error::msg)
.context("Failed to configure the daemon blob router")?,
);
let inbound_route: Arc<dyn BlobRoute> = resolver.clone();
let transport = Arc::new(
TcpBlobTransport::bind_advertised_route_with_config(
endpoint.native_endpoint.clone(),
store.clone(),
inbound_route,
blob_transport_config(request_timeout),
DAEMON_SAME_HOST_PROVIDER_PRIORITY,
)
.await
.context("Failed to advertise htree's blob resolver")?,
);
let peer_identities: Vec<_> = peers
.iter()
.filter_map(|peer| PeerIdentity::from_npub(&peer.npub).ok())
.collect();
if !peer_identities.is_empty() {
let max_provider_attempts = peer_identities.len().min(4);
let fips_route =
FipsBlobRoute::explicit(transport.clone(), peer_identities, max_provider_attempts)
.map_err(anyhow::Error::msg)
.context("Failed to configure the daemon FIPS blob route")?;
routes.push(BlobRouteEntry::new(
"configured-fips-peers",
Arc::new(fips_route),
));
}
resolver
.set_routes(routes)
.await
.map_err(anyhow::Error::msg)
.context("Failed to install daemon blob routes")?;
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;
struct RecordingStoreRoute {
store: Arc<hashtree_core::MemoryStore>,
requests: Arc<std::sync::Mutex<Vec<hashtree_core::BlobRequest>>>,
}
#[async_trait::async_trait]
impl BlobRoute for RecordingStoreRoute {
async fn route(
&self,
request: hashtree_core::BlobRequest,
) -> Result<hashtree_core::BlobReply, hashtree_core::StoreError> {
self.requests.lock().unwrap().push(request);
StoreBlobRoute::new(self.store.clone()).route(request).await
}
}
#[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 observer_store = Arc::new(hashtree_core::MemoryStore::new());
let observer_transport = Arc::new(
TcpBlobTransport::bind_route(
observer_endpoint.native_endpoint.clone(),
observer_store.clone(),
Arc::new(StoreBlobRoute::new(observer_store.clone())),
)
.await
.unwrap(),
);
let observer_fips_route = FipsBlobRoute::discovered(
observer_endpoint.native_endpoint.clone(),
observer_transport.clone(),
4,
)
.unwrap();
let observer_router = BlobRouter::new(
vec![BlobRouteEntry::new(
"same-host-fips",
Arc::new(observer_fips_route),
)],
Some(observer_store),
BlobRouterConfig::default(),
)
.unwrap();
assert_eq!(observer_router.get(&hash, None).await.unwrap(), Some(data));
drop(observer_router);
drop(observer_transport);
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_inbound_blob_route_forwards_to_configured_fips_peer() {
let scope = format!("htree-inbound-forwarding-{}", uuid::Uuid::new_v4());
let (upstream_endpoint, upstream_addr) = udp_endpoint(&scope).await;
let (provider_endpoint, provider_addr) = udp_endpoint(&scope).await;
let (observer_endpoint, observer_addr) = udp_endpoint(&scope).await;
let temp = tempfile::tempdir().unwrap();
let provider_store = Arc::new(HashtreeStore::new(temp.path().join("provider")).unwrap());
let upstream_store = Arc::new(hashtree_core::MemoryStore::new());
let data = b"served through the daemon's configured FIPS peer".to_vec();
let hash = Sha256::digest(&data).into();
upstream_store.put(hash, data.clone()).await.unwrap();
let upstream_requests = Arc::new(std::sync::Mutex::new(Vec::new()));
let upstream_transport = Arc::new(
TcpBlobTransport::bind_route(
upstream_endpoint.native_endpoint.clone(),
upstream_store.clone(),
Arc::new(RecordingStoreRoute {
store: upstream_store,
requests: upstream_requests.clone(),
}),
)
.await
.unwrap(),
);
let upstream_peer = FipsPeerConfig {
npub: upstream_endpoint.local_peer_id.clone(),
udp_addresses: vec![upstream_addr],
};
set_fips_peer_configs(
provider_endpoint.native_endpoint.as_ref(),
vec![
upstream_peer.clone(),
FipsPeerConfig {
npub: observer_endpoint.local_peer_id.clone(),
udp_addresses: vec![observer_addr],
},
],
)
.await
.unwrap();
set_fips_peer_configs(
upstream_endpoint.native_endpoint.as_ref(),
vec![FipsPeerConfig {
npub: provider_endpoint.local_peer_id.clone(),
udp_addresses: vec![provider_addr.clone()],
}],
)
.await
.unwrap();
set_fips_peer_configs(
observer_endpoint.native_endpoint.as_ref(),
vec![FipsPeerConfig {
npub: provider_endpoint.local_peer_id.clone(),
udp_addresses: vec![provider_addr],
}],
)
.await
.unwrap();
let (provider_resolver, provider_transport) = bind_daemon_blob_resolver(
&provider_endpoint,
provider_store.store_arc(),
&[upstream_peer],
Duration::from_secs(2),
)
.await
.unwrap();
let observer_store = Arc::new(hashtree_core::MemoryStore::new());
let observer_transport = Arc::new(
TcpBlobTransport::bind_route(
observer_endpoint.native_endpoint.clone(),
observer_store.clone(),
Arc::new(StoreBlobRoute::new(observer_store)),
)
.await
.unwrap(),
);
let observer_route = FipsBlobRoute::explicit(
observer_transport.clone(),
vec![PeerIdentity::from_npub(&provider_endpoint.local_peer_id).unwrap()],
1,
)
.unwrap();
wait_for_native_peer(
provider_endpoint.native_endpoint.as_ref(),
&upstream_endpoint.local_peer_id,
)
.await;
wait_for_native_peer(
observer_endpoint.native_endpoint.as_ref(),
&provider_endpoint.local_peer_id,
)
.await;
assert_eq!(
observer_route
.route(hashtree_core::BlobRequest { hash, htl: 10 })
.await
.unwrap(),
hashtree_core::BlobReply::Data(data),
);
assert_eq!(
upstream_requests.lock().unwrap().as_slice(),
&[hashtree_core::BlobRequest { hash, htl: 10 }],
"FIPS transport hops must preserve the Hashtree HTL",
);
drop(observer_route);
drop(observer_transport);
drop(provider_resolver);
drop(provider_transport);
drop(upstream_transport);
observer_endpoint.native_endpoint.shutdown().await.unwrap();
provider_endpoint.native_endpoint.shutdown().await.unwrap();
upstream_endpoint.native_endpoint.shutdown().await.unwrap();
}
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()
}
#[cfg(feature = "experimental-decentralized-pubsub")]
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));
}
}