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