rustls/conn/

receive.rs

use alloc::vec::Vec;
use core::marker::PhantomData;
use core::mem;
use core::ops::Range;
use std::io::{self, Read};

use super::SendOutput;
use crate::SideData;
use crate::common_state::{
    ConnectionOutput, Event, Output, OutputEvent, Side, UnborrowedPayload, maybe_send_fatal_alert,
};
use crate::conn::private::SideOutput;
use crate::conn::{ConnectionCore, StateMachine};
use crate::crypto::cipher::{Decrypted, DecryptionState, EncodedMessage, Payload};
use crate::enums::{ContentType, HandshakeType, ProtocolVersion};
use crate::error::{AlertDescription, Error, PeerMisbehaved};
use crate::log::{trace, warn};
use crate::msgs::{
    AlertLevel, AlertLevelName, AlertMessagePayload, Deframed, Deframer, Delocator,
    HandshakeAlignedProof, Locator, Message, MessagePayload,
};
use crate::quic::QuicOutput;

pub(crate) struct MessageIter<'a, 'm, Side: SideData> {
    input: &'m mut dyn TlsInputBuffer,
    recv: &'a mut ReceivePath,
    state: &'a mut Result<Side::State, Error>,
    output: JoinOutput<'a>,
}

impl<'a, 'm, Side: SideData> MessageIter<'a, 'm, Side> {
    pub(crate) fn new(
        input: &'m mut dyn TlsInputBuffer,
        quic: Option<&'a mut dyn QuicOutput>,
        conn: &'a mut ConnectionCore<Side>,
    ) -> Self {
        Self {
            recv: &mut conn.common.recv,
            input,
            state: &mut conn.state,
            output: JoinOutput {
                outputs: &mut conn.common.outputs,
                quic,
                send: &mut conn.common.send,
                side: &mut conn.side,
            },
        }
    }

    pub(super) fn receive(
        input: &'m mut dyn TlsInputBuffer,
        state: &'a mut Result<Side::State, Error>,
        recv: &'a mut ReceivePath,
        output: JoinOutput<'a>,
    ) -> Self {
        Self {
            recv,
            input,
            state,
            output,
        }
    }

    pub(crate) fn next(&mut self) -> Option<Result<UnborrowedPayload, Error>> {
        let mut st = match mem::replace(self.state, Err(Error::HandshakeNotComplete)) {
            Ok(state) => state,
            Err(e) => {
                *self.state = Err(e.clone());
                return Some(Err(e));
            }
        };

        let mut plaintext = None;
        while st.wants_input() {
            let buffer = self.input.slice_mut();
            let locator = Locator::new(buffer);
            let res = self.recv.deframe(buffer);

            let mut output = CaptureAppData {
                recv: self.recv,
                other: &mut self.output,
                plaintext_locator: &locator,
                received_plaintext: &mut plaintext,
                _message_lifetime: PhantomData,
            };

            let opt_msg = match res {
                Ok(opt_msg) => opt_msg,
                Err(e) => {
                    maybe_send_fatal_alert(output.other.send, &e);
                    if let Error::DecryptError = e {
                        st.handle_decrypt_error();
                    }
                    *self.state = Err(e.clone());
                    return Some(Err(e));
                }
            };

            let Some(msg) = opt_msg else {
                break;
            };

            let Decrypted {
                plaintext: msg,
                want_close_before_decrypt,
            } = msg;

            if want_close_before_decrypt {
                output
                    .other
                    .send
                    .send_alert(AlertLevel::Warning, AlertDescription::CloseNotify);
            } else if msg.payload.is_empty()
                && matches!(msg.typ, ContentType::Handshake | ContentType::Alert)
            {
                // <https://datatracker.ietf.org/doc/html/rfc8446#section-5.4>
                output
                    .other
                    .send
                    .send_alert(AlertLevel::Fatal, AlertDescription::UnexpectedMessage);
                return Some(Err(PeerMisbehaved::EmptyFragment.into()));
            }

            let hs_aligned = output.recv.deframer.aligned();
            let result = match output
                .recv
                .receive_message(msg, hs_aligned, output.other.send)
            {
                Ok(Some(input)) => st.handle(input, &mut output),
                Ok(None) => Ok(st),
                Err(e) => Err(e),
            };

            match result {
                Ok(new) => st = new,
                Err(e) => {
                    maybe_send_fatal_alert(output.other.send, &e);
                    *self.state = Err(e.clone());
                    return Some(Err(e));
                }
            }

            if self.recv.has_received_close_notify {
                // "Any data received after a closure alert has been received MUST be ignored."
                // -- <https://datatracker.ietf.org/doc/html/rfc8446#section-6.1>

                // First, discard actually-processed bytes.
                self.input
                    .discard(self.recv.deframer.take_discard());

                // Then the rest of any input data.
                let entirety = self.input.slice_mut().len();
                self.recv.deframer.set_discard(entirety);
                self.input.received_close_notify();
                break;
            }

            if let Some(payload) = plaintext.take() {
                *self.state = Ok(st);
                return Some(Ok(payload));
            }
        }

        *self.state = Ok(st);
        None
    }

