Drop frames whose header disagrees with the frame that arrived

The received frame header declares a payload length that nothing read. It now
gets compared against the frame the transport actually delivered, at the one
dispatch point every transport converges on, before that field can be used as
a parsing input.

This changes behaviour on a deployed line, so it is worth being exact about
what it drops. The stream transports read their frame boundary out of this
same field, so the comparison holds by construction and never fires for them.
The datagram transports deliver one whole frame per packet, where the length
is known exactly and nothing checked it before. A short read there is a
truncated frame, which already failed the tag or the exact-size parse; this
changes which reason it is dropped for, not whether it is dropped.

An unrecognised phase carries no fixed relationship, so it is left alone and
reaches the dispatch as before, rather than being rejected on a guess.

The drop gets its own rejection reason and its own counter. Reusing the
admission reason would have mis-attributed it: this is a framing rejection
decided before the phase dispatch, and it applies to data frames as well as
handshakes.

The dispatch function is now visible to the rest of the node module so a test
can drive one packet through it, which is the reach the handshake handlers
beside it already had.

src/node/handlers/rx_loop.rs

@@ -1,8 +1,10 @@
 //! RX event loop and packet dispatch.
 
 use crate::control::{ControlSocket, commands};
+use crate::node::reject::{RejectReason, TransportReject};
 use crate::node::wire::{
     COMMON_PREFIX_SIZE, CommonPrefix, FMP_VERSION, PHASE_ESTABLISHED, PHASE_MSG1, PHASE_MSG2,
+    expected_payload_len,
 };
 use crate::node::{Node, NodeError};
 use crate::transport::ReceivedPacket;
