1pub mod pac;
2pub mod reverse;
3pub mod routes;
4pub mod server;
5pub mod static_content;
6
7pub use eggress_config::compile::{PacConfig, StaticRoute};
8pub use reverse::{ReverseRegistry, ReverseServerEntry, ReverseServerId};
9pub use server::{
10 AdminServer, AdminSnapshot, AdminSnapshotProvider, AdminState, ListenerInfo,
11 StaticAdminSnapshot,
12};
13
14#[derive(Debug, thiserror::Error)]
15pub enum AdminError {
16 #[error("bind error: {0}")]
17 Bind(#[from] std::io::Error),
18
19 #[error("accept error: {0}")]
20 Accept(String),
21
22 #[error("server error: {0}")]
23 Server(String),
24}
25
26#[cfg(test)]
27mod tests {
28 use super::*;
29 use crate::pac::generate_pac;
30 use crate::server::AdminState;
31 use std::sync::Arc;
32 use std::time::Instant;
33
34 fn snapshot_with_router(router: Arc<eggress_routing::Router>) -> AdminSnapshot {
35 AdminSnapshot {
36 generation: 42,
37 router,
38 pac: Some(PacConfig {
39 path: "/pac".to_string(),
40 proxy_directive: "127.0.0.1:8080".to_string(),
41 direct_fallback: true,
42 direct_hosts: vec!["localhost".to_string()],
43 direct_suffixes: vec!["local".to_string()],
44 }),
45 static_routes: vec![StaticRoute {
46 path: "/test".to_string(),
47 content_type: "text/html".to_string(),
48 body: "<h1>Hello</h1>".to_string(),
49 }],
50 listeners: Vec::new(),
51 }
52 }
53
54 fn test_state() -> AdminState {
55 let router = Arc::new(eggress_routing::Router::new(
56 vec![],
57 eggress_routing::RouteActionSpec::Direct,
58 ));
59 let snap = snapshot_with_router(router);
60 AdminState {
61 metrics: Arc::new(eggress_metrics::MetricsRegistry::new()),
62 start_time: Instant::now(),
63 readiness: Arc::new(std::sync::atomic::AtomicBool::new(true)),
64 active_connections: None,
65 provider: Arc::new(StaticAdminSnapshot { snapshot: snap }),
66 udp_registry: Arc::new(eggress_udp::registry::UdpAssociationRegistry::new(
67 eggress_udp::limits::UdpLimits::default(),
68 )),
69 reverse_registry: Arc::new(ReverseRegistry::new()),
70 metrics_enabled: true,
71 auth: None,
72 }
73 }
74
75 fn test_state_no_pac() -> AdminState {
76 let router = Arc::new(eggress_routing::Router::new(
77 vec![],
78 eggress_routing::RouteActionSpec::Direct,
79 ));
80 let snap = AdminSnapshot {
81 generation: 0,
82 router,
83 pac: None,
84 static_routes: Vec::new(),
85 listeners: Vec::new(),
86 };
87 AdminState {
88 metrics: Arc::new(eggress_metrics::MetricsRegistry::new()),
89 start_time: Instant::now(),
90 readiness: Arc::new(std::sync::atomic::AtomicBool::new(true)),
91 active_connections: None,
92 provider: Arc::new(StaticAdminSnapshot { snapshot: snap }),
93 udp_registry: Arc::new(eggress_udp::registry::UdpAssociationRegistry::new(
94 eggress_udp::limits::UdpLimits::default(),
95 )),
96 reverse_registry: Arc::new(ReverseRegistry::new()),
97 metrics_enabled: true,
98 auth: None,
99 }
100 }
101
102 async fn start_server(state: AdminState) -> String {
103 let cancel = tokio_util::sync::CancellationToken::new();
104 let server = AdminServer::new("127.0.0.1:0", cancel.clone())
105 .await
106 .unwrap();
107 let addr = server.listener.local_addr().unwrap();
108 let bind = addr.to_string();
109 tokio::spawn(async move { server.run(state).await.unwrap() });
110 bind
111 }
112
113 async fn http_get(addr: &str, path: &str) -> (u16, String) {
114 let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
115
116 let request = format!("GET {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n");
117 tokio::io::AsyncWriteExt::write_all(&mut stream, request.as_bytes())
118 .await
119 .unwrap();
120 tokio::io::AsyncWriteExt::flush(&mut stream).await.unwrap();
121
122 let mut response = Vec::new();
123 loop {
124 let mut buf = [0u8; 4096];
125 match tokio::io::AsyncReadExt::read(&mut stream, &mut buf).await {
126 Ok(0) => break,
127 Ok(n) => response.extend_from_slice(&buf[..n]),
128 Err(_) => break,
129 }
130 }
131 let response = String::from_utf8_lossy(&response);
132
133 let status_line = response.lines().next().unwrap_or("");
134 let status = status_line
135 .split_whitespace()
136 .nth(1)
137 .and_then(|s| s.parse::<u16>().ok())
138 .unwrap_or(0);
139
140 let body = response.split("\r\n\r\n").nth(1).unwrap_or("").to_string();
141
142 (status, body)
143 }
144
145 #[tokio::test]
146 async fn health_returns_200() {
147 let state = test_state();
148 let addr = start_server(state).await;
149 let (status, body) = http_get(&addr, "/-/health").await;
150 assert_eq!(status, 200);
151 assert_eq!(body, "ok");
152 }
153
154 #[tokio::test]
155 async fn ready_returns_200() {
156 let state = test_state();
157 let addr = start_server(state).await;
158 let (status, body) = http_get(&addr, "/-/ready").await;
159 assert_eq!(status, 200);
160 assert_eq!(body, "ready");
161 }
162
163 #[tokio::test]
164 async fn status_returns_valid_json() {
165 let state = test_state();
166 let addr = start_server(state).await;
167 let (status, body) = http_get(&addr, "/-/status").await;
168 assert_eq!(status, 200);
169 let json: serde_json::Value = serde_json::from_str(&body).unwrap();
170 assert_eq!(json["version"], env!("CARGO_PKG_VERSION"));
171 assert_eq!(json["generation"], 42);
172 assert!(json["uptime_seconds"].is_number());
173 }
174
175 #[tokio::test]
176 async fn metrics_returns_prometheus_format() {
177 let state = test_state();
178 let addr = start_server(state).await;
179 let (status, body) = http_get(&addr, "/metrics").await;
180 assert_eq!(status, 200);
181 assert!(body.contains("eggress_connections_active"));
182 assert!(body.contains("eggress_connections_total"));
183 }
184
185 #[tokio::test]
186 async fn configured_auth_rejects_unauthenticated_requests() {
187 let mut state = test_state();
188 state.auth = Some(eggress_config::compile::AdminAuthConfig {
189 bearer_token: Some("secret".to_string()),
190 basic_username: None,
191 basic_password: None,
192 });
193 let addr = start_server(state).await;
194 let (status, body) = http_get(&addr, "/-/status").await;
195 assert_eq!(status, 401);
196 assert_eq!(body, "unauthorized");
197 }
198
199 #[tokio::test]
200 async fn static_content_serves_correct_content_type_and_body() {
201 let state = test_state();
202 let addr = start_server(state).await;
203 let (status, body) = http_get(&addr, "/test").await;
204 assert_eq!(status, 200);
205 assert_eq!(body, "<h1>Hello</h1>");
206 }
207
208 #[tokio::test]
209 async fn unknown_path_returns_404() {
210 let state = test_state();
211 let addr = start_server(state).await;
212 let (status, body) = http_get(&addr, "/nonexistent").await;
213 assert_eq!(status, 404);
214 assert_eq!(body, "not found");
215 }
216
217 #[tokio::test]
218 async fn pac_endpoint_returns_pac_when_configured() {
219 let state = test_state();
220 let addr = start_server(state).await;
221 let (status, body) = http_get(&addr, "/pac").await;
222 assert_eq!(status, 200);
223 assert!(body.contains("function FindProxyForURL"));
224 }
225
226 #[tokio::test]
227 async fn pac_endpoint_returns_404_when_not_configured() {
228 let state = test_state_no_pac();
229 let addr = start_server(state).await;
230 let (status, body) = http_get(&addr, "/pac").await;
231 assert_eq!(status, 404);
232 assert!(body.contains("pac not configured"));
233 }
234
235 #[tokio::test]
236 async fn routes_endpoint_returns_json() {
237 let state = test_state();
238 let addr = start_server(state).await;
239 let (status, body) = http_get(&addr, "/-/routes").await;
240 assert_eq!(status, 200);
241 let json: serde_json::Value = serde_json::from_str(&body).unwrap();
242 assert!(json.get("rules").is_some());
243 assert!(json.get("default_action").is_some());
244 assert_eq!(json["rule_count"], 0);
245 }
246
247 #[tokio::test]
248 async fn upstreams_endpoint_returns_json() {
249 let state = test_state();
250 let addr = start_server(state).await;
251 let (status, body) = http_get(&addr, "/-/upstreams").await;
252 assert_eq!(status, 200);
253 let json: serde_json::Value = serde_json::from_str(&body).unwrap();
254 assert!(json.is_array());
255 }
256
257 #[tokio::test]
258 async fn reverse_endpoint_empty_when_no_servers() {
259 let state = test_state();
260 let addr = start_server(state).await;
261 let (status, body) = http_get(&addr, "/-/reverse").await;
262 assert_eq!(status, 200);
263 let json: serde_json::Value = serde_json::from_str(&body).unwrap();
264 assert_eq!(json["totals"]["server_count"], 0);
265 assert!(json["servers"].is_array());
266 assert_eq!(json["servers"].as_array().unwrap().len(), 0);
267 }
268
269 #[tokio::test]
270 async fn reverse_endpoint_reports_registered_server_state() {
271 use crate::reverse::{ReverseRegistry, ReverseServerEntry, ReverseServerId};
272 use eggress_protocol_reverse::server::ReverseServerState;
273
274 let state = test_state();
275 let registry = ReverseRegistry::new();
276 let server_state = Arc::new(ReverseServerState::default());
277 server_state
278 .active_control
279 .store(3, std::sync::atomic::Ordering::Relaxed);
280 server_state
281 .active_streams
282 .store(7, std::sync::atomic::Ordering::Relaxed);
283 server_state
284 .denied_bind
285 .store(1, std::sync::atomic::Ordering::Relaxed);
286 server_state
287 .dropped_stream_limit
288 .store(2, std::sync::atomic::Ordering::Relaxed);
289 server_state
290 .dropped_pending_limit
291 .store(3, std::sync::atomic::Ordering::Relaxed);
292 registry.register(ReverseServerEntry {
293 id: ReverseServerId::from("rev-1"),
294 control_bind: "127.0.0.1:8080".to_string(),
295 state: server_state,
296 });
297 let mut state_with_rev = state;
298 state_with_rev.reverse_registry = Arc::new(registry);
299
300 let addr = start_server(state_with_rev).await;
301 let (status, body) = http_get(&addr, "/-/reverse").await;
302 assert_eq!(status, 200);
303 let json: serde_json::Value = serde_json::from_str(&body).unwrap();
304 assert_eq!(json["totals"]["server_count"], 1);
305 let servers = json["servers"].as_array().unwrap();
306 assert_eq!(servers.len(), 1);
307 assert_eq!(servers[0]["id"], "rev-1");
308 assert_eq!(servers[0]["control_bind"], "127.0.0.1:8080");
309 assert_eq!(servers[0]["active_control"], 3);
310 assert_eq!(servers[0]["active_streams"], 7);
311 assert_eq!(servers[0]["denied_bind"], 1);
312 assert_eq!(servers[0]["dropped_stream_limit"], 2);
313 assert_eq!(servers[0]["dropped_pending_limit"], 3);
314 }
315
316 #[test]
317 fn pac_generation_produces_valid_javascript() {
318 let config = PacConfig {
319 path: "/pac".to_string(),
320 proxy_directive: "127.0.0.1:8080".to_string(),
321 direct_fallback: true,
322 direct_hosts: vec!["localhost".to_string(), "127.0.0.1".to_string()],
323 direct_suffixes: vec!["local".to_string(), "internal".to_string()],
324 };
325 let pac = generate_pac(&config);
326 assert!(pac.contains("function FindProxyForURL(url, host)"));
327 assert!(pac.contains("isPlainHostName(host)"));
328 assert!(pac.contains("PROXY 127.0.0.1:8080; DIRECT"));
329 assert!(pac.contains("localhost"));
330 assert!(pac.contains("local"));
331 assert!(pac.contains("internal"));
332 }
333
334 #[test]
335 fn pac_escaping_of_quotes_and_backslashes() {
336 let config = PacConfig {
337 path: "/pac".to_string(),
338 proxy_directive: "proxy\"example.com:8080".to_string(),
339 direct_fallback: false,
340 direct_hosts: vec!["test\\host".to_string()],
341 direct_suffixes: vec![],
342 };
343 let pac = generate_pac(&config);
344 assert!(pac.contains("test\\\\host"));
345 assert!(pac.contains("proxy\\\"example.com:8080"));
346 }
347
348 #[test]
349 fn pac_no_fallback_when_disabled() {
350 let config = PacConfig {
351 path: "/pac".to_string(),
352 proxy_directive: "proxy:8080".to_string(),
353 direct_fallback: false,
354 direct_hosts: vec![],
355 direct_suffixes: vec![],
356 };
357 let pac = generate_pac(&config);
358 assert!(pac.contains("return \"PROXY proxy:8080\";"));
359 assert!(!pac.contains("; DIRECT"));
360 }
361
362 #[test]
363 fn pac_direct_hosts_sorted() {
364 let config = PacConfig {
365 path: "/pac".to_string(),
366 proxy_directive: "proxy:8080".to_string(),
367 direct_fallback: true,
368 direct_hosts: vec!["z.com".to_string(), "a.com".to_string()],
369 direct_suffixes: vec![],
370 };
371 let pac = generate_pac(&config);
372 let a_pos = pac.find("a.com").unwrap();
373 let z_pos = pac.find("z.com").unwrap();
374 assert!(a_pos < z_pos, "hosts should be sorted");
375 }
376
377 #[tokio::test]
378 async fn ready_returns_503_when_not_ready() {
379 let mut state = test_state();
380 state.readiness = Arc::new(std::sync::atomic::AtomicBool::new(false));
381 let addr = start_server(state).await;
382 let (status, body) = http_get(&addr, "/-/ready").await;
383 assert_eq!(status, 503);
384 assert_eq!(body, "not ready");
385 }
386
387 #[tokio::test]
388 async fn readiness_becomes_false_before_drain() {
389 let mut state = test_state();
390 state.readiness = Arc::new(std::sync::atomic::AtomicBool::new(true));
391 let addr = start_server(state.clone()).await;
392 let (status, body) = http_get(&addr, "/-/ready").await;
393 assert_eq!(status, 200);
394 assert_eq!(body, "ready");
395
396 state
397 .readiness
398 .store(false, std::sync::atomic::Ordering::Relaxed);
399 let (status, body) = http_get(&addr, "/-/ready").await;
400 assert_eq!(status, 503);
401 assert_eq!(body, "not ready");
402 }
403}