    pub(super) fn input(&mut self) -> &mut dyn TlsInputBuffer {
        self.input
    }

    pub(crate) fn state(&self) -> &Result<Side::State, Error> {
        self.state
    }
}

pub(crate) struct ReceivePath {
    side: Side,
    pub(crate) decrypt_state: DecryptionState,
    pub(crate) may_receive_application_data: bool,
    /// If the peer has signaled end of stream.
    pub(crate) has_received_close_notify: bool,
    temper_counters: TemperCounters,
    pub(crate) negotiated_version: Option<ProtocolVersion>,
    pub(crate) deframer: Deframer,

    /// We limit consecutive empty fragments to avoid a route for the peer to send
    /// us significant but fruitless traffic.
    seen_consecutive_empty_fragments: u8,

    pub(crate) tls13_tickets_received: u32,
}

impl ReceivePath {
    pub(crate) fn new(side: Side) -> Self {
        Self {
            side,
            decrypt_state: DecryptionState::new(),
            may_receive_application_data: false,
            has_received_close_notify: false,
            temper_counters: TemperCounters::default(),
            negotiated_version: None,
            deframer: Deframer::default(),
            seen_consecutive_empty_fragments: 0,
            tls13_tickets_received: 0,
        }
    }

    /// Pull a message out of the deframer and send any messages that need to be sent as a result.
    fn deframe<'b>(&mut self, buffer: &'b mut [u8]) -> Result<Option<Decrypted<'b>>, Error> {
        let locator = Locator::new(buffer);

        let mut want_close_before_decrypt = false;
        loop {
            // before processing any more of `buffer`, return any extant messages from `deframer`
            if let Some(span) = self.deframer.complete_span() {
                let plaintext = self.deframer.message(span, buffer);

                // trial decryption finishes with the first handshake message after it started.
                self.decrypt_state
                    .finish_trial_decryption();

                return Ok(Some(Decrypted {
                    plaintext,
                    want_close_before_decrypt,
                }));
            }

            let (message, bounds) = loop {
                match self.deframe_decrypted(buffer, &locator)? {
                    DeframeResult::Decrypted(decrypted, bounds) => break (decrypted, bounds),
                    DeframeResult::DecryptionFailed => continue,
                    DeframeResult::None => return Ok(None),
                }
            };

            want_close_before_decrypt = message.want_close_before_decrypt;
            let Decrypted {
                plaintext: message,
                want_close_before_decrypt: _,
            } = message;

            if self.deframer.aligned().is_none() && message.typ != ContentType::Handshake {
                // "Handshake messages MUST NOT be interleaved with other record
                // types.  That is, if a handshake message is split over two or more
                // records, there MUST NOT be any other records between them."
                // https://www.rfc-editor.org/rfc/rfc8446#section-5.1
                return Err(PeerMisbehaved::MessageInterleavedWithHandshakeMessage.into());
            }

            match message.payload.len() {
                0 => {
                    if self.seen_consecutive_empty_fragments
                        == ALLOWED_CONSECUTIVE_EMPTY_FRAGMENTS_MAX
                    {
                        return Err(PeerMisbehaved::TooManyEmptyFragments.into());
                    }
                    self.seen_consecutive_empty_fragments += 1;
                }
                _ => {
                    self.seen_consecutive_empty_fragments = 0;
                }
            };

            // do an end-run around the borrow checker, converting `message` (containing
            // a borrowed slice) to an unborrowed one (containing a `Range` into the
            // same buffer).  the reborrow happens inside the branch that returns the
            // message.
            //
            // is fixed by -Zpolonius
            // https://github.com/rust-lang/rfcs/blob/master/text/2094-nll.md#problem-case-3-conditional-control-flow-across-functions
            let unborrowed = InboundUnborrowedMessage::unborrow(&locator, message);

            if unborrowed.typ != ContentType::Handshake {
                let message = unborrowed.reborrow(&Delocator::new(buffer));
                self.deframer.discard_processed();
                return Ok(Some(Decrypted {
                    plaintext: message,
                    want_close_before_decrypt,
                }));
            }

            let message = unborrowed.reborrow(&Delocator::new(buffer));
            self.deframer
                .input_message(message.version, bounds, buffer);
            self.deframer.coalesce(buffer)?;
        }
    }

    fn deframe_decrypted<'b>(
        &mut self,
        buffer: &'b mut [u8],
        locator: &Locator,
    ) -> Result<DeframeResult<'b>, Error> {
        let (message, bounds) = match self.deframer.deframe(buffer) {
            Some(Ok(Deframed { message, bounds })) => (message, bounds),
            Some(Err(err)) => return Err(err),
            None => return Ok(DeframeResult::None),
        };

        let allowed_plaintext = match message.typ {
            // CCS messages are always plaintext.
            ContentType::ChangeCipherSpec => true,
            // Alerts are allowed to be plaintext if-and-only-if:
            // * The negotiated protocol version is TLS 1.3. - In TLS 1.2 it is unambiguous when
            //   keying changes based on the CCS message. Only TLS 1.3 requires these heuristics.
            // * We have not yet decrypted any messages from the peer - if we have we don't
            //   expect any plaintext.
            // * The payload size is indicative of a plaintext alert message.
            ContentType::Alert
                if matches!(self.negotiated_version, Some(ProtocolVersion::TLSv1_3))
                    && !self.decrypt_state.has_decrypted()
                    && message.payload.len() <= 2 =>
            {
                true
            }
            // In other circumstances, we expect all messages to be encrypted.
            _ => false,
        };

        if allowed_plaintext && !self.deframer.is_active() {
            return Ok(DeframeResult::Decrypted(
                Decrypted {
                    plaintext: message.into_plain_message(),
                    want_close_before_decrypt: false,
                },
                bounds,
            ));
        }

        match self
            .decrypt_state
            .decrypt_incoming(message)?
        {
            Some(decrypted) => {
                // After decryption, the payload is shorter
                let bounds = locator.locate(decrypted.plaintext.payload);
                Ok(DeframeResult::Decrypted(decrypted, bounds))
            }

            // failed decryption during trial decryption is not allowed to be
            // interleaved with partial handshake data.
            None if self.deframer.aligned().is_none() => {
                Err(PeerMisbehaved::RejectedEarlyDataInterleavedWithHandshakeMessage.into())
            }

            // failed decryption during trial decryption.
            None => Ok(DeframeResult::DecryptionFailed),
        }
    }

    /// Take a TLS message `msg` and map it into an `Input`
    ///
    /// `Input` is the input to our state machine.
    ///
    /// The message is mapped into `None` if it should be dropped with no further
    /// action.
    ///
    /// Otherwise the caller must present the returned `Input` to the state machine to
    /// progress the connection.
    pub(crate) fn receive_message<'a>(
        &mut self,
        msg: EncodedMessage<&'a [u8]>,
        aligned_handshake: Option<HandshakeAlignedProof>,
        send: &mut dyn SendOutput,
    ) -> Result<Option<Input<'a>>, Error> {
        // Drop CCS messages during handshake in TLS1.3
        if msg.typ == ContentType::ChangeCipherSpec && self.drop_tls13_ccs(&msg)? {
            trace!("Dropping CCS");
            return Ok(None);
        }

        // Now we can fully parse the message payload.
        let message = Message::try_from(msg)?;

        // For alerts, we have separate logic.
        if let MessagePayload::Alert(alert) = &message.payload {
            self.process_alert(alert)?;
            return Ok(None);
        }

        // For TLS1.2, outside of the handshake, send rejection alerts for
        // renegotiation requests.  These can occur any time.
        if self.reject_renegotiation_request(&message, send)? {
            return Ok(None);
        }

        Ok(Some(Input {
            message,
            aligned_handshake,
        }))
    }

    fn drop_tls13_ccs(&mut self, msg: &EncodedMessage<&'_ [u8]>) -> Result<bool, Error> {
        if self.may_receive_application_data
            || !matches!(self.negotiated_version, Some(ProtocolVersion::TLSv1_3))
        {
            return Ok(false);
        }

        if !msg.is_valid_ccs() {
            // "An implementation which receives any other change_cipher_spec value or
            //  which receives a protected change_cipher_spec record MUST abort the
            //  handshake with an "unexpected_message" alert."
            return Err(PeerMisbehaved::IllegalMiddleboxChangeCipherSpec.into());
        }

        self.temper_counters
            .received_tls13_change_cipher_spec()?;
        Ok(true)
    }

    fn reject_renegotiation_request(
        &mut self,
        msg: &Message<'_>,
        send: &mut dyn SendOutput,
    ) -> Result<bool, Error> {
        if !self.may_receive_application_data
            || matches!(self.negotiated_version, Some(ProtocolVersion::TLSv1_3))
        {
            return Ok(false);
        }

        let reject_ty = match self.side {
            Side::Client => HandshakeType::HelloRequest,
            Side::Server => HandshakeType::ClientHello,
        };

        if msg.handshake_type() != Some(reject_ty) {
            return Ok(false);
        }
        self.temper_counters
            .received_renegotiation_request()?;
        let desc = AlertDescription::NoRenegotiation;
        warn!("sending warning alert {desc:?}");
        send.send_alert(AlertLevel::Warning, desc);
        Ok(true)
    }

    fn process_alert(&mut self, alert: &AlertMessagePayload) -> Result<(), Error> {
        // Reject unknown AlertLevels.
        if AlertLevelName::try_from(alert.level).is_err() {
            return Err(PeerMisbehaved::IllegalAlertLevel(alert.level.0, alert.description).into());
        }

        // If we get a CloseNotify, make a note to declare EOF to our
        // caller.  But do not treat unauthenticated alerts like this.
        if self.may_receive_application_data && alert.description == AlertDescription::CloseNotify {
            self.has_received_close_notify = true;
            return Ok(());
        }

        // Warnings are nonfatal for TLS1.2, but outlawed in TLS1.3
        // (except, for no good reason, user_cancelled).
        let err = Error::AlertReceived(alert.description);
        if alert.level == AlertLevel::Warning {
            self.temper_counters
                .received_warning_alert()?;
            if matches!(self.negotiated_version, Some(ProtocolVersion::TLSv1_3))
                && alert.description != AlertDescription::UserCanceled
            {
                return Err(PeerMisbehaved::IllegalWarningAlert(alert.description).into());
            }

            // Some implementations send pointless `user_canceled` alerts, don't log them
            // in release mode (https://bugs.openjdk.org/browse/JDK-8323517).
            if alert.description != AlertDescription::UserCanceled || cfg!(debug_assertions) {
                warn!("TLS alert warning received: {alert:?}");
            }

            return Ok(());
        }

        Err(err)
    }
}

