rustls/conn/

receive.rs

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

use super::private::SideOutput;
use super::send::{SendOutput, SendPath};
use super::split::SendAdapter;
use super::{Connection, SEQ_SOFT_LIMIT, SideData, StateMachine, Transport};
use crate::common_state::{
    ConnectionOutput, Event, Output, OutputEvent, Side, UnborrowedPayload, maybe_send_fatal_alert,
};
use crate::crypto::cipher::{EncodableVersion, InboundOpaque, Payload, Record, RecordDecrypter};
use crate::enums::{ContentType, HandshakeType, ProtocolVersion};
use crate::error::{AlertDescription, Error, PeerMisbehaved};
use crate::msgs::{
    AlertLevel, AlertLevelName, AlertMessagePayload, Deframed, Deframer, Delocator,
    HandshakeAlignedProof, Locator, Message, MessagePayload,
};
use crate::quic::QuicOutput;
use crate::tracing::{trace, warn};

pub(crate) struct MessageIter<'a, 'm, Side: SideData, Send: SendOutput + 'a> {
    pub(super) input: &'m mut dyn TlsInputBuffer,
    pub(super) tls: &'a mut Vec<u8>,
    pub(super) recv: &'a mut ReceivePath,
    pub(super) state: &'a mut Result<Side::State, Error>,
    pub(super) output: JoinOutput<'a, Send>,
    pub(super) mode: MessageIterMode,
}

impl<'a, 'm, Side: SideData> MessageIter<'a, 'm, Side, SendPath> {
    pub(crate) fn new<T: Transport>(
        input: &'m mut dyn TlsInputBuffer,
        tls: &'a mut Vec<u8>,
        conn: &'a mut Connection<Side, T>,
        mode: MessageIterMode,
    ) -> Self {
        Self {
            input,
            tls,
            recv: &mut conn.common.recv,
            state: &mut conn.state,
            output: JoinOutput {
                outputs: &mut conn.common.outputs,
                quic: conn.transport.quic(),
                send: &mut conn.common.send,
                side: &mut conn.side,
            },
            mode,
        }
    }
}

impl<'a, 'm, 's, Side: SideData> MessageIter<'a, 'm, Side, SendAdapter<'s>> {
    pub(super) fn receive(
        input: &'m mut dyn TlsInputBuffer,
        tls: &'a mut Vec<u8>,
        state: &'a mut Result<Side::State, Error>,
        recv: &'a mut ReceivePath,
        output: JoinOutput<'a, SendAdapter<'s>>,
        mode: MessageIterMode,
    ) -> Self {
        Self {
            input,
            tls,
            recv,
            state,
            output,
            mode,
        }
    }
}

impl<'a, 'm, Side: SideData, Send: SendOutput + 'a> MessageIter<'a, 'm, Side, Send> {
    pub(crate) fn next(&mut self, early_only: bool) -> 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,
                tls: self.tls,
                other: &mut self.output,
                plaintext_locator: &locator,
                received_plaintext: &mut plaintext,
                _message_lifetime: PhantomData,
            };

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

            let Some(record) = opt_record else {
                break;
            };

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

            if want_close_before_decrypt {
                let _ = output.other.send.send_alert(
                    AlertLevel::Warning,
                    AlertDescription::CloseNotify,
                    output.tls,
                );
            } else if record.payload.is_empty()
                && matches!(record.typ, ContentType::Handshake | ContentType::Alert)
            {
                // <https://datatracker.ietf.org/doc/html/rfc9846#section-5.4>
                let _ = output.other.send.send_alert(
                    AlertLevel::Fatal,
                    AlertDescription::UnexpectedMessage,
                    output.tls,
                );

                let error = Error::from(PeerMisbehaved::EmptyFragment);
                *self.state = Err(error.clone());
                return Some(Err(error));
            }

            let hs_aligned = output.recv.deframer.aligned();
            let result =
                match output
                    .recv
                    .receive_record(record, hs_aligned, output.tls, 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) => {
                    let _ = maybe_send_fatal_alert(output.other.send, &e, output.tls);
                    *self.state = Err(e.clone());
                    return Some(Err(e));
                }
            }

            if matches!(self.mode, MessageIterMode::All) && !st.wants_input() {
                st = match st.handle_without_input(&mut output) {
                    Ok(st) => st,
                    Err(err) => {
                        let _ = maybe_send_fatal_alert(output.other.send, &err, output.tls);
                        *self.state = Err(err.clone());
                        return Some(Err(err));
                    }
                };
            }

            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/rfc9846#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;
            }

            match plaintext.take() {
                Some(DataKind::Traffic(payload)) if !early_only => {
                    *self.state = Ok(st);
                    return Some(Ok(payload));
                }
                Some(DataKind::Early(payload)) if early_only => {
                    *self.state = Ok(st);
                    return Some(Ok(payload));
                }
                _ => {}
            }

            // Stop before consuming traffic data if the caller only wants early data.
            if (early_only || matches!(self.mode, MessageIterMode::Handshake)) && st.is_traffic() {
                break;
            }
        }

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

