eggress_admin/
client.rs

1//! Minimal admin control-plane client.
2//!
3//! Owned by `eggress-admin` (not `eggress-cli`) so the route-explain command
4//! cannot silently reimplement the admin HTTP protocol inside a command
5//! handler. The implementation is deliberately small and HTTP/1.1-only to
6//! match the admin server contract: request construction, status parsing,
7//! body extraction, transport errors, and response deserialization all live
8//! here. Callers only validate user inputs, choose local vs remote
9//! explanation, render output, and map errors to process outcomes.
10
11use tokio::io::{AsyncReadExt, AsyncWriteExt};
12
13/// Errors from the admin route-explain client. Transport and protocol
14/// failures are typed so CLI handlers can map them without parsing strings.
15#[derive(Debug, thiserror::Error)]
16pub enum AdminClientError {
17    /// The `--admin` URL is malformed or uses an unsupported scheme.
18    #[error("invalid --admin '{url}': {reason}")]
19    InvalidUrl {
20        /// The user-supplied URL (never contains credentials in practice;
21        /// admin URLs carry no secrets).
22        url: String,
23        /// Why the URL was rejected.
24        reason: String,
25    },
26
27    /// TCP connection to the admin endpoint failed.
28    #[error("failed to connect to admin at {addr}: {reason}")]
29    Connect {
30        /// Resolved `host:port` dial string.
31        addr: String,
32        /// Underlying I/O reason.
33        reason: String,
34    },
35
36    /// Request send or response read failed.
37    #[error("admin request failed: {0}")]
38    Transport(String),
39
40    /// The admin server answered with a non-200 status.
41    #[error("admin returned {status}: {body}")]
42    Status {
43        /// HTTP status code.
44        status: u16,
45        /// Response body (truncated by the server contract).
46        body: String,
47    },
48
49    /// A 200 response carried a body that is not a route explanation.
50    #[error("failed to parse admin response: {0}")]
51    Parse(String),
52}
53
54/// Parsed admin endpoint: host without brackets, port, and absolute path.
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub struct AdminEndpoint {
57    /// Host without surrounding `[]` (IPv6 brackets are reapplied at dial).
58    pub host: String,
59    /// TCP port (default 9090).
60    pub port: u16,
61    /// Absolute request path (always starts with `/`).
62    pub path: String,
63}
64
65/// Parse an `--admin` URL into dial parts.
66///
67/// Accepts `http://host[:port][/path]` and bare `host[:port][/path]`.
68/// TLS admin URLs are rejected: the local admin contract is HTTP/1.1-only.
69pub fn parse_admin_url(url: &str) -> Result<AdminEndpoint, String> {
70    if url.starts_with("https://") {
71        return Err("TLS admin URLs are not supported; use http://".to_string());
72    }
73    let without_proto = url.strip_prefix("http://").unwrap_or(url);
74    let (host_port, path) = match without_proto.find('/') {
75        Some(i) => (&without_proto[..i], &without_proto[i..]),
76        None => (without_proto, "/"),
77    };
78    if host_port.is_empty() {
79        return Err("missing host in admin URL".to_string());
80    }
81    let (host, port) = if let Some(rest) = host_port.strip_prefix('[') {
82        let close = rest
83            .find(']')
84            .ok_or_else(|| "missing closing ']' in IPv6 admin host".to_string())?;
85        let host = rest[..close].to_string();
86        if host.is_empty() {
87            return Err("missing host in admin URL".to_string());
88        }
89        let after = &rest[close + 1..];
90        let port = match after.strip_prefix(':') {
91            Some(port_str) => port_str.parse::<u16>().map_err(|_| {
92                format!("invalid port '{port_str}' in admin URL (expected 1-65535)")
93            })?,
94            None if after.is_empty() => 9090,
95            None => {
96                return Err(format!(
97                    "invalid IPv6 admin host '{host_port}' (expected [host] or [host]:port)"
98                ));
99            }
100        };
101        (host, port)
102    } else {
103        match host_port.rfind(':') {
104            Some(i) => {
105                let port_str = &host_port[i + 1..];
106                let port = port_str.parse::<u16>().map_err(|_| {
107                    format!("invalid port '{port_str}' in admin URL (expected 1-65535)")
108                })?;
109                let host = host_port[..i].to_string();
110                if host.is_empty() {
111                    return Err("missing host in admin URL".to_string());
112                }
113                (host, port)
114            }
115            None => (host_port.to_string(), 9090),
116        }
117    };
118    Ok(AdminEndpoint {
119        host,
120        port,
121        path: path.to_string(),
122    })
123}
124
125impl AdminEndpoint {
126    fn dial_addr(&self) -> String {
127        if self.host.contains(':') {
128            format!("[{}]:{}", self.host, self.port)
129        } else {
130            format!("{}:{}", self.host, self.port)
131        }
132    }
133
134    fn host_header(&self) -> String {
135        self.dial_addr()
136    }
137}
138
139/// Explain a route through a live admin server.
140///
141/// Sends `POST <admin>/-/route-explain` with `{target, listener, protocol}`
142/// and returns the deserialized [`eggress_routing::RouteExplanation`].
143/// `protocol` must be one of `http`, `socks4`, `socks5`; validation stays
144/// with the caller so this client never reimplements CLI value policy.
145pub async fn route_explain(
146    admin_url: &str,
147    target: &str,
148    listener: &str,
149    protocol: &str,
150) -> Result<eggress_routing::RouteExplanation, AdminClientError> {
151    let endpoint = parse_admin_url(admin_url).map_err(|reason| AdminClientError::InvalidUrl {
152        url: admin_url.to_string(),
153        reason,
154    })?;
155    // The admin route-explain handler is mounted at `/-/route-explain`.
156    // A caller-supplied path prefix is honored; otherwise the canonical
157    // path is used.
158    let path = if endpoint.path == "/" {
159        "/-/route-explain".to_string()
160    } else {
161        endpoint.path.clone()
162    };
163    let body = serde_json::json!({
164        "target": target,
165        "listener": listener,
166        "protocol": protocol,
167    });
168    let body_str = body.to_string();
169    let request = format!(
170        "POST {path} HTTP/1.1\r\nHost: {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body_str}",
171        endpoint.host_header(),
172        body_str.len(),
173    );
174
175    let dial = endpoint.dial_addr();
176    let mut stream =
177        tokio::net::TcpStream::connect(&dial)
178            .await
179            .map_err(|e| AdminClientError::Connect {
180                addr: dial.clone(),
181                reason: e.to_string(),
182            })?;
183
184    stream
185        .write_all(request.as_bytes())
186        .await
187        .map_err(|e| AdminClientError::Transport(format!("failed to send request: {e}")))?;
188    stream
189        .flush()
190        .await
191        .map_err(|e| AdminClientError::Transport(format!("failed to send request: {e}")))?;
192    // Do not shut down the write half here: the server answers on this same
193    // connection (`Connection: close`) and a write-half shutdown races the
194    // response on some platforms, surfacing as an empty reply.
195
196    let mut response = Vec::new();
197    loop {
198        let mut buf = [0u8; 4096];
199        match stream.read(&mut buf).await {
200            Ok(0) => break,
201            Ok(n) => response.extend_from_slice(&buf[..n]),
202            Err(e) => {
203                return Err(AdminClientError::Transport(format!(
204                    "failed to read response: {e}"
205                )));
206            }
207        }
208    }
209    let text = String::from_utf8_lossy(&response).to_string();
210    let body_start = text.find("\r\n\r\n").map(|i| i + 4).unwrap_or(0);
211    let body = text[body_start..].to_string();
212    let status_line = text.lines().next().unwrap_or("");
213    let status = status_line
214        .split_whitespace()
215        .nth(1)
216        .and_then(|s| s.parse::<u16>().ok())
217        .unwrap_or(0);
218
219    if status != 200 {
220        return Err(AdminClientError::Status { status, body });
221    }
222    serde_json::from_str::<eggress_routing::RouteExplanation>(&body)
223        .map_err(|e| AdminClientError::Parse(e.to_string()))
224}
225
226#[cfg(test)]
227mod tests {
228    use super::*;
229
230    #[test]
231    fn parses_ipv4_with_default_port() {
232        let ep = parse_admin_url("http://127.0.0.1/admin").unwrap();
233        assert_eq!(
234            ep,
235            AdminEndpoint {
236                host: "127.0.0.1".to_string(),
237                port: 9090,
238                path: "/admin".to_string(),
239            }
240        );
241    }
242
243    #[test]
244    fn parses_bracketed_ipv6_loopback() {
245        let ep = parse_admin_url("http://[::1]/-/route-explain").unwrap();
246        assert_eq!(ep.host, "::1");
247        assert_eq!(ep.port, 9090);
248        assert_eq!(ep.path, "/-/route-explain");
249    }
250
251    #[test]
252    fn parses_bracketed_ipv6_with_port() {
253        let ep = parse_admin_url("http://[2001:db8::1]:8080/-/route-explain").unwrap();
254        assert_eq!(ep.host, "2001:db8::1");
255        assert_eq!(ep.port, 8080);
256    }
257
258    #[test]
259    fn parses_domain_with_port() {
260        let ep = parse_admin_url("http://admin.example.com:8080/-/x").unwrap();
261        assert_eq!(ep.host, "admin.example.com");
262        assert_eq!(ep.port, 8080);
263    }
264
265    #[test]
266    fn rejects_malformed_ports() {
267        assert!(parse_admin_url("http://host:notaport/path").is_err());
268        assert!(parse_admin_url("http://host:99999/path").is_err());
269        assert!(parse_admin_url("http://[::1]:notaport/admin").is_err());
270    }
271
272    #[test]
273    fn rejects_tls_admin_urls() {
274        let err = parse_admin_url("https://127.0.0.1:9090/-/route-explain").unwrap_err();
275        assert!(err.contains("TLS"), "unexpected error: {err}");
276    }
277
278    #[tokio::test]
279    async fn reports_connection_failure_without_panicking() {
280        // Port 1 is unroutable for TCP connect in tests; the client must
281        // return a typed Connect error rather than exiting or panicking.
282        let err = route_explain("http://127.0.0.1:1", "example.com:443", "cli", "http")
283            .await
284            .unwrap_err();
285        assert!(
286            matches!(err, AdminClientError::Connect { .. }),
287            "unexpected error: {err:?}"
288        );
289    }
290
291    #[tokio::test]
292    async fn surfaces_non_200_admin_responses() {
293        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
294        let addr = listener.local_addr().unwrap();
295        tokio::spawn(async move {
296            let (mut stream, _) = listener.accept().await.unwrap();
297            let mut buf = [0u8; 4096];
298            let _ = stream.read(&mut buf).await;
299            let body = r#"{"error":"missing 'target' field"}"#;
300            let response = format!(
301                "HTTP/1.1 400 Bad Request\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
302                body.len()
303            );
304            let _ = stream.write_all(response.as_bytes()).await;
305        });
306        let err = route_explain(&format!("http://{addr}"), "example.com:443", "cli", "http")
307            .await
308            .unwrap_err();
309        match err {
310            AdminClientError::Status { status, .. } => assert_eq!(status, 400),
311            other => panic!("unexpected error: {other:?}"),
312        }
313    }
314
315    #[tokio::test]
316    async fn rejects_malformed_200_bodies() {
317        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
318        let addr = listener.local_addr().unwrap();
319        tokio::spawn(async move {
320            let (mut stream, _) = listener.accept().await.unwrap();
321            let mut buf = [0u8; 4096];
322            let _ = stream.read(&mut buf).await;
323            let body = "not-json";
324            let response = format!(
325                "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
326                body.len()
327            );
328            let _ = stream.write_all(response.as_bytes()).await;
329        });
330        let err = route_explain(&format!("http://{addr}"), "example.com:443", "cli", "http")
331            .await
332            .unwrap_err();
333        assert!(
334            matches!(err, AdminClientError::Parse(_)),
335            "unexpected error: {err:?}"
336        );
337    }
338
339    #[tokio::test]
340    async fn round_trips_route_explain_against_live_admin() {
341        use std::sync::Arc;
342        use std::time::Instant;
343
344        let router = Arc::new(eggress_routing::Router::new(
345            vec![],
346            eggress_routing::RouteActionSpec::Direct,
347        ));
348        let snapshot = crate::server::AdminSnapshot {
349            generation: 7,
350            router,
351            pac: None,
352            static_routes: vec![],
353            listeners: vec![],
354        };
355        let state = crate::server::AdminState {
356            metrics: Arc::new(eggress_metrics::MetricsRegistry::new()),
357            start_time: Instant::now(),
358            readiness: Arc::new(std::sync::atomic::AtomicBool::new(true)),
359            active_connections: None,
360            provider: Arc::new(crate::server::StaticAdminSnapshot { snapshot }),
361            udp_registry: Arc::new(eggress_udp::registry::UdpAssociationRegistry::new(
362                eggress_udp::limits::UdpLimits::default(),
363            )),
364            reverse_registry: Arc::new(crate::reverse::ReverseRegistry::new()),
365            metrics_enabled: true,
366            auth: None,
367        };
368        let cancel = tokio_util::sync::CancellationToken::new();
369        let server = crate::server::AdminServer::new("127.0.0.1:0", cancel.clone())
370            .await
371            .unwrap();
372        let addr = server.listener.local_addr().unwrap().to_string();
373        tokio::spawn(async move { server.run(state).await.unwrap() });
374
375        let explanation =
376            route_explain(&format!("http://{addr}"), "example.com:443", "cli", "http")
377                .await
378                .expect("live admin route-explain must succeed");
379        assert_eq!(explanation.target, "example.com:443");
380    }
381}
eggress_admin/client.rs

eggress_admin/
client.rs