enum DeframeResult<'b> {
    Decrypted(Decrypted<'b>, Range<usize>),
    DecryptionFailed,
    None,
}

struct CaptureAppData<'a, 'j, 'm> {
    recv: &'a mut ReceivePath,
    other: &'a mut JoinOutput<'j>,
    /// Store a [`Locator`] initialized from the current receive buffer
    ///
    /// Allows received plaintext data to be unborrowed and stored in
    /// `received_plaintext` for in-place decryption.
    plaintext_locator: &'a Locator,
    /// Unborrowed received plaintext data
    ///
    /// Set if plaintext data was received.
    ///
    /// Plaintext data may be reborrowed using a [`Delocator`] which was
    /// initialized from the same slice as `plaintext_locator`.
    received_plaintext: &'a mut Option<UnborrowedPayload>,
    _message_lifetime: PhantomData<&'m ()>,
}

impl<'m> Output<'m> for CaptureAppData<'_, '_, 'm> {
    fn emit(&mut self, ev: Event<'_>) {
        self.other.side.emit(ev)
    }

    fn output(&mut self, ev: OutputEvent<'_>) {
        if let OutputEvent::ProtocolVersion(ver) = ev {
            self.recv.negotiated_version = Some(ver);
            self.other.send.negotiated_version(ver);
        }
        self.other.outputs.handle(ev);
    }

    fn send_msg(&mut self, m: Message<'_>, must_encrypt: bool) {
        match self.other.quic.as_deref_mut() {
            Some(quic) => quic.send_msg(m, must_encrypt),
            None => self
                .other
                .send
                .send_msg(m, must_encrypt),
        }
    }

    fn quic(&mut self) -> Option<&mut dyn QuicOutput> {
        match &mut self.other.quic {
            Some(quic) => Some(*quic),
            None => None,
        }
    }

    fn received_plaintext(&mut self, payload: Payload<'m>) {
        // Receive plaintext data [`Payload<'_>`].
        //
        // Since [`Context`] does not hold a lifetime to the receive buffer the
        // passed [`Payload`] will have it's lifetime erased by storing an index
        // into the receive buffer as an [`UnborrowedPayload`]. This enables the
        // data to be later reborrowed after it has been decrypted in-place.
        let previous = self
            .received_plaintext
            .replace(UnborrowedPayload::unborrow(self.plaintext_locator, payload));
        debug_assert!(previous.is_none(), "overwrote plaintext data");
    }

    fn start_traffic(&mut self) {
        self.recv.may_receive_application_data = true;
        self.other.send.start_traffic();
    }

    fn receive(&mut self) -> &mut ReceivePath {
        self.recv
    }

    fn send(&mut self) -> &mut dyn SendOutput {
        self.other.send
    }
}