pub(crate) enum MessageIterMode {
    /// Iterate through all available messages.
    All,

    /// Handshake mode.
    ///
    /// Stop iteration when when external input is needed to progress handshake, or the handshake ends.
    Handshake,
}

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.  That includes other record types too.
    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 potentially coalesced record out of the deframer.
    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 records from `deframer`
            if let Some(span) = self.deframer.complete_span() {
                let plaintext = self.deframer.record(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 (record, 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 = record.want_close_before_decrypt;
            let Decrypted {
                plaintext: record,
                want_close_before_decrypt: _,
            } = record;

            if self.deframer.aligned().is_none() && record.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/rfc9846#section-5.1
                return Err(PeerMisbehaved::MessageInterleavedWithHandshakeMessage.into());
            }

            match (record.payload.len(), record.typ) {
                (0, _) => {
                    if self.seen_consecutive_empty_fragments
                        == ALLOWED_CONSECUTIVE_EMPTY_FRAGMENTS_MAX
                    {
                        return Err(PeerMisbehaved::TooManyEmptyFragments.into());
                    }
                    self.seen_consecutive_empty_fragments += 1;
                }
                (_, ContentType::Handshake | ContentType::ApplicationData) => {
                    self.seen_consecutive_empty_fragments = 0;
                }
                (_, _) => {}
            };

            // do an end-run around the borrow checker, converting `record` (containing
            // a borrowed slice) to an unborrowed one (containing a `Range` into the
            // same buffer).  the reborrow happens inside the branch that returns the
            // record.
            //
            // 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 = InboundUnborrowedRecord::unborrow(&locator, record);

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

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

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

        let allowed_plaintext = match record.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 records 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()
                    && record.payload.len() <= 2 =>
            {
                true
            }
            // In other circumstances, we expect all records to be encrypted.
            _ => false,
        };

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

        match self
            .decrypt_state
            .decrypt_incoming(record)?
        {
            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 record and map it into an [`Input`]
    ///
    /// `Input` is the input to our state machine.
    ///
    /// The record 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_record<'a>(
        &mut self,
        record: Record<&'a [u8]>,
        aligned_handshake: Option<HandshakeAlignedProof>,
        tls: &mut Vec<u8>,
        send: &mut dyn SendOutput,
    ) -> Result<Option<Input<'a>>, Error> {
        // Drop CCS messages during handshake in TLS1.3
        if record.typ == ContentType::ChangeCipherSpec && self.drop_tls13_ccs(&record)? {
            trace!("Dropping CCS");
            return Ok(None);
        }

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

        // 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, tls, send)? {
            return Ok(None);
        }

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

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

        if !record.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<'_>,
        tls: &mut Vec<u8>,
        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, tls)?;
        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,
}

/// Record layer that tracks decryption keys.
pub(crate) struct DecryptionState {
    record_decrypter: Option<Box<dyn RecordDecrypter>>,
    read_seq: u64,
    has_decrypted: bool,

    // Records encrypted with other keys may be encountered, so failures
    // should be swallowed by the caller.  This struct tracks the amount
    // of record size this is allowed for.
    trial_decryption_len: Option<usize>,
}

impl DecryptionState {
    /// Create new record layer with no keys.
    pub(crate) fn new() -> Self {
        Self {
            record_decrypter: None,
            read_seq: 0,
            has_decrypted: false,
            trial_decryption_len: None,
        }
    }

