decdn_protocol/

framing.rs

//! Length-prefixed wire framing for deCDN ALPNs (ADR 013).
//!
//! Every message on every QUIC stream is `varint(len) || postcard_bytes[0..len]`.
//! The varint uses postcard's native continuation-bit encoding (7 data bits per
//! byte, MSB = continuation, little-endian, 1–5 bytes for `u32`).
//!
//! Receivers read the varint length, reject anything above the ALPN's frame
//! ceiling before allocation, allocate exactly that many bytes, then deserialize
//! the frame with [`decode_message`] (a thin wrapper over
//! [`postcard::take_from_bytes`] that preserves the trailing remainder for
//! future Tier-1 extension fields).
//!
//! Every read names its own ceiling, sized to the largest legitimate message
//! that can arrive at that read (ADR 013 § Wire Framing). The ceilings live
//! beside the messages they bound: [`crate::message::MAX_PROBE_FRAME`],
//! [`crate::client::MAX_CLIENT_CONTROL_FRAME`], [`crate::client::MAX_CHUNK_FRAME`]
//! and [`crate::dht::MAX_DHT_FRAME`]. There is no global ceiling, so no read can
//! fall back to a bound far above what its ALPN carries.

use serde::Serialize;
use serde::de::DeserializeOwned;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

/// Errors produced by the framing helpers. The handler-layer mapping to QUIC
/// application error codes lives in the node crate.
#[derive(Debug, thiserror::Error)]
pub enum FrameError {
    /// The underlying stream failed while reading or writing the frame.
    #[error("frame I/O error: {0}")]
    Io(#[from] std::io::Error),
    /// The frame is longer than the ceiling allows. A reader returns this for
    /// a length prefix above its ceiling, before it buffers anything.
    /// [`write_frame`] returns it for a payload too long for the `u32` length
    /// prefix, with `max` set to `u32::MAX`.
    #[error("frame length {len} exceeds the frame ceiling ({max})")]
    TooLarge {
        /// The announced (reader) or attempted (writer) payload length, in bytes.
        len: u64,
        /// The ceiling that applied, in bytes.
        max: u32,
    },
    /// The length prefix is not a well-formed varint — a non-terminating
    /// continuation run through all five bytes, or a fifth byte overflowing
    /// `u32`. A short or empty buffer reads as truncated, not malformed.
    #[error("malformed varint length prefix")]
    Varint,
    /// The frame's bytes are not a valid postcard encoding of the expected
    /// message.
    #[error("postcard decode error: {0}")]
    Decode(#[from] postcard::Error),
}

/// Serialize `msg` to a postcard buffer suitable for passing to [`write_frame`].
pub fn encode_message<M: Serialize>(msg: &M) -> Result<Vec<u8>, FrameError> {
    Ok(postcard::to_allocvec(msg)?)
}

/// Decode a frame into a message and the unconsumed remainder. The remainder
/// is reserved for Tier-1 extension fields (ADR 013 §Tier 1).
pub fn decode_message<M: DeserializeOwned>(frame: &[u8]) -> Result<(M, &[u8]), FrameError> {
    Ok(postcard::take_from_bytes::<M>(frame)?)
}

/// A top-level ALPN protocol enum (one per ALPN — [`crate::ProbeMessage`],
/// [`crate::ClientMessage`], [`crate::DhtMessage`]). Serialized as the
/// outermost postcard value in a frame, so its variant discriminant is the
/// frame's leading varint.
///
/// Implementors expose their declared-variant count so the framing layer can
/// tell an *unknown* discriminant (a variant this build does not know —
/// ADR 013 `UNSUPPORTED_MESSAGE`, `0x01`) apart from a genuine parse fault
/// (`MALFORMED_MESSAGE`, `0x03`) after a failed [`decode_message`].
pub trait TopLevelEnum {
    /// Number of declared variants. Postcard assigns discriminants
    /// `0..VARIANT_COUNT` in declaration order, so any leading discriminant
    /// `>= VARIANT_COUNT` names a variant this build does not know.
    const VARIANT_COUNT: u32;
}

/// Classify a failed top-level-enum decode per ADR 013: return `true` iff the
/// frame's leading discriminant names a variant at or beyond
/// `M::VARIANT_COUNT` — i.e. an unknown/unsupported variant the receiver must
/// answer with `UNSUPPORTED_MESSAGE` (`0x01`) rather than `MALFORMED_MESSAGE`
/// (`0x03`).
///
/// A frame that does not begin with a well-formed varint (empty, a
/// non-terminating continuation run, or a 5th byte that overflows `u32`) is a
/// genuine parse fault, so this returns `false` and the caller keeps the
/// `MALFORMED` classification. Callers should only consult this after
/// [`decode_message`] has actually failed; a valid in-range discriminant whose
/// inner payload is malformed also stays `MALFORMED` (this returns `false`).
pub fn is_unknown_variant<M: TopLevelEnum>(frame: &[u8]) -> bool {
    matches!(leading_varint_u32(frame), Some(d) if d >= M::VARIANT_COUNT)
}

/// Decode the leading postcard `u32` varint from `frame` (the top-level enum
/// discriminant) without consuming the remainder. Returns `None` when the
/// bytes do not begin with a well-formed varint — the synchronous mirror of
/// [`read_varint_u32`]'s validation, over an in-memory slice.
fn leading_varint_u32(frame: &[u8]) -> Option<u32> {
    let mut result: u32 = 0;
    for (i, &byte) in frame.iter().take(5).enumerate() {
        let data = u32::from(byte & 0x7F);
        // On the 5th byte (i == 4) only the low 4 bits are valid; reject overflow.
        if i == 4 && byte & 0x70 != 0 {
            return None;
        }
        let shift = u32::try_from(i).ok()?.saturating_mul(7);
        let shifted = data.checked_shl(shift)?;
        result |= shifted;
        if byte & 0x80 == 0 {
            return Some(result);
        }
    }
    None
}

/// Try to locate one complete length-prefixed frame at the front of `buf`
/// without consuming it, returning `(header_len, payload_len)` — the byte
/// counts of the varint prefix and the payload it announces. Returns `Ok(None)`
/// when `buf` does not yet hold a complete frame (the varint is truncated, or
/// fewer than `payload_len` payload bytes have arrived).
///
/// This is the synchronous, non-consuming counterpart to [`read_frame`], for a
/// caller that fills a byte buffer incrementally with **cancellation-safe**
/// reads and must split frames off it after each top-up — e.g. the per-voucher
/// reader's read-ahead (#1486), which reads ahead under a timeout that
/// [`read_frame`] (built on `read_exact`) could not survive without losing
/// partial bytes. The
/// caller slices the payload as `buf[header_len..header_len + payload_len]`,
/// decodes it with [`decode_message`], then drains `header_len + payload_len`
/// bytes.
///
/// The announced length is validated against `max` — the read's frame ceiling —
/// as soon as the varint is complete, before the caller waits for (or buffers)
/// the payload. This mirrors [`read_frame`]'s pre-allocation denial-of-service
/// bound.
///
/// # Errors
///
/// [`FrameError::Varint`] for a malformed length prefix (a non-terminating
/// continuation run within the first 5 bytes, or a 5th byte overflowing `u32`);
/// [`FrameError::TooLarge`] when the announced length exceeds `max`.
pub fn parse_frame(buf: &[u8], max: u32) -> Result<Option<(usize, usize)>, FrameError> {
    let Some((len, header_len)) = decode_varint_prefix(buf)? else {
        return Ok(None);
    };
    if len > max {
        return Err(FrameError::TooLarge {
            len: u64::from(len),
            max,
        });
    }
    let payload_len = len as usize;
    if buf.len().saturating_sub(header_len) < payload_len {
        return Ok(None);
    }
    Ok(Some((header_len, payload_len)))
}

/// Decode a postcard `u32` varint from the front of `buf`, returning
/// `(value, bytes_consumed)`, `Ok(None)` when the varint is truncated (fewer
/// than its continuation bytes have arrived), or [`FrameError::Varint`] when the
/// bytes present are already malformed. The synchronous mirror of
/// [`read_varint_u32`].
fn decode_varint_prefix(buf: &[u8]) -> Result<Option<(u32, usize)>, FrameError> {
    let mut result: u32 = 0;
    for (i, &byte) in buf.iter().take(5).enumerate() {
        let data = u32::from(byte & 0x7F);
        // On the 5th byte (i == 4) only the low 4 bits are valid; reject overflow.
        if i == 4 && byte & 0x70 != 0 {
            return Err(FrameError::Varint);
        }
        let shift = u32::try_from(i)
            .map_err(|_| FrameError::Varint)?
            .saturating_mul(7);
        let shifted = data.checked_shl(shift).ok_or(FrameError::Varint)?;
        result |= shifted;
        if byte & 0x80 == 0 {
            return Ok(Some((result, i + 1)));
        }
    }
    // Ran out of buffer mid-varint (< 5 bytes, all continuations) → truncated;
    // exactly 5 continuation bytes with the terminator missing is malformed.
    if buf.len() >= 5 {
        return Err(FrameError::Varint);
    }
    Ok(None)
}

/// Read one length-prefixed frame from an async reader.
///
/// The length prefix is validated against `max` *before* any payload
/// allocation, so a peer cannot make the reader buffer more than `max` bytes.
/// Pass the frame ceiling for the messages that can arrive at this read
/// (ADR 013 § Wire Framing): a ceiling sized to the largest legitimate message
/// keeps a stalled oversized frame from pinning memory.
///
/// # Errors
///
/// [`FrameError::TooLarge`] when the announced length exceeds `max`,
/// [`FrameError::Varint`] for a malformed length prefix, and
/// [`FrameError::Io`] when the stream fails or ends mid-frame.
pub async fn read_frame<R: AsyncRead + Unpin>(r: &mut R, max: u32) -> Result<Vec<u8>, FrameError> {
    let len = read_varint_u32(r).await?;
    if len > max {
        return Err(FrameError::TooLarge {
            len: u64::from(len),
            max,
        });
    }
    let mut buf = vec![0u8; len as usize];
    r.read_exact(&mut buf).await?;
    Ok(buf)
}

/// Write one length-prefixed frame to an async writer.
///
/// The writer checks only that the length fits the `u32` prefix. The chunk
/// encoders refuse a frame above [`crate::client::MAX_CHUNK_FRAME`]. Every other
/// message is small by shape, and the worst-case encode tests beside each
/// ceiling pin that it fits the read that receives it.
///
/// # Errors
///
/// [`FrameError::TooLarge`] when the payload length does not fit a `u32`, and
/// [`FrameError::Io`] when the stream fails.
pub async fn write_frame<W: AsyncWrite + Unpin>(
    w: &mut W,
    payload: &[u8],
) -> Result<(), FrameError> {
    let len = u32::try_from(payload.len()).map_err(|_| FrameError::TooLarge {
        len: u64::try_from(payload.len()).unwrap_or(u64::MAX),
        max: u32::MAX,
    })?;
    write_varint_u32(w, len).await?;
    w.write_all(payload).await?;
    Ok(())
}

async fn read_varint_u32<R: AsyncRead + Unpin>(r: &mut R) -> Result<u32, FrameError> {
    let mut result: u32 = 0;
    for i in 0u32..5 {
        let mut b = [0u8; 1];
        r.read_exact(&mut b).await?;
        let byte = b[0];
        let data = u32::from(byte & 0x7F);
        let shift = i.saturating_mul(7);
        // On the 5th byte (i=4) only the low 4 bits are valid; reject overflow.
        if i == 4 && byte & 0x70 != 0 {
            return Err(FrameError::Varint);
        }
        let shifted = data.checked_shl(shift).ok_or(FrameError::Varint)?;
        result |= shifted;
        if byte & 0x80 == 0 {
            return Ok(result);
        }
    }
    Err(FrameError::Varint)
}

/// Encode `value` as a postcard varint into the caller's stack buffer, returning
/// the number of bytes written (1..=5).
///
/// A `u32` encodes to at most 5 varint bytes (`ceil(32/7) = 5`), which is why the
/// buffer is a fixed `[u8; 5]` and why the count returned always fits it.
///
/// The `out.get_mut(idx)` guards exist to satisfy the `indexing_slicing` clippy lint.
/// Each one stops the encode rather than dropping a byte and counting it anyway: the
/// returned length is bytes *written*, never bytes intended, so no caller can be
/// handed a count that outruns the bytes behind it.
#[must_use]
pub(crate) fn encode_varint_u32(mut value: u32, out: &mut [u8; 5]) -> usize {
    let mut idx = 0usize;
    loop {
        let byte = (value & 0x7F) as u8;
        value >>= 7;
        let last = value == 0;
        let Some(slot) = out.get_mut(idx) else {
            break;
        };
        *slot = if last { byte } else { byte | 0x80 };
        idx = idx.saturating_add(1);
        if last {
            break;
        }
    }
    debug_assert!(idx <= 5, "u32 varint must fit in 5 bytes, idx={idx}");
    idx
}

async fn write_varint_u32<W: AsyncWrite + Unpin>(w: &mut W, value: u32) -> std::io::Result<()> {
    let mut buf = [0u8; 5];
    let idx = encode_varint_u32(value, &mut buf);
    // A length prefix that silently shortens to nothing is worse than a failed write:
    // `write_frame` would put the payload on the wire behind it, the peer would read
    // the payload's first byte as the length, and the stream — and the cumulative
    // wire-byte count vouchers are paid against — would desync with no error on
    // either side. Unreachable while `buf` is `[u8; 5]`; refuse rather than rely on it.
    let Some(bytes) = buf.get(..idx) else {
        return Err(std::io::Error::other(format!(
            "varint length prefix wants {idx} bytes but only {} were encoded",
            buf.len()
        )));
    };
    w.write_all(bytes).await
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::indexing_slicing
)]
mod proptests;

#[cfg(test)]
mod tests;