pub(super) struct JoinOutput<'a> {
    pub(super) outputs: &'a mut dyn ConnectionOutput,
    pub(super) quic: Option<&'a mut dyn QuicOutput>,
    pub(super) send: &'a mut dyn SendOutput,
    pub(super) side: &'a mut dyn SideOutput,
}

pub(super) struct Discard;

impl ConnectionOutput for Discard {
    fn handle(&mut self, _ev: OutputEvent<'_>) {}
}

impl SideOutput for Discard {
    fn emit(&mut self, _ev: Event<'_>) {}
}

/// Tracking technically-allowed protocol actions
/// that we limit to avoid denial-of-service vectors.
struct TemperCounters {
    allowed_warning_alerts: u8,
    allowed_renegotiation_requests: u8,
    allowed_middlebox_ccs: u8,
}

impl TemperCounters {
    fn received_warning_alert(&mut self) -> Result<(), Error> {
        match self.allowed_warning_alerts {
            0 => Err(PeerMisbehaved::TooManyWarningAlertsReceived.into()),
            _ => {
                self.allowed_warning_alerts -= 1;
                Ok(())
            }
        }
    }

    fn received_renegotiation_request(&mut self) -> Result<(), Error> {
        match self.allowed_renegotiation_requests {
            0 => Err(PeerMisbehaved::TooManyRenegotiationRequests.into()),
            _ => {
                self.allowed_renegotiation_requests -= 1;
                Ok(())
            }
        }
    }

