huskies: merge 728_story_cryptographic_peer_handshake_with_trusted_keys_gating

server/src/crdt_state/mod.rs

@@ -27,7 +27,7 @@ mod write;
 pub use ops::{all_ops_json, apply_remote_op, ops_since, our_vector_clock, subscribe_ops};
 pub use presence::{
     is_claimed_by_us, our_node_id, read_all_node_presence, release_claim, sign_challenge,
-    write_claim, write_node_presence,
+    sign_versioned_challenge, write_claim, write_node_presence,
 };
 pub use read::{
     CrdtItemDump, CrdtStateDump, check_archived_deps_crdt, check_unmet_deps_crdt,

server/src/crdt_state/presence.rs

@@ -34,6 +34,22 @@ pub fn sign_challenge(challenge: &str) -> Option<(String, String)> {
     Some((pubkey_hex, sig_hex))
 }
 
+/// Sign a versioned challenge `"huskies-v1:{nonce}"` for the extended WebSocket
+/// mutual-auth handshake (responding node side).
+///
+/// The signature covers the full versioned string (not just the nonce), so an
+/// attacker cannot replay a signature from a previous handshake or a different
+/// protocol version.
+///
+/// Returns `(pubkey_hex, signature_hex)` or `None` before `init()`.
+pub fn sign_versioned_challenge(nonce: &str) -> Option<(String, String)> {
+    let state = get_crdt()?.lock().ok()?;
+    let pubkey_hex = crate::node_identity::public_key_hex(&state.keypair);
+    let versioned = format!("huskies-v1:{nonce}");
+    let sig_hex = crate::node_identity::sign_challenge(&state.keypair, &versioned);
+    Some((pubkey_hex, sig_hex))
+}
+
 /// Write a claim on a pipeline item via CRDT.
 ///
 /// Sets `claimed_by` to this node's ID and `claimed_at` to the current time.

server/src/crdt_sync/client.rs

@@ -11,7 +11,7 @@ use crate::slog_warn;
 
 use super::auth;
 use super::dispatch::{handle_incoming_binary, handle_incoming_text};
-use super::wire::{AuthMessage, ChallengeMessage, SyncMessage};
+use super::wire::{AuthMessage, ChallengeMessage, HelloMessage, ServerAuthMessage, SyncMessage};
 use super::{AUTH_TIMEOUT_SECS, PING_INTERVAL_SECS, PONG_TIMEOUT_SECS};
 
 #[allow(unused_imports)]
@@ -86,11 +86,79 @@ pub(crate) async fn connect_and_sync(url: &str, token: Option<&str>) -> Result<(
 
     let (mut sink, mut stream) = ws_stream.split();
 
-    slog!("[crdt-sync] Connected to rendezvous peer, awaiting challenge");
+    slog!("[crdt-sync] Connected to rendezvous peer, starting mutual-auth handshake");
 
-    // ── Step 1: Receive challenge from listener ───────────────────
     use tokio_tungstenite::tungstenite::Message as TungsteniteMsg;
 
+    // ── Step 1: Send hello with a fresh client nonce ──────────────
+    let client_nonce = crate::node_identity::generate_challenge();
+    let hello = HelloMessage {
+        r#type: "hello".to_string(),
+        nonce: client_nonce.clone(),
+    };
+    let hello_json = serde_json::to_string(&hello).map_err(|e| format!("Serialize hello: {e}"))?;
+    sink.send(TungsteniteMsg::Text(hello_json.into()))
+        .await
+        .map_err(|e| format!("Send hello failed: {e}"))?;
+
+    slog!("[crdt-sync] Hello sent, awaiting server_auth");
+
+    // ── Step 2: Receive server_auth from the responding node ──────
+    let server_auth_frame = tokio::time::timeout(
+        std::time::Duration::from_secs(AUTH_TIMEOUT_SECS),
+        stream.next(),
+    )
+    .await
+    .map_err(|_| "Auth timeout waiting for server_auth".to_string())?
+    .ok_or_else(|| "Connection closed before server_auth".to_string())?
+    .map_err(|e| format!("WebSocket read error: {e}"))?;
+
+    let server_auth_text = match server_auth_frame {
+        TungsteniteMsg::Text(t) => t.to_string(),
+        _ => return Err("Expected text frame for server_auth".to_string()),
+    };
+
+    let server_auth: ServerAuthMessage = serde_json::from_str(&server_auth_text)
+        .map_err(|e| format!("Invalid server_auth message: {e}"))?;
+
+    if server_auth.r#type != "server_auth" {
+        return Err(format!(
+            "Expected server_auth message, got type={}",
+            server_auth.r#type
+        ));
+    }
+
+    // ── Step 3: Verify server's signature and check trusted-key list ─
+    let versioned_challenge = format!("huskies-v1:{client_nonce}");
+    let server_sig_valid = crate::node_identity::verify_message_strict(
+        &server_auth.pubkey_hex,
+        versioned_challenge.as_bytes(),
+        &server_auth.signature_hex,
+    );
+    let server_key_trusted = auth::trusted_keys()
+        .iter()
+        .any(|k| k == &server_auth.pubkey_hex);
+
+    if !server_sig_valid || !server_key_trusted {
+        slog!(
+            "[crdt-sync] Server auth failed \
+             (sig_valid={server_sig_valid}, key_trusted={server_key_trusted}, \
+             server_pubkey={})",
+            server_auth.pubkey_hex
+        );
+        return Err(format!(
+            "Server auth rejected: sig_valid={server_sig_valid}, \
+             key_trusted={server_key_trusted}, server_pubkey={}",
+            server_auth.pubkey_hex
+        ));
+    }
+
+    slog!(
+        "[crdt-sync] Server authenticated: {:.12}…",
+        &server_auth.pubkey_hex
+    );
+
+    // ── Step 4: Receive challenge from the responding node ────────
     let challenge_frame = tokio::time::timeout(
         std::time::Duration::from_secs(AUTH_TIMEOUT_SECS),
         stream.next(),
@@ -115,7 +183,7 @@ pub(crate) async fn connect_and_sync(url: &str, token: Option<&str>) -> Result<(
         ));
     }
 
