hashtree_cli/server/
auth.rs

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

hashtree_cli/server/
auth.rs