    fn received_tls13_change_cipher_spec(&mut self) -> Result<(), Error> {
        match self.allowed_middlebox_ccs {
            0 => Err(PeerMisbehaved::IllegalMiddleboxChangeCipherSpec.into()),
            _ => {
                self.allowed_middlebox_ccs -= 1;
                Ok(())
            }
        }
    }
}

impl Default for TemperCounters {
    fn default() -> Self {
        Self {
            // cf. BoringSSL `kMaxWarningAlerts`
            // <https://github.com/google/boringssl/blob/dec5989b793c56ad4dd32173bd2d8595ca78b398/ssl/tls_record.cc#L137-L139>
            allowed_warning_alerts: 4,

            // we rebuff renegotiation requests with a `NoRenegotiation` warning alerts.
            // a second request after this is fatal.
            allowed_renegotiation_requests: 1,

            // At most two CCS are allowed: one after each ClientHello (recall a second
            // ClientHello happens after a HelloRetryRequest).
            //
            // note BoringSSL allows up to 32.
            allowed_middlebox_ccs: 2,
        }
    }
}

pub(crate) struct TrafficTemperCounters {
    allowed_consecutive_handshake_messages: u8,
}

impl TrafficTemperCounters {
    pub(crate) fn received_handshake_message(&mut self) -> Result<(), Error> {
        match self.allowed_consecutive_handshake_messages {
            0 => Err(PeerMisbehaved::TooManyConsecutiveHandshakeMessagesAfterHandshake.into()),
            _ => {
                self.allowed_consecutive_handshake_messages -= 1;
                Ok(())
            }
        }
    }

