hatchery_server/
mesh_runtime.rs

1//! Production wiring for [`NodeOperatorMesh`] (A2).
2//!
3//! When drain + operator control are configured, core:
4//! 1. listens for operator clients on an owner-only local endpoint;
5//! 2. opens one drain consumer per configured node id (g4a control);
6//! 3. registers each [`OperatorLink`] on the mesh so operator requests reach the node.
7//!
8//! Optional: m4a channel consumers share the same node list when a hub is present.
9//! Secrets are read from named environment variables and scrubbed; never logged.
10
11use crate::drain_consumer::{DrainConsumer, NodeStream};
12use crate::node_controller::OperatorLink;
13use crate::operator_control::{serve_local, NodeOperatorMesh};
14use hatchery_c2_client::DrainClient;
15use hatchery_c2_protocol::nested::HqRecipientSecret;
16use hatchery_c2_protocol::ChannelKind;
17use std::net::SocketAddr;
18use std::path::{Path, PathBuf};
19use tokio::sync::{mpsc, watch};
20use tokio::task::JoinHandle;
21
22pub const DRAIN_ADDR_ENV: &str = "HATCHERY_CORE_DRAIN_ADDR";
23pub const DRAIN_TOKEN_VAR_ENV: &str = "HATCHERY_CORE_DRAIN_TOKEN_VAR";
24pub const OPERATOR_LISTEN_ENV: &str = "HATCHERY_CORE_OPERATOR_LISTEN";
25pub const OPERATOR_TOKEN_VAR_ENV: &str = "HATCHERY_CORE_OPERATOR_TOKEN_VAR";
26pub const NODE_IDS_ENV: &str = "HATCHERY_CORE_NODE_IDS";
27pub const RECIPIENT_SECRET_VAR_ENV: &str = "HATCHERY_CORE_RECIPIENT_SECRET_VAR";
28
29const DEFAULT_DRAIN_TOKEN_VAR: &str = "HATCHERY_CORE_DRAIN_TOKEN";
30const DEFAULT_OPERATOR_TOKEN_VAR: &str = "GATE4AGENT_CORE_OPERATOR_TOKEN";
31const G4A_CONTROL_CHANNEL: u32 = 1;
32const M4A_CHANNEL: u32 = 2;
33
34/// Optional production mesh configuration (all-or-nothing for the drain path).
35#[derive(Clone)]
36pub struct MeshConfig {
37    pub drain_addr: SocketAddr,
38    pub drain_token: String,
39    pub operator_listen: String,
40    pub operator_token: String,
41    pub node_ids: Vec<String>,
42    pub recipient: HqRecipientSecret,
43}
44
45impl MeshConfig {
46    /// Resolve from the environment. Returns `Ok(None)` when drain is not configured.
47    /// Token values are taken from the named vars and those vars are removed from the process env.
48    pub fn from_env(data_dir: &Path) -> Result<Option<Self>, String> {
49        let Some(drain_addr) = std::env::var(DRAIN_ADDR_ENV).ok().filter(|s| !s.is_empty()) else {
50            return Ok(None);
51        };
52        let drain_addr: SocketAddr = drain_addr
53            .parse()
54            .map_err(|e| format!("{DRAIN_ADDR_ENV}: {e}"))?;
55        let operator_listen = std::env::var(OPERATOR_LISTEN_ENV)
56            .map_err(|_| format!("{OPERATOR_LISTEN_ENV} is required with {DRAIN_ADDR_ENV}"))?;
57        if operator_listen.is_empty() {
58            return Err(format!("{OPERATOR_LISTEN_ENV} is empty"));
59        }
60        let node_ids = std::env::var(NODE_IDS_ENV)
61            .map_err(|_| format!("{NODE_IDS_ENV} is required with {DRAIN_ADDR_ENV}"))?
62            .split(',')
63            .map(str::trim)
64            .filter(|s| !s.is_empty())
65            .map(str::to_owned)
66            .collect::<Vec<_>>();
67        if node_ids.is_empty() {
68            return Err(format!("{NODE_IDS_ENV} must list at least one node id"));
69        }
70
71        let drain_token_var =
72            std::env::var(DRAIN_TOKEN_VAR_ENV).unwrap_or_else(|_| DEFAULT_DRAIN_TOKEN_VAR.into());
73        let operator_token_var = std::env::var(OPERATOR_TOKEN_VAR_ENV)
74            .unwrap_or_else(|_| DEFAULT_OPERATOR_TOKEN_VAR.into());
75        let drain_token = take_secret(&drain_token_var)?;
76        let operator_token = take_secret(&operator_token_var)?;
77        if drain_token.len() < 16 || operator_token.len() < 16 {
78            return Err("drain and operator tokens must be at least 16 bytes".into());
79        }
80        let recipient = load_or_create_recipient(data_dir)?;
81        Ok(Some(Self {
82            drain_addr,
83            drain_token,
84            operator_listen,
85            operator_token,
86            node_ids,
87            recipient,
88        }))
89    }
90}
91
92fn take_secret(var: &str) -> Result<String, String> {
93    let value = std::env::var(var).map_err(|_| format!("{var} is not set"))?;
94    std::env::remove_var(var);
95    if value.is_empty() {
96        return Err(format!("{var} is empty"));
97    }
98    Ok(value)
99}
100
101fn load_or_create_recipient(data_dir: &Path) -> Result<HqRecipientSecret, String> {
102    if let Ok(var) = std::env::var(RECIPIENT_SECRET_VAR_ENV) {
103        if !var.is_empty() {
104            let secret = take_secret(&var)?;
105            return parse_recipient_hex(&secret);
106        }
107    }
108    let path = data_dir.join("core-recipient.seed");
109    if path.exists() {
110        let bytes = std::fs::read(&path).map_err(|e| e.to_string())?;
111        if bytes.len() != 32 {
112            return Err(format!("{}: expected 32 bytes", path.display()));
113        }
114        let mut seed = [0u8; 32];
115        seed.copy_from_slice(&bytes);
116        return Ok(HqRecipientSecret::from_seed(seed));
117    }
118    let recipient = HqRecipientSecret::generate();
119    std::fs::create_dir_all(data_dir).map_err(|e| e.to_string())?;
120    // Persist only the seed bytes; never log them.
121    let seed = recipient.seed_bytes();
122    std::fs::write(&path, seed).map_err(|e| e.to_string())?;
123    #[cfg(unix)]
124    {
125        use std::os::unix::fs::PermissionsExt;
126        let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
127    }
128    Ok(recipient)
129}
130
131fn parse_recipient_hex(hex: &str) -> Result<HqRecipientSecret, String> {
132    let hex = hex.trim();
133    if hex.len() != 64 {
134        return Err("recipient secret must be 64 hex digits".into());
135    }
136    let mut seed = [0u8; 32];
137    for i in 0..32 {
138        seed[i] = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16)
139            .map_err(|_| "recipient secret is not hex".to_owned())?;
140    }
141    Ok(HqRecipientSecret::from_seed(seed))
142}
143
144/// Running mesh: operator control + per-node drain consumers.
145pub struct MeshRuntime {
146    pub mesh: NodeOperatorMesh,
147    tasks: Vec<JoinHandle<()>>,
148    stop: watch::Sender<bool>,
149}
150
151impl MeshRuntime {
152    pub fn start(config: MeshConfig) -> Self {
153        Self::start_with_m4a(config, None)
154    }
155
156    pub fn start_with_m4a(
157        config: MeshConfig,
158        m4a_handler: Option<crate::drain_consumer::CoreHandler>,
159    ) -> Self {
160        let mesh = NodeOperatorMesh::new();
161        let (stop, shutdown) = watch::channel(false);
162        let mut tasks = Vec::new();
163
164        let listen = config.operator_listen.clone();
165        let token = config.operator_token.clone();
166        let mesh_serve = mesh.clone();
167        let sd = shutdown.clone();
168        tasks.push(tokio::spawn(async move {
169            if let Err(err) = serve_local(listen, token, mesh_serve, sd).await {
170                tracing::error!(%err, "core operator control stopped");
171            }
172        }));
173
174        for node_id in &config.node_ids {
175            let (events_tx, _events_rx) = mpsc::unbounded_channel();
176            let (link, outbound) = OperatorLink::new(Some(events_tx));
177            mesh.register(node_id, link.clone());
178            let consumer = DrainConsumer::new(
179                DrainClient::new(config.drain_addr, config.drain_token.clone()),
180                config.recipient.clone(),
181                NodeStream {
182                    node_id: node_id.clone(),
183                    channel_id: G4A_CONTROL_CHANNEL,
184                    kind: ChannelKind::G4aControl,
185                },
186                link.handler(),
187            )
188            .with_outbound(outbound);
189            let sd = shutdown.clone();
190            tasks.push(tokio::spawn(async move {
191                consumer.run(sd).await;
192            }));
193
194            if let Some(handler) = m4a_handler.clone() {
195                let consumer = DrainConsumer::new(
196                    DrainClient::new(config.drain_addr, config.drain_token.clone()),
197                    config.recipient.clone(),
198                    NodeStream {
199                        node_id: node_id.clone(),
200                        channel_id: M4A_CHANNEL,
201                        kind: ChannelKind::M4a,
202                    },
203                    handler,
204                );
205                let sd = shutdown.clone();
206                tasks.push(tokio::spawn(async move {
207                    consumer.run(sd).await;
208                }));
209            }
210        }
211
212        tracing::info!(
213            nodes = config.node_ids.len(),
214            listen = %config.operator_listen,
215            "core operator mesh started (drain consumers registered)"
216        );
217        Self { mesh, tasks, stop }
218    }
219
220    pub fn shutdown(&self) {
221        let _ = self.stop.send(true);
222        for t in &self.tasks {
223            t.abort();
224        }
225    }
226}
227
228/// Public path of the persisted recipient seed (for operators to hand the pub key to C2).
229pub fn recipient_seed_path(data_dir: &Path) -> PathBuf {
230    data_dir.join("core-recipient.seed")
231}
hatchery_server/mesh_runtime.rs

hatchery_server/
mesh_runtime.rs
