Merge b6a30c1b51 into ad2057ad44

tls_front/emulator: hash compact cert info payload before TLS emulation

docs/QUICK_START_GUIDE.en.md

@@ -150,7 +150,7 @@ systemctl daemon-reload
 
 **7.** To get the link(s), enter:
 ```bash
-curl -s http://127.0.0.1:9091/v1/users | jq
+curl -s http://127.0.0.1:9091/v1/users | jq -r '.data[] | "User: \(.username)\n\(.links.tls[0] // empty)"'
 ```
 
 > Any number of people can use one link.

docs/QUICK_START_GUIDE.ru.md

@@ -150,7 +150,7 @@ systemctl daemon-reload
 
 **7.** Для получения ссылки/ссылок введите 
 ```bash
-curl -s http://127.0.0.1:9091/v1/users | jq
+curl -s http://127.0.0.1:9091/v1/users | jq -r '.data[] | "User: \(.username)\n\(.links.tls[0] // empty)"'
 ```
 > Одной ссылкой может пользоваться сколько угодно человек.
 

src/proxy/relay.rs

@@ -270,6 +270,7 @@ const QUOTA_NEAR_LIMIT_BYTES: u64 = 64 * 1024;
 const QUOTA_LARGE_CHARGE_BYTES: u64 = 16 * 1024;
 const QUOTA_ADAPTIVE_INTERVAL_MIN_BYTES: u64 = 4 * 1024;
 const QUOTA_ADAPTIVE_INTERVAL_MAX_BYTES: u64 = 64 * 1024;
