huskies: merge 1173 story Identity-aware fleet checks: cryptographic node identity in upgrade and health probes

2026-07-16 13:26:03 +00:00
parent 489c415fd9
commit 0ac68afa4c
14 changed files with 894 additions and 30 deletions
@@ -27,12 +27,69 @@
 //! on the sled restart.

 use crate::service::gateway::config::ProjectEntry;
+use crate::service::gateway::{IdentityCheck, check_identity, probe_identity};
 use std::collections::BTreeMap;
 use std::future::Future;
+use std::path::Path;
 use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 use tokio::sync::{Mutex, RwLock};

+/// Probe `/identity` on `sled_url`, verify it against the `expected_node_id`
+/// recorded for `project`, and capture a first-contact identity.
+///
+/// Fails open (returns `Ok`) when the probe is unreachable or the responder
+/// has no signature (legacy sled) — the existing `/health`-based liveness
+/// check remains the primary reachability gate. Only a confirmed identity
+/// mismatch or an invalid signature is treated as a hard failure, since both
+/// mean a verifiably different — or untrustworthy — container answered.
+async fn verify_sled_identity(
+    project: &str,
+    sled_url: &str,
+    projects_store: &Arc<RwLock<BTreeMap<String, ProjectEntry>>>,
+    config_dir: &Path,
+    client: &reqwest::Client,
+) -> Result<(), String> {
+    let container_name = format!("huskies-{project}");
+    let nonce = crate::node_identity::generate_challenge();
+    let Some(response) = probe_identity(client, sled_url, &nonce).await else {
+        return Ok(());
+    };
+
+    let expected = {
+        let projects = projects_store.read().await;
+        projects
+            .get(project)
+            .and_then(|e| e.expected_node_id.clone())
+    };
+
+    match check_identity(expected.as_deref(), &nonce, &response) {
+        IdentityCheck::Match | IdentityCheck::MissingSignature => Ok(()),
+        IdentityCheck::FirstContact { node_id } => {
+            {
+                let mut projects = projects_store.write().await;
+                if let Some(entry) = projects.get_mut(project) {
+                    entry.expected_node_id = Some(node_id);
+                }
+            }
+            let snapshot = projects_store.read().await.clone();
+            crate::service::gateway::io::save_config(&snapshot, config_dir).await;
+            Ok(())
+        }
+        IdentityCheck::Mismatch { responder_node_id } => Err(format!(
+            "**identity mismatch** for `{container_name}` at `{sled_url}`: expected node_id \
+             `{}`, but the container that answered identified as `{responder_node_id}`. \
+             Refusing to proceed — this may not be the sled you expect.",
+            expected.unwrap_or_default()
+        )),
+        IdentityCheck::InvalidSignature => Err(format!(
+            "**identity verification failed** for `{container_name}` at `{sled_url}`: the \
+             `/identity` response signature did not verify. Refusing to proceed — the \
+             container's identity cannot be trusted."
+        )),
+    }
+}
+
 // ── Serial lock ────────────────────────────────────────────────────────────────

 static UPGRADE_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
