fix(tcp): reset FIN-WAIT data after receive shutdown

src/iface/packet.rs

@@ -70,12 +70,7 @@ impl<'p> Packet<'p> {
         }
     }
 
-    pub fn emit_payload(
-        &self,
-        _ip_repr: &IpRepr,
-        payload: &mut [u8],
-        caps: &DeviceCapabilities,
-    ) {
+    pub fn emit_payload(&self, _ip_repr: &IpRepr, payload: &mut [u8], caps: &DeviceCapabilities) {
         match self.payload() {
             #[cfg(feature = "proto-ipv4")]
             IpPayload::Icmpv4(icmpv4_repr) => {

src/phy/fault_injector.rs

@@ -196,10 +196,12 @@ impl<D: Device> FaultInjector<D> {
 }
 
 impl<D: Device> Device for FaultInjector<D> {
-    type RxToken<'a> = RxToken<'a>
+    type RxToken<'a>
+        = RxToken<'a>
     where
         Self: 'a;
-    type TxToken<'a> = TxToken<'a, D::TxToken<'a>>
+    type TxToken<'a>
+        = TxToken<'a, D::TxToken<'a>>
     where
         Self: 'a;
 

src/phy/fuzz_injector.rs

@@ -47,10 +47,12 @@ where
     FTx: Fuzzer,
     FRx: Fuzzer,
 {
-    type RxToken<'a> = RxToken<'a, D::RxToken<'a>, FRx>
+    type RxToken<'a>
+        = RxToken<'a, D::RxToken<'a>, FRx>
     where
         Self: 'a;
-    type TxToken<'a> = TxToken<'a, D::TxToken<'a>, FTx>
+    type TxToken<'a>
+        = TxToken<'a, D::TxToken<'a>, FTx>
     where
         Self: 'a;
 

src/phy/pcap_writer.rs

@@ -165,10 +165,12 @@ impl<D: Device, S> Device for PcapWriter<D, S>
 where
     S: PcapSink,
 {
-    type RxToken<'a> = RxToken<'a, D::RxToken<'a>, S>
+    type RxToken<'a>
+        = RxToken<'a, D::RxToken<'a>, S>
     where
         Self: 'a;
-    type TxToken<'a> = TxToken<'a, D::TxToken<'a>, S>
+    type TxToken<'a>
+        = TxToken<'a, D::TxToken<'a>, S>
     where
         Self: 'a;
 

src/phy/raw_socket.rs

@@ -59,10 +59,12 @@ impl RawSocket {
 }
 
 impl Device for RawSocket {
-    type RxToken<'a> = RxToken
+    type RxToken<'a>
+        = RxToken
     where
         Self: 'a;
-    type TxToken<'a> = TxToken
+    type TxToken<'a>
+        = TxToken
     where
         Self: 'a;
 

src/phy/tracer.rs

@@ -42,10 +42,12 @@ impl<D: Device> Tracer<D> {
 }
 
 impl<D: Device> Device for Tracer<D> {
-    type RxToken<'a> = RxToken<D::RxToken<'a>>
+    type RxToken<'a>
+        = RxToken<D::RxToken<'a>>
     where
         Self: 'a;
-    type TxToken<'a> = TxToken<D::TxToken<'a>>
+    type TxToken<'a>
+        = TxToken<D::TxToken<'a>>
     where
         Self: 'a;
 

src/socket/icmp.rs

@@ -426,12 +426,7 @@ impl<'a> Socket<'a> {
     /// Accepted packets are enqueued into the socket's receive buffer.
     #[cfg(feature = "proto-ipv4")]
     #[inline]
-    pub fn accepts_v4(
-        &self,
-        cx: &mut Context,
-        ip_repr: &Ipv4Repr,
-        icmp_repr: &Icmpv4Repr,
-    ) -> bool {
+    pub fn accepts_v4(&self, cx: &mut Context, ip_repr: &Ipv4Repr, icmp_repr: &Icmpv4Repr) -> bool {
         match (&self.endpoint, icmp_repr) {
             // If we are bound to ICMP errors associated to a UDP port, only
             // accept Destination Unreachable or Time Exceeded messages with
@@ -468,12 +463,7 @@ impl<'a> Socket<'a> {
     /// Accepted packets are enqueued into the socket's receive buffer.
     #[cfg(feature = "proto-ipv6")]
     #[inline]
-    pub fn accepts_v6(
-        &self,
-        cx: &mut Context,
-        ip_repr: &Ipv6Repr,
-        icmp_repr: &Icmpv6Repr,
-    ) -> bool {
+    pub fn accepts_v6(&self, cx: &mut Context, ip_repr: &Ipv6Repr, icmp_repr: &Icmpv6Repr) -> bool {
         match (&self.endpoint, icmp_repr) {
             // If we are bound to ICMP errors associated to a UDP port, only
             // accept Destination Unreachable or Time Exceeded messages with
@@ -507,12 +497,7 @@ impl<'a> Socket<'a> {
     }
 
     #[cfg(feature = "proto-ipv4")]
-    pub fn process_v4(
-        &mut self,
-        _cx: &mut Context,
-        ip_repr: &Ipv4Repr,
-        icmp_repr: &Icmpv4Repr,
-    ) {
+    pub fn process_v4(&mut self, _cx: &mut Context, ip_repr: &Ipv4Repr, icmp_repr: &Icmpv4Repr) {
         net_trace!("icmp: receiving {} octets", icmp_repr.buffer_len());
 
         match self
@@ -533,12 +518,7 @@ impl<'a> Socket<'a> {
     }
 
     #[cfg(feature = "proto-ipv6")]
-    pub fn process_v6(
-        &mut self,
-        _cx: &mut Context,
-        ip_repr: &Ipv6Repr,
-        icmp_repr: &Icmpv6Repr,
-    ) {
+    pub fn process_v6(&mut self, _cx: &mut Context, ip_repr: &Ipv6Repr, icmp_repr: &Icmpv6Repr) {
         net_trace!("icmp: receiving {} octets", icmp_repr.buffer_len());
 
         match self

src/socket/tcp.rs

@@ -419,6 +419,12 @@ pub struct Socket<'a> {
     assembler: Assembler,
     rx_buffer: SocketBuffer<'a>,
     rx_fin_received: bool,
+    /// The local receive half has been shut down by the user.
+    ///
+    /// This models the BSD/Linux close(2) orphan case. Once the application has
+    /// closed the receive side, new data in FIN-WAIT states must reset the
+    /// connection instead of being queued indefinitely.
+    rx_shutdown: bool,
     tx_buffer: SocketBuffer<'a>,
     /// Interval after which, if no inbound packets are received, the connection is aborted.
     timeout: Option<Duration>,
@@ -524,6 +530,7 @@ impl<'a> Socket<'a> {
             tx_buffer,
             rx_buffer,
             rx_fin_received: false,
+            rx_shutdown: false,
             timeout: None,
             keep_alive: None,
             hop_limit: None,
@@ -899,6 +906,7 @@ impl<'a> Socket<'a> {
         self.tx_buffer.clear();
         self.rx_buffer.clear();
         self.rx_fin_received = false;
+        self.rx_shutdown = false;
         self.listen_endpoint = IpListenEndpoint::default();
         self.tuple = None;
         self.local_seq_no = TcpSeqNumber::default();
@@ -1102,6 +1110,22 @@ impl<'a> Socket<'a> {
         self.set_state(State::Closed);
     }
 
+    /// Shut down the receive half of the connection.
+    ///
+    /// This mirrors the transport effect of Linux `sk_shutdown & RCV_SHUTDOWN`.
+    /// In FIN-WAIT states, any new data that advances the peer sequence space
+    /// actively resets the connection instead of being queued.
+    #[inline]
+    pub fn shutdown_recv(&mut self) {
+        self.rx_shutdown = true;
+    }
+
+    /// Return whether the receive half has been explicitly shut down.
+    #[inline]
+    pub fn recv_shutdown(&self) -> bool {
+        self.rx_shutdown
+    }
+
     /// Return whether the socket is passively listening for incoming connections.
     ///
     /// In terms of the TCP state machine, the socket must be in the `LISTEN` state.
@@ -1182,6 +1206,10 @@ impl<'a> Socket<'a> {
     /// `FIN-WAIT-1`, or `FIN-WAIT-2` state, or have data in the receive buffer instead.
     #[inline]
     pub fn may_recv(&self) -> bool {
+        if self.rx_shutdown && self.rx_buffer.is_empty() {
+            return false;
+        }
+
         match self.state {
             State::Established => true,
             // In FIN-WAIT-1/2, we have closed our transmit half of the connection but
@@ -1517,6 +1545,19 @@ impl<'a> Socket<'a> {
         Some(self.ack_reply(ip_repr, repr))
     }
 
+    #[inline]
+    fn should_reset_receive_shutdown_data(
+        &self,
+        segment_start: TcpSeqNumber,
+        segment_end: TcpSeqNumber,
+        window_start: TcpSeqNumber,
+    ) -> bool {
+        self.rx_shutdown
+            && matches!(self.state, State::FinWait1 | State::FinWait2)
+            && segment_start != segment_end
+            && segment_end > window_start
+    }
+
     pub fn accepts(&self, _cx: &mut Context, ip_repr: &IpRepr, repr: &TcpRepr) -> bool {
         if self.state == State::Closed {
             return false;
@@ -1671,6 +1712,15 @@ impl<'a> Socket<'a> {
         let segment_start = repr.seq_number;
         let segment_end = repr.seq_number + repr.payload.len();
 
+        if self.should_reset_receive_shutdown_data(segment_start, segment_end, window_start) {
+            net_debug!(
+                "received data after receive shutdown in {}, sending RST",
+                self.state
+            );
+            self.set_state(State::Closed);
+            return Some(Self::rst_reply(ip_repr, repr));
+        }
+
         let (payload, payload_offset) = match self.state {
             // In LISTEN and SYN-SENT states, we have not yet synchronized with the remote end.
             State::Listen | State::SynSent => (&[][..], 0),
@@ -4807,6 +4857,37 @@ mod test {
         sanity!(s, socket_fin_wait_1());
     }
 
+    #[test]
+    fn test_fin_wait_receive_shutdown_data_resets() {
+        let mut s = socket_fin_wait_1();
+        s.shutdown_recv();
+
+        send!(
+            s,
+            TcpRepr {
+                seq_number: REMOTE_SEQ + 1,
+                ack_number: Some(LOCAL_SEQ + 1),
+                payload: &b"x"[..],
+                ..SEND_TEMPL
+            },
+            Some(TcpRepr {
+                src_port: LOCAL_PORT,
+                dst_port: REMOTE_PORT,
+                control: TcpControl::Rst,
+                seq_number: LOCAL_SEQ + 1,
+                ack_number: None,
+                window_len: 0,
+                window_scale: None,
+                max_seg_size: None,
+                sack_permitted: false,
+                sack_ranges: [None, None, None],
+                timestamp: None,
+                payload: &[],
+            })
+        );
+        assert_eq!(s.state, State::Closed);
+    }
+
     #[test]
     fn test_established_abort() {
         let mut s = socket_established();