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, HeaderMap, 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 event: Option<nostr::Event>,
119    pub cached_at: Instant,
120}
121
122pub type SharedBlobFetch = Shared<BoxFuture<'static, bool>>;
123pub type SharedBlobRead = Shared<BoxFuture<'static, Result<Option<Vec<u8>>, String>>>;
124
125#[derive(Debug, Clone, Copy, PartialEq, Eq)]
126pub enum WsProtocol {
127    HashtreeJson,
128    HashtreeMsgpack,
129    Unknown,
130}
131
132pub struct PendingRequest {
133    pub origin_id: u64,
134    pub hash: String,
135    pub found: bool,
136    pub origin_protocol: WsProtocol,
137}
138
139pub struct UpstreamNostrSubscription {
140    pub close_tx: watch::Sender<bool>,
141    pub tasks: Vec<JoinHandle<()>>,
142}
143
144#[derive(Debug, Clone, Default)]
145pub struct UpstreamBlossomFetchSnapshot {
146    pub lookup_attempts: u64,
147    pub hits: u64,
148    pub hit_bytes: u64,
149    pub explicit_misses: u64,
150    pub indeterminate_misses: u64,
151    pub miss_cache_hits: u64,
152    pub last_indeterminate_reason: Option<String>,
153}
154
155#[derive(Default)]
156pub struct UpstreamBlossomFetchMetrics {
157    lookup_attempts: AtomicU64,
158    hits: AtomicU64,
159    hit_bytes: AtomicU64,
160    explicit_misses: AtomicU64,
161    indeterminate_misses: AtomicU64,
162    miss_cache_hits: AtomicU64,
163    last_indeterminate_reason: StdMutex<Option<String>>,
164}
165
166impl UpstreamBlossomFetchMetrics {
167    pub fn note_lookup_attempt(&self) {
168        self.lookup_attempts.fetch_add(1, Ordering::Relaxed);
169    }
170
171    pub fn note_hit(&self, bytes: usize) {
172        self.hits.fetch_add(1, Ordering::Relaxed);
173        self.hit_bytes.fetch_add(bytes as u64, Ordering::Relaxed);
174    }
175
176    pub fn note_explicit_miss(&self) {
177        self.explicit_misses.fetch_add(1, Ordering::Relaxed);
178    }
179
180    pub fn note_indeterminate_miss(&self, reason: impl Into<String>) {
181        self.indeterminate_misses.fetch_add(1, Ordering::Relaxed);
182        let reason = reason.into();
183        if let Ok(mut last_reason) = self.last_indeterminate_reason.lock() {
184            *last_reason = Some(reason.chars().take(512).collect());
185        }
186    }
187
188    pub fn note_miss_cache_hit(&self) {
189        self.miss_cache_hits.fetch_add(1, Ordering::Relaxed);
190    }
191
192    pub fn snapshot(&self) -> UpstreamBlossomFetchSnapshot {
193        UpstreamBlossomFetchSnapshot {
194            lookup_attempts: self.lookup_attempts.load(Ordering::Relaxed),
195            hits: self.hits.load(Ordering::Relaxed),
196            hit_bytes: self.hit_bytes.load(Ordering::Relaxed),
197            explicit_misses: self.explicit_misses.load(Ordering::Relaxed),
198            indeterminate_misses: self.indeterminate_misses.load(Ordering::Relaxed),
199            miss_cache_hits: self.miss_cache_hits.load(Ordering::Relaxed),
200            last_indeterminate_reason: self
201                .last_indeterminate_reason
202                .lock()
203                .ok()
204                .and_then(|reason| reason.clone()),
205        }
206    }
207}
208
209pub struct WsRelayState {
210    pub clients: Mutex<HashMap<u64, mpsc::UnboundedSender<Message>>>,
211    pub pending: Mutex<HashMap<(u64, u32), PendingRequest>>,
212    pub client_protocols: Mutex<HashMap<u64, WsProtocol>>,
213    pub upstream_nostr_subscriptions: Mutex<HashMap<(u64, String), UpstreamNostrSubscription>>,
214    pub upstream_seen_events: Mutex<HashMap<(u64, String), HashSet<String>>>,
215    pub upstream_pending_eose: Mutex<HashMap<(u64, String), usize>>,
216    pub next_client_id: AtomicU64,
217    pub next_request_id: AtomicU32,
218    pub upstream_relay_bytes_sent: AtomicU64,
219    pub upstream_relay_bytes_received: AtomicU64,
220}
221
222impl WsRelayState {
223    pub fn new() -> Self {
224        Self {
225            clients: Mutex::new(HashMap::new()),
226            pending: Mutex::new(HashMap::new()),
227            client_protocols: Mutex::new(HashMap::new()),
228            upstream_nostr_subscriptions: Mutex::new(HashMap::new()),
229            upstream_seen_events: Mutex::new(HashMap::new()),
230            upstream_pending_eose: Mutex::new(HashMap::new()),
231            next_client_id: AtomicU64::new(1),
232            next_request_id: AtomicU32::new(1),
233            upstream_relay_bytes_sent: AtomicU64::new(0),
234            upstream_relay_bytes_received: AtomicU64::new(0),
235        }
236    }
237
238    pub fn next_id(&self) -> u64 {
239        self.next_client_id.fetch_add(1, Ordering::SeqCst)
240    }
241
242    pub fn next_request_id(&self) -> u32 {
243        self.next_request_id.fetch_add(1, Ordering::SeqCst)
244    }
245
246    pub fn note_upstream_relay_send(&self, bytes: usize) {
247        self.upstream_relay_bytes_sent
248            .fetch_add(bytes as u64, Ordering::Relaxed);
249    }
250
251    pub fn note_upstream_relay_receive(&self, bytes: usize) {
252        self.upstream_relay_bytes_received
253            .fetch_add(bytes as u64, Ordering::Relaxed);
254    }
255
256    pub fn upstream_relay_bandwidth(&self) -> (u64, u64) {
257        (
258            self.upstream_relay_bytes_sent.load(Ordering::Relaxed),
259            self.upstream_relay_bytes_received.load(Ordering::Relaxed),
260        )
261    }
262}
263
264#[derive(Clone)]
265pub struct AppState {
266    pub store: Arc<HashtreeStore>,
267    pub auth: Option<AuthCredentials>,
268    /// Unix timestamp when this daemon state was created.
269    pub daemon_started_at: u64,
270    pub peer_mode: crate::config::ServerMode,
271    pub hash_get_enabled: bool,
272    /// Whether HTTP cache misses should ask connected WebRTC peers before
273    /// falling back to upstream Blossom.
274    pub http_webrtc_fetch: bool,
275    /// WebRTC peer state for forwarding requests to connected P2P peers
276    pub webrtc_peers: Option<Arc<WebRTCState>>,
277    /// FIPS-backed Hashtree blob transport for peer fetches and responses.
278    pub fips_transport: Option<Arc<DaemonFipsTransport>>,
279    pub fetch_from_fips_peers: bool,
280    /// WebSocket relay state for /ws clients
281    pub ws_relay: Arc<WsRelayState>,
282    /// Maximum upload size in bytes for Blossom uploads (default: 5 MB)
283    pub max_upload_bytes: usize,
284    /// Allow anyone with valid Nostr auth to write (default: true)
285    /// When false, only allowed_pubkeys can write
286    pub public_writes: bool,
287    /// Allow public plaintext reads from mutable npub routes (default: false)
288    /// When false, only allowed_pubkeys or social graph approved pubkeys can read.
289    pub public_plaintext_reads: bool,
290    /// Require untrusted cached blob ingress to look like encrypted CHK blobs.
291    pub require_random_untrusted_ingest: bool,
292    /// Return from Blossom upload after validation while storage writes finish in
293    /// a bounded background queue.
294    pub optimistic_blossom_uploads: bool,
295    /// Background upload queue byte budget. Each queued body holds one permit per
296    /// byte until the storage write completes.
297    pub optimistic_upload_queue_bytes: usize,
298    pub optimistic_upload_queue: Arc<Semaphore>,
299    /// Pubkeys allowed to write (hex format, from config allowed_npubs)
300    pub allowed_pubkeys: HashSet<String>,
301    /// Upstream Blossom servers for cascade fetching
302    pub upstream_blossom: Vec<String>,
303    /// Shared HTTP client for upstream Blossom reads, so cold cache misses can
304    /// reuse connections instead of rebuilding a client per blob.
305    pub upstream_http_client: reqwest::Client,
306    /// Short cache for explicit upstream Blossom 404 misses. This only records
307    /// HTTP absence from configured Blossom upstreams; peer timeouts are not
308    /// treated as absence.
309    pub upstream_blossom_miss_cache: Arc<StdMutex<TimedLruCache<String, ()>>>,
310    /// Counters for upstream Blossom read-through decisions.
311    pub upstream_blossom_fetch_metrics: Arc<UpstreamBlossomFetchMetrics>,
312    /// Write-behind Blossom servers for blobs accepted by this server.
313    pub blossom_upload_replicas: Vec<String>,
314    /// Background replication queue byte budget. Each queued body holds one
315    /// permit per byte until the remote upload attempt finishes.
316    pub blossom_upload_replica_queue_bytes: usize,
317    pub blossom_upload_replica_queue: Arc<Semaphore>,
318    /// Signing key used for server-side write-behind replication auth.
319    pub blossom_upload_replica_keys: Option<Arc<Keys>>,
320    /// Per-server scheduler that can merge adjacent write-behind replica uploads.
321    pub blossom_upload_replica_scheduler:
322        Arc<crate::server::blossom::BlossomUploadReplicaScheduler>,
323    /// Social graph access control
324    pub social_graph: Option<Arc<socialgraph::SocialGraphAccessControl>>,
325    /// Social graph store handle for snapshot export
326    pub social_graph_store: Option<Arc<dyn socialgraph::SocialGraphBackend>>,
327    /// Social graph root pubkey bytes for snapshot export
328    pub social_graph_root: Option<[u8; 32]>,
329    /// Allow public access to social graph snapshot endpoint
330    pub socialgraph_snapshot_public: bool,
331    /// Nostr relay state for /ws and WebRTC Nostr messages
332    pub nostr_relay: Option<Arc<NostrRelay>>,
333    /// Selected provider for Hashtree Nostr root/site lookup and publication.
334    pub nostr_provider: Option<Arc<dyn nostr_pubsub::PubsubProvider>>,
335    /// Active upstream Nostr relays for HTTP resolver operations.
336    pub nostr_relay_urls: Vec<String>,
337    /// In-process cache for resolved mutable tree roots, keyed by npub/tree(+key)
338    pub tree_root_cache: Arc<StdMutex<HashMap<String, CachedTreeRootEntry>>>,
339    /// Shared in-flight blob fetches so concurrent misses only hit upstream once per hash
340    pub inflight_blob_fetches: Arc<Mutex<HashMap<String, SharedBlobFetch>>>,
341    /// Shared in-flight local blob reads so request bursts for the same hash
342    /// only spend one blocking storage read.
343    pub inflight_blob_reads: Arc<Mutex<HashMap<String, SharedBlobRead>>>,
344    /// Bounded hot cache for immutable blob bodies and metadata probes.
345    pub(super) blob_cache: Arc<BlobCache>,
346    /// Immutable directory listings keyed by CID
347    pub directory_listing_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<Vec<TreeEntry>>>>>,
348    /// Immutable resolved paths keyed by root CID + path
349    pub resolved_path_cache:
350        Arc<StdMutex<TimedLruCache<String, LookupResult<CachedResolvedPathEntry>>>>,
351    /// Immutable thumbnail alias resolutions keyed by root CID + alias path
352    pub thumbnail_path_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<String>>>>,
353    /// Immutable file sizes keyed by CID
354    pub cid_size_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<u64>>>>,
355}
356
357pub fn new_upstream_http_client() -> reqwest::Client {
358    reqwest::Client::builder()
359        .timeout(Duration::from_secs(10))
360        .build()
361        .expect("build upstream Blossom HTTP client")
362}
363
364#[derive(Clone)]
365pub struct AuthCredentials {
366    pub username: String,
367    pub password: String,
368}
369
370fn basic_auth_authorized(headers: &HeaderMap, auth: &AuthCredentials) -> bool {
371    let auth_header = headers
372        .get(header::AUTHORIZATION)
373        .and_then(|v| v.to_str().ok());
374
375    let Some(header_value) = auth_header else {
376        return false;
377    };
378    let Some(credentials) = header_value.strip_prefix("Basic ") else {
379        return false;
380    };
381
382    use base64::Engine;
383    let engine = base64::engine::general_purpose::STANDARD;
384    let Ok(decoded) = engine.decode(credentials) else {
385        return false;
386    };
387    let Ok(decoded_str) = String::from_utf8(decoded) else {
388        return false;
389    };
390    let expected = format!("{}:{}", auth.username, auth.password);
391    decoded_str == expected
392}
393
394fn unauthorized_basic_response() -> Response<Body> {
395    Response::builder()
396        .status(StatusCode::UNAUTHORIZED)
397        .header(header::WWW_AUTHENTICATE, "Basic realm=\"hashtree\"")
398        .body(Body::from("Unauthorized"))
399        .unwrap()
400}
401
402/// Auth middleware - validates HTTP Basic Auth
403pub async fn auth_middleware(
404    State(state): State<AppState>,
405    request: Request<Body>,
406    next: Next,
407) -> Result<Response<Body>, StatusCode> {
408    // If auth is not enabled, allow request
409    let Some(auth) = &state.auth else {
410        return Ok(next.run(request).await);
411    };
412
413    if basic_auth_authorized(request.headers(), auth) {
414        Ok(next.run(request).await)
415    } else {
416        Ok(unauthorized_basic_response())
417    }
418}
419
420/// Strict internal auth middleware - requires configured HTTP Basic Auth.
421pub async fn require_auth_middleware(
422    State(state): State<AppState>,
423    request: Request<Body>,
424    next: Next,
425) -> Result<Response<Body>, StatusCode> {
426    let Some(auth) = &state.auth else {
427        return Ok(Response::builder()
428            .status(StatusCode::FORBIDDEN)
429            .body(Body::from("Authentication is not configured"))
430            .unwrap());
431    };
432
433    if basic_auth_authorized(request.headers(), auth) {
434        Ok(next.run(request).await)
435    } else {
436        Ok(unauthorized_basic_response())
437    }
438}
hashtree_cli/server/auth.rs

hashtree_cli/server/
auth.rs
