Add adversarial tests for client, handshake, masking, and relay modules

- Introduced `client_adversarial_tests.rs` to stress test connection limits and IP tracker race conditions.
- Added `handshake_adversarial_tests.rs` for mutational bit-flipping tests and timing neutrality checks.
- Created `masking_adversarial_tests.rs` to validate probing indistinguishability and SSRF prevention.
- Implemented `relay_adversarial_tests.rs` to ensure HOL blocking prevention and data quota enforcement.
- Updated respective modules to include new test paths.

src/proxy/client.rs

@@ -1040,3 +1040,5 @@ impl RunningClientHandler {
 #[cfg(test)]
 #[path = "client_security_tests.rs"]
 mod security_tests;
+#[path = "client_adversarial_tests.rs"]
+mod adversarial_tests;

src/proxy/client_adversarial_tests.rs

@@ -0,0 +1,109 @@
+use super::*;
+use crate::config::ProxyConfig;
+use crate::stats::Stats;
+use crate::ip_tracker::UserIpTracker;
+use crate::error::ProxyError;
+use std::sync::Arc;
+use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+
+// ------------------------------------------------------------------
+// Priority 3: Massive Concurrency Stress (OWASP ASVS 5.1.6)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn client_stress_10k_connections_limit_strict() {
+    let user = "stress-user";
+    let limit = 512;
+    
+    let stats = Arc::new(Stats::new());
+    let ip_tracker = Arc::new(UserIpTracker::new());
+    
+    let mut config = ProxyConfig::default();
+    config.access.user_max_tcp_conns.insert(user.to_string(), limit);
+    
+    let iterations = 1000;
+    let mut tasks = Vec::new();
+
+    for i in 0..iterations {
+        let stats = Arc::clone(&stats);
+        let ip_tracker = Arc::clone(&ip_tracker);
+        let config = config.clone();
+        let user_str = user.to_string();
+        
+        tasks.push(tokio::spawn(async move {
+            let peer = SocketAddr::new(
+                IpAddr::V4(Ipv4Addr::new(127, 0, 0, (i % 254 + 1) as u8)),
+                10000 + (i % 1000) as u16,
+            );
+            
+            match RunningClientHandler::acquire_user_connection_reservation_static(
+                &user_str,
+                &config,
+                stats,
+                peer,
+                ip_tracker,
+            ).await {
+                Ok(res) => Ok(res),
+                Err(ProxyError::ConnectionLimitExceeded { .. }) => Err(()),
+                Err(e) => panic!("Unexpected error: {:?}", e),
+            }
+        }));
+    }
+
+    let results = futures::future::join_all(tasks).await;
+    let mut successes = 0;
+    let mut failures = 0;
+    let mut reservations = Vec::new();
+
+    for res in results {
+        match res.unwrap() {
+            Ok(r) => {
+                successes += 1;
+                reservations.push(r);
+            }
+            Err(_) => failures += 1,
+        }
+    }
+
+    assert_eq!(successes, limit, "Should allow exactly 'limit' connections");
+    assert_eq!(failures, iterations - limit, "Should fail the rest with LimitExceeded");
+    assert_eq!(stats.get_user_curr_connects(user), limit as u64);
+
+    drop(reservations);
+    
+    ip_tracker.drain_cleanup_queue().await;
+    
+    assert_eq!(stats.get_user_curr_connects(user), 0, "Stats must converge to 0 after all drops");
+    assert_eq!(ip_tracker.get_active_ip_count(user).await, 0, "IP tracker must converge to 0");
+}
+
+// ------------------------------------------------------------------
+// Priority 3: IP Tracker Race Stress
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn client_ip_tracker_race_condition_stress() {
+    let user = "race-user";
+    let ip_tracker = Arc::new(UserIpTracker::new());
+    ip_tracker.set_user_limit(user, 100).await;
+    
+    let iterations = 1000;
+    let mut tasks = Vec::new();
+
+    for i in 0..iterations {
+        let ip_tracker = Arc::clone(&ip_tracker);
+        let ip = IpAddr::V4(Ipv4Addr::new(10, 0, 0, (i % 254 + 1) as u8));
+        
+        tasks.push(tokio::spawn(async move {
+            for _ in 0..10 {
+                if let Ok(()) = ip_tracker.check_and_add("race-user", ip).await {
+                    ip_tracker.remove_ip("race-user", ip).await;
+                }
+            }
+        }));
+    }
+
+    futures::future::join_all(tasks).await;
+    
+    assert_eq!(ip_tracker.get_active_ip_count(user).await, 0, "IP count must be zero after balanced add/remove burst");
+}

