refactor: migrate from session nonces to UUID-based stream registry

src/client/actor/download_loop.rs

@@ -34,7 +34,7 @@ pub struct DownloadLoopActor {
     write_half: tokio::net::tcp::OwnedWriteHalf,
     cipher: Option<Arc<dyn FrameCipher>>,
     encoding: EncodingType,
-    cookie_val: String,
+    stream_id: String,
     http_client: Arc<wreq::Client>,
     state: Arc<SharedState>,
     max_bytes: Option<u64>,
@@ -46,7 +46,7 @@ impl DownloadLoopActor {
         initial_response: wreq::Response,
         write_half: tokio::net::tcp::OwnedWriteHalf,
         cipher: Option<Arc<AesFrameCipher>>,
-        cookie_val: String,
+        stream_id: String,
         http_client: Arc<wreq::Client>,
         state: Arc<SharedState>,
     ) -> Self {
@@ -58,7 +58,7 @@ impl DownloadLoopActor {
         let use_prefetch = prefetch_at.is_some_and(|at| at > 0);
 
         let (prefetch_trigger, prefetch_rx) = if rotate_enabled && use_prefetch {
-            let (tx, rx) = spawn_prefetch_continuation(&http_client, &state, &cookie_val);
+            let (tx, rx) = spawn_prefetch_continuation(&http_client, &state, &stream_id);
             (Some(tx), Some(rx))
         } else {
             (None, None)
@@ -68,7 +68,7 @@ impl DownloadLoopActor {
             write_half,
             cipher: cipher_dyn,
             encoding,
-            cookie_val,
+            stream_id,
             http_client,
             state,
             max_bytes,
@@ -151,15 +151,15 @@ impl DownloadLoopActor {
             match tokio::time::timeout(PREFETCH_ROTATE_TIMEOUT, rx).await {
                 Ok(Ok(Ok(resp))) => resp,
                 Ok(Ok(Err(_))) | Ok(Err(_)) => {
-                    send_continue_request(&self.http_client, &self.state, &self.cookie_val).await?
+                    send_continue_request(&self.http_client, &self.state, &self.stream_id).await?
                 }
                 Err(_elapsed) => {
                     warn!("prefetch timed out, falling back to synchronous continue");
-                    send_continue_request(&self.http_client, &self.state, &self.cookie_val).await?
+                    send_continue_request(&self.http_client, &self.state, &self.stream_id).await?
                 }
             }
         } else {
-            send_continue_request(&self.http_client, &self.state, &self.cookie_val).await?
+            send_continue_request(&self.http_client, &self.state, &self.stream_id).await?
         };
 
         let use_prefetch = self
@@ -168,7 +168,7 @@ impl DownloadLoopActor {
             .is_some_and(|at| at > 0);
         let (prefetch_trigger, prefetch_rx) = if use_prefetch {
             let (tx, rx) =
-                spawn_prefetch_continuation(&self.http_client, &self.state, &self.cookie_val);
+                spawn_prefetch_continuation(&self.http_client, &self.state, &self.stream_id);
             (Some(tx), Some(rx))
         } else {
             (None, None)
@@ -238,10 +238,10 @@ async fn download_single_response(
 async fn send_continue_request(
     http_client: &wreq::Client,
     state: &SharedState,
-    cookie_val: &str,
+    stream_id: &str,
 ) -> Result<wreq::Response> {
     let mut cookie = String::new();
-    utils::build_tunnel_cookie(&mut cookie, cookie_val);
+    utils::build_stream_cookie(&mut cookie, stream_id);
     let mut req = http_client
         .post(state.remote_str.as_str())
         .header("Cookie", cookie);
@@ -259,7 +259,7 @@ async fn send_continue_request(
 fn spawn_prefetch_continuation(
     http_client: &Arc<wreq::Client>,
     state: &Arc<SharedState>,
-    cookie_val: &str,
+    stream_id: &str,
 ) -> (
     oneshot::Sender<()>,
     oneshot::Receiver<Result<wreq::Response>>,
@@ -268,13 +268,13 @@ fn spawn_prefetch_continuation(
     let (result_tx, result_rx) = oneshot::channel();
     let pre_client = Arc::clone(http_client);
     let pre_state = Arc::clone(state);
-    let pre_cookie = cookie_val.to_owned();
+    let pre_stream_id = stream_id.to_owned();
     tokio::spawn(
         async move {
             if trigger_rx.await.is_err() {
                 return;
             }
-            match send_continue_request(&pre_client, &pre_state, &pre_cookie).await {
+            match send_continue_request(&pre_client, &pre_state, &pre_stream_id).await {
                 Ok(resp) => {
                     let _ = result_tx.send(Ok(resp));
                 }

src/client/actor/upload_loop.rs

@@ -35,7 +35,7 @@ enum Phase {
 pub struct UploadLoopActor {
     http_client: Arc<wreq::Client>,
     state: Arc<SharedState>,
-    session_cookie: String,
+    stream_id: String,
     shaped: ShaperStream,
     request_sem: Arc<Semaphore>,
     bytes_sem: Arc<Semaphore>,
@@ -50,7 +50,7 @@ impl UploadLoopActor {
         initial_payload: Bytes,
         read_half: tokio::net::tcp::OwnedReadHalf,
         cipher: Option<Arc<AesFrameCipher>>,
-        session_cookie: String,
+        stream_id: String,
         start_seq: u64,
     ) -> Self {
         let reader = AsyncReadExt::chain(std::io::Cursor::new(initial_payload), read_half);
@@ -65,7 +65,7 @@ impl UploadLoopActor {
         Self {
             http_client,
             state,
-            session_cookie,
+            stream_id,
             shaped,
             request_sem: Arc::new(Semaphore::new(UPLOAD_CONCURRENCY)),
             bytes_sem: Arc::new(Semaphore::new(MAX_IN_FLIGHT_BYTES)),
@@ -181,12 +181,12 @@ impl UploadLoopActor {
         let body = batch_buf.freeze();
         let http_client = Arc::clone(&self.http_client);
         let state_ref = Arc::clone(&self.state);
-        let session_val = self.session_cookie.clone();
+        let stream_id = self.stream_id.clone();
         self.tasks.spawn(
             async move {
                 let _req_guard = req_permit;
                 let _bytes = bytes_permits;
-                send_upload_post(&http_client, &state_ref, body, &session_val).await
+                send_upload_post(&http_client, &state_ref, body, &stream_id).await
             }
             .instrument(tracing::Span::current()),
         );
@@ -230,11 +230,11 @@ async fn send_upload_post(
     http_client: &wreq::Client,
     state: &SharedState,
     body: Bytes,
-    session_cookie_val: &str,
+    stream_id: &str,
 ) -> Result<()> {
     debug_assert!(!body.is_empty(), "empty upload body");
     let mut cookie = String::new();
-    utils::build_tunnel_cookie(&mut cookie, session_cookie_val);
+    utils::build_stream_cookie(&mut cookie, stream_id);
     let mut req = http_client
         .post(state.remote_str.as_str())
         .header("Accept-Encoding", "identity")

src/client/connection.rs

@@ -21,7 +21,7 @@ pub(crate) async fn handle_plain_proxy(
 ) -> Result<()> {
     let stream_id = uuid::Uuid::new_v4().to_string();
     let mut cookie = String::new();
-    utils::build_tunnel_cookie(&mut cookie, &stream_id);
+    utils::build_stream_cookie(&mut cookie, &stream_id);
 
     let (early_data, remaining_payload, frames_sent) = utils::encode_initial_payload(
         &payload,
@@ -30,7 +30,7 @@ pub(crate) async fn handle_plain_proxy(
         &state.traffic_config,
     )?;
 
-    info!(target = %target_host, "connection initiated");
+    info!(stream_id = %stream_id, target = %target_host, "connection initiated");
 
     let response = tokio::time::timeout(
         DOWNLOAD_CONNECT_TIMEOUT,
@@ -64,7 +64,7 @@ pub(crate) async fn handle_plain_proxy(
         response,
         write_half,
         None,
-        stream_id.to_owned(),
+        stream_id,
         Arc::clone(&http_client),
         Arc::clone(&state),
     );

src/client/handshake.rs

@@ -49,22 +49,23 @@ pub async fn try_pq_connect(
     let session_id = &ticket.session_id;
     info!(session_id = %session_id, target = %target_host, "session resumption: attempting to reuse session");
 
-    let conn_nonce: [u8; 16] = rand::rng().random();
+    let stream_id = uuid::Uuid::new_v4();
+    let stream_id_bytes: [u8; 16] = *stream_id.as_bytes();
     let (upload_key, download_key, target_key) =
-        crypto::derive_connection_keys(master, &conn_nonce);
+        crypto::derive_connection_keys(master, &stream_id_bytes);
     let upload_cipher = Arc::new(AesFrameCipher::new(&upload_key));
     let download_cipher = Arc::new(AesFrameCipher::new(&download_key));
 
     let enc_target = crypto::encrypt_bytes(&target_key, target_host.as_bytes())?;
 
-    let cookie_nonce_key = crypto::derive_cookie_nonce_key(master);
-    let enc_conn_nonce = crypto::encrypt_bytes(&cookie_nonce_key, &conn_nonce)?;
+    let cookie_stream_key = crypto::derive_cookie_stream_key(master);
+    let enc_stream_id = crypto::encrypt_bytes(&cookie_stream_key, &stream_id_bytes)?;
 
     let cookie_val = format!(
         "{}:{}:{}",
         session_id,
         base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(&enc_target),
-        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(&enc_conn_nonce)
+        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(&enc_stream_id)
     );
 
     let (early_data, remaining_payload, frames_sent) = utils::encode_initial_payload(
@@ -108,15 +109,15 @@ pub async fn try_pq_connect(
     let upload_client = Arc::clone(http_client);
     let upload_state = Arc::clone(state);
     let upload_cipher_clone = Arc::clone(&upload_cipher);
-    let session_cookie_val = cookie_val.clone();
+    let stream_id_str = stream_id.to_string();
 
     let upload_actor = UploadLoopActor::new(
         upload_client.clone(),
         upload_state.clone(),
         remaining_payload,
         read_half,
         Some(upload_cipher_clone),
-        session_cookie_val,
+        stream_id_str.clone(),
         frames_sent,
     );
     let upload_task =
@@ -126,7 +127,7 @@ pub async fn try_pq_connect(
         response,
         write_half,
         Some(download_cipher),
-        cookie_val.clone(),
+        stream_id_str,
         Arc::clone(http_client),
         Arc::clone(state),
     );

src/client/utils.rs

@@ -9,17 +9,28 @@ use crate::client::constants::{MIN_PADDING, PADDING_POOL};
 use crate::shaper::{self, FrameCipher};
 
 #[inline]
-pub fn build_tunnel_cookie(buf: &mut String, session_val: &str) {
+fn build_cookie_into(buf: &mut String, name: &str, value: &str) {
     buf.clear();
-    let cap = 8 + session_val.len() + MIN_PADDING + PADDING_POOL.len();
+    let cap = name.len() + 1 + value.len() + 2 + MIN_PADDING + PADDING_POOL.len();
     buf.reserve(cap);
-    buf.push_str("session=");
-    buf.push_str(session_val);
+    buf.push_str(name);
+    buf.push('=');
+    buf.push_str(value);
     buf.push_str("; ");
     let padding_len = rand::rng().random_range(MIN_PADDING..PADDING_POOL.len());
     buf.push_str(std::str::from_utf8(&PADDING_POOL[..padding_len]).expect("Invalid UTF-8"))
 }
 
+#[inline]
+pub fn build_tunnel_cookie(buf: &mut String, session_val: &str) {
+    build_cookie_into(buf, "session", session_val)
+}
+
+#[inline]
+pub fn build_stream_cookie(buf: &mut String, stream_id: &str) {
+    build_cookie_into(buf, "stream", stream_id)
+}
+
 pub fn encode_initial_payload(
     initial_payload: &[u8],
     max_bytes: usize,

src/crypto/handshake.rs

@@ -21,18 +21,18 @@ pub fn derive_initial_master(mlkem_ss: &[u8], x25519_ss: &[u8; 32]) -> Zeroizing
     master
 }
 
-pub fn derive_cookie_nonce_key(master: &[u8; 32]) -> Zeroizing<[u8; 32]> {
+pub fn derive_cookie_stream_key(master: &[u8; 32]) -> Zeroizing<[u8; 32]> {
     let hkdf = Hkdf::<Sha256>::new(None, master);
     let mut key = Zeroizing::new([0u8; 32]);
-    hkdf.expand(b"cookie_nonce_key", &mut *key)
+    hkdf.expand(b"cookie_stream_key", &mut *key)
         .expect("32 bytes is valid for HKDF");
     key
 }
 
-pub fn derive_connection_keys(master: &[u8; 32], conn_nonce: &[u8; 16]) -> super::ConnectionKeys {
+pub fn derive_connection_keys(master: &[u8; 32], stream_id: &[u8; 16]) -> super::ConnectionKeys {
     let hkdf = Hkdf::<Sha256>::new(None, master);
     let mut info = Vec::with_capacity(16 + 15);
-    info.extend_from_slice(conn_nonce);
+    info.extend_from_slice(stream_id);
     info.extend_from_slice(b"connection_keys");
     let mut buf = Zeroizing::new([0u8; 96]);
     hkdf.expand(&info, &mut *buf)

src/crypto/mod.rs

@@ -4,7 +4,7 @@ mod keys;
 
 pub use cipher::{AesFrameCipher, decrypt_bytes, encrypt_bytes};
 pub use handshake::{
-    derive_connection_keys, derive_cookie_nonce_key, derive_handshake_key, derive_initial_master,
+    derive_connection_keys, derive_cookie_stream_key, derive_handshake_key, derive_initial_master,
 };
 pub use keys::{
     b64_to_private_key, b64_to_public_key, bytes_to_encapsulation_key, diffie_hellman,