    /// Decrypt a TLS record.
    ///
    /// `encr` is a decoded record allegedly received from the peer.
    /// If it can be decrypted, its decryption is returned.  Otherwise,
    /// an error is returned.
    pub(crate) fn decrypt_incoming<'a>(
        &mut self,
        encr: Record<InboundOpaque<'a>>,
    ) -> Result<Option<Decrypted<'a>>, Error> {
        let Some(decrypter) = &mut self.record_decrypter else {
            return Ok(Some(Decrypted {
                want_close_before_decrypt: false,
                plaintext: encr.into_plain_record(),
            }));
        };

        // Set to `true` if the peer appears to getting close to encrypting
        // too many records with this key.
        //
        // Perhaps if we send an alert well before their counter wraps, a
        // buggy peer won't make a terrible mistake here?
        //
        // Note that there's no reason to refuse to decrypt: the security
        // failure has already happened.
        let want_close_before_decrypt = self.read_seq == SEQ_SOFT_LIMIT;

        let encrypted_len = encr.payload.len();
        match decrypter.decrypt(encr, self.read_seq) {
            Ok(plaintext) => {
                self.read_seq += 1;
                if !self.has_decrypted {
                    self.has_decrypted = true;
                }
                Ok(Some(Decrypted {
                    want_close_before_decrypt,
                    plaintext,
                }))
            }
            Err(Error::DecryptError) if self.doing_trial_decryption(encrypted_len) => {
                trace!("Dropping undecryptable record after aborted early_data");
                Ok(None)
            }
            Err(err) => Err(err),
        }
    }

    /// Set and start using the given `RecordDecrypter` for future incoming
    /// record decryption.
    pub(crate) fn set_record_decrypter(
        &mut self,
        cipher: Box<dyn RecordDecrypter>,
        _proof: &HandshakeAlignedProof,
    ) {
        self.record_decrypter = Some(cipher);
        self.read_seq = 0;
        self.trial_decryption_len = None;
    }

    /// Set and start using the given `RecordDecrypter` for future incoming
    /// record decryption, and enable "trial decryption" mode for when TLS1.3
    /// 0-RTT is attempted but rejected by the server.
    pub(crate) fn set_record_decrypter_with_trial_decryption(
        &mut self,
        cipher: Box<dyn RecordDecrypter>,
        max_length: usize,
        _proof: &HandshakeAlignedProof,
    ) {
        self.record_decrypter = Some(cipher);
        self.read_seq = 0;
        self.trial_decryption_len = Some(max_length);
    }

    pub(crate) fn finish_trial_decryption(&mut self) {
        self.trial_decryption_len = None;
    }

    /// Return true if we have ever decrypted a record. This is used in place
    /// of checking the read_seq since that will be reset on key updates.
    pub(crate) fn has_decrypted(&self) -> bool {
        self.has_decrypted
    }

    pub(crate) fn read_seq(&self) -> u64 {
        self.read_seq
    }

    fn doing_trial_decryption(&mut self, requested: usize) -> bool {
        match self
            .trial_decryption_len
            .and_then(|value| value.checked_sub(requested))
        {
            Some(remaining) => {
                self.trial_decryption_len = Some(remaining);
                true
            }
            _ => false,
        }
    }
}

/// Result of decryption.
pub(crate) struct Decrypted<'a> {
    /// Whether the peer appears to be getting close to encrypting too many records with this key.
    pub(crate) want_close_before_decrypt: bool,
    /// The decrypted record.
    pub(crate) plaintext: Record<&'a [u8]>,
}

struct CaptureAppData<'a, 'j, 'm, Send: SendOutput + 'a> {
    recv: &'a mut ReceivePath,
    other: &'a mut JoinOutput<'j, Send>,
    tls: &'a mut Vec<u8>,
    /// 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<DataKind<UnborrowedPayload>>,
    _message_lifetime: PhantomData<&'m ()>,
}

impl<'a, 'm, Send: SendOutput + 'a> Output<'m> for CaptureAppData<'a, '_, 'm, Send> {
    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) -> Result<(), Error> {
        match self.other.quic.as_deref_mut() {
            Some(quic) => {
                quic.send_msg(m, must_encrypt);
                Ok(())
            }
            None => self
                .other
                .send
                .send_msg(m, must_encrypt, self.tls),
        }
    }

    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: DataKind<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(match payload {
                DataKind::Early(p) => {
                    DataKind::Early(UnborrowedPayload::unborrow(self.plaintext_locator, p))
                }
                DataKind::Traffic(p) => {
                    DataKind::Traffic(UnborrowedPayload::unborrow(self.plaintext_locator, p))
                }
            });
        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(crate) enum DataKind<T> {
    Early(T),
    Traffic(T),
}

