hatchery_node/
door.rs

1//! Local door for agents on this node: mail and task operations without ever touching a key.
2//!
3//! The node holds its identity key; agents (other users, other processes) do not. This loopback
4//! listener accepts `POST /door {"op","payload"}` from them, forwards the call to core over the
5//! node's restreamer uplink (Noise to core; C2 sees opaque bytes), which applies the status policy. A node that is neutral, suspended or
6//! revoked simply gets "refused" back. The surface is exactly the `mail.*` and `task.*` operations.
7
8use hatchery_build::http;
9use hatchery_enroll::call::capability_for;
10use serde_json::{json, Value};
11use tokio::net::TcpListener;
12
13/// Process-wide door listen address (for harness MCP env injection).
14static DOOR_LISTEN: std::sync::OnceLock<std::net::SocketAddr> = std::sync::OnceLock::new();
15
16/// Loopback door address if the node started `--door-listen`, else None.
17pub fn door_listen_addr() -> Option<std::net::SocketAddr> {
18    DOOR_LISTEN.get().copied()
19}
20
21
22/// Node reports the door never forwards.
23pub const NODE_ONLY_OPS: &[&str] = &["task.session_bind", "task.session_git", "task.session_state", "task.session_signal", "task.provider_inventory", "task.provider_limits", "task.check_run"];
24
25/// `session.wake` (handled here, never forwarded): `{"token","nick","from","event_id","body"}`,
26/// the letter event m4a's `M4A_LETTER_HOOK` emits plus the node's wake token.
27pub async fn serve(listener: TcpListener, calls: crate::uplink_calls::NodeCalls, wake: Option<crate::server::node_wake::NodeWake>) {
28    if let Ok(addr) = listener.local_addr() {
29        let _ = DOOR_LISTEN.set(addr);
30    }
31    loop {
32        let Ok((mut stream, peer)) = listener.accept().await else { continue };
33        if !peer.ip().is_loopback() {
34            continue;
35        }
36        let (calls, wake) = (calls.clone(), wake.clone());
37        tokio::spawn(async move {
38            let Ok(Ok((method, path, _, body))) = tokio::time::timeout(std::time::Duration::from_secs(5), http::read_request(&mut stream, 64 * 1024)).await else {
39                let _ = http::respond(&mut stream, 400, br#"{"error":"bad request"}"#).await;
40                return;
41            };
42            if method != "POST" || path != "/door" {
43                let _ = http::respond(&mut stream, 404, br#"{"error":"not found"}"#).await;
44                return;
45            }
46            let Ok(request) = serde_json::from_slice::<Value>(&body) else {
47                let _ = http::respond(&mut stream, 400, br#"{"error":"malformed"}"#).await;
48                return;
49            };
50            let op = request.get("op").and_then(Value::as_str).unwrap_or_default();
51            if op.starts_with("local_server.") {
52                #[cfg(feature = "local-server")]
53                {
54                    let payload = request.get("payload").cloned().unwrap_or_else(|| json!({}));
55                    let (status, reply) = match crate::local_server::handle_op(op, &payload).await {
56                        Ok(v) => (200, v.to_string()),
57                        Err(e) => (400, json!({"error": e}).to_string()),
58                    };
59                    let _ = http::respond(&mut stream, status, reply.as_bytes()).await;
60                    return;
61                }
62                #[cfg(not(feature = "local-server"))]
63                {
64                    let _ = http::respond(&mut stream, 404, br#"{"error":"local-server feature off"}"#).await;
65                    return;
66                }
67            }
68            if op == "session.wake" {
69                let p = request.get("payload").cloned().unwrap_or_else(|| json!({}));
70                let field = |k: &str| p.get(k).and_then(Value::as_str).unwrap_or_default().to_owned();
71                let (status, reply) = match &wake {
72                    None => (404, json!({"error":"letter wake is off on this node"})),
73                    Some(w) if !w.authorized(&field("token")) => (403, json!({"error":"refused"})),
74                    Some(w) => {
75                        let letter = crate::server::node_wake::NodeLetter { nick: field("nick"), from: field("from"), event_id: field("event_id"), body: field("body"), own_spawn: false };
76                        match w.wake(letter).await {
77                            Ok(done) => (200, json!({"ok": done})),
78                            Err(why) => (409, json!({"error": why})),
79                        }
80                    }
81                };
82                let _ = http::respond(&mut stream, status, reply.to_string().as_bytes()).await;
83                return;
84            }
85            // What only the node itself reports (its sessions, their keys, git, limits) never
86            // comes through the door: a local process must not speak for a session.
87            if NODE_ONLY_OPS.contains(&op) {
88                let _ = http::respond(&mut stream, 403, br#"{"error":"refused"}"#).await;
89                return;
90            }
91            if capability_for(op).is_none() {
92                let _ = http::respond(&mut stream, 404, br#"{"error":"unknown operation"}"#).await;
93                return;
94            }
95            let payload = request.get("payload").cloned().unwrap_or_else(|| json!({}));
96            let (status, reply) = match calls.call(op, &payload).await {
97                Ok(value) => (200, value.to_string()),
98                Err(crate::uplink_calls::CallError::Refused) => (403, json!({"error":"refused"}).to_string()),
99                Err(crate::uplink_calls::CallError::Unavailable(why)) => {
100                    tracing::warn!(%op, reason = %why, "door call unavailable");
101                    (502, json!({"error":"unavailable"}).to_string())
102                }
103            };
104            let _ = http::respond(&mut stream, status, reply.as_bytes()).await;
105        });
106    }
107}
hatchery_node/door.rs

hatchery_node/
door.rs
