hatchery_c2/
enroll_service.rs

1//! Runtime side of enrollment: the public enroll listener, operator operations on the
2//! registry, WireGuard peer provisioning and the signed call-home decision.
3//!
4//! Secrets never appear in logs: neither enrollment codes, nor session tokens, nor keys.
5
6use std::collections::HashMap;
7use std::io;
8use std::net::{IpAddr, SocketAddr};
9use std::process::Command;
10use std::sync::{Arc, Mutex};
11use std::time::{Duration, Instant};
12
13use hatchery_enroll::{CallHomeChallenge, EnrollmentRequest, EnrollmentResponse, Policy, ANSWER_LEN};
14use serde::{Deserialize, Serialize};
15use tokio::io::{AsyncReadExt, AsyncWriteExt};
16use tokio::net::{TcpListener, TcpStream};
17use tokio::sync::{mpsc, watch};
18use tokio::time::timeout;
19
20use crate::enrollment::{EnrollmentError, EnrollmentRegistry, NodeRecord, NodeStatus, TunnelPool};
21use hatchery_enroll::call::ForwardedCall;
22
23const MAX_BODY: usize = 16 * 1024;
24const IO_DEADLINE: Duration = Duration::from_secs(5);
25const BUILD_DEADLINE: Duration = Duration::from_secs(20 * 60);
26const DEFAULT_CODE_TTL_MS: u64 = 15 * 60 * 1000;
27
28/// Answers "may this node do X right now?" for the enforcement points in the relay.
29/// `None` means the node is not an enrolled node (a token node), which policy does not govern.
30pub trait CapabilityGate: Send + Sync {
31    fn allows(&self, node_id: &str, capability: &str) -> Option<bool>;
32}
33
34/// Core-side applicator for admitted node calls (A2).
35///
36/// Invoked in-process after C2 policy checks. Must not dial core over the network —
37/// production wires this to a drain consumer; tests may apply directly to a Hub.
38#[async_trait::async_trait]
39pub trait NodeCallSink: Send + Sync {
40    async fn apply(&self, call: ForwardedCall) -> HttpReply;
41}
42
43/// Adds and removes node WireGuard peers.
44pub trait PeerProvisioner: Send + Sync {
45    fn add_peer(&self, wg_public_key: &str, tunnel_ip: &str) -> io::Result<()>;
46    fn remove_peer(&self, wg_public_key: &str) -> io::Result<()>;
47}
48
49/// Kernel WireGuard through the `wg` tool (`wg set IF peer KEY allowed-ips IP/32`). Needs
50/// CAP_NET_ADMIN. Runtime-only: C2 re-applies all active nodes at start.
51pub struct WgCli {
52    pub interface: String,
53}
54
55impl PeerProvisioner for WgCli {
56    fn add_peer(&self, key: &str, ip: &str) -> io::Result<()> {
57        run_wg(&["set", &self.interface, "peer", key, "allowed-ips", &format!("{ip}/32")])
58    }
59    fn remove_peer(&self, key: &str) -> io::Result<()> {
60        run_wg(&["set", &self.interface, "peer", key, "remove"])
61    }
62}
63
64fn run_wg(args: &[&str]) -> io::Result<()> {
65    let output = Command::new("wg").args(args).output()?;
66    if output.status.success() {
67        Ok(())
68    } else {
69        Err(io::Error::other(format!("wg exited with {}", output.status)))
70    }
71}
72
73/// Provisioner that does nothing (tests, nodes whose tunnel is managed elsewhere).
74pub struct NoopProvisioner;
75
76impl PeerProvisioner for NoopProvisioner {
77    fn add_peer(&self, _: &str, _: &str) -> io::Result<()> {
78        Ok(())
79    }
80    fn remove_peer(&self, _: &str) -> io::Result<()> {
81        Ok(())
82    }
83}
84
85/// Public values a node needs to bring its tunnel up. No secrets.
86#[derive(Clone, Default)]
87pub struct PublicInfo {
88    pub c2_tunnel_ip: Option<String>,
89    pub c2_wg_public_key: Option<String>,
90    pub c2_wg_endpoint: Option<String>,
91    pub c2_call_home: Option<String>,
92    /// A2: outer-layer C2 public key (base64) handed to nodes at enrollment.
93    pub upstream_c2_public_key: Option<String>,
94    /// A2: HQ recipient public key (base64) relayed to nodes; C2 has no private half.
95    pub upstream_recipient_public_key: Option<String>,
96    /// Messenger product URL for enrolled nodes (operator sets `HATCHERY_M4A_PRODUCT_URL` on C2).
97    pub m4a_product_url: Option<String>,
98}
99
100#[derive(Clone)]
101pub struct EnrollConfig {
102    pub registry_path: Option<std::path::PathBuf>,
103    /// Default true: enrollment only with a one-time operator code.
104    pub require_code: bool,
105    /// Code enrollments only from a loopback peer (the TLS proxy on this host). A code sent in
106    /// the clear from any other peer is burned (it was exposed) and the request refused.
107    pub tls_only: bool,
108    pub pool: Option<TunnelPool>,
109    /// Where nodes POST their enrollment request.
110    pub listen: SocketAddr,
111    pub provisioner: Arc<dyn PeerProvisioner>,
112    pub public: PublicInfo,
113    /// Requests per source address per minute (applies to every request, valid or not).
114    pub rate_limit_per_minute: u32,
115    /// Status -> capability policy file (JSON); `None` uses the built-in table.
116    pub policy_path: Option<std::path::PathBuf>,
117    /// Build host to forward admitted `POST /v1/build` requests to, and the bearer it expects.
118    pub build_host: Option<(SocketAddr, String)>,
119    /// Deprecated (A2): previously the core address C2 pushed node-calls to.
120    /// Ignored — C2 never dials core; retained so old configs still parse.
121    pub server_upstream: Option<(SocketAddr, String)>,
122    /// In-process sink for admitted node calls (no network push to core).
123    pub node_call_sink: Option<Arc<dyn NodeCallSink>>,
124}
125
126impl EnrollConfig {
127    pub fn new(listen: SocketAddr) -> Self {
128        Self {
129            registry_path: None,
130            require_code: true,
131            tls_only: false,
132            pool: None,
133            listen,
134            provisioner: Arc::new(NoopProvisioner),
135            public: PublicInfo::default(),
136            rate_limit_per_minute: 30,
137            policy_path: None,
138            build_host: None,
139            server_upstream: None,
140            node_call_sink: None,
141        }
142    }
143}
144
145/// What the runtime must do after an operator or node action.
146#[derive(Clone, Debug, Eq, PartialEq)]
147pub enum EnrollEvent {
148    /// Make sure a relay worker exists for this node.
149    Enrolled(String),
150    /// Status changed; nodes that lost the heartbeat capability are disconnected.
151    StatusChanged(String, NodeStatus),
152    /// The policy was reloaded; nodes that no longer hold the heartbeat capability are
153    /// disconnected.
154    PolicyChanged,
155}
156
157pub struct EnrollService {
158    registry: Mutex<EnrollmentRegistry>,
159    policy: std::sync::RwLock<Policy>,
160    policy_path: Option<std::path::PathBuf>,
161    provisioner: Arc<dyn PeerProvisioner>,
162    public: PublicInfo,
163    events: mpsc::UnboundedSender<EnrollEvent>,
164    rate: Mutex<HashMap<IpAddr, (Instant, u32)>>,
165    rate_limit: u32,
166    tls_only: bool,
167    build_host: Option<(SocketAddr, String)>,
168    server_upstream: Option<(SocketAddr, String)>,
169    node_call_sink: Option<Arc<dyn NodeCallSink>>,
170    build_nonces: Mutex<HashMap<(String, String), u64>>,
171    node_rate: Mutex<HashMap<(String, &'static str), (u64, u32)>>,
172    pub listen: SocketAddr,
173}
174
175#[derive(Serialize)]
176struct ErrorBody<'a> {
177    error: &'a str,
178}
179
180pub struct HttpReply {
181    pub status: u16,
182    pub body: Vec<u8>,
183}
184
185impl HttpReply {
186    fn json<T: Serialize>(status: u16, value: &T) -> Self {
187        Self { status, body: serde_json::to_vec(value).unwrap_or_default() }
188    }
189    fn error(status: u16, message: &str) -> Self {
190        Self::json(status, &ErrorBody { error: message })
191    }
192}
193
194fn unix_ms() -> u64 {
195    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as u64
196}
197
198fn is_active(status: NodeStatus) -> bool {
199    !matches!(status, NodeStatus::Suspended | NodeStatus::Revoked)
200}
201
202impl EnrollService {
203    pub fn start(config: &EnrollConfig) -> Result<(Arc<Self>, mpsc::UnboundedReceiver<EnrollEvent>), EnrollmentError> {
204        let mut registry = match &config.registry_path {
205            Some(path) => EnrollmentRegistry::open(path, config.require_code)?,
206            None => EnrollmentRegistry::in_memory(config.require_code),
207        };
208        if let Some(pool) = config.pool {
209            registry = registry.with_pool(pool);
210        }
211        let policy = match &config.policy_path {
212            Some(path) => Policy::load(path).map_err(|e| EnrollmentError::Storage(e.to_string()))?,
213            None => Policy::default(),
214        };
215        let (tx, rx) = mpsc::unbounded_channel();
216        let service = Arc::new(Self {
217            registry: Mutex::new(registry),
218            policy: std::sync::RwLock::new(policy),
219            policy_path: config.policy_path.clone(),
220            provisioner: Arc::clone(&config.provisioner),
221            public: config.public.clone(),
222            events: tx,
223            rate: Mutex::new(HashMap::new()),
224            rate_limit: config.rate_limit_per_minute.max(1),
225            tls_only: config.tls_only,
226            build_host: config.build_host.clone(),
227            server_upstream: config.server_upstream.clone(),
228            node_call_sink: config.node_call_sink.clone(),
229            build_nonces: Mutex::new(HashMap::new()),
230            node_rate: Mutex::new(HashMap::new()),
231            listen: config.listen,
232        });
233        // Restart persistence: re-apply every active node's peer and recreate its worker.
234        for record in service.registry.lock().expect("registry lock").list() {
235            if is_active(record.status) {
236                service.apply_peer(&record);
237                let _ = service.events.send(EnrollEvent::Enrolled(record.node_id.clone()));
238            }
239        }
240        Ok((service, rx))
241    }
242
243    fn apply_peer(&self, record: &NodeRecord) {
244        let Some(ip) = &record.tunnel_ip else { return };
245        let result = if is_active(record.status) {
246            self.provisioner.add_peer(&record.wg_public_key, ip)
247        } else {
248            self.provisioner.remove_peer(&record.wg_public_key)
249        };
250        if let Err(error) = result {
251            tracing::warn!(node_id = %record.node_id, cause = %error, "wireguard peer update failed");
252        }
253    }
254
255    fn allow(&self, peer: IpAddr) -> bool {
256        let mut rate = self.rate.lock().expect("rate lock");
257        let now = Instant::now();
258        let entry = rate.entry(peer).or_insert((now, 0));
259        if now.duration_since(entry.0) > Duration::from_secs(60) {
260            *entry = (now, 0);
261        }
262        entry.1 += 1;
263        entry.1 <= self.rate_limit
264    }
265
266    /// Public enroll request body -> reply. Errors are generic towards the caller.
267    pub fn handle_enroll(&self, body: &[u8], peer: IpAddr) -> HttpReply {
268        if !self.allow(peer) {
269            return HttpReply::error(429, "rate limited");
270        }
271        let Ok(request) = serde_json::from_slice::<EnrollmentRequest>(body) else {
272            return HttpReply::error(400, "malformed request");
273        };
274        if self.tls_only && !peer.is_loopback() {
275            if let Some(code) = request.code.as_deref() {
276                let burned = self.registry.lock().expect("registry lock").burn_code(code, unix_ms());
277                tracing::warn!(%peer, burned, "plaintext enrollment refused; the exposed code was burned (use https)");
278            }
279            return HttpReply::error(403, "enrollment requires https");
280        }
281        let result = self.registry.lock().expect("registry lock").submit(&request, unix_ms());
282        match result {
283            Ok(record) => {
284                self.apply_peer(&record);
285                let _ = self.events.send(EnrollEvent::Enrolled(record.node_id.clone()));
286                tracing::info!(node_id = %record.node_id, status = ?record.status, "node enrolled");
287                HttpReply::json(
288                    200,
289                    &EnrollmentResponse {
290                        node_id: record.node_id,
291                        status: format!("{:?}", record.status).to_lowercase(),
292                        tunnel_ip: record.tunnel_ip,
293                        c2_tunnel_ip: self.public.c2_tunnel_ip.clone(),
294                        c2_wg_public_key: self.public.c2_wg_public_key.clone(),
295                        c2_wg_endpoint: self.public.c2_wg_endpoint.clone(),
296                        c2_call_home: self.public.c2_call_home.clone(),
297                        upstream_c2_public_key: self.public.upstream_c2_public_key.clone(),
298                        upstream_recipient_public_key: self.public.upstream_recipient_public_key.clone(),
299                        m4a_product_url: self.public.m4a_product_url.clone(),
300                    },
301                )
302            }
303            Err(error) => {
304                tracing::warn!(cause = %error, %peer, "enrollment refused");
305                // Do not reveal which check failed beyond the broad class.
306                match error {
307                    EnrollmentError::Storage(_) => HttpReply::error(503, "enrollment unavailable"),
308                    EnrollmentError::Revoked => HttpReply::error(403, "refused"),
309                    _ => HttpReply::error(403, "refused"),
310                }
311            }
312        }
313    }
314
315    /// Per-node limit for signature-authenticated calls (applied after the signature checks, so
316    /// anonymous traffic cannot burn a node's budget): per minute.
317    fn node_allow(&self, node_id: &str, kind: &'static str, per_minute: u32) -> bool {
318        let now = unix_ms();
319        let mut map = self.node_rate.lock().expect("rate lock");
320        let slot = map.entry((node_id.to_owned(), kind)).or_insert((now, 0));
321        if now.saturating_sub(slot.0) >= 60_000 {
322            *slot = (now, 0);
323        }
324        slot.1 += 1;
325        slot.1 <= per_minute
326    }
327
328    /// `POST /v1/build`: a node-signed request to build and sign a configured recipe.
329    /// Checks, in order: shape, clock window, replay, node signature (enrolled identity key),
330    /// status/capability policy (`build`), then forwards to the build host with a bearer the
331    /// node never sees. Failures towards the node are generic.
332    pub async fn handle_build(&self, body: &[u8], peer: IpAddr) -> HttpReply {
333        use hatchery_build::wire::{HostRequest, SignedBuildRequest, MAX_SKEW_MS};
334        let Ok(request) = serde_json::from_slice::<SignedBuildRequest>(body) else {
335            return HttpReply::error(400, "malformed request");
336        };
337        if !request.well_formed() {
338            return HttpReply::error(400, "malformed request");
339        }
340        let now = unix_ms();
341        if now.abs_diff(request.ts_ms) > MAX_SKEW_MS {
342            return HttpReply::error(403, "refused");
343        }
344        let message = SignedBuildRequest::signing_bytes(&request.node_id, &request.recipe, request.source_ref.as_deref(), request.ts_ms, &request.nonce);
345        let verified = self.registry.lock().expect("registry lock").verify_node_message(&request.node_id, &message, &request.signature);
346        if verified.is_err() {
347            tracing::warn!(node_id = %request.node_id, %peer, "build request refused: signature");
348            return HttpReply::error(403, "refused");
349        }
350        if !self.node_allow(&request.node_id, "build", 30) {
351            return HttpReply::error(429, "rate limited");
352        }
353        {
354            let mut seen = self.build_nonces.lock().expect("nonce lock");
355            seen.retain(|_, ts| now.abs_diff(*ts) <= MAX_SKEW_MS * 2);
356            if seen.insert((request.node_id.clone(), request.nonce.clone()), request.ts_ms).is_some() {
357                tracing::warn!(node_id = %request.node_id, "build request refused: replay");
358                return HttpReply::error(403, "refused");
359            }
360        }
361        if self.allows(&request.node_id, "build") != Some(true) {
362            tracing::warn!(node_id = %request.node_id, "build request refused: status does not hold build");
363            return HttpReply::error(403, "refused");
364        }
365        let Some((addr, bearer)) = &self.build_host else {
366            return HttpReply::error(503, "build service unavailable");
367        };
368        let forward = HostRequest { node_id: request.node_id.clone(), recipe: request.recipe.clone(), source_ref: request.source_ref.clone() };
369        let payload = serde_json::to_vec(&forward).unwrap_or_default();
370        match hatchery_build::http::post(*addr, "/v1/build", Some(bearer), &payload, BUILD_DEADLINE).await {
371            Ok((status, body)) => {
372                tracing::info!(node_id = %request.node_id, recipe = %request.recipe, status, "build forwarded");
373                HttpReply { status, body }
374            }
375            Err(error) => {
376                tracing::warn!(node_id = %request.node_id, cause = %error, "build host unreachable");
377                HttpReply::error(502, "build service unavailable")
378            }
379        }
380    }
381
382    /// `POST /v1/node-call`: a node-signed mail/task operation. Same gate as builds (shape, clock,
383    /// replay, signature, status policy for the operation's capability), then forwarded to the
384    /// server's control API with a bearer only C2 holds. The server re-checks its own ledger.
385    pub async fn handle_node_call(&self, body: &[u8], peer: IpAddr) -> HttpReply {
386        use hatchery_enroll::call::{capability_for, SignedCall, MAX_SKEW_MS};
387        let Ok(call) = serde_json::from_slice::<SignedCall>(body) else {
388            return HttpReply::error(400, "malformed request");
389        };
390        if !call.well_formed() {
391            return HttpReply::error(400, "malformed request");
392        }
393        let now = unix_ms();
394        if now.abs_diff(call.ts_ms) > MAX_SKEW_MS {
395            return HttpReply::error(403, "refused");
396        }
397        let message = SignedCall::signing_bytes(&call.node_id, &call.op, &call.payload, call.ts_ms, &call.nonce);
398        if self.registry.lock().expect("registry lock").verify_node_message(&call.node_id, &message, &call.signature).is_err() {
399            tracing::warn!(node_id = %call.node_id, %peer, "node call refused: signature");
400            return HttpReply::error(403, "refused");
401        }
402        if !self.node_allow(&call.node_id, "call", 600) {
403            return HttpReply::error(429, "rate limited");
404        }
405        {
406            let mut seen = self.build_nonces.lock().expect("nonce lock");
407            seen.retain(|_, ts| now.abs_diff(*ts) <= MAX_SKEW_MS * 2);
408            if seen.insert((call.node_id.clone(), call.nonce.clone()), call.ts_ms).is_some() {
409                tracing::warn!(node_id = %call.node_id, "node call refused: replay");
410                return HttpReply::error(403, "refused");
411            }
412        }
413        if call.op == hatchery_enroll::call::NODE_CONFIG_OP {
414            // Config value only: the enrolled node gets C2's public settings. Nothing is relayed.
415            if self.registry.lock().expect("registry lock").get(&call.node_id).is_none() {
416                return HttpReply::error(403, "refused");
417            }
418            return HttpReply::json(200, &serde_json::json!({ "m4a_product_url": self.public.m4a_product_url }));
419        }
420        let capability = capability_for(&call.op).unwrap_or("none");
421        if self.allows(&call.node_id, capability) != Some(true) {
422            tracing::warn!(node_id = %call.node_id, op = %call.op, "node call refused: status does not hold the capability");
423            return HttpReply::error(403, "refused");
424        }
425        // A2: never dial core. Apply through an in-process sink (drain consumer / test hub).
426        let _ = capability;
427        let Ok(payload) = serde_json::from_str::<serde_json::Value>(&call.payload) else {
428            return HttpReply::error(400, "malformed payload");
429        };
430        let forward = ForwardedCall {
431            node_id: call.node_id.clone(),
432            op: call.op.clone(),
433            payload,
434        };
435        let Some(sink) = &self.node_call_sink else {
436            if self.server_upstream.is_some() {
437                tracing::warn!(
438                    node_id = %call.node_id,
439                    op = %call.op,
440                    "node-call push to core is removed (A2); configure node_call_sink"
441                );
442            }
443            return HttpReply::error(410, "use restreamer uplink");
444        };
445        sink.apply(forward).await
446    }
447
448    /// Operator endpoints under `/v1/enroll/...` (the caller has checked the bearer).
449    pub fn handle_operator(&self, method: &str, path: &str, body: &[u8]) -> HttpReply {
450        let rest = path.trim_start_matches("/v1/enroll").trim_matches('/');
451        let parts: Vec<&str> = rest.split('/').filter(|p| !p.is_empty()).collect();
452        match (method, parts.as_slice()) {
453            ("POST", ["codes"]) => {
454                #[derive(Deserialize, Default)]
455                struct Body {
456                    ttl_ms: Option<u64>,
457                }
458                let ttl = serde_json::from_slice::<Body>(body).unwrap_or_default().ttl_ms.unwrap_or(DEFAULT_CODE_TTL_MS);
459                match self.registry.lock().expect("registry lock").issue_code(unix_ms(), ttl.min(24 * 3600 * 1000), "operator") {
460                    Ok(code) => HttpReply::json(200, &serde_json::json!({ "code": code, "ttl_ms": ttl })),
461                    Err(_) => HttpReply::error(503, "storage failed"),
462                }
463            }
464            ("GET", ["nodes"]) => {
465                let nodes = self.registry.lock().expect("registry lock").list();
466                let policy = self.policy.read().expect("policy lock");
467                let nodes: Vec<serde_json::Value> = nodes
468                    .into_iter()
469                    .map(|n| {
470                        let grants = policy.grants(n.status);
471                        let mut value = serde_json::to_value(&n).unwrap_or_default();
472                        value["grants"] = serde_json::to_value(grants).unwrap_or_default();
473                        value
474                    })
475                    .collect();
476                HttpReply::json(200, &serde_json::json!({ "nodes": nodes }))
477            }
478            ("GET", ["policy"]) => HttpReply::json(200, &*self.policy.read().expect("policy lock")),
479            ("POST", ["policy", "reload"]) => match self.reload_policy() {
480                Ok(()) => HttpReply::json(200, &*self.policy.read().expect("policy lock")),
481                Err(message) => HttpReply::json(409, &serde_json::json!({ "error": message })),
482            },
483            ("POST", ["nodes", id, "allows"]) => {
484                #[derive(Deserialize)]
485                struct Body {
486                    capability: String,
487                }
488                match serde_json::from_slice::<Body>(body) {
489                    Ok(b) => match self.allows(id, &b.capability) {
490                        Some(allowed) => HttpReply::json(200, &serde_json::json!({ "allowed": allowed })),
491                        None => HttpReply::error(404, "unknown node"),
492                    },
493                    Err(_) => HttpReply::error(400, "body must be {\"capability\": ...}"),
494                }
495            }
496            ("POST", ["nodes", id, "status"]) => {
497                #[derive(Deserialize)]
498                struct Body {
499                    status: NodeStatus,
500                }
501                match serde_json::from_slice::<Body>(body) {
502                    Ok(b) => self.set_status(id, b.status),
503                    Err(_) => HttpReply::error(400, "body must be {\"status\": neutral|member|trusted|suspended|revoked}"),
504                }
505            }
506            ("POST", ["nodes", id, "revoke"]) => self.set_status(id, NodeStatus::Revoked),
507            _ => HttpReply::error(404, "not found"),
508        }
509    }
510
511    fn set_status(&self, id: &str, status: NodeStatus) -> HttpReply {
512        let result = self.registry.lock().expect("registry lock").set_status(id, status, "operator", unix_ms());
513        match result {
514            Ok(record) => {
515                self.apply_peer(&record);
516                let _ = self.events.send(EnrollEvent::StatusChanged(record.node_id.clone(), record.status));
517                tracing::info!(node_id = %record.node_id, status = ?record.status, "node status changed");
518                HttpReply::json(200, &record)
519            }
520            Err(EnrollmentError::Unknown) => HttpReply::error(404, "unknown node"),
521            Err(EnrollmentError::Revoked) => HttpReply::error(409, "node is revoked"),
522            Err(_) => HttpReply::error(503, "storage failed"),
523        }
524    }
525
526    /// Is this node enrolled and allowed to connect (neutral, member, trusted)?
527    pub fn callhome_allowed(&self, node_id: &str) -> bool {
528        self.allows(node_id, "heartbeat").unwrap_or(false)
529    }
530
531    /// Policy decision for an enrolled node: its current status must hold `capability`.
532    pub fn allows(&self, node_id: &str, capability: &str) -> Option<bool> {
533        let status = self.registry.lock().expect("registry lock").get(node_id).map(|r| r.status)?;
534        Some(self.policy.read().expect("policy lock").allows(status, capability))
535    }
536
537    /// Re-read the policy file. On any error the previous policy stays in force.
538    pub fn reload_policy(&self) -> Result<(), String> {
539        let path = self.policy_path.as_ref().ok_or_else(|| "no policy file is configured".to_owned())?;
540        let next = Policy::load(path).map_err(|e| e.to_string())?;
541        *self.policy.write().expect("policy lock") = next;
542        let _ = self.events.send(EnrollEvent::PolicyChanged);
543        tracing::info!("policy reloaded");
544        Ok(())
545    }
546
547    /// Enrolled nodes whose current status no longer holds `capability`.
548    pub fn nodes_without(&self, capability: &str) -> Vec<String> {
549        let policy = self.policy.read().expect("policy lock");
550        self.registry
551            .lock()
552            .expect("registry lock")
553            .list()
554            .into_iter()
555            .filter(|r| !policy.allows(r.status, capability))
556            .map(|r| r.node_id)
557            .collect()
558    }
559
560    /// Run the signed call-home exchange on a socket whose announce named `node_id`.
561    /// Returns the per-connection session token on success.
562    pub async fn signed_call_home(&self, stream: &mut TcpStream, node_id: &str) -> Result<String, String> {
563        if !self.callhome_allowed(node_id) {
564            return Err("node is not enrolled or not allowed".to_owned());
565        }
566        let challenge = CallHomeChallenge::new().map_err(|e| e.to_string())?;
567        timeout(IO_DEADLINE, stream.write_all(&challenge.frame())).await.map_err(|_| "challenge write timed out")?.map_err(|e| e.to_string())?;
568        let mut answer = [0_u8; ANSWER_LEN];
569        timeout(IO_DEADLINE, stream.read_exact(&mut answer)).await.map_err(|_| "answer timed out")?.map_err(|e| e.to_string())?;
570        let verified = challenge.accept_answer(&answer);
571        let status = self
572            .registry
573            .lock()
574            .expect("registry lock")
575            .verify_node_signature(node_id, &verified.material, &verified.signature_b64)
576            .map_err(|e| e.to_string())?;
577        if !is_active(status) || !self.policy.read().expect("policy lock").allows(status, "heartbeat") {
578            return Err("node status does not permit calling home".to_owned());
579        }
580        Ok(verified.session_token)
581    }
582}
583
584/// Public enroll listener: only `POST /v1/enroll`.
585pub async fn serve_enroll(listener: TcpListener, service: Arc<EnrollService>, mut shutdown: watch::Receiver<bool>) -> io::Result<()> {
586    tracing::info!(listen = %service.listen, "enrollment listener bound");
587    let permits = Arc::new(tokio::sync::Semaphore::new(64));
588    loop {
589        let (stream, peer) = tokio::select! {
590            accepted = listener.accept() => accepted?,
591            changed = shutdown.changed() => {
592                if changed.is_err() || *shutdown.borrow() { return Ok(()); }
593                continue;
594            }
595        };
596        let Ok(permit) = Arc::clone(&permits).try_acquire_owned() else { continue };
597        let service = Arc::clone(&service);
598        tokio::spawn(async move {
599            let _permit = permit;
600            let _ = serve_one(stream, peer.ip(), &service).await;
601        });
602    }
603}
604
605async fn serve_one(mut stream: TcpStream, peer: IpAddr, service: &EnrollService) -> io::Result<()> {
606    let request = timeout(IO_DEADLINE * 2, read_http(&mut stream)).await;
607    let reply = match request {
608        Ok(Ok((method, path, body))) if method == "POST" && path == "/v1/enroll" => service.handle_enroll(&body, peer),
609        Ok(Ok((method, path, body))) if method == "POST" && path == "/v1/build" => {
610            timeout(BUILD_DEADLINE + IO_DEADLINE, service.handle_build(&body, peer))
611                .await
612                .unwrap_or_else(|_| HttpReply::error(504, "build timed out"))
613        }
614        Ok(Ok((method, path, body))) if method == "POST" && path == "/v1/node-call" => {
615            timeout(Duration::from_secs(45), service.handle_node_call(&body, peer))
616                .await
617                .unwrap_or_else(|_| HttpReply::error(504, "timed out"))
618        }
619        Ok(Ok(_)) => HttpReply::error(404, "not found"),
620        _ => HttpReply::error(400, "bad request"),
621    };
622    let reason = match reply.status {
623        200 => "OK",
624        400 => "Bad Request",
625        403 => "Forbidden",
626        404 => "Not Found",
627        429 => "Too Many Requests",
628        422 => "Unprocessable Entity",
629        502 => "Bad Gateway",
630        504 => "Gateway Timeout",
631        _ => "Service Unavailable",
632    };
633    let head = format!(
634        "HTTP/1.1 {} {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
635        reply.status,
636        reason,
637        reply.body.len()
638    );
639    stream.write_all(head.as_bytes()).await?;
640    stream.write_all(&reply.body).await?;
641    stream.shutdown().await
642}
643
644/// Minimal HTTP/1.1 request reader: (method, path, body). Body limited to [`MAX_BODY`].
645pub async fn read_http(stream: &mut TcpStream) -> io::Result<(String, String, Vec<u8>)> {
646    let mut bytes = Vec::with_capacity(2048);
647    let mut chunk = [0_u8; 2048];
648    let header_end = loop {
649        let n = stream.read(&mut chunk).await?;
650        if n == 0 {
651            return Err(io::ErrorKind::UnexpectedEof.into());
652        }
653        bytes.extend_from_slice(&chunk[..n]);
654        if bytes.len() > MAX_BODY + 8192 {
655            return Err(io::Error::new(io::ErrorKind::InvalidData, "too large"));
656        }
657        if let Some(pos) = bytes.windows(4).position(|w| w == b"\r\n\r\n") {
658            break pos + 4;
659        }
660    };
661    let head = std::str::from_utf8(&bytes[..header_end]).map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "utf8"))?;
662    let mut lines = head.split("\r\n");
663    let mut first = lines.next().unwrap_or_default().split_whitespace();
664    let method = first.next().unwrap_or_default().to_owned();
665    let path = first.next().unwrap_or_default().to_owned();
666    let mut length = 0_usize;
667    for line in lines {
668        if let Some((name, value)) = line.split_once(':') {
669            if name.eq_ignore_ascii_case("content-length") {
670                length = value.trim().parse().map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "length"))?;
671            }
672        }
673    }
674    if length > MAX_BODY {
675        return Err(io::Error::new(io::ErrorKind::InvalidData, "body too large"));
676    }
677    let mut body = bytes[header_end..].to_vec();
678    while body.len() < length {
679        let n = stream.read(&mut chunk).await?;
680        if n == 0 {
681            return Err(io::ErrorKind::UnexpectedEof.into());
682        }
683        body.extend_from_slice(&chunk[..n]);
684    }
685    body.truncate(length);
686    Ok((method, path, body))
687}
688
689impl CapabilityGate for EnrollService {
690    fn allows(&self, node_id: &str, capability: &str) -> Option<bool> {
691        EnrollService::allows(self, node_id, capability)
692    }
693}
694
695#[cfg(test)]
696mod tls_only_tests {
697    use super::*;
698    use hatchery_enroll::{NodeIdentity, NodeKind};
699
700    fn code(service: &EnrollService) -> String {
701        let reply = service.handle_operator("POST", "/v1/enroll/codes", b"{}");
702        assert_eq!(reply.status, 200);
703        serde_json::from_slice::<serde_json::Value>(&reply.body).unwrap()["code"].as_str().unwrap().to_owned()
704    }
705
706    fn request(dir: &std::path::Path, id: &str, code: &str) -> Vec<u8> {
707        let identity = NodeIdentity::load_or_create(dir).unwrap();
708        serde_json::to_vec(&identity.sign_request(id, NodeKind::Local, Some(code), unix_ms()).unwrap()).unwrap()
709    }
710
711    #[test]
712    fn a_code_sent_in_the_clear_is_refused_and_burned_while_the_proxy_path_works() {
713        let dir = std::env::temp_dir().join(format!("c2-tls-only-{}", std::process::id()));
714        let mut config = EnrollConfig::new("127.0.0.1:0".parse().unwrap());
715        config.tls_only = true;
716        let (service, _events) = EnrollService::start(&config).unwrap();
717        let public: IpAddr = "203.0.113.9".parse().unwrap();
718        let loopback: IpAddr = "127.0.0.1".parse().unwrap();
719
720        let leaked = code(&service);
721        let body = request(&dir.join("a"), "alpha", &leaked);
722        assert_eq!(service.handle_enroll(&body, public).status, 403);
723        // The exposed code no longer works even through the proxy.
724        let body = request(&dir.join("a"), "alpha", &leaked);
725        assert_eq!(service.handle_enroll(&body, loopback).status, 403);
726
727        let fresh = code(&service);
728        let body = request(&dir.join("b"), "beta", &fresh);
729        assert_eq!(service.handle_enroll(&body, loopback).status, 200);
730        let _ = std::fs::remove_dir_all(&dir);
731    }
732
733    #[test]
734    fn without_the_flag_plaintext_still_enrolls() {
735        let dir = std::env::temp_dir().join(format!("c2-tls-off-{}", std::process::id()));
736        let (service, _events) = EnrollService::start(&EnrollConfig::new("127.0.0.1:0".parse().unwrap())).unwrap();
737        let c = code(&service);
738        let body = request(&dir, "gamma", &c);
739        assert_eq!(service.handle_enroll(&body, "203.0.113.9".parse().unwrap()).status, 200);
740        let _ = std::fs::remove_dir_all(&dir);
741    }
742
743    #[tokio::test]
744    async fn node_call_no_longer_pushes_to_core() {
745        use hatchery_enroll::call::SignedCall;
746        let dir = std::env::temp_dir().join(format!("c2-no-push-{}", std::process::id()));
747        let mut config = EnrollConfig::new("127.0.0.1:0".parse().unwrap());
748        // Even if an old launch line still sets upstream, C2 must not dial it.
749        config.server_upstream = Some(("127.0.0.1:9".parse().unwrap(), "token-not-used-here!".into()));
750        let (service, _events) = EnrollService::start(&config).unwrap();
751        let identity = NodeIdentity::load_or_create(&dir).unwrap();
752        let c = code(&service);
753        let body = request(&dir, "delta", &c);
754        assert_eq!(service.handle_enroll(&body, "127.0.0.1".parse().unwrap()).status, 200);
755        service.handle_operator(
756            "POST",
757            "/v1/enroll/nodes/delta/status",
758            br#"{"status":"trusted"}"#,
759        );
760        let ts = unix_ms();
761        let payload = "{}".to_owned();
762        let msg = SignedCall::signing_bytes("delta", "mail.inbox", &payload, ts, "noncea21");
763        let signed = SignedCall {
764            node_id: "delta".into(),
765            op: "mail.inbox".into(),
766            payload,
767            ts_ms: ts,
768            nonce: "noncea21".into(),
769            signature: identity.sign_bytes(&msg),
770        };
771        let reply = service
772            .handle_node_call(&serde_json::to_vec(&signed).unwrap(), "127.0.0.1".parse().unwrap())
773            .await;
774        assert_eq!(reply.status, 410, "push path removed");
775        let _ = std::fs::remove_dir_all(&dir);
776    }
777
778    #[tokio::test]
779    async fn node_config_answers_the_configured_product_url_only_to_enrolled_nodes() {
780        use hatchery_enroll::call::SignedCall;
781        let dir = std::env::temp_dir().join(format!("c2-node-config-{}", std::process::id()));
782        let _ = std::fs::remove_dir_all(&dir);
783        std::fs::create_dir_all(&dir).unwrap();
784        let mut config = EnrollConfig::new("127.0.0.1:0".parse().unwrap());
785        config.public.m4a_product_url = Some("https://example.invalid".into());
786        let (service, _events) = EnrollService::start(&config).unwrap();
787        let identity = NodeIdentity::load_or_create(&dir).unwrap();
788        let sign = |node: &str, nonce: &str| {
789            let ts = unix_ms();
790            let msg = SignedCall::signing_bytes(node, "node.config", "{}", ts, nonce);
791            serde_json::to_vec(&SignedCall {
792                node_id: node.into(),
793                op: "node.config".into(),
794                payload: "{}".into(),
795                ts_ms: ts,
796                nonce: nonce.into(),
797                signature: identity.sign_bytes(&msg),
798            })
799            .unwrap()
800        };
801        assert_eq!(
802            service
803                .handle_node_call(&sign("epsilon", "nc1"), "127.0.0.1".parse().unwrap())
804                .await
805                .status,
806            403
807        );
808        let c = code(&service);
809        assert_eq!(
810            service
811                .handle_enroll(&request(&dir, "epsilon", &c), "127.0.0.1".parse().unwrap())
812                .status,
813            200
814        );
815        let reply = service
816            .handle_node_call(&sign("epsilon", "nc2"), "127.0.0.1".parse().unwrap())
817            .await;
818        assert_eq!(reply.status, 200);
819        let v: serde_json::Value = serde_json::from_slice(&reply.body).unwrap();
820        assert_eq!(v["m4a_product_url"], "https://example.invalid");
821        let _ = std::fs::remove_dir_all(&dir);
822    }
823}
hatchery_c2/enroll_service.rs

hatchery_c2/
enroll_service.rs
