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 fn record_resend(&mut self, next_resend_at_ms: u64) {
409 self.resend_count += 1;
410 self.next_resend_at_ms = next_resend_at_ms;
411 }
412
413 pub fn start_handshake(
420 &mut self,
421 our_keypair: Keypair,
422 epoch: [u8; 8],
423 current_time_ms: u64,
424 ) -> Result<Vec<u8>, NoiseError> {
425 if self.direction != LinkDirection::Outbound {
426 return Err(NoiseError::WrongState {
427 expected: "outbound connection".to_string(),
428 got: "inbound connection".to_string(),
429 });
430 }
431
432 if self.handshake_state != HandshakeState::Initial {
433 return Err(NoiseError::WrongState {
434 expected: "initial state".to_string(),
435 got: self.handshake_state.to_string(),
436 });
437 }
438
439 let remote_static = self
440 .expected_identity
441 .as_ref()
442 .expect("outbound must have expected identity")
443 .pubkey_full();
444
445 let mut hs = noise::HandshakeState::new_initiator(our_keypair, remote_static);
446 hs.set_local_epoch(epoch);
447 let msg1 = hs.write_message_1()?;
448
449 self.noise_handshake = Some(hs);
450 self.handshake_state = HandshakeState::SentMsg1;
451 self.last_activity = current_time_ms;
452
453 Ok(msg1)
454 }
455
456 pub fn receive_handshake_init(
461 &mut self,
462 our_keypair: Keypair,
463 epoch: [u8; 8],
464 message: &[u8],
465 current_time_ms: u64,
466 ) -> Result<Vec<u8>, NoiseError> {
467 if self.direction != LinkDirection::Inbound {
468 return Err(NoiseError::WrongState {
469 expected: "inbound connection".to_string(),
470 got: "outbound connection".to_string(),
471 });
472 }
473
474 if self.handshake_state != HandshakeState::Initial {
475 return Err(NoiseError::WrongState {
476 expected: "initial state".to_string(),
477 got: self.handshake_state.to_string(),
478 });
479 }
480
481 let mut hs = noise::HandshakeState::new_responder(our_keypair);
482 hs.set_local_epoch(epoch);
483
484 hs.read_message_1(message)?;
486
487 let remote_static = *hs
489 .remote_static()
490 .expect("remote static available after msg1");
491 self.expected_identity = Some(PeerIdentity::from_pubkey_full(remote_static));
492
493 self.remote_epoch = hs.remote_epoch();
495
496 let msg2 = hs.write_message_2()?;
498
499 let session = hs.into_session()?;
501 self.noise_session = Some(session);
502 self.handshake_state = HandshakeState::Complete;
503 self.last_activity = current_time_ms;
504
505 Ok(msg2)
506 }
507
508 pub fn complete_handshake(
512 &mut self,
513 message: &[u8],
514 current_time_ms: u64,
515 ) -> Result<(), NoiseError> {
516 if self.handshake_state != HandshakeState::SentMsg1 {
517 return Err(NoiseError::WrongState {
518 expected: "sent_msg1 state".to_string(),
519 got: self.handshake_state.to_string(),
520 });
521 }
522
523 let mut hs = self
524 .noise_handshake
525 .take()
526 .expect("noise handshake must exist in SentMsg1 state");
527
528 hs.read_message_2(message)?;
529
530 self.remote_epoch = hs.remote_epoch();
532
533 let session = hs.into_session()?;
534 self.noise_session = Some(session);
535 self.handshake_state = HandshakeState::Complete;
536 self.last_activity = current_time_ms;
537
538 Ok(())
539 }
540
541 pub fn take_session(&mut self) -> Option<NoiseSession> {
546 if self.handshake_state == HandshakeState::Complete {
547 self.noise_session.take()
548 } else {
549 None
550 }
551 }
552
553 pub fn has_session(&self) -> bool {
555 self.handshake_state == HandshakeState::Complete && self.noise_session.is_some()
556 }
557
558 pub fn mark_failed(&mut self) {
562 self.handshake_state = HandshakeState::Failed;
563 self.noise_handshake = None;
564 }
565
566 pub fn touch(&mut self, current_time_ms: u64) {
568 self.last_activity = current_time_ms;
569 }
570
571 pub fn is_timed_out(&self, current_time_ms: u64, timeout_ms: u64) -> bool {
575 self.idle_time(current_time_ms) > timeout_ms
576 }
577}
578
579impl fmt::Debug for PeerConnection {
580 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
581 f.debug_struct("PeerConnection")
582 .field("link_id", &self.link_id)
583 .field("direction", &self.direction)
584 .field("handshake_state", &self.handshake_state)
585 .field("expected_identity", &self.expected_identity)
586 .field("has_noise_handshake", &self.noise_handshake.is_some())
587 .field("has_noise_session", &self.noise_session.is_some())
588 .field("our_index", &self.our_index)
589 .field("their_index", &self.their_index)
590 .field("transport_id", &self.transport_id)
591 .field("started_at", &self.started_at)
592 .field("last_activity", &self.last_activity)
593 .finish()
594 }
595}
596
597#[cfg(test)]
598mod tests {
599 use super::*;
600 use crate::Identity;
601 use rand::Rng;
602
603 fn make_peer_identity() -> PeerIdentity {
604 let identity = Identity::generate();
605 PeerIdentity::from_pubkey(identity.pubkey())
606 }
607
608 fn make_keypair() -> Keypair {
609 let identity = Identity::generate();
610 identity.keypair()
611 }
612
613 fn make_epoch() -> [u8; 8] {
614 let mut epoch = [0u8; 8];
615 rand::rng().fill_bytes(&mut epoch);
616 epoch
617 }
618
619 #[test]
620 fn test_handshake_state_properties() {
621 assert!(HandshakeState::Initial.is_in_progress());
622 assert!(HandshakeState::SentMsg1.is_in_progress());
623 assert!(HandshakeState::ReceivedMsg1.is_in_progress());
624 assert!(!HandshakeState::Complete.is_in_progress());
625 assert!(!HandshakeState::Failed.is_in_progress());
626
627 assert!(HandshakeState::Complete.is_complete());
628 assert!(HandshakeState::Failed.is_failed());
629 }
630
631 #[test]
632 fn test_outbound_connection() {
633 let identity = make_peer_identity();
634 let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000);
635
636 assert!(conn.is_outbound());
637 assert!(!conn.is_inbound());
638 assert_eq!(conn.handshake_state(), HandshakeState::Initial);
639 assert!(conn.expected_identity().is_some());
640 assert_eq!(conn.started_at(), 1000);
641 }
642
643 #[test]
644 fn test_inbound_connection() {
645 let conn = PeerConnection::inbound(LinkId::new(2), 2000);
646
647 assert!(conn.is_inbound());
648 assert!(!conn.is_outbound());
649 assert_eq!(conn.handshake_state(), HandshakeState::Initial);
650 assert!(conn.expected_identity().is_none());
651 assert_eq!(conn.started_at(), 2000);
652 }
653
654 #[test]
655 fn test_full_handshake_flow() {
656 let initiator_identity = Identity::generate();
658 let responder_identity = Identity::generate();
659
660 let initiator_keypair = initiator_identity.keypair();
661 let responder_keypair = responder_identity.keypair();
662 let initiator_epoch = make_epoch();
663 let responder_epoch = make_epoch();
664
665 let responder_peer_id = PeerIdentity::from_pubkey_full(responder_identity.pubkey_full());
667
668 let mut initiator_conn = PeerConnection::outbound(LinkId::new(1), responder_peer_id, 1000);
670 let mut responder_conn = PeerConnection::inbound(LinkId::new(2), 1000);
671
672 let msg1 = initiator_conn
674 .start_handshake(initiator_keypair, initiator_epoch, 1100)
675 .unwrap();
676 assert_eq!(initiator_conn.handshake_state(), HandshakeState::SentMsg1);
677
678 let msg2 = responder_conn
680 .receive_handshake_init(responder_keypair, responder_epoch, &msg1, 1200)
681 .unwrap();
682 assert_eq!(responder_conn.handshake_state(), HandshakeState::Complete);
683
684 let discovered = responder_conn.expected_identity().unwrap();
686 assert_eq!(discovered.pubkey(), initiator_identity.pubkey());
687
688 assert_eq!(responder_conn.remote_epoch(), Some(initiator_epoch));
690
691 initiator_conn.complete_handshake(&msg2, 1300).unwrap();
693 assert_eq!(initiator_conn.handshake_state(), HandshakeState::Complete);
694
695 assert_eq!(initiator_conn.remote_epoch(), Some(responder_epoch));
697
698 assert!(initiator_conn.has_session());
700 assert!(responder_conn.has_session());
701
702 let mut init_session = initiator_conn.take_session().unwrap();
704 let mut resp_session = responder_conn.take_session().unwrap();
705
706 let plaintext = b"test message";
708 let ciphertext = init_session.encrypt(plaintext).unwrap();
709 let decrypted = resp_session.decrypt(&ciphertext).unwrap();
710 assert_eq!(decrypted, plaintext);
711 }
712
713 #[test]
714 fn test_connection_timing() {
715 let identity = make_peer_identity();
716 let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000);
717
718 assert_eq!(conn.duration(1500), 500);
719 assert_eq!(conn.idle_time(1500), 500);
720 assert!(!conn.is_timed_out(1500, 1000));
721 assert!(conn.is_timed_out(2500, 1000));
722 }
723
724 #[test]
725 fn test_connection_failure() {
726 let identity = make_peer_identity();
727 let mut conn = PeerConnection::outbound(LinkId::new(1), identity, 1000);
728
729 conn.mark_failed();
730 assert!(conn.is_failed());
731 assert!(!conn.is_in_progress());
732 assert!(!conn.is_complete());
733 }
734
735 #[test]
736 fn test_wrong_direction_errors() {
737 let identity = make_peer_identity();
738 let keypair = make_keypair();
739
740 let mut outbound = PeerConnection::outbound(LinkId::new(1), identity, 1000);
742 assert!(
743 outbound
744 .receive_handshake_init(keypair, make_epoch(), &[0u8; 106], 1100)
745 .is_err()
746 );
747
748 let mut inbound = PeerConnection::inbound(LinkId::new(2), 1000);
750 assert!(
751 inbound
752 .start_handshake(keypair, make_epoch(), 1100)
753 .is_err()
754 );
755 }
756}