lace_sans_io/iltp/

stream_encoding.rs

//! ILTP stream encodings.
//!
//! These let each direction of a logical session carry its ILTP items
//! compressed. The direction's encoding is selected from the two raw prefaces
//! and covers every byte after the sender's preface. A compressed encoding ends
//! with its end mark, after which the decoder leaves raw bytes untouched for
//! the carrier: a fresh preface or EOF.
//!
//! Encoders and decoders are sans-IO and per direction. Callers own all wire
//! buffering: both read a caller slice and write into a caller-bounded output
//! slice.

#[cfg(feature = "deflate")]
mod deflate;
#[cfg(feature = "zstd")]
mod zstd;

use std::fmt;

/// Raw preset dictionary priming every fresh compressed encoder and decoder.
///
/// It holds ILTP item vocabulary: the operand body and generic rule
/// fragments, then proof, policy announcement and answers, listing, response,
/// change, request and fence text. The most frequent items come last, where
/// deflate's 4 KiB window keeps them longest. libzstd needs a raw-content
/// dictionary of at least 512 bytes; shorter ones degrade compression for the
/// whole stream.
pub const STREAM_ENCODING_DICTIONARY: &[u8] = concat!(
    // Operand body and the generic parts of compiled 040 rules.
    "::policy-body\n🖧: B.",
    ".H3\nData-Length: ",
    "\n\n",
    "ExposeRecord(",
    ") :- Have(",
    "), Field(",
    ",'By','0','V.",
    ".H3'), not Field(",
    ",'By','1',_).\n",
    "SelectAdvertisedCandidate(",
    ") :- Advertised(",
    "), AdvertisedField(",
    "','0','",
    "'), not AdvertisedField(",
    ",'1',_), PeerConfidential(",
    "SelectStoredRecord(",
    // Proof: none, or a Seal over the empty Blob.
    "::as unknown\n",
    "::as\n🖧: S.",
    ".H3\nBy: V.",
    ".H3\nMark: ",
    "\n🖧: P.",
    ".H3\nChallenge-side0: ",
    "\nChallenge-side1: ",
    "\nChannel-Binding: ",
    "\nProof-Profile: lace-iltp-peer-proof-v3\nProof-Role: side0\n",
    "🖧: B.svyLzSM7ffc91i~XDbkMnuOsdjsw_6GrXpTSckqHlpO.H3\nData-Length: 0\n\n\n",
    // Policy announcement and its answers.
    "::policy B.",
    ".H3\n::policy-known\n",
    "::policy-missing\n",
    // Listings, projected fields and symbols.
    "\n::offers\nS.",
    ".H3\n\tBy: V.",
    ".H3\n\t",
    ": ",
    "\nP.",
    ".H3\n\t:Length: ",
    "\n\t:Type: S\n\tData-Length: ",
    "\n\tMark: ",
    "\n\n::offers-unshared\n",
    "::symbols\n",
    " 1\n\n",
    // Responses.
    "::response\n🖧: S.",
    ".H3\nBy: V.",
    ".H3\nMark: ",
    "\n🖧: P.",
    ".H3\n",
    "\n🖧: B.",
    ".H3\nData-Length: ",
    "\n\n",
    // Later-round changes and requests.
    "::offers-changed\n+ S.",
    ".H3\n- S.",
    ".H3\n\n",
    "::want\nS.",
    ".H3\nP.",
    ".H3\n\n",
    // Fence and an unchanged round.
    "\n::round-end\n::offers-changed\n\n::want\n\n",
)
.as_bytes();

/// One direction's stream encoding, named by its preface token.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum StreamEncoding {
    /// Plain ILTP bytes. Always available and implied by every offer.
    Identity,
    /// Raw deflate (RFC 1951) with a 4 KiB window, sync flushes and the
    /// preset dictionary.
    #[cfg(feature = "deflate")]
    Deflate12c,
    /// Magicless zstd (RFC 8878) with at most a 1 MiB window, no checksum,
    /// content size or dictionary ID, and the raw-content preset dictionary.
    #[cfg(feature = "zstd")]
    Zstd20c,
}

