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 std::collections::{HashMap, HashSet};
18use std::hash::Hash;
19use std::num::NonZeroUsize;
20use std::sync::{
21    atomic::{AtomicU32, AtomicU64, Ordering},
22    Arc, Mutex as StdMutex,
23};
24use std::time::{Duration, Instant};
25use tokio::{
26    sync::{mpsc, watch, Mutex, Semaphore},
27    task::JoinHandle,
28};
29
30const LOOKUP_CACHE_CAPACITY: usize = 4096;
31const LOOKUP_CACHE_HIT_TTL: Duration = Duration::from_secs(300);
32const LOOKUP_CACHE_MISS_TTL: Duration = Duration::from_secs(1);
33
34#[derive(Debug, Clone)]
35pub enum LookupResult<T> {
36    Hit(T),
37    Miss,
38}
39
40impl<T> LookupResult<T> {
41    pub fn from_option(value: Option<T>) -> Self {
42        match value {
43            Some(value) => Self::Hit(value),
44            None => Self::Miss,
45        }
46    }
47
48    pub fn into_option(self) -> Option<T> {
49        match self {
50            Self::Hit(value) => Some(value),
51            Self::Miss => None,
52        }
53    }
54
55    pub fn ttl(&self) -> Duration {
56        match self {
57            Self::Hit(_) => LOOKUP_CACHE_HIT_TTL,
58            Self::Miss => LOOKUP_CACHE_MISS_TTL,
59        }
60    }
61}
62
63pub struct TimedLruCache<K, V> {
64    cache: LruCache<K, TimedValue<V>>,
65}
66
67#[derive(Clone)]
68struct TimedValue<V> {
69    value: V,
70    expires_at: Instant,
71}
72
73impl<K: Eq + Hash, V: Clone> TimedLruCache<K, V> {
74    pub fn new(capacity: usize) -> Self {
75        Self {
76            cache: LruCache::new(NonZeroUsize::new(capacity.max(1)).unwrap()),
77        }
78    }
79
80    pub fn get_cloned(&mut self, key: &K) -> Option<V> {
81        let now = Instant::now();
82        if let Some(entry) = self.cache.get(key) {
83            if entry.expires_at > now {
84                return Some(entry.value.clone());
85            }
86        }
87        self.cache.pop(key);
88        None
89    }
90
91    pub fn put(&mut self, key: K, value: V, ttl: Duration) {
92        self.cache.put(
93            key,
94            TimedValue {
95                value,
96                expires_at: Instant::now() + ttl,
97            },
98        );
99    }
100}
101
102pub fn new_lookup_cache<K: Eq + Hash, V: Clone>() -> TimedLruCache<K, V> {
103    TimedLruCache::new(LOOKUP_CACHE_CAPACITY)
104}
105
106#[derive(Debug, Clone)]
107pub struct CachedResolvedPathEntry {
108    pub cid: Cid,
109    pub link_type: LinkType,
110}
111
112#[derive(Debug, Clone)]
113pub struct CachedTreeRootEntry {
114    pub cid: Cid,
115    pub source: &'static str,
116    pub root_event: Option<PeerRootEvent>,
117}
118
119pub type SharedBlobFetch = Shared<BoxFuture<'static, bool>>;
120pub type SharedBlobRead = Shared<BoxFuture<'static, Result<Option<Vec<u8>>, String>>>;
121
122#[derive(Debug, Clone, Copy, PartialEq, Eq)]
123pub enum WsProtocol {
124    HashtreeJson,
125    HashtreeMsgpack,
126    Unknown,
127}
128
129pub struct PendingRequest {
130    pub origin_id: u64,
131    pub hash: String,
132    pub found: bool,
133    pub origin_protocol: WsProtocol,
134}
135
136pub struct UpstreamNostrSubscription {
137    pub close_tx: watch::Sender<bool>,
138    pub tasks: Vec<JoinHandle<()>>,
139}
140
141pub struct WsRelayState {
142    pub clients: Mutex<HashMap<u64, mpsc::UnboundedSender<Message>>>,
143    pub pending: Mutex<HashMap<(u64, u32), PendingRequest>>,
144    pub client_protocols: Mutex<HashMap<u64, WsProtocol>>,
145    pub upstream_nostr_subscriptions: Mutex<HashMap<(u64, String), UpstreamNostrSubscription>>,
146    pub upstream_seen_events: Mutex<HashMap<(u64, String), HashSet<String>>>,
147    pub upstream_pending_eose: Mutex<HashMap<(u64, String), usize>>,
148    pub next_client_id: AtomicU64,
149    pub next_request_id: AtomicU32,
150    pub upstream_relay_bytes_sent: AtomicU64,
151    pub upstream_relay_bytes_received: AtomicU64,
152}
153
154impl WsRelayState {
155    pub fn new() -> Self {
156        Self {
157            clients: Mutex::new(HashMap::new()),
158            pending: Mutex::new(HashMap::new()),
159            client_protocols: Mutex::new(HashMap::new()),
160            upstream_nostr_subscriptions: Mutex::new(HashMap::new()),
161            upstream_seen_events: Mutex::new(HashMap::new()),
162            upstream_pending_eose: Mutex::new(HashMap::new()),
163            next_client_id: AtomicU64::new(1),
164            next_request_id: AtomicU32::new(1),
165            upstream_relay_bytes_sent: AtomicU64::new(0),
166            upstream_relay_bytes_received: AtomicU64::new(0),
167        }
168    }
169
170    pub fn next_id(&self) -> u64 {
171        self.next_client_id.fetch_add(1, Ordering::SeqCst)
172    }
173
174    pub fn next_request_id(&self) -> u32 {
175        self.next_request_id.fetch_add(1, Ordering::SeqCst)
176    }
177
178    pub fn note_upstream_relay_send(&self, bytes: usize) {
179        self.upstream_relay_bytes_sent
180            .fetch_add(bytes as u64, Ordering::Relaxed);
181    }
182
183    pub fn note_upstream_relay_receive(&self, bytes: usize) {
184        self.upstream_relay_bytes_received
185            .fetch_add(bytes as u64, Ordering::Relaxed);
186    }
187
188    pub fn upstream_relay_bandwidth(&self) -> (u64, u64) {
189        (
190            self.upstream_relay_bytes_sent.load(Ordering::Relaxed),
191            self.upstream_relay_bytes_received.load(Ordering::Relaxed),
192        )
193    }
194}
195
196#[derive(Clone)]
197pub struct AppState {
198    pub store: Arc<HashtreeStore>,
199    pub auth: Option<AuthCredentials>,
200    /// Unix timestamp when this daemon state was created.
201    pub daemon_started_at: u64,
202    pub peer_mode: crate::config::ServerMode,
203    pub hash_get_enabled: bool,
204    /// Whether HTTP cache misses should ask connected WebRTC peers before
205    /// falling back to upstream Blossom.
206    pub http_webrtc_fetch: bool,
207    /// WebRTC peer state for forwarding requests to connected P2P peers
208    pub webrtc_peers: Option<Arc<WebRTCState>>,
209    /// FIPS-backed Hashtree blob transport for peer fetches and responses.
210    pub fips_transport: Option<Arc<DaemonFipsTransport>>,
211    pub fetch_from_fips_peers: bool,
212    /// WebSocket relay state for /ws clients
213    pub ws_relay: Arc<WsRelayState>,
214    /// Maximum upload size in bytes for Blossom uploads (default: 5 MB)
215    pub max_upload_bytes: usize,
216    /// Allow anyone with valid Nostr auth to write (default: true)
217    /// When false, only allowed_pubkeys can write
218    pub public_writes: bool,
219    /// Require untrusted cached blob ingress to look like encrypted CHK blobs.
220    pub require_random_untrusted_ingest: bool,
221    /// Return from Blossom upload after validation while storage writes finish in
222    /// a bounded background queue.
223    pub optimistic_blossom_uploads: bool,
224    /// Background upload queue byte budget. Each queued body holds one permit per
225    /// byte until the storage write completes.
226    pub optimistic_upload_queue_bytes: usize,
227    pub optimistic_upload_queue: Arc<Semaphore>,
228    /// Pubkeys allowed to write (hex format, from config allowed_npubs)
229    pub allowed_pubkeys: HashSet<String>,
230    /// Upstream Blossom servers for cascade fetching
231    pub upstream_blossom: Vec<String>,
232    /// Social graph access control
233    pub social_graph: Option<Arc<socialgraph::SocialGraphAccessControl>>,
234    /// Social graph store handle for snapshot export
235    pub social_graph_store: Option<Arc<dyn socialgraph::SocialGraphBackend>>,
236    /// Social graph root pubkey bytes for snapshot export
237    pub social_graph_root: Option<[u8; 32]>,
238    /// Allow public access to social graph snapshot endpoint
239    pub socialgraph_snapshot_public: bool,
240    /// Nostr relay state for /ws and WebRTC Nostr messages
241    pub nostr_relay: Option<Arc<NostrRelay>>,
242    /// Active upstream Nostr relays for HTTP resolver operations.
243    pub nostr_relay_urls: Vec<String>,
244    /// In-process cache for resolved mutable tree roots, keyed by npub/tree(+key)
245    pub tree_root_cache: Arc<StdMutex<HashMap<String, CachedTreeRootEntry>>>,
246    /// Shared in-flight blob fetches so concurrent misses only hit upstream once per hash
247    pub inflight_blob_fetches: Arc<Mutex<HashMap<String, SharedBlobFetch>>>,
248    /// Shared in-flight local blob reads so request bursts for the same hash
249    /// only spend one blocking storage read.
250    pub inflight_blob_reads: Arc<Mutex<HashMap<String, SharedBlobRead>>>,
251    /// Bounded hot cache for immutable blob bodies and metadata probes.
252    pub(super) blob_cache: Arc<BlobCache>,
253    /// Immutable directory listings keyed by CID
254    pub directory_listing_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<Vec<TreeEntry>>>>>,
255    /// Immutable resolved paths keyed by root CID + path
256    pub resolved_path_cache:
257        Arc<StdMutex<TimedLruCache<String, LookupResult<CachedResolvedPathEntry>>>>,
258    /// Immutable thumbnail alias resolutions keyed by root CID + alias path
259    pub thumbnail_path_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<String>>>>,
260    /// Immutable file sizes keyed by CID
261    pub cid_size_cache: Arc<StdMutex<TimedLruCache<String, LookupResult<u64>>>>,
262}
263
264#[derive(Clone)]
265pub struct AuthCredentials {
266    pub username: String,
267    pub password: String,
268}
269
270/// Auth middleware - validates HTTP Basic Auth
271pub async fn auth_middleware(
272    State(state): State<AppState>,
273    request: Request<Body>,
274    next: Next,
275) -> Result<Response<Body>, StatusCode> {
276    // If auth is not enabled, allow request
277    let Some(auth) = &state.auth else {
278        return Ok(next.run(request).await);
279    };
280
281    // Check Authorization header
282    let auth_header = request
283        .headers()
284        .get(header::AUTHORIZATION)
285        .and_then(|v| v.to_str().ok());
286
287    let authorized = if let Some(header_value) = auth_header {
288        if let Some(credentials) = header_value.strip_prefix("Basic ") {
289            use base64::Engine;
290            let engine = base64::engine::general_purpose::STANDARD;
291            if let Ok(decoded) = engine.decode(credentials) {
292                if let Ok(decoded_str) = String::from_utf8(decoded) {
293                    let expected = format!("{}:{}", auth.username, auth.password);
294                    decoded_str == expected
295                } else {
296                    false
297                }
298            } else {
299                false
300            }
301        } else {
302            false
303        }
304    } else {
305        false
306    };
307
308    if authorized {
309        Ok(next.run(request).await)
310    } else {
311        Ok(Response::builder()
312            .status(StatusCode::UNAUTHORIZED)
313            .header(header::WWW_AUTHENTICATE, "Basic realm=\"hashtree\"")
314            .body(Body::from("Unauthorized"))
315            .unwrap())
316    }
317}
hashtree_cli/server/auth.rs

hashtree_cli/server/
auth.rs
