reformat

Author: pseusys

typhoon/src/flow/client.rs

@@ -70,11 +70,7 @@ impl<T: IdentityType + Clone + 'static, AE: AsyncExecutor + 'static> ClientFlowM
 }
 
 impl<T: IdentityType + Clone + 'static, AE: AsyncExecutor + 'static> ProbeFlowSender for ClientFlowManager<T, AE> {
-    fn send_raw<'a>(
-        &'a self,
-        packet: DynamicByteBuffer,
-        _target: SocketAddr,
-    ) -> Pin<Box<dyn Future<Output = Result<(), SocketError>> + Send + 'a>> {
+    fn send_raw<'a>(&'a self, packet: DynamicByteBuffer, _target: SocketAddr) -> Pin<Box<dyn Future<Output = Result<(), SocketError>> + Send + 'a>> {
         Box::pin(async move { self.sock.send(packet).await.map(|_| ()) })
     }
 }
@@ -110,7 +106,7 @@ impl<T: IdentityType + Clone + 'static, AE: AsyncExecutor + 'static> FlowManager
 
             {
                 let mut lock = self.receive_internal.lock().await;
-                match lock.process_incoming(notified_packet.unwrap(), self.settings.pool())? {
+                match lock.process_incoming(notified_packet, self.settings.pool())? {
                     ProcessIncomingResult::Valid(result) => return Ok(result),
                     ProcessIncomingResult::Decoy => continue,
                     ProcessIncomingResult::Unexpected(pkt) => {

typhoon/src/flow/common.rs

@@ -140,18 +140,14 @@ impl<CP: FlowCryptoProvider> FlowReceiveInternal<CP> {
                         Ok(()) => {}
                         Err(err) => {
                             warn!("client flow: tailor verification failed: {err}");
-                            return Ok(ProcessIncomingResult::Unexpected(
-                                encrypted_packet.expand_end(encrypted_tailor_len),
-                            ));
+                            return Ok(ProcessIncomingResult::Unexpected(encrypted_packet.expand_end(encrypted_tailor_len)));
                         }
                     }
                     tailor
                 }
                 Err(err) => {
                     warn!("client flow: tailor decryption failed: {err}");
-                    return Ok(ProcessIncomingResult::Unexpected(
-                        encrypted_packet.expand_end(encrypted_tailor_len),
-                    ));
+                    return Ok(ProcessIncomingResult::Unexpected(encrypted_packet.expand_end(encrypted_tailor_len)));
                 }
             },
             Err(err) => return Err(FlowControllerError::MissingCache(err)),
@@ -163,8 +159,6 @@ impl<CP: FlowCryptoProvider> FlowReceiveInternal<CP> {
         }
 
         let payload_len = tailor.payload_length() as usize;
-        Ok(ProcessIncomingResult::Valid(
-            encrypted_packet.rebuffer_start(encrypted_packet.len() - payload_len).expand_end(full_tailor_len),
-        ))
+        Ok(ProcessIncomingResult::Valid(encrypted_packet.rebuffer_start(encrypted_packet.len() - payload_len).expand_end(full_tailor_len)))
     }
 }

typhoon/src/flow/probe.rs

@@ -23,11 +23,7 @@ use crate::utils::sync::AsyncExecutor;
 /// `ClientFlowManager` (forwards to `send` on the connected socket, ignoring `target`).
 pub trait ProbeFlowSender: Send + Sync {
     /// Send `packet` as raw bytes to `target`, bypassing all TYPHOON framing.
-    fn send_raw<'a>(
-        &'a self,
-        packet: DynamicByteBuffer,
-        target: SocketAddr,
-    ) -> Pin<Box<dyn Future<Output = Result<(), SocketError>> + Send + 'a>>;
+    fn send_raw<'a>(&'a self, packet: DynamicByteBuffer, target: SocketAddr) -> Pin<Box<dyn Future<Output = Result<(), SocketError>> + Send + 'a>>;
 }
 
 /// Handler for packets the flow manager could not identify (active probing protection).

typhoon/src/flow/server.rs

@@ -151,9 +151,7 @@ impl<T: IdentityType + Clone + Eq + Hash + Send + ToString + 'static, AE: AsyncE
                     Ok(result) => result,
                     Err(err) => {
                         warn!("server flow: tailor decryption failed from {source_addr}: {err}");
-                        self.probe_handler.lock().await.process(
-                            encrypted_packet.expand_end(identity_len + tailor_overhead), Some(source_addr),
-                        ).await;
+                        self.probe_handler.lock().await.process(encrypted_packet.expand_end(identity_len + tailor_overhead), Some(source_addr)).await;
                         continue;
                     }
                 };
@@ -162,9 +160,7 @@ impl<T: IdentityType + Clone + Eq + Hash + Send + ToString + 'static, AE: AsyncE
                     let identity = tailor.identity();
                     if let Err(err) = crypto.verify_tailor(&identity, transcript).await {
                         debug!("error verifying packet tailor: {err}");
-                        self.probe_handler.lock().await.process(
-                            encrypted_packet.expand_end(identity_len + tailor_overhead), Some(source_addr),
-                        ).await;
+                        self.probe_handler.lock().await.process(encrypted_packet.expand_end(identity_len + tailor_overhead), Some(source_addr)).await;
                         continue;
                     }
                 }
@@ -194,9 +190,7 @@ impl<T: IdentityType + Clone + Eq + Hash + Send + ToString + 'static, AE: AsyncE
                     }
                 } else {
                     // Non-handshake, non-decoy packet from an unregistered user.
-                    self.probe_handler.lock().await.process(
-                        encrypted_packet.expand_end(identity_len + tailor_overhead), Some(source_addr),
-                    ).await;
+                    self.probe_handler.lock().await.process(encrypted_packet.expand_end(identity_len + tailor_overhead), Some(source_addr)).await;
                     continue;
                 }
             }
@@ -226,11 +220,7 @@ impl<T: IdentityType + Clone + Eq + Hash + Send + ToString + 'static, AE: AsyncE
 }
 
 impl<T: IdentityType + Clone + Eq + Hash + Send + ToString + 'static, AE: AsyncExecutor + 'static> ProbeFlowSender for ServerFlowManager<T, AE> {
-    fn send_raw<'a>(
-        &'a self,
-        packet: DynamicByteBuffer,
-        target: SocketAddr,
-    ) -> Pin<Box<dyn Future<Output = Result<(), SocketError>> + Send + 'a>> {
+    fn send_raw<'a>(&'a self, packet: DynamicByteBuffer, target: SocketAddr) -> Pin<Box<dyn Future<Output = Result<(), SocketError>> + Send + 'a>> {
         Box::pin(async move { self.socks[0].send_to(packet, target).await.map(|_| ()) })
     }
 }