src/proxy/handshake.rs

@@ -971,6 +971,10 @@ mod security_tests;
 #[path = "handshake_gap_short_tls_probe_throttle_security_tests.rs"]
 mod gap_short_tls_probe_throttle_security_tests;
 
+#[cfg(test)]
+#[path = "handshake_adversarial_tests.rs"]
+mod adversarial_tests;
+
 /// Compile-time guard: HandshakeSuccess holds cryptographic key material and
 /// must never be Copy.  A Copy impl would allow silent key duplication,
 /// undermining the zeroize-on-drop guarantee.

src/proxy/handshake_adversarial_tests.rs

@@ -0,0 +1,231 @@
+use super::*;
+use std::sync::Arc;
+use std::net::{IpAddr, Ipv4Addr};
+use std::time::{Duration, Instant};
+use crate::crypto::sha256;
+
+fn make_valid_mtproto_handshake(secret_hex: &str, proto_tag: ProtoTag, dc_idx: i16) -> [u8; HANDSHAKE_LEN] {
+    let secret = hex::decode(secret_hex).expect("secret hex must decode");
+    let mut handshake = [0x5Au8; HANDSHAKE_LEN];
+    for (idx, b) in handshake[SKIP_LEN..SKIP_LEN + PREKEY_LEN + IV_LEN]
+        .iter_mut()
+        .enumerate()
+    {
+        *b = (idx as u8).wrapping_add(1);
+    }
+
+    let dec_prekey = &handshake[SKIP_LEN..SKIP_LEN + PREKEY_LEN];
+    let dec_iv_bytes = &handshake[SKIP_LEN + PREKEY_LEN..SKIP_LEN + PREKEY_LEN + IV_LEN];
+
+    let mut dec_key_input = Vec::with_capacity(PREKEY_LEN + secret.len());
+    dec_key_input.extend_from_slice(dec_prekey);
+    dec_key_input.extend_from_slice(&secret);
+    let dec_key = sha256(&dec_key_input);
+
+    let mut dec_iv_arr = [0u8; IV_LEN];
+    dec_iv_arr.copy_from_slice(dec_iv_bytes);
+    let dec_iv = u128::from_be_bytes(dec_iv_arr);
+
+    let mut stream = AesCtr::new(&dec_key, dec_iv);
+    let keystream = stream.encrypt(&[0u8; HANDSHAKE_LEN]);
+
+    let mut target_plain = [0u8; HANDSHAKE_LEN];
+    target_plain[PROTO_TAG_POS..PROTO_TAG_POS + 4].copy_from_slice(&proto_tag.to_bytes());
+    target_plain[DC_IDX_POS..DC_IDX_POS + 2].copy_from_slice(&dc_idx.to_le_bytes());
+
+    for idx in PROTO_TAG_POS..HANDSHAKE_LEN {
+        handshake[idx] = target_plain[idx] ^ keystream[idx];
+    }
+
+    handshake
+}
+
+fn auth_probe_test_guard() -> std::sync::MutexGuard<'static, ()> {
+    auth_probe_test_lock()
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+fn test_config_with_secret_hex(secret_hex: &str) -> ProxyConfig {
+    let mut cfg = ProxyConfig::default();
+    cfg.access.users.clear();
+    cfg.access.users.insert("user".to_string(), secret_hex.to_string());
+    cfg.access.ignore_time_skew = true;
+    cfg.general.modes.secure = true;
+    cfg
+}
+
+// ------------------------------------------------------------------
+// Mutational Bit-Flipping Tests (OWASP ASVS 5.1.4)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn mtproto_handshake_bit_flip_anywhere_rejected() {
+    let _guard = auth_probe_test_guard();
+    clear_auth_probe_state_for_testing();
+
+    let secret_hex = "11223344556677889900aabbccddeeff";
+    let base = make_valid_mtproto_handshake(secret_hex, ProtoTag::Secure, 2);
+    let config = test_config_with_secret_hex(secret_hex);
+    let replay_checker = ReplayChecker::new(128, Duration::from_secs(60));
+    let peer: SocketAddr = "192.0.2.1:12345".parse().unwrap();
+
+    // Baseline check
+    let res = handle_mtproto_handshake(&base, tokio::io::empty(), tokio::io::sink(), peer, &config, &replay_checker, false, None).await;
+    match res {
+        HandshakeResult::Success(_) => {},
+        _ => panic!("Baseline failed: expected Success"),
+    }
+
+    // Flip bits in the encrypted part (beyond the key material)
+    for byte_pos in SKIP_LEN..HANDSHAKE_LEN {
+        let mut h = base;
+        h[byte_pos] ^= 0x01; // Flip 1 bit
+        let res = handle_mtproto_handshake(&h, tokio::io::empty(), tokio::io::sink(), peer, &config, &replay_checker, false, None).await;
+        assert!(matches!(res, HandshakeResult::BadClient { .. }), "Flip at byte {byte_pos} bit 0 must be rejected");
+    }
+}
+
+// ------------------------------------------------------------------
+// Adversarial Probing / Timing Neutrality (OWASP ASVS 5.1.7)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn mtproto_handshake_timing_neutrality_mocked() {
+    let secret_hex = "00112233445566778899aabbccddeeff";
+    let base = make_valid_mtproto_handshake(secret_hex, ProtoTag::Secure, 1);
+    let config = test_config_with_secret_hex(secret_hex);
+    let replay_checker = ReplayChecker::new(128, Duration::from_secs(60));
+    let peer: SocketAddr = "192.0.2.2:54321".parse().unwrap();
+
+    const ITER: usize = 50;
+    
+    let mut start = Instant::now();
+    for _ in 0..ITER {
+        let _ = handle_mtproto_handshake(&base, tokio::io::empty(), tokio::io::sink(), peer, &config, &replay_checker, false, None).await;
+    }
+    let duration_success = start.elapsed();
+
+    start = Instant::now();
+    for i in 0..ITER {
+        let mut h = base;
+        h[SKIP_LEN + (i % 48)] ^= 0xFF; 
+        let _ = handle_mtproto_handshake(&h, tokio::io::empty(), tokio::io::sink(), peer, &config, &replay_checker, false, None).await;
+    }
+    let duration_fail = start.elapsed();
+
+    let avg_diff_ms = (duration_success.as_millis() as f64 - duration_fail.as_millis() as f64).abs() / ITER as f64;
+    
+    // Threshold (loose for CI)
+    assert!(avg_diff_ms < 100.0, "Timing difference too large: {} ms/iter", avg_diff_ms);
+}
+
+// ------------------------------------------------------------------
+// Stress Tests (OWASP ASVS 5.1.6)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn auth_probe_throttle_saturation_stress() {
+    let _guard = auth_probe_test_guard();
+    clear_auth_probe_state_for_testing();
+
+    let now = Instant::now();
+    
+    // Record enough failures for one IP to trigger backoff
+    let target_ip = IpAddr::V4(Ipv4Addr::new(1, 1, 1, 1));
+    for _ in 0..AUTH_PROBE_BACKOFF_START_FAILS {
+        auth_probe_record_failure(target_ip, now);
+    }
+    
+    assert!(auth_probe_is_throttled(target_ip, now));
+
+    // Stress test with many unique IPs
+    for i in 0..500u32 {
+        let ip = IpAddr::V4(Ipv4Addr::new(203, 0, 113, (i % 256) as u8));
+        auth_probe_record_failure(ip, now);
+    }
+
+    let tracked = AUTH_PROBE_STATE
+        .get()
+        .map(|state| state.len())
+        .unwrap_or(0);
+    assert!(
+        tracked <= AUTH_PROBE_TRACK_MAX_ENTRIES,
+        "auth probe state grew past hard cap: {tracked} > {AUTH_PROBE_TRACK_MAX_ENTRIES}"
+    );
+}
+
+#[tokio::test]
+async fn mtproto_handshake_abridged_prefix_rejected() {
+    let _guard = auth_probe_test_guard();
+    clear_auth_probe_state_for_testing();
+
+    let mut handshake = [0x5Au8; HANDSHAKE_LEN];
+    handshake[0] = 0xef; // Abridged prefix
+    let config = ProxyConfig::default();
+    let replay_checker = ReplayChecker::new(128, Duration::from_secs(60));
+    let peer: SocketAddr = "192.0.2.3:12345".parse().unwrap();
+
+    let res = handle_mtproto_handshake(&handshake, tokio::io::empty(), tokio::io::sink(), peer, &config, &replay_checker, false, None).await;
+    // MTProxy stops immediately on 0xef
+    assert!(matches!(res, HandshakeResult::BadClient { .. }));
+}
+
+#[tokio::test]
+async fn mtproto_handshake_preferred_user_mismatch_continues() {
+    let _guard = auth_probe_test_guard();
+    clear_auth_probe_state_for_testing();
+
+    let secret1_hex = "11111111111111111111111111111111";
+    let secret2_hex = "22222222222222222222222222222222";
+    
+    let base = make_valid_mtproto_handshake(secret2_hex, ProtoTag::Secure, 1);
+    let mut config = ProxyConfig::default();
+    config.access.users.insert("user1".to_string(), secret1_hex.to_string());
+    config.access.users.insert("user2".to_string(), secret2_hex.to_string());
+    config.access.ignore_time_skew = true;
+    config.general.modes.secure = true;
+
+    let replay_checker = ReplayChecker::new(128, Duration::from_secs(60));
+    let peer: SocketAddr = "192.0.2.4:12345".parse().unwrap();
+
+    // Even if we prefer user1, if user2 matches, it should succeed.
+    let res = handle_mtproto_handshake(&base, tokio::io::empty(), tokio::io::sink(), peer, &config, &replay_checker, false, Some("user1")).await;
+    if let HandshakeResult::Success((_, _, success)) = res {
+        assert_eq!(success.user, "user2");
+    } else {
+        panic!("Handshake failed even though user2 matched");
+    }
+}
+
+#[tokio::test]
+async fn mtproto_handshake_concurrent_flood_stability() {
+    let _guard = auth_probe_test_guard();
+    clear_auth_probe_state_for_testing();
+
+    let secret_hex = "00112233445566778899aabbccddeeff";
+    let base = make_valid_mtproto_handshake(secret_hex, ProtoTag::Secure, 1);
+    let mut config = test_config_with_secret_hex(secret_hex);
+    config.access.ignore_time_skew = true;
+    let replay_checker = Arc::new(ReplayChecker::new(1024, Duration::from_secs(60)));
+    let config = Arc::new(config);
+    
+    let mut tasks = Vec::new();
+    for i in 0..50 {
+        let base = base;
+        let config = Arc::clone(&config);
+        let replay_checker = Arc::clone(&replay_checker);
+        let peer: SocketAddr = format!("192.0.2.{}:12345", (i % 254) + 1).parse().unwrap();
+        
+        tasks.push(tokio::spawn(async move {
+            let res = handle_mtproto_handshake(&base, tokio::io::empty(), tokio::io::sink(), peer, &config, &replay_checker, false, None).await;
+            matches!(res, HandshakeResult::Success(_))
+        }));
+    }
+    
+    // We don't necessarily care if they all succeed (some might fail due to replay if they hit the same chunk),
+    // but the system must not panic or hang.
+    for task in tasks {
+        let _ = task.await.unwrap();
+    }
+}

