From 231555ce975be73c857b8611432c0d047602d6fd Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 14:09:25 -0500 Subject: [PATCH 01/10] add sizes for incoming data and use bounded lru instead of hashmap --- src/server.rs | 26 ++++++++++++++++++++++---- 1 file changed, 22 insertions(+), 4 deletions(-) diff --git a/src/server.rs b/src/server.rs index ae0c2d7..bc5980a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -47,12 +47,20 @@ impl QuiverInstance { } async fn handle_auth_stream(&mut self) -> Result { + const MAX_AUTH_TOKEN: u32 = 128; let Ok((mut send, mut recv)) = self.conn.accept_bi().await else { return Err(anyhow::anyhow!("connection closed before auth")); }; - - let api_key = String::from_utf8(recv.read_to_end(50).await?)?; - + let len = recv.read_u32().await?; + if len > MAX_AUTH_TOKEN { + return Err(anyhow::anyhow!("API key size of {} exceeds limit of {}", len, MAX_AUTH_TOKEN)); + } + if len == 0 { + return Err(anyhow::anyhow!("Received an empty API key.")); + } + let mut api_key_bytes = vec![0; len as usize]; + recv.read_exact(&mut api_key_bytes).await?; + let api_key = String::from_utf8(api_key_bytes)?; match self.authenticator.authenticate(&api_key).await { Ok((account_information, guard)) => { self.account_information = Some(account_information); @@ -71,11 +79,21 @@ impl QuiverInstance { } async fn handle_device_info_stream(&mut self) -> Result { + const MAX_DEVICE_INFO_SIZE: u32 = 4096; // 4 KB limit let Ok((mut send, mut recv)) = self.conn.accept_bi().await else { return Err(anyhow::anyhow!("connection closed before device info")); }; - let device_info = bincode::deserialize::(&recv.read_to_end(1024).await?)?; + // Read the length first and enforce limit + let len = recv.read_u32().await?; + if len > MAX_DEVICE_INFO_SIZE { + return Err(anyhow::anyhow!("Received device info exceeds size limit")); + } + + // Read the exact number of bytes + let mut buf = vec![0; len as usize]; + recv.read_exact(&mut buf).await?; + let device_info = bincode::deserialize::(&buf)?; let Some(api_key) = self.api_key.clone() else { return Err(anyhow::anyhow!("no api key")); }; From d49fd102ac67e5023c3e71e990da49d42012c34a Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 14:49:36 -0500 Subject: [PATCH 02/10] add length headers to client --- src/client.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/client.rs b/src/client.rs index de28e2c..fe5970a 100644 --- a/src/client.rs +++ b/src/client.rs @@ -49,7 +49,9 @@ impl QuiverClient { // --- Authentication Transaction --- info!("authenticating..."); let (mut send_auth, mut recv_auth) = self.conn.open_bi().await?; - send_auth.write_all(self.key.as_bytes()).await?; + let api_key_bytes = self.key.as_bytes(); + send_auth.write_u32(api_key_bytes.len() as u32).await?; + send_auth.write_all(api_key_bytes).await?; send_auth.finish()?; let auth_res = String::from_utf8(recv_auth.read_to_end(50).await?)?; if auth_res != "authenticated" { @@ -59,7 +61,9 @@ impl QuiverClient { // --- Device Info Transaction --- info!("sending device info..."); let (mut send_device, mut recv_device) = self.conn.open_bi().await?; - send_device.write_all(&bincode::serialize(&self.device_info)?).await?; + let device_info_bytes = bincode::serialize(&self.device_info)?; + send_device.write_u32(device_info_bytes.len() as u32).await?; + send_device.write_all(&device_info_bytes).await?; send_device.finish()?; let device_res = String::from_utf8(recv_device.read_to_end(50).await?)?; if device_res != "accepted" { From 028412b773b048d3521a569bb9ffc58ba653916c Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 14:53:26 -0500 Subject: [PATCH 03/10] read length from device info --- src/server.rs | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/src/server.rs b/src/server.rs index bc5980a..e790d05 100644 --- a/src/server.rs +++ b/src/server.rs @@ -83,17 +83,16 @@ impl QuiverInstance { let Ok((mut send, mut recv)) = self.conn.accept_bi().await else { return Err(anyhow::anyhow!("connection closed before device info")); }; - - // Read the length first and enforce limit let len = recv.read_u32().await?; if len > MAX_DEVICE_INFO_SIZE { - return Err(anyhow::anyhow!("Received device info exceeds size limit")); + return Err(anyhow::anyhow!("Device info size of {} exceeds limit of {}", len, MAX_DEVICE_INFO_SIZE)); } - - // Read the exact number of bytes - let mut buf = vec![0; len as usize]; - recv.read_exact(&mut buf).await?; - let device_info = bincode::deserialize::(&buf)?; + if len == 0 { + return Err(anyhow::anyhow!("Received empty device info.")); + } + let mut device_info_bytes = vec![0; len as usize]; + recv.read_exact(&mut device_info_bytes).await?; + let device_info = bincode::deserialize::(&device_info_bytes)?; let Some(api_key) = self.api_key.clone() else { return Err(anyhow::anyhow!("no api key")); }; From 5064c4b9641157607c29087133fa24bdecc810fe Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 15:44:28 -0500 Subject: [PATCH 04/10] longer timeout for server --- src/server.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/server.rs b/src/server.rs index e790d05..230bb9c 100644 --- a/src/server.rs +++ b/src/server.rs @@ -200,7 +200,7 @@ pub async fn run( let mut transport = TransportConfig::default(); transport.keep_alive_interval(Some(Duration::from_secs(2))); - transport.max_idle_timeout(Some(Duration::from_secs(5).try_into()?)); + transport.max_idle_timeout(Some(Duration::from_secs(60).try_into()?)); transport.max_concurrent_bidi_streams(quinn::VarInt::from_u32(50)); transport.max_concurrent_uni_streams(quinn::VarInt::from_u32(50)); transport.stream_receive_window(quinn::VarInt::from_u32(10_000_000)); From 52465b708943891a841339cee781e7ff68f14163 Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 17:40:23 -0500 Subject: [PATCH 05/10] fml --- src/server.rs | 8 -------- 1 file changed, 8 deletions(-) diff --git a/src/server.rs b/src/server.rs index 230bb9c..11bab27 100644 --- a/src/server.rs +++ b/src/server.rs @@ -47,14 +47,10 @@ impl QuiverInstance { } async fn handle_auth_stream(&mut self) -> Result { - const MAX_AUTH_TOKEN: u32 = 128; let Ok((mut send, mut recv)) = self.conn.accept_bi().await else { return Err(anyhow::anyhow!("connection closed before auth")); }; let len = recv.read_u32().await?; - if len > MAX_AUTH_TOKEN { - return Err(anyhow::anyhow!("API key size of {} exceeds limit of {}", len, MAX_AUTH_TOKEN)); - } if len == 0 { return Err(anyhow::anyhow!("Received an empty API key.")); } @@ -79,14 +75,10 @@ impl QuiverInstance { } async fn handle_device_info_stream(&mut self) -> Result { - const MAX_DEVICE_INFO_SIZE: u32 = 4096; // 4 KB limit let Ok((mut send, mut recv)) = self.conn.accept_bi().await else { return Err(anyhow::anyhow!("connection closed before device info")); }; let len = recv.read_u32().await?; - if len > MAX_DEVICE_INFO_SIZE { - return Err(anyhow::anyhow!("Device info size of {} exceeds limit of {}", len, MAX_DEVICE_INFO_SIZE)); - } if len == 0 { return Err(anyhow::anyhow!("Received empty device info.")); } From 62e8828f7e1f3b7a419aad6b216c7fc4990e2fe7 Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 20:18:23 -0500 Subject: [PATCH 06/10] add backward-compatible protocol versioning --- src/client.rs | 22 ++++++++++- src/server.rs | 104 ++++++++++++++++++++++++++++++++------------------ src/types.rs | 9 +++++ 3 files changed, 97 insertions(+), 38 deletions(-) diff --git a/src/client.rs b/src/client.rs index fe5970a..a81da85 100644 --- a/src/client.rs +++ b/src/client.rs @@ -14,7 +14,7 @@ use crate::types::{Template, Submission, SubmissionResponse}; use crate::new_job::NewJobConsumer; use crate::submission::{SubmissionProvider, SubmissionResponseHandler}; use crate::insecure::SkipServerVerification; - +use crate::types::CURRENT_VERSION; #[derive(Debug)] pub struct QuiverClient { @@ -45,7 +45,27 @@ impl QuiverClient { } } + async fn perform_version_handshake(&self) -> Result<()> { + info!("Negotiating protocol version..."); + let (mut send_version, mut recv_version) = self.conn.open_bi().await?; + + let ver_bytes = bincode::serialize(&CURRENT_VERSION)?; + send_version.write_u32(ver_bytes.len() as u32).await?; + send_version.write_all(&ver_bytes).await?; + send_version.finish()?; + + let response = String::from_utf8(recv_version.read_to_end(100).await?)?; + if response != "ok" { + return Err(anyhow::anyhow!("Protocol negotiation failed: {}", response)); + } + + info!("Protocol version compatible with server."); + Ok(()) + } + async fn serve(&mut self) -> Result<()> { + // attempt version handshake + self.perform_version_handshake().await?; // --- Authentication Transaction --- info!("authenticating..."); let (mut send_auth, mut recv_auth) = self.conn.open_bi().await?; diff --git a/src/server.rs b/src/server.rs index 11bab27..40f62ae 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,4 +1,4 @@ -use tracing::{info, error}; +use tracing::{info, warn, error}; use anyhow::Result; use quinn::{ServerConfig, Endpoint, Connection, TransportConfig}; use std::net::SocketAddr; @@ -14,6 +14,7 @@ use crate::device_info::{DeviceInfo, DeviceInfoUpdater}; use crate::new_job::NewJobProvider; use crate::types::{Template, Submission, SubmissionResponse, AccountInformation}; use crate::submission::SubmissionConsumer; +use crate::types::{CURRENT_VERSION, ProtocolVersion}; struct QuiverInstance { conn: Connection, @@ -46,34 +47,6 @@ impl QuiverInstance { } } - async fn handle_auth_stream(&mut self) -> Result { - let Ok((mut send, mut recv)) = self.conn.accept_bi().await else { - return Err(anyhow::anyhow!("connection closed before auth")); - }; - let len = recv.read_u32().await?; - if len == 0 { - return Err(anyhow::anyhow!("Received an empty API key.")); - } - let mut api_key_bytes = vec![0; len as usize]; - recv.read_exact(&mut api_key_bytes).await?; - let api_key = String::from_utf8(api_key_bytes)?; - match self.authenticator.authenticate(&api_key).await { - Ok((account_information, guard)) => { - self.account_information = Some(account_information); - self.connection_guard = Some(guard); - self.api_key = Some(api_key); - send.write_all(b"authenticated").await?; - send.finish()?; - Ok(true) - } - Err(_) => { - send.write_all(b"rejected").await?; - send.finish()?; - Ok(false) - } - } - } - async fn handle_device_info_stream(&mut self) -> Result { let Ok((mut send, mut recv)) = self.conn.accept_bi().await else { return Err(anyhow::anyhow!("connection closed before device info")); @@ -157,22 +130,79 @@ impl QuiverInstance { } async fn serve(&mut self) -> Result<()> { - // The first stream MUST be for authentication. - if !self.handle_auth_stream().await? { - // If auth fails, we close the connection immediately. - return Err(anyhow::anyhow!("Authentication failed")); + // The first stream must handle both versioning and auth for backward compatibility + let (mut send, mut recv) = self.conn.accept_bi().await?; + // heuristic to determine whether this is a handshake or legacy auth attempt + let initial_bytes = recv.read_u32().await; + const TOKEN_PREFIX: u32 = 1852793707; + + // We should remove this in the future + let api_key_bytes = match initial_bytes { + Ok(TOKEN_PREFIX) => { + warn!("Legacy authentication, no version handshake. Please update your miner!"); + let mut key_bytes = TOKEN_PREFIX.to_be_bytes().to_vec(); + key_bytes.extend(recv.read_to_end(50 - 4).await?); + key_bytes + }, + // assume this is a version handshake + Ok(len) => { + if len > 0 && len < 100 { + let mut ver_bytes = vec![0; len as usize]; + match recv.read_exact(&mut ver_bytes).await { + Ok(_) => { + match bincode::deserialize::(&ver_bytes) { + // Deserialized correctly and the major version matches + // If we have incompatible protocol changes in the future, fork here + Ok(client_version) if client_version.major == CURRENT_VERSION.major => { + info!("Handshake successful: v{}.{}.{}", client_version.major, client_version.minor, client_version.patch); + send.write_all(b"ok").await?; + send.finish()?; + let (_send_auth, mut recv_auth) = self.conn.accept_bi().await?; + let key_len = recv_auth.read_u32().await?; + // Check length + if key_len > 256 { + return Err(anyhow::anyhow!("Protocol error: new client key size of {} exceeds limit of 256", key_len)); + } + let mut key_bytes = vec![0; key_len as usize]; + recv_auth.read_exact(&mut key_bytes).await?; + key_bytes + }, + _ => return Err(anyhow::anyhow!("Incompatible protocol version or malformed handshake")), + } + }, + Err(_) => return Err(anyhow::anyhow!("Malformed protocol handshake. Please update your miner.")), + } + } else { + return Err(anyhow::anyhow!("Invalid protocol handshake length: {}", len)); + } + }, + Err(e) => { + error!("Failed to read initial handshake bytes: {}", e); + return Err(e.into()); + } + }; + + let api_key = String::from_utf8(api_key_bytes)?; + + // --- Authentication --- + match self.authenticator.authenticate(&api_key).await { + Ok((account_information, guard)) => { + self.account_information = Some(account_information); + self.connection_guard = Some(guard); + self.api_key = Some(api_key); + } + Err(e) => return Err(e), } - // After auth, receive device info. + // --- Device Info --- if !self.handle_device_info_stream().await? { return Err(anyhow::anyhow!("Device info rejected")); } - // take the last 8 characters of the api key let truncated_api_key = self.api_key.as_ref().unwrap().chars().rev().take(8).collect::(); - info!("{:?} put to work with key {:?}", self.account_information.as_ref().unwrap().user_uuid, truncated_api_key); + info!("{:?} device key {:?} put to work", self.account_information.as_ref().unwrap().user_uuid, truncated_api_key); - // Ready for work. + // --- Ready for work --- self.handle_work().await } } diff --git a/src/types.rs b/src/types.rs index 22a49e9..959119a 100644 --- a/src/types.rs +++ b/src/types.rs @@ -4,6 +4,15 @@ use bytes::Bytes; use nockapp::noun::slab::NounSlab; use uuid::Uuid; +pub const CURRENT_VERSION: ProtocolVersion = ProtocolVersion { major: 0, minor: 2, patch: 0 }; + +#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)] +pub struct ProtocolVersion { + pub major: u16, + pub minor: u16, + pub patch: u16, +} + #[derive(Serialize, Deserialize, Debug, Clone)] pub struct AccountInformation { pub user_uuid: Uuid, From 1a2fe50fc424d3f5440311fc3c887e57304b2cad Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 22:18:28 -0500 Subject: [PATCH 07/10] correctly handle device info --- src/server.rs | 121 ++++++++++++++++++++++++++++++-------------------- 1 file changed, 74 insertions(+), 47 deletions(-) diff --git a/src/server.rs b/src/server.rs index 40f62ae..1916295 100644 --- a/src/server.rs +++ b/src/server.rs @@ -130,52 +130,62 @@ impl QuiverInstance { } async fn serve(&mut self) -> Result<()> { - // The first stream must handle both versioning and auth for backward compatibility + // First stream: version (or legacy auth) let (mut send, mut recv) = self.conn.accept_bi().await?; - // heuristic to determine whether this is a handshake or legacy auth attempt - let initial_bytes = recv.read_u32().await; + let initial = recv.read_u32().await; const TOKEN_PREFIX: u32 = 1852793707; - - // We should remove this in the future - let api_key_bytes = match initial_bytes { + let mut send_auth_opt: Option = None; + let api_key_bytes: Vec = match initial { + // --- Legacy: API key was sent on the first stream --- Ok(TOKEN_PREFIX) => { warn!("Legacy authentication, no version handshake. Please update your miner!"); - let mut key_bytes = TOKEN_PREFIX.to_be_bytes().to_vec(); - key_bytes.extend(recv.read_to_end(50 - 4).await?); - key_bytes - }, - // assume this is a version handshake + // In legacy mode we reply on the same stream, so keep `send` as send_auth. + send_auth_opt = Some(send); + let mut key = TOKEN_PREFIX.to_be_bytes().to_vec(); + key.extend(recv.read_to_end(50 - 4).await?); + key + } + + // --- New protocol: version handshake, then separate auth stream --- Ok(len) => { - if len > 0 && len < 100 { - let mut ver_bytes = vec![0; len as usize]; - match recv.read_exact(&mut ver_bytes).await { - Ok(_) => { - match bincode::deserialize::(&ver_bytes) { - // Deserialized correctly and the major version matches - // If we have incompatible protocol changes in the future, fork here - Ok(client_version) if client_version.major == CURRENT_VERSION.major => { - info!("Handshake successful: v{}.{}.{}", client_version.major, client_version.minor, client_version.patch); - send.write_all(b"ok").await?; - send.finish()?; - let (_send_auth, mut recv_auth) = self.conn.accept_bi().await?; - let key_len = recv_auth.read_u32().await?; - // Check length - if key_len > 256 { - return Err(anyhow::anyhow!("Protocol error: new client key size of {} exceeds limit of 256", key_len)); - } - let mut key_bytes = vec![0; key_len as usize]; - recv_auth.read_exact(&mut key_bytes).await?; - key_bytes - }, - _ => return Err(anyhow::anyhow!("Incompatible protocol version or malformed handshake")), + if len == 0 || len >= 100 { + return Err(anyhow::anyhow!("Invalid protocol handshake length: {}", len)); + } + let mut ver_bytes = vec![0; len as usize]; + recv.read_exact(&mut ver_bytes).await + .map_err(|_| anyhow::anyhow!("Malformed protocol handshake. Please update your miner."))?; + match bincode::deserialize::(&ver_bytes) { + Ok(client_v) if client_v.major == CURRENT_VERSION.major => { + info!("Handshake successful: v{}.{}.{}", client_v.major, client_v.minor, client_v.patch); + // Complete the handshake stream so the client’s read_to_end() returns. + send.write_all(b"ok").await?; + send.finish()?; + + // Now accept auth stream + let (mut send_auth, mut recv_auth) = self.conn.accept_bi().await?; + // keep the send half so we can reply later + send_auth_opt = Some(send_auth); + + let key_len = recv_auth.read_u32().await?; + if key_len > 256 { + // Send an explicit failure and bail + if let Some(mut s) = send_auth_opt.take() { + let _ = s.write_all(b"invalid_key_len").await; + let _ = s.finish(); } - }, - Err(_) => return Err(anyhow::anyhow!("Malformed protocol handshake. Please update your miner.")), + return Err(anyhow::anyhow!( + "Protocol error: new client key size {} exceeds limit of 256", + key_len + )); + } + let mut key_bytes = vec![0; key_len as usize]; + recv_auth.read_exact(&mut key_bytes).await?; + key_bytes } - } else { - return Err(anyhow::anyhow!("Invalid protocol handshake length: {}", len)); + _ => return Err(anyhow::anyhow!("Incompatible protocol version or malformed handshake")), } - }, + } + Err(e) => { error!("Failed to read initial handshake bytes: {}", e); return Err(e.into()); @@ -185,24 +195,41 @@ impl QuiverInstance { let api_key = String::from_utf8(api_key_bytes)?; // --- Authentication --- - match self.authenticator.authenticate(&api_key).await { - Ok((account_information, guard)) => { - self.account_information = Some(account_information); - self.connection_guard = Some(guard); - self.api_key = Some(api_key); + let auth_result = self.authenticator.authenticate(&api_key).await; + // We must reply on the AUTH stream (legacy: first stream; new: dedicated auth stream) + if let Some(mut send_auth) = send_auth_opt.take() { + match auth_result { + Ok((account_information, guard)) => { + self.account_information = Some(account_information); + self.connection_guard = Some(guard); + self.api_key = Some(api_key); + send_auth.write_all(b"authenticated").await?; + send_auth.finish()?; + } + Err(e) => { + let _ = send_auth.write_all(b"unauthorized").await; + let _ = send_auth.finish(); + return Err(e); + } } - Err(e) => return Err(e), + } else { + // Shouldn't happen: we always set send_auth in both branches + return Err(anyhow::anyhow!("internal: missing auth send stream")); } - // --- Device Info --- + // --- Device Info (next bidi stream) --- if !self.handle_device_info_stream().await? { return Err(anyhow::anyhow!("Device info rejected")); } let truncated_api_key = self.api_key.as_ref().unwrap().chars().rev().take(8).collect::(); - info!("{:?} device key {:?} put to work", self.account_information.as_ref().unwrap().user_uuid, truncated_api_key); + info!( + "{:?} device key {:?} put to work", + self.account_information.as_ref().unwrap().user_uuid, + truncated_api_key + ); - // --- Ready for work --- + // --- Work loop --- self.handle_work().await } } From 33a1ace7d1182b78d490113a66af3aa9bebb6fcf Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 22:25:15 -0500 Subject: [PATCH 08/10] fix auth stream --- src/server.rs | 145 ++++++++++++++++++++++++++------------------------ 1 file changed, 75 insertions(+), 70 deletions(-) diff --git a/src/server.rs b/src/server.rs index 1916295..196ed7d 100644 --- a/src/server.rs +++ b/src/server.rs @@ -130,59 +130,93 @@ impl QuiverInstance { } async fn serve(&mut self) -> Result<()> { - // First stream: version (or legacy auth) - let (mut send, mut recv) = self.conn.accept_bi().await?; - let initial = recv.read_u32().await; + // 1) First bidi stream: version handshake OR legacy auth + let (mut send0, mut recv0) = self.conn.accept_bi().await?; const TOKEN_PREFIX: u32 = 1852793707; - let mut send_auth_opt: Option = None; - let api_key_bytes: Vec = match initial { - // --- Legacy: API key was sent on the first stream --- + + let first_u32 = recv0.read_u32().await; + match first_u32 { + // ----- LEGACY PATH: API key on first stream ----- Ok(TOKEN_PREFIX) => { warn!("Legacy authentication, no version handshake. Please update your miner!"); - // In legacy mode we reply on the same stream, so keep `send` as send_auth. - send_auth_opt = Some(send); + // Read the rest of the API key from the same stream. let mut key = TOKEN_PREFIX.to_be_bytes().to_vec(); - key.extend(recv.read_to_end(50 - 4).await?); - key + key.extend(recv0.read_to_end(50 - 4).await?); + let api_key = String::from_utf8(key)?; + + // Auth on the same stream + match self.authenticator.authenticate(&api_key).await { + Ok((acct, guard)) => { + self.account_information = Some(acct); + self.connection_guard = Some(guard); + self.api_key = Some(api_key); + send0.write_all(b"authenticated").await?; + send0.finish()?; + } + Err(e) => { + let _ = send0.write_all(b"unauthorized").await; + let _ = send0.finish(); + return Err(e); + } + } + + // Device-info stream (next bidi) + if !self.handle_device_info_stream().await? { + return Err(anyhow::anyhow!("Device info rejected")); + } } - // --- New protocol: version handshake, then separate auth stream --- + // ----- NEW PROTOCOL: version handshake, then separate auth stream ----- Ok(len) => { if len == 0 || len >= 100 { return Err(anyhow::anyhow!("Invalid protocol handshake length: {}", len)); } - let mut ver_bytes = vec![0; len as usize]; - recv.read_exact(&mut ver_bytes).await + let mut ver_buf = vec![0; len as usize]; + recv0.read_exact(&mut ver_buf).await .map_err(|_| anyhow::anyhow!("Malformed protocol handshake. Please update your miner."))?; - match bincode::deserialize::(&ver_bytes) { - Ok(client_v) if client_v.major == CURRENT_VERSION.major => { - info!("Handshake successful: v{}.{}.{}", client_v.major, client_v.minor, client_v.patch); - // Complete the handshake stream so the client’s read_to_end() returns. - send.write_all(b"ok").await?; - send.finish()?; + let client_v: ProtocolVersion = bincode::deserialize(&ver_buf) + .map_err(|_| anyhow::anyhow!("Malformed protocol handshake."))?; + if client_v.major != CURRENT_VERSION.major { + return Err(anyhow::anyhow!("Incompatible protocol version")); + } + info!("Handshake successful: v{}.{}.{}", client_v.major, client_v.minor, client_v.patch); + // Complete the handshake stream so client's read_to_end returns. + send0.write_all(b"ok").await?; + send0.finish()?; - // Now accept auth stream - let (mut send_auth, mut recv_auth) = self.conn.accept_bi().await?; - // keep the send half so we can reply later - send_auth_opt = Some(send_auth); + // 2) Auth stream: read key, authenticate, reply in same stream, finish stream. + let (mut send_auth, mut recv_auth) = self.conn.accept_bi().await?; + let key_len = recv_auth.read_u32().await?; + if key_len > 256 { + let _ = send_auth.write_all(b"invalid_key_len").await; + let _ = send_auth.finish(); + return Err(anyhow::anyhow!( + "Protocol error: new client key size {} exceeds limit of 256", + key_len + )); + } + let mut key_bytes = vec![0; key_len as usize]; + recv_auth.read_exact(&mut key_bytes).await?; + let api_key = String::from_utf8(key_bytes)?; - let key_len = recv_auth.read_u32().await?; - if key_len > 256 { - // Send an explicit failure and bail - if let Some(mut s) = send_auth_opt.take() { - let _ = s.write_all(b"invalid_key_len").await; - let _ = s.finish(); - } - return Err(anyhow::anyhow!( - "Protocol error: new client key size {} exceeds limit of 256", - key_len - )); - } - let mut key_bytes = vec![0; key_len as usize]; - recv_auth.read_exact(&mut key_bytes).await?; - key_bytes + match self.authenticator.authenticate(&api_key).await { + Ok((acct, guard)) => { + self.account_information = Some(acct); + self.connection_guard = Some(guard); + self.api_key = Some(api_key); + send_auth.write_all(b"authenticated").await?; + send_auth.finish()?; + } + Err(e) => { + let _ = send_auth.write_all(b"unauthorized").await; + let _ = send_auth.finish(); + return Err(e); } - _ => return Err(anyhow::anyhow!("Incompatible protocol version or malformed handshake")), + } + + // 3) Device-info stream + if !self.handle_device_info_stream().await? { + return Err(anyhow::anyhow!("Device info rejected")); } } @@ -190,38 +224,9 @@ impl QuiverInstance { error!("Failed to read initial handshake bytes: {}", e); return Err(e.into()); } - }; - - let api_key = String::from_utf8(api_key_bytes)?; - - // --- Authentication --- - let auth_result = self.authenticator.authenticate(&api_key).await; - // We must reply on the AUTH stream (legacy: first stream; new: dedicated auth stream) - if let Some(mut send_auth) = send_auth_opt.take() { - match auth_result { - Ok((account_information, guard)) => { - self.account_information = Some(account_information); - self.connection_guard = Some(guard); - self.api_key = Some(api_key); - send_auth.write_all(b"authenticated").await?; - send_auth.finish()?; - } - Err(e) => { - let _ = send_auth.write_all(b"unauthorized").await; - let _ = send_auth.finish(); - return Err(e); - } - } - } else { - // Shouldn't happen: we always set send_auth in both branches - return Err(anyhow::anyhow!("internal: missing auth send stream")); - } - - // --- Device Info (next bidi stream) --- - if !self.handle_device_info_stream().await? { - return Err(anyhow::anyhow!("Device info rejected")); } + // From here on, you’re authenticated and device info is accepted. let truncated_api_key = self.api_key.as_ref().unwrap().chars().rev().take(8).collect::(); info!( "{:?} device key {:?} put to work", @@ -229,9 +234,9 @@ impl QuiverInstance { truncated_api_key ); - // --- Work loop --- self.handle_work().await } + } pub async fn run( From 7b6a4362925133e78bf4ce78f5c609e23685f344 Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 23:00:14 -0500 Subject: [PATCH 09/10] length prefixing for negotiation and device info --- src/client.rs | 50 ++++++++++++++++++++++----------------- src/server.rs | 65 ++++++++++++++++++++++++++++++++------------------- 2 files changed, 70 insertions(+), 45 deletions(-) diff --git a/src/client.rs b/src/client.rs index a81da85..078793e 100644 --- a/src/client.rs +++ b/src/client.rs @@ -54,77 +54,85 @@ impl QuiverClient { send_version.write_all(&ver_bytes).await?; send_version.finish()?; - let response = String::from_utf8(recv_version.read_to_end(100).await?)?; + // read length-prefixed "ok" + let ok_len = recv_version.read_u32().await?; + let mut ok_buf = vec![0; ok_len as usize]; + recv_version.read_exact(&mut ok_buf).await?; + let response = String::from_utf8(ok_buf)?; + if response != "ok" { return Err(anyhow::anyhow!("Protocol negotiation failed: {}", response)); } - info!("Protocol version compatible with server."); Ok(()) } async fn serve(&mut self) -> Result<()> { - // attempt version handshake + // 1) Version handshake self.perform_version_handshake().await?; - // --- Authentication Transaction --- + + // 2) Authentication info!("authenticating..."); let (mut send_auth, mut recv_auth) = self.conn.open_bi().await?; let api_key_bytes = self.key.as_bytes(); send_auth.write_u32(api_key_bytes.len() as u32).await?; send_auth.write_all(api_key_bytes).await?; send_auth.finish()?; - let auth_res = String::from_utf8(recv_auth.read_to_end(50).await?)?; + + // read length-prefixed auth reply + let n = recv_auth.read_u32().await?; + let mut buf = vec![0; n as usize]; + recv_auth.read_exact(&mut buf).await?; + let auth_res = String::from_utf8(buf)?; if auth_res != "authenticated" { return Err(anyhow::anyhow!("Authentication failed: {}", auth_res)); } - // --- Device Info Transaction --- + // 3) Device info info!("sending device info..."); let (mut send_device, mut recv_device) = self.conn.open_bi().await?; let device_info_bytes = bincode::serialize(&self.device_info)?; send_device.write_u32(device_info_bytes.len() as u32).await?; send_device.write_all(&device_info_bytes).await?; send_device.finish()?; - let device_res = String::from_utf8(recv_device.read_to_end(50).await?)?; + + // read length-prefixed device-info reply + let n = recv_device.read_u32().await?; + let mut buf = vec![0; n as usize]; + recv_device.read_exact(&mut buf).await?; + let device_res = String::from_utf8(buf)?; if device_res != "accepted" { return Err(anyhow::anyhow!("Device info rejected: {}", device_res)); } info!("Client authenticated and ready for work."); - // Spawn the background task to receive new jobs. + // --- Your existing loop unchanged --- let mut job_handle = tokio::spawn(receive_jobs(self.conn.clone(), self.new_job_consumer.clone())); - loop { tokio::select! { - // Use a biased select to ensure we always check for connection - // closure and task failure before waiting on a new submission. biased; - // Branch 1: The connection dies for any reason. reason = self.conn.closed() => { error!("Connection closed: {:?}", reason); - job_handle.abort(); // Stop the background job stream. + job_handle.abort(); return Err(anyhow::anyhow!("Connection closed by server or network")); }, - // Branch 2: The background job receiver task fails or panics. res = &mut job_handle => { - match res { + match res { Ok(Ok(_)) => error!("Job stream closed unexpectedly."), Ok(Err(e)) => error!("Job stream failed: {}", e), Err(e) => error!("Job stream task panicked: {}", e), - } - return Err(anyhow::anyhow!("Job stream failed, disconnecting.")); + } + return Err(anyhow::anyhow!("Job stream failed, disconnecting.")); }, - // Branch 3: The submission provider (miner) has a new proof to send. submission_result = self.submission_provider.submit() => { match submission_result { Ok(submission) => { let conn = self.conn.clone(); let handler = self.submission_response_handler.clone(); - // Spawn a task to handle this specific submission. tokio::spawn(async move { if let Err(e) = handle_one_submission(conn, submission, handler).await { error!("Failed to process submission: {}", e); @@ -133,8 +141,8 @@ impl QuiverClient { } Err(e) => { error!("Submission provider failed critically: {}", e); - job_handle.abort(); // Clean up background task. - return Err(e); // Propagate critical errors. + job_handle.abort(); + return Err(e); } } }, diff --git a/src/server.rs b/src/server.rs index 196ed7d..a03c091 100644 --- a/src/server.rs +++ b/src/server.rs @@ -51,30 +51,38 @@ impl QuiverInstance { let Ok((mut send, mut recv)) = self.conn.accept_bi().await else { return Err(anyhow::anyhow!("connection closed before device info")); }; + let len = recv.read_u32().await?; - if len == 0 { - return Err(anyhow::anyhow!("Received empty device info.")); + if len == 0 || len > 1_000_000 { + return Err(anyhow::anyhow!("Device info size out of bounds: {}", len)); } + let mut device_info_bytes = vec![0; len as usize]; recv.read_exact(&mut device_info_bytes).await?; let device_info = bincode::deserialize::(&device_info_bytes)?; + let Some(api_key) = self.api_key.clone() else { return Err(anyhow::anyhow!("no api key")); }; match self.device_info_updater.update_device_info(&device_info, &api_key).await { Ok(_) => { - send.write_all(b"accepted").await?; + let msg = b"accepted"; + send.write_u32(msg.len() as u32).await?; + send.write_all(msg).await?; send.finish()?; Ok(true) } Err(_) => { - send.write_all(b"rejected").await?; + let msg = b"rejected"; + send.write_u32(msg.len() as u32).await?; + send.write_all(msg).await?; send.finish()?; Ok(false) } } } + async fn handle_work(&mut self) -> Result<()> { // New job pusher let new_job_provider = self.new_job_provider.clone(); @@ -130,43 +138,44 @@ impl QuiverInstance { } async fn serve(&mut self) -> Result<()> { - // 1) First bidi stream: version handshake OR legacy auth + // First bidi stream: version handshake OR legacy auth let (mut send0, mut recv0) = self.conn.accept_bi().await?; const TOKEN_PREFIX: u32 = 1852793707; let first_u32 = recv0.read_u32().await; match first_u32 { - // ----- LEGACY PATH: API key on first stream ----- + // ----- LEGACY: API key in first stream ----- Ok(TOKEN_PREFIX) => { warn!("Legacy authentication, no version handshake. Please update your miner!"); - // Read the rest of the API key from the same stream. let mut key = TOKEN_PREFIX.to_be_bytes().to_vec(); key.extend(recv0.read_to_end(50 - 4).await?); let api_key = String::from_utf8(key)?; - // Auth on the same stream match self.authenticator.authenticate(&api_key).await { Ok((acct, guard)) => { self.account_information = Some(acct); self.connection_guard = Some(guard); self.api_key = Some(api_key); - send0.write_all(b"authenticated").await?; + let msg = b"authenticated"; + send0.write_u32(msg.len() as u32).await?; + send0.write_all(msg).await?; send0.finish()?; } Err(e) => { - let _ = send0.write_all(b"unauthorized").await; + let msg = b"unauthorized"; + let _ = send0.write_u32(msg.len() as u32).await; + let _ = send0.write_all(msg).await; let _ = send0.finish(); return Err(e); } } - // Device-info stream (next bidi) if !self.handle_device_info_stream().await? { return Err(anyhow::anyhow!("Device info rejected")); } } - // ----- NEW PROTOCOL: version handshake, then separate auth stream ----- + // ----- NEW PROTOCOL: version then separate auth stream ----- Ok(len) => { if len == 0 || len >= 100 { return Err(anyhow::anyhow!("Invalid protocol handshake length: {}", len)); @@ -176,24 +185,28 @@ impl QuiverInstance { .map_err(|_| anyhow::anyhow!("Malformed protocol handshake. Please update your miner."))?; let client_v: ProtocolVersion = bincode::deserialize(&ver_buf) .map_err(|_| anyhow::anyhow!("Malformed protocol handshake."))?; + if client_v.major != CURRENT_VERSION.major { return Err(anyhow::anyhow!("Incompatible protocol version")); } + info!("Handshake successful: v{}.{}.{}", client_v.major, client_v.minor, client_v.patch); - // Complete the handshake stream so client's read_to_end returns. - send0.write_all(b"ok").await?; + + // send length-prefixed "ok" and finish handshake stream + let ok = b"ok"; + send0.write_u32(ok.len() as u32).await?; + send0.write_all(ok).await?; send0.finish()?; - // 2) Auth stream: read key, authenticate, reply in same stream, finish stream. + // Auth stream: read key, authenticate, reply, finish. let (mut send_auth, mut recv_auth) = self.conn.accept_bi().await?; let key_len = recv_auth.read_u32().await?; if key_len > 256 { - let _ = send_auth.write_all(b"invalid_key_len").await; + let msg = b"invalid_key_len"; + let _ = send_auth.write_u32(msg.len() as u32).await; + let _ = send_auth.write_all(msg).await; let _ = send_auth.finish(); - return Err(anyhow::anyhow!( - "Protocol error: new client key size {} exceeds limit of 256", - key_len - )); + return Err(anyhow::anyhow!("Protocol error: key size {} exceeds 256", key_len)); } let mut key_bytes = vec![0; key_len as usize]; recv_auth.read_exact(&mut key_bytes).await?; @@ -204,17 +217,21 @@ impl QuiverInstance { self.account_information = Some(acct); self.connection_guard = Some(guard); self.api_key = Some(api_key); - send_auth.write_all(b"authenticated").await?; + let msg = b"authenticated"; + send_auth.write_u32(msg.len() as u32).await?; + send_auth.write_all(msg).await?; send_auth.finish()?; } Err(e) => { - let _ = send_auth.write_all(b"unauthorized").await; + let msg = b"unauthorized"; + let _ = send_auth.write_u32(msg.len() as u32).await; + let _ = send_auth.write_all(msg).await; let _ = send_auth.finish(); return Err(e); } } - // 3) Device-info stream + // Device-info stream (next bidi) if !self.handle_device_info_stream().await? { return Err(anyhow::anyhow!("Device info rejected")); } @@ -226,7 +243,7 @@ impl QuiverInstance { } } - // From here on, you’re authenticated and device info is accepted. + // Ready for work let truncated_api_key = self.api_key.as_ref().unwrap().chars().rev().take(8).collect::(); info!( "{:?} device key {:?} put to work", From d2bf5b88f7a11f8ad334216f6350844ce408c165 Mon Sep 17 00:00:00 2001 From: reid Date: Sun, 19 Oct 2025 23:58:20 -0500 Subject: [PATCH 10/10] re-add backwards compat --- src/server.rs | 308 ++++++++++++++++++++++++++++---------------------- 1 file changed, 176 insertions(+), 132 deletions(-) diff --git a/src/server.rs b/src/server.rs index a03c091..ffefe29 100644 --- a/src/server.rs +++ b/src/server.rs @@ -46,43 +46,205 @@ impl QuiverInstance { submission_consumer, } } + + const TOKEN_PREFIX: u32 = 0x6e6f636b; // nock + + pub async fn serve(&mut self) -> Result<()> { + // First bidi stream: either legacy auth (v0) or version handshake (v1+) + let (mut send0, mut recv0) = self.conn.accept_bi().await?; + + // Peek first 4 bytes as u32 (legacy "nock" or length of version blob) + let first_u32 = recv0.read_u32().await; + + match first_u32 { + Ok(x) if x == Self::TOKEN_PREFIX => { + // v0 legacy path + self.serve_v1(send0, recv0).await + } + Ok(len) => { + // v1+ path (len = size of ProtocolVersion) + self.serve_v2(send0, recv0, len).await + } + Err(e) => { + error!("Failed to read initial handshake bytes: {}", e); + Err(e.into()) + } + } + } + + /// v0 (legacy) protocol: + /// - First stream carries the API key (raw, no length prefix), starting with "nock" + /// - Reply on same stream with raw "authenticated"/"unauthorized" + /// - Device info stream: raw bincode (no length), raw "accepted"/"rejected" + async fn serve_v1( + &mut self, + mut send0: quinn::SendStream, + mut recv0: quinn::RecvStream, + ) -> Result<()> { + warn!("Legacy authentication, no version handshake. Please update your miner!"); + // rebuild the full token: + let mut key = Self::TOKEN_PREFIX.to_be_bytes().to_vec(); + key.extend(recv0.read_to_end(50 - 4).await?); + let api_key = String::from_utf8(key)?; + + // Authenticate & reply on same stream + match self.authenticator.authenticate(&api_key).await { + Ok((acct, guard)) => { + self.account_information = Some(acct); + self.connection_guard = Some(guard); + self.api_key = Some(api_key); + send0.write_all(b"authenticated").await?; + send0.finish()?; + } + Err(e) => { + let _ = send0.write_all(b"unauthorized").await; + let _ = send0.finish(); + return Err(e); + } + } + + // Device info (legacy) + if !self.handle_device_info_stream(false).await? { + return Err(anyhow::anyhow!("Device info rejected")); + } + + self.after_ready_then_work().await + } - async fn handle_device_info_stream(&mut self) -> Result { + /// v2 protocol: + /// - First stream is version handshake (len-prefixed bincode), reply "ok" (len-prefixed) + /// - Second stream carries API key (len-prefixed), reply len-prefixed "authenticated"/"unauthorized" + /// - Device info stream: len-prefixed bincode, len-prefixed "accepted"/"rejected" + async fn serve_v2( + &mut self, + mut send0: quinn::SendStream, + mut recv0: quinn::RecvStream, + ver_len: u32, + ) -> Result<()> { + // Read version struct + if ver_len == 0 || ver_len >= 100 { + return Err(anyhow::anyhow!("Invalid protocol handshake length: {}", ver_len)); + } + let mut ver_buf = vec![0; ver_len as usize]; + recv0.read_exact(&mut ver_buf).await + .map_err(|_| anyhow::anyhow!("Malformed protocol handshake. Please update your miner."))?; + let client_v: ProtocolVersion = bincode::deserialize(&ver_buf) + .map_err(|_| anyhow::anyhow!("Malformed protocol handshake."))?; + if client_v.major != CURRENT_VERSION.major { + return Err(anyhow::anyhow!("Incompatible protocol version")); + } + info!("Handshake successful: v{}.{}.{}", client_v.major, client_v.minor, client_v.patch); + // Send len-prefixed "ok" and finish handshake stream + Self::lp_write(&mut send0, b"ok").await?; + send0.finish()?; + + // Auth stream + let (mut send_auth, mut recv_auth) = self.conn.accept_bi().await?; + let key = Self::lp_read(&mut recv_auth, 256).await?; // cap 256B + let api_key = String::from_utf8(key)?; + + match self.authenticator.authenticate(&api_key).await { + Ok((acct, guard)) => { + self.account_information = Some(acct); + self.connection_guard = Some(guard); + self.api_key = Some(api_key); + Self::lp_write(&mut send_auth, b"authenticated").await?; + send_auth.finish()?; + } + Err(e) => { + let _ = Self::lp_write(&mut send_auth, b"unauthorized").await; + let _ = send_auth.finish(); + return Err(e); + } + } + + // Device info v2 + if !self.handle_device_info_stream(true).await? { + return Err(anyhow::anyhow!("Device info rejected")); + } + + self.after_ready_then_work().await + } + + async fn after_ready_then_work(&mut self) -> Result<()> { + let truncated_api_key = self + .api_key + .as_ref() + .unwrap() + .chars() + .rev() + .take(8) + .collect::(); + + info!( + "{:?} device key {:?} put to work", + self.account_information.as_ref().unwrap().user_uuid, + truncated_api_key + ); + + self.handle_work().await + } + + /// len_prefixed: v2, expect/read/write length-prefixed frames + /// !len_prefixed: v1, legacy raw frames + async fn handle_device_info_stream(&mut self, len_prefixed: bool) -> Result { let Ok((mut send, mut recv)) = self.conn.accept_bi().await else { return Err(anyhow::anyhow!("connection closed before device info")); }; - let len = recv.read_u32().await?; - if len == 0 || len > 1_000_000 { - return Err(anyhow::anyhow!("Device info size out of bounds: {}", len)); - } + // Read device info bytes + let device_info_bytes = if len_prefixed { + Self::lp_read(&mut recv, 1_000_000).await? // 1MB cap + } else { + // legacy raw bincode, small bound + recv.read_to_end(4096).await? + }; - let mut device_info_bytes = vec![0; len as usize]; - recv.read_exact(&mut device_info_bytes).await?; let device_info = bincode::deserialize::(&device_info_bytes)?; - let Some(api_key) = self.api_key.clone() else { return Err(anyhow::anyhow!("no api key")); }; + // update & reply in matching format match self.device_info_updater.update_device_info(&device_info, &api_key).await { Ok(_) => { - let msg = b"accepted"; - send.write_u32(msg.len() as u32).await?; - send.write_all(msg).await?; + if len_prefixed { + Self::lp_write(&mut send, b"accepted").await?; + } else { + send.write_all(b"accepted").await?; + } send.finish()?; Ok(true) } Err(_) => { - let msg = b"rejected"; - send.write_u32(msg.len() as u32).await?; - send.write_all(msg).await?; + if len_prefixed { + Self::lp_write(&mut send, b"rejected").await?; + } else { + send.write_all(b"rejected").await?; + } send.finish()?; Ok(false) } } } + // helpers for v2 length-prefixed messages + async fn lp_read(recv: &mut quinn::RecvStream, max_len: u32) -> Result> { + let n = recv.read_u32().await?; + if n == 0 || n > max_len { + return Err(anyhow::anyhow!("frame size out of bounds: {}", n)); + } + let mut buf = vec![0; n as usize]; + recv.read_exact(&mut buf).await?; + Ok(buf) + } + + async fn lp_write(send: &mut quinn::SendStream, bytes: &[u8]) -> Result<()> { + send.write_u32(bytes.len() as u32).await?; + send.write_all(bytes).await?; + Ok(()) + } + async fn handle_work(&mut self) -> Result<()> { // New job pusher let new_job_provider = self.new_job_provider.clone(); @@ -136,124 +298,6 @@ impl QuiverInstance { } Ok(()) } - - async fn serve(&mut self) -> Result<()> { - // First bidi stream: version handshake OR legacy auth - let (mut send0, mut recv0) = self.conn.accept_bi().await?; - const TOKEN_PREFIX: u32 = 1852793707; - - let first_u32 = recv0.read_u32().await; - match first_u32 { - // ----- LEGACY: API key in first stream ----- - Ok(TOKEN_PREFIX) => { - warn!("Legacy authentication, no version handshake. Please update your miner!"); - let mut key = TOKEN_PREFIX.to_be_bytes().to_vec(); - key.extend(recv0.read_to_end(50 - 4).await?); - let api_key = String::from_utf8(key)?; - - match self.authenticator.authenticate(&api_key).await { - Ok((acct, guard)) => { - self.account_information = Some(acct); - self.connection_guard = Some(guard); - self.api_key = Some(api_key); - let msg = b"authenticated"; - send0.write_u32(msg.len() as u32).await?; - send0.write_all(msg).await?; - send0.finish()?; - } - Err(e) => { - let msg = b"unauthorized"; - let _ = send0.write_u32(msg.len() as u32).await; - let _ = send0.write_all(msg).await; - let _ = send0.finish(); - return Err(e); - } - } - - if !self.handle_device_info_stream().await? { - return Err(anyhow::anyhow!("Device info rejected")); - } - } - - // ----- NEW PROTOCOL: version then separate auth stream ----- - Ok(len) => { - if len == 0 || len >= 100 { - return Err(anyhow::anyhow!("Invalid protocol handshake length: {}", len)); - } - let mut ver_buf = vec![0; len as usize]; - recv0.read_exact(&mut ver_buf).await - .map_err(|_| anyhow::anyhow!("Malformed protocol handshake. Please update your miner."))?; - let client_v: ProtocolVersion = bincode::deserialize(&ver_buf) - .map_err(|_| anyhow::anyhow!("Malformed protocol handshake."))?; - - if client_v.major != CURRENT_VERSION.major { - return Err(anyhow::anyhow!("Incompatible protocol version")); - } - - info!("Handshake successful: v{}.{}.{}", client_v.major, client_v.minor, client_v.patch); - - // send length-prefixed "ok" and finish handshake stream - let ok = b"ok"; - send0.write_u32(ok.len() as u32).await?; - send0.write_all(ok).await?; - send0.finish()?; - - // Auth stream: read key, authenticate, reply, finish. - let (mut send_auth, mut recv_auth) = self.conn.accept_bi().await?; - let key_len = recv_auth.read_u32().await?; - if key_len > 256 { - let msg = b"invalid_key_len"; - let _ = send_auth.write_u32(msg.len() as u32).await; - let _ = send_auth.write_all(msg).await; - let _ = send_auth.finish(); - return Err(anyhow::anyhow!("Protocol error: key size {} exceeds 256", key_len)); - } - let mut key_bytes = vec![0; key_len as usize]; - recv_auth.read_exact(&mut key_bytes).await?; - let api_key = String::from_utf8(key_bytes)?; - - match self.authenticator.authenticate(&api_key).await { - Ok((acct, guard)) => { - self.account_information = Some(acct); - self.connection_guard = Some(guard); - self.api_key = Some(api_key); - let msg = b"authenticated"; - send_auth.write_u32(msg.len() as u32).await?; - send_auth.write_all(msg).await?; - send_auth.finish()?; - } - Err(e) => { - let msg = b"unauthorized"; - let _ = send_auth.write_u32(msg.len() as u32).await; - let _ = send_auth.write_all(msg).await; - let _ = send_auth.finish(); - return Err(e); - } - } - - // Device-info stream (next bidi) - if !self.handle_device_info_stream().await? { - return Err(anyhow::anyhow!("Device info rejected")); - } - } - - Err(e) => { - error!("Failed to read initial handshake bytes: {}", e); - return Err(e.into()); - } - } - - // Ready for work - let truncated_api_key = self.api_key.as_ref().unwrap().chars().rev().take(8).collect::(); - info!( - "{:?} device key {:?} put to work", - self.account_information.as_ref().unwrap().user_uuid, - truncated_api_key - ); - - self.handle_work().await - } - } pub async fn run(