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