    pub(crate) fn received_app_data(&mut self) {
        self.allowed_consecutive_handshake_messages = Self::MAX_CONSECUTIVE_HANDSHAKE_MESSAGES;
    }

    // cf. BoringSSL `kMaxKeyUpdates`
    // <https://github.com/google/boringssl/blob/dec5989b793c56ad4dd32173bd2d8595ca78b398/ssl/tls13_both.cc#L35-L38>
    const MAX_CONSECUTIVE_HANDSHAKE_MESSAGES: u8 = 32;
}

impl Default for TrafficTemperCounters {
    fn default() -> Self {
        Self {
            allowed_consecutive_handshake_messages: Self::MAX_CONSECUTIVE_HANDSHAKE_MESSAGES,
        }
    }
}

pub(crate) struct Input<'a> {
    pub(crate) message: Message<'a>,
    pub(crate) aligned_handshake: Option<HandshakeAlignedProof>,
}

impl Input<'_> {
    // Changing the keys must not span any fragmented handshake
    // messages.  Otherwise the defragmented messages will have
    // been protected with two different record layer protections,
    // which is illegal.  Not mentioned in RFC.
    pub(crate) fn check_aligned_handshake(&self) -> Result<HandshakeAlignedProof, Error> {
        self.aligned_handshake
            .ok_or_else(|| PeerMisbehaved::KeyEpochWithPendingFragment.into())
    }
}

/// An [`EncodedMessage<Payload<'_>>`] which does not borrow its payload, but
/// references a range that can later be borrowed.
struct InboundUnborrowedMessage {
    typ: ContentType,
    version: ProtocolVersion,
    bounds: Range<usize>,
}

impl InboundUnborrowedMessage {
    fn unborrow(locator: &Locator, msg: EncodedMessage<&'_ [u8]>) -> Self {
        Self {
            typ: msg.typ,
            version: msg.version,
            bounds: locator.locate(msg.payload),
        }
    }

