iso15118 0.5.0

//! Framing a byte stream into V2G messages, and back.
//!
//! Between the TCP (or TLS) stream and the message codecs sits a job that has
//! nothing to do with charging and everything to do with not getting it wrong:
//! V2GTP frames arrive split across reads and coalesced with their neighbours,
//! and the length field that says where one ends is attacker-controlled and
//! arrives before anything is authenticated.
//!
//! [`Connection`] does that and only that. It owns no session state, no timers
//! and no policy — it is the piece an EVCC and an SECC share exactly.

use alloc::collections::VecDeque;
use alloc::vec::Vec;
use core::fmt;

use crate::message::{Message, MessageError};
use crate::trace::trace_event;
use crate::v2gtp::{self, PayloadType, V2gtpError};
use crate::{MAX_EXI_PAYLOAD_LEN, Protocol};

/// How many complete-but-undrained V2GTP frames a [`Connection`] will hold.
///
/// V2G is strictly half-duplex: neither side sends again until the other has
/// answered, so one frame in hand is the normal case and a queue this deep is
/// already generous. The bound exists because the *raw* input buffer being
/// bounded by one frame only moves the problem — a peer that keeps sending
/// whole small frames without waiting would otherwise grow the decoded queue
/// without limit.
pub const MAX_PENDING_FRAMES: usize = 4;

/// Reassembles V2GTP frames from a byte stream and serialises messages back
/// into it.
///
/// ```
/// use iso15118::Protocol;
/// use iso15118::app_protocol::SupportedAppProtocolReq;
/// use iso15118::message::Message;
/// use iso15118::session::Connection;
///
/// let mut evcc = Connection::new();
/// let req = SupportedAppProtocolReq::advertising([Protocol::Iso20]);
/// evcc.send(&Message::AppProtocolReq(Box::new(req)))?;
/// let wire = evcc.take_transmit();
///
/// // The charger reads the stream one byte at a time and still sees one frame.
/// let mut secc = Connection::new();
/// for byte in &wire {
///     secc.receive(&[*byte])?;
/// }
/// assert_eq!(secc.next_message()?.map(|m| m.name()), Some("supportedAppProtocolReq"));
/// assert!(secc.next_message()?.is_none());
/// # Ok::<_, iso15118::session::ConnectionError>(())
/// ```
#[derive(Debug)]
pub struct Connection {
    protocol: Option<Protocol>,
    max_payload_len: usize,
    /// Bytes that are not yet a whole frame. Bounded by one frame's worth,
    /// because everything complete has already moved to `frames`.
    rx: Vec<u8>,
    /// Complete frames, framed but not yet decoded.
    ///
    /// Splitting framing from decoding is what lets the input buffer be bounded
    /// exactly — but it moves the bound rather than removing it, so this queue
    /// has a ceiling of its own. See [`Connection::with_limits`].
    frames: VecDeque<(PayloadType, Vec<u8>)>,
    max_pending_frames: usize,
    /// Frames skipped for a payload type this session does not process.
    /// \[V2G2-800\]
    ignored: u32,
    tx: VecDeque<u8>,
}

impl Default for Connection {
    fn default() -> Self {
        Self::new()
    }
}

impl Connection {
    /// A connection that has not yet negotiated a protocol.
    #[must_use]
    pub fn new() -> Self {
        Self::with_limit(MAX_EXI_PAYLOAD_LEN)
    }

    /// The same, with an explicit ceiling on one frame's payload.
    ///
    /// An embedded EVCC with 64 KiB of RAM should set this to what it can
    /// actually hold, not to the crate default: the limit is what stops a peer
    /// from making the receiver buffer more than it has.
    #[must_use]
    pub fn with_limit(max_payload_len: usize) -> Self {
        Self::with_limits(max_payload_len, MAX_PENDING_FRAMES)
    }

