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

hashtree_cli/server/
mod.rs
