1use crate::PeerIdentity;
8use crate::noise::{self, NoiseError, NoiseSession};
9use crate::transport::{LinkDirection, LinkId, LinkStats, TransportAddr, TransportId};
10use crate::utils::index::SessionIndex;
11use secp256k1::Keypair;
12use std::fmt;
13
14#[derive(Clone, Copy, Debug, PartialEq, Eq)]
20pub enum HandshakeState {
21 Initial,
23 SentMsg1,
25 ReceivedMsg1,
27 Complete,
29 Failed,
31}
32
33impl HandshakeState {
34 pub fn is_in_progress(&self) -> bool {
36 matches!(
37 self,
38 HandshakeState::Initial | HandshakeState::SentMsg1 | HandshakeState::ReceivedMsg1
39 )
40 }
41
42 pub fn is_complete(&self) -> bool {
44 matches!(self, HandshakeState::Complete)
45 }
46
47 pub fn is_failed(&self) -> bool {
49 matches!(self, HandshakeState::Failed)
50 }
51}
52
53impl fmt::Display for HandshakeState {
54 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
55 let s = match self {
56 HandshakeState::Initial => "initial",
57 HandshakeState::SentMsg1 => "sent_msg1",
58 HandshakeState::ReceivedMsg1 => "received_msg1",
59 HandshakeState::Complete => "complete",
60 HandshakeState::Failed => "failed",
61 };
62 write!(f, "{}", s)
63 }
64}
65
66pub struct PeerConnection {
71 link_id: LinkId,
74
75 direction: LinkDirection,
77
78 handshake_state: HandshakeState,
81
82 expected_identity: Option<PeerIdentity>,
85
86 noise_handshake: Option<noise::HandshakeState>,
88
89 noise_session: Option<NoiseSession>,
91
92 started_at: u64,
95
96 last_activity: u64,
98
99 link_stats: LinkStats,
102
103 our_index: Option<SessionIndex>,
108
109 their_index: Option<SessionIndex>,
113
114 transport_id: Option<TransportId>,
116
117 source_addr: Option<TransportAddr>,
119
120 preferred_send_addr: Option<TransportAddr>,
122
123 remote_epoch: Option<[u8; 8]>,
126
127 handshake_msg1: Option<Vec<u8>>,
130
131 handshake_msg2: Option<Vec<u8>>,
133
134 resend_count: u32,
136
137 next_resend_at_ms: u64,
139}
140
141impl PeerConnection {
142 pub fn outbound(
147 link_id: LinkId,
148 expected_identity: PeerIdentity,
149 current_time_ms: u64,
150 ) -> Self {
151 Self {
152 link_id,
153 direction: LinkDirection::Outbound,
154 handshake_state: HandshakeState::Initial,
155 expected_identity: Some(expected_identity),
156 noise_handshake: None,
157 noise_session: None,
158 started_at: current_time_ms,
159 last_activity: current_time_ms,
160
161 link_stats: LinkStats::new(),
162 our_index: None,
163 their_index: None,
164 transport_id: None,
165 source_addr: None,
166 preferred_send_addr: None,
167 remote_epoch: None,
168 handshake_msg1: None,
169 handshake_msg2: None,
170 resend_count: 0,
171 next_resend_at_ms: 0,
172 }
173 }
174
175 pub fn inbound(link_id: LinkId, current_time_ms: u64) -> Self {
180 Self {
181 link_id,
182 direction: LinkDirection::Inbound,
183 handshake_state: HandshakeState::Initial,
184 expected_identity: None,
185 noise_handshake: None,
186 noise_session: None,
187 started_at: current_time_ms,
188 last_activity: current_time_ms,
189
190 link_stats: LinkStats::new(),
191 our_index: None,
192 their_index: None,
193 transport_id: None,
194 source_addr: None,
195 preferred_send_addr: None,
196 remote_epoch: None,
197 handshake_msg1: None,
198 handshake_msg2: None,
199 resend_count: 0,
200 next_resend_at_ms: 0,
201 }
202 }
203
204 pub fn inbound_with_transport(
208 link_id: LinkId,
209 transport_id: TransportId,
210 source_addr: TransportAddr,
211 current_time_ms: u64,
212 ) -> Self {
213 Self {
214 link_id,
215 direction: LinkDirection::Inbound,
216 handshake_state: HandshakeState::Initial,
217 expected_identity: None,
218 noise_handshake: None,
219 noise_session: None,
220 started_at: current_time_ms,
221 last_activity: current_time_ms,
222
223 link_stats: LinkStats::new(),
224 our_index: None,
225 their_index: None,
226 transport_id: Some(transport_id),
227 source_addr: Some(source_addr),
228 preferred_send_addr: None,
229 remote_epoch: None,
230 handshake_msg1: None,
231 handshake_msg2: None,
232 resend_count: 0,
233 next_resend_at_ms: 0,
234 }
235 }
236
237 pub fn link_id(&self) -> LinkId {
241 self.link_id
242 }
243
244 pub fn direction(&self) -> LinkDirection {
246 self.direction
247 }
248
249 pub fn handshake_state(&self) -> HandshakeState {
251 self.handshake_state
252 }
253
254 pub fn expected_identity(&self) -> Option<&PeerIdentity> {
256 self.expected_identity.as_ref()
257 }
258
259 pub fn is_outbound(&self) -> bool {
261 self.direction == LinkDirection::Outbound
262 }
263
264 pub fn is_inbound(&self) -> bool {
266 self.direction == LinkDirection::Inbound
267 }
268
269 pub fn is_in_progress(&self) -> bool {
271 self.handshake_state.is_in_progress()
272 }
273
274 pub fn is_complete(&self) -> bool {
276 self.handshake_state.is_complete()
277 }
278
279 pub fn is_failed(&self) -> bool {
281 self.handshake_state.is_failed()
282 }
283
284 pub fn started_at(&self) -> u64 {
286 self.started_at
287 }
288
289 pub fn last_activity(&self) -> u64 {
291 self.last_activity
292 }
293
294 pub fn duration(&self, current_time_ms: u64) -> u64 {
296 current_time_ms.saturating_sub(self.started_at)
297 }
298
299 pub fn idle_time(&self, current_time_ms: u64) -> u64 {
301 current_time_ms.saturating_sub(self.last_activity)
302 }
303
304 pub fn link_stats(&self) -> &LinkStats {
306 &self.link_stats
307 }
308
309 pub fn link_stats_mut(&mut self) -> &mut LinkStats {
311 &mut self.link_stats
312 }
313
314 pub fn our_index(&self) -> Option<SessionIndex> {
318 self.our_index
319 }
320
321 pub fn set_our_index(&mut self, index: SessionIndex) {
323 self.our_index = Some(index);
324 }
325
326 pub fn their_index(&self) -> Option<SessionIndex> {
328 self.their_index
329 }
330
331 pub fn set_their_index(&mut self, index: SessionIndex) {
333 self.their_index = Some(index);
334 }
335
336 pub fn transport_id(&self) -> Option<TransportId> {
338 self.transport_id
339 }
340
341 pub fn set_transport_id(&mut self, id: TransportId) {
343 self.transport_id = Some(id);
344 }
345
346 pub fn source_addr(&self) -> Option<&TransportAddr> {
348 self.source_addr.as_ref()
349 }
350
351 pub fn set_source_addr(&mut self, addr: TransportAddr) {
353 self.source_addr = Some(addr);
354 }
355
356 pub fn preferred_send_addr(&self) -> Option<&TransportAddr> {
358 self.preferred_send_addr.as_ref()
359 }
360
361 pub fn set_preferred_send_addr(&mut self, addr: TransportAddr) {
363 self.preferred_send_addr = Some(addr);
364 }
365
366 pub fn remote_epoch(&self) -> Option<[u8; 8]> {
370 self.remote_epoch
371 }
372
373 pub fn set_handshake_msg1(&mut self, msg1: Vec<u8>, first_resend_at_ms: u64) {
377 self.handshake_msg1 = Some(msg1);
378 self.resend_count = 0;
379 self.next_resend_at_ms = first_resend_at_ms;
380 }
381
382 pub fn set_handshake_msg2(&mut self, msg2: Vec<u8>) {
384 self.handshake_msg2 = Some(msg2);
385 }
386
387 pub fn handshake_msg1(&self) -> Option<&[u8]> {
389 self.handshake_msg1.as_deref()
390 }
391
392 pub fn handshake_msg2(&self) -> Option<&[u8]> {
394 self.handshake_msg2.as_deref()
395 }
396
397 pub fn resend_count(&self) -> u32 {
399 self.resend_count
400 }
401
402 pub fn next_resend_at_ms(&self) -> u64 {
404 self.next_resend_at_ms
405 }
406
407 pub(crate) fn restart_handshake_resends(&mut self, now_ms: u64) {
414 if self.next_resend_at_ms > 0 {
415 self.resend_count = 0;
416 self.next_resend_at_ms = now_ms;
417 }
418 }
419
420 pub fn record_resend(&mut self, next_resend_at_ms: u64) {
422 self.resend_count += 1;
423 self.next_resend_at_ms = next_resend_at_ms;
424 }
425
426 pub fn start_handshake(
433 &mut self,
434 our_keypair: Keypair,
435 epoch: [u8; 8],
436 current_time_ms: u64,
437 ) -> Result<Vec<u8>, NoiseError> {
438 if self.direction != LinkDirection::Outbound {
439 return Err(NoiseError::WrongState {
440 expected: "outbound connection".to_string(),
441 got: "inbound connection".to_string(),
442 });
443 }
444
445 if self.handshake_state != HandshakeState::Initial {
446 return Err(NoiseError::WrongState {
447 expected: "initial state".to_string(),
448 got: self.handshake_state.to_string(),
449 });
450 }
451
452 let remote_static = self
453 .expected_identity
454 .as_ref()
455 .expect("outbound must have expected identity")
456 .pubkey_full();
457
458 let mut hs = noise::HandshakeState::new_initiator(our_keypair, remote_static);
459 hs.set_local_epoch(epoch);
460 let msg1 = hs.write_message_1()?;
461
462 self.noise_handshake = Some(hs);
463 self.handshake_state = HandshakeState::SentMsg1;
464 self.last_activity = current_time_ms;
465
466 Ok(msg1)
467 }
468
469 pub fn receive_handshake_init(
474 &mut self,
475 our_keypair: Keypair,
476 epoch: [u8; 8],
477 message: &[u8],
478 current_time_ms: u64,
479 ) -> Result<Vec<u8>, NoiseError> {
480 if self.direction != LinkDirection::Inbound {
481 return Err(NoiseError::WrongState {
482 expected: "inbound connection".to_string(),
483 got: "outbound connection".to_string(),
484 });
485 }
486
487 if self.handshake_state != HandshakeState::Initial {
488 return Err(NoiseError::WrongState {
489 expected: "initial state".to_string(),
490 got: self.handshake_state.to_string(),
491 });
492 }
493
494 let mut hs = noise::HandshakeState::new_responder(our_keypair);
495 hs.set_local_epoch(epoch);
496
497 hs.read_message_1(message)?;
499
500 let remote_static = *hs
502 .remote_static()
503 .expect("remote static available after msg1");
504 self.expected_identity = Some(PeerIdentity::from_pubkey_full(remote_static));
505
506 self.remote_epoch = hs.remote_epoch();
508
509 let msg2 = hs.write_message_2()?;
511
512 let session = hs.into_session()?;
514 self.noise_session = Some(session);
515 self.handshake_state = HandshakeState::Complete;
516 self.last_activity = current_time_ms;
517
518 Ok(msg2)
519 }
520
521 pub fn complete_handshake(
525 &mut self,
526 message: &[u8],
527 current_time_ms: u64,
528 ) -> Result<(), NoiseError> {
529 if self.handshake_state != HandshakeState::SentMsg1 {
530 return Err(NoiseError::WrongState {
531 expected: "sent_msg1 state".to_string(),
532 got: self.handshake_state.to_string(),
533 });
534 }
535
536 let mut hs = self
537 .noise_handshake
538 .take()
539 .expect("noise handshake must exist in SentMsg1 state");
540
541 hs.read_message_2(message)?;
542
543 self.remote_epoch = hs.remote_epoch();
545
546 let session = hs.into_session()?;
547 self.noise_session = Some(session);
548 self.handshake_state = HandshakeState::Complete;
549 self.last_activity = current_time_ms;
550
551 Ok(())
552 }
553
554 pub fn take_session(&mut self) -> Option<NoiseSession> {
559 if self.handshake_state == HandshakeState::Complete {
560 self.noise_session.take()
561 } else {
562 None
563 }
564 }
565
566 pub fn has_session(&self) -> bool {
568 self.handshake_state == HandshakeState::Complete && self.noise_session.is_some()
569 }
570
571 pub fn mark_failed(&mut self) {
575 self.handshake_state = HandshakeState::Failed;
576 self.noise_handshake = None;
577 }
578
579 pub fn touch(&mut self, current_time_ms: u64) {
581 self.last_activity = current_time_ms;
582 }
583
584 pub fn is_timed_out(&self, current_time_ms: u64, timeout_ms: u64) -> bool {
588 self.idle_time(current_time_ms) > timeout_ms
589 }
590}
591
592impl fmt::Debug for PeerConnection {
593 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
594 f.debug_struct("PeerConnection")
595 .field("link_id", &self.link_id)
596 .field("direction", &self.direction)
597 .field("handshake_state", &self.handshake_state)
598 .field("expected_identity", &self.expected_identity)
599 .field("has_noise_handshake", &self.noise_handshake.is_some())
600 .field("has_noise_session", &self.noise_session.is_some())
601 .field("our_index", &self.our_index)
602 .field("their_index", &self.their_index)
603 .field("transport_id", &self.transport_id)
604 .field("started_at", &self.started_at)
605 .field("last_activity", &self.last_activity)
606 .finish()
607 }
608}
609
610#[cfg(test)]
611mod tests {
612 use super::*;
613 use crate::Identity;
614 use rand::Rng;
615
616 fn make_peer_identity() -> PeerIdentity {
617 let identity = Identity::generate();
618 PeerIdentity::from_pubkey(identity.pubkey())
619 }
620
621 fn make_keypair() -> Keypair {
622 let identity = Identity::generate();
623 identity.keypair()
624 }
625
626 fn make_epoch() -> [u8; 8] {
627 let mut epoch = [0u8; 8];
628 rand::rng().fill_bytes(&mut epoch);
629 epoch
630 }
631
632 #[test]
633 fn test_handshake_state_properties() {
634 assert!(HandshakeState::Initial.is_in_progress());
635 assert!(HandshakeState::SentMsg1.is_in_progress());
636 assert!(HandshakeState::ReceivedMsg1.is_in_progress());
637 assert!(!HandshakeState::Complete.is_in_progress());
638 assert!(!HandshakeState::Failed.is_in_progress());
639
640 assert!(HandshakeState::Complete.is_complete());
641 assert!(HandshakeState::Failed.is_failed());
642 }
643
644 #[test]
645 fn test_outbound_connection() {
646 let identity = make_peer_identity();
647 let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000);
648
649 assert!(conn.is_outbound());
650 assert!(!conn.is_inbound());
651 assert_eq!(conn.handshake_state(), HandshakeState::Initial);
652 assert!(conn.expected_identity().is_some());
653 assert_eq!(conn.started_at(), 1000);
654 }
655
656 #[test]
657 fn test_inbound_connection() {
658 let conn = PeerConnection::inbound(LinkId::new(2), 2000);
659
660 assert!(conn.is_inbound());
661 assert!(!conn.is_outbound());
662 assert_eq!(conn.handshake_state(), HandshakeState::Initial);
663 assert!(conn.expected_identity().is_none());
664 assert_eq!(conn.started_at(), 2000);
665 }
666
667 #[test]
668 fn test_full_handshake_flow() {
669 let initiator_identity = Identity::generate();
671 let responder_identity = Identity::generate();
672
673 let initiator_keypair = initiator_identity.keypair();
674 let responder_keypair = responder_identity.keypair();
675 let initiator_epoch = make_epoch();
676 let responder_epoch = make_epoch();
677
678 let responder_peer_id = PeerIdentity::from_pubkey_full(responder_identity.pubkey_full());
680
681 let mut initiator_conn = PeerConnection::outbound(LinkId::new(1), responder_peer_id, 1000);
683 let mut responder_conn = PeerConnection::inbound(LinkId::new(2), 1000);
684
685 let msg1 = initiator_conn
687 .start_handshake(initiator_keypair, initiator_epoch, 1100)
688 .unwrap();
689 assert_eq!(initiator_conn.handshake_state(), HandshakeState::SentMsg1);
690
691 let msg2 = responder_conn
693 .receive_handshake_init(responder_keypair, responder_epoch, &msg1, 1200)
694 .unwrap();
695 assert_eq!(responder_conn.handshake_state(), HandshakeState::Complete);
696
697 let discovered = responder_conn.expected_identity().unwrap();
699 assert_eq!(discovered.pubkey(), initiator_identity.pubkey());
700
701 assert_eq!(responder_conn.remote_epoch(), Some(initiator_epoch));
703
704 initiator_conn.complete_handshake(&msg2, 1300).unwrap();
706 assert_eq!(initiator_conn.handshake_state(), HandshakeState::Complete);
707
708 assert_eq!(initiator_conn.remote_epoch(), Some(responder_epoch));
710
711 assert!(initiator_conn.has_session());
713 assert!(responder_conn.has_session());
714
715 let mut init_session = initiator_conn.take_session().unwrap();
717 let mut resp_session = responder_conn.take_session().unwrap();
718
719 let plaintext = b"test message";
721 let ciphertext = init_session.encrypt(plaintext).unwrap();
722 let decrypted = resp_session.decrypt(&ciphertext).unwrap();
723 assert_eq!(decrypted, plaintext);
724 }
725
726 #[test]
727 fn test_connection_timing() {
728 let identity = make_peer_identity();
729 let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000);
730
731 assert_eq!(conn.duration(1500), 500);
732 assert_eq!(conn.idle_time(1500), 500);
733 assert!(!conn.is_timed_out(1500, 1000));
734 assert!(conn.is_timed_out(2500, 1000));
735 }
736
737 #[test]
738 fn test_connection_failure() {
739 let identity = make_peer_identity();
740 let mut conn = PeerConnection::outbound(LinkId::new(1), identity, 1000);
741
742 conn.mark_failed();
743 assert!(conn.is_failed());
744 assert!(!conn.is_in_progress());
745 assert!(!conn.is_complete());
746 }
747
748 #[test]
749 fn test_wrong_direction_errors() {
750 let identity = make_peer_identity();
751 let keypair = make_keypair();
752
753 let mut outbound = PeerConnection::outbound(LinkId::new(1), identity, 1000);
755 assert!(
756 outbound
757 .receive_handshake_init(keypair, make_epoch(), &[0u8; 106], 1100)
758 .is_err()
759 );
760
761 let mut inbound = PeerConnection::inbound(LinkId::new(2), 1000);
763 assert!(
764 inbound
765 .start_handshake(keypair, make_epoch(), 1100)
766 .is_err()
767 );
768 }
769}