    /// The same, also bounding how many decoded-but-undrained frames may queue
    /// up.
    ///
    /// The two limits answer different questions. `max_payload_len` bounds one
    /// frame; `max_pending_frames` bounds how many whole frames a peer may push
    /// ahead of the reader. V2G is strictly half-duplex — neither side may send
    /// again before the other has answered — so a peer that pipelines is
    /// already outside the protocol, and without this second bound it could
    /// make the receiver hold an unbounded number of frames it will never be
    /// asked for.
    #[must_use]
    pub fn with_limits(max_payload_len: usize, max_pending_frames: usize) -> Self {
        Self {
            protocol: None,
            max_payload_len,
            rx: Vec::new(),
            frames: VecDeque::new(),
            max_pending_frames: max_pending_frames.max(1),
            ignored: 0,
            tx: VecDeque::new(),
        }
    }

    /// The protocol the session agreed on, once it has.
    #[must_use]
    pub const fn protocol(&self) -> Option<Protocol> {
        self.protocol
    }

    /// Records the outcome of the `supportedAppProtocol` handshake.
    ///
    /// Until this is called, payload type `0x8001` is read as the handshake;
    /// afterwards it is read as that protocol's message set. Nothing else can
    /// tell the two apart.
    pub const fn set_protocol(&mut self, protocol: Protocol) {
        self.protocol = Some(protocol);
    }

    /// Feeds bytes read from the wire. Any number, at any boundary.
    ///
    /// Whole frames are split off immediately and queued; only the trailing
    /// partial frame stays buffered as bytes. Call
    /// [`Connection::next_message`] afterwards to drain what became complete.
    ///
    /// Framing errors surface here rather than at decode time, because a stream
    /// that is not V2GTP cannot be resynchronised — there is no frame delimiter
    /// to scan for.
    pub fn receive(&mut self, data: &[u8]) -> Result<(), ConnectionError> {
        self.rx.extend_from_slice(data);
        loop {
            match v2gtp::split_frame(&self.rx, self.max_payload_len) {
                Ok((header, payload, rest)) => {
                    if self.frames.len() >= self.max_pending_frames {
                        // The peer is pushing frames faster than the protocol
                        // lets it. Nothing after this point can be answered in
                        // order, so the stream is finished.
                        self.rx.clear();
                        self.frames.clear();
                        return Err(ConnectionError::TooManyFrames {
                            limit: self.max_pending_frames,
                        });
                    }
                    let consumed = self.rx.len() - rest.len();
                    self.frames.push_back((header.payload_type, payload.to_vec()));
                    self.rx.drain(..consumed);
                }
                Err(V2gtpError::Incomplete) => break,
                Err(e) => {
                    // Nothing after a malformed header can be trusted; drop it
                    // rather than leave bytes that would be re-parsed as a
                    // frame boundary the peer never wrote.
                    self.rx.clear();
                    return Err(ConnectionError::Framing(e));
                }
            }
        }
        // What is left is a partial frame whose header already passed the
        // length limit, so the buffer is bounded by one frame's worth by
        // construction. The check is here anyway: it is one comparison, and it
        // turns a future mistake in the loop above into an error rather than
        // into unbounded growth on an unauthenticated peer's say-so.
        let ceiling = self.max_payload_len.saturating_add(v2gtp::HEADER_LEN);
        if self.rx.len() > ceiling {
            self.rx.clear();
            return Err(ConnectionError::Overflow { limit: ceiling });
        }
        Ok(())
    }

