From f0c7ea571cb5407f78c2c374e74fdfd2e2b32c56 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Fri, 29 Sep 2023 10:08:05 -0400 Subject: [PATCH] added a more flexible trait to handle sending --- performance/examples/basic_test.rs | 31 ++++++------ performance/examples/benchmark.rs | 33 ++++++------ performance/src/application.rs | 81 +++++++++++++++++++++++++----- performance/src/result.rs | 2 +- performance/src/zeta.rs | 26 +++++----- performance/src/zssp.rs | 42 ++++++++-------- 6 files changed, 136 insertions(+), 79 deletions(-) diff --git a/performance/examples/basic_test.rs b/performance/examples/basic_test.rs index 4bebc80..f7e2934 100644 --- a/performance/examples/basic_test.rs +++ b/performance/examples/basic_test.rs @@ -16,7 +16,6 @@ use zssp::application::{ use zssp::crypto::P384KeyPair; use zssp::crypto_impl::*; use zssp::result::ReceiveError; -use zssp::Session; const TEST_MTU: usize = 1500; @@ -26,6 +25,8 @@ struct TestApplication { ratchets: Mutex, } +type Session = zssp::Session; + struct Ratchets { rf_map: HashMap<[u8; RATCHET_SIZE], RatchetState>, peer_map: HashMap, @@ -64,9 +65,7 @@ impl CryptoLayer for TestApplication { type IncomingPacketBuffer = Vec; } #[allow(unused)] -impl ApplicationLayer for &TestApplication { - type Crypto = TestApplication; - +impl ApplicationLayer for &TestApplication { fn incoming_session(&mut self) -> IncomingSessionAction { IncomingSessionAction::Challenge } @@ -75,7 +74,7 @@ impl ApplicationLayer for &TestApplication { false } - fn initiator_disallows_downgrade(&mut self, session: &Arc>) -> bool { + fn initiator_disallows_downgrade(&mut self, session: &Arc) -> bool { true } @@ -162,7 +161,7 @@ fn alice_main( up = false; let result = context.open( alice_app, - |b| alice_out.send(b.to_vec()).is_ok(), + |b: &mut [u8]| alice_out.send(b.to_vec()).is_ok(), TEST_MTU, bob_pubkey.clone(), 0, @@ -181,9 +180,9 @@ fn alice_main( let mut output_data = Vec::new(); match context.receive( alice_app, - |b| alice_out.send(b.to_vec()).is_ok(), + |b: &mut [u8]| alice_out.send(b.to_vec()).is_ok(), TEST_MTU, - |_| Some((|b: &mut [u8]| alice_out.send(b.to_vec()).is_ok(), TEST_MTU)), + |_: &Arc| Some((|b: &mut [u8]| alice_out.send(b.to_vec()).is_ok(), TEST_MTU)), &0, pkt, &mut output_data, @@ -191,7 +190,7 @@ fn alice_main( Ok((Unassociated, _)) => { //println!("[alice] ok"); } - Ok((Session(_, event), _)) => match event { + Ok((SessionEvent(_, event), _)) => match event { Established => { up = true; } @@ -221,7 +220,7 @@ fn alice_main( context .send( alice_session.as_ref().unwrap(), - |b| alice_out.send(b.to_vec()).is_ok(), + |b: &mut [u8]| alice_out.send(b.to_vec()).is_ok(), &mut [0u8; TEST_MTU], &test_data[..1400 + ((OsRng.next_u64() as usize) % (test_data.len() - 1400))], ) @@ -236,7 +235,7 @@ fn alice_main( if current_time >= next_service { next_service = current_time - + context.service(alice_app, |_| { + + context.service(alice_app, |_: &Arc| { Some((|b: &mut [u8]| alice_out.send(b.to_vec()).is_ok(), TEST_MTU)) }); } @@ -272,15 +271,15 @@ fn bob_main( let mut output_data = Vec::new(); match context.receive( bob_app, - |b| bob_out.send(b.to_vec()).is_ok(), + |b: &mut [u8]| bob_out.send(b.to_vec()).is_ok(), TEST_MTU, - |_| Some((|b: &mut [u8]| bob_out.send(b.to_vec()).is_ok(), TEST_MTU)), + |_: &Arc| Some((|b: &mut [u8]| bob_out.send(b.to_vec()).is_ok(), TEST_MTU)), &0, pkt, &mut output_data, ) { Ok((Unassociated, _)) => {} - Ok((Session(s, event), _)) => match event { + Ok((SessionEvent(s, event), _)) => match event { NewSession | NewDowngradedSession => { println!("[bob] new session, took {}s", current_time as f32 / 1000.0); let _ = bob_session.replace(s); @@ -292,7 +291,7 @@ fn bob_main( context .send( &s, - |b| bob_out.send(b.to_vec()).is_ok(), + |b: &mut [u8]| bob_out.send(b.to_vec()).is_ok(), &mut [0u8; TEST_MTU], &output_data, ) @@ -325,7 +324,7 @@ fn bob_main( if current_time >= next_service { next_service = current_time - + context.service(bob_app, |_| { + + context.service(bob_app, |_: &Arc| { Some((|b: &mut [u8]| bob_out.send(b.to_vec()).is_ok(), TEST_MTU)) }); } diff --git a/performance/examples/benchmark.rs b/performance/examples/benchmark.rs index b86cc05..bbdf954 100644 --- a/performance/examples/benchmark.rs +++ b/performance/examples/benchmark.rs @@ -13,7 +13,6 @@ use zssp::application::{ use zssp::crypto::P384KeyPair; use zssp::crypto_impl::*; use zssp::result::ReceiveError; -use zssp::Session; const TEST_MTU: usize = 1500; @@ -56,10 +55,11 @@ impl DefaultCrypto for TestApplication { type SessionData = (); type IncomingPacketBuffer = PooledVec; } -#[allow(unused)] -impl ApplicationLayer for &TestApplication { - type Crypto = TestApplication; +type Session = zssp::Session; + +#[allow(unused)] +impl ApplicationLayer for &TestApplication { fn incoming_session(&mut self) -> IncomingSessionAction { IncomingSessionAction::Allow } @@ -68,7 +68,7 @@ impl ApplicationLayer for &TestApplication { false } - fn initiator_disallows_downgrade(&mut self, session: &Arc>) -> bool { + fn initiator_disallows_downgrade(&mut self, session: &Arc) -> bool { false } @@ -113,6 +113,7 @@ impl ApplicationLayer for &TestApplication { } } + #[allow(unused)] fn alice_main( run: &AtomicBool, @@ -131,7 +132,7 @@ fn alice_main( let result = context.open( alice_app, - |b| alice_out.send(alloc(b)).is_ok(), + |b: &mut [u8]| alice_out.send(alloc(b)).is_ok(), TEST_MTU, bob_pubkey.clone(), (), @@ -149,9 +150,9 @@ fn alice_main( output_data.clear(); match context.receive( alice_app, - |b| alice_out.send(alloc(b)).is_ok(), + |b: &mut [u8]| alice_out.send(alloc(b)).is_ok(), TEST_MTU, - |_| Some((|b: &mut [u8]| alice_out.send(alloc(b)).is_ok(), TEST_MTU)), + |_: &Arc>| Some((|b: &mut [u8]| alice_out.send(alloc(b)).is_ok(), TEST_MTU)), &0, pkt, &mut output_data, @@ -159,7 +160,7 @@ fn alice_main( Ok((Unassociated, _)) => { //println!("[alice] ok"); } - Ok((Session(_, event), _)) => match event { + Ok((SessionEvent(_, event), _)) => match event { Established => { up = true; } @@ -186,7 +187,7 @@ fn alice_main( context .send( alice_session.as_ref().unwrap(), - |b| alice_out.send(alloc(b)).is_ok(), + |b: &mut [u8]| alice_out.send(alloc(b)).is_ok(), &mut [0u8; TEST_MTU], &test_data[..1400 + ((OsRng.next_u64() as usize) % (test_data.len() - 1400))], ) @@ -197,7 +198,7 @@ fn alice_main( if current_time >= next_service { next_service = current_time - + context.service(alice_app, |_| { + + context.service(alice_app, |_: &Arc| { Some((|b: &mut [u8]| alice_out.send(alloc(b)).is_ok(), TEST_MTU)) }); } @@ -231,15 +232,15 @@ fn bob_main( output_data.clear(); match context.receive( bob_app, - |b| bob_out.send(alloc(b)).is_ok(), + |b: &mut [u8]| bob_out.send(alloc(b)).is_ok(), TEST_MTU, - |_| Some((|b: &mut [u8]| bob_out.send(alloc(b)).is_ok(), TEST_MTU)), + |_: &Arc| Some((|b: &mut [u8]| bob_out.send(alloc(b)).is_ok(), TEST_MTU)), &0, pkt, &mut output_data, ) { Ok((Unassociated, _)) => {} - Ok((Session(s, event), _)) => match event { + Ok((SessionEvent(s, event), _)) => match event { NewSession | NewDowngradedSession => { println!("[bob] new session, took {}s", current_time as f32 / 1000.0); let _ = bob_session.replace(s); @@ -251,7 +252,7 @@ fn bob_main( context .send( &s, - |b| bob_out.send(alloc(b)).is_ok(), + |b: &mut [u8]| bob_out.send(alloc(b)).is_ok(), &mut [0u8; TEST_MTU], &output_data, ) @@ -281,7 +282,7 @@ fn bob_main( if current_time >= next_service { next_service = current_time - + context.service(bob_app, |_| { + + context.service(bob_app, |_: &Arc| { Some((|b: &mut [u8]| bob_out.send(alloc(b)).is_ok(), TEST_MTU)) }); } diff --git a/performance/src/application.rs b/performance/src/application.rs index 86635b7..3487a18 100644 --- a/performance/src/application.rs +++ b/performance/src/application.rs @@ -142,16 +142,20 @@ pub trait CryptoLayer: Sized { type IncomingPacketBuffer: AsRef<[u8]> + AsMut<[u8]>; } -pub trait ApplicationLayer: Sized { - /// Specifies which concrete set of cryptography types will be used by this application. - type Crypto: CryptoLayer; - +pub trait ApplicationLayer: Sized { /// Should return the current time in milliseconds. Does not have to be monotonic, nor synced /// with remote peers (although both of these properties would help reliability slightly). /// Used to determine if any current handshakes should be resent or timed-out, or if a session /// should rekey. fn time(&mut self) -> i64; + /// This function will be called immediately after an anonymous Hello packet is received by Bob. + /// + /// Since the remote peer is anonymous at this stage of the handshake, this function is not + /// good for performing authentication and access control. Instead it should be used to mitigate + /// DDOS attacks by configuring it to return `Challenge` or `Drop` in response to an attacker's + /// Hello packet. If DDOS mitigation is not needed, this function can just be a single line that + /// returns `Allow`. fn incoming_session(&mut self) -> IncomingSessionAction; /// This function will be called whenever Alice's initial Hello packet contains the empty ratchet /// fingerprint. Brand new peers will always connect to Bob with the empty ratchet, but from @@ -178,7 +182,7 @@ pub trait ApplicationLayer: Sized { /// least one party is misconfigured and got their ratchet keys corrupted or lost, or Bob has /// been compromised and is being impersonated. An attacker must at least have Bob's private /// static key to be able to ask Alice to downgrade. - fn initiator_disallows_downgrade(&mut self, session: &Arc>) -> bool; + fn initiator_disallows_downgrade(&mut self, session: &Arc>) -> bool; /// Function to accept sessions after final negotiation. /// The second argument is the identity that the remote peer sent us. The application /// must verify this identity is associated with the remote peer's static key. @@ -187,9 +191,9 @@ pub trait ApplicationLayer: Sized { /// before returning. fn check_accept_session( &mut self, - remote_static_key: &::PublicKey, + remote_static_key: &Crypto::PublicKey, identity: &[u8], - ) -> AcceptAction; + ) -> AcceptAction; /// Lookup a specific ratchet state based on its ratchet fingerprint. /// This function will be called whenever Alice attempts to connect to us with a non-empty @@ -217,8 +221,8 @@ pub trait ApplicationLayer: Sized { /// function `ApplicationLayer::check_accept_session`. fn restore_by_identity( &mut self, - remote_static_key: &::PublicKey, - session_data: &::SessionData, + remote_static_key: &Crypto::PublicKey, + session_data: &Crypto::SessionData, ) -> Result, std::io::Error>; /// Atomically commit the update specified by `update_data` to storage, or return an error if /// the update could not be made. @@ -238,8 +242,8 @@ pub trait ApplicationLayer: Sized { /// Otherwise, when we restart, we will not be allowed to reconnect. fn save_ratchet_state( &mut self, - remote_static_key: &::PublicKey, - session_data: &::SessionData, + remote_static_key: &Crypto::PublicKey, + session_data: &Crypto::SessionData, update_data: RatchetUpdate<'_>, ) -> Result<(), std::io::Error>; @@ -248,13 +252,26 @@ pub trait ApplicationLayer: Sized { /// nothing else. Do not base protocol-level decisions upon the events passed to this function. #[cfg(feature = "logging")] #[allow(unused)] - fn event_log(&mut self, event: crate::LogEvent<'_, Self::Crypto>) {} + fn event_log(&mut self, event: crate::LogEvent<'_, Crypto>) {} } +/// Possible responses that can be made to Hello packets from an anonymous peer. #[derive(Debug, PartialEq, Eq, Clone)] pub enum IncomingSessionAction { + /// Allow the anonymous peer to continue connecting. + /// + /// In a later step, once forward secrecy is established, the peer will be forced to reveal + /// their identity. Allow, + /// Challenge the anonymous peer to complete a proof of work and IP/Address ownership before + /// they are allowed to consume our CPU resources to process their Hello packet. + /// + /// The challenge will only be effective if `Challenge` is consistently returned in response to + /// new Hello packets from the same peer or set of peers. + /// + /// If they complete the challenge they will be allowed to continue connecting. Challenge, + /// Drop the anonymous peer's Hello packet, preventing them from connecting. Drop, } @@ -276,3 +293,43 @@ pub struct AcceptAction { /// authentication checks. pub responder_silently_rejects: bool, } + +/// A trait to genericize the process of repeatedly sending packet fragments on some socket or +/// network interface. +/// +/// Is implemented by `FnMut(&mut [u8]) -> bool` closures. +pub trait Sender { + /// Send the given fragment on this interface and then return whether or not an error occured. + /// + /// If `true` is returned then sending is cancelled and this instance of `Sender` is dropped. + fn send_frag<'a>(&'a mut self, frag: &mut [u8]) -> bool; +} + +/// A trait to genericize the process of borrowing the resources necessary to repeatedly +/// send packet fragments on some socket or network interface. +/// +/// Is implemented by `FnMut(&Arc>) -> Option<(Sender, usize)>` closures. +pub trait SendTo { + /// Attempt to process and borrow the resources necessary to repeatedly send fragments of a + /// packet to the given session. + /// + /// If no error occurs this function should return a `Sender` instance configured to send to the + /// remote peer specified by `session`. It should also return the MTU of this link. This MTU can + /// be `usize::MAX`, in which case the packet is not fragmented and the `Sender` instance is + /// only called once. + /// + /// If `None` is returned then sending to this session is cancelled. + fn init_send<'a>(&'a mut self, session: &'a Arc>) -> Option<(S, usize)>; +} + +impl bool> Sender for F { + fn send_frag<'a>(&'a mut self, frag: &mut [u8]) -> bool { + self(frag) + } +} + +impl>) -> Option<(S, usize)>, S: Sender> SendTo for F { + fn init_send<'a>(&'a mut self, session: &'a Arc>) -> Option<(S, usize)> { + self(session) + } +} diff --git a/performance/src/result.rs b/performance/src/result.rs index a29bc86..f614238 100644 --- a/performance/src/result.rs +++ b/performance/src/result.rs @@ -191,7 +191,7 @@ pub enum ReceiveOk { /// or if it was a control packet that does not go through full Noise authentication. Unassociated, /// Packet was authentic and belongs to this specific session. - Session(Arc>, SessionEvent), + SessionEvent(Arc>, SessionEvent), } /// Something that can occur to an associated session when a packet is received successfully, /// including receiving a payload of decrypted, authenticated data. diff --git a/performance/src/zeta.rs b/performance/src/zeta.rs index d0d9fd5..1a0ffcd 100644 --- a/performance/src/zeta.rs +++ b/performance/src/zeta.rs @@ -343,7 +343,7 @@ fn create_a1_state( Some(Box::new(StateA1 { noise, e_secret, e1_secret, identity, x1 })) } /// Corresponds to Transition Algorithm 1 found in Section 4.3. -pub(crate) fn trans_to_a1>( +pub(crate) fn trans_to_a1>( mut app: App, ctx: &Arc>, s_remote: Crypto::PublicKey, @@ -443,7 +443,7 @@ pub(crate) fn respond_to_challenge( } } /// Corresponds to Transition Algorithm 2 found in Section 4.3. -pub(crate) fn received_x1_trans>( +pub(crate) fn received_x1_trans>( app: &mut App, ctx: &ContextInner, hash: &mut Crypto::Hash, @@ -595,7 +595,7 @@ pub(crate) fn received_x1_trans>( +pub(crate) fn received_x2_trans>( app: &mut App, ctx: &Arc>, session: &Arc>, @@ -817,7 +817,7 @@ fn send_control( } } /// Corresponds to Transition Algorithm 4 found in Section 4.3. -pub(crate) fn received_x3_trans>( +pub(crate) fn received_x3_trans>( app: &mut App, ctx: &Arc>, zeta: Arc>, @@ -994,7 +994,7 @@ pub(crate) fn received_x3_trans>( +pub(crate) fn received_c1_trans>( app: &mut App, ctx: &Arc>, session: &Arc>, @@ -1082,7 +1082,7 @@ pub(crate) fn received_c1_trans>( +pub(crate) fn received_c2_trans>( app: &mut App, ctx: &Arc>, session: &Arc>, @@ -1169,7 +1169,7 @@ pub(crate) fn received_d_trans( Ok(()) } // Corresponds to the timeout timer Transition Algorithm described in Section 4.1 - Definition 3. -fn timeout_trans>( +fn timeout_trans>( app: &mut App, ctx: &Arc>, session: &Arc>, @@ -1294,7 +1294,7 @@ fn timeout_trans>( } } /// Corresponds to the timer rules of the Zeta State Machine found in Section 4.1 - Definition 3. -pub(crate) fn process_timers>( +pub(crate) fn process_timers>( app: &mut App, ctx: &Arc>, session: &Arc>, @@ -1349,7 +1349,7 @@ pub(crate) fn process_timers>( +pub(crate) fn received_k1_trans>( app: &mut App, ctx: &Arc>, session: &Arc>, @@ -1505,7 +1505,7 @@ pub(crate) fn received_k1_trans>( +pub(crate) fn received_k2_trans>( app: &mut App, ctx: &Arc>, session: &Arc>, @@ -1638,7 +1638,7 @@ pub(crate) fn send_payload( ctx: &Arc>, session: &Session, payload: &[u8], - mut send: impl FnMut(&mut [u8]) -> bool, + mut send: impl Sender, mtu_sized_buffer: &mut [u8], ) -> Result { use SendError::*; @@ -1685,7 +1685,7 @@ pub(crate) fn send_payload( let header_auth = &mut mtu_sized_buffer[HEADER_AUTH_START..HEADER_AUTH_END]; state.hk_send.encrypt_in_place(header_auth.try_into().unwrap()); - if !send(&mut mtu_sized_buffer[..HEADER_SIZE + fragment_len]) { + if !send.send_frag(&mut mtu_sized_buffer[..HEADER_SIZE + fragment_len]) { return Ok(false); } i = j; @@ -1704,7 +1704,7 @@ pub(crate) fn send_payload( let header_auth = &mut mtu_sized_buffer[HEADER_AUTH_START..HEADER_AUTH_END]; state.hk_send.encrypt_in_place(header_auth.try_into().unwrap()); - if !send(&mut mtu_sized_buffer[..HEADER_SIZE + fragment_len]) { + if !send.send_frag(&mut mtu_sized_buffer[..HEADER_SIZE + fragment_len]) { return Ok(false); } diff --git a/performance/src/zssp.rs b/performance/src/zssp.rs index bc4b8bd..9cdda0c 100644 --- a/performance/src/zssp.rs +++ b/performance/src/zssp.rs @@ -82,7 +82,7 @@ fn parse_fragment_header(incoming_fragment: &[u8]) -> Result<(usize, usize, [u8; /// /// Corresponds to the fragmentation algorithm described in Section 6. fn send_with_fragmentation( - mut send: impl FnMut(&mut [u8]) -> bool, + mut send: impl Sender, mtu: usize, headered_packet: &mut [u8], hk_send: Option<&PrpEnc>, @@ -108,7 +108,7 @@ fn send_with_fragmentation( if let Some(hk_send) = hk_send { hk_send.encrypt_in_place((&mut fragment[HEADER_AUTH_START..HEADER_AUTH_END]).try_into().unwrap()); } - if !send(fragment) { + if !send.send_frag(fragment) { return false; } i = j; @@ -151,10 +151,10 @@ impl Context { /// object /// * `identity` - Payload to be sent to Bob that contains the information necessary /// for the upper protocol to authenticate and approve of Alice's identity. - pub fn open>( + pub fn open>( &self, app: App, - send: impl FnMut(&mut [u8]) -> bool, + send: impl Sender, mut mtu: usize, static_remote_key: Crypto::PublicKey, session_data: Crypto::SessionData, @@ -190,12 +190,12 @@ impl Context { /// * `remote_address` - Whatever the remote address is, as long as you can Hash it /// * `incoming_fragment_buf` - Buffer containing incoming wire packet (the context takes ownership) /// * `output_buffer` - Buffer to receive decrypted and authenticated object data - pub fn receive<'a, App: ApplicationLayer, SendFn: FnMut(&mut [u8]) -> bool>( + pub fn receive<'a, App: ApplicationLayer, S: Sender>( &self, mut app: App, - mut send_unassociated_reply: impl FnMut(&mut [u8]) -> bool, + mut send_unassociated_reply: impl Sender, mut send_unassociated_mtu: usize, - mut send_to: impl FnMut(&Arc>) -> Option<(SendFn, usize)>, + mut send_to: impl SendTo, remote_address: &impl Hash, mut incoming_fragment_buf: Crypto::IncomingPacketBuffer, output_buffer: impl Write, @@ -317,9 +317,9 @@ impl Context { }; let send_associated = |packet: &mut [u8], hk_send: Option<&Crypto::PrpEnc>| { - if let Some((send_fragment, mut mtu)) = send_to(&session) { + if let Some((sender, mut mtu)) = send_to.init_send(&session) { mtu = mtu.max(MIN_TRANSPORT_MTU); - send_with_fragmentation(send_fragment, mtu, packet, hk_send); + send_with_fragmentation(sender, mtu, packet, hk_send); } }; match packet_type { @@ -403,7 +403,7 @@ impl Context { _ => return Err(fault!(InvalidPacket, true)), // This is unreachable. } }; - Ok((ReceiveOk::Session(session, ret.0), ret.1)) + Ok((ReceiveOk::SessionEvent(session, ret.0), ret.1)) } else { // Check for and handle PACKET_TYPE_ALICE_NOISE_XK_PATTERN_3 let zeta = self.0.unassociated_handshake_states.get(kid_recv); @@ -463,7 +463,7 @@ impl Context { })?; log!(app, X3IsAuthSentKeyConfirm(&session)); Ok(( - ReceiveOk::Session( + ReceiveOk::SessionEvent( session, if should_warn_missing_ratchet { SessionEvent::NewDowngradedSession @@ -552,7 +552,7 @@ impl Context { .copy_from_slice(&nonce[..PACKET_NONCE_SIZE]); set_header(&mut challenge_packet, 0, &nonce); - send_unassociated_reply(&mut challenge_packet); + send_unassociated_reply.send_frag(&mut challenge_packet); // If we issue a challenge the first hello packet will always fail. return Err(fault!(FailedAuth, false)); } else { @@ -611,7 +611,7 @@ impl Context { pub fn send( &self, session: &Session, - send: impl FnMut(&mut [u8]) -> bool, + send: impl Sender, mtu_sized_buffer: &mut [u8], data: &[u8], ) -> Result { @@ -625,10 +625,10 @@ impl Context { /// /// * `app` - Interface to application using ZSSP /// * `send_to` - Function to get a sender and an MTU to send something over an active session - pub fn service, SendFn: FnMut(&mut [u8]) -> bool>( + pub fn service, S: Sender>( &self, mut app: App, - send_to: impl FnMut(&Arc>) -> Option<(SendFn, usize)>, + send_to: impl SendTo, ) -> i64 { let current_time = app.time(); let next_service_time = self.service_inner(app, send_to, current_time); @@ -649,18 +649,18 @@ impl Context { /// /// * `app` - Interface to application using ZSSP /// * `send_to` - Function to get a sender and an MTU to send something over an active session - pub fn service_scheduled, SendFn: FnMut(&mut [u8]) -> bool>( + pub fn service_scheduled, S: Sender>( &self, mut app: App, - send_to: impl FnMut(&Arc>) -> Option<(SendFn, usize)>, + send_to: impl SendTo ) -> i64 { let current_time = app.time(); self.service_inner(app, send_to, current_time) } - fn service_inner, SendFn: FnMut(&mut [u8]) -> bool>( + fn service_inner, S: Sender>( &self, mut app: App, - mut send_to: impl FnMut(&Arc>) -> Option<(SendFn, usize)>, + mut send_to: impl SendTo, current_time: i64, ) -> i64 { let ctx = &self.0; @@ -682,9 +682,9 @@ impl Context { } }; let result = process_timers(&mut app, ctx, &session, current_time, |packet, hk_send| { - if let Some((send_fragment, mut mtu)) = send_to(&session) { + if let Some((sender, mut mtu)) = send_to.init_send(&session) { mtu = mtu.max(MIN_TRANSPORT_MTU); - send_with_fragmentation(send_fragment, mtu, packet, hk_send); + send_with_fragmentation(sender, mtu, packet, hk_send); } }); if let Some(next_timer) = result {