pub(super) struct JoinOutput<'a, Send: SendOutput + 'a> {
    pub(super) outputs: &'a mut dyn ConnectionOutput,
    pub(super) quic: Option<&'a mut dyn QuicOutput>,
    pub(super) send: &'a mut Send,
    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 [`Record<Payload<'_>>`] which does not borrow its payload, but
/// references a range that can later be borrowed.
struct InboundUnborrowedRecord {
    typ: ContentType,
    version: EncodableVersion,
    bounds: Range<usize>,
}

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

    fn reborrow<'b>(self, delocator: &Delocator<'b>) -> Record<&'b [u8]> {
        Record {
            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)]
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.
    ///
    /// Once the buffer contains 64 kB of data, we will not read any more bytes until some
    /// are consumed by reading from the buffer via
    /// [`Connection::read_tls()`][super::Connection::read_tls()].
    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 record 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
    }
}

impl fmt::Debug for VecInput {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let Self {
            buf,
            used,
            has_seen_eof,
            received_close_notify,
        } = self;
        f.debug_struct("VecInput")
            .field("buf_len", &buf.len())
            .field("used", used)
            .field("has_seen_eof", has_seen_eof)
            .field("received_close_notify", received_close_notify)
            .finish_non_exhaustive()
    }
}

/// A borrowed version of [`VecInput`] that tracks discard operations
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
    }
}

impl fmt::Debug for SliceInput<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let Self {
            buf,
            discard,
            has_seen_eof,
            received_close_notify,
        } = self;
        f.debug_struct("SliceInput")
            .field("buf_len", &buf.len())
            .field("discard", discard)
            .field("has_seen_eof", has_seen_eof)
            .field("received_close_notify", received_close_notify)
            .finish_non_exhaustive()
    }
}

/// 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;

#[cfg(test)]
mod tests {
    use alloc::format;

    use super::*;

    #[test]
    fn test_has_decrypted() {
        struct PassThroughDecrypter;
        impl RecordDecrypter for PassThroughDecrypter {
            fn decrypt<'a>(
                &mut self,
                record: Record<InboundOpaque<'a>>,
                _: u64,
            ) -> Result<Record<&'a [u8]>, Error> {
                Ok(record.into_plain_record())
            }
        }

        // A record layer starts out invalid, having never decrypted.
        let mut record_layer = DecryptionState::new();
        assert!(record_layer.record_decrypter.is_none());
        assert_eq!(record_layer.read_seq, 0);
        assert!(!record_layer.has_decrypted());

        // Initializing the record layer should update the decrypt state, but shouldn't affect whether it
        // has decrypted.
        let deframer = Deframer::default();
        record_layer
            .set_record_decrypter(Box::new(PassThroughDecrypter), &deframer.aligned().unwrap());
        assert!(record_layer.record_decrypter.is_some());
        assert_eq!(record_layer.read_seq, 0);
        assert!(!record_layer.has_decrypted());

        // Decrypting a record should update the read_seq and track that we have now performed
        // a decryption.
        record_layer
            .decrypt_incoming(Record::new(
                ContentType::Handshake,
                EncodableVersion::Legacy(ProtocolVersion::TLSv1_2),
                InboundOpaque(&mut [0xC0, 0xFF, 0xEE]),
            ))
            .unwrap();
        assert_eq!(record_layer.read_seq, 1);
        assert!(record_layer.has_decrypted());

        // Resetting the record layer decrypter (as if a key update occurred) should reset
        // the read_seq number, but not our knowledge of whether we have decrypted previously.
        record_layer
            .set_record_decrypter(Box::new(PassThroughDecrypter), &deframer.aligned().unwrap());
        assert_eq!(record_layer.read_seq, 0);
        assert!(record_layer.has_decrypted());
    }

    #[test]
    fn debug_of_input_types() {
        let mut buffer = [b'a'; 32];
        let mut slice = SliceInput::new(&mut buffer);
        slice.discard(2);
        assert_eq!(
            format!("{slice:?}"),
            "SliceInput { buf_len: 32, discard: 2, has_seen_eof: false, received_close_notify: false, .. }"
        );

        let mut vec = VecInput::default();
        vec.prepare_read().unwrap();
        assert_eq!(
            format!("{vec:?}"),
            "VecInput { buf_len: 4096, used: 0, has_seen_eof: false, received_close_notify: false, .. }"
        );
    }
}