    fn reborrow<'b>(self, delocator: &Delocator<'b>) -> EncodedMessage<&'b [u8]> {
        EncodedMessage {
            typ: self.typ,
            version: self.version,
            payload: delocator.slice_from_range(&self.bounds),
        }
    }
}

/// A buffer of TLS bytes read from a socket, stored in a `Vec<u8>`.
#[derive(Default, Debug)]
pub struct VecInput {
    /// Buffer of data read from the socket, in the process of being parsed into messages.
    ///
    /// For buffer size management, checkout out the [`VecInput::prepare_read()`] method.
    buf: Vec<u8>,

    /// What size prefix of `buf` is used.
    used: usize,

    /// Whether we've seen a 0-byte read.
    has_seen_eof: bool,

    /// Whether a CloseNotify alert has been seen.
    received_close_notify: bool,
}

impl VecInput {
    /// Discard `taken` bytes from the start of our buffer.
    pub(crate) fn discard(&mut self, taken: usize) {
        if taken < self.used {
            /* Before:
             * +----------+----------+----------+
             * | taken    | pending  |xxxxxxxxxx|
             * +----------+----------+----------+
             * 0          ^ taken    ^ self.used
             *
             * After:
             * +----------+----------+----------+
             * | pending  |xxxxxxxxxxxxxxxxxxxxx|
             * +----------+----------+----------+
             * 0          ^ self.used
             */

            self.buf
                .copy_within(taken..self.used, 0);
            self.used -= taken;
        } else if taken >= self.used {
            self.used = 0;
        }
    }

    pub(crate) fn filled_mut(&mut self) -> &mut [u8] {
        &mut self.buf[..self.used]
    }

    /// Read some bytes from `rd`, and add them to the buffer.
    pub fn read(&mut self, rd: &mut dyn Read) -> io::Result<usize> {
        if self.received_close_notify {
            return Ok(0);
        } else if let Err(err) = self.prepare_read() {
            return Err(io::Error::new(io::ErrorKind::InvalidData, err));
        }

        // Try to do the largest reads possible. Note that if
        // we get a message with a length field out of range here,
        // we do a zero length read.  That looks like an EOF to
        // the next layer up, which is fine.
        let new_bytes = rd.read(&mut self.buf[self.used..])?;
        if new_bytes == 0 {
            self.has_seen_eof = true;
        }

        self.used += new_bytes;
        Ok(new_bytes)
    }

    /// Resize the internal `buf` if necessary for reading more bytes.
    fn prepare_read(&mut self) -> Result<(), &'static str> {
        /// TLS allows for handshake messages of up to 16MB.  We
        /// restrict that to 64KB to limit potential for denial-of-
        /// service.
        const MAX_HANDSHAKE_SIZE: usize = 0xffff;

        const READ_SIZE: usize = 4096;

        // We allow a maximum of 64k of buffered data. Given that the first read of such a
        // payload will only ever be 4k bytes, the next time we come around here we allow a
        // larger buffer size. Once the large message and any following handshake messages in
        // the same flight have been consumed, `pop()` will call `discard()` to reset `used`.
        // At this point, the buffer resizing logic below should reduce the buffer size.
        if self.used >= MAX_HANDSHAKE_SIZE {
            return Err("message buffer full");
        }

        // If we can and need to increase the buffer size to allow a 4k read, do so. After
        // dealing with a large handshake message (exceeding `MAX_HANDSHAKE_SIZE`),
        // make sure to reduce the buffer size again (large messages should be rare).
        // Also, reduce the buffer size if there are neither full nor partial messages in it,
        // which usually means that the other side suspended sending data.
        let need_capacity = Ord::min(MAX_HANDSHAKE_SIZE, self.used + READ_SIZE);
        if need_capacity > self.buf.len() {
            self.buf.resize(need_capacity, 0);
        } else if self.used == 0 || self.buf.len() > MAX_HANDSHAKE_SIZE {
            self.buf.resize(need_capacity, 0);
            self.buf.shrink_to(need_capacity);
        }

