eggress_admin/
server.rs

1use std::sync::atomic::{AtomicBool, AtomicU64};
2use std::sync::Arc;
3use std::time::Instant;
4
5use base64::Engine;
6use bytes::Bytes;
7use http_body_util::Full;
8use hyper::service::service_fn;
9use hyper_util::rt::TokioIo;
10use tokio::net::TcpListener;
11use tokio_util::sync::CancellationToken;
12
13use crate::reverse::ReverseRegistry;
14use crate::routes::handle_request;
15use crate::AdminError;
16use eggress_config::compile::{PacConfig, StaticRoute};
17
18pub struct AdminServer {
19    pub(crate) listener: TcpListener,
20    cancel: CancellationToken,
21}
22
23fn authorized(
24    req: &http::Request<hyper::body::Incoming>,
25    auth: &eggress_config::compile::AdminAuthConfig,
26) -> bool {
27    if let Some(expected) = auth.bearer_token.as_deref() {
28        return authorization_payload(req, "Bearer").is_some_and(|token| token == expected);
29    }
30
31    let Some(username) = auth.basic_username.as_deref() else {
32        return false;
33    };
34    let Some(password) = auth.basic_password.as_deref() else {
35        return false;
36    };
37    authorization_payload(req, "Basic")
38        .and_then(|value| base64::engine::general_purpose::STANDARD.decode(value).ok())
39        .and_then(|value| String::from_utf8(value).ok())
40        .and_then(|value| {
41            value
42                .split_once(':')
43                .map(|(user, pass)| (user.to_string(), pass.to_string()))
44        })
45        .is_some_and(|(user, pass)| user == username && pass == password)
46}
47
48fn authorization_payload<'a>(
49    req: &'a http::Request<hyper::body::Incoming>,
50    scheme: &str,
51) -> Option<&'a str> {
52    req.headers()
53        .get(http::header::AUTHORIZATION)
54        .and_then(|value| value.to_str().ok())
55        .and_then(|value| {
56            let (actual_scheme, payload) = value.split_once(' ')?;
57            actual_scheme
58                .eq_ignore_ascii_case(scheme)
59                .then_some(payload)
60        })
61}
62
63impl AdminServer {
64    pub async fn new(bind: &str, cancel: CancellationToken) -> Result<Self, AdminError> {
65        let listener = TcpListener::bind(bind).await?;
66        Ok(Self { listener, cancel })
67    }
68
69    pub fn local_addr(&self) -> Result<std::net::SocketAddr, std::io::Error> {
70        self.listener.local_addr()
71    }
72
73    pub async fn run(self, state: AdminState) -> Result<(), AdminError> {
74        loop {
75            tokio::select! {
76                result = self.listener.accept() => {
77                    let (stream, _addr) = result.map_err(|e| AdminError::Accept(e.to_string()))?;
78                    let state = state.clone();
79                    tokio::spawn(async move {
80                        let service = service_fn(move |req| {
81                            let state = state.clone();
82                            async move {
83                                let response = match state.auth.as_ref() {
84                                    Some(auth) if !authorized(&req, auth) => {
85                                        http::Response::builder()
86                                            .status(401)
87                                            .header(http::header::WWW_AUTHENTICATE, "Bearer, Basic")
88                                            .header(http::header::CONTENT_TYPE, "text/plain")
89                                            .body(Full::new(Bytes::from_static(b"unauthorized")))
90                                            .expect("static admin auth response")
91                                    }
92                                    _ => handle_request(req, &state).await,
93                                };
94                                Ok::<_, std::convert::Infallible>(response)
95                            }
96                        });
97                        let conn = hyper::server::conn::http1::Builder::new()
98                            .serve_connection(TokioIo::new(stream), service);
99                        match tokio::time::timeout(
100                            std::time::Duration::from_secs(30),
101                            conn,
102                        )
103                        .await
104                        {
105                            Err(_) => {
106                                tracing::debug!("admin connection timed out");
107                            }
108                            Ok(Err(e)) => {
109                                tracing::debug!("admin connection error: {e}");
110                            }
111                            Ok(Ok(())) => {}
112                        }
113                    });
114                }
115                _ = self.cancel.cancelled() => {
116                    break;
117                }
118            }
119        }
120        Ok(())
121    }
122}
123
124/// Live data the admin server reads per request.
125///
126/// Implementations wrap the current `CompiledRuntimeSnapshot` so reloads are
127/// reflected on the next request without restarting the admin server.
128#[derive(Clone)]
129pub struct AdminSnapshot {
130    pub generation: u64,
131    pub router: Arc<eggress_routing::Router>,
132    pub pac: Option<PacConfig>,
133    pub static_routes: Vec<StaticRoute>,
134    pub listeners: Vec<ListenerInfo>,
135}
136
137/// Source of admin-visible live data. Implemented by the runtime so that
138/// reloads immediately take effect on admin endpoints.
139pub trait AdminSnapshotProvider: Send + Sync + 'static {
140    fn snapshot(&self) -> AdminSnapshot;
141}
142
143/// A `AdminSnapshotProvider` backed by a fixed snapshot. Useful in tests
144/// that exercise admin endpoints without a full runtime.
145pub struct StaticAdminSnapshot {
146    pub snapshot: AdminSnapshot,
147}
148
149impl AdminSnapshotProvider for StaticAdminSnapshot {
150    fn snapshot(&self) -> AdminSnapshot {
151        self.snapshot.clone()
152    }
153}
154
155#[derive(Clone)]
156pub struct AdminState {
157    pub metrics: Arc<eggress_metrics::MetricsRegistry>,
158    pub start_time: Instant,
159    pub readiness: Arc<AtomicBool>,
160    pub active_connections: Option<Arc<AtomicU64>>,
161    pub provider: Arc<dyn AdminSnapshotProvider>,
162    pub udp_registry: Arc<eggress_udp::registry::UdpAssociationRegistry>,
163    /// Registry of reverse servers. Empty by default — populating it
164    /// enables the `/-/reverse` admin route.
165    pub reverse_registry: Arc<ReverseRegistry>,
166    /// Whether the `/metrics` endpoint is enabled.
167    pub metrics_enabled: bool,
168    pub auth: Option<eggress_config::compile::AdminAuthConfig>,
169}
170
171impl AdminState {
172    pub fn snapshot(&self) -> AdminSnapshot {
173        self.provider.snapshot()
174    }
175
176    pub fn generation(&self) -> u64 {
177        self.provider.snapshot().generation
178    }
179}
180
181pub type AdminResponse = http::Response<Full<Bytes>>;
182
183#[derive(Debug, Clone, serde::Serialize)]
184pub struct ListenerInfo {
185    pub name: String,
186    pub bind: String,
187    pub local_addr: String,
188    pub protocols: Vec<String>,
189    pub udp_enabled: bool,
190    #[serde(skip_serializing_if = "Option::is_none")]
191    pub mode: Option<String>,
192    #[serde(skip_serializing_if = "Option::is_none")]
193    pub capability_status: Option<String>,
194    #[serde(skip_serializing_if = "Option::is_none")]
195    pub original_dst_support: Option<bool>,
196    #[serde(skip_serializing_if = "Option::is_none")]
197    pub unix_socket_path: Option<String>,
198    #[serde(skip_serializing_if = "Option::is_none")]
199    pub unix_socket_unlink_existing: Option<bool>,
200}
201
202pub fn build_response(status: u16, body: impl Into<Bytes>, content_type: &str) -> AdminResponse {
203    http::Response::builder()
204        .status(status)
205        .header("content-type", content_type)
206        .body(Full::new(body.into()))
207        .unwrap()
208}
209
210pub fn build_json_response(status: u16, body: impl Into<Bytes>) -> AdminResponse {
211    build_response(status, body, "application/json")
212}
213
214pub fn build_text_response(status: u16, body: impl Into<Bytes>) -> AdminResponse {
215    build_response(status, body, "text/plain")
216}
217
218pub fn build_not_found() -> AdminResponse {
219    build_text_response(404, "not found")
220}
eggress_admin/server.rs

eggress_admin/
server.rs
