From 4445c46066dede85d31c285132a8625ea9f96108 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Tue, 10 Feb 2026 21:25:26 +0000 Subject: [PATCH] Fix secp256k1 parity in Noise IK, add disconnect protocol, cross-connection handling, timeout cleanup Noise IK parity fix: - Pre-message hash normalizes responder static key to even parity (0x02) so initiator and responder hash chains match regardless of actual parity - ECDH uses shared_secret_point() + SHA-256(x-only) instead of SharedSecret::new() which includes a parity-dependent version byte - Fixes handshake failure for ~50% of keys when initiator has only npub Graceful disconnect protocol (link message 0x50): - DisconnectReason enum with 8 reason codes - Disconnect struct with encode/decode - send_encrypted_link_message() reusable helper - handle_disconnect() with immediate peer removal - send_disconnect_to_all_peers() called during Node::stop() Cross-connection fix in handle_msg1(): - addr_to_link check now distinguishes inbound duplicates (reject) from outbound links (cross-connection, allow and resolve via tie-breaker) - remove_link() only clears addr_to_link if entry maps to same link_id - Link cleanup and addr_to_link restoration in cross-connection branches Handshake timeout cleanup: - RX loop uses tokio::select! with 1-second interval tick - check_timeouts() scans for stale (>30s) and failed connections - cleanup_stale_connection() removes all associated state Tests: 279 passing (4 new: cross-connection, stale cleanup, failed cleanup, odd-parity handshake) --- docs/design/fips-intro.md | 7 +- docs/design/fips-session-protocol.md | 10 + docs/design/fips-state-machines.md | 4 +- docs/design/fips-wire-protocol.md | 53 +++++- src/identity.rs | 30 +++ src/node/handlers.rs | 219 ++++++++++++++++++++-- src/node/lifecycle.rs | 45 ++++- src/node/mod.rs | 77 +++++++- src/node/tests.rs | 266 +++++++++++++++++++++++++++ src/noise.rs | 105 ++++++++++- src/protocol.rs | 198 ++++++++++++++++++++ 11 files changed, 975 insertions(+), 39 deletions(-) diff --git a/docs/design/fips-intro.md b/docs/design/fips-intro.md index 143244a..6d6ac02 100644 --- a/docs/design/fips-intro.md +++ b/docs/design/fips-intro.md @@ -465,10 +465,11 @@ Exchanged between directly connected peers, encrypted with link session keys: | Type | Name | Purpose | |------|----------------|--------------------------------------------| | 0x10 | TreeAnnounce | Spanning tree state (parent, ancestry) | -| 0x11 | FilterAnnounce | Bloom filter reachability update | -| 0x12 | LookupRequest | Query for node's tree coordinates | -| 0x13 | LookupResponse | Response with coordinates and proof | +| 0x20 | FilterAnnounce | Bloom filter reachability update | +| 0x30 | LookupRequest | Query for node's tree coordinates | +| 0x31 | LookupResponse | Response with coordinates and proof | | 0x40 | SessionDatagram| Carries end-to-end encrypted payloads | +| 0x50 | Disconnect | Orderly disconnect notification | ### Session Layer Messages diff --git a/docs/design/fips-session-protocol.md b/docs/design/fips-session-protocol.md index 938e839..16d7871 100644 --- a/docs/design/fips-session-protocol.md +++ b/docs/design/fips-session-protocol.md @@ -411,6 +411,16 @@ Secp256k1 and SHA-256 are already used for Nostr identities, and ChaCha20-Poly1305 matches NIP-44 encryption. Lightning's BOLT 8 provides a proven reference for adapting Noise Protocol to secp256k1. +**secp256k1 parity normalization**: Nostr npubs encode x-only public keys +(32 bytes, no y-coordinate parity). The Noise IK pre-message mixes the +responder's static key as a 33-byte compressed key, and the default +secp256k1 ECDH hash includes a parity-dependent version byte. Since the +initiator may only have the x-only key, both operations are normalized to be +parity-independent: the pre-message hash uses even parity (`0x02` prefix), +and ECDH hashes only the x-coordinate of the result point. This ensures +handshakes succeed regardless of the responder's actual key parity. See +`secp256k1-parity-fix.md` for detailed analysis. + ### 6.4 Handshake Integration with SessionSetup The Noise handshake messages embed in SessionSetup/SessionAck: diff --git a/docs/design/fips-state-machines.md b/docs/design/fips-state-machines.md index 7029312..0037526 100644 --- a/docs/design/fips-state-machines.md +++ b/docs/design/fips-state-machines.md @@ -232,9 +232,9 @@ PeerSlot::Connecting(PeerConnection) ▼ PeerSlot::Active(ActivePeer) │ - │ Link failure / explicit disconnect + │ Disconnect message (0x50) / link failure / timeout ▼ -[removed from peers map] +[removed from peers map, index freed, link cleaned up] ``` **PeerConnection** contains: diff --git a/docs/design/fips-wire-protocol.md b/docs/design/fips-wire-protocol.md index 119ca9e..cdd770a 100644 --- a/docs/design/fips-wire-protocol.md +++ b/docs/design/fips-wire-protocol.md @@ -156,6 +156,53 @@ Indices SHOULD be: 4. **Rotated on rekey**: When a session rekeys, allocate new indices to prevent cross-session correlation. +### 2.6 Link Control Messages + +Link control messages are sent inside encrypted frames (discriminator 0x00) and +use the 0x50–0x5F message type range. The first defined control message is +Disconnect (0x50). + +#### Disconnect (0x50) + +Orderly disconnect notification sent before closing a peer link: + +```text +ENCRYPTED FRAME (discriminator 0x00): + [receiver_idx][counter][ENCRYPTED_PAYLOAD + tag] + +DECRYPTED PLAINTEXT: +┌──────────┬──────────┐ +│ 0x50 │ reason │ +│ 1 byte │ 1 byte │ +└──────────┴──────────┘ + +Total plaintext: 2 bytes +``` + +**Reason codes:** + +| Code | Name | Description | +|------|---------------------|------------------------------------------| +| 0x00 | Shutdown | Normal operator-requested stop | +| 0x01 | Restart | Restarting, may reconnect soon | +| 0x02 | ProtocolError | Protocol error encountered | +| 0x03 | TransportFailure | Transport failure | +| 0x04 | ResourceExhaustion | Memory or connection limit | +| 0x05 | SecurityViolation | Authentication or policy violation | +| 0x06 | ConfigurationChange | Peer removed from configuration | +| 0x07 | Timeout | Keepalive or stale detection timeout | +| 0xFF | Other | Unspecified reason | + +**Semantics:** + +- **Best-effort delivery**: If the transport is broken, the message won't arrive. + Timeout-based detection (stale peer, keepalive failure) remains the fallback. +- **Receiver action**: Immediately remove the peer from the peer table, free the + session index, remove the link, and clean up address mappings. If the departed + peer was a tree parent, trigger parent reselection. +- **Shutdown sequence**: On node shutdown, Disconnect is sent to all active peers + *before* transports are stopped. + --- ## 3. Packet Dispatch @@ -846,9 +893,9 @@ Post-handshake data packets between authenticated peers. │ └────────┴──────────────────┴───────────┴───────────────────────────────┘ │ │ │ │ Link message types: │ -│ 0x10 = TreeAnnounce 0x12 = LookupRequest │ -│ 0x11 = FilterAnnounce 0x13 = LookupResponse │ -│ 0x40 = SessionDatagram │ +│ 0x10 = TreeAnnounce 0x30 = LookupRequest │ +│ 0x20 = FilterAnnounce 0x31 = LookupResponse │ +│ 0x40 = SessionDatagram 0x50 = Disconnect │ │ │ └─────────────────────────────────────────────────────────────────────────────┘ ``` diff --git a/src/identity.rs b/src/identity.rs index 7454497..4e374d3 100644 --- a/src/identity.rs +++ b/src/identity.rs @@ -897,3 +897,33 @@ mod tests { assert_eq!(identity.node_addr(), identity_from_bytes.node_addr()); } } + +#[cfg(test)] +mod conversion_tests { + use super::*; + + #[test] + fn test_hex_conversion_case1() { + let hex_str = "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20"; + let identity = Identity::from_secret_str(hex_str).unwrap(); + let npub = identity.npub(); + println!("Hex: {}", hex_str); + println!("NPub: {}", npub); + println!("NodeAddr: {}", identity.node_addr()); + println!("FipsAddress: {}", identity.address()); + assert!(npub.starts_with("npub1")); + } + + #[test] + fn test_hex_conversion_case2() { + let hex_str = "b102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fb0"; + let identity = Identity::from_secret_str(hex_str).unwrap(); + let npub = identity.npub(); + println!("Hex: {}", hex_str); + println!("NPub: {}", npub); + println!("NodeAddr: {}", identity.node_addr()); + println!("FipsAddress: {}", identity.address()); + assert!(npub.starts_with("npub1")); + } + +} diff --git a/src/node/handlers.rs b/src/node/handlers.rs index e2bd2ce..79cdb8e 100644 --- a/src/node/handlers.rs +++ b/src/node/handlers.rs @@ -1,6 +1,7 @@ //! RX event loop and message handlers. use super::*; +use crate::rate_limit::HANDSHAKE_TIMEOUT_SECS; impl Node { // === RX Event Loop === @@ -13,16 +14,32 @@ impl Node { /// - 0x01: Handshake message 1 (initiator -> responder) /// - 0x02: Handshake message 2 (responder -> initiator) /// + /// Also runs a periodic tick (1s) to clean up stale handshake connections + /// that never received a response. This prevents resource leaks when peers + /// are unreachable. + /// /// This method takes ownership of the packet_rx channel and runs /// until the channel is closed (typically when stop() is called). pub async fn run_rx_loop(&mut self) -> Result<(), NodeError> { let mut packet_rx = self.packet_rx.take() .ok_or(NodeError::NotStarted)?; + let mut tick = tokio::time::interval(Duration::from_secs(1)); + info!("RX event loop started"); - while let Some(packet) = packet_rx.recv().await { - self.process_packet(packet).await; + loop { + tokio::select! { + packet = packet_rx.recv() => { + match packet { + Some(p) => self.process_packet(p).await, + None => break, // channel closed + } + } + _ = tick.tick() => { + self.check_timeouts(); + } + } } info!("RX event loop stopped (channel closed)"); @@ -159,16 +176,32 @@ impl Node { } }; - // Check for existing connection from this address + // Check for existing connection from this address. + // + // If we already have an *inbound* link from this address, drop the msg1 + // (duplicate or replay). But if we have an *outbound* link to this address + // (we initiated to them AND they initiated to us), this is a cross-connection. + // Allow it to proceed — promote_connection() will resolve via tie-breaker. let addr_key = (packet.transport_id, packet.remote_addr.clone()); - if self.addr_to_link.contains_key(&addr_key) { - self.msg1_rate_limiter.complete_handshake(); - debug!( - transport_id = %packet.transport_id, - remote_addr = %packet.remote_addr, - "Already have connection from this address" - ); - return; + if let Some(&existing_link_id) = self.addr_to_link.get(&addr_key) { + if let Some(link) = self.links.get(&existing_link_id) { + if link.direction() == LinkDirection::Inbound { + self.msg1_rate_limiter.complete_handshake(); + debug!( + transport_id = %packet.transport_id, + remote_addr = %packet.remote_addr, + "Already have inbound connection from this address" + ); + return; + } + // Outbound link to this address — cross-connection, allow msg1 + debug!( + transport_id = %packet.transport_id, + remote_addr = %packet.remote_addr, + existing_link_id = %existing_link_id, + "Cross-connection detected: have outbound, received inbound msg1" + ); + } } // === CRYPTO COST PAID HERE === @@ -279,13 +312,22 @@ impl Node { ); } PromotionResult::CrossConnectionWon { loser_link_id, node_addr } => { + // Clean up the losing connection's link + self.remove_link(&loser_link_id); info!( node_addr = %node_addr, loser_link_id = %loser_link_id, - "Inbound cross-connection won" + "Inbound cross-connection won, loser link cleaned up" ); } PromotionResult::CrossConnectionLost { winner_link_id } => { + // This connection lost — clean up its link + self.remove_link(&link_id); + // Restore addr_to_link for the winner's link + self.addr_to_link.insert( + (packet.transport_id, packet.remote_addr.clone()), + winner_link_id, + ); info!( winner_link_id = %winner_link_id, "Inbound cross-connection lost, keeping existing" @@ -300,9 +342,7 @@ impl Node { "Failed to promote inbound connection" ); // Clean up on promotion failure - self.links.remove(&link_id); - self.addr_to_link - .remove(&(packet.transport_id, packet.remote_addr)); + self.remove_link(&link_id); let _ = self.index_allocator.free(our_index); } } @@ -392,16 +432,30 @@ impl Node { ); } PromotionResult::CrossConnectionWon { loser_link_id, node_addr } => { + // Clean up the losing connection's link + self.remove_link(&loser_link_id); + // Ensure addr_to_link points to the winning link + self.addr_to_link.insert( + (packet.transport_id, packet.remote_addr.clone()), + link_id, + ); info!( node_addr = %node_addr, loser_link_id = %loser_link_id, - "Cross-connection won" + "Outbound cross-connection won, loser link cleaned up" ); } PromotionResult::CrossConnectionLost { winner_link_id } => { + // This connection lost — clean up its link + self.remove_link(&link_id); + // Ensure addr_to_link points to the winner's link + self.addr_to_link.insert( + (packet.transport_id, packet.remote_addr.clone()), + winner_link_id, + ); info!( winner_link_id = %winner_link_id, - "Cross-connection lost" + "Outbound cross-connection lost, keeping existing" ); } } @@ -575,15 +629,15 @@ impl Node { /// Dispatch a decrypted link message to the appropriate handler. /// /// Link messages are protocol messages exchanged between authenticated peers. - async fn dispatch_link_message(&mut self, _from: &NodeAddr, plaintext: &[u8]) { + async fn dispatch_link_message(&mut self, from: &NodeAddr, plaintext: &[u8]) { if plaintext.is_empty() { return; } let msg_type = plaintext[0]; - let _payload = &plaintext[1..]; + let payload = &plaintext[1..]; - // TODO: Implement link message handlers + // TODO: Implement remaining link message handlers match msg_type { 0x10 => { // TreeAnnounce @@ -605,9 +659,134 @@ impl Node { // SessionDatagram debug!("Received SessionDatagram (not yet implemented)"); } + 0x50 => { + // Disconnect + self.handle_disconnect(from, payload); + } _ => { debug!(msg_type = msg_type, "Unknown link message type"); } } } + + /// Handle a Disconnect notification from a peer. + /// + /// The peer is signaling an orderly departure. We immediately remove + /// them from all state rather than waiting for timeout detection. + fn handle_disconnect(&mut self, from: &NodeAddr, payload: &[u8]) { + let disconnect = match crate::protocol::Disconnect::decode(payload) { + Ok(msg) => msg, + Err(e) => { + debug!(from = %from, error = %e, "Malformed disconnect message"); + return; + } + }; + + info!( + node_addr = %from, + reason = %disconnect.reason, + "Peer sent disconnect notification" + ); + + self.remove_active_peer(from); + } + + /// Remove an active peer and clean up all associated state. + /// + /// Frees session index, removes link and address mappings. Used for + /// both graceful disconnect and timeout-based eviction. + pub(super) fn remove_active_peer(&mut self, node_addr: &NodeAddr) { + let peer = match self.peers.remove(node_addr) { + Some(p) => p, + None => { + debug!(node_addr = %node_addr, "Peer already removed"); + return; + } + }; + + let link_id = peer.link_id(); + + // Free session index + if let (Some(tid), Some(idx)) = (peer.transport_id(), peer.our_index()) { + self.peers_by_index.remove(&(tid, idx.as_u32())); + let _ = self.index_allocator.free(idx); + } + + // Remove link and address mapping + self.remove_link(&link_id); + + info!( + node_addr = %node_addr, + link_id = %link_id, + "Peer removed and state cleaned up" + ); + } + + // === Timeout Management === + + /// Check for timed-out handshake connections and clean them up. + /// + /// Called periodically by the RX event loop. Removes connections that have + /// been idle longer than HANDSHAKE_TIMEOUT_SECS or are in Failed state. + pub(super) fn check_timeouts(&mut self) { + if self.connections.is_empty() { + return; + } + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let timeout_ms = HANDSHAKE_TIMEOUT_SECS * 1000; + + let stale: Vec = self.connections.iter() + .filter(|(_, conn)| conn.is_timed_out(now_ms, timeout_ms) || conn.is_failed()) + .map(|(link_id, _)| *link_id) + .collect(); + + for link_id in stale { + self.cleanup_stale_connection(link_id, now_ms); + } + } + + /// Remove a stale or failed handshake connection and all associated state. + /// + /// Frees the session index, removes pending_outbound entry, and cleans up + /// the link and address mapping. + fn cleanup_stale_connection(&mut self, link_id: LinkId, now_ms: u64) { + let conn = match self.connections.remove(&link_id) { + Some(c) => c, + None => return, + }; + + let direction = conn.direction(); + let idle_ms = conn.idle_time(now_ms); + let is_failed = conn.is_failed(); + + // Free session index and pending_outbound if allocated + if let Some(idx) = conn.our_index() { + if let Some(tid) = conn.transport_id() { + self.pending_outbound.remove(&(tid, idx.as_u32())); + } + let _ = self.index_allocator.free(idx); + } + + // Remove link and addr_to_link + self.remove_link(&link_id); + + if is_failed { + info!( + link_id = %link_id, + direction = %direction, + "Failed handshake connection cleaned up" + ); + } else { + info!( + link_id = %link_id, + direction = %direction, + idle_secs = idle_ms / 1000, + "Stale handshake connection timed out" + ); + } + } } diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 8836660..e541e48 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -1,6 +1,7 @@ //! Node lifecycle management: start, stop, and peer connection initiation. use super::*; +use crate::protocol::{Disconnect, DisconnectReason}; impl Node { /// Initiate connections to configured static peers. @@ -316,7 +317,10 @@ impl Node { self.state = NodeState::Stopping; info!(state = %self.state, "Node stopping"); - // Shutdown transports first (they're packet producers) + // Send disconnect notifications to all active peers before closing transports + self.send_disconnect_to_all_peers(DisconnectReason::Shutdown).await; + + // Shutdown transports (they're packet producers) let transport_ids: Vec<_> = self.transports.keys().cloned().collect(); for transport_id in transport_ids { if let Some(mut handle) = self.transports.remove(&transport_id) { @@ -368,4 +372,43 @@ impl Node { info!(state = %self.state, "Node stopped"); Ok(()) } + + /// Send disconnect notifications to all active peers. + /// + /// Best-effort: send failures are logged and ignored since the transport + /// may already be degraded. This runs before transports are shut down. + async fn send_disconnect_to_all_peers(&mut self, reason: DisconnectReason) { + let disconnect = Disconnect::new(reason); + let plaintext = disconnect.encode(); + + // Collect node_addrs to avoid borrow conflict with send helper + let peer_addrs: Vec = self.peers.iter() + .filter(|(_, peer)| peer.can_send() && peer.has_session()) + .map(|(addr, _)| *addr) + .collect(); + + if peer_addrs.is_empty() { + debug!( + total_peers = self.peers.len(), + "No sendable peers for disconnect notification" + ); + return; + } + + let mut sent = 0usize; + for node_addr in &peer_addrs { + match self.send_encrypted_link_message(node_addr, &plaintext).await { + Ok(()) => sent += 1, + Err(e) => { + debug!( + node_addr = %node_addr, + error = %e, + "Failed to send disconnect (transport may be down)" + ); + } + } + } + + info!(sent, total = peer_addrs.len(), reason = %reason, "Sent disconnect notifications"); + } } diff --git a/src/node/mod.rs b/src/node/mod.rs index 36f40e4..2c42550 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -24,7 +24,7 @@ use crate::transport::udp::UdpTransport; use crate::tree::TreeState; use crate::tun::{run_tun_reader, shutdown_tun_interface, TunDevice, TunError, TunState, TunTx}; use crate::wire::{ - build_msg1, build_msg2, EncryptedHeader, Msg1Header, Msg2Header, + build_encrypted, build_msg1, build_msg2, EncryptedHeader, Msg1Header, Msg2Header, DISCRIMINATOR_ENCRYPTED, DISCRIMINATOR_MSG1, DISCRIMINATOR_MSG2, }; use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity}; @@ -89,6 +89,9 @@ pub enum NodeError { #[error("promotion failed for link {link_id}: {reason}")] PromotionFailed { link_id: LinkId, reason: String }, + #[error("send failed to {node_addr}: {reason}")] + SendFailed { node_addr: NodeAddr, reason: String }, + #[error("config error: {0}")] Config(#[from] ConfigError), @@ -594,11 +597,17 @@ impl Node { } /// Remove a link. + /// + /// Only removes the addr_to_link reverse lookup if it still points to this + /// link. In cross-connection scenarios, a newer link may have replaced the + /// entry for the same address. pub fn remove_link(&mut self, link_id: &LinkId) -> Option { if let Some(link) = self.links.remove(link_id) { - // Clean up reverse lookup + // Clean up reverse lookup only if it still maps to this link let key = (link.transport_id(), link.remote_addr().clone()); - self.addr_to_link.remove(&key); + if self.addr_to_link.get(&key) == Some(link_id) { + self.addr_to_link.remove(&key); + } Some(link) } else { None @@ -708,6 +717,68 @@ impl Node { pub fn tun_tx(&self) -> Option<&TunTx> { self.tun_tx.as_ref() } + + // === Sending === + + /// Encrypt and send a link-layer message to an authenticated peer. + /// + /// The plaintext should include the message type byte followed by the + /// message-specific payload (e.g., `[0x50, reason]` for Disconnect). + /// + /// This is the standard path for sending any link-layer control message + /// to a peer over their encrypted Noise session. + pub(super) async fn send_encrypted_link_message( + &mut self, + node_addr: &NodeAddr, + plaintext: &[u8], + ) -> Result<(), NodeError> { + let peer = self.peers.get_mut(node_addr) + .ok_or(NodeError::PeerNotFound(*node_addr))?; + + let their_index = peer.their_index().ok_or_else(|| NodeError::SendFailed { + node_addr: *node_addr, + reason: "no their_index".into(), + })?; + let transport_id = peer.transport_id().ok_or_else(|| NodeError::SendFailed { + node_addr: *node_addr, + reason: "no transport_id".into(), + })?; + let remote_addr = peer.current_addr().cloned().ok_or_else(|| NodeError::SendFailed { + node_addr: *node_addr, + reason: "no current_addr".into(), + })?; + + let session = peer.noise_session_mut().ok_or_else(|| NodeError::SendFailed { + node_addr: *node_addr, + reason: "no noise session".into(), + })?; + + // Get counter before encrypt (encrypt increments it) + let counter = session.current_send_counter(); + let ciphertext = session.encrypt(plaintext).map_err(|e| NodeError::SendFailed { + node_addr: *node_addr, + reason: format!("encryption failed: {}", e), + })?; + + let wire_packet = build_encrypted(their_index, counter, &ciphertext); + + // Re-borrow peer for stats update after sending + let transport = self.transports.get(&transport_id) + .ok_or(NodeError::TransportNotFound(transport_id))?; + + let bytes_sent = transport.send(&remote_addr, &wire_packet).await + .map_err(|e| NodeError::SendFailed { + node_addr: *node_addr, + reason: format!("transport send: {}", e), + })?; + + // Update send statistics + if let Some(peer) = self.peers.get_mut(node_addr) { + peer.link_stats_mut().record_sent(bytes_sent); + } + + Ok(()) + } } impl fmt::Debug for Node { diff --git a/src/node/tests.rs b/src/node/tests.rs index 209a65f..b68da44 100644 --- a/src/node/tests.rs +++ b/src/node/tests.rs @@ -896,3 +896,269 @@ async fn test_run_rx_loop_handshake() { t.stop().await.ok(); } } + +/// Integration test: simultaneous cross-connection (both nodes initiate). +/// +/// Simulates the live scenario where both nodes have auto_connect to each other. +/// Both send msg1 simultaneously, creating a cross-connection that must be +/// resolved by the tie-breaker rule. Exercises the addr_to_link fix that allows +/// inbound msg1 when an outbound link to the same address already exists. +#[tokio::test] +async fn test_cross_connection_both_initiate() { + use crate::config::UdpConfig; + use crate::transport::udp::UdpTransport; + use crate::wire::build_msg1; + use tokio::time::{timeout, Duration}; + + // === Setup: Two nodes with UDP transports on localhost === + + let mut node_a = make_node(); + let mut node_b = make_node(); + + let transport_id_a = TransportId::new(1); + let transport_id_b = TransportId::new(1); + + let udp_config = UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + mtu: Some(1280), + }; + + let (packet_tx_a, mut packet_rx_a) = packet_channel(64); + let (packet_tx_b, mut packet_rx_b) = packet_channel(64); + + let mut transport_a = + UdpTransport::new(transport_id_a, None, udp_config.clone(), packet_tx_a); + let mut transport_b = + UdpTransport::new(transport_id_b, None, udp_config, packet_tx_b); + + transport_a.start_async().await.unwrap(); + transport_b.start_async().await.unwrap(); + + let addr_a = transport_a.local_addr().unwrap(); + let addr_b = transport_b.local_addr().unwrap(); + let remote_addr_b = TransportAddr::from_string(&addr_b.to_string()); + let remote_addr_a = TransportAddr::from_string(&addr_a.to_string()); + + node_a + .transports + .insert(transport_id_a, TransportHandle::Udp(transport_a)); + node_b + .transports + .insert(transport_id_b, TransportHandle::Udp(transport_b)); + + // Peer identities (must use full key for ECDH parity) + let peer_b_identity = + PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full()); + let peer_b_node_addr = *peer_b_identity.node_addr(); + let peer_a_identity = + PeerIdentity::from_pubkey_full(node_a.identity.pubkey_full()); + let peer_a_node_addr = *peer_a_identity.node_addr(); + + // === Phase 1: Both nodes initiate handshakes (simulate auto_connect) === + + // Node A initiates to Node B + let link_id_a_out = node_a.allocate_link_id(); + let mut conn_a = PeerConnection::outbound(link_id_a_out, peer_b_identity.clone(), 1000); + let our_index_a = node_a.index_allocator.allocate().unwrap(); + let our_keypair_a = node_a.identity.keypair(); + let noise_msg1_a = conn_a.start_handshake(our_keypair_a, 1000).unwrap(); + conn_a.set_our_index(our_index_a); + conn_a.set_transport_id(transport_id_a); + conn_a.set_source_addr(remote_addr_b.clone()); + + let wire_msg1_a = build_msg1(our_index_a, &noise_msg1_a); + + let link_a_out = Link::connectionless( + link_id_a_out, transport_id_a, remote_addr_b.clone(), + LinkDirection::Outbound, Duration::from_millis(100), + ); + node_a.links.insert(link_id_a_out, link_a_out); + node_a.addr_to_link.insert((transport_id_a, remote_addr_b.clone()), link_id_a_out); + node_a.connections.insert(link_id_a_out, conn_a); + node_a.pending_outbound.insert((transport_id_a, our_index_a.as_u32()), link_id_a_out); + + // Node B initiates to Node A + let link_id_b_out = node_b.allocate_link_id(); + let mut conn_b = PeerConnection::outbound(link_id_b_out, peer_a_identity.clone(), 1000); + let our_index_b = node_b.index_allocator.allocate().unwrap(); + let our_keypair_b = node_b.identity.keypair(); + let noise_msg1_b = conn_b.start_handshake(our_keypair_b, 1000).unwrap(); + conn_b.set_our_index(our_index_b); + conn_b.set_transport_id(transport_id_b); + conn_b.set_source_addr(remote_addr_a.clone()); + + let wire_msg1_b = build_msg1(our_index_b, &noise_msg1_b); + + let link_b_out = Link::connectionless( + link_id_b_out, transport_id_b, remote_addr_a.clone(), + LinkDirection::Outbound, Duration::from_millis(100), + ); + node_b.links.insert(link_id_b_out, link_b_out); + node_b.addr_to_link.insert((transport_id_b, remote_addr_a.clone()), link_id_b_out); + node_b.connections.insert(link_id_b_out, conn_b); + node_b.pending_outbound.insert((transport_id_b, our_index_b.as_u32()), link_id_b_out); + + // Both send msg1 over UDP + let transport = node_a.transports.get(&transport_id_a).unwrap(); + transport.send(&remote_addr_b, &wire_msg1_a).await.expect("A send msg1"); + + let transport = node_b.transports.get(&transport_id_b).unwrap(); + transport.send(&remote_addr_a, &wire_msg1_b).await.expect("B send msg1"); + + // === Phase 2: Both nodes receive the other's msg1 === + // Before the fix, addr_to_link would reject these because outbound links + // already exist for these addresses. + + // B receives A's msg1 + let packet_at_b = timeout(Duration::from_secs(1), packet_rx_b.recv()) + .await.expect("Timeout").expect("Channel closed"); + node_b.handle_msg1(packet_at_b).await; + + // B should have promoted the inbound connection + assert_eq!(node_b.peer_count(), 1, "Node B should have 1 peer after processing A's msg1"); + assert!(node_b.get_peer(&peer_a_node_addr).is_some(), "Node B should have peer A"); + + // A receives B's msg1 + let packet_at_a = timeout(Duration::from_secs(1), packet_rx_a.recv()) + .await.expect("Timeout").expect("Channel closed"); + node_a.handle_msg1(packet_at_a).await; + + // A should have promoted the inbound connection + assert_eq!(node_a.peer_count(), 1, "Node A should have 1 peer after processing B's msg1"); + assert!(node_a.get_peer(&peer_b_node_addr).is_some(), "Node A should have peer B"); + + // === Phase 3: Both nodes receive msg2 responses === + // The msg2 was sent during handle_msg1 processing. When handle_msg2 + // processes it, it will detect the cross-connection and resolve. + + // A receives B's msg2 (response to A's original msg1) + let msg2_at_a = timeout(Duration::from_secs(1), packet_rx_a.recv()) + .await.expect("Timeout waiting for msg2 at A").expect("Channel closed"); + node_a.handle_msg2(msg2_at_a).await; + + // B receives A's msg2 (response to B's original msg1) + let msg2_at_b = timeout(Duration::from_secs(1), packet_rx_b.recv()) + .await.expect("Timeout waiting for msg2 at B").expect("Channel closed"); + node_b.handle_msg2(msg2_at_b).await; + + // === Verification === + // Both nodes should have exactly 1 peer each after cross-connection resolution + assert_eq!(node_a.peer_count(), 1, "Node A should have exactly 1 peer after cross-connection"); + assert_eq!(node_b.peer_count(), 1, "Node B should have exactly 1 peer after cross-connection"); + + let peer_b_on_a = node_a.get_peer(&peer_b_node_addr).expect("A should have peer B"); + let peer_a_on_b = node_b.get_peer(&peer_a_node_addr).expect("B should have peer A"); + + assert!(peer_b_on_a.has_session(), "Peer B on A should have session"); + assert!(peer_a_on_b.has_session(), "Peer A on B should have session"); + assert!(peer_b_on_a.can_send(), "Peer B on A should be sendable"); + assert!(peer_a_on_b.can_send(), "Peer A on B should be sendable"); + + // Clean up transports + for (_, t) in node_a.transports.iter_mut() { + t.stop().await.ok(); + } + for (_, t) in node_b.transports.iter_mut() { + t.stop().await.ok(); + } +} + +/// Test that stale handshake connections are cleaned up by check_timeouts(). +/// +/// Simulates the scenario where a node initiates a handshake to a peer that +/// isn't running. The outbound connection should be cleaned up after the +/// handshake timeout expires. +#[tokio::test] +async fn test_stale_connection_cleanup() { + let mut node = make_node(); + let transport_id = TransportId::new(1); + + let peer_identity = make_peer_identity(); + let remote_addr = TransportAddr::from_string("10.0.0.2:4000"); + + // Create outbound connection with a timestamp far in the past + let past_time_ms = 1000; // A very early timestamp + let link_id = node.allocate_link_id(); + let mut conn = PeerConnection::outbound(link_id, peer_identity.clone(), past_time_ms); + + // Allocate session index and set transport info + let our_index = node.index_allocator.allocate().unwrap(); + let our_keypair = node.identity.keypair(); + let _noise_msg1 = conn.start_handshake(our_keypair, past_time_ms).unwrap(); + conn.set_our_index(our_index); + conn.set_transport_id(transport_id); + conn.set_source_addr(remote_addr.clone()); + + // Set up all the state that initiate_peer_connection would create + let link = Link::connectionless( + link_id, transport_id, remote_addr.clone(), + LinkDirection::Outbound, Duration::from_millis(100), + ); + node.links.insert(link_id, link); + node.addr_to_link.insert((transport_id, remote_addr.clone()), link_id); + node.connections.insert(link_id, conn); + node.pending_outbound.insert((transport_id, our_index.as_u32()), link_id); + + // Verify state before timeout check + assert_eq!(node.connection_count(), 1); + assert_eq!(node.link_count(), 1); + assert!(node.pending_outbound.contains_key(&(transport_id, our_index.as_u32()))); + assert_eq!(node.index_allocator.count(), 1); + + // Connection was created at time 1000ms. check_timeouts uses SystemTime::now(), + // which is far beyond the 30s timeout. The connection should be cleaned up. + node.check_timeouts(); + + // Verify everything was cleaned up + assert_eq!(node.connection_count(), 0, "Stale connection should be removed"); + assert_eq!(node.link_count(), 0, "Stale link should be removed"); + assert!(!node.pending_outbound.contains_key(&(transport_id, our_index.as_u32())), + "pending_outbound should be cleaned up"); + assert_eq!(node.index_allocator.count(), 0, "Session index should be freed"); + assert!(node.addr_to_link.get(&(transport_id, remote_addr)).is_none(), + "addr_to_link should be cleaned up"); +} + +/// Test that failed connections are cleaned up by check_timeouts(). +#[tokio::test] +async fn test_failed_connection_cleanup() { + let mut node = make_node(); + let transport_id = TransportId::new(1); + + let peer_identity = make_peer_identity(); + let remote_addr = TransportAddr::from_string("10.0.0.2:4000"); + + // Create a connection and mark it failed (simulating a send failure) + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let link_id = node.allocate_link_id(); + let mut conn = PeerConnection::outbound(link_id, peer_identity.clone(), now_ms); + + let our_index = node.index_allocator.allocate().unwrap(); + let our_keypair = node.identity.keypair(); + let _noise_msg1 = conn.start_handshake(our_keypair, now_ms).unwrap(); + conn.set_our_index(our_index); + conn.set_transport_id(transport_id); + conn.set_source_addr(remote_addr.clone()); + conn.mark_failed(); // Simulate send failure + + let link = Link::connectionless( + link_id, transport_id, remote_addr.clone(), + LinkDirection::Outbound, Duration::from_millis(100), + ); + node.links.insert(link_id, link); + node.addr_to_link.insert((transport_id, remote_addr.clone()), link_id); + node.connections.insert(link_id, conn); + node.pending_outbound.insert((transport_id, our_index.as_u32()), link_id); + + assert_eq!(node.connection_count(), 1); + + // Failed connections should be cleaned up immediately regardless of age + node.check_timeouts(); + + assert_eq!(node.connection_count(), 0, "Failed connection should be removed"); + assert_eq!(node.link_count(), 0, "Failed link should be removed"); + assert_eq!(node.index_allocator.count(), 0, "Session index should be freed"); +} diff --git a/src/noise.rs b/src/noise.rs index 1a8cd99..0003045 100644 --- a/src/noise.rs +++ b/src/noise.rs @@ -37,7 +37,7 @@ use chacha20poly1305::{ }; use hkdf::Hkdf; use rand::RngCore; -use secp256k1::{ecdh::SharedSecret, Keypair, PublicKey, Secp256k1, SecretKey, XOnlyPublicKey}; +use secp256k1::{ecdh::shared_secret_point, Keypair, PublicKey, Secp256k1, SecretKey, XOnlyPublicKey}; use sha2::{Digest, Sha256}; use std::fmt; use thiserror::Error; @@ -561,6 +561,22 @@ pub struct HandshakeState { } impl HandshakeState { + /// Normalize a compressed public key to even parity for pre-message hashing. + /// + /// Nostr npubs encode x-only keys (no parity). The Noise IK pre-message + /// mixes the responder's static key into the hash before any messages. + /// Both sides must mix identical bytes. Since the initiator may only have + /// the x-only key (from an npub), we normalize to even parity (0x02 prefix) + /// so the hash chain matches regardless of the key's actual parity. + /// + /// This does NOT affect ECDH operations (which use x-coordinate-only output) + /// or the keys sent in handshake messages (which use actual parity). + fn normalize_for_premessage(pubkey: &PublicKey) -> [u8; PUBKEY_SIZE] { + let mut bytes = pubkey.serialize(); + bytes[0] = 0x02; // Force even parity + bytes + } + /// Create a new handshake as initiator. /// /// The initiator knows the responder's static key and will send first. @@ -578,8 +594,10 @@ impl HandshakeState { }; // Mix in pre-message: <- s (responder's static is known) - let remote_static_bytes = remote_static.serialize(); - state.symmetric.mix_hash(&remote_static_bytes); + // Normalize to even parity so initiator and responder hash chains match + // even when the initiator only has the x-only key (from npub). + let normalized = Self::normalize_for_premessage(&remote_static); + state.symmetric.mix_hash(&normalized); state } @@ -602,8 +620,9 @@ impl HandshakeState { }; // Mix in pre-message: <- s (our static, since we're responder) - let our_static_pubkey = state.static_keypair.public_key().serialize(); - state.symmetric.mix_hash(&our_static_pubkey); + // Normalize to even parity to match initiator's hash chain. + let normalized = Self::normalize_for_premessage(&state.static_keypair.public_key()); + state.symmetric.mix_hash(&normalized); state } @@ -640,10 +659,22 @@ impl HandshakeState { } /// Perform ECDH between our secret and their public key. + /// + /// Uses x-only hashing (SHA-256 of just the x-coordinate) to produce + /// a parity-independent shared secret. This is necessary because Nostr + /// npubs encode x-only keys without parity information, so the initiator + /// may have the wrong parity for the responder's static key. Since P and + /// -P produce ECDH result points with the same x-coordinate, hashing + /// only x ensures both sides derive the same shared secret. fn ecdh(&self, our_secret: &SecretKey, their_public: &PublicKey) -> [u8; 32] { - let shared = SharedSecret::new(their_public, our_secret); + // Get raw (x, y) coordinates (64 bytes) without any hashing + let point = shared_secret_point(their_public, our_secret); + // Hash only the x-coordinate (first 32 bytes), ignoring y/parity + let mut hasher = Sha256::new(); + hasher.update(&point[..32]); + let hash = hasher.finalize(); let mut result = [0u8; 32]; - result.copy_from_slice(shared.as_ref()); + result.copy_from_slice(&hash); result } @@ -1024,6 +1055,7 @@ impl fmt::Debug for NoiseSession { #[cfg(test)] mod tests { use super::*; + use secp256k1::Parity; fn generate_keypair() -> Keypair { let secp = Secp256k1::new(); @@ -1381,4 +1413,63 @@ mod tests { // Check method alone also detects replay assert!(receiver.check_replay(counter).is_err()); } + + #[test] + fn test_handshake_with_odd_parity_responder() { + // Node B's secret key produces an odd-parity public key (0x03 prefix). + // When the initiator only has the npub (x-only), PeerIdentity::pubkey_full() + // returns even parity (0x02). The pre-message mix_hash must normalize + // parity so both sides produce matching hash chains. + let secp = Secp256k1::new(); + + // Node B (responder) - odd parity key + let sk_b = SecretKey::from_slice( + &hex::decode("b102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fb0") + .unwrap(), + ) + .unwrap(); + let kp_b = Keypair::from_secret_key(&secp, &sk_b); + let (xonly_b, parity_b) = kp_b.public_key().x_only_public_key(); + assert_eq!(parity_b, Parity::Odd, "Test requires odd-parity responder key"); + + // Node A (initiator) - even parity key + let sk_a = SecretKey::from_slice( + &hex::decode("0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20") + .unwrap(), + ) + .unwrap(); + let kp_a = Keypair::from_secret_key(&secp, &sk_a); + + // Simulate the production path: initiator gets responder's key via npub + // (x-only → assumed even parity) + let assumed_even_b = xonly_b.public_key(Parity::Even); + assert_ne!( + assumed_even_b, kp_b.public_key(), + "Even assumption should differ from actual odd key" + ); + + // Handshake using assumed-even key (as production code does) + let mut initiator = HandshakeState::new_initiator(kp_a, assumed_even_b); + let mut responder = HandshakeState::new_responder(kp_b); + + let msg1 = initiator.write_message_1().unwrap(); + responder.read_message_1(&msg1).unwrap(); + + let msg2 = responder.write_message_2().unwrap(); + initiator.read_message_2(&msg2).unwrap(); + + assert!(initiator.is_complete()); + assert!(responder.is_complete()); + + // Verify sessions can communicate + let mut sender = initiator.into_session().unwrap(); + let mut receiver = responder.into_session().unwrap(); + + let counter = sender.current_send_counter(); + let ciphertext = sender.encrypt(b"parity test").unwrap(); + let plaintext = receiver + .decrypt_with_replay_check(&ciphertext, counter) + .unwrap(); + assert_eq!(plaintext, b"parity test"); + } } diff --git a/src/protocol.rs b/src/protocol.rs index 7549b9c..6377674 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -116,6 +116,10 @@ pub enum LinkMessageType { /// Encapsulated session-layer datagram for forwarding. /// Payload is opaque to intermediate nodes (end-to-end encrypted). SessionDatagram = 0x40, + + // Link Control (0x50-0x5F) + /// Orderly disconnect notification before link closure. + Disconnect = 0x50, } impl LinkMessageType { @@ -127,6 +131,7 @@ impl LinkMessageType { 0x30 => Some(LinkMessageType::LookupRequest), 0x31 => Some(LinkMessageType::LookupResponse), 0x40 => Some(LinkMessageType::SessionDatagram), + 0x50 => Some(LinkMessageType::Disconnect), _ => None, } } @@ -145,11 +150,126 @@ impl fmt::Display for LinkMessageType { LinkMessageType::LookupRequest => "LookupRequest", LinkMessageType::LookupResponse => "LookupResponse", LinkMessageType::SessionDatagram => "SessionDatagram", + LinkMessageType::Disconnect => "Disconnect", }; write!(f, "{}", name) } } +// ============================================================================ +// Disconnect Reason Codes +// ============================================================================ + +/// Reason for an orderly disconnect notification. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(u8)] +pub enum DisconnectReason { + /// Normal shutdown (operator requested). + Shutdown = 0x00, + /// Restarting (may reconnect soon). + Restart = 0x01, + /// Protocol error encountered. + ProtocolError = 0x02, + /// Transport failure. + TransportFailure = 0x03, + /// Resource exhaustion (memory, connections). + ResourceExhaustion = 0x04, + /// Authentication or security policy violation. + SecurityViolation = 0x05, + /// Configuration change (peer removed from config). + ConfigurationChange = 0x06, + /// Timeout or keepalive failure. + Timeout = 0x07, + /// Unspecified reason. + Other = 0xFF, +} + +impl DisconnectReason { + /// Try to convert from a byte. + pub fn from_byte(b: u8) -> Option { + match b { + 0x00 => Some(DisconnectReason::Shutdown), + 0x01 => Some(DisconnectReason::Restart), + 0x02 => Some(DisconnectReason::ProtocolError), + 0x03 => Some(DisconnectReason::TransportFailure), + 0x04 => Some(DisconnectReason::ResourceExhaustion), + 0x05 => Some(DisconnectReason::SecurityViolation), + 0x06 => Some(DisconnectReason::ConfigurationChange), + 0x07 => Some(DisconnectReason::Timeout), + 0xFF => Some(DisconnectReason::Other), + _ => None, + } + } + + /// Convert to a byte. + pub fn to_byte(self) -> u8 { + self as u8 + } +} + +impl fmt::Display for DisconnectReason { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let name = match self { + DisconnectReason::Shutdown => "Shutdown", + DisconnectReason::Restart => "Restart", + DisconnectReason::ProtocolError => "ProtocolError", + DisconnectReason::TransportFailure => "TransportFailure", + DisconnectReason::ResourceExhaustion => "ResourceExhaustion", + DisconnectReason::SecurityViolation => "SecurityViolation", + DisconnectReason::ConfigurationChange => "ConfigurationChange", + DisconnectReason::Timeout => "Timeout", + DisconnectReason::Other => "Other", + }; + write!(f, "{}", name) + } +} + +// ============================================================================ +// Disconnect Message +// ============================================================================ + +/// Orderly disconnect notification sent before closing a peer link. +/// +/// Sent as a link-layer message (type 0x50) inside an encrypted frame. +/// Allows the receiving peer to immediately clean up state rather than +/// waiting for timeout-based detection. +/// +/// ## Wire Format +/// +/// | Offset | Field | Size | Notes | +/// |--------|----------|--------|------------------------| +/// | 0 | msg_type | 1 byte | 0x50 | +/// | 1 | reason | 1 byte | DisconnectReason value | +#[derive(Clone, Debug)] +pub struct Disconnect { + /// Reason for disconnection. + pub reason: DisconnectReason, +} + +impl Disconnect { + /// Create a new Disconnect message. + pub fn new(reason: DisconnectReason) -> Self { + Self { reason } + } + + /// Encode as link-layer plaintext (msg_type + reason). + pub fn encode(&self) -> [u8; 2] { + [LinkMessageType::Disconnect.to_byte(), self.reason.to_byte()] + } + + /// Decode from link-layer payload (after msg_type byte has been consumed). + pub fn decode(payload: &[u8]) -> Result { + if payload.is_empty() { + return Err(ProtocolError::MessageTooShort { + expected: 1, + got: 0, + }); + } + let reason = DisconnectReason::from_byte(payload[0]).unwrap_or(DisconnectReason::Other); + Ok(Self { reason }) + } +} + // ============================================================================ // Session Layer Message Types (end-to-end, between FIPS addresses) // ============================================================================ @@ -902,6 +1022,7 @@ mod tests { LinkMessageType::LookupRequest, LinkMessageType::LookupResponse, LinkMessageType::SessionDatagram, + LinkMessageType::Disconnect, ]; for ty in types { @@ -917,6 +1038,83 @@ mod tests { assert!(LinkMessageType::from_byte(0x00).is_none()); } + // ===== DisconnectReason Tests ===== + + #[test] + fn test_disconnect_reason_roundtrip() { + let reasons = [ + DisconnectReason::Shutdown, + DisconnectReason::Restart, + DisconnectReason::ProtocolError, + DisconnectReason::TransportFailure, + DisconnectReason::ResourceExhaustion, + DisconnectReason::SecurityViolation, + DisconnectReason::ConfigurationChange, + DisconnectReason::Timeout, + DisconnectReason::Other, + ]; + + for reason in reasons { + let byte = reason.to_byte(); + let restored = DisconnectReason::from_byte(byte); + assert_eq!(restored, Some(reason)); + } + } + + #[test] + fn test_disconnect_reason_unknown_byte() { + // Unrecognized bytes return None + assert!(DisconnectReason::from_byte(0x08).is_none()); + assert!(DisconnectReason::from_byte(0x80).is_none()); + assert!(DisconnectReason::from_byte(0xFE).is_none()); + } + + // ===== Disconnect Message Tests ===== + + #[test] + fn test_disconnect_encode_decode() { + let msg = Disconnect::new(DisconnectReason::Shutdown); + let encoded = msg.encode(); + + assert_eq!(encoded.len(), 2); + assert_eq!(encoded[0], 0x50); // LinkMessageType::Disconnect + assert_eq!(encoded[1], 0x00); // DisconnectReason::Shutdown + + // Decode from payload (after msg_type byte) + let decoded = Disconnect::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.reason, DisconnectReason::Shutdown); + } + + #[test] + fn test_disconnect_all_reasons() { + let reasons = [ + DisconnectReason::Shutdown, + DisconnectReason::Restart, + DisconnectReason::ProtocolError, + DisconnectReason::Other, + ]; + + for reason in reasons { + let msg = Disconnect::new(reason); + let encoded = msg.encode(); + let decoded = Disconnect::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.reason, reason); + } + } + + #[test] + fn test_disconnect_decode_empty_payload() { + let result = Disconnect::decode(&[]); + assert!(result.is_err()); + } + + #[test] + fn test_disconnect_decode_unknown_reason() { + // Unknown reason codes fall back to Other + let decoded = Disconnect::decode(&[0x80]).unwrap(); + assert_eq!(decoded.reason, DisconnectReason::Other); + } + // ===== SessionMessageType Tests ===== #[test]