diff --git a/src/rust/Cargo.lock b/src/rust/Cargo.lock index 99aa51f3..d91b7a98 100644 --- a/src/rust/Cargo.lock +++ b/src/rust/Cargo.lock @@ -1896,7 +1896,6 @@ version = "0.1.0" dependencies = [ "allocative", "allocative_derive", - "bincode", "lqos_config", "lqos_utils", "nix", diff --git a/src/rust/lqos_bus/Cargo.toml b/src/rust/lqos_bus/Cargo.toml index 3a5d141f..84222a72 100644 --- a/src/rust/lqos_bus/Cargo.toml +++ b/src/rust/lqos_bus/Cargo.toml @@ -11,7 +11,6 @@ equinix_tests = [] [dependencies] serde = { workspace = true } -bincode = { workspace = true } thiserror = { workspace = true } lqos_config = { path = "../lqos_config" } lqos_utils = { path = "../lqos_utils" } diff --git a/src/rust/lqos_bus/README.md b/src/rust/lqos_bus/README.md index 525a295f..c063b286 100644 --- a/src/rust/lqos_bus/README.md +++ b/src/rust/lqos_bus/README.md @@ -5,8 +5,8 @@ The bus acts as an intermediary through which `lqosd` and various clients commun The bus is intentionally not stable between releases. So if you integrate with it, you will need to update your integration *every time* you update `lqosd`. This: * Allows for rapid iteration on the back-end. -* Avoids the need for version negotiation. -* Permits the use of `bincode` for serialization, which is very fast. +* Avoids the need for version negotiation (CBOR tolerates unknown enum variants). +* Uses CBOR serialization with length-prefixed chunked framing for large payloads. > If you'd like a stable API, please consider the HTTP API provided with Insight. diff --git a/src/rust/lqos_bus/src/bus/client.rs b/src/rust/lqos_bus/src/bus/client.rs index ccd0fa39..8320035d 100644 --- a/src/rust/lqos_bus/src/bus/client.rs +++ b/src/rust/lqos_bus/src/bus/client.rs @@ -8,6 +8,8 @@ use tokio::{ }; use tracing::error; +use super::protocol::{decode_reply_cbor, encode_session_cbor, read_frame, write_frame}; + pub(crate) const MAGIC_NUMBER: [u8; 4] = [0x4C, 0x52, 0x45, 0x51]; // "LREQ" pub(crate) const MAGIC_RESPONSE: [u8; 4] = [0x4C, 0x52, 0x45, 0x50]; // "LREP" @@ -63,65 +65,19 @@ impl LibreqosBusClient { // Mirror the code in unix_socket_server::listen let session = BusSession { requests }; - let Ok(session_bytes) = bincode::serialize(&session) else { - error!("Unable to serialize session."); - return Err(BusClientError::EncodingError); - }; - let size = session_bytes.len(); - self.stream.write_u64_le(request_id).await.map_err(|_| { - error!("Unable to write request ID to {BUS_SOCKET_PATH} stream."); - BusClientError::StreamWriteError - })?; - self.stream.write_u64_le(size as u64).await.map_err(|_| { - error!("Unable to write session size to {BUS_SOCKET_PATH} stream."); - BusClientError::StreamWriteError - })?; - self.stream.write_all(&session_bytes).await.map_err(|_| { - error!("Unable to write session to {BUS_SOCKET_PATH} stream."); - BusClientError::StreamWriteError - })?; + let session_bytes = encode_session_cbor(&session)?; + write_frame(&mut self.stream, request_id, &session_bytes).await?; // Read the response - let mut response_id = [0u8; 8]; - self.stream - .read_exact(&mut response_id) - .await - .map_err(|_| { - error!("Unable to read response ID from {BUS_SOCKET_PATH} stream."); - BusClientError::StreamReadError - })?; - let response_id = u64::from_le_bytes(response_id); + let (response_id, response_bytes) = read_frame(&mut self.stream).await?; if response_id != request_id { error!("Received response ID {response_id} does not match request ID {request_id}."); return Err(BusClientError::StreamReadError); } - let mut response_size = [0u8; 8]; - self.stream - .read_exact(&mut response_size) - .await - .map_err(|_| { - error!("Unable to read response size from {BUS_SOCKET_PATH} stream."); - BusClientError::StreamReadError - })?; - let response_size = u64::from_le_bytes(response_size) as usize; - if response_size == 0 { + if response_bytes.is_empty() { return Ok(Vec::new()); } - let mut response_bytes = vec![0u8; response_size]; - self.stream - .read_exact(&mut response_bytes) - .await - .map_err(|_| { - error!("Unable to read response from {BUS_SOCKET_PATH} stream."); - BusClientError::StreamReadError - })?; - let response: BusReply = match bincode::deserialize(&response_bytes) { - Ok(response) => response, - Err(e) => { - error!("Unable to deserialize response: {:?}", e); - return Err(BusClientError::DecodingError); - } - }; + let response: BusReply = decode_reply_cbor(&response_bytes)?; if response.responses.is_empty() { return Ok(Vec::new()); } diff --git a/src/rust/lqos_bus/src/bus/mod.rs b/src/rust/lqos_bus/src/bus/mod.rs index bc812426..fb042258 100644 --- a/src/rust/lqos_bus/src/bus/mod.rs +++ b/src/rust/lqos_bus/src/bus/mod.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: AGPL-3.0-or-later WITH LicenseRef-LibreQoS-Exception mod client; +mod protocol; mod queue_data; mod reply; mod request; diff --git a/src/rust/lqos_bus/src/bus/protocol.rs b/src/rust/lqos_bus/src/bus/protocol.rs new file mode 100644 index 00000000..057453d3 --- /dev/null +++ b/src/rust/lqos_bus/src/bus/protocol.rs @@ -0,0 +1,279 @@ +// SPDX-FileCopyrightText: 2025 LibreQoE support@libreqos.io +// SPDX-License-Identifier: AGPL-3.0-or-later WITH LicenseRef-LibreQoS-Exception + +use super::{BusClientError, BusReply, BusSession}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tracing::error; + +pub(crate) const BUS_CHUNK_SIZE: usize = 8 * 1024; +pub(crate) const MAX_FRAME_BYTES: usize = 64 * 1024 * 1024; +pub(crate) const CHUNK_LEN_BYTES: usize = 4; + +pub(crate) fn encode_session_cbor(session: &BusSession) -> Result, BusClientError> { + serde_cbor::to_vec(session).map_err(|e| { + error!("Unable to serialize session to CBOR: {:?}", e); + BusClientError::EncodingError + }) +} + +pub(crate) fn decode_session_cbor(bytes: &[u8]) -> Result { + serde_cbor::from_slice(bytes).map_err(|e| { + error!("Unable to deserialize session from CBOR: {:?}", e); + BusClientError::DecodingError + }) +} + +pub(crate) fn encode_reply_cbor(reply: &BusReply) -> Result, BusClientError> { + serde_cbor::to_vec(reply).map_err(|e| { + error!("Unable to serialize reply to CBOR: {:?}", e); + BusClientError::EncodingError + }) +} + +pub(crate) fn decode_reply_cbor(bytes: &[u8]) -> Result { + serde_cbor::from_slice(bytes).map_err(|e| { + error!("Unable to deserialize reply from CBOR: {:?}", e); + BusClientError::DecodingError + }) +} + +pub(crate) async fn write_frame( + writer: &mut W, + request_id: u64, + payload: &[u8], +) -> Result<(), BusClientError> { + if payload.len() > MAX_FRAME_BYTES { + error!( + "Payload size {} exceeds MAX_FRAME_BYTES {}.", + payload.len(), + MAX_FRAME_BYTES + ); + return Err(BusClientError::EncodingError); + } + + writer + .write_u64_le(request_id) + .await + .map_err(|_| BusClientError::StreamWriteError)?; + writer + .write_u64_le(payload.len() as u64) + .await + .map_err(|_| BusClientError::StreamWriteError)?; + + if payload.is_empty() { + return Ok(()); + } + + for chunk in payload.chunks(BUS_CHUNK_SIZE) { + let chunk_len = u32::try_from(chunk.len()).map_err(|_| { + error!("Chunk length exceeds u32 capacity."); + BusClientError::EncodingError + })?; + let mut len_buf = [0u8; CHUNK_LEN_BYTES]; + len_buf.copy_from_slice(&chunk_len.to_le_bytes()); + writer + .write_all(&len_buf) + .await + .map_err(|_| BusClientError::StreamWriteError)?; + writer + .write_all(chunk) + .await + .map_err(|_| BusClientError::StreamWriteError)?; + } + + Ok(()) +} + +pub(crate) async fn read_frame( + reader: &mut R, +) -> Result<(u64, Vec), BusClientError> { + let request_id = reader + .read_u64_le() + .await + .map_err(|_| BusClientError::StreamReadError)?; + let payload_len = reader + .read_u64_le() + .await + .map_err(|_| BusClientError::StreamReadError)?; + + if payload_len == 0 { + return Ok((request_id, Vec::new())); + } + + let payload_len = usize::try_from(payload_len).map_err(|_| { + error!("Payload size exceeds usize capacity."); + BusClientError::DecodingError + })?; + if payload_len > MAX_FRAME_BYTES { + error!( + "Payload size {} exceeds MAX_FRAME_BYTES {}.", + payload_len, MAX_FRAME_BYTES + ); + return Err(BusClientError::DecodingError); + } + + let mut payload = Vec::with_capacity(payload_len); + let mut remaining = payload_len; + let mut chunk_buf = vec![0u8; BUS_CHUNK_SIZE]; + + while remaining > 0 { + let mut len_buf = [0u8; CHUNK_LEN_BYTES]; + reader + .read_exact(&mut len_buf) + .await + .map_err(|_| BusClientError::StreamReadError)?; + let chunk_len = u32::from_le_bytes(len_buf) as usize; + if chunk_len == 0 { + error!("Chunk length of 0 is invalid for non-empty payloads."); + return Err(BusClientError::DecodingError); + } + if chunk_len > BUS_CHUNK_SIZE { + error!( + "Chunk length {} exceeds BUS_CHUNK_SIZE {}.", + chunk_len, BUS_CHUNK_SIZE + ); + return Err(BusClientError::DecodingError); + } + if chunk_len > remaining { + error!( + "Chunk length {} exceeds remaining payload {}.", + chunk_len, remaining + ); + return Err(BusClientError::DecodingError); + } + reader + .read_exact(&mut chunk_buf[..chunk_len]) + .await + .map_err(|_| BusClientError::StreamReadError)?; + payload.extend_from_slice(&chunk_buf[..chunk_len]); + remaining -= chunk_len; + } + + Ok((request_id, payload)) +} + +#[cfg(test)] +mod tests { + use super::{ + BUS_CHUNK_SIZE, MAX_FRAME_BYTES, decode_reply_cbor, decode_session_cbor, encode_reply_cbor, + encode_session_cbor, read_frame, write_frame, + }; + use crate::{BusReply, BusRequest, BusResponse, BusSession, bus::BusClientError}; + use tokio::io::{AsyncWriteExt, duplex}; + + #[test] + fn cbor_round_trip_session() { + let session = BusSession { + requests: vec![BusRequest::Ping], + }; + let bytes = encode_session_cbor(&session).expect("encode_session_cbor"); + let decoded = decode_session_cbor(&bytes).expect("decode_session_cbor"); + assert_eq!(decoded.requests, session.requests); + } + + #[test] + fn cbor_round_trip_reply() { + let reply = BusReply { + responses: vec![BusResponse::Ack], + }; + let bytes = encode_reply_cbor(&reply).expect("encode_reply_cbor"); + let decoded = decode_reply_cbor(&bytes).expect("decode_reply_cbor"); + assert_eq!(decoded.responses, reply.responses); + } + + #[tokio::test] + async fn frame_round_trip_small_payload() { + let (mut client, mut server) = duplex(128 * 1024); + let payload = vec![0xAB; BUS_CHUNK_SIZE / 2]; + let expected = payload.clone(); + + let write = async { + write_frame(&mut client, 7, &payload) + .await + .expect("write_frame"); + }; + let read = async { + read_frame(&mut server).await.expect("read_frame") + }; + + let (_, (request_id, read_payload)) = tokio::join!(write, read); + assert_eq!(request_id, 7); + assert_eq!(read_payload, expected); + } + + #[tokio::test] + async fn frame_round_trip_large_payload() { + let (mut client, mut server) = duplex(256 * 1024); + let payload = vec![0xCD; BUS_CHUNK_SIZE * 10 + 123]; + let expected = payload.clone(); + + let write = async { + write_frame(&mut client, 11, &payload) + .await + .expect("write_frame"); + }; + let read = async { + read_frame(&mut server).await.expect("read_frame") + }; + + let (_, (request_id, read_payload)) = tokio::join!(write, read); + assert_eq!(request_id, 11); + assert_eq!(read_payload, expected); + } + + #[tokio::test] + async fn frame_rejects_oversized_on_write() { + let (mut client, _server) = duplex(128 * 1024); + let payload = vec![0xEF; MAX_FRAME_BYTES + 1]; + let result = write_frame(&mut client, 1, &payload).await; + assert!(matches!(result, Err(BusClientError::EncodingError))); + } + + #[tokio::test] + async fn frame_rejects_oversized_on_read() { + let (mut client, mut server) = duplex(128 * 1024); + let write = async { + client + .write_u64_le(5) + .await + .expect("write request id"); + client + .write_u64_le((MAX_FRAME_BYTES as u64) + 1) + .await + .expect("write oversized len"); + }; + let read = async { + read_frame(&mut server).await + }; + + let (_, result) = tokio::join!(write, read); + assert!(matches!(result, Err(BusClientError::DecodingError))); + } + + #[tokio::test] + async fn frame_multiple_back_to_back() { + let (mut client, mut server) = duplex(256 * 1024); + let payload_a = vec![0x01; BUS_CHUNK_SIZE + 5]; + let payload_b = vec![0x02; BUS_CHUNK_SIZE * 2 + 7]; + + let write = async { + write_frame(&mut client, 100, &payload_a) + .await + .expect("write_frame a"); + write_frame(&mut client, 101, &payload_b) + .await + .expect("write_frame b"); + }; + let read = async { + let first = read_frame(&mut server).await.expect("read_frame a"); + let second = read_frame(&mut server).await.expect("read_frame b"); + (first, second) + }; + + let (_, ((id_a, data_a), (id_b, data_b))) = tokio::join!(write, read); + assert_eq!(id_a, 100); + assert_eq!(data_a, payload_a); + assert_eq!(id_b, 101); + assert_eq!(data_b, payload_b); + } +} diff --git a/src/rust/lqos_bus/src/bus/unix_socket_server.rs b/src/rust/lqos_bus/src/bus/unix_socket_server.rs index 826d3630..854b19d8 100644 --- a/src/rust/lqos_bus/src/bus/unix_socket_server.rs +++ b/src/rust/lqos_bus/src/bus/unix_socket_server.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: AGPL-3.0-or-later WITH LicenseRef-LibreQoS-Exception use crate::{ - BUS_SOCKET_PATH, BusReply, BusRequest, BusResponse, BusSession, + BUS_SOCKET_PATH, BusReply, BusRequest, BusResponse, bus::client::{MAGIC_NUMBER, MAGIC_RESPONSE}, }; use std::{ffi::CString, fs::remove_file}; @@ -14,15 +14,12 @@ use tokio::{ use tracing::{debug, error, info, warn}; use super::BUS_SOCKET_DIRECTORY; +use super::protocol::{decode_session_cbor, encode_reply_cbor, read_frame, write_frame}; /// Implements a Tokio-friendly server using Unix Sockets and the bus protocol. /// Requests are handled and then forwarded to the handler. pub struct UnixSocketServer {} -static RECEIVE_BUFFER: tokio::sync::OnceCell>> = - tokio::sync::OnceCell::const_new(); -const RECEIVE_BUFFER_SIZE: usize = 1024 * 1024 * 512; // 512 MiB - impl UnixSocketServer { /// Creates a new `UnixSocketServer`. Will delete any pre-existing /// socket file. @@ -96,10 +93,6 @@ impl UnixSocketServer { BusRequest, )>, ) -> Result<(), UnixSocketServerError> { - RECEIVE_BUFFER - .set(tokio::sync::Mutex::new(vec![0; RECEIVE_BUFFER_SIZE])) // 512 MiB - .map_err(|_| UnixSocketServerError::MkDirFail)?; - // Set up the listener and grant permissions to it let listener = UnixListener::bind(BUS_SOCKET_PATH); let Ok(listener) = listener else { @@ -156,75 +149,46 @@ impl UnixSocketServer { } loop { - // Read the request ID (64 bits) and the Size (64 bits) - // This is the size of the request in bytes, not the number of requests - let mut id_buf = [0; 16]; - let bytes_read = socket.read_exact(&mut id_buf).await; - if bytes_read.is_err() { - debug!("Unable to read request ID from client socket. Server remains alive."); - debug!("This is probably harmless."); - debug!("{:?}", bytes_read); - break; // Escape out of the thread - } - let Ok(id_bytes): Result<[u8; 8], _> = id_buf[0..8].try_into() else { - warn!("Invalid request header for ID; closing client socket."); - break; // Escape out of the thread + let (request_id, request_bytes) = match read_frame(&mut socket).await { + Ok(frame) => frame, + Err(e) => { + debug!("Unable to read request frame from client socket."); + debug!("This is probably harmless."); + debug!("{:?}", e); + break; + } }; - let request_id = u64::from_le_bytes(id_bytes); - - let Ok(size_bytes): Result<[u8; 8], _> = id_buf[8..16].try_into() else { - warn!("Invalid request header for size; closing client socket."); - break; // Escape out of the thread - }; - let request_size = u64::from_le_bytes(size_bytes); - debug!("Received request ID: {request_id}, Size: {request_size}"); - if request_size > RECEIVE_BUFFER_SIZE as u64 { - warn!("Received request size ({request_size}) exceeds buffer size ({RECEIVE_BUFFER_SIZE})."); - break; // Escape out of the thread - } - - // Read the request data - //let mut buf = vec![0; request_size as usize]; - let Some(buf_cell) = RECEIVE_BUFFER.get() else { - error!("Receive buffer not initialized; closing client socket."); - break; // Escape out of the thread - }; - let mut buf = buf_cell.lock().await; - - let bytes_read = socket.read_exact(&mut buf[0 .. request_size as usize]).await; - if bytes_read.is_err() { - debug!("Unable to read request data from client socket. Server remains alive."); - debug!("This is probably harmless."); - debug!("{:?}", bytes_read); - break; // Escape out of the thread + if request_bytes.is_empty() { + warn!("Received empty request payload; closing client socket."); + break; } + debug!( + "Received request ID: {request_id}, Size: {}", + request_bytes.len() + ); // Decode the request - let Ok(request) = bincode::deserialize::(&buf[0..request_size as usize]) else { + let Ok(request) = decode_session_cbor(&request_bytes) else { warn!("Invalid data on local socket"); break; }; debug!("Received request: {:?}", request); - drop(buf); // Release the lock on the buffer // Handle the request and build the response let mut response = BusReply { responses: Vec::with_capacity(8) }; handle_bus_requests(&request.requests, &mut response.responses); // Encode the response - let Ok(encoded_response) = bincode::serialize(&response) else { + let Ok(encoded_response) = encode_reply_cbor(&response) else { warn!("Unable to encode response for request ID: {request_id}"); break; }; debug!("Sending response for request ID: {request_id}"); // Send the response back to the client - let response_size = encoded_response.len() as u64; - let mut response_buf = vec![0; 16 + encoded_response.len()]; - response_buf[0..8].copy_from_slice(&request_id.to_le_bytes()); - response_buf[8..16].copy_from_slice(&response_size.to_le_bytes()); - response_buf[16..].copy_from_slice(&encoded_response); - if let Err(e) = socket.write_all(&response_buf).await { + if let Err(e) = + write_frame(&mut socket, request_id, &encoded_response).await + { debug!("Unable to write response to client socket. Server remains alive."); debug!("This is probably harmless."); debug!("{:?}", e); diff --git a/src/rust/lqos_bus/src/lib.rs b/src/rust/lqos_bus/src/lib.rs index 490dc09a..dd0f4c86 100644 --- a/src/rust/lqos_bus/src/lib.rs +++ b/src/rust/lqos_bus/src/lib.rs @@ -7,10 +7,13 @@ //! with the LibreQoS system. //! //! A normal session consists of connecting and sending a single `BusSession` -//! object (serialized with `bincode`), that must contain one or more -//! `BusRequest` objects. Replies are then batched inside a `BusReply` -//! object, containing one or more `BusResponse` detail objects. -//! The session then terminates. +//! object (serialized with CBOR), that must contain one or more +//! `BusRequest` objects. Payloads are framed with a header and chunked +//! into length-prefixed blocks for transport. Replies are then batched +//! inside a `BusReply` object, containing one or more `BusResponse` +//! detail objects. The session then terminates. +//! +//! Protocol versioning/negotiation is intentionally skipped. #![deny(clippy::unwrap_used)] #![warn(missing_docs)] @@ -32,11 +35,6 @@ pub use bus::{ }; pub use tc_handle::TcHandle; -/// Re-export bincode -pub mod bincode { - pub use bincode::*; -} - /// Re-export CBOR pub mod cbor { pub use serde_cbor::*; diff --git a/src/update_api.sh b/src/update_api.sh index 2031e0be..f71a1793 100755 --- a/src/update_api.sh +++ b/src/update_api.sh @@ -3,7 +3,7 @@ set -euo pipefail # Fetch and install the lqos_api binary into src/bin -# - Downloads https://download.libreqos.com/api.zip +# - Downloads https://download.libreqos.com/api2.zip # - Extracts the single file lqos_api # - Places it into ./bin alongside other binaries @@ -12,7 +12,11 @@ BIN_DIR="$SCRIPT_DIR/bin" TMP_ZIP="$(mktemp /tmp/lqos_api.XXXXXX.zip)" echo "Downloading lqos_api..." -curl -fsSL -o "$TMP_ZIP" "https://download.libreqos.com/api.zip" +if ! curl -fsSL -o "$TMP_ZIP" "https://download.libreqos.com/api2.zip"; then + echo "Warning: Failed to download lqos_api; leaving existing binary untouched." + rm -f "$TMP_ZIP" + exit 0 +fi # Ensure unzip is available if ! command -v unzip >/dev/null 2>&1; then