SYN in TIME_WAIT is dropped when the stack accepts through tcp.Forwarder #15013

// Reproducer: with a tcp.Forwarder as the TCP protocol handler, a SYN that
// reuses the 4-tuple of an endpoint in TIME_WAIT is dropped without a reply
// until TIME_WAIT expires, even though its sequence number is above the old
// connection's RcvNxt (RFC 1122's reopen case).
//
// The stack is driven through a channel endpoint: the test plays the peer,
// injecting segments and reading what the stack sends.
package repro

import (
        "context"
        "testing"
        "time"

        "gvisor.dev/gvisor/pkg/buffer"
        "gvisor.dev/gvisor/pkg/tcpip"
        "gvisor.dev/gvisor/pkg/tcpip/header"
        "gvisor.dev/gvisor/pkg/tcpip/link/channel"
        "gvisor.dev/gvisor/pkg/tcpip/network/ipv4"
        "gvisor.dev/gvisor/pkg/tcpip/stack"
        "gvisor.dev/gvisor/pkg/tcpip/transport/tcp"
        "gvisor.dev/gvisor/pkg/waiter"
)

var (
        peerAddr  = tcpip.AddrFrom4([4]byte{10, 0, 0, 1}) // the remote side (this test)
        stackAddr = tcpip.AddrFrom4([4]byte{10, 0, 0, 2}) // the gVisor side
)

const peerPort, stackPort = 40000, 80

// newStack builds an IPv4 stack whose TCP connections are all accepted by a
// tcp.Forwarder. The forwarder's handler accepts and closes at once, so the
// gVisor side sends the first FIN and ends in TIME_WAIT. timeWait == 0 keeps
// gVisor's default (60 s).
func newStack(t *testing.T, timeWait time.Duration) *channel.Endpoint {
        return newStackWith(t, timeWait, false)
}

// newStackWith is newStack, or with listener set, a stack that accepts the
// same connections through an ordinary listening endpoint on stackPort
// instead of a forwarder, closing each one at once in the same way.
func newStackWith(t *testing.T, timeWait time.Duration, listener bool) *channel.Endpoint {
        t.Helper()
        s := stack.New(stack.Options{
                NetworkProtocols:   []stack.NetworkProtocolFactory{ipv4.NewProtocol},
                TransportProtocols: []stack.TransportProtocolFactory{tcp.NewProtocol},
        })
        t.Cleanup(s.Close)
        if timeWait > 0 {
                opt := tcpip.TCPTimeWaitTimeoutOption(timeWait)
                if err := s.SetTransportProtocolOption(tcp.ProtocolNumber, &opt); err != nil {
                        t.Fatal(err)
                }
        }
        if !listener {
                fwd := tcp.NewForwarder(s, 0, 16, func(r *tcp.ForwarderRequest) {
                        var wq waiter.Queue
                        ep, err := r.CreateEndpoint(&wq)
                        if err != nil {
                                r.Complete(true)
                                return
                        }
                        r.Complete(false)
                        ep.Close()
                })
                s.SetTransportProtocolHandler(tcp.ProtocolNumber, fwd.HandlePacket)
        }

        link := channel.New(64, 1500, "")
        if err := s.CreateNIC(1, link); err != nil {
                t.Fatal(err)
        }
        if err := s.AddProtocolAddress(1, tcpip.ProtocolAddress{
                Protocol:          ipv4.ProtocolNumber,
                AddressWithPrefix: stackAddr.WithPrefix(),
        }, stack.AddressProperties{}); err != nil {
                t.Fatal(err)
        }
        s.SetRouteTable([]tcpip.Route{{Destination: header.IPv4EmptySubnet, NIC: 1}})
        if listener {
                var wq waiter.Queue
                ln, err := s.NewEndpoint(tcp.ProtocolNumber, ipv4.ProtocolNumber, &wq)
                if err != nil {
                        t.Fatal(err)
                }
                if err := ln.Bind(tcpip.FullAddress{Addr: stackAddr, Port: stackPort}); err != nil {
                        t.Fatal(err)
                }
                if err := ln.Listen(16); err != nil {
                        t.Fatal(err)
                }
                we, ch := waiter.NewChannelEntry(waiter.ReadableEvents)
                wq.EventRegister(&we)
                go func() {
                        for {
                                ep, _, err := ln.Accept(nil)
                                if err == nil {
                                        ep.Close()
                                        continue
                                }
                                if _, ok := err.(*tcpip.ErrWouldBlock); !ok {
                                        return
                                }
                                <-ch
                        }
                }()
                t.Cleanup(ln.Close)
        }
        return link
}