@@ -299,7 +301,11 @@ impl Node {
     /// Process a single received packet.
     ///
     /// Dispatches based on the phase field in the 4-byte common prefix.
-    async fn process_packet(&mut self, packet: ReceivedPacket) {
+    ///
+    /// Visible to the rest of `crate::node` so tests can drive a single
+    /// packet through the dispatch, the same reach `handle_msg1` and
+    /// `handle_msg2` already have.
+    pub(in crate::node) async fn process_packet(&mut self, packet: ReceivedPacket) {
         if packet.data.len() < COMMON_PREFIX_SIZE {
             return; // Drop packets too short for common prefix
         }
@@ -346,6 +352,39 @@ impl Node {
             return;
         }
 
+        // Drop a frame whose declared payload length disagrees with the
+        // frame that arrived, before that field can be used as a parsing
+        // input.
+        //
+        // Every transport's packets converge here, but the two families
+        // reach this line differently. TCP, Tor and Nym read their frame
+        // boundary out of this same field, so for them the comparison holds
+        // by construction and never fires. UDP, Ethernet and BLE deliver one
+        // whole frame per packet, where the arrived length is known exactly
+        // and nothing compares the two today. A short read on those
+        // transports is a truncated frame, which fails the AEAD tag or the
+        // exact-size handshake parse already; this changes which reason it
+        // is dropped for, not whether it is dropped.
+        //
+        // A `None` means the phase carries no fixed relationship and the
+        // frame is left alone rather than rejected, so an unrecognised phase
+        // still reaches the dispatch below and is handled there.
+        if let Some(expected) = expected_payload_len(prefix.phase, packet.data.len())
+            && prefix.payload_len != expected
+        {
+            debug!(
+                phase = prefix.phase,
+                declared = prefix.payload_len,
+                expected,
+                len = packet.data.len(),
+                transport_id = %packet.transport_id,
+                "FMP payload_len disagrees with frame length, dropping"
+            );
+            self.stats_mut()
+                .record_reject(RejectReason::Transport(TransportReject::PayloadLenMismatch));
+            return;
+        }
+
         match prefix.phase {
             PHASE_ESTABLISHED => {
                 self.handle_encrypted_frame(packet).await;

src/node/reject.rs

@@ -298,10 +298,10 @@ pub enum ForwardingReject {
 
 /// Transport-layer rejection reasons.
 ///
-/// Currently covers the admission cap-hit path at the TCP and Tor
-/// accept loops. Additional transport-side rejection variants
-/// (framing errors, connection failures wired through to the node
-/// stats path) can be added incrementally.
+/// Covers the admission cap-hit path at the TCP and Tor accept loops
+/// and the frame-length check at the receive dispatch point. Additional
+/// transport-side rejection variants (connection failures wired through
+/// to the node stats path) can be added incrementally.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 #[non_exhaustive]
 pub enum TransportReject {
@@ -310,6 +310,13 @@ pub enum TransportReject {
     /// (`max_inbound_connections`) was already reached. Tracked via
     /// [`TransportStats::inbound_cap_exceeded`](crate::node::stats::TransportStats).
     InboundCapExceeded,
+    /// Inbound FMP frame dropped because the payload length its header
+    /// declares does not match the frame the transport delivered. This
+    /// is a framing rejection, not an admission one: it applies to
+    /// established data frames as well as to handshake frames, and it
+    /// is decided before the phase dispatch. Tracked via
+    /// [`TransportStats::payload_len_mismatch`](crate::node::stats::TransportStats).
+    PayloadLenMismatch,
 }
 
 #[cfg(test)]
@@ -403,4 +410,18 @@ mod tests {
             RejectReason::Transport(TransportReject::InboundCapExceeded)
         ));
     }
+
+    #[test]
+    fn transport_reject_payload_len_mismatch_round_trips() {
+        let r = RejectReason::Transport(TransportReject::PayloadLenMismatch);
+        assert!(matches!(
+            r,
+            RejectReason::Transport(TransportReject::PayloadLenMismatch)
+        ));
+        assert_ne!(
+            r,
+            RejectReason::Transport(TransportReject::InboundCapExceeded),
+            "a framing drop must not compare equal to an admission rejection"
+        );
+    }
 }

src/node/stats.rs

@@ -202,6 +202,10 @@ impl MmpStats {
 /// typed-rejection enum stays the canonical entry point and so a
 /// future transport-to-node bridge (event or sampling) has a
 /// well-known destination.
+///
+/// `payload_len_mismatch` is different in kind: it counts a framing
+/// drop the node itself performs at the receive dispatch point, so it
+/// has a live writer and can be non-zero on a running node.
 #[derive(Default)]
 pub struct TransportStats {
     /// Reserved for node-side inbound-cap-exceeded admission rejection
@@ -209,18 +213,24 @@ pub struct TransportStats {
     /// on the transport-level stats (`TcpStats::connections_rejected`,
     /// `TorStats::connections_rejected`) directly.
     pub inbound_cap_exceeded: u64,
+    /// Inbound FMP frames dropped because the payload length declared
+    /// in the common prefix did not match the frame the transport
+    /// delivered.
+    pub payload_len_mismatch: u64,
 }
 
 impl TransportStats {
     pub fn snapshot(&self) -> TransportStatsSnapshot {
         TransportStatsSnapshot {
             inbound_cap_exceeded: self.inbound_cap_exceeded,
+            payload_len_mismatch: self.payload_len_mismatch,
         }
     }
 
     pub(super) fn record_reject(&mut self, reason: TransportReject) {
         match reason {
             TransportReject::InboundCapExceeded => self.inbound_cap_exceeded += 1,
+            TransportReject::PayloadLenMismatch => self.payload_len_mismatch += 1,
         }
     }
 }
@@ -396,6 +406,7 @@ pub struct MmpStatsSnapshot {
 #[derive(Clone, Debug, Default, Serialize)]
 pub struct TransportStatsSnapshot {
     pub inbound_cap_exceeded: u64,
+    pub payload_len_mismatch: u64,
 }
 
 #[derive(Clone, Debug, Default, Serialize)]
@@ -577,4 +588,25 @@ mod tests {
         stats.record_reject(RejectReason::Transport(TransportReject::InboundCapExceeded));
         assert_eq!(stats.transport.inbound_cap_exceeded, 1);
     }
+
+    /// Records both transport reasons so a swapped or shared arm shows up
+    /// as a mis-attributed counter rather than as a plausible total.
+    #[test]
+    fn transport_stats_record_reject_keeps_the_two_reasons_on_separate_counters() {
+        let mut s = TransportStats::default();
+        s.record_reject(TransportReject::PayloadLenMismatch);
+        s.record_reject(TransportReject::PayloadLenMismatch);
+        s.record_reject(TransportReject::InboundCapExceeded);
+        assert_eq!(s.payload_len_mismatch, 2);
+        assert_eq!(s.inbound_cap_exceeded, 1);
+    }
+
+    #[test]
+    fn node_stats_record_reject_dispatches_payload_len_mismatch_to_transport() {
+        let mut stats = NodeStats::new();
+        stats.record_reject(RejectReason::Transport(TransportReject::PayloadLenMismatch));
+        assert_eq!(stats.transport.payload_len_mismatch, 1);
+        assert_eq!(stats.transport.inbound_cap_exceeded, 0);
+        assert_eq!(stats.handshake.bad_state, 0);
+    }
 }

src/node/tests/handshake.rs

@@ -1151,3 +1151,137 @@ async fn test_should_admit_msg1_admits_rekey_when_addr_form_differs() {
     assert!(node.is_established_link_msg1(transport_id, &numeric_addr));
     assert!(!node.is_established_link_msg1(transport_id, &stranger_addr));
 }
+
+// ============================================================================
+// Frame-length validation at the dispatch point
+// ============================================================================
+
+/// Build a promoted peer and return the node, the peer's address, and the
+/// session index inbound frames must name to reach it.
+///
+/// `handle_encrypted_frame` looks a frame up by `(transport_id, receiver_idx)`
+/// in `peers_by_index`, so a frame carrying this index reaches the decrypt and
+/// bumps the peer's failure counter. That counter is how the tests below tell
+/// "the frame reached its handler" apart from "the frame was dropped before
+/// the dispatch": an unknown session is dropped silently and counts nothing.
+fn node_with_promoted_peer(transport_id: TransportId) -> (Node, NodeAddr, SessionIndex) {
+    let mut node = make_node();
+    let link_id = LinkId::new(1);
+    let (conn, identity) = make_completed_connection(&mut node, link_id, transport_id, 1_000);
+    let node_addr = *identity.node_addr();
+    node.add_connection(conn).unwrap();
+    node.promote_connection(link_id, identity, 2_000).unwrap();
+    let our_index = node
+        .get_peer(&node_addr)
+        .and_then(|p| p.our_index())
+        .expect("promoted peer must have our_index");
+    (node, node_addr, our_index)
+}
+
+/// A well-formed established frame carrying 40 bytes of inner plaintext.
+///
+/// 16-byte header + 40 + 16-byte tag = 72 bytes on the wire, declaring 40.
+/// The ciphertext is filler: these tests are about the length check, and
+/// every one of them stops before or at the AEAD.
+fn established_frame_declaring_40(receiver_idx: SessionIndex) -> Vec<u8> {
+    use crate::node::wire::{build_encrypted, build_established_header};
+    use crate::noise::TAG_SIZE;
+
+    let header = build_established_header(receiver_idx, 0, 0, 40);
+    build_encrypted(&header, &[0u8; 40 + TAG_SIZE])
+}
+
+#[tokio::test]
+async fn an_established_frame_whose_declared_payload_len_disagrees_with_its_length_is_dropped() {
+    let transport_id = TransportId::new(1);
+    let (mut node, node_addr, our_index) = node_with_promoted_peer(transport_id);
+
+    let mut frame = established_frame_declaring_40(our_index);
+    // Bytes 2-3 are the little-endian payload_len. 68 is what a validator
+    // written to the field's looser description would compute for this frame
+    // (72 on the wire minus the 4-byte common prefix), so it is both a wrong
+    // value and the specific wrong value worth naming.
+    frame[2..4].copy_from_slice(&68u16.to_le_bytes());
+
+    node.process_packet(ReceivedPacket::new(
+        transport_id,
+        TransportAddr::from_string("127.0.0.1:2121"),
+        frame,
+    ))
+    .await;
+
+    assert_eq!(
+        node.stats().transport.payload_len_mismatch,
+        1,
+        "the frame must be counted as a framing drop"
+    );
+    assert_eq!(
+        node.get_peer(&node_addr)
+            .expect("peer must survive a dropped frame")
+            .consecutive_decrypt_failures(),
+        0,
+        "the frame must be dropped before the phase dispatch, so the \
+         encrypted-frame handler never sees it"
+    );
+}
+
+#[tokio::test]
+async fn an_established_frame_with_a_correct_payload_len_is_not_dropped() {
+    let transport_id = TransportId::new(1);
+    let (mut node, node_addr, our_index) = node_with_promoted_peer(transport_id);
+
+    let frame = established_frame_declaring_40(our_index);
+
+    node.process_packet(ReceivedPacket::new(
+        transport_id,
+        TransportAddr::from_string("127.0.0.1:2121"),
+        frame,
+    ))
+    .await;
+
+    assert_eq!(
+        node.stats().transport.payload_len_mismatch,
+        0,
+        "a frame whose header agrees with its length must not be dropped"
+    );
+    assert_eq!(
+        node.get_peer(&node_addr)
+            .expect("peer must survive a failed decrypt below the threshold")
+            .consecutive_decrypt_failures(),
+        1,
+        "the frame must reach the encrypted-frame handler, where the filler \
+         ciphertext fails the AEAD tag"
+    );
+}
+
+#[tokio::test]
+async fn a_msg1_with_a_correct_payload_len_is_not_dropped() {
+    use crate::node::wire::build_msg1;
+
+    let mut node = make_node();
+    let transport_id = TransportId::new(1);
+
+    // A real-shaped msg1 with filler Noise bytes: `build_msg1` writes the
+    // only payload_len the msg1 arm accepts, and the handshake fails one
+    // step later at the DH, which is the observable that it got there.
+    let frame = build_msg1(SessionIndex::new(1), &[0u8; 106]);
+
+    node.process_packet(ReceivedPacket::new(
+        transport_id,
+        TransportAddr::from_string("127.0.0.1:2121"),
+        frame,
+    ))
+    .await;
+
+    assert_eq!(
+        node.stats().transport.payload_len_mismatch,
+        0,
+        "a msg1 built by the encoder must not be dropped by the length check"
+    );
+    assert_eq!(
+        node.stats().handshake.bad_state,
+        1,
+        "the msg1 must reach handle_msg1, which rejects the filler Noise \
+         payload at the DH"
+    );
+}

src/node/wire.rs

@@ -87,11 +87,6 @@ pub const FLAG_SP: u8 = 0x04;
 /// A `None` from the established arm means "no fixed relationship", so a caller
 /// skips such a frame rather than rejecting it. That is the safe direction: a
 /// truncating cast here would produce a false rejection instead.
-// Nothing calls this yet. The dispatch point that calls it arrives in the
-// following commit, which removes this annotation with it. `CommonPrefix::flags`
-// and `EncryptedHeader::payload_len` below carry the annotation for the same
-// reason.
-#[allow(dead_code)]
 pub fn expected_payload_len(phase: u8, total: usize) -> Option<u16> {
     match phase {
         PHASE_MSG1 => Some((MSG1_WIRE_SIZE - COMMON_PREFIX_SIZE) as u16),

src/transport/ble/io.rs

@@ -23,6 +23,15 @@ pub trait BleStream: Send + Sync {
     /// Receive data from the L2CAP connection.
     ///
     /// Returns the number of bytes read into `buf`.
+    ///
+    /// A single call must never return bytes drawn from more than one SDU.
+    /// The receive loop emits what one call returns as one FMP frame. A
+    /// concatenation is caught by the frame-length check in the node's
+    /// dispatch only when the leading frame is an established one; where it
+    /// is a handshake frame, that check passes and the buffer is dropped one
+    /// step later by that frame kind's exact-size parse. Returning less than
+    /// a whole SDU is allowed: a truncated frame fails the AEAD tag or the
+    /// exact-size parse regardless.
     fn recv(
         &self,
         buf: &mut [u8],