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

hashtree_cli/server/
auth.rs