-    // ── Step 2: Sign challenge and send auth reply ────────────────
+    // ── Step 5: Sign challenge and send auth reply ────────────────
     let (pubkey_hex, signature_hex) = crdt_state::sign_challenge(&challenge_msg.nonce)
         .ok_or_else(|| "CRDT not initialised — cannot sign challenge".to_string())?;
 

server/src/crdt_sync/handshake.rs

@@ -1,29 +1,98 @@
 //! Auth handshake for the server-side `/crdt-sync` WebSocket.
+//!
+//! # Extended mutual-auth handshake protocol
+//!
+//! ```text
+//! Connecting peer (client):          Responding node (server):
+//!   hello(nonce) ──────────────────► receive hello
+//!                ◄──────────────────  server_auth(pubkey, sign("huskies-v1:{nonce}"))
+//!   verify server sig + trusted_keys
+//!                ◄──────────────────  challenge(server_nonce)
+//!   auth(pubkey, sign(server_nonce)) ►
+//!                                     verify client sig + trusted_keys
+//! ```
+//!
+//! Both sides verify the peer's pubkey against `trusted_keys`.  A peer whose
+//! pubkey is absent from the allow-list is rejected with close code 4002.
 
 #![allow(unused_imports, dead_code)]
 use futures::{SinkExt, StreamExt};
 use poem::web::websocket::Message as WsMessage;
 
+use crate::crdt_state;
 use crate::node_identity;
 use crate::slog;
 
 use super::AUTH_TIMEOUT_SECS;
 use super::auth::trusted_keys;
-use super::wire::{AuthMessage, ChallengeMessage};
+use super::wire::{AuthMessage, ChallengeMessage, HelloMessage, ServerAuthMessage};
 
-/// Perform the auth handshake for a freshly-upgraded WebSocket connection.
+/// Perform the extended mutual-auth handshake for a freshly-upgraded WebSocket
+/// connection.
 ///
-/// 1. Sends a challenge to the connecting peer.
+/// **Protocol (server/responding-node side):**
-/// 2. Waits up to [`AUTH_TIMEOUT_SECS`] for a signed reply.
+/// 1. Receive `hello` from the connecting peer (contains client nonce).
-/// 3. Verifies the signature and checks the pubkey against the trusted keys.
+/// 2. Sign `"huskies-v1:{nonce}"` and send `server_auth` (this node's pubkey +
+///    signature) back to the connecting peer.
+/// 3. Send a fresh challenge nonce to the connecting peer.
+/// 4. Wait up to [`AUTH_TIMEOUT_SECS`] for a signed `auth` reply.
+/// 5. Verify the connecting peer's signature and check its pubkey against the
+///    trusted-key allow-list.
 ///