    /// Takes the next complete message, if one has arrived.
    ///
    /// `Ok(None)` means "not yet, read more". An error means a frame this
    /// session *should* have been able to decode did not decode, and it is
    /// terminal: everything still buffered is dropped with it, because the peer
    /// that produced the fault also chose where the frames behind it began.
    ///
    /// A frame whose **payload type** is not one this session processes is not
    /// an error: \[V2G2-800\] has a V2GTP entity *ignore* such a message, and
    /// ignoring is possible precisely because the header was well-formed — the
    /// sync pattern matched and the length says where the next frame starts. So
    /// those are skipped and counted ([`Connection::ignored_frames`]), and the
    /// search continues with the frame behind them.
    ///
    /// That covers three things a conforming peer may legitimately put on the
    /// wire: a manufacturer-specific extension (`0xA000..=0xFFFF`, which
    /// Table 10 reserves for exactly that), a payload type from a generation
    /// this session did not negotiate, and one from a schema set this build
    /// does not have. Ending a charging session over any of them would be a
    /// station that cannot be extended and a vehicle that cannot talk to one
    /// that has been.
    ///
    /// The security property is unchanged, because ignoring is what it wanted:
    /// a peer still cannot get a -20 message processed in a session that
    /// negotiated -2. It simply does not also get to end the session by trying.
    pub fn next_message(&mut self) -> Result<Option<Message>, ConnectionError> {
        while let Some((payload_type, payload)) = self.next_frame() {
            match Message::decode(self.protocol, payload_type, &payload) {
                Ok(message) => return Ok(Some(message)),
                Err(MessageError::NotAMessage(_) | MessageError::NotForThisSession { .. }) => {
                    // Bounded: the queue this drains is itself bounded by
                    // `max_pending_frames`, and an ignored frame never advances
                    // the session, so the sequence timer still closes a peer
                    // that sends nothing else.
                    self.ignored = self.ignored.saturating_add(1);
                    trace_event!(payload_type = payload_type.as_u16(), "frame ignored");
                }
                // A frame whose type this session *does* process and whose body
                // will not decode is a different thing: the peer and this side
                // disagree about a grammar they have both agreed to use. There
                // is no reading of the bytes left to try, so what is behind it —
                // framed by the same peer, at boundaries the same peer chose —
                // goes with it rather than waiting to be decoded on the next
                // call as though nothing had happened.
                Err(e) => {
                    self.reset_input();
                    return Err(e.into());
                }
            }
        }
        Ok(None)
    }

    /// How many frames have been ignored for their payload type.
    ///
    /// Worth surfacing rather than swallowing: a session that ignores a lot is
    /// talking to a peer that believes in an extension this build has never
    /// heard of, and that is a fact about a deployment rather than about a
    /// message.
    #[must_use]
    pub const fn ignored_frames(&self) -> u32 {
        self.ignored
    }

    /// Takes the next complete frame without decoding it.
    ///
    /// The same reassembly and the same limits as
    /// [`Connection::next_message`], stopping one layer short: a payload type
    /// and the bytes under it. `None` means "not yet, read more".
    ///
    /// This is the seam for a message set this crate does not have. The
    /// reassembly, the attacker-controlled length field and both ceilings are
    /// identical for every generation, so a consumer with its own DIN SPEC
    /// 70121 codec can drive this and supply only the message set.
    ///
    /// ```
    /// use iso15118::session::Connection;
    /// use iso15118::v2gtp::PayloadType;
    ///
    /// let mut conn = Connection::new();
    /// conn.send_frame(PayloadType::ExiEncodedV2gMessage, b"not ours to decode")?;
    /// let wire = conn.take_transmit();
    ///
    /// let mut peer = Connection::new();
    /// peer.receive(&wire)?;
    /// assert_eq!(
    ///     peer.next_frame(),
    ///     Some((PayloadType::ExiEncodedV2gMessage, b"not ours to decode".to_vec())),
    /// );
    /// # Ok::<_, iso15118::session::ConnectionError>(())
    /// ```
    #[must_use]
    pub fn next_frame(&mut self) -> Option<(PayloadType, Vec<u8>)> {
        self.frames.pop_front()
    }

    /// Queues a message for the wire.
    ///
    /// Borrowed rather than consumed: encoding only reads, and the session
    /// drivers need the message afterwards — to advance the ordering graph
    /// once, and only once, the bytes actually exist.
    pub fn send(&mut self, message: &Message) -> Result<(), ConnectionError> {
        let (payload_type, payload) = message.encode()?;
        self.send_frame(payload_type, &payload)
    }

    /// Queues an already-encoded payload for the wire, under `payload_type`.
    ///
    /// The sending half of [`Connection::next_frame`], and subject to the same
    /// `max_payload_len` ceiling — which is checked here rather than left to
    /// the caller, because a payload too large to send is also one the peer
    /// will refuse to receive.
    pub fn send_frame(
        &mut self,
        payload_type: PayloadType,
        payload: &[u8],
    ) -> Result<(), ConnectionError> {
        if payload.len() > self.max_payload_len {
            return Err(ConnectionError::Overflow { limit: self.max_payload_len });
        }
        let mut frame = alloc::vec![0u8; v2gtp::HEADER_LEN + payload.len()];
        let n = v2gtp::write_frame(payload_type, payload, &mut frame)
            .map_err(ConnectionError::Framing)?;
        self.tx.extend(frame[..n].iter().copied());
        Ok(())
    }

