From 874ffd6a2655821ca5cbcf5b274fbb7395c61fe1 Mon Sep 17 00:00:00 2001 From: Bhasker Hariharan Date: Thu, 19 May 2022 16:01:35 -0700 Subject: [PATCH] Fix SYN-COOKIE based connection wakeup. When SYN-COOKIE is in use an ACK that completes the handshake could be followed by other segments that carry data and potentially get queued to the listening endpoint. This happens because the first ACK may not have been processed by the listening endpoint when the other segments arrived and as a result the listening endpoint is the closest matching endpoint for the incoming segment. What this means is when processing an ACK that has a valid cookie we must first check if the ACK now matches a fully established connection and in such a case reflect the segment to the fully established connections endpoint queue and stop further processing in the listen context for that segment. Further we missed removing endpoints from pendingEndpoints when a passive handshake fails. In such cases overtime failed handshakes would cause SYN-COOKIE to be used overtime even when the server is not slow to accept connections. This can lead to degraded performance as SYN-COOKIE based connections today in netstack drop support for window scaling as well as SACK etc. PiperOrigin-RevId: 449852701 --- pkg/tcpip/transport/tcp/accept.go | 35 ++- pkg/tcpip/transport/tcp/connect.go | 2 +- pkg/tcpip/transport/tcp/dispatcher.go | 5 + pkg/tcpip/transport/tcp/test/e2e/tcp_test.go | 269 ++++++++++++++---- .../transport/tcp/testing/context/context.go | 7 +- 5 files changed, 256 insertions(+), 62 deletions(-) diff --git a/pkg/tcpip/transport/tcp/accept.go b/pkg/tcpip/transport/tcp/accept.go index 27994925a..76e610101 100644 --- a/pkg/tcpip/transport/tcp/accept.go +++ b/pkg/tcpip/transport/tcp/accept.go @@ -573,6 +573,39 @@ func (e *endpoint) handleListenSegment(ctx *listenContext, s *segment) tcpip.Err return replyWithReset(e.stack, s, e.sendTOS, e.ipv4TTL, e.ipv6HopLimit) } + // As an edge case when SYN-COOKIES are in use and we receive a + // segment that has data and is valid we should check if it + // already matches a created endpoint and redirect the segment + // rather than try and create a new endpoint. This can happen + // where the final ACK for the handshake and other data packets + // arrive at the same time and are queued to the listening + // endpoint before the listening endpoint has had time to + // process the first ACK and create the endpoint that matches + // the incoming packet's full 5 tuple. + netProtos := []tcpip.NetworkProtocolNumber{s.netProto} + // If the local address is an IPv4 Address then also look for IPv6 + // dual stack endpoints. + if s.id.LocalAddress.To4() != "" { + netProtos = []tcpip.NetworkProtocolNumber{header.IPv4ProtocolNumber, header.IPv6ProtocolNumber} + } + for _, netProto := range netProtos { + if newEP := e.stack.FindTransportEndpoint(netProto, ProtocolNumber, s.id, s.nicID); newEP != nil && newEP != e { + tcpEP := newEP.(*endpoint) + if !tcpEP.EndpointState().connected() { + continue + } + if !tcpEP.enqueueSegment(s) { + // Just silently drop the segment as we failed + // to queue, we don't want to generate a RST + // further below or try and create a new + // endpoint etc. + return nil + } + tcpEP.notifyProcessor() + return nil + } + } + // Keep hold of acceptMu until the new endpoint is in the accept queue (or // if there is an error), to guarantee that we will keep our spot in the // queue even if another handshake from the syn queue completes. @@ -668,7 +701,7 @@ func (e *endpoint) handleListenSegment(ctx *listenContext, s *segment) tcpip.Err n.mu.Unlock() // Requeue the segment if the ACK completing the handshake has more info - // to be procesed by the newly established endpoint. + // to be processed by the newly established endpoint. if (s.flags.Contains(header.TCPFlagFin) || s.data.Size() > 0) && n.enqueueSegment(s) { n.notifyProcessor() } diff --git a/pkg/tcpip/transport/tcp/connect.go b/pkg/tcpip/transport/tcp/connect.go index 8f5b000e6..79d887a0a 100644 --- a/pkg/tcpip/transport/tcp/connect.go +++ b/pkg/tcpip/transport/tcp/connect.go @@ -485,7 +485,7 @@ func (h *handshake) synRcvdState(s *segment) tcpip.Error { h.transitionToStateEstablishedLocked(s) // Requeue the segment if the ACK completing the handshake has more info - // to be procesed by the newly established endpoint. + // to be processed by the newly established endpoint. if (s.flags.Contains(header.TCPFlagFin) || s.data.Size() > 0) && h.ep.enqueueSegment(s) { h.ep.protocol.dispatcher.selectProcessor(h.ep.ID).queueEndpoint(h.ep) diff --git a/pkg/tcpip/transport/tcp/dispatcher.go b/pkg/tcpip/transport/tcp/dispatcher.go index c066b29a6..4622feb2e 100644 --- a/pkg/tcpip/transport/tcp/dispatcher.go +++ b/pkg/tcpip/transport/tcp/dispatcher.go @@ -148,6 +148,11 @@ func (p *processor) handleConnecting(ep *endpoint) { if err := ep.h.processSegments(); err != nil { // +checklocksforce:ep.h.ep.mu // handshake failed. clean up the tcp endpoint and handshake // state. + if lEP := ep.h.listenEP; lEP != nil { + lEP.acceptMu.Lock() + delete(lEP.acceptQueue.pendingEndpoints, ep) + lEP.acceptMu.Unlock() + } ep.handshakeFailed(err) cleanup() return diff --git a/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go b/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go index 49414f15c..cc23e2a67 100644 --- a/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go +++ b/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go @@ -1175,12 +1175,11 @@ func TestSendRstOnListenerRxSynAckV6(t *testing.T) { checker.TCPSeqNum(200))) } -// TestTCPAckBeforeAcceptV4 tests that once the 3-way handshake is complete, -// peers can send data and expect a response within a reasonable ammount of time -// without calling Accept on the listening endpoint first. -// -// This test uses IPv4. -func TestTCPAckBeforeAcceptV4(t *testing.T) { +// TestNoSynCookieWithoutOverflow tests that SYN-COOKIEs are not issued when the +// queue is not overflowing. That is as long as newly completed connections are being +// accepted we do not see a SYN-COOKIE even > 2x listen backlog number of connections +// are accepted. +func TestNoSynCookieWithoutOverflow(t *testing.T) { c := context.New(t, e2e.DefaultMTU) defer c.Cleanup() @@ -1190,27 +1189,160 @@ func TestTCPAckBeforeAcceptV4(t *testing.T) { t.Fatal("Bind failed:", err) } - if err := c.EP.Listen(10); err != nil { + const backlog = 10 + if err := c.EP.Listen(backlog); err != nil { t.Fatal("Listen failed:", err) } - irs, iss := executeHandshake(t, c, context.TestPort, false /* synCookiesInUse */) + doOne := func(portIndex int) { + // Try to accept the connection. + we, ch := waiter.NewChannelEntry(waiter.ReadableEvents) + c.WQ.EventRegister(&we) + defer c.WQ.EventUnregister(&we) - // Send data before accepting the connection. - c.SendPacket([]byte{1, 2, 3, 4}, &context.Headers{ - SrcPort: context.TestPort, - DstPort: context.StackPort, - Flags: header.TCPFlagAck, - SeqNum: irs + 1, - AckNum: iss + 1, - }) + _, _ = executeHandshake(t, c, context.TestPort+uint16(portIndex), false /* synCookiesInUse */) - // Receive ACK for the data we sent. - checker.IPv4(t, c.GetPacket(), checker.TCP( - checker.DstPort(context.TestPort), - checker.TCPFlags(header.TCPFlagAck), - checker.TCPSeqNum(uint32(iss+1)), - checker.TCPAckNum(uint32(irs+5)))) + _, _, err := c.EP.Accept(nil) + if err == nil { + return + } + switch { + case cmp.Equal(&tcpip.ErrWouldBlock{}, err): + { + select { + case <-ch: + _, _, err = c.EP.Accept(nil) + if err != nil { + t.Fatalf("Accept failed: %s", err) + } + case <-time.After(1 * time.Second): + t.Fatalf("Timed out waiting for accept") + } + } + default: + t.Fatalf("Accept failed: %s", err) + } + } + + for i := 0; i < backlog*5; i++ { + doOne(i) + } +} + +// TestNoSynCookieOnFailedHandshakes tests that failed handshakes clear +// endpoints from the pending queue. This is tested by verifying that the +// SYN-ACK from the stack carries a valid window scale despite > 2xbacklog +// handshakes failing. +// +// If a failed handshake was not resulting in removal from pending endpoints +// list for the accepting endpoint then it will eventually result in a +// SYN-COOKIE which we can identify with a SYN-ACK w/ a WS of -1. +func TestNoSynCookieOnFailedHandshakes(t *testing.T) { + c := context.New(t, e2e.DefaultMTU) + defer c.Cleanup() + + c.Create(-1) + + if err := c.EP.Bind(tcpip.FullAddress{Port: context.StackPort}); err != nil { + t.Fatal("Bind failed:", err) + } + + const backlog = 10 + if err := c.EP.Listen(backlog); err != nil { + t.Fatal("Listen failed:", err) + } + + doOne := func() { + // Send a SYN request. + options := []byte{header.TCPOptionWS, 3, 0, header.TCPOptionNOP} + irs := seqnum.Value(context.TestInitialSequenceNumber) + c.SendPacket(nil, &context.Headers{ + SrcPort: context.TestPort, + DstPort: context.StackPort, + Flags: header.TCPFlagSyn, + SeqNum: irs, + RcvWnd: 30000, + TCPOpts: options, + }) + + // Receive the SYN-ACK reply. + b := c.GetPacket() + tcpHdr := header.TCP(header.IPv4(b).Payload()) + iss := seqnum.Value(tcpHdr.SequenceNumber()) + tcpCheckers := []checker.TransportChecker{ + checker.SrcPort(context.StackPort), + checker.DstPort(context.TestPort), + checker.TCPFlags(header.TCPFlagAck | header.TCPFlagSyn), + checker.TCPAckNum(uint32(irs) + 1), + checker.TCPSynOptions(header.TCPSynOptions{ + WS: tcp.FindWndScale(tcp.DefaultReceiveBufferSize), + MSS: c.MSSWithoutOptions(), + }), + } + + checker.IPv4(t, b, checker.TCP(tcpCheckers...)) + + // Send a RST to abort the handshake. + c.SendPacket(nil, &context.Headers{ + SrcPort: context.TestPort, + DstPort: context.StackPort, + Flags: header.TCPFlagRst, + SeqNum: irs + 1, + AckNum: iss + 1, + RcvWnd: 0, + }) + + } + + for i := 0; i < backlog*5; i++ { + doOne() + } +} + +// TestTCPAckBeforeAcceptV4 tests that once the 3-way handshake is complete, +// peers can send data and expect a response within a reasonable ammount of time +// without calling Accept on the listening endpoint first. +// +// This test uses IPv4. +func TestTCPAckBeforeAcceptV4(t *testing.T) { + for _, cookieEnabled := range []tcpip.TCPAlwaysUseSynCookies{false, true} { + t.Run(fmt.Sprintf("syn-cookies enabled: %t", cookieEnabled), func(t *testing.T) { + c := context.New(t, e2e.DefaultMTU) + defer c.Cleanup() + + if err := c.Stack().SetTransportProtocolOption(header.TCPProtocolNumber, &cookieEnabled); err != nil { + panic(fmt.Sprintf("SetTransportProtocolOption(%d, %T) = %s", header.TCPProtocolNumber, cookieEnabled, err)) + } + + c.Create(-1) + + if err := c.EP.Bind(tcpip.FullAddress{Port: context.StackPort}); err != nil { + t.Fatal("Bind failed:", err) + } + + if err := c.EP.Listen(10); err != nil { + t.Fatal("Listen failed:", err) + } + + irs, iss := executeHandshake(t, c, context.TestPort, bool(cookieEnabled)) + + // Send data before accepting the connection. + c.SendPacket([]byte{1, 2, 3, 4}, &context.Headers{ + SrcPort: context.TestPort, + DstPort: context.StackPort, + Flags: header.TCPFlagAck, + SeqNum: irs + 1, + AckNum: iss + 1, + }) + + // Receive ACK for the data we sent. + checker.IPv4(t, c.GetPacket(), checker.TCP( + checker.DstPort(context.TestPort), + checker.TCPFlags(header.TCPFlagAck), + checker.TCPSeqNum(uint32(iss+1)), + checker.TCPAckNum(uint32(irs+5)))) + }) + } } // TestTCPAckBeforeAcceptV6 tests that once the 3-way handshake is complete, @@ -1219,36 +1351,43 @@ func TestTCPAckBeforeAcceptV4(t *testing.T) { // // This test uses IPv6. func TestTCPAckBeforeAcceptV6(t *testing.T) { - c := context.New(t, e2e.DefaultMTU) - defer c.Cleanup() + for _, cookieEnabled := range []tcpip.TCPAlwaysUseSynCookies{false, true} { + t.Run(fmt.Sprintf("syn-cookies enabled: %t", cookieEnabled), func(t *testing.T) { + c := context.New(t, e2e.DefaultMTU) + defer c.Cleanup() - c.CreateV6Endpoint(true) + if err := c.Stack().SetTransportProtocolOption(header.TCPProtocolNumber, &cookieEnabled); err != nil { + panic(fmt.Sprintf("SetTransportProtocolOption(%d, %T) = %s", header.TCPProtocolNumber, cookieEnabled, err)) + } + c.CreateV6Endpoint(true) - if err := c.EP.Bind(tcpip.FullAddress{Port: context.StackPort}); err != nil { - t.Fatal("Bind failed:", err) + if err := c.EP.Bind(tcpip.FullAddress{Port: context.StackPort}); err != nil { + t.Fatal("Bind failed:", err) + } + + if err := c.EP.Listen(10); err != nil { + t.Fatal("Listen failed:", err) + } + + irs, iss := executeV6Handshake(t, c, context.TestPort, bool(cookieEnabled)) + + // Send data before accepting the connection. + c.SendV6Packet([]byte{1, 2, 3, 4}, &context.Headers{ + SrcPort: context.TestPort, + DstPort: context.StackPort, + Flags: header.TCPFlagAck, + SeqNum: irs + 1, + AckNum: iss + 1, + }) + + // Receive ACK for the data we sent. + checker.IPv6(t, c.GetV6Packet(), checker.TCP( + checker.DstPort(context.TestPort), + checker.TCPFlags(header.TCPFlagAck), + checker.TCPSeqNum(uint32(iss+1)), + checker.TCPAckNum(uint32(irs+5)))) + }) } - - if err := c.EP.Listen(10); err != nil { - t.Fatal("Listen failed:", err) - } - - irs, iss := executeV6Handshake(t, c, context.TestPort, false /* synCookiesInUse */) - - // Send data before accepting the connection. - c.SendV6Packet([]byte{1, 2, 3, 4}, &context.Headers{ - SrcPort: context.TestPort, - DstPort: context.StackPort, - Flags: header.TCPFlagAck, - SeqNum: irs + 1, - AckNum: iss + 1, - }) - - // Receive ACK for the data we sent. - checker.IPv6(t, c.GetV6Packet(), checker.TCP( - checker.DstPort(context.TestPort), - checker.TCPFlags(header.TCPFlagAck), - checker.TCPSeqNum(uint32(iss+1)), - checker.TCPAckNum(uint32(irs+5)))) } func TestSendRstOnListenerRxAckV4(t *testing.T) { @@ -1452,7 +1591,7 @@ func TestListenCloseWhileConnect(t *testing.T) { c.WQ.EventRegister(&waitEntry) defer c.WQ.EventUnregister(&waitEntry) - executeHandshake(t, c, context.TestPort, false /* synCookiesInUse */) + executeHandshake(t, c, context.TestPort, true /* synCookiesInUse */) // Wait for the new endpoint created because of handshake to be delivered // to the listening endpoint's accept queue. <-notifyCh @@ -5624,7 +5763,9 @@ func TestKeepalive(t *testing.T) { func executeHandshake(t *testing.T, c *context.Context, srcPort uint16, synCookieInUse bool) (irs, iss seqnum.Value) { t.Helper() + // Send a SYN request. + options := []byte{header.TCPOptionWS, 3, 0, header.TCPOptionNOP} irs = seqnum.Value(context.TestInitialSequenceNumber) c.SendPacket(nil, &context.Headers{ SrcPort: srcPort, @@ -5632,12 +5773,13 @@ func executeHandshake(t *testing.T, c *context.Context, srcPort uint16, synCooki Flags: header.TCPFlagSyn, SeqNum: irs, RcvWnd: 30000, + TCPOpts: options, }) // Receive the SYN-ACK reply. b := c.GetPacket() - tcp := header.TCP(header.IPv4(b).Payload()) - iss = seqnum.Value(tcp.SequenceNumber()) + tcpHdr := header.TCP(header.IPv4(b).Payload()) + iss = seqnum.Value(tcpHdr.SequenceNumber()) tcpCheckers := []checker.TransportChecker{ checker.SrcPort(context.StackPort), checker.DstPort(srcPort), @@ -5651,6 +5793,11 @@ func executeHandshake(t *testing.T, c *context.Context, srcPort uint16, synCooki WS: -1, MSS: c.MSSWithoutOptions(), })) + } else { + tcpCheckers = append(tcpCheckers, checker.TCPSynOptions(header.TCPSynOptions{ + WS: tcp.FindWndScale(tcp.DefaultReceiveBufferSize), + MSS: c.MSSWithoutOptions(), + })) } checker.IPv4(t, b, checker.TCP(tcpCheckers...)) @@ -5669,7 +5816,9 @@ func executeHandshake(t *testing.T, c *context.Context, srcPort uint16, synCooki func executeV6Handshake(t *testing.T, c *context.Context, srcPort uint16, synCookieInUse bool) (irs, iss seqnum.Value) { t.Helper() + // Send a SYN request. + options := []byte{header.TCPOptionWS, 3, 0, header.TCPOptionNOP} irs = seqnum.Value(context.TestInitialSequenceNumber) c.SendV6Packet(nil, &context.Headers{ SrcPort: srcPort, @@ -5677,12 +5826,13 @@ func executeV6Handshake(t *testing.T, c *context.Context, srcPort uint16, synCoo Flags: header.TCPFlagSyn, SeqNum: irs, RcvWnd: 30000, + TCPOpts: options, }) // Receive the SYN-ACK reply. b := c.GetV6Packet() - tcp := header.TCP(header.IPv6(b).Payload()) - iss = seqnum.Value(tcp.SequenceNumber()) + tcpHdr := header.TCP(header.IPv6(b).Payload()) + iss = seqnum.Value(tcpHdr.SequenceNumber()) tcpCheckers := []checker.TransportChecker{ checker.SrcPort(context.StackPort), checker.DstPort(srcPort), @@ -5696,6 +5846,11 @@ func executeV6Handshake(t *testing.T, c *context.Context, srcPort uint16, synCoo WS: -1, MSS: c.MSSWithoutOptionsV6(), })) + } else { + tcpCheckers = append(tcpCheckers, checker.TCPSynOptions(header.TCPSynOptions{ + WS: tcp.FindWndScale(tcp.DefaultReceiveBufferSize), + MSS: c.MSSWithoutOptionsV6(), + })) } checker.IPv6(t, b, checker.TCP(tcpCheckers...)) @@ -6151,7 +6306,7 @@ func TestListenBacklogFullSynCookieInUse(t *testing.T) { t.Fatalf("Listen failed: %s", err) } - executeHandshake(t, c, context.TestPort, false) + executeHandshake(t, c, context.TestPort, true) // Wait for this to be delivered to the accept queue. time.Sleep(50 * time.Millisecond) @@ -6375,7 +6530,7 @@ func TestPassiveConnectionAttemptIncrement(t *testing.T) { want := stats.TCP.PassiveConnectionOpenings.Value() + 1 srcPort := uint16(context.TestPort) - executeHandshake(t, c, srcPort+1, false) + executeHandshake(t, c, srcPort+1, true /* synCookiesInUse */) we, ch := waiter.NewChannelEntry(waiter.ReadableEvents) c.WQ.EventRegister(&we) @@ -6421,7 +6576,7 @@ func TestPassiveFailedConnectionAttemptIncrement(t *testing.T) { srcPort := uint16(context.TestPort) // Now attempt a handshakes it will fill up the accept backlog. - executeHandshake(t, c, srcPort, false) + executeHandshake(t, c, srcPort, true /* synCookesInUse */) // Give time for the final ACK to be processed as otherwise the next handshake could // get accepted before the previous one based on goroutine scheduling. diff --git a/pkg/tcpip/transport/tcp/testing/context/context.go b/pkg/tcpip/transport/tcp/testing/context/context.go index 549590526..e58d8b40d 100644 --- a/pkg/tcpip/transport/tcp/testing/context/context.go +++ b/pkg/tcpip/transport/tcp/testing/context/context.go @@ -650,13 +650,14 @@ func (c *Context) SendV6Packet(payload []byte, h *Headers) { // addresses. func (c *Context) SendV6PacketWithAddrs(payload []byte, h *Headers, src, dst tcpip.Address) { // Allocate a buffer for data and headers. - buf := buffer.NewView(header.TCPMinimumSize + header.IPv6MinimumSize + len(payload)) + buf := buffer.NewView(header.TCPMinimumSize + header.IPv6MinimumSize + len(h.TCPOpts) + len(payload)) copy(buf[len(buf)-len(payload):], payload) + copy(buf[len(buf)-len(payload)-len(h.TCPOpts):], h.TCPOpts) // Initialize the IP header. ip := header.IPv6(buf) ip.Encode(&header.IPv6Fields{ - PayloadLength: uint16(header.TCPMinimumSize + len(payload)), + PayloadLength: uint16(header.TCPMinimumSize + len(h.TCPOpts) + len(payload)), TransportProtocol: tcp.ProtocolNumber, HopLimit: 65, SrcAddr: src, @@ -670,7 +671,7 @@ func (c *Context) SendV6PacketWithAddrs(payload []byte, h *Headers, src, dst tcp DstPort: h.DstPort, SeqNum: uint32(h.SeqNum), AckNum: uint32(h.AckNum), - DataOffset: header.TCPMinimumSize, + DataOffset: uint8(header.TCPMinimumSize + len(h.TCPOpts)), Flags: h.Flags, WindowSize: uint16(h.RcvWnd), })