+const QUOTA_RESERVE_SPIN_RETRIES: usize = 64;
 
 #[inline]
 fn quota_adaptive_interval_bytes(remaining_before: u64) -> u64 {
@@ -314,6 +315,50 @@ impl<S: AsyncRead + Unpin> AsyncRead for StatsIo<S> {
                 if n > 0 {
                     let n_to_charge = n as u64;
 
+                    if let (Some(limit), Some(remaining)) = (this.quota_limit, remaining_before) {
+                        let mut reserved_total = None;
+                        let mut reserve_rounds = 0usize;
+                        while reserved_total.is_none() {
+                            for _ in 0..QUOTA_RESERVE_SPIN_RETRIES {
+                                match this.user_stats.quota_try_reserve(n_to_charge, limit) {
+                                    Ok(total) => {
+                                        reserved_total = Some(total);
+                                        break;
+                                    }
+                                    Err(crate::stats::QuotaReserveError::LimitExceeded) => {
+                                        this.quota_exceeded.store(true, Ordering::Release);
+                                        buf.set_filled(before);
+                                        return Poll::Ready(Err(quota_io_error()));
+                                    }
+                                    Err(crate::stats::QuotaReserveError::Contended) => {
+                                        std::hint::spin_loop();
+                                    }
+                                }
+                            }
+                            reserve_rounds = reserve_rounds.saturating_add(1);
+                            if reserved_total.is_none() && reserve_rounds >= 8 {
+                                this.quota_exceeded.store(true, Ordering::Release);
+                                buf.set_filled(before);
+                                return Poll::Ready(Err(quota_io_error()));
+                            }
+                        }
+
+                        if should_immediate_quota_check(remaining, n_to_charge) {
+                            this.quota_bytes_since_check = 0;
+                        } else {
+                            this.quota_bytes_since_check =
+                                this.quota_bytes_since_check.saturating_add(n_to_charge);
+                            let interval = quota_adaptive_interval_bytes(remaining);
+                            if this.quota_bytes_since_check >= interval {
+                                this.quota_bytes_since_check = 0;
+                            }
+                        }
+
+                        if reserved_total.unwrap_or(0) >= limit {
+                            this.quota_exceeded.store(true, Ordering::Release);
+                        }
+                    }
+
                     // C→S: client sent data
                     this.counters
                         .c2s_bytes
@@ -326,27 +371,6 @@ impl<S: AsyncRead + Unpin> AsyncRead for StatsIo<S> {
                     this.stats
                         .increment_user_msgs_from_handle(this.user_stats.as_ref());
 
-                    if let (Some(limit), Some(remaining)) = (this.quota_limit, remaining_before) {
-                        this.stats
-                            .quota_charge_post_write(this.user_stats.as_ref(), n_to_charge);
-                        if should_immediate_quota_check(remaining, n_to_charge) {
-                            this.quota_bytes_since_check = 0;
-                            if this.user_stats.quota_used() >= limit {
-                                this.quota_exceeded.store(true, Ordering::Release);
-                            }
-                        } else {
-                            this.quota_bytes_since_check =
-                                this.quota_bytes_since_check.saturating_add(n_to_charge);
-                            let interval = quota_adaptive_interval_bytes(remaining);
-                            if this.quota_bytes_since_check >= interval {
-                                this.quota_bytes_since_check = 0;
-                                if this.user_stats.quota_used() >= limit {
-                                    this.quota_exceeded.store(true, Ordering::Release);
-                                }
-                            }
-                        }
-                    }
-
                     trace!(user = %this.user, bytes = n, "C->S");
                 }
                 Poll::Ready(Ok(()))
@@ -368,7 +392,44 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
         }
 
         let mut remaining_before = None;
+        let mut reserved_bytes = 0u64;
+        let mut write_buf = buf;
         if let Some(limit) = this.quota_limit {
+            if !buf.is_empty() {
+                let mut reserve_rounds = 0usize;
+                while reserved_bytes == 0 {
+                    let used_before = this.user_stats.quota_used();
+                    let remaining = limit.saturating_sub(used_before);
+                    if remaining == 0 {
+                        this.quota_exceeded.store(true, Ordering::Release);
+                        return Poll::Ready(Err(quota_io_error()));
+                    }
+                    remaining_before = Some(remaining);
+
+                    let desired = remaining.min(buf.len() as u64);
+                    for _ in 0..QUOTA_RESERVE_SPIN_RETRIES {
+                        match this.user_stats.quota_try_reserve(desired, limit) {
+                            Ok(_) => {
+                                reserved_bytes = desired;
+                                write_buf = &buf[..desired as usize];
+                                break;
+                            }
+                            Err(crate::stats::QuotaReserveError::LimitExceeded) => {
+                                break;
+                            }
+                            Err(crate::stats::QuotaReserveError::Contended) => {
+                                std::hint::spin_loop();
+                            }
+                        }
+                    }
+
+                    reserve_rounds = reserve_rounds.saturating_add(1);
+                    if reserved_bytes == 0 && reserve_rounds >= 8 {
+                        this.quota_exceeded.store(true, Ordering::Release);
+                        return Poll::Ready(Err(quota_io_error()));
+                    }
+                }
+            } else {
                 let used_before = this.user_stats.quota_used();
                 let remaining = limit.saturating_sub(used_before);
                 if remaining == 0 {
@@ -377,9 +438,27 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
                 }
                 remaining_before = Some(remaining);
             }
+        }
 
-        match Pin::new(&mut this.inner).poll_write(cx, buf) {
+        match Pin::new(&mut this.inner).poll_write(cx, write_buf) {
             Poll::Ready(Ok(n)) => {
+                if reserved_bytes > n as u64 {
+                    let refund = reserved_bytes - n as u64;
+                    let mut current = this.user_stats.quota_used.load(Ordering::Relaxed);
+                    loop {
+                        let next = current.saturating_sub(refund);
+                        match this.user_stats.quota_used.compare_exchange_weak(
+                            current,
+                            next,
+                            Ordering::Relaxed,
+                            Ordering::Relaxed,
+                        ) {
+                            Ok(_) => break,
+                            Err(observed) => current = observed,
+                        }
+                    }
+                }
+
                 if n > 0 {
                     let n_to_charge = n as u64;
 
@@ -396,8 +475,6 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
                         .increment_user_msgs_to_handle(this.user_stats.as_ref());
 
                     if let (Some(limit), Some(remaining)) = (this.quota_limit, remaining_before) {
-                        this.stats
-                            .quota_charge_post_write(this.user_stats.as_ref(), n_to_charge);
                         if should_immediate_quota_check(remaining, n_to_charge) {
                             this.quota_bytes_since_check = 0;
                             if this.user_stats.quota_used() >= limit {
@@ -420,7 +497,42 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
                 }
                 Poll::Ready(Ok(n))
             }
-            other => other,
+            Poll::Ready(Err(err)) => {
+                if reserved_bytes > 0 {
+                    let mut current = this.user_stats.quota_used.load(Ordering::Relaxed);
+                    loop {
+                        let next = current.saturating_sub(reserved_bytes);
+                        match this.user_stats.quota_used.compare_exchange_weak(
+                            current,
+                            next,
+                            Ordering::Relaxed,
+                            Ordering::Relaxed,
+                        ) {
+                            Ok(_) => break,
+                            Err(observed) => current = observed,
+                        }
+                    }
+                }
+                Poll::Ready(Err(err))
+            }
+            Poll::Pending => {
+                if reserved_bytes > 0 {
+                    let mut current = this.user_stats.quota_used.load(Ordering::Relaxed);
+                    loop {
+                        let next = current.saturating_sub(reserved_bytes);
+                        match this.user_stats.quota_used.compare_exchange_weak(
+                            current,
+                            next,
+                            Ordering::Relaxed,
+                            Ordering::Relaxed,
+                        ) {
+                            Ok(_) => break,
+                            Err(observed) => current = observed,
+                        }
+                    }
+                }
+                Poll::Pending
+            }
         }
     }
 

src/tls_front/emulator.rs

@@ -1,5 +1,6 @@
 #![allow(clippy::too_many_arguments)]
 
+use crc32fast::Hasher;
 use crate::crypto::{SecureRandom, sha256_hmac};
 use crate::protocol::constants::{
     MAX_TLS_CIPHERTEXT_SIZE, TLS_RECORD_APPLICATION, TLS_RECORD_CHANGE_CIPHER,
@@ -98,6 +99,31 @@ fn build_compact_cert_info_payload(cert_info: &ParsedCertificateInfo) -> Option<
     Some(payload)
 }
 
+fn hash_compact_cert_info_payload(cert_payload: Vec<u8>) -> Option<Vec<u8>> {
+    if cert_payload.is_empty() {
+        return None;
+    }
+
+    let mut hashed = Vec::with_capacity(cert_payload.len());
+    let mut seed_hasher = Hasher::new();
+    seed_hasher.update(&cert_payload);
+    let mut state = seed_hasher.finalize();
+
+    while hashed.len() < cert_payload.len() {
+        let mut hasher = Hasher::new();
+        hasher.update(&state.to_le_bytes());
+        hasher.update(&cert_payload);
+        state = hasher.finalize();
+
+        let block = state.to_le_bytes();
+        let remaining = cert_payload.len() - hashed.len();
+        let copy_len = remaining.min(block.len());
+        hashed.extend_from_slice(&block[..copy_len]);
+    }
+
+    Some(hashed)
+}
+
 /// Build a ServerHello + CCS + ApplicationData sequence using cached TLS metadata.
 pub fn build_emulated_server_hello(
     secret: &[u8],
@@ -190,7 +216,8 @@ pub fn build_emulated_server_hello(
     let compact_payload = cached
         .cert_info
         .as_ref()
-        .and_then(build_compact_cert_info_payload);
+        .and_then(build_compact_cert_info_payload)
+        .and_then(hash_compact_cert_info_payload);
     let selected_payload: Option<&[u8]> = if use_full_cert_payload {
         cached
             .cert_payload
@@ -221,7 +248,6 @@ pub fn build_emulated_server_hello(
             marker.extend_from_slice(proto);
             marker
         });
-    let mut payload_offset = 0usize;
     for (idx, size) in sizes.into_iter().enumerate() {
         let mut rec = Vec::with_capacity(5 + size);
         rec.push(TLS_RECORD_APPLICATION);
@@ -231,11 +257,10 @@ pub fn build_emulated_server_hello(
         if let Some(payload) = selected_payload {
             if size > 17 {
                 let body_len = size - 17;
-                let remaining = payload.len().saturating_sub(payload_offset);
+                let remaining = payload.len();
                 let copy_len = remaining.min(body_len);
                 if copy_len > 0 {
-                    rec.extend_from_slice(&payload[payload_offset..payload_offset + copy_len]);
+                    rec.extend_from_slice(&payload[..copy_len]);
-                    payload_offset += copy_len;
                 }
                 if body_len > copy_len {
                     rec.extend_from_slice(&rng.bytes(body_len - copy_len));
@@ -317,7 +342,9 @@ mod tests {
         CachedTlsData, ParsedServerHello, TlsBehaviorProfile, TlsCertPayload, TlsProfileSource,
     };
 
-    use super::build_emulated_server_hello;
+    use super::{
+        build_compact_cert_info_payload, build_emulated_server_hello, hash_compact_cert_info_payload,
+    };
     use crate::crypto::SecureRandom;
     use crate::protocol::constants::{
         TLS_RECORD_APPLICATION, TLS_RECORD_CHANGE_CIPHER, TLS_RECORD_HANDSHAKE,
@@ -432,7 +459,21 @@ mod tests {
         );
 
         let payload = first_app_data_payload(&response);
-        assert!(payload.starts_with(b"CN=example.com"));
+        let expected_hashed_payload = build_compact_cert_info_payload(
+            cached
+                .cert_info
+                .as_ref()
+                .expect("test fixture must provide certificate info"),
+        )
+        .and_then(hash_compact_cert_info_payload)
+        .expect("compact certificate info payload must be present for this test");
+        let copied_prefix_len = expected_hashed_payload
+            .len()
+            .min(payload.len().saturating_sub(17));
+        assert_eq!(
+            &payload[..copied_prefix_len],
+            &expected_hashed_payload[..copied_prefix_len]
+        );
     }
 
     #[test]