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;
15const MAX_BODY_BYTES: usize = 128 * 1024;
16
17pub async fn bind(addr: SocketAddr) -> std::io::Result<TcpListener> {
18    assert!(addr.ip().is_loopback(), "control listener must be loopback");
19    TcpListener::bind(addr).await
20}
21
22pub async fn serve(listener: TcpListener, supervisor: Supervisor, token: Option<String>) {
23    serve_with_hub(listener, supervisor, token, None).await
24}
25
26/// Same as [`serve`], plus the P6 hub routes (`/v1/...`, bearer required): node calls forwarded by
27/// C2, operator mail and tasks.
28#[cfg(feature = "m4a")]
29pub type HubHandle = Option<std::sync::Arc<crate::hub::Hub>>;
30#[cfg(not(feature = "m4a"))]
31pub type HubHandle = Option<()>;
32
33pub async fn serve_with_hub(listener: TcpListener, supervisor: Supervisor, token: Option<String>, hub: HubHandle) {
34    loop {
35        let Ok((stream, peer)) = listener.accept().await else { continue };
36        if !peer.ip().is_loopback() {
37            continue;
38        }
39        let supervisor = supervisor.clone();
40        let token = token.clone();
41        let hub = hub.clone();
42        tokio::spawn(async move {
43            let _ = handle(stream, supervisor, token, hub).await;
44        });
45    }
46}
47
48async fn handle(
49    mut stream: TcpStream,
50    supervisor: Supervisor,
51    token: Option<String>,
52    hub: HubHandle,
53) -> std::io::Result<()> {
54    let mut buffer = vec![0u8; MAX_REQUEST_BYTES];
55    let mut filled = 0;
56    loop {
57        let read = stream.read(&mut buffer[filled..]).await?;
58        if read == 0 {
59            return Ok(());
60        }
61        filled += read;
62        if buffer[..filled].windows(4).any(|w| w == b"\r\n\r\n") || filled == buffer.len() {
63            break;
64        }
65    }
66    let head_end = buffer[..filled].windows(4).position(|w| w == b"\r\n\r\n").map(|p| p + 4).unwrap_or(filled);
67    let head = String::from_utf8_lossy(&buffer[..head_end]).into_owned();
68    let mut request_body = buffer[head_end..filled].to_vec();
69    let length = head
70        .lines()
71        .find_map(|l| l.split_once(':').filter(|(n, _)| n.eq_ignore_ascii_case("content-length")).and_then(|(_, v)| v.trim().parse::<usize>().ok()))
72        .unwrap_or(0);
73    if length > MAX_BODY_BYTES {
74        return respond(&mut stream, "413 Payload Too Large", json!({ "error": "too large" }).to_string()).await;
75    }
76    while request_body.len() < length {
77        let mut chunk = [0u8; 4096];
78        let n = stream.read(&mut chunk).await?;
79        if n == 0 {
80            break;
81        }
82        request_body.extend_from_slice(&chunk[..n]);
83    }
84    request_body.truncate(length);
85    let first = head.lines().next().unwrap_or("");
86    let mut fp = first.split_whitespace();
87    let (method, path) = (fp.next().unwrap_or(""), fp.next().unwrap_or(""));
88    let (status, body) = if path.starts_with("/v1/") {
89        // The mailbox store blocks on its connection lock; never run it on a runtime thread.
90        let (method, path, head, hub, token) = (method.to_owned(), path.to_owned(), head.clone(), hub.clone(), token.clone());
91        tokio::task::spawn_blocking(move || v1(&method, &path, &head, &request_body, token.as_deref(), &hub))
92            .await
93            .unwrap_or(("500 Internal Server Error", json!({ "error": "internal" }).to_string()))
94    } else {
95        route(&head, &supervisor, token.as_deref())
96    };
97    let response = format!(
98        "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
99        body.len()
100    );
101    stream.write_all(response.as_bytes()).await?;
102    stream.shutdown().await
103}
104
105async fn respond(stream: &mut TcpStream, status: &str, body: String) -> std::io::Result<()> {
106    let response = format!("HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len());
107    stream.write_all(response.as_bytes()).await?;
108    stream.shutdown().await
109}
110
111fn bearer_of(head: &str) -> Option<String> {
112    head.lines()
113        .find_map(|line| {
114            let (name, value) = line.split_once(':')?;
115            name.eq_ignore_ascii_case("authorization").then(|| value.trim().to_owned())
116        })
117        .and_then(|value| value.strip_prefix("Bearer ").map(str::to_owned))
118}
119
120fn status_line(code: u16) -> &'static str {
121    match code {
122        200 => "200 OK",
123        400 => "400 Bad Request",
124        403 => "403 Forbidden",
125        404 => "404 Not Found",
126        409 => "409 Conflict",
127        _ => "500 Internal Server Error",
128    }
129}
130
131#[cfg(feature = "m4a")]
132fn v1(method: &str, path: &str, head: &str, body: &[u8], token: Option<&str>, hub: &HubHandle) -> (&'static str, String) {
133    let authorized = match (token, bearer_of(head).as_deref()) {
134        (Some(expected), Some(given)) => constant_time_eq(expected.as_bytes(), given.as_bytes()),
135        _ => false,
136    };
137    if !authorized {
138        return ("401 Unauthorized", json!({ "error": "unauthorized" }).to_string());
139    }
140    let Some(hub) = hub else {
141        return ("404 Not Found", json!({ "error": "messenger is not enabled" }).to_string());
142    };
143    let value: serde_json::Value = if body.is_empty() { json!({}) } else {
144        match serde_json::from_slice(body) {
145            Ok(v) => v,
146            Err(_) => return ("400 Bad Request", json!({ "error": "malformed json" }).to_string()),
147        }
148    };
149    let result = if method == "POST" && path == "/v1/node-call" {
150        match serde_json::from_value::<hatchery_enroll::call::ForwardedCall>(value) {
151            Ok(call) => hub.node_call(&call.node_id, &call.op, &call.payload),
152            Err(_) => return ("400 Bad Request", json!({ "error": "malformed call" }).to_string()),
153        }
154    } else {
155        hub.operator_call(method, path, &value)
156    };
157    match result {
158        Ok(v) => ("200 OK", v.to_string()),
159        Err(e) => (status_line(e.status), json!({ "error": e.message }).to_string()),
160    }
161}
162
163#[cfg(not(feature = "m4a"))]
164fn v1(_: &str, _: &str, _: &str, _: &[u8], _: Option<&str>, _: &HubHandle) -> (&'static str, String) {
165    ("404 Not Found", json!({ "error": "messenger is not built in" }).to_string())
166}
167
168pub fn route(head: &str, supervisor: &Supervisor, token: Option<&str>) -> (&'static str, String) {
169    let mut lines = head.lines();
170    let request_line = lines.next().unwrap_or("");
171    let mut parts = request_line.split_whitespace();
172    let method = parts.next().unwrap_or("");
173    let path = parts.next().unwrap_or("");
174    if method == "GET" && path == "/health" {
175        return ("200 OK", json!({ "status": "ok", "subservices": supervisor.statuses() }).to_string());
176    }
177    let segments: Vec<&str> = path.trim_matches('/').split('/').collect();
178    if method == "POST" && segments.len() == 3 && segments[0] == "sub" {
179        let presented = lines
180            .find_map(|line| {
181                let (name, value) = line.split_once(':')?;
182                name.eq_ignore_ascii_case("authorization").then(|| value.trim().to_owned())
183            })
184            .and_then(|value| value.strip_prefix("Bearer ").map(str::to_owned));
185        let authorized = match (token, presented.as_deref()) {
186            (Some(expected), Some(given)) => constant_time_eq(expected.as_bytes(), given.as_bytes()),
187            _ => false,
188        };
189        if !authorized {
190            return ("401 Unauthorized", json!({ "error": "unauthorized" }).to_string());
191        }
192        let result = match segments[2] {
193            "start" => supervisor.start_named(segments[1]),
194            "stop" => supervisor.stop_named(segments[1]),
195            _ => return ("404 Not Found", json!({ "error": "not found" }).to_string()),
196        };
197        return match result {
198            Ok(status) => ("200 OK", json!(status).to_string()),
199            Err(error) => ("409 Conflict", json!({ "error": error.to_string() }).to_string()),
200        };
201    }
202    ("404 Not Found", json!({ "error": "not found" }).to_string())
203}
204
205fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
206    if left.len() != right.len() {
207        return false;
208    }
209    left.iter().zip(right).fold(0u8, |acc, (a, b)| acc | (a ^ b)) == 0
210}
hatchery_server/control.rs

hatchery_server/
control.rs
