diff --git a/lynx/agent/src/conflict.rs b/lynx/agent/src/conflict.rs index 11d744e..90bf071 100644 --- a/lynx/agent/src/conflict.rs +++ b/lynx/agent/src/conflict.rs @@ -1,5 +1,5 @@ use crate::state::AppState; -use std::{process::Command, sync::atomic::Ordering, time::Duration}; +use std::{process::Command, time::Duration}; use tokio::time::interval; const CHECK_INTERVAL_SECS: u64 = 300; @@ -71,7 +71,7 @@ async fn check_and_remove(state: &AppState) { ); notify_dashboard(state, software.name, &format!("removal_failed: {e}")).await; record_audit(state, software.name, &format!("removal_failed: {e}")).await; - state.lockdown.store(true, Ordering::SeqCst); + state.set_lockdown(crate::state::LockdownReason::IncompatibleSoftware); return; } } diff --git a/lynx/agent/src/handlers/system.rs b/lynx/agent/src/handlers/system.rs index a6dda5d..ab4af20 100644 --- a/lynx/agent/src/handlers/system.rs +++ b/lynx/agent/src/handlers/system.rs @@ -203,9 +203,7 @@ async fn command_dispatch( // Handled here so WS path can also process it via run_verified_command. "agent.heartbeat_ack" => { *state.last_heartbeat.lock().unwrap() = std::time::Instant::now(); - state - .lockdown - .store(false, std::sync::atomic::Ordering::SeqCst); + state.clear_lockdown_if_heartbeat(); Ok(json!({ "ok": true })) } other => { diff --git a/lynx/agent/src/main.rs b/lynx/agent/src/main.rs index a6240c6..987c5f5 100644 --- a/lynx/agent/src/main.rs +++ b/lynx/agent/src/main.rs @@ -196,6 +196,7 @@ async fn main() -> anyhow::Result<()> { db, config: Arc::new(config), lockdown: lockdown.clone(), + lockdown_reason: Arc::new(std::sync::Mutex::new(None)), nft_checksum: Arc::new(std::sync::Mutex::new(None)), nft_chain_checksums: Arc::new(std::sync::Mutex::new([None, None, None])), nft_last_ruleset: Arc::new(std::sync::Mutex::new(None)), @@ -344,14 +345,15 @@ async fn main() -> anyhow::Result<()> { && !state_db.is_locked_down() { tracing::error!("PostgreSQL unreachable — entering lockdown"); - state_db - .lockdown - .store(true, std::sync::atomic::Ordering::SeqCst); + state_db.set_lockdown(crate::state::LockdownReason::PgUnreachable); } } }); } + // Startup health guard: poll /health for 30s; restore .prev and write CRITICAL if unhealthy. + update::spawn_startup_health_guard(); + // WebSocket client — persistent connection to dashboard tokio::spawn(ws_client::run_ws_client(state.clone())); @@ -372,7 +374,6 @@ async fn main() -> anyhow::Result<()> { // Heartbeat watchdog task let heartbeat_state = state.clone(); - let lockdown_clone = lockdown.clone(); tokio::spawn(async move { let mut ticker = interval(Duration::from_secs(30)); loop { @@ -383,9 +384,9 @@ async fn main() -> anyhow::Result<()> { .unwrap() .elapsed() .as_secs(); - if elapsed > HEARTBEAT_TIMEOUT_SECS && !lockdown_clone.load(Ordering::SeqCst) { + if elapsed > HEARTBEAT_TIMEOUT_SECS && !heartbeat_state.is_locked_down() { tracing::warn!(elapsed_secs = elapsed, "heartbeat lost — entering lockdown"); - lockdown_clone.store(true, Ordering::SeqCst); + heartbeat_state.set_lockdown(crate::state::LockdownReason::Heartbeat); } } }); @@ -450,7 +451,7 @@ async fn heartbeat_handler( *state.last_heartbeat.lock().unwrap() = std::time::Instant::now(); let is_lockdown = state.lockdown.load(Ordering::SeqCst); - state.lockdown.store(false, Ordering::SeqCst); + state.clear_lockdown_if_heartbeat(); let body = serde_json::json!({ "agent_id": state.config.agent_id, diff --git a/lynx/agent/src/nftables/divergence.rs b/lynx/agent/src/nftables/divergence.rs index 92f5326..67956a9 100644 --- a/lynx/agent/src/nftables/divergence.rs +++ b/lynx/agent/src/nftables/divergence.rs @@ -58,9 +58,7 @@ async fn check_once(state: &AppState) { if let Err(e2) = super::apply_emergency() { error!(error = %e2, "emergency ruleset also failed — lockdown"); } - state - .lockdown - .store(true, std::sync::atomic::Ordering::SeqCst); + state.set_lockdown(crate::state::LockdownReason::NftablesFailure); } else { info!("nftables auto-restored successfully"); } diff --git a/lynx/agent/src/state.rs b/lynx/agent/src/state.rs index 9ce0edc..4919174 100644 --- a/lynx/agent/src/state.rs +++ b/lynx/agent/src/state.rs @@ -6,12 +6,25 @@ use std::sync::{ }; use std::time::Instant; +/// Tracks why the agent entered lockdown. +/// Only `Heartbeat` (and `None`) can be cleared by a `heartbeat_ack`. +/// All other reasons require a manual service restart to clear. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LockdownReason { + Heartbeat, + PgUnreachable, + IncompatibleSoftware, + NftablesFailure, +} + #[derive(Clone)] pub struct AppState { pub db: PgPool, pub config: Arc, - /// Set to true when heartbeat is lost — agent enters lockdown + /// Set to true when the agent enters lockdown. pub lockdown: Arc, + /// The reason the agent entered lockdown, if any. + pub lockdown_reason: Arc>>, /// Last known-good nftables checksum after apply(). None = no ruleset applied yet. pub nft_checksum: Arc>>, /// Per-chain checksums captured after each successful apply() — used for divergence attribution. @@ -48,6 +61,26 @@ impl AppState { self.lockdown.load(Ordering::SeqCst) } + /// Enter lockdown with an explicit reason. + pub fn set_lockdown(&self, reason: LockdownReason) { + self.lockdown.store(true, Ordering::SeqCst); + *self.lockdown_reason.lock().unwrap() = Some(reason); + } + + /// Clear lockdown only when the reason is `Heartbeat` or `None`. + /// Reasons such as `PgUnreachable`, `IncompatibleSoftware`, and + /// `NftablesFailure` require a manual service restart to clear. + pub fn clear_lockdown_if_heartbeat(&self) { + let mut guard = self.lockdown_reason.lock().unwrap(); + match *guard { + None | Some(LockdownReason::Heartbeat) => { + self.lockdown.store(false, Ordering::SeqCst); + *guard = None; + } + _ => {} + } + } + /// Returns true if the command is within the 100/min limit, false if it should be rejected. pub fn check_cmd_rate(&self) -> bool { let now = std::time::SystemTime::now() diff --git a/lynx/agent/src/update/mod.rs b/lynx/agent/src/update/mod.rs index 236e3d8..9ebec7f 100644 --- a/lynx/agent/src/update/mod.rs +++ b/lynx/agent/src/update/mod.rs @@ -2,9 +2,12 @@ pub mod fallback; use anyhow::{Context, Result}; use ed25519_dalek::{Signature, Verifier, VerifyingKey}; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; -/// Download new binary, verify Ed25519 signature, atomic swap, then exec into new process. +const AGENT_BINARY: &str = "/etc/lynx/bin/lynx-agent"; +const CRITICAL_FILE: &str = "/etc/lynx/CRITICAL"; + +/// Download new binary, verify Ed25519 signature, backup to .prev, atomic swap, restart via systemd. /// /// The release verify key (`RELEASE_VERIFY_KEY_B64`) is compiled into the binary and is distinct /// from the dashboard command-signing key. The corresponding private key lives only in GitHub @@ -41,24 +44,28 @@ pub async fn perform_update(version: &str, download_url: &str, sig_url: &str) -> tracing::info!(version, bytes = binary_bytes.len(), "signature verified"); - // Write new binary to a temp path beside the current executable - let current_exe = std::env::current_exe().context("resolve current exe")?; - let tmp_path = tmp_path(¤t_exe); + let target = PathBuf::from(AGENT_BINARY); + let prev = PathBuf::from(format!("{AGENT_BINARY}.prev")); + let tmp = PathBuf::from(format!("{AGENT_BINARY}.new")); - std::fs::write(&tmp_path, &binary_bytes).with_context(|| format!("write to {tmp_path:?}"))?; + std::fs::write(&tmp, &binary_bytes).with_context(|| format!("write to {tmp:?}"))?; // Make it executable #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; - let mut perms = std::fs::metadata(&tmp_path)?.permissions(); + let mut perms = std::fs::metadata(&tmp)?.permissions(); perms.set_mode(0o755); - std::fs::set_permissions(&tmp_path, perms)?; + std::fs::set_permissions(&tmp, perms)?; + } + + // Back up current binary to .prev before swap + if target.exists() { + std::fs::copy(&target, &prev).context("backup agent binary to .prev")?; } - // Atomic rename: tmp → current exe path (POSIX atomic on same filesystem) - std::fs::rename(&tmp_path, ¤t_exe) - .with_context(|| format!("rename {tmp_path:?} → {current_exe:?}"))?; + // Atomic rename: tmp → canonical path (POSIX atomic on same filesystem) + std::fs::rename(&tmp, &target).with_context(|| format!("rename {tmp:?} → {target:?}"))?; tracing::info!(version, "binary swapped — restarting via systemd"); @@ -67,6 +74,65 @@ pub async fn perform_update(version: &str, download_url: &str, sig_url: &str) -> std::process::exit(0); } +/// Spawn a background task that monitors agent startup health. +/// +/// Polls `http://127.0.0.1:9090/health` every 2s for 30s. +/// If still unhealthy → attempt `.prev` restore and exit 1 (systemd restarts with old binary). +/// If `.prev` unavailable or restore fails → write `/etc/lynx/CRITICAL` and exit 1. +/// On healthy startup → delete `/etc/lynx/CRITICAL` if present (recovery from prior critical state). +pub fn spawn_startup_health_guard() { + tokio::spawn(async move { + let client = match reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(3)) + .build() + { + Ok(c) => c, + Err(_) => return, + }; + + for _ in 0..15 { + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + if client + .get("http://127.0.0.1:9090/health") // audit-urls: ok — self health check, not a download + .send() + .await + .map(|r| r.status().is_success()) + .unwrap_or(false) + { + // Healthy — clear any leftover CRITICAL file from a previous failed startup. + let _ = std::fs::remove_file(CRITICAL_FILE); + return; + } + } + + // Still unhealthy after 30s — attempt .prev restore. + tracing::error!("startup health check failed — restoring .prev binary"); + let target = PathBuf::from(AGENT_BINARY); + let prev = PathBuf::from(format!("{AGENT_BINARY}.prev")); + + let restore_ok = if prev.exists() { + std::fs::copy(&prev, &target).is_ok() + } else { + false + }; + + let reason = if restore_ok { + "new binary failed health check; restored .prev" + } else { + "new binary failed health check; .prev unavailable — MANUAL RECOVERY REQUIRED" + }; + + let ts = chrono::Utc::now().to_rfc3339(); + let _ = std::fs::write( + CRITICAL_FILE, + format!("timestamp={ts}\ncomponent=lynx-agent\nreason={reason}\n"), + ); + + tracing::error!(reason, "critical state — exiting for systemd restart"); + std::process::exit(1); + }); +} + async fn download_bytes(client: &reqwest::Client, url: &str) -> Result> { let resp = client .get(url) @@ -116,16 +182,6 @@ fn load_verify_key() -> Result<[u8; 32]> { .map_err(|_| anyhow::anyhow!("release verify key must be 32 bytes")) } -fn tmp_path(exe: &Path) -> PathBuf { - let mut p = exe.to_path_buf(); - let name = exe - .file_name() - .and_then(|n| n.to_str()) - .unwrap_or("lynx-agent"); - p.set_file_name(format!("{name}.new")); - p -} - fn validate_github_url(url: &str) -> Result<()> { let allowed = [ "https://github.com/",