    /// True when there is nothing waiting to be written.
    #[must_use]
    pub fn transmit_is_empty(&self) -> bool {
        self.tx.is_empty()
    }

    /// Takes everything queued for the wire.
    #[must_use]
    pub fn take_transmit(&mut self) -> Vec<u8> {
        self.tx.drain(..).collect()
    }

    /// Number of buffered input bytes not yet part of a complete frame.
    #[must_use]
    pub fn pending_input(&self) -> usize {
        self.rx.len()
    }

    /// Number of complete frames waiting to be decoded.
    #[must_use]
    pub fn pending_frames(&self) -> usize {
        self.frames.len()
    }

    /// Drops everything buffered in both directions.
    pub fn reset(&mut self) {
        self.reset_input();
        self.tx.clear();
    }

    /// Drops what has arrived and not yet been read, leaving the outgoing queue
    /// alone.
    ///
    /// What a session does when the stream stops being one: the bytes behind a
    /// framing or decode fault were framed by the peer that produced the fault,
    /// and V2GTP has no delimiter to resynchronise on, so there is no honest way
    /// to say where the next message would have begun. Anything already queued
    /// *for* the wire was built before the fault and is still the caller's to
    /// flush — a refusal the peer is owed, most of all.
    pub fn reset_input(&mut self) {
        self.rx.clear();
        self.frames.clear();
    }
}

/// Why a byte stream could not be framed or a message could not be built.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ConnectionError {
    /// The V2GTP framing was malformed.
    Framing(V2gtpError),
    /// The framed payload was not a message this build can decode.
    Message(MessageError),
    /// A frame, or the buffer waiting for one, exceeded the configured limit.
    Overflow {
        /// The limit that was exceeded, in bytes.
        limit: usize,
    },
    /// The peer pipelined more frames than the half-duplex protocol allows.
    TooManyFrames {
        /// How many undrained frames the connection will hold.
        limit: usize,
    },
}

impl From<V2gtpError> for ConnectionError {
    fn from(e: V2gtpError) -> Self {
        Self::Framing(e)
    }
}

impl From<MessageError> for ConnectionError {
    fn from(e: MessageError) -> Self {
        Self::Message(e)
    }
}

impl fmt::Display for ConnectionError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Framing(e) => write!(f, "V2GTP: {e}"),
            Self::Message(e) => write!(f, "{e}"),
            Self::Overflow { limit } => write!(f, "frame exceeds the {limit} byte limit"),
            Self::TooManyFrames { limit } => {
                write!(f, "the peer pipelined more than {limit} unanswered V2GTP frames")
            }
        }
    }
}