        Ok(())
    }
}

impl TlsInputBuffer for VecInput {
    fn slice_mut(&mut self) -> &mut [u8] {
        self.filled_mut()
    }

    fn discard(&mut self, num_bytes: usize) {
        self.discard(num_bytes)
    }

    fn received_close_notify(&mut self) {
        self.received_close_notify = true;
    }

    fn has_seen_eof(&self) -> bool {
        self.has_seen_eof
    }
}

/// A borrowed version of [`VecInput`] that tracks discard operations
#[derive(Debug)]
pub struct SliceInput<'a> {
    // a fully initialized buffer that will be deframed
    buf: &'a mut [u8],
    // number of bytes to discard from the front of `buf` at a later time
    discard: usize,
    /// Whether we've seen a 0-byte read.
    has_seen_eof: bool,
    /// Whether a CloseNotify alert has been seen.
    received_close_notify: bool,
}

impl<'a> SliceInput<'a> {
    /// Create a new [`SliceInput`] from a mutable slice of bytes.
    pub fn new(buf: &'a mut [u8]) -> Self {
        Self {
            buf,
            discard: 0,
            has_seen_eof: false,
            received_close_notify: false,
        }
    }

    /// Returns how many bytes were consumed at the start of the original buffer.
    pub fn into_used(self) -> usize {
        self.discard
    }
}

impl TlsInputBuffer for SliceInput<'_> {
    fn slice_mut(&mut self) -> &mut [u8] {
        &mut self.buf[self.discard..]
    }

    fn discard(&mut self, num_bytes: usize) {
        self.discard += num_bytes;
    }

    fn received_close_notify(&mut self) {
        self.received_close_notify = true;
    }

    fn has_seen_eof(&self) -> bool {
        self.has_seen_eof
    }
}

/// An abstraction over received data buffers (either owned or borrowed)
pub trait TlsInputBuffer {
    /// Return the buffer which contains the received data.
    ///
    /// If no data is available, return the empty slice.
    ///
    /// This is mutable, because the buffer is used for in-place decryption
    /// and coalescing of TLS records.  Coalescing of TLS records can happen
    /// incrementally over multiple calls into rustls.  As a result the
    /// contents of this buffer must not be altered except to add new bytes
    /// at the end.
    fn slice_mut(&mut self) -> &mut [u8];

    /// Discard `num_bytes` from the front of the buffer returned by `slice_mut()`.
    ///
    /// Multiple calls to `discard()` are cumulative, rather than "last wins".  In
    /// other words, `discard(1)` followed by `discard(1)` gives the same result
    /// as `discard(2)`.
    ///
    /// The next call to `slice_mut()` must reflect all previous `discard()`s. In
    /// other words, if `slice_mut()` returns slice `[p..q]`, it should then
    /// return `[p+n..q]` after `discard(n)`.
    ///
    /// Rustls guarantees it will not `discard()` more bytes than are returned
    /// from `slice_mut()`.
    fn discard(&mut self, num_bytes: usize);

    /// Signal that the connection has received a TLS `close_notify` alert.
    ///
    /// The buffer should not accept any more data, because the peer has closed the connection.
    fn received_close_notify(&mut self);

    /// Whether the buffer has seen a TCP EOF.
    ///
    /// This is not a TCP-level event, but it is signalled to the TLS state via the input buffer.
    fn has_seen_eof(&self) -> bool;
}

/// cf. BoringSSL's `kMaxEmptyRecords`
/// <https://github.com/google/boringssl/blob/dec5989b793c56ad4dd32173bd2d8595ca78b398/ssl/tls_record.cc#L124-L128>
const ALLOWED_CONSECUTIVE_EMPTY_FRAGMENTS_MAX: u8 = 32;