src/proxy/masking.rs

@@ -317,3 +317,7 @@ async fn consume_client_data<R: AsyncRead + Unpin>(mut reader: R) {
 #[cfg(test)]
 #[path = "masking_security_tests.rs"]
 mod security_tests;
+
+#[cfg(test)]
+#[path = "masking_adversarial_tests.rs"]
+mod adversarial_tests;

src/proxy/masking_adversarial_tests.rs

@@ -0,0 +1,213 @@
+use super::*;
+use std::sync::Arc;
+use tokio::io::duplex;
+use tokio::net::TcpListener;
+use tokio::time::{Instant, Duration};
+use crate::config::ProxyConfig;
+use crate::stats::beobachten::BeobachtenStore;
+
+// ------------------------------------------------------------------
+// Probing Indistinguishability (OWASP ASVS 5.1.7)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn masking_probes_indistinguishable_timing() {
+    let mut config = ProxyConfig::default();
+    config.censorship.mask = true;
+    config.censorship.mask_host = Some("127.0.0.1".to_string());
+    config.censorship.mask_port = 80; // Should timeout/refuse
+    
+    let peer: SocketAddr = "192.0.2.10:443".parse().unwrap();
+    let local_addr: SocketAddr = "127.0.0.1:443".parse().unwrap();
+    let beobachten = BeobachtenStore::new();
+
+    // Test different probe types
+    let probes = vec![
+        (b"GET / HTTP/1.1\r\nHost: x\r\n\r\n".to_vec(), "HTTP"),
+        (b"SSH-2.0-probe".to_vec(), "SSH"),
+        (vec![0x16, 0x03, 0x03, 0x00, 0x05, 0x01, 0x00, 0x00, 0x01, 0x00], "TLS-scanner"),
+        (vec![0x42; 5], "port-scanner"),
+    ];
+
+    for (probe, type_name) in probes {
+        let (client_reader, _client_writer) = duplex(256);
+        let (_client_visible_reader, client_visible_writer) = duplex(256);
+        
+        let start = Instant::now();
+        handle_bad_client(
+            client_reader,
+            client_visible_writer,
+            &probe,
+            peer,
+            local_addr,
+            &config,
+            &beobachten,
+        ).await;
+        
+        let elapsed = start.elapsed();
+        
+        // We expect any outcome to take roughly MASK_TIMEOUT (50ms in tests)
+        // to mask whether the backend was reachable or refused.
+        assert!(elapsed >= Duration::from_millis(30), "Probe {type_name} finished too fast: {elapsed:?}");
+    }
+}
+
+// ------------------------------------------------------------------
+// Masking Budget Stress Tests (OWASP ASVS 5.1.6)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn masking_budget_stress_under_load() {
+    let mut config = ProxyConfig::default();
+    config.censorship.mask = true;
+    config.censorship.mask_host = Some("127.0.0.1".to_string());
+    config.censorship.mask_port = 1; // Unlikely port
+
+    let peer: SocketAddr = "192.0.2.20:443".parse().unwrap();
+    let local_addr: SocketAddr = "127.0.0.1:443".parse().unwrap();
+    let beobachten = Arc::new(BeobachtenStore::new());
+
+    let mut tasks = Vec::new();
+    for _ in 0..50 {
+        let (client_reader, _client_writer) = duplex(256);
+        let (_client_visible_reader, client_visible_writer) = duplex(256);
+        let config = config.clone();
+        let beobachten = Arc::clone(&beobachten);
+        
+        tasks.push(tokio::spawn(async move {
+            let start = Instant::now();
+            handle_bad_client(
+                client_reader,
+                client_visible_writer,
+                b"probe",
+                peer,
+                local_addr,
+                &config,
+                &beobachten,
+            ).await;
+            start.elapsed()
+        }));
+    }
+
+    for task in tasks {
+        let elapsed = task.await.unwrap();
+        assert!(elapsed >= Duration::from_millis(30), "Stress probe finished too fast: {elapsed:?}");
+    }
+}
+
+// ------------------------------------------------------------------
+// detect_client_type Fingerprint Check
+// ------------------------------------------------------------------
+
+#[test]
+fn test_detect_client_type_boundary_cases() {
+    // 9 bytes = port-scanner
+    assert_eq!(detect_client_type(&[0x42; 9]), "port-scanner");
+    // 10 bytes = unknown
+    assert_eq!(detect_client_type(&[0x42; 10]), "unknown");
+    
+    // HTTP verbs without trailing space
+    assert_eq!(detect_client_type(b"GET/"), "port-scanner"); // because len < 10
+    assert_eq!(detect_client_type(b"GET /path"), "HTTP"); 
+}
+
+// ------------------------------------------------------------------
+// Priority 2: Slowloris and Slow Read Attacks (OWASP ASVS 5.1.5)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn masking_slowloris_client_idle_timeout_rejected() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let backend_addr = listener.local_addr().unwrap();
+    let initial = b"GET / HTTP/1.1\r\nHost: front.example\r\n\r\n".to_vec();
+
+    let accept_task = tokio::spawn({
+        let initial = initial.clone();
+        async move {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            let mut observed = vec![0u8; initial.len()];
+            stream.read_exact(&mut observed).await.unwrap();
+            assert_eq!(observed, initial);
+
+            let mut drip = [0u8; 1];
+            let drip_read = tokio::time::timeout(Duration::from_millis(220), stream.read_exact(&mut drip)).await;
+            assert!(
+                drip_read.is_err() || drip_read.unwrap().is_err(),
+                "backend must not receive post-timeout slowloris drip bytes"
+            );
+        }
+    });
+
+    let mut config = ProxyConfig::default();
+    config.censorship.mask = true;
+    config.censorship.mask_host = Some("127.0.0.1".to_string());
+    config.censorship.mask_port = backend_addr.port();
+
+    let beobachten = BeobachtenStore::new();
+    let peer: SocketAddr = "192.0.2.10:12345".parse().unwrap();
+    let local: SocketAddr = "192.0.2.1:443".parse().unwrap();
+
+    let (mut client_writer, client_reader) = duplex(1024);
+    let (_client_visible_reader, client_visible_writer) = duplex(1024);
+
+    let handle = tokio::spawn(async move {
+        handle_bad_client(
+            client_reader,
+            client_visible_writer,
+            &initial,
+            peer,
+            local,
+            &config,
+            &beobachten,
+        )
+        .await;
+    });
+
+    tokio::time::sleep(Duration::from_millis(160)).await;
+    let _ = client_writer.write_all(b"X").await;
+
+    handle.await.unwrap();
+    accept_task.await.unwrap();
+}
+
+// ------------------------------------------------------------------
+// Priority 2: Fallback Server Down / Fingerprinting (OWASP ASVS 5.1.7)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn masking_fallback_down_mimics_timeout() {
+    let mut config = ProxyConfig::default();
+    config.censorship.mask = true;
+    config.censorship.mask_host = Some("127.0.0.1".to_string());
+    config.censorship.mask_port = 1; // Unlikely port
+    
+    let (server_reader, server_writer) = duplex(1024);
+    let beobachten = BeobachtenStore::new();
+    let peer: SocketAddr = "192.0.2.12:12345".parse().unwrap();
+    let local: SocketAddr = "192.0.2.1:443".parse().unwrap();
+
+    let start = Instant::now();
+    handle_bad_client(server_reader, server_writer, b"GET / HTTP/1.1\r\n", peer, local, &config, &beobachten).await;
+    
+    let elapsed = start.elapsed();
+    // It should wait for MASK_TIMEOUT (50ms in tests) even if connection was refused immediately
+    assert!(elapsed >= Duration::from_millis(40), "Must respect connect budget even on failure: {:?}", elapsed);
+}
+
+// ------------------------------------------------------------------
+// Priority 2: SSRF Prevention (OWASP ASVS 5.1.2)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn masking_ssrf_resolve_internal_ranges_blocked() {
+    use crate::network::dns_overrides::resolve_socket_addr;
+
+    let blocked_ips = ["127.0.0.1", "169.254.169.254", "10.0.0.1", "192.168.1.1", "0.0.0.0"];
+
+    for ip in blocked_ips {
+        assert!(
+            resolve_socket_addr(ip, 80).is_none(),
+            "runtime DNS overrides must not resolve unconfigured literal host targets"
+        );
+    }
+}