@@ -157,6 +214,7 @@ fn resolve_artifact_source(gateway_port: Option<u16>) -> Result<(String, Option<
 pub async fn handle_upgrade_all<F, Fut>(
    projects_store: &Arc<RwLock<BTreeMap<String, ProjectEntry>>>,
    gateway_port: Option<u16>,
+    config_dir: &Path,
    send_phase: F,
 ) -> String
 where
@@ -178,7 +236,7 @@ where

    let mut results: Vec<String> = Vec::with_capacity(names.len());
    for name in &names {
-        let outcome = handle_sled_upgrade(name, projects_store, gateway_port, |msg| {
+        let outcome = handle_sled_upgrade(name, projects_store, gateway_port, config_dir, |msg| {
            send_phase(format!("**{name}** {msg}"))
        })
        .await;
@@ -205,6 +263,7 @@ pub async fn handle_sled_upgrade<F, Fut>(
    project: &str,
    projects_store: &Arc<RwLock<BTreeMap<String, ProjectEntry>>>,
    gateway_port: Option<u16>,
+    config_dir: &Path,
    send_phase: F,
 ) -> String
 where
@@ -236,17 +295,29 @@ where
        Err(e) => return e,
    };

-    run_sled_upgrade(project, &sled_url, &source_url, expected_hash, send_phase).await
+    run_sled_upgrade(
+        project,
+        &sled_url,
+        &source_url,
+        expected_hash,
+        projects_store,
+        config_dir,
+        send_phase,
+    )
+    .await
 }

 /// Run the four-phase upgrade against a sled whose source URL is already
 /// resolved.  Split from [`handle_sled_upgrade`] so tests can drive the wire
 /// behaviour without a published artifact on the host.
+#[allow(clippy::too_many_arguments)]
 async fn run_sled_upgrade<F, Fut>(
    project: &str,
    sled_url: &str,
    source_url: &str,
    expected_hash: Option<String>,
+    projects_store: &Arc<RwLock<BTreeMap<String, ProjectEntry>>>,
+    config_dir: &Path,
    send_phase: F,
 ) -> String
 where
@@ -263,6 +334,13 @@ where
        .build()
        .unwrap_or_default();

+    // ── Verify identity before triggering ────────────────────────────────────
+    if let Err(e) =
+        verify_sled_identity(project, sled_url, projects_store, config_dir, &client).await
+    {
+        return e;
+    }
+
    // ── Phase 1: downloading ─────────────────────────────────────────────────
    send_phase("[1/4] downloading\u{2026}".to_string()).await;

@@ -316,6 +394,13 @@ where
    // ── Phase 4: reconnected ─────────────────────────────────────────────────
    send_phase("[4/4] reconnected to gateway".to_string()).await;

+    // ── Verify identity during convergence ───────────────────────────────────
+    if let Err(e) =
+        verify_sled_identity(project, sled_url, projects_store, config_dir, &client).await
+    {
+        return format!("upgraded and reconnected, but {e}");
+    }
+
    // ── Verify convergence ───────────────────────────────────────────────────
    match fetch_sled_version(&client, sled_url).await {
        Some((version, git_hash)) => match expected_hash {
@@ -446,6 +531,7 @@ mod tests {
                auth_token: None,
                ssh_port: None,
                host_path: None,
+                expected_node_id: None,
            },
        );
        map.insert(
@@ -455,6 +541,7 @@ mod tests {
                auth_token: None,
                ssh_port: None,
                host_path: None,
+                expected_node_id: None,
            },
        );
        let store = Arc::new(RwLock::new(map));
@@ -470,10 +557,17 @@ mod tests {
        let store: Arc<RwLock<BTreeMap<String, ProjectEntry>>> =
            Arc::new(RwLock::new(BTreeMap::new()));
        let phases: std::sync::Mutex<Vec<String>> = std::sync::Mutex::new(vec![]);
-        let result = handle_sled_upgrade("nonexistent", &store, Some(3000), |msg| {
-            phases.lock().unwrap().push(msg);
-            async {}
-        })
+        let config_dir = tempfile::tempdir().unwrap();
+        let result = handle_sled_upgrade(
+            "nonexistent",
+            &store,
+            Some(3000),
+            config_dir.path(),
+            |msg| {
+                phases.lock().unwrap().push(msg);
+                async {}
+            },
+        )
        .await;
        assert!(
            result.contains("not found"),
@@ -496,10 +590,19 @@ mod tests {
                auth_token: None,
                ssh_port: None,
                host_path: None,
+                expected_node_id: None,
            },
        );
        let store = Arc::new(RwLock::new(map));
-        let result = handle_sled_upgrade("myapp", &store, Some(3000), |_msg| async {}).await;
+        let config_dir = tempfile::tempdir().unwrap();
+        let result = handle_sled_upgrade(
+            "myapp",
+            &store,
+            Some(3000),
+            config_dir.path(),
+            |_msg| async {},
+        )
+        .await;
        assert!(
            result.contains("not found"),
            "project with no URL should say not found: {result}"
@@ -518,7 +621,8 @@ mod tests {
    async fn upgrade_all_empty_store_reports_no_projects() {
        let store: Arc<RwLock<BTreeMap<String, ProjectEntry>>> =
            Arc::new(RwLock::new(BTreeMap::new()));
-        let msg = handle_upgrade_all(&store, Some(3000), |_msg| async {}).await;
+        let config_dir = tempfile::tempdir().unwrap();
+        let msg = handle_upgrade_all(&store, Some(3000), config_dir.path(), |_msg| async {}).await;
        assert!(
            msg.contains("No projects"),
            "empty store should say no projects: {msg}"
@@ -528,11 +632,16 @@ mod tests {
    #[tokio::test]
    async fn upgrade_unreachable_sled_reports_failure() {
        let phases: std::sync::Mutex<Vec<String>> = std::sync::Mutex::new(vec![]);
+        let store: Arc<RwLock<BTreeMap<String, ProjectEntry>>> =
+            Arc::new(RwLock::new(BTreeMap::new()));
+        let config_dir = tempfile::tempdir().unwrap();
        let result = run_sled_upgrade(
            "myapp",
            "http://127.0.0.1:1", // port 1 is never listening
            "http://127.0.0.1:1/api/artifacts/huskies-linux-arm64",
            None,
+            &store,
+            config_dir.path(),
            |msg| {
                phases.lock().unwrap().push(msg);
                async {}
@@ -590,4 +699,169 @@ mod tests {
        let ok = wait_for_health(&client, "http://127.0.0.1:1/health", 1).await;
        assert!(!ok, "should return false when health probe never succeeds");
    }
+
+    // ── verify_sled_identity ──────────────────────────────────────────────────
+
+    /// Spawn a one-shot TCP listener that answers a single `GET
+    /// /identity?nonce=<hex>` request with a JSON body signed by `kp` over
+    /// whatever nonce the caller actually sent — mirrors the real
+    /// `/identity` handler without needing a full HTTP server.
+    fn spawn_identity_responder(
+        listener: tokio::net::TcpListener,
+        kp: bft_json_crdt::keypair::Ed25519KeyPair,
+    ) {
+        tokio::spawn(async move {
+            if let Ok((mut stream, _)) = listener.accept().await {
+                use tokio::io::{AsyncReadExt, AsyncWriteExt};
+                let mut buf = [0u8; 4096];
+                let n = stream.read(&mut buf).await.unwrap_or(0);
+                let req = String::from_utf8_lossy(&buf[..n]);
+                let nonce = req
+                    .lines()
+                    .next()
+                    .unwrap_or("")
+                    .split("nonce=")
+                    .nth(1)
+                    .and_then(|s| s.split_whitespace().next())
+                    .unwrap_or("")
+                    .to_string();
+                let pubkey = crate::node_identity::public_key_hex(&kp);
+                let sig = crate::node_identity::sign_challenge(&kp, &nonce);
+                let body = serde_json::json!({
+                    "node_id": pubkey,
+                    "pubkey": pubkey,
+                    "signature": sig,
+                })
+                .to_string();
+                let response = format!(
+                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}",
+                    body.len(),
+                    body
+                );
+                let _ = stream.write_all(response.as_bytes()).await;
+            }
+        });
+    }
+
+    #[tokio::test]
+    async fn verify_sled_identity_first_contact_captures_and_persists() {
+        let kp = bft_json_crdt::keypair::make_keypair();
+        let pubkey = crate::node_identity::public_key_hex(&kp);
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+        spawn_identity_responder(listener, kp);
+
+        let sled_url = format!("http://127.0.0.1:{port}");
+        let mut map = BTreeMap::new();
+        map.insert(
+            "myapp".to_string(),
+            ProjectEntry {
+                url: Some(sled_url.clone()),
+                auth_token: None,
+                ssh_port: None,
+                host_path: None,
+                expected_node_id: None,
+            },
+        );
+        let store = Arc::new(RwLock::new(map));
+        let config_dir = tempfile::tempdir().unwrap();
+        let client = reqwest::Client::new();
+
+        let result =
+            verify_sled_identity("myapp", &sled_url, &store, config_dir.path(), &client).await;
+        assert!(result.is_ok(), "first contact should succeed: {result:?}");
+
+        let captured = store
+            .read()
+            .await
+            .get("myapp")
+            .and_then(|e| e.expected_node_id.clone());
+        assert_eq!(
+            captured,
+            Some(pubkey.clone()),
+            "expected_node_id should be captured on first contact"
+        );
+
+        let toml_content = tokio::fs::read_to_string(config_dir.path().join("projects.toml")).await;
+        assert!(
+            toml_content.unwrap_or_default().contains(&pubkey),
+            "captured node_id should be persisted to projects.toml"
+        );
+    }
+
+    #[tokio::test]
+    async fn verify_sled_identity_match_succeeds() {
+        let kp = bft_json_crdt::keypair::make_keypair();
+        let pubkey = crate::node_identity::public_key_hex(&kp);
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+        spawn_identity_responder(listener, kp);
+
+        let sled_url = format!("http://127.0.0.1:{port}");
+        let mut map = BTreeMap::new();
+        map.insert(
+            "myapp".to_string(),
+            ProjectEntry {
+                url: Some(sled_url.clone()),
+                auth_token: None,
+                ssh_port: None,
+                host_path: None,
+                expected_node_id: Some(pubkey),
+            },
+        );
+        let store = Arc::new(RwLock::new(map));
+        let config_dir = tempfile::tempdir().unwrap();
+        let client = reqwest::Client::new();
+
+        let result =
+            verify_sled_identity("myapp", &sled_url, &store, config_dir.path(), &client).await;
+        assert!(
+            result.is_ok(),
+            "matching identity should not fail the upgrade: {result:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn verify_sled_identity_mismatch_fails_loudly() {
+        let kp = bft_json_crdt::keypair::make_keypair();
+        let responder_pubkey = crate::node_identity::public_key_hex(&kp);
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+        spawn_identity_responder(listener, kp);
+
+        let sled_url = format!("http://127.0.0.1:{port}");
+        let mut map = BTreeMap::new();
+        map.insert(
+            "myapp".to_string(),
+            ProjectEntry {
+                url: Some(sled_url.clone()),
+                auth_token: None,
+                ssh_port: None,
+                host_path: None,
+                expected_node_id: Some("ab".repeat(32)),
+            },
+        );
+        let store = Arc::new(RwLock::new(map));
+        let config_dir = tempfile::tempdir().unwrap();
+        let client = reqwest::Client::new();
+
+        let result =
+            verify_sled_identity("myapp", &sled_url, &store, config_dir.path(), &client).await;
+        let err = result.expect_err("mismatched identity must fail loudly");
+        assert!(
+            err.contains("identity mismatch"),
+            "error should name the failure class: {err}"
+        );
+        assert!(
+            err.contains("huskies-myapp"),
+            "error should name the container: {err}"
+        );
+        assert!(
+            err.contains(&responder_pubkey),
+            "error should name the node_id that actually answered: {err}"
+        );
+    }
 }