reddb_wire/redwire/
bulk_binary.rs

1//! RedWire binary bulk payload codec.
2//!
3//! Layout:
4//! `[collection_len:u16][collection][column_count:u16]`
5//! `([column_len:u16][column_name])*`
6//! `[row_count:u32]([legacy WireValue])*(column_count * row_count)`.
7
8use crate::legacy::{encode_value, try_decode_value, WireValue};
9use std::fmt;
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum BulkBinaryFlavor {
13    Binary,
14    Prevalidated,
15}
16
17#[derive(Debug, Clone, PartialEq)]
18pub struct BulkBinaryPayload {
19    pub collection: String,
20    pub columns: Vec<String>,
21    pub rows: Vec<Vec<WireValue>>,
22}
23
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub enum BulkBinaryError {
26    PayloadTooShort(BulkBinaryFlavor),
27    MissingCollectionLength(BulkBinaryFlavor),
28    TruncatedCollectionName(BulkBinaryFlavor),
29    InvalidCollectionName(BulkBinaryFlavor),
30    MissingColumnCount(BulkBinaryFlavor),
31    MissingColumnNameLength(BulkBinaryFlavor),
32    TruncatedColumnName(BulkBinaryFlavor),
33    InvalidColumnName(BulkBinaryFlavor),
34    MissingRowCount(BulkBinaryFlavor),
35    Value(BulkBinaryFlavor, &'static str),
36    LengthOverflow(&'static str),
37    RowWidthMismatch { got: usize, expected: usize },
38    RowCountOverflow,
39}
40
41impl fmt::Display for BulkBinaryError {
42    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
43        match self {
44            Self::PayloadTooShort(flavor) => write!(f, "{}: payload too short", prefix(*flavor)),
45            Self::MissingCollectionLength(flavor) => {
46                write!(f, "{}: missing collection length", label(*flavor))
47            }
48            Self::TruncatedCollectionName(flavor) => {
49                write!(f, "{}: truncated collection name", label(*flavor))
50            }
51            Self::InvalidCollectionName(flavor) => {
52                write!(f, "{}: invalid collection name", label(*flavor))
53            }
54            Self::MissingColumnCount(flavor) => {
55                write!(f, "{}: missing column count", label(*flavor))
56            }
57            Self::MissingColumnNameLength(flavor) => {
58                write!(f, "{}: missing column name length", label(*flavor))
59            }
60            Self::TruncatedColumnName(flavor) => {
61                write!(f, "{}: truncated column name", label(*flavor))
62            }
63            Self::InvalidColumnName(flavor) => {
64                write!(f, "{}: invalid column name", label(*flavor))
65            }
66            Self::MissingRowCount(flavor) => write!(f, "{}: missing row count", label(*flavor)),
67            Self::Value(flavor, err) => write!(f, "{}: {err}", label(*flavor)),
68            Self::LengthOverflow(field) => {
69                write!(f, "{field} is too large for RedWire binary bulk")
70            }
71            Self::RowWidthMismatch { got, expected } => {
72                write!(f, "row had {got} values for {expected} columns")
73            }
74            Self::RowCountOverflow => write!(f, "row count is too large for RedWire binary bulk"),
75        }
76    }
77}
78
79impl std::error::Error for BulkBinaryError {}
80
81pub fn encode_bulk_binary_payload(
82    collection: &str,
83    columns: &[&str],
84    rows: &[Vec<WireValue>],
85) -> Result<Vec<u8>, BulkBinaryError> {
86    write_len_u16(collection.len(), "collection")?;
87    write_len_u16(columns.len(), "columns")?;
88    if rows.len() > u32::MAX as usize {
89        return Err(BulkBinaryError::RowCountOverflow);
90    }
91    let mut out = Vec::with_capacity(64 + rows.len() * columns.len() * 16);
92    write_string_u16(&mut out, collection, "collection")?;
93    out.extend_from_slice(&(columns.len() as u16).to_le_bytes());
94    for column in columns {
95        write_string_u16(&mut out, column, "column")?;
96    }
97    out.extend_from_slice(&(rows.len() as u32).to_le_bytes());
98    for row in rows {
99        if row.len() != columns.len() {
100            return Err(BulkBinaryError::RowWidthMismatch {
101                got: row.len(),
102                expected: columns.len(),
103            });
104        }
105        for value in row {
106            encode_value(&mut out, value);
107        }
108    }
109    Ok(out)
110}
111
112pub fn decode_bulk_binary_payload(
113    payload: &[u8],
114    flavor: BulkBinaryFlavor,
115) -> Result<BulkBinaryPayload, BulkBinaryError> {
116    let mut pos = 0;
117    if payload.len() < 6 {
118        return Err(BulkBinaryError::PayloadTooShort(flavor));
119    }
120    let coll_len = read_u16(
121        payload,
122        &mut pos,
123        BulkBinaryError::MissingCollectionLength(flavor),
124    )? as usize;
125    let collection = read_string(
126        payload,
127        &mut pos,
128        coll_len,
129        BulkBinaryError::TruncatedCollectionName(flavor),
130        BulkBinaryError::InvalidCollectionName(flavor),
131    )?;
132    let ncols = read_u16(
133        payload,
134        &mut pos,
135        BulkBinaryError::MissingColumnCount(flavor),
136    )? as usize;
137    let mut columns = Vec::with_capacity(ncols);
138    for _ in 0..ncols {
139        let name_len = read_u16(
140            payload,
141            &mut pos,
142            BulkBinaryError::MissingColumnNameLength(flavor),
143        )? as usize;
144        columns.push(read_string(
145            payload,
146            &mut pos,
147            name_len,
148            BulkBinaryError::TruncatedColumnName(flavor),
149            BulkBinaryError::InvalidColumnName(flavor),
150        )?);
151    }
152    let nrows = read_u32(payload, &mut pos, BulkBinaryError::MissingRowCount(flavor))? as usize;
153    // With `ncols` zero the value loop consumes nothing, so the truncation
154    // error that normally terminates decoding never fires and the row loop
155    // runs up to `u32::MAX` times pushing empty vectors. A zero-column bulk
156    // insert is meaningless anyway.
157    if ncols == 0 {
158        return Err(BulkBinaryError::RowWidthMismatch {
159            got: 0,
160            expected: 1,
161        });
162    }
163    // `nrows` is attacker-controlled. Reject a count the payload cannot hold
164    // (every value costs at least one byte) rather than only trimming the
165    // capacity hint — clamping the allocation still left the loop free to run.
166    let remaining = payload.len().saturating_sub(pos);
167    if nrows > remaining / ncols {
168        return Err(BulkBinaryError::RowCountOverflow);
169    }
170    let mut rows = Vec::with_capacity(nrows);
171    for _ in 0..nrows {
172        let mut values = Vec::with_capacity(ncols);
173        for _ in 0..ncols {
174            values.push(
175                try_decode_value(payload, &mut pos)
176                    .map_err(|err| BulkBinaryError::Value(flavor, err))?,
177            );
178        }
179        rows.push(values);
180    }
181    Ok(BulkBinaryPayload {
182        collection,
183        columns,
184        rows,
185    })
186}
187
188fn prefix(flavor: BulkBinaryFlavor) -> &'static str {
189    match flavor {
190        BulkBinaryFlavor::Binary => "binary bulk",
191        BulkBinaryFlavor::Prevalidated => "binary bulk prevalidated",
192    }
193}
194
195fn label(flavor: BulkBinaryFlavor) -> &'static str {
196    match flavor {
197        BulkBinaryFlavor::Binary => "binary bulk",
198        BulkBinaryFlavor::Prevalidated => "prevalidated",
199    }
200}
201
202fn write_string_u16(
203    out: &mut Vec<u8>,
204    value: &str,
205    field: &'static str,
206) -> Result<(), BulkBinaryError> {
207    write_len_u16(value.len(), field)?;
208    out.extend_from_slice(&(value.len() as u16).to_le_bytes());
209    out.extend_from_slice(value.as_bytes());
210    Ok(())
211}
212
213fn write_len_u16(len: usize, field: &'static str) -> Result<(), BulkBinaryError> {
214    if len > u16::MAX as usize {
215        return Err(BulkBinaryError::LengthOverflow(field));
216    }
217    Ok(())
218}
219
220fn read_u16(payload: &[u8], pos: &mut usize, err: BulkBinaryError) -> Result<u16, BulkBinaryError> {
221    let bytes = read_bytes(payload, pos, 2, err)?;
222    Ok(u16::from_le_bytes([bytes[0], bytes[1]]))
223}
224
225fn read_u32(payload: &[u8], pos: &mut usize, err: BulkBinaryError) -> Result<u32, BulkBinaryError> {
226    let bytes = read_bytes(payload, pos, 4, err)?;
227    Ok(u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
228}
229
230fn read_string(
231    payload: &[u8],
232    pos: &mut usize,
233    len: usize,
234    truncated_err: BulkBinaryError,
235    utf8_err: BulkBinaryError,
236) -> Result<String, BulkBinaryError> {
237    let bytes = read_bytes(payload, pos, len, truncated_err)?;
238    std::str::from_utf8(bytes)
239        .map(str::to_owned)
240        .map_err(|_| utf8_err)
241}
242
243fn read_bytes<'a>(
244    payload: &'a [u8],
245    pos: &mut usize,
246    len: usize,
247    err: BulkBinaryError,
248) -> Result<&'a [u8], BulkBinaryError> {
249    let end = pos.saturating_add(len);
250    if end > payload.len() {
251        return Err(err);
252    }
253    let bytes = &payload[*pos..end];
254    *pos = end;
255    Ok(bytes)
256}
257
258#[cfg(test)]
259mod tests {
260    use super::*;
261
262    #[test]
263    fn binary_bulk_payload_round_trips_values() {
264        let rows = vec![vec![
265            WireValue::I64(7),
266            WireValue::Text("Ada".into()),
267            WireValue::Bool(true),
268        ]];
269        let bytes = encode_bulk_binary_payload("users", &["id", "name", "active"], &rows).unwrap();
270        let decoded = decode_bulk_binary_payload(&bytes, BulkBinaryFlavor::Binary).unwrap();
271        assert_eq!(decoded.collection, "users");
272        assert_eq!(decoded.columns, vec!["id", "name", "active"]);
273        assert_eq!(decoded.rows, rows);
274    }
275
276    #[test]
277    fn binary_bulk_decode_preserves_error_prefixes() {
278        assert_eq!(
279            decode_bulk_binary_payload(&[0; 5], BulkBinaryFlavor::Binary)
280                .unwrap_err()
281                .to_string(),
282            "binary bulk: payload too short"
283        );
284        let payload = vec![1, 0, b't', 1, 0, 1, 0, b'x', 1, 0, 0, 0, 1];
285        assert_eq!(
286            decode_bulk_binary_payload(&payload, BulkBinaryFlavor::Prevalidated)
287                .unwrap_err()
288                .to_string(),
289            "prevalidated: truncated i64 value"
290        );
291    }
292}
reddb_wire/redwire/bulk_binary.rs

reddb_wire/redwire/
bulk_binary.rs
