soli_proxy/admin/
mod.rs

1pub mod handlers;
2
3use crate::app::AppManager;
4use crate::circuit_breaker::SharedCircuitBreaker;
5use crate::config::ConfigManager;
6use crate::metrics::SharedMetrics;
7use anyhow::Result;
8use bytes::Bytes;
9use http_body_util::BodyExt;
10use hyper::body::Incoming;
11use hyper::service::service_fn;
12use hyper::{Method, Request, Response};
13use hyper_util::client::legacy::connect::HttpConnector;
14use hyper_util::client::legacy::Client;
15use hyper_util::rt::{TokioExecutor, TokioIo};
16use std::sync::Arc;
17use std::time::Instant;
18use tokio::io::{AsyncReadExt, AsyncWriteExt};
19use tokio::net::{TcpListener, TcpStream};
20
21type BoxBody = http_body_util::combinators::BoxBody<Bytes, std::convert::Infallible>;
22
23pub struct AdminState {
24    pub config_manager: Arc<ConfigManager>,
25    pub metrics: SharedMetrics,
26    pub start_time: Instant,
27    pub circuit_breaker: SharedCircuitBreaker,
28    pub app_manager: Option<Arc<AppManager>>,
29}
30
31pub(crate) fn cors_headers(
32    builder: hyper::http::response::Builder,
33) -> hyper::http::response::Builder {
34    builder
35        .header("Access-Control-Allow-Origin", "*")
36        .header(
37            "Access-Control-Allow-Methods",
38            "GET, POST, PUT, DELETE, OPTIONS",
39        )
40        .header("Access-Control-Allow-Headers", "Content-Type, X-Api-Key")
41}
42
43fn json_response(status: u16, body: serde_json::Value) -> Response<BoxBody> {
44    let bytes = Bytes::from(serde_json::to_string(&body).unwrap());
45    cors_headers(Response::builder())
46        .status(status)
47        .header("Content-Type", "application/json")
48        .body(http_body_util::Full::new(bytes).boxed())
49        .unwrap()
50}
51
52fn ok_response(data: serde_json::Value) -> Response<BoxBody> {
53    json_response(200, serde_json::json!({ "ok": true, "data": data }))
54}
55
56fn created_response(data: serde_json::Value) -> Response<BoxBody> {
57    json_response(201, serde_json::json!({ "ok": true, "data": data }))
58}
59
60fn no_content_response() -> Response<BoxBody> {
61    cors_headers(Response::builder())
62        .status(204)
63        .body(http_body_util::Full::new(Bytes::new()).boxed())
64        .unwrap()
65}
66
67fn preflight_response() -> Response<BoxBody> {
68    cors_headers(Response::builder())
69        .status(204)
70        .header("Access-Control-Max-Age", "86400")
71        .body(http_body_util::Full::new(Bytes::new()).boxed())
72        .unwrap()
73}
74
75fn error_response(status: u16, message: &str) -> Response<BoxBody> {
76    json_response(status, serde_json::json!({ "ok": false, "error": message }))
77}
78
79fn check_auth(req: &Request<Incoming>, api_key: &Option<String>) -> bool {
80    match api_key {
81        None => true, // No auth configured = dev mode
82        Some(key) if key.is_empty() => true,
83        Some(key) => req
84            .headers()
85            .get("X-Api-Key")
86            .and_then(|v| v.to_str().ok())
87            .is_some_and(|v| v == key),
88    }
89}
90
91/// Extract route index from path like /api/v1/routes/3
92fn extract_route_index(path: &str) -> Option<usize> {
93    path.strip_prefix("/api/v1/routes/")
94        .and_then(|s| s.parse::<usize>().ok())
95}
96
97async fn handle_admin_request(
98    req: Request<Incoming>,
99    state: Arc<AdminState>,
100) -> Result<Response<BoxBody>, std::convert::Infallible> {
101    let api_key = state.config_manager.get_config().admin.api_key.clone();
102    if !check_auth(&req, &api_key) {
103        return Ok(error_response(401, "Unauthorized"));
104    }
105
106    let method = req.method().clone();
107    let path = req.uri().path().to_string();
108
109    // Handle CORS preflight
110    if method == Method::OPTIONS {
111        return Ok(preflight_response());
112    }
113
114    let response = match (method.clone(), path.as_str()) {
115        // Phase 1: Read-only endpoints
116        (Method::GET, "/api/v1/status") => handlers::get_status(&state),
117        (Method::GET, "/api/v1/config") => handlers::get_config(&state),
118        (Method::GET, "/api/v1/routes") => handlers::get_routes(&state),
119        (Method::GET, "/api/v1/metrics") => handlers::get_metrics(&state),
120        (Method::GET, "/api/v1/app-metrics") => handlers::get_all_app_metrics(&state),
121        (Method::POST, "/api/v1/reload") => handlers::post_reload(&state).await,
122
123        // App management endpoints
124        (Method::GET, "/api/v1/apps") => handlers::get_apps(&state).await,
125        (_, p) if p.starts_with("/api/v1/apps/") => {
126            let app_name = p.strip_prefix("/api/v1/apps/").unwrap_or("");
127            if method == Method::GET && !app_name.is_empty() && !app_name.contains('/') {
128                handlers::get_app(&state, app_name).await
129            } else if method == Method::GET && app_name.ends_with("/metrics") {
130                let name = app_name.strip_suffix("/metrics").unwrap_or("");
131                if name.is_empty() {
132                    error_response(400, "Invalid app name")
133                } else {
134                    handlers::get_app_metrics(&state, name).await
135                }
136            } else if app_name.ends_with("/deploy") {
137                let name = app_name.strip_suffix("/deploy").unwrap_or("");
138                if name.is_empty() {
139                    error_response(400, "Invalid app name")
140                } else {
141                    handlers::post_app_deploy(&state, name).await
142                }
143            } else if app_name.ends_with("/restart") {
144                let name = app_name.strip_suffix("/restart").unwrap_or("");
145                if name.is_empty() {
146                    error_response(400, "Invalid app name")
147                } else {
148                    handlers::post_app_restart(&state, name).await
149                }
150            } else if app_name.ends_with("/rollback") {
151                let name = app_name.strip_suffix("/rollback").unwrap_or("");
152                if name.is_empty() {
153                    error_response(400, "Invalid app name")
154                } else {
155                    handlers::post_app_rollback(&state, name).await
156                }
157            } else if app_name.ends_with("/stop") {
158                let name = app_name.strip_suffix("/stop").unwrap_or("");
159                if name.is_empty() {
160                    error_response(400, "Invalid app name")
161                } else {
162                    handlers::post_app_stop(&state, name).await
163                }
164            } else if app_name.ends_with("/logs") {
165                let name = app_name.strip_suffix("/logs").unwrap_or("");
166                if name.is_empty() {
167                    error_response(400, "Invalid app name")
168                } else {
169                    handlers::get_app_logs(&state, name).await
170                }
171            } else {
172                error_response(404, "Not found")
173            }
174        }
175
176        // Phase 2: Mutation endpoints
177        (Method::POST, "/api/v1/routes") => {
178            let body = read_body(req).await;
179            handlers::post_route(&state, &body)
180        }
181        (Method::PUT, "/api/v1/config") => {
182            let body = read_body(req).await;
183            handlers::put_config(&state, &body)
184        }
185
186        // Routes with index parameter
187        (Method::GET, p) if p.starts_with("/api/v1/routes/") => match extract_route_index(p) {
188            Some(idx) => handlers::get_route(&state, idx),
189            None => error_response(400, "Invalid route index"),
190        },
191        (Method::PUT, p) if p.starts_with("/api/v1/routes/") => match extract_route_index(p) {
192            Some(idx) => {
193                let body = read_body(req).await;
194                handlers::put_route(&state, idx, &body)
195            }
196            None => error_response(400, "Invalid route index"),
197        },
198        (Method::DELETE, p) if p.starts_with("/api/v1/routes/") => match extract_route_index(p) {
199            Some(idx) => handlers::delete_route(&state, idx),
200            None => error_response(400, "Invalid route index"),
201        },
202
203        // Circuit breaker endpoints
204        (Method::GET, "/api/v1/circuit-breaker") => handlers::get_circuit_breaker(&state),
205        (Method::POST, "/api/v1/circuit-breaker/reset") => handlers::reset_circuit_breaker(&state),
206
207        // Everything else → proxy to _admin app (UI, static assets, etc.)
208        _ if state.app_manager.is_some() => {
209            let is_ws = req
210                .headers()
211                .get("upgrade")
212                .and_then(|v| v.to_str().ok())
213                .is_some_and(|v| v.eq_ignore_ascii_case("websocket"));
214            if is_ws {
215                proxy_websocket_to_admin_app(req, &state).await
216            } else {
217                proxy_to_admin_app(req, &state).await
218            }
219        }
220        _ => error_response(404, "Not found"),
221    };
222
223    Ok(response)
224}
225
226async fn proxy_to_admin_app(req: Request<Incoming>, state: &Arc<AdminState>) -> Response<BoxBody> {
227    let port = match resolve_admin_port(state).await {
228        Ok(p) => p,
229        Err(resp) => return resp,
230    };
231
232    let path = req.uri().path();
233    let query = req
234        .uri()
235        .query()
236        .map(|q| format!("?{}", q))
237        .unwrap_or_default();
238    let target_uri = format!("http://localhost:{}{}{}", port, path, query);
239
240    let (mut parts, body) = req.into_parts();
241    parts.uri = match target_uri.parse() {
242        Ok(uri) => uri,
243        Err(_) => return error_response(500, "Failed to build proxy URI"),
244    };
245
246    let proxy_req = Request::from_parts(parts, body);
247
248    let mut connector = HttpConnector::new();
249    connector.set_connect_timeout(Some(std::time::Duration::from_secs(3)));
250
251    let client: Client<HttpConnector, Incoming> =
252        Client::builder(TokioExecutor::new()).build(connector);
253
254    match client.request(proxy_req).await {
255        Ok(resp) => {
256            let (parts, body) = resp.into_parts();
257            let mapped = body.map_err(|_| -> std::convert::Infallible { unreachable!() });
258            Response::from_parts(parts, mapped.boxed())
259        }
260        Err(e) => {
261            tracing::error!("Failed to proxy to _admin app: {}", e);
262            error_response(502, &format!("Admin app not reachable on port {} — deploy it first: POST /api/v1/apps/_admin/deploy", port))
263        }
264    }
265}
266
267/// Resolve the _admin app's backend port, or return an error response.
268async fn resolve_admin_port(state: &Arc<AdminState>) -> Result<u16, Response<BoxBody>> {
269    let app_manager = match &state.app_manager {
270        Some(m) => m,
271        None => return Err(error_response(501, "App management not configured")),
272    };
273
274    let app = match app_manager.get_app("_admin").await {
275        Some(a) => a,
276        None => return Err(error_response(502, "_admin app not found")),
277    };
278
279    let port = if app.current_slot == "blue" {
280        app.blue.port
281    } else {
282        app.green.port
283    };
284
285    if port == 0 {
286        return Err(error_response(502, "_admin app not deployed"));
287    }
288
289    Ok(port)
290}
291
292async fn proxy_websocket_to_admin_app(
293    req: Request<Incoming>,
294    state: &Arc<AdminState>,
295) -> Response<BoxBody> {
296    let port = match resolve_admin_port(state).await {
297        Ok(p) => p,
298        Err(resp) => return resp,
299    };
300
301    let path = req.uri().path().to_string();
302    let query = req
303        .uri()
304        .query()
305        .map(|q| format!("?{}", q))
306        .unwrap_or_default();
307
308    // Build the raw HTTP upgrade request to send to the backend
309    let ws_key = req
310        .headers()
311        .get("sec-websocket-key")
312        .and_then(|v| v.to_str().ok())
313        .unwrap_or("")
314        .to_string();
315    let ws_version = req
316        .headers()
317        .get("sec-websocket-version")
318        .and_then(|v| v.to_str().ok())
319        .unwrap_or("13")
320        .to_string();
321    let ws_protocol = req
322        .headers()
323        .get("sec-websocket-protocol")
324        .and_then(|v| v.to_str().ok())
325        .map(|s| s.to_string());
326
327    // Connect to the backend
328    let backend = match TcpStream::connect(format!("127.0.0.1:{}", port)).await {
329        Ok(s) => s,
330        Err(e) => {
331            tracing::error!("Failed to connect to _admin backend for WebSocket: {}", e);
332            return error_response(502, "Admin app not reachable");
333        }
334    };
335
336    // Send the upgrade request to the backend
337    let mut handshake = format!(
338        "GET {}{} HTTP/1.1\r\n\
339         Host: 127.0.0.1:{}\r\n\
340         Upgrade: websocket\r\n\
341         Connection: Upgrade\r\n\
342         Sec-WebSocket-Key: {}\r\n\
343         Sec-WebSocket-Version: {}\r\n",
344        path, query, port, ws_key, ws_version,
345    );
346    if let Some(proto) = &ws_protocol {
347        handshake.push_str(&format!("Sec-WebSocket-Protocol: {}\r\n", proto));
348    }
349    handshake.push_str("\r\n");
350
351    let (mut backend_read, mut backend_write) = backend.into_split();
352    if let Err(e) = backend_write.write_all(handshake.as_bytes()).await {
353        tracing::error!("Failed to send WebSocket handshake to backend: {}", e);
354        return error_response(502, "Failed to initiate WebSocket with backend");
355    }
356
357    // Read the backend's 101 response
358    let mut response_buf = vec![0u8; 4096];
359    let n = match backend_read.read(&mut response_buf).await {
360        Ok(n) if n > 0 => n,
361        _ => {
362            tracing::error!("No response from backend for WebSocket upgrade");
363            return error_response(502, "Backend did not respond to WebSocket upgrade");
364        }
365    };
366
367    let response_str = String::from_utf8_lossy(&response_buf[..n]);
368    if !response_str.contains("101") {
369        tracing::error!(
370            "Backend rejected WebSocket upgrade: {}",
371            response_str.lines().next().unwrap_or("")
372        );
373        return error_response(502, "Backend rejected WebSocket upgrade");
374    }
375
376    // Extract headers from backend 101 response to forward to client
377    let mut accept_key = String::new();
378    let mut resp_protocol = None;
379    for line in response_str.lines().skip(1) {
380        if line.trim().is_empty() {
381            break;
382        }
383        if let Some((name, value)) = line.split_once(':') {
384            let name_lower = name.trim().to_lowercase();
385            let value = value.trim().to_string();
386            if name_lower == "sec-websocket-accept" {
387                accept_key = value;
388            } else if name_lower == "sec-websocket-protocol" {
389                resp_protocol = Some(value);
390            }
391        }
392    }
393
394    // Use hyper::upgrade::on to get the client-side stream after we return 101
395    let client_upgrade = hyper::upgrade::on(req);
396
397    // Reunite the backend halves
398    let backend_stream = backend_read.reunite(backend_write).unwrap();
399
400    // Spawn the bidirectional copy task
401    tokio::spawn(async move {
402        match client_upgrade.await {
403            Ok(upgraded) => {
404                let mut client_stream = TokioIo::new(upgraded);
405                let (mut br, mut bw) = tokio::io::split(backend_stream);
406                let (mut cr, mut cw) = tokio::io::split(&mut client_stream);
407                let _ = tokio::join!(
408                    tokio::io::copy(&mut br, &mut cw),
409                    tokio::io::copy(&mut cr, &mut bw),
410                );
411            }
412            Err(e) => {
413                tracing::error!("WebSocket client upgrade failed: {}", e);
414            }
415        }
416    });
417
418    // Return 101 Switching Protocols to the client
419    let mut resp = Response::builder()
420        .status(101)
421        .header("Upgrade", "websocket")
422        .header("Connection", "Upgrade")
423        .header("Sec-WebSocket-Accept", accept_key);
424    if let Some(proto) = resp_protocol {
425        resp = resp.header("Sec-WebSocket-Protocol", proto);
426    }
427    resp.body(http_body_util::Full::new(Bytes::new()).boxed())
428        .unwrap()
429}
430
431async fn read_body(req: Request<Incoming>) -> String {
432    match req.into_body().collect().await {
433        Ok(collected) => String::from_utf8_lossy(&collected.to_bytes()).to_string(),
434        Err(_) => String::new(),
435    }
436}
437
438pub async fn run_admin_server(state: Arc<AdminState>) -> Result<()> {
439    let bind = state.config_manager.get_config().admin.bind.clone();
440    let addr: std::net::SocketAddr = bind.parse()?;
441    let listener = TcpListener::bind(addr).await?;
442
443    tracing::info!("Admin API listening on {}", addr);
444
445    loop {
446        match listener.accept().await {
447            Ok((stream, _)) => {
448                let state = state.clone();
449                tokio::spawn(async move {
450                    let io = TokioIo::new(stream);
451                    let svc = service_fn(move |req| {
452                        let state = state.clone();
453                        async move { handle_admin_request(req, state).await }
454                    });
455                    if let Err(e) = hyper::server::conn::http1::Builder::new()
456                        .serve_connection(io, svc)
457                        .with_upgrades()
458                        .await
459                    {
460                        tracing::debug!("Admin connection error: {}", e);
461                    }
462                });
463            }
464            Err(e) => {
465                tracing::error!("Admin accept error: {}", e);
466            }
467        }
468    }
469}
soli_proxy/admin/mod.rs

soli_proxy/admin/
mod.rs
