diff --git a/src/proxy/masking.rs b/src/proxy/masking.rs index c7f335e..c346b88 100644 --- a/src/proxy/masking.rs +++ b/src/proxy/masking.rs @@ -235,9 +235,12 @@ async fn consume_client_data_with_timeout_and_cap( ) where R: AsyncRead + Unpin, { - if timeout(relay_timeout, consume_client_data(reader, byte_cap, idle_timeout)) - .await - .is_err() + if timeout( + relay_timeout, + consume_client_data(reader, byte_cap, idle_timeout), + ) + .await + .is_err() { debug!("Timed out while consuming client data on masking fallback path"); } diff --git a/src/proxy/tests/masking_consume_idle_timeout_security_tests.rs b/src/proxy/tests/masking_consume_idle_timeout_security_tests.rs index 227db73..fcd2e79 100644 --- a/src/proxy/tests/masking_consume_idle_timeout_security_tests.rs +++ b/src/proxy/tests/masking_consume_idle_timeout_security_tests.rs @@ -57,9 +57,12 @@ async fn fast_reader_drains_to_eof() { let data = vec![0xAAu8; 32 * 1024]; let reader = std::io::Cursor::new(data); - tokio::time::timeout(MASK_RELAY_TIMEOUT, consume_client_data(reader, usize::MAX, MASK_RELAY_IDLE_TIMEOUT)) - .await - .expect("consume_client_data did not complete for fast EOF reader"); + tokio::time::timeout( + MASK_RELAY_TIMEOUT, + consume_client_data(reader, usize::MAX, MASK_RELAY_IDLE_TIMEOUT), + ) + .await + .expect("consume_client_data did not complete for fast EOF reader"); } #[tokio::test] diff --git a/src/proxy/tests/masking_consume_stress_adversarial_tests.rs b/src/proxy/tests/masking_consume_stress_adversarial_tests.rs index 4ef283e..7579a9c 100644 --- a/src/proxy/tests/masking_consume_stress_adversarial_tests.rs +++ b/src/proxy/tests/masking_consume_stress_adversarial_tests.rs @@ -34,7 +34,11 @@ async fn consume_stall_stress_finishes_within_idle_budget() { set.spawn(async { tokio::time::timeout( MASK_RELAY_TIMEOUT, - consume_client_data(OneByteThenStall { sent: false }, usize::MAX, MASK_RELAY_IDLE_TIMEOUT), + consume_client_data( + OneByteThenStall { sent: false }, + usize::MAX, + MASK_RELAY_IDLE_TIMEOUT, + ), ) .await .expect("consume_client_data exceeded relay timeout under stall load"); diff --git a/src/proxy/tests/masking_production_cap_regression_security_tests.rs b/src/proxy/tests/masking_production_cap_regression_security_tests.rs index 32ecdbb..c5d542e 100644 --- a/src/proxy/tests/masking_production_cap_regression_security_tests.rs +++ b/src/proxy/tests/masking_production_cap_regression_security_tests.rs @@ -127,8 +127,14 @@ async fn positive_copy_with_production_cap_stops_exactly_at_budget() { let mut reader = FinitePatternReader::new(PROD_CAP_BYTES + (256 * 1024), 4096, read_calls); let mut writer = CountingWriter::default(); - let outcome = - copy_with_idle_timeout(&mut reader, &mut writer, PROD_CAP_BYTES, true, MASK_RELAY_IDLE_TIMEOUT).await; + let outcome = copy_with_idle_timeout( + &mut reader, + &mut writer, + PROD_CAP_BYTES, + true, + MASK_RELAY_IDLE_TIMEOUT, + ) + .await; assert_eq!( outcome.total, PROD_CAP_BYTES, @@ -146,7 +152,13 @@ async fn negative_consume_with_zero_cap_performs_no_reads() { let read_calls = Arc::new(AtomicUsize::new(0)); let reader = FinitePatternReader::new(1024, 64, Arc::clone(&read_calls)); - consume_client_data_with_timeout_and_cap(reader, 0, MASK_RELAY_TIMEOUT, MASK_RELAY_IDLE_TIMEOUT).await; + consume_client_data_with_timeout_and_cap( + reader, + 0, + MASK_RELAY_TIMEOUT, + MASK_RELAY_IDLE_TIMEOUT, + ) + .await; assert_eq!( read_calls.load(Ordering::Relaxed), @@ -162,8 +174,14 @@ async fn edge_copy_below_cap_reports_eof_without_overread() { let mut reader = FinitePatternReader::new(payload, 3072, read_calls); let mut writer = CountingWriter::default(); - let outcome = - copy_with_idle_timeout(&mut reader, &mut writer, PROD_CAP_BYTES, true, MASK_RELAY_IDLE_TIMEOUT).await; + let outcome = copy_with_idle_timeout( + &mut reader, + &mut writer, + PROD_CAP_BYTES, + true, + MASK_RELAY_IDLE_TIMEOUT, + ) + .await; assert_eq!(outcome.total, payload); assert_eq!(writer.written, payload); @@ -177,7 +195,13 @@ async fn edge_copy_below_cap_reports_eof_without_overread() { async fn adversarial_blackhat_never_ready_reader_is_bounded_by_timeout_guards() { let started = Instant::now(); - consume_client_data_with_timeout_and_cap(NeverReadyReader, PROD_CAP_BYTES, MASK_RELAY_TIMEOUT, MASK_RELAY_IDLE_TIMEOUT).await; + consume_client_data_with_timeout_and_cap( + NeverReadyReader, + PROD_CAP_BYTES, + MASK_RELAY_TIMEOUT, + MASK_RELAY_IDLE_TIMEOUT, + ) + .await; assert!( started.elapsed() < Duration::from_millis(350), @@ -192,7 +216,12 @@ async fn integration_consume_path_honors_production_cap_for_large_payload() { let bounded = timeout( Duration::from_millis(350), - consume_client_data_with_timeout_and_cap(reader, PROD_CAP_BYTES, MASK_RELAY_TIMEOUT, MASK_RELAY_IDLE_TIMEOUT), + consume_client_data_with_timeout_and_cap( + reader, + PROD_CAP_BYTES, + MASK_RELAY_TIMEOUT, + MASK_RELAY_IDLE_TIMEOUT, + ), ) .await; @@ -208,7 +237,13 @@ async fn adversarial_consume_path_never_reads_beyond_declared_byte_cap() { let total_read = Arc::new(AtomicUsize::new(0)); let reader = BudgetProbeReader::new(256 * 1024, Arc::clone(&total_read)); - consume_client_data_with_timeout_and_cap(reader, byte_cap, MASK_RELAY_TIMEOUT, MASK_RELAY_IDLE_TIMEOUT).await; + consume_client_data_with_timeout_and_cap( + reader, + byte_cap, + MASK_RELAY_TIMEOUT, + MASK_RELAY_IDLE_TIMEOUT, + ) + .await; assert!( total_read.load(Ordering::Relaxed) <= byte_cap, @@ -233,7 +268,9 @@ async fn light_fuzz_cap_and_payload_matrix_preserves_min_budget_invariant() { let mut reader = FinitePatternReader::new(payload, chunk, read_calls); let mut writer = CountingWriter::default(); - let outcome = copy_with_idle_timeout(&mut reader, &mut writer, cap, true, MASK_RELAY_IDLE_TIMEOUT).await; + let outcome = + copy_with_idle_timeout(&mut reader, &mut writer, cap, true, MASK_RELAY_IDLE_TIMEOUT) + .await; let expected = payload.min(cap); assert_eq!( @@ -263,7 +300,14 @@ async fn stress_parallel_copy_tasks_with_production_cap_complete_without_leaks() read_calls, ); let mut writer = CountingWriter::default(); - copy_with_idle_timeout(&mut reader, &mut writer, PROD_CAP_BYTES, true, MASK_RELAY_IDLE_TIMEOUT).await + copy_with_idle_timeout( + &mut reader, + &mut writer, + PROD_CAP_BYTES, + true, + MASK_RELAY_IDLE_TIMEOUT, + ) + .await })); }