eggress_admin/
server.rs

1use std::sync::atomic::{AtomicBool, AtomicU64};
2use std::sync::{Arc, Mutex};
3use std::time::{Duration, Instant};
4
5use base64::Engine;
6use bytes::Bytes;
7use http_body_util::Full;
8use hyper::service::service_fn;
9use hyper_util::rt::TokioIo;
10use subtle::ConstantTimeEq;
11use tokio::net::TcpListener;
12use tokio::sync::Semaphore;
13use tokio_util::sync::CancellationToken;
14use zeroize::Zeroizing;
15
16use crate::reverse::ReverseRegistry;
17use crate::routes::handle_request;
18use crate::AdminError;
19use eggress_config::compile::{PacConfig, StaticRoute};
20
21/// Upper bound on concurrently served admin connections. The admin surface is
22/// local tooling; an unbounded accept loop would otherwise turn the admin port
23/// into a trivially cheap DoS target.
24const MAX_ADMIN_CONNECTIONS: usize = 64;
25const AUTH_FAILURE_LIMIT: u32 = 5;
26const AUTH_FAILURE_WINDOW: Duration = Duration::from_secs(60);
27const AUTH_FAILURE_BLOCK: Duration = Duration::from_secs(30);
28const MAX_AUTH_FAILURE_ENTRIES: usize = 4096;
29
30#[derive(Default)]
31struct AuthFailure {
32    window_started: Option<Instant>,
33    failures: u32,
34    blocked_until: Option<Instant>,
35}
36
37#[derive(Default)]
38struct AuthFailureLimiter {
39    entries: Mutex<std::collections::HashMap<std::net::IpAddr, AuthFailure>>,
40}
41
42impl AuthFailureLimiter {
43    fn lock_entries(
44        &self,
45    ) -> std::sync::MutexGuard<'_, std::collections::HashMap<std::net::IpAddr, AuthFailure>> {
46        self.entries.lock().unwrap_or_else(|e| {
47            tracing::warn!("auth failure limiter was poisoned; clearing it: {e}");
48            let mut entries = e.into_inner();
49            entries.clear();
50            self.entries.clear_poison();
51            entries
52        })
53    }
54
55    fn blocked_for(&self, ip: std::net::IpAddr) -> Option<Duration> {
56        let entries = self.lock_entries();
57        entries.get(&ip).and_then(|entry| {
58            entry
59                .blocked_until
60                .and_then(|until| until.checked_duration_since(Instant::now()))
61        })
62    }
63
64    fn record_failure(&self, ip: std::net::IpAddr) {
65        let now = Instant::now();
66        let mut entries = self.lock_entries();
67        if entries.len() >= MAX_AUTH_FAILURE_ENTRIES && !entries.contains_key(&ip) {
68            entries.retain(|_, entry| {
69                entry
70                    .window_started
71                    .is_some_and(|started| now.duration_since(started) < AUTH_FAILURE_WINDOW)
72            });
73            if entries.len() >= MAX_AUTH_FAILURE_ENTRIES {
74                // Evict the entry with the oldest window_started (LRU-by-time)
75                // rather than HashMap's arbitrary iteration order.
76                let victim = entries
77                    .iter()
78                    .min_by_key(|(_, entry)| entry.window_started.unwrap_or(now))
79                    .map(|(ip, _)| *ip);
80                if let Some(victim) = victim {
81                    entries.remove(&victim);
82                }
83            }
84        }
85        let entry = entries.entry(ip).or_default();
86        if entry
87            .window_started
88            .is_none_or(|started| now.duration_since(started) >= AUTH_FAILURE_WINDOW)
89        {
90            entry.window_started = Some(now);
91            entry.failures = 0;
92        }
93        entry.failures = entry.failures.saturating_add(1);
94        if entry.failures >= AUTH_FAILURE_LIMIT {
95            entry.blocked_until = Some(now + AUTH_FAILURE_BLOCK);
96        }
97    }
98
99    fn record_success(&self, ip: std::net::IpAddr) {
100        self.lock_entries().remove(&ip);
101    }
102}
103
104pub struct AdminServer {
105    pub(crate) listener: TcpListener,
106    cancel: CancellationToken,
107}
108
109fn authorized(
110    req: &http::Request<hyper::body::Incoming>,
111    auth: &eggress_config::compile::AdminAuthConfig,
112) -> bool {
113    if let Some(expected) = auth.bearer_token.as_deref() {
114        return authorization_payload(req, "Bearer")
115            .is_some_and(|token| token.as_bytes().ct_eq(expected.as_bytes()).unwrap_u8() == 1);
116    }
117
118    let Some(username) = auth.basic_username.as_deref() else {
119        return false;
120    };
121    let Some(password) = auth.basic_password.as_deref() else {
122        return false;
123    };
124    authorization_payload(req, "Basic")
125        .and_then(|value| {
126            let value = Zeroizing::new(
127                base64::engine::general_purpose::STANDARD
128                    .decode(value)
129                    .ok()?,
130            );
131            let separator = value.iter().position(|&byte| byte == b':')?;
132            Some((value, separator))
133        })
134        .is_some_and(|(value, separator)| {
135            (value[..separator].ct_eq(username.as_bytes())
136                & value[separator + 1..].ct_eq(password.as_bytes()))
137            .unwrap_u8()
138                == 1
139        })
140}
141
142fn authorization_payload<'a>(
143    req: &'a http::Request<hyper::body::Incoming>,
144    scheme: &str,
145) -> Option<&'a str> {
146    req.headers()
147        .get(http::header::AUTHORIZATION)
148        .and_then(|value| value.to_str().ok())
149        .and_then(|value| {
150            let (actual_scheme, payload) = value.split_once(' ')?;
151            actual_scheme
152                .eq_ignore_ascii_case(scheme)
153                .then_some(payload)
154        })
155}
156
157impl AdminServer {
158    pub async fn new(bind: &str, cancel: CancellationToken) -> Result<Self, AdminError> {
159        let listener = TcpListener::bind(bind).await?;
160        if let Ok(addr) = listener.local_addr() {
161            if !addr.ip().is_loopback() {
162                tracing::warn!(
163                    "admin listener bound to non-loopback address {addr}: \
164                     status, metrics, and topology are exposed to the network; \
165                     prefer a loopback bind or configure admin auth"
166                );
167            }
168        }
169        Ok(Self { listener, cancel })
170    }
171
172    pub fn local_addr(&self) -> Result<std::net::SocketAddr, std::io::Error> {
173        self.listener.local_addr()
174    }
175
176    pub async fn run(self, state: AdminState) -> Result<(), AdminError> {
177        let permits = Arc::new(Semaphore::new(MAX_ADMIN_CONNECTIONS));
178        let auth_failures = Arc::new(AuthFailureLimiter::default());
179        loop {
180            tokio::select! {
181                result = self.listener.accept() => {
182                    let (stream, addr) = result.map_err(|e| AdminError::Accept(e.to_string()))?;
183                    let Ok(permit) = Arc::clone(&permits).try_acquire_owned() else {
184                        tracing::warn!("admin connection limit ({MAX_ADMIN_CONNECTIONS}) reached; rejecting connection");
185                        drop(stream);
186                        continue;
187                    };
188                    let state = state.clone();
189                    let auth_failures = auth_failures.clone();
190                    tokio::spawn(async move {
191                        let _permit = permit;
192                        let service = service_fn(move |req| {
193                            let state = state.clone();
194                            let auth_failures = auth_failures.clone();
195                            async move {
196                                let response = match state.auth.as_ref() {
197                                    Some(auth) if !authorized(&req, auth) => {
198                                        let blocked_for = auth_failures.blocked_for(addr.ip());
199                                        if blocked_for.is_none() {
200                                            auth_failures.record_failure(addr.ip());
201                                        }
202                                        let mut response = http::Response::new(Full::new(
203                                            Bytes::from_static(if blocked_for.is_some() {
204                                                b"too many authentication failures"
205                                            } else {
206                                                b"unauthorized"
207                                            }),
208                                        ));
209                                        *response.status_mut() = if blocked_for.is_some() {
210                                            http::StatusCode::TOO_MANY_REQUESTS
211                                        } else {
212                                            http::StatusCode::UNAUTHORIZED
213                                        };
214                                        response.headers_mut().insert(
215                                            http::header::WWW_AUTHENTICATE,
216                                            http::HeaderValue::from_static("Bearer, Basic"),
217                                        );
218                                        if let Some(duration) = blocked_for {
219                                            let seconds = duration.as_secs().saturating_add(1).max(1);
220                                            if let Ok(value) = http::HeaderValue::from_str(&seconds.to_string()) {
221                                                response.headers_mut().insert(http::header::RETRY_AFTER, value);
222                                            }
223                                        }
224                                        response.headers_mut().insert(
225                                            http::header::CONTENT_TYPE,
226                                            http::HeaderValue::from_static("text/plain"),
227                                        );
228                                        response
229                                    }
230                                    _ => {
231                                        auth_failures.record_success(addr.ip());
232                                        handle_request(req, &state).await
233                                    }
234                                };
235                                Ok::<_, std::convert::Infallible>(response)
236                            }
237                        });
238                        let conn = hyper::server::conn::http1::Builder::new()
239                            .serve_connection(TokioIo::new(stream), service);
240                        match tokio::time::timeout(
241                            std::time::Duration::from_secs(30),
242                            conn,
243                        )
244                        .await
245                        {
246                            Err(_) => {
247                                tracing::debug!("admin connection timed out");
248                            }
249                            Ok(Err(e)) => {
250                                tracing::debug!("admin connection error: {e}");
251                            }
252                            Ok(Ok(())) => {}
253                        }
254                    });
255                }
256                _ = self.cancel.cancelled() => {
257                    break;
258                }
259            }
260        }
261        Ok(())
262    }
263}
264
265/// Live data the admin server reads per request.
266///
267/// Implementations wrap the current `CompiledRuntimeSnapshot` so reloads are
268/// reflected on the next request without restarting the admin server.
269#[derive(Clone)]
270pub struct AdminSnapshot {
271    pub generation: u64,
272    pub router: Arc<eggress_routing::Router>,
273    pub pac: Option<PacConfig>,
274    pub static_routes: Vec<StaticRoute>,
275    pub listeners: Vec<ListenerInfo>,
276}
277
278/// Source of admin-visible live data. Implemented by the runtime so that
279/// reloads immediately take effect on admin endpoints.
280pub trait AdminSnapshotProvider: Send + Sync + 'static {
281    fn snapshot(&self) -> AdminSnapshot;
282
283    fn generation(&self) -> u64 {
284        self.snapshot().generation
285    }
286}
287
288/// A `AdminSnapshotProvider` backed by a fixed snapshot. Useful in tests
289/// that exercise admin endpoints without a full runtime.
290pub struct StaticAdminSnapshot {
291    pub snapshot: AdminSnapshot,
292}
293
294impl AdminSnapshotProvider for StaticAdminSnapshot {
295    fn snapshot(&self) -> AdminSnapshot {
296        self.snapshot.clone()
297    }
298}
299
300#[derive(Clone)]
301pub struct AdminState {
302    pub metrics: Arc<eggress_metrics::MetricsRegistry>,
303    pub start_time: Instant,
304    pub readiness: Arc<AtomicBool>,
305    pub active_connections: Option<Arc<AtomicU64>>,
306    pub provider: Arc<dyn AdminSnapshotProvider>,
307    pub udp_registry: Arc<eggress_udp::registry::UdpAssociationRegistry>,
308    /// Registry of reverse servers. Empty by default — populating it
309    /// enables the `/-/reverse` admin route.
310    pub reverse_registry: Arc<ReverseRegistry>,
311    /// Whether the `/metrics` endpoint is enabled.
312    pub metrics_enabled: bool,
313    pub auth: Option<eggress_config::compile::AdminAuthConfig>,
314}
315
316#[cfg(test)]
317mod tests {
318    use super::{AuthFailureLimiter, AUTH_FAILURE_LIMIT};
319    use std::net::{IpAddr, Ipv4Addr};
320
321    #[test]
322    fn auth_failure_limiter_blocks_and_resets() {
323        let limiter = AuthFailureLimiter::default();
324        let ip = IpAddr::V4(Ipv4Addr::LOCALHOST);
325
326        for _ in 0..AUTH_FAILURE_LIMIT {
327            limiter.record_failure(ip);
328        }
329        assert!(limiter.blocked_for(ip).is_some());
330
331        limiter.record_success(ip);
332        assert!(limiter.blocked_for(ip).is_none());
333    }
334}
335
336impl AdminState {
337    pub fn snapshot(&self) -> AdminSnapshot {
338        self.provider.snapshot()
339    }
340
341    pub fn generation(&self) -> u64 {
342        self.provider.generation()
343    }
344}
345
346pub type AdminResponse = http::Response<Full<Bytes>>;
347
348#[derive(Debug, Clone, serde::Serialize)]
349pub struct ListenerInfo {
350    pub name: String,
351    pub bind: String,
352    pub local_addr: String,
353    pub protocols: Vec<String>,
354    pub udp_enabled: bool,
355    #[serde(skip_serializing_if = "Option::is_none")]
356    pub mode: Option<String>,
357    #[serde(skip_serializing_if = "Option::is_none")]
358    pub capability_status: Option<String>,
359    #[serde(skip_serializing_if = "Option::is_none")]
360    pub original_dst_support: Option<bool>,
361    #[serde(skip_serializing_if = "Option::is_none")]
362    pub unix_socket_path: Option<String>,
363    #[serde(skip_serializing_if = "Option::is_none")]
364    pub unix_socket_unlink_existing: Option<bool>,
365}
366
367pub fn build_response(status: u16, body: impl Into<Bytes>, content_type: &str) -> AdminResponse {
368    let status = if (100..=599).contains(&status) {
369        match http::StatusCode::from_u16(status) {
370            Ok(status) => status,
371            Err(_) => http::StatusCode::INTERNAL_SERVER_ERROR,
372        }
373    } else {
374        http::StatusCode::INTERNAL_SERVER_ERROR
375    };
376    let content_type = http::HeaderValue::from_str(content_type)
377        .unwrap_or_else(|_| http::HeaderValue::from_static("application/octet-stream"));
378    let mut response = http::Response::new(Full::new(body.into()));
379    *response.status_mut() = status;
380    response
381        .headers_mut()
382        .insert(http::header::CONTENT_TYPE, content_type);
383    response
384}
385
386pub fn build_json_response(status: u16, body: impl Into<Bytes>) -> AdminResponse {
387    build_response(status, body, "application/json")
388}
389
390pub fn build_text_response(status: u16, body: impl Into<Bytes>) -> AdminResponse {
391    build_response(status, body, "text/plain")
392}
393
394pub fn build_not_found() -> AdminResponse {
395    build_text_response(404, "not found")
396}
eggress_admin/server.rs

eggress_admin/
server.rs