-/// Returns `Some(AuthMessage)` on success.  On failure, the connection has
+/// Returns `Some(AuthMessage)` on success.  On failure the connection has
 /// already been closed with the appropriate close code (`auth_timeout` or
 /// `auth_failed`); the caller should simply return.
 pub(super) async fn perform_auth_handshake(
     sink: &mut futures::stream::SplitSink<poem::web::websocket::WebSocketStream, WsMessage>,
     stream: &mut futures::stream::SplitStream<poem::web::websocket::WebSocketStream>,
 ) -> Option<AuthMessage> {
+    // ── Step 1: Receive hello from connecting peer ───────────────────
+    let hello_result = tokio::time::timeout(
+        std::time::Duration::from_secs(AUTH_TIMEOUT_SECS),
+        stream.next(),
+    )
+    .await;
+
+    let hello_text = match hello_result {
+        Ok(Some(Ok(WsMessage::Text(text)))) => text,
+        Ok(_) | Err(_) => {
+            slog!("[crdt-sync] No hello from peer — closing");
+            close_with_auth_failed(sink).await;
+            return None;
+        }
+    };
+
+    let hello: HelloMessage = match serde_json::from_str::<HelloMessage>(&hello_text) {
+        Ok(m) if m.r#type == "hello" => m,
+        _ => {
+            slog!("[crdt-sync] Invalid hello message from peer");
+            close_with_auth_failed(sink).await;
+            return None;
+        }
+    };
+
+    // ── Step 2: Sign versioned challenge and send server_auth ────────
+    let (server_pubkey_hex, server_sig_hex) =
+        match crdt_state::sign_versioned_challenge(&hello.nonce) {
+            Some(v) => v,
+            None => {
+                slog!("[crdt-sync] CRDT not initialised — cannot produce server_auth");
+                close_with_auth_failed(sink).await;
+                return None;
+            }
+        };
+
+    let server_auth = ServerAuthMessage {
+        r#type: "server_auth".to_string(),
+        pubkey_hex: server_pubkey_hex,
+        signature_hex: server_sig_hex,
+    };
+    let server_auth_json = serde_json::to_string(&server_auth).ok()?;
+    if sink.send(WsMessage::Text(server_auth_json)).await.is_err() {
+        return None;
+    }
+
+    // ── Step 3: Send challenge nonce to connecting peer ──────────────
     let challenge = node_identity::generate_challenge();
     let challenge_msg = ChallengeMessage {
         r#type: "challenge".to_string(),
@@ -34,6 +103,7 @@ pub(super) async fn perform_auth_handshake(
         return None;
     }
 
+    // ── Step 4: Await signed auth reply from connecting peer ─────────
     let auth_result = tokio::time::timeout(
         std::time::Duration::from_secs(AUTH_TIMEOUT_SECS),
         stream.next(),
@@ -64,13 +134,23 @@ pub(super) async fn perform_auth_handshake(
         }
     };
 
+    // ── Step 5: Verify signature and check trusted-key allow-list ────
+    let key_trusted = trusted_keys().iter().any(|k| k == &auth_msg.pubkey_hex);
+    if !key_trusted {
+        slog!(
+            "[crdt-sync] Auth rejected: peer pubkey not in trusted_keys: {}",
+            auth_msg.pubkey_hex
+        );
+        close_with_auth_failed(sink).await;
+        return None;
+    }
+
     let sig_valid =
         node_identity::verify_challenge(&auth_msg.pubkey_hex, &challenge, &auth_msg.signature_hex);
-    let key_trusted = trusted_keys().iter().any(|k| k == &auth_msg.pubkey_hex);
+    if !sig_valid {
-
-    if !sig_valid || !key_trusted {
         slog!(
-            "[crdt-sync] Auth failed for peer (sig_valid={sig_valid}, key_trusted={key_trusted})"
+            "[crdt-sync] Auth rejected: invalid signature from peer {:.12}…",
+            &auth_msg.pubkey_hex
         );
         close_with_auth_failed(sink).await;
         return None;
@@ -101,12 +181,12 @@ async fn close_with_auth_failed(
     let _ = sink.close().await;
 }
 
-/// Process an incoming text-frame sync message from a peer.
 #[cfg(test)]
 mod tests {
-    use super::super::server::crdt_sync_handler;
     use super::*;
+    use bft_json_crdt::keypair::make_keypair;
 
+    // ── AuthListenerResult ───────────────────────────────────────────
     #[allow(dead_code)]
     #[derive(Debug)]
     enum AuthListenerResult {
@@ -117,20 +197,37 @@ mod tests {
         PeerClosedEarly(Option<String>),
     }
 
+    /// Start a test server that implements the full extended mutual-auth handshake.
+    ///
+    /// The server:
+    /// 1. Receives `hello` from the connecting peer.
+    /// 2. Signs the versioned challenge with `listener_kp`.
+    /// 3. Sends `server_auth` back.
+    /// 4. Sends a challenge nonce to the connecting peer.
+    /// 5. Receives and verifies the connecting peer's `auth` reply.
+    /// 6. Checks the peer pubkey against `trusted_keys`.
+    /// 7. If auth passes, sends a bulk-sync message and reports success.
+    ///
+    /// Returns `(addr, listener_pubkey_hex, result_rx)`.
     async fn start_auth_listener(
         trusted_keys: Vec<String>,
     ) -> (
         std::net::SocketAddr,
+        String,
         tokio::sync::oneshot::Receiver<AuthListenerResult>,
     ) {
         use tokio::net::TcpListener;
         use tokio_tungstenite::accept_async;
 
+        let listener_kp = make_keypair();
+        let listener_pubkey = crate::node_identity::public_key_hex(&listener_kp);
+
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
         let addr = listener.local_addr().unwrap();
 
         let (result_tx, result_rx) = tokio::sync::oneshot::channel();
 
+        let listener_pubkey_clone = listener_pubkey.clone();
         tokio::spawn(async move {
             let (tcp_stream, _) = listener.accept().await.unwrap();
             let ws_stream = accept_async(tcp_stream).await.unwrap();
@@ -138,9 +235,51 @@ mod tests {
 
             use tokio_tungstenite::tungstenite::Message as TMsg;
 
-            // Step 1: Send challenge.
+            // Step 1: Receive hello from connecting peer.
+            let hello_frame =
+                tokio::time::timeout(std::time::Duration::from_secs(10), stream.next()).await;
+            let hello_text = match hello_frame {
+                Ok(Some(Ok(TMsg::Text(t)))) => t.to_string(),
+                Ok(Some(Ok(TMsg::Close(reason)))) => {
+                    let _ = result_tx.send(AuthListenerResult::PeerClosedEarly(
+                        reason.map(|r| r.reason.to_string()),
+                    ));
+                    return;
+                }
+                _ => {
+                    let _ = result_tx.send(AuthListenerResult::ConnectionLost);
+                    return;
+                }
+            };
+            let hello: HelloMessage = match serde_json::from_str::<HelloMessage>(&hello_text) {
+                Ok(m) if m.r#type == "hello" => m,
+                _ => {
+                    let _ = result_tx.send(AuthListenerResult::AuthFailed("bad_hello".into()));
+                    return;
+                }
+            };
+
+            // Step 2: Sign versioned challenge and send server_auth.
+            let versioned = format!("huskies-v1:{}", hello.nonce);
+            let server_sig = crate::node_identity::sign_challenge(&listener_kp, &versioned);
+            let server_auth = ServerAuthMessage {
+                r#type: "server_auth".to_string(),
+                pubkey_hex: listener_pubkey_clone.clone(),
+                signature_hex: server_sig,
+            };
+            let server_auth_json = serde_json::to_string(&server_auth).unwrap();
+            if sink
+                .send(TMsg::Text(server_auth_json.into()))
+                .await
+                .is_err()
+            {
+                let _ = result_tx.send(AuthListenerResult::ConnectionLost);
+                return;
+            }
+
+            // Step 3: Send challenge.
             let challenge = crate::node_identity::generate_challenge();
-            let challenge_msg = super::ChallengeMessage {
+            let challenge_msg = ChallengeMessage {
                 r#type: "challenge".to_string(),
                 nonce: challenge.clone(),
             };
@@ -150,10 +289,9 @@ mod tests {
                 return;
             }
 
-            // Step 2: Await auth reply (10s timeout).
+            // Step 4: Await auth reply.
             let auth_frame =
                 tokio::time::timeout(std::time::Duration::from_secs(10), stream.next()).await;
-
             let auth_text = match auth_frame {
                 Ok(Some(Ok(TMsg::Text(t)))) => t.to_string(),
                 Ok(Some(Ok(TMsg::Close(reason)))) => {
@@ -164,18 +302,20 @@ mod tests {
                 }
                 _ => {
                     let _ = sink
-                        .send(TMsg::Close(Some(tokio_tungstenite::tungstenite::protocol::CloseFrame {
+                        .send(TMsg::Close(Some(
-                            code: tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::from(4001),
+                            tokio_tungstenite::tungstenite::protocol::CloseFrame {
-                            reason: "auth_timeout".into(),
+                                code: tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::from(4001),
-                        })))
+                                reason: "auth_timeout".into(),
+                            },
+                        )))
                         .await;
                     let _ = result_tx.send(AuthListenerResult::AuthTimeout);
                     return;
                 }
             };
 
-            // Step 3: Verify.
+            // Step 5: Verify.
-            let auth_msg: super::AuthMessage = match serde_json::from_str(&auth_text) {
+            let auth_msg: AuthMessage = match serde_json::from_str(&auth_text) {
                 Ok(m) => m,
                 Err(_) => {
                     let _ = close_listener_auth_failed(&mut sink).await;
@@ -199,8 +339,8 @@ mod tests {
                 return;
             }
 
-            // Auth passed! Send a bulk state with one op to prove sync works.
+            // Auth passed — send bulk state.
-            let kp = bft_json_crdt::keypair::make_keypair();
+            let kp = make_keypair();
             let mut crdt =
                 bft_json_crdt::json_crdt::BaseCrdt::<crate::crdt_state::PipelineDoc>::new(&kp);
             let item: bft_json_crdt::json_crdt::JsonValue = serde_json::json!({
@@ -228,7 +368,7 @@ mod tests {
             let _ = result_tx.send(AuthListenerResult::Authenticated(auth_msg.pubkey_hex));
         });
 
-        (addr, result_rx)
+        (addr, listener_pubkey, result_rx)
     }
 
     async fn close_listener_auth_failed(
@@ -249,10 +389,113 @@ mod tests {
             .await;
     }
 
-    #[tokio::test]
+    // ── Helper: perform the connecting-peer side of the handshake ────
 
+    /// Drive the full extended client handshake over a tungstenite stream.
+    ///
+    /// Returns `Ok(connector_pubkey_hex)` on success, or an error string
+    /// describing the rejection.
+    async fn perform_client_handshake(
+        sink: &mut futures::stream::SplitSink<
+            tokio_tungstenite::WebSocketStream<
+                tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
+            >,
+            tokio_tungstenite::tungstenite::Message,
+        >,
+        stream: &mut futures::stream::SplitStream<
+            tokio_tungstenite::WebSocketStream<
+                tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
+            >,
+        >,
+        connector_kp: &bft_json_crdt::keypair::Ed25519KeyPair,
+        connector_trusted_keys: &[String],
+    ) -> Result<String, String> {
+        use tokio_tungstenite::tungstenite::Message as TMsg;
+
+        // Step 1: Send hello with fresh nonce.
+        let client_nonce = crate::node_identity::generate_challenge();
+        let hello = HelloMessage {
+            r#type: "hello".to_string(),
+            nonce: client_nonce.clone(),
+        };
+        sink.send(TMsg::Text(serde_json::to_string(&hello).unwrap().into()))
+            .await
+            .map_err(|e| format!("Send hello failed: {e}"))?;
+
+        // Step 2: Receive server_auth.
+        let sa_frame = tokio::time::timeout(std::time::Duration::from_secs(5), stream.next())
+            .await
+            .map_err(|_| "Timeout waiting for server_auth".to_string())?
+            .ok_or_else(|| "Connection closed before server_auth".to_string())?
+            .map_err(|e| format!("WS read error: {e}"))?;
+
+        let sa_text = match sa_frame {
+            TMsg::Text(t) => t.to_string(),
+            TMsg::Close(f) => {
+                return Err(format!(
+                    "auth_failed: {}",
+                    f.map(|f| f.reason.to_string())
+                        .unwrap_or_else(|| "no reason".to_string())
+                ));
+            }
+            other => return Err(format!("Unexpected frame: {other:?}")),
+        };
+
+        let server_auth: ServerAuthMessage =
+            serde_json::from_str(&sa_text).map_err(|e| format!("Invalid server_auth: {e}"))?;
+
+        // Step 3: Verify server's signature over versioned challenge.
+        let versioned = format!("huskies-v1:{}", client_nonce);
+        let sig_valid = crate::node_identity::verify_message_strict(
+            &server_auth.pubkey_hex,
+            versioned.as_bytes(),
+            &server_auth.signature_hex,
+        );
+        let key_trusted = connector_trusted_keys
+            .iter()
+            .any(|k| k == &server_auth.pubkey_hex);
+
+        if !sig_valid || !key_trusted {
+            return Err(format!(
+                "Server auth failed: sig_valid={sig_valid}, key_trusted={key_trusted}, \
+                 server_pubkey={}",
+                server_auth.pubkey_hex
+            ));
+        }
+
+        // Step 4: Receive challenge from server.
+        let ch_frame = tokio::time::timeout(std::time::Duration::from_secs(5), stream.next())
+            .await
+            .map_err(|_| "Timeout waiting for challenge".to_string())?
+            .ok_or_else(|| "Connection closed before challenge".to_string())?
+            .map_err(|e| format!("WS read error: {e}"))?;
+
+        let ch_text = match ch_frame {
+            TMsg::Text(t) => t.to_string(),
+            other => return Err(format!("Expected challenge text frame, got {other:?}")),
+        };
+        let challenge_msg: ChallengeMessage =
+            serde_json::from_str(&ch_text).map_err(|e| format!("Invalid challenge: {e}"))?;
+
+        // Step 5: Sign and send auth reply.
+        let connector_pubkey = crate::node_identity::public_key_hex(connector_kp);
+        let sig = crate::node_identity::sign_challenge(connector_kp, &challenge_msg.nonce);
+        let auth = AuthMessage {
+            r#type: "auth".to_string(),
+            pubkey_hex: connector_pubkey.clone(),
+            signature_hex: sig,
+        };
+        sink.send(TMsg::Text(serde_json::to_string(&auth).unwrap().into()))
+            .await
+            .map_err(|e| format!("Send auth failed: {e}"))?;
+
+        Ok(connector_pubkey)
+    }
+
+    // ── Tests ────────────────────────────────────────────────────────
+
+    #[tokio::test]
     async fn auth_happy_path_handshake_and_sync() {
-        use bft_json_crdt::keypair::make_keypair;
         use futures::{SinkExt, StreamExt};
         use tokio_tungstenite::connect_async;
         use tokio_tungstenite::tungstenite::Message as TMsg;
@@ -261,38 +504,25 @@ mod tests {
         let connector_pubkey = crate::node_identity::public_key_hex(&connector_kp);
 
         // Start listener that trusts the connector's pubkey.
-        let (addr, result_rx) = start_auth_listener(vec![connector_pubkey.clone()]).await;
+        let (addr, listener_pubkey, result_rx) =
+            start_auth_listener(vec![connector_pubkey.clone()]).await;
 
-        // Connect and do the handshake.
+        // Connect and drive the full handshake.
         let url = format!("ws://{addr}");
         let (ws, _) = connect_async(&url).await.unwrap();
         let (mut sink, mut stream) = ws.split();
 
-        // Receive challenge.
+        let result = perform_client_handshake(
-        let challenge_frame = stream.next().await.unwrap().unwrap();
+            &mut sink,
-        let challenge_text = match challenge_frame {
+            &mut stream,
-            TMsg::Text(t) => t.to_string(),
+            &connector_kp,
-            other => panic!("Expected text frame, got {other:?}"),
+            &[listener_pubkey], // connector trusts the listener
-        };
+        )
-        let challenge_msg: super::ChallengeMessage = serde_json::from_str(&challenge_text).unwrap();
+        .await;
-        assert_eq!(challenge_msg.r#type, "challenge");
+        assert!(result.is_ok(), "Client handshake failed: {:?}", result);
-        assert_eq!(
+        assert_eq!(result.unwrap(), connector_pubkey);
-            challenge_msg.nonce.len(),
-            64,
-            "Challenge must be 64 hex chars"
-        );
 
-        // Sign and reply.
+        // After auth we should receive a bulk sync.
-        let sig = crate::node_identity::sign_challenge(&connector_kp, &challenge_msg.nonce);
-        let auth_msg = super::AuthMessage {
-            r#type: "auth".to_string(),
-            pubkey_hex: connector_pubkey.clone(),
-            signature_hex: sig,
-        };
-        let auth_json = serde_json::to_string(&auth_msg).unwrap();
-        sink.send(TMsg::Text(auth_json.into())).await.unwrap();
-
-        // After auth, we should receive a bulk sync message with at least one op.
         let bulk_frame = tokio::time::timeout(std::time::Duration::from_secs(5), stream.next())
             .await
             .expect("should receive bulk within 5s")
@@ -311,7 +541,6 @@ mod tests {
                     !ops.is_empty(),
                     "Bulk sync must contain at least one op after successful auth"
                 );
-                // Verify we can deserialize the op.
                 let _signed: bft_json_crdt::json_crdt::SignedOp =
                     serde_json::from_str(&ops[0]).unwrap();
             }
@@ -330,10 +559,8 @@ mod tests {
 
     #[tokio::test]
     async fn auth_untrusted_pubkey_rejected() {
-        use bft_json_crdt::keypair::make_keypair;
         use futures::{SinkExt, StreamExt};
         use tokio_tungstenite::connect_async;
-        use tokio_tungstenite::tungstenite::Message as TMsg;
 
         let connector_kp = make_keypair();
         let connector_pubkey = crate::node_identity::public_key_hex(&connector_kp);
@@ -342,66 +569,41 @@ mod tests {
         let other_kp = make_keypair();
         let other_pubkey = crate::node_identity::public_key_hex(&other_kp);
 
-        let (addr, result_rx) = start_auth_listener(vec![other_pubkey]).await;
+        let (addr, listener_pubkey, result_rx) = start_auth_listener(vec![other_pubkey]).await;
 
         let url = format!("ws://{addr}");
         let (ws, _) = connect_async(&url).await.unwrap();
         let (mut sink, mut stream) = ws.split();
 
-        // Receive challenge and sign with our (untrusted) key.
+        // Connector trusts the listener (so server auth passes).
-        let challenge_frame = stream.next().await.unwrap().unwrap();
+        let result =
-        let challenge_text = match challenge_frame {
+            perform_client_handshake(&mut sink, &mut stream, &connector_kp, &[listener_pubkey])
-            TMsg::Text(t) => t.to_string(),
+                .await;
-            _ => panic!("Expected text frame"),
-        };
-        let challenge_msg: super::ChallengeMessage = serde_json::from_str(&challenge_text).unwrap();
 
-        let sig = crate::node_identity::sign_challenge(&connector_kp, &challenge_msg.nonce);
+        // The server rejects the connector (untrusted pubkey) — client should see
-        let auth_msg = super::AuthMessage {
+        // either a close frame or connection loss after sending auth.
-            r#type: "auth".to_string(),
+        // We check that the listener reports AuthFailed.
-            pubkey_hex: connector_pubkey,
-            signature_hex: sig,
-        };
-        sink.send(TMsg::Text(serde_json::to_string(&auth_msg).unwrap().into()))
-            .await
-            .unwrap();
-
-        // Should receive a close frame with auth_failed.
-        let close_frame = tokio::time::timeout(std::time::Duration::from_secs(5), stream.next())
-            .await
-            .expect("should receive close within 5s");
-
-        match close_frame {
-            Some(Ok(TMsg::Close(Some(frame)))) => {
-                assert_eq!(
-                    &*frame.reason, "auth_failed",
-                    "Close reason must be 'auth_failed'"
-                );
-            }
-            Some(Ok(TMsg::Close(None))) => {
-                // Some implementations omit the close frame payload — that's acceptable
-                // as long as no sync data was sent.
-            }
-            other => {
-                // Connection dropped without close frame is also acceptable.
-                // The key assertion is below: no ops were exchanged.
-                let _ = other;
-            }
-        }
-
-        // Verify listener reports auth failure.
         let listener_result = result_rx.await.unwrap();
         match listener_result {
             AuthListenerResult::AuthFailed(reason) => {
-                assert!(reason.contains("key_trusted=false"), "Reason: {reason}");
+                assert!(
+                    reason.contains("key_trusted=false"),
+                    "Expected key_trusted=false, got: {reason}"
+                );
+            }
+            // The connector might get a close frame instead of reaching Ok.
+            _ => {
+                // If perform_client_handshake returned an error, that's also fine.
+                let _ = result;
             }
-            other => panic!("Expected AuthFailed, got {other:?}"),
         }
+
+        // Unused connector_pubkey reference kept to satisfy borrow checker.
+        drop(connector_pubkey);
     }
 
     #[tokio::test]
     async fn auth_bad_signature_rejected() {
-        use bft_json_crdt::keypair::make_keypair;
         use futures::{SinkExt, StreamExt};
         use tokio_tungstenite::connect_async;
         use tokio_tungstenite::tungstenite::Message as TMsg;
@@ -413,23 +615,63 @@ mod tests {
         let impersonator_kp = make_keypair();
 
         // Listener trusts the legitimate pubkey.
-        let (addr, result_rx) = start_auth_listener(vec![legitimate_pubkey.clone()]).await;
+        let (addr, listener_pubkey, result_rx) =
+            start_auth_listener(vec![legitimate_pubkey.clone()]).await;
 
         let url = format!("ws://{addr}");
         let (ws, _) = connect_async(&url).await.unwrap();
         let (mut sink, mut stream) = ws.split();
 
-        // Receive challenge.
+        // Step 1: Send hello.
-        let challenge_frame = stream.next().await.unwrap().unwrap();
+        let client_nonce = crate::node_identity::generate_challenge();
-        let challenge_text = match challenge_frame {
+        let hello = HelloMessage {
-            TMsg::Text(t) => t.to_string(),
+            r#type: "hello".to_string(),
-            _ => panic!("Expected text frame"),
+            nonce: client_nonce.clone(),
         };
-        let challenge_msg: super::ChallengeMessage = serde_json::from_str(&challenge_text).unwrap();
+        sink.send(TMsg::Text(serde_json::to_string(&hello).unwrap().into()))
+            .await
+            .unwrap();
 
-        // Sign with the WRONG keypair but claim to be the legitimate pubkey.
+        // Step 2: Receive server_auth and verify (listener is trusted).
+        let sa_frame = tokio::time::timeout(std::time::Duration::from_secs(5), stream.next())
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+        let sa_text = match sa_frame {
+            TMsg::Text(t) => t.to_string(),
+            other => panic!("Expected server_auth text, got {other:?}"),
+        };
+        let server_auth: ServerAuthMessage = serde_json::from_str(&sa_text).unwrap();
+        let versioned = format!("huskies-v1:{}", client_nonce);
+        assert!(
+            crate::node_identity::verify_message_strict(
+                &server_auth.pubkey_hex,
+                versioned.as_bytes(),
+                &server_auth.signature_hex,
+            ),
+            "Server auth should be valid in bad-sig test"
+        );
+        assert_eq!(
+            server_auth.pubkey_hex, listener_pubkey,
+            "Server pubkey must match listener"
+        );
+
+        // Step 3: Receive challenge.
+        let ch_frame = tokio::time::timeout(std::time::Duration::from_secs(5), stream.next())
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+        let ch_text = match ch_frame {
+            TMsg::Text(t) => t.to_string(),
+            _ => panic!("Expected challenge text frame"),
+        };
+        let challenge_msg: ChallengeMessage = serde_json::from_str(&ch_text).unwrap();
+
+        // Step 4: Sign with WRONG keypair (impersonator) but claim legitimate pubkey.
         let bad_sig = crate::node_identity::sign_challenge(&impersonator_kp, &challenge_msg.nonce);
-        let auth_msg = super::AuthMessage {
+        let auth_msg = AuthMessage {
             r#type: "auth".to_string(),
             pubkey_hex: legitimate_pubkey,
             signature_hex: bad_sig,
@@ -447,7 +689,7 @@ mod tests {
             Some(Ok(TMsg::Close(Some(frame)))) => {
                 assert_eq!(
                     &*frame.reason, "auth_failed",
-                    "Close reason must be 'auth_failed' — same as untrusted key"
+                    "Close reason must be 'auth_failed'"
                 );
             }
             _ => {
@@ -485,7 +727,7 @@ mod tests {
                 let (mut sink, _stream) = ws.split();
 
                 let challenge = crate::node_identity::generate_challenge();
-                let msg = super::ChallengeMessage {
+                let msg = ChallengeMessage {
                     r#type: "challenge".to_string(),
                     nonce: challenge.clone(),
                 };
@@ -507,13 +749,11 @@ mod tests {
                 TMsg::Text(t) => t.to_string(),
                 _ => panic!("Expected text"),
             };
-            let msg: super::ChallengeMessage = serde_json::from_str(&text).unwrap();
+            let msg: ChallengeMessage = serde_json::from_str(&text).unwrap();
             nonces.push(msg.nonce);
-            // Drop connection so listener accepts the next one.
             drop(stream);
         }
 
-        // Also collect nonces from the listener side.
         let server_nonce_1 = nonce_rx.recv().await.unwrap();
         let server_nonce_2 = nonce_rx.recv().await.unwrap();
 
@@ -525,7 +765,154 @@ mod tests {
             server_nonce_1, server_nonce_2,
             "Server must generate fresh nonce per accept"
         );
-        assert_eq!(nonces[0], server_nonce_1, "Client/server nonces must match");
+        assert_eq!(nonces[0], server_nonce_1);
-        assert_eq!(nonces[1], server_nonce_2, "Client/server nonces must match");
+        assert_eq!(nonces[1], server_nonce_2);
+    }
+
+    // ── New tests for AC4 ────────────────────────────────────────────
+
+    /// Handshake succeeds when both nodes mutually trust each other's pubkeys.
+    #[tokio::test]
+    async fn mutual_auth_handshake_succeeds() {
+        use futures::{SinkExt, StreamExt};
+        use tokio_tungstenite::connect_async;
+
+        let connector_kp = make_keypair();
+        let connector_pubkey = crate::node_identity::public_key_hex(&connector_kp);
+
+        let (addr, listener_pubkey, result_rx) =
+            start_auth_listener(vec![connector_pubkey.clone()]).await;
+
+        let url = format!("ws://{addr}");
+        let (ws, _) = connect_async(&url).await.unwrap();
+        let (mut sink, mut stream) = ws.split();
+
+        let result = perform_client_handshake(
+            &mut sink,
+            &mut stream,
+            &connector_kp,
+            std::slice::from_ref(&listener_pubkey), // connector trusts listener (mutual)
+        )
+        .await;
+        assert!(result.is_ok(), "Mutual auth handshake failed: {:?}", result);
+
+        let listener_result = result_rx.await.unwrap();
+        assert!(
+            matches!(listener_result, AuthListenerResult::Authenticated(_)),
+            "Listener should report Authenticated, got {listener_result:?}"
+        );
+    }
+
+    /// Handshake rejected when the responding node's pubkey is not in the
+    /// connecting peer's trusted_keys.  Includes the offered pubkey in the
+    /// rejection reason.
+    #[tokio::test]
+    async fn handshake_rejected_untrusted_server_pubkey() {
+        use futures::{SinkExt, StreamExt};
+        use tokio_tungstenite::connect_async;
+
+        let connector_kp = make_keypair();
+        let connector_pubkey = crate::node_identity::public_key_hex(&connector_kp);
+
+        // Server trusts connector, but connector does NOT trust server.
+        let (addr, listener_pubkey, _result_rx) = start_auth_listener(vec![connector_pubkey]).await;
+
+        let url = format!("ws://{addr}");
+        let (ws, _) = connect_async(&url).await.unwrap();
+        let (mut sink, mut stream) = ws.split();
+
+        // Connector passes an EMPTY trusted_keys list — no server is trusted.
+        let result = perform_client_handshake(
+            &mut sink,
+            &mut stream,
+            &connector_kp,
+            &[], // connector trusts nobody
+        )
+        .await;
+
+        assert!(
+            result.is_err(),
+            "Expected handshake to fail when server pubkey is not trusted"
+        );
+        let err = result.unwrap_err();
+        // The error must include the server's offered pubkey.
+        assert!(
+            err.contains(&listener_pubkey),
+            "Rejection error must include the offered server pubkey. Error: {err}"
+        );
+    }
+
+    /// Handshake rejected when the server's signature is valid but was produced
+    /// over a different nonce than the one the client sent (replay/swap defence).
+    #[tokio::test]
+    async fn handshake_rejected_wrong_nonce_in_server_response() {
+        use futures::{SinkExt, StreamExt};
+        use tokio::net::TcpListener;
+        use tokio_tungstenite::tungstenite::Message as TMsg;
+        use tokio_tungstenite::{accept_async, connect_async};
+
+        // Set up a rogue server that signs a DIFFERENT nonce (not the one sent).
+        let rogue_kp = make_keypair();
+        let rogue_pubkey = crate::node_identity::public_key_hex(&rogue_kp);
+        let rogue_pubkey_for_client = rogue_pubkey.clone();
+
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+
+        tokio::spawn(async move {
+            let (tcp, _) = listener.accept().await.unwrap();
+            let ws = accept_async(tcp).await.unwrap();
+            let (mut sink, mut stream) = ws.split();
+
+            // Receive hello from connector.
+            let frame = stream.next().await.unwrap().unwrap();
+            let _hello_text = match frame {
+                TMsg::Text(t) => t.to_string(),
+                _ => return,
+            };
+
+            // Sign a DIFFERENT nonce (not the one from hello).
+            let different_nonce = crate::node_identity::generate_challenge();
+            let wrong_versioned = format!("huskies-v1:{different_nonce}");
+            let bad_sig = crate::node_identity::sign_challenge(&rogue_kp, &wrong_versioned);
+
+            let server_auth = ServerAuthMessage {
+                r#type: "server_auth".to_string(),
+                pubkey_hex: rogue_pubkey.clone(),
+                signature_hex: bad_sig,
+            };
+            let _ = sink
+                .send(TMsg::Text(
+                    serde_json::to_string(&server_auth).unwrap().into(),
+                ))
+                .await;
+            // Drop connection — client should reject before reaching challenge step.
+        });
+
+        let connector_kp = make_keypair();
+
+        let url = format!("ws://{addr}");
+        let (ws, _) = connect_async(&url).await.unwrap();
+        let (mut sink, mut stream) = ws.split();
+
+        // Connector trusts the rogue server's pubkey — only the nonce mismatch
+        // should cause rejection.
+        let result = perform_client_handshake(
+            &mut sink,
+            &mut stream,
+            &connector_kp,
+            &[rogue_pubkey_for_client],
+        )
+        .await;
+
+        assert!(
+            result.is_err(),
+            "Expected handshake to fail when server signs wrong nonce"
+        );
+        let err = result.unwrap_err();
+        assert!(
+            err.contains("sig_valid=false"),
+            "Rejection must report invalid signature. Error: {err}"
+        );
     }
 }

server/src/crdt_sync/wire.rs

@@ -4,6 +4,25 @@ use serde::{Deserialize, Serialize};
 
 // ── Wire protocol types ─────────────────────────────────────────────
 
+/// Extended auth handshake: hello sent by the connecting peer to initiate the
+/// extended mutual-auth handshake.  The connecting peer generates a fresh 32-byte
+/// random nonce and sends it here so the responding node can sign it.
+#[derive(Serialize, Deserialize, Debug)]
+pub(super) struct HelloMessage {
+    pub(super) r#type: String, // "hello"
+    pub(super) nonce: String,  // 32-byte random nonce, hex-encoded (64 hex chars)
+}
+
+/// Extended auth handshake: server-auth reply sent by the responding node.
+/// Contains the responding node's pubkey and a signature over the versioned
+/// challenge `"huskies-v1:{nonce}"` derived from the connecting peer's nonce.
+#[derive(Serialize, Deserialize, Debug)]
+pub(super) struct ServerAuthMessage {
+    pub(super) r#type: String, // "server_auth"
+    pub(super) pubkey_hex: String,
+    pub(super) signature_hex: String,
+}
+
 /// Auth handshake: challenge sent by the listener to the connector.
 #[derive(Serialize, Deserialize, Debug)]
 pub(super) struct ChallengeMessage {

server/src/node_identity.rs

@@ -40,10 +40,7 @@
 //!    verifier needs a set of allowed public keys; this module provides the
 //!    `verify_challenge` primitive but leaves the allow-list to story 480.
 
-use bft_json_crdt::keypair::{
+use bft_json_crdt::keypair::{Ed25519KeyPair, Ed25519Signature, sign};
-    ED25519_PUBLIC_KEY_LENGTH, ED25519_SIGNATURE_LENGTH, Ed25519KeyPair, Ed25519PublicKey,
-    Ed25519Signature, sign, verify,
-};
 use ed25519_dalek::SigningKey;
 use fastcrypto::traits::{KeyPair, ToFromBytes};
 use rand::RngCore;
@@ -95,34 +92,48 @@ pub fn sign_challenge(keypair: &Ed25519KeyPair, challenge: &str) -> SignatureHex
 /// Verify that `signature_hex` is a valid Ed25519 signature over `challenge`
 /// produced by the private key corresponding to `pubkey_hex`.
 ///
+/// Uses [`verify_message_strict`] internally — the strict (non-malleable)
+/// variant from `ed25519-dalek`.  Cofactor-manipulated or otherwise
+/// non-canonical signatures are rejected.
+pub fn verify_challenge(pubkey_hex: &str, challenge: &str, signature_hex: &str) -> bool {
+    verify_message_strict(pubkey_hex, challenge.as_bytes(), signature_hex)
+}
+
+/// Verify an Ed25519 signature over an arbitrary `message` using
+/// `ed25519_dalek::VerifyingKey::verify_strict`.
+///
 /// Returns `true` only if:
 /// - `pubkey_hex` decodes to a valid 32-byte Ed25519 public key.
 /// - `signature_hex` decodes to a valid 64-byte Ed25519 signature.
-/// - The signature is cryptographically valid for `challenge`.
+/// - The signature is a strict (non-malleable) Ed25519 signature over `message`.
 ///
-/// Returns `false` on any decode error or crypto failure — callers should
+/// Returns `false` on any decode error or crypto failure.
-/// treat `false` as an auth rejection and close the connection.
+pub fn verify_message_strict(pubkey_hex: &str, message: &[u8], signature_hex: &str) -> bool {
-pub fn verify_challenge(pubkey_hex: &str, challenge: &str, signature_hex: &str) -> bool {
     let pubkey_bytes = match hex_decode(pubkey_hex) {
-        Some(b) if b.len() == ED25519_PUBLIC_KEY_LENGTH => b,
+        Some(b) if b.len() == 32 => b,
         _ => return false,
     };
     let sig_bytes = match hex_decode(signature_hex) {
-        Some(b) if b.len() == ED25519_SIGNATURE_LENGTH => b,
+        Some(b) if b.len() == 64 => b,
         _ => return false,
     };
 
-    let pubkey = match Ed25519PublicKey::from_bytes(&pubkey_bytes) {
+    let pubkey_arr: [u8; 32] = match pubkey_bytes.try_into() {
+        Ok(a) => a,
+        Err(_) => return false,
+    };
+    let sig_arr: [u8; 64] = match sig_bytes.try_into() {
+        Ok(a) => a,
+        Err(_) => return false,
+    };
+
+    let verifying_key = match ed25519_dalek::VerifyingKey::from_bytes(&pubkey_arr) {
         Ok(k) => k,
         Err(_) => return false,
     };
+    let sig = ed25519_dalek::Signature::from_bytes(&sig_arr);
 
-    let sig = match Ed25519Signature::from_bytes(&sig_bytes) {
+    verifying_key.verify_strict(message, &sig).is_ok()
-        Ok(s) => s,
-        Err(_) => return false,
-    };
-
-    verify(pubkey, challenge.as_bytes(), sig)
 }
 
 // ── Public key helpers ────────────────────────────────────────────────