// send injects one TCP segment from the peer.
func send(link *channel.Endpoint, seq, ack uint32, flags header.TCPFlags) {
        pkt := make([]byte, header.IPv4MinimumSize+header.TCPMinimumSize)
        ip := header.IPv4(pkt)
        ip.Encode(&header.IPv4Fields{
                TotalLength: uint16(len(pkt)),
                TTL:         64,
                Protocol:    uint8(tcp.ProtocolNumber),
                SrcAddr:     peerAddr,
                DstAddr:     stackAddr,
        })
        ip.SetChecksum(^ip.CalculateChecksum())
        th := header.TCP(pkt[header.IPv4MinimumSize:])
        th.Encode(&header.TCPFields{
                SrcPort:    peerPort,
                DstPort:    stackPort,
                SeqNum:     seq,
                AckNum:     ack,
                DataOffset: header.TCPMinimumSize,
                Flags:      flags,
                WindowSize: 65535,
        })
        xsum := header.PseudoHeaderChecksum(tcp.ProtocolNumber, peerAddr, stackAddr, header.TCPMinimumSize)
        th.SetChecksum(^th.CalculateChecksum(xsum))
        pb := stack.NewPacketBuffer(stack.PacketBufferOptions{Payload: buffer.MakeWithData(pkt)})
        link.InjectInbound(ipv4.ProtocolNumber, pb)
        pb.DecRef()
}

// expect waits up to d for a segment from the stack carrying all of want and
// returns its sequence number; ok is false if none came.
func expect(link *channel.Endpoint, want header.TCPFlags, d time.Duration) (seq uint32, ok bool) {
        ctx, cancel := context.WithTimeout(context.Background(), d)
        defer cancel()
        for {
                pb := link.ReadContext(ctx)
                if pb == nil {
                        return 0, false
                }
                v := pb.ToView()
                th := header.TCP(header.IPv4(v.AsSlice()).Payload())
                flags, s := th.Flags(), th.SequenceNumber()
                v.Release()
                pb.DecRef()
                if flags&want == want {
                        return s, true
                }
        }
}

// reopen runs one connection that the gVisor side closes first, then a new
// SYN on the same 4-tuple with a higher sequence number after wait, and
// reports whether that SYN was answered within two seconds.
func reopen(t *testing.T, timeWait, wait time.Duration) bool {
        return reopenWith(t, timeWait, wait, false)
}

func reopenWith(t *testing.T, timeWait, wait time.Duration, listener bool) bool {
        link := newStackWith(t, timeWait, listener)

        send(link, 1000, 0, header.TCPFlagSyn)
        synAck, ok := expect(link, header.TCPFlagSyn|header.TCPFlagAck, 2*time.Second)
        if !ok {
                t.Fatal("first SYN not answered")
        }
        send(link, 1001, synAck+1, header.TCPFlagAck)
        fin, ok := expect(link, header.TCPFlagFin, 2*time.Second)
        if !ok {
                t.Fatal("stack did not close first")
        }
        send(link, 1001, fin+1, header.TCPFlagAck|header.TCPFlagFin)
        // The gVisor endpoint is now in TIME_WAIT.

        time.Sleep(wait)
        send(link, 5000, 0, header.TCPFlagSyn) // 5000 > old RcvNxt (1002)
        _, ok = expect(link, header.TCPFlagSyn|header.TCPFlagAck, 2*time.Second)
        return ok
}

// Fails: the new SYN hits the TIME_WAIT endpoint, handleTimeWaitSegment
// reports newSyn, handleTimeWaitSegments looks for a listening endpoint,
// there is none (connections are accepted by the forwarder), and the SYN
// is dropped. It stays dropped for the 60 s default TIME_WAIT.
func TestReopenAfterTimeWaitWithForwarder(t *testing.T) {
        if !reopen(t, 0, 500*time.Millisecond) {
                t.Fatal("new SYN on the 4-tuple of a TIME_WAIT endpoint was not answered")
        }
}

// Control: with TIME_WAIT shortened below the wait, the same SYN is answered,
// so the sequence above is otherwise valid.
func TestReopenAfterShortTimeWait(t *testing.T) {
        if !reopen(t, 200*time.Millisecond, 500*time.Millisecond) {
                t.Fatal("new SYN not answered even after TIME_WAIT ended")
        }
}

// Control: the same sequence against an ordinary listening endpoint, default
// TIME_WAIT. The reopen is honoured -- the SYN is redirected to the
// listener -- so the drop is specific to connections accepted by the
// forwarder.
func TestReopenAfterTimeWaitWithListener(t *testing.T) {
        if !reopenWith(t, 0, 500*time.Millisecond, true) {
                t.Fatal("new SYN not answered with a listener either")
        }
}

tw := tcpip.TCPTimeWaitTimeoutOption(time.Second)
s.SetTransportProtocolOption(tcp.ProtocolNumber, &tw)