hashtree_cli/server/
mod.rs

1mod auth;
2mod blob_read;
3pub mod blossom;
4mod handlers;
5mod ingest_filter;
6mod mime;
7mod nostr_query;
8mod peer_status;
9mod request_paths;
10mod status_metrics;
11#[cfg(feature = "p2p")]
12pub mod stun;
13mod ui;
14pub mod ws_relay;
15
16use crate::nostr_relay::NostrRelay;
17use crate::socialgraph;
18use crate::storage::HashtreeStore;
19use crate::webrtc::WebRTCState;
20use anyhow::Result;
21use axum::{
22    body::Body,
23    extract::DefaultBodyLimit,
24    http::Request,
25    middleware,
26    response::Response,
27    routing::{get, post, put},
28    Router,
29};
30use futures::{future::poll_fn, pin_mut, FutureExt};
31use hashtree_core::Cid;
32use hyper::body::Incoming;
33use hyper_util::{
34    rt::{TokioExecutor, TokioIo, TokioTimer},
35    server::conn::auto::Builder as HyperBuilder,
36    service::TowerToHyperService,
37};
38use socket2::{SockRef, TcpKeepalive};
39use std::collections::{HashMap, HashSet};
40use std::convert::Infallible;
41use std::future;
42use std::io;
43use std::net::SocketAddr;
44use std::sync::{Arc, OnceLock, RwLock};
45use std::time::{Duration, Instant};
46use tokio::sync::watch;
47use tower::{Service, ServiceExt as _};
48use tower_http::cors::CorsLayer;
49use tracing::{debug, error, trace};
50
51pub use auth::{
52    new_lookup_cache, new_upstream_http_client, AppState, AuthCredentials, CachedTreeRootEntry,
53};
54
55static VIRTUAL_TREE_HOSTS: OnceLock<RwLock<HashMap<String, String>>> = OnceLock::new();
56const DEFAULT_OPTIMISTIC_UPLOAD_QUEUE_BYTES: usize = 256 * 1024 * 1024;
57const DEFAULT_BLOSSOM_UPLOAD_REPLICA_QUEUE_BYTES: usize = 256 * 1024 * 1024;
58const INTERNAL_JSON_BODY_LIMIT_BYTES: usize = 64 * 1024;
59
60#[cfg(not(test))]
61const HTTP1_HEADER_READ_TIMEOUT: Duration = Duration::from_secs(30);
62#[cfg(test)]
63const HTTP1_HEADER_READ_TIMEOUT: Duration = Duration::from_millis(200);
64const HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
65const HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(10);
66const TCP_KEEPALIVE_TIME: Duration = Duration::from_secs(60);
67const TCP_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15);
68
69pub fn bounded_upload_queue_bytes(bytes: u64) -> usize {
70    usize::try_from(bytes)
71        .unwrap_or(usize::MAX)
72        .clamp(1, tokio::sync::Semaphore::MAX_PERMITS)
73}
74
75fn virtual_tree_hosts() -> &'static RwLock<HashMap<String, String>> {
76    VIRTUAL_TREE_HOSTS.get_or_init(|| RwLock::new(HashMap::new()))
77}
78
79fn normalize_virtual_tree_host(host: &str) -> Option<String> {
80    let trimmed = host.trim().trim_end_matches('.').to_ascii_lowercase();
81    if trimmed.is_empty() {
82        return None;
83    }
84
85    if let Some(stripped) = trimmed
86        .strip_prefix('[')
87        .and_then(|value| value.split_once(']'))
88    {
89        let host_only = stripped.0.trim();
90        if host_only.is_empty() {
91            return None;
92        }
93        return Some(host_only.to_string());
94    }
95
96    if let Some((host_only, port)) = trimmed.rsplit_once(':') {
97        if !host_only.is_empty() && !port.is_empty() && port.chars().all(|ch| ch.is_ascii_digit()) {
98            return Some(host_only.to_string());
99        }
100    }
101
102    Some(trimmed)
103}
104
105pub fn register_virtual_tree_host(host: &str, internal_root: &str) {
106    let Some(normalized_host) = normalize_virtual_tree_host(host) else {
107        return;
108    };
109
110    let normalized_root = internal_root.trim().trim_end_matches('/');
111    if normalized_root.is_empty() {
112        return;
113    }
114
115    if let Ok(mut hosts) = virtual_tree_hosts().write() {
116        hosts.insert(normalized_host, normalized_root.to_string());
117    }
118}
119
120pub fn resolve_virtual_tree_host(host: &str) -> Option<String> {
121    let normalized_host = normalize_virtual_tree_host(host)?;
122    let configured = virtual_tree_hosts()
123        .read()
124        .ok()
125        .and_then(|hosts| hosts.get(&normalized_host).cloned());
126    configured.or_else(|| resolve_iris_localhost_tree_root(&normalized_host))
127}
128
129fn resolve_iris_localhost_tree_root(host: &str) -> Option<String> {
130    let labels: Vec<&str> = host.strip_suffix(".iris.localhost")?.split('.').collect();
131    match labels.as_slice() {
132        [nhash] if nhash.starts_with("nhash1") => Some(format!("/htree/{nhash}")),
133        [site, npub] if !site.is_empty() && npub.starts_with("npub1") => {
134            Some(format!("/htree/{npub}/{site}"))
135        }
136        _ => None,
137    }
138}
139
140#[cfg(test)]
141pub fn clear_virtual_tree_hosts_for_test() {
142    if let Ok(mut hosts) = virtual_tree_hosts().write() {
143        hosts.clear();
144    }
145}
146
147pub struct HashtreeServer {
148    state: AppState,
149    addr: String,
150    extra_routes: Option<Router<AppState>>,
151    cors: Option<CorsLayer>,
152}
153
154impl HashtreeServer {
155    pub fn new(store: Arc<HashtreeStore>, addr: String) -> Self {
156        Self {
157            state: AppState {
158                store,
159                auth: None,
160                daemon_started_at: current_unix_secs(),
161                peer_mode: crate::config::ServerMode::Normal,
162                hash_get_enabled: true,
163                http_webrtc_fetch: true,
164                webrtc_peers: None,
165                fips_endpoint: None,
166                fips_blob_resolver: None,
167                fetch_from_fips_peers: true,
168                ws_relay: Arc::new(auth::WsRelayState::new()),
169                max_upload_bytes: 5 * 1024 * 1024, // 5 MB default
170                public_writes: true,               // Allow anyone with valid Nostr auth by default
171                public_plaintext_reads: false,
172                require_random_untrusted_ingest: true,
173                optimistic_blossom_uploads: false,
174                optimistic_upload_queue_bytes: DEFAULT_OPTIMISTIC_UPLOAD_QUEUE_BYTES,
175                optimistic_upload_queue: Arc::new(tokio::sync::Semaphore::new(
176                    DEFAULT_OPTIMISTIC_UPLOAD_QUEUE_BYTES,
177                )),
178                allowed_pubkeys: HashSet::new(), // No pubkeys allowed by default (use public_writes)
179                upstream_blossom: Vec::new(),
180                upstream_http_client: new_upstream_http_client(),
181                upstream_blossom_miss_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
182                upstream_blossom_fetch_metrics: Arc::new(
183                    auth::UpstreamBlossomFetchMetrics::default(),
184                ),
185                blossom_upload_replicas: Vec::new(),
186                blossom_upload_replica_queue_bytes: DEFAULT_BLOSSOM_UPLOAD_REPLICA_QUEUE_BYTES,
187                blossom_upload_replica_queue: Arc::new(tokio::sync::Semaphore::new(
188                    DEFAULT_BLOSSOM_UPLOAD_REPLICA_QUEUE_BYTES,
189                )),
190                blossom_upload_replica_keys: None,
191                blossom_upload_replica_scheduler: Arc::new(
192                    blossom::BlossomUploadReplicaScheduler::new(),
193                ),
194                social_graph: None,
195                social_graph_store: None,
196                social_graph_root: None,
197                socialgraph_snapshot_public: false,
198                nostr_relay: None,
199                nostr_provider: None,
200                nostr_relay_urls: Vec::new(),
201                tree_root_cache: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
202                inflight_blob_fetches: Arc::new(tokio::sync::Mutex::new(
203                    std::collections::HashMap::new(),
204                )),
205                inflight_blob_reads: Arc::new(tokio::sync::Mutex::new(
206                    std::collections::HashMap::new(),
207                )),
208                blob_cache: Arc::new(crate::blob_cache::BlobCache::from_env()),
209                directory_listing_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
210                resolved_path_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
211                thumbnail_path_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
212                cid_size_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
213            },
214            addr,
215            extra_routes: None,
216            cors: None,
217        }
218    }
219
220    /// Set maximum upload size for Blossom uploads
221    pub fn with_max_upload_bytes(mut self, bytes: usize) -> Self {
222        self.state.max_upload_bytes = bytes;
223        self
224    }
225
226    /// Set whether to allow public writes (anyone with valid Nostr auth)
227    /// When false, only social graph members can write
228    pub fn with_public_writes(mut self, public: bool) -> Self {
229        self.state.public_writes = public;
230        self
231    }
232
233    /// Set whether mutable npub routes serve plaintext for unapproved pubkeys
234    pub fn with_public_plaintext_reads(mut self, public: bool) -> Self {
235        self.state.public_plaintext_reads = public;
236        self
237    }
238
239    pub fn with_require_random_untrusted_ingest(mut self, require: bool) -> Self {
240        self.state.require_random_untrusted_ingest = require;
241        self
242    }
243
244    pub fn with_optimistic_blossom_uploads(mut self, enabled: bool) -> Self {
245        self.state.optimistic_blossom_uploads = enabled;
246        self
247    }
248
249    pub fn with_server_mode(mut self, mode: crate::config::ServerMode) -> Self {
250        self.state.peer_mode = mode;
251        self
252    }
253
254    pub fn with_hash_get_enabled(mut self, enabled: bool) -> Self {
255        self.state.hash_get_enabled = enabled;
256        self
257    }
258
259    pub fn with_http_webrtc_fetch(mut self, enabled: bool) -> Self {
260        self.state.http_webrtc_fetch = enabled;
261        self
262    }
263
264    pub fn with_fetch_from_fips_peers(mut self, enabled: bool) -> Self {
265        self.state.fetch_from_fips_peers = enabled;
266        self
267    }
268
269    pub fn with_fips_endpoint(
270        mut self,
271        endpoint: Arc<hashtree_fips_transport::FipsEndpoint>,
272    ) -> Self {
273        self.state.fips_endpoint = Some(endpoint);
274        self
275    }
276
277    pub fn with_fips_blob_resolver(
278        mut self,
279        resolver: Arc<crate::fips_transport::DaemonBlobResolver>,
280    ) -> Self {
281        self.state.fips_blob_resolver = Some(resolver);
282        self
283    }
284
285    /// Set WebRTC state for P2P peer queries
286    pub fn with_webrtc_peers(mut self, webrtc_state: Arc<WebRTCState>) -> Self {
287        self.state.webrtc_peers = Some(webrtc_state);
288        self
289    }
290
291    pub fn with_auth(mut self, username: String, password: String) -> Self {
292        self.state.auth = Some(AuthCredentials { username, password });
293        self
294    }
295
296    /// Set allowed pubkeys for blossom write access (hex format)
297    pub fn with_allowed_pubkeys(mut self, pubkeys: HashSet<String>) -> Self {
298        self.state.allowed_pubkeys = pubkeys;
299        self
300    }
301
302    /// Set upstream Blossom servers for cascade fetching
303    pub fn with_upstream_blossom(mut self, servers: Vec<String>) -> Self {
304        self.state.upstream_blossom = servers;
305        self
306    }
307
308    /// Set write-behind Blossom servers for blobs accepted by this server.
309    pub fn with_blossom_upload_replicas(
310        mut self,
311        servers: Vec<String>,
312        queue_bytes: usize,
313        keys: nostr::Keys,
314    ) -> Self {
315        let queue_bytes = queue_bytes.clamp(1, tokio::sync::Semaphore::MAX_PERMITS);
316        let mut servers: Vec<String> = servers
317            .into_iter()
318            .map(|server| server.trim().trim_end_matches('/').to_string())
319            .filter(|server| !server.is_empty())
320            .collect();
321        servers.sort();
322        servers.dedup();
323        let replica_keys = (!servers.is_empty()).then(|| Arc::new(keys));
324        self.state.blossom_upload_replicas = servers;
325        self.state.blossom_upload_replica_queue_bytes = queue_bytes;
326        self.state.blossom_upload_replica_queue =
327            Arc::new(tokio::sync::Semaphore::new(queue_bytes));
328        self.state.blossom_upload_replica_keys = replica_keys;
329        self
330    }
331
332    /// Set social graph access control
333    pub fn with_social_graph(mut self, sg: Arc<socialgraph::SocialGraphAccessControl>) -> Self {
334        self.state.social_graph = Some(sg);
335        self
336    }
337
338    /// Configure social graph snapshot export (store handle + root)
339    pub fn with_socialgraph_snapshot(
340        mut self,
341        store: Arc<dyn socialgraph::SocialGraphBackend>,
342        root: [u8; 32],
343        public: bool,
344    ) -> Self {
345        self.state.social_graph_store = Some(store);
346        self.state.social_graph_root = Some(root);
347        self.state.socialgraph_snapshot_public = public;
348        self
349    }
350
351    /// Set Nostr relay state (shared for /ws and WebRTC)
352    pub fn with_nostr_relay(mut self, relay: Arc<NostrRelay>) -> Self {
353        self.state.nostr_relay = Some(relay);
354        self
355    }
356
357    pub fn with_nostr_provider(mut self, provider: Arc<dyn nostr_pubsub::PubsubProvider>) -> Self {
358        self.state.nostr_provider = Some(provider);
359        self
360    }
361
362    /// Set active upstream Nostr relays for HTTP resolver operations.
363    pub fn with_nostr_relay_urls(mut self, relays: Vec<String>) -> Self {
364        self.state.nostr_relay_urls = relays;
365        self
366    }
367
368    /// Seed mutable root cache entries before the server starts.
369    pub fn with_cached_tree_roots(self, roots: Vec<(String, Cid)>) -> Self {
370        if let Ok(mut cache) = self.state.tree_root_cache.lock() {
371            let now = Instant::now();
372            for (key, cid) in roots {
373                cache.insert(
374                    key,
375                    CachedTreeRootEntry {
376                        cid,
377                        source: "embedded-bootstrap",
378                        root_event: None,
379                        event: None,
380                        cached_at: now,
381                    },
382                );
383            }
384        }
385        self
386    }
387
388    /// Merge extra routes into the daemon router (e.g. Tauri embeds /nip07).
389    pub fn with_extra_routes(mut self, routes: Router<AppState>) -> Self {
390        self.extra_routes = Some(routes);
391        self
392    }
393
394    /// Apply a CORS layer to all routes (used by embedded clients like Tauri).
395    pub fn with_cors(mut self, cors: CorsLayer) -> Self {
396        self.cors = Some(cors);
397        self
398    }
399
400    pub async fn run(self) -> Result<()> {
401        let listener = tokio::net::TcpListener::bind(&self.addr).await?;
402        let _ = self.run_with_listener(listener).await?;
403        Ok(())
404    }
405
406    pub async fn run_with_listener(self, listener: tokio::net::TcpListener) -> Result<u16> {
407        self.run_with_listener_until(listener, future::pending::<()>())
408            .await
409    }
410
411    pub async fn run_with_listener_until<F>(
412        self,
413        listener: tokio::net::TcpListener,
414        shutdown: F,
415    ) -> Result<u16>
416    where
417        F: std::future::Future<Output = ()> + Send + 'static,
418    {
419        let local_addr = listener.local_addr()?;
420
421        // Public endpoints (no auth required)
422        // Note: /:id serves raw SHA256 blobs only. Logical tree/file assembly
423        // stays on explicitly configured mutable routes such as approved npub.
424        let state = self.state.clone();
425        let public_routes = Router::new()
426            .route("/", get(handlers::serve_root_or_virtual_host))
427            .route("/ws", get(ws_relay::ws_data))
428            .route("/ws/", get(ws_relay::ws_data))
429            .route(
430                "/__iris/store/:hash",
431                get(handlers::iris_store_get).head(handlers::iris_store_head),
432            )
433            .route(
434                "/htree/test",
435                get(handlers::htree_test).head(handlers::htree_test),
436            )
437            // /htree/nhash1...[/path] - content-addressed (immutable)
438            .route("/htree/nhash1:nhash", get(handlers::htree_nhash))
439            .route("/htree/nhash1:nhash/", get(handlers::htree_nhash))
440            .route("/htree/nhash1:nhash/*path", get(handlers::htree_nhash_path))
441            // /htree/npub1.../tree[/path] - mutable (resolver-backed)
442            .route("/htree/npub1:npub/:treename", get(handlers::htree_npub))
443            .route("/htree/npub1:npub/:treename/", get(handlers::htree_npub))
444            .route(
445                "/htree/npub1:npub/:treename/*path",
446                get(handlers::htree_npub_path),
447            )
448            // Nostr resolver endpoints - resolve npub/treename to content
449            .route("/n/:pubkey/:treename", get(handlers::resolve_and_serve))
450            // Direct npub route (clients should parse nhash and request by hex hash)
451            .route("/npub1:rest", get(handlers::serve_npub))
452            .route("/npub1:rest/*path", get(handlers::serve_npub))
453            // Blossom endpoints (BUD-01, BUD-02)
454            .route(
455                "/:id",
456                get(handlers::serve_content_or_blob)
457                    .head(blossom::head_blob)
458                    .delete(blossom::delete_blob)
459                    .options(blossom::cors_preflight),
460            )
461            .route(
462                "/upload",
463                put(blossom::upload_blob)
464                    .layer::<_, std::convert::Infallible>(middleware::from_fn(
465                        blossom::require_upload_auth_middleware,
466                    ))
467                    .layer(DefaultBodyLimit::max(blossom::MAX_SINGLE_UPLOAD_BODY_BYTES))
468                    .head(blossom::head_upload)
469                    .options(blossom::cors_preflight),
470            )
471            .route(
472                "/upload/batch",
473                post(blossom::upload_blob_batch)
474                    .layer::<_, std::convert::Infallible>(middleware::from_fn(
475                        blossom::require_upload_auth_middleware,
476                    ))
477                    .options(blossom::cors_preflight)
478                    .layer(DefaultBodyLimit::max(
479                        blossom::MAX_BATCH_UPLOAD_JSON_BODY_BYTES,
480                    )),
481            )
482            .route(
483                "/upload/batch-binary",
484                post(blossom::upload_blob_batch_binary)
485                    .layer::<_, std::convert::Infallible>(middleware::from_fn(
486                        blossom::require_upload_auth_middleware,
487                    ))
488                    .options(blossom::cors_preflight)
489                    .layer(DefaultBodyLimit::max(
490                        blossom::MAX_BATCH_UPLOAD_BINARY_BODY_BYTES,
491                    )),
492            )
493            .route(
494                "/upload/check",
495                post(blossom::upload_check).options(blossom::cors_preflight),
496            )
497            .route(
498                "/blob/batch",
499                post(handlers::download_blob_batch).options(blossom::cors_preflight),
500            )
501            .route(
502                "/list/:pubkey",
503                get(blossom::list_blobs).options(blossom::cors_preflight),
504            )
505            // Hashtree API endpoints
506            .route("/health", get(handlers::health_check))
507            .route("/api/pins", get(handlers::list_pins))
508            .route("/api/stats", get(handlers::storage_stats))
509            .route("/api/peers", get(handlers::webrtc_peers))
510            .route("/api/status", get(handlers::daemon_status))
511            .route("/api/p2p/signal", post(handlers::p2p_signal))
512            .route("/api/socialgraph", get(handlers::socialgraph_stats))
513            .route(
514                "/api/socialgraph/snapshot",
515                get(handlers::socialgraph_snapshot),
516            )
517            .route(
518                "/api/socialgraph/distance/:pubkey",
519                get(handlers::follow_distance),
520            )
521            // Resolver API endpoints
522            .route(
523                "/api/resolve/:pubkey/:treename",
524                get(handlers::resolve_to_hash),
525            )
526            .route(
527                "/api/nostr/resolve/:pubkey/:treename",
528                get(handlers::resolve_to_hash),
529            )
530            .route("/api/nostr/profile/:pubkey", get(handlers::nostr_profile))
531            .route(
532                "/api/nostr/events",
533                post(handlers::publish_nostr_event)
534                    .layer(DefaultBodyLimit::max(INTERNAL_JSON_BODY_LIMIT_BYTES)),
535            )
536            .route("/api/trees/:pubkey", get(handlers::list_trees))
537            .fallback(get(handlers::serve_virtual_host_fallback))
538            .with_state(state.clone());
539
540        // Protected endpoints (require auth if enabled)
541        let protected_routes = Router::new()
542            .route("/upload", post(handlers::upload_file))
543            .route("/api/pin/:cid", post(handlers::pin_cid))
544            .route("/api/unpin/:cid", post(handlers::unpin_cid))
545            .route("/api/gc", post(handlers::garbage_collect))
546            .layer(middleware::from_fn_with_state(
547                state.clone(),
548                auth::auth_middleware,
549            ))
550            .with_state(state.clone());
551
552        // Internal mutating endpoints require configured Basic auth. These
553        // routes stay closed even when optional API auth is disabled.
554        let internal_routes = Router::new()
555            .route(
556                "/__iris/store/:hash",
557                put(handlers::iris_store_put)
558                    .delete(handlers::iris_store_delete)
559                    .layer(DefaultBodyLimit::max(blossom::MAX_BATCH_UPLOAD_BYTES)),
560            )
561            .route(
562                "/api/pin-tree",
563                post(handlers::pin_tree)
564                    .layer(DefaultBodyLimit::max(INTERNAL_JSON_BODY_LIMIT_BYTES)),
565            )
566            .route(
567                "/api/cache-tree-root",
568                post(handlers::cache_tree_root)
569                    .layer(DefaultBodyLimit::max(INTERNAL_JSON_BODY_LIMIT_BYTES)),
570            )
571            .route(
572                "/api/clear-tree-root-cache",
573                post(handlers::clear_tree_root_cache)
574                    .layer(DefaultBodyLimit::max(INTERNAL_JSON_BODY_LIMIT_BYTES)),
575            )
576            .layer(middleware::from_fn_with_state(
577                state.clone(),
578                auth::require_auth_middleware,
579            ))
580            .with_state(state.clone());
581
582        let mut app = public_routes
583            .merge(protected_routes)
584            .merge(internal_routes)
585            .layer(DefaultBodyLimit::max(10 * 1024 * 1024 * 1024)) // 10GB limit
586            .layer(middleware::from_fn(status_metrics::record_http_status));
587
588        if let Some(extra) = self.extra_routes {
589            app = app.merge(extra.with_state(state));
590        }
591
592        if let Some(cors) = self.cors {
593            app = app.layer(cors);
594        }
595
596        let make_service = app.into_make_service_with_connect_info::<std::net::SocketAddr>();
597        serve_with_connection_limits(listener, make_service, shutdown).await?;
598
599        Ok(local_addr.port())
600    }
601
602    pub fn addr(&self) -> &str {
603        &self.addr
604    }
605}
606
607async fn serve_with_connection_limits<M, S, F>(
608    listener: tokio::net::TcpListener,
609    mut make_service: M,
610    shutdown: F,
611) -> io::Result<()>
612where
613    M: Service<SocketAddr, Error = Infallible, Response = S> + Send + 'static,
614    M::Future: Send,
615    S: Service<Request<Body>, Response = Response, Error = Infallible> + Clone + Send + 'static,
616    S::Future: Send,
617    F: std::future::Future<Output = ()> + Send + 'static,
618{
619    let (signal_tx, signal_rx) = watch::channel(());
620    let signal_tx = Arc::new(signal_tx);
621    tokio::spawn(async move {
622        shutdown.await;
623        trace!("received graceful shutdown signal; stopping daemon listener");
624        drop(signal_rx);
625    });
626
627    let (close_tx, close_rx) = watch::channel(());
628
629    loop {
630        let (tcp_stream, remote_addr) = tokio::select! {
631            accepted = accept_tcp(&listener) => {
632                match accepted? {
633                    Some(connection) => connection,
634                    None => continue,
635                }
636            }
637            _ = signal_tx.closed() => {
638                trace!("shutdown signal received; no longer accepting daemon connections");
639                break;
640            }
641        };
642
643        configure_tcp_stream(&tcp_stream);
644        let tcp_stream = TokioIo::new(tcp_stream);
645
646        poll_fn(|cx| make_service.poll_ready(cx))
647            .await
648            .unwrap_or_else(|err| match err {});
649
650        let tower_service = make_service
651            .call(remote_addr)
652            .await
653            .unwrap_or_else(|err| match err {})
654            .map_request(|req: Request<Incoming>| req.map(Body::new));
655        let hyper_service = TowerToHyperService::new(tower_service);
656
657        let signal_tx = Arc::clone(&signal_tx);
658        let close_rx = close_rx.clone();
659
660        tokio::spawn(async move {
661            let mut builder = HyperBuilder::new(TokioExecutor::new());
662            builder
663                .http1()
664                .timer(TokioTimer::new())
665                .header_read_timeout(HTTP1_HEADER_READ_TIMEOUT);
666            builder
667                .http2()
668                .timer(TokioTimer::new())
669                .keep_alive_interval(Some(HTTP2_KEEPALIVE_INTERVAL))
670                .keep_alive_timeout(HTTP2_KEEPALIVE_TIMEOUT);
671
672            let conn = builder.serve_connection_with_upgrades(tcp_stream, hyper_service);
673            pin_mut!(conn);
674
675            let signal_closed = signal_tx.closed().fuse();
676            pin_mut!(signal_closed);
677
678            loop {
679                tokio::select! {
680                    result = conn.as_mut() => {
681                        if let Err(err) = result {
682                            trace!("daemon connection closed with error: {err:#}");
683                        }
684                        break;
685                    }
686                    _ = &mut signal_closed => {
687                        trace!("shutdown signal received by connection task");
688                        conn.as_mut().graceful_shutdown();
689                    }
690                }
691            }
692
693            drop(close_rx);
694        });
695    }
696
697    drop(close_rx);
698    drop(listener);
699    close_tx.closed().await;
700
701    Ok(())
702}
703
704fn configure_tcp_stream(tcp_stream: &tokio::net::TcpStream) {
705    if let Err(err) = tcp_stream.set_nodelay(true) {
706        debug!("failed to set TCP_NODELAY on daemon connection: {err:#}");
707    }
708
709    let socket = SockRef::from(tcp_stream);
710    if let Err(err) = socket.set_tcp_keepalive(
711        &TcpKeepalive::new()
712            .with_time(TCP_KEEPALIVE_TIME)
713            .with_interval(TCP_KEEPALIVE_INTERVAL),
714    ) {
715        debug!("failed to set TCP keepalive on daemon connection: {err:#}");
716    }
717}
718
719async fn accept_tcp(
720    listener: &tokio::net::TcpListener,
721) -> io::Result<Option<(tokio::net::TcpStream, SocketAddr)>> {
722    match listener.accept().await {
723        Ok(connection) => Ok(Some(connection)),
724        Err(err) => {
725            if is_connection_error(&err) {
726                return Ok(None);
727            }
728            if is_resource_exhaustion_error(&err) {
729                error!(
730                    "daemon accept failed due to file descriptor exhaustion; exiting for supervisor restart: {err}"
731                );
732                return Err(err);
733            }
734            error!("daemon accept error: {err}");
735            tokio::time::sleep(Duration::from_secs(1)).await;
736            Ok(None)
737        }
738    }
739}
740
741fn is_resource_exhaustion_error(err: &io::Error) -> bool {
742    matches!(
743        err.raw_os_error(),
744        Some(code) if code == libc::EMFILE || code == libc::ENFILE
745    )
746}
747
748fn is_connection_error(err: &io::Error) -> bool {
749    matches!(
750        err.kind(),
751        io::ErrorKind::ConnectionRefused
752            | io::ErrorKind::ConnectionAborted
753            | io::ErrorKind::ConnectionReset
754    )
755}
756
757fn current_unix_secs() -> u64 {
758    std::time::SystemTime::now()
759        .duration_since(std::time::UNIX_EPOCH)
760        .unwrap_or(std::time::Duration::ZERO)
761        .as_secs()
762}
763
764#[cfg(test)]
765mod tests {
766    use super::*;
767    use crate::nostr_relay::{NostrRelay, NostrRelayConfig};
768    use crate::storage::HashtreeStore;
769    use async_trait::async_trait;
770    use hashtree_core::{
771        from_hex, nhash_encode, nhash_encode_full, sha256, to_hex, DirEntry, HashTree,
772        HashTreeConfig, LinkType, NHashData,
773    };
774    use nostr::{nips::nip19::ToBech32, EventBuilder, Keys, Kind, Timestamp};
775    use nostr_pubsub::{
776        EventBus, EventSource, PublishReport, PubsubProvider, PubsubProviderMode, QueryEvent,
777        QueryOptions, QueryReport, VerifiedEvent,
778    };
779    use serde_json::json;
780    use std::sync::atomic::{AtomicUsize, Ordering};
781    use tempfile::TempDir;
782    use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
783
784    struct StaticProvider {
785        event: Option<VerifiedEvent>,
786        queries: AtomicUsize,
787        publishes: AtomicUsize,
788        mode: PubsubProviderMode,
789    }
790
791    #[async_trait]
792    impl EventBus for StaticProvider {
793        async fn publish(
794            &self,
795            _event: VerifiedEvent,
796            _source: EventSource,
797        ) -> nostr_pubsub::Result<PublishReport> {
798            self.publishes.fetch_add(1, Ordering::Relaxed);
799            Ok(PublishReport {
800                accepted: true,
801                priority: 0,
802                reason: None,
803            })
804        }
805
806        async fn query(
807            &self,
808            _filters: Vec<nostr_pubsub::Filter>,
809            _options: QueryOptions,
810        ) -> nostr_pubsub::Result<QueryReport> {
811            self.queries.fetch_add(1, Ordering::Relaxed);
812            Ok(QueryReport {
813                events: self
814                    .event
815                    .clone()
816                    .map(|event| QueryEvent {
817                        event,
818                        source: EventSource::fips_endpoint("browser-router"),
819                        priority: 0,
820                    })
821                    .into_iter()
822                    .collect(),
823            })
824        }
825    }
826
827    impl PubsubProvider for StaticProvider {
828        fn mode(&self) -> PubsubProviderMode {
829            self.mode
830        }
831    }
832
833    #[test]
834    fn resource_exhaustion_errors_are_fatal_accept_errors() {
835        assert!(is_resource_exhaustion_error(&io::Error::from_raw_os_error(
836            libc::EMFILE
837        )));
838        assert!(is_resource_exhaustion_error(&io::Error::from_raw_os_error(
839            libc::ENFILE
840        )));
841        assert!(!is_resource_exhaustion_error(
842            &io::Error::from_raw_os_error(libc::ECONNRESET)
843        ));
844    }
845
846    #[test]
847    fn upload_queue_semaphores_fit_all_tokio_targets() -> Result<()> {
848        let temp_dir = TempDir::new()?;
849        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
850        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string());
851
852        assert_eq!(
853            server.state.optimistic_upload_queue.available_permits(),
854            256 * 1024 * 1024
855        );
856        assert_eq!(
857            server
858                .state
859                .blossom_upload_replica_queue
860                .available_permits(),
861            256 * 1024 * 1024
862        );
863        assert_eq!(
864            bounded_upload_queue_bytes(u64::MAX),
865            tokio::sync::Semaphore::MAX_PERMITS
866        );
867        Ok(())
868    }
869
870    #[test]
871    fn server_builder_seeds_initial_tree_roots() -> Result<()> {
872        let temp_dir = TempDir::new()?;
873        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
874        let hash = from_hex("1111111111111111111111111111111111111111111111111111111111111111")?;
875        let key = from_hex("2222222222222222222222222222222222222222222222222222222222222222")?;
876        let cid = Cid {
877            hash,
878            key: Some(key),
879        };
880
881        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
882            .with_cached_tree_roots(vec![("npub1example/sites".to_string(), cid.clone())]);
883        let cached = server
884            .state
885            .tree_root_cache
886            .lock()
887            .unwrap()
888            .get("npub1example/sites")
889            .cloned()
890            .expect("seeded root");
891
892        assert_eq!(cached.cid, cid);
893        assert_eq!(cached.source, "embedded-bootstrap");
894        assert!(cached.root_event.is_none());
895        Ok(())
896    }
897
898    #[test]
899    fn iris_localhost_hosts_map_to_existing_tree_routes() {
900        assert_eq!(
901            resolve_virtual_tree_host("NHASH1EXAMPLE.iris.localhost:8080"),
902            Some("/htree/nhash1example".to_string())
903        );
904        assert_eq!(
905            resolve_virtual_tree_host("audio.NPUB1EXAMPLE.iris.localhost:8080"),
906            Some("/htree/npub1example/audio".to_string())
907        );
908        assert_eq!(
909            resolve_virtual_tree_host("audio.extra.npub1example.iris.localhost"),
910            None
911        );
912        assert_eq!(
913            resolve_virtual_tree_host("nhash1example.htree.localhost"),
914            None
915        );
916    }
917
918    #[tokio::test]
919    async fn test_server_serve_file() -> Result<()> {
920        let temp_dir = TempDir::new()?;
921        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
922
923        // Create and upload a test file
924        let test_file = temp_dir.path().join("test.txt");
925        std::fs::write(&test_file, b"Hello, Hashtree!")?;
926
927        let cid = store.upload_file(&test_file)?;
928        let hash = from_hex(&cid)?;
929
930        // Verify we can get it
931        let content = store.get_file(&hash)?;
932        assert!(content.is_some());
933        assert_eq!(content.unwrap(), b"Hello, Hashtree!");
934
935        Ok(())
936    }
937
938    #[tokio::test]
939    async fn test_server_list_pins() -> Result<()> {
940        let temp_dir = TempDir::new()?;
941        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
942
943        let test_file = temp_dir.path().join("test.txt");
944        std::fs::write(&test_file, b"Test")?;
945
946        let cid = store.upload_file(&test_file)?;
947        let hash = from_hex(&cid)?;
948
949        let pins = store.list_pins_raw()?;
950        assert_eq!(pins.len(), 1);
951        assert_eq!(pins[0], hash);
952
953        Ok(())
954    }
955
956    async fn spawn_test_server(
957        store: Arc<HashtreeStore>,
958    ) -> Result<(u16, tokio::task::JoinHandle<Result<()>>)> {
959        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
960        let port = listener.local_addr()?.port();
961        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string());
962        let handle =
963            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
964        Ok((port, handle))
965    }
966
967    async fn spawn_test_server_with_auth(
968        store: Arc<HashtreeStore>,
969    ) -> Result<(u16, tokio::task::JoinHandle<Result<()>>)> {
970        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
971        let port = listener.local_addr()?.port();
972        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
973            .with_auth("test-user".to_string(), "test-password".to_string());
974        let handle =
975            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
976        Ok((port, handle))
977    }
978
979    async fn encrypted_test_directory(store: &HashtreeStore) -> Result<(Cid, Cid, Vec<u8>)> {
980        let tree = HashTree::new(
981            HashTreeConfig::new(store.store_arc())
982                .with_chunk_size(4)
983                .with_max_links(2),
984        );
985        let content = b"encrypted descendant content spanning chunks".to_vec();
986        let (file_cid, file_size) = tree.put(&content).await?;
987        let nested_cid = tree
988            .put_directory(vec![DirEntry::from_cid("post.json", &file_cid)
989                .with_size(file_size)
990                .with_link_type(LinkType::File)])
991            .await?;
992        let root_cid = tree
993            .put_directory(vec![
994                DirEntry::from_cid("events", &nested_cid).with_link_type(LinkType::Dir)
995            ])
996            .await?;
997        Ok((root_cid, file_cid, content))
998    }
999
1000    async fn spawn_test_server_with_nostr_relay(
1001        store: Arc<HashtreeStore>,
1002        relay: Arc<NostrRelay>,
1003    ) -> Result<(u16, tokio::task::JoinHandle<Result<()>>)> {
1004        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1005        let port = listener.local_addr()?.port();
1006        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string()).with_nostr_relay(relay);
1007        let handle =
1008            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1009        Ok((port, handle))
1010    }
1011
1012    #[tokio::test]
1013    async fn unauthenticated_native_store_mutation_is_rejected() -> Result<()> {
1014        let temp_dir = TempDir::new()?;
1015        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1016        let body = b"unauthorized native store write";
1017        let hash = sha256(body);
1018        let hash_hex = to_hex(&hash);
1019        let (port, handle) = spawn_test_server(Arc::clone(&store)).await?;
1020
1021        let response = reqwest::Client::new()
1022            .put(format!("http://127.0.0.1:{port}/__iris/store/{hash_hex}"))
1023            .body(body.to_vec())
1024            .send()
1025            .await?;
1026
1027        assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
1028        assert!(store.get_blob(&hash)?.is_none());
1029        handle.abort();
1030        Ok(())
1031    }
1032
1033    #[tokio::test]
1034    async fn unauthenticated_cache_tree_root_is_rejected() -> Result<()> {
1035        let temp_dir = TempDir::new()?;
1036        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1037        let (port, handle) = spawn_test_server(store).await?;
1038
1039        let response = reqwest::Client::new()
1040            .post(format!("http://127.0.0.1:{port}/api/cache-tree-root"))
1041            .json(&json!({
1042                "npub": "npub1example",
1043                "treeName": "video",
1044                "hash": "988db3f24dc222715f1c1e1fa5876690d3147122243d72d85fd44283867cd61a",
1045                "visibility": "public"
1046            }))
1047            .send()
1048            .await?;
1049
1050        assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
1051        handle.abort();
1052        Ok(())
1053    }
1054
1055    #[tokio::test]
1056    async fn pin_tree_requires_configured_valid_auth() -> Result<()> {
1057        let temp_dir = TempDir::new()?;
1058        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1059        let nhash = nhash_encode(&sha256(b"missing"))?;
1060        let client = reqwest::Client::new();
1061
1062        let (port, handle) = spawn_test_server(Arc::clone(&store)).await?;
1063        let response = client
1064            .post(format!("http://127.0.0.1:{port}/api/pin-tree"))
1065            .json(&json!({"nhash": nhash}))
1066            .send()
1067            .await?;
1068        assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
1069        handle.abort();
1070
1071        let (port, handle) = spawn_test_server_with_auth(store).await?;
1072        let response = client
1073            .post(format!("http://127.0.0.1:{port}/api/pin-tree"))
1074            .json(&json!({"nhash": nhash}))
1075            .send()
1076            .await?;
1077        assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);
1078        handle.abort();
1079        Ok(())
1080    }
1081
1082    #[tokio::test]
1083    async fn pin_tree_rejects_malformed_noncanonical_and_missing_roots() -> Result<()> {
1084        let temp_dir = TempDir::new()?;
1085        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1086        let missing_hash = sha256(b"not stored");
1087        let canonical = nhash_encode(&missing_hash)?;
1088        let (port, handle) = spawn_test_server_with_auth(Arc::clone(&store)).await?;
1089        let client = reqwest::Client::new();
1090
1091        for nhash in ["not-an-nhash".to_string(), format!("hashtree:{canonical}")] {
1092            let response = client
1093                .post(format!("http://127.0.0.1:{port}/api/pin-tree"))
1094                .basic_auth("test-user", Some("test-password"))
1095                .json(&json!({"nhash": nhash}))
1096                .send()
1097                .await?;
1098            assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);
1099        }
1100
1101        let response = client
1102            .post(format!("http://127.0.0.1:{port}/api/pin-tree"))
1103            .basic_auth("test-user", Some("test-password"))
1104            .json(&json!({"nhash": canonical}))
1105            .send()
1106            .await?;
1107        assert_eq!(response.status(), reqwest::StatusCode::NOT_FOUND);
1108        assert!(!store.is_pinned(&missing_hash)?);
1109        assert!(store.get_tree_meta(&missing_hash)?.is_none());
1110        handle.abort();
1111        Ok(())
1112    }
1113
1114    #[tokio::test]
1115    async fn pin_tree_indexes_encrypted_descendants_and_is_idempotent() -> Result<()> {
1116        let temp_dir = TempDir::new()?;
1117        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1118        let (root_cid, file_cid, expected_content) = encrypted_test_directory(&store).await?;
1119        let nhash = nhash_encode_full(&NHashData {
1120            hash: root_cid.hash,
1121            decrypt_key: root_cid.key,
1122        })?;
1123        let (port, handle) = spawn_test_server_with_auth(Arc::clone(&store)).await?;
1124        let client = reqwest::Client::new();
1125
1126        for expected_already_pinned in [false, true] {
1127            let response = client
1128                .post(format!("http://127.0.0.1:{port}/api/pin-tree"))
1129                .basic_auth("test-user", Some("test-password"))
1130                .json(&json!({"nhash": nhash}))
1131                .send()
1132                .await?;
1133            assert_eq!(response.status(), reqwest::StatusCode::OK);
1134            let body: serde_json::Value = response.json().await?;
1135            assert_eq!(body["already_pinned"], expected_already_pinned);
1136            assert!(body["indexed_hashes"].as_u64().unwrap_or_default() > 2);
1137        }
1138
1139        assert!(store.is_pinned(&root_cid.hash)?);
1140        assert_eq!(store.list_pins_raw()?, vec![root_cid.hash]);
1141        assert_eq!(store.list_indexed_trees()?.len(), 1);
1142        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()));
1143        assert_eq!(
1144            tree.get(&file_cid, None).await?,
1145            Some(expected_content),
1146            "encrypted descendant remains readable"
1147        );
1148        handle.abort();
1149        Ok(())
1150    }
1151
1152    #[tokio::test]
1153    async fn pin_tree_missing_descendant_leaves_no_pin_or_index() -> Result<()> {
1154        let temp_dir = TempDir::new()?;
1155        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1156        let (root_cid, file_cid, _) = encrypted_test_directory(&store).await?;
1157        assert!(store.router().delete_sync(&file_cid.hash)?);
1158        let nhash = nhash_encode_full(&NHashData {
1159            hash: root_cid.hash,
1160            decrypt_key: root_cid.key,
1161        })?;
1162        let (port, handle) = spawn_test_server_with_auth(Arc::clone(&store)).await?;
1163
1164        let response = reqwest::Client::new()
1165            .post(format!("http://127.0.0.1:{port}/api/pin-tree"))
1166            .basic_auth("test-user", Some("test-password"))
1167            .json(&json!({"nhash": nhash}))
1168            .send()
1169            .await?;
1170
1171        assert_eq!(response.status(), reqwest::StatusCode::UNPROCESSABLE_ENTITY);
1172        assert!(!store.is_pinned(&root_cid.hash)?);
1173        assert!(store.get_tree_meta(&root_cid.hash)?.is_none());
1174        handle.abort();
1175        Ok(())
1176    }
1177
1178    #[tokio::test]
1179    async fn pin_tree_resource_failure_leaves_no_pin_or_index() -> Result<()> {
1180        let temp_dir = TempDir::new()?;
1181        let store = Arc::new(HashtreeStore::with_options(
1182            temp_dir.path().join("db"),
1183            None,
1184            8,
1185        )?);
1186        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()));
1187        let (cid, _) = tree.put(b"larger than the pin tree byte budget").await?;
1188        let nhash = nhash_encode_full(&NHashData {
1189            hash: cid.hash,
1190            decrypt_key: cid.key,
1191        })?;
1192        let (port, handle) = spawn_test_server_with_auth(Arc::clone(&store)).await?;
1193
1194        let response = reqwest::Client::new()
1195            .post(format!("http://127.0.0.1:{port}/api/pin-tree"))
1196            .basic_auth("test-user", Some("test-password"))
1197            .json(&json!({"nhash": nhash}))
1198            .send()
1199            .await?;
1200
1201        assert_eq!(response.status(), reqwest::StatusCode::PAYLOAD_TOO_LARGE);
1202        assert!(!store.is_pinned(&cid.hash)?);
1203        assert!(store.get_tree_meta(&cid.hash)?.is_none());
1204        handle.abort();
1205        Ok(())
1206    }
1207
1208    #[tokio::test]
1209    async fn unauthenticated_upload_batch_is_rejected_before_json_extraction() -> Result<()> {
1210        let temp_dir = TempDir::new()?;
1211        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1212        let (port, handle) = spawn_test_server(store).await?;
1213
1214        let response = reqwest::Client::new()
1215            .post(format!("http://127.0.0.1:{port}/upload/batch"))
1216            .header(reqwest::header::CONTENT_TYPE, "application/json")
1217            .body("not valid json")
1218            .send()
1219            .await?;
1220
1221        assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);
1222        handle.abort();
1223        Ok(())
1224    }
1225
1226    #[tokio::test]
1227    async fn unauthenticated_single_upload_is_rejected_before_body_limit() -> Result<()> {
1228        let temp_dir = TempDir::new()?;
1229        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1230        let (port, handle) = spawn_test_server(store).await?;
1231        let declared_bytes = blossom::MAX_SINGLE_UPLOAD_BODY_BYTES + 1;
1232        let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port)).await?;
1233        stream
1234            .write_all(
1235                format!(
1236                    "PUT /upload HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nContent-Length: {declared_bytes}\r\nConnection: close\r\n\r\n"
1237                )
1238                .as_bytes(),
1239            )
1240            .await?;
1241
1242        // Send headers only: receiving a response proves authentication ran
1243        // before either buffering the declared body or enforcing its limit.
1244        let mut status_line = String::new();
1245        BufReader::new(stream).read_line(&mut status_line).await?;
1246        assert_eq!(status_line.trim_end(), "HTTP/1.1 401 Unauthorized");
1247        handle.abort();
1248        Ok(())
1249    }
1250
1251    #[tokio::test]
1252    async fn upload_options_preflight_does_not_require_auth() -> Result<()> {
1253        let temp_dir = TempDir::new()?;
1254        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1255        let (port, handle) = spawn_test_server(store).await?;
1256
1257        let response = reqwest::Client::new()
1258            .request(
1259                reqwest::Method::OPTIONS,
1260                format!("http://127.0.0.1:{port}/upload"),
1261            )
1262            .send()
1263            .await?;
1264
1265        assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
1266        handle.abort();
1267        Ok(())
1268    }
1269
1270    #[tokio::test]
1271    async fn virtual_tree_hosts_serve_root_assets_and_spa_fallbacks() -> Result<()> {
1272        clear_virtual_tree_hosts_for_test();
1273
1274        let temp_dir = TempDir::new()?;
1275        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1276        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
1277
1278        let (index_cid, _) = tree
1279            .put(b"<!doctype html><title>Virtual host ok</title>")
1280            .await?;
1281        let (favicon_cid, _) = tree.put(b"ico").await?;
1282        let (main_js_cid, _) = tree.put(b"console.log('ok');").await?;
1283        let assets_dir = tree
1284            .put_directory(vec![
1285                DirEntry::from_cid("main.js", &main_js_cid).with_link_type(LinkType::File)
1286            ])
1287            .await?;
1288        let root_cid = tree
1289            .put_directory(vec![
1290                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File),
1291                DirEntry::from_cid("favicon.ico", &favicon_cid).with_link_type(LinkType::File),
1292                DirEntry::from_cid("assets", &assets_dir).with_link_type(LinkType::Dir),
1293            ])
1294            .await?;
1295        let nhash = nhash_encode(&root_cid.hash)?;
1296        let host = "tree-test.htree.localhost";
1297        register_virtual_tree_host(host, &format!("/htree/{nhash}"));
1298
1299        let (port, handle) = spawn_test_server(store).await?;
1300        let base_url = format!("http://127.0.0.1:{port}");
1301        let host_header = format!("{host}:{port}");
1302        let client = reqwest::Client::new();
1303
1304        let root_response = client
1305            .get(format!("{base_url}/"))
1306            .header("Host", &host_header)
1307            .header("Accept", "text/html")
1308            .send()
1309            .await?;
1310        assert_eq!(root_response.status(), reqwest::StatusCode::OK);
1311        assert_eq!(
1312            root_response.bytes().await?.as_ref(),
1313            b"<!doctype html><title>Virtual host ok</title>"
1314        );
1315
1316        let favicon_response = client
1317            .get(format!("{base_url}/favicon.ico"))
1318            .header("Host", &host_header)
1319            .send()
1320            .await?;
1321        assert_eq!(favicon_response.status(), reqwest::StatusCode::OK);
1322        assert_eq!(favicon_response.bytes().await?.as_ref(), b"ico");
1323
1324        let js_response = client
1325            .get(format!("{base_url}/assets/main.js"))
1326            .header("Host", &host_header)
1327            .send()
1328            .await?;
1329        assert_eq!(js_response.status(), reqwest::StatusCode::OK);
1330        assert_eq!(js_response.bytes().await?.as_ref(), b"console.log('ok');");
1331
1332        let profile_response = client
1333            .get(format!("{base_url}/users/npub1example"))
1334            .header("Host", &host_header)
1335            .header("Accept", "text/html")
1336            .send()
1337            .await?;
1338        assert_eq!(profile_response.status(), reqwest::StatusCode::OK);
1339        assert_eq!(
1340            profile_response.bytes().await?.as_ref(),
1341            b"<!doctype html><title>Virtual host ok</title>"
1342        );
1343
1344        handle.abort();
1345        clear_virtual_tree_hosts_for_test();
1346
1347        Ok(())
1348    }
1349
1350    #[tokio::test]
1351    async fn iris_localhost_hosts_serve_immutable_and_named_sites() -> Result<()> {
1352        let temp_dir = TempDir::new()?;
1353        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1354        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
1355
1356        let (index_cid, _) = tree
1357            .put(b"<!doctype html><title>Iris localhost ok</title>")
1358            .await?;
1359        let (main_js_cid, _) = tree.put(b"console.log('iris localhost');").await?;
1360        let assets_dir = tree
1361            .put_directory(vec![
1362                DirEntry::from_cid("main.js", &main_js_cid).with_link_type(LinkType::File)
1363            ])
1364            .await?;
1365        let root_cid = tree
1366            .put_directory(vec![
1367                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File),
1368                DirEntry::from_cid("assets", &assets_dir).with_link_type(LinkType::Dir),
1369            ])
1370            .await?;
1371
1372        let nhash = nhash_encode(&root_cid.hash)?;
1373        let npub = Keys::generate().public_key().to_bech32()?;
1374        let site = "audio";
1375        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1376        let port = listener.local_addr()?.port();
1377        let server = HashtreeServer::new(Arc::clone(&store), "127.0.0.1:0".to_string())
1378            .with_public_plaintext_reads(true)
1379            .with_cached_tree_roots(vec![(format!("{npub}/{site}"), root_cid)]);
1380        let handle =
1381            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1382        let client = reqwest::Client::new();
1383        let base_url = format!("http://127.0.0.1:{port}");
1384
1385        for host in [
1386            format!("{nhash}.iris.localhost:{port}"),
1387            format!("{site}.{npub}.iris.localhost:{port}"),
1388        ] {
1389            let root_response = client
1390                .get(format!("{base_url}/"))
1391                .header("Host", &host)
1392                .header("Accept", "text/html")
1393                .send()
1394                .await?;
1395            assert_eq!(root_response.status(), reqwest::StatusCode::OK, "{host}");
1396            assert_eq!(
1397                root_response.bytes().await?.as_ref(),
1398                b"<!doctype html><title>Iris localhost ok</title>"
1399            );
1400
1401            let asset_response = client
1402                .get(format!("{base_url}/assets/main.js"))
1403                .header("Host", &host)
1404                .send()
1405                .await?;
1406            assert_eq!(asset_response.status(), reqwest::StatusCode::OK, "{host}");
1407            assert_eq!(
1408                asset_response.bytes().await?.as_ref(),
1409                b"console.log('iris localhost');"
1410            );
1411        }
1412
1413        handle.abort();
1414        Ok(())
1415    }
1416
1417    #[tokio::test]
1418    async fn named_iris_site_resolves_from_fips_event_provider_without_relays() -> Result<()> {
1419        let temp_dir = TempDir::new()?;
1420        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1421        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
1422        let (index_cid, _) = tree.put(b"fips event provider site").await?;
1423        let root_cid = tree
1424            .put_directory(vec![
1425                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File)
1426            ])
1427            .await?;
1428        let owner = Keys::generate();
1429        let site = "radio";
1430        let root_event = EventBuilder::new(Kind::Custom(30064), "")
1431            .tags(vec![
1432                nostr::Tag::identifier(site),
1433                nostr::Tag::custom(nostr::TagKind::custom("l"), vec!["hashtree".to_string()]),
1434                nostr::Tag::custom(nostr::TagKind::custom("hash"), vec![to_hex(&root_cid.hash)]),
1435            ])
1436            .sign_with_keys(&owner)?;
1437        let provider = Arc::new(StaticProvider {
1438            event: Some(VerifiedEvent::try_from(root_event)?),
1439            queries: AtomicUsize::new(0),
1440            publishes: AtomicUsize::new(0),
1441            mode: PubsubProviderMode::LocalOnly,
1442        });
1443
1444        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1445        let port = listener.local_addr()?.port();
1446        let server = HashtreeServer::new(Arc::clone(&store), "127.0.0.1:0".to_string())
1447            .with_public_plaintext_reads(true)
1448            .with_nostr_provider(provider.clone());
1449        assert!(server.state.nostr_relay_urls.is_empty());
1450        assert_eq!(
1451            server.state.nostr_provider.as_ref().unwrap().mode(),
1452            PubsubProviderMode::LocalOnly
1453        );
1454        let handle =
1455            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1456
1457        let npub = owner.public_key().to_bech32()?;
1458        let response = reqwest::Client::new()
1459            .get(format!("http://127.0.0.1:{port}/"))
1460            .header("Host", format!("{site}.{npub}.iris.localhost"))
1461            .header("Accept", "text/html")
1462            .send()
1463            .await?;
1464
1465        assert_eq!(response.status(), reqwest::StatusCode::OK);
1466        assert_eq!(
1467            response.bytes().await?.as_ref(),
1468            b"fips event provider site"
1469        );
1470        assert_eq!(provider.queries.load(Ordering::Relaxed), 1);
1471        handle.abort();
1472        Ok(())
1473    }
1474
1475    #[tokio::test]
1476    async fn loopback_event_publication_uses_configured_provider() -> Result<()> {
1477        let temp_dir = TempDir::new()?;
1478        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1479        let keys = Keys::generate();
1480        let event = EventBuilder::new(Kind::TextNote, "provider test").sign_with_keys(&keys)?;
1481        let provider = Arc::new(StaticProvider {
1482            event: Some(VerifiedEvent::try_from(event.clone())?),
1483            queries: AtomicUsize::new(0),
1484            publishes: AtomicUsize::new(0),
1485            mode: PubsubProviderMode::LocalOnly,
1486        });
1487        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1488        let port = listener.local_addr()?.port();
1489        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
1490            .with_nostr_provider(provider.clone());
1491        assert!(server.state.nostr_relay_urls.is_empty());
1492        let handle =
1493            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1494
1495        let response = reqwest::Client::new()
1496            .post(format!("http://127.0.0.1:{port}/api/nostr/events"))
1497            .json(&event)
1498            .send()
1499            .await?;
1500
1501        assert_eq!(response.status(), reqwest::StatusCode::ACCEPTED);
1502        assert_eq!(provider.publishes.load(Ordering::Relaxed), 1);
1503        handle.abort();
1504        Ok(())
1505    }
1506
1507    #[tokio::test]
1508    async fn loopback_root_publication_is_immediately_resolvable_without_provider_replay(
1509    ) -> Result<()> {
1510        let temp_dir = TempDir::new()?;
1511        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1512        let keys = Keys::generate();
1513        let tree_name = "webvm-e2e";
1514        let root_hash = "11".repeat(32);
1515        let encryption_key = "22".repeat(32);
1516        let event = EventBuilder::new(Kind::Custom(30064), &root_hash)
1517            .tags(vec![
1518                nostr::Tag::identifier(tree_name),
1519                nostr::Tag::custom(nostr::TagKind::custom("l"), vec!["hashtree".to_string()]),
1520                nostr::Tag::custom(nostr::TagKind::custom("l"), vec!["git".to_string()]),
1521                nostr::Tag::custom(nostr::TagKind::custom("hash"), vec![root_hash.clone()]),
1522                nostr::Tag::custom(nostr::TagKind::custom("key"), vec![encryption_key.clone()]),
1523            ])
1524            .sign_with_keys(&keys)?;
1525        let provider = Arc::new(StaticProvider {
1526            event: None,
1527            queries: AtomicUsize::new(0),
1528            publishes: AtomicUsize::new(0),
1529            mode: PubsubProviderMode::LocalOnly,
1530        });
1531        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1532        let port = listener.local_addr()?.port();
1533        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
1534            .with_nostr_provider(provider.clone());
1535        assert!(server.state.nostr_relay_urls.is_empty());
1536        let handle =
1537            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1538        let client = reqwest::Client::new();
1539
1540        let publish = client
1541            .post(format!("http://127.0.0.1:{port}/api/nostr/events"))
1542            .json(&event)
1543            .send()
1544            .await?;
1545        assert_eq!(publish.status(), reqwest::StatusCode::ACCEPTED);
1546
1547        let npub = keys.public_key().to_bech32()?;
1548        let resolved = client
1549            .get(format!(
1550                "http://127.0.0.1:{port}/api/nostr/resolve/{npub}/{tree_name}?refresh=1"
1551            ))
1552            .send()
1553            .await?
1554            .json::<serde_json::Value>()
1555            .await?;
1556
1557        assert_eq!(resolved["hash"], root_hash);
1558        assert_eq!(resolved["key_tag"], encryption_key);
1559        assert_eq!(resolved["source"], "local-relay");
1560        assert_eq!(provider.publishes.load(Ordering::Relaxed), 1);
1561        assert_eq!(provider.queries.load(Ordering::Relaxed), 1);
1562        handle.abort();
1563        Ok(())
1564    }
1565
1566    #[tokio::test]
1567    async fn nostr_profile_route_returns_latest_metadata_event() -> Result<()> {
1568        let temp_dir = TempDir::new()?;
1569        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1570        let graph_store = {
1571            let _guard = crate::socialgraph::test_lock().await;
1572            crate::socialgraph::open_test_social_graph_store_with_mapsize(
1573                &temp_dir.path().join("relay-db"),
1574                Some(128 * 1024 * 1024),
1575            )?
1576        };
1577        let backend: Arc<dyn crate::socialgraph::SocialGraphBackend> = graph_store;
1578        let relay = Arc::new(NostrRelay::new(
1579            backend,
1580            temp_dir.path().to_path_buf(),
1581            HashSet::new(),
1582            None,
1583            NostrRelayConfig {
1584                spambox_db_max_bytes: 0,
1585                ..Default::default()
1586            },
1587        )?);
1588
1589        let author = Keys::generate();
1590        let older = EventBuilder::new(
1591            Kind::Metadata,
1592            json!({ "name": "older", "about": "before" }).to_string(),
1593        )
1594        .custom_created_at(Timestamp::from_secs(10))
1595        .sign_with_keys(&author)?;
1596        let newer = EventBuilder::new(
1597            Kind::Metadata,
1598            json!({ "name": "newer", "about": "after" }).to_string(),
1599        )
1600        .custom_created_at(Timestamp::from_secs(20))
1601        .sign_with_keys(&author)?;
1602
1603        relay.ingest_trusted_event(older).await?;
1604        relay.ingest_trusted_event(newer.clone()).await?;
1605
1606        let (port, handle) = spawn_test_server_with_nostr_relay(store, relay).await?;
1607        let response = reqwest::get(format!(
1608            "http://127.0.0.1:{port}/api/nostr/profile/{}",
1609            author.public_key().to_hex()
1610        ))
1611        .await?;
1612
1613        assert_eq!(response.status(), reqwest::StatusCode::OK);
1614        let payload: serde_json::Value = response.json().await?;
1615        assert_eq!(payload["profile"]["name"].as_str(), Some("newer"),);
1616        assert_eq!(payload["profile"]["about"].as_str(), Some("after"));
1617        assert_eq!(payload["created_at"].as_u64(), Some(20));
1618        let expected_event_id = newer.id.to_hex();
1619        assert_eq!(
1620            payload["event_id"].as_str(),
1621            Some(expected_event_id.as_str())
1622        );
1623
1624        handle.abort();
1625        Ok(())
1626    }
1627}
hashtree_cli/server/mod.rs

hashtree_cli/server/
mod.rs