#[cfg(feature = "std")]
impl std::error::Error for ConnectionError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Framing(e) => Some(e),
            Self::Message(e) => Some(e),
            Self::Overflow { .. } | Self::TooManyFrames { .. } => None,
        }
    }
}

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

    use super::*;
    use crate::app_protocol::SupportedAppProtocolReq;

    fn handshake() -> Message {
        Message::AppProtocolReq(Box::new(SupportedAppProtocolReq::advertising([
            Protocol::Iso20,
            Protocol::Iso2,
        ])))
    }

    #[test]
    fn a_frame_split_across_reads_is_reassembled() {
        let mut out = Connection::new();
        out.send(&handshake()).unwrap();
        let wire = out.take_transmit();

        let mut c = Connection::new();
        for chunk in wire.chunks(3) {
            c.receive(chunk).unwrap();
        }
        assert!(c.next_message().unwrap().is_some());
        assert_eq!(c.pending_input(), 0);
    }

    #[test]
    fn two_frames_in_one_read_come_out_as_two_messages() {
        let mut out = Connection::new();
        out.send(&handshake()).unwrap();
        out.send(&handshake()).unwrap();
        let wire = out.take_transmit();

        let mut c = Connection::new();
        c.receive(&wire).unwrap();
        assert!(c.next_message().unwrap().is_some());
        assert!(c.next_message().unwrap().is_some());
        assert!(c.next_message().unwrap().is_none());
    }

    #[test]
    fn an_incomplete_frame_is_not_an_error() {
        let mut out = Connection::new();
        out.send(&handshake()).unwrap();
        let wire = out.take_transmit();

        let mut c = Connection::new();
        c.receive(&wire[..wire.len() - 1]).unwrap();
        assert_eq!(c.next_message().unwrap(), None, "still waiting, not broken");
    }

    #[test]
    fn a_forged_length_is_refused_before_anything_is_buffered_for_it() {
        // A header claiming 4 GiB, with nothing behind it.
        let mut c = Connection::with_limit(4096);
        assert!(matches!(
            c.receive(&[0x01, 0xFE, 0x80, 0x01, 0xFF, 0xFF, 0xFF, 0xFF]),
            Err(ConnectionError::Framing(V2gtpError::PayloadTooLarge { .. }))
        ));
        assert_eq!(c.pending_input(), 0, "nothing is kept from a stream we cannot parse");
    }

    /// The buffer a peer can grow is bounded by one legal frame, whatever it
    /// sends and however it splits it.
    #[test]
    fn a_peer_that_never_finishes_a_frame_cannot_grow_the_buffer() {
        const LIMIT: usize = 64;
        let ceiling = LIMIT + v2gtp::HEADER_LEN;
        let mut c = Connection::with_limit(LIMIT);
        // A header claiming the largest legal payload, then a trickle that
        // stops one byte short of completing it, forever.
        c.receive(&[0x01, 0xFE, 0x80, 0x01, 0x00, 0x00, 0x00, u8::try_from(LIMIT).unwrap()])
            .unwrap();
        for _ in 0..LIMIT - 1 {
            c.receive(&[0u8]).unwrap();
            assert!(c.pending_input() <= ceiling, "buffered {} bytes", c.pending_input());
        }
        assert_eq!(c.pending_frames(), 0, "the frame is still one byte short");
    }

    #[test]
    fn garbage_is_rejected_rather_than_skipped() {
        let mut c = Connection::new();
        assert!(c.receive(&[0xDE, 0xAD, 0xBE, 0xEF, 0, 0, 0, 0]).is_err());
    }

    /// Several whole frames in one read stay inside the byte limit — that one
    /// bounds a frame, not a read — but the queue they land in has a ceiling of
    /// its own.
    #[test]
    fn whole_frames_in_one_read_are_not_a_byte_overflow() {
        let mut out = Connection::new();
        for _ in 0..MAX_PENDING_FRAMES {
            out.send(&handshake()).unwrap();
        }
        let wire = out.take_transmit();

        // A byte limit that one frame fits inside and four do not.
        let mut c = Connection::with_limit(wire.len() / MAX_PENDING_FRAMES);
        c.receive(&wire).unwrap();
        assert_eq!(c.pending_frames(), MAX_PENDING_FRAMES);
        assert_eq!(c.pending_input(), 0);
        for _ in 0..MAX_PENDING_FRAMES {
            assert!(c.next_message().unwrap().is_some());
        }
    }

    /// V2G is half-duplex: a peer that keeps sending without waiting for an
    /// answer is outside the protocol, and bounding only the *byte* buffer
    /// would have let it grow the decoded queue instead.
    #[test]
    fn a_peer_that_pipelines_without_waiting_is_cut_off() {
        let mut out = Connection::new();
        for _ in 0..=MAX_PENDING_FRAMES {
            out.send(&handshake()).unwrap();
        }
        let wire = out.take_transmit();

        let mut c = Connection::new();
        assert_eq!(
            c.receive(&wire),
            Err(ConnectionError::TooManyFrames { limit: MAX_PENDING_FRAMES })
        );
        assert_eq!(c.pending_frames(), 0, "nothing is kept from a stream we will not follow");
        assert_eq!(c.pending_input(), 0);
    }

    /// ...and draining as you go is not pipelining, however many frames the
    /// session carries in total.
    #[test]
    fn draining_between_reads_has_no_ceiling() {
        let mut out = Connection::new();
        out.send(&handshake()).unwrap();
        let one = out.take_transmit();

        let mut c = Connection::with_limits(4096, 1);
        for _ in 0..64 {
            c.receive(&one).unwrap();
            assert!(c.next_message().unwrap().is_some());
        }
    }

    /// \[V2G2-800\]: a manufacturer-specific frame is one to ignore, not one to
    /// end a session over.
    ///
    /// `0xA000..=0xFFFF` is reserved by Table 10 for exactly this, so a peer
    /// sending a vendor extension is conforming. A station that dropped the
    /// session on one would be a station nobody could extend.
    #[test]
    fn a_manufacturer_specific_frame_is_skipped_and_the_next_one_still_arrives() {
        let mut out = Connection::new();
        out.send_frame(PayloadType::ManufacturerSpecific(0xA123), b"vendor").unwrap();
        out.send(&handshake()).unwrap();
        let wire = out.take_transmit();

        let mut c = Connection::new();
        c.receive(&wire).unwrap();
        assert_eq!(
            c.next_message().unwrap().map(|m| m.name()),
            Some("supportedAppProtocolReq"),
            "the extension is stepped over, not stumbled over"
        );
        assert_eq!(c.ignored_frames(), 1);
        assert!(c.next_message().unwrap().is_none());
    }

    /// The same for a reserved code, which is not even legitimate — the header
    /// was still well-formed, so the frame boundary is still trustworthy.
    #[test]
    fn a_reserved_payload_type_is_skipped_rather_than_desynchronising_the_stream() {
        let mut out = Connection::new();
        out.send_frame(PayloadType::Reserved(0x8007), b"?").unwrap();
        out.send(&handshake()).unwrap();
        let wire = out.take_transmit();

        let mut c = Connection::new();
        c.receive(&wire).unwrap();
        assert!(c.next_message().unwrap().is_some());
        assert_eq!(c.ignored_frames(), 1);
    }

    /// A payload type belonging to a generation this session did not negotiate
    /// is ignored — which is the *same* security outcome as refusing it, minus
    /// the peer's ability to end the session by trying.
    #[test]
    fn a_message_for_another_generation_is_ignored_not_processed() {
        let mut out = Connection::new();
        out.send_frame(PayloadType::Part20Main, b"not for a -2 session").unwrap();
        let wire = out.take_transmit();

        let mut c = Connection::new();
        c.set_protocol(Protocol::Iso2);
        c.receive(&wire).unwrap();
        assert_eq!(c.next_message().unwrap(), None, "nothing decoded");
        assert_eq!(c.ignored_frames(), 1, "and nothing fatal happened either");
    }

    /// ...but a payload type that *does* belong to this session and has no
    /// codec is a different fault, and is reported. Ignoring it would look like
    /// a peer that had gone quiet.
    #[test]
    fn a_missing_codec_for_this_sessions_own_payload_type_is_reported() {
        let mut out = Connection::new();
        out.send_frame(PayloadType::ExiEncodedV2gMessage, b"a DIN message").unwrap();
        let wire = out.take_transmit();

        let mut c = Connection::new();
        c.set_protocol(Protocol::Din70121);
        c.receive(&wire).unwrap();
        assert!(matches!(c.next_message(), Err(ConnectionError::Message(_))));
        assert_eq!(c.ignored_frames(), 0);
    }

    /// Payload type `0x8001` means the handshake before negotiation and a -2
    /// message afterwards. Nothing in the frame says which.
    #[test]
    fn the_negotiated_protocol_decides_how_0x8001_is_read() {
        let mut out = Connection::new();
        out.send(&handshake()).unwrap();
        let wire = out.take_transmit();

        let mut c = Connection::new();
        c.set_protocol(Protocol::Iso2);
        c.receive(&wire).unwrap();
        assert!(
            c.next_message().is_err(),
            "a handshake read as a -2 message must fail, not decode to nonsense"
        );
    }
}