mongreldb_protocol/
envelope.rs

1//! Versioned network message envelope (spec section 4.10, S1D-001).
2//!
3//! Every message on every MongrelDB wire protocol travels inside a
4//! [`ProtocolEnvelope`], never as an unversioned payload. The canonical
5//! encoding is:
6//!
7//! ```text
8//! protocol_version : u32 LE
9//! message_type     : u32 LE
10//! payload_len      : u32 LE
11//! payload          : [u8; payload_len]
12//! payload_crc32    : u32 LE
13//! ```
14//!
15//! The CRC-32 (IEEE, reflected) covers the protocol version, the message
16//! type, the payload length, and the payload. This mirrors the fail-closed
17//! shape of `mongreldb-log`'s `CommandEnvelope`: unknown protocol versions,
18//! oversized payloads, truncated frames, trailing bytes, and checksum
19//! mismatches are all decode errors (spec section 4.10: unknown required
20//! fields or incompatible versions fail closed).
21//!
22//! Unlike `CommandEnvelope` the checksum is a CRC-32 rather than SHA-256:
23//! this crate's dependency set is frozen (`serde`, `thiserror`,
24//! `mongreldb-types`), and transport integrity plus peer authentication are
25//! provided by TLS 1.3 (S1D-002), so the checksum's job here is only
26//! framing sanity, not cryptographic authentication.
27//!
28//! Encoding is deterministic: [`ProtocolEnvelope::encode`] is a pure
29//! function of the envelope fields, so equal envelopes produce
30//! byte-identical frames and `decode(encode(e))` is the identity.
31//!
32//! Payload evolution rules (spec section 4.10):
33//!
34//! - `message_type` discriminants and payload field numbers are never
35//!   reused; new message types are allocated fresh numbers.
36//! - The envelope never interprets payload bytes; payload decoding is the
37//!   adapter's job (Protobuf control frames, Arrow IPC result frames, per
38//!   S1D-002 and ADR 0005).
39//! - Unknown protocol versions fail closed with
40//!   [`EnvelopeError::UnsupportedVersion`].
41
42use core::fmt;
43
44/// The protocol version this build writes.
45pub const PROTOCOL_VERSION: u32 = 1;
46/// The oldest protocol version this build accepts.
47pub const MIN_SUPPORTED_PROTOCOL_VERSION: u32 = 1;
48/// Upper bound on a single message payload.
49pub const MAX_MESSAGE_PAYLOAD_BYTES: usize = 64 * 1024 * 1024;
50/// Encoded length of the fixed header preceding the payload.
51pub const HEADER_LEN: usize = 4 + 4 + 4;
52/// Encoded length of the trailing checksum.
53pub const CHECKSUM_LEN: usize = 4;
54
55/// Errors produced while verifying or decoding a [`ProtocolEnvelope`].
56#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
57pub enum EnvelopeError {
58    /// The protocol version is outside the supported range.
59    #[error("unsupported protocol version {found} (supported {min}..={max})")]
60    UnsupportedVersion {
61        /// Version found in the envelope.
62        found: u32,
63        /// Oldest version this build accepts.
64        min: u32,
65        /// Newest version this build accepts.
66        max: u32,
67    },
68    /// The stored checksum does not match the recomputed one.
69    #[error("protocol-envelope checksum mismatch")]
70    ChecksumMismatch,
71    /// The byte slice ended before a complete envelope was read.
72    #[error("protocol envelope truncated: expected at least {expected} bytes, got {actual}")]
73    Truncated {
74        /// Minimum number of bytes required.
75        expected: usize,
76        /// Number of bytes actually present.
77        actual: usize,
78    },
79    /// Extra bytes followed an otherwise complete envelope.
80    #[error("protocol envelope has {0} trailing bytes")]
81    TrailingBytes(usize),
82    /// The payload exceeds [`MAX_MESSAGE_PAYLOAD_BYTES`].
83    #[error("message payload too large: {0} bytes")]
84    PayloadTooLarge(usize),
85}
86
87/// IEEE CRC-32 (reflected, polynomial 0xEDB88320) lookup table, built at
88/// compile time.
89const fn build_crc32_table() -> [u32; 256] {
90    let mut table = [0u32; 256];
91    let mut i = 0;
92    while i < 256 {
93        let mut c = i as u32;
94        let mut k = 0;
95        while k < 8 {
96            c = if c & 1 != 0 {
97                0xEDB8_8320 ^ (c >> 1)
98            } else {
99                c >> 1
100            };
101            k += 1;
102        }
103        table[i] = c;
104        i += 1;
105    }
106    table
107}
108
109static CRC32_TABLE: [u32; 256] = build_crc32_table();
110
111fn crc32_update(mut crc: u32, bytes: &[u8]) -> u32 {
112    for &byte in bytes {
113        crc = CRC32_TABLE[((crc ^ u32::from(byte)) & 0xff) as usize] ^ (crc >> 8);
114    }
115    crc
116}
117
118/// CRC-32 over the concatenation of `parts`.
119fn crc32(parts: &[&[u8]]) -> u32 {
120    let mut crc = 0xffff_ffffu32;
121    for part in parts {
122        crc = crc32_update(crc, part);
123    }
124    crc ^ 0xffff_ffff
125}
126
127/// The versioned, checksummed form of every protocol message.
128#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
129pub struct ProtocolEnvelope {
130    /// Protocol version; see [`PROTOCOL_VERSION`].
131    pub protocol_version: u32,
132    /// Discriminant of the payload message; numbers are never reused.
133    pub message_type: u32,
134    /// Encoded message payload. The envelope never interprets these bytes.
135    pub payload: Vec<u8>,
136    /// CRC-32 over version, type, length, and payload.
137    pub payload_crc32: u32,
138}
139
140impl ProtocolEnvelope {
141    /// Builds an envelope at the current protocol version, computing the
142    /// checksum.
143    pub fn new(message_type: u32, payload: Vec<u8>) -> Self {
144        let payload_crc32 = Self::checksum(PROTOCOL_VERSION, message_type, &payload);
145        Self {
146            protocol_version: PROTOCOL_VERSION,
147            message_type,
148            payload,
149            payload_crc32,
150        }
151    }
152
153    /// Computes the checksum covering version, type, length, and payload.
154    pub fn checksum(protocol_version: u32, message_type: u32, payload: &[u8]) -> u32 {
155        crc32(&[
156            &protocol_version.to_le_bytes(),
157            &message_type.to_le_bytes(),
158            &(payload.len() as u64).to_le_bytes(),
159            payload,
160        ])
161    }
162
163    /// Fails closed unless the version is supported and the checksum matches.
164    pub fn verify(&self) -> Result<(), EnvelopeError> {
165        if !(MIN_SUPPORTED_PROTOCOL_VERSION..=PROTOCOL_VERSION).contains(&self.protocol_version) {
166            return Err(EnvelopeError::UnsupportedVersion {
167                found: self.protocol_version,
168                min: MIN_SUPPORTED_PROTOCOL_VERSION,
169                max: PROTOCOL_VERSION,
170            });
171        }
172        if self.payload.len() > MAX_MESSAGE_PAYLOAD_BYTES {
173            return Err(EnvelopeError::PayloadTooLarge(self.payload.len()));
174        }
175        let expected = Self::checksum(self.protocol_version, self.message_type, &self.payload);
176        if expected != self.payload_crc32 {
177            return Err(EnvelopeError::ChecksumMismatch);
178        }
179        Ok(())
180    }
181
182    /// Serializes to the canonical deterministic encoding: equal envelopes
183    /// always produce byte-identical output.
184    pub fn encode(&self) -> Vec<u8> {
185        debug_assert!(
186            self.payload.len() <= MAX_MESSAGE_PAYLOAD_BYTES,
187            "payload exceeds MAX_MESSAGE_PAYLOAD_BYTES"
188        );
189        let mut out = Vec::with_capacity(HEADER_LEN + self.payload.len() + CHECKSUM_LEN);
190        out.extend_from_slice(&self.protocol_version.to_le_bytes());
191        out.extend_from_slice(&self.message_type.to_le_bytes());
192        out.extend_from_slice(&(self.payload.len() as u32).to_le_bytes());
193        out.extend_from_slice(&self.payload);
194        out.extend_from_slice(&self.payload_crc32.to_le_bytes());
195        out
196    }
197
198    /// Parses one envelope, verifying version and checksum (fails closed).
199    pub fn decode(bytes: &[u8]) -> Result<Self, EnvelopeError> {
200        if bytes.len() < HEADER_LEN + CHECKSUM_LEN {
201            return Err(EnvelopeError::Truncated {
202                expected: HEADER_LEN + CHECKSUM_LEN,
203                actual: bytes.len(),
204            });
205        }
206        let protocol_version = u32::from_le_bytes(bytes[0..4].try_into().expect("slice len"));
207        let message_type = u32::from_le_bytes(bytes[4..8].try_into().expect("slice len"));
208        let payload_len = u32::from_le_bytes(bytes[8..12].try_into().expect("slice len")) as usize;
209        // Fail closed on an oversize length prefix before copying any payload
210        // bytes, regardless of how long the supplied frame actually is.
211        if payload_len > MAX_MESSAGE_PAYLOAD_BYTES {
212            return Err(EnvelopeError::PayloadTooLarge(payload_len));
213        }
214        let expected_total = HEADER_LEN + payload_len + CHECKSUM_LEN;
215        if bytes.len() < expected_total {
216            return Err(EnvelopeError::Truncated {
217                expected: expected_total,
218                actual: bytes.len(),
219            });
220        }
221        if bytes.len() > expected_total {
222            return Err(EnvelopeError::TrailingBytes(bytes.len() - expected_total));
223        }
224        let payload = bytes[HEADER_LEN..HEADER_LEN + payload_len].to_vec();
225        let payload_crc32 = u32::from_le_bytes(
226            bytes[HEADER_LEN + payload_len..expected_total]
227                .try_into()
228                .expect("slice len"),
229        );
230        let envelope = Self {
231            protocol_version,
232            message_type,
233            payload,
234            payload_crc32,
235        };
236        envelope.verify()?;
237        Ok(envelope)
238    }
239}
240
241impl fmt::Debug for ProtocolEnvelope {
242    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
243        f.debug_struct("ProtocolEnvelope")
244            .field("protocol_version", &self.protocol_version)
245            .field("message_type", &self.message_type)
246            .field("payload_len", &self.payload.len())
247            .finish_non_exhaustive()
248    }
249}
250
251#[cfg(test)]
252mod tests {
253    use super::*;
254
255    #[test]
256    fn crc32_matches_reference_vector() {
257        // The canonical IEEE CRC-32 check value.
258        assert_eq!(crc32(&[b"123456789"]), 0xCBF4_3926);
259        assert_eq!(crc32(&[]), 0);
260        // Multi-part input equals the one-shot form.
261        assert_eq!(crc32(&[b"1234", b"56789"]), crc32(&[b"123456789"]));
262    }
263
264    #[test]
265    fn round_trip() {
266        let envelope = ProtocolEnvelope::new(7, b"hello".to_vec());
267        let bytes = envelope.encode();
268        assert_eq!(ProtocolEnvelope::decode(&bytes).unwrap(), envelope);
269        // Encoding is deterministic.
270        assert_eq!(envelope.encode(), bytes);
271    }
272
273    #[test]
274    fn bit_flip_breaks_checksum() {
275        let envelope = ProtocolEnvelope::new(1, vec![9u8; 64]);
276        let mut bytes = envelope.encode();
277        bytes[HEADER_LEN] ^= 0x01;
278        assert_eq!(
279            ProtocolEnvelope::decode(&bytes),
280            Err(EnvelopeError::ChecksumMismatch)
281        );
282    }
283
284    #[test]
285    fn unknown_version_fails_closed() {
286        let mut envelope = ProtocolEnvelope::new(1, vec![]);
287        envelope.protocol_version = PROTOCOL_VERSION + 1;
288        envelope.payload_crc32 =
289            ProtocolEnvelope::checksum(envelope.protocol_version, 1, &envelope.payload);
290        assert!(matches!(
291            envelope.verify(),
292            Err(EnvelopeError::UnsupportedVersion { .. })
293        ));
294        // The wire form fails closed too, not just the in-memory form.
295        let bytes = envelope.encode();
296        assert!(matches!(
297            ProtocolEnvelope::decode(&bytes),
298            Err(EnvelopeError::UnsupportedVersion { .. })
299        ));
300    }
301
302    #[test]
303    fn truncation_and_trailing_bytes_fail() {
304        let envelope = ProtocolEnvelope::new(3, vec![1, 2, 3]);
305        let bytes = envelope.encode();
306        assert!(matches!(
307            ProtocolEnvelope::decode(&bytes[..bytes.len() - 1]),
308            Err(EnvelopeError::Truncated { .. })
309        ));
310        // A frame shorter than header + checksum is truncated.
311        assert!(matches!(
312            ProtocolEnvelope::decode(&bytes[..HEADER_LEN]),
313            Err(EnvelopeError::Truncated { .. })
314        ));
315        let mut longer = bytes.clone();
316        longer.push(0);
317        assert_eq!(
318            ProtocolEnvelope::decode(&longer),
319            Err(EnvelopeError::TrailingBytes(1))
320        );
321    }
322
323    #[test]
324    fn oversize_length_prefix_fails_before_copying() {
325        let mut bytes = Vec::new();
326        bytes.extend_from_slice(&PROTOCOL_VERSION.to_le_bytes());
327        bytes.extend_from_slice(&1u32.to_le_bytes());
328        bytes.extend_from_slice(&(MAX_MESSAGE_PAYLOAD_BYTES as u32 + 1).to_le_bytes());
329        bytes.extend_from_slice(&[0u8; CHECKSUM_LEN]);
330        assert!(matches!(
331            ProtocolEnvelope::decode(&bytes),
332            Err(EnvelopeError::PayloadTooLarge(_))
333        ));
334    }
335
336    #[test]
337    fn envelope_serde_round_trip() {
338        crate::test_support::assert_serde_round_trip(&ProtocolEnvelope::new(42, vec![1, 2, 3]));
339    }
340}
mongreldb_protocol/envelope.rs

mongreldb_protocol/
envelope.rs