impl StreamEncoding {
    /// Compiled-in compressed encodings, most preferred first.
    pub const COMPILED: &'static [Self] = &[
        #[cfg(feature = "zstd")]
        Self::Zstd20c,
        #[cfg(feature = "deflate")]
        Self::Deflate12c,
    ];

    pub fn token(self) -> &'static str {
        match self {
            Self::Identity => "identity",
            #[cfg(feature = "deflate")]
            Self::Deflate12c => "deflate12c",
            #[cfg(feature = "zstd")]
            Self::Zstd20c => "zstd20c",
        }
    }

    /// Parse a token; tokens unknown to this build return `None`.
    pub fn from_token(token: &str) -> Option<Self> {
        match token {
            "identity" => Some(Self::Identity),
            #[cfg(feature = "deflate")]
            "deflate12c" => Some(Self::Deflate12c),
            #[cfg(feature = "zstd")]
            "zstd20c" => Some(Self::Zstd20c),
            _ => None,
        }
    }

    /// Parse one configured token list: the encode preference and decode set
    /// of a carrier. Tokens unknown to this build, `identity`, and repeats are
    /// errors.
    pub fn parse_list<'a>(
        tokens: impl IntoIterator<Item = &'a str>,
    ) -> Result<Vec<Self>, &'static str> {
        let encodings = tokens
            .into_iter()
            .map(|token| {
                Self::from_token(token)
                    .ok_or("stream encoding token is unknown or not compiled into this build")
            })
            .collect::<Result<Vec<_>, _>>()?;
        StreamOffers::new(&encodings)?;
        Ok(encodings)
    }

    /// The encoding of one direction: the first entry of the sender's
    /// `Encode` list that the receiver's `Decode` list contains, else
    /// `Identity`, which both lists imply.
    pub fn select(sender_encode: &[Self], receiver_decode: &[Self]) -> Self {
        sender_encode
            .iter()
            .copied()
            .find(|encoding| *encoding == Self::Identity || receiver_decode.contains(encoding))
            .unwrap_or(Self::Identity)
    }
}

/// The 065 offers one preface carries, restricted to encodings this build
/// implements. `identity` ends both lists implicitly and is never listed.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct StreamOffers {
    /// Encodings the sender decodes, in ascending token order.
    decode: Vec<StreamEncoding>,
    /// Encodings the sender produces, most preferred first.
    encode: Vec<StreamEncoding>,
}

impl StreamOffers {
    /// The offers of a carrier configured with one ordered list, used as its
    /// encode preference and its decode set.
    pub fn new(encodings: &[StreamEncoding]) -> Result<Self, &'static str> {
        if encodings.contains(&StreamEncoding::Identity) {
            return Err("stream encodings must not list identity, which is always implied");
        }
        for (index, encoding) in encodings.iter().enumerate() {
            if encodings[..index].contains(encoding) {
                return Err("stream encodings must not repeat a token");
            }
        }
        let mut decode = encodings.to_vec();
        decode.sort_by_key(|encoding| encoding.token());
        Ok(Self {
            decode,
            encode: encodings.to_vec(),
        })
    }

    /// The encodings of validated preface tokens. Tokens this build does not
    /// implement cannot match a local list, so dropping them keeps selection.
    pub(super) fn known(tokens: Vec<&str>) -> Vec<StreamEncoding> {
        tokens
            .into_iter()
            .filter_map(StreamEncoding::from_token)
            .collect()
    }

    pub(super) fn from_lists(decode: Vec<StreamEncoding>, encode: Vec<StreamEncoding>) -> Self {
        Self { decode, encode }
    }

    /// Encodings the sender decodes: its preface's `Decode` field.
    pub fn decode(&self) -> &[StreamEncoding] {
        &self.decode
    }

    /// Encodings the sender produces, most preferred first: its preface's
    /// `Encode` field.
    pub fn encode(&self) -> &[StreamEncoding] {
        &self.encode
    }

    /// The encoding of the direction from `sender` to `receiver`.
    pub fn select(sender: &Self, receiver: &Self) -> StreamEncoding {
        StreamEncoding::select(&sender.encode, &receiver.decode)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StreamEncodingError {
    /// Malformed encoded peer input, including a declared window above the
    /// token's maximum or a back-reference beyond retained history: the
    /// window plus, for zlib, output already written by the same decode
    /// call. Maps to `PeerProtocol`.
    Malformed(String),
    /// Local encoder or decoder failure: `Integration`.
    Local(String),
}

impl fmt::Display for StreamEncodingError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Malformed(detail) => write!(f, "malformed ILTP stream encoding: {detail}"),
            Self::Local(detail) => write!(f, "ILTP stream encoding failed: {detail}"),
        }
    }
}

impl std::error::Error for StreamEncodingError {}

/// Encoder for one direction of one logical session.
pub struct StreamEncoder(Encoder);

enum Encoder {
    Identity,
    #[cfg(feature = "deflate")]
    Deflate(deflate::Encoder),
    #[cfg(feature = "zstd")]
    Zstd(zstd::Encoder),
}

// Without compressed encodings, levels have no use.
#[cfg_attr(
    not(any(feature = "deflate", feature = "zstd")),
    allow(unused_variables)
)]
impl StreamEncoder {
    /// Fresh encoder at the default level: deflate 9, zstd 3.
    pub fn new(encoding: StreamEncoding) -> Result<Self, StreamEncodingError> {
        let level = match encoding {
            StreamEncoding::Identity => 0,
            #[cfg(feature = "deflate")]
            StreamEncoding::Deflate12c => deflate::DEFAULT_LEVEL,
            #[cfg(feature = "zstd")]
            StreamEncoding::Zstd20c => zstd::DEFAULT_LEVEL,
        };
        Self::with_level(encoding, level)
    }

