hashtree_cli/server/
auth.rs

1use crate::blob_cache::BlobCache;
2use crate::fips_transport::DaemonFipsTransport;
3use crate::nostr_relay::NostrRelay;
4use crate::socialgraph;
5use crate::storage::HashtreeStore;
6use crate::webrtc::{PeerRootEvent, WebRTCState};
7use axum::{
8    body::Body,
9    extract::ws::Message,
10    extract::State,
11    http::{header, Request, Response, StatusCode},
12    middleware::Next,
13};
14use futures::future::{BoxFuture, Shared};
15use hashtree_core::{Cid, LinkType, TreeEntry};
16use lru::LruCache;
17use nostr::Keys;
18use std::collections::{HashMap, HashSet};
19use std::hash::Hash;
20use std::num::NonZeroUsize;
21use std::sync::{
22    atomic::{AtomicU32, AtomicU64, Ordering},
23    Arc, Mutex as StdMutex,
24};
25use std::time::{Duration, Instant};
26use tokio::{
27    sync::{mpsc, watch, Mutex, Semaphore},
28    task::JoinHandle,
29};
30
31const LOOKUP_CACHE_CAPACITY: usize = 4096;
32const LOOKUP_CACHE_HIT_TTL: Duration = Duration::from_secs(300);
33const LOOKUP_CACHE_MISS_TTL: Duration = Duration::from_secs(1);
34
35#[derive(Debug, Clone)]
36pub enum LookupResult<T> {
37    Hit(T),
38    Miss,
39}
40
41impl<T> LookupResult<T> {
42    pub fn from_option(value: Option<T>) -> Self {
43        match value {
44            Some(value) => Self::Hit(value),
45            None => Self::Miss,
46        }
47    }
48
49    pub fn into_option(self) -> Option<T> {
50        match self {
51            Self::Hit(value) => Some(value),
52            Self::Miss => None,
53        }
54    }
55
56    pub fn ttl(&self) -> Duration {
57        match self {
58            Self::Hit(_) => LOOKUP_CACHE_HIT_TTL,
59            Self::Miss => LOOKUP_CACHE_MISS_TTL,
60        }
61    }
62}
63
64pub struct TimedLruCache<K, V> {
65    cache: LruCache<K, TimedValue<V>>,
66}
67
68#[derive(Clone)]
69struct TimedValue<V> {
70    value: V,
71    expires_at: Instant,
72}
73
74impl<K: Eq + Hash, V: Clone> TimedLruCache<K, V> {
75    pub fn new(capacity: usize) -> Self {
76        Self {
77            cache: LruCache::new(NonZeroUsize::new(capacity.max(1)).unwrap()),
78        }
79    }
80
81    pub fn get_cloned(&mut self, key: &K) -> Option<V> {
82        let now = Instant::now();
83        if let Some(entry) = self.cache.get(key) {
84            if entry.expires_at > now {
85                return Some(entry.value.clone());
86            }
87        }
88        self.cache.pop(key);
89        None
90    }
91
92    pub fn put(&mut self, key: K, value: V, ttl: Duration) {
93        self.cache.put(
94            key,
95            TimedValue {
96                value,
97                expires_at: Instant::now() + ttl,
98            },
99        );
100    }
101}
102
103pub fn new_lookup_cache<K: Eq + Hash, V: Clone>() -> TimedLruCache<K, V> {
104    TimedLruCache::new(LOOKUP_CACHE_CAPACITY)
105}
106
107#[derive(Debug, Clone)]
108pub struct CachedResolvedPathEntry {
109    pub cid: Cid,
110    pub link_type: LinkType,
111}
112
113#[derive(Debug, Clone)]
114pub struct CachedTreeRootEntry {
115    pub cid: Cid,
116    pub source: &'static str,
117    pub root_event: Option<PeerRootEvent>,
118    pub cached_at: Instant,
119}
120
121pub type SharedBlobFetch = Shared<BoxFuture<'static, bool>>;
122pub type SharedBlobRead = Shared<BoxFuture<'static, Result<Option<Vec<u8>>, String>>>;
123
124#[derive(Debug, Clone, Copy, PartialEq, Eq)]
125pub enum WsProtocol {
126    HashtreeJson,
127    HashtreeMsgpack,
128    Unknown,
129}
130
131pub struct PendingRequest {
132    pub origin_id: u64,
133    pub hash: String,
134    pub found: bool,
135    pub origin_protocol: WsProtocol,
136}
137
138pub struct UpstreamNostrSubscription {
139    pub close_tx: watch::Sender<bool>,
140    pub tasks: Vec<JoinHandle<()>>,
141}
142
143#[derive(Debug, Clone, Default)]
144pub struct UpstreamBlossomFetchSnapshot {
145    pub lookup_attempts: u64,
146    pub hits: u64,
147    pub hit_bytes: u64,
148    pub explicit_misses: u64,
149    pub indeterminate_misses: u64,
150    pub miss_cache_hits: u64,
151    pub last_indeterminate_reason: Option<String>,
152}
153
154#[derive(Default)]
155pub struct UpstreamBlossomFetchMetrics {
156    lookup_attempts: AtomicU64,
157    hits: AtomicU64,
158    hit_bytes: AtomicU64,
159    explicit_misses: AtomicU64,
160    indeterminate_misses: AtomicU64,
161    miss_cache_hits: AtomicU64,
162    last_indeterminate_reason: StdMutex<Option<String>>,
163}
164
165impl UpstreamBlossomFetchMetrics {
166    pub fn note_lookup_attempt(&self) {
167        self.lookup_attempts.fetch_add(1, Ordering::Relaxed);
168    }
169
170    pub fn note_hit(&self, bytes: usize) {
171        self.hits.fetch_add(1, Ordering::Relaxed);
172        self.hit_bytes.fetch_add(bytes as u64, Ordering::Relaxed);
173    }
174
175    pub fn note_explicit_miss(&self) {
176        self.explicit_misses.fetch_add(1, Ordering::Relaxed);
177    }
178
179    pub fn note_indeterminate_miss(&self, reason: impl Into<String>) {
180        self.indeterminate_misses.fetch_add(1, Ordering::Relaxed);
181        let reason = reason.into();
182        if let Ok(mut last_reason) = self.last_indeterminate_reason.lock() {
183            *last_reason = Some(reason.chars().take(512).collect());
184        }
185    }
186
187    pub fn note_miss_cache_hit(&self) {
188        self.miss_cache_hits.fetch_add(1, Ordering::Relaxed);
189    }
190
191    pub fn snapshot(&self) -> UpstreamBlossomFetchSnapshot {
192        UpstreamBlossomFetchSnapshot {
193            lookup_attempts: self.lookup_attempts.load(Ordering::Relaxed),
194            hits: self.hits.load(Ordering::Relaxed),
195            hit_bytes: self.hit_bytes.load(Ordering::Relaxed),
196            explicit_misses: self.explicit_misses.load(Ordering::Relaxed),
197            indeterminate_misses: self.indeterminate_misses.load(Ordering::Relaxed),
198            miss_cache_hits: self.miss_cache_hits.load(Ordering::Relaxed),
199            last_indeterminate_reason: self
200                .last_indeterminate_reason
201                .lock()
202                .ok()
203                .and_then(|reason| reason.clone()),
204        }
205    }
206}
207
208pub struct WsRelayState {
209    pub clients: Mutex<HashMap<u64, mpsc::UnboundedSender<Message>>>,
210    pub pending: Mutex<HashMap<(u64, u32), PendingRequest>>,
211    pub client_protocols: Mutex<HashMap<u64, WsProtocol>>,
212    pub upstream_nostr_subscriptions: Mutex<HashMap<(u64, String), UpstreamNostrSubscription>>,
213    pub upstream_seen_events: Mutex<HashMap<(u64, String), HashSet<String>>>,
214    pub upstream_pending_eose: Mutex<HashMap<(u64, String), usize>>,
215    pub next_client_id: AtomicU64,
216    pub next_request_id: AtomicU32,
217    pub upstream_relay_bytes_sent: AtomicU64,
218    pub upstream_relay_bytes_received: AtomicU64,
219}
220
221impl WsRelayState {
222    pub fn new() -> Self {
223        Self {
224            clients: Mutex::new(HashMap::new()),
225            pending: Mutex::new(HashMap::new()),
226            client_protocols: Mutex::new(HashMap::new()),
227            upstream_nostr_subscriptions: Mutex::new(HashMap::new()),
228            upstream_seen_events: Mutex::new(HashMap::new()),
229            upstream_pending_eose: Mutex::new(HashMap::new()),
230            next_client_id: AtomicU64::new(1),
231            next_request_id: AtomicU32::new(1),
232            upstream_relay_bytes_sent: AtomicU64::new(0),
233            upstream_relay_bytes_received: AtomicU64::new(0),
234        }
235    }
236
237    pub fn next_id(&self) -> u64 {
238        self.next_client_id.fetch_add(1, Ordering::SeqCst)
239    }
240
241    pub fn next_request_id(&self) -> u32 {
242        self.next_request_id.fetch_add(1, Ordering::SeqCst)
243    }
244
245    pub fn note_upstream_relay_send(&self, bytes: usize) {
246        self.upstream_relay_bytes_sent
247            .fetch_add(bytes as u64, Ordering::Relaxed);
248    }
249
250    pub fn note_upstream_relay_receive(&self, bytes: usize) {
251        self.upstream_relay_bytes_received
252            .fetch_add(bytes as u64, Ordering::Relaxed);
253    }
254
255    pub fn upstream_relay_bandwidth(&self) -> (u64, u64) {
256        (
257            self.upstream_relay_bytes_sent.load(Ordering::Relaxed),
258            self.upstream_relay_bytes_received.load(Ordering::Relaxed),
259        )
260    }
261}
262
263#[derive(Clone)]
264pub struct AppState {
265    pub store: Arc<HashtreeStore>,
266    pub auth: Option<AuthCredentials>,
267    /// Unix timestamp when this daemon state was created.
268    pub daemon_started_at: u64,
269    pub peer_mode: crate::config::ServerMode,
270    pub hash_get_enabled: bool,
271    /// Whether HTTP cache misses should ask connected WebRTC peers before
272    /// falling back to upstream Blossom.
273    pub http_webrtc_fetch: bool,
274    /// WebRTC peer state for forwarding requests to connected P2P peers
275    pub webrtc_peers: Option<Arc<WebRTCState>>,
276    /// FIPS-backed Hashtree blob transport for peer fetches and responses.
277    pub fips_transport: Option<Arc<DaemonFipsTransport>>,
278    pub fetch_from_fips_peers: bool,
279    /// WebSocket relay state for /ws clients
280    pub ws_relay: Arc<WsRelayState>,
281    /// Maximum upload size in bytes for Blossom uploads (default: 5 MB)
282    pub max_upload_bytes: usize,
283    /// Allow anyone with valid Nostr auth to write (default: true)
284    /// When false, only allowed_pubkeys can write
285    pub public_writes: bool,
286    /// Allow public plaintext reads from mutable npub routes (default: true)
287    /// When false, only allowed_pubkeys or social graph approved pubkeys can read.
288    pub public_plaintext_reads: bool,
289    /// Require untrusted cached blob ingress to look like encrypted CHK blobs.
290    pub require_random_untrusted_ingest: bool,
291    /// Return from Blossom upload after validation while storage writes finish in
292    /// a bounded background queue.
293    pub optimistic_blossom_uploads: bool,
294    /// Background upload queue byte budget. Each queued body holds one permit per
295    /// byte until the storage write completes.
296    pub optimistic_upload_queue_bytes: usize,
297    pub optimistic_upload_queue: Arc<Semaphore>,
298    /// Pubkeys allowed to write (hex format, from config allowed_npubs)
299    pub allowed_pubkeys: HashSet<String>,
300    /// Upstream Blossom servers for cascade fetching
301    pub upstream_blossom: Vec<String>,
302    /// Shared HTTP client for upstream Blossom reads, so cold cache misses can
303    /// reuse connections instead of rebuilding a client per blob.
304    pub upstream_http_client: reqwest::Client,
305    /// Short cache for explicit upstream Blossom 404 misses. This only records
306    /// HTTP absence from configured Blossom upstreams; peer timeouts are not
307    /// treated as absence.
308    pub upstream_blossom_miss_cache: Arc<StdMutex<TimedLruCache<String, ()>>>,
309    /// Counters for upstream Blossom read-through decisions.
310    pub upstream_blossom_fetch_metrics: Arc<UpstreamBlossomFetchMetrics>,
311    /// Write-behind Blossom servers for blobs accepted by this server.
312    pub blossom_upload_replicas: Vec<String>,
313    /// Background replication queue byte budget. Each queued body holds one
314    /// permit per byte until the remote upload attempt finishes.
315    pub blossom_upload_replica_queue_bytes: usize,
316    pub blossom_upload_replica_queue: Arc<Semaphore>,
317    /// Signing key used for server-side write-behind replication auth.
318    pub blossom_upload_replica_keys: Option<Arc<Keys>>,
319    /// Per-server scheduler that can merge adjacent write-behind replica uploads.
320    pub blossom_upload_replica_scheduler:
321        Arc<crate::server::blossom::BlossomUploadReplicaScheduler>,
322    /// Social graph access control
323    pub social_graph: Option<Arc<socialgraph::SocialGraphAccessControl>>,
324    /// Social graph store handle for snapshot export
325    pub social_graph_store: Option<Arc<dyn socialgraph::SocialGraphBackend>>,
326    /// Social graph root pubkey bytes for snapshot export
327    pub social_graph_root: Option<[u8; 32]>,
328    /// Allow public access to social graph snapshot endpoint
329    pub socialgraph_snapshot_public: bool,
330    /// Nostr relay state for /ws and WebRTC Nostr messages
331    pub nostr_relay: Option<Arc<NostrRelay>>,
332    /// Active upstream Nostr relays for HTTP resolver operations.
333    pub nostr_relay_urls: Vec<String>,
334    /// In-process cache for resolved mutable tree roots, keyed by npub/tree(+key)
335    pub tree_root_cache: Arc<StdMutex<HashMap<String, CachedTreeRootEntry>>>,
336    /// Shared in-flight blob fetches so concurrent misses only hit upstream once per hash
337    pub inflight_blob_fetches: Arc<Mutex<HashMap<String, SharedBlobFetch>>>,
338    /// Shared in-flight local blob reads so request bursts for the same hash
339    /// only spend one blocking storage read.
340    pub inflight_blob_reads: Arc<Mutex<HashMap<String, SharedBlobRead>>>,
341    /// Bounded hot cache for immutable blob bodies and metadata probes.
342    pub(super) blob_cache: Arc<BlobCache>,
343    /// Immutable directory listings keyed by CID
344    pub directory_listing_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<Vec<TreeEntry>>>>>,
345    /// Immutable resolved paths keyed by root CID + path
346    pub resolved_path_cache:
347        Arc<StdMutex<TimedLruCache<String, LookupResult<CachedResolvedPathEntry>>>>,
348    /// Immutable thumbnail alias resolutions keyed by root CID + alias path
349    pub thumbnail_path_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<String>>>>,
350    /// Immutable file sizes keyed by CID
351    pub cid_size_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<u64>>>>,
352}
353
354pub fn new_upstream_http_client() -> reqwest::Client {
355    reqwest::Client::builder()
356        .timeout(Duration::from_secs(10))
357        .build()
358        .expect("build upstream Blossom HTTP client")
359}
360
361#[derive(Clone)]
362pub struct AuthCredentials {
363    pub username: String,
364    pub password: String,
365}
366
367/// Auth middleware - validates HTTP Basic Auth
368pub async fn auth_middleware(
369    State(state): State<AppState>,
370    request: Request<Body>,
371    next: Next,
372) -> Result<Response<Body>, StatusCode> {
373    // If auth is not enabled, allow request
374    let Some(auth) = &state.auth else {
375        return Ok(next.run(request).await);
376    };
377
378    // Check Authorization header
379    let auth_header = request
380        .headers()
381        .get(header::AUTHORIZATION)
382        .and_then(|v| v.to_str().ok());
383
384    let authorized = if let Some(header_value) = auth_header {
385        if let Some(credentials) = header_value.strip_prefix("Basic ") {
386            use base64::Engine;
387            let engine = base64::engine::general_purpose::STANDARD;
388            if let Ok(decoded) = engine.decode(credentials) {
389                if let Ok(decoded_str) = String::from_utf8(decoded) {
390                    let expected = format!("{}:{}", auth.username, auth.password);
391                    decoded_str == expected
392                } else {
393                    false
394                }
395            } else {
396                false
397            }
398        } else {
399            false
400        }
401    } else {
402        false
403    };
404
405    if authorized {
406        Ok(next.run(request).await)
407    } else {
408        Ok(Response::builder()
409            .status(StatusCode::UNAUTHORIZED)
410            .header(header::WWW_AUTHENTICATE, "Basic realm=\"hashtree\"")
411            .body(Body::from("Unauthorized"))
412            .unwrap())
413    }
414}
hashtree_cli/server/auth.rs

hashtree_cli/server/
auth.rs
