hashtree_cli/server/
auth.rs

1use crate::blob_cache::BlobCache;
2use crate::fips_transport::DaemonBlobResolver;
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, Result<bool, String>>>;
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    /// Native authenticated FIPS endpoint used by blob routes and services.
278    pub fips_endpoint: Option<Arc<hashtree_fips_transport::FipsEndpoint>>,
279    /// Canonical Hashtree resolver whose peer routes run over fips-tcp.
280    pub fips_blob_resolver: Option<Arc<DaemonBlobResolver>>,
281    pub fetch_from_fips_peers: bool,
282    /// WebSocket relay state for /ws clients
283    pub ws_relay: Arc<WsRelayState>,
284    /// Maximum upload size in bytes for Blossom uploads (default: 5 MB)
285    pub max_upload_bytes: usize,
286    /// Allow anyone with valid Nostr auth to write (default: true)
287    /// When false, only allowed_pubkeys can write
288    pub public_writes: bool,
289    /// Allow public plaintext reads from mutable npub routes (default: false)
290    /// When false, only allowed_pubkeys or social graph approved pubkeys can read.
291    pub public_plaintext_reads: bool,
292    /// Require untrusted cached blob ingress to look like encrypted CHK blobs.
293    pub require_random_untrusted_ingest: bool,
294    /// Return from Blossom upload after validation while storage writes finish in
295    /// a bounded background queue.
296    pub optimistic_blossom_uploads: bool,
297    /// Background upload queue byte budget. Each queued body holds one permit per
298    /// byte until the storage write completes.
299    pub optimistic_upload_queue_bytes: usize,
300    pub optimistic_upload_queue: Arc<Semaphore>,
301    /// Pubkeys allowed to write (hex format, from config allowed_npubs)
302    pub allowed_pubkeys: HashSet<String>,
303    /// Upstream Blossom servers for cascade fetching
304    pub upstream_blossom: Vec<String>,
305    /// Shared HTTP client for upstream Blossom reads, so cold cache misses can
306    /// reuse connections instead of rebuilding a client per blob.
307    pub upstream_http_client: reqwest::Client,
308    /// Short cache for explicit upstream Blossom 404 misses. This only records
309    /// HTTP absence from configured Blossom upstreams; peer timeouts are not
310    /// treated as absence.
311    pub upstream_blossom_miss_cache: Arc<StdMutex<TimedLruCache<String, ()>>>,
312    /// Counters for upstream Blossom read-through decisions.
313    pub upstream_blossom_fetch_metrics: Arc<UpstreamBlossomFetchMetrics>,
314    /// Write-behind Blossom servers for blobs accepted by this server.
315    pub blossom_upload_replicas: Vec<String>,
316    /// Background replication queue byte budget. Each queued body holds one
317    /// permit per byte until the remote upload attempt finishes.
318    pub blossom_upload_replica_queue_bytes: usize,
319    pub blossom_upload_replica_queue: Arc<Semaphore>,
320    /// Signing key used for server-side write-behind replication auth.
321    pub blossom_upload_replica_keys: Option<Arc<Keys>>,
322    /// Per-server scheduler that can merge adjacent write-behind replica uploads.
323    pub blossom_upload_replica_scheduler:
324        Arc<crate::server::blossom::BlossomUploadReplicaScheduler>,
325    /// Social graph access control
326    pub social_graph: Option<Arc<socialgraph::SocialGraphAccessControl>>,
327    /// Social graph store handle for snapshot export
328    pub social_graph_store: Option<Arc<dyn socialgraph::SocialGraphBackend>>,
329    /// Social graph root pubkey bytes for snapshot export
330    pub social_graph_root: Option<[u8; 32]>,
331    /// Allow public access to social graph snapshot endpoint
332    pub socialgraph_snapshot_public: bool,
333    /// Nostr relay state for /ws and WebRTC Nostr messages
334    pub nostr_relay: Option<Arc<NostrRelay>>,
335    /// Selected provider for Hashtree Nostr root/site lookup and publication.
336    pub nostr_provider: Option<Arc<dyn nostr_pubsub::PubsubProvider>>,
337    /// Active upstream Nostr relays for HTTP resolver operations.
338    pub nostr_relay_urls: Vec<String>,
339    /// In-process cache for resolved mutable tree roots, keyed by npub/tree(+key)
340    pub tree_root_cache: Arc<StdMutex<HashMap<String, CachedTreeRootEntry>>>,
341    /// Shared in-flight blob fetches so concurrent misses only hit upstream once per hash
342    pub inflight_blob_fetches: Arc<Mutex<HashMap<String, SharedBlobFetch>>>,
343    /// Shared in-flight local blob reads so request bursts for the same hash
344    /// only spend one blocking storage read.
345    pub inflight_blob_reads: Arc<Mutex<HashMap<String, SharedBlobRead>>>,
346    /// Bounded hot cache for immutable blob bodies and metadata probes.
347    pub(super) blob_cache: Arc<BlobCache>,
348    /// Immutable directory listings keyed by CID
349    pub directory_listing_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<Vec<TreeEntry>>>>>,
350    /// Immutable resolved paths keyed by root CID + path
351    pub resolved_path_cache:
352        Arc<StdMutex<TimedLruCache<String, LookupResult<CachedResolvedPathEntry>>>>,
353    /// Immutable thumbnail alias resolutions keyed by root CID + alias path
354    pub thumbnail_path_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<String>>>>,
355    /// Immutable file sizes keyed by CID
356    pub cid_size_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<u64>>>>,
357}
358
359pub fn new_upstream_http_client() -> reqwest::Client {
360    reqwest::Client::builder()
361        .timeout(Duration::from_secs(10))
362        .build()
363        .expect("build upstream Blossom HTTP client")
364}
365
366#[derive(Clone)]
367pub struct AuthCredentials {
368    pub username: String,
369    pub password: String,
370}
371
372fn basic_auth_authorized(headers: &HeaderMap, auth: &AuthCredentials) -> bool {
373    let auth_header = headers
374        .get(header::AUTHORIZATION)
375        .and_then(|v| v.to_str().ok());
376
377    let Some(header_value) = auth_header else {
378        return false;
379    };
380    let Some(credentials) = header_value.strip_prefix("Basic ") else {
381        return false;
382    };
383
384    use base64::Engine;
385    let engine = base64::engine::general_purpose::STANDARD;
386    let Ok(decoded) = engine.decode(credentials) else {
387        return false;
388    };
389    let Ok(decoded_str) = String::from_utf8(decoded) else {
390        return false;
391    };
392    let expected = format!("{}:{}", auth.username, auth.password);
393    decoded_str == expected
394}
395
396fn unauthorized_basic_response() -> Response<Body> {
397    Response::builder()
398        .status(StatusCode::UNAUTHORIZED)
399        .header(header::WWW_AUTHENTICATE, "Basic realm=\"hashtree\"")
400        .body(Body::from("Unauthorized"))
401        .unwrap()
402}
403
404/// Auth middleware - validates HTTP Basic Auth
405pub async fn auth_middleware(
406    State(state): State<AppState>,
407    request: Request<Body>,
408    next: Next,
409) -> Result<Response<Body>, StatusCode> {
410    // If auth is not enabled, allow request
411    let Some(auth) = &state.auth else {
412        return Ok(next.run(request).await);
413    };
414
415    if basic_auth_authorized(request.headers(), auth) {
416        Ok(next.run(request).await)
417    } else {
418        Ok(unauthorized_basic_response())
419    }
420}
421
422/// Strict internal auth middleware - requires configured HTTP Basic Auth.
423pub async fn require_auth_middleware(
424    State(state): State<AppState>,
425    request: Request<Body>,
426    next: Next,
427) -> Result<Response<Body>, StatusCode> {
428    let Some(auth) = &state.auth else {
429        return Ok(Response::builder()
430            .status(StatusCode::FORBIDDEN)
431            .body(Body::from("Authentication is not configured"))
432            .unwrap());
433    };
434
435    if basic_auth_authorized(request.headers(), auth) {
436        Ok(next.run(request).await)
437    } else {
438        Ok(unauthorized_basic_response())
439    }
440}
hashtree_cli/server/auth.rs

hashtree_cli/server/
auth.rs