src/proxy/relay.rs

@@ -659,3 +659,5 @@ where
 #[cfg(test)]
 #[path = "relay_security_tests.rs"]
 mod security_tests;
+#[path = "relay_adversarial_tests.rs"]
+mod adversarial_tests;

src/proxy/relay_adversarial_tests.rs

@@ -0,0 +1,122 @@
+use super::*;
+use crate::error::ProxyError;
+use crate::stats::Stats;
+use crate::stream::BufferPool;
+use std::sync::Arc;
+use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
+use tokio::time::{Duration, Instant, timeout};
+
+// ------------------------------------------------------------------
+// Priority 3: Async Relay HOL Blocking Prevention (OWASP ASVS 5.1.5)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn relay_hol_blocking_prevention_regression() {
+    let stats = Arc::new(Stats::new());
+    let user = "hol-user";
+    
+    let (client_peer, relay_client) = duplex(65536);
+    let (relay_server, server_peer) = duplex(65536);
+
+    let (client_reader, client_writer) = tokio::io::split(relay_client);
+    let (server_reader, server_writer) = tokio::io::split(relay_server);
+    let (mut cp_reader, mut cp_writer) = tokio::io::split(client_peer);
+    let (mut sp_reader, mut sp_writer) = tokio::io::split(server_peer);
+
+    let relay_task = tokio::spawn(relay_bidirectional(
+        client_reader,
+        client_writer,
+        server_reader,
+        server_writer,
+        8192,
+        8192,
+        user,
+        Arc::clone(&stats),
+        None,
+        Arc::new(BufferPool::new()),
+    ));
+
+    let payload_size = 1024 * 10;
+    let s2c_payload = vec![0x41; payload_size];
+    let c2s_payload = vec![0x42; payload_size];
+
+    let s2c_handle = tokio::spawn(async move {
+        sp_writer.write_all(&s2c_payload).await.unwrap();
+        
+        let mut total_read = 0;
+        let mut buf = [0u8; 10];
+        while total_read < payload_size {
+            let n = cp_reader.read(&mut buf).await.unwrap();
+            total_read += n;
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+    });
+
+    let start = Instant::now();
+    cp_writer.write_all(&c2s_payload).await.unwrap();
+    
+    let mut server_buf = vec![0u8; payload_size];
+    sp_reader.read_exact(&mut server_buf).await.unwrap();
+    let elapsed = start.elapsed();
+
+    assert!(elapsed < Duration::from_millis(1000), "C->S must not be blocked by slow S->C (HOL blocking): {:?}", elapsed);
+    assert_eq!(server_buf, c2s_payload);
+
+    s2c_handle.abort();
+    relay_task.abort();
+}
+
+// ------------------------------------------------------------------
+// Priority 3: Data Quota Mid-Session Cutoff (OWASP ASVS 5.1.6)
+// ------------------------------------------------------------------
+
+#[tokio::test]
+async fn relay_quota_mid_session_cutoff() {
+    let stats = Arc::new(Stats::new());
+    let user = "quota-mid-user";
+    let quota = 5000;
+    
+    let (client_peer, relay_client) = duplex(8192);
+    let (relay_server, server_peer) = duplex(8192);
+
+    let (client_reader, client_writer) = tokio::io::split(relay_client);
+    let (server_reader, server_writer) = tokio::io::split(relay_server);
+    let (mut _cp_reader, mut cp_writer) = tokio::io::split(client_peer);
+    let (mut sp_reader, _sp_writer) = tokio::io::split(server_peer);
+
+    let relay_task = tokio::spawn(relay_bidirectional(
+        client_reader,
+        client_writer,
+        server_reader,
+        server_writer,
+        1024,
+        1024,
+        user,
+        Arc::clone(&stats),
+        Some(quota),
+        Arc::new(BufferPool::new()),
+    ));
+
+    // Send 4000 bytes (Ok)
+    let buf1 = vec![0x42; 4000];
+    cp_writer.write_all(&buf1).await.unwrap();
+    let mut server_recv = vec![0u8; 4000];
+    sp_reader.read_exact(&mut server_recv).await.unwrap();
+
+    // Send another 2000 bytes (Total 6000 > 5000)
+    let buf2 = vec![0x42; 2000];
+    let _ = cp_writer.write_all(&buf2).await;
+    
+    let relay_res = timeout(Duration::from_secs(1), relay_task).await.unwrap();
+    
+    match relay_res {
+        Ok(Err(ProxyError::DataQuotaExceeded { .. })) => {
+            // Expected
+        }
+        other => panic!("Expected DataQuotaExceeded error, got: {:?}", other),
+    }
+
+    let mut small_buf = [0u8; 1];
+    let n = sp_reader.read(&mut small_buf).await.unwrap();
+    assert_eq!(n, 0, "Server must see EOF after quota reached");
+}