hatchery_server/
provision.rs

1//! P2: provisioning of enrolled nodes on the server side.
2//!
3//! The server reads the node list from C2 (operator API, bearer from an environment
4//! variable) and, whenever a node enrolls or its status changes, brings the server side in
5//! line:
6//!
7//! * an m4a identity per node (ed25519 key pair generated in the vault, master key from the
8//!   environment; the private seed never leaves the vault and no agent sees it);
9//! * a mailbox address (stub: recorded in the ledger, the mail server consumes the ledger);
10//! * rooms and capabilities by status (stub grants, see [`policy`]).
11//!
12//! Reconciliation is idempotent: the same input changes nothing. A revoked node loses its
13//! identity (keys deleted from the vault) and every grant; revocation is final on C2 as well.
14//! The ledger holds public data only (key id, public key, grants) and is written 0600.
15
16use crate::error::ServerError;
17use crate::subservice::{State, Status, Subservice};
18use m4a_agent::vault::EnvMasterKey;
19use m4a_agent::{BackendKind, FileVault, IdentityStore, KeyVault};
20use serde::{Deserialize, Serialize};
21use std::collections::BTreeMap;
22use std::path::{Path, PathBuf};
23use std::sync::{Arc, Mutex};
24use std::time::Duration;
25use tokio::io::{AsyncReadExt, AsyncWriteExt};
26
27pub use hatchery_enroll::{Grants, NodeStatus, Policy};
28
29/// What C2 tells us about a node.
30#[derive(Clone, Debug, Deserialize)]
31pub struct NodeInfo {
32    pub node_id: String,
33    pub status: NodeStatus,
34    #[serde(default)]
35    pub kind: Option<String>,
36}
37
38/// One node as the server provisioned it. Public data only.
39#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
40pub struct LedgerEntry {
41    pub node_id: String,
42    pub status: NodeStatus,
43    pub kind: Option<String>,
44    /// m4a identity (absent after revocation).
45    pub m4a_key_id: Option<String>,
46    pub m4a_public_key: Option<String>,
47    pub mailbox: Option<String>,
48    pub grants: Grants,
49    pub updated_at_ms: u64,
50}
51
52/// Read-only shape of a ledger entry for consumers (the mail server, tests): status as text.
53#[derive(Clone, Debug, Deserialize)]
54pub struct LedgerEntryView {
55    pub node_id: String,
56    pub status: String,
57    pub m4a_public_key: Option<String>,
58    pub mailbox: Option<String>,
59    pub grants: Grants,
60}
61
62/// What a reconcile pass did to one node.
63#[derive(Clone, Debug, Eq, PartialEq)]
64pub enum Change {
65    Created(String),
66    Updated(String, NodeStatus),
67    Revoked(String),
68}
69
70#[derive(Default, Serialize, Deserialize)]
71struct LedgerFile {
72    nodes: BTreeMap<String, LedgerEntry>,
73}
74
75/// Read the ledger a [`Provisioner`] wrote (consumers such as the messenger and task manager).
76pub fn read_ledger(data_dir: &Path) -> Result<Vec<LedgerEntry>, ServerError> {
77    let path = data_dir.join("provision").join("ledger.json");
78    match std::fs::read(&path) {
79        Ok(bytes) => {
80            let file: LedgerFile = serde_json::from_slice(&bytes).map_err(|e| ServerError::sub("provision", format!("ledger: {e}")))?;
81            Ok(file.nodes.into_values().collect())
82        }
83        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
84        Err(error) => Err(error.into()),
85    }
86}
87
88pub struct Provisioner {
89    ids: IdentityStore,
90    ledger_path: PathBuf,
91    ledger: LedgerFile,
92    policy: Policy,
93    server_ref: String,
94    mail_domain: String,
95}
96
97fn unix_ms() -> u64 {
98    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as u64
99}
100
101impl Provisioner {
102    pub fn new(vault: Arc<dyn KeyVault>, data_dir: &Path, server_ref: &str, mail_domain: &str) -> Result<Self, ServerError> {
103        let dir = data_dir.join("provision");
104        std::fs::create_dir_all(&dir)?;
105        let ledger_path = dir.join("ledger.json");
106        let ledger = match std::fs::read(&ledger_path) {
107            Ok(bytes) => serde_json::from_slice(&bytes).map_err(|e| ServerError::sub("provision", format!("ledger: {e}")))?,
108            Err(error) if error.kind() == std::io::ErrorKind::NotFound => LedgerFile::default(),
109            Err(error) => return Err(error.into()),
110        };
111        Ok(Self {
112            ids: IdentityStore::new(vault),
113            ledger_path,
114            ledger,
115            policy: Policy::default(),
116            server_ref: server_ref.to_owned(),
117            mail_domain: mail_domain.to_owned(),
118        })
119    }
120
121    /// Replace the status -> grants policy; the next `reconcile` applies the difference.
122    pub fn set_policy(&mut self, policy: Policy) {
123        self.policy = policy;
124    }
125
126    pub fn entries(&self) -> Vec<LedgerEntry> {
127        self.ledger.nodes.values().cloned().collect()
128    }
129
130    pub fn entry(&self, node_id: &str) -> Option<&LedgerEntry> {
131        self.ledger.nodes.get(node_id)
132    }
133
134    /// Bring the server side in line with `nodes`. Idempotent; a node already revoked stays so.
135    pub fn reconcile(&mut self, nodes: &[NodeInfo]) -> Result<Vec<Change>, ServerError> {
136        let mut changes = Vec::new();
137        for node in nodes {
138            let existing = self.ledger.nodes.get(&node.node_id).cloned();
139            if existing.as_ref().is_some_and(|e| e.status == NodeStatus::Revoked) {
140                continue; // final
141            }
142            let grants = self.policy.grants(node.status);
143            if existing.as_ref().is_some_and(|e| e.status == node.status && e.kind == node.kind && e.grants == grants) {
144                continue;
145            }
146            let entry = if node.status == NodeStatus::Revoked {
147                self.ids.forget(&node.node_id).map_err(|e| ServerError::sub("provision", e.to_string()))?;
148                LedgerEntry {
149                    node_id: node.node_id.clone(),
150                    status: node.status,
151                    kind: node.kind.clone(),
152                    m4a_key_id: None,
153                    m4a_public_key: None,
154                    mailbox: None,
155                    grants,
156                    updated_at_ms: unix_ms(),
157                }
158            } else {
159                let mut identity = self
160                    .ids
161                    .resolve(&node.node_id, BackendKind::Server, &self.server_ref)
162                    .map_err(|e| ServerError::sub("provision", e.to_string()))?;
163                // The nick is the node id; frozen after the first call.
164                self.ids.adopt_nick(&mut identity, &node.node_id).map_err(|e| ServerError::sub("provision", e.to_string()))?;
165                LedgerEntry {
166                    node_id: node.node_id.clone(),
167                    status: node.status,
168                    kind: node.kind.clone(),
169                    m4a_key_id: Some(identity.key_id.clone()),
170                    m4a_public_key: Some(identity.public_key.clone()),
171                    mailbox: grants.mailbox.then(|| format!("{}@{}", node.node_id, self.mail_domain)),
172                    grants,
173                    updated_at_ms: unix_ms(),
174                }
175            };
176            changes.push(match (&existing, node.status) {
177                (_, NodeStatus::Revoked) => Change::Revoked(node.node_id.clone()),
178                (None, _) => Change::Created(node.node_id.clone()),
179                _ => Change::Updated(node.node_id.clone(), node.status),
180            });
181            self.ledger.nodes.insert(node.node_id.clone(), entry);
182        }
183        if !changes.is_empty() {
184            self.save()?;
185        }
186        Ok(changes)
187    }
188
189    fn save(&self) -> Result<(), ServerError> {
190        let bytes = serde_json::to_vec_pretty(&self.ledger).map_err(|e| ServerError::sub("provision", e.to_string()))?;
191        let tmp = self.ledger_path.with_extension("tmp");
192        {
193            use std::io::Write;
194            let mut options = std::fs::OpenOptions::new();
195            options.write(true).create(true).truncate(true);
196            #[cfg(unix)]
197            {
198                use std::os::unix::fs::OpenOptionsExt;
199                options.mode(0o600);
200            }
201            options.open(&tmp)?.write_all(&bytes)?;
202        }
203        std::fs::rename(tmp, &self.ledger_path)?;
204        Ok(())
205    }
206}
207
208/// Where the node list comes from: the C2 operator API (`GET /v1/enroll/nodes`, bearer).
209#[derive(Clone)]
210pub struct C2Source {
211    pub addr: String,
212    pub bearer: String,
213}
214
215impl C2Source {
216    pub async fn fetch(&self) -> Result<Vec<NodeInfo>, ServerError> {
217        #[derive(Deserialize)]
218        struct Reply {
219            nodes: Vec<NodeInfo>,
220        }
221        let io = async {
222            let mut stream = tokio::net::TcpStream::connect(&self.addr).await?;
223            let request = format!(
224                "GET /v1/enroll/nodes HTTP/1.1\r\nHost: c2\r\nAuthorization: Bearer {}\r\nConnection: close\r\n\r\n",
225                self.bearer
226            );
227            stream.write_all(request.as_bytes()).await?;
228            let mut raw = Vec::new();
229            stream.read_to_end(&mut raw).await?;
230            Ok::<_, std::io::Error>(raw)
231        };
232        let raw = tokio::time::timeout(Duration::from_secs(10), io)
233            .await
234            .map_err(|_| ServerError::sub("provision", "C2 request timed out"))??;
235        let split = raw
236            .windows(4)
237            .position(|w| w == b"\r\n\r\n")
238            .ok_or_else(|| ServerError::sub("provision", "C2 reply has no header end"))?;
239        let head = String::from_utf8_lossy(&raw[..split]);
240        let status: u16 = head.split_whitespace().nth(1).and_then(|s| s.parse().ok()).unwrap_or(0);
241        if status != 200 {
242            return Err(ServerError::sub("provision", format!("C2 answered {status}")));
243        }
244        let reply: Reply = serde_json::from_slice(&raw[split + 4..]).map_err(|e| ServerError::sub("provision", format!("C2 reply: {e}")))?;
245        Ok(reply.nodes)
246    }
247}
248
249#[derive(Clone, Debug, Default)]
250struct Summary {
251    nodes: usize,
252    last_error: Option<String>,
253    last_pass_ms: u64,
254}
255
256pub struct ProvisionConfig {
257    pub data_dir: PathBuf,
258    pub master_key_var: String,
259    pub c2_addr: String,
260    pub c2_token_var: String,
261    pub interval: Duration,
262    pub server_ref: String,
263    pub mail_domain: String,
264    /// Same policy file C2 enforces (`HATCHERY_POLICY_FILE`); `None` uses the built-in table.
265    pub policy_path: Option<PathBuf>,
266}
267
268/// Runs the poll loop as a supervised subservice.
269pub struct ProvisionSubservice {
270    config: ProvisionConfig,
271    task: Option<tokio::task::JoinHandle<()>>,
272    summary: Arc<Mutex<Summary>>,
273    failed: Option<String>,
274}
275
276impl ProvisionSubservice {
277    pub fn new(config: ProvisionConfig) -> Self {
278        Self { config, task: None, summary: Arc::new(Mutex::new(Summary::default())), failed: None }
279    }
280
281    fn build(&self) -> Result<(Provisioner, C2Source), ServerError> {
282        let c = &self.config;
283        let bearer = std::env::var(&c.c2_token_var)
284            .ok()
285            .filter(|v| !v.is_empty())
286            .ok_or_else(|| ServerError::sub("provision", "the C2 operator token variable is not set"))?;
287        let dir = c.data_dir.join("m4a").join(".m4a-agent");
288        std::fs::create_dir_all(&dir)?;
289        let vault = FileVault::open(dir.join("vault.enc"), Box::new(EnvMasterKey { var: c.master_key_var.clone() }))
290            .map_err(|e| ServerError::sub("provision", format!("key vault: {e}")))?;
291        let provisioner = Provisioner::new(Arc::new(vault), &c.data_dir, &c.server_ref, &c.mail_domain)?;
292        Ok((provisioner, C2Source { addr: c.c2_addr.clone(), bearer }))
293    }
294}
295
296impl Subservice for ProvisionSubservice {
297    fn name(&self) -> &'static str {
298        "provision"
299    }
300
301    fn start(&mut self) -> Result<(), ServerError> {
302        if self.task.is_some() {
303            return Ok(());
304        }
305        let (mut provisioner, source) = match self.build() {
306            Ok(parts) => parts,
307            Err(error) => {
308                self.failed = Some(error.to_string());
309                return Err(error);
310            }
311        };
312        self.failed = None;
313        let summary = Arc::clone(&self.summary);
314        let interval = self.config.interval;
315        let policy_path = self.config.policy_path.clone();
316        self.task = Some(tokio::spawn(async move {
317            let mut policy_stamp: Option<std::time::SystemTime> = None;
318            loop {
319                // Policy reload: a changed (or newly valid) file is applied on the next pass; an
320                // invalid file keeps the previous policy and is reported.
321                let mut policy_error = None;
322                if let Some(path) = &policy_path {
323                    let stamp = std::fs::metadata(path).and_then(|m| m.modified()).ok();
324                    if stamp != policy_stamp {
325                        match Policy::load(path) {
326                            Ok(policy) => {
327                                provisioner.set_policy(policy);
328                                policy_stamp = stamp;
329                                tracing::info!("provisioning policy loaded");
330                            }
331                            Err(error) => policy_error = Some(format!("policy file: {error}")),
332                        }
333                    }
334                }
335                let outcome = match source.fetch().await {
336                    Ok(nodes) => provisioner.reconcile(&nodes).map(|changes| (nodes.len(), changes)),
337                    Err(error) => Err(error),
338                };
339                {
340                    let mut s = summary.lock().unwrap_or_else(|p| p.into_inner());
341                    s.last_pass_ms = unix_ms();
342                    match outcome {
343                        Ok((count, changes)) => {
344                            s.nodes = count;
345                            s.last_error = policy_error;
346                            for change in changes {
347                                tracing::info!(?change, "node provisioned");
348                            }
349                        }
350                        Err(error) => {
351                            tracing::warn!(%error, "provisioning pass failed");
352                            s.last_error = Some(error.to_string());
353                        }
354                    }
355                }
356                tokio::time::sleep(interval).await;
357            }
358        }));
359        Ok(())
360    }
361
362    fn stop(&mut self) {
363        if let Some(task) = self.task.take() {
364            task.abort();
365        }
366    }
367
368    fn status(&mut self) -> Status {
369        let summary = self.summary.lock().unwrap_or_else(|p| p.into_inner()).clone();
370        let running = self.task.as_ref().is_some_and(|t| !t.is_finished());
371        let state = if running && summary.last_error.is_none() {
372            State::Running
373        } else if self.failed.is_some() || summary.last_error.is_some() {
374            State::Failed
375        } else {
376            State::Stopped
377        };
378        let detail = self.failed.clone().or(summary.last_error).or_else(|| running.then(|| format!("{} nodes", summary.nodes)));
379        Status { name: "provision", state, port: None, detail }
380    }
381}
382
383#[cfg(test)]
384mod tests {
385    use super::*;
386    use m4a_agent::MemoryVault;
387
388    fn policy(status: NodeStatus) -> Grants {
389        Policy::default().grants(status)
390    }
391
392    fn node(id: &str, status: NodeStatus) -> NodeInfo {
393        NodeInfo { node_id: id.to_owned(), status, kind: Some("qemu".to_owned()) }
394    }
395
396    fn provisioner(dir: &Path, vault: Arc<dyn KeyVault>) -> Provisioner {
397        Provisioner::new(vault, dir, "https://messenger.test", "nodes.test").unwrap()
398    }
399
400    #[test]
401    fn a_policy_change_re_grants_without_changing_status_or_key() {
402        let dir = tempfile::tempdir().unwrap();
403        let vault: Arc<dyn KeyVault> = Arc::new(MemoryVault::new());
404        let mut p = provisioner(dir.path(), Arc::clone(&vault));
405        p.reconcile(&[node("n", NodeStatus::Member)]).unwrap();
406        let key = p.entry("n").unwrap().m4a_public_key.clone();
407        p.set_policy(Policy::from_json(r#"{"member":{"mailbox":false,"rooms":["general"],"capabilities":["heartbeat"]}}"#).unwrap());
408        let c = p.reconcile(&[node("n", NodeStatus::Member)]).unwrap();
409        assert_eq!(c, [Change::Updated("n".into(), NodeStatus::Member)]);
410        let e = p.entry("n").unwrap();
411        assert!(e.mailbox.is_none() && e.grants.rooms == ["general"] && e.m4a_public_key == key);
412        assert!(p.reconcile(&[node("n", NodeStatus::Member)]).unwrap().is_empty());
413    }
414
415    #[test]
416    fn policy_grows_with_status_and_is_empty_when_out() {
417        let n = policy(NodeStatus::Neutral);
418        let m = policy(NodeStatus::Member);
419        let t = policy(NodeStatus::Trusted);
420        assert!(!n.mailbox && n.rooms == ["lobby"] && n.capabilities == ["heartbeat"]);
421        assert!(m.mailbox && m.rooms.len() > n.rooms.len() && m.capabilities.contains(&"mail".to_owned()));
422        assert!(t.rooms.len() > m.rooms.len() && t.capabilities.contains(&"tasks".to_owned()));
423        assert_eq!(policy(NodeStatus::Suspended), Grants::default());
424        assert_eq!(policy(NodeStatus::Revoked), Grants::default());
425    }
426
427    #[test]
428    fn enroll_creates_identity_neutral_gets_almost_nothing_then_flips() {
429        let dir = tempfile::tempdir().unwrap();
430        let vault: Arc<dyn KeyVault> = Arc::new(MemoryVault::new());
431        let mut p = provisioner(dir.path(), Arc::clone(&vault));
432        let c = p.reconcile(&[node("lab1", NodeStatus::Neutral)]).unwrap();
433        assert_eq!(c, [Change::Created("lab1".into())]);
434        let e = p.entry("lab1").unwrap().clone();
435        assert!(e.m4a_public_key.is_some() && e.mailbox.is_none());
436        assert_eq!(e.grants, policy(NodeStatus::Neutral));
437        // the key pair really is in the vault and the ledger holds only the public half
438        assert!(vault.get("identity-key/lab1").unwrap().is_some());
439        let text = std::fs::read_to_string(dir.path().join("provision/ledger.json")).unwrap();
440        assert!(!text.contains("identity-key"));
441        // idempotent
442        assert!(p.reconcile(&[node("lab1", NodeStatus::Neutral)]).unwrap().is_empty());
443        // approval: same key, more grants, mailbox appears
444        let c = p.reconcile(&[node("lab1", NodeStatus::Member)]).unwrap();
445        assert_eq!(c, [Change::Updated("lab1".into(), NodeStatus::Member)]);
446        let m = p.entry("lab1").unwrap();
447        assert_eq!(m.m4a_public_key, e.m4a_public_key, "the identity is stable across status changes");
448        assert_eq!(m.mailbox.as_deref(), Some("lab1@nodes.test"));
449        p.reconcile(&[node("lab1", NodeStatus::Trusted)]).unwrap();
450        assert!(p.entry("lab1").unwrap().grants.rooms.contains(&"ops".to_owned()));
451    }
452
453    #[test]
454    fn suspend_keeps_the_key_but_removes_every_grant() {
455        let dir = tempfile::tempdir().unwrap();
456        let vault: Arc<dyn KeyVault> = Arc::new(MemoryVault::new());
457        let mut p = provisioner(dir.path(), Arc::clone(&vault));
458        p.reconcile(&[node("n", NodeStatus::Trusted)]).unwrap();
459        p.reconcile(&[node("n", NodeStatus::Suspended)]).unwrap();
460        let e = p.entry("n").unwrap();
461        assert!(e.m4a_public_key.is_some() && e.mailbox.is_none() && e.grants == Grants::default());
462        assert!(vault.get("identity-key/n").unwrap().is_some());
463        p.reconcile(&[node("n", NodeStatus::Member)]).unwrap();
464        assert!(p.entry("n").unwrap().mailbox.is_some(), "a suspended node can be restored");
465    }
466
467    #[test]
468    fn revoke_deletes_the_identity_and_is_final() {
469        let dir = tempfile::tempdir().unwrap();
470        let vault: Arc<dyn KeyVault> = Arc::new(MemoryVault::new());
471        let mut p = provisioner(dir.path(), Arc::clone(&vault));
472        p.reconcile(&[node("n", NodeStatus::Member)]).unwrap();
473        let c = p.reconcile(&[node("n", NodeStatus::Revoked)]).unwrap();
474        assert_eq!(c, [Change::Revoked("n".into())]);
475        assert!(vault.get("identity-key/n").unwrap().is_none(), "keys are gone from the vault");
476        let e = p.entry("n").unwrap();
477        assert!(e.m4a_public_key.is_none() && e.grants == Grants::default());
478        // nothing brings it back, even if the source says member again
479        assert!(p.reconcile(&[node("n", NodeStatus::Member)]).unwrap().is_empty());
480        assert_eq!(p.entry("n").unwrap().status, NodeStatus::Revoked);
481        assert!(vault.get("identity-key/n").unwrap().is_none());
482    }
483
484    #[test]
485    fn ledger_survives_a_restart_and_is_private() {
486        let dir = tempfile::tempdir().unwrap();
487        let vault: Arc<dyn KeyVault> = Arc::new(MemoryVault::new());
488        {
489            let mut p = provisioner(dir.path(), Arc::clone(&vault));
490            p.reconcile(&[node("a", NodeStatus::Member)]).unwrap();
491        }
492        let mut p = provisioner(dir.path(), Arc::clone(&vault));
493        assert_eq!(p.entry("a").unwrap().status, NodeStatus::Member);
494        assert!(p.reconcile(&[node("a", NodeStatus::Member)]).unwrap().is_empty());
495        #[cfg(unix)]
496        {
497            use std::os::unix::fs::PermissionsExt;
498            let mode = std::fs::metadata(dir.path().join("provision/ledger.json")).unwrap().permissions().mode();
499            assert_eq!(mode & 0o777, 0o600);
500        }
501    }
502
503    #[test]
504    fn real_file_vault_needs_the_env_master_key() {
505        let dir = tempfile::tempdir().unwrap();
506        let var = "HATCHERY_TEST_PROVISION_MASTER";
507        std::env::set_var(var, "test-master-secret-0123456789");
508        let config = |v: &str| ProvisionConfig {
509            data_dir: dir.path().to_owned(),
510            master_key_var: v.to_owned(),
511            c2_addr: "127.0.0.1:1".into(),
512            c2_token_var: var.to_owned(),
513            interval: Duration::from_secs(1),
514            server_ref: "https://messenger.test".into(),
515            mail_domain: "nodes.test".into(),
516            policy_path: None,
517        };
518        let (mut p, _) = ProvisionSubservice::new(config(var)).build().unwrap();
519        p.reconcile(&[node("k", NodeStatus::Member)]).unwrap();
520        assert!(dir.path().join("m4a/.m4a-agent/vault.enc").exists());
521        // the seed is not readable from the ledger or the directory listing of provision/
522        for f in std::fs::read_dir(dir.path().join("provision")).unwrap() {
523            let text = std::fs::read(f.unwrap().path()).unwrap();
524            assert!(!String::from_utf8_lossy(&text).contains("test-master-secret"));
525        }
526        std::env::remove_var(var);
527    }
528}
hatchery_server/provision.rs

hatchery_server/
provision.rs
