kipuka/ha/
health.rs

1//! Periodic health probing with state machine transitions.
2//!
3//! Implements RHELBU-3536 R4: health state machine
4//! `Healthy -> Degraded -> Unavailable -> Recovering -> Healthy`
5//! with configurable probe intervals and alert generation via audit log.
6
7use std::sync::Arc;
8use std::time::{Duration, Instant};
9
10use serde::{Deserialize, Serialize};
11use tracing::{debug, info, warn};
12
13use super::pool::{CaId, CaPool};
14
15/// Health state of a single CA backend (RHELBU-3536 R4).
16///
17/// Transitions follow the state machine:
18/// ```text
19/// Healthy -> Degraded -> Unavailable -> Recovering -> Healthy
20///                ^                          |
21///                +---- (probe failure) ------+
22/// ```
23#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
24pub enum HealthState {
25    /// CA is responding normally within latency thresholds.
26    Healthy,
27    /// CA is responding but with elevated latency or intermittent errors.
28    Degraded,
29    /// CA is not responding; circuit breaker is open.
30    Unavailable,
31    /// CA was unavailable and is being re-probed after cooldown.
32    Recovering,
33}
34
35impl HealthState {
36    /// Whether this state allows routing requests to the CA.
37    pub fn is_available(&self) -> bool {
38        matches!(self, Self::Healthy | Self::Degraded | Self::Recovering)
39    }
40}
41
42/// Configuration for the health checker.
43#[derive(Debug, Clone)]
44pub struct HealthConfig {
45    /// Interval between probe rounds (default: 30 seconds).
46    pub probe_interval: Duration,
47    /// Timeout for a single probe request.
48    pub probe_timeout: Duration,
49    /// Number of consecutive successes required to transition from
50    /// `Recovering` back to `Healthy`.
51    pub recovery_threshold: u32,
52    /// Latency threshold (ms) above which a CA is considered degraded.
53    pub degraded_latency_ms: u64,
54}
55
56impl Default for HealthConfig {
57    fn default() -> Self {
58        Self {
59            probe_interval: Duration::from_secs(30),
60            probe_timeout: Duration::from_secs(5),
61            recovery_threshold: 2,
62            degraded_latency_ms: 2000,
63        }
64    }
65}
66
67/// Per-CA probe metrics tracked between probe rounds.
68#[derive(Debug, Clone)]
69pub struct ProbeMetrics {
70    /// Timestamp of the last completed probe.
71    pub last_check: Option<Instant>,
72    /// Number of consecutive probe failures.
73    pub consecutive_failures: u32,
74    /// Number of consecutive probe successes (used during recovery).
75    pub consecutive_successes: u32,
76    /// Last observed response latency.
77    pub last_latency: Option<Duration>,
78}
79
80impl ProbeMetrics {
81    fn new() -> Self {
82        Self {
83            last_check: None,
84            consecutive_failures: 0,
85            consecutive_successes: 0,
86            last_latency: None,
87        }
88    }
89}
90
91pub struct LocalProbeContext {
92    pub config: Arc<crate::config::Config>,
93    pub cas: Arc<indexmap::IndexMap<String, Arc<crate::state::CaState>>>,
94    pub hsm: Option<Arc<kipuka_hsm::HsmContext>>,
95}
96
97/// One-shot liveness check for a single locally-hosted CA.
98///
99/// Verifies that the CA's signing key resolves and matches its certificate
100/// (or, for an HSM-backed CA, that the PKCS#11 session is live), and that the
101/// CA certificate is currently within its validity window.
102///
103/// This is the exact assertion the background probe makes for a `local:` CA,
104/// factored out so callers can run it on demand without a running
105/// [`HealthChecker`] — notably the admin health endpoint when the HA
106/// subsystem is disabled, so that its healthy-CA count reflects a real check
107/// rather than an unconditional assumption.
108///
109/// Returns `Ok(())` when the CA is live, or `Err` with a human-readable
110/// reason otherwise.
111pub fn probe_local_ca(
112    ca_cfg: &crate::config::CaConfig,
113    ca_state: &crate::state::CaState,
114    hsm: Option<&Arc<kipuka_hsm::HsmContext>>,
115) -> Result<(), String> {
116    let key = crate::ca::issue::resolve_signing_key_sync(ca_cfg, hsm).map_err(|e| e.to_string())?;
117    let cert = openssl::x509::X509::from_der(&ca_state.cert_der).map_err(|e| e.to_string())?;
118    match key {
119        crate::ca::issue::ResolvedSigningKey::Pem(pem) => {
120            let key = openssl::pkey::PKey::private_key_from_pem(&pem).map_err(|e| e.to_string())?;
121            let public_key = cert.public_key().map_err(|e| e.to_string())?;
122            if !key.public_eq(&public_key) {
123                return Err("CA key does not match its certificate".into());
124            }
125        }
126        crate::ca::issue::ResolvedSigningKey::Hsm { context, .. } => {
127            context.health_check().map_err(|e| e.to_string())?
128        }
129    }
130    let now = openssl::asn1::Asn1Time::days_from_now(0).map_err(|e| e.to_string())?;
131    if cert.not_before() > now || cert.not_after() <= now {
132        return Err("CA certificate is outside its validity period".into());
133    }
134    Ok(())
135}
136
137/// Runs periodic health probes against each CA backend.
138///
139/// The checker is cloneable (behind `Arc`) and designed to run in a
140/// background tokio task managed by [`super::HaManager`].
141#[derive(Clone)]
142pub struct HealthChecker {
143    pool: Arc<CaPool>,
144    config: HealthConfig,
145    local: Option<Arc<LocalProbeContext>>,
146    /// Per-CA probe metrics, keyed by CaId.
147    metrics: Arc<parking_lot::RwLock<std::collections::HashMap<CaId, ProbeMetrics>>>,
148}
149
150impl HealthChecker {
151    /// Create a new health checker for the given pool.
152    pub fn new(pool: Arc<CaPool>, config: HealthConfig) -> Self {
153        let mut metrics = std::collections::HashMap::new();
154        for conn in pool.connections() {
155            metrics.insert(conn.id.clone(), ProbeMetrics::new());
156        }
157
158        Self {
159            pool,
160            config,
161            local: None,
162            metrics: Arc::new(parking_lot::RwLock::new(metrics)),
163        }
164    }
165
166    pub fn with_local(mut self, local: Arc<LocalProbeContext>) -> Self {
167        self.local = Some(local);
168        self
169    }
170
171    /// Configured probe interval.
172    pub fn interval(&self) -> Duration {
173        self.config.probe_interval
174    }
175
176    /// Execute one round of probes against all registered CAs.
177    pub async fn run_probes(&self) {
178        debug!("starting health probe round");
179
180        for conn in self.pool.connections() {
181            let start = Instant::now();
182            let result = self.probe_ca(&conn.id).await;
183            let elapsed = start.elapsed();
184
185            let mut metrics = self.metrics.write();
186            let m = metrics
187                .entry(conn.id.clone())
188                .or_insert_with(ProbeMetrics::new);
189            m.last_check = Some(Instant::now());
190            m.last_latency = Some(elapsed);
191
192            match result {
193                Ok(()) => self.handle_probe_success(&conn.id, elapsed, m),
194                Err(e) => self.handle_probe_failure(&conn.id, e, m),
195            }
196        }
197    }
198
199    /// Probe a single CA backend.
200    ///
201    /// Issues an HTTP GET to the CA's health endpoint to verify it is
202    /// responding.  For Dogtag CAs this hits `/ca/admin/ca/getStatus`;
203    /// for generic HTTP CAs it performs a simple connectivity check
204    /// against the configured endpoint.
205    ///
206    /// The probe respects the configured timeout ([`HealthConfig::probe_timeout`])
207    /// and returns `Err` with a human-readable reason on failure.
208    async fn probe_ca(&self, id: &CaId) -> Result<(), String> {
209        // If the CA is unavailable, check whether cooldown has elapsed.
210        // If cooldown hasn't elapsed yet, don't probe — report as still failed.
211        let current_snapshot = self.pool.status_snapshot();
212        let is_unavailable = current_snapshot
213            .get(id)
214            .map(|s| s.health == HealthState::Unavailable)
215            .unwrap_or(false);
216
217        if is_unavailable && !self.pool.should_reprobe(id) {
218            return Err("circuit breaker open, cooldown not elapsed".to_string());
219        }
220
221        if is_unavailable {
222            debug!(ca = %id, "cooldown elapsed, attempting re-probe");
223        }
224
225        // Find the endpoint URL for this CA.
226        let endpoint = self
227            .pool
228            .connections()
229            .iter()
230            .find(|c| c.id == *id)
231            .map(|c| c.endpoint.clone())
232            .ok_or_else(|| format!("CA {id} not found in pool"))?;
233
234        if endpoint.starts_with("local:") {
235            let context = self.local.as_ref().ok_or("local probe context missing")?;
236            let ca = context.cas.get(&id.0).ok_or("local CA state missing")?;
237            let config = context
238                .config
239                .cas
240                .iter()
241                .find(|ca| ca.id == id.0)
242                .ok_or("local CA configuration missing")?;
243            return probe_local_ca(config, ca, context.hsm.as_ref());
244        }
245
246        // Build the health check URL.
247        // For Dogtag CAs (endpoint contains /ca), use the Dogtag status API.
248        // For generic CAs, try a simple GET against the endpoint root.
249        let health_url = if endpoint.contains("/ca") {
250            // Dogtag CA: use the agent status endpoint
251            format!("{}/admin/ca/getStatus", endpoint.trim_end_matches('/'))
252        } else {
253            // Generic CA: probe the endpoint root
254            endpoint.clone()
255        };
256
257        debug!(ca = %id, url = %health_url, "probing CA health endpoint");
258
259        let client = reqwest::Client::builder()
260            .timeout(self.config.probe_timeout)
261            .redirect(reqwest::redirect::Policy::none())
262            .build()
263            .map_err(|e| format!("HTTP client build failed: {e}"))?;
264
265        let response = client.get(&health_url).send().await.map_err(|e| {
266            if e.is_timeout() {
267                format!(
268                    "health probe timed out after {:?}",
269                    self.config.probe_timeout
270                )
271            } else if e.is_connect() {
272                format!("health probe connection refused: {e}")
273            } else {
274                format!("health probe failed: {e}")
275            }
276        })?;
277
278        let status = response.status();
279        if !status.is_success() {
280            return Err(format!("CA returned server error: HTTP {status}"));
281        }
282
283        // For Dogtag, verify the response indicates the CA subsystem is
284        // running.  A 200 OK from /admin/ca/getStatus with a body
285        // containing "running" confirms the CA is operational.
286        if health_url.contains("getStatus") {
287            let mut response = response;
288            let mut bytes = Vec::new();
289            while let Some(chunk) = response.chunk().await.map_err(|e| e.to_string())? {
290                if bytes.len() + chunk.len() > 65536 {
291                    return Err("health response exceeds 64 KiB".into());
292                }
293                bytes.extend_from_slice(&chunk);
294            }
295            let body = String::from_utf8_lossy(&bytes);
296            if !body.to_lowercase().contains("running") {
297                return Err(format!(
298                    "Dogtag CA reports unhealthy status: {}",
299                    body.chars().take(200).collect::<String>()
300                ));
301            }
302        }
303
304        debug!(ca = %id, status = %status, "CA health probe succeeded");
305        Ok(())
306    }
307
308    /// Handle a successful probe, applying state transitions.
309    fn handle_probe_success(&self, id: &CaId, latency: Duration, metrics: &mut ProbeMetrics) {
310        metrics.consecutive_failures = 0;
311        metrics.consecutive_successes += 1;
312
313        let current_snapshot = self.pool.status_snapshot();
314        let current_health = current_snapshot
315            .get(id)
316            .map(|s| s.health.clone())
317            .unwrap_or(HealthState::Healthy);
318
319        let latency_ms = latency.as_millis() as u64;
320
321        let new_state = match current_health {
322            HealthState::Unavailable => {
323                info!(ca = %id, "CA responding again, entering recovery");
324                metrics.consecutive_successes = 1;
325                HealthState::Recovering
326            }
327            HealthState::Recovering => {
328                if metrics.consecutive_successes >= self.config.recovery_threshold {
329                    info!(ca = %id, "CA recovery confirmed, marking healthy");
330                    HealthState::Healthy
331                } else {
332                    debug!(
333                        ca = %id,
334                        successes = metrics.consecutive_successes,
335                        needed = self.config.recovery_threshold,
336                        "CA still recovering"
337                    );
338                    HealthState::Recovering
339                }
340            }
341            HealthState::Degraded | HealthState::Healthy => {
342                if latency_ms > self.config.degraded_latency_ms {
343                    debug!(ca = %id, latency_ms, "CA responding slowly, marking degraded");
344                    HealthState::Degraded
345                } else {
346                    HealthState::Healthy
347                }
348            }
349        };
350
351        self.pool.set_health(id, new_state);
352        self.pool.record_success(id, latency);
353    }
354
355    /// Handle a failed probe, applying state transitions.
356    fn handle_probe_failure(&self, id: &CaId, error: String, metrics: &mut ProbeMetrics) {
357        metrics.consecutive_failures += 1;
358        metrics.consecutive_successes = 0;
359
360        warn!(
361            ca = %id,
362            failures = metrics.consecutive_failures,
363            error = %error,
364            "health probe failed"
365        );
366
367        self.pool.record_failure(id);
368    }
369
370    /// Snapshot of probe metrics for monitoring.
371    pub fn metrics_snapshot(&self) -> std::collections::HashMap<CaId, ProbeMetrics> {
372        self.metrics.read().clone()
373    }
374}
375
376#[cfg(test)]
377mod probe_local_ca_tests {
378    //! Regression tests for the on-demand local-CA liveness check.
379    //!
380    //! These pin the honest behaviour the admin health endpoint depends on when
381    //! the HA subsystem is disabled: a CA is only counted "healthy" if its
382    //! signing key actually resolves, matches its certificate, and that
383    //! certificate is currently in date. Previously the endpoint assumed every
384    //! configured CA was healthy without checking anything.
385
386    use super::*;
387    use openssl::asn1::Asn1Time;
388    use openssl::bn::{BigNum, MsbOption};
389    use openssl::ec::{EcGroup, EcKey};
390    use openssl::hash::MessageDigest;
391    use openssl::nid::Nid;
392    use openssl::pkey::{PKey, Private};
393    use openssl::x509::{X509, X509NameBuilder};
394    use std::io::Write;
395
396    fn gen_ec_key() -> PKey<Private> {
397        let group = EcGroup::from_curve_name(Nid::X9_62_PRIME256V1).unwrap();
398        let ec = EcKey::generate(&group).unwrap();
399        PKey::from_ec_key(ec).unwrap()
400    }
401
402    /// Self-sign a CA certificate for `key`. `valid_days` sets `notAfter`
403    /// relative to now — pass a negative value to build an expired certificate.
404    fn self_signed(key: &PKey<Private>, valid_days: i64) -> X509 {
405        let mut nb = X509NameBuilder::new().unwrap();
406        nb.append_entry_by_text("CN", "probe-test-ca").unwrap();
407        let name = nb.build();
408
409        let mut b = X509::builder().unwrap();
410        b.set_version(2).unwrap();
411
412        let mut serial = BigNum::new().unwrap();
413        serial.rand(128, MsbOption::MAYBE_ZERO, false).unwrap();
414        b.set_serial_number(&serial.to_asn1_integer().unwrap())
415            .unwrap();
416
417        b.set_subject_name(&name).unwrap();
418        b.set_issuer_name(&name).unwrap();
419        b.set_pubkey(key).unwrap();
420
421        let now = std::time::SystemTime::now()
422            .duration_since(std::time::UNIX_EPOCH)
423            .unwrap()
424            .as_secs() as i64;
425        b.set_not_before(&Asn1Time::from_unix(now - 3600).unwrap())
426            .unwrap();
427        b.set_not_after(&Asn1Time::from_unix(now + valid_days * 86_400).unwrap())
428            .unwrap();
429
430        b.sign(key, MessageDigest::sha256()).unwrap();
431        b.build()
432    }
433
434    fn ca_state(cert_der: Vec<u8>) -> crate::state::CaState {
435        crate::state::CaState {
436            id: "probe-test".into(),
437            key_type: "ec:P-256".into(),
438            cert_der,
439            cert_chain: vec![],
440            hash_algorithm: "sha256".into(),
441            validity_days: 90,
442            crl_url: None,
443            ocsp_url: None,
444            crl_cache: parking_lot::Mutex::new(None),
445            cab_forum_compliant: false,
446        }
447    }
448
449    fn ca_config(key_path: &std::path::Path) -> crate::config::CaConfig {
450        // Build via TOML so the fixture tracks CaConfig's real defaults; the
451        // probe only reads key_file (cert comes from CaState), so cert_file is
452        // a throwaway.
453        let toml = format!(
454            "id = \"probe-test\"\nkey_file = \"{}\"\ncert_file = \"/dev/null\"\n",
455            key_path.display()
456        );
457        toml::from_str(&toml).expect("valid CaConfig toml")
458    }
459
460    fn write_key(key: &PKey<Private>) -> tempfile::NamedTempFile {
461        let mut f = tempfile::NamedTempFile::new().unwrap();
462        f.write_all(&key.private_key_to_pem_pkcs8().unwrap())
463            .unwrap();
464        f.flush().unwrap();
465        f
466    }
467
468    #[test]
469    fn healthy_ca_passes_liveness_check() {
470        let key = gen_ec_key();
471        let cert = self_signed(&key, 365);
472        let key_file = write_key(&key);
473        let cfg = ca_config(key_file.path());
474        let state = ca_state(cert.to_der().unwrap());
475
476        assert!(
477            probe_local_ca(&cfg, &state, None).is_ok(),
478            "a CA whose on-disk key matches its in-date certificate must be reported live"
479        );
480    }
481
482    #[test]
483    fn mismatched_key_and_cert_fails_liveness_check() {
484        // The certificate is bound to key2's public key, but the config points
485        // key_file at key1 — precisely the "healthy" fabrication the old health
486        // endpoint could not detect.
487        let key1 = gen_ec_key();
488        let key2 = gen_ec_key();
489        let cert = self_signed(&key2, 365);
490        let key_file = write_key(&key1);
491        let cfg = ca_config(key_file.path());
492        let state = ca_state(cert.to_der().unwrap());
493
494        let err = probe_local_ca(&cfg, &state, None)
495            .expect_err("a CA whose key does not match its certificate must fail");
496        assert!(err.contains("does not match"), "unexpected reason: {err}");
497    }
498
499    #[test]
500    fn expired_ca_cert_fails_liveness_check() {
501        let key = gen_ec_key();
502        let cert = self_signed(&key, -1); // notAfter one day in the past
503        let key_file = write_key(&key);
504        let cfg = ca_config(key_file.path());
505        let state = ca_state(cert.to_der().unwrap());
506
507        let err = probe_local_ca(&cfg, &state, None)
508            .expect_err("an expired CA certificate must fail the liveness check");
509        assert!(err.contains("validity period"), "unexpected reason: {err}");
510    }
511}
kipuka/ha/health.rs

kipuka/ha/
health.rs
