hatchery_server/
control.rs

1//! Loopback control endpoint (minimal HTTP/1.1, no framework).
2//!
3//! - `GET /health`            : open on loopback, lists subservice status.
4//! - `POST /sub/<name>/start` : bearer required (`HATCHERY_SERVER_CONTROL_TOKEN`).
5//! - `POST /sub/<name>/stop`  : bearer required.
6//! The bearer is a node-side credential; agents never receive it.
7
8use crate::supervisor::Supervisor;
9use serde_json::json;
10use std::net::SocketAddr;
11use tokio::io::{AsyncReadExt, AsyncWriteExt};
12use tokio::net::{TcpListener, TcpStream};
13
14const MAX_REQUEST_BYTES: usize = 8 * 1024;
15
16pub async fn bind(addr: SocketAddr) -> std::io::Result<TcpListener> {
17    assert!(addr.ip().is_loopback(), "control listener must be loopback");
18    TcpListener::bind(addr).await
19}
20
21pub async fn serve(listener: TcpListener, supervisor: Supervisor, token: Option<String>) {
22    loop {
23        let Ok((stream, peer)) = listener.accept().await else { continue };
24        if !peer.ip().is_loopback() {
25            continue;
26        }
27        let supervisor = supervisor.clone();
28        let token = token.clone();
29        tokio::spawn(async move {
30            let _ = handle(stream, supervisor, token).await;
31        });
32    }
33}
34
35async fn handle(
36    mut stream: TcpStream,
37    supervisor: Supervisor,
38    token: Option<String>,
39) -> std::io::Result<()> {
40    let mut buffer = vec![0u8; MAX_REQUEST_BYTES];
41    let mut filled = 0;
42    loop {
43        let read = stream.read(&mut buffer[filled..]).await?;
44        if read == 0 {
45            return Ok(());
46        }
47        filled += read;
48        if buffer[..filled].windows(4).any(|w| w == b"\r\n\r\n") || filled == buffer.len() {
49            break;
50        }
51    }
52    let head = String::from_utf8_lossy(&buffer[..filled]).into_owned();
53    let (status, body) = route(&head, &supervisor, token.as_deref());
54    let response = format!(
55        "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
56        body.len()
57    );
58    stream.write_all(response.as_bytes()).await?;
59    stream.shutdown().await
60}
61
62pub fn route(head: &str, supervisor: &Supervisor, token: Option<&str>) -> (&'static str, String) {
63    let mut lines = head.lines();
64    let request_line = lines.next().unwrap_or("");
65    let mut parts = request_line.split_whitespace();
66    let method = parts.next().unwrap_or("");
67    let path = parts.next().unwrap_or("");
68    if method == "GET" && path == "/health" {
69        return ("200 OK", json!({ "status": "ok", "subservices": supervisor.statuses() }).to_string());
70    }
71    let segments: Vec<&str> = path.trim_matches('/').split('/').collect();
72    if method == "POST" && segments.len() == 3 && segments[0] == "sub" {
73        let presented = lines
74            .find_map(|line| {
75                let (name, value) = line.split_once(':')?;
76                name.eq_ignore_ascii_case("authorization").then(|| value.trim().to_owned())
77            })
78            .and_then(|value| value.strip_prefix("Bearer ").map(str::to_owned));
79        let authorized = match (token, presented.as_deref()) {
80            (Some(expected), Some(given)) => constant_time_eq(expected.as_bytes(), given.as_bytes()),
81            _ => false,
82        };
83        if !authorized {
84            return ("401 Unauthorized", json!({ "error": "unauthorized" }).to_string());
85        }
86        let result = match segments[2] {
87            "start" => supervisor.start_named(segments[1]),
88            "stop" => supervisor.stop_named(segments[1]),
89            _ => return ("404 Not Found", json!({ "error": "not found" }).to_string()),
90        };
91        return match result {
92            Ok(status) => ("200 OK", json!(status).to_string()),
93            Err(error) => ("409 Conflict", json!({ "error": error.to_string() }).to_string()),
94        };
95    }
96    ("404 Not Found", json!({ "error": "not found" }).to_string())
97}
98
99fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
100    if left.len() != right.len() {
101        return false;
102    }
103    left.iter().zip(right).fold(0u8, |acc, (a, b)| acc | (a ^ b)) == 0
104}
hatchery_server/control.rs

hatchery_server/
control.rs
