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

hatchery_c2/
enroll_service.rs