    /// Fresh encoder at a sender-chosen level. The decoder needs only the
    /// encoding. `Identity` ignores the level.
    pub fn with_level(encoding: StreamEncoding, level: i32) -> Result<Self, StreamEncodingError> {
        Ok(Self(match encoding {
            StreamEncoding::Identity => Encoder::Identity,
            #[cfg(feature = "deflate")]
            StreamEncoding::Deflate12c => Encoder::Deflate(deflate::Encoder::new(level)?),
            #[cfg(feature = "zstd")]
            StreamEncoding::Zstd20c => Encoder::Zstd(zstd::Encoder::new(level)?),
        }))
    }

    /// Encode a prefix of `input` into at most `output.len()` bytes.
    ///
    /// `Write` takes plaintext, `Flush` completes all plaintext written so far
    /// so the peer can decode it from the wire bytes alone (every batch end),
    /// and `End` writes the end mark, after which the encoder takes nothing
    /// more. Callers repeat a step with fresh output room until it is done;
    /// a flushing step needs at least [`ENCODE_MIN_OUTPUT`] bytes of room.
    pub fn encode(
        &mut self,
        input: &[u8],
        mode: EncodeMode,
        output: &mut [u8],
    ) -> Result<EncodeProgress, StreamEncodingError> {
        match &mut self.0 {
            Encoder::Identity => {
                let count = match mode {
                    EncodeMode::Write => input.len().min(output.len()),
                    EncodeMode::Flush | EncodeMode::End => 0,
                };
                output[..count].copy_from_slice(&input[..count]);
                Ok(EncodeProgress {
                    consumed: count,
                    produced: count,
                    done: mode != EncodeMode::Write || count == input.len(),
                })
            }
            #[cfg(feature = "deflate")]
            Encoder::Deflate(encoder) => encoder.encode(input, mode, output),
            #[cfg(feature = "zstd")]
            Encoder::Zstd(encoder) => encoder.encode(input, mode, output),
        }
    }

    /// Whether this encoder passes plaintext through unchanged.
    pub fn is_identity(&self) -> bool {
        matches!(self.0, Encoder::Identity)
    }
}

/// The step one [`StreamEncoder::encode`] call takes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EncodeMode {
    Write,
    Flush,
    End,
}

/// Output room a flushing [`StreamEncoder::encode`] step needs: zlib repeats
/// sync-flush markers when a flush ends with no room left.
pub const ENCODE_MIN_OUTPUT: usize = 16;

/// Bytes one [`StreamEncoder::encode`] call took and wrote, and whether its
/// step is complete: all input taken for `Write`, everything flushed or ended
/// for `Flush` and `End`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EncodeProgress {
    pub consumed: usize,
    pub produced: usize,
    pub done: bool,
}

/// Bytes one [`StreamDecoder::decode`] call took from its input and wrote to
/// its output.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DecodeProgress {
    pub consumed: usize,
    pub produced: usize,
}

/// Decoder for one direction of one logical session.
pub struct StreamDecoder(Decoder);

enum Decoder {
    Identity,
    #[cfg(feature = "deflate")]
    Deflate(deflate::Decoder),
    #[cfg(feature = "zstd")]
    Zstd(zstd::Decoder),
}

impl StreamDecoder {
    pub fn new(encoding: StreamEncoding) -> Result<Self, StreamEncodingError> {
        Ok(Self(match encoding {
            StreamEncoding::Identity => Decoder::Identity,
            #[cfg(feature = "deflate")]
            StreamEncoding::Deflate12c => Decoder::Deflate(deflate::Decoder::new()?),
            #[cfg(feature = "zstd")]
            StreamEncoding::Zstd20c => Decoder::Zstd(zstd::Decoder::new()?),
        }))
    }

    /// Decode a prefix of `input` into at most `output.len()` bytes.
    ///
    /// Bytes not consumed stay with the caller. Zero progress means the
    /// decoder needs more input or more output space. After the end mark the
    /// decoder consumes nothing, so the input following it remains raw.
    pub fn decode(
        &mut self,
        input: &[u8],
        output: &mut [u8],
    ) -> Result<DecodeProgress, StreamEncodingError> {
        match &mut self.0 {
            Decoder::Identity => {
                let count = input.len().min(output.len());
                output[..count].copy_from_slice(&input[..count]);
                Ok(DecodeProgress {
                    consumed: count,
                    produced: count,
                })
            }
            #[cfg(feature = "deflate")]
            Decoder::Deflate(decoder) => decoder.decode(input, output),
            #[cfg(feature = "zstd")]
            Decoder::Zstd(decoder) => decoder.decode(input, output),
        }
    }

    /// Whether the encoding is at its end: always for `Identity`, after the
    /// end mark for a compressed encoding.
    pub fn is_ended(&self) -> bool {
        match &self.0 {
            Decoder::Identity => true,
            #[cfg(feature = "deflate")]
            Decoder::Deflate(decoder) => decoder.is_ended(),
            #[cfg(feature = "zstd")]
            Decoder::Zstd(decoder) => decoder.is_ended(),
        }
    }
}